From 4d4f4b89ae1f45d339db74fe6d82cb5c1e52ca09 Mon Sep 17 00:00:00 2001 From: Sungsoo Ha Date: Thu, 27 Aug 2026 16:27:26 -0700 Subject: [PATCH 1/6] fix(frontend): report reasoning token usage on the vllm chat-processor path `--dyn-chat-processor vllm` selects the Python chat processor, which never implements reasoning-token usage. #12181 added it, but only to the Rust `OpenAIPreprocessor` (`lib/llm/src/preprocessor.rs`), which this path bypasses entirely. The vLLM worker's `_build_completion_usage()` emits no `completion_tokens_details` key at all, so `reasoning_tokens` is **absent** from usage rather than zero -- clients that default a missing field then report 0 reasoning tokens for a response with visibly non-empty `reasoning_content`. Port #12181's estimator to the Python path, kept deliberately behaviour-identical to the Rust one -- including its documented chunk-granular overcount (a chunk carrying both reasoning and visible content counts entirely as reasoning) -- so the two paths cannot disagree. A positive backend-supplied count stays authoritative. Measured on Inkling-NVFP4 (vLLM 0.28.0, GB300 TP4): `reasoning_content` 518 chars -> `usage.reasoning_tokens` 131; `reasoning_tokens <= completion_tokens`; 0 when `reasoning_effort: none` produces no reasoning; correct in streaming. Adds unit tests covering chunk accounting, the active-choice gap, usage-only frames, tool calls as visible output, and the annotate contract (fills when absent, preserves a positive backend value, replaces a zero, never mutates its input). Signed-off-by: Sungsoo Ha --- .../tests/test_vllm_processor_unit.py | 77 +++++++++++++++++++ .../src/dynamo/frontend/vllm_processor.py | 74 +++++++++++++++++- 2 files changed, 150 insertions(+), 1 deletion(-) diff --git a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py index b58888b4e878..76bc8c349255 100644 --- a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py +++ b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py @@ -2539,3 +2539,80 @@ def test_shared_thinking_policy(self, tokenizer, case): tool_parser_class=None, ) assert kwargs == case.expected + + +class TestReasoningUsageEstimator: + """Reasoning-token usage on the Python chat-processor path. + + dynamo #12181 added this only to the Rust OpenAIPreprocessor, which + `--dyn-chat-processor vllm` bypasses, so `reasoning_tokens` was absent from usage + entirely (the worker's _build_completion_usage emits no completion_tokens_details). + """ + + @staticmethod + def _est(): + from dynamo.frontend.vllm_processor import _ReasoningUsageEstimator + + return _ReasoningUsageEstimator() + + @staticmethod + def _choice(index=0, reasoning=None, content=None, tools=None): + delta = {} + if reasoning is not None: + delta["reasoning_content"] = reasoning + if content is not None: + delta["content"] = content + if tools is not None: + delta["tool_calls"] = tools + return {"index": index, "delta": delta} + + def test_counts_only_reasoning_chunks(self): + e = self._est() + e.observe([self._choice(reasoning="a")], 10) + e.observe([self._choice(reasoning="b")], 5) + e.observe([self._choice(content="answer")], 7) + assert e.total == 15 + + def test_active_choice_without_visible_output_still_counts(self): + e = self._est() + e.observe([self._choice(reasoning="a")], 4) + e.observe([self._choice()], 3) + e.observe([self._choice(content="x")], 9) + assert e.total == 7 + + def test_usage_only_frame_while_reasoning_counts(self): + e = self._est() + e.observe([self._choice(reasoning="a")], 4) + e.observe([], 2) + assert e.total == 6 + + def test_content_only_counts_nothing(self): + e = self._est() + e.observe([self._choice(content="hi")], 12) + assert e.total == 0 + + def test_tool_call_is_visible_output_and_stops_reasoning(self): + e = self._est() + e.observe([self._choice(reasoning="a")], 5) + e.observe([self._choice(tools=[{"id": "1"}])], 6) + assert e.total == 5 + + def test_annotate_fills_when_absent_and_does_not_mutate_input(self): + e = self._est() + e.total = 42 + usage = {"completion_tokens": 100} + out = e.annotate(usage) + assert out["completion_tokens_details"]["reasoning_tokens"] == 42 + assert "completion_tokens_details" not in usage + + def test_annotate_preserves_a_positive_backend_value(self): + e = self._est() + e.total = 42 + usage = {"completion_tokens_details": {"reasoning_tokens": 7}} + assert e.annotate(usage)["completion_tokens_details"]["reasoning_tokens"] == 7 + + def test_annotate_replaces_a_zero_backend_value(self): + e = self._est() + e.total = 42 + usage = {"completion_tokens_details": {"reasoning_tokens": 0}} + assert e.annotate(usage)["completion_tokens_details"]["reasoning_tokens"] == 42 diff --git a/components/src/dynamo/frontend/vllm_processor.py b/components/src/dynamo/frontend/vllm_processor.py index cb5ded3fe439..82fb819044cb 100644 --- a/components/src/dynamo/frontend/vllm_processor.py +++ b/components/src/dynamo/frontend/vllm_processor.py @@ -64,6 +64,73 @@ } +class _ReasoningUsageEstimator: + """Chunk-granular reasoning-token estimate for the PYTHON chat-processor path. + + NVBug 6678449b. dynamo #12181 added this, but only to the RUST OpenAIPreprocessor + (lib/llm/src/preprocessor.rs). `--dyn-chat-processor vllm` selects this Python path, + which bypasses that code entirely -- and the worker's _build_completion_usage() + emits no `completion_tokens_details` key at all, so `reasoning_tokens` is ABSENT + (not null) from usage. This is a port of that estimator, kept deliberately + behaviour-identical to the Rust so the two paths cannot disagree. + + Chunk-granular, exactly as upstream: if a single chunk carries both reasoning and + visible content, ALL of its tokens count as reasoning. Upstream documents that + overcount and accepts it; matching it is more useful than being cleverer here. + A positive backend-supplied count stays authoritative. + """ + + __slots__ = ("total", "_active") + + def __init__(self) -> None: + self.total = 0 + self._active: set[Any] = set() + + @staticmethod + def _has_visible_output(choice: dict[str, Any]) -> bool: + delta = choice.get("delta") or {} + return delta.get("content") is not None or bool(delta.get("tool_calls")) + + @staticmethod + def _is_reasoning(choice: dict[str, Any]) -> bool: + return ((choice.get("delta") or {}).get("reasoning_content")) is not None + + def observe(self, choices: list[dict[str, Any]], chunk_tokens: Any) -> None: + try: + count = int(chunk_tokens or 0) + except (TypeError, ValueError): + count = 0 + + has_reasoning = any(self._is_reasoning(c) for c in choices) + active_without_visible = any( + c.get("index") in self._active and not self._has_visible_output(c) + for c in choices + ) + # A usage-only frame (no choices) that arrives while a choice is still + # reasoning still carries reasoning tokens. + usage_frame_while_reasoning = not choices and bool(self._active) + + if count > 0 and ( + has_reasoning or active_without_visible or usage_frame_while_reasoning + ): + self.total += count + + for choice in choices: + if self._is_reasoning(choice): + self._active.add(choice.get("index")) + elif self._has_visible_output(choice): + self._active.discard(choice.get("index")) + + def annotate(self, usage: dict[str, Any]) -> dict[str, Any]: + """Return a COPY of usage carrying reasoning_tokens; never mutates the input.""" + annotated = dict(usage) + details = dict(annotated.get("completion_tokens_details") or {}) + if not details.get("reasoning_tokens"): + details["reasoning_tokens"] = self.total + annotated["completion_tokens_details"] = details + return annotated + + def map_finish_reason(raw_reason: str | None) -> FinishReason | None: if raw_reason is None: return None @@ -838,6 +905,9 @@ async def _generate_and_stream( # content-part counts here too (else frontend metrics report zero media). input_tokens = len(tokens) cumulative_output_tokens = 0 + # Per-request reasoning-token estimate (NVBug 6678449b); see + # _ReasoningUsageEstimator. Must be per-request, never module-level. + reasoning_usage = _ReasoningUsageEstimator() _mm_counts, _ = extract_mm_urls(request.get("messages") or []) _mm_counts = _mm_counts or {} image_count = len(_mm_counts.get("image_url", [])) @@ -950,6 +1020,8 @@ async def _generate_and_stream( if postprocess_error: continue + reasoning_usage.observe(choices, chunk_tokens) + # One envelope per iteration carries both data and metrics so # client cancellation can't drop the annotation between yields. envelope: dict[str, Any] = {"_dynamo_annotated": True} @@ -962,7 +1034,7 @@ async def _generate_and_stream( "object": "chat.completion.chunk", } if usage := engine_response.get("completion_usage"): - dynamo_out["usage"] = usage + dynamo_out["usage"] = reasoning_usage.annotate(usage) envelope["data"] = dynamo_out metrics = { From cca33a542f20c3c7e11969032b1d28995d5a6b5a Mon Sep 17 00:00:00 2001 From: Sungsoo Ha Date: Thu, 27 Aug 2026 21:25:54 -0700 Subject: [PATCH 2/6] fix(frontend): account reasoning tokens at parser classification Addresses both P1 review findings. The estimator observed the choices this path EMITS, so any projection that dropped or deferred the reasoning lost the count: * include_reasoning=false makes StreamingPostProcessor suppress every reasoning delta, so a response with hidden reasoning reported 0; * the non-streaming tool path emits no choice per generated chunk and releases the aggregated reasoning on a terminal delta whose chunk_tokens is 0, so the buffered reasoning chunks reported 0. Move the accounting to where the reasoning parser CLASSIFIES its output. StreamingPostProcessor now owns `reasoning_token_total`, incremented before the include_reasoning projection on the streaming path and, on the buffered path (which has no per-chunk classification), computed exactly from the parsed reasoning text before the same projection. _ReasoningUsageEstimator becomes _ReasoningUsageAnnotator, which only sums those totals -- it no longer derives anything from the emitted choices. Streaming stays chunk-granular, matching the Rust: a chunk carrying reasoning and visible content counts wholly as reasoning. The buffered path counts the reasoning text exactly, which is more precise than the Rust rather than divergent from it. Measured on Inkling-NVFP4 / GB300, vLLM 0.28.0, against the same image with only this change overlaid. Reported vs an independent tokenization of the returned reasoning_content, 3 runs each: non-streaming tool call before 0.03/0.03/0.02 after 1.00/1.00/1.00 plain reasoning request before 1.04/1.03/1.02 after 1.00/1.03/1.04 Before, the tool case reported a constant 7 whatever the real count was (212, 242, 295) -- the single terminal frame. The plain path is unchanged. Unit tests cover both regressions plus the buffered-and-suppressed combination. Signed-off-by: Sungsoo Ha --- components/src/dynamo/frontend/prepost.py | 30 +++ .../tests/test_vllm_processor_unit.py | 229 +++++++++++++----- .../src/dynamo/frontend/vllm_processor.py | 77 +++--- 3 files changed, 227 insertions(+), 109 deletions(-) diff --git a/components/src/dynamo/frontend/prepost.py b/components/src/dynamo/frontend/prepost.py index 512450a987be..969f6a964212 100644 --- a/components/src/dynamo/frontend/prepost.py +++ b/components/src/dynamo/frontend/prepost.py @@ -714,6 +714,19 @@ def __init__( self._reasoning_parser_streaming_started = False self._tool_parser_streaming_started = False + # Reasoning tokens attributed to THIS choice, accumulated where the + # reasoning parser CLASSIFIES output -- deliberately not where the + # response is projected. Two projections would otherwise lose the count + # entirely (NVBug 6678449b): + # * `_suppress_reasoning_output` (include_reasoning=false) drops the + # parsed reasoning from the emitted delta, so a response with hidden + # reasoning would report zero; + # * the non-streaming tool path buffers every generated chunk and + # emits once on the terminal delta, so per-chunk observation sees + # only a zero-token terminal frame. + # Counting at classification is invariant to both. + self.reasoning_token_total = 0 + self._control_markers = tuple( t for t in getattr(tokenizer, "all_special_tokens", ()) if t ) @@ -1072,6 +1085,15 @@ def _process_non_streaming_tool_output(self, output: Any) -> dict[str, Any] | No current_text, request=self.request_for_sampling, ) + # This path buffers every generated chunk and parses once here, so + # there is no per-chunk classification to count. The reasoning text + # IS known now, so count it exactly -- and do it before the + # include_reasoning projection on the next line, which would + # otherwise hide it. + if saved_reasoning: + self.reasoning_token_total += len( + self.tokenizer.encode(saved_reasoning, add_special_tokens=False) + ) if not self.request_for_sampling.include_reasoning: saved_reasoning = None @@ -1264,6 +1286,14 @@ def process_output(self, output: Any) -> dict[str, Any] | None: ) self._tool_parser_streaming_started = False + # Account BEFORE the projection below. `delta_message` is the parser's + # classification; `reasoning` is what survives include_reasoning. Reading + # the latter is what reported zero for hidden reasoning. + # Chunk-granular, matching the Rust estimator: a chunk carrying both + # reasoning and visible content counts entirely as reasoning. + if delta_message is not None and delta_message.reasoning: + self.reasoning_token_total += len(delta_token_ids) + choice = None if delta_message is None: if self.in_progress_tool_calls: diff --git a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py index 76bc8c349255..da9fca08845b 100644 --- a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py +++ b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py @@ -2541,78 +2541,189 @@ def test_shared_thinking_policy(self, tokenizer, case): assert kwargs == case.expected -class TestReasoningUsageEstimator: - """Reasoning-token usage on the Python chat-processor path. +class _AccountingReasoningParser: + """Emits reasoning until a chunk contains "", then visible content.""" + + engine_based_streaming = False + + def __init__(self, tokenizer, *, chat_template_kwargs=None): + self.tokenizer = tokenizer + self.chat_template_kwargs = chat_template_kwargs or {} + + def is_reasoning_end(self, prompt_token_ids): + return False + + def adjust_initial_state_from_prompt(self, prompt_token_ids): + return None + + def extract_reasoning_streaming( + self, + previous_text, + current_text, + delta_text, + previous_token_ids, + current_token_ids, + delta_token_ids, + ): + if "" in delta_text: + return prepost_module.DeltaMessage(content=delta_text.split("")[-1]) + return prepost_module.DeltaMessage(reasoning=delta_text) + + def is_reasoning_end_streaming(self, current_token_ids, delta_token_ids): + return False + + def extract_reasoning(self, text, request=None): + if "" in text: + reasoning, _, content = text.partition("") + return reasoning, content + return text, "" + + +class _PassthroughToolParser: + """Finds no tool calls; returns the text as content. + + The buffered non-streaming path only needs `tool_parser is not None` to engage, + so this keeps the reasoning-accounting test focused on reasoning. + """ + + def __init__(self, tokenizer, *args, **kwargs): + self.tokenizer = tokenizer + + def extract_tool_calls(self, text, request=None): + return SimpleNamespace(tools_called=False, tool_calls=[], content=text) + + +class TestReasoningTokenAccounting: + """Reasoning-token usage on the Python chat-processor path (NVBug 6678449b). dynamo #12181 added this only to the Rust OpenAIPreprocessor, which `--dyn-chat-processor vllm` bypasses, so `reasoning_tokens` was absent from usage entirely (the worker's _build_completion_usage emits no completion_tokens_details). + + The count is accumulated where the reasoning parser CLASSIFIES output, NOT where + the response is projected. The two regression tests below both reported 0 when it + was derived from the emitted choices. """ @staticmethod - def _est(): - from dynamo.frontend.vllm_processor import _ReasoningUsageEstimator + def _post(tokenizer, *, include_reasoning=True, stream_response=True, tools=False): + request = SimpleNamespace( + include_reasoning=include_reasoning, + tool_choice="auto" if tools else "none", + model_fields=frozenset(), + ) + return StreamingPostProcessor( + tokenizer=tokenizer, + request_for_sampling=request, + sampling_params=SamplingParams(), + prompt_token_ids=[], + tool_parser=_PassthroughToolParser(tokenizer) if tools else None, + reasoning_parser_class=_AccountingReasoningParser, + chat_template_kwargs={}, + stream_response=stream_response, + ) - return _ReasoningUsageEstimator() + @staticmethod + def _out(text, token_ids, finish_reason=None, index=0): + return SimpleNamespace( + index=index, + text=text, + token_ids=token_ids, + finish_reason=finish_reason, + logprobs=None, + ) @staticmethod - def _choice(index=0, reasoning=None, content=None, tools=None): - delta = {} - if reasoning is not None: - delta["reasoning_content"] = reasoning - if content is not None: - delta["content"] = content - if tools is not None: - delta["tool_calls"] = tools - return {"index": index, "delta": delta} - - def test_counts_only_reasoning_chunks(self): - e = self._est() - e.observe([self._choice(reasoning="a")], 10) - e.observe([self._choice(reasoning="b")], 5) - e.observe([self._choice(content="answer")], 7) - assert e.total == 15 - - def test_active_choice_without_visible_output_still_counts(self): - e = self._est() - e.observe([self._choice(reasoning="a")], 4) - e.observe([self._choice()], 3) - e.observe([self._choice(content="x")], 9) - assert e.total == 7 - - def test_usage_only_frame_while_reasoning_counts(self): - e = self._est() - e.observe([self._choice(reasoning="a")], 4) - e.observe([], 2) - assert e.total == 6 - - def test_content_only_counts_nothing(self): - e = self._est() - e.observe([self._choice(content="hi")], 12) - assert e.total == 0 - - def test_tool_call_is_visible_output_and_stops_reasoning(self): - e = self._est() - e.observe([self._choice(reasoning="a")], 5) - e.observe([self._choice(tools=[{"id": "1"}])], 6) - assert e.total == 5 - - def test_annotate_fills_when_absent_and_does_not_mutate_input(self): - e = self._est() - e.total = 42 + def _annotator(*posts): + from dynamo.frontend.vllm_processor import _ReasoningUsageAnnotator + + return _ReasoningUsageAnnotator({i: p for i, p in enumerate(posts)}) + + # -- streaming classification ------------------------------------------ + + def test_streaming_counts_reasoning_chunks_only(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("thinking", [1, 2, 3])) + post.process_output(self._out("more", [4, 5])) + post.process_output(self._out("answer", [6, 7, 8, 9])) + assert post.reasoning_token_total == 5 + + # -- P1 #1: account before the include_reasoning projection ------------ + + def test_hidden_reasoning_is_counted_when_include_reasoning_false( + self, tokenizer + ): + """include_reasoning=false suppresses the reasoning DELTA, not the tokens. + + Observing emitted choices saw no reasoning_content here and reported 0 for a + response that really did spend tokens reasoning. + """ + post = self._post(tokenizer, include_reasoning=False) + choice = post.process_output(self._out("hidden", [1, 2, 3])) + + # The projection still hides it -- that part is intended... + assert choice is None or "reasoning_content" not in (choice.get("delta") or {}) + # ...but the tokens are accounted anyway. + assert post.reasoning_token_total == 3 + annotated = self._annotator(post).annotate({"completion_tokens": 10}) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 3 + + # -- P1 #2: survive non-streaming tool buffering ------------------------ + + def test_non_streaming_tool_buffering_preserves_count(self, tokenizer): + """The non-streaming tool path emits nothing per chunk and releases on the + terminal delta, where chunk_tokens is 0. Per-chunk observation of emitted + choices therefore saw only zeros and reported 0. + """ + post = self._post(tokenizer, stream_response=False, tools=True) + for chunk in ("think a", "think b", "think c"): + assert post.process_output(self._out(chunk, [1, 2])) is None + post.process_output(self._out("answer", [3], finish_reason="stop")) + + assert post.reasoning_token_total > 0 + annotated = self._annotator(post).annotate({"completion_tokens": 20}) + assert annotated["completion_tokens_details"]["reasoning_tokens"] > 0 + + def test_non_streaming_buffering_counts_even_when_suppressed(self, tokenizer): + """Both projections at once: buffered AND include_reasoning=false.""" + post = self._post( + tokenizer, stream_response=False, tools=True, include_reasoning=False + ) + post.process_output(self._out("thinking hard", [1, 2])) + post.process_output(self._out("answer", [3], finish_reason="stop")) + assert post.reasoning_token_total > 0 + + def test_content_only_counts_nothing(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("hi", [1, 2, 3, 4])) + assert post.reasoning_token_total == 0 + + # -- annotation --------------------------------------------------------- + + def test_total_sums_across_choices(self, tokenizer): + a, b = self._post(tokenizer), self._post(tokenizer) + a.process_output(self._out("x", [1, 2])) + b.process_output(self._out("y", [3, 4, 5])) + assert self._annotator(a, b).total == 5 + + def test_annotate_fills_when_absent_and_does_not_mutate_input(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) usage = {"completion_tokens": 100} - out = e.annotate(usage) - assert out["completion_tokens_details"]["reasoning_tokens"] == 42 + out = self._annotator(post).annotate(usage) + assert out["completion_tokens_details"]["reasoning_tokens"] == 2 assert "completion_tokens_details" not in usage - def test_annotate_preserves_a_positive_backend_value(self): - e = self._est() - e.total = 42 + def test_annotate_preserves_a_positive_backend_value(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) usage = {"completion_tokens_details": {"reasoning_tokens": 7}} - assert e.annotate(usage)["completion_tokens_details"]["reasoning_tokens"] == 7 + annotated = self._annotator(post).annotate(usage) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 7 - def test_annotate_replaces_a_zero_backend_value(self): - e = self._est() - e.total = 42 + def test_annotate_replaces_a_zero_backend_value(self, tokenizer): + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) usage = {"completion_tokens_details": {"reasoning_tokens": 0}} - assert e.annotate(usage)["completion_tokens_details"]["reasoning_tokens"] == 42 + annotated = self._annotator(post).annotate(usage) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 2 diff --git a/components/src/dynamo/frontend/vllm_processor.py b/components/src/dynamo/frontend/vllm_processor.py index 82fb819044cb..bb71aa67ba9a 100644 --- a/components/src/dynamo/frontend/vllm_processor.py +++ b/components/src/dynamo/frontend/vllm_processor.py @@ -64,62 +64,40 @@ } -class _ReasoningUsageEstimator: - """Chunk-granular reasoning-token estimate for the PYTHON chat-processor path. +class _ReasoningUsageAnnotator: + """Report reasoning-token usage on the PYTHON chat-processor path. NVBug 6678449b. dynamo #12181 added this, but only to the RUST OpenAIPreprocessor (lib/llm/src/preprocessor.rs). `--dyn-chat-processor vllm` selects this Python path, which bypasses that code entirely -- and the worker's _build_completion_usage() emits no `completion_tokens_details` key at all, so `reasoning_tokens` is ABSENT - (not null) from usage. This is a port of that estimator, kept deliberately - behaviour-identical to the Rust so the two paths cannot disagree. - - Chunk-granular, exactly as upstream: if a single chunk carries both reasoning and - visible content, ALL of its tokens count as reasoning. Upstream documents that - overcount and accepts it; matching it is more useful than being cleverer here. + (not null) from usage. + + The count is NOT derived here. Each StreamingPostProcessor accumulates + `reasoning_token_total` where the reasoning parser CLASSIFIES its output; this + class only sums those totals and annotates usage. Deriving it here -- from the + choices this path emits -- is what reported zero whenever the response projection + dropped or deferred the reasoning (include_reasoning=false, and the non-streaming + tool path's terminal-delta buffering). + + Streaming classification stays chunk-granular, matching the Rust: a chunk carrying + both reasoning and visible content counts entirely as reasoning. The non-streaming + tool path has no per-chunk classification, so it counts the parsed reasoning text + exactly; that is more precise than the Rust rather than divergent from it. A positive backend-supplied count stays authoritative. """ - __slots__ = ("total", "_active") - - def __init__(self) -> None: - self.total = 0 - self._active: set[Any] = set() + __slots__ = ("_post_processors",) - @staticmethod - def _has_visible_output(choice: dict[str, Any]) -> bool: - delta = choice.get("delta") or {} - return delta.get("content") is not None or bool(delta.get("tool_calls")) + def __init__(self, post_processors: dict[int, StreamingPostProcessor]) -> None: + self._post_processors = post_processors - @staticmethod - def _is_reasoning(choice: dict[str, Any]) -> bool: - return ((choice.get("delta") or {}).get("reasoning_content")) is not None - - def observe(self, choices: list[dict[str, Any]], chunk_tokens: Any) -> None: - try: - count = int(chunk_tokens or 0) - except (TypeError, ValueError): - count = 0 - - has_reasoning = any(self._is_reasoning(c) for c in choices) - active_without_visible = any( - c.get("index") in self._active and not self._has_visible_output(c) - for c in choices + @property + def total(self) -> int: + return sum( + getattr(post, "reasoning_token_total", 0) + for post in self._post_processors.values() ) - # A usage-only frame (no choices) that arrives while a choice is still - # reasoning still carries reasoning tokens. - usage_frame_while_reasoning = not choices and bool(self._active) - - if count > 0 and ( - has_reasoning or active_without_visible or usage_frame_while_reasoning - ): - self.total += count - - for choice in choices: - if self._is_reasoning(choice): - self._active.add(choice.get("index")) - elif self._has_visible_output(choice): - self._active.discard(choice.get("index")) def annotate(self, usage: dict[str, Any]) -> dict[str, Any]: """Return a COPY of usage carrying reasoning_tokens; never mutates the input.""" @@ -905,9 +883,10 @@ async def _generate_and_stream( # content-part counts here too (else frontend metrics report zero media). input_tokens = len(tokens) cumulative_output_tokens = 0 - # Per-request reasoning-token estimate (NVBug 6678449b); see - # _ReasoningUsageEstimator. Must be per-request, never module-level. - reasoning_usage = _ReasoningUsageEstimator() + # Per-request reasoning-token usage (NVBug 6678449b); see + # _ReasoningUsageAnnotator. Must be per-request, never module-level. + # The counts live on the post-processors, which are per-request too. + reasoning_usage = _ReasoningUsageAnnotator(post_processors) _mm_counts, _ = extract_mm_urls(request.get("messages") or []) _mm_counts = _mm_counts or {} image_count = len(_mm_counts.get("image_url", [])) @@ -1020,8 +999,6 @@ async def _generate_and_stream( if postprocess_error: continue - reasoning_usage.observe(choices, chunk_tokens) - # One envelope per iteration carries both data and metrics so # client cancellation can't drop the annotation between yields. envelope: dict[str, Any] = {"_dynamo_annotated": True} From 07aa422971bd575bb2dd32da57fe88bb4d33a4f2 Mon Sep 17 00:00:00 2001 From: Sungsoo Ha Date: Thu, 27 Aug 2026 21:33:39 -0700 Subject: [PATCH 3/6] style(frontend): black formatting for the reasoning-accounting tests pre-commit black (pinned 23.1.0) collapsed one test signature onto a single line. Formatting only, no behaviour change. Signed-off-by: Sungsoo Ha --- .../src/dynamo/frontend/tests/test_vllm_processor_unit.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py index da9fca08845b..d6df9cbe1b2b 100644 --- a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py +++ b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py @@ -2650,9 +2650,7 @@ def test_streaming_counts_reasoning_chunks_only(self, tokenizer): # -- P1 #1: account before the include_reasoning projection ------------ - def test_hidden_reasoning_is_counted_when_include_reasoning_false( - self, tokenizer - ): + def test_hidden_reasoning_is_counted_when_include_reasoning_false(self, tokenizer): """include_reasoning=false suppresses the reasoning DELTA, not the tokens. Observing emitted choices saw no reasoning_content here and reported 0 for a From 4c489b71a56e152432bed3961c243422ef8373fd Mon Sep 17 00:00:00 2001 From: Sungsoo Ha Date: Thu, 27 Aug 2026 21:44:32 -0700 Subject: [PATCH 4/6] test(frontend): let the accounting test's parser accept model_config Forward-compatible with #12541, which threads model_config into the reasoning-parser constructor. Optional kwarg, so it is a no-op on this branch and does not break once #12541 lands. Signed-off-by: Sungsoo Ha --- .../src/dynamo/frontend/tests/test_vllm_processor_unit.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py index d6df9cbe1b2b..2bdb63a671ea 100644 --- a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py +++ b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py @@ -2546,9 +2546,10 @@ class _AccountingReasoningParser: engine_based_streaming = False - def __init__(self, tokenizer, *, chat_template_kwargs=None): + def __init__(self, tokenizer, *, chat_template_kwargs=None, model_config=None): self.tokenizer = tokenizer self.chat_template_kwargs = chat_template_kwargs or {} + self.model_config = model_config def is_reasoning_end(self, prompt_token_ids): return False From 6a02e4959b5b12f8d287167dba4d0cd75d195bd2 Mon Sep 17 00:00:00 2001 From: Sungsoo Ha Date: Thu, 27 Aug 2026 23:14:25 -0700 Subject: [PATCH 5/6] fix(frontend): only treat a POSITIVE backend reasoning_tokens as authoritative Two CodeRabbit findings on #13947. 1. `annotate()` used a falsiness check, so a negative backend-supplied `reasoning_tokens` survived (-1 is truthy) and was reported to the client. The documented contract is that only a POSITIVE backend count is authoritative; missing, zero and negative now all fall back to our own tally. Covered by a parametrised regression test (-1, -100, None, "7"). 2. `total` used `getattr(post, "reasoning_token_total", 0)`. The values are StreamingPostProcessor instances and the attribute is initialised in their constructor, so the defensive default could only mask a broken processor as zero usage. Direct attribute access instead. A third finding -- move the `_ReasoningUsageAnnotator` import in the test helper to module scope -- is NOT applied. This test module imports dynamo.frontend.vllm_processor lazily by design (see the NOTE at the top of the file): a module-level import pulls `vllm.tasks` / `vllm.v1.engine.parallel_sampling` in during pytest collection and breaks the pytest-marker-report pre-commit hook, whose vllm stub list does not cover those submodules. Verified: adding the module-scope import fails that hook with `ImportError ... from vllm.exceptions import VLLMClientError`. The local import matches the five other lazy imports already in this file. Signed-off-by: Sungsoo Ha --- .../frontend/tests/test_vllm_processor_unit.py | 13 +++++++++++++ components/src/dynamo/frontend/vllm_processor.py | 9 ++++++--- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py index 2bdb63a671ea..071b92c5d02f 100644 --- a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py +++ b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py @@ -2726,3 +2726,16 @@ def test_annotate_replaces_a_zero_backend_value(self, tokenizer): usage = {"completion_tokens_details": {"reasoning_tokens": 0}} annotated = self._annotator(post).annotate(usage) assert annotated["completion_tokens_details"]["reasoning_tokens"] == 2 + + @pytest.mark.parametrize("backend", [-1, -100, None, "7"]) + def test_annotate_replaces_a_non_positive_backend_value(self, tokenizer, backend): + """Only a POSITIVE backend count is authoritative. + + -1 is truthy, so a plain falsiness check preserved it and reported a + negative reasoning_tokens to the client. + """ + post = self._post(tokenizer) + post.process_output(self._out("x", [1, 2])) + usage = {"completion_tokens_details": {"reasoning_tokens": backend}} + annotated = self._annotator(post).annotate(usage) + assert annotated["completion_tokens_details"]["reasoning_tokens"] == 2 diff --git a/components/src/dynamo/frontend/vllm_processor.py b/components/src/dynamo/frontend/vllm_processor.py index bb71aa67ba9a..e68773687c00 100644 --- a/components/src/dynamo/frontend/vllm_processor.py +++ b/components/src/dynamo/frontend/vllm_processor.py @@ -95,15 +95,18 @@ def __init__(self, post_processors: dict[int, StreamingPostProcessor]) -> None: @property def total(self) -> int: return sum( - getattr(post, "reasoning_token_total", 0) - for post in self._post_processors.values() + post.reasoning_token_total for post in self._post_processors.values() ) def annotate(self, usage: dict[str, Any]) -> dict[str, Any]: """Return a COPY of usage carrying reasoning_tokens; never mutates the input.""" annotated = dict(usage) details = dict(annotated.get("completion_tokens_details") or {}) - if not details.get("reasoning_tokens"): + # Only a POSITIVE backend count is authoritative. Missing, zero and + # negative all fall back to our own tally -- a negative would otherwise + # survive, since -1 is truthy. + backend = details.get("reasoning_tokens") + if not isinstance(backend, int) or backend <= 0: details["reasoning_tokens"] = self.total annotated["completion_tokens_details"] = details return annotated From b7f1f105d02910de038b7b5886ed7cf19a7f5b6b Mon Sep 17 00:00:00 2001 From: Sungsoo Ha Date: Fri, 28 Aug 2026 00:43:01 -0700 Subject: [PATCH 6/6] fix(frontend): count reasoning tokens with the parser's own counter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the P1 review finding: accounting was derived from parser output even though the pinned vLLM already exposes `ReasoningParser.count_reasoning_tokens(token_ids)`. StreamingPostProcessor now keeps a per-choice ledger of the ORIGINAL generated token ids, appended before any branch in process_output so every path feeds it -- including the ones that emit nothing for a chunk. `reasoning_token_total` becomes a property over the parser counter. Both previous mechanisms are gone: the `len(delta_token_ids)` per-chunk classifier and the terminal `tokenizer.encode(saved_reasoning)` retokenisation. That removes three separate error sources: * the response projections drop or defer the reasoning (`_suppress_reasoning_output`, and the non-streaming tool path's terminal-delta buffering); * a per-chunk classifier over-counts a chunk carrying both kinds; * re-encoding decoded text drifts from the original ids, because detokenisation is not injective. Both reviewer counterexamples reproduce exactly and are now regression tests: * ``, `abc`, `<`, `x` -> ids [151667, 13683, 27, 87]; the parser counts 3, the old classifier counted 2. * generated id 77150 decodes to '１０', which RE-ENCODES to two tokens; the old retokenisation reported 2 for a single generated token. Also avoids synchronous tokenizer work on the async loop. Validated on Inkling-NVFP4 / GB300 (vLLM 0.28.0, MTP 8) on BOTH topologies -- 2-replica AGG and 1P1D disagg -- with the change overlaid on the shipped image. Reported reasoning_tokens against an independent tokenisation of the returned reasoning_content: ratio 1.00 on both, where the pre-fix code scored 0.02 on a non-streaming tool request. Accounting suite 15/15. Signed-off-by: Sungsoo Ha --- components/src/dynamo/frontend/prepost.py | 50 +++---- .../tests/test_vllm_processor_unit.py | 131 ++++++++++++++++-- 2 files changed, 143 insertions(+), 38 deletions(-) diff --git a/components/src/dynamo/frontend/prepost.py b/components/src/dynamo/frontend/prepost.py index 969f6a964212..60b24379eba3 100644 --- a/components/src/dynamo/frontend/prepost.py +++ b/components/src/dynamo/frontend/prepost.py @@ -714,18 +714,16 @@ def __init__( self._reasoning_parser_streaming_started = False self._tool_parser_streaming_started = False - # Reasoning tokens attributed to THIS choice, accumulated where the - # reasoning parser CLASSIFIES output -- deliberately not where the - # response is projected. Two projections would otherwise lose the count - # entirely (NVBug 6678449b): - # * `_suppress_reasoning_output` (include_reasoning=false) drops the - # parsed reasoning from the emitted delta, so a response with hidden - # reasoning would report zero; - # * the non-streaming tool path buffers every generated chunk and - # emits once on the terminal delta, so per-chunk observation sees - # only a zero-token terminal frame. - # Counting at classification is invariant to both. - self.reasoning_token_total = 0 + # Every ORIGINAL generated token id for this choice, in order (NVBug + # 6678449b). The count is delegated to the pinned vLLM parser's own + # `count_reasoning_tokens`, so it is never re-derived from parser output. + # Deriving it was wrong three ways: the response projections drop or + # defer it (`_suppress_reasoning_output`, and the non-streaming tool + # path's terminal-delta buffering); a per-chunk classifier over-counts a + # chunk carrying both kinds; and re-encoding decoded text drifts from the + # original ids, because detokenisation is not injective -- one generated + # id can re-encode to several. + self._reasoning_token_ids: list[int] = [] self._control_markers = tuple( t for t in getattr(tokenizer, "all_special_tokens", ()) if t @@ -746,6 +744,13 @@ def __init__( self._tool_text_buffer: str | None = None self._dynamo_json_fallback_chunks: dict[int, list[str]] = {} + @property + def reasoning_token_total(self) -> int: + """Reasoning tokens for this choice, per the parser's own counter.""" + if self.reasoning_parser is None: + return 0 + return self.reasoning_parser.count_reasoning_tokens(self._reasoning_token_ids) + def _decode_dynamo_json_fallback_tool_calls( self, text: str ) -> list[dict[str, Any]]: @@ -1085,15 +1090,6 @@ def _process_non_streaming_tool_output(self, output: Any) -> dict[str, Any] | No current_text, request=self.request_for_sampling, ) - # This path buffers every generated chunk and parses once here, so - # there is no per-chunk classification to count. The reasoning text - # IS known now, so count it exactly -- and do it before the - # include_reasoning projection on the next line, which would - # otherwise hide it. - if saved_reasoning: - self.reasoning_token_total += len( - self.tokenizer.encode(saved_reasoning, add_special_tokens=False) - ) if not self.request_for_sampling.include_reasoning: saved_reasoning = None @@ -1119,6 +1115,10 @@ def _process_non_streaming_tool_output(self, output: Any) -> dict[str, Any] | No return self._build_choice(output, delta) def process_output(self, output: Any) -> dict[str, Any] | None: + # BEFORE any branch below: every path must feed the ledger, including + # the ones that emit nothing for this chunk. + if self.reasoning_parser is not None: + self._reasoning_token_ids.extend(output.token_ids or []) if self._uses_dynamo_json_tool_call_fallback: return self._process_dynamo_json_fallback_tool_calls(output) if self._should_buffer_for_non_streaming_tool_parse(): @@ -1286,14 +1286,6 @@ def process_output(self, output: Any) -> dict[str, Any] | None: ) self._tool_parser_streaming_started = False - # Account BEFORE the projection below. `delta_message` is the parser's - # classification; `reasoning` is what survives include_reasoning. Reading - # the latter is what reported zero for hidden reasoning. - # Chunk-granular, matching the Rust estimator: a chunk carrying both - # reasoning and visible content counts entirely as reasoning. - if delta_message is not None and delta_message.reasoning: - self.reasoning_token_total += len(delta_token_ids) - choice = None if delta_message is None: if self.in_progress_tool_calls: diff --git a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py index 071b92c5d02f..67384313620b 100644 --- a/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py +++ b/components/src/dynamo/frontend/tests/test_vllm_processor_unit.py @@ -2579,6 +2579,21 @@ def extract_reasoning(self, text, request=None): return reasoning, content return text, "" + def count_reasoning_tokens(self, token_ids): + """Everything up to and including the end marker counts as reasoning. + + The production code delegates to this, so the fake has to provide it -- + a fake that omits it inherits ReasoningParser's base, which returns 0. + """ + end = self.tokenizer.encode("", add_special_tokens=False) + end_id = end[-1] if end else None + count = 0 + for tid in token_ids: + count += 1 + if tid == end_id: + break + return count + class _PassthroughToolParser: """Finds no tool calls; returns the text as content. @@ -2642,12 +2657,21 @@ def _annotator(*posts): # -- streaming classification ------------------------------------------ - def test_streaming_counts_reasoning_chunks_only(self, tokenizer): - post = self._post(tokenizer) - post.process_output(self._out("thinking", [1, 2, 3])) - post.process_output(self._out("more", [4, 5])) - post.process_output(self._out("answer", [6, 7, 8, 9])) - assert post.reasoning_token_total == 5 + def test_streaming_counts_reasoning_only_not_the_answer(self, tokenizer): + """Real parser, real ids: the visible answer must not be counted.""" + from vllm.reasoning import ReasoningParserManager + + reasoning = tokenizer.encode("abc", add_special_tokens=False) + answer = tokenizer.encode("the answer", add_special_tokens=False) + native = ReasoningParserManager.get_reasoning_parser("qwen3")( + tokenizer + ).count_reasoning_tokens(reasoning + answer) + + post = self._qwen3_post(tokenizer) + for tid in reasoning + answer: + post.process_output(self._out(tokenizer.decode([tid]), [tid])) + assert post.reasoning_token_total == native + assert 0 < native < len(reasoning + answer) # -- P1 #1: account before the include_reasoning projection ------------ @@ -2692,11 +2716,100 @@ def test_non_streaming_buffering_counts_even_when_suppressed(self, tokenizer): post.process_output(self._out("answer", [3], finish_reason="stop")) assert post.reasoning_token_total > 0 - def test_content_only_counts_nothing(self, tokenizer): - post = self._post(tokenizer) - post.process_output(self._out("hi", [1, 2, 3, 4])) + def test_no_reasoning_parser_counts_nothing(self, tokenizer): + """Without a reasoning parser there is nothing to delegate to -> 0. + + Note the counterpart is NOT true: with a parser, plain content is not + automatically zero. Qwen3's parser counts an unopened span as reasoning + (the model is trained to reason first), so delegating means inheriting + that contract -- which is the point, since it is what vLLM itself + reports. + """ + request = SimpleNamespace( + include_reasoning=True, tool_choice="none", model_fields=frozenset() + ) + post = StreamingPostProcessor( + tokenizer=tokenizer, + request_for_sampling=request, + sampling_params=SamplingParams(), + prompt_token_ids=[], + tool_parser=None, + reasoning_parser_class=None, + chat_template_kwargs={}, + stream_response=True, + ) + for tid in tokenizer.encode("just an answer", add_special_tokens=False): + post.process_output(self._out(tokenizer.decode([tid]), [tid])) assert post.reasoning_token_total == 0 + # -- the parser's OWN counter, not a re-derivation ----------------------- + + @staticmethod + def _qwen3_post(tokenizer): + from vllm.reasoning import ReasoningParserManager + + request = SimpleNamespace( + include_reasoning=True, tool_choice="none", model_fields=frozenset() + ) + return StreamingPostProcessor( + tokenizer=tokenizer, + request_for_sampling=request, + sampling_params=SamplingParams(), + prompt_token_ids=[], + tool_parser=None, + reasoning_parser_class=ReasoningParserManager.get_reasoning_parser("qwen3"), + chat_template_kwargs={}, + stream_response=True, + ) + + def test_count_matches_the_parser_for_split_marker_tokens(self, tokenizer): + """``, `abc`, `<`, `x` -- the parser counts 3. + + A per-chunk classifier that adds len(delta_token_ids) whenever the chunk + carried reasoning reported 2 here. Delegating to the parser's own + count_reasoning_tokens is what makes this exact. + """ + from vllm.reasoning import ReasoningParserManager + + ids = [ + i + for piece in ("", "abc", "<", "x") + for i in tokenizer.encode(piece, add_special_tokens=False) + ] + native = ReasoningParserManager.get_reasoning_parser("qwen3")( + tokenizer + ).count_reasoning_tokens(ids) + assert native == 3, f"fixture drift: parser now counts {native}" + + post = self._qwen3_post(tokenizer) + for tid in ids: + post.process_output(self._out(tokenizer.decode([tid]), [tid])) + assert post.reasoning_token_total == native == 3 + + def test_count_survives_non_injective_detokenisation(self, tokenizer): + """Generated id 77150 decodes to '１０', which RE-ENCODES to two tokens. + + Counting by re-encoding the decoded reasoning text therefore reported 2 + for a single generated token. The ledger keeps the original ids, so the + parser sees exactly what was generated. + """ + think = tokenizer.encode("", add_special_tokens=False) + ids = think + [77150] + assert ( + len(tokenizer.encode(tokenizer.decode([77150]), add_special_tokens=False)) + == 2 + ), "fixture drift: 77150 no longer re-encodes to 2 tokens" + + from vllm.reasoning import ReasoningParserManager + + native = ReasoningParserManager.get_reasoning_parser("qwen3")( + tokenizer + ).count_reasoning_tokens(ids) + post = self._qwen3_post(tokenizer) + for tid in ids: + post.process_output(self._out(tokenizer.decode([tid]), [tid])) + assert post.reasoning_token_total == native == 1 + # -- annotation --------------------------------------------------------- def test_total_sums_across_choices(self, tokenizer):