From c79576c32eef327ff852e8b35637ec6145622266 Mon Sep 17 00:00:00 2001
From: Kevin <kwang3170@berkeley.edu>
Date: Thu, 23 Jul 2026 02:27:43 +1000
Subject: [PATCH 1/3] Serving latency, collective signal, and verification
 export
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three signal gaps in the loop, each a separate concern sharing one file
(gitm/scheduler/loop.py), so they land as one commit rather than a broken
intermediate state.

1. Serving latency (TTFT/TPOT/goodput) + request join
   The loop measured only aggregate tokens/sec and had no per-request view.
   vllm_stats gains RequestRecord (TTFT/TPOT derived from vLLM's
   RequestOutput.metrics), ServingSummary, and summarize_requests. TPOT
   divides by n-1 (the first token's cost is TTFT) and is None for a
   one-token response rather than a rate from a zero-length gap.

   The join: SchedulerStatsSampler now takes a wall clock alongside its
   monotonic one, surfaced as SchedulerStatsSummary.t0_wall_ns, putting the
   scheduler series, request timestamps, and Trace.captured_at_ns on one
   clock. Persisted to scheduler_stats.json.

   Signals only — the A/B objective is untouched.

2. Collective (NCCL) time as a ranked cause
   NCCL kernels count as busy time, so a communication-bound run reported as
   well-utilized. collective_signal reuses node_rollup's matcher and interval
   math and adds only the ranking: CollectiveCause (field-for-field identical
   to SchedulerCause) plus two rules, exposed_collective and
   collective_dominant. Emitted into residuals.json.

   worst_device_comm splits by device_id first: device_comm_stats assumes a
   single-device trace, and on a merged multi-GPU capture one GPU's compute
   masks another's communication, collapsing exposed comm to zero on a fully
   blocked GPU.

3. Customer-verification export
   EngineABResult was read for two numbers and discarded. verification_export
   writes it to verification.json with the full engine config on both sides,
   reps, scatter, and environment.

   kept follows ApplyResult.rolled_back (the gate) rather than
   EngineABResult.kept (a measure-time delta>=0 indicator) so the export never
   claims a lever was kept that was rolled back. agreement_band is floored at
   MIN_NOISE_BAND because a single-rep A/B reports zero scatter, which would
   make ordinary jitter read as a failed reproduction.

   Export only — no generated reproduce script, report.py untouched.

30 new tests. Full suite unchanged: the 5 pre-existing failures (missing
pyarrow, HFT dataset generation) fail identically on main.

Not verified on GPU: whether RequestOutput.metrics populates on the target
vLLM build (degrades to None fields), and whether gitm_llm_kwargs survives
the restart path in a real structural-knob A/B.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 gitm/optimizer/collective_signal.py   | 147 ++++++++++++++++++++
 gitm/optimizer/verification_export.py | 188 +++++++++++++++++++++++++
 gitm/scheduler/loop.py                |  67 ++++++++-
 gitm/tracer/vllm_stats.py             | 192 +++++++++++++++++++++++++-
 gitm/workloads.py                     |  11 +-
 tests/test_collective_signal.py       | 126 +++++++++++++++++
 tests/test_serving_latency.py         | 149 ++++++++++++++++++++
 tests/test_verification_export.py     | 144 +++++++++++++++++++
 8 files changed, 1016 insertions(+), 8 deletions(-)
 create mode 100644 gitm/optimizer/collective_signal.py
 create mode 100644 gitm/optimizer/verification_export.py
 create mode 100644 tests/test_collective_signal.py
 create mode 100644 tests/test_serving_latency.py
 create mode 100644 tests/test_verification_export.py

diff --git a/gitm/optimizer/collective_signal.py b/gitm/optimizer/collective_signal.py
new file mode 100644
index 0000000..8755533
--- /dev/null
+++ b/gitm/optimizer/collective_signal.py
@@ -0,0 +1,147 @@
+"""Collective (NCCL) time → causal attribution.
+
+The kernel-level residuals explain *compute*; on a multi-GPU run a large share
+of step time can instead be collective communication — all-reduces and
+all-gathers synchronizing partial results. Those are ordinary CUDA kernels on
+the timeline, so the busy-fraction math counts them as work: a run blocked on
+communication reads as a well-utilized GPU with nothing to fix.
+
+:func:`collective_causes` turns the comm stats already computed by
+:func:`gitm.importers.node_rollup.device_comm_stats` into ranked
+:class:`CollectiveCause` hypotheses, so communication enters attribution the
+same way engine-scheduler signals do (see
+:mod:`gitm.optimizer.scheduler_attribution`) and each cause names the levers it
+argues for.
+
+The measurement that carries the information is **exposed** communication —
+comm that does *not* overlap compute. Overlapped comm is hidden behind useful
+work and costs nothing; only exposed comm is a bottleneck. Both are already
+computed by ``device_comm_stats``; this module only ranks them.
+
+Like the scheduler causes, these are rule-based severity-ranked observations
+grounded in what the timeline shows, deliberately not dressed up as statistical
+tests over a handful of kernels.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from gitm.importers.node_rollup import DeviceCommStats, device_comm_stats
+from gitm.tracer.schema import Trace
+
+# Levers a communication bottleneck argues for. Unlike the scheduler knobs these
+# are topology-level and structural (they need an engine rebuild), which is why
+# the notes say "consider" rather than promising a hot-swap.
+_TOPOLOGY_KNOBS = [
+    "tensor_parallel_size",
+    "pipeline_parallel_size",
+    "distributed_executor_backend",
+]
+
+
+def worst_device_comm(trace: Trace) -> DeviceCommStats | None:
+    """Comm stats for the device with the most *exposed* communication.
+
+    ``device_comm_stats`` assumes a single-device trace — the importer path
+    splits per device before calling it. A live CUPTI capture of a multi-GPU run
+    holds every device's kernels in one trace, and calling it directly there is
+    quietly wrong: one GPU's compute overlaps another GPU's communication, so
+    exposed comm collapses toward zero and a communication-bound run reports
+    clean. Splitting by ``device_id`` first keeps the overlap math inside a
+    single device, where "was compute running while this GPU communicated?" is a
+    real question.
+
+    The worst device is the one reported because a collective is a barrier: the
+    step waits on the GPU that spent longest exposed.
+    """
+    kernels = trace.kernels()
+    if not kernels:
+        return None
+    by_dev: dict[int, list] = {}
+    for k in kernels:
+        by_dev.setdefault(k.device_id, []).append(k)
+    if len(by_dev) == 1:
+        return device_comm_stats(trace)
+    per_device = [
+        device_comm_stats(trace.model_copy(update={"events": evs}))
+        for _, evs in sorted(by_dev.items())
+    ]
+    return max(per_device, key=lambda c: c.exposed_comm_share_of_wall)
+
+
+@dataclass
+class CollectiveCause:
+    """One communication-level causal hypothesis, ranked by ``severity`` (0..1).
+
+    Mirrors :class:`~gitm.optimizer.scheduler_attribution.SchedulerCause` field
+    for field so both kinds of cause serialize and rank identically.
+    """
+
+    signal: str  # e.g. "exposed_collective"
+    effect: str  # the symptom it explains
+    severity: float  # 0..1, how strongly the condition held
+    note: str  # human-readable cause → implied fix
+    motivates_knobs: list[str]  # library knobs this cause argues for
+
+
+def collective_causes(
+    stats: DeviceCommStats | None,
+    *,
+    exposed_floor: float = 0.05,
+    dominant_floor: float = 0.20,
+) -> list[CollectiveCause]:
+    """Rank communication causes from comm stats (empty when there is no comm).
+
+    ``exposed_floor`` is the share of wall time spent in comm that no compute
+    was hiding; ``dominant_floor`` is the share of *busy* time that is comm at
+    all. Conservative defaults — a few percent of exposed comm is normal even on
+    a healthy run.
+    """
+    if stats is None or stats.comm_ns <= 0:
+        return []
+
+    causes: list[CollectiveCause] = []
+
+    # Exposed comm — the GPU had nothing else to run while communicating. This is
+    # the actionable one: it is time that overlap could have reclaimed.
+    if stats.exposed_comm_share_of_wall > exposed_floor:
+        over = stats.exposed_comm_share_of_wall - exposed_floor
+        causes.append(
+            CollectiveCause(
+                signal="exposed_collective",
+                effect="step time inflated by non-overlapped communication",
+                # Normalize against the remaining headroom above the floor, so a
+                # run that is *entirely* exposed comm saturates at 1.0.
+                severity=min(1.0, over / max(1.0 - exposed_floor, 1e-6)),
+                note=(
+                    f"{stats.exposed_comm_share_of_wall:.0%} of wall time is collective "
+                    "communication with no compute overlapping it. Consider a smaller "
+                    "tensor_parallel_size (less cross-GPU traffic per step) or an "
+                    "executor backend that overlaps comm with compute."
+                ),
+                motivates_knobs=list(_TOPOLOGY_KNOBS),
+            )
+        )
+
+    # Comm-dominant — a large fraction of everything the GPU did was communication,
+    # overlapped or not. Argues the parallelism topology is over-split.
+    if stats.comm_share_of_busy > dominant_floor:
+        over = stats.comm_share_of_busy - dominant_floor
+        causes.append(
+            CollectiveCause(
+                signal="collective_dominant",
+                effect="GPU busy time dominated by communication, not compute",
+                severity=min(1.0, over / max(1.0 - dominant_floor, 1e-6)),
+                note=(
+                    f"{stats.comm_share_of_busy:.0%} of GPU busy time is collective "
+                    "kernels. The parallelism topology may be over-split for this "
+                    "model size — a lower tensor_parallel_size trades communication "
+                    "for per-GPU memory."
+                ),
+                motivates_knobs=list(_TOPOLOGY_KNOBS),
+            )
+        )
+
+    causes.sort(key=lambda c: c.severity, reverse=True)
+    return causes
diff --git a/gitm/optimizer/verification_export.py b/gitm/optimizer/verification_export.py
new file mode 100644
index 0000000..1067fc0
--- /dev/null
+++ b/gitm/optimizer/verification_export.py
@@ -0,0 +1,188 @@
+"""Customer-verification export — the A/B result as data, not prose.
+
+The provenance report (:mod:`gitm.optimizer.report`) is written for a human: it
+says a lever was kept and by how much. That is not enough for a customer who
+wants to re-measure the claim on their own harness, because the report gives a
+percentage and a sentence — not the configuration each number was measured
+under, nor how much the measurement itself scattered.
+
+This module writes the same A/B as structured data: both throughputs, the full
+engine configuration on each side, the scatter and rep count behind them, and
+the environment they were measured in. Nothing here measures anything new — the
+numbers all come from :class:`~gitm.optimizer.apply.EngineABResult`, which the
+loop already produces and currently reads two fields from.
+
+Two things are deliberate:
+
+* **``kept`` comes from the gate, not the measurement.** ``EngineABResult.kept``
+  is a measure-time ``delta >= 0`` indicator; the authoritative keep/rollback
+  decision is ``ApplyResult.rolled_back``, which applies the caller's
+  ``min_keep_delta``. Exporting the indicator would occasionally claim a lever
+  was kept that the gate actually rolled back.
+* **An agreement band travels with every number.** At ``reps=1`` the measured
+  scatter is exactly 0, so a customer re-measuring would "disagree" with us over
+  ordinary run-to-run jitter. :data:`MIN_NOISE_BAND` floors the band so the
+  export states the precision it actually has instead of implying certainty it
+  doesn't.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from gitm.kernels.spec import InterventionSpec
+    from gitm.optimizer.apply import ApplyResult, EngineABResult
+    from gitm.optimizer.report import Provenance
+
+#: Export format version. Bump on any breaking field change so a consumer can
+#: tell which shape it is reading.
+SCHEMA = 1
+
+#: Floor on the relative band within which a re-measurement counts as agreeing
+#: with ours. A single-rep A/B reports zero scatter, which would imply a
+#: precision no GPU benchmark has; 2% is a conservative stand-in for ordinary
+#: run-to-run variation on decode throughput.
+MIN_NOISE_BAND = 0.02
+
+
+@dataclass
+class VerificationRecord:
+    """One baseline↔candidate comparison, in full.
+
+    ``baseline_config`` / ``candidate_config`` are the engine's complete kwargs
+    on each side, not just the knob that moved: a customer reproducing at a
+    different ``gpu_memory_utilization`` measures a different system, and
+    without both configs neither side can tell a real disagreement from a setup
+    difference.
+    """
+
+    intervention_name: str
+    summary: str
+    knob: str
+    value: Any
+    source: str  # the citation the lever came from
+
+    baseline_tps: float
+    candidate_tps: float
+    speedup: float
+    delta: float  # speedup - 1, the signed change
+
+    baseline_std: float
+    candidate_std: float
+    reps: int
+    agreement_band: float  # relative; a re-measurement inside this agrees
+    significant: bool  # the gain cleared the measured noise band
+
+    kept: bool  # from the rollback gate, not the measure-time indicator
+    via: str  # "hot-swap" | "restart"
+
+    baseline_config: dict[str, Any] = field(default_factory=dict)
+    candidate_config: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def build_record(
+    spec: InterventionSpec,
+    ab: EngineABResult,
+    apply_result: ApplyResult,
+    *,
+    baseline_config: dict[str, Any] | None = None,
+    candidate_config: dict[str, Any] | None = None,
+) -> VerificationRecord:
+    """Assemble one record from a live A/B and its gate decision.
+
+    ``baseline_config`` must be captured *before* the apply — the engine's
+    kwargs are mutated in place by a hot-swap and replaced entirely by a
+    restart, so reading them afterwards yields the candidate on both sides.
+    """
+    return VerificationRecord(
+        intervention_name=spec.name,
+        summary=spec.summary,
+        knob=spec.knob,
+        value=spec.value,
+        source=spec.source,
+        baseline_tps=ab.baseline_tps,
+        candidate_tps=ab.candidate_tps,
+        speedup=ab.speedup,
+        delta=ab.speedup - 1.0,
+        baseline_std=ab.baseline_std,
+        candidate_std=ab.candidate_std,
+        reps=ab.reps,
+        agreement_band=max(ab.rel_std, MIN_NOISE_BAND),
+        significant=ab.significant,
+        # The gate decides, not the measurement — see the module docstring.
+        kept=not apply_result.rolled_back,
+        via=ab.via,
+        baseline_config=dict(baseline_config or {}),
+        candidate_config=dict(candidate_config or {}),
+    )
+
+
+def _environment(gpu_sku: str | None) -> dict[str, Any]:
+    """Best-effort description of the box the numbers were measured on.
+
+    Every field degrades to ``None`` rather than a guess: an export that names
+    the wrong GPU is worse than one that admits it doesn't know.
+    """
+    from gitm.cuda_env import driver_cuda, torch_cuda
+
+    def _ver(v: tuple[int, int] | None) -> str | None:
+        return f"{v[0]}.{v[1]}" if v else None
+
+    return {
+        "gpu_sku": gpu_sku,
+        "driver_cuda": _ver(driver_cuda()),
+        "torch_cuda": _ver(torch_cuda()),
+    }
+
+
+def build_export(
+    records: list[VerificationRecord],
+    provenance: Provenance,
+    *,
+    gpu_sku: str | None = None,
+) -> dict[str, Any]:
+    """The full export document: provenance + environment + every comparison."""
+    return {
+        "schema": SCHEMA,
+        "provenance": {
+            "workload_id": provenance.workload_id,
+            "fingerprint": provenance.fingerprint,
+            "run_id": provenance.run_id,
+            "git_sha": provenance.git_sha,
+            "gitm_version": provenance.gitm_version,
+        },
+        "environment": _environment(gpu_sku),
+        "protocol": {
+            "metric": "decode throughput (tokens/sec)",
+            "reps": "each side benchmarked `reps` times; std is the sample stdev",
+            "agreement_band": (
+                "relative band around our numbers within which a re-measurement "
+                f"agrees; floored at {MIN_NOISE_BAND:.0%} because a single-rep A/B "
+                "reports zero scatter"
+            ),
+            "kept": "decided by the rollback gate (min_keep_delta), not by delta >= 0",
+        },
+        "results": [r.to_dict() for r in records],
+    }
+
+
+def write_verification(
+    records: list[VerificationRecord],
+    provenance: Provenance,
+    out_path: str | Path,
+    *,
+    gpu_sku: str | None = None,
+) -> str:
+    """Write the export as JSON to ``out_path``. Returns the path written."""
+    out_path = Path(out_path)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    doc = build_export(records, provenance, gpu_sku=gpu_sku)
+    out_path.write_text(json.dumps(doc, indent=2, default=str) + "\n")
+    return str(out_path)
diff --git a/gitm/scheduler/loop.py b/gitm/scheduler/loop.py
index 1f39309..b47cee2 100644
--- a/gitm/scheduler/loop.py
+++ b/gitm/scheduler/loop.py
@@ -35,6 +35,7 @@
     apply_intervention,
 )
 from gitm.optimizer.attribution import attribute
