diff --git a/docs/serving/online_serving/derenderer.md b/docs/serving/online_serving/derenderer.md
index e6d1221fd4..cf57f5d8b2 100644
--- a/docs/serving/online_serving/derenderer.md
+++ b/docs/serving/online_serving/derenderer.md
@@ -23,6 +23,10 @@ endpoints.
 
 The derender step needs more than the engine's `token_ids`. It also consumes the original `chat_request`/`completion_request` and `prompt_tokens` carried over from the render step (see [Request format](#request-format)) so the tool and reasoning parsers have the context they need. The chat streaming path additionally needs `prompt_token_ids` (see [Streaming cost](#streaming-cost)) when a parser is configured.
 
+All derender endpoints also accept an optional `prompt_token_ids` (the `token_ids` of the `GenerateRequest` returned by `/render`). It's used to seed detokenization from the end of the prompt, the same way the engine does. Without it, the first output token is decoded as if it started a new sequence. On SentencePiece tokenizers (Llama-2, Phi-3) that drops its leading space compared to `/v1/chat/completions` and `/v1/completions`. If it's omitted, derender falls back to the generate response's `prompt_token_ids` (set when the `GenerateRequest` has `return_token_ids`), then to decoding without prompt context.
+
+Seeding only reads the last few IDs (7 today), so a suffix of the prompt is enough and a long prompt doesn't need to travel back to derender. The exception is streaming chat with a tool or reasoning parser configured. There the parser also reads `prompt_token_ids` to decide whether the prompt left reasoning open and it can look back to any point in the prompt. In this case, send the full list (see [Streaming cost](#streaming-cost)).
+
 ## API Reference
 
 - Chat Completions Derender API (`/v1/chat/completions/derender`)
