From 20a64afb8311501334ea4cd8975df9c52800ad08 Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Fri, 23 Oct 2026 04:32:08 +0000 Subject: [PATCH 1/4] [None][fix] Anthropic Messages: read the chat stream to its end so its metrics are recorded A streamed /v1/messages request is served by the chat stream and reframed into Anthropic events. The reframer stopped reading at [DONE], so the chat stream stayed suspended at its last yield and was closed with GeneratorExit: the code it runs after [DONE], which records the request's metrics (perf metrics and the Prometheus request metrics), never ran, and its bare `except:` logged the GeneratorExit as an error traceback. Every streamed Messages request (every Claude Code turn) was missing from the metrics and logged a traceback. The reframer now finishes the Anthropic stream at [DONE] as before and then reads the upstream to its end, dropping anything after [DONE]. A relayed byte stream from a disaggregated worker likewise ends at the worker's EOF. The chat stream logs only exceptions: a consumer that stops reading (GeneratorExit) or a client that disconnects (CancelledError) is not an error of the stream. Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- tensorrt_llm/serve/anthropic_adapter.py | 13 +++++- tensorrt_llm/serve/openai_server.py | 5 +- .../llmapi/apps/test_anthropic_adapter.py | 46 +++++++++++++++++++ 3 files changed, 62 insertions(+), 2 deletions(-) diff --git a/tensorrt_llm/serve/anthropic_adapter.py b/tensorrt_llm/serve/anthropic_adapter.py index c03928825e1b..3b3b53277dcf 100644 --- a/tensorrt_llm/serve/anthropic_adapter.py +++ b/tensorrt_llm/serve/anthropic_adapter.py @@ -1062,8 +1062,16 @@ async def reframe_openai_stream( ) -> AsyncIterator[str]: """Translate an OpenAI SSE stream into Anthropic SSE frames.""" reframer = AnthropicStreamReframer(model=model) + done = False try: async for line in _iter_openai_sse_lines(openai_sse): + if done: + # The upstream is read to its end, not abandoned at [DONE]: the + # chat stream records the request's metrics after its last + # frame, and a stream left suspended at [DONE] is closed instead + # (GeneratorExit), so that work never runs. A relayed byte + # stream likewise ends cleanly at the worker's EOF. + continue line = line.strip() if not line.startswith("data:"): continue @@ -1073,13 +1081,16 @@ async def reframe_openai_stream( if data == "[DONE]": for frame in reframer.finish(): yield frame - return + done = True + continue try: chunk = ChatCompletionStreamResponse(**json.loads(data)) except (json.JSONDecodeError, ValueError) as e: raise AnthropicResponseError("Malformed upstream stream chunk") from e for frame in reframer.process_chunk(chunk): yield frame + if done: + return # Upstream ended without [DONE]. Both streaming producers emit it only # after the response is complete, so its absence means the stream was # cut short. Calling finish() here would close the message with diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 547541f4b63f..903aa62bfce4 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -2033,7 +2033,10 @@ async def chat_stream_generator( yield "data: [DONE]\n\n" await self._extract_metrics(res, raw_request) nvtx_mark("generation ends") - except: + except Exception: + # Only failures are logged: a consumer that stops reading closes + # the stream with GeneratorExit, and a client that disconnects + # cancels it, and neither is an error of this stream. logger.error(traceback.format_exc()) raise diff --git a/tests/unittest/llmapi/apps/test_anthropic_adapter.py b/tests/unittest/llmapi/apps/test_anthropic_adapter.py index 2e9bb12fc656..a56b95363abd 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_adapter.py +++ b/tests/unittest/llmapi/apps/test_anthropic_adapter.py @@ -1132,6 +1132,52 @@ async def collect(): assert "".join(text_deltas) == "hello" +def test_stream_reframer_reads_the_upstream_to_its_end_after_done(): + """The upstream runs past [DONE], so the work it does there runs. + + The chat stream records the request's metrics after its last frame. A + reframer that stopped reading at [DONE] left it suspended there, to be + closed with GeneratorExit: every streamed /v1/messages request lost its + metrics, and the server logged a traceback for it. + """ + after_done = [] + + async def source(): + yield f"data: {chunk({'content': 'hello'}).model_dump_json()}\n\n" + yield f"data: {chunk({}, finish_reason='stop').model_dump_json()}\n\n" + yield "data: [DONE]\n\n" + after_done.append("ran") + + async def collect(): + return [frame async for frame in reframe_openai_stream(source(), MODEL)] + + events = parse_frames(asyncio.run(collect())) + assert_event_invariants(events) + assert after_done == ["ran"] + + +def test_stream_reframer_drops_anything_after_done(): + """Lines after [DONE] are read to reach the end, never reframed.""" + + async def source(): + yield f"data: {chunk({'content': 'hello'}).model_dump_json()}\n\n" + yield f"data: {chunk({}, finish_reason='stop').model_dump_json()}\n\n" + yield "data: [DONE]\n\n" + yield f"data: {chunk({'content': 'late'}).model_dump_json()}\n\n" + + async def collect(): + return [frame async for frame in reframe_openai_stream(source(), MODEL)] + + events = parse_frames(asyncio.run(collect())) + assert_event_invariants(events) + text_deltas = [ + payload["delta"]["text"] + for name, payload in events + if name == "content_block_delta" and payload["delta"]["type"] == "text_delta" + ] + assert "".join(text_deltas) == "hello" + + def test_stream_reframer_splits_multibyte_characters(): """A chunk boundary landing inside a UTF-8 sequence must not corrupt output. From 9b00debc190b9759777c72be9b52a57a7a31c0b7 Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Fri, 23 Oct 2026 05:34:46 +0000 Subject: [PATCH 2/4] [None][fix] Anthropic Messages: add no event after message_stop when the drain fails Once [DONE] has been reframed into message_stop the message is closed, so a failure while reading the rest of the upstream is logged and the stream ends without an error event after message_stop. Tests: the upstream failing after [DONE], a byte stream whose [DONE] spans two chunks being read to its end, and the chat stream behind /v1/messages recording its metrics when read to its end, logging nothing when its consumer closes it or a disconnect cancels it, and still logging a real failure. Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- tensorrt_llm/serve/anthropic_adapter.py | 4 + .../llmapi/apps/test_anthropic_adapter.py | 40 +++++ .../llmapi/apps/test_anthropic_routes.py | 163 ++++++++++++++++++ 3 files changed, 207 insertions(+) diff --git a/tensorrt_llm/serve/anthropic_adapter.py b/tensorrt_llm/serve/anthropic_adapter.py index 3b3b53277dcf..2c896c65726c 100644 --- a/tensorrt_llm/serve/anthropic_adapter.py +++ b/tensorrt_llm/serve/anthropic_adapter.py @@ -1108,5 +1108,9 @@ async def reframe_openai_stream( yield frame except Exception as e: # noqa: BLE001 - stream must end with an event logger.error(f"Anthropic stream reframing failed: {e}\n{traceback.format_exc()}") + if done: + # message_stop has already closed the message, so a failure while + # reading the rest of the upstream is logged but adds no event. + return for frame in reframer.error("Internal server error"): yield frame diff --git a/tests/unittest/llmapi/apps/test_anthropic_adapter.py b/tests/unittest/llmapi/apps/test_anthropic_adapter.py index a56b95363abd..0a48fab2dd55 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_adapter.py +++ b/tests/unittest/llmapi/apps/test_anthropic_adapter.py @@ -1178,6 +1178,46 @@ async def collect(): assert "".join(text_deltas) == "hello" +def test_stream_reframer_adds_no_event_when_the_upstream_fails_after_done(): + """message_stop stays the last event when reading past [DONE] fails.""" + + async def source(): + yield f"data: {chunk({'content': 'hello'}).model_dump_json()}\n\n" + yield f"data: {chunk({}, finish_reason='stop').model_dump_json()}\n\n" + yield "data: [DONE]\n\n" + raise RuntimeError("failed after the last frame") + + async def collect(): + return [frame async for frame in reframe_openai_stream(source(), MODEL)] + + events = parse_frames(asyncio.run(collect())) + assert_event_invariants(events) + assert "error" not in [name for name, _ in events] + + +def test_stream_reframer_reads_a_byte_stream_to_its_end_when_done_is_split(): + """A relayed byte stream is read to its EOF however [DONE] is chunked.""" + body = ( + f"data: {chunk({'content': 'hello'}).model_dump_json()}\n\n" + f"data: {chunk({}, finish_reason='stop').model_dump_json()}\n\n" + "data: [DONE]\n\n" + ).encode() + split = body.index(b"[DONE]") + len(b"[DO") + after_last_chunk = [] + + async def source(): + yield body[:split] + yield body[split:] + after_last_chunk.append("ran") + + async def collect(): + return [frame async for frame in reframe_openai_stream(source(), MODEL)] + + events = parse_frames(asyncio.run(collect())) + assert_event_invariants(events) + assert after_last_chunk == ["ran"] + + def test_stream_reframer_splits_multibyte_characters(): """A chunk boundary landing inside a UTF-8 sequence must not corrupt output. diff --git a/tests/unittest/llmapi/apps/test_anthropic_routes.py b/tests/unittest/llmapi/apps/test_anthropic_routes.py index 6c04971f764b..1d23ba54f149 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_routes.py +++ b/tests/unittest/llmapi/apps/test_anthropic_routes.py @@ -27,8 +27,10 @@ from fastapi.testclient import TestClient from tensorrt_llm.inputs.chat_template_guard import UnusedChatTemplateKwargsError +from tensorrt_llm.serve import openai_server as openai_server_module from tensorrt_llm.serve.openai_disagg_server import OpenAIDisaggServer from tensorrt_llm.serve.openai_protocol import ( + ChatCompletionRequest, ChatCompletionResponse, ChatCompletionResponseChoice, ChatCompletionStreamResponse, @@ -38,6 +40,7 @@ UsageInfo, ) from tensorrt_llm.serve.openai_server import OpenAIServer +from tensorrt_llm.serve.render import chat as render_chat_module # These tests are CPU-only (no GPU, engine or sockets) and run in the # CPU-Generic CI stage, which selects with `-m cpu_only`. @@ -271,6 +274,166 @@ def test_messages_route_reframes_streaming_response(server_kind): assert backend.await_args.args[0].stream +class _StreamedGeneration: + """A streamed generation whose steps are already post-processed frames. + + After its steps it ends, raises ``then``, or (``hang=True``) waits forever + for the next step, as a generation does while the model is still running. + """ + + prompt_token_ids = [1, 2, 3] + + def __init__(self, steps, *, then=None, hang=False): + self._steps = list(steps) + self._then = then + self._hang = hang + + def __aiter__(self): + return self + + async def __anext__(self): + if self._steps: + frames = self._steps.pop(0) + return SimpleNamespace( + finished=not self._steps, + outputs=[SimpleNamespace(_postprocess_result=frames)], + ) + if self._then is not None: + raise self._then + if self._hang: + await asyncio.Event().wait() + raise StopAsyncIteration + + +class _ModelConfig: + model_type = "llama" + vocab_size = 1024 + + +async def _chat_stream(monkeypatch, generation): + """The body of a streamed /v1/chat/completions response, and the server. + + The real ``openai_chat`` handler runs on a server built with + ``object.__new__``; prompt rendering and the engine are stubbed, and the + engine's output is already post-processed into SSE frames. + """ + + async def render(**kwargs): + return "rendered prompt" + + monkeypatch.setattr(render_chat_module, "async_apply_chat_template", render) + server = object.__new__(OpenAIServer) + server.model = MODEL + server.allow_request_chat_template = False + server.model_config = _ModelConfig() + server.processor = None + server.tokenizer = SimpleNamespace(tokenizer=SimpleNamespace(vocab_size=1024)) + server.chat_template = None + server.tool_parser = None + server.tool_call_id_type = "random" + server.multimodal_server_config = None + server._is_visual_gen = False + server._adjusted_steady_clock = SimpleNamespace(now=time.monotonic) + server.generator = SimpleNamespace( + args=SimpleNamespace( + gather_generation_logits=False, + reasoning_parser=None, + backend="pytorch", + guided_decoding_backend=None, + num_postprocess_workers=1, + ), + generate_async=Mock(return_value=generation), + ) + server.await_disconnected = AsyncMock() + server._extract_metrics = AsyncMock() + request = ChatCompletionRequest( + model=MODEL, messages=[{"role": "user", "content": "hi"}], stream=True + ) + raw_request = SimpleNamespace(state=SimpleNamespace(), headers={}) + response = await server.openai_chat(request, raw_request) + assert isinstance(response, StreamingResponse), response.body + return response.body_iterator, server + + +@pytest.fixture +def server_logger(monkeypatch): + log = Mock() + monkeypatch.setattr(openai_server_module, "logger", log) + return log + + +def test_chat_stream_read_to_its_end_records_the_metrics(monkeypatch, server_logger): + """Past [DONE] the chat stream records the request's metrics.""" + + async def run(): + generation = _StreamedGeneration([["data: one\n\n"], ["data: two\n\n"]]) + stream, server = await _chat_stream(monkeypatch, generation) + frames = [frame async for frame in stream] + return frames, server + + frames, server = asyncio.run(run()) + + assert frames == ["data: one\n\n", "data: two\n\n", "data: [DONE]\n\n"] + server._extract_metrics.assert_awaited_once() + server_logger.error.assert_not_called() + + +def test_chat_stream_closed_by_its_consumer_logs_no_error(monkeypatch, server_logger): + """A consumer that stops reading (GeneratorExit) is not a stream error.""" + + async def run(): + generation = _StreamedGeneration([["data: one\n\n"], ["data: two\n\n"]]) + stream, server = await _chat_stream(monkeypatch, generation) + assert await stream.__anext__() == "data: one\n\n" + await stream.aclose() + return server + + server = asyncio.run(run()) + + server._extract_metrics.assert_not_awaited() + server_logger.error.assert_not_called() + + +def test_chat_stream_cancelled_by_a_disconnect_logs_no_error(monkeypatch, server_logger): + """A client that disconnects cancels the stream; that is not a stream error.""" + + async def run(): + generation = _StreamedGeneration([["data: one\n\n"]], hang=True) + stream, _ = await _chat_stream(monkeypatch, generation) + first_frame = asyncio.Event() + + async def consume(): + async for _ in stream: + first_frame.set() + + consumer = asyncio.create_task(consume()) + await first_frame.wait() + # The stream is now waiting on the generation for its next step. + consumer.cancel() + with pytest.raises(asyncio.CancelledError): + await consumer + + asyncio.run(run()) + + server_logger.error.assert_not_called() + + +def test_chat_stream_logs_a_failure_and_raises_it(monkeypatch, server_logger): + """A failure of the generation itself is still logged and raised.""" + + async def run(): + generation = _StreamedGeneration([["data: one\n\n"]], then=RuntimeError("engine failed")) + stream, _ = await _chat_stream(monkeypatch, generation) + with pytest.raises(RuntimeError, match="engine failed"): + async for _ in stream: + pass + + asyncio.run(run()) + + server_logger.error.assert_called_once() + assert "engine failed" in server_logger.error.call_args.args[0] + + def test_standard_and_disagg_register_messages_route(monkeypatch, tmp_path): # The disagg server's register_routes mounts a prometheus multiprocess # collector, which errors unless PROMETHEUS_MULTIPROC_DIR points at a real From 052f0c1781e56eec02678c6282214dfe7491a755 Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Fri, 23 Oct 2026 05:49:57 +0000 Subject: [PATCH 3/4] [None][fix] Anthropic Messages: bound the wait for the first frame in the cancellation test A stream that failed before its first frame would leave the test waiting forever; it now fails after five seconds. Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- tests/unittest/llmapi/apps/test_anthropic_routes.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/unittest/llmapi/apps/test_anthropic_routes.py b/tests/unittest/llmapi/apps/test_anthropic_routes.py index 1d23ba54f149..7c2df325f79f 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_routes.py +++ b/tests/unittest/llmapi/apps/test_anthropic_routes.py @@ -407,7 +407,9 @@ async def consume(): first_frame.set() consumer = asyncio.create_task(consume()) - await first_frame.wait() + # Bounded, so a stream that fails before its first frame fails the + # test instead of hanging it. + await asyncio.wait_for(first_frame.wait(), timeout=5.0) # The stream is now waiting on the generation for its next step. consumer.cancel() with pytest.raises(asyncio.CancelledError): From 5bc273ab1d2cc9f5323ab7f64aea73d79b951f8d Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:34:27 +0000 Subject: [PATCH 4/4] [None][fix] Anthropic Messages: close the message only once the upstream has ended The frames that close the message (message_delta and message_stop) are now held from [DONE] until the upstream has been read to its end. A client may disconnect as soon as it has message_stop; had message_stop gone out first, the chat stream would be left suspended at [DONE] and its metrics never recorded. If reading past [DONE] fails, the failure is logged and the message still closes. Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- tensorrt_llm/serve/anthropic_adapter.py | 25 ++++++++++------ .../llmapi/apps/test_anthropic_adapter.py | 30 +++++++++++++++++++ 2 files changed, 46 insertions(+), 9 deletions(-) diff --git a/tensorrt_llm/serve/anthropic_adapter.py b/tensorrt_llm/serve/anthropic_adapter.py index 2c896c65726c..ff80c53eec6a 100644 --- a/tensorrt_llm/serve/anthropic_adapter.py +++ b/tensorrt_llm/serve/anthropic_adapter.py @@ -1063,14 +1063,18 @@ async def reframe_openai_stream( """Translate an OpenAI SSE stream into Anthropic SSE frames.""" reframer = AnthropicStreamReframer(model=model) done = False + # The frames that close the message, held from [DONE] until the upstream + # has been read to its end. + closing_frames: List[str] = [] try: async for line in _iter_openai_sse_lines(openai_sse): if done: - # The upstream is read to its end, not abandoned at [DONE]: the - # chat stream records the request's metrics after its last - # frame, and a stream left suspended at [DONE] is closed instead - # (GeneratorExit), so that work never runs. A relayed byte - # stream likewise ends cleanly at the worker's EOF. + # The upstream is read to its end before the message is closed: + # the chat stream records the request's metrics after its last + # frame, and a consumer that stops reading at message_stop would + # leave it suspended at [DONE], to be closed there (GeneratorExit) + # without doing that work. A relayed byte stream likewise ends + # cleanly at the worker's EOF. continue line = line.strip() if not line.startswith("data:"): @@ -1079,8 +1083,7 @@ async def reframe_openai_stream( if not data: continue if data == "[DONE]": - for frame in reframer.finish(): - yield frame + closing_frames = reframer.finish() done = True continue try: @@ -1090,6 +1093,8 @@ async def reframe_openai_stream( for frame in reframer.process_chunk(chunk): yield frame if done: + for frame in closing_frames: + yield frame return # Upstream ended without [DONE]. Both streaming producers emit it only # after the response is complete, so its absence means the stream was @@ -1109,8 +1114,10 @@ async def reframe_openai_stream( except Exception as e: # noqa: BLE001 - stream must end with an event logger.error(f"Anthropic stream reframing failed: {e}\n{traceback.format_exc()}") if done: - # message_stop has already closed the message, so a failure while - # reading the rest of the upstream is logged but adds no event. + # The response was complete at [DONE], so a failure while reading + # the rest of the upstream is logged and the message still closes. + for frame in closing_frames: + yield frame return for frame in reframer.error("Internal server error"): yield frame diff --git a/tests/unittest/llmapi/apps/test_anthropic_adapter.py b/tests/unittest/llmapi/apps/test_anthropic_adapter.py index 0a48fab2dd55..bbbc80254eed 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_adapter.py +++ b/tests/unittest/llmapi/apps/test_anthropic_adapter.py @@ -1156,6 +1156,36 @@ async def collect(): assert after_done == ["ran"] +def test_stream_reframer_reads_the_upstream_to_its_end_before_message_stop(): + """A consumer that stops reading at message_stop still lets the upstream end. + + A client may disconnect as soon as it has message_stop. Had message_stop + gone out before the upstream was drained, the chat stream would be left + suspended at [DONE] and its metrics never recorded. + """ + after_done = [] + + async def source(): + yield f"data: {chunk({'content': 'hello'}).model_dump_json()}\n\n" + yield f"data: {chunk({}, finish_reason='stop').model_dump_json()}\n\n" + yield "data: [DONE]\n\n" + after_done.append("ran") + + async def read_until_message_stop(): + stream = reframe_openai_stream(source(), MODEL) + frames = [] + async for frame in stream: + frames.append(frame) + if frame.startswith("event: message_stop"): + break + await stream.aclose() + return frames + + events = parse_frames(asyncio.run(read_until_message_stop())) + assert_event_invariants(events) + assert after_done == ["ran"] + + def test_stream_reframer_drops_anything_after_done(): """Lines after [DONE] are read to reach the end, never reframed."""