[None][feat] Responses: name the engine request behind each reply

docs/source/commands/trtllm-serve/trtllm-serve.rst

@@ -142,6 +142,8 @@ Another example uses ``curl``:
     :language: bash
     :linenos:
 
+A reply's ``id`` is a random ``resp_`` id, while the server's log lines, per-request perf metrics and ``--post_processor_hook`` name a request by the id the engine gave it. Each ``/v1/responses`` reply, streamed or not, therefore carries that engine request id in an ``x-trtllm-engine-request-id`` response header, so a reply can be matched with them. ``trtllm-serve disaggregated`` does not forward this header; there the reply ``id`` names the request instead; see the Responses API section of :doc:`Disaggregated Serving <../../features/disagg-serving>`.
+
 
 More openai compatible examples can be found in the `compatibility examples <https://github.com/NVIDIA/TensorRT-LLM/tree/main/examples/serve/compatibility>`_ directory.
 

docs/source/features/disagg-serving.md

@@ -263,6 +263,8 @@ The disaggregated server also serves `/v1/responses`, with both `context_first`
 - 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.
 
+A reply's `id` is `resp_<disagg_request_id>_<16 hex digits>`. Its first part is the [disaggregated request ID](#unique-global-request-id) the server gives the request, which the server's per-request perf metrics record as `disagg_request_id`, so the two can be matched. If a context request is retried under a new disaggregated request ID, the reply keeps the first one while the perf metrics record the new one. A `request_id` set by the client is used as the reply id unchanged. The disaggregated server does not forward the workers' response headers, so a reply carries no `x-trtllm-engine-request-id` header, unlike a reply from a single `trtllm-serve` instance.
+
 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.

docs/source/release-notes.md

@@ -19,6 +19,7 @@ All published functionality in the Release Notes has been fully tested and verif
 ### 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.
