[https://nvbugs/6781271][fix] Responses API: agent-client fixes and disaggregated serving

docs/source/features/disagg-serving.md

@@ -226,6 +226,17 @@ curl http://localhost:8000/v1/completions \
     }' -w "\n"
 ```
 
+#### Responses API
+
+The disaggregated server also serves `/v1/responses`, with both `context_first` and `generation_first` scheduling, and `/v1/models`, which it answers from a context worker. A streamed response carries the generation worker's events unchanged. Unlike a single `trtllm-serve` instance, the disaggregated server keeps no response history:
+
+- A request with `previous_response_id` is rejected with a 400, because a stored response lives in one worker's memory and a later request may reach another worker. Send the earlier turns in `input` instead.
+- `store` is ignored: nothing is persisted, and the server logs a warning once when a client asks for it.
+
+One limitation applies to `/v1/responses` on every server, disaggregated or not:
+
+- `parallel_tool_calls: false`, and a `tool_choice` other than `auto` or `none` on non-Harmony models, are accepted but not enforced. The server logs a warning once.
+
 #### Launching disaggregated servers on SLURM clusters
 
 Please refer to [Disaggregated Inference Benchmark Scripts](source:examples/disaggregated/slurm).

docs/source/release-notes.md

@@ -12,11 +12,13 @@ All published functionality in the Release Notes has been fully tested and verif
   - **First-launch compile cost.** The kernels are JIT-compiled by the CuTe DSL and cached on disk under the FlashInfer workspace. The first start on a host (or after a FlashInfer / CuTe DSL upgrade) compiles the kernel variants the deployment's generation row counts select during `LLM()` startup, about 5 to 10 minutes on B200 for DeepSeek-V4; later starts load the cache. Under `trtllm-llmapi-launch` and the default `LLM()` MPI session every rank has its own FlashInfer workspace (`TRTLLM_FLASHINFER_WORKSPACE_PER_PROCESS=1`), so each rank compiles its own copy; set `TRTLLM_FLASHINFER_WORKSPACE_PER_PROCESS=0` to share one cache between the ranks of a host.
   - **Warm-up controls.** `TLLM_PRIMTS_WARM_WORKERS` sets the compile subprocesses per rank (default 4, reduced so that the ranks on a host, counted from the model's parallel mapping, start at most 16 together; `0` compiles in-process), `TLLM_PRIMTS_WARM_TIMEOUT` bounds the wait for them (default 1800 s), and `TLLM_PRIMTS_WARM_STRICT=1` turns a worker failure into an error instead of an in-process retry. Each worker imports torch and the CuTe DSL and opens a CUDA context, so lower the worker count on nodes with little host memory.
   - **`FLASHINFER_DISABLE_JIT=1`** now also refuses to compile these kernels (with the on-disk CuTe DSL cache enabled, its default): startup then fails on a variant missing from the cache instead of falling back to TRTLLM-Gen, so a deployment that sets it must pre-populate the cache (one start without the variable) or opt out of the library.
+- Disaggregated serving supports the Responses API: `trtllm-serve disaggregated` serves `/v1/responses` with context-first and generation-first scheduling, and `/v1/models`. `previous_response_id` is rejected and `store` is ignored there, since no response history is kept. See [Disaggregated Serving](features/disagg-serving.md#responses-api).
 
 ### Infrastructure Changes
 
 ### API Changes
 
+- `/v1/responses`: a response cut short by `max_output_tokens` now reports `incomplete_details.reason` `max_output_tokens`, and when streamed it ends with `response.incomplete` instead of `response.completed`, as in the OpenAI API.
 - <span style="color: red">**[BREAKING CHANGE] Removed obsolete C++ executor orchestration helpers.**</span> The internal `RequestWithId` type, its `serializeReqWithIds` / `deserializeReqWithIds` functions, all production `insertRequestInOrder` overloads, and the unused `requestUtils.h` and `orchestratorUtils.h` helpers have been removed. These helpers belonged to the retired C++ executor orchestration path; downstream code linking their exported symbols must stop using them and rebuild. The current `Request` / `Response` interfaces and serialization remain available. Executor factory and worker constructors no longer accept the unused `engine` and `batched_logits_processor` arguments; callers should omit them and continue passing model configuration through `llm_args` and `hf_model_dir`. Unused decoder, request-cancellation, executor, legacy-model, orchestration, and speculative-decoding MPI tag names have also been removed; values 1–23, 127–129, and 1023–1025 are retired. The default, KV-cache event, and context-transfer tags retain their existing values. Internal `LogprobParams` and `LogProbsResult` tuples now contain only prompt-logprob fields; generation logprobs remain supplied by the sampler. The unused `TLLM_EXECUTOR_PERIODICAL_RESP_IN_AWAIT` environment variable has been removed.
 - <span style="color: red">**[BREAKING CHANGE] Unified multimodal processor metadata.**</span>
   Multimodal processor extensions now use `get_mm_item_metadata()` and

tensorrt_llm/serve/openai_client.py

@@ -51,6 +51,8 @@
     ChatCompletionResponse,
     CompletionRequest,
     CompletionResponse,
+    ResponsesRequest,
+    ResponsesResponse,
     UCompletionRequest,
     UCompletionResponse,
 )
@@ -76,14 +78,39 @@
 _msgpack_encoder = msgspec.msgpack.Encoder()
 MSGPACK_HEADERS = {"Content-Type": "application/json", "X-TRTLLM-Msgpack": "1"}
 
+# A Responses stream opens with these events before the model emits anything.
+_RESPONSES_OPENING_EVENTS = frozenset((b"response.created", b"response.in_progress"))
+
+
+def _has_responses_output_event(chunk: bytes) -> bool:
+    """Whether an SSE chunk of a Responses stream carries an event past its opening."""
+    return any(
+        line.startswith(b"event: ")
+        and line[len(b"event: ") :].strip() not in _RESPONSES_OPENING_EVENTS
+        for line in chunk.split(b"\n")
+    )
+
 
 # post_json's request and response are ordinary pydantic models, but not the
 # UCompletion* union: it carries auxiliary payloads such as count-tokens. The
 # protocol names the one method actually called, and the TypeVar ties the
 # returned object to the type the caller asked for, so a mismatch is a type
 # error rather than an Any that silently propagates.
 class _SerializableRequest(Protocol):
-    def model_dump(self, *, mode: str = ..., exclude_unset: bool = ...) -> dict: ...
+    def model_dump(
+        self, *, mode: str = ..., exclude_unset: bool = ..., by_alias: bool = ...
+    ) -> dict: ...
+
+
+def _encode_request_body(request: _SerializableRequest) -> bytes:
+    """Encode a request forwarded to a worker.
+
+    Fields go out under their wire names, e.g. ``schema``, which the model
+    holds as ``schema_``.
+    """
+    return _msgpack_encoder.encode(
+        request.model_dump(mode="json", exclude_unset=True, by_alias=True)
+    )
 
 
 _ResponseT = TypeVar("_ResponseT", bound=BaseModel)
@@ -114,6 +141,10 @@ async def send_request(
                 hooks,
                 req_id,
             )
+        elif isinstance(request, ResponsesRequest):
+            return await self._send_request(
+                "v1/responses", request, ResponsesResponse, server, hooks, req_id
+            )
         else:
             raise ValueError(f"Invalid request type: {type(request)}")
 
@@ -154,6 +185,16 @@ async def post_json(
         """
         ...
 
+    @abstractmethod
+    async def get_json(
+        self,
+        endpoint: str,
+        response_type: Type[_ResponseT],
+        server: str,
+    ) -> _ResponseT:
+        """GET a worker endpoint, outside routing state, like post_json."""
+        ...
+
     async def shutdown(self) -> None: ...
 
     @abstractmethod
@@ -236,7 +277,7 @@ async def post_json(
         # same way -- hence the same msgpack transport as _send_request. The
         # worker decodes it because _MsgspecRoute is the app's route_class, so
         # it covers every route rather than just the completion ones.
-        body = _msgpack_encoder.encode(request.model_dump(mode="json", exclude_unset=True))
+        body = _encode_request_body(request)
         headers = dict(MSGPACK_HEADERS)
         async with self._session.post(url, data=body, headers=headers) as response:
             if response.status >= 400:
@@ -250,6 +291,26 @@ async def post_json(
                 )
             return response_type(**await response.json())
 
+    async def get_json(
+        self,
+        endpoint: str,
+        response_type: Type[_ResponseT],
+        server: str,
+    ) -> _ResponseT:
+        server_url = server if server.startswith("http") else f"http://{server}"
+        url = f"{server_url.rstrip('/')}/{endpoint}"
+        async with self._session.get(url) as response:
+            if response.status >= 400:
+                error_body = await response.text()
+                raise aiohttp.ClientResponseError(
+                    response.request_info,
+                    response.history,
+                    status=response.status,
+                    message=f"{response.reason}: {error_body[:2048]}",
+                    headers=response.headers,
+                )
+            return response_type(**await response.json())
+
     async def _send_request(
         self,
         endpoint: str,
@@ -391,7 +452,7 @@ async def _post_with_retry(
             # Serialize once on the orchestrator's single event-loop thread.
             # Content-Type stays application/json so FastAPI still routes the
             # body through Request.json(); the header picks the decoder.
-            body = _msgpack_encoder.encode(request.model_dump(mode="json", exclude_unset=True))
+            body = _encode_request_body(request)
             headers = dict(MSGPACK_HEADERS)
             if self._request_perf_metrics:
                 headers[RETURN_METRICS_HEADER] = "1"
@@ -528,13 +589,18 @@ async def _response_generator(
         try:
             last_token_time = start_time
             chunk_count = 0
+            # The first token of a Responses stream is the first frame after
+            # its opening events, which the worker sends before generating.
+            awaiting_output = isinstance(request, ResponsesRequest)
             marker = f"event: {SSE_METRICS_EVENT}\n".encode()
             event_delimiter = b"\n\n"
             pending = b""
             metrics_event = b""
             async for chunk in http_response.content.iter_any():
                 now_time = get_steady_clock_now_in_seconds()
-                if chunk:
+                if awaiting_output and chunk:
+                    awaiting_output = not _has_responses_output_event(chunk)
+                if chunk and not awaiting_output:
                     if chunk_count == 0 and hooks:
                         hooks.on_first_token(server, request)
                     latency = now_time - last_token_time

tensorrt_llm/serve/openai_disagg_server.py

@@ -61,14 +61,16 @@
     OpenAIDisaggregatedService, ResponseHooks)
 from tensorrt_llm.serve.openai_protocol import (
     ChatCompletionRequest, ChatCompletionResponse, CompletionRequest,
-    ConversationParams, ErrorResponse, UCompletionRequest, UCompletionResponse,
-    ensure_request_chat_template_allowed)
+    ConversationParams, ErrorResponse, ResponsesRequest, UCompletionRequest,
+    UCompletionResponse, ensure_request_chat_template_allowed)
 from tensorrt_llm.serve.perf_metrics import (DisaggPerfMetricsCollector,
                                              PerfMetricsJsonlWriter,
                                              PerfMetricsMiddleware,
                                              combine_disagg_metrics)
 from tensorrt_llm.serve.responses_utils import (ServerArrivalTimeMiddleware,
-                                                get_steady_clock_now_in_seconds)
+                                                get_steady_clock_now_in_seconds,
+                                                guard_responses_stream,
+                                                stream_error_event)
 from tensorrt_llm.serve.router import Router
 from tensorrt_llm.usage import TerminalOutcome, record_termination_observation
 from tensorrt_llm.version import __version__ as VERSION
@@ -345,6 +347,8 @@ def register_routes(self):
         self.app.add_api_route("/v1/chat/completions", self._wrap_entry_point(self._service.openai_chat_completion, ChatCompletionRequest), methods=["POST"])
         self.app.add_api_route("/v1/messages", self.anthropic_messages, methods=["POST"])
         self.app.add_api_route("/v1/messages/count_tokens", self.anthropic_count_tokens, methods=["POST"])
+        self.app.add_api_route("/v1/responses", self._wrap_entry_point(self._service.openai_responses, ResponsesRequest), methods=["POST"])
+        self.app.add_api_route("/v1/models", self.get_model, methods=["GET"])
         self.app.add_api_route("/health", self.health, methods=["GET"])
         self.app.add_api_route("/cluster_info", self.cluster_info, methods=["GET"])
         self.app.add_api_route("/version", self.version, methods=["GET"])
@@ -437,10 +441,18 @@ async def wrapper(req: request_type, raw_req: Request) -> Response:
                 response_or_generator = await entry_point(req, hooks)
                 self._perf_metrics_collector.total_responses.inc()
                 if req.stream:
+                    stream = response_or_generator
+                    if isinstance(req, ResponsesRequest):
+                        # A relayed Responses stream that fails early ends with
+                        # an `error` event; other protocols end on a sentinel.
+                        stream = guard_responses_stream(stream,
+                                                        stream_error_event)
                     return StreamingResponse(
-                        content=response_or_generator,
+                        content=stream,
                         media_type="text/event-stream")
-                return JSONResponse(content=response_or_generator.model_dump())
+                # by_alias: e.g. text.format.schema goes out under its wire name.
+                return JSONResponse(
+                    content=response_or_generator.model_dump(by_alias=True))
             except Exception as e:
                 # Usually raises; returns a Response for worker errors that
                 # carry a machine-readable code (context_length_exceeded).
@@ -497,6 +509,26 @@ async def anthropic_messages(self, request: AnthropicMessagesRequest,
         return JSONResponse(content=anthropic_response.model_dump(
             exclude_none=True))
 
+    async def get_model(self) -> Response:
+        """List the served model, as a context worker reports it."""
+        try:
+            return JSONResponse(
+                content=(await self._service.get_model()).model_dump())
+        except aiohttp.ClientResponseError as error:
+            status, message = error.status or 500, _upstream_error_message(
+                error)
+        except (RuntimeError, ValueError) as error:
+            # No context worker yet; the client may retry.
+            status, message = 503, str(error)
+        return JSONResponse(
+            status_code=status,
+            content=ErrorResponse(
+                message=message,
+                type=("invalid_request_error"
+                      if 400 <= status < 500 else "api_error"),
+                code=status,
+            ).model_dump())
+
     async def anthropic_count_tokens(
             self, request: AnthropicCountTokensRequest) -> Response:
         """Count Anthropic input tokens on a context worker."""