+from gitm.optimizer.collective_signal import collective_causes, worst_device_comm
 from gitm.optimizer.deviation import deviation_summary, deviation_trace, write_deviation_jsonl
 from gitm.optimizer.dr import attribute_dr
 from gitm.optimizer.measure import measure_trace, measurement_claims, measurement_summary
@@ -42,6 +43,11 @@
 from gitm.optimizer.qualification import qualify
 from gitm.optimizer.report import Claim, build_provenance, write_report
 from gitm.optimizer.scheduler_attribution import scheduler_causes
+from gitm.optimizer.verification_export import (
+    VerificationRecord,
+    build_record,
+    write_verification,
+)
 from gitm.optimizer.vllm_knobs import (
     expand_relative_candidates,
     knob_kind,
@@ -51,7 +57,7 @@
 from gitm.planner.graph import predict_graph
 from gitm.safety.audit import AuditLog, _write_report
 from gitm.tracer.capture import capture
-from gitm.tracer.vllm_stats import sample_scheduler_stats
+from gitm.tracer.vllm_stats import sample_scheduler_stats, summarize_requests
 from gitm.workloads import WorkloadRunner, get_factory, sync_device
 
 _BUDGET_RE = re.compile(r"^\s*(\d+(?:\.\d+)?)\s*([smhd])\s*$")
@@ -297,13 +303,14 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]:
     # Sample the engine scheduler (queue depth, batch occupancy, preemptions)
     # over the same window as the CUPTI capture — engine-level telemetry the GPU
     # trace can't see. A no-op when no engine is attached (empty series).
