feat(metrics): add use_legacy_loadgen_qps_metrics for LoadGen-parity QPS/TPS

AGENTS.md

@@ -114,7 +114,7 @@ Multi-process, event-loop design optimized for throughput:
 The aggregator is a separate process (`python -m inference_endpoint.async_utils.services.metrics_aggregator`) that subscribes to events and publishes `MetricsSnapshot` messages. Key facts for working in this layer:
 
 - **Series storage**: each `SeriesSampler` keeps three parallel views: O(1) cheap rollups (count/total/min/max/sum_sq, exact), an HDR Histogram (cheap live percentiles), and an in-memory `array.array` of raw values (for exact percentiles in the `COMPLETE` snapshot). Hot path is `registry.record(name, value)` — no allocation, no I/O.
-- **Counter API**: `registry.increment(name, delta=1)` for sample-event counters. `registry.set_counter(name, value)` only for the two duration counters (`total_duration_ns` max-of-elapsed, `tracked_duration_ns` sum-of-blocks).
+- **Counter API**: `registry.increment(name, delta=1)` for sample-event counters. `registry.set_counter(name, value)` only for the three derived-duration counters (`total_duration_ns` max-of-elapsed, `tracked_duration_ns` sum-of-blocks, `legacy_loadgen_window_duration_ns` first-issue→last-issued-completion span for LoadGen-parity QPS/TPS).
 - **Lifecycle**: `INITIALIZE` (constructed, awaiting first `STARTED`) → `LIVE` (run in progress, ticking every `--publish-interval` seconds) → `DRAINING` (set on `ENDED`; tick continues; bounded by `--drain-timeout` budget, default 60 s) → terminal: `COMPLETE` (clean end via `publish_final`, exact stats) **or** `INTERRUPTED` (signal-handler-triggered final via SIGTERM/SIGINT; best-effort partial stats). Drain timeout detected by consumers as `state == COMPLETE and n_pending_tasks > 0`; interrupted runs are detected as `state == INTERRUPTED` directly.
 - **Final delivery is dual-path with separated concerns**: `publish_final` atomically writes `final_snapshot.json` (`tmp + fsync(file) + rename + fsync(parent_dir)`) — this is the **primary** Report source — AND emits the terminal-state snapshot over pub/sub as a TUI shutdown signal. Each path is wrapped in its own try/except so one failure cannot suppress the other. Main process consumer reads `final_snapshot.json` (via `json.loads` to dict, no Struct decode); falls back to the subscriber's `latest` live snapshot only if the file is missing (e.g. SIGKILL / OOM before the signal handler ran). The dict form is the canonical consumer contract (see `snapshot_to_dict`).
 - **Histogram bucket edges are dynamic per snapshot**: log-spaced over the observed `[min, max]`. Bucket count is fixed at construction; consumers MUST re-render from the snapshot's `(lo, hi, count)` triples each frame and MUST NOT track bucket-by-index across snapshots.

scripts/regenerate_templates.py

