# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Unit tests for streaming derender.
Tests are split into two layers:
1. Unit tests (no server): covers `_detokenize_delta` correctness
(chunked == one-shot) and `derender_completion_stream` /
`derender_chat_stream` logic via a real tokenizer on a tiny model.
The parser path is covered both with a deterministic stub parser and
with the real `HarmonyParser` (skipped without `openai_harmony`).
2. Integration tests (require a running render server): covers the full
HTTP round-trip through the streaming endpoint. Marked with
`@pytest.mark.asyncio` and gated by the `server` / `client`
fixtures from the sibling `test_derender.py`.
"""
import json
import threading
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
import pytest_asyncio
from pydantic import ValidationError
from tokenizers.decoders import DecodeStream
import vllm.renderers.online_derenderer as online_derenderer_module
from tests.entrypoints.scale_out.derender.utils import stream_chat_derender
from tests.utils import RemoteLaunchRenderServer
from vllm.entrypoints.generate.base.protocol import (
DeltaFunctionCall,
DeltaMessage,
DeltaToolCall,
PerRequestMetrics,
SpeculativeDecodingMetrics,
)
from vllm.entrypoints.openai.chat_completion.protocol import (
ChatCompletionRequest,
ChatCompletionStreamResponse,
)
from vllm.entrypoints.openai.completion.protocol import CompletionRequest
from vllm.entrypoints.scale_out.derender.serving import ServingDerender
from vllm.entrypoints.scale_out.token_in_token_out.protocol import (
DerenderChatRequest,
DerenderChatStreamRequest,
DerenderCompletionRequest,
DerenderCompletionStreamRequest,
DerenderStreamState,
GenerateStreamResponse,
GenerateTokensChoice,
GenerateTokensResponse,
GenerateTokensStreamChoice,
GenerateTokensStreamResponse,
)
from vllm.entrypoints.serve.engine.protocol import ErrorResponse, UsageInfo
from vllm.parser import Parser
from vllm.renderers.online_derenderer import (
OnlineDerenderer,
_decode_params,
_seed_stream_state,
)
from vllm.tokenizers import get_tokenizer
from vllm.utils import random_uuid
MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
def _merge_fake_deltas(
accumulated: DeltaMessage | None, new: DeltaMessage | None
) -> DeltaMessage | None:
"""Fold one token's worth of `_FakeParser` output into a per call
accumulator, the same way a real parser merges multiple tokens'
effects into the single `DeltaMessage` it returns per `parse_delta`
call.
"""
if new is None:
return accumulated
if accumulated is None:
accumulated = DeltaMessage()
if new.content:
accumulated.content = (accumulated.content or "") + new.content
if new.reasoning:
accumulated.reasoning = (accumulated.reasoning or "") + new.reasoning
for tc in new.tool_calls:
starts_new_call = tc.id is not None or (
tc.function is not None and tc.function.name is not None
)
existing = (
None
if starts_new_call
else next((t for t in accumulated.tool_calls if t.index == tc.index), None)
)
if existing is None:
accumulated.tool_calls.append(tc)
continue
if existing.function is None:
existing.function = DeltaFunctionCall()
if tc.function is not None and tc.function.arguments:
existing.function.arguments = (
existing.function.arguments or ""
) + tc.function.arguments
return accumulated
class _FakeParser(Parser):
"""Deterministic `Parser` stub for unit testing the replay/merge/pin
mechanics of `OnlineDerenderer._derender_chat_stream_parsed` in
isolation from any real reasoning/tool parser's markup grammar.
Processes every id in `delta_token_ids` in order and folds their
effects into one `DeltaMessage`, like a real parser handling a
multi-token delta in a single call. Ignores `delta_text`:
- `TOOL_START` opens a new tool call at index 0 (id + name).
- `TOOL_ARG` appends one `"a"` to that tool call's arguments.
- `REASON` emits one `"r"` of reasoning (suppressed when the request
has `include_reasoning=False`, mirroring the real parsers).
- `CONTENT` emits one `"c"` of content.
- An empty `delta_token_ids` with `finished=True` (the finish only
flush call) emits a `"FLUSH"` content sentinel so tests can confirm
it happened.
- Anything else emits nothing.
"""
TOOL_START = 9
TOOL_ARG = 10
REASON = 11
CONTENT = 12
def parse_delta(
self,
delta_text,
delta_token_ids,
request,
prompt_token_ids=None,
*,
finished,
):
if not delta_token_ids:
return DeltaMessage(content="FLUSH") if finished else None
accumulated: DeltaMessage | None = None
for tok in delta_token_ids:
accumulated = _merge_fake_deltas(
accumulated, self._parse_one_token(tok, request)
)
return accumulated
def _parse_one_token(self, tok: int, request) -> DeltaMessage | None:
if tok == self.TOOL_START:
return DeltaMessage(
tool_calls=[
DeltaToolCall(
id=f"call-{random_uuid()}",
type="function",
function=DeltaFunctionCall(name="get_weather", arguments=""),
index=0,
)
]
)
if tok == self.TOOL_ARG:
return DeltaMessage(
tool_calls=[
DeltaToolCall(index=0, function=DeltaFunctionCall(arguments="a"))
]
)
if tok == self.REASON:
if not request.include_reasoning:
return None
return DeltaMessage(reasoning="r")
if tok == self.CONTENT:
return DeltaMessage(content="c")
return None
# ---------------------------------------------------------------------------
# Helpers shared across tests
# ---------------------------------------------------------------------------
def _make_stream_chunk(
token_ids: list[int],
index: int = 0,
finish_reason: str | None = None,
request_id: str = "test-req",
usage: dict | None = None,
metrics: PerRequestMetrics | None = None,
logprobs: dict | None = None,
prompt_token_ids: list[int] | None = None,
) -> GenerateStreamResponse:
"""Build a GenerateStreamResponse SSE chunk."""
return GenerateTokensStreamResponse(
request_id=request_id,
choices=[
GenerateTokensStreamChoice(
index=index,
token_ids=token_ids,
finish_reason=finish_reason,
logprobs=logprobs,
)
],
usage=UsageInfo(**usage) if usage else None,
metrics=metrics,
prompt_token_ids=prompt_token_ids,
)
def _placeholder_logprobs(token_ids: list[int]) -> dict:
"""Per-token logprob entries using token_id:N placeholders, as sent by
the generate worker."""
return {
"content": [
{
"token": f"token_id:{tid}",
"logprob": -0.5,
"bytes": None,
"top_logprobs": [],
}
for tid in token_ids
]
}
def _make_usage_chunk(
completion_tokens: int,
prompt_tokens: int = 0,
request_id: str = "test-req",
) -> GenerateStreamResponse:
"""Build a usage only final SSE chunk (empty choices)."""
return GenerateTokensStreamResponse(
request_id=request_id,
choices=[],
usage=UsageInfo(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
),
)
# ---------------------------------------------------------------------------
# Unit tests — no running server
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def tokenizer():
"""Load the tiny tokenizer used across unit tests."""
return get_tokenizer(MODEL_NAME)
@pytest.fixture(scope="module")
def derenderer(tokenizer, request):
"""Construct a minimal OnlineDerenderer backed by a stub renderer."""
renderer = MagicMock()
renderer.get_tokenizer.return_value = tokenizer
executor = ThreadPoolExecutor(max_workers=1)
request.addfinalizer(executor.shutdown)
renderer._executor = executor
model_config = MagicMock()
model_config.hf_config.model_type = "llama"
model_config.model = MODEL_NAME
return OnlineDerenderer(
model_config=model_config,
renderer=renderer,
request_logger=None,
chat_template=None,
chat_template_content_format="string",
trust_request_chat_template=False,
enable_auto_tools=False,
tool_parser=None,
reasoning_parser=None,
)
@pytest.fixture(scope="module")
def parsed_derenderer(tokenizer):
"""OnlineDerenderer with `_FakeParser` wired in as `self.parser`.
The parser-configured path runs `_derender_chat_stream_parsed` via
`make_async(..., executor=renderer._executor)`, so
`renderer._executor` must be a real `ThreadPoolExecutor`.
`loop.run_in_executor` cannot submit work to a `MagicMock`.
"""
renderer = MagicMock()
renderer.get_tokenizer.return_value = tokenizer
renderer._executor = ThreadPoolExecutor(max_workers=2)
model_config = MagicMock()
model_config.hf_config.model_type = "llama"
model_config.model = MODEL_NAME
dr = OnlineDerenderer(
model_config=model_config,
renderer=renderer,
request_logger=None,
chat_template=None,
chat_template_content_format="string",
)
dr.parser = _FakeParser
return dr
def test_non_streaming_derender_initializes_parser_prefix(
derenderer, tokenizer, monkeypatch
):
from unittest.mock import MagicMock
from vllm.entrypoints.generate.base.protocol import FunctionCall
from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
parser = MagicMock()
parser.parse.return_value = (
None,
None,
[FunctionCall(name="tool", arguments="{}")],
)
monkeypatch.setattr(derenderer, "parser", MagicMock(return_value=parser))
generated_ids = tokenizer.encode("answer", add_special_tokens=False)
response = GenerateTokensResponse(
request_id="test",
prompt_token_ids=[11, 12],
choices=[
GenerateTokensChoice(
index=0,
token_ids=generated_ids,
finish_reason="length",
)
],
)
request = ChatCompletionRequest(
model=MODEL_NAME,
messages=[{"role": "user", "content": "question"}],
)
choices = derenderer._derender_chat(response, request)
parser.set_prompt_token_ids.assert_called_once_with([11, 12])
assert choices[0].message.tool_calls[0].function.name == "tool"
assert choices[0].finish_reason == "length"
class TestDetokenizeDelta:
"""_detokenize_delta: chunked decode must equal one shot decode."""
def _one_shot(self, tokenizer, token_ids: list[int]) -> str:
return tokenizer.decode(token_ids, skip_special_tokens=True)
def _chunked(self, derenderer, tokenizer, chunks: list[list[int]]) -> str:
state = DerenderStreamState()
parts: list[str] = []
for delta in chunks:
text, state = derenderer._detokenize_delta(
tokenizer, delta, state, skip_special_tokens=True
)
parts.append(text)
return "".join(parts)
def test_single_chunk(self, derenderer, tokenizer):
"""All tokens in one chunk == one shot decode."""
token_ids = tokenizer.encode("Hello world")[:8]
assert self._chunked(derenderer, tokenizer, [token_ids]) == self._one_shot(
tokenizer, token_ids
)
def test_two_equal_chunks(self, derenderer, tokenizer):
"""Split in half and reassemble == one shot."""
token_ids = tokenizer.encode("Hello world from streaming derender")[:12]
mid = len(token_ids) // 2
chunks = [token_ids[:mid], token_ids[mid:]]
assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
tokenizer, token_ids
)
def test_single_token_per_chunk(self, derenderer, tokenizer):
"""One token per chunk (most granular streaming) == one shot."""
token_ids = tokenizer.encode("incremental detokenization test")[:10]
chunks = [[t] for t in token_ids]
assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
tokenizer, token_ids
)
def test_empty_delta_passthrough(self, derenderer, tokenizer):
"""An empty delta (usage only chunk) emits empty string and preserves state."""
token_ids = tokenizer.encode("Hello")[:4]
_, state = derenderer._detokenize_delta(
tokenizer, token_ids, DerenderStreamState(), skip_special_tokens=True
)
text, new_state = derenderer._detokenize_delta(
tokenizer, [], state, skip_special_tokens=True
)
assert text == ""
assert new_state.prev_tokens == state.prev_tokens
assert new_state.prefix_offset == state.prefix_offset
assert new_state.read_offset == state.read_offset
def test_multibyte_char_split_across_chunks(self, derenderer, tokenizer):
"""A CJK/emoji char straddling chunk boundaries == one shot.
Regression test for held back trailing incomplete UTF-8 byte
sequences being dropped when the rebuild window marks them as
already read (see #46159).
"""
token_ids = tokenizer.encode("Hello ✅ world 日本語")[:16]
chunks = [[t] for t in token_ids]
assert self._chunked(derenderer, tokenizer, chunks) == self._one_shot(
tokenizer, token_ids
)
def test_state_carries_across_calls(self, derenderer, tokenizer):
"""Decode state threads across calls. Text still matches one shot."""
t1 = tokenizer.encode("Hello")[:2]
t2 = tokenizer.encode(" world")[:2]
state = DerenderStreamState()
text1, state = derenderer._detokenize_delta(tokenizer, t1, state)
text2, state = derenderer._detokenize_delta(tokenizer, t2, state)
assert text1 + text2 == self._one_shot(tokenizer, t1 + t2)
# Offsets are rebased to the carried tail each chunk
assert state.prefix_offset == 0
def test_state_window_stays_bounded(self, derenderer, tokenizer):
"""prev_tokens must not grow with the number of chunks (bounded transport).
Guards that the carried decode window is a small constant
tail, so cumulative `stream_state` transport is O(n) and not O(n^2).
"""
token_ids = tokenizer.encode(
"a reasonably long ascii stream of tokens used to exercise the "
"window bound across many single token chunks so the carried "
"state cannot grow linearly with the generation length"
)
assert len(token_ids) > 32
state = DerenderStreamState()
max_window = 0
for tok in token_ids:
_, state = derenderer._detokenize_delta(tokenizer, [tok], state)
max_window = max(max_window, len(state.prev_tokens))
# Bounded by a small constant independent of len(token_ids)
assert max_window <= 32
def test_truncated_multibyte_char_is_held_back(self, derenderer, tokenizer):
"""Output ending mid-character emits no U+FFFD: batch == stream == engine."""
emoji_ids = tokenizer.encode("🌈", add_special_tokens=False)
byte_ids = emoji_ids[-4:]
assert len(byte_ids) == 4
output_ids = byte_ids[:-1]
batch, _ = derenderer._detokenize_delta(
tokenizer, output_ids, DerenderStreamState()
)
streamed = self._chunked(derenderer, tokenizer, [[t] for t in output_ids])
engine = _decode_stream_expected(tokenizer, [], output_ids)
assert "�" not in batch
assert batch == streamed == engine
full, _ = derenderer._detokenize_delta(
tokenizer, byte_ids, DerenderStreamState()
)
assert full.endswith("🌈")
@pytest.mark.parametrize(
("skip", "spaces", "preserve", "fast", "expected"),
[
(True, True, False, True, (True, True)),
(True, False, False, True, (True, True)),
(True, False, False, False, (True, False)),
(False, False, False, True, (False, False)),
(False, False, True, True, (False, False)),
(True, False, True, True, (False, False)),
(True, True, True, True, (False, True)),
],
)
def test_decode_params_mirror_engine(
self, monkeypatch, skip, spaces, preserve, fast, expected
):
"""Flags match the engine's fast or slow detokenizer for the tokenizer."""
monkeypatch.setattr(
online_derenderer_module, "uses_fast_detokenizer", lambda _: fast
)
request = CompletionRequest(
model=MODEL_NAME,
prompt="x",
skip_special_tokens=skip,
spaces_between_special_tokens=spaces,
)
assert _decode_params(None, request, preserve_special=preserve) == expected
def test_decode_params_default_without_request(self, tokenizer):
assert _decode_params(tokenizer, None) == (True, True)
def test_n_independent_streams_same_result(self, derenderer, tokenizer):
"""N parallel streams with the same token sequence give the same text."""
token_ids = tokenizer.encode("parallel streams")[:8]
mid = len(token_ids) // 2
results = []
for _ in range(3):
state = DerenderStreamState()
text, state = derenderer._detokenize_delta(
tokenizer, token_ids[:mid], state
)
text2, _ = derenderer._detokenize_delta(tokenizer, token_ids[mid:], state)
results.append(text + text2)
assert len(set(results)) == 1, "All independent streams must produce same text"
assert results[0] == self._one_shot(tokenizer, token_ids)
# ---------------------------------------------------------------------------
# Prompt seeded detokenization (matches the engine's primed detokenizer)
# ---------------------------------------------------------------------------
def _decode_stream_expected(
tokenizer, prompt_ids: list[int], output_ids: list[int]
) -> str:
"""Engine-equivalent decode via `DecodeStream`, primed by stepping the
prompt (`DecodeStream(ids=...)` needs tokenizers >= 0.22)."""
stream = DecodeStream(skip_special_tokens=True)
for tid in prompt_ids:
stream.step(tokenizer.backend_tokenizer, tid)
return "".join(
s
for tid in output_ids
if (s := stream.step(tokenizer.backend_tokenizer, tid)) is not None
)
@pytest.fixture(scope="module")
def leading_space_ids(tokenizer):
"""(prompt_ids, output_ids, expected) where seeded and unseeded decode
differ by the leading space."""
prompt_ids = tokenizer.encode("[INST] Hi [/INST]", add_special_tokens=False)
output_ids = tokenizer.encode("Hello there, output", add_special_tokens=False)
expected = _decode_stream_expected(tokenizer, prompt_ids, output_ids)
unseeded = tokenizer.decode(output_ids, skip_special_tokens=True)
assert expected.startswith(" ") and not unseeded.startswith(" "), (
"fixture no longer reproduces the leading-space divergence; "
"pick a different prompt/output pair"
)
return prompt_ids, output_ids, expected
class TestSeedStreamState:
"""`_seed_stream_state`: pure helper building the initial decode state."""
def test_seeds_prev_tokens_from_prompt_tail(self, tokenizer, leading_space_ids):
prompt_ids, _, _ = leading_space_ids
state = _seed_stream_state(tokenizer, prompt_ids, skip_special_tokens=True)
assert state.prev_tokens
assert state.read_offset == len(state.prev_tokens)
def test_none_prompt_token_ids_returns_empty_state(self, tokenizer):
assert (
_seed_stream_state(tokenizer, None, skip_special_tokens=True)
== DerenderStreamState()
)
def test_empty_prompt_token_ids_returns_empty_state(self, tokenizer):
assert (
_seed_stream_state(tokenizer, [], skip_special_tokens=True)
== DerenderStreamState()
)
class TestPromptSeededLeadingSpace:
"""Every decode entry point keeps the first token's leading space when
given prompt_token_ids."""
@pytest.mark.asyncio
async def test_chat_batch_seeded_via_explicit_prompt_token_ids(
self, derenderer, leading_space_ids
):
prompt_ids, output_ids, expected = leading_space_ids
choices = await derenderer.derender_chat(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
),
prompt_token_ids=prompt_ids,
)
assert choices[0].message.content == expected
@pytest.mark.asyncio
async def test_chat_batch_seeded_via_generate_response_fallback(
self, derenderer, leading_space_ids
):
prompt_ids, output_ids, expected = leading_space_ids
choices = await derenderer.derender_chat(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
prompt_token_ids=prompt_ids,
),
)
assert choices[0].message.content == expected
@pytest.mark.asyncio
async def test_chat_batch_explicit_takes_precedence_over_fallback(
self, derenderer, tokenizer, leading_space_ids
):
prompt_ids, output_ids, expected = leading_space_ids
wrong_ids = tokenizer.encode("a different prompt", add_special_tokens=False)
choices = await derenderer.derender_chat(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
prompt_token_ids=wrong_ids,
),
prompt_token_ids=prompt_ids,
)
assert choices[0].message.content == expected
@pytest.mark.asyncio
async def test_chat_batch_without_prompt_context_stays_unseeded(
self, derenderer, leading_space_ids
):
_, output_ids, _ = leading_space_ids
choices = await derenderer.derender_chat(
GenerateTokensResponse(
request_id="t",
choices=[
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
),
)
assert not choices[0].message.content.startswith(" ")
@pytest.mark.asyncio
async def test_completion_batch_seeded(self, derenderer, leading_space_ids):
prompt_ids, output_ids, expected = leading_space_ids
choices, _, _ = await derenderer.derender_completion(
[
GenerateTokensResponse(
request_id="t",
choices=[
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
)
],
prompt_token_ids=[prompt_ids],
)
assert choices[0].text == expected
@pytest.mark.asyncio
@pytest.mark.parametrize("from_generate_chunk", [False, True])
async def test_chat_stream_first_chunk_seeded(
self, derenderer, leading_space_ids, from_generate_chunk
):
prompt_ids, output_ids, expected = leading_space_ids
mid = len(output_ids) // 2
chunk1, state = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
output_ids[:mid],
prompt_token_ids=prompt_ids if from_generate_chunk else None,
),
prompt_token_ids=None if from_generate_chunk else prompt_ids,
)
# Only needed on the first chunk; the carried state already holds
# the seeded window from here on.
chunk2, _ = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"),
state=state,
)
streamed = (chunk1.choices[0].delta.content or "") + (
chunk2.choices[0].delta.content or ""
)
assert streamed == expected
@pytest.mark.asyncio
@pytest.mark.parametrize("from_generate_chunk", [False, True])
async def test_completion_stream_first_chunk_seeded(
self, derenderer, leading_space_ids, from_generate_chunk
):
prompt_ids, output_ids, expected = leading_space_ids
mid = len(output_ids) // 2
chunk1, state = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
output_ids[:mid],
prompt_token_ids=prompt_ids if from_generate_chunk else None,
),
prompt_token_ids=None if from_generate_chunk else prompt_ids,
)
chunk2, _ = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(output_ids[mid:], finish_reason="stop"),
state=state,
)
assert chunk1.choices[0].text + chunk2.choices[0].text == expected
@pytest.mark.asyncio
async def test_parser_path_ephemeral_state_seeded(
self, parsed_derenderer, monkeypatch, leading_space_ids
):
"""The parser path's per-call detok window is seeded too, so
`parse_delta` sees the same text as coupled serving."""
class _EchoParser(Parser):
def parse_delta(
self,
delta_text,
delta_token_ids,
request,
prompt_token_ids=None,
*,
finished,
):
return DeltaMessage(content=delta_text) if delta_text else None
monkeypatch.setattr(parsed_derenderer, "parser", _EchoParser)
prompt_ids, output_ids, expected = leading_space_ids
chunk, _ = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(output_ids, finish_reason="stop"),
chat_request=_chat_request(),
prompt_token_ids=prompt_ids,
)
assert chunk.choices[0].delta.content == expected
@pytest.mark.asyncio
@pytest.mark.parametrize(
"stream_method",
["derender_chat_stream", "derender_completion_stream"],
)
async def test_streaming_detokenization_runs_off_event_loop(
derenderer, monkeypatch, stream_method
):
"""Streaming detokenization runs on the renderer executor."""
event_loop_thread_id = threading.get_ident()
executor_thread_id = derenderer.renderer._executor.submit(
threading.get_ident
).result()
detokenize_thread_id = None
original_detokenize = online_derenderer_module.detokenize_incrementally
def record_detokenize_thread(*args, **kwargs):
nonlocal detokenize_thread_id
detokenize_thread_id = threading.get_ident()
return original_detokenize(*args, **kwargs)
monkeypatch.setattr(
online_derenderer_module,
"detokenize_incrementally",
record_detokenize_thread,
)
await getattr(derenderer, stream_method)(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([10]),
)
assert detokenize_thread_id is not None
assert detokenize_thread_id == executor_thread_id
assert detokenize_thread_id != event_loop_thread_id
class TestDerenderCompletionStream:
"""derender_completion_stream: streaming output parity with one shot."""
@pytest.mark.asyncio
async def test_chunked_equals_oneshot(self, derenderer, tokenizer):
"""Sum of streaming text chunks == one shot tokenizer.decode."""
token_ids = tokenizer.encode("streaming completion test")[:10]
mid = len(token_ids) // 2
state = DerenderStreamState()
chunk1, state = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids[:mid]),
state=state,
)
chunk2, _ = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids[mid:], finish_reason="stop"),
state=state,
)
streamed_text = chunk1.choices[0].text + chunk2.choices[0].text
one_shot = tokenizer.decode(token_ids, skip_special_tokens=True)
assert streamed_text == one_shot
@pytest.mark.asyncio
async def test_usage_chunk_passthrough(self, derenderer, tokenizer):
"""Usage only final chunk (empty choices) is passed through correctly."""
usage_chunk = _make_usage_chunk(completion_tokens=10, prompt_tokens=5)
chunk, state = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=usage_chunk,
)
assert chunk.choices == []
assert chunk.usage is not None
assert chunk.usage.completion_tokens == 10
assert chunk.usage.prompt_tokens == 5
@pytest.mark.asyncio
async def test_prompt_tokens_in_usage(self, derenderer, tokenizer):
"""prompt_tokens is correctly forwarded into usage on a usage chunk."""
token_ids = tokenizer.encode("hello")[:3]
usage_chunk = _make_usage_chunk(
completion_tokens=len(token_ids), prompt_tokens=7
)
chunk, _ = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=usage_chunk,
prompt_tokens=7,
)
assert chunk.usage is not None
assert chunk.usage.prompt_tokens == 7
@pytest.mark.asyncio
async def test_none_state_initialises_correctly(self, derenderer, tokenizer):
"""Passing state=None (first call) initialises an empty DerenderStreamState."""
token_ids = tokenizer.encode("hello")[:4]
chunk, state = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids),
state=None,
)
assert isinstance(state, DerenderStreamState)
assert chunk.choices[0].text == tokenizer.decode(
token_ids, skip_special_tokens=True
)
@pytest.mark.asyncio
async def test_skip_special_tokens_threaded(self, derenderer, tokenizer):
"""completion_request.skip_special_tokens is honored (not hardcoded True)."""
eos = tokenizer.eos_token_id
if eos is None:
pytest.skip("tokenizer has no eos token to exercise special stripping")
token_ids = tokenizer.encode("hi")[:2] + [eos]
async def _text(skip: bool) -> str:
req = CompletionRequest(
model=MODEL_NAME, prompt="x", skip_special_tokens=skip
)
chunk, _ = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids),
completion_request=req,
)
return chunk.choices[0].text
# skip=False must retain the special token; skip=True must strip it.
assert await _text(False) != await _text(True)
@pytest.mark.asyncio
async def test_finish_reason_forwarded(self, derenderer, tokenizer):
"""finish_reason from the generate chunk reaches the derendered choice."""
token_ids = tokenizer.encode("done")[:2]
chunk, _ = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids, finish_reason="length"),
)
assert chunk.choices[0].finish_reason == "length"
class TestStreamLogprobs:
"""Streaming derender must carry per-chunk logprobs with placeholders
resolved, matching what the generate streaming path emits."""
@pytest.mark.asyncio
async def test_chat_stream_logprobs_resolved_per_chunk(self, derenderer, tokenizer):
"""Each streamed chunk carries logprobs with token_id:N resolved."""
token_ids = tokenizer.encode("hello world")[:6]
mid = len(token_ids) // 2
state = None
for part in (token_ids[:mid], token_ids[mid:]):
chunk, state = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
part, logprobs=_placeholder_logprobs(part)
),
state=state,
)
logprobs = chunk.choices[0].logprobs
assert logprobs is not None and logprobs.content is not None
assert len(logprobs.content) == len(part)
for entry in logprobs.content:
assert not entry.token.startswith("token_id:"), (
f"placeholder not resolved: {entry.token!r}"
)
@pytest.mark.asyncio
async def test_chat_stream_logprobs_multibyte_across_chunks(
self, derenderer, tokenizer
):
"""Byte-fallback correction works when a multi-byte character's
tokens are split across chunks (context carried in stream_state)."""
token_ids = tokenizer.encode("👍", add_special_tokens=False)
if len(token_ids) < 2 or "�" not in tokenizer.decode([token_ids[-1]]):
pytest.skip("Tokenizer does not byte-split this character")
state = None
last_chunk = None
# One token per chunk: every entry after the first needs cross-chunk
# context to resolve without U+FFFD.
for i, tid in enumerate(token_ids):
last_chunk, state = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[tid], logprobs=_placeholder_logprobs([tid])
),
state=state,
)
assert last_chunk is not None
final_entry = last_chunk.choices[0].logprobs.content[0]
assert not final_entry.token.endswith("�"), (
"byte-fallback correction failed across chunk boundary: "
f"{final_entry.token!r}"
)
@pytest.mark.asyncio
async def test_completion_stream_logprobs_text_offset_absolute(
self, derenderer, tokenizer
):
"""text_offset continues across chunks instead of restarting at 0."""
token_ids = tokenizer.encode("hello world")[:6]
mid = len(token_ids) // 2
chunk1, state = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
token_ids[:mid], logprobs=_placeholder_logprobs(token_ids[:mid])
),
)
chunk2, _ = await derenderer.derender_completion_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
token_ids[mid:], logprobs=_placeholder_logprobs(token_ids[mid:])
),
state=state,
)
assert chunk1.choices[0].logprobs.text_offset[0] == 0
assert chunk2.choices[0].logprobs.text_offset[0] == len(chunk1.choices[0].text)
@pytest.mark.asyncio
async def test_stream_logprob_state_stays_bounded(self, derenderer, tokenizer):
"""Carried logprob context never exceeds the 4-token window."""
token_ids = tokenizer.encode("the quick brown fox jumps over the lazy dog")
state = None
for tid in token_ids:
_, state = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[tid], logprobs=_placeholder_logprobs([tid])
),
state=state,
)
assert state is not None
assert len(state.logprob_context_token_ids) <= 4
@pytest.mark.asyncio
async def test_stream_without_logprobs_unchanged(self, derenderer, tokenizer):
"""Chunks without logprobs keep logprobs=None on the output choice."""
token_ids = tokenizer.encode("hello")[:3]
chunk, _ = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids),
)
assert chunk.choices[0].logprobs is None
class TestDerenderChatStream:
"""derender_chat_stream: plain detok branch (no parser)."""
@pytest.mark.asyncio
async def test_role_on_first_chunk_only(self, derenderer, tokenizer):
"""role='assistant' appears in the first chunk, not subsequent ones."""
token_ids = tokenizer.encode("hello world")[:6]
mid = len(token_ids) // 2
state = DerenderStreamState()
chunk1, state = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids[:mid]),
state=state,
)
chunk2, _ = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids[mid:], finish_reason="stop"),
state=state,
)
assert chunk1.choices[0].delta.role == "assistant"
assert chunk2.choices[0].delta.role is None
@pytest.mark.asyncio
async def test_metrics_passthrough(self, derenderer):
metrics = PerRequestMetrics(
speculative_decoding=SpeculativeDecodingMetrics(
mean_acceptance_length=2.0,
draft_acceptance_rate=0.5,
acceptance_histogram=[0, 1],
num_spec_steps=1,
num_accepted_draft_tokens=1,
num_draft_tokens=2,
num_spec_tokens=1,
)
)
chunk, _ = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([], metrics=metrics),
state=DerenderStreamState(),
)
assert chunk.metrics == metrics
@pytest.mark.asyncio
async def test_chunked_equals_oneshot(self, derenderer, tokenizer):
"""Sum of streaming content deltas == one shot decode."""
token_ids = tokenizer.encode("streaming chat derender text")[:10]
mid = len(token_ids) // 2
state = DerenderStreamState()
chunk1, state = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids[:mid]),
state=state,
)
chunk2, _ = await derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(token_ids[mid:]),
state=state,
)
streamed = (chunk1.choices[0].delta.content or "") + (
chunk2.choices[0].delta.content or ""
)
one_shot = tokenizer.decode(token_ids, skip_special_tokens=True)
assert streamed == one_shot
@pytest.mark.asyncio
async def test_parser_configured_missing_chat_request_raises(
self, parsed_derenderer
):
"""A parser configured model must never fall through to plain detok
even when `chat_request` is omitted. If allowed this would leak raw
reasoning/tool markup into `delta.content`. `ServingDerender`
pre-checks this too (400 before touching the tokenizer). This pins
the `OnlineDerenderer` level backstop."""
with pytest.raises(ValueError, match="chat_request"):
await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
state=None,
chat_request=None,
)
def _chat_request(**kwargs):
kwargs.setdefault("messages", [{"role": "user", "content": "hi"}])
kwargs.setdefault("model", MODEL_NAME)
return ChatCompletionRequest(**kwargs)
class TestDerenderChatStreamParsed:
"""derender_chat_stream: parser branch (replay + parse_delta) exercised
against the deterministic `_FakeParser` so these pin OnlineDerenderer's
own replay/merge/pin/finish_reason logic independent of any real
parser's markup grammar (covered separately by the parser_server backed
integration tests below)."""
@pytest.mark.asyncio
async def test_dispatches_and_emits_content(self, parsed_derenderer):
chunk, state = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[_FakeParser.CONTENT], finish_reason="stop"
),
chat_request=_chat_request(),
)
assert chunk.choices[0].delta.content == "c"
assert chunk.choices[0].delta.role == "assistant"
assert chunk.choices[0].finish_reason == "stop"
assert state.output_token_ids == [_FakeParser.CONTENT]
@pytest.mark.asyncio
async def test_role_sent_once(self, parsed_derenderer):
chat_request = _chat_request()
chunk1, state = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
chat_request=chat_request,
)
chunk2, _ = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[_FakeParser.CONTENT], finish_reason="stop"
),
state=state,
chat_request=chat_request,
)
assert chunk1.choices[0].delta.role == "assistant"
assert chunk2.choices[0].delta.role is None
@pytest.mark.asyncio
async def test_finish_only_chunk_flushes(self, parsed_derenderer):
"""A finish only chunk (no new tokens) still calls `parse_delta`
once with `finished=True` to flush buffered state."""
chat_request = _chat_request()
_, state = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([_FakeParser.CONTENT]),
chat_request=chat_request,
)
chunk2, _ = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([], finish_reason="stop"),
state=state,
chat_request=chat_request,
)
assert chunk2.choices[0].delta.content == "FLUSH"
assert chunk2.choices[0].finish_reason == "stop"
@pytest.mark.asyncio
async def test_chunking_invariance(self, parsed_derenderer):
"""1 token per chunk vs a single whole chunk assemble identically.
Every chunk (live or replayed) is fed to `parse_delta` at its own
producer granularity (see `_derender_chat_stream_parsed`'s
docstring). This proves that choice doesn't change the assembled
output for a well behaved incremental parser. Each call's effects
fold together the same way regardless of how many tokens land in
one call vs. many."""
token_ids = [
_FakeParser.REASON,
_FakeParser.REASON,
_FakeParser.TOOL_START,
_FakeParser.TOOL_ARG,
_FakeParser.TOOL_ARG,
_FakeParser.CONTENT,
]
chat_request = _chat_request(
tools=[{"type": "function", "function": {"name": "get_weather"}}],
tool_choice="auto",
include_reasoning=True,
)
async def _assemble(chunks: list[list[int]]) -> dict:
state = None
content = ""
reasoning = ""
tool_args = ""
for i, tids in enumerate(chunks):
finish = "stop" if i == len(chunks) - 1 else None
chunk, state = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(tids, finish_reason=finish),
state=state,
chat_request=chat_request,
)
delta = chunk.choices[0].delta
content += delta.content or ""
reasoning += delta.reasoning or ""
for tc in delta.tool_calls:
if tc.function and tc.function.arguments:
tool_args += tc.function.arguments
return {"content": content, "reasoning": reasoning, "tool_args": tool_args}
whole = await _assemble([token_ids])
one_at_a_time = await _assemble([[t] for t in token_ids])
assert whole == one_at_a_time
assert whole == {"content": "c", "reasoning": "rr", "tool_args": "aa"}
@pytest.mark.asyncio
async def test_replay_preserves_chunk_boundaries(
self, parsed_derenderer, monkeypatch
):
"""A multi-token chunk (e.g. a speculative decoding step) must be
replayed as one `parse_delta` call, not one per token. Parsers
with per-call state such as `history_tool_call_cnt` would
otherwise diverge from standard serving."""
class _CallCountingParser(_FakeParser):
"""Emits how many `parse_delta` calls it has seen."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.num_calls = 0
def parse_delta(
self,
delta_text,
delta_token_ids,
request,
prompt_token_ids=None,
*,
finished,
):
self.num_calls += 1
return DeltaMessage(content=str(self.num_calls))
monkeypatch.setattr(parsed_derenderer, "parser", _CallCountingParser)
chat_request = _chat_request()
_, state = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([_FakeParser.CONTENT] * 3),
chat_request=chat_request,
)
assert state.output_chunk_lens == [3]
chunk, _ = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[_FakeParser.CONTENT], finish_reason="stop"
),
state=state,
chat_request=chat_request,
)
assert chunk.choices[0].delta.content == "2"
@pytest.mark.asyncio
async def test_tool_call_id_pinned_across_chunks(self, parsed_derenderer):
"""Once an index's ID is recorded in `last_tool_call_ids`, a later
id bearing delta for that same index is pinned to the recorded
value rather than a freshly (re-)generated one."""
chat_request = _chat_request(
tools=[{"type": "function", "function": {"name": "get_weather"}}],
tool_choice="auto",
)
chunk1, state = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([_FakeParser.TOOL_START]),
chat_request=chat_request,
)
first_id = chunk1.choices[0].delta.tool_calls[0].id
assert first_id is not None
assert state.last_tool_call_ids == [first_id]
chunk2, state = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[_FakeParser.TOOL_START], finish_reason="stop"
),
state=state,
chat_request=chat_request,
)
assert chunk2.choices[0].delta.tool_calls[0].id == first_id
assert state.last_tool_call_ids == [first_id]
@pytest.mark.asyncio
async def test_finish_reason_rewritten_to_tool_calls(self, parsed_derenderer):
chat_request = _chat_request(
tools=[{"type": "function", "function": {"name": "get_weather"}}],
tool_choice="auto",
)
chunk, _ = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[_FakeParser.TOOL_START, _FakeParser.TOOL_ARG], finish_reason="stop"
),
chat_request=chat_request,
)
assert chunk.choices[0].finish_reason == "tool_calls"
@pytest.mark.asyncio
async def test_finish_reason_stop_for_named_tool_choice(self, parsed_derenderer):
# ChatCompletionRequest's `check_tool_usage` is a mode="before"
# validator, so it sees the raw value and only accepts "auto",
# "required" or a dict, never an already built
# ChatCompletionNamedToolChoiceParam.
chat_request = _chat_request(
tools=[{"type": "function", "function": {"name": "get_weather"}}],
tool_choice={"type": "function", "function": {"name": "get_weather"}},
)
chunk, _ = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[_FakeParser.TOOL_START, _FakeParser.TOOL_ARG], finish_reason="stop"
),
chat_request=chat_request,
)
assert chunk.choices[0].finish_reason == "stop"
@pytest.mark.asyncio
async def test_include_reasoning_false_suppresses_reasoning(
self, parsed_derenderer
):
chat_request = _chat_request(include_reasoning=False)
chunk, _ = await parsed_derenderer.derender_chat_stream(
model=MODEL_NAME,
generate_chunk=_make_stream_chunk(
[_FakeParser.REASON, _FakeParser.CONTENT], finish_reason="stop"
),
chat_request=chat_request,
)
delta = chunk.choices[0].delta
assert delta.reasoning is None
assert delta.content == "c"
# ---------------------------------------------------------------------------
# Harmony / GPT-OSS replay — unit, no server
# ---------------------------------------------------------------------------
HARMONY_MODEL = "openai/gpt-oss-20b"
HARMONY_REASONING = "The user wants 2 plus 3."
HARMONY_ANSWER = "The answer is 5."
HARMONY_PROMPT = "<|start|>user<|message|>Add 2 and 3.<|end|><|start|>assistant"
HARMONY_OUTPUT = (
f"<|channel|>analysis<|message|>{HARMONY_REASONING}<|end|>"
f"<|start|>assistant<|channel|>final<|message|>{HARMONY_ANSWER}<|return|>"
)
@pytest.fixture(scope="module")
def harmony_encode():
"""Encoder for canned GPT-OSS harmony token sequences."""
pytest.importorskip("openai_harmony")
# Pre-caches the o200k_base BPE file that openai-harmony's Rust backend
# downloads on first use, same as the sibling suite's E2E harmony tests.
from tests.entrypoints.scale_out.derender.test_derender import (
_ensure_harmony_vocab,
)
from vllm.entrypoints.openai.parser.harmony_utils import get_encoding
_ensure_harmony_vocab()
encoding = get_encoding()
def _encode(harmony_str: str) -> list[int]:
return encoding.encode(harmony_str, allowed_special="all")
return _encode
@pytest.fixture(scope="module")
def harmony_tokenizer():
pytest.importorskip("openai_harmony")
return get_tokenizer(HARMONY_MODEL, trust_remote_code=True)
@pytest.fixture(scope="module")
def harmony_derenderer(harmony_tokenizer):
"""OnlineDerenderer whose parser resolves to the real `HarmonyParser`.
Same shape as `parsed_derenderer` (mocked renderer, real executor) but
with a real tokenizer and parser, so the replay path is exercised
against Harmony's actual channel grammar rather than a stub.
"""
from vllm.parser.harmony import HarmonyParser
renderer = MagicMock()
renderer.get_tokenizer.return_value = harmony_tokenizer
renderer._executor = ThreadPoolExecutor(max_workers=2)
model_config = MagicMock()
model_config.model = HARMONY_MODEL
model_config.hf_config.model_type = "gpt_oss"
model_config.hf_text_config.model_type = "gpt_oss"
model_config.hf_overrides = None
dr = OnlineDerenderer(
model_config=model_config,
renderer=renderer,
request_logger=None,
chat_template=None,
chat_template_content_format="string",
enable_auto_tools=True,
tool_parser="openai",
reasoning_parser="openai_gptoss",
)
assert dr.use_harmony
assert dr.parser is HarmonyParser
return dr
async def _stream_harmony_deltas(
derenderer,
chat_request,
output_ids: list[int],
prompt_ids: list[int],
chunk_size: int,
) -> list[DeltaMessage]:
"""Drive `output_ids` through the parser path in fixed size chunks.
Returns the per chunk `DeltaMessage`s in order so callers can assert on
intermediate emissions, not just the assembled result.
"""
deltas: list[DeltaMessage] = []
state = None
for start in range(0, len(output_ids), chunk_size):
tids = output_ids[start : start + chunk_size]
is_last = start + chunk_size >= len(output_ids)
chunk, state = await derenderer.derender_chat_stream(
model=HARMONY_MODEL,
generate_chunk=_make_stream_chunk(
tids, finish_reason="stop" if is_last else None
),
state=state,
chat_request=chat_request,
prompt_token_ids=prompt_ids,
)
deltas.append(chunk.choices[0].delta)
return deltas
class TestDerenderChatStreamHarmony:
"""derender_chat_stream: replay + `parse_delta` against real HarmonyParser.
Skipped where `openai_harmony` is not installed. `prompt_token_ids` is
threaded through for call shape fidelity only, since HarmonyParser
derives its state from the output tokens alone.
"""
@pytest.mark.asyncio
async def test_reasoning_never_leaks_as_content_midstream(
self, harmony_derenderer, harmony_encode
):
"""Regression guard for the RFC's original replay + diff design.
`HarmonyParser.parse()` always flushes to EOS. Mid-stream that
raises `HarmonyError` and the recovery branch re-emits the in
flight message on the `final` channel, so partial analysis surfaces
as content. Replay + `parse_delta` must never do that. Driven one
token per chunk, the finest granularity a client can produce.
"""
output_ids = harmony_encode(HARMONY_OUTPUT)
prompt_ids = harmony_encode(HARMONY_PROMPT)
chat_request = _chat_request(model=HARMONY_MODEL, include_reasoning=True)
deltas = await _stream_harmony_deltas(
harmony_derenderer, chat_request, output_ids, prompt_ids, chunk_size=1
)
reasoning = ""
content = ""
for i, delta in enumerate(deltas):
reasoning += delta.reasoning or ""
content += delta.content or ""
# Any analysis text emitted as content breaks the prefix
# property, since the two channels share no prefix here.
assert HARMONY_ANSWER.startswith(content), (
f"content after chunk {i} is not a prefix of the final "
f"channel text: {content!r}"
)
assert HARMONY_REASONING.startswith(reasoning)
assert not delta.tool_calls
assert reasoning == HARMONY_REASONING
assert content == HARMONY_ANSWER
@pytest.mark.asyncio
async def test_stream_matches_batch(self, harmony_derenderer, harmony_encode):
"""Streamed assembly equals one shot `/derender` over the same IDs."""
output_ids = harmony_encode(HARMONY_OUTPUT)
prompt_ids = harmony_encode(HARMONY_PROMPT)
chat_request = _chat_request(model=HARMONY_MODEL, include_reasoning=True)
deltas = await _stream_harmony_deltas(
harmony_derenderer, chat_request, output_ids, prompt_ids, chunk_size=3
)
reasoning = "".join(d.reasoning or "" for d in deltas)
content = "".join(d.content or "" for d in deltas)
batch_choices = await harmony_derenderer.derender_chat(
GenerateTokensResponse(
request_id="test-harmony-batch",
choices=[
GenerateTokensChoice(
index=0, token_ids=output_ids, finish_reason="stop"
)
],
),
chat_request,
)
message = batch_choices[0].message
assert reasoning == message.reasoning
assert content == message.content
class TestDerenderStreamStateValidation:
"""DerenderStreamState rejects malformed caller supplied offsets/lengths."""
def test_negative_prefix_offset_rejected(self):
with pytest.raises(ValidationError):
DerenderStreamState(prefix_offset=-1)
def test_negative_read_offset_rejected(self):
with pytest.raises(ValidationError):
DerenderStreamState(read_offset=-1)
def test_prev_tokens_over_cap_rejected(self):
with pytest.raises(ValidationError):
DerenderStreamState(prev_tokens=["a"] * 1025)
def test_prev_tokens_at_cap_accepted(self):
state = DerenderStreamState(prev_tokens=["a"] * 1024)
assert len(state.prev_tokens) == 1024
def test_output_chunk_lens_mismatch_rejected(self):
with pytest.raises(ValidationError):
DerenderStreamState(output_token_ids=[1, 2, 3], output_chunk_lens=[1, 1])
def test_output_chunk_lens_zero_entry_rejected(self):
with pytest.raises(ValidationError):
DerenderStreamState(output_token_ids=[1, 2], output_chunk_lens=[0, 2])
def test_output_chunk_lens_matching_accepted(self):
state = DerenderStreamState(
output_token_ids=[1, 2, 3], output_chunk_lens=[1, 2]
)
assert state.output_chunk_lens == [1, 2]
_GEN = {"request_id": "t", "choices": []}
_GEN_BAD = {**_GEN, "prompt_token_ids": [-1]}
@pytest.mark.parametrize(
"model_cls,payload",
[
(DerenderChatRequest, {"generate_response": _GEN, "prompt_token_ids": [-1]}),
(DerenderChatRequest, {"generate_response": _GEN_BAD}),
(
DerenderCompletionRequest,
{"generate_responses": [_GEN], "prompt_token_ids": [[-1]]},
),
(
DerenderChatStreamRequest,
{"stream": True, "generate_chunk": _GEN, "prompt_token_ids": [-1]},
),
(
DerenderCompletionStreamRequest,
{"stream": True, "generate_chunk": _GEN_BAD},
),
],
)
def test_negative_prompt_token_ids_rejected(model_cls, payload):
"""Negative prompt ids would reach convert_ids_to_tokens, so they get a
400 at parse time like `GenerateRequest.token_ids` does."""
with pytest.raises(ValidationError, match="prompt_token_ids"):
model_cls.model_validate(payload)
class TestServingDerenderStreamErrorHandling:
"""Malformed stream_state must surface as 400 and not an unhandled 500."""
def _make_serving(self, side_effect: Exception):
models = MagicMock()
models.is_base_model.return_value = True
models.model_config = MagicMock()
models.model_config.max_model_len = 100_000
online_derenderer = MagicMock()
online_derenderer.parser = None
online_derenderer.derender_completion_stream = AsyncMock(
side_effect=side_effect
)
online_derenderer.derender_chat_stream = AsyncMock(side_effect=side_effect)
return ServingDerender(models=models, online_derenderer=online_derenderer)
@pytest.mark.asyncio
@pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
async def test_completion_stream_bad_state_returns_400(self, exc):
serving = self._make_serving(exc)
request = DerenderCompletionStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1, 2]),
stream_state=DerenderStreamState(),
)
result = await serving.derender_completion_stream_response(request)
assert isinstance(result, ErrorResponse)
assert result.error.code == 400
@pytest.mark.asyncio
@pytest.mark.parametrize("exc", [KeyError("bad byte"), IndexError("oob")])
async def test_chat_stream_bad_state_returns_400(self, exc):
serving = self._make_serving(exc)
request = DerenderChatStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1, 2]),
stream_state=DerenderStreamState(),
)
result = await serving.derender_chat_stream_response(request)
assert isinstance(result, ErrorResponse)
assert result.error.code == 400
class TestServingDerenderStreamValidation:
"""If checks cannot run in `derender_chat_stream_response`: all must
reject with 400 before ever calling into `online_derenderer` (i.e.
before touching the tokenizer)."""
def _make_serving(self, *, parser_configured: bool, max_model_len: int = 100_000):
models = MagicMock()
models.is_base_model.return_value = True
models.model_config = MagicMock()
models.model_config.max_model_len = max_model_len
online_derenderer = MagicMock()
online_derenderer.parser = MagicMock() if parser_configured else None
online_derenderer.derender_chat_stream = AsyncMock(
return_value=(
ChatCompletionStreamResponse(id="t", model=MODEL_NAME, choices=[]),
DerenderStreamState(),
)
)
return ServingDerender(models=models, online_derenderer=online_derenderer)
@pytest.mark.asyncio
async def test_missing_chat_request_with_parser_rejected(self):
serving = self._make_serving(parser_configured=True)
request = DerenderChatStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1, 2]),
)
result = await serving.derender_chat_stream_response(request)
assert isinstance(result, ErrorResponse)
assert result.error.code == 400
assert "chat_request" in result.error.message
serving.online_derenderer.derender_chat_stream.assert_not_called()
@pytest.mark.asyncio
async def test_missing_chat_request_without_parser_ok(self):
serving = self._make_serving(parser_configured=False)
request = DerenderChatStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1, 2]),
)
result = await serving.derender_chat_stream_response(request)
assert not isinstance(result, ErrorResponse)
@pytest.mark.asyncio
async def test_missing_prompt_token_ids_with_parser_rejected(self):
"""A parser configured model must reject a missing prompt_token_ids
the same way it rejects a missing chat_request. Without it,
parse_delta cannot tell whether the prompt left reasoning open and
would misclassify reasoning content as plain content."""
serving = self._make_serving(parser_configured=True)
request = DerenderChatStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1, 2]),
chat_request=_chat_request(),
)
result = await serving.derender_chat_stream_response(request)
assert isinstance(result, ErrorResponse)
assert result.error.code == 400
assert "prompt_token_ids" in result.error.message
serving.online_derenderer.derender_chat_stream.assert_not_called()
@pytest.mark.asyncio
async def test_prompt_token_ids_present_with_parser_ok(self):
serving = self._make_serving(parser_configured=True)
request = DerenderChatStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1, 2]),
chat_request=_chat_request(),
prompt_token_ids=[1, 2, 3],
)
result = await serving.derender_chat_stream_response(request)
assert not isinstance(result, ErrorResponse)
@pytest.mark.asyncio
async def test_oversized_output_token_ids_rejected(self):
serving = self._make_serving(parser_configured=False, max_model_len=4)
request = DerenderChatStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1, 2, 3]),
stream_state=DerenderStreamState(
output_token_ids=[1, 2], output_chunk_lens=[2]
),
)
result = await serving.derender_chat_stream_response(request)
assert isinstance(result, ErrorResponse)
assert result.error.code == 400
assert "max_model_len" in result.error.message
serving.online_derenderer.derender_chat_stream.assert_not_called()
@pytest.mark.asyncio
async def test_oversized_prompt_token_ids_rejected(self):
"""prompt_token_ids is caller supplied and otherwise unbounded. A
parser configured deployment rescans it in full on every chunk
(is_reasoning_end / adjust_initial_state_from_prompt), so it must be
bounded the same way output_token_ids is."""
serving = self._make_serving(parser_configured=True, max_model_len=4)
request = DerenderChatStreamRequest(
stream=True,
model=MODEL_NAME,
generate_chunk=_make_stream_chunk([1]),
chat_request=_chat_request(),
prompt_token_ids=[1, 2, 3, 4, 5],
)
result = await serving.derender_chat_stream_response(request)
assert isinstance(result, ErrorResponse)
assert result.error.code == 400
assert "max_model_len" in result.error.message
serving.online_derenderer.derender_chat_stream.assert_not_called()
@pytest.mark.asyncio
async def test_too_many_choices_rejected(self):
"""Each streamed chunk contains at most one choice (a single
DerenderStreamState is threaded through every choice). The check
could never fire since derender_chat_stream itself rejects anything
above 1 first."""
serving = self._make_serving(parser_configured=False)
two_choices = GenerateTokensStreamResponse(
request_id="t",
choices=[
GenerateTokensStreamChoice(index=i, token_ids=[1]) for i in range(2)
],
)
request = DerenderChatStreamRequest(
stream=True, model=MODEL_NAME, generate_chunk=two_choices
)
result = await serving.derender_chat_stream_response(request)
assert isinstance(result, ErrorResponse)
assert result.error.code == 400
assert "at most one choice" in result.error.message
serving.online_derenderer.derender_chat_stream.assert_not_called()
# ---------------------------------------------------------------------------
# Integration tests — require a live render server
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def server():
with RemoteLaunchRenderServer(MODEL_NAME, []) as remote_server:
yield remote_server
@pytest_asyncio.fixture
async def client(server):
async with httpx.AsyncClient(
base_url=server.url_for(""), timeout=30.0
) as http_client:
yield http_client
async def _render_chat(client) -> dict:
"""Render a minimal chat request and return the GenerateRequest dict."""
resp = await client.post(
"/v1/chat/completions/render",
json={
"model": MODEL_NAME,
"messages": [{"role": "user", "content": "Hello"}],
},
)
assert resp.status_code == 200
return resp.json()
@pytest.mark.asyncio
async def test_streaming_completion_derender_roundtrip(client):
"""Streaming completions derender: chunked text == non streaming text."""
gen_req = await _render_chat(client)
token_ids: list[int] = gen_req["token_ids"][:8]
mid = len(token_ids) // 2
chunk1_ids, chunk2_ids = token_ids[:mid], token_ids[mid:]
# Non streaming baseline.
non_stream_resp = await client.post(
"/v1/completions/derender",
json={
"model": MODEL_NAME,
"generate_responses": [
{
"request_id": "test-ns",
"choices": [
{
"index": 0,
"token_ids": token_ids,
"finish_reason": "stop",
}
],
}
],
},
)
assert non_stream_resp.status_code == 200
expected_text = non_stream_resp.json()["choices"][0]["text"]
# Streaming call 1.
r1 = await client.post(
"/v1/completions/derender",
json={
"stream": True,
"model": MODEL_NAME,
"generate_chunk": {
"request_id": "test-s",
"choices": [
{"index": 0, "token_ids": chunk1_ids, "finish_reason": None}
],
},
"stream_state": None,
},
)
assert r1.status_code == 200
d1 = r1.json()
text1 = d1["chunk"]["choices"][0]["text"]
state1 = d1["stream_state"]
# Streaming call 2 (final chunk).
r2 = await client.post(
"/v1/completions/derender",
json={
"stream": True,
"model": MODEL_NAME,
"generate_chunk": {
"request_id": "test-s",
"choices": [
{"index": 0, "token_ids": chunk2_ids, "finish_reason": "stop"}
],
},
"stream_state": state1,
},
)
assert r2.status_code == 200
text2 = r2.json()["chunk"]["choices"][0]["text"]
assert text1 + text2 == expected_text
@pytest.mark.asyncio
async def test_streaming_chat_derender_roundtrip(client):
"""Streaming chat derender (plain detok): chunked text == non streaming text."""
gen_req = await _render_chat(client)
token_ids: list[int] = gen_req["token_ids"][:8]
mid = len(token_ids) // 2
chunk1_ids, chunk2_ids = token_ids[:mid], token_ids[mid:]
# Non streaming baseline.
ns = await client.post(
"/v1/chat/completions/derender",
json={
"model": MODEL_NAME,
"generate_response": {
"request_id": "test-ns",
"choices": [
{
"index": 0,
"token_ids": token_ids,
"finish_reason": "stop",
}
],
},
},
)
assert ns.status_code == 200
expected_content = ns.json()["choices"][0]["message"]["content"]
# Streaming call 1.
r1 = await client.post(
"/v1/chat/completions/derender",
json={
"stream": True,
"model": MODEL_NAME,
"generate_chunk": {
"request_id": "test-s",
"choices": [
{"index": 0, "token_ids": chunk1_ids, "finish_reason": None}
],
},
"stream_state": None,
},
)
assert r1.status_code == 200
d1 = r1.json()
text1 = d1["chunk"]["choices"][0]["delta"].get("content") or ""
state1 = d1["stream_state"]
# role=assistant on the first chunk
assert d1["chunk"]["choices"][0]["delta"].get("role") == "assistant"
# Streaming call 2.
r2 = await client.post(
"/v1/chat/completions/derender",
json={
"stream": True,
"model": MODEL_NAME,
"generate_chunk": {
"request_id": "test-s",
"choices": [
{"index": 0, "token_ids": chunk2_ids, "finish_reason": "stop"}
],
},
"stream_state": state1,
},
)
assert r2.status_code == 200
d2 = r2.json()
text2 = d2["chunk"]["choices"][0]["delta"].get("content") or ""
# role must NOT be repeated on subsequent chunks
assert d2["chunk"]["choices"][0]["delta"].get("role") is None
assert text1 + text2 == expected_content
@pytest.mark.asyncio
async def test_streaming_derender_invalid_body_returns_400(client):
"""Missing required field in streaming request returns 400."""
r = await client.post(
"/v1/completions/derender",
json={
"stream": True,
# missing required 'model' and 'generate_chunk'
},
)
assert r.status_code == 400
@pytest.mark.asyncio
async def test_streaming_derender_non_object_body_returns_400(client):
"""A non object JSON body (e.g. a list) returns 400, not a 500."""
r = await client.post(
"/v1/completions/derender",
json=[1, 2, 3],
)
assert r.status_code == 400
@pytest.mark.asyncio
async def test_streaming_usage_chunk(client):
"""Usage only final chunk is forwarded with correct token counts."""
gen_req = await _render_chat(client)
token_ids: list[int] = gen_req["token_ids"][:6]
state: dict = {}
# Send content chunk first.
r1 = await client.post(
"/v1/completions/derender",
json={
"stream": True,
"model": MODEL_NAME,
"generate_chunk": {
"request_id": "usage-test",
"choices": [
{"index": 0, "token_ids": token_ids, "finish_reason": "stop"}
],
},
"stream_state": None,
},
)
assert r1.status_code == 200
state = r1.json()["stream_state"]
# Send usage only final chunk.
r2 = await client.post(
"/v1/completions/derender",
json={
"stream": True,
"model": MODEL_NAME,
"generate_chunk": {
"request_id": "usage-test",
"choices": [],
"usage": {
"prompt_tokens": 10,
"completion_tokens": len(token_ids),
"total_tokens": 10 + len(token_ids),
},
},
"stream_state": state,
"prompt_tokens": 10,
},
)
assert r2.status_code == 200
d2 = r2.json()
assert d2["chunk"]["choices"] == []
assert d2["chunk"]["usage"]["prompt_tokens"] == 10
assert d2["chunk"]["usage"]["completion_tokens"] == len(token_ids)
# ---------------------------------------------------------------------------
# Integration tests for parser configured (reasoning + tool calls) require a
# live render server. Mirrors the parser_server / parser_tokenizer pattern
# from test_derender.py.
# ---------------------------------------------------------------------------
PARSER_MODEL = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"
_PARSER_TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
},
},
}
]
@pytest.fixture(scope="module")
def parser_server():
args = [
"--enable-auto-tool-choice",
"--tool-call-parser",
"hermes",
"--reasoning-parser",
"deepseek_r1",
]
with RemoteLaunchRenderServer(PARSER_MODEL, args) as remote_server:
yield remote_server
@pytest_asyncio.fixture
async def parser_client(parser_server):
async with httpx.AsyncClient(
base_url=parser_server.url_for(""), timeout=60.0
) as http_client:
yield http_client
@pytest.fixture(scope="module")
def parser_tokenizer():
return get_tokenizer(PARSER_MODEL)
def _require_parser_markers(tokenizer, text: str, *markers: str) -> list[int]:
"""Encode text and skip the test if any marker is lost in roundtrip."""
ids = tokenizer.encode(text, add_special_tokens=False)
decoded = tokenizer.decode(ids, skip_special_tokens=False)
for m in markers:
if m not in decoded:
pytest.skip(f"Marker {m!r} lost in encode->decode roundtrip")
return ids
async def _render_parser_chat(client, messages: list[dict]) -> dict:
resp = await client.post(
"/v1/chat/completions/render",
json={"model": PARSER_MODEL, "messages": messages},
)
assert resp.status_code == 200, resp.text
return resp.json()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"chunking",
[
"whole",
"per_token",
pytest.param(
"triples",
marks=pytest.mark.xfail(
reason=" leaks into reasoning when it shares a "
"multi-token delta with reasoning text (vllm#55195)"
),
),
],
)
async def test_stream_parsed_matches_batch_reasoning(
parser_client, parser_tokenizer, chunking
):
"""Streamed == batch: assembled reasoning/content equal the non
streaming `/derender` result over the same token IDs, for every
chunking of the same output (chunking invariance)."""
messages = [{"role": "user", "content": "What is 2+3?"}]
gen_req = await _render_parser_chat(parser_client, messages)
reasoning_text = "The user wants 2 plus 3. That is 5."
answer_text = "The answer is 5."
output_text = f"{reasoning_text}{answer_text}"
output_ids = _require_parser_markers(parser_tokenizer, output_text, "")
chat_request = {
"model": PARSER_MODEL,
"messages": messages,
"include_reasoning": True,
}
batch_resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": {
"request_id": "batch",
"choices": [
{"index": 0, "token_ids": output_ids, "finish_reason": "stop"}
],
},
"prompt_tokens": len(gen_req["token_ids"]),
"chat_request": chat_request,
},
)
assert batch_resp.status_code == 200, batch_resp.text
batch_msg = batch_resp.json()["choices"][0]["message"]
n = len(output_ids)
chunk_sizes = {
"whole": [n],
"per_token": [1] * n,
"triples": [3] * (n // 3) + [n % 3],
}[chunking]
chunk_sizes = [c for c in chunk_sizes if c > 0]
streamed = await stream_chat_derender(
parser_client,
output_ids,
chunk_sizes,
chat_request,
len(gen_req["token_ids"]),
gen_req["token_ids"],
)
assert streamed["content"] == batch_msg["content"]
assert streamed["reasoning"] == batch_msg.get("reasoning")
@pytest.mark.asyncio
async def test_stream_parsed_matches_batch_tool_call(parser_client, parser_tokenizer):
"""Tool call name+id+arguments parity between streamed and batch, plus
the finish_reason -> "tool_calls" rewrite for auto tool choice."""
messages = [{"role": "user", "content": "Weather in Paris?"}]
gen_req = await _render_parser_chat(parser_client, messages)
output_text = (
"Let me check."
'\n{"name": "get_weather", '
'"arguments": {"city": "Paris"}}\n'
)
output_ids = _require_parser_markers(
parser_tokenizer, output_text, "", "", ""
)
chat_request = {
"model": PARSER_MODEL,
"messages": messages,
"tools": _PARSER_TOOLS,
"tool_choice": "auto",
}
batch_resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": {
"request_id": "batch",
"choices": [
{"index": 0, "token_ids": output_ids, "finish_reason": "stop"}
],
},
"prompt_tokens": len(gen_req["token_ids"]),
"chat_request": chat_request,
},
)
assert batch_resp.status_code == 200, batch_resp.text
batch_choice = batch_resp.json()["choices"][0]
if not batch_choice["message"]["tool_calls"]:
pytest.skip("Model did not emit a tool call")
for chunk_sizes in ([len(output_ids)], [1] * len(output_ids)):
# Snapshot of tool_calls[0]["arguments"] after every chunk, to check
# it only ever grows by appending. It never retracts or duplicates
# already streamed text. That's the exact failure mode the
# replay + parse_delta design (over diff-based streaming) exists to
# avoid. Parity with the batch result alone wouldn't catch a
# transient mid-stream regression.
arg_snapshots: list[str] = []
def _record(
tool_calls: list[dict], _snapshots: list[str] = arg_snapshots
) -> None:
if tool_calls and tool_calls[0]["arguments"]:
_snapshots.append(tool_calls[0]["arguments"])
streamed = await stream_chat_derender(
parser_client,
output_ids,
chunk_sizes,
chat_request,
len(gen_req["token_ids"]),
gen_req["token_ids"],
on_chunk=_record,
)
assert streamed["tool_calls"]
assert streamed["tool_calls"][0]["name"] == "get_weather"
assert streamed["tool_calls"][0]["id"] is not None
assert json.loads(streamed["tool_calls"][0]["arguments"]) == json.loads(
batch_choice["message"]["tool_calls"][0]["function"]["arguments"]
)
assert streamed["finish_reason"] == "tool_calls"
assert arg_snapshots, "expected at least one tool-call argument delta"
for prev, curr in zip(arg_snapshots, arg_snapshots[1:]):
assert curr.startswith(prev), (
f"tool-call arguments regressed: {prev!r} -> {curr!r}"
)
assert arg_snapshots[-1] == streamed["tool_calls"][0]["arguments"]
@pytest.mark.asyncio
async def test_stream_parsed_cjk_across_chunk_boundaries(
parser_client, parser_tokenizer
):
"""CJK/emoji tokens split across chunk boundaries with a parser active
across the reasoning -> content transition (parser path regression
guard analogous to #46159 in the plain detok path)."""
messages = [{"role": "user", "content": "Reply in Chinese"}]
gen_req = await _render_parser_chat(parser_client, messages)
reasoning_text = "思考中"
answer_text = "你好世界 😀"
output_text = f"{reasoning_text}{answer_text}"
output_ids = _require_parser_markers(parser_tokenizer, output_text, "")
chat_request = {
"model": PARSER_MODEL,
"messages": messages,
"include_reasoning": True,
}
streamed = await stream_chat_derender(
parser_client,
output_ids,
[1] * len(output_ids),
chat_request,
len(gen_req["token_ids"]),
gen_req["token_ids"],
)
assert "�" not in (streamed["content"] or "")
assert "�" not in (streamed["reasoning"] or "")
assert answer_text in (streamed["content"] or "")
@pytest.mark.asyncio
async def test_stream_parsed_include_reasoning_false(parser_client, parser_tokenizer):
"""include_reasoning=False emits no reasoning deltas on the streaming
parser path."""
messages = [{"role": "user", "content": "What is 2+3?"}]
gen_req = await _render_parser_chat(parser_client, messages)
output_text = "reasoning hereThe answer is 5."
output_ids = _require_parser_markers(parser_tokenizer, output_text, "")
chat_request = {
"model": PARSER_MODEL,
"messages": messages,
"include_reasoning": False,
}
streamed = await stream_chat_derender(
parser_client,
output_ids,
[1] * len(output_ids),
chat_request,
len(gen_req["token_ids"]),
gen_req["token_ids"],
)
assert streamed["reasoning"] is None
@pytest.mark.asyncio
async def test_stream_parsed_missing_chat_request_rejected(parser_client):
"""Parser configured + no chat_request on the streaming endpoint -> 400
(the live-server counterpart to the mocked
`TestServingDerenderStreamValidation` checks above)."""
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"stream": True,
"model": PARSER_MODEL,
"generate_chunk": {
"request_id": "reject-test",
"choices": [{"index": 0, "token_ids": [1, 2], "finish_reason": None}],
},
},
)
assert resp.status_code == 400
assert "chat_request" in resp.json()["error"]["message"]
@pytest.mark.asyncio
async def test_stream_parsed_missing_prompt_token_ids_rejected(parser_client):
"""Parser configured + chat_request given but no prompt_token_ids on the
streaming endpoint -> 400 (the live-server counterpart to the mocked
`test_missing_prompt_token_ids_with_parser_rejected` above). Without
prompt_token_ids, parse_delta cannot tell whether the prompt left
reasoning open and would silently misclassify reasoning as content."""
messages = [{"role": "user", "content": "Hello"}]
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"stream": True,
"model": PARSER_MODEL,
"generate_chunk": {
"request_id": "reject-test",
"choices": [{"index": 0, "token_ids": [1, 2], "finish_reason": None}],
},
"chat_request": {
"model": PARSER_MODEL,
"messages": messages,
"include_reasoning": True,
},
},
)
assert resp.status_code == 400
assert "prompt_token_ids" in resp.json()["error"]["message"]