fix: add endpoint liveness watchdog

Keep the guard disabled by default; recommend a 300-second or longer deadline when enabled. Preserve standard artifacts after a liveness failure and cover session-wide progress semantics.

Author: roborluo

README.md

@@ -90,6 +90,18 @@ Dataset Manager ──> Load Generator ──> Endpoint Client ──> External
 - **Online** (`poisson`): Fixed QPS with Poisson arrival distribution for latency profiling
 - **Concurrency**: Fixed concurrent request count
 
+### Endpoint liveness
+
+Set an endpoint response idle timeout to fail the run when in-flight work stops responding. Use `>=300` seconds; raise it for long requests:
+
+```yaml
+settings:
+  timeouts:
+    endpoint_response_idle_timeout_s: 300
+```
+
+Or pass `--endpoint-response-idle-timeout 300`. The guard is client-side; see [the configuration guide](docs/config/DESIGN.md#endpoint-response-idle-timeout) for tuning.
+
 ### Performance Design
 
 The hot path is optimized for minimal overhead:

docs/CLI_DESIGN.md

@@ -61,6 +61,8 @@ Both paths produce the **same subclass with the same defaults**. A YAML file wit
 
 3. **Datasets injected after construction.** `--dataset` strings are parsed by a `BeforeValidator` on the `datasets` field, then merged via `config.with_updates(datasets=...)`.
 
+Fields get their dotted path automatically; those needing a shorter operational spelling also declare a `cyclopts.Parameter(alias=...)`. For example `settings.timeouts.endpoint_response_idle_timeout_s` accepts either `--timeouts.endpoint-response-idle-timeout-s` or the alias `--endpoint-response-idle-timeout`; it defaults to off and, when enabled, we recommend `>=300` seconds.
+
 ### YAML path
 
 1. **`from_yaml_file(path)`** loads YAML, resolves `${VAR}` env vars on parsed values, then passes the dict to a Pydantic `TypeAdapter` with `Discriminator`.

docs/CLI_QUICK_REFERENCE.md

@@ -300,6 +300,7 @@ settings:
   timeouts:
     run_timeout_s: null # Whole-run watchdog; null = off
     performance_drain_timeout_s: null # null = unlimited
+    endpoint_response_idle_timeout_s: 300 # Endpoint liveness deadline; use >=300 s, null = off
   load_pattern:
     type: "max_throughput"
     target_qps: 10.0

docs/config/DESIGN.md

@@ -48,7 +48,7 @@ Key nested models:
 | Model            | Purpose                                             |
 | ---------------- | --------------------------------------------------- |
 | `LoadPattern`    | Pattern type + parameters (target QPS, concurrency) |
-| `RuntimeConfig`  | Duration, sample count, RNG seeds                   |
+| `RuntimeConfig`  | Workload sizing, issue-duration cap, and RNG seeds  |
 | `ClientSettings` | Worker count and HTTP client settings               |
 | `EndpointConfig` | Endpoint URLs, API key                              |
 | `Dataset`        | Dataset path, type (performance / accuracy)         |
@@ -71,6 +71,16 @@ Immutable snapshot of all parameters needed to execute a run.
 
 Once constructed, `RuntimeSettings` cannot be modified. All consumers receive the same instance.
 
+### Endpoint response idle timeout
+
+`settings.timeouts.endpoint_response_idle_timeout_s` is a client-side liveness deadline, `None` by default. When enabled, use at least 300 seconds in normal deployments. Once a phase has an in-flight request, `BenchmarkSession` fails the run if no streamed chunk or final response arrives within the interval. The clock starts at local issuance — when the request is handed to the client transport. There is no server-acceptance acknowledgment in the client contract, so worker scheduling, connection acquisition, and local transport queueing all consume the deadline before the server even sees the request. It does not replace an engine's own KV-transfer deadline.
+
+**What counts as progress.** Only messages that reach the main process reset the clock. Under the default `client.stream_all_chunks: false`, workers forward a request's first streamed chunk and its final result — intermediate chunks never cross the IPC boundary. The largest gap a healthy streaming request can therefore produce is its full first-chunk-to-completion span, not a per-token gap. With `stream_all_chunks: true` every chunk resets the clock, at the cost of per-chunk IPC traffic.
+
+**Choosing a value.** Use `>=300` seconds and set it above the deployment's longest expected gap between main-process-visible events. Under the default `stream_all_chunks: false` — and for any non-streaming endpoint — that is the full latency of the longest single request, including pre-accept overhead; use `600` seconds or more for reasoning or long-output workloads. For TensorRT-LLM disaggregated serving, match the executor `hang_detection_timeout` (default 300 s), not `cache_transceiver_config.kv_transfer_timeout_ms`; raise the value if the normal p99 gap between visible events is longer.
+
+**On failure**, the run still drains the metrics pipeline and writes the standard artifacts (`performance/result_summary.json`, `report.txt`) with the report marked `INTERRUPTED`, then exits non-zero.
+
 ### `BenchmarkSuiteRuleset` (abstract base)
 
 Extension point for competition-specific constraints.

docs/load_generator/DESIGN.md

@@ -35,6 +35,7 @@ each producing an independent report.
 BenchmarkSession.run(phases)
     |
     +-- STARTED
+    +-- [endpoint response idle watchdog]   armed only if endpoint_response_idle_timeout_s is set (default null)
     +-- [warmup]     strategy.execute() → drain_after=False (keep in-flight saturated)
     +-- [perf phase 1]   START_PERFORMANCE_TRACKING → strategy.execute() → drain → STOP_PERFORMANCE_TRACKING
     +-- [warmup]     strategy.execute() → drain_after=False (keep in-flight saturated)
@@ -173,7 +174,7 @@ class BenchmarkSession:
 **`run(phases)`** lifecycle:
 
 1. Publish `SessionEventType.STARTED`
-2. Start receiver coroutine (`_receive_responses`)
+2. Start receiver coroutine (`_receive_responses`). When the endpoint response idle timeout is set, a single session-liveness timer is armed while work is in flight and fails the session if every in-flight request is silent for the interval. Configure `>=300` seconds; see `docs/config/DESIGN.md` for tuning.
 3. For each phase:
    a. Create `SampleOrder` and `LoadStrategy` from phase settings
    b. Set `self._current_dataset` to phase dataset
@@ -296,13 +297,17 @@ async def _receive_responses(self):
                 self._strategy_task.cancel()
             break
         self._handle_response(resp)
+        if self._endpoint_response_idle_timeout_ns is not None:
+            self._last_response_progress_ns = time.monotonic_ns()
 ```
 
 Uses `recv()` exclusively — no `poll()` spin. The ZMQ fd is registered with
 the event loop, so `recv()` wakes exactly when a response is available with
 zero CPU overhead. Each `recv()` call yields to the event loop, ensuring
 strategy coroutines (call_at callbacks, semaphore waiters) are never starved.
 
+The endpoint response idle guard rides this same path to catch a client-visible worker or transport stall. It is not a per-request engine timeout: any chunk or final response counts as progress, and only a fully silent endpoint fails the run. A request may remain stuck while other in-flight requests respond; once those requests drain, the remaining stuck request causes the session to fail after the interval. The receiver only stamps a timestamp, and only when the deadline is set — streaming frame rate is QPS x output length, so per-chunk work is paid per token. The armed `call_later` timer re-derives its own deadline from that stamp when it fires, and re-arms for the remainder if progress raced it.
+
 For `ConcurrencyStrategy`, `_handle_response` calls `strategy.on_query_complete()`
 which releases the semaphore. Since `recv()` returns as soon as the fd is readable
 and `eager_task_factory` executes the woken semaphore waiter synchronously, there

src/inference_endpoint/commands/benchmark/execute.py

@@ -101,6 +101,7 @@
 from inference_endpoint.load_generator.conversation_manager import ConversationManager
 from inference_endpoint.load_generator.session import (
     BenchmarkSession,
+    EndpointResponseIdleTimeoutError,
     PhaseConfig,
     PhaseType,
     SessionResult,
@@ -170,6 +171,9 @@ class BenchmarkResult:
     profiling: dict[str, Any] | None = None
     run_timed_out: bool = False
     user_interrupted: bool = False
+    # Endpoint response idle timeout message when the endpoint stalled; the run still drains
+    # and finalizes its artifacts, then run_benchmark fails with this message.
+    stall_error: str | None = None
 
 
 @dataclass
@@ -871,6 +875,7 @@ async def _run_benchmark_async(
                     loop=loop,
                     on_sample_complete=on_sample_complete,
                     session_id=session_id,
+                    endpoint_response_idle_timeout_s=ctx.config.settings.timeouts.endpoint_response_idle_timeout_s,
                 )
                 watchdog.bind_session(session)
                 sigint.bind_session(session, abort_event)
@@ -883,6 +888,7 @@ async def _run_benchmark_async(
                 )
                 _perf_cap_done = False
                 session_completed_normally = False
+                stall_error: str | None = None
 
                 def _on_perf_phase_timeout() -> None:
                     if not _perf_cap_done:
@@ -929,6 +935,23 @@ def _on_phase_start(phase: PhaseConfig) -> None:
                             start_time_ns=0,
                             end_time_ns=0,
                         )
+                    elif isinstance(e, EndpointResponseIdleTimeoutError):
+                        # Endpoint stall is fatal but the drained data is still
+                        # worth keeping: fall through to the drain below so the
+                        # interrupted report and standard artifacts are written,
+                        # then run_benchmark fails with this message.
+                        logger.error(
+                            "Endpoint stalled — finalizing partial results "
+                            "before failing the run: %s",
+                            e,
+                        )
+                        stall_error = str(e)
+                        result = SessionResult(
+                            session_id=session_id,
+                            phase_results=[],
+                            start_time_ns=0,
+                            end_time_ns=0,
+                        )
                     else:
                         raise ExecutionError(f"Benchmark execution failed: {e}") from e
                 finally:
@@ -1010,6 +1033,7 @@ def _on_phase_start(phase: PhaseConfig) -> None:
         profiling=profiler.payload(),
         run_timed_out=watchdog.fired,
         user_interrupted=sigint.interrupted,
+        stall_error=stall_error,
     )
 
 
@@ -1157,6 +1181,7 @@ def finalize_benchmark(ctx: BenchmarkContext, bench: BenchmarkResult) -> None:
     aborted = (
         bench.run_timed_out
         or bench.user_interrupted
+        or bench.stall_error is not None
         or (report is not None and report.state == "interrupted")
     )
     if report is not None and aborted and report.state != "interrupted":
@@ -1253,6 +1278,10 @@ def run_benchmark(
                     f"Run timeout ({run_timeout_s}s) reached; run aborted and "
                     "report marked INTERRUPTED"
                 )
+            if bench.stall_error is not None:
+                raise ExecutionError(
+                    f"{bench.stall_error}; run aborted and report marked INTERRUPTED"
+                )
             if bench.report is None:
                 raise ExecutionError("Benchmark produced no usable metrics report")
             if not (report := bench.report).complete:

src/inference_endpoint/config/schema.py

@@ -801,6 +801,21 @@ class Timeouts(WithUpdatesMixin, BaseModel):
         gt=0,
         description="Whole-run watchdog seconds (None = off).",
     )
+    endpoint_response_idle_timeout_s: Annotated[
+        float | None,
+        cyclopts.Parameter(
+            alias="--endpoint-response-idle-timeout",
+        ),
+    ] = Field(
+        None,
+        gt=0,
+        description=(
+            "Endpoint liveness deadline: fail the run when work is in flight but "
+            "no response chunk or completion arrives for this many seconds "
+            "(None = off). When enabled, recommend >=300 seconds and size it above the longest "
+            "single-request latency. See docs/config/DESIGN.md."
+        ),
+    )
     interrupted_teardown_grace_s: float = Field(
         30.0,
         gt=0,

src/inference_endpoint/config/templates/concurrency_template_full.yaml

@@ -86,6 +86,7 @@ settings:
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   timeouts:  # All global waits and deadlines (see config/schema.py)
     run_timeout_s: null  # Whole-run watchdog seconds (None = off).
+    endpoint_response_idle_timeout_s: null  # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). When enabled, recommend >=300 seconds and size it above the longest single-request latency. See docs/config/DESIGN.md.
     interrupted_teardown_grace_s: 30.0  # Abort drain grace before remaining services are killed.
     service_ready_timeout_s: 30.0  # Service startup wait in seconds.
     warmup_drain_timeout_s: 240.0  # Warmup drain seconds (None = unlimited).

src/inference_endpoint/config/templates/offline_template_full.yaml

@@ -86,6 +86,7 @@ settings:
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   timeouts:  # All global waits and deadlines (see config/schema.py)
     run_timeout_s: null  # Whole-run watchdog seconds (None = off).
+    endpoint_response_idle_timeout_s: null  # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). When enabled, recommend >=300 seconds and size it above the longest single-request latency. See docs/config/DESIGN.md.
     interrupted_teardown_grace_s: 30.0  # Abort drain grace before remaining services are killed.
     service_ready_timeout_s: 30.0  # Service startup wait in seconds.
     warmup_drain_timeout_s: 240.0  # Warmup drain seconds (None = unlimited).

src/inference_endpoint/config/templates/online_template_full.yaml

@@ -87,6 +87,7 @@ settings:
     worker_gc_mode: relaxed  # Worker GC strategy | options: disabled, relaxed, system
   timeouts:  # All global waits and deadlines (see config/schema.py)
     run_timeout_s: null  # Whole-run watchdog seconds (None = off).
+    endpoint_response_idle_timeout_s: null  # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). When enabled, recommend >=300 seconds and size it above the longest single-request latency. See docs/config/DESIGN.md.
     interrupted_teardown_grace_s: 30.0  # Abort drain grace before remaining services are killed.
     service_ready_timeout_s: 30.0  # Service startup wait in seconds.
     warmup_drain_timeout_s: 240.0  # Warmup drain seconds (None = unlimited).