[Frontend] Add stream reasoning and tool calls from the derender endpoint (#50550)
Signed-off-by: Martin Hickey <martin.hickey@ie.ibm.com>
Martin Hickey committed
Sep 18, 2026 at 15:36 UTC
70164bdadc048ce2697ea92b2dbb2d39f881e65e
6 files changed
+1652
-98
docs/serving/online_serving/derenderer.md
+18
-3
@@ -6,7 +6,7 @@ This closes the loop for a token-in / token-out engine in disaggregated serving:
6
7
- **GPU less post processing**: Detokenization, reasoning parsing, and tool call parsing run on the same GPU less frontend that hosts `/render`
8
- **Parser parity**: The derenderer reuses vLLM's tool and reasoning parsers, so a disaggregated deployment produces the same `content`/`reasoning`/ `tool_calls` split as a standard `vllm serve` server
9
-- **Non-streaming**: The endpoints expect a complete `GenerateResponse` with all token IDs present and perform one-shot parsing. Streaming derender would require a separate endpoint design and is not currently supported but is in the pipeline
9
+- **Streaming and one-shot parsing**: Non-streaming calls send a complete `GenerateResponse` with all token IDs present and get one-shot parsing. Both endpoints also accept `stream: true`, taking one `GenerateStreamResponse` delta plus a client carried `stream_state` and returning `{chunk, stream_state}`. The chat endpoint's streaming path supports reasoning and tool call parsing while emitting the same `reasoning`/`content`/`tool_calls` deltas the generate streaming path would for the same token IDs
10
11
Both endpoints are hosted by the GPU less rendering server started with [`vllm launch render`](../../cli/launch/render.md), alongside the `/render`
12
endpoints.
@@ -21,7 +21,7 @@ endpoints.
21
└─────────────── request + prompt_tokens ──┘
22
```
23
24
-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.
24
+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.
25
26
## API Reference
27
@@ -52,6 +52,21 @@ Each request wraps the engine's `GenerateResponse`(s) together with the caller m
52
53
Oversized payloads are rejected with a `400` before any `tokenizer.decode()` or parser runs.
54
55
+## Streaming cost
56
+
57
+Streaming derender threads a client carried `stream_state` across per-chunk calls instead of keeping session state on the server.
58
+
59
+Plain detokenization (no parser configured) carries only a small, bounded incremental decode window in `stream_state` independent of generation length.
60
+
61
+When a tool or reasoning parser is configured, parser internal state (buffered markup, reasoning/tool phase) can't be serialized. `stream_state` therefore instead carries the full `output_token_ids` seen so far plus `output_chunk_lens`, the token count of each chunk they arrived in. Each chunk rebuilds a fresh parser and replays that history through `parse_delta`, one call per original chunk, before processing the new tokens for real. For the parser path only, this means:
62
+
63
+- **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.
64
+- **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.
65
+- 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.
66
+- 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.
67
+
68
+`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.
69
+
70
## Example
71
72
The example below drives the full `render → generate → derender` round trip for a chat request against a GPU less render server (`/render`, `/derender`) and a token-in / token-out engine (`/inference/v1/generate`).
@@ -102,4 +117,4 @@ with httpx.Client(timeout=60.0) as client:
117
print(response["choices"][0]["message"]["content"])
118
```
119
105
-Passing `chat_request` lets the derenderer run the configured tool and reasoning parsers. This means `response["choices"][0]["message"]` carries the same `content` / `reasoning` / `tool_calls` split a `vllm serve` server would produce. Omit `chat_request` for plain detokenization only.
120
+Passing `chat_request` lets the derenderer run the configured tool and reasoning parsers. This means `response["choices"][0]["message"]` carries the same `content` / `reasoning` / `tool_calls` split a `vllm serve` server would produce. `chat_request` can only be omitted for a model with no tool or reasoning parser configured. A parser configured model rejects a missing `chat_request` with a 400 rather than silently falling back to plain detokenization.
tests/entrypoints/scale_out/derender/test_derender.py
+32
-14
@@ -800,7 +800,12 @@ def _e2e_generate_response(
800
801
@pytest.mark.asyncio
802
async def test_e2e_plain_roundtrip(parser_client, parser_tokenizer):
803
- """Plain text without reasoning markers roundtrips correctly."""
803
+ """Plain text without reasoning markers roundtrips correctly.
804
+
805
+ Markerless output has no ``</think>``, which deepseek_r1 classifies
806
+ wholly as reasoning, so the text lands there rather than in content.
807
+ What this pins is detokenization fidelity through the parser path.
808
+ """
809
messages = [{"role": "user", "content": "What is 2+2?"}]
810
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
811
@@ -814,16 +819,22 @@ async def test_e2e_plain_roundtrip(parser_client, parser_tokenizer):
819
"model": PARSER_MODEL,
820
"generate_response": _e2e_generate_response(output_ids),
821
"prompt_tokens": len(gen_req["token_ids"]),
822
+ "chat_request": {"model": PARSER_MODEL, "messages": messages},
823
},
824
)
825
assert resp.status_code == 200, resp.text
820
- content = resp.json()["choices"][0]["message"]["content"]
821
- assert content == expected
826
+ message = resp.json()["choices"][0]["message"]
827
+ assert message["reasoning"] == expected
828
+ assert message["content"] is None
829
830
831
@pytest.mark.asyncio
832
async def test_e2e_token_identity(parser_client, parser_tokenizer):
826
- """encode(derender(token_ids)) == token_ids (RL invariant)."""
833
+ """encode(derender(token_ids)) == token_ids (RL invariant).
834
+
835
+ Markerless output comes back as reasoning (see
836
+ ``test_e2e_plain_roundtrip``), so re-encode that.
837
+ """
838
messages = [{"role": "user", "content": "Hi"}]
839
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
840
@@ -836,17 +847,22 @@ async def test_e2e_token_identity(parser_client, parser_tokenizer):
847
"model": PARSER_MODEL,
848
"generate_response": _e2e_generate_response(output_ids),
849
"prompt_tokens": len(gen_req["token_ids"]),
850
+ "chat_request": {"model": PARSER_MODEL, "messages": messages},
851
},
852
)
853
assert resp.status_code == 200
842
- content = resp.json()["choices"][0]["message"]["content"]
843
- re_encoded = _encode(parser_tokenizer, content)
854
+ reasoning = resp.json()["choices"][0]["message"]["reasoning"]
855
+ re_encoded = _encode(parser_tokenizer, reasoning)
856
assert output_ids == re_encoded
857
858
859
@pytest.mark.asyncio
860
async def test_e2e_non_ascii_roundtrip(parser_client, parser_tokenizer):
849
- """CJK + emoji roundtrip without U+FFFD."""
861
+ """CJK + emoji roundtrip without U+FFFD.
862
+
863
+ Markerless output comes back as reasoning (see
864
+ ``test_e2e_plain_roundtrip``).
865
+ """
866
messages = [{"role": "user", "content": "Reply in Chinese"}]
867
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
868
@@ -859,11 +875,12 @@ async def test_e2e_non_ascii_roundtrip(parser_client, parser_tokenizer):
875
"model": PARSER_MODEL,
876
"generate_response": _e2e_generate_response(output_ids),
877
"prompt_tokens": len(gen_req["token_ids"]),
878
+ "chat_request": {"model": PARSER_MODEL, "messages": messages},
879
},
880
)
881
assert resp.status_code == 200
865
- content = resp.json()["choices"][0]["message"]["content"]
866
- assert "�" not in content
882
+ reasoning = resp.json()["choices"][0]["message"]["reasoning"]
883
+ assert "�" not in reasoning
884
885
886
@pytest.mark.asyncio
@@ -976,8 +993,10 @@ async def test_e2e_parsed_reasoning_and_tool_call(parser_client, parser_tokenize
993
994
995
@pytest.mark.asyncio
979
-async def test_e2e_no_chat_request_fallback(parser_client, parser_tokenizer):
980
- """Without chat_request, derender falls back to plain detokenization."""
996
+async def test_e2e_no_chat_request_rejected(parser_client, parser_tokenizer):
997
+ """Without chat_request a parser configured model rejects with 400
998
+ rather than silently falling back to plain detokenization. This is to
999
+ prevent the leak of raw reasoning/tool markup into content."""
1000
messages = [{"role": "user", "content": "Hello"}]
1001
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
1002
@@ -992,9 +1011,8 @@ async def test_e2e_no_chat_request_fallback(parser_client, parser_tokenizer):
1011
"prompt_tokens": len(gen_req["token_ids"]),
1012
},
1013
)
995
- assert resp.status_code == 200
996
- content = resp.json()["choices"][0]["message"]["content"]
997
- assert "Hi" in content
1014
+ assert resp.status_code == 400
1015
+ assert "chat_request" in resp.json()["error"]["message"]
1016
1017
1018
# ---------------------------------------------------------------------------
tests/entrypoints/scale_out/derender/test_derender_stream.py
+1204
-32
@@ -8,6 +8,8 @@ Tests are split into two layers:
8
1. Unit tests (no server): covers ``_detokenize_delta`` correctness
9
(chunked == one-shot) and ``derender_completion_stream`` /
10
``derender_chat_stream`` logic via a real tokenizer on a tiny model.
11
+ The parser path is covered both with a deterministic stub parser and
12
+ with the real ``HarmonyParser`` (skipped without ``openai_harmony``).
13
14
2. Integration tests (require a running render server): covers the full
15
HTTP round-trip through the streaming endpoint. Marked with
@@ -15,19 +17,142 @@ Tests are split into two layers:
17
fixtures from the sibling ``test_derender.py``.
18
"""
19
20
+import json
21
+from collections.abc import Callable
22
+
23
import pytest
24
import pytest_asyncio
25
26
+from vllm.entrypoints.generate.base.protocol import (
27
+ DeltaFunctionCall,
28
+ DeltaMessage,
29
+ DeltaToolCall,
30
+)
31
from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
32
DerenderStreamState,
33
+ GenerateResponse,
34
+ GenerateResponseChoice,
35
GenerateResponseStreamChoice,
36
GenerateStreamResponse,
37
)
38
from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo
39
+from vllm.parser import Parser
40
+from vllm.utils import random_uuid
41
42
MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
43
44
45
+def _merge_fake_deltas(
46
+ accumulated: DeltaMessage | None, new: DeltaMessage | None
47
+) -> DeltaMessage | None:
48
+ """Fold one token's worth of `_FakeParser` output into a per call
49
+ accumulator, the same way a real parser merges multiple tokens'
50
+ effects into the single `DeltaMessage` it returns per `parse_delta`
51
+ call.
52
+ """
53
+ if new is None:
54
+ return accumulated
55
+ if accumulated is None:
56
+ accumulated = DeltaMessage()
57
+
58
+ if new.content:
59
+ accumulated.content = (accumulated.content or "") + new.content
60
+ if new.reasoning:
61
+ accumulated.reasoning = (accumulated.reasoning or "") + new.reasoning
62
+
63
+ for tc in new.tool_calls:
64
+ starts_new_call = tc.id is not None or (
65
+ tc.function is not None and tc.function.name is not None
66
+ )
67
+ existing = (
68
+ None
69
+ if starts_new_call
70
+ else next((t for t in accumulated.tool_calls if t.index == tc.index), None)
71
+ )
72
+ if existing is None:
73
+ accumulated.tool_calls.append(tc)
74
+ continue
75
+ if existing.function is None:
76
+ existing.function = DeltaFunctionCall()
77
+ if tc.function is not None and tc.function.arguments:
78
+ existing.function.arguments = (
79
+ existing.function.arguments or ""
80
+ ) + tc.function.arguments
81
+
82
+ return accumulated
83
+
84
+
85
+class _FakeParser(Parser):
86
+ """Deterministic `Parser` stub for unit testing the replay/merge/pin
87
+ mechanics of `OnlineDerenderer._derender_chat_stream_parsed` in
88
+ isolation from any real reasoning/tool parser's markup grammar.
89
+
90
+ Processes every id in `delta_token_ids` in order and folds their
91
+ effects into one `DeltaMessage`, like a real parser handling a
92
+ multi-token delta in a single call. Ignores `delta_text`:
93
+
94
+ - `TOOL_START` opens a new tool call at index 0 (id + name).
95
+ - `TOOL_ARG` appends one `"a"` to that tool call's arguments.
96
+ - `REASON` emits one `"r"` of reasoning (suppressed when the request
97
+ has `include_reasoning=False`, mirroring the real parsers).
98
+ - `CONTENT` emits one `"c"` of content.
99
+ - An empty `delta_token_ids` with `finished=True` (the finish only
100
+ flush call) emits a `"FLUSH"` content sentinel so tests can confirm
101
+ it happened.
102
+ - Anything else emits nothing.
103
+ """
104
+
105
+ TOOL_START = 9
106
+ TOOL_ARG = 10
107
+ REASON = 11
108
+ CONTENT = 12
109
+
110
+ def parse_delta(
111
+ self,
112
+ delta_text,
113
+ delta_token_ids,
114
+ request,
115
+ prompt_token_ids=None,
116
+ *,
117
+ finished,
118
+ ):
119
+ if not delta_token_ids:
120
+ return DeltaMessage(content="FLUSH") if finished else None
121
+
122
+ accumulated: DeltaMessage | None = None
123
+ for tok in delta_token_ids:
124
+ accumulated = _merge_fake_deltas(
125
+ accumulated, self._parse_one_token(tok, request)
126
+ )
127
+ return accumulated
128
+
129
+ def _parse_one_token(self, tok: int, request) -> DeltaMessage | None:
130
+ if tok == self.TOOL_START:
131
+ return DeltaMessage(
132
+ tool_calls=[
133
+ DeltaToolCall(
134
+ id=f"call-{random_uuid()}",
135
+ type="function",
136
+ function=DeltaFunctionCall(name="get_weather", arguments=""),
137
+ index=0,
138
+ )
139
+ ]
140
+ )
141
+ if tok == self.TOOL_ARG:
142
+ return DeltaMessage(
143
+ tool_calls=[
144
+ DeltaToolCall(index=0, function=DeltaFunctionCall(arguments="a"))
145
+ ]
146
+ )
147
+ if tok == self.REASON:
148
+ if not request.include_reasoning:
149
+ return None
150
+ return DeltaMessage(reasoning="r")
151
+ if tok == self.CONTENT:
152
+ return DeltaMessage(content="c")
153
+ return None
154
+
155
+
156
# ---------------------------------------------------------------------------
157
# Helpers shared across tests
158
# ---------------------------------------------------------------------------
@@ -111,6 +236,40 @@ def derenderer(tokenizer):
236
)
237
238
239
+@pytest.fixture(scope="module")
240
+def parsed_derenderer(tokenizer):
241
+ """OnlineDerenderer with `_FakeParser` wired in as `self.parser`.
242
+
243
+ The parser configured path runs `_derender_chat_stream_parsed` via
244
+ `make_async(..., executor=renderer._executor)`, so unlike
245
+ `derenderer` above (whose streaming paths never touch the executor),
246
+ `renderer._executor` must be a real `ThreadPoolExecutor` here.
247
+ `loop.run_in_executor` cannot submit work to a `MagicMock`.
248
+ """
249
+ from concurrent.futures import ThreadPoolExecutor
250
+ from unittest.mock import MagicMock
251
+
252
+ from vllm.renderers.online_derenderer import OnlineDerenderer
253
+
254
+ renderer = MagicMock()
255
+ renderer.get_tokenizer.return_value = tokenizer
256
+ renderer._executor = ThreadPoolExecutor(max_workers=2)
257
+
258
+ model_config = MagicMock()
259
+ model_config.hf_config.model_type = "llama"
260
+ model_config.model = MODEL_NAME
261
+
262
+ dr = OnlineDerenderer(
263
+ model_config=model_config,
264
+ renderer=renderer,
265
+ request_logger=None,
266
+ chat_template=None,
267
+ chat_template_content_format="string",
268
+ )
269
+ dr.parser = _FakeParser
270
+ return dr
271
+
272
+
273
class TestDetokenizeDelta:
274
"""_detokenize_delta: chunked decode must equal one shot decode."""
275
@@ -378,47 +537,454 @@ class TestDerenderChatStream:
537
assert streamed == one_shot
538
539
@pytest.mark.asyncio
381
- @pytest.mark.parametrize("with_chat_request", [True, False])
382
- async def test_parser_raises_not_implemented(self, tokenizer, with_chat_request):
383
- """Stream chat with a parser active must fail closed (NotImplementedError).
540
+ async def test_parser_configured_missing_chat_request_raises(
541
+ self, parsed_derenderer
542
+ ):
543
+ """A parser configured model must never fall through to plain detok
544
+ even when `chat_request` is omitted. If allowed this would leak raw
545
+ reasoning/tool markup into `delta.content`. `ServingDerender`
546
+ pre-checks this too (400 before touching the tokenizer). This pins
547
+ the `OnlineDerenderer` level backstop."""
548
+ with pytest.raises(ValueError, match="chat_request"):
549
+ await parsed_derenderer.derender_chat_stream(
550
+ model=MODEL_NAME,
551
+ generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
552
+ state=None,
553
+ chat_request=None,
554
+ )
555
385
- Checks that a parser configured model must 501 even when ``chat_request`` is
386
- omitted, otherwise reasoning/tool markup would leak into ``delta.content`
387
- via the plain detok fallback.
388
- """
389
- from unittest.mock import MagicMock
556
391
- from vllm.renderers.online_derenderer import OnlineDerenderer
557
+def _chat_request(**kwargs):
558
+ from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
559
+
560
+ kwargs.setdefault("messages", [{"role": "user", "content": "hi"}])
561
+ kwargs.setdefault("model", MODEL_NAME)
562
+ return ChatCompletionRequest(**kwargs)
563
393
- renderer = MagicMock()
394
- renderer.get_tokenizer.return_value = tokenizer
564
396
- model_config = MagicMock()
397
- model_config.hf_config.model_type = "llama"
398
- model_config.model = MODEL_NAME
565
+class TestDerenderChatStreamParsed:
566
+ """derender_chat_stream: parser branch (replay + parse_delta) exercised
567
+ against the deterministic `_FakeParser` so these pin OnlineDerenderer's
568
+ own replay/merge/pin/finish_reason logic independent of any real
569
+ parser's markup grammar (covered separately by the parser_server backed
570
+ integration tests below)."""
571
400
- # Construct a derenderer WITH a tool/reasoning parser active.
401
- # ParserManager.get_parser returns None when no parser is named, so
402
- # we inject a mock parser class directly.
403
- dr = OnlineDerenderer(
404
- model_config=model_config,
405
- renderer=renderer,
406
- request_logger=None,
407
- chat_template=None,
408
- chat_template_content_format="string",
572
+ @pytest.mark.asyncio
573
+ async def test_dispatches_and_emits_content(self, parsed_derenderer):
574
+ chunk, state = await parsed_derenderer.derender_chat_stream(
575
+ model=MODEL_NAME,
576
+ generate_chunk=_make_stream_chunk(
577
+ [_FakeParser.CONTENT], finish_reason="stop"
578
+ ),
579
+ chat_request=_chat_request(),
580
)
410
- dr.parser = MagicMock() # simulate a parser being active
581
+ assert chunk.choices[0].delta.content == "c"
582
+ assert chunk.choices[0].delta.role == "assistant"
583
+ assert chunk.choices[0].finish_reason == "stop"
584
+ assert state.output_token_ids == [_FakeParser.CONTENT]
585
412
- chat_request = MagicMock() if with_chat_request else None
413
- token_ids = tokenizer.encode("hello")[:3]
586
+ @pytest.mark.asyncio
587
+ async def test_role_sent_once(self, parsed_derenderer):
588
+ chat_request = _chat_request()
589
+ chunk1, state = await parsed_derenderer.derender_chat_stream(
590
+ model=MODEL_NAME,
591
+ generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
592
+ chat_request=chat_request,
593
+ )
594
+ chunk2, _ = await parsed_derenderer.derender_chat_stream(
595
+ model=MODEL_NAME,
596
+ generate_chunk=_make_stream_chunk(
597
+ [_FakeParser.CONTENT], finish_reason="stop"
598
+ ),
599
+ state=state,
600
+ chat_request=chat_request,
601
+ )
602
+ assert chunk1.choices[0].delta.role == "assistant"
603
+ assert chunk2.choices[0].delta.role is None
604
415
- with pytest.raises(NotImplementedError):
416
- await dr.derender_chat_stream(
417
- model=MODEL_NAME,
418
- generate_chunk=_make_stream_chunk(token_ids),
419
- state=None,
420
- chat_request=chat_request,
605
+ @pytest.mark.asyncio
606
+ async def test_finish_only_chunk_flushes(self, parsed_derenderer):
607
+ """A finish only chunk (no new tokens) still calls `parse_delta`
608
+ once with `finished=True` to flush buffered state."""
609
+ chat_request = _chat_request()
610
+ _, state = await parsed_derenderer.derender_chat_stream(
611
+ model=MODEL_NAME,
612
+ generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
613
+ chat_request=chat_request,
614
+ )
615
+ chunk2, _ = await parsed_derenderer.derender_chat_stream(
616
+ model=MODEL_NAME,
617
+ generate_chunk=_make_stream_chunk([], finish_reason="stop"),
618
+ state=state,
619
+ chat_request=chat_request,
620
+ )
621
+ assert chunk2.choices[0].delta.content == "FLUSH"
622
+ assert chunk2.choices[0].finish_reason == "stop"
623
+
624
+ @pytest.mark.asyncio
625
+ async def test_chunking_invariance(self, parsed_derenderer):
626
+ """1 token per chunk vs a single whole chunk assemble identically.
627
+
628
+ Every chunk (live or replayed) is fed to `parse_delta` at its own
629
+ producer granularity (see `_derender_chat_stream_parsed`'s
630
+ docstring). This proves that choice doesn't change the assembled
631
+ output for a well behaved incremental parser. Each call's effects
632
+ fold together the same way regardless of how many tokens land in
633
+ one call vs. many."""
634
+ token_ids = [
635
+ _FakeParser.REASON,
636
+ _FakeParser.REASON,
637
+ _FakeParser.TOOL_START,
638
+ _FakeParser.TOOL_ARG,
639
+ _FakeParser.TOOL_ARG,
640
+ _FakeParser.CONTENT,
641
+ ]
642
+ chat_request = _chat_request(
643
+ tools=[{"type": "function", "function": {"name": "get_weather"}}],
644
+ tool_choice="auto",
645
+ include_reasoning=True,
646
+ )
647
+
648
+ async def _assemble(chunks: list[list[int]]) -> dict:
649
+ state = None
650
+ content = ""
651
+ reasoning = ""
652
+ tool_args = ""
653
+ for i, tids in enumerate(chunks):
654
+ finish = "stop" if i == len(chunks) - 1 else None
655
+ chunk, state = await parsed_derenderer.derender_chat_stream(
656
+ model=MODEL_NAME,
657
+ generate_chunk=_make_stream_chunk(tids, finish_reason=finish),
658
+ state=state,
659
+ chat_request=chat_request,
660
+ )
661
+ delta = chunk.choices[0].delta
662
+ content += delta.content or ""
663
+ reasoning += delta.reasoning or ""
664
+ for tc in delta.tool_calls:
665
+ if tc.function and tc.function.arguments:
666
+ tool_args += tc.function.arguments
667
+ return {"content": content, "reasoning": reasoning, "tool_args": tool_args}
668
+
669
+ whole = await _assemble([token_ids])
670
+ one_at_a_time = await _assemble([[t] for t in token_ids])
671
+
672
+ assert whole == one_at_a_time
673
+ assert whole == {"content": "c", "reasoning": "rr", "tool_args": "aa"}
674
+
675
+ @pytest.mark.asyncio
676
+ async def test_replay_preserves_chunk_boundaries(
677
+ self, parsed_derenderer, monkeypatch
678
+ ):
679
+ """A multi-token chunk (e.g. a speculative decoding step) must be
680
+ replayed as one `parse_delta` call, not one per token. Parsers
681
+ with per-call state such as `history_tool_call_cnt` would
682
+ otherwise diverge from standard serving."""
683
+
684
+ class _CallCountingParser(_FakeParser):
685
+ """Emits how many `parse_delta` calls it has seen."""
686
+
687
+ def __init__(self, *args, **kwargs):
688
+ super().__init__(*args, **kwargs)
689
+ self.num_calls = 0
690
+
691
+ def parse_delta(
692
+ self,
693
+ delta_text,
694
+ delta_token_ids,
695
+ request,
696
+ prompt_token_ids=None,
697
+ *,
698
+ finished,
699
+ ):
700
+ self.num_calls += 1
701
+ return DeltaMessage(content=str(self.num_calls))
702
+
703
+ monkeypatch.setattr(parsed_derenderer, "parser", _CallCountingParser)
704
+ chat_request = _chat_request()
705
+
706
+ _, state = await parsed_derenderer.derender_chat_stream(
707
+ model=MODEL_NAME,
708
+ generate_chunk=_make_stream_chunk([_FakeParser.CONTENT] * 3),
709
+ chat_request=chat_request,
710
+ )
711
+ assert state.output_chunk_lens == [3]
712
+
713
+ chunk, _ = await parsed_derenderer.derender_chat_stream(
714
+ model=MODEL_NAME,
715
+ generate_chunk=_make_stream_chunk(
716
+ [_FakeParser.CONTENT], finish_reason="stop"
717
+ ),
718
+ state=state,
719
+ chat_request=chat_request,
720
+ )
721
+ assert chunk.choices[0].delta.content == "2"
722
+
723
+ @pytest.mark.asyncio
724
+ async def test_tool_call_id_pinned_across_chunks(self, parsed_derenderer):
725
+ """Once an index's ID is recorded in `last_tool_call_ids`, a later
726
+ id bearing delta for that same index is pinned to the recorded
727
+ value rather than a freshly (re-)generated one."""
728
+ chat_request = _chat_request(
729
+ tools=[{"type": "function", "function": {"name": "get_weather"}}],
730
+ tool_choice="auto",
731
+ )
732
+ chunk1, state = await parsed_derenderer.derender_chat_stream(
733
+ model=MODEL_NAME,
734
+ generate_chunk=_make_stream_chunk([_FakeParser.TOOL_START]),
735
+ chat_request=chat_request,
736
+ )
737
+ first_id = chunk1.choices[0].delta.tool_calls[0].id
738
+ assert first_id is not None
739
+ assert state.last_tool_call_ids == [first_id]
740
+
741
+ chunk2, state = await parsed_derenderer.derender_chat_stream(
742
+ model=MODEL_NAME,
743
+ generate_chunk=_make_stream_chunk(
744
+ [_FakeParser.TOOL_START], finish_reason="stop"
745
+ ),
746
+ state=state,
747
+ chat_request=chat_request,
748
+ )
749
+ assert chunk2.choices[0].delta.tool_calls[0].id == first_id
750
+ assert state.last_tool_call_ids == [first_id]
751
+
752
+ @pytest.mark.asyncio
753
+ async def test_finish_reason_rewritten_to_tool_calls(self, parsed_derenderer):
754
+ chat_request = _chat_request(
755
+ tools=[{"type": "function", "function": {"name": "get_weather"}}],
756
+ tool_choice="auto",
757
+ )
758
+ chunk, _ = await parsed_derenderer.derender_chat_stream(
759
+ model=MODEL_NAME,
760
+ generate_chunk=_make_stream_chunk(
761
+ [_FakeParser.TOOL_START, _FakeParser.TOOL_ARG], finish_reason="stop"
762
+ ),
763
+ chat_request=chat_request,
764
+ )
765
+ assert chunk.choices[0].finish_reason == "tool_calls"
766
+
767
+ @pytest.mark.asyncio
768
+ async def test_finish_reason_stop_for_named_tool_choice(self, parsed_derenderer):
769
+ # ChatCompletionRequest's `check_tool_usage` is a mode="before"
770
+ # validator, so it sees the raw value and only accepts "auto",
771
+ # "required" or a dict, never an already built
772
+ # ChatCompletionNamedToolChoiceParam.
773
+ chat_request = _chat_request(
774
+ tools=[{"type": "function", "function": {"name": "get_weather"}}],
775
+ tool_choice={"type": "function", "function": {"name": "get_weather"}},
776
+ )
777
+ chunk, _ = await parsed_derenderer.derender_chat_stream(
778
+ model=MODEL_NAME,
779
+ generate_chunk=_make_stream_chunk(
780
+ [_FakeParser.TOOL_START, _FakeParser.TOOL_ARG], finish_reason="stop"
781
+ ),
782
+ chat_request=chat_request,
783
+ )
784
+ assert chunk.choices[0].finish_reason == "stop"
785
+
786
+ @pytest.mark.asyncio
787
+ async def test_include_reasoning_false_suppresses_reasoning(
788
+ self, parsed_derenderer
789
+ ):
790
+ chat_request = _chat_request(include_reasoning=False)
791
+ chunk, _ = await parsed_derenderer.derender_chat_stream(
792
+ model=MODEL_NAME,
793
+ generate_chunk=_make_stream_chunk(
794
+ [_FakeParser.REASON, _FakeParser.CONTENT], finish_reason="stop"
795
+ ),
796
+ chat_request=chat_request,
797
+ )
798
+ delta = chunk.choices[0].delta
799
+ assert delta.reasoning is None
800
+ assert delta.content == "c"
801
+
802
+
803
+# ---------------------------------------------------------------------------
804
+# Harmony / GPT-OSS replay — unit, no server
805
+# ---------------------------------------------------------------------------
806
+
807
+HARMONY_MODEL = "openai/gpt-oss-20b"
808
+HARMONY_REASONING = "The user wants 2 plus 3."
809
+HARMONY_ANSWER = "The answer is 5."
810
+HARMONY_PROMPT = "<|start|>user<|message|>Add 2 and 3.<|end|><|start|>assistant"
811
+HARMONY_OUTPUT = (
812
+ f"<|channel|>analysis<|message|>{HARMONY_REASONING}<|end|>"
813
+ f"<|start|>assistant<|channel|>final<|message|>{HARMONY_ANSWER}<|return|>"
814
+)
815
+
816
+
817
+@pytest.fixture(scope="module")
818
+def harmony_encode():
819
+ """Encoder for canned GPT-OSS harmony token sequences."""
820
+ pytest.importorskip("openai_harmony")
821
+
822
+ # Pre-caches the o200k_base BPE file that openai-harmony's Rust backend
823
+ # downloads on first use, same as the sibling suite's E2E harmony tests.
824
+ from tests.entrypoints.scale_out.derender.test_derender import (
825
+ _ensure_harmony_vocab,
826
+ )
827
+ from vllm.entrypoints.openai.parser.harmony_utils import get_encoding
828
+
829
+ _ensure_harmony_vocab()
830
+ encoding = get_encoding()
831
+
832
+ def _encode(harmony_str: str) -> list[int]:
833
+ return encoding.encode(harmony_str, allowed_special="all")
834
+
835
+ return _encode
836
+
837
+
838
+@pytest.fixture(scope="module")
839
+def harmony_tokenizer():
840
+ pytest.importorskip("openai_harmony")
841
+ from vllm.tokenizers import get_tokenizer
842
+
843
+ return get_tokenizer(HARMONY_MODEL, trust_remote_code=True)
844
+
845
+
846
+@pytest.fixture(scope="module")
847
+def harmony_derenderer(harmony_tokenizer):
848
+ """OnlineDerenderer whose parser resolves to the real `HarmonyParser`.
849
+
850
+ Same shape as `parsed_derenderer` (mocked renderer, real executor) but
851
+ with a real tokenizer and parser, so the replay path is exercised
852
+ against Harmony's actual channel grammar rather than a stub.
853
+ """
854
+ from concurrent.futures import ThreadPoolExecutor
855
+ from unittest.mock import MagicMock
856
+
857
+ from vllm.parser.harmony import HarmonyParser
858
+ from vllm.renderers.online_derenderer import OnlineDerenderer
859
+
860
+ renderer = MagicMock()
861
+ renderer.get_tokenizer.return_value = harmony_tokenizer
862
+ renderer._executor = ThreadPoolExecutor(max_workers=2)
863
+
864
+ model_config = MagicMock()
865
+ model_config.model = HARMONY_MODEL
866
+ model_config.hf_config.model_type = "gpt_oss"
867
+ model_config.hf_text_config.model_type = "gpt_oss"
868
+ model_config.hf_overrides = None
869
+
870
+ dr = OnlineDerenderer(
871
+ model_config=model_config,
872
+ renderer=renderer,
873
+ request_logger=None,
874
+ chat_template=None,
875
+ chat_template_content_format="string",
876
+ enable_auto_tools=True,
877
+ tool_parser="openai",
878
+ reasoning_parser="openai_gptoss",
879
+ )
880
+ assert dr.use_harmony
881
+ assert dr.parser is HarmonyParser
882
+ return dr
883
+
884
+
885
+async def _stream_harmony_deltas(
886
+ derenderer,
887
+ chat_request,
888
+ output_ids: list[int],
889
+ prompt_ids: list[int],
890
+ chunk_size: int,
891
+) -> list[DeltaMessage]:
892
+ """Drive `output_ids` through the parser path in fixed size chunks.
893
+
894
+ Returns the per chunk `DeltaMessage`s in order so callers can assert on
895
+ intermediate emissions, not just the assembled result.
896
+ """
897
+ deltas: list[DeltaMessage] = []
898
+ state = None
899
+ for start in range(0, len(output_ids), chunk_size):
900
+ tids = output_ids[start : start + chunk_size]
901
+ is_last = start + chunk_size >= len(output_ids)
902
+ chunk, state = await derenderer.derender_chat_stream(
903
+ model=HARMONY_MODEL,
904
+ generate_chunk=_make_stream_chunk(
905
+ tids, finish_reason="stop" if is_last else None
906
+ ),
907
+ state=state,
908
+ chat_request=chat_request,
909
+ prompt_token_ids=prompt_ids,
910
+ )
911
+ deltas.append(chunk.choices[0].delta)
912
+ return deltas
913
+
914
+
915
+class TestDerenderChatStreamHarmony:
916
+ """derender_chat_stream: replay + `parse_delta` against real HarmonyParser.
917
+
918
+ Skipped where `openai_harmony` is not installed. `prompt_token_ids` is
919
+ threaded through for call shape fidelity only, since HarmonyParser
920
+ derives its state from the output tokens alone.
921
+ """
922
+
923
+ @pytest.mark.asyncio
924
+ async def test_reasoning_never_leaks_as_content_midstream(
925
+ self, harmony_derenderer, harmony_encode
926
+ ):
927
+ """Regression guard for the RFC's original replay + diff design.
928
+
929
+ `HarmonyParser.parse()` always flushes to EOS. Mid-stream that
930
+ raises `HarmonyError` and the recovery branch re-emits the in
931
+ flight message on the `final` channel, so partial analysis surfaces
932
+ as content. Replay + `parse_delta` must never do that. Driven one
933
+ token per chunk, the finest granularity a client can produce.
934
+ """
935
+ output_ids = harmony_encode(HARMONY_OUTPUT)
936
+ prompt_ids = harmony_encode(HARMONY_PROMPT)
937
+ chat_request = _chat_request(model=HARMONY_MODEL, include_reasoning=True)
938
+
939
+ deltas = await _stream_harmony_deltas(
940
+ harmony_derenderer, chat_request, output_ids, prompt_ids, chunk_size=1
941
+ )
942
+
943
+ reasoning = ""
944
+ content = ""
945
+ for i, delta in enumerate(deltas):
946
+ reasoning += delta.reasoning or ""
947
+ content += delta.content or ""
948
+ # Any analysis text emitted as content breaks the prefix
949
+ # property, since the two channels share no prefix here.
950
+ assert HARMONY_ANSWER.startswith(content), (
951
+ f"content after chunk {i} is not a prefix of the final "
952
+ f"channel text: {content!r}"
953
)
954
+ assert HARMONY_REASONING.startswith(reasoning)
955
+ assert not delta.tool_calls
956
+
957
+ assert reasoning == HARMONY_REASONING
958
+ assert content == HARMONY_ANSWER
959
+
960
+ @pytest.mark.asyncio
961
+ async def test_stream_matches_batch(self, harmony_derenderer, harmony_encode):
962
+ """Streamed assembly equals one shot `/derender` over the same IDs."""
963
+ output_ids = harmony_encode(HARMONY_OUTPUT)
964
+ prompt_ids = harmony_encode(HARMONY_PROMPT)
965
+ chat_request = _chat_request(model=HARMONY_MODEL, include_reasoning=True)
966
+
967
+ deltas = await _stream_harmony_deltas(
968
+ harmony_derenderer, chat_request, output_ids, prompt_ids, chunk_size=3
969
+ )
970
+ reasoning = "".join(d.reasoning or "" for d in deltas)
971
+ content = "".join(d.content or "" for d in deltas)
972
+
973
+ batch_choices = await harmony_derenderer.derender_chat(
974
+ GenerateResponse(
975
+ request_id="test-harmony-batch",
976
+ choices=[
977
+ GenerateResponseChoice(
978
+ index=0, token_ids=output_ids, finish_reason="stop"
979
+ )
980
+ ],
981
+ ),
982
+ chat_request,
983
+ )
984
+ message = batch_choices[0].message
985
+
986
+ assert reasoning == message.reasoning
987
+ assert content == message.content
988
989
990
class TestDerenderStreamStateValidation:
@@ -446,6 +1012,24 @@ class TestDerenderStreamStateValidation:
1012
state = DerenderStreamState(prev_tokens=["a"] * 1024)
1013
assert len(state.prev_tokens) == 1024
1014
1015
+ def test_output_chunk_lens_mismatch_rejected(self):
1016
+ from pydantic import ValidationError
1017
+
1018
+ with pytest.raises(ValidationError):
1019
+ DerenderStreamState(output_token_ids=[1, 2, 3], output_chunk_lens=[1, 1])
1020
+
1021
+ def test_output_chunk_lens_zero_entry_rejected(self):
1022
+ from pydantic import ValidationError
1023
+
1024
+ with pytest.raises(ValidationError):
1025
+ DerenderStreamState(output_token_ids=[1, 2], output_chunk_lens=[0, 2])
1026
+
1027
+ def test_output_chunk_lens_matching_accepted(self):
1028
+ state = DerenderStreamState(
1029
+ output_token_ids=[1, 2, 3], output_chunk_lens=[1, 2]
1030
+ )
1031
+ assert state.output_chunk_lens == [1, 2]
1032
+
1033
1034
class TestServingDerenderStreamErrorHandling:
1035
"""Malformed stream_state must surface as 400 and not an unhandled 500."""
@@ -458,8 +1042,10 @@ class TestServingDerenderStreamErrorHandling:
1042
models = MagicMock()
1043
models.is_base_model.return_value = True
1044
models.model_config = MagicMock()
1045
+ models.model_config.max_model_len = 100_000
1046
1047
online_derenderer = MagicMock()
1048
+ online_derenderer.parser = None
1049
online_derenderer.derender_completion_stream = AsyncMock(
1050
side_effect=side_effect
1051
)
@@ -504,6 +1090,179 @@ class TestServingDerenderStreamErrorHandling:
1090
assert result.error.code == 400
1091
1092
1093
+class TestServingDerenderStreamValidation:
1094
+ """If checks cannot run in `derender_chat_stream_response`: all must
1095
+ reject with 400 before ever calling into `online_derenderer` (i.e.
1096
+ before touching the tokenizer)."""
1097
+
1098
+ def _make_serving(self, *, parser_configured: bool, max_model_len: int = 100_000):
1099
+ from unittest.mock import AsyncMock, MagicMock
1100
+
1101
+ from vllm.entrypoints.openai.chat_completion.protocol import (
1102
+ ChatCompletionStreamResponse,
1103
+ )
1104
+ from vllm.entrypoints.scale_out.derender.serving import ServingDerender
1105
+
1106
+ models = MagicMock()
1107
+ models.is_base_model.return_value = True
1108
+ models.model_config = MagicMock()
1109
+ models.model_config.max_model_len = max_model_len
1110
+
1111
+ online_derenderer = MagicMock()
1112
+ online_derenderer.parser = MagicMock() if parser_configured else None
1113
+ online_derenderer.derender_chat_stream = AsyncMock(
1114
+ return_value=(
1115
+ ChatCompletionStreamResponse(id="t", model=MODEL_NAME, choices=[]),
1116
+ DerenderStreamState(),
1117
+ )
1118
+ )
1119
+ return ServingDerender(models=models, online_derenderer=online_derenderer)
1120
+
1121
+ @pytest.mark.asyncio
1122
+ async def test_missing_chat_request_with_parser_rejected(self):
1123
+ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
1124
+ DerenderChatStreamRequest,
1125
+ )
1126
+
1127
+ serving = self._make_serving(parser_configured=True)
1128
+ request = DerenderChatStreamRequest(
1129
+ stream=True,
1130
+ model=MODEL_NAME,
1131
+ generate_chunk=_make_stream_chunk([1, 2]),
1132
+ )
1133
+ result = await serving.derender_chat_stream_response(request)
1134
+ assert isinstance(result, ErrorResponse)
1135
+ assert result.error.code == 400
1136
+ assert "chat_request" in result.error.message
1137
+ serving.online_derenderer.derender_chat_stream.assert_not_called()
1138
+
1139
+ @pytest.mark.asyncio
1140
+ async def test_missing_chat_request_without_parser_ok(self):
1141
+ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
1142
+ DerenderChatStreamRequest,
1143
+ )
1144
+
1145
+ serving = self._make_serving(parser_configured=False)
1146
+ request = DerenderChatStreamRequest(
1147
+ stream=True,
1148
+ model=MODEL_NAME,
1149
+ generate_chunk=_make_stream_chunk([1, 2]),
1150
+ )
1151
+ result = await serving.derender_chat_stream_response(request)
1152
+ assert not isinstance(result, ErrorResponse)
1153
+
1154
+ @pytest.mark.asyncio
1155
+ async def test_missing_prompt_token_ids_with_parser_rejected(self):
1156
+ """A parser configured model must reject a missing prompt_token_ids
1157
+ the same way it rejects a missing chat_request. Without it,
1158
+ parse_delta cannot tell whether the prompt left reasoning open and
1159
+ would misclassify reasoning content as plain content."""
1160
+ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
1161
+ DerenderChatStreamRequest,
1162
+ )
1163
+
1164
+ serving = self._make_serving(parser_configured=True)
1165
+ request = DerenderChatStreamRequest(
1166
+ stream=True,
1167
+ model=MODEL_NAME,
1168
+ generate_chunk=_make_stream_chunk([1, 2]),
1169
+ chat_request=_chat_request(),
1170
+ )
1171
+ result = await serving.derender_chat_stream_response(request)
1172
+ assert isinstance(result, ErrorResponse)
1173
+ assert result.error.code == 400
1174
+ assert "prompt_token_ids" in result.error.message
1175
+ serving.online_derenderer.derender_chat_stream.assert_not_called()
1176
+
1177
+ @pytest.mark.asyncio
1178
+ async def test_prompt_token_ids_present_with_parser_ok(self):
1179
+ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
1180
+ DerenderChatStreamRequest,
1181
+ )
1182
+
1183
+ serving = self._make_serving(parser_configured=True)
1184
+ request = DerenderChatStreamRequest(
1185
+ stream=True,
1186
+ model=MODEL_NAME,
1187
+ generate_chunk=_make_stream_chunk([1, 2]),
1188
+ chat_request=_chat_request(),
1189
+ prompt_token_ids=[1, 2, 3],
1190
+ )
1191
+ result = await serving.derender_chat_stream_response(request)
1192
+ assert not isinstance(result, ErrorResponse)
1193
+
1194
+ @pytest.mark.asyncio
1195
+ async def test_oversized_output_token_ids_rejected(self):
1196
+ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
1197
+ DerenderChatStreamRequest,
1198
+ )
1199
+
1200
+ serving = self._make_serving(parser_configured=False, max_model_len=4)
1201
+ request = DerenderChatStreamRequest(
1202
+ stream=True,
1203
+ model=MODEL_NAME,
1204
+ generate_chunk=_make_stream_chunk([1, 2, 3]),
1205
+ stream_state=DerenderStreamState(
1206
+ output_token_ids=[1, 2], output_chunk_lens=[2]
1207
+ ),
1208
+ )
1209
+ result = await serving.derender_chat_stream_response(request)
1210
+ assert isinstance(result, ErrorResponse)
1211
+ assert result.error.code == 400
1212
+ assert "max_model_len" in result.error.message
1213
+ serving.online_derenderer.derender_chat_stream.assert_not_called()
1214
+
1215
+ @pytest.mark.asyncio
1216
+ async def test_oversized_prompt_token_ids_rejected(self):
1217
+ """prompt_token_ids is caller supplied and otherwise unbounded. A
1218
+ parser configured deployment rescans it in full on every chunk
1219
+ (is_reasoning_end / adjust_initial_state_from_prompt), so it must be
1220
+ bounded the same way output_token_ids is."""
1221
+ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
1222
+ DerenderChatStreamRequest,
1223
+ )
1224
+
1225
+ serving = self._make_serving(parser_configured=True, max_model_len=4)
1226
+ request = DerenderChatStreamRequest(
1227
+ stream=True,
1228
+ model=MODEL_NAME,
1229
+ generate_chunk=_make_stream_chunk([1]),
1230
+ chat_request=_chat_request(),
1231
+ prompt_token_ids=[1, 2, 3, 4, 5],
1232
+ )
1233
+ result = await serving.derender_chat_stream_response(request)
1234
+ assert isinstance(result, ErrorResponse)
1235
+ assert result.error.code == 400
1236
+ assert "max_model_len" in result.error.message
1237
+ serving.online_derenderer.derender_chat_stream.assert_not_called()
1238
+
1239
+ @pytest.mark.asyncio
1240
+ async def test_too_many_choices_rejected(self):
1241
+ """Each streamed chunk contains at most one choice (a single
1242
+ DerenderStreamState is threaded through every choice). The check
1243
+ could never fire since derender_chat_stream itself rejects anything
1244
+ above 1 first."""
1245
+ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
1246
+ DerenderChatStreamRequest,
1247
+ )
1248
+
1249
+ serving = self._make_serving(parser_configured=False)
1250
+ two_choices = GenerateStreamResponse(
1251
+ request_id="t",
1252
+ choices=[
1253
+ GenerateResponseStreamChoice(index=i, token_ids=[1]) for i in range(2)
1254
+ ],
1255
+ )
1256
+ request = DerenderChatStreamRequest(
1257
+ stream=True, model=MODEL_NAME, generate_chunk=two_choices
1258
+ )
1259
+ result = await serving.derender_chat_stream_response(request)
1260
+ assert isinstance(result, ErrorResponse)
1261
+ assert result.error.code == 400
1262
+ assert "at most one choice" in result.error.message
1263
+ serving.online_derenderer.derender_chat_stream.assert_not_called()
1264
+
1265
+
1266
# ---------------------------------------------------------------------------
1267
# Integration tests — require a live render server
1268
# ---------------------------------------------------------------------------
@@ -757,3 +1516,416 @@ async def test_streaming_usage_chunk(client):
1516
assert d2["chunk"]["choices"] == []
1517
assert d2["chunk"]["usage"]["prompt_tokens"] == 10
1518
assert d2["chunk"]["usage"]["completion_tokens"] == len(token_ids)
1519
+
1520
+
1521
+# ---------------------------------------------------------------------------
1522
+# Integration tests for parser configured (reasoning + tool calls) require a
1523
+# live render server. Mirrors the parser_server / parser_tokenizer pattern
1524
+# from test_derender.py.
1525
+# ---------------------------------------------------------------------------
1526
+
1527
+PARSER_MODEL = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"
1528
+
1529
+_PARSER_TOOLS = [
1530
+ {
1531
+ "type": "function",
1532
+ "function": {
1533
+ "name": "get_weather",
1534
+ "description": "Get weather for a city",
1535
+ "parameters": {
1536
+ "type": "object",
1537
+ "properties": {"city": {"type": "string"}},
1538
+ },
1539
+ },
1540
+ }
1541
+]
1542
+
1543
+
1544
+@pytest.fixture(scope="module")
1545
+def parser_server():
1546
+ from tests.utils import RemoteLaunchRenderServer
1547
+
1548
+ args = [
1549
+ "--enable-auto-tool-choice",
1550
+ "--tool-call-parser",
1551
+ "hermes",
1552
+ "--reasoning-parser",
1553
+ "deepseek_r1",
1554
+ ]
1555
+ with RemoteLaunchRenderServer(PARSER_MODEL, args) as remote_server:
1556
+ yield remote_server
1557
+
1558
+
1559
+@pytest_asyncio.fixture
1560
+async def parser_client(parser_server):
1561
+ import httpx
1562
+
1563
+ async with httpx.AsyncClient(
1564
+ base_url=parser_server.url_for(""), timeout=60.0
1565
+ ) as http_client:
1566
+ yield http_client
1567
+
1568
+
1569
+@pytest.fixture(scope="module")
1570
+def parser_tokenizer():
1571
+ from vllm.tokenizers import get_tokenizer
1572
+
1573
+ return get_tokenizer(PARSER_MODEL)
1574
+
1575
+
1576
+def _require_parser_markers(tokenizer, text: str, *markers: str) -> list[int]:
1577
+ """Encode text and skip the test if any marker is lost in roundtrip."""
1578
+ ids = tokenizer.encode(text, add_special_tokens=False)
1579
+ decoded = tokenizer.decode(ids, skip_special_tokens=False)
1580
+ for m in markers:
1581
+ if m not in decoded:
1582
+ pytest.skip(f"Marker {m!r} lost in encode->decode roundtrip")
1583
+ return ids
1584
+
1585
+
1586
+async def _render_parser_chat(client, messages: list[dict]) -> dict:
1587
+ resp = await client.post(
1588
+ "/v1/chat/completions/render",
1589
+ json={"model": PARSER_MODEL, "messages": messages},
1590
+ )
1591
+ assert resp.status_code == 200, resp.text
1592
+ return resp.json()
1593
+
1594
+
1595
+async def _stream_chat_derender(
1596
+ client,
1597
+ output_ids: list[int],
1598
+ chunk_sizes: list[int],
1599
+ chat_request: dict,
1600
+ prompt_tokens: int,
1601
+ prompt_token_ids: list[int],
1602
+ on_chunk: Callable[[list[dict]], None] | None = None,
1603
+) -> dict:
1604
+ """Feed `output_ids` through the streaming chat derender endpoint in
1605
+ the given `chunk_sizes`, threading `stream_state` across calls and
1606
+ return the assembled message.
1607
+
1608
+ If `on_chunk` is given, it is called after every chunk with a
1609
+ snapshot (deep copy) of the `tool_calls` accumulator so far, letting
1610
+ callers assert properties of the intermediate deltas (e.g. monotonic
1611
+ argument growth) rather than only the final assembled result.
1612
+ """
1613
+ state = None
1614
+ content = ""
1615
+ reasoning = ""
1616
+ tool_calls: list[dict] = []
1617
+ finish_reason = None
1618
+
1619
+ pos = 0
1620
+ for i, size in enumerate(chunk_sizes):
1621
+ tids = output_ids[pos : pos + size]
1622
+ pos += size
1623
+ is_last = i == len(chunk_sizes) - 1
1624
+ resp = await client.post(
1625
+ "/v1/chat/completions/derender",
1626
+ json={
1627
+ "stream": True,
1628
+ "model": PARSER_MODEL,
1629
+ "generate_chunk": {
1630
+ "request_id": "stream-test",
1631
+ "choices": [
1632
+ {
1633
+ "index": 0,
1634
+ "token_ids": tids,
1635
+ "finish_reason": "stop" if is_last else None,
1636
+ }
1637
+ ],
1638
+ },
1639
+ "stream_state": state,
1640
+ "prompt_tokens": prompt_tokens,
1641
+ "prompt_token_ids": prompt_token_ids,
1642
+ "chat_request": chat_request,
1643
+ },
1644
+ )
1645
+ assert resp.status_code == 200, resp.text
1646
+ data = resp.json()
1647
+ state = data["stream_state"]
1648
+ delta = data["chunk"]["choices"][0]["delta"]
1649
+ content += delta.get("content") or ""
1650
+ reasoning += delta.get("reasoning") or ""
1651
+ for tc in delta.get("tool_calls") or []:
1652
+ idx = tc["index"]
1653
+ while len(tool_calls) <= idx:
1654
+ tool_calls.append({"id": None, "name": None, "arguments": ""})
1655
+ if tc.get("id"):
1656
+ tool_calls[idx]["id"] = tc["id"]
1657
+ fn = tc.get("function") or {}
1658
+ if fn.get("name"):
1659
+ tool_calls[idx]["name"] = fn["name"]
1660
+ if fn.get("arguments"):
1661
+ tool_calls[idx]["arguments"] += fn["arguments"]
1662
+ if is_last:
1663
+ finish_reason = data["chunk"]["choices"][0]["finish_reason"]
1664
+ if on_chunk is not None:
1665
+ on_chunk([dict(tc) for tc in tool_calls])
1666
+
1667
+ return {
1668
+ "content": content or None,
1669
+ "reasoning": reasoning or None,
1670
+ "tool_calls": tool_calls,
1671
+ "finish_reason": finish_reason,
1672
+ }
1673
+
1674
+
1675
+@pytest.mark.asyncio
1676
+@pytest.mark.parametrize(
1677
+ "chunking",
1678
+ [
1679
+ "whole",
1680
+ "per_token",
1681
+ pytest.param(
1682
+ "triples",
1683
+ marks=pytest.mark.xfail(
1684
+ reason="<think> leaks into reasoning when it shares a "
1685
+ "multi-token delta with reasoning text (vllm#55195)"
1686
+ ),
1687
+ ),
1688
+ ],
1689
+)
1690
+async def test_stream_parsed_matches_batch_reasoning(
1691
+ parser_client, parser_tokenizer, chunking
1692
+):
1693
+ """Streamed == batch: assembled reasoning/content equal the non
1694
+ streaming `/derender` result over the same token IDs, for every
1695
+ chunking of the same output (chunking invariance)."""
1696
+ messages = [{"role": "user", "content": "What is 2+3?"}]
1697
+ gen_req = await _render_parser_chat(parser_client, messages)
1698
+
1699
+ reasoning_text = "The user wants 2 plus 3. That is 5."
1700
+ answer_text = "The answer is 5."
1701
+ output_text = f"<think>{reasoning_text}</think>{answer_text}"
1702
+ output_ids = _require_parser_markers(parser_tokenizer, output_text, "</think>")
1703
+
1704
+ chat_request = {
1705
+ "model": PARSER_MODEL,
1706
+ "messages": messages,
1707
+ "include_reasoning": True,
1708
+ }
1709
+
1710
+ batch_resp = await parser_client.post(
1711
+ "/v1/chat/completions/derender",
1712
+ json={
1713
+ "model": PARSER_MODEL,
1714
+ "generate_response": {
1715
+ "request_id": "batch",
1716
+ "choices": [
1717
+ {"index": 0, "token_ids": output_ids, "finish_reason": "stop"}
1718
+ ],
1719
+ },
1720
+ "prompt_tokens": len(gen_req["token_ids"]),
1721
+ "chat_request": chat_request,
1722
+ },
1723
+ )
1724
+ assert batch_resp.status_code == 200, batch_resp.text
1725
+ batch_msg = batch_resp.json()["choices"][0]["message"]
1726
+
1727
+ n = len(output_ids)
1728
+ chunk_sizes = {
1729
+ "whole": [n],
1730
+ "per_token": [1] * n,
1731
+ "triples": [3] * (n // 3) + [n % 3],
1732
+ }[chunking]
1733
+ chunk_sizes = [c for c in chunk_sizes if c > 0]
1734
+ streamed = await _stream_chat_derender(
1735
+ parser_client,
1736
+ output_ids,
1737
+ chunk_sizes,
1738
+ chat_request,
1739
+ len(gen_req["token_ids"]),
1740
+ gen_req["token_ids"],
1741
+ )
1742
+ assert streamed["content"] == batch_msg["content"]
1743
+ assert streamed["reasoning"] == batch_msg.get("reasoning")
1744
+
1745
+
1746
+@pytest.mark.asyncio
1747
+async def test_stream_parsed_matches_batch_tool_call(parser_client, parser_tokenizer):
1748
+ """Tool call name+id+arguments parity between streamed and batch, plus
1749
+ the finish_reason -> "tool_calls" rewrite for auto tool choice."""
1750
+ messages = [{"role": "user", "content": "Weather in Paris?"}]
1751
+ gen_req = await _render_parser_chat(parser_client, messages)
1752
+
1753
+ output_text = (
1754
+ "<think>Let me check.</think>"
1755
+ '<tool_call>\n{"name": "get_weather", '
1756
+ '"arguments": {"city": "Paris"}}\n</tool_call>'
1757
+ )
1758
+ output_ids = _require_parser_markers(
1759
+ parser_tokenizer, output_text, "</think>", "<tool_call>", "</tool_call>"
1760
+ )
1761
+ chat_request = {
1762
+ "model": PARSER_MODEL,
1763
+ "messages": messages,
1764
+ "tools": _PARSER_TOOLS,
1765
+ "tool_choice": "auto",
1766
+ }
1767
+
1768
+ batch_resp = await parser_client.post(
1769
+ "/v1/chat/completions/derender",
1770
+ json={
1771
+ "model": PARSER_MODEL,
1772
+ "generate_response": {
1773
+ "request_id": "batch",
1774
+ "choices": [
1775
+ {"index": 0, "token_ids": output_ids, "finish_reason": "stop"}
1776
+ ],
1777
+ },
1778
+ "prompt_tokens": len(gen_req["token_ids"]),
1779
+ "chat_request": chat_request,
1780
+ },
1781
+ )
1782
+ assert batch_resp.status_code == 200, batch_resp.text
1783
+ batch_choice = batch_resp.json()["choices"][0]
1784
+ if not batch_choice["message"]["tool_calls"]:
1785
+ pytest.skip("Model did not emit a tool call")
1786
+
1787
+ for chunk_sizes in ([len(output_ids)], [1] * len(output_ids)):
1788
+ # Snapshot of tool_calls[0]["arguments"] after every chunk, to check
1789
+ # it only ever grows by appending. It never retracts or duplicates
1790
+ # already streamed text. That's the exact failure mode the
1791
+ # replay + parse_delta design (over diff-based streaming) exists to
1792
+ # avoid. Parity with the batch result alone wouldn't catch a
1793
+ # transient mid-stream regression.
1794
+ arg_snapshots: list[str] = []
1795
+
1796
+ def _record(
1797
+ tool_calls: list[dict], _snapshots: list[str] = arg_snapshots
1798
+ ) -> None:
1799
+ if tool_calls and tool_calls[0]["arguments"]:
1800
+ _snapshots.append(tool_calls[0]["arguments"])
1801
+
1802
+ streamed = await _stream_chat_derender(
1803
+ parser_client,
1804
+ output_ids,
1805
+ chunk_sizes,
1806
+ chat_request,
1807
+ len(gen_req["token_ids"]),
1808
+ gen_req["token_ids"],
1809
+ on_chunk=_record,
1810
+ )
1811
+ assert streamed["tool_calls"]
1812
+ assert streamed["tool_calls"][0]["name"] == "get_weather"
1813
+ assert streamed["tool_calls"][0]["id"] is not None
1814
+ assert json.loads(streamed["tool_calls"][0]["arguments"]) == json.loads(
1815
+ batch_choice["message"]["tool_calls"][0]["function"]["arguments"]
1816
+ )
1817
+ assert streamed["finish_reason"] == "tool_calls"
1818
+
1819
+ assert arg_snapshots, "expected at least one tool-call argument delta"
1820
+ for prev, curr in zip(arg_snapshots, arg_snapshots[1:]):
1821
+ assert curr.startswith(prev), (
1822
+ f"tool-call arguments regressed: {prev!r} -> {curr!r}"
1823
+ )
1824
+ assert arg_snapshots[-1] == streamed["tool_calls"][0]["arguments"]
1825
+
1826
+
1827
+@pytest.mark.asyncio
1828
+async def test_stream_parsed_cjk_across_chunk_boundaries(
1829
+ parser_client, parser_tokenizer
1830
+):
1831
+ """CJK/emoji tokens split across chunk boundaries with a parser active
1832
+ across the reasoning -> content transition (parser path regression
1833
+ guard analogous to #46159 in the plain detok path)."""
1834
+ messages = [{"role": "user", "content": "Reply in Chinese"}]
1835
+ gen_req = await _render_parser_chat(parser_client, messages)
1836
+
1837
+ reasoning_text = "思考中"
1838
+ answer_text = "你好世界 😀"
1839
+ output_text = f"<think>{reasoning_text}</think>{answer_text}"
1840
+ output_ids = _require_parser_markers(parser_tokenizer, output_text, "</think>")
1841
+
1842
+ chat_request = {
1843
+ "model": PARSER_MODEL,
1844
+ "messages": messages,
1845
+ "include_reasoning": True,
1846
+ }
1847
+ streamed = await _stream_chat_derender(
1848
+ parser_client,
1849
+ output_ids,
1850
+ [1] * len(output_ids),
1851
+ chat_request,
1852
+ len(gen_req["token_ids"]),
1853
+ gen_req["token_ids"],
1854
+ )
1855
+ assert "�" not in (streamed["content"] or "")
1856
+ assert "�" not in (streamed["reasoning"] or "")
1857
+ assert answer_text in (streamed["content"] or "")
1858
+
1859
+
1860
+@pytest.mark.asyncio
1861
+async def test_stream_parsed_include_reasoning_false(parser_client, parser_tokenizer):
1862
+ """include_reasoning=False emits no reasoning deltas on the streaming
1863
+ parser path."""
1864
+ messages = [{"role": "user", "content": "What is 2+3?"}]
1865
+ gen_req = await _render_parser_chat(parser_client, messages)
1866
+
1867
+ output_text = "<think>reasoning here</think>The answer is 5."
1868
+ output_ids = _require_parser_markers(parser_tokenizer, output_text, "</think>")
1869
+
1870
+ chat_request = {
1871
+ "model": PARSER_MODEL,
1872
+ "messages": messages,
1873
+ "include_reasoning": False,
1874
+ }
1875
+ streamed = await _stream_chat_derender(
1876
+ parser_client,
1877
+ output_ids,
1878
+ [1] * len(output_ids),
1879
+ chat_request,
1880
+ len(gen_req["token_ids"]),
1881
+ gen_req["token_ids"],
1882
+ )
1883
+ assert streamed["reasoning"] is None
1884
+
1885
+
1886
+@pytest.mark.asyncio
1887
+async def test_stream_parsed_missing_chat_request_rejected(parser_client):
1888
+ """Parser configured + no chat_request on the streaming endpoint -> 400
1889
+ (the live-server counterpart to the mocked
1890
+ `TestServingDerenderStreamValidation` checks above)."""
1891
+ resp = await parser_client.post(
1892
+ "/v1/chat/completions/derender",
1893
+ json={
1894
+ "stream": True,
1895
+ "model": PARSER_MODEL,
1896
+ "generate_chunk": {
1897
+ "request_id": "reject-test",
1898
+ "choices": [{"index": 0, "token_ids": [1, 2], "finish_reason": None}],
1899
+ },
1900
+ },
1901
+ )
1902
+ assert resp.status_code == 400
1903
+ assert "chat_request" in resp.json()["error"]["message"]
1904
+
1905
+
1906
+@pytest.mark.asyncio
1907
+async def test_stream_parsed_missing_prompt_token_ids_rejected(parser_client):
1908
+ """Parser configured + chat_request given but no prompt_token_ids on the
1909
+ streaming endpoint -> 400 (the live-server counterpart to the mocked
1910
+ `test_missing_prompt_token_ids_with_parser_rejected` above). Without
1911
+ prompt_token_ids, parse_delta cannot tell whether the prompt left
1912
+ reasoning open and would silently misclassify reasoning as content."""
1913
+ messages = [{"role": "user", "content": "Hello"}]
1914
+ resp = await parser_client.post(
1915
+ "/v1/chat/completions/derender",
1916
+ json={
1917
+ "stream": True,
1918
+ "model": PARSER_MODEL,
1919
+ "generate_chunk": {
1920
+ "request_id": "reject-test",
1921
+ "choices": [{"index": 0, "token_ids": [1, 2], "finish_reason": None}],
1922
+ },
1923
+ "chat_request": {
1924
+ "model": PARSER_MODEL,
1925
+ "messages": messages,
1926
+ "include_reasoning": True,
1927
+ },
1928
+ },
1929
+ )
1930
+ assert resp.status_code == 400
1931
+ assert "prompt_token_ids" in resp.json()["error"]["message"]
vllm/entrypoints/scale_out/derender/serving.py
+77
-11
@@ -138,6 +138,13 @@ class ServingDerender(BaseServing):
138
if bounds_error is not None:
139
return bounds_error
140
141
+ if self.online_derenderer.parser is not None and request.chat_request is None:
142
+ return self.create_error_response(
143
+ "chat_request is required when a tool or reasoning parser is "
144
+ "configured because plain detokenization would leak raw parser "
145
+ "markup into content."
146
+ )
147
+
148
try:
149
choices = await self.online_derenderer.derender_chat(
150
request.generate_response, request.chat_request
@@ -248,14 +255,67 @@ class ServingDerender(BaseServing):
255
256
Processes one ``GenerateStreamResponse`` chunk and returns the
257
derendered chunk together with the updated client carried state.
251
-
252
- ``parser is None`` or no ``chat_request`` until reasoning/tool call
253
- functionality added in future PR.
258
"""
259
error_check_ret = await self._check_model(request)
260
if error_check_ret is not None:
261
return error_check_ret
262
263
+ if self.online_derenderer.parser is not None and request.chat_request is None:
264
+ return self.create_error_response(
265
+ "chat_request is required when a tool or reasoning parser is "
266
+ "configured because plain detokenization would leak raw parser "
267
+ "markup into content."
268
+ )
269
+
270
+ if (
271
+ self.online_derenderer.parser is not None
272
+ and request.prompt_token_ids is None
273
+ ):
274
+ return self.create_error_response(
275
+ "prompt_token_ids is required when a tool or reasoning "
276
+ "parser is configured. Without it parse_delta cannot tell "
277
+ "whether the prompt left reasoning open (e.g. a chat "
278
+ "template that pre-opens <think>) and would misclassify "
279
+ "reasoning content as plain content."
280
+ )
281
+
282
+ # Each streamed chunk contains at most one choice per SSE event.
283
+ # A single DerenderStreamState is threaded
284
+ # through every choice in the chunk, so >1 would corrupt detok/parser
285
+ # state across choices. See the matching check in
286
+ # OnlineDerenderer.derender_chat_stream which this fails ahead of to
287
+ # avoid touching the tokenizer at all on a malformed chunk.
288
+ num_choices = len(request.generate_chunk.choices)
289
+ if num_choices > 1:
290
+ return self.create_error_response(
291
+ f"derender_chat_stream expects at most one choice per chunk "
292
+ f"(got {num_choices})."
293
+ )
294
+
295
+ # prompt_token_ids is folded in here too. It is caller supplied and
296
+ # otherwise unbounded. A parser configured deployment rescans it
297
+ # in full on every chunk (is_reasoning_end /
298
+ # adjust_initial_state_from_prompt, see abstract_parser.parse_delta)
299
+ # since parser state cannot be carried across calls.
300
+ prompt_len = (
301
+ len(request.prompt_token_ids) if request.prompt_token_ids is not None else 0
302
+ )
303
+ prior_len = (
304
+ len(request.stream_state.output_token_ids)
305
+ if request.stream_state is not None
306
+ else 0
307
+ )
308
+ delta_len = sum(
309
+ len(c.token_ids) for c in request.generate_chunk.choices if c.token_ids
310
+ )
311
+ max_model_len = self.model_config.max_model_len
312
+ if prompt_len + prior_len + delta_len > max_model_len:
313
+ return self.create_error_response(
314
+ f"prompt_token_ids length ({prompt_len}) plus "
315
+ f"output_token_ids length ({prior_len}) plus delta "
316
+ f"({delta_len}) exceeds max_model_len ({max_model_len})."
317
+ )
318
+
319
model_name = request.model or self.models.model_name()
320
try:
321
chunk, updated_state = await self.online_derenderer.derender_chat_stream(
@@ -264,15 +324,21 @@ class ServingDerender(BaseServing):
324
state=request.stream_state,
325
chat_request=request.chat_request,
326
prompt_tokens=request.prompt_tokens,
327
+ prompt_token_ids=request.prompt_token_ids,
328
)
268
- except NotImplementedError as exc:
269
- return self.create_error_response(exc)
270
- except ValueError as exc:
271
- return self.create_error_response(str(exc))
272
- except (KeyError, IndexError) as exc:
273
- return self.create_error_response(
274
- f"invalid stream_state: detokenization failed ({exc!r})"
329
+ except Exception as exc:
330
+ # The two ValueErrors derender_chat_stream can raise directly
331
+ # (missing chat_request, >1 choice per chunk) are already
332
+ # pre-checked above, so anything reaching here comes from the
333
+ # parser replaying entirely client controlled state. Hermes
334
+ # swallows its own streaming exceptions but finalize_generation
335
+ # and engine based parsers don't, so an unexpected parser
336
+ # failure on malformed (but well typed) state must surface as
337
+ # a 400 here rather than an unhandled 500.
338
+ logger.debug(
339
+ "derender stream replay failed: %s: %s", type(exc).__name__, exc
340
)
341
+ return self.create_error_response("invalid stream_state: derender failed")
342
343
logger.debug(
344
"derender_chat_stream request_id=%s model=%s delta_tokens=%d",
@@ -313,7 +379,7 @@ class ServingDerender(BaseServing):
379
return self.create_error_response(str(exc))
380
except (KeyError, IndexError) as exc:
381
return self.create_error_response(
316
- f"invalid stream_state: detokenization failed ({exc!r})"
382
+ f"invalid stream_state: detokenization failed ({exc})"
383
)
384
385
logger.debug(
vllm/entrypoints/scale_out/token_in_token_out/protocol.py
+76
-17
@@ -1,6 +1,6 @@
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
-from typing import Any, Literal, TypeAlias
3
+from typing import Annotated, Any, Literal, TypeAlias
4
5
from pydantic import (
6
BaseModel,
@@ -483,16 +483,32 @@ class DerenderStreamState(BaseModel):
483
streaming derender endpoint. All fields are plain JSON serializable data.
484
No opaque tokenizer or parser internals are stored here.
485
486
+ Two separate sets of fields support two different streaming modes:
487
+
488
+ - For plain streaming (no parser configured including the completions path):
489
+ `prev_tokens`, `prefix_offset` and `read_offset` maintain a bounded incremental
490
+ decoding window. This requires O(window) transport and O(delta) computation
491
+ per chunk.
492
+ - For parser enabled chat streaming: `output_token_ids`, `output_chunk_lens`,
493
+ `tools_streamed` and `last_tool_call_ids` are used to replay `parse_delta()`
494
+ from scratch on every chunk because parser state cannot be serialized. This
495
+ incurs O(n) transport per chunk (O(n²) per generation) and O(n²)
496
+ `parse_delta()` calls per generation. Since many parsers re-scan the
497
+ entire accumulated text on each invocation, `parse_delta()` itself is
498
+ O(n) yielding a true worst case compute cost of O(n³) per generation.
499
+ No caching is performed. Work is bounded by `max_model_len`.
500
+ See `OnlineDerenderer._derender_chat_stream_parsed`.
501
+
502
The detokenization strategy carries the incremental decode offsets
503
directly rather than re-sending the whole token history each chunk.
488
- ``detokenize_incrementally`` only ever reads the trailing token window
489
- ``prev_tokens[prefix_offset:]``, so we carry just that tail plus the two
504
+ `detokenize_incrementally` only ever reads the trailing token window
505
+ `prev_tokens[prefix_offset:]`, so we carry just that tail plus the two
506
offsets. Each chunk resumes exactly where the last one stopped, including
491
- any partially processed multi-byte character (tracked by ``read_offset``),
507
+ any partially processed multi-byte character (tracked by `read_offset`),
508
then trims and rebases the window so it never grows with generation length.
509
510
Performance:
495
- - Compute per chunk is O(delta). One ``detokenize_incrementally`` call per
511
+ - Compute per chunk is O(delta). One `detokenize_incrementally` call per
512
new token, independent of how many tokens preceded it.
513
- Transport per chunk is O(window). The carried tail is bounded by the
514
incremental detokenization offset, so cumulative bytes over the wire are
@@ -500,18 +516,18 @@ class DerenderStreamState(BaseModel):
516
"""
517
518
prev_tokens: list[str] = Field(default_factory=list)
503
- """Trailing decode window. Token strings from ``prefix_offset`` onward.
519
+ """Trailing decode window. Token strings from `prefix_offset` onward.
520
521
Bounded, trimmed and rebased each chunk to the tail
506
- ``detokenize_incrementally`` still reads, so it does not grow with the
522
+ `detokenize_incrementally` still reads, so it does not grow with the
523
number of chunks.
524
"""
525
526
prefix_offset: int = Field(default=0, ge=0)
511
- """Prefix offset into ``prev_tokens`` for incremental detokenization."""
527
+ """Prefix offset into `prev_tokens` for incremental detokenization."""
528
529
read_offset: int = Field(default=0, ge=0)
514
- """Read offset into ``prev_tokens`` for incremental detokenization."""
530
+ """Read offset into `prev_tokens` for incremental detokenization."""
531
532
@field_validator("prev_tokens")
533
@classmethod
@@ -525,23 +541,53 @@ class DerenderStreamState(BaseModel):
541
return v
542
543
role_sent: bool = False
528
- """True once the initial ``role: "assistant"`` delta has been emitted.
544
+ """True once the initial `role: "assistant"` delta has been emitted.
545
546
Prevents re-emitting the role on subsequent chunks even when the detok
547
window is transiently empty (e.g. usage only final chunk).
548
"""
549
534
- # TODO: Properties used in follow on PR for tool call parsing
535
- last_content: str | None = None
536
- """Last emitted cumulative assistant content text."""
550
+ output_token_ids: list[int] = Field(default_factory=list)
551
+ """All output tokens seen so far. Parser path only.
552
+
553
+ Replay buffer: each chunk rebuilds a fresh parser and replays every
554
+ token in here through `parse_delta` (discarding the result) before
555
+ processing the current chunk's tokens since parser internal state
556
+ cannot be serialized into this stateless model. Unavoidably O(n)
557
+ bounded by ``max_model_len`` (enforced server side, not by a field
558
+ validator here since the bound is model dependent).
559
+ """
560
+
561
+ output_chunk_lens: list[Annotated[int, Field(gt=0)]] = Field(default_factory=list)
562
+ """Token count of each chunk in `output_token_ids`. Parser path only.
563
+
564
+ Replay uses these to reproduce the original `parse_delta` call
565
+ boundaries. Must sum to `len(output_token_ids)`.
566
+ """
567
+
568
+ @model_validator(mode="after")
569
+ def _validate_output_chunk_lens(self) -> "DerenderStreamState":
570
+ total = sum(self.output_chunk_lens)
571
+ if total != len(self.output_token_ids):
572
+ raise ValueError(
573
+ f"output_chunk_lens must sum to len(output_token_ids) "
574
+ f"(got sum={total}, len(output_token_ids)="
575
+ f"{len(self.output_token_ids)})"
576
+ )
577
+ return self
578
+
579
+ tools_streamed: bool = False
580
+ """True once a tool call delta has been emitted. Parser path only.
581
538
- last_reasoning: str | None = None
539
- """Last emitted cumulative reasoning text."""
582
+ Drives the `finish_reason` -> `"tool_calls"` rewrite on the final
583
+ chunk mirroring the generate streaming path.
584
+ """
585
586
last_tool_call_ids: list[str] = Field(default_factory=list)
542
- """Stable tool-call IDs, assigned once when each call first appears.
587
+ """Stable tool call IDs, assigned once when each call first appears.
588
544
- Prevents ID regeneration across re-parsing.
589
+ Indexed by tool call index. Parser path only. Prevents ID regeneration
590
+ when replay reprocesses a tool call that already has a pinned ID.
591
"""
592
593
@@ -570,6 +616,19 @@ class DerenderChatStreamRequest(BaseModel):
616
prompt_tokens: int | None = None
617
"""Prompt token count for usage. Forwarded from the render step."""
618
619
+ prompt_token_ids: list[int] | None = None
620
+ """Prompt token IDs. Required by the parser path's `parse_delta` to
621
+ settle its initial reasoning state (e.g. chat templates that pre-open
622
+ ``<think>``). `prompt_tokens` is a usage count and cannot serve this
623
+ purpose. Sourced from `GenerateRequest.token_ids` at the render step.
624
+
625
+ Rejected with a 400 (by `ServingDerender`) when a tool or reasoning
626
+ parser is configured and this is omitted. Without it, `parse_delta`
627
+ cannot tell whether the prompt left reasoning open and would silently
628
+ misclassify reasoning content as plain content. Unused on the plain
629
+ detokenization path.
630
+ """
631
+
632
chat_request: ChatCompletionRequest | None = None
633
"""The original (post adjust_request) ChatCompletionRequest from /render."""
634
vllm/renderers/online_derenderer.py
+245
-21
@@ -4,7 +4,10 @@ from typing import Any
4
5
from vllm.config import ModelConfig
6
from vllm.entrypoints.chat_utils import ChatTemplateContentFormatOption
7
-from vllm.entrypoints.generate.base.protocol import DeltaMessage, ToolCall
7
+from vllm.entrypoints.generate.base.protocol import (
8
+ DeltaMessage,
9
+ ToolCall,
10
+)
11
from vllm.entrypoints.generate.base.serving import resolve_token_id_placeholder
12
from vllm.entrypoints.openai.chat_completion.protocol import (
13
ChatCompletionLogProbs,
@@ -29,6 +32,9 @@ from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
32
)
33
from vllm.entrypoints.serve.engine.protocol import UsageInfo
34
from vllm.entrypoints.serve.utils.request_logger import RequestLogger
35
+from vllm.entrypoints.serve.utils.tool_calls_utils import (
36
+ maybe_filter_parallel_tool_calls,
37
+)
38
from vllm.logger import init_logger
39
from vllm.parser import Parser, ParserManager
40
from vllm.renderers import BaseRenderer
@@ -95,6 +101,11 @@ class OnlineDerenderer:
101
self._derender_completion_async = make_async(
102
self._derender_completion, executor=renderer._executor
103
)
104
+ # Replay is O(n) per chunk (unlike the O(delta) detok-only paths),
105
+ # so it must not run on the event loop either.
106
+ self._derender_chat_stream_parsed_async = make_async(
107
+ self._derender_chat_stream_parsed, executor=renderer._executor
108
+ )
109
110
async def derender_chat(
111
self,
@@ -144,6 +155,7 @@ class OnlineDerenderer:
155
tokenizer,
156
chat_request.tools,
157
chat_template_kwargs=chat_template_kwargs,
158
+ model_config=self.model_config,
159
)
160
reasoning, content, tool_calls = parser.parse(
161
decoded_text,
@@ -278,23 +290,30 @@ class OnlineDerenderer:
290
state: DerenderStreamState | None = None,
291
chat_request: ChatCompletionRequest | None = None,
292
prompt_tokens: int | None = None,
293
+ prompt_token_ids: list[int] | None = None,
294
) -> tuple[ChatCompletionStreamResponse, DerenderStreamState]:
295
"""Process one GenerateStreamResponse chunk for streaming chat derender.
296
284
- TODO: parse path for reasoning and tool calls is implemented in future PR.
285
-
286
- Unlike OpenAI's API, which always emits ``role: "assistant"`` on the
297
+ Unlike OpenAI's API, which always emits `role: "assistant"` on the
298
very first chunk, this emits it on the first chunk with a non empty
288
- ``choices`` list. A leading usage only chunk therefore defers the
299
+ `choices` list. A leading usage only chunk therefore defers the
300
role to the following content chunk instead of sending an empty
301
role only delta.
302
303
Args:
304
model: Model name for the response object.
294
- generate_chunk: One SSE chunk from ``/inference/v1/generate``.
295
- state: Client carried detok state (``None`` for first call).
296
- chat_request: Original ChatCompletionRequest from ``/render``.
305
+ generate_chunk: One SSE chunk from `/inference/v1/generate`.
306
+ state: Client carried detok state (`None` for first call).
307
+ chat_request: Original ChatCompletionRequest from `/render`.
308
+ Required when a reasoning or tool parser is configured
309
+ (validated by the caller — see `ServingDerender`) because
310
+ plain detokenization would leak raw parser markup into `content`.
311
prompt_tokens: Prompt token count for the usage chunk.
312
+ prompt_token_ids: Prompt token IDs. Parser path only. Lets
313
+ `parse_delta` settle its initial reasoning state. Required
314
+ when a reasoning or tool parser is configured (validated by
315
+ the caller — see ``ServingDerender``). Without it reasoning
316
+ left open by the prompt would be misclassified as content.
317
318
Returns:
319
(chunk, updated_state) — the derendered SSE chunk and the state
@@ -304,28 +323,36 @@ class OnlineDerenderer:
323
if state is None:
324
state = DerenderStreamState()
325
307
- if self.parser is not None:
308
- # TODO: Follow on PR will implement the parse path. Check on the
309
- # parser alone (fail closed). A parser configured model must never
310
- # fall through to plain detok on the streaming path, even when
311
- # ``chat_request`` is omitted or reasoning/tool markup would leak
312
- # into ``delta.content``.
313
- raise NotImplementedError(
314
- "Streaming chat derender is not yet supported for models with "
315
- "a reasoning or tool parser configured. Use the non-streaming "
316
- "derender endpoint (stream=false) for parsed output."
317
- )
318
-
326
# A single DerenderStreamState is threaded through every choice in
327
# this chunk. Correct only when there is at most one choice per SSE
328
# event (n=1, one call per index), as the streaming derender
329
# protocol assumes. Multiple choices sharing one chunk would corrupt
323
- # each other's detok window.
330
+ # each other's detok/parser state.
331
if len(generate_chunk.choices) > 1:
332
raise ValueError(
333
"derender_chat_stream expects at most one choice per chunk"
334
)
335
336
+ parser_cls = self.parser
337
+ if parser_cls is not None:
338
+ # Fail (mirrors ServingDerender's pre-check) because a parser
339
+ # configured model must never fall through to plain detok or
340
+ # reasoning/tool markup would leak into `delta.content`.
341
+ if chat_request is None:
342
+ raise ValueError(
343
+ "chat_request is required for streaming chat derender "
344
+ "when a tool or reasoning parser is configured"
345
+ )
346
+ return await self._derender_chat_stream_parsed_async(
347
+ parser_cls,
348
+ model,
349
+ generate_chunk,
350
+ state,
351
+ chat_request,
352
+ prompt_tokens,
353
+ prompt_token_ids,
354
+ )
355
+
356
tokenizer = self.renderer.get_tokenizer()
357
skip_special = (
358
chat_request.skip_special_tokens if chat_request is not None else True
@@ -374,6 +401,203 @@ class OnlineDerenderer:
401
)
402
return chunk, updated_state
403
404
+ def _derender_chat_stream_parsed(
405
+ self,
406
+ parser_cls: type[Parser],
407
+ model: str,
408
+ generate_chunk: GenerateStreamResponse,
409
+ state: DerenderStreamState,
410
+ chat_request: ChatCompletionRequest,
411
+ prompt_tokens: int | None,
412
+ prompt_token_ids: list[int] | None,
413
+ ) -> tuple[ChatCompletionStreamResponse, DerenderStreamState]:
414
+ """Parser path for streaming chat derender: replay + `parse_delta`.
415
+
416
+ Parser internal state (buffered markup, reasoning/tool phase, etc.)
417
+ cannot be serialized into `DerenderStreamState`, so each call
418
+ builds a fresh parser and replays every prior output token through
419
+ `parse_delta` (discarding the result) before processing this
420
+ chunk's tokens for real.
421
+
422
+ Every chunk, replayed or live, goes through one `parse_delta` call
423
+ with the tokens it arrived with (more than one under e.g.
424
+ speculative decoding). Replay recovers those boundaries from
425
+ `state.output_chunk_lens`, so the rebuilt parser state matches
426
+ standard serving, which calls `parse_delta` once per engine step.
427
+ Text comes from a fresh incremental detokenizer with special tokens
428
+ preserved (``skip_special_tokens=False``).
429
+ """
430
+ tokenizer = self.renderer.get_tokenizer()
431
+
432
+ chat_template_kwargs: dict[str, Any] = {}
433
+ if not self.use_harmony:
434
+ chat_template_kwargs = (
435
+ chat_request.build_chat_params(
436
+ self.chat_template,
437
+ self.chat_template_content_format,
438
+ )
439
+ .with_defaults(self.default_chat_template_kwargs)
440
+ .chat_template_kwargs
441
+ )
442
+
443
+ parser = parser_cls(
444
+ tokenizer,
445
+ chat_request.tools,
446
+ chat_template_kwargs=chat_template_kwargs,
447
+ model_config=self.model_config,
448
+ )
449
+
450
+ # Ephemeral incremental detok window, local to this call. Threaded
451
+ # across both the replay and current chunk phases (via `nonlocal`)
452
+ # so multi-byte characters split across that boundary still decode
453
+ # correctly. Discarded once the call returns.
454
+ detok_state = DerenderStreamState()
455
+
456
+ def _replay(token_ids: list[int], chunk_lens: list[int]) -> None:
457
+ """Replay prior chunks through `parse_delta` to rebuild parser
458
+ state, discarding the result. Never `finished` since that only
459
+ applies to the current chunk.
460
+ """
461
+ nonlocal detok_state
462
+ start = 0
463
+ for chunk_len in chunk_lens:
464
+ chunk = token_ids[start : start + chunk_len]
465
+ start += chunk_len
466
+ text, detok_state = self._detokenize_delta(
467
+ tokenizer, chunk, detok_state, skip_special_tokens=False
468
+ )
469
+ parser.parse_delta(
470
+ text,
471
+ chunk,
472
+ chat_request,
473
+ prompt_token_ids=prompt_token_ids,
474
+ finished=False,
475
+ )
476
+
477
+ # Replay history to reconstruct parser state. The result is thrown
478
+ # away and only the current chunk's emission goes to the client.
479
+ _replay(state.output_token_ids, state.output_chunk_lens)
480
+
481
+ stream_choices: list[ChatCompletionResponseStreamChoice] = []
482
+ output_token_ids = list(state.output_token_ids)
483
+ output_chunk_lens = list(state.output_chunk_lens)
484
+ role_sent = state.role_sent
485
+ tools_streamed = state.tools_streamed
486
+ last_tool_call_ids = list(state.last_tool_call_ids)
487
+
488
+ # At most one choice: the caller (derender_chat_stream) already
489
+ # rejects >1 before dispatching here. role_sent/tools_streamed/
490
+ # output_token_ids below are updated for that single choice, not
491
+ # accumulated across choices. Looping over generate_chunk.choices
492
+ # could silently corrupt output if chunks were ever allowed to
493
+ # contain multiple choices.
494
+ if generate_chunk.choices:
495
+ choice = generate_chunk.choices[0]
496
+ delta_tids = choice.token_ids or []
497
+ is_finished = choice.finish_reason is not None
498
+
499
+ if delta_tids:
500
+ # One parse_delta call for the whole chunk (producer
501
+ # granularity), not one per token. See the granularity note
502
+ # in this method's docstring.
503
+ text, detok_state = self._detokenize_delta(
504
+ tokenizer, delta_tids, detok_state, skip_special_tokens=False
505
+ )
506
+ delta_message = parser.parse_delta(
507
+ text,
508
+ delta_tids,
509
+ chat_request,
510
+ prompt_token_ids=prompt_token_ids,
511
+ finished=is_finished,
512
+ )
513
+ elif is_finished:
514
+ # Finish only chunk (no new tokens). Still flush any
515
+ # buffered tool call arguments.
516
+ delta_message = parser.parse_delta(
517
+ "",
518
+ [],
519
+ chat_request,
520
+ prompt_token_ids=prompt_token_ids,
521
+ finished=True,
522
+ )
523
+ else:
524
+ delta_message = None
525
+
526
+ if delta_tids:
527
+ output_token_ids.extend(delta_tids)
528
+ output_chunk_lens.append(len(delta_tids))
529
+
530
+ if delta_message is None:
531
+ delta_message = DeltaMessage()
532
+
533
+ if delta_message.tool_calls:
534
+ tools_streamed = True
535
+ for tc in delta_message.tool_calls:
536
+ if tc.id is None:
537
+ continue
538
+ if tc.index < len(last_tool_call_ids):
539
+ # Pin: reuse the ID already recorded for this index
540
+ # rather than one a from scratch replay regenerated.
541
+ # Real trigger not just defensive with
542
+ # tool_choice="required",
543
+ # extract_required_tool_call_streaming resets
544
+ # function_name_returned to False whenever the
545
+ # partial JSON transiently fails to parse which
546
+ # re-emits id+name for the same index on replay.
547
+ tc.id = last_tool_call_ids[tc.index]
548
+ else:
549
+ last_tool_call_ids.append(tc.id)
550
+
551
+ if not role_sent:
552
+ delta_message.role = "assistant"
553
+ role_sent = True
554
+
555
+ finish_reason = choice.finish_reason
556
+ if finish_reason is not None:
557
+ is_named_tool_choice = (
558
+ type(chat_request.tool_choice) is ChatCompletionNamedToolChoiceParam
559
+ )
560
+ if tools_streamed and not is_named_tool_choice:
561
+ finish_reason = "tool_calls"
562
+
563
+ stream_choice = ChatCompletionResponseStreamChoice(
564
+ index=choice.index,
565
+ delta=delta_message,
566
+ finish_reason=finish_reason,
567
+ )
568
+ stream_choices.append(
569
+ maybe_filter_parallel_tool_calls(stream_choice, chat_request)
570
+ )
571
+
572
+ updated_state = state.model_copy(
573
+ update={
574
+ "output_token_ids": output_token_ids,
575
+ "output_chunk_lens": output_chunk_lens,
576
+ "role_sent": role_sent,
577
+ "tools_streamed": tools_streamed,
578
+ "last_tool_call_ids": last_tool_call_ids,
579
+ }
580
+ )
581
+
582
+ usage: UsageInfo | None = None
583
+ if generate_chunk.usage is not None:
584
+ u = generate_chunk.usage
585
+ pt = prompt_tokens if prompt_tokens is not None else (u.prompt_tokens or 0)
586
+ ct = u.completion_tokens or 0
587
+ usage = UsageInfo(
588
+ prompt_tokens=pt,
589
+ completion_tokens=ct,
590
+ total_tokens=pt + ct,
591
+ )
592
+
593
+ chunk = ChatCompletionStreamResponse(
594
+ id=generate_chunk.request_id,
595
+ model=model,
596
+ choices=stream_choices,
597
+ usage=usage,
598
+ )
599
+ return chunk, updated_state
600
+
601
async def derender_completion(
602
self,
603
generate_responses: list[GenerateResponse],