feat(trace): add -vvv binary trace events + live dashboard

Adds a low-overhead binary trace pipeline for inspecting per-request
lifecycle latency end-to-end (issue -> worker_received -> conn_acquired
-> write -> headers -> 1st chunk -> final chunk -> main_received ->
complete), plus per-process asyncio event-loop lag sampling.

- New `inference_endpoint.utils.trace` lock-free SPSC emitter (~90 ns
  per event), FIFO transport with O_NONBLOCK writes and 1 MiB kernel
  pipe buffer, per-process `emit_loop_lag` async task drives flush at
  0.3 s cadence — sole flush driver, no locks needed.
- New `scripts/trace_dashboard.py` rich.Live TUI reading the FIFO and
  the loadgen `final_snapshot.json` written by the metrics aggregator.
  Renders REQUEST LIFECYCLE / LOADGEN vs TRACE / EVENT LOOP LAG with
  one-row-per-metric layout (skips rows with no data on either side).
- Worker + main-process emit sites for the 9-stage lifecycle.
- 40 unit tests for the dashboard's count, fold, drop, loop-lag,
  loadgen comparison, and row-skip behavior.

Disabled by default; enabled with `-vvv` on `inference-endpoint`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: viraatc

scripts/trace_dashboard.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""CLI entry point for the -vvv trace dashboard.
+
+Reads fixed-size binary frames from the FIFO opened by
+:func:`inference_endpoint.utils.trace.bootstrap` and renders the
+dashboard via ``rich.Live``. Dashboard aggregation lives in
+:mod:`inference_endpoint.utils.trace_dashboard` so it can be unit
+tested without standing up a TUI.
+
+Linux only: timestamps are compared across processes and rely on
+``CLOCK_MONOTONIC`` being system-wide (per ``man 7 time``).
+"""
+
+# ruff: noqa: I001
+# The pre-commit ruff hook is pinned to v0.3.3 (see
+# .pre-commit-config.yaml's "TODO: sync rev with ruff version"), which
+# does not auto-detect `inference_endpoint` as a first-party package
+# and therefore disagrees with the project's local ruff (v0.15.8) on
+# import order in this file. File-level noqa keeps both versions quiet
+# until the rev is synced.
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import threading
+import time
+
+from inference_endpoint.utils.trace import FRAME_SIZE, snapshot_sidecar_path
+from inference_endpoint.utils.trace_dashboard import (
+    DASHBOARD_THEME,
+    READ_CHUNK,
+    REFRESH_HZ,
+    Dashboard,
+)
+from rich.console import Console
+from rich.live import Live
+
+
+def _try_load_snapshot(path: str) -> dict | None:
+    """Best-effort read of the loadgen snapshot sidecar. Returns None
+    if the file is missing or transiently mid-rename (atomic write may
+    briefly produce a half-rename window that json.load tolerates)."""
+    try:
+        with open(path) as f:
+            return json.load(f)
+    except (OSError, json.JSONDecodeError):
+        return None
+
+
+class _FrameReader(threading.Thread):
+    """Blocking read loop; ingests whole frames into the Dashboard."""
+
+    def __init__(self, fd: int, dash: Dashboard) -> None:
+        super().__init__(daemon=True, name="trace-reader")
+        self._fd = fd
+        self._dash = dash
+        self._pending = bytearray()
+        self._eof = threading.Event()
+
+    @property
+    def eof(self) -> bool:
+        return self._eof.is_set()
+
+    def run(self) -> None:
+        try:
+            while True:
+                try:
+                    chunk = os.read(self._fd, READ_CHUNK)
+                except OSError:
+                    return
+                if not chunk:
+                    return
+                self._pending.extend(chunk)
+                whole = (len(self._pending) // FRAME_SIZE) * FRAME_SIZE
+                if whole:
+                    self._dash.ingest_frames(bytes(self._pending[:whole]))
+                    del self._pending[:whole]
+        finally:
+            self._eof.set()
+
+
+def _open_trace_input(pipe_path: str | None) -> int:
+    if pipe_path:
+        return os.open(pipe_path, os.O_RDONLY)
+    return sys.stdin.fileno()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--trace-pipe",
+        help="FIFO path to read binary trace frames from (default: stdin).",
+    )
+    args = parser.parse_args()
+
+    dash = Dashboard()
+    # Dashboard renders to stderr because the parent benchmark process
+    # has redirected its own stdout/stderr to a log file (see trace.bootstrap).
+    console = Console(file=sys.stderr, force_terminal=True, theme=DASHBOARD_THEME)
+    reader = _FrameReader(_open_trace_input(args.trace_pipe), dash)
+    reader.start()
+
+    # Snapshot sidecar path is convention-named after the parent's pid
+    # (main proc that spawned us). The benchmark writes it periodically
+    # under -vvv so we can live-update the LOADGEN vs TRACE panel.
+    snap_path = snapshot_sidecar_path(os.getppid())
+
+    # screen=True uses the alternate-screen buffer so updates redraw
+    # cleanly without scrollback noise. When Live() exits the alt
+    # screen is torn down — to keep the final frame visible we capture
+    # it BEFORE leaving the context and print it to the normal buffer
+    # afterward.
+    final_frame = None
+    with Live(
+        dash.render(),
+        console=console,
+        refresh_per_second=REFRESH_HZ,
+        screen=True,
+        transient=False,
+    ) as live:
+        while not reader.eof:
+            snap = _try_load_snapshot(snap_path)
+            if snap is not None:
+                dash.attach_loadgen_snapshot(snap)
+            live.update(dash.render())
+            time.sleep(1.0 / REFRESH_HZ)
+        snap = _try_load_snapshot(snap_path)
+        if snap is not None:
+            dash.attach_loadgen_snapshot(snap)
+        # Bypass the per-tick fold-defer window so the final render
+        # captures COMPLETE frames that landed within the last 300 ms
+        # — otherwise they sit queued and the closing frame shows
+        # stale stage histograms / verdict.
+        dash.flush_pending_folds()
+        final_frame = dash.render()
+        live.update(final_frame)
+    # Now we're back on the normal screen — print the last snapshot
+    # so the user can see the totals + verdict after the run ends.
+    if final_frame is not None:
+        console.print(final_frame)
+        console.print("[dim]── trace finished ──[/dim]")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/inference_endpoint/commands/benchmark/execute.py

@@ -26,6 +26,7 @@
 import asyncio
 import json
 import logging
+import os
 import random
 import shutil
 import signal
@@ -96,6 +97,7 @@
     SessionResult,
 )
 from inference_endpoint.metrics.report import Report
+from inference_endpoint.utils import trace
 
 transformers_logging.set_verbosity_error()
 
@@ -556,6 +558,14 @@ async def _run_benchmark_async(
     config = ctx.config
     session_id = f"cli_benchmark_{uuid.uuid4().hex[:8]}"
 
+    # Trace mode (-vvv): drive periodic emitter flush + main-proc loop
+    # lag sampling on this loop. The same task does both so we have
+    # exactly one trace consumer per process on the producer's thread.
+    # Held in a local so the teardown block below can cancel it.
+    lag_task: asyncio.Task[None] | None = None
+    if trace.is_enabled():
+        lag_task = loop.create_task(trace.emit_loop_lag(trace.MAIN_PROC_LOOP_ID))
+
     # Progress bar + response collector
     pbar = tqdm(
         desc=f"{config.model_params.name} (Streaming: {ctx.enable_streaming})",
@@ -603,6 +613,19 @@ async def _run_benchmark_async(
         )
         metrics_subscriber.start()
 
+        # -vvv only: tap the latest snapshot to a JSON sidecar so the
+        # trace dashboard can show "LOADGEN vs TRACE" live. Path is
+        # derived from our pid; dashboard finds it via os.getppid().
+        # Held in a local so we can cancel it in the teardown block —
+        # otherwise it would race the subscriber's close().
+        tap_task: asyncio.Task[None] | None = None
+        if trace.is_enabled():
+            tap_task = loop.create_task(
+                _snapshot_tap_loop(
+                    metrics_subscriber, trace.snapshot_sidecar_path(os.getpid())
+                )
+            )
+
         # Launch service subprocesses
         launcher = ServiceLauncher(zmq_ctx)
         aggregator_args: list[str] = [
@@ -663,7 +686,15 @@ async def _run_benchmark_async(
                 event_logs_dir=ctx.report_dir,
                 cpu_affinity=ctx.affinity_plan,
             )
-            http_client = await HTTPEndpointClient.create(http_config, loop)
+            # -vvv on main proc → tell HTTPEndpointClient which FIFO
+            # workers should attach to. Threaded as an explicit arg so
+            # the trace channel stays out of the public CLI / YAML
+            # config schema (would otherwise leak into generated YAML
+            # templates and create a CLI flag users could misuse).
+            trace_pipe = trace.fifo_path(os.getpid()) if trace.is_enabled() else None
+            http_client = await HTTPEndpointClient.create(
+                http_config, loop, trace_pipe_path=trace_pipe
+            )
             issuer = HttpClientSampleIssuer(http_client)
         except Exception as e:
             pbar.close()
@@ -813,6 +844,33 @@ def _on_phase_start(phase: PhaseConfig) -> None:
                 except Exception as e:  # noqa: BLE001 — best-effort report build.
                     logger.warning(f"Failed to build report from snapshot: {e}")
 
+            # Trace-mode (-vvv) only: flush the terminal snapshot to
+            # the sidecar BEFORE cancelling the tap task. The tap loop
+            # ticks every 0.5s, so a short run can finalize between
+            # ticks and the dashboard would read stale data for its
+            # final LOADGEN vs TRACE frame.
+            if trace.is_enabled() and snap_dict is not None:
+                snap_path = trace.snapshot_sidecar_path(os.getpid())
+                tmp_path = f"{snap_path}.tmp"
+                try:
+                    with open(tmp_path, "w") as f:
+                        json.dump(snap_dict, f)
+                    os.replace(tmp_path, snap_path)
+                except OSError as e:
+                    logger.warning(f"final snapshot sidecar write failed: {e}")
+
+            # Cancel trace-mode helper tasks before closing the
+            # subscriber so they don't tick on a closed pipe.
+            for t in (tap_task, lag_task):
+                if t is not None and not t.done():
+                    t.cancel()
+                    try:
+                        await t
+                    except (asyncio.CancelledError, Exception):
+                        # CancelledError is the normal cancel ack;
+                        # other exceptions during shutdown are logged
+                        # by the task itself and not worth surfacing.
+                        pass
             metrics_subscriber.close()
             pbar.close()
 
@@ -824,6 +882,35 @@ def _on_phase_start(phase: PhaseConfig) -> None:
     )
 
 
+async def _snapshot_tap_loop(
+    subscriber: MetricsSnapshotSubscriber,
+    out_path: str,
+    period_s: float = 0.5,
+) -> None:
+    """Trace-mode helper. Periodically serialize the latest
+    ``MetricsSnapshot`` from the subscriber to ``out_path`` as JSON via
+    atomic tmp + rename so the dashboard's poll never reads a partial
+    file. Stops naturally when the task is cancelled at benchmark
+    teardown.
+    """
+    logger.debug("snapshot tap → %s every %.2fs", out_path, period_s)
+    tmp_path = f"{out_path}.tmp"
+    while True:
+        try:
+            await asyncio.sleep(period_s)
+            snap = subscriber.latest
+            if snap is None:
+                continue
+            payload = snapshot_to_dict(snap)
+            with open(tmp_path, "w") as f:
+                json.dump(payload, f)
+            os.replace(tmp_path, out_path)
+        except asyncio.CancelledError:
+            return
+        except Exception:  # noqa: BLE001 — telemetry, never crash the bench
+            logger.debug("snapshot tap write failed", exc_info=True)
+
+
 def run_benchmark_async(ctx: BenchmarkContext) -> BenchmarkResult:
     """Run async benchmark. Sync entry point — drives the event loop."""
     loop = LoopManager().default_loop

src/inference_endpoint/endpoint_client/http.py

@@ -791,13 +791,14 @@ def build_request(
 
 @dataclass(slots=True)
 class InFlightRequest:
-    """State for a single HTTP request through its lifecycle:
+    """State for a single HTTP request through its lifecycle.
 
     Attributes:
         query_id: Correlates response back to original Query.
         http_bytes: Serialized HTTP request for socket.write().
         is_streaming: Whether this is a streaming (SSE) request or not.
-        connection: PooledConnection assigned to this request (set once request is fired).
+        connection: PooledConnection assigned to this request (set once
+            request is fired).
     """
 
     query_id: str