@@ -96,6 +100,7 @@ The server keeps no state between calls. Everything the next call needs is in `s
 - **`n > 1` is N streams.** Generate interleaves chunks for different choices and tags each with its `index`. Keep one `stream_state` per index and send each chunk with the state for its index. Each index gets its own `role` delta on its first chunk.
 - **Send every chunk through, including the last ones.** The finish reason usually arrives with the final tokens. A chunk with a `finish_reason` and no tokens is accepted too and flushes any buffered tool call arguments. With `stream_options: {"include_usage": true}`, generate ends with a usage only chunk (`choices: []`). Send it through derender like any other chunk to get the usage chunk for the response. `usage.prompt_tokens` is the request's `prompt_tokens` if set, otherwise the generate chunk's.
 - **Send the same context on every call.** `chat_request` and `prompt_token_ids` aren't kept between calls, so they go with every chunk including the usage chunk. Both are required when a tool or reasoning parser is configured. `prompt_token_ids` is the `token_ids` of the `GenerateRequest` returned by `/render`.
+- **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.
@@ -110,7 +115,7 @@ When a tool or reasoning parser is configured, parser internal state (buffered m
 
 - **Transport**: `output_token_ids` and `output_chunk_lens` round-trip in full in both directions on every call. `output_chunk_lens` has one entry per chunk, which is one per token without speculative decoding. This means O(n) bytes per chunk, O(n²) bytes over a full generation. Bounded by `max_model_len`. `prompt_token_ids` is sent in full on every call too and it isn't trimmed as `output_token_ids` grows. This means that for most of a stream it dominates the per chunk payload. A 100k token prompt with 1k tokens of output means `prompt_token_ids` is ~99% of the request body on every chunk.
 - **Compute**: replay is O(n) `parse_delta` calls per chunk (O(n²) per generation). `parse_delta` itself is O(n) for parsers that re-scan accumulated text (e.g. Hermes tool-call JSON, DeepSeek-R1 reasoning). The per-generation cost is O(n³) character work, not O(n²). This is a deliberately minimal first implementation with no caching layer.
-- The parser path also requires `prompt_token_ids` so `parse_delta` can settle whether the prompt left reasoning open or not. Since parser state can't be carried across calls, it re-scans the full prompt once per chunk.
+- The parser path also requires `prompt_token_ids` so `parse_delta` can settle whether the prompt left reasoning open or not. Since parser state can't be carried across calls, it re-scans the full prompt once per chunk. Send the full list and not a suffix because the last reasoning marker can be anywhere in the prompt.
 - Replay runs off the event loop on the renderer's executor (`renderer_num_workers`, default `1`). Size it for the expected number of concurrent parser configured streams.
 
 `output_token_ids` and `prompt_token_ids` are both bounded by `max_model_len` but callers streaming long reasoning traces through a parser configured model should expect materially more state transport and CPU cost than the plain detokenization path.
@@ -153,7 +158,7 @@ with httpx.Client(timeout=60.0) as client:
     generate_request = client.post(
         f"{RENDER}/v1/chat/completions/render", json=chat_request
     ).json()
-    prompt_tokens = len(generate_request["token_ids"])
+    prompt_token_ids = generate_request["token_ids"]
 
     # 2. Generate: token IDs -> token IDs (token-in / token-out engine)
     generate_response = client.post(
@@ -166,7 +171,8 @@ with httpx.Client(timeout=60.0) as client:
         json={
             "model": MODEL,
             "generate_response": generate_response,
-            "prompt_tokens": prompt_tokens,
+            "prompt_tokens": len(prompt_token_ids),
+            "prompt_token_ids": prompt_token_ids,
             "chat_request": chat_request,
         },
     ).json()
diff --git a/requirements/common.txt b/requirements/common.txt
index 70bfa93253..8e483cc712 100644
--- a/requirements/common.txt
+++ b/requirements/common.txt
@@ -9,7 +9,7 @@ blake3
 py-cpuinfo
 transformers >= 5.16.1
 huggingface_hub >= 1.31.0  # Required for the `huggingface_hub.utils.httpx` re-export
-tokenizers >= 0.21.1  # Required for fast incremental detokenization.
+tokenizers >= 0.22.0  # Required for fast incremental detokenization.
 safetensors >= 0.6.2  # MXFP4/MXFP6 dtype support (F8_E8M0, F4) added in 0.6.0: https://github.com/huggingface/safetensors/pull/611
 protobuf >= 5.29.6, !=6.30.*, !=6.31.*, !=6.32.*, !=6.33.0.*, !=6.33.1.*, !=6.33.2.*, !=6.33.3.*, !=6.33.4.*; platform_machine != "s390x"
 protobuf == 7.36.1; platform_machine == "s390x"
diff --git a/tests/entrypoints/scale_out/derender/test_derender.py b/tests/entrypoints/scale_out/derender/test_derender.py
index 80ecc0f49c..659aaed1aa 100644
--- a/tests/entrypoints/scale_out/derender/test_derender.py
+++ b/tests/entrypoints/scale_out/derender/test_derender.py
@@ -27,6 +27,11 @@ async def client(server):
         yield http_client
 
 
+@pytest.fixture(scope="module")
+def tokenizer():
+    return get_tokenizer(MODEL_NAME)
+
+
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
@@ -372,6 +377,58 @@ async def test_derender_chat_model_omitted_resolves_served_name(client):
     assert response.json()["model"] == MODEL_NAME
 
 
+@pytest.mark.asyncio
+async def test_derender_chat_leading_space_seeded_from_prompt(client, tokenizer):
+    """prompt_token_ids keeps the first token's leading space on this
+    Metaspace tokenizer, matching the coupled path."""
+    gen_req = await _render_chat(client)
+    prompt_token_ids = gen_req["token_ids"]
+    output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False)
+
+    unseeded = await client.post(
+        "/v1/chat/completions/derender",
+        json={
+            "model": MODEL_NAME,
+            "generate_response": _make_generate_response(output_ids),
+        },
+    )
+    seeded = await client.post(
+        "/v1/chat/completions/derender",
+        json={
+            "model": MODEL_NAME,
+            "generate_response": _make_generate_response(output_ids),
+            "prompt_token_ids": prompt_token_ids,
+        },
+    )
+    assert unseeded.status_code == 200
+    assert seeded.status_code == 200
+    unseeded_content = unseeded.json()["choices"][0]["message"]["content"]
+    seeded_content = seeded.json()["choices"][0]["message"]["content"]
+
+    assert not unseeded_content.startswith(" ")
+    assert seeded_content.startswith(" ")
+    assert seeded_content.lstrip(" ") == unseeded_content
+
+
+@pytest.mark.asyncio
+async def test_derender_chat_oversized_prompt_token_ids_rejected(client):
+    """prompt_token_ids longer than max_model_len returns 400."""
+    gen_req = await _render_chat(client)
+    synthetic_ids = gen_req["token_ids"][:3]
+    oversized_prompt_ids = [42] * 1_000_000
+
+    response = await client.post(
+        "/v1/chat/completions/derender",
+        json={
+            "model": MODEL_NAME,
+            "generate_response": _make_generate_response(synthetic_ids),
+            "prompt_token_ids": oversized_prompt_ids,
+        },
+    )
+    assert response.status_code == 400
+    assert "max_model_len" in response.json()["error"]["message"]
+
+
 # ---------------------------------------------------------------------------
 # Completion derender tests
 # ---------------------------------------------------------------------------
@@ -486,6 +543,84 @@ async def test_derender_completion_prompt_tokens_length_mismatch(client):
     assert response.status_code == 400
 
 
+@pytest.mark.asyncio
+async def test_derender_completion_prompt_token_ids_length_mismatch(client):
+    """len(prompt_token_ids) != len(generate_responses) returns 400, the
+    same as the existing prompt_tokens length check."""
+    gr1 = await _render_completion(client, "Hello")
+    ids1 = gr1["token_ids"][:3]
+
+    response = await client.post(
+        "/v1/completions/derender",
+        json={
+            "model": MODEL_NAME,
+            "generate_responses": [
+                _make_completion_generate_response(ids1, gr1["request_id"]),
+            ],
+            "prompt_token_ids": [[1, 2], [3, 4]],
+        },
+    )
+    assert response.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_derender_completion_leading_space_seeded_from_prompt(client, tokenizer):
+    """Completions counterpart of
+    test_derender_chat_leading_space_seeded_from_prompt."""
+    gr1 = await _render_completion(client, "Hello world")
+    prompt_token_ids = gr1["token_ids"]
+    output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False)
+
+    unseeded = await client.post(
+        "/v1/completions/derender",
+        json={
+            "model": MODEL_NAME,
+            "generate_responses": [
+                _make_completion_generate_response(output_ids, gr1["request_id"]),
+            ],
+        },
+    )
+    seeded = await client.post(
+        "/v1/completions/derender",
+        json={
+            "model": MODEL_NAME,
+            "generate_responses": [
+                _make_completion_generate_response(output_ids, gr1["request_id"]),
+            ],
+            "prompt_token_ids": [prompt_token_ids],
+        },
+    )
+    assert unseeded.status_code == 200
+    assert seeded.status_code == 200
+    unseeded_text = unseeded.json()["choices"][0]["text"]
+    seeded_text = seeded.json()["choices"][0]["text"]
+
+    assert not unseeded_text.startswith(" ")
+    assert seeded_text.startswith(" ")
+    assert seeded_text.lstrip(" ") == unseeded_text
+
+
+@pytest.mark.asyncio
+async def test_derender_completion_oversized_prompt_token_ids_rejected(client):
+    """A prompt_token_ids entry longer than max_model_len returns 400."""
+    gr1 = await _render_completion(client, "Hello")
+    ids1 = gr1["token_ids"][:3]
+    oversized_prompt_ids = [42] * 1_000_000
+
+    response = await client.post(
+        "/v1/completions/derender",
+        json={
+            "model": MODEL_NAME,
+            "generate_responses": [
+                _make_completion_generate_response(ids1, gr1["request_id"]),
+            ],
+            "prompt_token_ids": [oversized_prompt_ids],
+        },
+    )
+    assert response.status_code == 400
+    assert "max_model_len" in response.json()["error"]["message"]
+
+
 @pytest.mark.asyncio
 async def test_derender_completion_empty_generate_responses(client):
     """Empty generate_responses list returns 400."""
diff --git a/tests/entrypoints/scale_out/derender/test_derender_stream.py b/tests/entrypoints/scale_out/derender/test_derender_stream.py
index b1db1193a2..5862256835 100644
--- a/tests/entrypoints/scale_out/derender/test_derender_stream.py
+++ b/tests/entrypoints/scale_out/derender/test_derender_stream.py
@@ -20,11 +20,17 @@ Tests are split into two layers:
 import json
 import threading
 from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import AsyncMock, MagicMock
 
+import httpx
 import pytest
 import pytest_asyncio
+from pydantic import ValidationError
+from tokenizers.decoders import DecodeStream
 
+import vllm.renderers.online_derenderer as online_derenderer_module
 from tests.entrypoints.scale_out.derender.utils import stream_chat_derender
+from tests.utils import RemoteLaunchRenderServer
 from vllm.entrypoints.generate.base.protocol import (
     DeltaFunctionCall,
     DeltaMessage,
@@ -32,7 +38,17 @@ from vllm.entrypoints.generate.base.protocol import (
     PerRequestMetrics,
     SpeculativeDecodingMetrics,
 )
+from vllm.entrypoints.openai.chat_completion.protocol import (
+    ChatCompletionRequest,
+    ChatCompletionStreamResponse,
+)
+from vllm.entrypoints.openai.completion.protocol import CompletionRequest
+from vllm.entrypoints.scale_out.derender.serving import ServingDerender
 from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
+    DerenderChatRequest,
+    DerenderChatStreamRequest,
+    DerenderCompletionRequest,
+    DerenderCompletionStreamRequest,
     DerenderStreamState,
     GenerateResponse,
     GenerateResponseChoice,
@@ -41,6 +57,12 @@ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
 )
 from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo
 from vllm.parser import Parser
+from vllm.renderers.online_derenderer import (
+    OnlineDerenderer,
+    _decode_params,
+    _seed_stream_state,
+)
+from vllm.tokenizers import get_tokenizer
 from vllm.utils import random_uuid
 
 MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
@@ -170,6 +192,7 @@ def _make_stream_chunk(
     usage: dict | None = None,
     metrics: PerRequestMetrics | None = None,
     logprobs: dict | None = None,
+    prompt_token_ids: list[int] | None = None,
 ) -> GenerateStreamResponse:
     """Build a GenerateStreamResponse SSE chunk."""
     return GenerateStreamResponse(
@@ -184,6 +207,7 @@ def _make_stream_chunk(
         ],
         usage=UsageInfo(**usage) if usage else None,
         metrics=metrics,
+        prompt_token_ids=prompt_token_ids,
     )
 
 
@@ -228,18 +252,12 @@ def _make_usage_chunk(
 @pytest.fixture(scope="module")
 def tokenizer():
     """Load the tiny tokenizer used across unit tests."""
-    from vllm.tokenizers import get_tokenizer
-
     return get_tokenizer(MODEL_NAME)
 
 
 @pytest.fixture(scope="module")
 def derenderer(tokenizer, request):
     """Construct a minimal OnlineDerenderer backed by a stub renderer."""
-    from unittest.mock import MagicMock
-
-    from vllm.renderers.online_derenderer import OnlineDerenderer
-
     renderer = MagicMock()
     renderer.get_tokenizer.return_value = tokenizer
 
@@ -273,10 +291,6 @@ def parsed_derenderer(tokenizer):
     `renderer._executor` must be a real `ThreadPoolExecutor`.
     `loop.run_in_executor` cannot submit work to a `MagicMock`.
     """
-    from unittest.mock import MagicMock
-
-    from vllm.renderers.online_derenderer import OnlineDerenderer
-
     renderer = MagicMock()
     renderer.get_tokenizer.return_value = tokenizer
     renderer._executor = ThreadPoolExecutor(max_workers=2)
@@ -433,6 +447,57 @@ class TestDetokenizeDelta:
         # Bounded by a small constant independent of len(token_ids)
         assert max_window <= 32
 
+    def test_truncated_multibyte_char_is_held_back(self, derenderer, tokenizer):
+        """Output ending mid-character emits no U+FFFD: batch == stream == engine."""
+        emoji_ids = tokenizer.encode("🌈", add_special_tokens=False)
+        byte_ids = emoji_ids[-4:]
+        assert len(byte_ids) == 4
+        output_ids = byte_ids[:-1]
+
+        batch, _ = derenderer._detokenize_delta(
+            tokenizer, output_ids, DerenderStreamState()
+        )
+        streamed = self._chunked(derenderer, tokenizer, [[t] for t in output_ids])
+        engine = _decode_stream_expected(tokenizer, [], output_ids)
+
+        assert "�" not in batch
+        assert batch == streamed == engine
+
+        full, _ = derenderer._detokenize_delta(
+            tokenizer, byte_ids, DerenderStreamState()
+        )
+        assert full.endswith("🌈")
+
+    @pytest.mark.parametrize(
+        ("skip", "spaces", "preserve", "fast", "expected"),
+        [
+            (True, True, False, True, (True, True)),
+            (True, False, False, True, (True, True)),
+            (True, False, False, False, (True, False)),
+            (False, False, False, True, (False, False)),
+            (False, False, True, True, (False, False)),
+            (True, False, True, True, (False, False)),
+            (True, True, True, True, (False, True)),
+        ],
+    )
+    def test_decode_params_mirror_engine(
+        self, monkeypatch, skip, spaces, preserve, fast, expected
+    ):
+        """Flags match the engine's fast or slow detokenizer for the tokenizer."""
+        monkeypatch.setattr(
+            online_derenderer_module, "uses_fast_detokenizer", lambda _: fast
+        )
+        request = CompletionRequest(
+            model=MODEL_NAME,
+            prompt="x",
+            skip_special_tokens=skip,
+            spaces_between_special_tokens=spaces,
+        )
+        assert _decode_params(None, request, preserve_special=preserve) == expected
+
+    def test_decode_params_default_without_request(self, tokenizer):
+        assert _decode_params(tokenizer, None) == (True, True)
+
     def test_n_independent_streams_same_result(self, derenderer, tokenizer):
         """N parallel streams with the same token sequence give the same text."""
         token_ids = tokenizer.encode("parallel streams")[:8]
@@ -451,6 +516,242 @@ class TestDetokenizeDelta:
         assert results[0] == self._one_shot(tokenizer, token_ids)
 
 
+# ---------------------------------------------------------------------------
+# Prompt seeded detokenization (matches the engine's primed detokenizer)
+# ---------------------------------------------------------------------------
+
+
+def _decode_stream_expected(
+    tokenizer, prompt_ids: list[int], output_ids: list[int]
+) -> str:
+    """Engine-equivalent decode via `DecodeStream`, primed by stepping the
+    prompt (`DecodeStream(ids=...)` needs tokenizers >= 0.22)."""
+    stream = DecodeStream(skip_special_tokens=True)
+    for tid in prompt_ids:
+        stream.step(tokenizer.backend_tokenizer, tid)
+    return "".join(
+        s
+        for tid in output_ids
+        if (s := stream.step(tokenizer.backend_tokenizer, tid)) is not None
+    )
+
+
+@pytest.fixture(scope="module")
+def leading_space_ids(tokenizer):
+    """(prompt_ids, output_ids, expected) where seeded and unseeded decode
+    differ by the leading space."""
+    prompt_ids = tokenizer.encode("[INST] Hi [/INST]", add_special_tokens=False)
+    output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False)
+    expected = _decode_stream_expected(tokenizer, prompt_ids, output_ids)
+    unseeded = tokenizer.decode(output_ids, skip_special_tokens=True)
+    assert expected.startswith(" ") and not unseeded.startswith(" "), (
+        "fixture no longer reproduces the leading-space divergence; "
+        "pick a different prompt/output pair"
+    )
+    return prompt_ids, output_ids, expected
+
+
+class TestSeedStreamState:
+    """`_seed_stream_state`: pure helper building the initial decode state."""
+
+    def test_seeds_prev_tokens_from_prompt_tail(self, tokenizer, leading_space_ids):
+        prompt_ids, _, _ = leading_space_ids
+        state = _seed_stream_state(tokenizer, prompt_ids, skip_special_tokens=True)
+        assert state.prev_tokens
+        assert state.read_offset == len(state.prev_tokens)
+
+    def test_none_prompt_token_ids_returns_empty_state(self, tokenizer):
+        assert (
+            _seed_stream_state(tokenizer, None, skip_special_tokens=True)
+            == DerenderStreamState()
+        )
+
+    def test_empty_prompt_token_ids_returns_empty_state(self, tokenizer):
+        assert (
+            _seed_stream_state(tokenizer, [], skip_special_tokens=True)
+            == DerenderStreamState()
+        )
+
+
+class TestPromptSeededLeadingSpace:
+    """Every decode entry point keeps the first token's leading space when
+    given prompt_token_ids."""
+
+    @pytest.mark.asyncio
+    async def test_chat_batch_seeded_via_explicit_prompt_token_ids(
+        self, derenderer, leading_space_ids
+    ):
+        prompt_ids, output_ids, expected = leading_space_ids
+        choices = await derenderer.derender_chat(
+            GenerateResponse(
+                request_id="t",
+                choices=[
+                    GenerateResponseChoice(
+                        index=0, token_ids=output_ids, finish_reason="stop"
+                    )
+                ],
+            ),
+            prompt_token_ids=prompt_ids,
+        )
+        assert choices[0].message.content == expected
+
+    @pytest.mark.asyncio
+    async def test_chat_batch_seeded_via_generate_response_fallback(
+        self, derenderer, leading_space_ids
+    ):
+        prompt_ids, output_ids, expected = leading_space_ids
+        choices = await derenderer.derender_chat(
+            GenerateResponse(
+                request_id="t",
+                choices=[
+                    GenerateResponseChoice(
+                        index=0, token_ids=output_ids, finish_reason="stop"
+                    )
+                ],
+                prompt_token_ids=prompt_ids,
+            ),
+        )
+        assert choices[0].message.content == expected
+
+    @pytest.mark.asyncio
+    async def test_chat_batch_explicit_takes_precedence_over_fallback(
+        self, derenderer, tokenizer, leading_space_ids
+    ):
+        prompt_ids, output_ids, expected = leading_space_ids
+        wrong_ids = tokenizer.encode("a different prompt", add_special_tokens=False)
+        choices = await derenderer.derender_chat(
+            GenerateResponse(
+                request_id="t",
+                choices=[
+                    GenerateResponseChoice(
+                        index=0, token_ids=output_ids, finish_reason="stop"
+                    )
+                ],
+                prompt_token_ids=wrong_ids,
+            ),
+            prompt_token_ids=prompt_ids,
+        )
+        assert choices[0].message.content == expected
+
+    @pytest.mark.asyncio
+    async def test_chat_batch_without_prompt_context_stays_unseeded(
+        self, derenderer, leading_space_ids
+    ):
+        _, output_ids, _ = leading_space_ids
+        choices = await derenderer.derender_chat(
+            GenerateResponse(
+                request_id="t",
+                choices=[
+                    GenerateResponseChoice(
+                        index=0, token_ids=output_ids, finish_reason="stop"
+                    )
+                ],
+            ),
+        )
+        assert not choices[0].message.content.startswith(" ")
+
+    @pytest.mark.asyncio
+    async def test_completion_batch_seeded(self, derenderer, leading_space_ids):
+        prompt_ids, output_ids, expected = leading_space_ids
+        choices, _, _ = await derenderer.derender_completion(
+            [
+                GenerateResponse(
+                    request_id="t",
+                    choices=[
+                        GenerateResponseChoice(
+                            index=0, token_ids=output_ids, finish_reason="stop"
+                        )
+                    ],
+                )
+            ],
+            prompt_token_ids=[prompt_ids],
+        )
+        assert choices[0].text == expected
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("from_generate_chunk", [False, True])
+    async def test_chat_stream_first_chunk_seeded(
+        self, derenderer, leading_space_ids, from_generate_chunk
+    ):
+        prompt_ids, output_ids, expected = leading_space_ids
+        mid = len(output_ids) // 2
+
+        chunk1, state = await derenderer.derender_chat_stream(
+            model=MODEL_NAME,
+            generate_chunk=_make_stream_chunk(
+                output_ids[:mid],
+                prompt_token_ids=prompt_ids if from_generate_chunk else None,
+            ),
+            prompt_token_ids=None if from_generate_chunk else prompt_ids,
+        )
+        # Only needed on the first chunk; the carried state already holds
+        # the seeded window from here on.
+        chunk2, _ = await derenderer.derender_chat_stream(
+            model=MODEL_NAME,
+            generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"),
+            state=state,
+        )
+
+        streamed = (chunk1.choices[0].delta.content or "") + (
+            chunk2.choices[0].delta.content or ""
+        )
+        assert streamed == expected
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("from_generate_chunk", [False, True])
+    async def test_completion_stream_first_chunk_seeded(
+        self, derenderer, leading_space_ids, from_generate_chunk
+    ):
+        prompt_ids, output_ids, expected = leading_space_ids
+        mid = len(output_ids) // 2
+
+        chunk1, state = await derenderer.derender_completion_stream(
+            model=MODEL_NAME,
+            generate_chunk=_make_stream_chunk(
+                output_ids[:mid],
+                prompt_token_ids=prompt_ids if from_generate_chunk else None,
+            ),
+            prompt_token_ids=None if from_generate_chunk else prompt_ids,
+        )
+        chunk2, _ = await derenderer.derender_completion_stream(
+            model=MODEL_NAME,
+            generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"),
+            state=state,
+        )
+
+        assert chunk1.choices[0].text + chunk2.choices[0].text == expected
+
+    @pytest.mark.asyncio
+    async def test_parser_path_ephemeral_state_seeded(
+        self, parsed_derenderer, monkeypatch, leading_space_ids
+    ):
+        """The parser path's per-call detok window is seeded too, so
+        `parse_delta` sees the same text as coupled serving."""
+
+        class _EchoParser(Parser):
+            def parse_delta(
+                self,
+                delta_text,
+                delta_token_ids,
+                request,
+                prompt_token_ids=None,
+                *,
+                finished,
+            ):
+                return DeltaMessage(content=delta_text) if delta_text else None
+
+        monkeypatch.setattr(parsed_derenderer, "parser", _EchoParser)
+        prompt_ids, output_ids, expected = leading_space_ids
+
+        chunk, _ = await parsed_derenderer.derender_chat_stream(
+            model=MODEL_NAME,
+            generate_chunk=_make_stream_chunk(output_ids, finish_reason="stop"),
+            chat_request=_chat_request(),
+            prompt_token_ids=prompt_ids,
+        )
+        assert chunk.choices[0].delta.content == expected
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "stream_method",
@@ -460,8 +761,6 @@ async def test_streaming_detokenization_runs_off_event_loop(
     derenderer, monkeypatch, stream_method
 ):
     """Streaming detokenization runs on the renderer executor."""
-    import vllm.renderers.online_derenderer as online_derenderer_module
-
     event_loop_thread_id = threading.get_ident()
     executor_thread_id = derenderer.renderer._executor.submit(
         threading.get_ident
@@ -561,8 +860,6 @@ class TestDerenderCompletionStream:
     @pytest.mark.asyncio
     async def test_skip_special_tokens_threaded(self, derenderer, tokenizer):
         """completion_request.skip_special_tokens is honored (not hardcoded True)."""
-        from vllm.entrypoints.openai.completion.protocol import CompletionRequest
-
         eos = tokenizer.eos_token_id
         if eos is None:
             pytest.skip("tokenizer has no eos token to exercise special stripping")
@@ -789,8 +1086,6 @@ class TestDerenderChatStream:
 
 
 def _chat_request(**kwargs):
-    from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
-
     kwargs.setdefault("messages", [{"role": "user", "content": "hi"}])
     kwargs.setdefault("model", MODEL_NAME)
     return ChatCompletionRequest(**kwargs)
@@ -1072,8 +1367,6 @@ def harmony_encode():
 @pytest.fixture(scope="module")
 def harmony_tokenizer():
     pytest.importorskip("openai_harmony")
-    from vllm.tokenizers import get_tokenizer
-
     return get_tokenizer(HARMONY_MODEL, trust_remote_code=True)
 
 
@@ -1085,10 +1378,7 @@ def harmony_derenderer(harmony_tokenizer):
     with a real tokenizer and parser, so the replay path is exercised
     against Harmony's actual channel grammar rather than a stub.
     """
-    from unittest.mock import MagicMock
-
     from vllm.parser.harmony import HarmonyParser
-    from vllm.renderers.online_derenderer import OnlineDerenderer
 
     renderer = MagicMock()
     renderer.get_tokenizer.return_value = harmony_tokenizer
@@ -1224,20 +1514,14 @@ class TestDerenderStreamStateValidation:
     """DerenderStreamState rejects malformed caller supplied offsets/lengths."""
 
     def test_negative_prefix_offset_rejected(self):
-        from pydantic import ValidationError
-
         with pytest.raises(ValidationError):
             DerenderStreamState(prefix_offset=-1)
 
     def test_negative_read_offset_rejected(self):
-        from pydantic import ValidationError
-
         with pytest.raises(ValidationError):
             DerenderStreamState(read_offset=-1)
 
     def test_prev_tokens_over_cap_rejected(self):
-        from pydantic import ValidationError
-
         with pytest.raises(ValidationError):
             DerenderStreamState(prev_tokens=["a"] * 1025)
 
@@ -1246,14 +1530,10 @@ class TestDerenderStreamStateValidation:
         assert len(state.prev_tokens) == 1024
 
     def test_output_chunk_lens_mismatch_rejected(self):
-        from pydantic import ValidationError
-
         with pytest.raises(ValidationError):
             DerenderStreamState(output_token_ids=[1, 2, 3], output_chunk_lens=[1, 1])
 
     def test_output_chunk_lens_zero_entry_rejected(self):
-        from pydantic import ValidationError
-
         with pytest.raises(ValidationError):
             DerenderStreamState(output_token_ids=[1, 2], output_chunk_lens=[0, 2])
 
@@ -1264,14 +1544,40 @@ class TestDerenderStreamStateValidation:
         assert state.output_chunk_lens == [1, 2]
 
 
+_GEN = {"request_id": "t", "choices": []}
+_GEN_BAD = {**_GEN, "prompt_token_ids": [-1]}
+
+
+@pytest.mark.parametrize(
+    "model_cls,payload",
+    [
+        (DerenderChatRequest, {"generate_response": _GEN, "prompt_token_ids": [-1]}),
+        (DerenderChatRequest, {"generate_response": _GEN_BAD}),
+        (
+            DerenderCompletionRequest,
+            {"generate_responses": [_GEN], "prompt_token_ids": [[-1]]},
+        ),
+        (
+            DerenderChatStreamRequest,
+            {"stream": True, "generate_chunk": _GEN, "prompt_token_ids": [-1]},
+        ),
+        (
+            DerenderCompletionStreamRequest,
+            {"stream": True, "generate_chunk": _GEN_BAD},
+        ),
+    ],
+)
+def test_negative_prompt_token_ids_rejected(model_cls, payload):
+    """Negative prompt ids would reach convert_ids_to_tokens, so they get a
+    400 at parse time like `GenerateRequest.token_ids` does."""
+    with pytest.raises(ValidationError, match="prompt_token_ids"):
+        model_cls.model_validate(payload)
+
+
 class TestServingDerenderStreamErrorHandling:
     """Malformed stream_state must surface as 400 and not an unhandled 500."""
 
     def _make_serving(self, side_effect: Exception):
-        from unittest.mock import AsyncMock, MagicMock
-
-        from vllm.entrypoints.scale_out.derender.serving import ServingDerender
-
         models = MagicMock()
         models.is_base_model.return_value = True
         models.model_config = MagicMock()
@@ -1289,10 +1595,6 @@ class TestServingDerenderStreamErrorHandling:
     @pytest.mark.asyncio
     @pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
     async def test_completion_stream_bad_state_returns_400(self, exc):
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderCompletionStreamRequest,
-        )
-
         serving = self._make_serving(exc)
         request = DerenderCompletionStreamRequest(
             stream=True,
@@ -1307,10 +1609,6 @@ class TestServingDerenderStreamErrorHandling:
     @pytest.mark.asyncio
     @pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
     async def test_chat_stream_bad_state_returns_400(self, exc):
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(exc)
         request = DerenderChatStreamRequest(
             stream=True,
@@ -1329,13 +1627,6 @@ class TestServingDerenderStreamValidation:
     before touching the tokenizer)."""
 
     def _make_serving(self, *, parser_configured: bool, max_model_len: int = 100_000):
-        from unittest.mock import AsyncMock, MagicMock
-
-        from vllm.entrypoints.openai.chat_completion.protocol import (
-            ChatCompletionStreamResponse,
-        )
-        from vllm.entrypoints.scale_out.derender.serving import ServingDerender
-
         models = MagicMock()
         models.is_base_model.return_value = True
         models.model_config = MagicMock()
@@ -1353,10 +1644,6 @@ class TestServingDerenderStreamValidation:
 
     @pytest.mark.asyncio
     async def test_missing_chat_request_with_parser_rejected(self):
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(parser_configured=True)
         request = DerenderChatStreamRequest(
             stream=True,
@@ -1371,10 +1658,6 @@ class TestServingDerenderStreamValidation:
 
     @pytest.mark.asyncio
     async def test_missing_chat_request_without_parser_ok(self):
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(parser_configured=False)
         request = DerenderChatStreamRequest(
             stream=True,
@@ -1390,10 +1673,6 @@ class TestServingDerenderStreamValidation:
         the same way it rejects a missing chat_request. Without it,
         parse_delta cannot tell whether the prompt left reasoning open and
         would misclassify reasoning content as plain content."""
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(parser_configured=True)
         request = DerenderChatStreamRequest(
             stream=True,
@@ -1409,10 +1688,6 @@ class TestServingDerenderStreamValidation:
 
     @pytest.mark.asyncio
     async def test_prompt_token_ids_present_with_parser_ok(self):
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(parser_configured=True)
         request = DerenderChatStreamRequest(
             stream=True,
@@ -1426,10 +1701,6 @@ class TestServingDerenderStreamValidation:
 
     @pytest.mark.asyncio
     async def test_oversized_output_token_ids_rejected(self):
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(parser_configured=False, max_model_len=4)
         request = DerenderChatStreamRequest(
             stream=True,
@@ -1451,10 +1722,6 @@ class TestServingDerenderStreamValidation:
         parser configured deployment rescans it in full on every chunk
         (is_reasoning_end / adjust_initial_state_from_prompt), so it must be
         bounded the same way output_token_ids is."""
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(parser_configured=True, max_model_len=4)
         request = DerenderChatStreamRequest(
             stream=True,
@@ -1475,10 +1742,6 @@ class TestServingDerenderStreamValidation:
         DerenderStreamState is threaded through every choice). The check
         could never fire since derender_chat_stream itself rejects anything
         above 1 first."""
-        from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
-            DerenderChatStreamRequest,
-        )
-
         serving = self._make_serving(parser_configured=False)
         two_choices = GenerateStreamResponse(
             request_id="t",
@@ -1503,16 +1766,12 @@ class TestServingDerenderStreamValidation:
 
 @pytest.fixture(scope="module")
 def server():
-    from tests.utils import RemoteLaunchRenderServer
-
     with RemoteLaunchRenderServer(MODEL_NAME, []) as remote_server:
         yield remote_server
 
 
 @pytest_asyncio.fixture
 async def client(server):
-    import httpx
-
     async with httpx.AsyncClient(
         base_url=server.url_for(""), timeout=30.0
     ) as http_client:
@@ -1776,8 +2035,6 @@ _PARSER_TOOLS = [
 
 @pytest.fixture(scope="module")
 def parser_server():
-    from tests.utils import RemoteLaunchRenderServer
-
     args = [
         "--enable-auto-tool-choice",
         "--tool-call-parser",
@@ -1791,8 +2048,6 @@ def parser_server():
 
 @pytest_asyncio.fixture
 async def parser_client(parser_server):
-    import httpx
-
     async with httpx.AsyncClient(
         base_url=parser_server.url_for(""), timeout=60.0
     ) as http_client:
@@ -1801,8 +2056,6 @@ async def parser_client(parser_server):
 
 @pytest.fixture(scope="module")
 def parser_tokenizer():
-    from vllm.tokenizers import get_tokenizer
-
     return get_tokenizer(PARSER_MODEL)
 
 
diff --git a/vllm/entrypoints/scale_out/derender/serving.py b/vllm/entrypoints/scale_out/derender/serving.py
index 5da7366403..0783d3a78f 100644
--- a/vllm/entrypoints/scale_out/derender/serving.py
+++ b/vllm/entrypoints/scale_out/derender/serving.py
@@ -117,6 +117,20 @@ class ServingDerender(BaseServing):
 
         return None
 
+    def _validate_prompt_token_ids(
+        self,
+        prompt_token_ids: list[list[int] | None],
+    ) -> ErrorResponse | None:
+        """Reject caller-supplied prompt_token_ids longer than max_model_len."""
+        max_model_len = self.model_config.max_model_len
+        for ids in prompt_token_ids:
+            if ids is not None and len(ids) > max_model_len:
+                return self.create_error_response(
+                    f"prompt_token_ids length ({len(ids)}) exceeds "
+                    f"max_model_len ({max_model_len})."
+                )
+        return None
+
     async def derender_chat_response(
         self,
         request: DerenderChatRequest,
@@ -138,6 +152,10 @@ class ServingDerender(BaseServing):
         if bounds_error is not None:
             return bounds_error
 
+        prompt_ids_error = self._validate_prompt_token_ids([request.prompt_token_ids])
+        if prompt_ids_error is not None:
+            return prompt_ids_error
+
         if self.online_derenderer.parser is not None and request.chat_request is None:
             return self.create_error_response(
                 "chat_request is required when a tool or reasoning parser is "
@@ -147,7 +165,9 @@ class ServingDerender(BaseServing):
 
         try:
             choices = await self.online_derenderer.derender_chat(
-                request.generate_response, request.chat_request
+                request.generate_response,
+                request.chat_request,
+                request.prompt_token_ids,
             )
         except ValueError as exc:
             return self.create_error_response(str(exc))
@@ -203,6 +223,11 @@ class ServingDerender(BaseServing):
         if bounds_error is not None:
             return bounds_error
 
+        if request.prompt_token_ids is not None:
+            prompt_ids_error = self._validate_prompt_token_ids(request.prompt_token_ids)
+            if prompt_ids_error is not None:
+                return prompt_ids_error
+
         (
             choices,
             total_prompt_tokens,
@@ -211,6 +236,7 @@ class ServingDerender(BaseServing):
             request.generate_responses,
             request.prompt_tokens,
             completion_request=request.completion_request,
+            prompt_token_ids=request.prompt_token_ids,
         )
 
         first = request.generate_responses[0]
@@ -367,6 +393,10 @@ class ServingDerender(BaseServing):
         if error_check_ret is not None:
             return error_check_ret
 
+        prompt_ids_error = self._validate_prompt_token_ids([request.prompt_token_ids])
+        if prompt_ids_error is not None:
+            return prompt_ids_error
+
         model_name = request.model or self.models.model_name()
         try:
             (
@@ -378,6 +408,7 @@ class ServingDerender(BaseServing):
                 state=request.stream_state,
                 prompt_tokens=request.prompt_tokens,
                 completion_request=request.completion_request,
+                prompt_token_ids=request.prompt_token_ids,
             )
         except ValueError as exc:
             return self.create_error_response(str(exc))
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 94604b013f..60cffd07c2 100644
--- a/vllm/entrypoints/scale_out/token_in_token_out/protocol.py
+++ b/vllm/entrypoints/scale_out/token_in_token_out/protocol.py
@@ -5,6 +5,7 @@ from typing import Annotated, Any, Literal, TypeAlias
 from pydantic import (
     BaseModel,
     Field,
+    NonNegativeInt,
     PrivateAttr,
     field_validator,
     model_validator,
@@ -362,7 +363,7 @@ class GenerateStreamResponse(BaseModel):
     )
     choices: list[GenerateResponseStreamChoice]
     usage: UsageInfo | None = Field(default=None)
-    prompt_token_ids: list[int] | None = None
+    prompt_token_ids: list[NonNegativeInt] | None = None
     mm_placeholders: dict[str, list[PlaceholderRangeInfo]] | None = None
     metrics: PerRequestMetrics | None = None
 
@@ -382,7 +383,7 @@ class GenerateResponse(BaseModel):
     usage: UsageInfo | None = Field(default=None)
     prompt_logprobs: list[dict[int, Logprob] | None] | None = None
     prompt_token_id_logprobs: str | None = None
-    prompt_token_ids: list[int] | None = None
+    prompt_token_ids: list[NonNegativeInt] | None = None
     mm_placeholders: dict[str, list[PlaceholderRangeInfo]] | None = None
     metrics: PerRequestMetrics | None = None
 
@@ -421,6 +422,15 @@ class DerenderChatRequest(BaseModel):
     len(GenerateRequest.token_ids) from the render step.
     """
 
+    prompt_token_ids: list[NonNegativeInt] | None = None
+    """Prompt token IDs (`GenerateRequest.token_ids` from /render). Seeds
+    detokenization from the prompt tail so the first output token keeps its
+    leading space on SentencePiece tokenizers. Falls back to
+    `generate_response.prompt_token_ids`, then to unseeded decoding.
+
+    Only the last few IDs are read, so a suffix of the prompt is enough.
+    """
+
     chat_request: ChatCompletionRequest | None = None
     """The original (post-adjust_request) ChatCompletionRequest from /render.
 
@@ -455,6 +465,14 @@ class DerenderCompletionRequest(BaseModel):
     If provided, len(prompt_tokens) must equal len(generate_responses).
     """
 
+    prompt_token_ids: list[list[NonNegativeInt] | None] | None = None
+    """One prompt token ID list per response, used to seed detokenization.
+    See `DerenderChatRequest.prompt_token_ids`.
+
+    If provided, len(prompt_token_ids) must equal len(generate_responses).
+    A `None` entry falls back to `generate_responses[i].prompt_token_ids`.
+    """
+
     completion_request: CompletionRequest | None = None
     """The original (post-adjust_request) CompletionRequest from /render.
 
@@ -472,6 +490,13 @@ class DerenderCompletionRequest(BaseModel):
                 f"prompt_tokens length ({len(self.prompt_tokens)}) must equal "
                 f"generate_responses length ({len(self.generate_responses)})"
             )
+        if self.prompt_token_ids is not None and len(self.prompt_token_ids) != len(
+            self.generate_responses
+        ):
+            raise ValueError(
+                f"prompt_token_ids length ({len(self.prompt_token_ids)}) must "
+                f"equal generate_responses length ({len(self.generate_responses)})"
+            )
         return self
 
 
@@ -627,7 +652,7 @@ class DerenderChatStreamRequest(BaseModel):
     prompt_tokens: int | None = None
     """Prompt token count for usage. Forwarded from the render step."""
 
-    prompt_token_ids: list[int] | None = None
+    prompt_token_ids: list[NonNegativeInt] | None = None
     """Prompt token IDs. Required by the parser path's `parse_delta` to
     settle its initial reasoning state (e.g. chat templates that pre-open
     ``<think>``). `prompt_tokens` is a usage count and cannot serve this
@@ -636,8 +661,15 @@ class DerenderChatStreamRequest(BaseModel):
     Rejected with a 400 (by `ServingDerender`) when a tool or reasoning
     parser is configured and this is omitted. Without it, `parse_delta`
     cannot tell whether the prompt left reasoning open and would silently
-    misclassify reasoning content as plain content. Unused on the plain
-    detokenization path.
+    misclassify reasoning content as plain content. On all paths it also
+    seeds detokenization on the first chunk, falling back to
+    `generate_chunk.prompt_token_ids` when omitted (see
+    `DerenderChatRequest.prompt_token_ids`).
+
+    With a parser configured, send the full list. Parsers look back to the
+    last reasoning marker which can be anywhere in the prompt, so a suffix
+    can flip the initial reasoning state. Without a parser, a suffix is
+    enough.
     """
 
     chat_request: ChatCompletionRequest | None = None
@@ -666,6 +698,12 @@ class DerenderCompletionStreamRequest(BaseModel):
     prompt_tokens: int | None = None
     """Prompt token count for usage."""
 
+    prompt_token_ids: list[NonNegativeInt] | None = None
+    """Prompt token IDs, used on the first chunk to seed detokenization.
+    Falls back to `generate_chunk.prompt_token_ids`. See
+    `DerenderChatRequest.prompt_token_ids`.
+    """
+
     completion_request: CompletionRequest | None = None
     """The original (post adjust_request) CompletionRequest from /render."""
     # --8<-- [end:derender-completion-stream-request]
diff --git a/vllm/renderers/online_derenderer.py b/vllm/renderers/online_derenderer.py
index c1f175798d..0261401dd4 100644
--- a/vllm/renderers/online_derenderer.py
+++ b/vllm/renderers/online_derenderer.py
@@ -40,9 +40,14 @@ from vllm.logger import init_logger
 from vllm.parser import Parser, ParserManager
 from vllm.renderers import BaseRenderer
 from vllm.tokenizers import TokenizerLike
-from vllm.tokenizers.detokenizer_utils import detokenize_incrementally
+from vllm.tokenizers.detokenizer_utils import (
+    convert_prompt_ids_to_tokens,
+    detokenize_incrementally,
+    get_leading_space_marker,
+)
 from vllm.utils import random_uuid
 from vllm.utils.async_utils import make_async
+from vllm.v1.engine.detokenizer import uses_fast_detokenizer
 
 logger = init_logger(__name__)
 
@@ -115,17 +120,37 @@ class OnlineDerenderer:
         self,
         generate_response: GenerateResponse,
         chat_request: ChatCompletionRequest | None = None,
+        prompt_token_ids: list[int] | None = None,
     ) -> list[ChatCompletionResponseChoice]:
-        return await self._derender_chat_async(generate_response, chat_request)
+        return await self._derender_chat_async(
+            generate_response, chat_request, prompt_token_ids
+        )
 
     def _derender_chat(
         self,
         generate_response: GenerateResponse,
         chat_request: ChatCompletionRequest | None = None,
+        prompt_token_ids: list[int] | None = None,
     ) -> list[ChatCompletionResponseChoice]:
         tokenizer = self.renderer.get_tokenizer()
         choices: list[ChatCompletionResponseChoice] = []
 
+        has_parser = self.parser is not None and chat_request is not None
+        skip_special, spaces_between = _decode_params(
+            tokenizer, chat_request, preserve_special=has_parser
+        )
+        seed_ids = (
+            prompt_token_ids
+            if prompt_token_ids is not None
+            else generate_response.prompt_token_ids
+        )
+        # Choices continue the same prompt, so they share one seeded window.
+        seed_state = _seed_stream_state(
+            tokenizer,
+            seed_ids,
+            skip_special_tokens=skip_special,
+        )
+
         for choice in generate_response.choices:
             if not choice.token_ids:
                 raise ValueError(f"choice {choice.index} has empty or null token_ids")
@@ -136,12 +161,17 @@ class OnlineDerenderer:
                 else None
             )
 
-            if self.parser is not None and chat_request is not None:
+            if has_parser:
+                assert self.parser is not None and chat_request is not None
                 # Parser path: decode with special tokens preserved
                 # so the parser can see markers like </think>,
                 # <tool_call>, or Harmony channel tokens.
-                decoded_text = tokenizer.decode(
-                    choice.token_ids, skip_special_tokens=False
+                decoded_text, _ = self._detokenize_delta(
+                    tokenizer,
+                    choice.token_ids,
+                    seed_state,
+                    skip_special_tokens=skip_special,
+                    spaces_between_special_tokens=spaces_between,
                 )
 
                 chat_template_kwargs: dict[str, Any] = {}
@@ -201,13 +231,12 @@ class OnlineDerenderer:
             else:
                 # No parser: plain detokenization honouring the request's
                 # skip_special_tokens (default True when no request was given).
-                skip_special = (
-                    chat_request.skip_special_tokens
-                    if chat_request is not None
-                    else True
-                )
-                decoded_text = tokenizer.decode(
-                    choice.token_ids, skip_special_tokens=skip_special
+                decoded_text, _ = self._detokenize_delta(
+                    tokenizer,
+                    choice.token_ids,
+                    seed_state,
+                    skip_special_tokens=skip_special,
+                    spaces_between_special_tokens=spaces_between,
                 )
                 message = ChatMessage(role="assistant", content=decoded_text)
 
@@ -239,7 +268,7 @@ class OnlineDerenderer:
         (tracked by ``read_offset``). The delta tokens are fed straight onto it.
 
         The window is bounded. ``detokenize_incrementally`` never reads before
-        ``prefix_offset``, so after processing we trim ``prev_tokens`` to that
+        ``prefix_offset``, so after each token we trim ``prev_tokens`` to that
         tail and rebase the offsets to it. State transport therefore stays
         O(window) per chunk instead of re-sending the full token history.
 
@@ -273,18 +302,20 @@ class OnlineDerenderer:
                 skip_special_tokens=skip_special_tokens,
                 spaces_between_special_tokens=spaces_between_special_tokens,
             )
-            prev_tokens = prev_tokens + new_toks
+            # Trim to the tail still readable by detokenize_incrementally
+            # (everything before prefix_offset is dead) and rebase the
+            # offsets, so the window stays bounded within a long batch
+            # decode as well as across chunks.
+            prev_tokens = (prev_tokens + new_toks)[prefix_offset:]
+            read_offset -= prefix_offset
+            prefix_offset = 0
             text_parts.append(text)
 
-        # Trim to the tail still readable by detokenize_incrementally
-        # (everything before prefix_offset is dead) and rebase the offsets so
-        # the carried window stays bounded regardless of generation length.
-        trimmed = prev_tokens[prefix_offset:]
         updated_state = state.model_copy(
             update={
-                "prev_tokens": trimmed,
-                "prefix_offset": 0,
-                "read_offset": read_offset - prefix_offset,
+                "prev_tokens": prev_tokens,
+                "prefix_offset": prefix_offset,
+                "read_offset": read_offset,
             }
         )
         return "".join(text_parts), updated_state
@@ -315,20 +346,16 @@ class OnlineDerenderer:
                 (validated by the caller — see `ServingDerender`) because
                 plain detokenization would leak raw parser markup into `content`.
             prompt_tokens: Prompt token count for the usage chunk.
-            prompt_token_ids: Prompt token IDs. Parser path only. Lets
-                `parse_delta` settle its initial reasoning state. Required
-                when a reasoning or tool parser is configured (validated by
-                the caller — see ``ServingDerender``). Without it reasoning
-                left open by the prompt would be misclassified as content.
+            prompt_token_ids: Prompt token IDs. Required on the parser path
+                (validated by the caller) and optional otherwise, falling
+                back to ``generate_chunk.prompt_token_ids``. See
+                `DerenderChatStreamRequest.prompt_token_ids`.
 
         Returns:
             (chunk, updated_state) — the derendered SSE chunk and the state
             the client must pass to the next call.
 
         """
-        if state is None:
-            state = DerenderStreamState()
-
         # A single DerenderStreamState is threaded through every choice in
         # this chunk. Correct only when there is at most one choice per SSE
         # event (n=1, one call per index), as the streaming derender
@@ -353,23 +380,34 @@ class OnlineDerenderer:
                 parser_cls,
                 model,
                 generate_chunk,
-                state,
+                state if state is not None else DerenderStreamState(),
                 chat_request,
                 prompt_tokens,
                 prompt_token_ids,
             )
 
         tokenizer = self.renderer.get_tokenizer()
-        skip_special = (
-            chat_request.skip_special_tokens if chat_request is not None else True
-        )
+        skip_special, spaces_between = _decode_params(tokenizer, chat_request)
+        # Seed on the first chunk only. A carried state already has the prompt.
+        if state is None:
+            state = _seed_stream_state(
+                tokenizer,
+                prompt_token_ids
+                if prompt_token_ids is not None
+                else generate_chunk.prompt_token_ids,
+                skip_special_tokens=skip_special,
+            )
         stream_choices: list[ChatCompletionResponseStreamChoice] = []
         updated_state = state
 
         for choice in generate_chunk.choices:
             delta_tids = choice.token_ids or []
             new_text, updated_state = await self._detokenize_delta_async(
-                tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special
+                tokenizer,
+                delta_tids,
+                updated_state,
+                skip_special_tokens=skip_special,
+                spaces_between_special_tokens=spaces_between,
             )
 
             # NOTE: parser-configured servers dispatch to
@@ -454,7 +492,8 @@ class OnlineDerenderer:
         `state.output_chunk_lens`, so the rebuilt parser state matches
         standard serving, which calls `parse_delta` once per engine step.
         Text comes from a fresh incremental detokenizer with special tokens
-        preserved (``skip_special_tokens=False``).
+        preserved (``skip_special_tokens=False``), seeded from the prompt
+        tail on every call since replay starts from scratch.
         """
         tokenizer = self.renderer.get_tokenizer()
 
@@ -480,7 +519,12 @@ class OnlineDerenderer:
         # across both the replay and current chunk phases (via `nonlocal`)
         # so multi-byte characters split across that boundary still decode
         # correctly. Discarded once the call returns.
-        detok_state = DerenderStreamState()
+        skip_special, spaces_between = _decode_params(
+            tokenizer, chat_request, preserve_special=True
+        )
+        detok_state = _seed_stream_state(
+            tokenizer, prompt_token_ids, skip_special_tokens=skip_special
+        )
 
         def _replay(token_ids: list[int], chunk_lens: list[int]) -> None:
             """Replay prior chunks through `parse_delta` to rebuild parser
@@ -493,7 +537,11 @@ class OnlineDerenderer:
                 chunk = token_ids[start : start + chunk_len]
                 start += chunk_len
                 text, detok_state = self._detokenize_delta(
-                    tokenizer, chunk, detok_state, skip_special_tokens=False
+                    tokenizer,
+                    chunk,
+                    detok_state,
+                    skip_special_tokens=skip_special,
+                    spaces_between_special_tokens=spaces_between,
                 )
                 parser.parse_delta(
                     text,
@@ -530,7 +578,11 @@ class OnlineDerenderer:
                 # granularity), not one per token. See the granularity note
                 # in this method's docstring.
                 text, detok_state = self._detokenize_delta(
-                    tokenizer, delta_tids, detok_state, skip_special_tokens=False
+                    tokenizer,
+                    delta_tids,
+                    detok_state,
+                    skip_special_tokens=skip_special,
+                    spaces_between_special_tokens=spaces_between,
                 )
                 delta_message = parser.parse_delta(
                     text,
@@ -633,9 +685,10 @@ class OnlineDerenderer:
         generate_responses: list[GenerateResponse],
         prompt_tokens: list[int] | None = None,
         completion_request: CompletionRequest | None = None,
+        prompt_token_ids: list[list[int] | None] | None = None,
     ) -> tuple[list[CompletionResponseChoice], int, int]:
         return await self._derender_completion_async(
-            generate_responses, prompt_tokens, completion_request
+            generate_responses, prompt_tokens, completion_request, prompt_token_ids
         )
 
     def _derender_completion(
@@ -643,24 +696,34 @@ class OnlineDerenderer:
         generate_responses: list[GenerateResponse],
         prompt_tokens: list[int] | None = None,
         completion_request: CompletionRequest | None = None,
+        prompt_token_ids: list[list[int] | None] | None = None,
     ) -> tuple[list[CompletionResponseChoice], int, int]:
         n = len(generate_responses)
         prompt_tokens_list: list[int] = (
             prompt_tokens if prompt_tokens is not None else [0] * n
         )
-
-        skip_special = (
-            completion_request.skip_special_tokens
-            if completion_request is not None
-            else True
+        prompt_token_ids_list: list[list[int] | None] = (
+            prompt_token_ids if prompt_token_ids is not None else [None] * n
         )
+
         tokenizer = self.renderer.get_tokenizer()
+        skip_special, spaces_between = _decode_params(tokenizer, completion_request)
         choices: list[CompletionResponseChoice] = []
         total_prompt_tokens = 0
         total_completion_tokens = 0
         index = 0
 
-        for gen, pt in zip(generate_responses, prompt_tokens_list):
+        for gen, pt, seed_ids_override in zip(
+            generate_responses, prompt_tokens_list, prompt_token_ids_list
+        ):
+            seed_ids = (
+                seed_ids_override
+                if seed_ids_override is not None
+                else gen.prompt_token_ids
+            )
+            seed_state = _seed_stream_state(
+                tokenizer, seed_ids, skip_special_tokens=skip_special
+            )
             for choice in gen.choices:
                 if not choice.token_ids:
                     raise ValueError(
@@ -668,8 +731,12 @@ class OnlineDerenderer:
                         "has empty or null token_ids"
                     )
 
-                decoded_text = tokenizer.decode(
-                    choice.token_ids, skip_special_tokens=skip_special
+                decoded_text, _ = self._detokenize_delta(
+                    tokenizer,
+                    choice.token_ids,
+                    seed_state,
+                    skip_special_tokens=skip_special,
+                    spaces_between_special_tokens=spaces_between,
                 )
                 completion_logprobs = None
                 if choice.logprobs is not None:
@@ -698,6 +765,7 @@ class OnlineDerenderer:
         state: DerenderStreamState | None = None,
         prompt_tokens: int | None = None,
         completion_request: CompletionRequest | None = None,
+        prompt_token_ids: list[int] | None = None,
     ) -> tuple[CompletionStreamResponse, DerenderStreamState]:
         """Process one GenerateStreamResponse chunk for streaming completions.
 
@@ -716,14 +784,13 @@ class OnlineDerenderer:
             prompt_tokens: Prompt token count for usage (from the render step).
             completion_request: Original CompletionRequest from ``/render``;
                 supplies ``skip_special_tokens``.
+            prompt_token_ids: Seeds the first chunk's decode. Falls back to
+                ``generate_chunk.prompt_token_ids``.
 
         Returns:
             (chunk, updated_state) — the derendered chunk and updated state.
 
         """
-        if state is None:
-            state = DerenderStreamState()
-
         # See the equivalent check in derender_chat_stream: a single
         # DerenderStreamState is threaded through every choice in this
         # chunk, so more than one choice per chunk would corrupt the
@@ -734,18 +801,27 @@ class OnlineDerenderer:
             )
 
         tokenizer = self.renderer.get_tokenizer()
-        skip_special = (
-            completion_request.skip_special_tokens
-            if completion_request is not None
-            else True
-        )
+        skip_special, spaces_between = _decode_params(tokenizer, completion_request)
+        # Seed on the first chunk only. A carried state already has the prompt.
+        if state is None:
+            state = _seed_stream_state(
+                tokenizer,
+                prompt_token_ids
+                if prompt_token_ids is not None
+                else generate_chunk.prompt_token_ids,
+                skip_special_tokens=skip_special,
+            )
         stream_choices: list[CompletionResponseStreamChoice] = []
         updated_state = state
 
         for choice in generate_chunk.choices:
             delta_tids = choice.token_ids or []
             new_text, updated_state = await self._detokenize_delta_async(
-                tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special
+                tokenizer,
+                delta_tids,
+                updated_state,
+                skip_special_tokens=skip_special,
+                spaces_between_special_tokens=spaces_between,
             )
 
             completion_logprobs = None
@@ -810,6 +886,65 @@ def _logprob_context_tail(
     return (list(context_token_ids) + list(delta_token_ids))[-_LOGPROB_CONTEXT_WINDOW:]
 
 
+def _decode_params(
+    tokenizer: TokenizerLike,
+    request: ChatCompletionRequest | CompletionRequest | None,
+    preserve_special: bool = False,
+) -> tuple[bool, bool]:
+    """Derive `(skip_special_tokens, spaces_between_special_tokens)` the way
+    the engine does (`IncrementalDetokenizer`).
+
+    Args:
+        tokenizer: Picks the engine's fast or slow detokenizer rules.
+        request: The original request, if the caller supplied one.
+        preserve_special: Keep special tokens so a parser can see markers.
+            The serving side does this via `adjust_request`.
+
+    """
+    skip_special = (
+        False
+        if preserve_special
+        else (request.skip_special_tokens if request is not None else True)
+    )
+    spaces_between = (
+        request.spaces_between_special_tokens if request is not None else True
+    )
+    # Only the fast detokenizer forces spaces on when skipping special tokens.
+    if skip_special and uses_fast_detokenizer(tokenizer):
+        spaces_between = True
+    return skip_special, spaces_between
+
+
+def _seed_stream_state(
+    tokenizer: TokenizerLike,
+    prompt_token_ids: list[int] | None,
+    skip_special_tokens: bool,
+) -> DerenderStreamState:
+    """Build the initial decode state from the prompt tail, the same way the
+    engine primes its incremental detokenizer. Without it, Metaspace
+    tokenizers drop the first output token's leading space.
+
+    Returns an empty state when `prompt_token_ids` is empty or omitted.
+    """
+    if not prompt_token_ids:
+        if prompt_token_ids is None and get_leading_space_marker(tokenizer) is not None:
+            logger.warning_once(
+                "derender got no prompt_token_ids, so the first output token "
+                "may lose its leading space compared to the coupled endpoint. "
+                "Pass prompt_token_ids from the render step to avoid this."
+            )
+        return DerenderStreamState()
+
+    prev_tokens, prefix_offset, read_offset = convert_prompt_ids_to_tokens(
+        tokenizer, prompt_token_ids, skip_special_tokens=skip_special_tokens
+    )
+    return DerenderStreamState(
+        prev_tokens=prev_tokens,
+        prefix_offset=prefix_offset,
+        read_offset=read_offset,
+    )
+
+
 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:"):
diff --git a/vllm/tokenizers/detokenizer_utils.py b/vllm/tokenizers/detokenizer_utils.py
index 35e73bc410..ddaf512442 100644
--- a/vllm/tokenizers/detokenizer_utils.py
+++ b/vllm/tokenizers/detokenizer_utils.py
@@ -63,7 +63,7 @@ _CACHED_MARKER_KEY = "_vllm_space_marker_cache"
 _NOT_CACHED = "__not_computed__"
 
 
-def _get_leading_space_marker(tokenizer: TokenizerLike) -> str | None:
+def get_leading_space_marker(tokenizer: TokenizerLike) -> str | None:
     """Read the space marker from the tokenizer's pre_tokenizer config.
 
     Only Metaspace pre_tokenizers (used by SentencePiece-based models like
@@ -160,7 +160,7 @@ def convert_ids_list_to_tokens(
     """
     if not token_ids:
         return []
-    marker = _get_leading_space_marker(tokenizer)
+    marker = get_leading_space_marker(tokenizer)
     if marker is None:
         return [tokenizer.decode([tid]) or "" for tid in token_ids]
     raw_tokens = tokenizer.convert_ids_to_tokens(token_ids)
diff --git a/vllm/v1/engine/detokenizer.py b/vllm/v1/engine/detokenizer.py
index 9658c030ef..c27e01895c 100644
--- a/vllm/v1/engine/detokenizer.py
+++ b/vllm/v1/engine/detokenizer.py
@@ -3,9 +3,7 @@
 import sys
 from abc import ABC, abstractmethod
 
-import tokenizers
 import tokenizers.decoders
-from packaging import version
 from tokenizers import Tokenizer
 from transformers import TokenizersBackend
 
@@ -20,14 +18,15 @@ from vllm.v1.engine import EngineCoreRequest
 
 logger = init_logger(__name__)
 
-# Only tokenizers >= 0.22.0 supports DecodeStream with native prefill
-# (ids parameter) used for FastIncrementalDetokenizer.
-USE_FAST_DETOKENIZER = version.parse(tokenizers.__version__) >= version.parse("0.22.0")
-
 # Error string from https://github.com/huggingface/tokenizers/blob/909fdde2a4ffedd9295206f705eb612be2a91b12/tokenizers/src/tokenizer/mod.rs#L1042
 INVALID_PREFIX_ERR_MSG = "Invalid prefix encountered"
 
 
+def uses_fast_detokenizer(tokenizer: TokenizerLike) -> bool:
+    """Whether the engine detokenizes with `FastIncrementalDetokenizer`."""
+    return isinstance(tokenizer, TokenizersBackend)
+
+
 class IncrementalDetokenizer:
     def __init__(self):
         self.token_ids: list[int] = []
@@ -58,7 +57,7 @@ class IncrementalDetokenizer:
             # No tokenizer => skipping detokenization.
             return IncrementalDetokenizer()
 
-        if USE_FAST_DETOKENIZER and isinstance(tokenizer, TokenizersBackend):
+        if uses_fast_detokenizer(tokenizer):
             # Fast tokenizer => use tokenizers library DecodeStream.
             return FastIncrementalDetokenizer(tokenizer, request)