+    run_out: Any = None
     with (
         capture(trace_path, workload_id=workload, run_id=run_id) as trace,
         sample_scheduler_stats(cfg.engine) as sched_stats,
     ):
         if runner is not None:
             try:
-                runner()
+                run_out = runner()
                 sync_device()  # ensure all kernels land in the trace before stop
             except Exception as exc:
                 runner_error = f"workload run failed: {exc}"
@@ -313,10 +320,23 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]:
     # evidence below) — empty when no engine produced samples.
     sched_summary = sched_stats.summary()
     sched_causes = scheduler_causes(sched_summary)
-    if sched_stats.samples:
+    # Per-request serving latency, when the runner reported request records
+    # (vllm-decode does; synthetic runners don't). Same window as the scheduler
+    # series and the trace — joined via SchedulerStatsSummary.t0_wall_ns.
+    req_records = list(run_out.get("requests") or []) if isinstance(run_out, dict) else []
+    serving_summary = summarize_requests(req_records) if req_records else None
+    # Collective-communication causes from the same trace — ranked beside the
+    # scheduler causes below. Empty when the trace holds no collective kernels.
+    coll_causes = collective_causes(worst_device_comm(trace))
+    if sched_stats.samples or serving_summary is not None:
         (run_dir / "scheduler_stats.json").write_text(
             json.dumps(
-                {"summary": asdict(sched_summary), "samples": sched_stats.to_records()},
+                {
+                    "summary": asdict(sched_summary),
+                    "samples": sched_stats.to_records(),
+                    "serving": asdict(serving_summary) if serving_summary else None,
+                    "requests": [asdict(r) for r in req_records],
+                },
                 indent=2,
             )
         )
@@ -486,6 +506,14 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]:
                      "note": c.note, "motivates_knobs": c.motivates_knobs}
                     for c in sched_causes
                 ],
+                # Collective (NCCL) causes — communication time the kernel-time
+                # residuals can't distinguish from compute. Empty on single-GPU
+                # runs and on any trace with no collective kernels.
+                "collective_causes": [
+                    {"signal": c.signal, "effect": c.effect, "severity": c.severity,
+                     "note": c.note, "motivates_knobs": c.motivates_knobs}
+                    for c in coll_causes
+                ],
             },
             indent=2,
         )
@@ -556,6 +584,10 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]:
     claims: list[Claim] = []
     rolled_back: list[str] = []
     rejected: list[str] = []
+    # Customer-verification records: the full A/B behind each claim, captured as
+    # it happens. EngineABResult lives on applicator.last_result and is
+    # overwritten by the next candidate, so it has to be taken per-iteration.
+    verification: list[VerificationRecord] = []
     # Aggregate kernel-time residual for the report (was hardcoded 0.0). Same for
     # every claim in a run — it describes the run's gap vs the predicted graph.
     kt_residual = _agg_kt_residual(res)
@@ -570,10 +602,29 @@ def run_loop(cfg: LoopConfig) -> dict[str, Any]:
         if cfg.engine is not None and live_restart_fn is None and knob_kind(c.spec.knob) == "structural":
             rejected.append(f"{c.spec.name} (structural knob: needs engine restart, no restart_fn)")
             continue
+        # Snapshot the engine config BEFORE the apply: a hot-swap mutates these
+        # kwargs in place and a restart replaces the engine outright, so reading
+        # them afterwards would report the candidate on both sides of the diff.
+        baseline_cfg = dict(getattr(cfg.engine, "gitm_llm_kwargs", None) or {})
         result = apply_intervention(c.spec, applicator, min_keep_delta=0.0)
         ab = getattr(applicator, "last_result", None)
         if result.rolled_back:
             rolled_back.append(c.spec.name)
+        if ab is not None:
+            # Read the candidate config off whichever engine the applicator is
+            # holding now — for a restart A/B that is the rebuilt engine, not
+            # the original. Falls back to the knob delta over the baseline.
+            live_engine = getattr(applicator, "engine", cfg.engine)
+            candidate_cfg = dict(getattr(live_engine, "gitm_llm_kwargs", None) or {})
+            if not candidate_cfg and baseline_cfg:
+                candidate_cfg = {**baseline_cfg, **(c.spec.knobs or {c.spec.knob: c.spec.value})}
+            verification.append(
+                build_record(
+                    c.spec, ab, result,
+                    baseline_config=baseline_cfg,
+                    candidate_config=candidate_cfg,
+                )
+            )
         # Causal evidence: the measured A/B verdict when live, else the Granger
         # signal that motivated the candidate. The kept/rolled-back wording comes
         # from the authoritative ApplyResult (the real gate decision), not from
@@ -710,6 +761,14 @@ def _unenactable(spec: Any) -> str | None:
     provenance.rejected_candidates = rejected
     provenance.rolled_back = rolled_back
 