+- `/v1/responses`: a `trtllm-serve` reply now carries an `x-trtllm-engine-request-id` response header with the engine request id that the server's logs and perf metrics use. On `trtllm-serve disaggregated`, a reply id is now `resp_<disagg_request_id>_<16 hex digits>` instead of `resp_<32 hex digits>`, matching the `disagg_request_id` in the server's perf metrics; worker response headers, this one included, are not forwarded there. See [`trtllm-serve`](commands/trtllm-serve/trtllm-serve.rst) and [Disaggregated Serving](features/disagg-serving.md#responses-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_disagg_service.py

@@ -99,6 +99,29 @@ def _drop_ctx_handoff(response: UCompletionResponse) -> None:
         response.finish_reason = None
 
 
+def _name_responses_reply(request: UCompletionRequest, disagg_request_id: int) -> None:
+    """Name the reply to a Responses request ``resp_<disagg_request_id>_<16 hex>``.
+
+    A worker answers a Responses request under the request's ``request_id``, by
+    default a random ``resp_`` id. Each worker numbers its own engine requests,
+    so an engine request id can name different requests on different workers;
+    the disaggregated request id is unique across the deployment, and both
+    workers and this server record it with the request. The random part keeps
+    the name unique should a disaggregated request id repeat.
+
+    The name is given once, before any worker sees the request, so it stays the
+    same if a retried context request gets a new disaggregated request id. A
+    ``request_id`` the client set is kept, and other requests are left alone.
+
+    The workers get the name only because assigning a field adds it to
+    pydantic's ``model_fields_set``: ``_encode_request_body`` forwards set
+    fields only (``exclude_unset=True``), and an unset ``request_id`` would
+    leave each worker to draw its own from the field's default factory.
+    """
+    if isinstance(request, ResponsesRequest) and "request_id" not in request.model_fields_set:
+        request.request_id = f"resp_{disagg_request_id}_{os.urandom(8).hex()}"
+
+
 class _SseErrorObserver:
     """Observe worker error events without changing the forwarded SSE bytes."""
 
@@ -300,6 +323,7 @@ async def _send_disagg_request_ctx_first(
         disagg_request_id = await self._coordinator.get_disagg_request_id()
         if hooks:
             hooks.on_disagg_request_id(disagg_request_id)
+        _name_responses_reply(request, disagg_request_id)
         # reserve a gen_server if conditional disagg is needed
         gen_server, need_ctx = await self._check_conditional_disagg(request, disagg_request_id)
         # Context retries may replace disagg_request_id for the KV-transfer
@@ -604,6 +628,7 @@ async def _send_disagg_request_gen_first(
         disagg_request_id = await self._coordinator.get_disagg_request_id()
         if hooks:
             hooks.on_disagg_request_id(disagg_request_id)
+        _name_responses_reply(request, disagg_request_id)
         if need_ctx:
             # arrival->here = pre-ctx wait in the orchestrator/fleet.
             if hooks:

tensorrt_llm/serve/openai_server.py

@@ -208,6 +208,23 @@ class _MsgspecRoute(KvHintsRoute):
 
 TIMEOUT_KEEP_ALIVE = 5  # seconds.
 
+# The header in which a /v1/responses reply names the engine request it
+# answers; see _engine_request_id_headers.
+ENGINE_REQUEST_ID_HEADER = "x-trtllm-engine-request-id"
+
+
+def _engine_request_id_headers(promise: RequestOutput) -> Dict[str, str]:
+    """The response headers that name the engine request behind ``promise``.
+
+    A /v1/responses reply is named with a random ``resp_`` id, while the
+    server's log lines, perf metrics and post-processor hook name a request by
+    the id the engine gave it. ENGINE_REQUEST_ID_HEADER carries that id, so a
+    reply can be matched with them. The executor gives a request its id when
+    it accepts it, so ``promise`` always has one.
+    """
+    return {ENGINE_REQUEST_ID_HEADER: str(promise.request_id)}
+
+
 _warned_unresolvable_thinking = False
 
 
@@ -3558,6 +3575,7 @@ async def create_streaming_generator(promise: RequestOutput,
 
             asyncio.create_task(self.await_disconnected(raw_request, promise))
 
+            headers = _engine_request_id_headers(promise)
             if request.stream:
                 return ClosingStreamingResponse(
                     content=keep_responses_stream_alive(
@@ -3566,7 +3584,8 @@ async def create_streaming_generator(promise: RequestOutput,
                                                        postproc_params),
                             streaming_processor.get_stream_failed_events,
                         )),
-                    media_type="text/event-stream")
+                    media_type="text/event-stream",
+                    headers=headers)
             else:
                 response = await create_response(promise, postproc_params)
                 # Context-only: the orchestrator may ask for the prompt as a
@@ -3580,7 +3599,8 @@ async def create_streaming_generator(promise: RequestOutput,
                                    dtype=np.int32).tobytes()).decode("ascii")
                     response.prompt_token_ids = None
                 # by_alias: e.g. text.format.schema goes out under its wire name.
-                return JSONResponse(content=response.model_dump(by_alias=True))
+                return JSONResponse(content=response.model_dump(by_alias=True),
+                                    headers=headers)
         except CppExecutorError:
             logger.error(traceback.format_exc())
             # If internal executor error is raised, shutdown the server

tests/integration/test_lists/test-db/l0_cpu.yml

@@ -293,6 +293,7 @@ l0_cpu:
   - unittest/llmapi/apps/test_request_metrics.py::TestJsonlWriterSurvival
   - unittest/llmapi/apps/test_responses_custom_tools.py
   - unittest/llmapi/apps/test_responses_deepseek_v41_namespaces.py
+  - unittest/llmapi/apps/test_responses_engine_request_id.py
   - unittest/llmapi/apps/test_responses_input_preprocess.py
   - unittest/llmapi/apps/test_responses_store_replay.py
   - unittest/llmapi/apps/test_responses_stream_final_consistency.py

tests/unittest/disaggregated/test_disagg_responses_api.py

@@ -327,6 +327,7 @@ async def nothing(*args, **kwargs):
         return None
 
     promise = SimpleNamespace(
+        request_id=1,
         outputs=[SimpleNamespace(_postprocess_result=postprocessed)],
         prompt_token_ids=[5, 6, 7],
         aresult=nothing,

tests/unittest/disaggregated/test_openai_disagg_service.py

@@ -47,7 +47,7 @@
 from tensorrt_llm.serve.conversation_id import get_request_routing_id
 from tensorrt_llm.serve.disagg_auto_scaling import DisaggClusterManager, WorkerInfo
 from tensorrt_llm.serve.disagg_coordinator import DisaggCoordinatorService
-from tensorrt_llm.serve.openai_client import OpenAIHttpClient
+from tensorrt_llm.serve.openai_client import OpenAIHttpClient, _encode_request_body
 from tensorrt_llm.serve.openai_disagg_server import OpenAIDisaggServer
 from tensorrt_llm.serve.openai_disagg_service import OpenAIDisaggregatedService, _SseErrorObserver
 from tensorrt_llm.serve.openai_protocol import (
@@ -997,6 +997,130 @@ async def _ctx_response(request, *_args, **_kwargs):
     assert gen_call.args[0].disaggregated_params.ctx_request_id == 202
 
 
+def _as_a_worker_parses(request):
+    """``request`` as a worker parses it from the body it is sent."""
+    return type(request).model_validate(msgspec.msgpack.decode(_encode_request_body(request)))
+
+
+async def _context_worker_answer(request, *_args, **_kwargs):
+    """A context worker's answer; a Responses one is named after the request_id it parsed."""
+    received = _as_a_worker_parses(request)
+    disagg_request_id = received.disaggregated_params.disagg_request_id
+    if not isinstance(received, ResponsesRequest):
+        return _make_chat_response("length", disagg_request_id=disagg_request_id)
+    response = _make_responses_response(status="incomplete")
+    response.id = received.request_id
+    response.finish_reason = "length"
+    response.prompt_token_ids = [1, 2, 3]
+    response.disaggregated_params = DisaggregatedParams(
+        request_type="context_only",
+        ctx_request_id=disagg_request_id,
+        disagg_request_id=disagg_request_id,
+    )
+    return response
+
+
+async def _generation_worker_answer(request, *_args, **_kwargs):
+    """A generation worker's answer; a chat completion is named after its own engine request."""
+    received = _as_a_worker_parses(request)
+    if not isinstance(received, ResponsesRequest):
+        response = _make_chat_response("stop")
+        response.id = "chatcmpl-7"
+        return response
+    response = _make_responses_response()
+    response.id = received.request_id
+    return response
+
+
+async def _generation_worker_stream(request, *_args, **_kwargs):
+    """A generation worker's Responses stream, opened under the request_id it parsed."""
+    received = _as_a_worker_parses(request)
+    created = {
+        "type": "response.created",
+        "sequence_number": 0,
+        "response": {"id": received.request_id},
+    }
+    return _mock_streaming_response([b"data: " + msgspec.json.encode(created) + b"\n\n"])
+
+
+def _service_with_disagg_request_id_101(monkeypatch, schedule_style, generation_worker):
+    """A service that gives every request disaggregated request id 101."""
+    monkeypatch.delenv("TRTLLM_DISAGG_BENCHMARK_GEN_ONLY", raising=False)
+    service = _make_service(schedule_style)
+    service.is_ready = AsyncMock(return_value=True)
+    service._coordinator.get_disagg_request_id = AsyncMock(return_value=101)
+    service._ctx_router.get_next_server = AsyncMock(return_value=("ctx:9000", {"server_info": {}}))
+    service._gen_router.get_next_server = AsyncMock(return_value=("gen:9001", {"server_info": {}}))
+    service._ctx_client = AsyncMock()
+    service._gen_client = AsyncMock()
+    service._ctx_client.send_request = AsyncMock(side_effect=_context_worker_answer)
+    service._gen_client.send_request = AsyncMock(side_effect=generation_worker)
+    return service
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("schedule_style", ["context_first", "generation_first"])
+async def test_only_a_responses_reply_is_named_after_the_disagg_request_id(
+    monkeypatch, schedule_style
+):
+    """Both workers answer a Responses request as ``resp_<disagg_request_id>_<16 hex>``.
+
+    The disaggregated request id is unique across the deployment, unlike each
+    worker's engine request id. A Responses request the client named, and a chat
+    request, go to the workers as they came.
+    """
+    service = _service_with_disagg_request_id_101(
+        monkeypatch, schedule_style, _generation_worker_answer
+    )
+
+    def responses(**fields):
+        request = ResponsesRequest(model="test-model", input="hello", **fields)
+        return service.openai_responses(request)
+
+    first = await responses()
+    second = await responses()
+    client_named = await responses(request_id="resp_client")
+    chat = await service.openai_chat_completion(
+        ChatCompletionRequest(model="test-model", messages=[{"role": "user", "content": "hello"}])
+    )
+
+    assert re.fullmatch(r"resp_101_[0-9a-f]{16}", first.id)
+    assert re.fullmatch(r"resp_101_[0-9a-f]{16}", second.id)
+    assert first.id != second.id
+    # The context worker is sent the same name, for a request its phase finishes.
+    context_names = [
+        _as_a_worker_parses(call.args[0]).request_id
+        for call in service._ctx_client.send_request.call_args_list
+        if isinstance(call.args[0], ResponsesRequest)
+    ]
+    assert context_names == [first.id, second.id, "resp_client"]
+    assert client_named.id == "resp_client"
+    assert chat.id == "chatcmpl-7"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("schedule_style", ["context_first", "generation_first"])
+async def test_a_streamed_responses_reply_is_named_after_the_disagg_request_id(
+    monkeypatch, schedule_style
+):
+    """A streamed reply opens under ``resp_<disagg_request_id>_<16 hex>``, the name both workers get."""
+    service = _service_with_disagg_request_id_101(
+        monkeypatch, schedule_style, _generation_worker_stream
+    )
+
+    stream = await service.openai_responses(
+        ResponsesRequest(model="test-model", input="hello", stream=True)
+    )
+    events = [event async for event in stream]
+
+    assert len(events) == 1, events
+    created = msgspec.json.decode(events[0].removeprefix(b"data: "))
+    name = created["response"]["id"]
+    assert re.fullmatch(r"resp_101_[0-9a-f]{16}", name)
+    for worker in (service._ctx_client, service._gen_client):
+        assert _as_a_worker_parses(worker.send_request.call_args.args[0]).request_id == name
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("schedule_style", ["context_first", "generation_first"])
 async def test_gen_request_carries_orig_elapsed(monkeypatch, schedule_style):

tests/unittest/llmapi/apps/test_responses_engine_request_id.py

@@ -0,0 +1,97 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Offline tests for the engine request id header of /v1/responses replies.
+
+A Responses reply is named with a random ``resp_`` id, so the header is what
+ties it to the server's records of the request, which use the engine's id.
+"""
+
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import tensorrt_llm.serve.openai_server as server_module
+from tensorrt_llm.executor.result import CompletionOutput
+
+from .openai_server_stub import make_stub_openai_server
+
+pytestmark = pytest.mark.cpu_only
+
+_HEADER = "x-trtllm-engine-request-id"
+
+
+class _FinishedGeneration:
+    """A generation the engine finished in one step, under ``request_id``."""
+
+    def __init__(self, request_id):
+        self.request_id = request_id
+        self.id = request_id
+        self.prompt_token_ids = [1, 2, 3]
+        self.cached_tokens = 0
+        self.outputs = [
+            CompletionOutput(index=0, text="Four.", token_ids=[4, 5], finish_reason="stop")
+        ]
+        self.finished = self._done = True
+        self.error = None
+        self._steps = [self]
+
+    async def aresult(self):
+        return self
+
+    def __await__(self):
+        return self.aresult().__await__()
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if not self._steps:
+            raise StopAsyncIteration
+        return self._steps.pop()
+
+
+def _responses_client(tmp_path, monkeypatch, request_id):
+    """A client of /v1/responses on a server whose engine names its request ``request_id``."""
+
+    async def preprocess(**kwargs):
+        return [1, 2, 3], kwargs["request"].to_sampling_params()
+
+    monkeypatch.setattr(server_module, "responses_api_request_preprocess", preprocess)
+    model_config = type("ModelConfig", (), {"model_type": "qwen3", "vocab_size": 1024})()
+    server = make_stub_openai_server(
+        tmp_path,
+        "test-model",
+        model_config,
+        lambda **kwargs: _FinishedGeneration(request_id),
+    )
+    server.await_disconnected = AsyncMock()
+    server._extract_metrics = AsyncMock()
+    app = FastAPI()
+    app.add_api_route("/v1/responses", server.openai_responses, methods=["POST"])
+    return TestClient(app)
+
+
+@pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"])
+def test_a_reply_names_its_engine_request_in_a_header(tmp_path, monkeypatch, stream):
+    """The header carries the engine's id for the request."""
+    client = _responses_client(tmp_path, monkeypatch, request_id=42)
+    body = {"model": "test-model", "input": "What is 2 + 2?", "stream": stream}
+
+    reply = client.post("/v1/responses", json=body)
+
+    assert reply.status_code == 200, reply.text
+    assert reply.headers.get(_HEADER) == "42"