[TRTLLM-17272][feat] Add public runtime accessors for the remaining serving-frontend private accesses

pyproject.toml

@@ -862,6 +862,7 @@ files = [
     "tensorrt_llm/_torch/pyexecutor/sampler/token_ban.py",
     "tensorrt_llm/_torch/pyexecutor/sampler/ops/flashinfer.py",
     "tensorrt_llm/_torch/pyexecutor/sampler/ops/vanilla.py",
+    "tensorrt_llm/llmapi/engine_client.py",
     "tests/unittest/_torch/sampler/test_torch_sampler.py",
     "tests/unittest/_torch/sampler/test_token_ban.py",
     "tests/unittest/_torch/sampler/test_beam_search.py",

tensorrt_llm/_torch/pyexecutor/py_executor.py

@@ -14,8 +14,8 @@
 from contextlib import ExitStack, contextmanager, nullcontext
 from enum import IntEnum
 from queue import Queue
-from typing import (TYPE_CHECKING, Callable, Dict, Iterable, Iterator, List,
-                    Literal, NoReturn, Optional, Tuple, Union)
+from typing import (TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator,
+                    List, Literal, NoReturn, Optional, Tuple, Union)
 
 import torch
 from strenum import StrEnum
@@ -2598,6 +2598,19 @@ def get_openengine_discovery(self) -> dict:
         """Return immutable-at-startup OpenEngine node and publisher discovery."""
         return self._openengine_discovery
 
+    def set_openengine_discovery(self, discovery: dict) -> None:
+        """Record OpenEngine node and KV-event publisher discovery metadata.
+
+        Called once at startup by the OpenEngine node-discovery hook
+        (tensorrt_llm/grpc/openengine/node.py); get_openengine_discovery()
+        returns it.
+        """
+        self._openengine_discovery = discovery
+
+    def set_openengine_node_server(self, server: Any) -> None:
+        """Adopt the OpenEngine node metadata server; shutdown closes it."""
+        self._openengine_node_server = server
+
     def get_kv_cache_load(self) -> dict:
         """Return the scheduler-owned KV load snapshot.
 

tensorrt_llm/commands/serve.py

@@ -555,8 +555,8 @@ def _bind_launcher_uds(llm) -> Tuple[socket.socket, str]:
     It lives in the multi-frontend ipc dir, which is private to the user and
     removed, socket file included, when the executor proxy shuts down.
     """
-    # Only the classic IPC proxy has the attribute.
-    ipc_dir = getattr(llm._executor, "_multi_frontend_ipc_dir", None)
+    # Only the classic IPC proxy in multi-frontend mode has one.
+    ipc_dir = llm.multi_frontend_ipc_dir
     if not ipc_dir:
         raise ValueError(
             "num_serve_frontends > 1 requires the classic IPC executor "
@@ -2661,7 +2661,7 @@ def _run_fleet_worker_impl():
     """Build and run one fleet worker inside its telemetry boundary."""
     _init_fleet_worker_process()
     server = _build_disagg_server_from_env()
-    host, port = server._config.hostname, server._config.port
+    host, port = server.address
 
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

tensorrt_llm/executor/executor.py

@@ -337,6 +337,20 @@ def is_shutdown(self) -> bool:
         """Return True if the executor is shutting down or fatally errored."""
         return self.doing_shutdown or self._fatal_error is not None
 
+    @property
+    def fatal_error(self) -> Optional[BaseException]:
+        """The unrecoverable engine error recorded by ``_set_fatal_error``, or None."""
+        return self._fatal_error
+
+    def record_fatal_error(self, error: BaseException) -> None:
+        """Record an unrecoverable engine error; the first recorded error wins (see ``fatal_error``)."""
+        self._set_fatal_error(error)
+
+    @property
+    def engine_dead(self) -> bool:
+        """Whether the engine is known dead (an unrecoverable error was recorded)."""
+        return self._fatal_error is not None
+
     def check_health(self) -> bool:
         """Check whether the executor is healthy and able to process requests.
 
@@ -405,6 +419,11 @@ def resource_governor_queue(self):
         """Return the resource governor queue if this executor supports it."""
         return None
 
+    @property
+    def multi_frontend_ipc_dir(self) -> Optional[str]:
+        """The multi-frontend IPC directory if this executor runs one, else None."""
+        return None
+
     @property
     def enable_postprocess_parallel(self) -> bool:
         return self.postproc_config.enabled

tensorrt_llm/executor/proxy.py

@@ -480,6 +480,15 @@ def check_health(self) -> bool:
 
         return True
 
+    @property
+    def engine_dead(self) -> bool:
+        """Whether the engine is known dead (a worker died or a fatal error was recorded).
+
+        ``_mark_engine_dead`` can set the sticky flag before the fatal error
+        is recorded, so check both.
+        """
+        return self._engine_dead or self._fatal_error is not None
+
     def _set_fatal_error(self, error: BaseException) -> None:
         """Record the fatal error, then unblock every pending request.
 
@@ -671,6 +680,19 @@ def _setup_queues(self) -> WorkerCommIpcAddrs:
             frontend_result_queue_addrs=frontend_result_addrs,
         )
 
+    @property
+    def multi_frontend_ipc_dir(self) -> Optional[str]:
+        """The launcher-owned directory of the multi-frontend IPC endpoints.
+
+        None unless multi-frontend mode is active. The launcher places its
+        own Unix sockets (the attached frontends' forwarding endpoint) in it;
+        the directory is private to the user and removed, socket files
+        included, when this proxy shuts down.
+        """
+        # GenerationExecutorFrontendProxy skips this class's __init__, so the
+        # attribute can be missing there; an attached frontend owns no ipc dir.
+        return getattr(self, "_multi_frontend_ipc_dir", None)
+
     def multi_frontend_attach_info(self) -> Optional[dict]:
         """The attach payload consumed by attached serving frontends.
 
@@ -1684,8 +1706,8 @@ def __init__(
         self.garbage_collection_gen0_threshold = None
         self.workers_started = False
         self.dispatch_result_thread: Optional[ManagedThread] = None
-        # Must be None: OpenAIServer reads the resource_governor_queue
-        # property at init; the governor lives with the launcher only.
+        # Must be None: LLM.resource_governor_queue reads this property at
+        # server init; the governor lives with the launcher only.
         self._resource_governor_queue = None
 
         hmac_key = bytes.fromhex(attach_info["hmac_key"])

tensorrt_llm/executor/result.py

@@ -67,7 +67,7 @@ class LogProbsResult(NamedTuple):
     prompt: Optional[TokenLogprobs | SimpleTokenLogprobs] = None
 
 
-class _SpecDecCounters(NamedTuple):
+class SpecDecCounters(NamedTuple):
     """Speculative-decoding counters reported by the request behind one sequence.
 
     Field names match the executor response so a formatter can read either.
@@ -128,6 +128,8 @@ class CompletionOutput:
         logprobs_diff (TokenLogprobs | SimpleTokenLogprobs): Logprobs of newly generated tokens.
         text_diff (str): Newly generated tokens.
         routed_experts (Optional[torch.Tensor]): Per-token pre-EPLB logical top-k MoE expert ids (Router Replay / R3), or None when not requested.
+        spec_dec_counters (tensorrt_llm.executor.result.SpecDecCounters, optional): This sequence's speculative-decoding counters, or None.
+        text_diff_offset (int): Offset in text at which text_diff starts.
     """
     index: int
     text: str = ""
@@ -157,9 +159,9 @@ class CompletionOutput:
     # the result of result_handler passed to postprocess workers
     _postprocess_result: Any = None
     # this sequence's own spec-decode counters; see _handle_sequence
-    _spec_dec_counters: Optional[_SpecDecCounters] = field(default=None,
-                                                           init=False,
-                                                           repr=False)
+    _spec_dec_counters: Optional[SpecDecCounters] = field(default=None,
+                                                          init=False,
+                                                          repr=False)
 
     def __getstate__(self) -> dict:
         # _incremental_states holds a tokenizers.DecodeStream (a Rust object,
@@ -208,6 +210,16 @@ def routed_experts(self) -> Optional[torch.Tensor]:
     def text_diff(self) -> str:
         return self.text[self._last_text_len:]
 
+    @property
+    def text_diff_offset(self) -> int:
+        """Offset in ``text`` at which ``text_diff`` starts."""
+        return self._last_text_len
+
+    @property
+    def spec_dec_counters(self) -> Optional[SpecDecCounters]:
+        """This sequence's own speculative-decoding counters, or None if the request did not draft."""
+        return self._spec_dec_counters
+
     @property
     def token_ids_diff(self) -> List[int]:
         return self.token_ids[self._last_token_ids_len:]
@@ -382,7 +394,7 @@ def _handle_sequence(self,
         output.disaggregated_params = self.disaggregated_params
         output._last_token_ids_len = len(output.token_ids)
         output._last_logprobs_len = len(output.logprobs)
-        output._spec_dec_counters = _SpecDecCounters()
+        output._spec_dec_counters = SpecDecCounters()
         decoder_output_prefix = ()
         if (self.sampling_params.exclude_input_from_output
                 or getattr(self, "_streaming", False)):
@@ -463,7 +475,7 @@ def _handle_sequence(self,
             # counters, while the request-level copies on self are overwritten by
             # every candidate's response. Capture this sequence's here so a choice
             # never reports whichever candidate happened to respond last.
-            output._spec_dec_counters = _SpecDecCounters(
+            output._spec_dec_counters = SpecDecCounters(
                 per_pos_drafted=response_tensors.per_pos_drafted,
                 per_pos_accepted=response_tensors.per_pos_accepted,
                 spec_dec_totals=response_tensors.spec_dec_totals)

tensorrt_llm/grpc/openengine/capabilities.py

@@ -11,6 +11,7 @@
 from tensorrt_llm.inputs.registry import (
     MULTIMODAL_PLACEHOLDER_REGISTRY,
     BaseMultimodalInputProcessor,
+    registered_model_type,
 )
 
 from .kv_events import events_config
@@ -67,7 +68,7 @@ def supported_guides(backend: str | None) -> frozenset[str]:
 def _loaded_model_type(llm: Any) -> str:
     """Return the top-level model type for the loaded input processor."""
     input_processor = getattr(llm, "input_processor", None)
-    config = getattr(input_processor, "config", None) or getattr(llm, "_hf_model_config", None)
+    config = getattr(input_processor, "config", None) or getattr(llm, "hf_model_config", None)
     if config is None:
         return ""
     return resolve_top_level_model_type(config)
@@ -102,8 +103,7 @@ def resolve_media_capabilities(
     if not model_type:
         return MediaCapabilities({})
     if not isinstance(input_processor, BaseMultimodalInputProcessor):
-        registered_model_type = getattr(type(input_processor), "_registered_model_type", None)
-        if registered_model_type != model_type:
+        if registered_model_type(type(input_processor)) != model_type:
             return MediaCapabilities({})
         if not callable(input_processor):
             return MediaCapabilities({})

tensorrt_llm/grpc/openengine/control.py

@@ -131,32 +131,26 @@ async def _get_kv_cache_load(self) -> dict:
         async with self._load_snapshot_lock:
             now = time.monotonic()
             if now >= self._load_snapshot_deadline:
-                executor = getattr(self._llm, "_executor", None)
-                getter = getattr(executor, "get_kv_cache_load", None)
-                self._load_snapshot = await asyncio.to_thread(getter) if callable(getter) else {}
+                self._load_snapshot = await asyncio.to_thread(self._llm.get_kv_cache_load)
                 self._load_snapshot_deadline = time.monotonic() + _LOAD_SNAPSHOT_CACHE_SECONDS
         return self._load_snapshot
 
     def _engine_is_healthy(self) -> bool:
         """Readiness via the predicate the engine's own HTTP /health uses.
 
-        `_check_health` is stronger than the `is_shutdown()` guard generate_async
+        `check_health()` is stronger than the `is_shutdown()` guard generate_async
         raises on: it drains the executor's error queue, so a fatal error queued
         but not yet promoted is caught, and the proxy polls MPI worker liveness.
-        It also covers encode-only engines, where `_executor` is None.
+        It also covers encode-only engines, where there is no generation executor.
         """
-        check = getattr(self._llm, "_check_health", None)
-        if callable(check):
-            try:
-                return bool(check())
-            except (RuntimeError, ValueError, OSError) as error:
-                logger.warning(f"OpenEngine health check reported a failure: {error}")
-                return False
-            except (TypeError, AttributeError) as error:
-                logger.error(f"OpenEngine health check contract failed: {error}")
-                return False
-        executor = getattr(self._llm, "_executor", None)
-        return executor is not None and not executor.is_shutdown()
+        try:
+            return bool(self._llm.check_health())
+        except (RuntimeError, ValueError, OSError) as error:
+            logger.warning(f"OpenEngine health check reported a failure: {error}")
+            return False
+        except (TypeError, AttributeError) as error:
+            logger.error(f"OpenEngine health check contract failed: {error}")
+            return False
 
     # -- Runtime metadata ---------------------------------------------------
 
@@ -222,15 +216,17 @@ async def GetServerInfo(
         block_size = _positive_int(self._llm.args.kv_cache_config.tokens_per_block)
         if block_size is not None:
             capacity.kv_block_size = block_size
-        executor = getattr(self._llm, "_executor", None)
-        capacity_getter = getattr(executor, "get_kv_cache_capacity", None)
-        if callable(capacity_getter):
-            kv_capacity = await asyncio.to_thread(capacity_getter)
-            local_blocks = _positive_int(kv_capacity.get("maxNumBlocks"))
-            if local_blocks is not None and data_parallel_size(self._llm) == 1:
-                capacity.total_kv_blocks = local_blocks
-        discovery_getter = getattr(executor, "get_openengine_discovery", None)
-        discovery = await asyncio.to_thread(discovery_getter) if callable(discovery_getter) else {}
+        # An encode-only LLM has no generation executor and no KV capacity;
+        # LLM.get_kv_cache_capacity() raises for it, so skip the query.
+        kv_capacity = (
+            {}
+            if self._llm.args.encode_only is True
+            else await asyncio.to_thread(self._llm.get_kv_cache_capacity)
+        )
+        local_blocks = _positive_int(kv_capacity.get("maxNumBlocks"))
+        if local_blocks is not None and data_parallel_size(self._llm) == 1:
+            capacity.total_kv_blocks = local_blocks
+        discovery = await asyncio.to_thread(self._llm.get_openengine_discovery)
         events = events_config(self._llm)
         discovery_required = (
             getattr(self._llm.args, "_openengine_discovery", None) is not None
@@ -660,9 +656,7 @@ async def _requested_sources(
         """
         if events_config(self._llm) is None:
             return []
-        executor = getattr(self._llm, "_executor", None)
-        getter = getattr(executor, "get_openengine_discovery", None)
-        discovery = await asyncio.to_thread(getter) if callable(getter) else {}
+        discovery = await asyncio.to_thread(self._llm.get_openengine_discovery)
         descriptors = discovery.get("sources")
         if descriptors is None:
             await context.abort(

tensorrt_llm/grpc/openengine/formatting.py

@@ -116,15 +116,15 @@ def _stop_texts(sampling_params: SamplingParams) -> list[str]:
 def _token_holdback(sampling_params: SamplingParams) -> int:
     """Number of trailing tokens to withhold so a stop token id is not streamed.
 
-    NOTE: this depends on the private ``SamplingParams._stop_word_ids``, which the
-    LLM engine populates with the tokenized stop words. It is read after
+    NOTE: this depends on ``SamplingParams.stop_word_ids``, which the LLM
+    engine populates with the tokenized stop words. It is read after
     ``generate_async`` returns, when it is expected to be set. If the engine ever
     stops populating it (or populates it later), this degrades to 0 (no holdback)
     rather than failing — ``test_token_holdback_*`` guards the current behavior.
     """
     if sampling_params.include_stop_str_in_output:
         return 0
-    stop_word_ids = getattr(sampling_params, "_stop_word_ids", None) or []
+    stop_word_ids = sampling_params.stop_word_ids or []
     if not stop_word_ids:
         return 0
     return max(1, max(len(token_ids) for token_ids in stop_word_ids) - 1)

tensorrt_llm/grpc/openengine/node.py

@@ -151,7 +151,7 @@ def initialize_node_discovery(executor) -> None:
         ),
         "source_owners": source_owners,
     }
-    executor._openengine_discovery = {"node": metadata, "sources": sources}
+    executor.set_openengine_discovery({"node": metadata, "sources": sources})
     info = server_pb2.ServerInfo(
         engine_name="tensorrt_llm",
         engine_version=__version__,
@@ -190,4 +190,4 @@ def initialize_node_discovery(executor) -> None:
         if server is not None:
             server.close()
         raise RuntimeError(f"OpenEngine node discovery startup failed: {errors}")
-    executor._openengine_node_server = server
+    executor.set_openengine_node_server(server)

tensorrt_llm/grpc/openengine/request_mapping.py

@@ -347,7 +347,7 @@ def sampling_params_from_request(
     # `_record_sampling_params_request_fields`). Without this, the materialized
     # defaults above would look client-provided and suppress a model's
     # generation_config.json sampling defaults under generation_config="auto".
-    sampling_params._set_request_provided_fields(provided_fields)
+    sampling_params.set_request_provided_fields(provided_fields)
     return sampling_params