test_derender_stream.py · tduncan/Nvidia-Optimization-Stack-VLLM

@tduncan / Nvidia-Optimization-Stack-VLLM

Code Commits Issues Pulls Cloud Agent CI/CD

main

py 2,337 lines 85.2 KB

Raw

1	# SPDX-License-Identifier: Apache-2.0
2	# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4	"""Unit tests for streaming derender.
5
6	Tests are split into two layers:
7
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
16	`@pytest.mark.asyncio` and gated by the `server` / `client`
17	fixtures from the sibling `test_derender.py`.
18	"""
19
20	import json
21	import threading
22	from concurrent.futures import ThreadPoolExecutor
23	from unittest.mock import AsyncMock, MagicMock
24
25	import httpx
26	import pytest
27	import pytest_asyncio
28	from pydantic import ValidationError
29	from tokenizers.decoders import DecodeStream
30
31	import vllm.renderers.online_derenderer as online_derenderer_module
32	from tests.entrypoints.scale_out.derender.utils import stream_chat_derender
33	from tests.utils import RemoteLaunchRenderServer
34	from vllm.entrypoints.generate.base.protocol import (
35	DeltaFunctionCall,
36	DeltaMessage,
37	DeltaToolCall,
38	PerRequestMetrics,
39	SpeculativeDecodingMetrics,
40	)
41	from vllm.entrypoints.openai.chat_completion.protocol import (
42	ChatCompletionRequest,
43	ChatCompletionStreamResponse,
44	)
45	from vllm.entrypoints.openai.completion.protocol import CompletionRequest
46	from vllm.entrypoints.scale_out.derender.serving import ServingDerender
47	from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
48	DerenderChatRequest,
49	DerenderChatStreamRequest,
50	DerenderCompletionRequest,
51	DerenderCompletionStreamRequest,
52	DerenderStreamState,
53	GenerateResponse,
54	GenerateResponseChoice,
55	GenerateResponseStreamChoice,
56	GenerateStreamResponse,
57	)
58	from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo
59	from vllm.parser import Parser
60	from vllm.renderers.online_derenderer import (
61	OnlineDerenderer,
62	_decode_params,
63	_seed_stream_state,
64	)
65	from vllm.tokenizers import get_tokenizer
66	from vllm.utils import random_uuid
67
68	MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
69
70
71	def _merge_fake_deltas(
72	accumulated: DeltaMessage \| None, new: DeltaMessage \| None
73	) -> DeltaMessage \| None:
74	"""Fold one token's worth of `_FakeParser` output into a per call
75	accumulator, the same way a real parser merges multiple tokens'
76	effects into the single `DeltaMessage` it returns per `parse_delta`
77	call.
78	"""
79	if new is None:
80	return accumulated
81	if accumulated is None:
82	accumulated = DeltaMessage()
83
84	if new.content:
85	accumulated.content = (accumulated.content or "") + new.content
86	if new.reasoning:
87	accumulated.reasoning = (accumulated.reasoning or "") + new.reasoning
88
89	for tc in new.tool_calls:
90	starts_new_call = tc.id is not None or (
91	tc.function is not None and tc.function.name is not None
92	)
93	existing = (
94	None
95	if starts_new_call
96	else next((t for t in accumulated.tool_calls if t.index == tc.index), None)
97	)
98	if existing is None:
99	accumulated.tool_calls.append(tc)
100	continue
101	if existing.function is None:
102	existing.function = DeltaFunctionCall()
103	if tc.function is not None and tc.function.arguments:
104	existing.function.arguments = (
105	existing.function.arguments or ""
106	) + tc.function.arguments
107
108	return accumulated
109
110
111	class _FakeParser(Parser):
112	"""Deterministic `Parser` stub for unit testing the replay/merge/pin
113	mechanics of `OnlineDerenderer._derender_chat_stream_parsed` in
114	isolation from any real reasoning/tool parser's markup grammar.
115
116	Processes every id in `delta_token_ids` in order and folds their
117	effects into one `DeltaMessage`, like a real parser handling a
118	multi-token delta in a single call. Ignores `delta_text`:
119
120	- `TOOL_START` opens a new tool call at index 0 (id + name).
121	- `TOOL_ARG` appends one `"a"` to that tool call's arguments.
122	- `REASON` emits one `"r"` of reasoning (suppressed when the request
123	has `include_reasoning=False`, mirroring the real parsers).
124	- `CONTENT` emits one `"c"` of content.
125	- An empty `delta_token_ids` with `finished=True` (the finish only
126	flush call) emits a `"FLUSH"` content sentinel so tests can confirm
127	it happened.
128	- Anything else emits nothing.
129	"""
130
131	TOOL_START = 9
132	TOOL_ARG = 10
133	REASON = 11
134	CONTENT = 12
135
136	def parse_delta(
137	self,
138	delta_text,
139	delta_token_ids,
140	request,
141	prompt_token_ids=None,
142	*,
143	finished,
144	):
145	if not delta_token_ids:
146	return DeltaMessage(content="FLUSH") if finished else None
147
148	accumulated: DeltaMessage \| None = None
149	for tok in delta_token_ids:
150	accumulated = _merge_fake_deltas(
151	accumulated, self._parse_one_token(tok, request)
152	)
153	return accumulated
154
155	def _parse_one_token(self, tok: int, request) -> DeltaMessage \| None:
156	if tok == self.TOOL_START:
157	return DeltaMessage(
158	tool_calls=[
159	DeltaToolCall(
160	id=f"call-{random_uuid()}",
161	type="function",
162	function=DeltaFunctionCall(name="get_weather", arguments=""),
163	index=0,
164	)
165	]
166	)
167	if tok == self.TOOL_ARG:
168	return DeltaMessage(
169	tool_calls=[
170	DeltaToolCall(index=0, function=DeltaFunctionCall(arguments="a"))
171	]
172	)
173	if tok == self.REASON:
174	if not request.include_reasoning:
175	return None
176	return DeltaMessage(reasoning="r")
177	if tok == self.CONTENT:
178	return DeltaMessage(content="c")
179	return None
180
181
182	# ---------------------------------------------------------------------------
183	# Helpers shared across tests
184	# ---------------------------------------------------------------------------
185
186
187	def _make_stream_chunk(
188	token_ids: list[int],
189	index: int = 0,
190	finish_reason: str \| None = None,
191	request_id: str = "test-req",
192	usage: dict \| None = None,
193	metrics: PerRequestMetrics \| None = None,
194	logprobs: dict \| None = None,
195	prompt_token_ids: list[int] \| None = None,
196	) -> GenerateStreamResponse:
197	"""Build a GenerateStreamResponse SSE chunk."""
198	return GenerateStreamResponse(
199	request_id=request_id,
200	choices=[
201	GenerateResponseStreamChoice(
202	index=index,
203	token_ids=token_ids,
204	finish_reason=finish_reason,
205	logprobs=logprobs,
206	)
207	],
208	usage=UsageInfo(**usage) if usage else None,
209	metrics=metrics,
210	prompt_token_ids=prompt_token_ids,
211	)
212
213
214	def _placeholder_logprobs(token_ids: list[int]) -> dict:
215	"""Per-token logprob entries using token_id:N placeholders, as sent by
216	the generate worker."""
217	return {
218	"content": [
219	{
220	"token": f"token_id:{tid}",
221	"logprob": -0.5,
222	"bytes": None,
223	"top_logprobs": [],
224	}
225	for tid in token_ids
226	]
227	}
228
229
230	def _make_usage_chunk(
231	completion_tokens: int,
232	prompt_tokens: int = 0,
233	request_id: str = "test-req",
234	) -> GenerateStreamResponse:
235	"""Build a usage only final SSE chunk (empty choices)."""
236	return GenerateStreamResponse(
237	request_id=request_id,
238	choices=[],
239	usage=UsageInfo(
240	prompt_tokens=prompt_tokens,
241	completion_tokens=completion_tokens,
242	total_tokens=prompt_tokens + completion_tokens,
243	),
244	)
245
246
247	# ---------------------------------------------------------------------------
248	# Unit tests — no running server
249	# ---------------------------------------------------------------------------
250
251
252	@pytest.fixture(scope="module")
253	def tokenizer():
254	"""Load the tiny tokenizer used across unit tests."""
255	return get_tokenizer(MODEL_NAME)
256
257
258	@pytest.fixture(scope="module")
259	def derenderer(tokenizer, request):
260	"""Construct a minimal OnlineDerenderer backed by a stub renderer."""
261	renderer = MagicMock()
262	renderer.get_tokenizer.return_value = tokenizer
263
264	executor = ThreadPoolExecutor(max_workers=1)
265	request.addfinalizer(executor.shutdown)
266	renderer._executor = executor
267
268	model_config = MagicMock()
269	model_config.hf_config.model_type = "llama"
270	model_config.model = MODEL_NAME
271
272	return OnlineDerenderer(
273	model_config=model_config,
274	renderer=renderer,
275	request_logger=None,
276	chat_template=None,
277	chat_template_content_format="string",
278	trust_request_chat_template=False,
279	enable_auto_tools=False,
280	tool_parser=None,
281	reasoning_parser=None,
282	)
283
284
285	@pytest.fixture(scope="module")
286	def parsed_derenderer(tokenizer):
287	"""OnlineDerenderer with `_FakeParser` wired in as `self.parser`.
288
289	The parser-configured path runs `_derender_chat_stream_parsed` via
290	`make_async(..., executor=renderer._executor)`, so
291	`renderer._executor` must be a real `ThreadPoolExecutor`.
292	`loop.run_in_executor` cannot submit work to a `MagicMock`.
293	"""
294	renderer = MagicMock()
295	renderer.get_tokenizer.return_value = tokenizer
296	renderer._executor = ThreadPoolExecutor(max_workers=2)
297
298	model_config = MagicMock()
299	model_config.hf_config.model_type = "llama"
300	model_config.model = MODEL_NAME
301
302	dr = OnlineDerenderer(
303	model_config=model_config,
304	renderer=renderer,
305	request_logger=None,
306	chat_template=None,
307	chat_template_content_format="string",
308	)
309	dr.parser = _FakeParser
310	return dr
311
312
313	def test_non_streaming_derender_initializes_parser_prefix(
314	derenderer, tokenizer, monkeypatch
315	):
316	from unittest.mock import MagicMock
317
318	from vllm.entrypoints.generate.base.protocol import FunctionCall
319	from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
320
321	parser = MagicMock()
322	parser.parse.return_value = (
323	None,
324	None,
325	[FunctionCall(name="tool", arguments="{}")],
326	)
327	monkeypatch.setattr(derenderer, "parser", MagicMock(return_value=parser))
328	generated_ids = tokenizer.encode("answer", add_special_tokens=False)
329	response = GenerateResponse(
330	request_id="test",
331	prompt_token_ids=[11, 12],
332	choices=[
333	GenerateResponseChoice(
334	index=0,
335	token_ids=generated_ids,
336	finish_reason="length",
337	)
338	],
339	)
340	request = ChatCompletionRequest(
341	model=MODEL_NAME,
342	messages=[{"role": "user", "content": "question"}],
343	)
344
345	choices = derenderer._derender_chat(response, request)
346
347	parser.set_prompt_token_ids.assert_called_once_with([11, 12])
348	assert choices[0].message.tool_calls[0].function.name == "tool"
349	assert choices[0].finish_reason == "length"
350
351
352	class TestDetokenizeDelta:
353	"""_detokenize_delta: chunked decode must equal one shot decode."""
354
355	def _one_shot(self, tokenizer, token_ids: list[int]) -> str:
356	return tokenizer.decode(token_ids, skip_special_tokens=True)
357
358	def _chunked(self, derenderer, tokenizer, chunks: list[list[int]]) -> str:
359	state = DerenderStreamState()
360	parts: list[str] = []
361	for delta in chunks:
362	text, state = derenderer._detokenize_delta(
363	tokenizer, delta, state, skip_special_tokens=True
364	)
365	parts.append(text)
366	return "".join(parts)
367
368	def test_single_chunk(self, derenderer, tokenizer):
369	"""All tokens in one chunk == one shot decode."""
370	token_ids = tokenizer.encode("Hello world")[:8]
371	assert self._chunked(derenderer, tokenizer, [token_ids]) == self._one_shot(
372	tokenizer, token_ids
373	)
374
375	def test_two_equal_chunks(self, derenderer, tokenizer):
376	"""Split in half and reassemble == one shot."""
377	token_ids = tokenizer.encode("Hello world from streaming derender")[:12]
378	mid = len(token_ids) // 2
379	chunks = [token_ids[:mid], token_ids[mid:]]
380	assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
381	tokenizer, token_ids
382	)
383
384	def test_single_token_per_chunk(self, derenderer, tokenizer):
385	"""One token per chunk (most granular streaming) == one shot."""
386	token_ids = tokenizer.encode("incremental detokenization test")[:10]
387	chunks = [[t] for t in token_ids]
388	assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
389	tokenizer, token_ids
390	)
391
392	def test_empty_delta_passthrough(self, derenderer, tokenizer):
393	"""An empty delta (usage only chunk) emits empty string and preserves state."""
394	token_ids = tokenizer.encode("Hello")[:4]
395	_, state = derenderer._detokenize_delta(
396	tokenizer, token_ids, DerenderStreamState(), skip_special_tokens=True
397	)
398	text, new_state = derenderer._detokenize_delta(
399	tokenizer, [], state, skip_special_tokens=True
400	)
401	assert text == ""
402	assert new_state.prev_tokens == state.prev_tokens
403	assert new_state.prefix_offset == state.prefix_offset
404	assert new_state.read_offset == state.read_offset
405
406	def test_multibyte_char_split_across_chunks(self, derenderer, tokenizer):
407	"""A CJK/emoji char straddling chunk boundaries == one shot.
408
409	Regression test for held back trailing incomplete UTF-8 byte
410	sequences being dropped when the rebuild window marks them as
411	already read (see #46159).
412	"""
413	token_ids = tokenizer.encode("Hello ✅ world 日本語")[:16]
414	chunks = [[t] for t in token_ids]
415	assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
416	tokenizer, token_ids
417	)
418
419	def test_state_carries_across_calls(self, derenderer, tokenizer):
420	"""Decode state threads across calls. Text still matches one shot."""
421	t1 = tokenizer.encode("Hello")[:2]
422	t2 = tokenizer.encode(" world")[:2]
423	state = DerenderStreamState()
424	text1, state = derenderer._detokenize_delta(tokenizer, t1, state)
425	text2, state = derenderer._detokenize_delta(tokenizer, t2, state)
426	assert text1 + text2 == self._one_shot(tokenizer, t1 + t2)
427	# Offsets are rebased to the carried tail each chunk
428	assert state.prefix_offset == 0
429
430	def test_state_window_stays_bounded(self, derenderer, tokenizer):
431	"""prev_tokens must not grow with the number of chunks (bounded transport).
432
433	Guards that the carried decode window is a small constant
434	tail, so cumulative `stream_state` transport is O(n) and not O(n^2).
435	"""
436	token_ids = tokenizer.encode(
437	"a reasonably long ascii stream of tokens used to exercise the "
438	"window bound across many single token chunks so the carried "
439	"state cannot grow linearly with the generation length"
440	)
441	assert len(token_ids) > 32
442	state = DerenderStreamState()
443	max_window = 0
444	for tok in token_ids:
445	_, state = derenderer._detokenize_delta(tokenizer, [tok], state)
446	max_window = max(max_window, len(state.prev_tokens))
447	# Bounded by a small constant independent of len(token_ids)
448	assert max_window <= 32
449
450	def test_truncated_multibyte_char_is_held_back(self, derenderer, tokenizer):
451	"""Output ending mid-character emits no U+FFFD: batch == stream == engine."""
452	emoji_ids = tokenizer.encode("🌈", add_special_tokens=False)
453	byte_ids = emoji_ids[-4:]
454	assert len(byte_ids) == 4
455	output_ids = byte_ids[:-1]
456
457	batch, _ = derenderer._detokenize_delta(
458	tokenizer, output_ids, DerenderStreamState()
459	)
460	streamed = self._chunked(derenderer, tokenizer, [[t] for t in output_ids])
461	engine = _decode_stream_expected(tokenizer, [], output_ids)
462
463	assert "�" not in batch
464	assert batch == streamed == engine
465
466	full, _ = derenderer._detokenize_delta(
467	tokenizer, byte_ids, DerenderStreamState()
468	)
469	assert full.endswith("🌈")
470
471	@pytest.mark.parametrize(
472	("skip", "spaces", "preserve", "fast", "expected"),
473	[
474	(True, True, False, True, (True, True)),
475	(True, False, False, True, (True, True)),
476	(True, False, False, False, (True, False)),
477	(False, False, False, True, (False, False)),
478	(False, False, True, True, (False, False)),
479	(True, False, True, True, (False, False)),
480	(True, True, True, True, (False, True)),
481	],
482	)
483	def test_decode_params_mirror_engine(
484	self, monkeypatch, skip, spaces, preserve, fast, expected
485	):
486	"""Flags match the engine's fast or slow detokenizer for the tokenizer."""
487	monkeypatch.setattr(
488	online_derenderer_module, "uses_fast_detokenizer", lambda _: fast
489	)
490	request = CompletionRequest(
491	model=MODEL_NAME,
492	prompt="x",
493	skip_special_tokens=skip,
494	spaces_between_special_tokens=spaces,
495	)
496	assert _decode_params(None, request, preserve_special=preserve) == expected
497
498	def test_decode_params_default_without_request(self, tokenizer):
499	assert _decode_params(tokenizer, None) == (True, True)
500
501	def test_n_independent_streams_same_result(self, derenderer, tokenizer):
502	"""N parallel streams with the same token sequence give the same text."""
503	token_ids = tokenizer.encode("parallel streams")[:8]
504	mid = len(token_ids) // 2
505
506	results = []
507	for _ in range(3):
508	state = DerenderStreamState()
509	text, state = derenderer._detokenize_delta(
510	tokenizer, token_ids[:mid], state
511	)
512	text2, _ = derenderer._detokenize_delta(tokenizer, token_ids[mid:], state)
513	results.append(text + text2)
514
515	assert len(set(results)) == 1, "All independent streams must produce same text"
516	assert results[0] == self._one_shot(tokenizer, token_ids)
517
518
519	# ---------------------------------------------------------------------------
520	# Prompt seeded detokenization (matches the engine's primed detokenizer)
521	# ---------------------------------------------------------------------------
522
523
524	def _decode_stream_expected(
525	tokenizer, prompt_ids: list[int], output_ids: list[int]
526	) -> str:
527	"""Engine-equivalent decode via `DecodeStream`, primed by stepping the
528	prompt (`DecodeStream(ids=...)` needs tokenizers >= 0.22)."""
529	stream = DecodeStream(skip_special_tokens=True)
530	for tid in prompt_ids:
531	stream.step(tokenizer.backend_tokenizer, tid)
532	return "".join(
533	s
534	for tid in output_ids
535	if (s := stream.step(tokenizer.backend_tokenizer, tid)) is not None
536	)
537
538
539	@pytest.fixture(scope="module")
540	def leading_space_ids(tokenizer):
541	"""(prompt_ids, output_ids, expected) where seeded and unseeded decode
542	differ by the leading space."""
543	prompt_ids = tokenizer.encode("[INST] Hi [/INST]", add_special_tokens=False)
544	output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False)
545	expected = _decode_stream_expected(tokenizer, prompt_ids, output_ids)
546	unseeded = tokenizer.decode(output_ids, skip_special_tokens=True)
547	assert expected.startswith(" ") and not unseeded.startswith(" "), (
548	"fixture no longer reproduces the leading-space divergence; "
549	"pick a different prompt/output pair"
550	)
551	return prompt_ids, output_ids, expected
552
553
554	class TestSeedStreamState:
555	"""`_seed_stream_state`: pure helper building the initial decode state."""
556
557	def test_seeds_prev_tokens_from_prompt_tail(self, tokenizer, leading_space_ids):
558	prompt_ids, _, _ = leading_space_ids
559	state = _seed_stream_state(tokenizer, prompt_ids, skip_special_tokens=True)
560	assert state.prev_tokens
561	assert state.read_offset == len(state.prev_tokens)
562
563	def test_none_prompt_token_ids_returns_empty_state(self, tokenizer):
564	assert (
565	_seed_stream_state(tokenizer, None, skip_special_tokens=True)
566	== DerenderStreamState()
567	)
568
569	def test_empty_prompt_token_ids_returns_empty_state(self, tokenizer):
570	assert (
571	_seed_stream_state(tokenizer, [], skip_special_tokens=True)
572	== DerenderStreamState()
573	)
574
575
576	class TestPromptSeededLeadingSpace:
577	"""Every decode entry point keeps the first token's leading space when
578	given prompt_token_ids."""
579
580	@pytest.mark.asyncio
581	async def test_chat_batch_seeded_via_explicit_prompt_token_ids(
582	self, derenderer, leading_space_ids
583	):
584	prompt_ids, output_ids, expected = leading_space_ids
585	choices = await derenderer.derender_chat(
586	GenerateResponse(
587	request_id="t",
588	choices=[
589	GenerateResponseChoice(
590	index=0, token_ids=output_ids, finish_reason="stop"
591	)
592	],
593	),
594	prompt_token_ids=prompt_ids,
595	)
596	assert choices[0].message.content == expected
597
598	@pytest.mark.asyncio
599	async def test_chat_batch_seeded_via_generate_response_fallback(
600	self, derenderer, leading_space_ids
601	):
602	prompt_ids, output_ids, expected = leading_space_ids
603	choices = await derenderer.derender_chat(
604	GenerateResponse(
605	request_id="t",
606	choices=[
607	GenerateResponseChoice(
608	index=0, token_ids=output_ids, finish_reason="stop"
609	)
610	],
611	prompt_token_ids=prompt_ids,
612	),
613	)
614	assert choices[0].message.content == expected
615
616	@pytest.mark.asyncio
617	async def test_chat_batch_explicit_takes_precedence_over_fallback(
618	self, derenderer, tokenizer, leading_space_ids
619	):
620	prompt_ids, output_ids, expected = leading_space_ids
621	wrong_ids = tokenizer.encode("a different prompt", add_special_tokens=False)
622	choices = await derenderer.derender_chat(
623	GenerateResponse(
624	request_id="t",
625	choices=[
626	GenerateResponseChoice(
627	index=0, token_ids=output_ids, finish_reason="stop"
628	)
629	],
630	prompt_token_ids=wrong_ids,
631	),
632	prompt_token_ids=prompt_ids,
633	)
634	assert choices[0].message.content == expected
635
636	@pytest.mark.asyncio
637	async def test_chat_batch_without_prompt_context_stays_unseeded(
638	self, derenderer, leading_space_ids
639	):
640	_, output_ids, _ = leading_space_ids
641	choices = await derenderer.derender_chat(
642	GenerateResponse(
643	request_id="t",
644	choices=[
645	GenerateResponseChoice(
646	index=0, token_ids=output_ids, finish_reason="stop"
647	)
648	],
649	),
650	)
651	assert not choices[0].message.content.startswith(" ")
652
653	@pytest.mark.asyncio
654	async def test_completion_batch_seeded(self, derenderer, leading_space_ids):
655	prompt_ids, output_ids, expected = leading_space_ids
656	choices, _, _ = await derenderer.derender_completion(
657	[
658	GenerateResponse(
659	request_id="t",
660	choices=[
661	GenerateResponseChoice(
662	index=0, token_ids=output_ids, finish_reason="stop"
663	)
664	],
665	)
666	],
667	prompt_token_ids=[prompt_ids],
668	)
669	assert choices[0].text == expected
670
671	@pytest.mark.asyncio
672	@pytest.mark.parametrize("from_generate_chunk", [False, True])
673	async def test_chat_stream_first_chunk_seeded(
674	self, derenderer, leading_space_ids, from_generate_chunk
675	):
676	prompt_ids, output_ids, expected = leading_space_ids
677	mid = len(output_ids) // 2
678
679	chunk1, state = await derenderer.derender_chat_stream(
680	model=MODEL_NAME,
681	generate_chunk=_make_stream_chunk(
682	output_ids[:mid],
683	prompt_token_ids=prompt_ids if from_generate_chunk else None,
684	),
685	prompt_token_ids=None if from_generate_chunk else prompt_ids,
686	)
687	# Only needed on the first chunk; the carried state already holds
688	# the seeded window from here on.
689	chunk2, _ = await derenderer.derender_chat_stream(
690	model=MODEL_NAME,
691	generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"),
692	state=state,
693	)
694
695	streamed = (chunk1.choices[0].delta.content or "") + (
696	chunk2.choices[0].delta.content or ""
697	)
698	assert streamed == expected
699
700	@pytest.mark.asyncio
701	@pytest.mark.parametrize("from_generate_chunk", [False, True])
702	async def test_completion_stream_first_chunk_seeded(
703	self, derenderer, leading_space_ids, from_generate_chunk
704	):
705	prompt_ids, output_ids, expected = leading_space_ids
706	mid = len(output_ids) // 2
707
708	chunk1, state = await derenderer.derender_completion_stream(
709	model=MODEL_NAME,
710	generate_chunk=_make_stream_chunk(
711	output_ids[:mid],
712	prompt_token_ids=prompt_ids if from_generate_chunk else None,
713	),
714	prompt_token_ids=None if from_generate_chunk else prompt_ids,
715	)
716	chunk2, _ = await derenderer.derender_completion_stream(
717	model=MODEL_NAME,
718	generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"),
719	state=state,
720	)
721
722	assert chunk1.choices[0].text + chunk2.choices[0].text == expected
723
724	@pytest.mark.asyncio
725	async def test_parser_path_ephemeral_state_seeded(
726	self, parsed_derenderer, monkeypatch, leading_space_ids
727	):
728	"""The parser path's per-call detok window is seeded too, so
729	`parse_delta` sees the same text as coupled serving."""
730
731	class _EchoParser(Parser):
732	def parse_delta(
733	self,
734	delta_text,
735	delta_token_ids,
736	request,
737	prompt_token_ids=None,
738	*,
739	finished,
740	):
741	return DeltaMessage(content=delta_text) if delta_text else None
742
743	monkeypatch.setattr(parsed_derenderer, "parser", _EchoParser)
744	prompt_ids, output_ids, expected = leading_space_ids
745
746	chunk, _ = await parsed_derenderer.derender_chat_stream(
747	model=MODEL_NAME,
748	generate_chunk=_make_stream_chunk(output_ids, finish_reason="stop"),
749	chat_request=_chat_request(),
750	prompt_token_ids=prompt_ids,
751	)
752	assert chunk.choices[0].delta.content == expected
753
754
755	@pytest.mark.asyncio
756	@pytest.mark.parametrize(
757	"stream_method",
758	["derender_chat_stream", "derender_completion_stream"],
759	)
760	async def test_streaming_detokenization_runs_off_event_loop(
761	derenderer, monkeypatch, stream_method
762	):
763	"""Streaming detokenization runs on the renderer executor."""
764	event_loop_thread_id = threading.get_ident()
765	executor_thread_id = derenderer.renderer._executor.submit(
766	threading.get_ident
767	).result()
768
769	detokenize_thread_id = None
770	original_detokenize = online_derenderer_module.detokenize_incrementally
771
772	def record_detokenize_thread(args, *kwargs):
773	nonlocal detokenize_thread_id
774	detokenize_thread_id = threading.get_ident()
775	return original_detokenize(args, *kwargs)
776
777	monkeypatch.setattr(
778	online_derenderer_module,
779	"detokenize_incrementally",
780	record_detokenize_thread,
781	)
782
783	await getattr(derenderer, stream_method)(
784	model=MODEL_NAME,
785	generate_chunk=_make_stream_chunk([10]),
786	)
787
788	assert detokenize_thread_id is not None
789	assert detokenize_thread_id == executor_thread_id
790	assert detokenize_thread_id != event_loop_thread_id
791
792
793	class TestDerenderCompletionStream:
794	"""derender_completion_stream: streaming output parity with one shot."""
795
796	@pytest.mark.asyncio
797	async def test_chunked_equals_oneshot(self, derenderer, tokenizer):
798	"""Sum of streaming text chunks == one shot tokenizer.decode."""
799	token_ids = tokenizer.encode("streaming completion test")[:10]
800	mid = len(token_ids) // 2
801
802	state = DerenderStreamState()
803	chunk1, state = await derenderer.derender_completion_stream(
804	model=MODEL_NAME,
805	generate_chunk=_make_stream_chunk(token_ids[:mid]),
806	state=state,
807	)
808	chunk2, _ = await derenderer.derender_completion_stream(
809	model=MODEL_NAME,
810	generate_chunk=_make_stream_chunk(token_ids[mid:], finish_reason="stop"),
811	state=state,
812	)
813
814	streamed_text = chunk1.choices[0].text + chunk2.choices[0].text
815	one_shot = tokenizer.decode(token_ids, skip_special_tokens=True)
816	assert streamed_text == one_shot
817
818	@pytest.mark.asyncio
819	async def test_usage_chunk_passthrough(self, derenderer, tokenizer):
820	"""Usage only final chunk (empty choices) is passed through correctly."""
821	usage_chunk = _make_usage_chunk(completion_tokens=10, prompt_tokens=5)
822	chunk, state = await derenderer.derender_completion_stream(
823	model=MODEL_NAME,
824	generate_chunk=usage_chunk,
825	)
826	assert chunk.choices == []
827	assert chunk.usage is not None
828	assert chunk.usage.completion_tokens == 10
829	assert chunk.usage.prompt_tokens == 5
830
831	@pytest.mark.asyncio
832	async def test_prompt_tokens_in_usage(self, derenderer, tokenizer):
833	"""prompt_tokens is correctly forwarded into usage on a usage chunk."""
834	token_ids = tokenizer.encode("hello")[:3]
835	usage_chunk = _make_usage_chunk(
836	completion_tokens=len(token_ids), prompt_tokens=7
837	)
838	chunk, _ = await derenderer.derender_completion_stream(
839	model=MODEL_NAME,
840	generate_chunk=usage_chunk,
841	prompt_tokens=7,
842	)
843	assert chunk.usage is not None
844	assert chunk.usage.prompt_tokens == 7
845
846	@pytest.mark.asyncio
847	async def test_none_state_initialises_correctly(self, derenderer, tokenizer):
848	"""Passing state=None (first call) initialises an empty DerenderStreamState."""
849	token_ids = tokenizer.encode("hello")[:4]
850	chunk, state = await derenderer.derender_completion_stream(
851	model=MODEL_NAME,
852	generate_chunk=_make_stream_chunk(token_ids),
853	state=None,
854	)
855	assert isinstance(state, DerenderStreamState)
856	assert chunk.choices[0].text == tokenizer.decode(
857	token_ids, skip_special_tokens=True
858	)
859
860	@pytest.mark.asyncio
861	async def test_skip_special_tokens_threaded(self, derenderer, tokenizer):
862	"""completion_request.skip_special_tokens is honored (not hardcoded True)."""
863	eos = tokenizer.eos_token_id
864	if eos is None:
865	pytest.skip("tokenizer has no eos token to exercise special stripping")
866	token_ids = tokenizer.encode("hi")[:2] + [eos]
867
868	async def _text(skip: bool) -> str:
869	req = CompletionRequest(
870	model=MODEL_NAME, prompt="x", skip_special_tokens=skip
871	)
872	chunk, _ = await derenderer.derender_completion_stream(
873	model=MODEL_NAME,
874	generate_chunk=_make_stream_chunk(token_ids),
875	completion_request=req,
876	)
877	return chunk.choices[0].text
878
879	# skip=False must retain the special token; skip=True must strip it.
880	assert await _text(False) != await _text(True)
881
882	@pytest.mark.asyncio
883	async def test_finish_reason_forwarded(self, derenderer, tokenizer):
884	"""finish_reason from the generate chunk reaches the derendered choice."""
885	token_ids = tokenizer.encode("done")[:2]
886	chunk, _ = await derenderer.derender_completion_stream(
887	model=MODEL_NAME,
888	generate_chunk=_make_stream_chunk(token_ids, finish_reason="length"),
889	)
890	assert chunk.choices[0].finish_reason == "length"
891
892
893	class TestStreamLogprobs:
894	"""Streaming derender must carry per-chunk logprobs with placeholders
895	resolved, matching what the generate streaming path emits."""
896
897	@pytest.mark.asyncio
898	async def test_chat_stream_logprobs_resolved_per_chunk(self, derenderer, tokenizer):
899	"""Each streamed chunk carries logprobs with token_id:N resolved."""
900	token_ids = tokenizer.encode("hello world")[:6]
901	mid = len(token_ids) // 2
902
903	state = None
904	for part in (token_ids[:mid], token_ids[mid:]):
905	chunk, state = await derenderer.derender_chat_stream(
906	model=MODEL_NAME,
907	generate_chunk=_make_stream_chunk(
908	part, logprobs=_placeholder_logprobs(part)
909	),
910	state=state,
911	)
912	logprobs = chunk.choices[0].logprobs
913	assert logprobs is not None and logprobs.content is not None
914	assert len(logprobs.content) == len(part)
915	for entry in logprobs.content:
916	assert not entry.token.startswith("token_id:"), (
917	f"placeholder not resolved: {entry.token!r}"
918	)
919
920	@pytest.mark.asyncio
921	async def test_chat_stream_logprobs_multibyte_across_chunks(
922	self, derenderer, tokenizer
923	):
924	"""Byte-fallback correction works when a multi-byte character's
925	tokens are split across chunks (context carried in stream_state)."""
926	token_ids = tokenizer.encode("👍", add_special_tokens=False)
927	if len(token_ids) < 2 or "�" not in tokenizer.decode([token_ids[-1]]):
928	pytest.skip("Tokenizer does not byte-split this character")
929
930	state = None
931	last_chunk = None
932	# One token per chunk: every entry after the first needs cross-chunk
933	# context to resolve without U+FFFD.
934	for i, tid in enumerate(token_ids):
935	last_chunk, state = await derenderer.derender_chat_stream(
936	model=MODEL_NAME,
937	generate_chunk=_make_stream_chunk(
938	[tid], logprobs=_placeholder_logprobs([tid])
939	),
940	state=state,
941	)
942
943	assert last_chunk is not None
944	final_entry = last_chunk.choices[0].logprobs.content[0]
945	assert not final_entry.token.endswith("�"), (
946	"byte-fallback correction failed across chunk boundary: "
947	f"{final_entry.token!r}"
948	)
949
950	@pytest.mark.asyncio
951	async def test_completion_stream_logprobs_text_offset_absolute(
952	self, derenderer, tokenizer
953	):
954	"""text_offset continues across chunks instead of restarting at 0."""
955	token_ids = tokenizer.encode("hello world")[:6]
956	mid = len(token_ids) // 2
957
958	chunk1, state = await derenderer.derender_completion_stream(
959	model=MODEL_NAME,
960	generate_chunk=_make_stream_chunk(
961	token_ids[:mid], logprobs=_placeholder_logprobs(token_ids[:mid])
962	),
963	)
964	chunk2, _ = await derenderer.derender_completion_stream(
965	model=MODEL_NAME,
966	generate_chunk=_make_stream_chunk(
967	token_ids[mid:], logprobs=_placeholder_logprobs(token_ids[mid:])
968	),
969	state=state,
970	)
971
972	assert chunk1.choices[0].logprobs.text_offset[0] == 0
973	assert chunk2.choices[0].logprobs.text_offset[0] == len(chunk1.choices[0].text)
974
975	@pytest.mark.asyncio
976	async def test_stream_logprob_state_stays_bounded(self, derenderer, tokenizer):
977	"""Carried logprob context never exceeds the 4-token window."""
978	token_ids = tokenizer.encode("the quick brown fox jumps over the lazy dog")
979	state = None
980	for tid in token_ids:
981	_, state = await derenderer.derender_chat_stream(
982	model=MODEL_NAME,
983	generate_chunk=_make_stream_chunk(
984	[tid], logprobs=_placeholder_logprobs([tid])
985	),
986	state=state,
987	)
988	assert state is not None
989	assert len(state.logprob_context_token_ids) <= 4
990
991	@pytest.mark.asyncio
992	async def test_stream_without_logprobs_unchanged(self, derenderer, tokenizer):
993	"""Chunks without logprobs keep logprobs=None on the output choice."""
994	token_ids = tokenizer.encode("hello")[:3]
995	chunk, _ = await derenderer.derender_chat_stream(
996	model=MODEL_NAME,
997	generate_chunk=_make_stream_chunk(token_ids),
998	)
999	assert chunk.choices[0].logprobs is None
1000
1001
1002	class TestDerenderChatStream:
1003	"""derender_chat_stream: plain detok branch (no parser)."""
1004
1005	@pytest.mark.asyncio
1006	async def test_role_on_first_chunk_only(self, derenderer, tokenizer):
1007	"""role='assistant' appears in the first chunk, not subsequent ones."""
1008	token_ids = tokenizer.encode("hello world")[:6]
1009	mid = len(token_ids) // 2
1010
1011	state = DerenderStreamState()
1012	chunk1, state = await derenderer.derender_chat_stream(
1013	model=MODEL_NAME,
1014	generate_chunk=_make_stream_chunk(token_ids[:mid]),
1015	state=state,
1016	)
1017	chunk2, _ = await derenderer.derender_chat_stream(
1018	model=MODEL_NAME,
1019	generate_chunk=_make_stream_chunk(token_ids[mid:], finish_reason="stop"),
1020	state=state,
1021	)
1022
1023	assert chunk1.choices[0].delta.role == "assistant"
1024	assert chunk2.choices[0].delta.role is None
1025
1026	@pytest.mark.asyncio
1027	async def test_metrics_passthrough(self, derenderer):
1028	metrics = PerRequestMetrics(
1029	speculative_decoding=SpeculativeDecodingMetrics(
1030	mean_acceptance_length=2.0,
1031	draft_acceptance_rate=0.5,
1032	acceptance_histogram=[0, 1],
1033	num_spec_steps=1,
1034	num_accepted_draft_tokens=1,
1035	num_draft_tokens=2,
1036	num_spec_tokens=1,
1037	)
1038	)
1039	chunk, _ = await derenderer.derender_chat_stream(
1040	model=MODEL_NAME,
1041	generate_chunk=_make_stream_chunk([], metrics=metrics),
1042	state=DerenderStreamState(),
1043	)
1044	assert chunk.metrics == metrics
1045
1046	@pytest.mark.asyncio
1047	async def test_chunked_equals_oneshot(self, derenderer, tokenizer):
1048	"""Sum of streaming content deltas == one shot decode."""
1049	token_ids = tokenizer.encode("streaming chat derender text")[:10]
1050	mid = len(token_ids) // 2
1051
1052	state = DerenderStreamState()
1053	chunk1, state = await derenderer.derender_chat_stream(
1054	model=MODEL_NAME,
1055	generate_chunk=_make_stream_chunk(token_ids[:mid]),
1056	state=state,
1057	)
1058	chunk2, _ = await derenderer.derender_chat_stream(
1059	model=MODEL_NAME,
1060	generate_chunk=_make_stream_chunk(token_ids[mid:]),
1061	state=state,
1062	)
1063
1064	streamed = (chunk1.choices[0].delta.content or "") + (
1065	chunk2.choices[0].delta.content or ""
1066	)
1067	one_shot = tokenizer.decode(token_ids, skip_special_tokens=True)
1068	assert streamed == one_shot
1069
1070	@pytest.mark.asyncio
1071	async def test_parser_configured_missing_chat_request_raises(
1072	self, parsed_derenderer
1073	):
1074	"""A parser configured model must never fall through to plain detok
1075	even when `chat_request` is omitted. If allowed this would leak raw
1076	reasoning/tool markup into `delta.content`. `ServingDerender`
1077	pre-checks this too (400 before touching the tokenizer). This pins
1078	the `OnlineDerenderer` level backstop."""
1079	with pytest.raises(ValueError, match="chat_request"):
1080	await parsed_derenderer.derender_chat_stream(
1081	model=MODEL_NAME,
1082	generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
1083	state=None,
1084	chat_request=None,
1085	)
1086
1087
1088	def _chat_request(**kwargs):
1089	kwargs.setdefault("messages", [{"role": "user", "content": "hi"}])
1090	kwargs.setdefault("model", MODEL_NAME)
1091	return ChatCompletionRequest(**kwargs)
1092
1093
1094	class TestDerenderChatStreamParsed:
1095	"""derender_chat_stream: parser branch (replay + parse_delta) exercised
1096	against the deterministic `_FakeParser` so these pin OnlineDerenderer's
1097	own replay/merge/pin/finish_reason logic independent of any real
1098	parser's markup grammar (covered separately by the parser_server backed
1099	integration tests below)."""
1100
1101	@pytest.mark.asyncio
1102	async def test_dispatches_and_emits_content(self, parsed_derenderer):
1103	chunk, state = await parsed_derenderer.derender_chat_stream(
1104	model=MODEL_NAME,
1105	generate_chunk=_make_stream_chunk(
1106	[_FakeParser.CONTENT], finish_reason="stop"
1107	),
1108	chat_request=_chat_request(),
1109	)
1110	assert chunk.choices[0].delta.content == "c"
1111	assert chunk.choices[0].delta.role == "assistant"
1112	assert chunk.choices[0].finish_reason == "stop"
1113	assert state.output_token_ids == [_FakeParser.CONTENT]
1114
1115	@pytest.mark.asyncio
1116	async def test_role_sent_once(self, parsed_derenderer):
1117	chat_request = _chat_request()
1118	chunk1, state = await parsed_derenderer.derender_chat_stream(
1119	model=MODEL_NAME,
1120	generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
1121	chat_request=chat_request,
1122	)
1123	chunk2, _ = await parsed_derenderer.derender_chat_stream(
1124	model=MODEL_NAME,
1125	generate_chunk=_make_stream_chunk(
1126	[_FakeParser.CONTENT], finish_reason="stop"
1127	),
1128	state=state,
1129	chat_request=chat_request,
1130	)
1131	assert chunk1.choices[0].delta.role == "assistant"
1132	assert chunk2.choices[0].delta.role is None
1133
1134	@pytest.mark.asyncio
1135	async def test_finish_only_chunk_flushes(self, parsed_derenderer):
1136	"""A finish only chunk (no new tokens) still calls `parse_delta`
1137	once with `finished=True` to flush buffered state."""
1138	chat_request = _chat_request()
1139	_, state = await parsed_derenderer.derender_chat_stream(
1140	model=MODEL_NAME,
1141	generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
1142	chat_request=chat_request,
1143	)
1144	chunk2, _ = await parsed_derenderer.derender_chat_stream(
1145	model=MODEL_NAME,
1146	generate_chunk=_make_stream_chunk([], finish_reason="stop"),
1147	state=state,
1148	chat_request=chat_request,
1149	)
1150	assert chunk2.choices[0].delta.content == "FLUSH"
1151	assert chunk2.choices[0].finish_reason == "stop"
1152
1153	@pytest.mark.asyncio
1154	async def test_chunking_invariance(self, parsed_derenderer):
1155	"""1 token per chunk vs a single whole chunk assemble identically.
1156
1157	Every chunk (live or replayed) is fed to `parse_delta` at its own
1158	producer granularity (see `_derender_chat_stream_parsed`'s
1159	docstring). This proves that choice doesn't change the assembled
1160	output for a well behaved incremental parser. Each call's effects
1161	fold together the same way regardless of how many tokens land in
1162	one call vs. many."""
1163	token_ids = [
1164	_FakeParser.REASON,
1165	_FakeParser.REASON,
1166	_FakeParser.TOOL_START,
1167	_FakeParser.TOOL_ARG,
1168	_FakeParser.TOOL_ARG,
1169	_FakeParser.CONTENT,
1170	]
1171	chat_request = _chat_request(
1172	tools=[{"type": "function", "function": {"name": "get_weather"}}],
1173	tool_choice="auto",
1174	include_reasoning=True,
1175	)
1176
1177	async def _assemble(chunks: list[list[int]]) -> dict:
1178	state = None
1179	content = ""
1180	reasoning = ""
1181	tool_args = ""
1182	for i, tids in enumerate(chunks):
1183	finish = "stop" if i == len(chunks) - 1 else None
1184	chunk, state = await parsed_derenderer.derender_chat_stream(
1185	model=MODEL_NAME,
1186	generate_chunk=_make_stream_chunk(tids, finish_reason=finish),
1187	state=state,
1188	chat_request=chat_request,
1189	)
1190	delta = chunk.choices[0].delta
1191	content += delta.content or ""
1192	reasoning += delta.reasoning or ""
1193	for tc in delta.tool_calls:
1194	if tc.function and tc.function.arguments:
1195	tool_args += tc.function.arguments
1196	return {"content": content, "reasoning": reasoning, "tool_args": tool_args}
1197
1198	whole = await _assemble([token_ids])
1199	one_at_a_time = await _assemble([[t] for t in token_ids])
1200
1201	assert whole == one_at_a_time
1202	assert whole == {"content": "c", "reasoning": "rr", "tool_args": "aa"}
1203
1204	@pytest.mark.asyncio
1205	async def test_replay_preserves_chunk_boundaries(
1206	self, parsed_derenderer, monkeypatch
1207	):
1208	"""A multi-token chunk (e.g. a speculative decoding step) must be
1209	replayed as one `parse_delta` call, not one per token. Parsers
1210	with per-call state such as `history_tool_call_cnt` would
1211	otherwise diverge from standard serving."""
1212
1213	class _CallCountingParser(_FakeParser):
1214	"""Emits how many `parse_delta` calls it has seen."""
1215
1216	def __init__(self, args, *kwargs):
1217	super().__init__(args, *kwargs)
1218	self.num_calls = 0
1219
1220	def parse_delta(
1221	self,
1222	delta_text,
1223	delta_token_ids,
1224	request,
1225	prompt_token_ids=None,
1226	*,
1227	finished,
1228	):
1229	self.num_calls += 1
1230	return DeltaMessage(content=str(self.num_calls))
1231
1232	monkeypatch.setattr(parsed_derenderer, "parser", _CallCountingParser)
1233	chat_request = _chat_request()
1234
1235	_, state = await parsed_derenderer.derender_chat_stream(
1236	model=MODEL_NAME,
1237	generate_chunk=_make_stream_chunk([_FakeParser.CONTENT] * 3),
1238	chat_request=chat_request,
1239	)
1240	assert state.output_chunk_lens == [3]
1241
1242	chunk, _ = await parsed_derenderer.derender_chat_stream(
1243	model=MODEL_NAME,
1244	generate_chunk=_make_stream_chunk(
1245	[_FakeParser.CONTENT], finish_reason="stop"
1246	),
1247	state=state,
1248	chat_request=chat_request,
1249	)
1250	assert chunk.choices[0].delta.content == "2"
1251
1252	@pytest.mark.asyncio
1253	async def test_tool_call_id_pinned_across_chunks(self, parsed_derenderer):
1254	"""Once an index's ID is recorded in `last_tool_call_ids`, a later
1255	id bearing delta for that same index is pinned to the recorded
1256	value rather than a freshly (re-)generated one."""
1257	chat_request = _chat_request(
1258	tools=[{"type": "function", "function": {"name": "get_weather"}}],
1259	tool_choice="auto",
1260	)
1261	chunk1, state = await parsed_derenderer.derender_chat_stream(
1262	model=MODEL_NAME,
1263	generate_chunk=_make_stream_chunk([_FakeParser.TOOL_START]),
1264	chat_request=chat_request,
1265	)
1266	first_id = chunk1.choices[0].delta.tool_calls[0].id
1267	assert first_id is not None
1268	assert state.last_tool_call_ids == [first_id]
1269
1270	chunk2, state = await parsed_derenderer.derender_chat_stream(
1271	model=MODEL_NAME,
1272	generate_chunk=_make_stream_chunk(
1273	[_FakeParser.TOOL_START], finish_reason="stop"
1274	),
1275	state=state,
1276	chat_request=chat_request,
1277	)
1278	assert chunk2.choices[0].delta.tool_calls[0].id == first_id
1279	assert state.last_tool_call_ids == [first_id]
1280
1281	@pytest.mark.asyncio
1282	async def test_finish_reason_rewritten_to_tool_calls(self, parsed_derenderer):
1283	chat_request = _chat_request(
1284	tools=[{"type": "function", "function": {"name": "get_weather"}}],
1285	tool_choice="auto",
1286	)
1287	chunk, _ = await parsed_derenderer.derender_chat_stream(
1288	model=MODEL_NAME,
1289	generate_chunk=_make_stream_chunk(
1290	[_FakeParser.TOOL_START, _FakeParser.TOOL_ARG], finish_reason="stop"
1291	),
1292	chat_request=chat_request,
1293	)
1294	assert chunk.choices[0].finish_reason == "tool_calls"
1295
1296	@pytest.mark.asyncio
1297	async def test_finish_reason_stop_for_named_tool_choice(self, parsed_derenderer):
1298	# ChatCompletionRequest's `check_tool_usage` is a mode="before"
1299	# validator, so it sees the raw value and only accepts "auto",
1300	# "required" or a dict, never an already built
1301	# ChatCompletionNamedToolChoiceParam.
1302	chat_request = _chat_request(
1303	tools=[{"type": "function", "function": {"name": "get_weather"}}],
1304	tool_choice={"type": "function", "function": {"name": "get_weather"}},
1305	)
1306	chunk, _ = await parsed_derenderer.derender_chat_stream(
1307	model=MODEL_NAME,
1308	generate_chunk=_make_stream_chunk(
1309	[_FakeParser.TOOL_START, _FakeParser.TOOL_ARG], finish_reason="stop"
1310	),
1311	chat_request=chat_request,
1312	)
1313	assert chunk.choices[0].finish_reason == "stop"
1314
1315	@pytest.mark.asyncio
1316	async def test_include_reasoning_false_suppresses_reasoning(
1317	self, parsed_derenderer
1318	):
1319	chat_request = _chat_request(include_reasoning=False)
1320	chunk, _ = await parsed_derenderer.derender_chat_stream(
1321	model=MODEL_NAME,
1322	generate_chunk=_make_stream_chunk(
1323	[_FakeParser.REASON, _FakeParser.CONTENT], finish_reason="stop"
1324	),
1325	chat_request=chat_request,
1326	)
1327	delta = chunk.choices[0].delta
1328	assert delta.reasoning is None
1329	assert delta.content == "c"
1330
1331
1332	# ---------------------------------------------------------------------------
1333	# Harmony / GPT-OSS replay — unit, no server
1334	# ---------------------------------------------------------------------------
1335
1336	HARMONY_MODEL = "openai/gpt-oss-20b"
1337	HARMONY_REASONING = "The user wants 2 plus 3."
1338	HARMONY_ANSWER = "The answer is 5."
1339	HARMONY_PROMPT = "<\|start\|>user<\|message\|>Add 2 and 3.<\|end\|><\|start\|>assistant"
1340	HARMONY_OUTPUT = (
1341	f"<\|channel\|>analysis<\|message\|>{HARMONY_REASONING}<\|end\|>"
1342	f"<\|start\|>assistant<\|channel\|>final<\|message\|>{HARMONY_ANSWER}<\|return\|>"
1343	)
1344
1345
1346	@pytest.fixture(scope="module")
1347	def harmony_encode():
1348	"""Encoder for canned GPT-OSS harmony token sequences."""
1349	pytest.importorskip("openai_harmony")
1350
1351	# Pre-caches the o200k_base BPE file that openai-harmony's Rust backend
1352	# downloads on first use, same as the sibling suite's E2E harmony tests.
1353	from tests.entrypoints.scale_out.derender.test_derender import (
1354	_ensure_harmony_vocab,
1355	)
1356	from vllm.entrypoints.openai.parser.harmony_utils import get_encoding
1357
1358	_ensure_harmony_vocab()
1359	encoding = get_encoding()
1360
1361	def _encode(harmony_str: str) -> list[int]:
1362	return encoding.encode(harmony_str, allowed_special="all")
1363
1364	return _encode
1365
1366
1367	@pytest.fixture(scope="module")
1368	def harmony_tokenizer():
1369	pytest.importorskip("openai_harmony")
1370	return get_tokenizer(HARMONY_MODEL, trust_remote_code=True)
1371
1372
1373	@pytest.fixture(scope="module")
1374	def harmony_derenderer(harmony_tokenizer):
1375	"""OnlineDerenderer whose parser resolves to the real `HarmonyParser`.
1376
1377	Same shape as `parsed_derenderer` (mocked renderer, real executor) but
1378	with a real tokenizer and parser, so the replay path is exercised
1379	against Harmony's actual channel grammar rather than a stub.
1380	"""
1381	from vllm.parser.harmony import HarmonyParser
1382
1383	renderer = MagicMock()
1384	renderer.get_tokenizer.return_value = harmony_tokenizer
1385	renderer._executor = ThreadPoolExecutor(max_workers=2)
1386
1387	model_config = MagicMock()
1388	model_config.model = HARMONY_MODEL
1389	model_config.hf_config.model_type = "gpt_oss"
1390	model_config.hf_text_config.model_type = "gpt_oss"
1391	model_config.hf_overrides = None
1392
1393	dr = OnlineDerenderer(
1394	model_config=model_config,
1395	renderer=renderer,
1396	request_logger=None,
1397	chat_template=None,
1398	chat_template_content_format="string",
1399	enable_auto_tools=True,
1400	tool_parser="openai",
1401	reasoning_parser="openai_gptoss",
1402	)
1403	assert dr.use_harmony
1404	assert dr.parser is HarmonyParser
1405	return dr
1406
1407
1408	async def _stream_harmony_deltas(
1409	derenderer,
1410	chat_request,
1411	output_ids: list[int],
1412	prompt_ids: list[int],
1413	chunk_size: int,
1414	) -> list[DeltaMessage]:
1415	"""Drive `output_ids` through the parser path in fixed size chunks.
1416
1417	Returns the per chunk `DeltaMessage`s in order so callers can assert on
1418	intermediate emissions, not just the assembled result.
1419	"""
1420	deltas: list[DeltaMessage] = []
1421	state = None
1422	for start in range(0, len(output_ids), chunk_size):
1423	tids = output_ids[start : start + chunk_size]
1424	is_last = start + chunk_size >= len(output_ids)
1425	chunk, state = await derenderer.derender_chat_stream(
1426	model=HARMONY_MODEL,
1427	generate_chunk=_make_stream_chunk(
1428	tids, finish_reason="stop" if is_last else None
1429	),
1430	state=state,
1431	chat_request=chat_request,
1432	prompt_token_ids=prompt_ids,
1433	)
1434	deltas.append(chunk.choices[0].delta)
1435	return deltas
1436
1437
1438	class TestDerenderChatStreamHarmony:
1439	"""derender_chat_stream: replay + `parse_delta` against real HarmonyParser.
1440
1441	Skipped where `openai_harmony` is not installed. `prompt_token_ids` is
1442	threaded through for call shape fidelity only, since HarmonyParser
1443	derives its state from the output tokens alone.
1444	"""
1445
1446	@pytest.mark.asyncio
1447	async def test_reasoning_never_leaks_as_content_midstream(
1448	self, harmony_derenderer, harmony_encode
1449	):
1450	"""Regression guard for the RFC's original replay + diff design.
1451
1452	`HarmonyParser.parse()` always flushes to EOS. Mid-stream that
1453	raises `HarmonyError` and the recovery branch re-emits the in
1454	flight message on the `final` channel, so partial analysis surfaces
1455	as content. Replay + `parse_delta` must never do that. Driven one
1456	token per chunk, the finest granularity a client can produce.
1457	"""
1458	output_ids = harmony_encode(HARMONY_OUTPUT)
1459	prompt_ids = harmony_encode(HARMONY_PROMPT)
1460	chat_request = _chat_request(model=HARMONY_MODEL, include_reasoning=True)
1461
1462	deltas = await _stream_harmony_deltas(
1463	harmony_derenderer, chat_request, output_ids, prompt_ids, chunk_size=1
1464	)
1465
1466	reasoning = ""
1467	content = ""
1468	for i, delta in enumerate(deltas):
1469	reasoning += delta.reasoning or ""
1470	content += delta.content or ""
1471	# Any analysis text emitted as content breaks the prefix
1472	# property, since the two channels share no prefix here.
1473	assert HARMONY_ANSWER.startswith(content), (
1474	f"content after chunk {i} is not a prefix of the final "
1475	f"channel text: {content!r}"
1476	)
1477	assert HARMONY_REASONING.startswith(reasoning)
1478	assert not delta.tool_calls
1479
1480	assert reasoning == HARMONY_REASONING
1481	assert content == HARMONY_ANSWER
1482
1483	@pytest.mark.asyncio
1484	async def test_stream_matches_batch(self, harmony_derenderer, harmony_encode):
1485	"""Streamed assembly equals one shot `/derender` over the same IDs."""
1486	output_ids = harmony_encode(HARMONY_OUTPUT)
1487	prompt_ids = harmony_encode(HARMONY_PROMPT)
1488	chat_request = _chat_request(model=HARMONY_MODEL, include_reasoning=True)
1489
1490	deltas = await _stream_harmony_deltas(
1491	harmony_derenderer, chat_request, output_ids, prompt_ids, chunk_size=3
1492	)
1493	reasoning = "".join(d.reasoning or "" for d in deltas)
1494	content = "".join(d.content or "" for d in deltas)
1495
1496	batch_choices = await harmony_derenderer.derender_chat(
1497	GenerateResponse(
1498	request_id="test-harmony-batch",
1499	choices=[
1500	GenerateResponseChoice(
1501	index=0, token_ids=output_ids, finish_reason="stop"
1502	)
1503	],
1504	),
1505	chat_request,
1506	)
1507	message = batch_choices[0].message
1508
1509	assert reasoning == message.reasoning
1510	assert content == message.content
1511
1512
1513	class TestDerenderStreamStateValidation:
1514	"""DerenderStreamState rejects malformed caller supplied offsets/lengths."""
1515
1516	def test_negative_prefix_offset_rejected(self):
1517	with pytest.raises(ValidationError):
1518	DerenderStreamState(prefix_offset=-1)
1519
1520	def test_negative_read_offset_rejected(self):
1521	with pytest.raises(ValidationError):
1522	DerenderStreamState(read_offset=-1)
1523
1524	def test_prev_tokens_over_cap_rejected(self):
1525	with pytest.raises(ValidationError):
1526	DerenderStreamState(prev_tokens=["a"] * 1025)
1527
1528	def test_prev_tokens_at_cap_accepted(self):
1529	state = DerenderStreamState(prev_tokens=["a"] * 1024)
1530	assert len(state.prev_tokens) == 1024
1531
1532	def test_output_chunk_lens_mismatch_rejected(self):
1533	with pytest.raises(ValidationError):
1534	DerenderStreamState(output_token_ids=[1, 2, 3], output_chunk_lens=[1, 1])
1535
1536	def test_output_chunk_lens_zero_entry_rejected(self):
1537	with pytest.raises(ValidationError):
1538	DerenderStreamState(output_token_ids=[1, 2], output_chunk_lens=[0, 2])
1539
1540	def test_output_chunk_lens_matching_accepted(self):
1541	state = DerenderStreamState(
1542	output_token_ids=[1, 2, 3], output_chunk_lens=[1, 2]
1543	)
1544	assert state.output_chunk_lens == [1, 2]
1545
1546
1547	_GEN = {"request_id": "t", "choices": []}
1548	_GEN_BAD = {**_GEN, "prompt_token_ids": [-1]}
1549
1550
1551	@pytest.mark.parametrize(
1552	"model_cls,payload",
1553	[
1554	(DerenderChatRequest, {"generate_response": _GEN, "prompt_token_ids": [-1]}),
1555	(DerenderChatRequest, {"generate_response": _GEN_BAD}),
1556	(
1557	DerenderCompletionRequest,
1558	{"generate_responses": [_GEN], "prompt_token_ids": [[-1]]},
1559	),
1560	(
1561	DerenderChatStreamRequest,
1562	{"stream": True, "generate_chunk": _GEN, "prompt_token_ids": [-1]},
1563	),
1564	(
1565	DerenderCompletionStreamRequest,
1566	{"stream": True, "generate_chunk": _GEN_BAD},
1567	),
1568	],
1569	)
1570	def test_negative_prompt_token_ids_rejected(model_cls, payload):
1571	"""Negative prompt ids would reach convert_ids_to_tokens, so they get a
1572	400 at parse time like `GenerateRequest.token_ids` does."""
1573	with pytest.raises(ValidationError, match="prompt_token_ids"):
1574	model_cls.model_validate(payload)
1575
1576
1577	class TestServingDerenderStreamErrorHandling:
1578	"""Malformed stream_state must surface as 400 and not an unhandled 500."""
1579
1580	def _make_serving(self, side_effect: Exception):
1581	models = MagicMock()
1582	models.is_base_model.return_value = True
1583	models.model_config = MagicMock()
1584	models.model_config.max_model_len = 100_000
1585
1586	online_derenderer = MagicMock()
1587	online_derenderer.parser = None
1588	online_derenderer.derender_completion_stream = AsyncMock(
1589	side_effect=side_effect
1590	)
1591	online_derenderer.derender_chat_stream = AsyncMock(side_effect=side_effect)
1592
1593	return ServingDerender(models=models, online_derenderer=online_derenderer)
1594
1595	@pytest.mark.asyncio
1596	@pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
1597	async def test_completion_stream_bad_state_returns_400(self, exc):
1598	serving = self._make_serving(exc)
1599	request = DerenderCompletionStreamRequest(
1600	stream=True,
1601	model=MODEL_NAME,
1602	generate_chunk=_make_stream_chunk([1, 2]),
1603	stream_state=DerenderStreamState(),
1604	)
1605	result = await serving.derender_completion_stream_response(request)
1606	assert isinstance(result, ErrorResponse)
1607	assert result.error.code == 400
1608
1609	@pytest.mark.asyncio
1610	@pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
1611	async def test_chat_stream_bad_state_returns_400(self, exc):
1612	serving = self._make_serving(exc)
1613	request = DerenderChatStreamRequest(
1614	stream=True,
1615	model=MODEL_NAME,
1616	generate_chunk=_make_stream_chunk([1, 2]),
1617	stream_state=DerenderStreamState(),
1618	)
1619	result = await serving.derender_chat_stream_response(request)
1620	assert isinstance(result, ErrorResponse)
1621	assert result.error.code == 400
1622
1623
1624	class TestServingDerenderStreamValidation:
1625	"""If checks cannot run in `derender_chat_stream_response`: all must
1626	reject with 400 before ever calling into `online_derenderer` (i.e.
1627	before touching the tokenizer)."""
1628
1629	def _make_serving(self, *, parser_configured: bool, max_model_len: int = 100_000):
1630	models = MagicMock()
1631	models.is_base_model.return_value = True
1632	models.model_config = MagicMock()
1633	models.model_config.max_model_len = max_model_len
1634
1635	online_derenderer = MagicMock()
1636	online_derenderer.parser = MagicMock() if parser_configured else None
1637	online_derenderer.derender_chat_stream = AsyncMock(
1638	return_value=(
1639	ChatCompletionStreamResponse(id="t", model=MODEL_NAME, choices=[]),
1640	DerenderStreamState(),
1641	)
1642	)
1643	return ServingDerender(models=models, online_derenderer=online_derenderer)
1644
1645	@pytest.mark.asyncio
1646	async def test_missing_chat_request_with_parser_rejected(self):
1647	serving = self._make_serving(parser_configured=True)
1648	request = DerenderChatStreamRequest(
1649	stream=True,
1650	model=MODEL_NAME,
1651	generate_chunk=_make_stream_chunk([1, 2]),
1652	)
1653	result = await serving.derender_chat_stream_response(request)
1654	assert isinstance(result, ErrorResponse)
1655	assert result.error.code == 400
1656	assert "chat_request" in result.error.message
1657	serving.online_derenderer.derender_chat_stream.assert_not_called()
1658
1659	@pytest.mark.asyncio
1660	async def test_missing_chat_request_without_parser_ok(self):
1661	serving = self._make_serving(parser_configured=False)
1662	request = DerenderChatStreamRequest(
1663	stream=True,
1664	model=MODEL_NAME,
1665	generate_chunk=_make_stream_chunk([1, 2]),
1666	)
1667	result = await serving.derender_chat_stream_response(request)
1668	assert not isinstance(result, ErrorResponse)
1669
1670	@pytest.mark.asyncio
1671	async def test_missing_prompt_token_ids_with_parser_rejected(self):
1672	"""A parser configured model must reject a missing prompt_token_ids
1673	the same way it rejects a missing chat_request. Without it,
1674	parse_delta cannot tell whether the prompt left reasoning open and
1675	would misclassify reasoning content as plain content."""
1676	serving = self._make_serving(parser_configured=True)
1677	request = DerenderChatStreamRequest(
1678	stream=True,
1679	model=MODEL_NAME,
1680	generate_chunk=_make_stream_chunk([1, 2]),
1681	chat_request=_chat_request(),
1682	)
1683	result = await serving.derender_chat_stream_response(request)
1684	assert isinstance(result, ErrorResponse)
1685	assert result.error.code == 400
1686	assert "prompt_token_ids" in result.error.message
1687	serving.online_derenderer.derender_chat_stream.assert_not_called()
1688
1689	@pytest.mark.asyncio
1690	async def test_prompt_token_ids_present_with_parser_ok(self):
1691	serving = self._make_serving(parser_configured=True)
1692	request = DerenderChatStreamRequest(
1693	stream=True,
1694	model=MODEL_NAME,
1695	generate_chunk=_make_stream_chunk([1, 2]),
1696	chat_request=_chat_request(),
1697	prompt_token_ids=[1, 2, 3],
1698	)
1699	result = await serving.derender_chat_stream_response(request)
1700	assert not isinstance(result, ErrorResponse)
1701
1702	@pytest.mark.asyncio
1703	async def test_oversized_output_token_ids_rejected(self):
1704	serving = self._make_serving(parser_configured=False, max_model_len=4)
1705	request = DerenderChatStreamRequest(
1706	stream=True,
1707	model=MODEL_NAME,
1708	generate_chunk=_make_stream_chunk([1, 2, 3]),
1709	stream_state=DerenderStreamState(
1710	output_token_ids=[1, 2], output_chunk_lens=[2]
1711	),
1712	)
1713	result = await serving.derender_chat_stream_response(request)
1714	assert isinstance(result, ErrorResponse)
1715	assert result.error.code == 400
1716	assert "max_model_len" in result.error.message
1717	serving.online_derenderer.derender_chat_stream.assert_not_called()
1718
1719	@pytest.mark.asyncio
1720	async def test_oversized_prompt_token_ids_rejected(self):
1721	"""prompt_token_ids is caller supplied and otherwise unbounded. A
1722	parser configured deployment rescans it in full on every chunk
1723	(is_reasoning_end / adjust_initial_state_from_prompt), so it must be
1724	bounded the same way output_token_ids is."""
1725	serving = self._make_serving(parser_configured=True, max_model_len=4)
1726	request = DerenderChatStreamRequest(
1727	stream=True,
1728	model=MODEL_NAME,
1729	generate_chunk=_make_stream_chunk([1]),
1730	chat_request=_chat_request(),
1731	prompt_token_ids=[1, 2, 3, 4, 5],
1732	)
1733	result = await serving.derender_chat_stream_response(request)
1734	assert isinstance(result, ErrorResponse)
1735	assert result.error.code == 400
1736	assert "max_model_len" in result.error.message
1737	serving.online_derenderer.derender_chat_stream.assert_not_called()
1738
1739	@pytest.mark.asyncio
1740	async def test_too_many_choices_rejected(self):
1741	"""Each streamed chunk contains at most one choice (a single
1742	DerenderStreamState is threaded through every choice). The check
1743	could never fire since derender_chat_stream itself rejects anything
1744	above 1 first."""
1745	serving = self._make_serving(parser_configured=False)
1746	two_choices = GenerateStreamResponse(
1747	request_id="t",
1748	choices=[
1749	GenerateResponseStreamChoice(index=i, token_ids=[1]) for i in range(2)
1750	],
1751	)
1752	request = DerenderChatStreamRequest(
1753	stream=True, model=MODEL_NAME, generate_chunk=two_choices
1754	)
1755	result = await serving.derender_chat_stream_response(request)
1756	assert isinstance(result, ErrorResponse)
1757	assert result.error.code == 400
1758	assert "at most one choice" in result.error.message
1759	serving.online_derenderer.derender_chat_stream.assert_not_called()
1760
1761
1762	# ---------------------------------------------------------------------------
1763	# Integration tests — require a live render server
1764	# ---------------------------------------------------------------------------
1765
1766
1767	@pytest.fixture(scope="module")
1768	def server():
1769	with RemoteLaunchRenderServer(MODEL_NAME, []) as remote_server:
1770	yield remote_server
1771
1772
1773	@pytest_asyncio.fixture
1774	async def client(server):
1775	async with httpx.AsyncClient(
1776	base_url=server.url_for(""), timeout=30.0
1777	) as http_client:
1778	yield http_client
1779
1780
1781	async def _render_chat(client) -> dict:
1782	"""Render a minimal chat request and return the GenerateRequest dict."""
1783	resp = await client.post(
1784	"/v1/chat/completions/render",
1785	json={
1786	"model": MODEL_NAME,
1787	"messages": [{"role": "user", "content": "Hello"}],
1788	},
1789	)
1790	assert resp.status_code == 200
1791	return resp.json()
1792
1793
1794	@pytest.mark.asyncio
1795	async def test_streaming_completion_derender_roundtrip(client):
1796	"""Streaming completions derender: chunked text == non streaming text."""
1797	gen_req = await _render_chat(client)
1798	token_ids: list[int] = gen_req["token_ids"][:8]
1799	mid = len(token_ids) // 2
1800	chunk1_ids, chunk2_ids = token_ids[:mid], token_ids[mid:]
1801
1802	# Non streaming baseline.
1803	non_stream_resp = await client.post(
1804	"/v1/completions/derender",
1805	json={
1806	"model": MODEL_NAME,
1807	"generate_responses": [
1808	{
1809	"request_id": "test-ns",
1810	"choices": [
1811	{
1812	"index": 0,
1813	"token_ids": token_ids,
1814	"finish_reason": "stop",
1815	}
1816	],
1817	}
1818	],
1819	},
1820	)
1821	assert non_stream_resp.status_code == 200
1822	expected_text = non_stream_resp.json()["choices"][0]["text"]
1823
1824	# Streaming call 1.
1825	r1 = await client.post(
1826	"/v1/completions/derender",
1827	json={
1828	"stream": True,
1829	"model": MODEL_NAME,
1830	"generate_chunk": {
1831	"request_id": "test-s",
1832	"choices": [
1833	{"index": 0, "token_ids": chunk1_ids, "finish_reason": None}
1834	],
1835	},
1836	"stream_state": None,
1837	},
1838	)
1839	assert r1.status_code == 200
1840	d1 = r1.json()
1841	text1 = d1["chunk"]["choices"][0]["text"]
1842	state1 = d1["stream_state"]
1843
1844	# Streaming call 2 (final chunk).
1845	r2 = await client.post(
1846	"/v1/completions/derender",
1847	json={
1848	"stream": True,
1849	"model": MODEL_NAME,
1850	"generate_chunk": {
1851	"request_id": "test-s",
1852	"choices": [
1853	{"index": 0, "token_ids": chunk2_ids, "finish_reason": "stop"}
1854	],
1855	},
1856	"stream_state": state1,
1857	},
1858	)
1859	assert r2.status_code == 200
1860	text2 = r2.json()["chunk"]["choices"][0]["text"]
1861
1862	assert text1 + text2 == expected_text
1863
1864
1865	@pytest.mark.asyncio
1866	async def test_streaming_chat_derender_roundtrip(client):
1867	"""Streaming chat derender (plain detok): chunked text == non streaming text."""
1868	gen_req = await _render_chat(client)
1869	token_ids: list[int] = gen_req["token_ids"][:8]
1870	mid = len(token_ids) // 2
1871	chunk1_ids, chunk2_ids = token_ids[:mid], token_ids[mid:]
1872
1873	# Non streaming baseline.
1874	ns = await client.post(
1875	"/v1/chat/completions/derender",
1876	json={
1877	"model": MODEL_NAME,
1878	"generate_response": {
1879	"request_id": "test-ns",
1880	"choices": [
1881	{
1882	"index": 0,
1883	"token_ids": token_ids,
1884	"finish_reason": "stop",
1885	}
1886	],
1887	},
1888	},
1889	)
1890	assert ns.status_code == 200
1891	expected_content = ns.json()["choices"][0]["message"]["content"]
1892
1893	# Streaming call 1.
1894	r1 = await client.post(
1895	"/v1/chat/completions/derender",
1896	json={
1897	"stream": True,
1898	"model": MODEL_NAME,
1899	"generate_chunk": {
1900	"request_id": "test-s",
1901	"choices": [
1902	{"index": 0, "token_ids": chunk1_ids, "finish_reason": None}
1903	],
1904	},
1905	"stream_state": None,
1906	},
1907	)
1908	assert r1.status_code == 200
1909	d1 = r1.json()
1910	text1 = d1["chunk"]["choices"][0]["delta"].get("content") or ""
1911	state1 = d1["stream_state"]
1912	# role=assistant on the first chunk
1913	assert d1["chunk"]["choices"][0]["delta"].get("role") == "assistant"
1914
1915	# Streaming call 2.
1916	r2 = await client.post(
1917	"/v1/chat/completions/derender",
1918	json={
1919	"stream": True,
1920	"model": MODEL_NAME,
1921	"generate_chunk": {
1922	"request_id": "test-s",
1923	"choices": [
1924	{"index": 0, "token_ids": chunk2_ids, "finish_reason": "stop"}
1925	],
1926	},
1927	"stream_state": state1,
1928	},
1929	)
1930	assert r2.status_code == 200
1931	d2 = r2.json()
1932	text2 = d2["chunk"]["choices"][0]["delta"].get("content") or ""
1933	# role must NOT be repeated on subsequent chunks
1934	assert d2["chunk"]["choices"][0]["delta"].get("role") is None
1935
1936	assert text1 + text2 == expected_content
1937
1938
1939	@pytest.mark.asyncio
1940	async def test_streaming_derender_invalid_body_returns_400(client):
1941	"""Missing required field in streaming request returns 400."""
1942	r = await client.post(
1943	"/v1/completions/derender",
1944	json={
1945	"stream": True,
1946	# missing required 'model' and 'generate_chunk'
1947	},
1948	)
1949	assert r.status_code == 400
1950
1951
1952	@pytest.mark.asyncio
1953	async def test_streaming_derender_non_object_body_returns_400(client):
1954	"""A non object JSON body (e.g. a list) returns 400, not a 500."""
1955	r = await client.post(
1956	"/v1/completions/derender",
1957	json=[1, 2, 3],
1958	)
1959	assert r.status_code == 400
1960
1961
1962	@pytest.mark.asyncio
1963	async def test_streaming_usage_chunk(client):
1964	"""Usage only final chunk is forwarded with correct token counts."""
1965	gen_req = await _render_chat(client)
1966	token_ids: list[int] = gen_req["token_ids"][:6]
1967	state: dict = {}
1968
1969	# Send content chunk first.
1970	r1 = await client.post(
1971	"/v1/completions/derender",
1972	json={
1973	"stream": True,
1974	"model": MODEL_NAME,
1975	"generate_chunk": {
1976	"request_id": "usage-test",
1977	"choices": [
1978	{"index": 0, "token_ids": token_ids, "finish_reason": "stop"}
1979	],
1980	},
1981	"stream_state": None,
1982	},
1983	)
1984	assert r1.status_code == 200
1985	state = r1.json()["stream_state"]
1986
1987	# Send usage only final chunk.
1988	r2 = await client.post(
1989	"/v1/completions/derender",
1990	json={
1991	"stream": True,
1992	"model": MODEL_NAME,
1993	"generate_chunk": {
1994	"request_id": "usage-test",
1995	"choices": [],
1996	"usage": {
1997	"prompt_tokens": 10,
1998	"completion_tokens": len(token_ids),
1999	"total_tokens": 10 + len(token_ids),
2000	},
2001	},
2002	"stream_state": state,
2003	"prompt_tokens": 10,
2004	},
2005	)
2006	assert r2.status_code == 200
2007	d2 = r2.json()
2008	assert d2["chunk"]["choices"] == []
2009	assert d2["chunk"]["usage"]["prompt_tokens"] == 10
2010	assert d2["chunk"]["usage"]["completion_tokens"] == len(token_ids)
2011
2012
2013	# ---------------------------------------------------------------------------
2014	# Integration tests for parser configured (reasoning + tool calls) require a
2015	# live render server. Mirrors the parser_server / parser_tokenizer pattern
2016	# from test_derender.py.
2017	# ---------------------------------------------------------------------------
2018
2019	PARSER_MODEL = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"
2020
2021	_PARSER_TOOLS = [
2022	{
2023	"type": "function",
2024	"function": {
2025	"name": "get_weather",
2026	"description": "Get weather for a city",
2027	"parameters": {
2028	"type": "object",
2029	"properties": {"city": {"type": "string"}},
2030	},
2031	},
2032	}
2033	]
2034
2035
2036	@pytest.fixture(scope="module")
2037	def parser_server():
2038	args = [
2039	"--enable-auto-tool-choice",
2040	"--tool-call-parser",
2041	"hermes",
2042	"--reasoning-parser",
2043	"deepseek_r1",
2044	]
2045	with RemoteLaunchRenderServer(PARSER_MODEL, args) as remote_server:
2046	yield remote_server
2047
2048
2049	@pytest_asyncio.fixture
2050	async def parser_client(parser_server):
2051	async with httpx.AsyncClient(
2052	base_url=parser_server.url_for(""), timeout=60.0
2053	) as http_client:
2054	yield http_client
2055
2056
2057	@pytest.fixture(scope="module")
2058	def parser_tokenizer():
2059	return get_tokenizer(PARSER_MODEL)
2060
2061
2062	def _require_parser_markers(tokenizer, text: str, *markers: str) -> list[int]:
2063	"""Encode text and skip the test if any marker is lost in roundtrip."""
2064	ids = tokenizer.encode(text, add_special_tokens=False)
2065	decoded = tokenizer.decode(ids, skip_special_tokens=False)
2066	for m in markers:
2067	if m not in decoded:
2068	pytest.skip(f"Marker {m!r} lost in encode->decode roundtrip")
2069	return ids
2070
2071
2072	async def _render_parser_chat(client, messages: list[dict]) -> dict:
2073	resp = await client.post(
2074	"/v1/chat/completions/render",
2075	json={"model": PARSER_MODEL, "messages": messages},
2076	)
2077	assert resp.status_code == 200, resp.text
2078	return resp.json()
2079
2080
2081	@pytest.mark.asyncio
2082	@pytest.mark.parametrize(
2083	"chunking",
2084	[
2085	"whole",
2086	"per_token",
2087	pytest.param(
2088	"triples",
2089	marks=pytest.mark.xfail(
2090	reason="<think> leaks into reasoning when it shares a "
2091	"multi-token delta with reasoning text (vllm#55195)"
2092	),
2093	),
2094	],
2095	)
2096	async def test_stream_parsed_matches_batch_reasoning(
2097	parser_client, parser_tokenizer, chunking
2098	):
2099	"""Streamed == batch: assembled reasoning/content equal the non
2100	streaming `/derender` result over the same token IDs, for every
2101	chunking of the same output (chunking invariance)."""
2102	messages = [{"role": "user", "content": "What is 2+3?"}]
2103	gen_req = await _render_parser_chat(parser_client, messages)
2104
2105	reasoning_text = "The user wants 2 plus 3. That is 5."
2106	answer_text = "The answer is 5."
2107	output_text = f"<think>{reasoning_text}</think>{answer_text}"
2108	output_ids = _require_parser_markers(parser_tokenizer, output_text, "</think>")
2109
2110	chat_request = {
2111	"model": PARSER_MODEL,
2112	"messages": messages,
2113	"include_reasoning": True,
2114	}
2115
2116	batch_resp = await parser_client.post(
2117	"/v1/chat/completions/derender",
2118	json={
2119	"model": PARSER_MODEL,
2120	"generate_response": {
2121	"request_id": "batch",
2122	"choices": [
2123	{"index": 0, "token_ids": output_ids, "finish_reason": "stop"}
2124	],
2125	},
2126	"prompt_tokens": len(gen_req["token_ids"]),
2127	"chat_request": chat_request,
2128	},
2129	)
2130	assert batch_resp.status_code == 200, batch_resp.text
2131	batch_msg = batch_resp.json()["choices"][0]["message"]
2132
2133	n = len(output_ids)
2134	chunk_sizes = {
2135	"whole": [n],
2136	"per_token": [1] * n,
2137	"triples": [3] * (n // 3) + [n % 3],
2138	}[chunking]
2139	chunk_sizes = [c for c in chunk_sizes if c > 0]
2140	streamed = await stream_chat_derender(
2141	parser_client,
2142	output_ids,
2143	chunk_sizes,
2144	chat_request,
2145	len(gen_req["token_ids"]),
2146	gen_req["token_ids"],
2147	)
2148	assert streamed["content"] == batch_msg["content"]
2149	assert streamed["reasoning"] == batch_msg.get("reasoning")
2150
2151
2152	@pytest.mark.asyncio
2153	async def test_stream_parsed_matches_batch_tool_call(parser_client, parser_tokenizer):
2154	"""Tool call name+id+arguments parity between streamed and batch, plus
2155	the finish_reason -> "tool_calls" rewrite for auto tool choice."""
2156	messages = [{"role": "user", "content": "Weather in Paris?"}]
2157	gen_req = await _render_parser_chat(parser_client, messages)
2158
2159	output_text = (
2160	"<think>Let me check.</think>"
2161	'<tool_call>\n{"name": "get_weather", '
2162	'"arguments": {"city": "Paris"}}\n</tool_call>'
2163	)
2164	output_ids = _require_parser_markers(
2165	parser_tokenizer, output_text, "</think>", "<tool_call>", "</tool_call>"
2166	)
2167	chat_request = {
2168	"model": PARSER_MODEL,
2169	"messages": messages,
2170	"tools": _PARSER_TOOLS,
2171	"tool_choice": "auto",
2172	}
2173
2174	batch_resp = await parser_client.post(
2175	"/v1/chat/completions/derender",
2176	json={
2177	"model": PARSER_MODEL,
2178	"generate_response": {
2179	"request_id": "batch",
2180	"choices": [
2181	{"index": 0, "token_ids": output_ids, "finish_reason": "stop"}
2182	],
2183	},
2184	"prompt_tokens": len(gen_req["token_ids"]),
2185	"chat_request": chat_request,
2186	},
2187	)
2188	assert batch_resp.status_code == 200, batch_resp.text
2189	batch_choice = batch_resp.json()["choices"][0]
2190	if not batch_choice["message"]["tool_calls"]:
2191	pytest.skip("Model did not emit a tool call")
2192
2193	for chunk_sizes in ([len(output_ids)], [1] * len(output_ids)):
2194	# Snapshot of tool_calls[0]["arguments"] after every chunk, to check
2195	# it only ever grows by appending. It never retracts or duplicates
2196	# already streamed text. That's the exact failure mode the
2197	# replay + parse_delta design (over diff-based streaming) exists to
2198	# avoid. Parity with the batch result alone wouldn't catch a
2199	# transient mid-stream regression.
2200	arg_snapshots: list[str] = []
2201
2202	def _record(
2203	tool_calls: list[dict], _snapshots: list[str] = arg_snapshots
2204	) -> None:
2205	if tool_calls and tool_calls[0]["arguments"]:
2206	_snapshots.append(tool_calls[0]["arguments"])
2207
2208	streamed = await stream_chat_derender(
2209	parser_client,
2210	output_ids,
2211	chunk_sizes,
2212	chat_request,
2213	len(gen_req["token_ids"]),
2214	gen_req["token_ids"],
2215	on_chunk=_record,
2216	)
2217	assert streamed["tool_calls"]
2218	assert streamed["tool_calls"][0]["name"] == "get_weather"
2219	assert streamed["tool_calls"][0]["id"] is not None
2220	assert json.loads(streamed["tool_calls"][0]["arguments"]) == json.loads(
2221	batch_choice["message"]["tool_calls"][0]["function"]["arguments"]
2222	)
2223	assert streamed["finish_reason"] == "tool_calls"
2224
2225	assert arg_snapshots, "expected at least one tool-call argument delta"
2226	for prev, curr in zip(arg_snapshots, arg_snapshots[1:]):
2227	assert curr.startswith(prev), (
2228	f"tool-call arguments regressed: {prev!r} -> {curr!r}"
2229	)
2230	assert arg_snapshots[-1] == streamed["tool_calls"][0]["arguments"]
2231
2232
2233	@pytest.mark.asyncio
2234	async def test_stream_parsed_cjk_across_chunk_boundaries(
2235	parser_client, parser_tokenizer
2236	):
2237	"""CJK/emoji tokens split across chunk boundaries with a parser active
2238	across the reasoning -> content transition (parser path regression
2239	guard analogous to #46159 in the plain detok path)."""
2240	messages = [{"role": "user", "content": "Reply in Chinese"}]
2241	gen_req = await _render_parser_chat(parser_client, messages)
2242
2243	reasoning_text = "思考中"
2244	answer_text = "你好世界 😀"
2245	output_text = f"<think>{reasoning_text}</think>{answer_text}"
2246	output_ids = _require_parser_markers(parser_tokenizer, output_text, "</think>")
2247
2248	chat_request = {
2249	"model": PARSER_MODEL,
2250	"messages": messages,
2251	"include_reasoning": True,
2252	}
2253	streamed = await stream_chat_derender(
2254	parser_client,
2255	output_ids,
2256	[1] * len(output_ids),
2257	chat_request,
2258	len(gen_req["token_ids"]),
2259	gen_req["token_ids"],
2260	)
2261	assert "�" not in (streamed["content"] or "")
2262	assert "�" not in (streamed["reasoning"] or "")
2263	assert answer_text in (streamed["content"] or "")
2264
2265
2266	@pytest.mark.asyncio
2267	async def test_stream_parsed_include_reasoning_false(parser_client, parser_tokenizer):
2268	"""include_reasoning=False emits no reasoning deltas on the streaming
2269	parser path."""
2270	messages = [{"role": "user", "content": "What is 2+3?"}]
2271	gen_req = await _render_parser_chat(parser_client, messages)
2272
2273	output_text = "<think>reasoning here</think>The answer is 5."
2274	output_ids = _require_parser_markers(parser_tokenizer, output_text, "</think>")
2275
2276	chat_request = {
2277	"model": PARSER_MODEL,
2278	"messages": messages,
2279	"include_reasoning": False,
2280	}
2281	streamed = await stream_chat_derender(
2282	parser_client,
2283	output_ids,
2284	[1] * len(output_ids),
2285	chat_request,
2286	len(gen_req["token_ids"]),
2287	gen_req["token_ids"],
2288	)
2289	assert streamed["reasoning"] is None
2290
2291
2292	@pytest.mark.asyncio
2293	async def test_stream_parsed_missing_chat_request_rejected(parser_client):
2294	"""Parser configured + no chat_request on the streaming endpoint -> 400
2295	(the live-server counterpart to the mocked
2296	`TestServingDerenderStreamValidation` checks above)."""
2297	resp = await parser_client.post(
2298	"/v1/chat/completions/derender",
2299	json={
2300	"stream": True,
2301	"model": PARSER_MODEL,
2302	"generate_chunk": {
2303	"request_id": "reject-test",
2304	"choices": [{"index": 0, "token_ids": [1, 2], "finish_reason": None}],
2305	},
2306	},
2307	)
2308	assert resp.status_code == 400
2309	assert "chat_request" in resp.json()["error"]["message"]
2310
2311
2312	@pytest.mark.asyncio
2313	async def test_stream_parsed_missing_prompt_token_ids_rejected(parser_client):
2314	"""Parser configured + chat_request given but no prompt_token_ids on the
2315	streaming endpoint -> 400 (the live-server counterpart to the mocked
2316	`test_missing_prompt_token_ids_with_parser_rejected` above). Without
2317	prompt_token_ids, parse_delta cannot tell whether the prompt left
2318	reasoning open and would silently misclassify reasoning as content."""
2319	messages = [{"role": "user", "content": "Hello"}]
2320	resp = await parser_client.post(
2321	"/v1/chat/completions/derender",
2322	json={
2323	"stream": True,
2324	"model": PARSER_MODEL,
2325	"generate_chunk": {
2326	"request_id": "reject-test",
2327	"choices": [{"index": 0, "token_ids": [1, 2], "finish_reason": None}],
2328	},
2329	"chat_request": {
2330	"model": PARSER_MODEL,
2331	"messages": messages,
2332	"include_reasoning": True,
2333	},
2334	},
2335	)
2336	assert resp.status_code == 400
2337	assert "prompt_token_ids" in resp.json()["error"]["message"]