From a5690c9cb90ad3355080af6d9863fcfb9b53d89a Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Tue, 27 Oct 2026 12:05:38 +0000 Subject: [PATCH 1/4] [None][fix] Anthropic Messages: answer an over-length prompt with "prompt is too long", before any stream Anthropic rejects a prompt longer than the context window with HTTP 400 invalid_request_error, "prompt is too long: N tokens > M maximum", before any stream, and Claude Code compacts its conversation on that wording. /v1/messages answered in OpenAI's wording instead, and a streamed request on the default deployment (executor in worker processes) got HTTP 200 and an "Internal server error" event: the rejection reaches the server only as the failure of the stream's first frame. - context_length_exceeded_counts reads the context and prompt lengths from the server's rejection text. - prompt_too_long_message words it as Anthropic does. M is the longest prompt the server accepts (the context length minus one), so N > M also holds for a prompt that fills the context exactly. - anthropic_chat_error_response maps a chat-pipeline failure to the Anthropic error, over-length to that 400, everything else as before; both servers' /v1/messages handlers use it. - wait_for_stream_start: a streamed request waits up to 1 s for the first frame before its response starts, so a rejection becomes the 400. A slow first token is not waited for; a rejection after the wait ends the stream with an error event in the same wording. Other routes keep their wording and codes. Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- tensorrt_llm/executor/utils.py | 16 +- tensorrt_llm/serve/anthropic_adapter.py | 95 ++++++ tensorrt_llm/serve/openai_disagg_server.py | 13 +- tensorrt_llm/serve/openai_server.py | 17 +- .../llmapi/apps/test_anthropic_routes.py | 295 +++++++++++++++++- 5 files changed, 414 insertions(+), 22 deletions(-) diff --git a/tensorrt_llm/executor/utils.py b/tensorrt_llm/executor/utils.py index f488a03682dc..248af5e627ee 100644 --- a/tensorrt_llm/executor/utils.py +++ b/tensorrt_llm/executor/utils.py @@ -155,8 +155,20 @@ def context_length_exceeded_message(max_context_length: int, _CONTEXT_LENGTH_EXCEEDED_RE = re.compile( - r"This model's maximum context length is \d+ tokens\. " - r"However, your messages resulted in \d+ tokens\.") + r"This model's maximum context length is (\d+) tokens\. " + r"However, your messages resulted in (\d+) tokens\.") + + +def context_length_exceeded_counts(message: str) -> tuple[int, int] | None: + """``(max_context_length, num_prompt_tokens)`` of a ``context_length_exceeded_message``. + + None if ``message`` contains none. Searched for like + ``is_context_length_exceeded_message`` does. + """ + match = _CONTEXT_LENGTH_EXCEEDED_RE.search(message) + if match is None: + return None + return int(match.group(1)), int(match.group(2)) def is_context_length_exceeded_message(message: str) -> bool: diff --git a/tensorrt_llm/serve/anthropic_adapter.py b/tensorrt_llm/serve/anthropic_adapter.py index cc07ae2f1b51..3b9d25126319 100644 --- a/tensorrt_llm/serve/anthropic_adapter.py +++ b/tensorrt_llm/serve/anthropic_adapter.py @@ -24,6 +24,7 @@ chat templates, or the engine. """ +import asyncio import base64 import hashlib import json @@ -35,6 +36,7 @@ from fastapi.responses import JSONResponse +from tensorrt_llm.executor.utils import RequestError, context_length_exceeded_counts from tensorrt_llm.logger import logger from tensorrt_llm.serve.anthropic_protocol import ( AnthropicContentBlockDeltaEvent, @@ -95,6 +97,99 @@ def anthropic_error_response( return JSONResponse(content=envelope.model_dump(exclude_none=True), status_code=status_code) +def prompt_too_long_message(message: str) -> Optional[str]: + """Anthropic's wording for this server's over-length rejection, or None. + + Anthropic answers a prompt longer than the context window with an + ``invalid_request_error`` that reads ``prompt is too long: N tokens > M + maximum``, and Claude Code compacts its conversation on that wording. The + server's own rejection (``context_length_exceeded_message``) is recognized + anywhere in ``message``. M is the longest prompt the server accepts, one + token below its context length, so the inequality also holds for a prompt + that fills the context exactly. + """ + counts = context_length_exceeded_counts(message) + if counts is None: + return None + max_context_length, num_prompt_tokens = counts + return f"prompt is too long: {num_prompt_tokens} tokens > {max_context_length - 1} maximum" + + +def anthropic_chat_error_response(message: str, status_code: int) -> JSONResponse: + """The Anthropic error for a failure the chat pipeline reported. + + A 4xx is the client's ``invalid_request_error`` and anything else an + ``api_error``. A prompt longer than the context is answered as Anthropic + answers it: a 400 ``invalid_request_error`` in Anthropic's wording. + """ + too_long = prompt_too_long_message(message) + if too_long is not None: + return anthropic_error_response(too_long, "invalid_request_error", 400) + error_type = "invalid_request_error" if 400 <= status_code < 500 else "api_error" + return anthropic_error_response(message, error_type, status_code) + + +# Seconds a streamed request is held before its response starts, for the +# engine's verdict on its prompt. The engine rejects a prompt longer than its +# context when it admits the request, well within this; a first token can take +# far longer, and the stream then starts without waiting for it. +STREAM_ADMISSION_WAIT_S = 1.0 + + +async def wait_for_stream_start( + openai_sse: AsyncIterator[Union[str, bytes]], +) -> Union[JSONResponse, AsyncIterator[Union[str, bytes]]]: + """Wait up to ``STREAM_ADMISSION_WAIT_S`` for the first frame of a chat stream. + + An engine in a worker process rejects a prompt longer than its context only + as the failure of the stream's first frame, when a response started at once + would already be under way. Anthropic answers such a prompt with HTTP 400 + before any stream, and Claude Code compacts its conversation on that error, + so the rejection is returned here as that 400. Otherwise the stream is + returned with its first frame, to be read as before; a rejection that + arrives after the wait ends the stream with an error event in the same + wording. + """ + first = asyncio.ensure_future(openai_sse.__anext__()) + try: + await asyncio.wait((first,), timeout=STREAM_ADMISSION_WAIT_S) + except asyncio.CancelledError: + first.cancel() + raise + if first.done() and not first.cancelled() and isinstance(first.exception(), RequestError): + too_long = prompt_too_long_message(str(first.exception())) + if too_long is not None: + return anthropic_error_response(too_long, "invalid_request_error", 400) + return _resume_stream(first, openai_sse) + + +async def _resume_stream( + first: asyncio.Future, rest: AsyncIterator[Union[str, bytes]] +) -> AsyncIterator[Union[str, bytes]]: + """Yield the frame ``first`` reads from ``rest``, then the remaining frames of ``rest``. + + Cancelling a read that waits on ``first`` cancels ``first`` too, and + closing this stream closes ``rest``. + """ + try: + try: + frame = await first + except StopAsyncIteration: + return + except RequestError as e: + too_long = prompt_too_long_message(str(e)) + if too_long is None: + raise + raise AnthropicResponseError(too_long) from e + yield frame + async for frame in rest: + yield frame + finally: + aclose = getattr(rest, "aclose", None) + if aclose is not None: + await aclose() + + # --------------------------------------------------------------------------- # Request conversion: Anthropic -> ChatCompletionRequest # --------------------------------------------------------------------------- diff --git a/tensorrt_llm/serve/openai_disagg_server.py b/tensorrt_llm/serve/openai_disagg_server.py index 686612b2aa81..f34bb60e56e8 100644 --- a/tensorrt_llm/serve/openai_disagg_server.py +++ b/tensorrt_llm/serve/openai_disagg_server.py @@ -43,6 +43,7 @@ from tensorrt_llm.serve._telemetry import create_uvicorn_server from tensorrt_llm.serve.anthropic_adapter import (AnthropicRequestError, AnthropicResponseError, + anthropic_chat_error_response, anthropic_error_response, convert_anthropic_request, convert_chat_response, @@ -472,10 +473,7 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest, openai_response = await self._wrap_entry_point( self._service.openai_chat_completion)(chat_request, raw_request) except HTTPException as e: - error_type = ("invalid_request_error" - if 400 <= e.status_code < 500 else "api_error") - return anthropic_error_response(str(e.detail), error_type, - e.status_code) + return anthropic_chat_error_response(str(e.detail), e.status_code) if isinstance(openai_response, StreamingResponse): return StreamingResponse( @@ -492,9 +490,10 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest, or payload.get("error") or json.dumps(payload)) except (json.JSONDecodeError, AttributeError, TypeError): message = "Internal server error" - error_type = ("invalid_request_error" - if 400 <= status < 500 else "api_error") - return anthropic_error_response(str(message), error_type, status) + # The context phase never streams, so a context worker's over-length + # rejection arrives here, before any stream, for a streamed request + # too. + return anthropic_chat_error_response(str(message), status) try: chat_response = ChatCompletionResponse( diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 8c4ba0f9315a..7ebf244652f7 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -68,9 +68,10 @@ get_effective_kv_cache_event_hash_algo from tensorrt_llm.sampling_params import GuidedDecodingParams, SamplingParams from tensorrt_llm.serve.anthropic_adapter import ( - AnthropicRequestError, AnthropicResponseError, anthropic_error_response, + AnthropicRequestError, AnthropicResponseError, + anthropic_chat_error_response, anthropic_error_response, convert_anthropic_count_tokens_request, convert_anthropic_request, - convert_chat_response, reframe_openai_stream) + convert_chat_response, reframe_openai_stream, wait_for_stream_start) from tensorrt_llm.serve.anthropic_batches import (AnthropicBatchStore, BatchStoreFullError, results_to_jsonl) @@ -2550,9 +2551,13 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest, "api_error", 500) if isinstance(response, StreamingResponse): + # A prompt the engine rejects as too long is answered with a 400, + # not with a stream (see wait_for_stream_start). + stream = await wait_for_stream_start(response.body_iterator) + if isinstance(stream, JSONResponse): + return stream return StreamingResponse( - content=reframe_openai_stream(response.body_iterator, - model=self.model), + content=reframe_openai_stream(stream, model=self.model), media_type="text/event-stream", ) @@ -2563,9 +2568,7 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest, message = payload.get("message") or json.dumps(payload) except (json.JSONDecodeError, AttributeError): message = "Internal server error" - err_type = ("invalid_request_error" - if 400 <= status < 500 else "api_error") - return anthropic_error_response(message, err_type, status) + return anthropic_chat_error_response(message, status) try: chat_response = ChatCompletionResponse(**json.loads(response.body)) diff --git a/tests/unittest/llmapi/apps/test_anthropic_routes.py b/tests/unittest/llmapi/apps/test_anthropic_routes.py index b7e6cf62a3d6..5bc03023f450 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_routes.py +++ b/tests/unittest/llmapi/apps/test_anthropic_routes.py @@ -17,17 +17,20 @@ import asyncio import json import time +from http import HTTPStatus from types import SimpleNamespace from unittest.mock import AsyncMock, Mock import aiohttp import pytest -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse, StreamingResponse from fastapi.testclient import TestClient +from tensorrt_llm.executor.utils import RequestError, context_length_exceeded_message from tensorrt_llm.inputs.chat_template_guard import UnusedChatTemplateKwargsError from tensorrt_llm.serve import openai_server as openai_server_module +from tensorrt_llm.serve.anthropic_protocol import AnthropicMessagesRequest from tensorrt_llm.serve.openai_disagg_server import OpenAIDisaggServer from tensorrt_llm.serve.openai_protocol import ( ChatCompletionRequest, @@ -324,12 +327,12 @@ class _ModelConfig: vocab_size = 1024 -async def _chat_stream(monkeypatch, generation): - """The body of a streamed /v1/chat/completions response, and the server. +def _chat_server(monkeypatch, generation): + """A server whose real chat handlers run over ``generation``. - 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. + The server is 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): @@ -360,6 +363,12 @@ async def render(**kwargs): ) server.await_disconnected = AsyncMock() server._extract_metrics = AsyncMock() + return server + + +async def _chat_stream(monkeypatch, generation): + """The body of a streamed /v1/chat/completions response, and the server.""" + server = _chat_server(monkeypatch, generation) request = ChatCompletionRequest( model=MODEL, messages=[{"role": "user", "content": "hi"}], stream=True ) @@ -450,6 +459,280 @@ async def run(): assert "engine failed" in server_logger.error.call_args.args[0] +# --------------------------------------------------------------------------- +# A prompt longer than the context +# --------------------------------------------------------------------------- + +# The engine's rejection of a 210000-token prompt against a 200000-token context. +_OVER_LENGTH = context_length_exceeded_message(max_context_length=200000, num_prompt_tokens=210000) + +# Anthropic's answer to such a prompt, the one Claude Code compacts its history on. +_PROMPT_TOO_LONG = { + "type": "error", + "error": { + "type": "invalid_request_error", + "message": "prompt is too long: 210000 tokens > 199999 maximum", + }, +} + + +class _GatedGeneration(_StreamedGeneration): + """A ``_StreamedGeneration`` whose first step waits for ``gate``, like a slow prefill.""" + + def __init__(self, steps, gate, **kwargs): + super().__init__(steps, **kwargs) + self._gate = gate + + async def __anext__(self): + await self._gate.wait() + return await super().__anext__() + + +def _reply_steps(): + """A one-word reply as post-processed chat frames, one engine step each.""" + chunks = [ + ChatCompletionStreamResponse( + model=MODEL, + choices=[{"index": 0, "delta": {"role": "assistant", "content": "hello"}}], + ), + ChatCompletionStreamResponse( + model=MODEL, + choices=[{"index": 0, "delta": {}, "finish_reason": "stop"}], + usage=UsageInfo(prompt_tokens=3, completion_tokens=1, total_tokens=4), + ), + ] + return [[f"data: {chunk.model_dump_json()}\n\n"] for chunk in chunks] + + +async def _messages_response(server, stream): + """The response of ``server`` to a /v1/messages request, its body not read yet.""" + return await server.anthropic_messages( + AnthropicMessagesRequest(**_request(stream=stream)), + SimpleNamespace(state=SimpleNamespace(), headers={}), + ) + + +async def _read(response): + """The body of ``response``, read to its end.""" + if isinstance(response, StreamingResponse): + return "".join([frame async for frame in response.body_iterator]) + return response.body.decode() + + +async def _messages(server, stream): + """The response of ``server`` to a /v1/messages request, and its body.""" + response = await _messages_response(server, stream) + return response, await _read(response) + + +@pytest.fixture +def short_admission_wait(monkeypatch): + """Shorten how long a streamed request waits for a rejection before it starts.""" + monkeypatch.setattr( + "tensorrt_llm.serve.anthropic_adapter.STREAM_ADMISSION_WAIT_S", 0.01, raising=False + ) + + +def _rejecting_engine(where, stream): + """``generate_async`` of an engine that rejects the prompt as too long. + + An engine in this process rejects it when the request is submitted; one in + a worker process (the default) rejects it as the first result, which a + streamed request reads as the failure of its first frame. + """ + error = RequestError(_OVER_LENGTH) + if where == "in_process": + return Mock(side_effect=error) + if stream: + return Mock(return_value=_StreamedGeneration([], then=error)) + return Mock( + return_value=SimpleNamespace( + prompt_token_ids=[1, 2, 3], aresult=AsyncMock(side_effect=error) + ) + ) + + +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("where", ["in_worker", "in_process"]) +def test_over_length_prompt_is_answered_as_anthropic_answers_it(monkeypatch, where, stream): + """An HTTP 400 invalid_request_error in Anthropic's wording, with no stream. + + Claude Code compacts its conversation when a request fails that way. A + stream that opens and then fails, or OpenAI's wording, reads to it as some + other error. + """ + server = _chat_server(monkeypatch, generation=None) + server.generator.generate_async = _rejecting_engine(where, stream) + + response, body = asyncio.run(_messages(server, stream)) + + assert not isinstance(response, StreamingResponse) + assert response.status_code == 400 + assert json.loads(body) == _PROMPT_TOO_LONG + + +def test_stream_starts_without_waiting_for_a_slow_first_token(monkeypatch, short_admission_wait): + """The wait for a rejection is bounded; a long prefill does not hold the stream back.""" + + async def run(): + gate = asyncio.Event() + server = _chat_server(monkeypatch, _GatedGeneration(_reply_steps(), gate)) + response = await _messages_response(server, stream=True) + # Started while the first token is still pending. + assert isinstance(response, StreamingResponse) + gate.set() + return await _read(response) + + body = asyncio.run(run()) + + assert body.startswith("event: message_start\n") + assert '"text":"hello"' in body + assert body.rstrip().endswith('data: {"type":"message_stop"}') + + +def test_rejection_after_the_wait_ends_the_stream_in_the_same_wording( + monkeypatch, short_admission_wait +): + """A rejection slower than the wait still says the prompt is too long.""" + + async def run(): + gate = asyncio.Event() + generation = _GatedGeneration([], gate, then=RequestError(_OVER_LENGTH)) + response = await _messages_response(_chat_server(monkeypatch, generation), stream=True) + assert isinstance(response, StreamingResponse) + gate.set() + return await _read(response) + + body = asyncio.run(run()) + + # The stream has started, so the error event has the type every stream failure has. + event, data = body.strip().split("\n") + assert event == "event: error" + assert json.loads(data[len("data: ") :])["error"] == { + "type": "api_error", + "message": _PROMPT_TOO_LONG["error"]["message"], + } + + +def test_other_failures_before_the_first_token_stay_in_the_stream(monkeypatch): + """Only an over-length rejection turns into a 400; other failures are reported as before.""" + generation = _StreamedGeneration([], then=RequestError("engine failed")) + server = _chat_server(monkeypatch, generation) + + response, body = asyncio.run(_messages(server, stream=True)) + + assert isinstance(response, StreamingResponse) + assert response.status_code == 200 + assert body == ( + "event: error\n" + 'data: {"type":"error","error":{"type":"api_error","message":"Internal server error"}}\n\n' + ) + + +def test_client_gone_before_the_first_token_leaves_nothing_running( + monkeypatch, server_logger, short_admission_wait +): + """A disconnect cancels the read the stream is waiting on, and logs no error.""" + + async def run(): + server = _chat_server(monkeypatch, _StreamedGeneration([], hang=True)) + response = await _messages_response(server, stream=True) + consumer = asyncio.ensure_future(response.body_iterator.__anext__()) + await asyncio.sleep(0.05) + consumer.cancel() + with pytest.raises(asyncio.CancelledError): + await consumer + return [task for task in asyncio.all_tasks() if task is not asyncio.current_task()] + + assert asyncio.run(run()) == [] + server_logger.error.assert_not_called() + + +@pytest.mark.parametrize( + ("message", "reply"), + [ + pytest.param( + context_length_exceeded_message(8192, 8192), + "prompt is too long: 8192 tokens > 8191 maximum", + id="prompt_fills_the_context", + ), + pytest.param( + f"Generation failed: {context_length_exceeded_message(8192, 9000)}", + "prompt is too long: 9000 tokens > 8191 maximum", + id="wrapped_rejection", + ), + ], +) +@pytest.mark.parametrize("server_kind", ["standard", "disagg"]) +def test_prompt_too_long_carries_the_counts(server_kind, message, reply): + """N and M are the prompt and the longest prompt the server accepts, so N > M holds.""" + client, _ = _make_route_client(server_kind, OpenAIServer.create_error_response(message)) + + response = client.post("/v1/messages", json=_request(stream=True)) + + assert response.status_code == 400 + assert response.json()["error"] == {"type": "invalid_request_error", "message": reply} + + +def _relayed_by_disagg_frontend(worker_response): + """What the disaggregated frontend hands its routes for a worker's error response.""" + frontend = object.__new__(OpenAIDisaggServer) + frontend._perf_metrics_collector = SimpleNamespace( + http_exceptions=SimpleNamespace(inc=Mock()), + internal_errors=SimpleNamespace(inc=Mock()), + ) + upstream = aiohttp.ClientResponseError( + request_info=Mock(), + history=(), + status=worker_response.status_code, + message=f"Bad Request: {worker_response.body.decode()}", + ) + try: + return frontend._handle_exception(upstream) + except HTTPException as e: + return e + + +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize( + ("message", "status", "error"), + [ + pytest.param(_OVER_LENGTH, 400, _PROMPT_TOO_LONG["error"], id="over_length"), + pytest.param( + "`max_tokens` (0) must be greater than 0", + 400, + {"type": "invalid_request_error", "message": "`max_tokens` (0) must be greater than 0"}, + id="other_client_error", + ), + pytest.param( + "engine overloaded", + 503, + {"type": "api_error", "message": "engine overloaded"}, + id="server_error", + ), + ], +) +def test_disagg_route_answers_a_worker_rejection(stream, message, status, error): + """A context worker's rejection reaches the client before any stream, streamed or not. + + The context phase never streams, so the frontend holds the worker's error + before it starts a response, as a response for an over-length prompt and + as an HTTPException for anything else. + """ + worker_response = OpenAIServer.create_error_response(message, status_code=HTTPStatus(status)) + relayed = _relayed_by_disagg_frontend(worker_response) + client, backend = _make_route_client("disagg", None) + if isinstance(relayed, HTTPException): + backend.side_effect = relayed + else: + backend.return_value = relayed + + response = client.post("/v1/messages", json=_request(stream=stream)) + + assert response.status_code == status + assert response.json() == {"type": "error", "error": error} + + @pytest.fixture def prometheus_multiproc_dir(tmp_path, monkeypatch): """PROMETHEUS_MULTIPROC_DIR for one test. From dbf4266323ac9a3f45e162415676474fe183ed37 Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Tue, 27 Oct 2026 12:11:20 +0000 Subject: [PATCH 2/4] [None][test] Anthropic Messages route tests: give the generation stand-in a request id A generation result carries a request_id. The non-streamed stand-in now has one too, so the over-length tests do not depend on which attributes of the result the chat handler reads. Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- tests/unittest/llmapi/apps/test_anthropic_routes.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unittest/llmapi/apps/test_anthropic_routes.py b/tests/unittest/llmapi/apps/test_anthropic_routes.py index 5bc03023f450..a17d5ce2ac52 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_routes.py +++ b/tests/unittest/llmapi/apps/test_anthropic_routes.py @@ -547,7 +547,7 @@ def _rejecting_engine(where, stream): return Mock(return_value=_StreamedGeneration([], then=error)) return Mock( return_value=SimpleNamespace( - prompt_token_ids=[1, 2, 3], aresult=AsyncMock(side_effect=error) + request_id=1, prompt_token_ids=[1, 2, 3], aresult=AsyncMock(side_effect=error) ) ) From 1c0637f825427fa7f217aaa280b8a9b9227e2472 Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:54:35 +0000 Subject: [PATCH 3/4] [None][test] Anthropic Messages: the admission-wait fixture fails if its constant is renamed Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- tests/unittest/llmapi/apps/test_anthropic_routes.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/unittest/llmapi/apps/test_anthropic_routes.py b/tests/unittest/llmapi/apps/test_anthropic_routes.py index a17d5ce2ac52..bf475fbcf00f 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_routes.py +++ b/tests/unittest/llmapi/apps/test_anthropic_routes.py @@ -528,9 +528,7 @@ async def _messages(server, stream): @pytest.fixture def short_admission_wait(monkeypatch): """Shorten how long a streamed request waits for a rejection before it starts.""" - monkeypatch.setattr( - "tensorrt_llm.serve.anthropic_adapter.STREAM_ADMISSION_WAIT_S", 0.01, raising=False - ) + monkeypatch.setattr("tensorrt_llm.serve.anthropic_adapter.STREAM_ADMISSION_WAIT_S", 0.01) def _rejecting_engine(where, stream): From 4d28ff00e1d9933daccabf3817c1562924f70bfb Mon Sep 17 00:00:00 2001 From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> Date: Fri, 30 Oct 2026 01:36:55 +0000 Subject: [PATCH 4/4] [None][test] Check exact Anthropic stream replay and event order Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com> --- .../llmapi/apps/test_anthropic_routes.py | 35 +++++++++++++++---- 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/tests/unittest/llmapi/apps/test_anthropic_routes.py b/tests/unittest/llmapi/apps/test_anthropic_routes.py index bf475fbcf00f..f6311195edef 100644 --- a/tests/unittest/llmapi/apps/test_anthropic_routes.py +++ b/tests/unittest/llmapi/apps/test_anthropic_routes.py @@ -569,23 +569,46 @@ def test_over_length_prompt_is_answered_as_anthropic_answers_it(monkeypatch, whe assert json.loads(body) == _PROMPT_TOO_LONG -def test_stream_starts_without_waiting_for_a_slow_first_token(monkeypatch, short_admission_wait): - """The wait for a rejection is bounded; a long prefill does not hold the stream back.""" +@pytest.mark.parametrize("immediate", [True, False], ids=["immediate", "delayed"]) +def test_stream_starts_without_waiting_for_a_slow_first_token( + monkeypatch, short_admission_wait, immediate +): + """Resume the stream exactly once, whether its first frame is ready or still pending.""" async def run(): gate = asyncio.Event() + if immediate: + gate.set() server = _chat_server(monkeypatch, _GatedGeneration(_reply_steps(), gate)) response = await _messages_response(server, stream=True) - # Started while the first token is still pending. + # The delayed case starts while the first token is still pending. assert isinstance(response, StreamingResponse) gate.set() return await _read(response) body = asyncio.run(run()) - assert body.startswith("event: message_start\n") - assert '"text":"hello"' in body - assert body.rstrip().endswith('data: {"type":"message_stop"}') + events = [ + json.loads(line.removeprefix("data: ")) + for line in body.splitlines() + if line.startswith("data: ") + ] + assert [event["type"] for event in events] == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + text = "".join( + event["content_block"]["text"] + if event["type"] == "content_block_start" + else event["delta"]["text"] + for event in events + if event["type"] in ("content_block_start", "content_block_delta") + ) + assert text == "hello" def test_rejection_after_the_wait_ends_the_stream_in_the_same_wording(