+    # Customer-verification export: the A/B numbers the markdown report states as
+    # a percentage, emitted as structured data with the config each side ran
+    # under. Written only when a live A/B actually produced results.
+    if verification:
+        write_verification(
+            verification, provenance, run_dir / "verification.json", gpu_sku=pctx.sku
+        )
+
     sched_note = _scheduler_note(sched_summary)
     report_md = write_report(
         claims=claims,
diff --git a/gitm/tracer/vllm_stats.py b/gitm/tracer/vllm_stats.py
index fd096f2..b67e4f0 100644
--- a/gitm/tracer/vllm_stats.py
+++ b/gitm/tracer/vllm_stats.py
@@ -61,6 +61,184 @@ class SchedulerStatsSummary:
     total_preemptions: int | None  # delta of cumulative counter over the window
     peak_gpu_cache_usage: float | None
     peak_swapped: int | None
+    # Wall clock (``time.time_ns``) at sampling start. Samples carry ``t_ns``
+    # relative to that instant, so ``t0_wall_ns + sample.t_ns`` puts the scheduler
+    # series on the same wall clock as vLLM's per-request timestamps and as
+    # ``Trace.captured_at_ns`` — the join across the three views. 0 when unset.
+    t0_wall_ns: int = 0
+
+
+# SLO defaults for goodput. Serving-shaped starting points, not tuned: a request
+# is "good" only if it met both. Callers override per workload.
+DEFAULT_TTFT_SLO_S = 1.0
+DEFAULT_TPOT_SLO_S = 0.05
+
+
+@dataclass
+class RequestRecord:
+    """One request's lifecycle, in wall-clock seconds as vLLM reports them.
+
+    vLLM exposes these as ``time.time()`` floats on ``RequestOutput.metrics``.
+    Any field is ``None`` when the build didn't populate it — V1 leaves
+    ``metrics`` unset on some point releases — so a missing timestamp reads as
+    "unknown" and drops out of the percentiles, never as a zero that would make
+    latency look better than it was.
+    """
+
+    arrival_wall_s: float | None = None
+    first_token_wall_s: float | None = None
+    finished_wall_s: float | None = None
+    n_output_tokens: int = 0
+
+    @property
+    def ttft_s(self) -> float | None:
+        """Time to first token — the queue+prefill wait the user actually feels."""
+        if self.arrival_wall_s is None or self.first_token_wall_s is None:
+            return None
+        return max(self.first_token_wall_s - self.arrival_wall_s, 0.0)
+
+    @property
+    def tpot_s(self) -> float | None:
+        """Mean time per output token after the first.
+
+        The first token's cost is TTFT, so the inter-token rate covers the
+        ``n - 1`` tokens that followed it. A single-token response has no
+        inter-token interval at all and yields ``None`` rather than a number
+        derived from a zero-length gap.
+        """
+        if self.first_token_wall_s is None or self.finished_wall_s is None:
+            return None
+        if self.n_output_tokens < 2:
+            return None
+        span = max(self.finished_wall_s - self.first_token_wall_s, 0.0)
+        return span / (self.n_output_tokens - 1)
+
+    def meets_slo(self, ttft_slo_s: float, tpot_slo_s: float) -> bool:
+        """True if every *measurable* latency component cleared its SLO.
+
+        TTFT must be present — without it there is no evidence the request was
+        served in time, so it cannot count toward goodput. TPOT is allowed to be
+        ``None`` (a one-token response), which must not be penalised for lacking
+        an interval it could never have had.
+        """
+        ttft = self.ttft_s
+        if ttft is None or ttft > ttft_slo_s:
+            return False
+        tpot = self.tpot_s
+        return tpot is None or tpot <= tpot_slo_s
+
+
+@dataclass
+class ServingSummary:
+    """Per-request latency + goodput over the window, the serving-side companion
+    to :class:`SchedulerStatsSummary`.
+
+    ``n_ttft`` / ``n_tpot`` report how many requests actually yielded each
+    measurement, so a percentile computed from two samples is visibly weak rather
+    than silently authoritative.
+    """
+
+    n_requests: int
+    n_ttft: int
+    n_tpot: int
+    ttft_p50_s: float | None
+    ttft_p95_s: float | None
+    ttft_p99_s: float | None
+    tpot_p50_s: float | None
+    tpot_p95_s: float | None
+    tpot_p99_s: float | None
+    n_met_slo: int
+    goodput_rps: float | None  # SLO-meeting requests per second over the window
+    window_s: float | None
+    ttft_slo_s: float
+    tpot_slo_s: float
+
+
+def _percentile(vals: list[float], q: float) -> float | None:
+    """Nearest-rank percentile of ``vals`` at quantile ``q`` (0..1).
+
+    Deliberately stdlib-only: this module is imported on the capture path and
+    stays free of numpy so a tracer-only install keeps working.
+    """
+    if not vals:
+        return None
+    ordered = sorted(vals)
+    idx = min(max(int(round(q * (len(ordered) - 1))), 0), len(ordered) - 1)
+    return ordered[idx]
+
+
+def request_records_from_outputs(outputs: Any) -> list[RequestRecord]:
+    """Build :class:`RequestRecord`s from vLLM ``RequestOutput`` objects.
+
+    Duck-typed and best-effort in the same spirit as the scheduler reads: a
+    build that exposes no ``metrics`` still yields one record per request
+    carrying the token count, which is the honest "requests ran, latency
+    unavailable" outcome rather than an empty series or an invented timestamp.
+    """
+    records: list[RequestRecord] = []
+    for o in outputs or []:
+        rec = RequestRecord()
+        outs = getattr(o, "outputs", None)
+        if outs:
+            try:
+                rec.n_output_tokens = len(outs[0].token_ids)
+            except (AttributeError, TypeError, IndexError):
+                pass
+        metrics = getattr(o, "metrics", None)
+        if metrics is not None:
+            for field_name, attr in (
+                ("arrival_wall_s", "arrival_time"),
+                ("first_token_wall_s", "first_token_time"),
+                ("finished_wall_s", "finished_time"),
+            ):
+                val = getattr(metrics, attr, None)
+                if isinstance(val, int | float):
+                    setattr(rec, field_name, float(val))
+        records.append(rec)
+    return records
+
+
+def summarize_requests(
+    records: list[RequestRecord],
+    *,
+    ttft_slo_s: float = DEFAULT_TTFT_SLO_S,
+    tpot_slo_s: float = DEFAULT_TPOT_SLO_S,
+) -> ServingSummary:
+    """Aggregate per-request records into TTFT/TPOT percentiles and goodput.
+
+    Goodput is SLO-meeting requests per second over the observed window
+    (first arrival → last completion). Without those bounds the rate has no
+    denominator, so it is reported as ``None`` rather than divided by a guess.
+    """
+    ttfts = [t for t in (r.ttft_s for r in records) if t is not None]
+    tpots = [t for t in (r.tpot_s for r in records) if t is not None]
+    met = [r for r in records if r.meets_slo(ttft_slo_s, tpot_slo_s)]
+
+    arrivals = [r.arrival_wall_s for r in records if r.arrival_wall_s is not None]
+    finishes = [r.finished_wall_s for r in records if r.finished_wall_s is not None]
+    window_s: float | None = None
+    if arrivals and finishes:
+        window_s = max(max(finishes) - min(arrivals), 0.0)
+    # A zero-length window (single instantaneous request, or clock granularity)
+    # has no meaningful rate — report the count, not a division by ~0.
+    goodput = (len(met) / window_s) if window_s else None
+
+    return ServingSummary(
+        n_requests=len(records),
+        n_ttft=len(ttfts),
+        n_tpot=len(tpots),
+        ttft_p50_s=_percentile(ttfts, 0.50),
+        ttft_p95_s=_percentile(ttfts, 0.95),
+        ttft_p99_s=_percentile(ttfts, 0.99),
+        tpot_p50_s=_percentile(tpots, 0.50),
+        tpot_p95_s=_percentile(tpots, 0.95),
+        tpot_p99_s=_percentile(tpots, 0.99),
+        n_met_slo=len(met),
+        goodput_rps=goodput,
+        window_s=window_s,
+        ttft_slo_s=ttft_slo_s,
+        tpot_slo_s=tpot_slo_s,
+    )
 
 
 def _first_attr(obj: Any, *paths: str) -> Any:
@@ -262,11 +440,16 @@ def __init__(self, engine: Any, *, interval_s: float = 0.05) -> None:
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
         self._t0_ns: int = 0
+        self._t0_wall_ns: int = 0
 
     def start(self) -> None:
         if self._thread is not None or self.engine is None:
             return
+        # Two clocks taken together: monotonic for sample spacing (immune to wall
+        # clock jumps), wall for the join against request timestamps and the
+        # trace. Captured adjacently so the offset between them is exact.
         self._t0_ns = time.perf_counter_ns()
+        self._t0_wall_ns = time.time_ns()
         # Take one snapshot synchronously so even a decode shorter than one
         # sampling interval still yields a scheduler reading (no start/stop race).
         try:
@@ -300,20 +483,22 @@ def summary(self) -> SchedulerStatsSummary:
         # Snapshot first: stop() joins with a timeout, so in the pathological case
         # where the daemon thread is still alive, summarize must iterate a stable
         # copy rather than a list being appended to concurrently.
-        return summarize(list(self.samples))
+        return summarize(list(self.samples), t0_wall_ns=self._t0_wall_ns)
 
     def to_records(self) -> list[dict[str, Any]]:
         """Samples as plain dicts, ready for JSONL alongside the kernel trace."""
         return [asdict(s) for s in list(self.samples)]
 
 
-def summarize(samples: list[SchedulerSample]) -> SchedulerStatsSummary:
+def summarize(
+    samples: list[SchedulerSample], *, t0_wall_ns: int = 0
+) -> SchedulerStatsSummary:
     """Aggregate a sample series into the compact summary attribution consumes."""
     if not samples:
         return SchedulerStatsSummary(
             n_samples=0, duration_s=0.0, peak_queue_depth=None, mean_running=None,
             peak_running=None, mean_batch_occupancy=None, total_preemptions=None,
-            peak_gpu_cache_usage=None, peak_swapped=None,
+            peak_gpu_cache_usage=None, peak_swapped=None, t0_wall_ns=t0_wall_ns,
         )
 
     def _vals(attr: str) -> list[float]:
@@ -340,6 +525,7 @@ def _vals(attr: str) -> list[float]:
         total_preemptions=int(max(preempt) - min(preempt)) if preempt else None,
         peak_gpu_cache_usage=max(cache) if cache else None,
         peak_swapped=int(max(swapped)) if swapped else None,
+        t0_wall_ns=t0_wall_ns,
     )
 
 
