[None][fix] Disaggregated Anthropic Messages: start the stream and ping while the context phase runs

tensorrt_llm/serve/openai_disagg_server.py

@@ -21,7 +21,7 @@
 import traceback
 import uuid
 from contextlib import asynccontextmanager
-from typing import Callable, Optional
+from typing import AsyncIterator, Awaitable, Callable, Optional, Union
 
 import aiohttp
 import uvicorn
@@ -67,7 +67,8 @@
                                              PerfMetricsJsonlWriter,
                                              PerfMetricsMiddleware,
                                              combine_disagg_metrics)
-from tensorrt_llm.serve.responses_utils import (ServerArrivalTimeMiddleware,
+from tensorrt_llm.serve.responses_utils import (ClosingStreamingResponse,
+                                                ServerArrivalTimeMiddleware,
                                                 get_steady_clock_now_in_seconds,
                                                 guard_responses_stream,
                                                 stream_error_event)
@@ -108,6 +109,15 @@
 # strictly better than applying it.
 _CLOCK_SYNC_MAX_DELAY_SECONDS = 0.010
 
+# Seconds a streamed /v1/messages request waits for its chat pipeline before
+# the response starts. The pipeline has a stream to return only once its
+# context phase is done, and until the response starts the client receives
+# nothing, not even the status line, which clients and proxies take for a dead
+# connection after a while. A pipeline that answers within the wait is relayed
+# or, if it failed, answered with its error status; one still running has its
+# stream started and kept alive with pings until it answers.
+_MESSAGES_STREAM_START_WAIT_S = 1.0
+
 class RawRequestResponseHooks(ResponseHooks):
     def __init__(self, raw_req: Request, queue_latency_metric,
                  collect_perf_metrics: bool):
@@ -211,6 +221,20 @@ def _upstream_error_message(error: aiohttp.ClientResponseError) -> str:
     return f"context worker rejected the request with status {error.status}"
 
 
+class _MessagesStreamingResponse(ClosingStreamingResponse):
+    """Cancel the chat pipeline even when a disconnect precedes body iteration."""
+
+    def __init__(self, chat: asyncio.Future, content: AsyncIterator[str]) -> None:
+        super().__init__(content=content, media_type="text/event-stream")
+        self._chat = chat
+
+    async def __call__(self, scope, receive, send) -> None:
+        try:
+            await super().__call__(scope, receive, send)
+        finally:
+            self._chat.cancel()
+
+
 class OpenAIDisaggServer:
     def __init__(self,
                  config: DisaggServerConfig,
@@ -469,21 +493,51 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest,
             return anthropic_error_response(str(e), "invalid_request_error",
                                             400)
 
+        chat = self._wrap_entry_point(self._service.openai_chat_completion)(
+            chat_request, raw_request)
+        if request.stream:
+            chat = asyncio.ensure_future(chat)
+            try:
+                await asyncio.wait((chat, ),
+                                   timeout=_MESSAGES_STREAM_START_WAIT_S)
+            except asyncio.CancelledError:
+                # Cancelling the request cancels its pipeline.
+                chat.cancel()
+                raise
+            if not chat.done():
+                # The reframer pings while it waits for the pipeline.
+                return _MessagesStreamingResponse(
+                    chat=chat,
+                    content=reframe_openai_stream(
+                        self._openai_stream_when_done(chat),
+                        model=request.model),
+                )
+
+        answer = await self._messages_answer(chat)
+        if isinstance(answer, StreamingResponse):
+            return StreamingResponse(
+                content=reframe_openai_stream(answer.body_iterator,
+                                              model=request.model),
+                media_type="text/event-stream",
+            )
+        return answer
+
+    async def _messages_answer(self, chat: Awaitable[Response]) -> Response:
+        """Await a /v1/messages request's chat pipeline and answer from it.
+
+        A stream is returned as the pipeline's OpenAI stream, anything else as
+        the Anthropic response.
+        """
         try:
-            openai_response = await self._wrap_entry_point(
-                self._service.openai_chat_completion)(chat_request, raw_request)
+            openai_response = await chat
         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)
 
         if isinstance(openai_response, StreamingResponse):
-            return StreamingResponse(
-                content=reframe_openai_stream(openai_response.body_iterator,
-                                              model=request.model),
-                media_type="text/event-stream",
-            )
+            return openai_response
 
         status = getattr(openai_response, "status_code", 500)
         if status != 200:
@@ -510,6 +564,32 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest,
         return JSONResponse(content=anthropic_response.model_dump(
             exclude_none=True))
 
+    async def _openai_stream_when_done(
+            self, chat: asyncio.Future) -> AsyncIterator[Union[str, bytes]]:
+        """The OpenAI stream of a chat pipeline that outlasted the start wait.
+
+        The status line is already out, so a pipeline that fails raises
+        AnthropicResponseError with the message of its error response, and
+        the reframer ends the stream with an ``error`` event. A stream
+        abandoned before the pipeline answers, as when the client disconnects,
+        cancels the pipeline.
+        """
+        try:
+            answer = await self._messages_answer(chat)
+        finally:
+            chat.cancel()
+        if not isinstance(answer, StreamingResponse):
+            raise AnthropicResponseError(
+                json.loads(answer.body)["error"]["message"])
+        payloads = answer.body_iterator
+        try:
+            async for payload in payloads:
+                yield payload
+        finally:
+            aclose = getattr(payloads, "aclose", None)
+            if aclose is not None:
+                await aclose()
+
     async def get_model(self) -> Response:
         """List the served model, as a context worker reports it."""
         try:

tensorrt_llm/serve/openai_disagg_service.py

@@ -333,7 +333,7 @@ async def _send_disagg_request_ctx_first(
                     if hooks:
                         hooks.on_disagg_request_id(disagg_request_id)
                 gen_req = self._get_gen_request(request, ctx_response, disagg_request_id)
-            except Exception:
+            except (Exception, asyncio.CancelledError):
                 if gen_server:
                     await self._gen_router.finish_request(
                         request, success=False, req_id=gen_reservation_id