[TRTLLM-11123][fix] Propagate real errors to disagg server

tensorrt_llm/executor/postproc_worker.py

@@ -11,6 +11,7 @@
 from ..bindings import executor as tllm
 from ..llmapi.tokenizer import TransformersTokenizer, load_hf_tokenizer
 from ..llmapi.utils import print_traceback_on_error
+from ..logger import logger
 from ..sampling_params import SamplingParams
 from .ipc import ZeroMqQueue
 from .utils import ErrorResponse, is_llm_response
@@ -69,7 +70,6 @@ class Output(NamedTuple):
         client_id: int
         res: Any
         is_final: bool
-        error: str = ""
         metrics: Optional[dict[str, float]] = None
         request_perf_metrics: Any = None
         disaggregated_params: Any = None
@@ -193,24 +193,36 @@ async def handle_single_input(inp: PostprocWorker.Input,
                 batch.append(inp.rsp)
                 self._records.pop(client_id, None)
                 return
-            is_final = inp.rsp.result.is_final if is_llm_response(
-                inp.rsp) else True
-            res, metrics, perf_metrics, disaggregated_params = await self._handle_input(
-                inp)
-            record = self._records.get(client_id)
-            should_abort = record._aborted if record else False
-            batch.append(
-                PostprocWorker.Output(
-                    client_id=client_id,
-                    res=res,
-                    is_final=is_final,
-                    metrics=metrics,
-                    request_perf_metrics=perf_metrics,
-                    disaggregated_params=disaggregated_params,
-                    should_abort=should_abort,
-                ))
-            if is_final:
-                self._records.pop(client_id)
+            try:
+                is_final = inp.rsp.result.is_final if is_llm_response(
+                    inp.rsp) else True
+                res, metrics, perf_metrics, disaggregated_params = await self._handle_input(
+                    inp)
+                record = self._records.get(client_id)
+                should_abort = record._aborted if record else False
+                batch.append(
+                    PostprocWorker.Output(
+                        client_id=client_id,
+                        res=res,
+                        is_final=is_final,
+                        metrics=metrics,
+                        request_perf_metrics=perf_metrics,
+                        disaggregated_params=disaggregated_params,
+                        should_abort=should_abort,
+                    ))
+                if is_final:
+                    self._records.pop(client_id)
+            except Exception as e:
+                logger.error(
+                    f"Postprocessing error for client {client_id}: {e}\n"
+                    f"{traceback.format_exc()}")
+                batch.append(
+                    ErrorResponse(
+                        client_id=client_id,
+                        error_msg=f"Postprocessing error: {e}",
+                        request_id=getattr(inp.rsp, 'request_id', -1),
+                    ))
+                self._records.pop(client_id, None)
 
         while not self._to_stop.is_set():
             batch = []

tensorrt_llm/executor/result.py

@@ -172,6 +172,7 @@ def __init__(self,
         self.id = id
         self.sampling_params = sampling_params
         self.postproc_params = postproc_params
+        self._error_msg: Optional[str] = None
         self._disaggregated_params = None
         self.decoding_iter = 0
         self.cached_tokens = 0
@@ -259,6 +260,11 @@ def disaggregated_params(self) -> Optional[DisaggregatedParams]:
         """Returns the disaggregated params."""
         return self._disaggregated_params
 
+    @property
+    def error(self) -> Optional[str]:
+        """Return the error message if this result completed with an error."""
+        return self._error_msg
+
     def _handle_sequence(self,
                          finish_reasons,
                          response_tensors,
@@ -446,15 +452,14 @@ def _handle_response(self,
             if response.should_abort and not self._aborted:
                 self.abort()
 
-            if response.error:
-                if self._background_error_handler is not None and (
-                        handler := self._background_error_handler()):
-                    handler(response.error)
         elif is_llm_response(response):
             if response.has_error():
+                self._error_msg = response.error_msg
+                self._done = True
                 if self._background_error_handler is not None and (
                         handler := self._background_error_handler()):
                     handler(response.error_msg)
+                return  # Never fall through to response.result
 
             response_result = response.result
             if hasattr(response_result, "_result") and isinstance(
@@ -546,6 +551,7 @@ def _handle_response(self,
                     handler := self._background_error_handler()):
                 handler()
         elif isinstance(response, ErrorResponse):
+            self._error_msg = response.error_msg
             self._done = True
             if self._background_error_handler is not None and (
                     handler := self._background_error_handler()):

tensorrt_llm/llmapi/llm.py

@@ -64,6 +64,7 @@ class RequestOutput(DetokenizedGenerationResultBase, GenerationResult):
         context_logits (torch.Tensor, optional): The logits on the prompt token ids.
         disaggregated_params (DisaggregatedParams, optional): Parameters for disaggregated serving, including multimodal embedding handles.
         finished (bool): Whether the whole request is finished.
+        error (str, optional): The error message if this result completed with an error.
     """
 
     def __init__(self) -> None:

tensorrt_llm/serve/openai_client.py

@@ -16,7 +16,7 @@
 import asyncio
 import traceback
 from abc import ABC, abstractmethod
-from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple, Type
+from typing import Any, AsyncGenerator, Callable, Dict, List, Optional, Tuple, Type
 
 import aiohttp
 
@@ -99,6 +99,7 @@ def __init__(
         max_retries: int = 1,
         retry_interval_sec: int = 1,
         session: Optional[aiohttp.ClientSession] = None,
+        disagg_id_generator: Optional[Callable[[], int]] = None,
     ):
         self._router = router
         self._role = role
@@ -115,6 +116,7 @@ def __init__(
         )
         self._max_retries = max_retries
         self._retry_interval_sec = retry_interval_sec
+        self._disagg_id_generator = disagg_id_generator
 
     async def _send_request(
         self,
@@ -161,9 +163,14 @@ async def _post_with_retry(
         request: UCompletionRequest,
         hooks: Optional[ResponseHooks] = None,
     ) -> AsyncGenerator[Any, None]:
-        json_data = request.model_dump(exclude_unset=True, mode="json")
         is_stream = request.stream
         for attempt in range(self._max_retries + 1):
+            # Regenerate disagg_request_id on retry to avoid ID collision on workers
+            if attempt > 0 and self._disagg_id_generator is not None:
+                dp = getattr(request, "disaggregated_params", None)
+                if dp is not None and getattr(dp, "disagg_request_id", None) is not None:
+                    dp.disagg_request_id = self._disagg_id_generator()
+            json_data = request.model_dump(exclude_unset=True, mode="json")
             try:
                 lines_yielded = 0
                 start_time = get_steady_clock_now_in_seconds()
@@ -183,7 +190,15 @@ async def _post_with_retry(
                             yield line
                         # don't finish the request here since the response generator is not done yet
                     else:
-                        http_response.raise_for_status()
+                        if http_response.status >= 400:
+                            error_body = await http_response.text()
+                            raise aiohttp.ClientResponseError(
+                                http_response.request_info,
+                                http_response.history,
+                                status=http_response.status,
+                                message=f"{http_response.reason}: {error_body[:2048]}",
+                                headers=http_response.headers,
+                            )
                         response_dict = await http_response.json()
                         # yield here since python forbids return statements in async generators
                         yield response_dict

tensorrt_llm/serve/openai_disagg_server.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2025, NVIDIA CORPORATION.
+# Copyright (c) 2025-2026, NVIDIA CORPORATION.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -33,7 +33,8 @@
 from tensorrt_llm.llmapi import tracing
 from tensorrt_llm.llmapi.disagg_utils import (DisaggServerConfig,
                                               MetadataServerConfig, ServerRole,
-                                              get_ctx_gen_server_addrs)
+                                              get_ctx_gen_server_addrs,
+                                              get_global_disagg_request_id)
 from tensorrt_llm.logger import logger
 from tensorrt_llm.serve.cluster_storage import (HttpClusterStorageServer,
                                                 create_cluster_storage)
@@ -141,7 +142,10 @@ async def validation_exception_handler(_, exc):
         self.register_routes()
 
     def _create_client(self, router: Router, role: ServerRole, max_retries: int = 1) -> OpenAIClient:
-        client = OpenAIHttpClient(router, role, self._req_timeout_secs, max_retries)
+        node_id = self._config.node_id
+        client = OpenAIHttpClient(
+            router, role, self._req_timeout_secs, max_retries,
+            disagg_id_generator=lambda: get_global_disagg_request_id(node_id))
         self._perf_metrics_collector.add_client(client)
         return client
 

tensorrt_llm/serve/openai_disagg_service.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2025, NVIDIA CORPORATION.
+# Copyright (c) 2025-2026, NVIDIA CORPORATION.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -380,13 +380,23 @@ async def _verify_ctx_response(self, ctx_response: UCompletionResponse) -> None:
                 raise ValueError(
                     f"Context server returned {len(ctx_response.choices)} choices, expecting 1."
                 )
-            if ctx_response.choices[0].disaggregated_params is None:
-                raise ValueError("Context server did not return disaggregated params")
-            if ctx_response.choices[0].disaggregated_params.ctx_request_id is None:
-                raise ValueError("Invalid disaggregated params in context phase response.")
-            if ctx_response.choices[0].disaggregated_params.disagg_request_id is None:
+            choice = ctx_response.choices[0]
+            if choice.disaggregated_params is None:
                 raise ValueError(
-                    "Invalid disaggregated params in context phase response. disagg_request_id is None"
+                    f"Context server did not return disaggregated params."
+                    f" finish_reason={choice.finish_reason!r}"
+                )
+            if choice.disaggregated_params.ctx_request_id is None:
+                raise ValueError(
+                    f"Invalid disaggregated params: ctx_request_id is None."
+                    f" finish_reason={choice.finish_reason!r},"
+                    f" disagg_request_id={choice.disaggregated_params.disagg_request_id!r}"
+                )
+            if choice.disaggregated_params.disagg_request_id is None:
+                raise ValueError(
+                    f"Invalid disaggregated params: disagg_request_id is None."
+                    f" finish_reason={choice.finish_reason!r},"
+                    f" ctx_request_id={choice.disaggregated_params.ctx_request_id!r}"
                 )
             return ctx_response
 

tensorrt_llm/serve/openai_server.py

@@ -939,6 +939,8 @@ async def _create_chat_response(
         disaggregated_params: Optional[LlmDisaggregatedParams] = None
     ) -> ChatCompletionResponse:
         await promise.aresult()
+        if promise.error is not None:
+            raise RuntimeError(f"Generation failed: {promise.error}")
         if self.postproc_worker_enabled:
             chat_response = promise.outputs[0]._postprocess_result
         else:
@@ -1265,6 +1267,8 @@ async def completion_response(
                 postproc_params: Optional[PostprocParams]
         ) -> CompletionResponse:
             response = await promise
+            if response.error is not None:
+                raise RuntimeError(f"Generation failed: {response.error}")
             if not self.postproc_worker_enabled:
                 post_processor, args = postproc_params.post_processor, postproc_params.postproc_args
                 pp_result = post_processor(response, args)

tests/unittest/api_stability/references/request_output.yaml

@@ -32,4 +32,7 @@ methods:
         annotation: Optional[dict[str, float]]
         default: None
     return_annotation: None
-properties: {}
+properties:
+  error:
+    annotation: Optional[str]
+    default: inspect._empty