diff --git a/gitm/workloads.py b/gitm/workloads.py
index b9924b2..80d15e7 100644
--- a/gitm/workloads.py
+++ b/gitm/workloads.py
@@ -685,7 +685,16 @@ def run() -> dict[str, Any]:
         outputs = active.generate(prompts, params)
         produced = sum(len(o.outputs[0].token_ids) for o in outputs)
         sync_device()
-        return {"prompts": len(prompts), "generated_tokens": produced, "model": model}
+        # Per-request lifecycle alongside the token count: the loop turns these
+        # into TTFT/TPOT/goodput. Empty on builds that don't populate metrics.
+        from gitm.tracer.vllm_stats import request_records_from_outputs
+
+        return {
+            "prompts": len(prompts),
+            "generated_tokens": produced,
+            "model": model,
+            "requests": request_records_from_outputs(outputs),
+        }
 
     def _throughput(eng: Any) -> float:
         """Decode-throughput probe (tokens/sec) for the Phase-4 A/B.
diff --git a/tests/test_collective_signal.py b/tests/test_collective_signal.py
new file mode 100644
index 0000000..e211d25
--- /dev/null
+++ b/tests/test_collective_signal.py
@@ -0,0 +1,126 @@
+"""Collective (NCCL) time as a ranked causal signal."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from gitm.importers.torch_trace import import_torch_trace
+from gitm.optimizer.collective_signal import collective_causes, worst_device_comm
+from gitm.tracer.schema import KernelEvent, Trace
+
+FIXTURES = Path(__file__).parent / "fixtures" / "importers"
+
+
+def _trace(events: list[KernelEvent], duration_ns: int) -> Trace:
+    return Trace(
+        workload_id="test",
+        fingerprint="fp",
+        run_id="r",
+        device_count=1,
+        vendor="nvidia",
+        captured_at_ns=0,
+        duration_ns=duration_ns,
+        events=events,
+    )
+
+
+def _kernel(name: str, start: int, end: int, *, device: int = 0, stream: int = 0) -> KernelEvent:
+    return KernelEvent(
+        name=name, start_ns=start, end_ns=end, stream_id=stream, device_id=device
+    )
+
+
+# --------------------------------------------------------------------------- #
+# exposed vs overlapped — the distinction the signal exists to make           #
+# --------------------------------------------------------------------------- #
+def test_exposed_collective_raises_a_cause():
+    # AllReduce runs alone: nothing overlaps it, so it is pure exposed cost.
+    trace = _trace(
+        [
+            _kernel("gemm", 0, 1_000_000),
+            _kernel("ncclDevKernel_AllReduce_Sum_f32", 1_000_000, 3_000_000),
+        ],
+        duration_ns=3_000_000,
+    )
+    causes = collective_causes(worst_device_comm(trace))
+    assert any(c.signal == "exposed_collective" for c in causes)
+
+
+def test_fully_overlapped_collective_raises_no_exposed_cause():
+    # Same 2ms AllReduce, but compute runs concurrently on another stream the
+    # whole time. That comm is hidden behind useful work and costs nothing —
+    # this is the case that proves the overlap math is actually doing something.
+    trace = _trace(
+        [
+            _kernel("gemm", 0, 3_000_000, stream=0),
+            _kernel("ncclDevKernel_AllReduce_Sum_f32", 1_000_000, 3_000_000, stream=1),
+        ],
+        duration_ns=3_000_000,
+    )
+    stats = worst_device_comm(trace)
+    assert stats.comm_ns > 0  # the comm did happen
+    assert stats.exposed_comm_ns == 0  # ...but none of it was exposed
+    assert not any(c.signal == "exposed_collective" for c in collective_causes(stats))
+
+
+def test_no_collective_kernels_yields_no_causes():
+    trace = _trace([_kernel("gemm", 0, 1_000_000)], duration_ns=1_000_000)
+    assert collective_causes(worst_device_comm(trace)) == []
+
+
+def test_empty_trace_and_none_are_safe():
+    assert worst_device_comm(_trace([], duration_ns=0)) is None
+    assert collective_causes(None) == []
+
+
+def test_severity_is_bounded():
+    # An entirely-collective trace must saturate at 1.0, not exceed it.
+    trace = _trace(
+        [_kernel("ncclDevKernel_AllReduce_Sum_f32", 0, 1_000_000)], duration_ns=1_000_000
+    )
+    causes = collective_causes(worst_device_comm(trace))
+    assert causes and all(0.0 <= c.severity <= 1.0 for c in causes)
+
+
+def test_causes_name_topology_knobs():
+    trace = _trace(
+        [_kernel("ncclDevKernel_AllReduce_Sum_f32", 0, 1_000_000)], duration_ns=1_000_000
+    )
+    for c in collective_causes(worst_device_comm(trace)):
+        assert "tensor_parallel_size" in c.motivates_knobs
+
+
+# --------------------------------------------------------------------------- #
+# multi-device: a live capture holds every GPU's kernels in one trace          #
+# --------------------------------------------------------------------------- #
+def test_cross_device_compute_does_not_mask_exposed_comm():
+    # dev0 communicates while dev1 computes. Treating this as one flat timeline
+    # would let dev1's gemm "cover" dev0's AllReduce and report zero exposed
+    # comm. Splitting per device keeps the overlap question inside a GPU.
+    trace = _trace(
+        [
+            _kernel("ncclDevKernel_AllReduce_Sum_f32", 0, 2_000_000, device=0),
+            _kernel("gemm", 0, 2_000_000, device=1),
+        ],
+        duration_ns=2_000_000,
+    )
+    stats = worst_device_comm(trace)
+    assert stats.device_id == 0
+    assert stats.exposed_comm_ns == 2_000_000
+
+
+def test_nccl_fixture_produces_a_collective_cause():
+    # The real 4xA100 NCCL fixture, merged back into a single trace the way a
+    # live CUPTI capture of a multi-GPU run would present it.
+    traces, _ = import_torch_trace(FIXTURES / "synthetic_4xA100_nccl.json", run_id="coll")
+    merged_events = [e for t in traces for e in t.events]
+    merged = traces[0].model_copy(
+        update={
+            "events": merged_events,
+            "duration_ns": max(t.duration_ns for t in traces),
+            "device_count": len(traces),
+        }
+    )
+    stats = worst_device_comm(merged)
+    assert stats is not None and stats.comm_ns > 0
+    assert collective_causes(stats)
diff --git a/tests/test_serving_latency.py b/tests/test_serving_latency.py
new file mode 100644
index 0000000..635a5f6
--- /dev/null
+++ b/tests/test_serving_latency.py
@@ -0,0 +1,149 @@
+"""Per-request serving latency: TTFT/TPOT/goodput and the wall-clock join."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from gitm.tracer.vllm_stats import (
+    RequestRecord,
+    request_records_from_outputs,
+    summarize_requests,
+)
+
+
+def _rec(arrival, first, finished, n_tokens):
+    return RequestRecord(
+        arrival_wall_s=arrival,
+        first_token_wall_s=first,
+        finished_wall_s=finished,
+        n_output_tokens=n_tokens,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# per-request derivations                                                     #
+# --------------------------------------------------------------------------- #
+def test_ttft_and_tpot_from_timestamps():
+    # arrival 0, first token at 0.2, finished at 1.2 with 11 tokens →
+    # TTFT 0.2s; the 10 tokens after the first span 1.0s → TPOT 0.1s.
+    r = _rec(0.0, 0.2, 1.2, 11)
+    assert r.ttft_s == 0.2
+    assert abs(r.tpot_s - 0.1) < 1e-9
+
+
+def test_single_token_response_has_no_tpot():
+    # One token means there is no inter-token interval to measure. It must read
+    # as unknown, not as a rate derived from a zero-length gap.
+    r = _rec(0.0, 0.2, 0.2, 1)
+    assert r.ttft_s == 0.2
+    assert r.tpot_s is None
+
+
+def test_missing_timestamps_yield_none_not_zero():
+    r = RequestRecord(n_output_tokens=8)
+    assert r.ttft_s is None and r.tpot_s is None
+
+
+# --------------------------------------------------------------------------- #
+# summary: percentiles exclude unmeasurable requests                          #
+# --------------------------------------------------------------------------- #
+def test_unmeasurable_requests_drop_out_of_percentiles():
+    records = [
+        _rec(0.0, 0.1, 1.1, 11),
+        _rec(0.0, 0.3, 1.3, 11),
+        RequestRecord(n_output_tokens=5),  # no metrics on this build
+    ]
+    s = summarize_requests(records)
+    assert s.n_requests == 3
+    # Only the two with timestamps contribute — a missing sample must not be
+    # imputed as a 0s TTFT, which would make latency look better than it was.
+    assert s.n_ttft == 2
+    assert s.ttft_p50_s in (0.1, 0.3)
+    assert s.ttft_p50_s > 0.0
+
+
+def test_empty_records_summarize_to_zeros_not_crash():
+    s = summarize_requests([])
+    assert s.n_requests == 0 and s.n_ttft == 0
+    assert s.ttft_p50_s is None and s.goodput_rps is None
+
+
+# --------------------------------------------------------------------------- #
+# goodput                                                                     #
+# --------------------------------------------------------------------------- #
+def test_goodput_counts_only_slo_meeting_requests():
+    # Two fast requests inside the SLO, one with a 5s TTFT that blows it. The
+    # window spans arrival 0 → last finish 6s.
+    records = [
+        _rec(0.0, 0.1, 1.1, 11),
+        _rec(0.0, 0.2, 1.2, 11),
+        _rec(0.0, 5.0, 6.0, 11),
+    ]
+    s = summarize_requests(records, ttft_slo_s=1.0, tpot_slo_s=0.5)
+    assert s.n_met_slo == 2
+    assert s.window_s == 6.0
+    assert abs(s.goodput_rps - 2 / 6.0) < 1e-9
+
+
+def test_tpot_violation_alone_fails_the_slo():
+    # TTFT is fine but the stream crawls: 11 tokens over 10s → TPOT 1.0s.
+    records = [_rec(0.0, 0.1, 10.1, 11)]
+    s = summarize_requests(records, ttft_slo_s=1.0, tpot_slo_s=0.05)
+    assert s.n_met_slo == 0
+
+
+def test_single_token_request_not_penalised_for_absent_tpot():
+    # No TPOT to measure, TTFT well inside the SLO — this is a good request.
+    records = [_rec(0.0, 0.1, 0.1, 1)]
+    s = summarize_requests(records, ttft_slo_s=1.0, tpot_slo_s=0.05)
+    assert s.n_met_slo == 1
+
+
+# --------------------------------------------------------------------------- #
+# vLLM RequestOutput adaptation                                               #
+# --------------------------------------------------------------------------- #
+def test_records_from_vllm_outputs():
+    out = SimpleNamespace(
+        outputs=[SimpleNamespace(token_ids=[1, 2, 3])],
+        metrics=SimpleNamespace(arrival_time=10.0, first_token_time=10.5, finished_time=12.5),
+    )
+    (rec,) = request_records_from_outputs([out])
+    assert rec.n_output_tokens == 3
+    assert rec.ttft_s == 0.5
+
+
+def test_outputs_without_metrics_still_yield_a_record():
+    # V1 builds may leave `metrics` unset. We still know a request ran and how
+    # many tokens it produced — that is the honest partial record.
+    out = SimpleNamespace(outputs=[SimpleNamespace(token_ids=[1, 2])], metrics=None)
+    (rec,) = request_records_from_outputs([out])
+    assert rec.n_output_tokens == 2
+    assert rec.ttft_s is None
+
+
+def test_records_from_empty_outputs():
+    assert request_records_from_outputs([]) == []
+    assert request_records_from_outputs(None) == []
+
+
+# --------------------------------------------------------------------------- #
+# the join: scheduler samples carry a wall-clock anchor                       #
+# --------------------------------------------------------------------------- #
+def test_sampler_summary_carries_wall_clock_anchor():
+    from gitm.tracer.vllm_stats import sample_scheduler_stats
+
+    engine = SimpleNamespace(scheduler=SimpleNamespace(running=[1], waiting=[]))
+    with sample_scheduler_stats(engine, interval_s=0.002) as sampler:
+        pass
+    s = sampler.summary()
+    # Without this anchor the scheduler series (monotonic t_ns) cannot be placed
+    # on the same clock as request timestamps or the trace.
+    assert s.t0_wall_ns > 0
+
+
+def test_summary_without_engine_has_no_anchor():
+    from gitm.tracer.vllm_stats import sample_scheduler_stats
+
+    with sample_scheduler_stats(None) as sampler:
+        pass
+    assert sampler.summary().t0_wall_ns == 0
diff --git a/tests/test_verification_export.py b/tests/test_verification_export.py
new file mode 100644
index 0000000..f64494b
--- /dev/null
+++ b/tests/test_verification_export.py
@@ -0,0 +1,144 @@
+"""Customer-verification export: the A/B as structured, re-measurable data."""
+
+from __future__ import annotations
+
+import json
+
+from gitm.kernels.spec import InterventionSpec
+from gitm.optimizer.apply import ApplyResult, EngineABResult
+from gitm.optimizer.report import Provenance
+from gitm.optimizer.verification_export import (
+    MIN_NOISE_BAND,
+    SCHEMA,
+    build_export,
+    build_record,
+    write_verification,
+)
+
+
+def _spec(name="raise-batch", knob="max_num_seqs", value=512) -> InterventionSpec:
+    return InterventionSpec(
+        name=name,
+        summary="Raise max concurrent sequences",
+        knob=knob,
+        value=value,
+        expected_delta_mean=0.08,
+        expected_delta_lo=0.02,
+        expected_delta_hi=0.14,
+        source="https://docs.vllm.ai/example",
+    )
+
+
+def _ab(**over) -> EngineABResult:
+    base = dict(
+        knob="max_num_seqs", value=512, baseline_tps=100.0, candidate_tps=112.0,
+        speedup=1.12, kept=True, via="hot-swap", baseline_std=1.0,
+        candidate_std=1.0, reps=3, significant=True,
+    )
+    base.update(over)
+    return EngineABResult(**base)
+
+
+def _prov() -> Provenance:
+    return Provenance(
+        workload_id="vllm-decode", fingerprint="fp", run_id="run-1",
+        git_sha="abc1234", gitm_version="0.1.11",
+        started_at_ns=0, ended_at_ns=1,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# the measured numbers survive intact                                         #
+# --------------------------------------------------------------------------- #
+def test_record_carries_both_sides_not_just_the_delta():
+    rec = build_record(_spec(), _ab(), ApplyResult(True, False, 0.12))
+    # The markdown report states only the percentage; re-measuring needs the
+    # absolute numbers on both sides.
+    assert rec.baseline_tps == 100.0
+    assert rec.candidate_tps == 112.0
+    assert abs(rec.delta - 0.12) < 1e-9
+    assert rec.reps == 3
+    assert rec.source.startswith("https://")
+
+
+def test_full_config_travels_on_both_sides():
+    rec = build_record(
+        _spec(), _ab(), ApplyResult(True, False, 0.12),
+        baseline_config={"max_num_seqs": 256, "gpu_memory_utilization": 0.9},
+        candidate_config={"max_num_seqs": 512, "gpu_memory_utilization": 0.9},
+    )
+    # Not just the knob that moved: reproducing at a different
+    # gpu_memory_utilization measures a different system.
+    assert rec.baseline_config["gpu_memory_utilization"] == 0.9
+    assert rec.baseline_config["max_num_seqs"] == 256
+    assert rec.candidate_config["max_num_seqs"] == 512
+
+
+# --------------------------------------------------------------------------- #
+# `kept` follows the gate, not the measurement                                #
+# --------------------------------------------------------------------------- #
+def test_kept_follows_the_rollback_gate_not_the_ab_indicator():
+    # EngineABResult.kept is a measure-time delta>=0 indicator. The gate can
+    # still roll the candidate back (min_keep_delta, a failed correctness
+    # check). The export must report what actually happened to the workload.
+    ab = _ab(kept=True)
+    rec = build_record(_spec(), ab, ApplyResult(True, rolled_back=True, measured_delta=0.001))
+    assert ab.kept is True
+    assert rec.kept is False
+
+
+def test_kept_true_when_gate_kept_it():
+    rec = build_record(_spec(), _ab(kept=True), ApplyResult(True, False, 0.12))
+    assert rec.kept is True
+
+
+# --------------------------------------------------------------------------- #
+# agreement band                                                              #
+# --------------------------------------------------------------------------- #
+def test_single_rep_gets_a_floored_agreement_band():
+    # reps=1 → zero measured scatter. Publishing a 0% band would imply a
+    # precision no GPU benchmark has, and any customer jitter would read as
+    # a failed reproduction.
+    rec = build_record(
+        _spec(),
+        _ab(reps=1, baseline_std=0.0, candidate_std=0.0),
+        ApplyResult(True, False, 0.12),
+    )
+    assert rec.agreement_band == MIN_NOISE_BAND
+
+
+def test_real_scatter_widens_the_band_beyond_the_floor():
+    # baseline 100 tok/s with std 5+5 → 10% scatter, well above the floor.
+    rec = build_record(
+        _spec(),
+        _ab(reps=5, baseline_std=5.0, candidate_std=5.0),
+        ApplyResult(True, False, 0.12),
+    )
+    assert abs(rec.agreement_band - 0.10) < 1e-9
+
+
+# --------------------------------------------------------------------------- #
+# document shape                                                              #
+# --------------------------------------------------------------------------- #
+def test_export_is_json_serializable_with_provenance():
+    rec = build_record(_spec(), _ab(), ApplyResult(True, False, 0.12))
+    doc = build_export([rec], _prov(), gpu_sku="NVIDIA A100-SXM4-80GB")
+    round_tripped = json.loads(json.dumps(doc, default=str))
+    assert round_tripped["schema"] == SCHEMA
+    assert round_tripped["provenance"]["git_sha"] == "abc1234"
+    assert round_tripped["environment"]["gpu_sku"] == "NVIDIA A100-SXM4-80GB"
+    assert len(round_tripped["results"]) == 1
+
+
+def test_unknown_environment_reports_none_not_a_guess():
+    doc = build_export([], _prov())
+    # No SKU passed and (on a CPU box) no driver — these must read as unknown.
+    assert doc["environment"]["gpu_sku"] is None
+
+
+def test_write_verification_writes_readable_json(tmp_path):
+    rec = build_record(_spec(), _ab(), ApplyResult(True, False, 0.12))
+    out = write_verification([rec], _prov(), tmp_path / "verification.json")
+    doc = json.loads(open(out).read())
+    assert doc["results"][0]["knob"] == "max_num_seqs"
+    assert doc["protocol"]["metric"].startswith("decode throughput")

From 26c43ec89397e5904aadb1dd963bfa3c942343e4 Mon Sep 17 00:00:00 2001
From: Kevin <kwang3170@berkeley.edu>
Date: Fri, 24 Jul 2026 04:06:28 +1000
Subject: [PATCH 2/3] Fix pre-existing repo lint + two review nits

The CI lint job runs `ruff check .` over the whole repo and was already red on
main (12 errors, all in importer/metrics files this branch never touched).
Clearing them here so the PR's lint check goes green; squash-merge folds this in.

- ruff --fix: 9 unused imports / import-order in gitm/importers/* and metrics.py
- 3 UP038: isinstance((A, B)) -> isinstance(A | B), valid on our py310 target

Two points from review that were actually true:
- vllm_stats: the clock-anchor comment claimed the two-clock offset was "exact";
  it's sub-microsecond, not exact. Reworded with the real bound and why it's
  negligible against the 50ms sampling interval.
- test_serving_latency: added the arrival-without-first-token case (TTFT
  unmeasurable -> no SLO can be met -> goodput 0), which the existing tests
  didn't lock in.

Review points rejected as inaccurate: DeviceCommStats.device_id does exist;
the rel_std test matches the real formula; collective_signal's two floors
measure different denominators (wall vs busy) and aren't comparable.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 gitm/importers/_common.py          |  2 +-
 gitm/importers/analyze.py          |  2 +-
 gitm/importers/node_rollup.py      |  2 +-
 gitm/importers/nsys.py             |  3 ---
 gitm/importers/torch_trace.py      |  4 ++--
 gitm/optimizer/metrics.py          |  2 +-
 gitm/tracer/vllm_stats.py          |  4 +++-
 tests/test_importers_hypothesis.py |  4 ++--
 tests/test_serving_latency.py      | 14 ++++++++++++++
 9 files changed, 25 insertions(+), 12 deletions(-)

diff --git a/gitm/importers/_common.py b/gitm/importers/_common.py
index 996f5d1..07882ea 100644
--- a/gitm/importers/_common.py
+++ b/gitm/importers/_common.py
@@ -3,11 +3,11 @@
 from __future__ import annotations
 
 import os
+import sys
 import tempfile
 import warnings
 from collections.abc import Iterable
 from dataclasses import dataclass, field
-import sys
 from pathlib import Path
 from typing import Any
 
diff --git a/gitm/importers/analyze.py b/gitm/importers/analyze.py
index 84a02ee..178799f 100644
--- a/gitm/importers/analyze.py
+++ b/gitm/importers/analyze.py
@@ -18,7 +18,7 @@
 from gitm.optimizer.qualification import QualificationResult, qualify
 from gitm.planner.context import peak_for_sku
 from gitm.tracer.capture import write_trace_jsonl
-from gitm.tracer.schema import KernelEvent, Trace
+from gitm.tracer.schema import Trace
 
 # Default catalogue peak when SKU is unknown (A100 PCIe figures; called out in caveats).
 _DEFAULT_PEAK = HardwarePeak(name="A100", peak_flops=312e12, peak_bw_bytes_s=1555e9)
diff --git a/gitm/importers/node_rollup.py b/gitm/importers/node_rollup.py
index d433d07..c30bfd4 100644
--- a/gitm/importers/node_rollup.py
+++ b/gitm/importers/node_rollup.py
@@ -11,7 +11,7 @@
 from typing import Any
 
 from gitm.optimizer.metrics import _merge_intervals
-from gitm.tracer.schema import KernelEvent, Trace
+from gitm.tracer.schema import Trace
 
 # Collective kernel name patterns (case-insensitive). One module-level table;
 # each entry is a substring match against the demangled/exported kernel name.
diff --git a/gitm/importers/nsys.py b/gitm/importers/nsys.py
index 68035bd..27a1af7 100644
--- a/gitm/importers/nsys.py
+++ b/gitm/importers/nsys.py
@@ -17,11 +17,8 @@
     ImportError,
     ImportStats,
     as_int,
-    device_count_from_events,
     file_mtime_ns,
-    filter_device,
     finish_trace,
-    per_device_kernel_counts,
 )
 from gitm.tracer.schema import KernelEvent, MemcpyEvent, SyncEvent, Trace, TraceEvent
 
diff --git a/gitm/importers/torch_trace.py b/gitm/importers/torch_trace.py
index c2c8be4..efe7413 100644
--- a/gitm/importers/torch_trace.py
+++ b/gitm/importers/torch_trace.py
@@ -24,7 +24,7 @@
     finish_trace,
     per_device_kernel_counts,
 )
-from gitm.tracer.schema import KernelEvent, MemcpyEvent, Trace, TraceEvent
+from gitm.tracer.schema import MemcpyEvent, Trace, TraceEvent
 
 # Max decompressed size for .json.gz (customer dumps can be multi-GB).
 DEFAULT_MAX_DECOMPRESSED_BYTES = 20 * 1024 ** 3  # 20 GiB
@@ -91,7 +91,7 @@ def _arg_get(args: dict[str, Any] | None, keys: tuple[str, ...]) -> Any:
 def _as_dims(val: Any) -> tuple[int, int, int]:
     if val is None:
         return (1, 1, 1)
-    if isinstance(val, (list, tuple)) and len(val) >= 1:
+    if isinstance(val, list | tuple) and len(val) >= 1:
         x = as_int(val[0], 1) if len(val) > 0 else 1
         y = as_int(val[1], 1) if len(val) > 1 else 1
         z = as_int(val[2], 1) if len(val) > 2 else 1
diff --git a/gitm/optimizer/metrics.py b/gitm/optimizer/metrics.py
index 47a0968..8d93542 100644
--- a/gitm/optimizer/metrics.py
+++ b/gitm/optimizer/metrics.py
@@ -26,7 +26,7 @@
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 
-from gitm.tracer.schema import KernelEvent, MemcpyEvent, SyncEvent, Trace
+from gitm.tracer.schema import KernelEvent, Trace
 
 FlopsModel = Callable[[KernelEvent], float]
 
diff --git a/gitm/tracer/vllm_stats.py b/gitm/tracer/vllm_stats.py
index b67e4f0..0fc4418 100644
--- a/gitm/tracer/vllm_stats.py
+++ b/gitm/tracer/vllm_stats.py
@@ -447,7 +447,9 @@ def start(self) -> None:
             return
         # Two clocks taken together: monotonic for sample spacing (immune to wall
         # clock jumps), wall for the join against request timestamps and the
-        # trace. Captured adjacently so the offset between them is exact.
+        # trace. Captured back-to-back so the offset between the two clocks is
+        # sub-microsecond — far below the 50ms sampling interval and the
+        # second-scale request latencies this anchor is used to align.
         self._t0_ns = time.perf_counter_ns()
         self._t0_wall_ns = time.time_ns()
         # Take one snapshot synchronously so even a decode shorter than one
diff --git a/tests/test_importers_hypothesis.py b/tests/test_importers_hypothesis.py
index 7a343fb..e45eded 100644
--- a/tests/test_importers_hypothesis.py
+++ b/tests/test_importers_hypothesis.py
@@ -41,7 +41,7 @@ def test_prop_us_to_ns_monotonic(ts: float, dur: float):
         }
     )
     # Default import path uses LightKernel for scale; --strict uses KernelEvent.
-    assert isinstance(ev, (KernelEvent, LightKernel))
+    assert isinstance(ev, KernelEvent | LightKernel)
     assert ev.start_ns == int(ts * 1000.0)
     assert ev.end_ns == int((ts + dur) * 1000.0)
     assert ev.end_ns >= ev.start_ns or dur == 0.0
@@ -74,7 +74,7 @@ def test_prop_missing_args_never_crash(cat, name, has_grid, has_stream, has_devi
     )
     # May be None for weird name/cat combos; must not raise.
     if ev is not None:
-        assert isinstance(ev, (KernelEvent, MemcpyEvent, LightKernel, LightMemcpy))
+        assert isinstance(ev, KernelEvent | MemcpyEvent | LightKernel | LightMemcpy)
         assert getattr(ev, "kind", None) in ("kernel", "memcpy")
         assert ev.end_ns >= ev.start_ns
         # Light events are not pydantic; only validate the full Trace path
diff --git a/tests/test_serving_latency.py b/tests/test_serving_latency.py
index 635a5f6..89db662 100644
--- a/tests/test_serving_latency.py
+++ b/tests/test_serving_latency.py
@@ -68,6 +68,20 @@ def test_empty_records_summarize_to_zeros_not_crash():
     assert s.ttft_p50_s is None and s.goodput_rps is None
 
 
+def test_arrival_without_first_token_meets_no_slo():
+    # A build that reports arrival but never first_token_time: TTFT is
+    # unmeasurable, so no request can be shown to have met its latency SLO.
+    # Goodput must be 0-per-window, never credited for absent evidence.
+    records = [
+        RequestRecord(arrival_wall_s=0.0, finished_wall_s=1.0, n_output_tokens=8),
+        RequestRecord(arrival_wall_s=0.0, finished_wall_s=2.0, n_output_tokens=8),
+    ]
+    s = summarize_requests(records)
+    assert s.n_ttft == 0
+    assert s.n_met_slo == 0
+    assert s.goodput_rps == 0.0  # window is known (2.0s), zero requests met SLO
+
+
 # --------------------------------------------------------------------------- #
 # goodput                                                                     #
 # --------------------------------------------------------------------------- #

From 779e59a8aa9c769d7007d1b9345d706823455813 Mon Sep 17 00:00:00 2001
From: Kevin <kwang3170@berkeley.edu>
Date: Fri, 24 Jul 2026 04:14:21 +1000
Subject: [PATCH 3/3] Add hypothesis to dev extras so CI can collect the
 property tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

CI installs `.[dev,bench]` then runs `pytest`. tests/test_importers_hypothesis.py
does a plain `from hypothesis import ...`, but hypothesis was never listed in any
extra — so collection errored and the whole pytest job failed on every Python
version. Pre-existing on main (this branch touched neither pyproject nor that
import); it surfaced here because the PR gates on the check.

With hypothesis present the full suite is 564 passed on 3.11.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 pyproject.toml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pyproject.toml b/pyproject.toml
index 230b098..c10a623 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -59,6 +59,7 @@ dev = [
     "mypy>=1.10",
     "build>=1.0",  # `python -m build` for the wheel-build verification check
     "pre-commit>=3.5",  # local lint/test gate; see .pre-commit-config.yaml
+    "hypothesis>=6.0",  # property tests in tests/test_importers_hypothesis.py
 ]
 
 [project.scripts]