src/inference_endpoint/endpoint_client/http_client.py

@@ -49,9 +49,11 @@ def __init__(
         self,
         config: HTTPClientConfig,
         loop: asyncio.AbstractEventLoop | None = None,
+        trace_pipe_path: str | None = None,
     ):
         self.client_id = uuid.uuid4().hex[:8]
         self.config = config
+        self._trace_pipe_path = trace_pipe_path
         self._worker_cycle = cycle(range(self.config.num_workers))
 
         # Use provided loop or create one via LoopManager (uvloop + eager task factory)
@@ -87,17 +89,24 @@ async def create(
         cls,
         config: HTTPClientConfig,
         loop: asyncio.AbstractEventLoop,
+        trace_pipe_path: str | None = None,
     ) -> "HTTPEndpointClient":
         """Async factory for shared-loop usage.
 
         Use this instead of __init__ when the caller is already running on
         the target event loop (e.g., inside run_benchmark_async). The regular
         constructor uses run_coroutine_threadsafe().result() which deadlocks
         when called from the same loop.
+
+        ``trace_pipe_path`` opts each worker into the -vvv trace channel
+        (the FIFO main proc created in ``trace.bootstrap``). It rides
+        with the client rather than ``config`` so the trace plumbing
+        stays out of the public CLI / YAML schema.
         """
         self = cls.__new__(cls)
         self.client_id = uuid.uuid4().hex[:8]
         self.config = config
+        self._trace_pipe_path = trace_pipe_path
         self._worker_cycle = cycle(range(config.num_workers))
         self._owns_loop = False
         self._loop_name = None
@@ -118,7 +127,9 @@ async def _initialize(self) -> None:
         self._dropped_requests: int = 0
 
         assert self.loop is not None
-        self.worker_manager = WorkerManager(self.config, self.loop)
+        self.worker_manager = WorkerManager(
+            self.config, self.loop, trace_pipe_path=self._trace_pipe_path
+        )
         await self.worker_manager.initialize()
         self.pool = self.worker_manager.pool_transport