diff --git a/docs/serving/online_serving/derenderer.md b/docs/serving/online_serving/derenderer.md
index 9585031312..f85433a963 100644
--- a/docs/serving/online_serving/derenderer.md
+++ b/docs/serving/online_serving/derenderer.md
@@ -105,7 +105,7 @@ The server keeps no state between calls. Everything the next call needs is in `s
 - **Send `prompt_token_ids` on the first chunk, even without a parser.** It seeds detokenization so the first token's leading space matches the coupled endpoint (see [Request format](#request-format)). Later chunks already carry that context in `stream_state`.
 - **Don't forward `[DONE]`.** It marks the end of the generate stream and isn't a chunk.
 
-Streaming derender resolves `logprobs` on the plain detokenization path, including `token_id:N` placeholders, same as the non-streaming endpoints (see [Streaming state and logprobs](#streaming-state-and-logprobs)). The parser path doesn't resolve them yet and drops them.
+`/inference/v1/generate` returns `GenerateLogProbs` with integer `token_id`s (the generate server has no tokenizer), and derender turns those into `ChatCompletionLogProbs` / `CompletionLogProbs`, filling `token` and `bytes` from the tokenizer with the usual U+FFFD byte-fallback correction. Streaming derender does the same per chunk on the plain detokenization path (see [Streaming state and logprobs](#streaming-state-and-logprobs)). The parser path doesn't resolve them yet and drops them.
 
 ## Streaming cost
 
@@ -129,7 +129,7 @@ Streaming derender is stateless on the server side: all mutable state lives in t
 - `logprob_context_token_ids`: the trailing sampled token IDs (at most 4) from previous chunks, used to seed byte-fallback (U+FFFD) correction so multi-byte characters whose tokens split across chunk boundaries still resolve to real strings
 - `logprob_text_offset`: the cumulative emitted text length, so `text_offset` in completion streaming logprobs stays absolute across chunks instead of restarting at 0
 
-When a streamed `GenerateTokensStreamChoice` carries `logprobs`, the `token_id:N` placeholders are resolved per chunk and the resolved logprobs are attached to the corresponding streamed choice. For chat it is as `ChatCompletionLogProbs` and for completions it is converted to the flat `CompletionLogProbs` lists. Chunks without `logprobs` produce choices with `logprobs: null`.
+When a streamed `GenerateTokensStreamChoice` carries `logprobs`, the integer `token_id`s are decoded per chunk and the resolved logprobs are attached to the corresponding streamed choice. For chat it is as `ChatCompletionLogProbs` and for completions it is converted to the flat `CompletionLogProbs` lists. Chunks without `logprobs` produce choices with `logprobs: null`.
 
 ## Example
 
diff --git a/docs/serving/online_serving/renderer.md b/docs/serving/online_serving/renderer.md
index fc6579c258..79d0ade85b 100644
--- a/docs/serving/online_serving/renderer.md
+++ b/docs/serving/online_serving/renderer.md
@@ -79,6 +79,60 @@ for the existing authentication boundaries.
 
 For the post processing counterpart that turns generated token IDs back into OpenAI compatible responses, see the [Derenderer APIs](derenderer.md).
 
+## Generate Output Logprobs
+
+`/inference/v1/generate` is a token in / token out API, so its output logprobs
+identify tokens by integer ID rather than by the OpenAI string token. With
+`sampling_params.logprobs` set and `output_mode: "tokens"` (the default), each
+choice carries a `GenerateLogProbs`; `output_mode: "text"` returns decoded
+`ChatCompletionLogProbs` instead (see [token in, token out](token_in_token_out.md)):
+
+```json
+{
+  "logprobs": {
+    "content": [
+      {
+        "token_id": 262,
+        "logprob": -0.10,
+        "rank": 1,
+        "top_logprobs": [
+          {"token_id": 262, "logprob": -0.10, "rank": 1},
+          {"token_id": 257, "logprob": -1.20, "rank": 2}
+        ]
+      }
+    ]
+  }
+}
+```
+
+- `content` has one entry per generated token, in generation order.
+- `top_logprobs` is a list, not a dict: JSON turns dict keys into strings and
+  the ordering would be implicit. It follows the engine's order: the sampled
+  token first, then the remaining candidates in rank order. With non-greedy
+  sampling the sampled token can sit outside the top k (for example ranks
+  `[5, 1, 2]` at `logprobs=2`); it then takes one of the `logprobs` slots and
+  the rank-k candidate is left out, as on the OpenAI endpoints.
+- `rank` is the token's rank in the vocabulary distribution (1 = most likely)
+  on every entry, the sampled one included; a top-k candidate's rank is its
+  top-k position. The list is not sorted by it, so sort by `rank` if you need
+  rank order. It is `null` when the engine could not rank the token (a NaN
+  logprob, which is sent as `-9999.0`).
+- There is no `token` or `bytes` field. The generate server has no tokenizer;
+  [derender](derenderer.md) fills those in when it converts the response to the
+  OpenAI shapes.
+- `prompt_logprobs` on the same response is unchanged
+  (`list[dict[int, Logprob] | None]`).
+
+!!! warning "Changed in this release"
+    Output logprobs used to be `ChatCompletionLogProbs` with every token written
+    as a `"token_id:N"` placeholder string, and `bytes` set to the UTF-8 bytes of
+    that placeholder by the Rust frontend but left unset by the Python one.
+    Clients that read only `content[i].logprob` are unaffected. Clients that
+    parsed the placeholder should read `content[i].token_id` instead.
+    `return_tokens_as_token_ids` on `/v1/chat/completions` and `/v1/completions`
+    is unchanged: it is a user-facing OpenAI option and still uses the
+    `token_id:N` format.
+
 ## Multimodal Render Features
 
 Multimodal render responses include a `features` object with per-modality
diff --git a/docs/serving/online_serving/token_in_token_out.md b/docs/serving/online_serving/token_in_token_out.md
index b4e0d465bd..e22d490615 100644
--- a/docs/serving/online_serving/token_in_token_out.md
+++ b/docs/serving/online_serving/token_in_token_out.md
@@ -43,7 +43,7 @@ print(response["choices"][0]["token_ids"])
 
 ### Logprobs
 
-With `output_mode: "tokens"`, logprob entries carry `token_id:N` placeholders. With `output_mode: "text"`, they carry the decoded token strings and their UTF-8 `bytes` for the sampled token and every entry in `top_logprobs`. A server started with `--return-tokens-as-token-ids` always returns the placeholders, as `/v1/completions` does.
+With `output_mode: "tokens"`, logprobs are `GenerateLogProbs` with an integer `token_id` and `rank` per entry and no `token` or `bytes` (see [Generate Output Logprobs](renderer.md#generate-output-logprobs)). With `output_mode: "text"`, they are `ChatCompletionLogProbs` carrying the decoded token strings and their UTF-8 `bytes` for the sampled token and every entry in `top_logprobs`; a server started with `--return-tokens-as-token-ids` returns `token_id:N` placeholders there instead, as `/v1/completions` does.
 
 ### Errors
 
diff --git a/rust/src/server/src/routes/inference/generate.rs b/rust/src/server/src/routes/inference/generate.rs
index 0e682f7c45..99d7aec1b7 100644
--- a/rust/src/server/src/routes/inference/generate.rs
+++ b/rust/src/server/src/routes/inference/generate.rs
@@ -5,7 +5,7 @@ mod convert;
 mod types;
 mod validate;
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::convert::Infallible;
 use std::result::Result;
 use std::sync::Arc;
@@ -31,16 +31,17 @@ use vllm_llm::{
 
 use self::convert::{ResponseOptions, prepare_generate_request};
 use self::types::{
-    GenerateLogprob, GenerateResponse, GenerateResponseChoice, GenerateResponseStreamChoice,
-    GenerateStreamResponse, MultiModalPlaceholders, PerRequestMetrics, PlaceholderRangeInfo,
-    SpeculativeDecodingMetrics, StreamingSpeculativeDecodingMetrics,
+    GenerateLogProb, GenerateLogProbs, GenerateLogProbsContent, GenerateLogprob, GenerateResponse,
+    GenerateResponseChoice, GenerateResponseStreamChoice, GenerateStreamResponse,
+    MultiModalPlaceholders, PerRequestMetrics, PlaceholderRangeInfo, SpeculativeDecodingMetrics,
+    StreamingSpeculativeDecodingMetrics,
 };
 pub(crate) use self::types::{GenerateRequest, GenerateSamplingParams};
 pub(crate) use self::validate::validate_request_compat;
 use crate::config::ApiServerOptions;
 use crate::error::{ApiError, bail_server_error, server_error, text_submit_error};
 use crate::routes::openai::utils::logprobs::clamp_logprob;
-use crate::routes::openai::utils::types::{ChatLogProbs, ChatLogProbsContent, TopLogProb, Usage};
+use crate::routes::openai::utils::types::Usage;
 use crate::routes::openai::utils::validated_json::ValidatedJson;
 use crate::state::AppState;
 use crate::utils::{resolve_request_context, sse_response};
@@ -157,7 +158,7 @@ async fn generate_chunk_stream(
     ResponseOptions {
         include_usage,
         include_continuous_usage,
-        include_logprobs,
+        logprobs: output_logprobs,
         // Ignored: raw generate streaming has no prompt-logprobs wire shape.
         include_prompt_logprobs: _,
         return_token_ids,
@@ -212,13 +213,15 @@ async fn generate_chunk_stream(
                     continue;
                 }
 
-                let logprobs = if include_logprobs && !token_ids.is_empty() {
+                let logprobs = if let Some(requested) = output_logprobs
+                    && !token_ids.is_empty()
+                {
                     let logprobs = output.logprobs.as_ref().ok_or_else(|| {
                         server_error!(
                             "raw generate stream requested logprobs but generation returned none"
                         )
                     })?;
-                    Some(raw_logprobs_to_openai_chat(logprobs)?)
+                    Some(raw_logprobs_to_generate(logprobs, requested)?)
                 } else {
                     None
                 };
@@ -282,19 +285,19 @@ fn collect_generate(
         include_usage: _,
         // Ignored: continuous usage is a streaming-only option.
         include_continuous_usage: _,
-        include_logprobs,
+        logprobs: output_logprobs,
         include_prompt_logprobs,
         return_token_ids,
     }: ResponseOptions,
     mm_placeholders: Option<MultiModalPlaceholders>,
 ) -> Result<GenerateResponse, ApiError> {
-    let logprobs = if include_logprobs {
+    let logprobs = if let Some(requested) = output_logprobs {
         let logprobs = collected.logprobs.as_ref().ok_or_else(|| {
             ApiError::server_error(
                 "raw generate response requested logprobs but generation returned none".to_string(),
             )
         })?;
-        Some(raw_logprobs_to_openai_chat(logprobs)?)
+        Some(raw_logprobs_to_generate(logprobs, requested)?)
     } else {
         None
     };
@@ -371,14 +374,17 @@ fn extract_mm_placeholders(features: Option<&[MmFeatureSpec]>) -> Option<MultiMo
     Some(placeholders)
 }
 
-fn raw_logprobs_to_openai_chat(logprobs: &Logprobs) -> Result<ChatLogProbs, ApiError> {
+fn raw_logprobs_to_generate(
+    logprobs: &Logprobs,
+    requested: i32,
+) -> Result<GenerateLogProbs, ApiError> {
     let content = logprobs
         .positions
         .iter()
-        .map(position_to_chat_logprobs_content)
+        .map(|position| position_to_generate_logprobs_content(position, requested))
         .collect::<Result<Vec<_>, _>>()?;
 
-    Ok(ChatLogProbs {
+    Ok(GenerateLogProbs {
         content: Some(content),
     })
 }
@@ -396,35 +402,47 @@ fn raw_prompt_logprobs_to_maps(
         .collect()
 }
 
-fn position_to_chat_logprobs_content(
+/// `top_logprobs` matches the Python generate server: the engine's entries (the
+/// sampled token first, then ranks 1..k) with repeated token ids dropped, cut to
+/// `max(requested, 1)` entries, or all of them for `-1`.
+fn position_to_generate_logprobs_content(
     position: &PositionLogprobs,
-) -> Result<ChatLogProbsContent, ApiError> {
+    requested: i32,
+) -> Result<GenerateLogProbsContent, ApiError> {
     let chosen = position.entries.first().ok_or_else(|| {
         ApiError::server_error(
             "raw generate logprobs position unexpectedly had no token candidates".to_string(),
         )
     })?;
-    let token = format_token_id(chosen.token_id);
 
-    Ok(ChatLogProbsContent {
-        token: token.clone(),
+    // One pass: the first entry of each token id wins, as in Python's dict. The
+    // sampled token is the only one the engine can list twice.
+    let mut seen = HashSet::with_capacity(position.entries.len());
+    Ok(GenerateLogProbsContent {
+        token_id: chosen.token_id,
         logprob: clamp_logprob(chosen.logprob),
-        bytes: Some(token.as_bytes().to_vec()),
+        rank: wire_rank(chosen.rank),
         top_logprobs: position
             .entries
             .iter()
-            .map(|entry| {
-                let token = format_token_id(entry.token_id);
-                TopLogProb {
-                    token: token.clone(),
-                    logprob: clamp_logprob(entry.logprob),
-                    bytes: Some(token.into_bytes()),
-                }
+            .filter(|entry| seen.insert(entry.token_id))
+            .take(usize::try_from(requested).map_or(usize::MAX, |n| n.max(1)))
+            .map(|entry| GenerateLogProb {
+                token_id: entry.token_id,
+                logprob: clamp_logprob(entry.logprob),
+                rank: wire_rank(entry.rank),
             })
             .collect(),
     })
 }
 
+/// The engine reports rank 0 for a sampled token whose logprob is NaN (no value
+/// compares >= NaN). That is not a rank, so it goes on the wire as `None`, the
+/// way `clamp_logprob` handles the logprob itself.
+fn wire_rank(rank: u32) -> Option<u32> {
+    (rank > 0).then_some(rank)
+}
+
 fn position_to_logprob_map(position: &PositionLogprobs) -> HashMap<u32, GenerateLogprob> {
     position
         .entries
@@ -523,6 +541,80 @@ mod tests {
 
     use super::*;
 
+    fn position(entries: &[(u32, f32, u32)]) -> PositionLogprobs {
+        PositionLogprobs {
+            entries: entries
+                .iter()
+                .map(|&(token_id, logprob, rank)| {
+                    vllm_engine_core_client::protocol::logprobs::TokenLogprob {
+                        token_id,
+                        logprob,
+                        rank,
+                    }
+                })
+                .collect(),
+        }
+    }
+
+    fn top(position: &PositionLogprobs, requested: i32) -> Vec<(u32, Option<u32>)> {
+        position_to_generate_logprobs_content(position, requested)
+            .unwrap()
+            .top_logprobs
+            .iter()
+            .map(|entry| (entry.token_id, entry.rank))
+            .collect()
+    }
+
+    #[test]
+    fn generate_top_logprobs_drop_the_repeated_sampled_token() {
+        // Sampled token 7 is also top-1: the engine row is [7, 7, 8].
+        let pos = position(&[(7, -0.1, 1), (7, -0.1, 1), (8, -2.0, 2)]);
+        assert_eq!(top(&pos, 2), vec![(7, Some(1)), (8, Some(2))]);
+    }
+
+    #[test]
+    fn generate_top_logprobs_keep_sampled_first_when_outside_top_k() {
+        // Sampled token 50 has vocab rank 5; it takes a slot and rank 2 drops
+        // out, as on the Python generate server and the OpenAI endpoints.
+        let pos = position(&[(50, -3.0, 5), (10, -0.2, 1), (20, -1.0, 2)]);
+        assert_eq!(top(&pos, 2), vec![(50, Some(5)), (10, Some(1))]);
+    }
+
+    #[test]
+    fn generate_top_logprobs_respect_zero_and_all() {
+        let pos = position(&[(7, -0.1, 1), (7, -0.1, 1), (8, -2.0, 2), (9, -3.0, 3)]);
+        assert_eq!(top(&pos, 0), vec![(7, Some(1))]);
+        assert_eq!(
+            top(&pos, -1),
+            vec![(7, Some(1)), (8, Some(2)), (9, Some(3))]
+        );
+    }
+
+    #[test]
+    fn generate_top_logprobs_full_vocab_is_linear() {
+        // logprobs = -1 returns the whole vocabulary; deduping it must not be
+        // quadratic in the number of entries.
+        const VOCAB: u32 = 200_000;
+        let mut entries = vec![(5, -0.5, 1)];
+        entries.extend((0..VOCAB).map(|t| (t, -1.0 - t as f32, t + 1)));
+        let pos = position(&entries);
+        let start = std::time::Instant::now();
+        let got = top(&pos, -1);
+        assert!(start.elapsed() < std::time::Duration::from_secs(5));
+        assert_eq!(got.len(), VOCAB as usize);
+        assert_eq!(got[0], (5, Some(1)));
+    }
+
+    #[test]
+    fn generate_rank_zero_from_a_nan_logprob_is_none() {
+        // The engine reports rank 0 for a sampled token whose logprob is NaN.
+        let pos = position(&[(7, f32::NAN, 0), (8, -0.2, 1)]);
+        let content = position_to_generate_logprobs_content(&pos, 1).unwrap();
+        assert_eq!(content.rank, None);
+        assert_eq!(content.logprob, -9999.0);
+        assert_eq!(top(&pos, 1), vec![(7, None)]);
+    }
+
     #[tokio::test]
     async fn generate_chunk_stream_captures_late_prompt_info() {
         let stream = stream::iter(vec![
diff --git a/rust/src/server/src/routes/inference/generate/convert.rs b/rust/src/server/src/routes/inference/generate/convert.rs
index 4c54619ed4..36a497f6dd 100644
--- a/rust/src/server/src/routes/inference/generate/convert.rs
+++ b/rust/src/server/src/routes/inference/generate/convert.rs
@@ -26,8 +26,10 @@ pub(super) struct ResponseOptions {
     pub include_usage: bool,
     /// Whether the caller asked for usage on every streamed chunk.
     pub include_continuous_usage: bool,
-    /// Whether the caller requested output logprobs on generate choices.
-    pub include_logprobs: bool,
+    /// Requested output logprobs (`sampling_params.logprobs`): `None` omits
+    /// them, `Some(n)` returns `max(n, 1)` candidates per position, `Some(-1)`
+    /// all of them.
+    pub logprobs: Option<i32>,
     /// Whether the caller requested top-level prompt logprobs.
     pub include_prompt_logprobs: bool,
     /// Whether the caller requested final prompt token metadata.
@@ -56,7 +58,7 @@ pub(super) fn prepare_generate_request(
             .as_ref()
             .and_then(|options| options.continuous_usage_stats)
             .unwrap_or(false);
-    let include_logprobs = request.sampling_params.inner.logprobs.is_some();
+    let logprobs = request.sampling_params.inner.logprobs;
     let include_prompt_logprobs = request.sampling_params.inner.prompt_logprobs.is_some();
     let return_token_ids = request.return_token_ids.unwrap_or(false);
     let mut sampling_params = request.sampling_params.inner;
@@ -96,7 +98,7 @@ pub(super) fn prepare_generate_request(
         options: ResponseOptions {
             include_usage,
             include_continuous_usage,
-            include_logprobs,
+            logprobs,
             include_prompt_logprobs,
             return_token_ids,
         },
diff --git a/rust/src/server/src/routes/inference/generate/types.rs b/rust/src/server/src/routes/inference/generate/types.rs
index b0b93d7080..bd277ef34a 100644
--- a/rust/src/server/src/routes/inference/generate/types.rs
+++ b/rust/src/server/src/routes/inference/generate/types.rs
@@ -10,7 +10,7 @@ use validator::Validate;
 use vllm_engine_core_client::protocol::output::RequestSpecDecodeMetrics;
 use vllm_text::SamplingParams;
 
-use crate::routes::openai::utils::types::{ChatLogProbs, Normalizable, StreamOptions, Usage};
+use crate::routes::openai::utils::types::{Normalizable, StreamOptions, Usage};
 
 /// Sampling parameters for the token-in/token-out generate API.
 ///
@@ -61,7 +61,7 @@ impl Normalizable for GenerateRequest {}
 #[derive(Debug, Clone, Serialize)]
 pub(super) struct GenerateResponseChoice {
     pub index: u32,
-    pub logprobs: Option<ChatLogProbs>,
+    pub logprobs: Option<GenerateLogProbs>,
     pub finish_reason: Option<String>,
     pub token_ids: Vec<u32>,
     pub sampling_mask: Option<Vec<Vec<u32>>>,
@@ -72,7 +72,7 @@ pub(super) struct GenerateResponseChoice {
 #[derive(Debug, Clone, Serialize)]
 pub(super) struct GenerateResponseStreamChoice {
     pub index: u32,
-    pub logprobs: Option<ChatLogProbs>,
+    pub logprobs: Option<GenerateLogProbs>,
     pub finish_reason: Option<String>,
     pub token_ids: Vec<u32>,
     pub sampling_mask: Option<Vec<Vec<u32>>>,
@@ -197,6 +197,32 @@ pub(super) struct PlaceholderRangeInfo {
     pub length: usize,
 }
 
+/// Mirrors the Python vLLM `GenerateLogProbs` class: output logprobs for one
+/// choice, carrying integer token ids rather than the OpenAI string token.
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct GenerateLogProbs {
+    pub content: Option<Vec<GenerateLogProbsContent>>,
+}
+
+/// Mirrors the Python vLLM `GenerateLogProbsContent` class: the sampled token
+/// at one position plus its top-k candidates, in the engine's order: the
+/// sampled token first, then the remaining candidates in rank order.
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct GenerateLogProbsContent {
+    pub token_id: u32,
+    pub logprob: f32,
+    pub rank: Option<u32>,
+    pub top_logprobs: Vec<GenerateLogProb>,
+}
+
+/// Mirrors the Python vLLM `GenerateLogProb` class: one candidate token.
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct GenerateLogProb {
+    pub token_id: u32,
+    pub logprob: f32,
+    pub rank: Option<u32>,
+}
+
 /// Mirrors the Python vLLM `Logprob` class used in prompt-logprobs payloads.
 #[derive(Debug, Clone, Serialize)]
 pub(super) struct GenerateLogprob {
diff --git a/rust/src/server/src/routes/tests.rs b/rust/src/server/src/routes/tests.rs
index 31c328451b..0ed2ea4d45 100644
--- a/rust/src/server/src/routes/tests.rs
+++ b/rust/src/server/src/routes/tests.rs
@@ -4453,13 +4453,15 @@ async fn non_stream_raw_generate_returns_token_output_envelope() {
     assert_eq!(json["choices"][0]["index"], 0);
     assert_eq!(json["choices"][0]["token_ids"], json!([33, 44]));
     assert_eq!(json["choices"][0]["finish_reason"], "stop");
+    // Output logprobs carry integer token ids and rank, not the OpenAI
+    // "token_id:N" placeholder string; `bytes` is gone.
+    assert_eq!(json["choices"][0]["logprobs"]["content"][0]["token_id"], 33);
+    assert_eq!(json["choices"][0]["logprobs"]["content"][0]["rank"], 1);
+    assert!(json["choices"][0]["logprobs"]["content"][0]["token"].is_null());
+    assert!(json["choices"][0]["logprobs"]["content"][0]["bytes"].is_null());
     assert_eq!(
-        json["choices"][0]["logprobs"]["content"][0]["token"],
-        "token_id:33"
-    );
-    assert_eq!(
-        json["choices"][0]["logprobs"]["content"][1]["top_logprobs"][0]["token"],
-        "token_id:44"
+        json["choices"][0]["logprobs"]["content"][1]["top_logprobs"][0]["token_id"],
+        44
     );
     assert_eq!(json["prompt_logprobs"][0], serde_json::Value::Null);
     assert_eq!(
@@ -4581,9 +4583,11 @@ async fn stream_raw_generate_returns_sse_chunks_and_usage() {
     assert_eq!(first["choices"][0]["index"], 0);
     assert_eq!(first["choices"][0]["token_ids"], json!([33]));
     assert_eq!(
-        first["choices"][0]["logprobs"]["content"][0]["token"],
-        "token_id:33"
+        first["choices"][0]["logprobs"]["content"][0]["token_id"],
+        33
     );
+    assert_eq!(first["choices"][0]["logprobs"]["content"][0]["rank"], 1);
+    assert!(first["choices"][0]["logprobs"]["content"][0]["token"].is_null());
     assert_eq!(first["usage"]["prompt_tokens"], 2);
     assert_eq!(first["usage"]["completion_tokens"], 1);
 
diff --git a/tests/entrypoints/scale_out/derender/test_derender.py b/tests/entrypoints/scale_out/derender/test_derender.py
index 40b051304e..d29aa3f53f 100644
--- a/tests/entrypoints/scale_out/derender/test_derender.py
+++ b/tests/entrypoints/scale_out/derender/test_derender.py
@@ -74,14 +74,13 @@ def _make_generate_response(
     }
 
 
-def _make_logprobs_with_placeholders(token_id: int = 1234) -> dict:
+def _make_generate_logprobs(token_id: int = 1234) -> dict:
+    """A `GenerateLogProbs` payload: integer token ids, no token/bytes."""
     entry = {
-        "token": f"token_id:{token_id}",
+        "token_id": token_id,
         "logprob": -1.0,
-        "bytes": None,
-        "top_logprobs": [
-            {"token": f"token_id:{token_id + 1}", "logprob": -2.0, "bytes": None}
-        ],
+        "rank": 1,
+        "top_logprobs": [{"token_id": token_id + 1, "logprob": -2.0, "rank": 2}],
     }
     return {"content": [entry]}
 
@@ -154,7 +153,7 @@ async def test_derender_chat_usage_default(client):
 
 @pytest.mark.asyncio
 async def test_derender_chat_logprobs(client):
-    """token_id:N placeholders in content.token are resolved to real strings."""
+    """Integer token ids are decoded to real strings by derender."""
     gen_req = await _render_chat(client)
     synthetic_ids = gen_req["token_ids"][:3]
     token_id = synthetic_ids[0]
@@ -165,7 +164,7 @@ async def test_derender_chat_logprobs(client):
             "model": MODEL_NAME,
             "generate_response": _make_generate_response(
                 synthetic_ids,
-                logprobs=_make_logprobs_with_placeholders(token_id),
+                logprobs=_make_generate_logprobs(token_id),
             ),
         },
     )
@@ -176,8 +175,8 @@ async def test_derender_chat_logprobs(client):
     content = logprobs["content"]
     assert content is not None and len(content) == 1
     token_str = content[0]["token"]
-    assert not token_str.startswith("token_id:"), (
-        f"Placeholder was not resolved: {token_str!r}"
+    assert token_str and not token_str.startswith("token_id:"), (
+        f"token id was not decoded: {token_str!r}"
     )
 
 
@@ -194,7 +193,7 @@ async def test_derender_chat_logprobs_bytes(client):
             "model": MODEL_NAME,
             "generate_response": _make_generate_response(
                 synthetic_ids,
-                logprobs=_make_logprobs_with_placeholders(token_id),
+                logprobs=_make_generate_logprobs(token_id),
             ),
         },
     )
@@ -208,7 +207,7 @@ async def test_derender_chat_logprobs_bytes(client):
 
 @pytest.mark.asyncio
 async def test_derender_chat_top_logprobs(client):
-    """top_logprobs entries also have their placeholders resolved."""
+    """top_logprobs entries are decoded too."""
     gen_req = await _render_chat(client)
     synthetic_ids = gen_req["token_ids"][:3]
     token_id = synthetic_ids[0]
@@ -219,7 +218,7 @@ async def test_derender_chat_top_logprobs(client):
             "model": MODEL_NAME,
             "generate_response": _make_generate_response(
                 synthetic_ids,
-                logprobs=_make_logprobs_with_placeholders(token_id),
+                logprobs=_make_generate_logprobs(token_id),
             ),
         },
     )
@@ -228,7 +227,7 @@ async def test_derender_chat_top_logprobs(client):
     top = content[0]["top_logprobs"]
     assert len(top) == 1
     assert not top[0]["token"].startswith("token_id:"), (
-        f"top_logprobs placeholder not resolved: {top[0]['token']!r}"
+        f"top_logprobs token id not decoded: {top[0]['token']!r}"
     )
 
 
@@ -633,8 +632,8 @@ async def test_derender_completion_empty_generate_responses(client):
 
 @pytest.mark.asyncio
 async def test_derender_completion_logprobs(client):
-    """token_id:N placeholders in logprobs are resolved; CompletionLogProbs
-    flat-list structure is returned with non-empty tokens and text_offsets."""
+    """Integer token ids are decoded; CompletionLogProbs flat-list structure is
+    returned with non-empty tokens and text_offsets."""
     gr1 = await _render_completion(client, "Hello world")
     ids1 = gr1["token_ids"][:3]
     token_id = ids1[0]
@@ -647,7 +646,7 @@ async def test_derender_completion_logprobs(client):
                 _make_completion_generate_response(
                     ids1,
                     gr1["request_id"],
-                    logprobs=_make_logprobs_with_placeholders(token_id),
+                    logprobs=_make_generate_logprobs(token_id),
                 ),
             ],
         },
@@ -658,7 +657,7 @@ async def test_derender_completion_logprobs(client):
     tokens = logprobs["tokens"]
     assert len(tokens) == 1
     assert not tokens[0].startswith("token_id:"), (
-        f"Placeholder was not resolved: {tokens[0]!r}"
+        f"token id was not decoded: {tokens[0]!r}"
     )
     assert len(logprobs["token_logprobs"]) == 1
     assert isinstance(logprobs["token_logprobs"][0], float)
@@ -794,7 +793,7 @@ async def test_derender_chat_oversized_logprobs_rejected(client):
     """logprobs.content longer than max_model_len returns 400."""
     oversized_logprobs: dict = {
         "content": [
-            {"token": "x", "logprob": -1.0, "bytes": None, "top_logprobs": []}
+            {"token_id": 42, "logprob": -1.0, "rank": 1, "top_logprobs": []}
             for _ in range(1_000_000)
         ]
     }
@@ -825,11 +824,11 @@ async def test_derender_chat_oversized_top_logprobs_rejected(client):
     oversized_top_logprobs = {
         "content": [
             {
-                "token": "x",
+                "token_id": 42,
                 "logprob": -1.0,
-                "bytes": None,
+                "rank": 1,
                 "top_logprobs": [
-                    {"token": f"t{i}", "logprob": -float(i), "bytes": None}
+                    {"token_id": i, "logprob": -float(i), "rank": i + 1}
                     for i in range(25)
                 ],
             }
diff --git a/tests/entrypoints/scale_out/derender/test_derender_parity.py b/tests/entrypoints/scale_out/derender/test_derender_parity.py
index 66d3e83969..4ea3bc2461 100644
--- a/tests/entrypoints/scale_out/derender/test_derender_parity.py
+++ b/tests/entrypoints/scale_out/derender/test_derender_parity.py
@@ -237,21 +237,47 @@ async def test_parity_reasoning_and_tool_call(client):
     assert _tool_sig(disagg["choices"][0])
 
 
+def _to_generate_logprobs(openai_logprobs: dict) -> dict:
+    """Turn a coupled response's `token_id:N` logprobs into `GenerateLogProbs`.
+
+    A real GPU-less worker emits integer ids directly; here the coupled server
+    with `return_tokens_as_token_ids=True` is the only source of the same ids,
+    so the placeholders are parsed back out to build the generate-shaped
+    payload that `/derender` now consumes.
+    """
+
+    def _id(token: str) -> int:
+        assert token.startswith("token_id:"), token
+        return int(token.removeprefix("token_id:"))
+
+    return {
+        "content": [
+            {
+                "token_id": _id(entry["token"]),
+                "logprob": entry["logprob"],
+                "top_logprobs": [
+                    {"token_id": _id(top["token"]), "logprob": top["logprob"]}
+                    for top in entry["top_logprobs"]
+                ],
+            }
+            for entry in openai_logprobs["content"]
+        ]
+    }
+
+
 @pytest.mark.asyncio
 async def test_parity_logprobs(client):
-    """token_id:N resolution parity vs. the coupled server's real strings.
+    """Integer-id logprob decoding parity vs. the coupled server's strings.
 
-    A real disaggregated worker only has token IDs so it emits logprobs
-    with `token_id:N` placeholders (`return_tokens_as_token_ids=True`
-    reproduces that shape here). `/derender` must resolve those
-    placeholders to the same token strings/bytes the coupled server
-    resolves them to directly.
+    A real disaggregated worker only has token IDs, so it emits
+    `GenerateLogProbs` with integer `token_id`s. `/derender` must decode those
+    to the same token strings/bytes the coupled server produces directly.
     """
     messages = [{"role": "user", "content": "What is 2+2?"}]
     extra = {"logprobs": True, "top_logprobs": 3}
 
-    # What a real GPU less worker would hand to /derender is token IDs plus
-    # logprobs still in token_id:N placeholder form
+    # The coupled server with placeholders is used only as a source of the
+    # token ids a GPU-less worker would have had.
     placeholder = await _coupled(
         client, messages, return_tokens_as_token_ids=True, **extra
     )
@@ -263,7 +289,7 @@ async def test_parity_logprobs(client):
         placeholder["usage"]["prompt_tokens"],
         ch["finish_reason"],
         chat_request,
-        logprobs=ch["logprobs"],
+        logprobs=_to_generate_logprobs(ch["logprobs"]),
     )
 
     # The coupled server resolving the same greedy generation
diff --git a/tests/entrypoints/scale_out/derender/test_derender_stream.py b/tests/entrypoints/scale_out/derender/test_derender_stream.py
index 0e4b168677..0a7d96f3cf 100644
--- a/tests/entrypoints/scale_out/derender/test_derender_stream.py
+++ b/tests/entrypoints/scale_out/derender/test_derender_stream.py
@@ -64,6 +64,7 @@ from vllm.renderers.online_derenderer import (
     _seed_stream_state,
 )
 from vllm.tokenizers import get_tokenizer
+from vllm.tokenizers.detokenizer_utils import convert_ids_list_to_tokens
 from vllm.utils import random_uuid
 
 MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
@@ -212,17 +213,12 @@ def _make_stream_chunk(
     )
 
 
-def _placeholder_logprobs(token_ids: list[int]) -> dict:
-    """Per-token logprob entries using token_id:N placeholders, as sent by
-    the generate worker."""
+def _generate_logprobs(token_ids: list[int]) -> dict:
+    """Per-token `GenerateLogProbs` entries (integer token ids, no token or
+    bytes), as sent by the generate worker."""
     return {
         "content": [
-            {
-                "token": f"token_id:{tid}",
-                "logprob": -0.5,
-                "bytes": None,
-                "top_logprobs": [],
-            }
+            {"token_id": tid, "logprob": -0.5, "rank": 1, "top_logprobs": []}
             for tid in token_ids
         ]
     }
@@ -892,12 +888,12 @@ class TestDerenderCompletionStream:
 
 
 class TestStreamLogprobs:
-    """Streaming derender must carry per-chunk logprobs with placeholders
-    resolved, matching what the generate streaming path emits."""
+    """Streaming derender must carry per-chunk logprobs with token ids
+    decoded, matching what the generate streaming path emits."""
 
     @pytest.mark.asyncio
     async def test_chat_stream_logprobs_resolved_per_chunk(self, derenderer, tokenizer):
-        """Each streamed chunk carries logprobs with token_id:N resolved."""
+        """Each streamed chunk carries logprobs with its token ids decoded."""
         token_ids = tokenizer.encode("hello world")[:6]
         mid = len(token_ids) // 2
 
@@ -906,17 +902,15 @@ class TestStreamLogprobs:
             chunk, state = await derenderer.derender_chat_stream(
                 model=MODEL_NAME,
                 generate_chunk=_make_stream_chunk(
-                    part, logprobs=_placeholder_logprobs(part)
+                    part, logprobs=_generate_logprobs(part)
                 ),
                 state=state,
             )
             logprobs = chunk.choices[0].logprobs
             assert logprobs is not None and logprobs.content is not None
-            assert len(logprobs.content) == len(part)
-            for entry in logprobs.content:
-                assert not entry.token.startswith("token_id:"), (
-                    f"placeholder not resolved: {entry.token!r}"
-                )
+            assert [entry.token for entry in logprobs.content] == (
+                convert_ids_list_to_tokens(tokenizer, part)
+            )
 
     @pytest.mark.asyncio
     async def test_chat_stream_logprobs_multibyte_across_chunks(
@@ -936,7 +930,7 @@ class TestStreamLogprobs:
             last_chunk, state = await derenderer.derender_chat_stream(
                 model=MODEL_NAME,
                 generate_chunk=_make_stream_chunk(
-                    [tid], logprobs=_placeholder_logprobs([tid])
+                    [tid], logprobs=_generate_logprobs([tid])
                 ),
                 state=state,
             )
@@ -959,13 +953,13 @@ class TestStreamLogprobs:
         chunk1, state = await derenderer.derender_completion_stream(
             model=MODEL_NAME,
             generate_chunk=_make_stream_chunk(
-                token_ids[:mid], logprobs=_placeholder_logprobs(token_ids[:mid])
+                token_ids[:mid], logprobs=_generate_logprobs(token_ids[:mid])
             ),
         )
         chunk2, _ = await derenderer.derender_completion_stream(
             model=MODEL_NAME,
             generate_chunk=_make_stream_chunk(
-                token_ids[mid:], logprobs=_placeholder_logprobs(token_ids[mid:])
+                token_ids[mid:], logprobs=_generate_logprobs(token_ids[mid:])
             ),
             state=state,
         )
@@ -982,7 +976,7 @@ class TestStreamLogprobs:
             _, state = await derenderer.derender_chat_stream(
                 model=MODEL_NAME,
                 generate_chunk=_make_stream_chunk(
-                    [tid], logprobs=_placeholder_logprobs([tid])
+                    [tid], logprobs=_generate_logprobs([tid])
                 ),
                 state=state,
             )
diff --git a/tests/entrypoints/scale_out/derender/test_generate_logprobs_conversion.py b/tests/entrypoints/scale_out/derender/test_generate_logprobs_conversion.py
new file mode 100644
index 0000000000..96d937ad5a
--- /dev/null
+++ b/tests/entrypoints/scale_out/derender/test_generate_logprobs_conversion.py
@@ -0,0 +1,172 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Unit tests for the generate -> OpenAI logprob conversion in derender.
+
+These exercise `_resolve_logprobs` directly. A stub tokenizer covers the
+byte-fallback correction path; a real SentencePiece tokenizer covers the
+leading-space restoration.
+"""
+
+from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
+    GenerateLogProb,
+    GenerateLogProbs,
+    GenerateLogProbsContent,
+)
+from vllm.renderers.online_derenderer import (
+    _convert_chat_logprobs_to_completion_logprobs,
+    _resolve_logprobs,
+)
+from vllm.tokenizers import get_tokenizer
+from vllm.tokenizers.detokenizer_utils import convert_ids_list_to_tokens
+
+# A two-byte character split across two byte-fallback tokens: neither decodes
+# to anything printable on its own, but the pair decodes to "é".
+FIRST_BYTE = 10
+SECOND_BYTE = 11
+PLAIN = 12
+
+VOCAB = {FIRST_BYTE: "<0xC3>", SECOND_BYTE: "<0xA9>", PLAIN: "ok"}
+
+
+class _StubTokenizer:
+    """Decodes byte-fallback tokens only when both halves are present."""
+
+    def convert_ids_to_tokens(self, ids: list[int]) -> list[str | None]:
+        return [VOCAB.get(i) for i in ids]
+
+    def convert_tokens_to_string(self, tokens: list[str]) -> str:
+        return self.decode([k for t in tokens for k, v in VOCAB.items() if v == t])
+
+    def decode(self, ids: list[int], **kwargs) -> str:
+        out = ""
+        i = 0
+        while i < len(ids):
+            if ids[i] == FIRST_BYTE and i + 1 < len(ids) and ids[i + 1] == SECOND_BYTE:
+                out += "é"
+                i += 2
+            elif ids[i] in (FIRST_BYTE, SECOND_BYTE):
+                out += "�"
+                i += 1
+            else:
+                out += VOCAB.get(ids[i], "")
+                i += 1
+        return out
+
+
+def test_token_ids_are_decoded_with_bytes_and_order_preserved():
+    logprobs = GenerateLogProbs(
+        content=[
+            GenerateLogProbsContent(
+                token_id=PLAIN,
+                logprob=-0.5,
+                rank=1,
+                top_logprobs=[
+                    GenerateLogProb(token_id=PLAIN, logprob=-0.5, rank=1),
+                    GenerateLogProb(token_id=FIRST_BYTE, logprob=-2.0, rank=2),
+                ],
+            )
+        ]
+    )
+
+    resolved = _resolve_logprobs(logprobs, _StubTokenizer())
+
+    assert resolved.content is not None
+    entry = resolved.content[0]
+    assert entry.token == "ok"
+    assert entry.logprob == -0.5
+    assert entry.bytes == list(b"ok")
+    # Rank order survives the conversion (the OpenAI shape has no rank field).
+    assert [t.logprob for t in entry.top_logprobs] == [-0.5, -2.0]
+    assert entry.top_logprobs[0].token == "ok"
+
+
+def test_byte_fallback_second_half_uses_preceding_token_as_context():
+    """U+FFFD correction needs the preceding sampled ids, which the integer
+    shape carries directly (no placeholder parsing)."""
+    logprobs = GenerateLogProbs(
+        content=[
+            GenerateLogProbsContent(token_id=FIRST_BYTE, logprob=-0.1),
+            GenerateLogProbsContent(token_id=SECOND_BYTE, logprob=-0.2),
+        ]
+    )
+
+    resolved = _resolve_logprobs(logprobs, _StubTokenizer())
+
+    assert resolved.content is not None
+    # The first half has no preceding context to repair it with, so it stays
+    # empty (existing `_correct_decoded_token` behaviour). The second resolves
+    # to the complete character once the first is used as context.
+    assert resolved.content[0].token == ""
+    assert resolved.content[0].bytes == []
+    assert resolved.content[1].token == "é"
+    assert resolved.content[1].bytes == list("é".encode())
+
+
+def test_content_none_is_not_an_error():
+    resolved = _resolve_logprobs(GenerateLogProbs(), _StubTokenizer())
+    assert resolved.content is None
+
+
+def test_sentencepiece_leading_space_matches_engine():
+    """`▁true` and `true` are distinct tokens and must stay distinct: the
+    strings match the engine's per-token detokenization (leading space
+    restored), so `/v1/completions/derender` top_logprobs keys do not collide.
+    Regression test for #59241."""
+    tokenizer = get_tokenizer("hmellor/tiny-random-LlamaForCausalLM")
+    spaced = tokenizer.convert_tokens_to_ids("▁true")
+    bare = tokenizer.convert_tokens_to_ids("true")
+    assert spaced != bare
+    logprobs = GenerateLogProbs(
+        content=[
+            GenerateLogProbsContent(
+                token_id=spaced,
+                logprob=-0.5,
+                rank=1,
+                top_logprobs=[
+                    GenerateLogProb(token_id=spaced, logprob=-0.5, rank=1),
+                    GenerateLogProb(token_id=bare, logprob=-1.5, rank=2),
+                ],
+            )
+        ]
+    )
+
+    resolved = _resolve_logprobs(logprobs, tokenizer)
+
+    assert resolved.content is not None
+    entry = resolved.content[0]
+    assert entry.token == " true"
+    assert entry.bytes == list(b" true")
+    assert [t.token for t in entry.top_logprobs] == convert_ids_list_to_tokens(
+        tokenizer, [spaced, bare]
+    )
+    completion = _convert_chat_logprobs_to_completion_logprobs(resolved)
+    assert completion.top_logprobs == [{" true": -0.5, "true": -1.5}]
+
+
+def test_unknown_top_id_does_not_shift_the_other_candidates():
+    """Ids are decoded in one batch per position; an id with no vocab entry
+    becomes an empty token without shifting its neighbours."""
+    unknown = 99
+    logprobs = GenerateLogProbs(
+        content=[
+            GenerateLogProbsContent(
+                token_id=PLAIN,
+                logprob=-0.5,
+                rank=1,
+                top_logprobs=[
+                    GenerateLogProb(token_id=unknown, logprob=-1.0, rank=1),
+                    GenerateLogProb(token_id=PLAIN, logprob=-0.5, rank=2),
+                ],
+            )
+        ]
+    )
+
+    resolved = _resolve_logprobs(logprobs, _StubTokenizer())
+
+    assert resolved.content is not None
+    entry = resolved.content[0]
+    assert entry.token == "ok"
+    assert [(t.token, t.bytes) for t in entry.top_logprobs] == [
+        ("", None),
+        ("ok", list(b"ok")),
+    ]
diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py
index 7e7f74de9f..7f0cd12860 100644
--- a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py
+++ b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py
@@ -1123,7 +1123,8 @@ async def test_stream_with_logprobs():
         lp = dc["choices"][0]["logprobs"]
         assert lp is not None
         assert len(lp["content"]) == 1
-        assert lp["content"][0]["token"].startswith("token_id:")
+        assert isinstance(lp["content"][0]["token_id"], int)
+        assert "token" not in lp["content"][0]
 
 
 @pytest.mark.asyncio
diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py b/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py
index 1027dcef03..0919431aba 100644
--- a/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py
+++ b/tests/entrypoints/scale_out/token_in_token_out/test_serving_tokens.py
@@ -183,7 +183,7 @@ async def test_generate_sampling_mask(client):
         assert token_id in support
         # processed_logprobs: exactly the support carries finite probability.
         for top in entry["top_logprobs"]:
-            in_support = int(top["token"].removeprefix("token_id:")) in support
+            in_support = top["token_id"] in support
             assert in_support == (top["logprob"] > -9999.0)
 
 
diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py b/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py
index c417a5fc2b..05dba167fe 100644
--- a/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py
+++ b/tests/entrypoints/scale_out/token_in_token_out/test_tokens_logprobs.py
@@ -1,22 +1,62 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
+import json
 from unittest.mock import MagicMock
 
+import pytest
+from pydantic import ValidationError
+
+from vllm.entrypoints.scale_out.token_in_token_out.protocol import GenerateLogProb
 from vllm.entrypoints.scale_out.token_in_token_out.serving import ServingTokens
 from vllm.logprobs import Logprob
 
 
 def test_top_logprobs_alternatives_have_own_token_ids():
-    """Each top_logprobs alternative must carry its own token_id placeholder."""
+    """Each top_logprobs alternative must carry its own integer token id."""
     result = ServingTokens._create_tokens_logprobs(
         None,
         token_ids=[262],
         top_logprobs=[{262: Logprob(-0.1), 257: Logprob(-1.2), 428: Logprob(-2.3)}],
         num_output_top_logprobs=2,
     )
-    tokens = {e.token for e in result.content[0].top_logprobs}
-    assert tokens == {"token_id:262", "token_id:257"}, f"got {tokens}"
+    token_ids = {e.token_id for e in result.content[0].top_logprobs}
+    assert token_ids == {262, 257}, f"got {token_ids}"
+
+
+def test_sampled_entry_carries_token_id_and_rank():
+    """The sampled token is identified by id, and the engine's rank survives."""
+    result = ServingTokens._create_tokens_logprobs(
+        None,
+        token_ids=[262],
+        top_logprobs=[
+            {262: Logprob(-0.1, rank=1), 257: Logprob(-1.2, rank=2)},
+        ],
+        num_output_top_logprobs=2,
+    )
+    entry = result.content[0]
+    assert entry.token_id == 262
+    assert entry.logprob == -0.1
+    assert entry.rank == 1
+    assert [(t.token_id, t.rank) for t in entry.top_logprobs] == [(262, 1), (257, 2)]
+    # No tokenizer on the generate server: no string token, no bytes.
+    assert not hasattr(entry, "token")
+    assert not hasattr(entry, "bytes")
+
+
+def test_sampled_token_absent_from_topk_uses_sentinel():
+    """A sampled token missing from the engine's map keeps the -9999.0 sentinel."""
+    result = ServingTokens._create_tokens_logprobs(
+        None,
+        token_ids=[5],
+        top_logprobs=[{7: Logprob(-0.9)}],
+        num_output_top_logprobs=1,
+    )
+    entry = result.content[0]
+    assert entry.token_id == 5
+    assert entry.logprob == -9999.0
+    assert entry.rank is None
+    assert entry.top_logprobs == []
 
 
 def test_logprobs_zero_emits_sampled_token():
@@ -40,21 +80,23 @@ def test_logprobs_minus_one_emits_all_tokens():
     assert len(result.content[0].top_logprobs) == 2
 
 
-def test_tokens_level_logprobs_have_no_bytes():
-    result = ServingTokens._create_tokens_logprobs(
+def test_text_logprobs_without_tokenizer_are_placeholders_without_bytes():
+    """--return-tokens-as-token-ids keeps `token_id:N` placeholders in text mode."""
+    result = ServingTokens._create_text_logprobs(
         None,
         token_ids=[7],
         top_logprobs=[{7: Logprob(-0.9), 8: Logprob(-1.1)}],
         num_output_top_logprobs=2,
     )
     entry = result.content[0]
+    assert entry.token == "token_id:7"
     assert entry.bytes is None
     assert all(t.bytes is None for t in entry.top_logprobs)
 
 
 def test_resolved_logprobs_use_engine_decoded_tokens_and_bytes():
     tokenizer = MagicMock()
-    result = ServingTokens._create_tokens_logprobs(
+    result = ServingTokens._create_text_logprobs(
         None,
         token_ids=[7],
         top_logprobs=[
@@ -75,7 +117,7 @@ def test_resolved_logprobs_use_engine_decoded_tokens_and_bytes():
 def test_resolved_logprobs_decode_tokens_the_engine_did_not():
     tokenizer = MagicMock()
     tokenizer.decode.side_effect = lambda ids: f"<{ids[0]}>"
-    result = ServingTokens._create_tokens_logprobs(
+    result = ServingTokens._create_text_logprobs(
         None,
         token_ids=[7, 9],
         top_logprobs=[{7: Logprob(-0.9)}, None],
@@ -85,3 +127,59 @@ def test_resolved_logprobs_decode_tokens_the_engine_did_not():
     assert [e.token for e in result.content] == ["<7>", "<9>"]
     assert result.content[0].top_logprobs[0].token == "<7>"
     assert result.content[1].bytes == list(b"<9>")
+
+
+def test_sampled_token_outside_topk_comes_first():
+    """The engine puts the sampled token first, then ranks 1..k. When the
+    sampled token is outside the top k it takes one of the k slots, so rank k
+    is left out (same as the OpenAI endpoints)."""
+    result = ServingTokens._create_tokens_logprobs(
+        None,
+        token_ids=[50],
+        top_logprobs=[
+            {
+                50: Logprob(-3.0, rank=5),
+                10: Logprob(-0.2, rank=1),
+                20: Logprob(-1.0, rank=2),
+            }
+        ],
+        num_output_top_logprobs=2,
+    )
+    assert [t.rank for t in result.content[0].top_logprobs] == [5, 1]
+
+
+def test_logprob_is_required_on_the_wire():
+    """A payload missing logprob is rejected rather than read as -9999."""
+    with pytest.raises(ValidationError):
+        GenerateLogProb.model_validate({"token_id": 1})
+
+
+def test_rank_zero_from_a_nan_logprob_is_none():
+    """The engine reports rank 0 for a sampled token whose logprob is NaN."""
+    result = ServingTokens._create_tokens_logprobs(
+        None,
+        token_ids=[7],
+        top_logprobs=[{7: Logprob(float("nan"), rank=0), 8: Logprob(-0.2, rank=1)}],
+        num_output_top_logprobs=1,
+    )
+    entry = result.content[0]
+    assert entry.rank is None
+    assert entry.top_logprobs[0].rank is None
+    # Clamped like the Rust frontend (max(nan, x) is nan in Python).
+    assert entry.logprob == -9999.0
+    assert entry.top_logprobs[0].logprob == -9999.0
+    json.dumps(result.model_dump(), allow_nan=False)
+
+
+def test_text_logprobs_clamp_a_nan_logprob():
+    result = ServingTokens._create_text_logprobs(
+        None,
+        token_ids=[7],
+        top_logprobs=[{7: Logprob(float("nan"), rank=0, decoded_token="a")}],
+        num_output_top_logprobs=1,
+        tokenizer=MagicMock(),
+    )
+    entry = result.content[0]
+    assert entry.logprob == -9999.0
+    assert entry.top_logprobs[0].logprob == -9999.0
+    json.dumps(result.model_dump(), allow_nan=False)
diff --git a/vllm/entrypoints/generate/base/serving.py b/vllm/entrypoints/generate/base/serving.py
index 5142346df6..d1f3e94b00 100644
--- a/vllm/entrypoints/generate/base/serving.py
+++ b/vllm/entrypoints/generate/base/serving.py
@@ -32,6 +32,7 @@ from vllm.logger import init_logger
 from vllm.logprobs import Logprob, PromptLogprobs
 from vllm.lora.request import LoRARequest
 from vllm.tokenizers import TokenizerLike
+from vllm.tokenizers.detokenizer_utils import convert_ids_list_to_tokens
 from vllm.tracing import (
     contains_trace_headers,
     extract_trace_headers,
@@ -346,32 +347,33 @@ def format_token_id_placeholder(token_id: int) -> str:
     return f"token_id:{token_id}"
 
 
-def resolve_token_id_placeholder(
-    token: str, tokenizer: TokenizerLike
-) -> tuple[str, list[int] | None]:
-    """Decode a 'token_id:N' placeholder back to a token string and UTF-8 bytes.
+def decode_token_ids(
+    token_ids: list[int], tokenizer: TokenizerLike
+) -> list[tuple[str, list[int] | None]]:
+    """Decode token ids individually to their token strings and UTF-8 bytes.
 
-    Returns (token, None) unchanged if token is not a placeholder.
-    This is the inverse of format_token_id_placeholder / _get_decoded_token
-    when return_as_token_id=True.
+    Uses the engine's per-token detokenization, which restores the
+    SentencePiece leading space that `convert_tokens_to_string` drops, so the
+    strings match the coupled endpoints. Ids are decoded in one batch (callers
+    pass a position's sampled id together with its top-k ids). An id with no
+    vocab entry decodes to ("", None).
     """
-    suffix = token.removeprefix("token_id:")
-    if suffix == token:
-        return token, None
-    try:
-        token_id = int(suffix)
-    except ValueError:
-        return token, None
-    token_repr = tokenizer.convert_ids_to_tokens([token_id])[0]
-    if token_repr is None:
-        logger.warning_once(
-            "resolve_token_id_placeholder: token_id %d has no vocab entry; "
-            "substituting empty string",
-            token_id,
-        )
-        return "", None
-    token_str = tokenizer.convert_tokens_to_string([token_repr])
-    return token_str, list(token_str.encode("utf-8", errors="replace"))
+    pieces = tokenizer.convert_ids_to_tokens(token_ids)
+    known = [tid for tid, piece in zip(token_ids, pieces) if piece is not None]
+    decoded = iter(convert_ids_list_to_tokens(tokenizer, known))
+    out: list[tuple[str, list[int] | None]] = []
+    for tid, piece in zip(token_ids, pieces):
+        if piece is None:
+            logger.warning_once(
+                "decode_token_ids: token_id %d has no vocab entry; "
+                "substituting empty string",
+                tid,
+            )
+            out.append(("", None))
+            continue
+        token_str = next(decoded)
+        out.append((token_str, list(token_str.encode("utf-8", errors="replace"))))
+    return out
 
 
 def clamp_prompt_logprobs(
diff --git a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py
index 4e29f7e59d..84a6a2c623 100644
--- a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py
+++ b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py
@@ -360,6 +360,39 @@ class GenerateRequest(BaseModel):
         )
 
 
+class GenerateLogProb(BaseModel):
+    """A single (token, logprob) candidate on the generate wire protocol.
+
+    Unlike the OpenAI logprob shapes this carries the integer token id: the
+    generate server has no tokenizer, so decoding to a string belongs in
+    derender (or the coupled chat/completions path), not here.
+    """
+
+    token_id: int
+    logprob: float
+    rank: int | None = None
+
+
+class GenerateLogProbsContent(GenerateLogProb):
+    """The sampled token at one position, plus its top-k candidates.
+
+    ``top_logprobs`` is a list, not a dict: JSON turns dict keys into strings
+    and the order would be implicit. It is in the engine's order: the sampled
+    token first, then the remaining candidates in rank order.
+    """
+
+    top_logprobs: list[GenerateLogProb] = []
+
+
+class GenerateLogProbs(BaseModel):
+    """Output logprobs for one choice.
+
+    ``content`` holds one entry per generated token.
+    """
+
+    content: list[GenerateLogProbsContent] | None = None
+
+
 class GenerateChoiceBase(BaseModel):
     """Fields shared by every `output_mode` of a non-streaming choice."""
 
@@ -388,8 +421,8 @@ class GenerateChoiceBase(BaseModel):
 
 
 class GenerateTokensChoice(GenerateChoiceBase):
-    logprobs: ChatCompletionLogProbs | None = None
-    """Logprobs whose tokens are `token_id:N` placeholders."""
+    logprobs: GenerateLogProbs | None = None
+    """Logprobs with integer token ids (no tokenizer on the generate server)."""
 
 
 class GenerateTextChoice(GenerateChoiceBase):
@@ -413,7 +446,7 @@ class GenerateStreamChoiceBase(BaseModel):
 
 
 class GenerateTokensStreamChoice(GenerateStreamChoiceBase):
-    logprobs: ChatCompletionLogProbs | None = None
+    logprobs: GenerateLogProbs | None = None
 
 
 class GenerateTextStreamChoice(GenerateStreamChoiceBase):
diff --git a/vllm/entrypoints/scale_out/token_in_token_out/serving.py b/vllm/entrypoints/scale_out/token_in_token_out/serving.py
index e7afa0401b..b6d7bf19ba 100644
--- a/vllm/entrypoints/scale_out/token_in_token_out/serving.py
+++ b/vllm/entrypoints/scale_out/token_in_token_out/serving.py
@@ -3,6 +3,7 @@
 
 
 import asyncio
+import math
 import time
 from collections.abc import AsyncGenerator
 from collections.abc import Sequence as GenericSequence
@@ -57,6 +58,9 @@ from .mm_features import (
     placeholder_ranges_from_engine_input,
 )
 from .protocol import (
+    GenerateLogProb,
+    GenerateLogProbs,
+    GenerateLogProbsContent,
     GenerateRequest,
     GenerateResponse,
     GenerateTextChoice,
@@ -72,6 +76,15 @@ from .protocol import (
 logger = init_logger(__name__)
 
 
+def _clamp_logprob(logprob: float) -> float:
+    """The OpenAI shapes' floor for a logprob: ``-inf`` and NaN become ``-9999.0``.
+
+    ``max(nan, -9999.0)`` is NaN in Python, so NaN needs its own check; NaN
+    would otherwise fail ``JSONResponse`` or reach the client as ``null``.
+    """
+    return -9999.0 if math.isnan(logprob) else max(logprob, -9999.0)
+
+
 def _logprob_token(
     token_id: int, logprob: Logprob | None, tokenizer: TokenizerLike | None
 ) -> tuple[str, list[int] | None]:
@@ -94,7 +107,7 @@ def _top_logprob(
 ) -> ChatCompletionLogProb:
     token, token_bytes = _logprob_token(token_id, logprob, tokenizer)
     return ChatCompletionLogProb(
-        token=token, logprob=max(logprob.logprob, -9999.0), bytes=token_bytes
+        token=token, logprob=_clamp_logprob(logprob.logprob), bytes=token_bytes
     )
 
 
@@ -406,12 +419,20 @@ class ServingTokens(GenerateBaseServing):
             # This is top_logprobs in completions API
             if sampling_params.logprobs is not None:
                 assert out_logprobs is not None, "Did not output logprobs"
-                logprobs = self._create_tokens_logprobs(
-                    token_ids=token_ids,
-                    top_logprobs=out_logprobs,
-                    num_output_top_logprobs=sampling_params.logprobs,
-                    tokenizer=tokenizer,
-                )
+                logprobs: GenerateLogProbs | ChatCompletionLogProbs | None
+                if text_mode:
+                    logprobs = self._create_text_logprobs(
+                        token_ids=token_ids,
+                        top_logprobs=out_logprobs,
+                        num_output_top_logprobs=sampling_params.logprobs,
+                        tokenizer=tokenizer,
+                    )
+                else:
+                    logprobs = self._create_tokens_logprobs(
+                        token_ids=token_ids,
+                        top_logprobs=out_logprobs,
+                        num_output_top_logprobs=sampling_params.logprobs,
+                    )
             else:
                 logprobs = None
 
@@ -571,12 +592,20 @@ class ServingTokens(GenerateBaseServing):
                     if sampling_params.logprobs is not None:
                         out_logprobs = output.logprobs
                         assert out_logprobs is not None, "Did not output logprobs"
-                        logprobs = self._create_tokens_logprobs(
-                            token_ids=delta_token_ids,
-                            top_logprobs=out_logprobs,
-                            num_output_top_logprobs=sampling_params.logprobs,
-                            tokenizer=tokenizer,
-                        )
+                        logprobs: GenerateLogProbs | ChatCompletionLogProbs | None
+                        if text_mode:
+                            logprobs = self._create_text_logprobs(
+                                token_ids=delta_token_ids,
+                                top_logprobs=out_logprobs,
+                                num_output_top_logprobs=sampling_params.logprobs,
+                                tokenizer=tokenizer,
+                            )
+                        else:
+                            logprobs = self._create_tokens_logprobs(
+                                token_ids=delta_token_ids,
+                                top_logprobs=out_logprobs,
+                                num_output_top_logprobs=sampling_params.logprobs,
+                            )
                     else:
                         logprobs = None
 
@@ -691,17 +720,18 @@ class ServingTokens(GenerateBaseServing):
             return None
         return self.renderer.tokenizer
 
-    def _create_tokens_logprobs(
+    def _create_text_logprobs(
         self,
         token_ids: GenericSequence[int],
         top_logprobs: GenericSequence[dict[int, Logprob] | None],
         num_output_top_logprobs: int | None = None,
         tokenizer: TokenizerLike | None = None,
     ) -> ChatCompletionLogProbs:
-        """Create OpenAI-style logprobs.
+        """Create OpenAI-style logprobs for ``output_mode="text"``.
 
-        Tokens are ``token_id:N`` placeholders. With a ``tokenizer`` they are
-        decoded strings and carry ``bytes``.
+        With a ``tokenizer`` the tokens are decoded strings and carry ``bytes``;
+        without one (``--return-tokens-as-token-ids``) they are ``token_id:N``
+        placeholders.
         """
         logprobs_content: list[ChatCompletionLogProbsContent] = []
 
@@ -719,7 +749,7 @@ class ServingTokens(GenerateBaseServing):
                 logprobs_content.append(
                     ChatCompletionLogProbsContent(
                         token=token,
-                        logprob=max(step_token.logprob, -9999.0),
+                        logprob=_clamp_logprob(step_token.logprob),
                         bytes=token_bytes,
                         top_logprobs=[
                             _top_logprob(top_id, logprob, tokenizer)
@@ -736,3 +766,52 @@ class ServingTokens(GenerateBaseServing):
                 )
 
         return ChatCompletionLogProbs(content=logprobs_content)
+
+    def _create_tokens_logprobs(
+        self,
+        token_ids: GenericSequence[int],
+        top_logprobs: GenericSequence[dict[int, Logprob] | None],
+        num_output_top_logprobs: int | None = None,
+    ) -> GenerateLogProbs:
+        """Create generate-shaped logprobs (integer token ids, no tokenizer).
+
+        The engine reports rank 0 for a sampled token whose logprob is NaN; that
+        is not a rank, so it is sent as ``None`` (the logprob is clamped).
+        """
+        logprobs_content: list[GenerateLogProbsContent] = []
+
+        for i, token_id in enumerate(token_ids):
+            step_top_logprobs = top_logprobs[i]
+            if step_top_logprobs is None or step_top_logprobs.get(token_id) is None:
+                # Same sentinel the OpenAI shapes use when the sampled token
+                # has no entry in the engine's top-k map.
+                logprobs_content.append(
+                    GenerateLogProbsContent(token_id=token_id, logprob=-9999.0)
+                )
+            else:
+                step_token = step_top_logprobs[token_id]
+
+                logprobs_content.append(
+                    GenerateLogProbsContent(
+                        token_id=token_id,
+                        logprob=_clamp_logprob(step_token.logprob),
+                        rank=step_token.rank or None,
+                        top_logprobs=[
+                            GenerateLogProb(
+                                token_id=top_token_id,
+                                logprob=_clamp_logprob(top_logprob.logprob),
+                                rank=top_logprob.rank or None,
+                            )
+                            for rank_index, (top_token_id, top_logprob) in enumerate(
+                                step_top_logprobs.items()
+                            )
+                            if num_output_top_logprobs is not None
+                            and (
+                                num_output_top_logprobs == -1
+                                or rank_index < max(num_output_top_logprobs, 1)
+                            )
+                        ],
+                    )
+                )
+
+        return GenerateLogProbs(content=logprobs_content)
diff --git a/vllm/renderers/online_derenderer.py b/vllm/renderers/online_derenderer.py
index 1c6875159a..bd5bf88fbe 100644
--- a/vllm/renderers/online_derenderer.py
+++ b/vllm/renderers/online_derenderer.py
@@ -9,9 +9,11 @@ from vllm.entrypoints.generate.base.protocol import (
     DeltaMessage,
     ToolCall,
 )
-from vllm.entrypoints.generate.base.serving import resolve_token_id_placeholder
+from vllm.entrypoints.generate.base.serving import decode_token_ids
 from vllm.entrypoints.openai.chat_completion.protocol import (
+    ChatCompletionLogProb,
     ChatCompletionLogProbs,
+    ChatCompletionLogProbsContent,
     ChatCompletionNamedToolChoiceParam,
     ChatCompletionRequest,
     ChatCompletionResponseChoice,
@@ -28,6 +30,7 @@ from vllm.entrypoints.openai.completion.protocol import (
 )
 from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
     DerenderStreamState,
+    GenerateLogProbs,
     GenerateTokensResponse,
     GenerateTokensStreamResponse,
 )
@@ -963,16 +966,6 @@ def _seed_stream_state(
     )
 
 
-def _parse_token_id_placeholder(token: str) -> int | None:
-    """Extract token ID from a 'token_id:N' placeholder string."""
-    if not token.startswith("token_id:"):
-        return None
-    try:
-        return int(token[len("token_id:") :])
-    except ValueError:
-        return None
-
-
 def _correct_decoded_token(
     token_id: int, context_token_ids: list[int], tokenizer: TokenizerLike
 ) -> str:
@@ -1012,53 +1005,63 @@ def _correct_decoded_token(
 
 
 def _resolve_logprobs(
-    logprobs: ChatCompletionLogProbs,
+    logprobs: GenerateLogProbs,
     tokenizer: TokenizerLike,
     initial_context_token_ids: Sequence[int] = (),
 ) -> ChatCompletionLogProbs:
-    """Resolve token_id:N placeholders in a ChatCompletionLogProbs object.
+    """Convert generate's integer-id logprobs to the OpenAI chat shape.
 
-    ``initial_context_token_ids`` seeds the byte-fallback correction context
-    with sampled IDs from preceding chunks (streaming), so multi-byte
-    characters split across chunk boundaries still resolve.
+    The generate server has no tokenizer, so `token` and `bytes` are filled
+    here, with the same U+FFFD byte-fallback correction the coupled path
+    applies (`_correct_decoded_token`, which needs the preceding sampled token
+    ids as context). ``initial_context_token_ids`` seeds that context with
+    sampled IDs from preceding chunks (streaming), so multi-byte characters
+    split across chunk boundaries still resolve.
     """
     if logprobs.content is None:
-        return logprobs
+        return ChatCompletionLogProbs()
 
     context_token_ids: list[int] = list(initial_context_token_ids)
     resolved_content = []
 
     for entry in logprobs.content:
-        token_str, token_bytes = resolve_token_id_placeholder(entry.token, tokenizer)
-        sampled_id = _parse_token_id_placeholder(entry.token)
+        # One batch per position: the sampled id and its top-k ids.
+        (token_str, token_bytes), *top_decoded = decode_token_ids(
+            [entry.token_id, *(top.token_id for top in entry.top_logprobs)],
+            tokenizer,
+        )
 
-        if token_str.endswith("�") and sampled_id is not None:
-            token_str = _correct_decoded_token(sampled_id, context_token_ids, tokenizer)
+        if token_str.endswith("\ufffd"):
+            token_str = _correct_decoded_token(
+                entry.token_id, context_token_ids, tokenizer
+            )
             token_bytes = list(token_str.encode("utf-8"))
 
         resolved_top = []
-        for top in entry.top_logprobs:
-            top_str, top_bytes = resolve_token_id_placeholder(top.token, tokenizer)
-            top_id = _parse_token_id_placeholder(top.token)
-            if top_str.endswith("�") and top_id is not None:
-                top_str = _correct_decoded_token(top_id, context_token_ids, tokenizer)
+        for top, (top_str, top_bytes) in zip(entry.top_logprobs, top_decoded):
+            if top_str.endswith("\ufffd"):
+                top_str = _correct_decoded_token(
+                    top.token_id, context_token_ids, tokenizer
+                )
                 top_bytes = list(top_str.encode("utf-8"))
             resolved_top.append(
-                top.model_copy(update={"token": top_str, "bytes": top_bytes})
+                ChatCompletionLogProb(
+                    token=top_str,
+                    logprob=top.logprob,
+                    bytes=top_bytes,
+                )
             )
 
         resolved_content.append(
-            entry.model_copy(
-                update={
-                    "token": token_str,
-                    "bytes": token_bytes,
-                    "top_logprobs": resolved_top,
-                }
+            ChatCompletionLogProbsContent(
+                token=token_str,
+                logprob=entry.logprob,
+                bytes=token_bytes,
+                top_logprobs=resolved_top,
             )
         )
 
-        if sampled_id is not None:
-            context_token_ids.append(sampled_id)
+        context_token_ids.append(entry.token_id)
 
     return ChatCompletionLogProbs(content=resolved_content)