@@ -280,9 +280,9 @@ def _add_comments(text: str, comments: dict[str, str]) -> str:
     for key, comment in sorted(comments.items(), key=lambda x: -len(x[0])):
         text = re.sub(
             rf"^(\s*{re.escape(key)}.*)$",
-            lambda m, c=comment: m.group(0)
-            if "#" in m.group(0)
-            else f"{m.group(0)}  {c}",
+            lambda m, c=comment: (
+                m.group(0) if "#" in m.group(0) else f"{m.group(0)}  {c}"
+            ),
             text,
             count=0,
             flags=re.MULTILINE,
@@ -326,6 +326,15 @@ def _build_full(model_cls: type[BenchmarkConfig], overrides: dict) -> dict:
     if overrides:
         data = _deep_merge(data, overrides)
 
+    # Mirror LoadPattern's model_serializer: use_legacy_loadgen_qps_metrics
+    # applies only to poisson, so drop it from other patterns' templates.
+    # TODO(vir): remove this prune when use_legacy_loadgen_qps_metrics is removed.
+    settings = data.get("settings")
+    if isinstance(settings, dict):
+        load_pattern = settings.get("load_pattern")
+        if isinstance(load_pattern, dict) and load_pattern.get("type") != "poisson":
+            load_pattern.pop("use_legacy_loadgen_qps_metrics", None)
+
     # Resolve streaming AUTO → off/on (mirrors schema validator)
     test_type = data.get("type")
     mp = data.get("model_params", {})

src/inference_endpoint/async_utils/services/metrics_aggregator/__main__.py

@@ -65,14 +65,20 @@ def _make_sigterm_handler(
     async def _signal_finalize() -> None:
         try:
             # Mirror the ENDED-driven path: refresh tracked_duration_ns
-            # from the table BEFORE publish_final, otherwise an
-            # interrupted run whose STOP_PERFORMANCE_TRACKING never
-            # fired would report duration_ns=0 and QPS=N/A in the final
-            # report even after processing many tracked samples.
+            # and the legacy LoadGen window from the table BEFORE
+            # publish_final, otherwise an interrupted run whose
+            # STOP_PERFORMANCE_TRACKING never fired would report
+            # duration_ns=0 (QPS=N/A) and a zero legacy window (silent
+            # native fallback) in the final report even after processing
+            # many tracked samples.
             registry.set_counter(
                 MetricCounterKey.TRACKED_DURATION_NS.value,
                 table.total_tracked_duration_ns,
             )
+            registry.set_counter(
+                MetricCounterKey.LEGACY_LOADGEN_WINDOW_DURATION_NS.value,
+                table.total_loadgen_window_ns,
+            )
             await publisher.publish_final(
                 registry,
                 n_pending_tasks=table.in_flight_tasks_count,

src/inference_endpoint/async_utils/services/metrics_aggregator/aggregator.py

@@ -70,6 +70,8 @@ class MetricCounterKey(str, Enum):
     # tracked row still exists when the aggregator sees the ERROR.
     TRACKED_SAMPLES_FAILED = "tracked_samples_failed"
     TRACKED_DURATION_NS = "tracked_duration_ns"
+    # Legacy MLPerf LoadGen Server "completed" window (final_query_all_samples_done_time).
+    LEGACY_LOADGEN_WINDOW_DURATION_NS = "legacy_loadgen_window_duration_ns"
     # Total wall-clock duration since session start. Updated on every event as
     # max(current, event_timestamp - session_start). Stored as a counter
     # rather than computed from (now - start) at read time because
@@ -320,6 +322,10 @@ async def process(self, records: list[EventRecord]) -> None:
                             MetricCounterKey.TRACKED_DURATION_NS.value,
                             table.total_tracked_duration_ns,
                         )
+                        registry.set_counter(
+                            MetricCounterKey.LEGACY_LOADGEN_WINDOW_DURATION_NS.value,
+                            table.total_loadgen_window_ns,
+                        )
                 logger.debug("Session event: %s", ev)
                 continue
 
@@ -392,6 +398,10 @@ async def process(self, records: list[EventRecord]) -> None:
                 MetricCounterKey.TRACKED_DURATION_NS.value,
                 table.total_tracked_duration_ns,
             )
+            registry.set_counter(
+                MetricCounterKey.LEGACY_LOADGEN_WINDOW_DURATION_NS.value,
+                table.total_loadgen_window_ns,
+            )
             try:
                 await self._publisher.publish_final(registry, n_pending_tasks=n_pending)
             finally:

src/inference_endpoint/async_utils/services/metrics_aggregator/metrics_table.py

@@ -451,6 +451,13 @@ def __init__(self, registry: MetricsRegistry) -> None:
         self.session_started_ns: int | None = None
         self.tracked_blocks: list[TrackedBlock] = []
 
+        # LoadGen window anchors: start at the FIRST issued tracked request
+        # (LoadGen t=0), end at the completion of the last-issued request that
+        # completed. See total_loadgen_window_ns.
+        self._loadgen_max_issued_ns: int = -1
+        self._loadgen_window_end_ns: int | None = None
+        self._loadgen_window_start_ns: int | None = None
+
     # --- Trigger registration ---
 
     def add_trigger(self, field_name: str, trigger: EmitTrigger) -> None:
@@ -501,6 +508,16 @@ def total_completed_tracked_samples(self) -> int:
         """Total samples completed across all tracking blocks."""
         return sum(b.completed_samples for b in self.tracked_blocks)
 
+    @property
+    def total_loadgen_window_ns(self) -> int:
+        """Window from the first issued tracked request (LoadGen t=0) to the
+        completion of the last-issued request that completed — the legacy MLPerf
+        LoadGen ``final_query_all_samples_done_time`` analog. Returns 0 (=>
+        the legacy metric falls back to native) when no tracked request completed."""
+        if self._loadgen_window_end_ns is None or self._loadgen_window_start_ns is None:
+            return 0
+        return max(0, self._loadgen_window_end_ns - self._loadgen_window_start_ns)
+
     # --- Field updates ---
 
     def set_field(
@@ -526,6 +543,8 @@ def set_field(
                 return
             row = self._create_row(sample_uuid)
             row.tracked_block_idx = len(self.tracked_blocks) - 1
+            if self._loadgen_window_start_ns is None:
+                self._loadgen_window_start_ns = value
         else:
             row = self._in_flight.get(sample_uuid)
             if row is None:
@@ -608,3 +627,8 @@ def _update_tracked_block(self, row: SampleRow, complete_ns: int) -> None:
             if complete_ns > block.last_complete_ns:
                 block.last_complete_ns = complete_ns
             block.completed_samples += 1
+        # End the legacy LoadGen window at the completion of the last-issued
+        # (largest issued_ns) request that completed.
+        if row.issued_ns is not None and row.issued_ns > self._loadgen_max_issued_ns:
+            self._loadgen_max_issued_ns = row.issued_ns
+            self._loadgen_window_end_ns = complete_ns

src/inference_endpoint/commands/benchmark/execute.py

@@ -967,20 +967,33 @@ def _on_phase_start(phase: PhaseConfig) -> None:
                 try:
                     runtime = ctx.config.settings.runtime
                     warmup = ctx.config.settings.warmup
+                    load_pattern = ctx.config.settings.load_pattern
                     report = Report.from_snapshot(
                         snap_dict,
                         seeds={
                             "scheduler_random_seed": runtime.scheduler_random_seed,
                             "dataloader_random_seed": runtime.dataloader_random_seed,
                             "warmup_random_seed": warmup.warmup_random_seed,
                         },
+                        use_legacy_loadgen_qps_metrics=(
+                            load_pattern.type == LoadPatternType.POISSON
+                            and load_pattern.use_legacy_loadgen_qps_metrics
+                        ),
                     )
                     if not report.complete:
                         logger.warning(
                             "Report is incomplete (state=%s, n_pending_tasks=%d)",
                             report.state,
                             snap_dict.get("n_pending_tasks", 0),
                         )
+                    if report.legacy_loadgen_window_duration_ns is not None:
+                        logger.warning(
+                            "Reporting QPS/TPS with the legacy MLPerf LoadGen Server "
+                            "'completed' definition (deprecated; to be removed once a "
+                            "formal tail-cutting mechanism lands). Pass "
+                            "--no-use-legacy-loadgen-qps-metrics for endpoints-native "
+                            "metrics."
+                        )
                 except Exception as e:  # noqa: BLE001 — best-effort report build.
                     logger.warning(f"Failed to build report from snapshot: {e}")
 

src/inference_endpoint/config/schema.py

@@ -34,9 +34,11 @@
     ConfigDict,
     Discriminator,
     Field,
+    SerializerFunctionWrapHandler,
     Tag,
     TypeAdapter,
     field_validator,
+    model_serializer,
     model_validator,
 )
 
@@ -466,6 +468,31 @@ class LoadPattern(BaseModel):
         cyclopts.Parameter(alias="--concurrency", help="Concurrent requests"),
     ] = Field(None, gt=0)
 
+    # TODO(vir): remove once the formal tail-cutting mechanism lands.
+    use_legacy_loadgen_qps_metrics: Annotated[
+        bool,
+        cyclopts.Parameter(
+            negative="--no-use-legacy-loadgen-qps-metrics",
+            help=(
+                "Only applies to the poisson load pattern. Report QPS/TPS using "
+                "the legacy MLPerf LoadGen Server 'completed' definition — (completed-1)/T "
+                "and tokens/T, T = first issued request to completion of the "
+                "last-issued request (see mlcommons/inference loadgen/results.cc). "
+                "--no-... uses endpoints-native completed/duration. Ignored for "
+                "non-poisson patterns."
+            ),
+        ),
+    ] = True
+
+    @model_serializer(mode="wrap")
+    def _serialize(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
+        # use_legacy_loadgen_qps_metrics only applies to poisson; drop it from
+        # the serialized form (and thus YAML templates) for other patterns.
+        data = handler(self)
+        if self.type != LoadPatternType.POISSON:
+            data.pop("use_legacy_loadgen_qps_metrics", None)
+        return data
+
     @model_validator(mode="after")
     def _validate_completeness(self) -> Self:
         if self.type == LoadPatternType.POISSON and (

src/inference_endpoint/config/templates/online_template_full.yaml

@@ -53,6 +53,7 @@ settings:
     type: poisson  # Load pattern type | options: max_throughput, poisson, concurrency, agentic_inference, burst, step
     target_qps: 10.0  # Target QPS
     target_concurrency: null  # Concurrent requests
+    use_legacy_loadgen_qps_metrics: true  # Only applies to the poisson load pattern. Report QPS/TPS using the legacy MLPerf LoadGen Server 'completed' definition — (completed-1)/T and tokens/T, T = first issued request to completion of the last-issued request (see mlcommons/inference loadgen/results.cc). --no-... uses endpoints-native completed/duration. Ignored for non-poisson patterns.
   client:
     num_workers: -1  # Worker processes (-1=auto)
     log_level: INFO  # Worker log level

src/inference_endpoint/metrics/report.py

@@ -130,6 +130,14 @@ class Report(msgspec.Struct, frozen=True):  # type: ignore[call-arg]
     tpot: dict[str, Any]
     latency: dict[str, Any]
     output_sequence_lengths: dict[str, Any]
+    # Legacy MLPerf LoadGen Server "completed" window (poisson only): first
+    # issued request -> completion of the last-issued request
+    # (final_query_all_samples_done_time analog; see mlcommons/inference
+    # loadgen/results.cc). Not None iff QPS/TPS were computed over this window;
+    # None means the endpoints-native full-run window was used. Recorded so
+    # result_summary.json is self-describing about which view it holds.
+    # TODO(vir): deprecate once endpoints has a formal tail-cutting mechanism.
+    legacy_loadgen_window_duration_ns: int | None = None
 
     # Derived throughput, computed once in from_snapshot so the serialized
     # report (result_summary.json) is self-complete. qps is None without a
@@ -146,7 +154,11 @@ class Report(msgspec.Struct, frozen=True):  # type: ignore[call-arg]
 
     @classmethod
     def from_snapshot(
-        cls, snap: dict[str, Any], *, seeds: dict[str, int] | None = None
+        cls,
+        snap: dict[str, Any],
+        *,
+        seeds: dict[str, int] | None = None,
+        use_legacy_loadgen_qps_metrics: bool = True,
     ) -> Report:
         """Build a Report from a snapshot dict.
 
@@ -169,6 +181,13 @@ def from_snapshot(
         honest "incomplete" report on missing fields instead of crashing:
         missing ``state`` defaults to ``"interrupted"`` (worst-case),
         missing counters / series to zero / empty.
+
+        The snapshot always carries BOTH ``tracked_duration_ns`` and
+        ``legacy_loadgen_window_duration_ns``, so it stays config-agnostic and
+        fully reinterpretable either way. Which window the reported QPS/TPS use
+        is decided by the run config (``use_legacy_loadgen_qps_metrics``,
+        recorded in ``config.yaml`` and in this Report's serialized JSON), not
+        by the snapshot.
         """
         counters: dict[str, int | float] = {}
         series: dict[str, dict[str, Any]] = {}
@@ -197,13 +216,39 @@ def _series_dict(key: str) -> dict[str, Any]:
         n_completed = _counter("tracked_samples_completed")
         osl = _series_dict("osl")
 
-        # Derived throughput. qps needs a duration; tps additionally needs OSL.
-        if duration_ns is None:
-            qps = tps = None
-        else:
+        # Legacy MLPerf LoadGen Server "completed" window (poisson only): first
+        # issued request -> completion of the last-issued request
+        # (final_query_all_samples_done_time analog; see mlcommons/inference
+        # loadgen/results.cc).
+        # TODO(vir): deprecate once endpoints has a formal tail-cutting mechanism.
+        raw_loadgen_window_ns = _counter("legacy_loadgen_window_duration_ns")
+
+        # Derived throughput, computed once so result_summary.json is
+        # self-complete. The legacy LoadGen window drives the headline QPS/TPS
+        # only when it is enabled, available, AND there are >=2 completions
+        # (QPS = (completed-1)/window is undefined below 2). If any of those
+        # fail, BOTH QPS and TPS fall back to the native window so they always
+        # share one window, and legacy_loadgen_window_duration_ns stays None so
+        # the serialized report honestly records which view it holds.
+        use_legacy_window = (
+            use_legacy_loadgen_qps_metrics
+            and raw_loadgen_window_ns > 0
+            and n_completed >= 2
+        )
+        legacy_loadgen_window_duration_ns = (
+            raw_loadgen_window_ns if use_legacy_window else None
+        )
+        if use_legacy_window:
+            window_s = raw_loadgen_window_ns / 1e9
+            qps = (n_completed - 1) / window_s
+            tps = (osl.get("total", 0) / window_s) if osl else None
+        elif duration_ns is not None:
             duration_s = duration_ns / 1e9
             qps = n_completed / duration_s
             tps = (osl.get("total", 0) / duration_s) if osl else None
+        else:
+            qps = None
+            tps = None
 
         # Default missing state to "interrupted" — a malformed / partial
         # snapshot dict is treated as worst-case (run did not reach a
@@ -226,6 +271,7 @@ def _series_dict(key: str) -> dict[str, Any]:
             tpot=_series_dict("tpot_ns"),
             latency=_series_dict("sample_latency_ns"),
             output_sequence_lengths=osl,
+            legacy_loadgen_window_duration_ns=legacy_loadgen_window_duration_ns,
             qps=qps,
             tps=tps,
             seeds=seeds,