Harden trace replay measurement integrity

scripts/collectivespec/summarize_fixed_grid.py

@@ -31,7 +31,18 @@ def num(value: Any) -> float | None:
     return value if math.isfinite(value) else None
 
 
-def read_one(root: Path, k: int, expected_tokens: int | None) -> dict[str, Any]:
+def nonnegative_int(value: Any) -> int | None:
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        return None
+    return value
+
+
+def read_one(
+    root: Path,
+    k: int,
+    expected_tokens: int | None,
+    expected_request_count: int | None,
+) -> dict[str, Any]:
     record: dict[str, Any] = {"k": k, "label": "NoSpec" if k == 0 else f"K={k}"}
     store = root / "stores" / f"k{k}"
     result_path = first(store, "*/trials/trial-*/result.json")
@@ -70,14 +81,36 @@ def read_one(root: Path, k: int, expected_tokens: int | None) -> dict[str, Any]:
     probe = history[0]
     details = detail_rows[0]
     outcomes = details.get("outcomes") if isinstance(details.get("outcomes"), list) else []
-    request_count = int(probe.get("request_count", -1))
+    request_count = nonnegative_int(probe.get("request_count"))
+    replayed_request_count = nonnegative_int(probe.get("replayed_request_count"))
+    if request_count is None:
+        failures.append("invalid_request_count")
+        request_count = -1
+    if replayed_request_count is None:
+        failures.append("invalid_replayed_request_count")
+        replayed_request_count = -1
     if details.get("early_stopped") or probe.get("early_stopped"):
         failures.append("early_stopped")
     if len(outcomes) != request_count:
         failures.append(f"outcome_count={len(outcomes)}_expected={request_count}")
+    if replayed_request_count != request_count:
+        failures.append(
+            f"replayed_request_count={replayed_request_count}_expected={request_count}"
+        )
+    if expected_request_count is not None and request_count != expected_request_count:
+        failures.append(
+            f"materialized_request_count={expected_request_count}_replayed={request_count}"
+        )
     successful = sum(bool(item.get("success")) for item in outcomes if isinstance(item, dict))
     tpot_present = sum(item.get("tpot_ms") is not None for item in outcomes if isinstance(item, dict))
     ttft_present = sum(item.get("ttft_ms") is not None for item in outcomes if isinstance(item, dict))
+    nonpositive_tpot = sum(
+        isinstance(item, dict)
+        and num(item.get("tpot_ms")) is not None
+        and num(item.get("tpot_ms")) <= 0
+        and (nonnegative_int(item.get("completion_tokens")) or 0) > 1
+        for item in outcomes
+    )
     tokens_verified = all(
         isinstance(item, dict)
         and item.get("completion_tokens_source") == "usage"
@@ -91,6 +124,8 @@ def read_one(root: Path, k: int, expected_tokens: int | None) -> dict[str, Any]:
         failures.append(f"tpot_count={tpot_present}_expected={request_count}")
     if ttft_present != request_count:
         failures.append(f"ttft_count={ttft_present}_expected={request_count}")
+    if nonpositive_tpot:
+        failures.append(f"nonpositive_tpot_for_multi_token_completion={nonpositive_tpot}")
     if expected_tokens is not None and not tokens_verified:
         failures.append("completion_tokens_not_all_usage_verified")
     latency = probe.get("latency_summary") if isinstance(probe.get("latency_summary"), dict) else {}
@@ -101,12 +136,15 @@ def read_one(root: Path, k: int, expected_tokens: int | None) -> dict[str, Any]:
             "integrity_failures": failures,
             "threshold": num(probe.get("threshold")),
             "request_count": request_count,
+            "replayed_request_count": replayed_request_count,
+            "materialized_request_count": expected_request_count,
             "request_rate": num(probe.get("request_rate")),
             "feasible": bool(probe.get("feasible")),
             "pass_rate": num(probe.get("pass_rate")),
             "success_count": successful,
             "ttft_count": ttft_present,
             "tpot_count": tpot_present,
+            "nonpositive_tpot_count": nonpositive_tpot,
             "tpot_ms": {name: num(tpot.get(name)) for name in ("mean", "p50", "p90", "p95", "p99")},
         }
     )
@@ -157,26 +195,57 @@ def data_sanity(records: list[dict[str, Any]]) -> dict[str, Any]:
                 and record.get("tpot_count") == record.get("request_count")
                 for record in valid
             ),
+            "all_materialized_request_counts_match": all(
+                record.get("materialized_request_count") is None
+                or record.get("request_count") == record.get("materialized_request_count")
+                for record in valid
+            ),
+            "no_nonpositive_tpot_for_multi_token_completion": all(
+                record.get("nonpositive_tpot_count", 0) == 0 for record in valid
+            ),
         },
     }
 
 
+def materialized_request_count(manifest: dict[str, Any]) -> int | None:
+    windows_path = manifest.get("trace_windows_path")
+    window_id = manifest.get("trace_window_id")
+    if not isinstance(windows_path, str) or not isinstance(window_id, str):
+        return None
+    path = Path(windows_path)
+    if not path.is_file():
+        return None
+    payload = load_json(path)
+    windows = payload.get("windows") if isinstance(payload, dict) else payload
+    if not isinstance(windows, list):
+        return None
+    for window in windows:
+        if not isinstance(window, dict) or window.get("window_id") != window_id:
+            continue
+        value = window.get("num_requests")
+        if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
+            return value
+    return None
+
+
 def markdown(records: list[dict[str, Any]], sanity: dict[str, Any]) -> str:
     lines = [
         "# CollectiveSpec fresh-engine fixed-grid summary",
         "",
-        "| configuration | valid | feasible | offered req/s | completed | pass rate | p95 TPOT (ms) | p99 TPOT (ms) | integrity failures |",
+        "| configuration | valid | feasible | offered req/s | materialized/replayed | completed | pass rate | p95 TPOT (ms) | p99 TPOT (ms) | integrity failures |",
-        "|---|---:|---:|---:|---:|---:|---:|---:|---|",
+        "|---|---:|---:|---:|---:|---:|---:|---:|---:|---|",
     ]
     for record in records:
         tpot = record.get("tpot_ms") or {}
         fmt = lambda value: "—" if value is None else f"{value:.6g}"
         lines.append(
-            "| {label} | {valid} | {feasible} | {rate} | {completed}/{count} | {pass_rate} | {p95} | {p99} | {failures} |".format(
+            "| {label} | {valid} | {feasible} | {rate} | {materialized}/{replayed} | {completed}/{count} | {pass_rate} | {p95} | {p99} | {failures} |".format(
                 label=record["label"],
                 valid=record.get("valid", False),
                 feasible=record.get("feasible", "—"),
                 rate=fmt(record.get("request_rate")),
+                materialized=record.get("materialized_request_count", "—"),
+                replayed=record.get("replayed_request_count", "—"),
                 completed=record.get("success_count", "—"),
                 count=record.get("request_count", "—"),
                 pass_rate=fmt(record.get("pass_rate")),
@@ -199,8 +268,9 @@ def main() -> int:
     expected_tokens = manifest.get("completion_tokens_override")
     if isinstance(expected_tokens, bool) or not isinstance(expected_tokens, int):
         expected_tokens = None
+    expected_request_count = materialized_request_count(manifest)
     records = [
-        read_one(args.root, int(k), expected_tokens)
+        read_one(args.root, int(k), expected_tokens, expected_request_count)
         for k in sorted(int(k) for k in manifest["k_values"])
     ]
     sanity = data_sanity(records)

src/aituner/http_client.py

@@ -315,7 +315,15 @@ def stream_chat_completion(
     ):
         tpot_ms = None
     else:
-        tpot_ms = ((last_token_at - first_token_at) / max(used_tokens - 1, 1)) * 1000.0
+        # A response that delivers all content-bearing chunks at one observed
+        # instant has no measurable inter-token interval.  Reporting 0 ms here
+        # would turn an unobservable TPOT into an artificial SLO pass.
+        elapsed_s = last_token_at - first_token_at
+        tpot_ms = (
+            (elapsed_s / max(used_tokens - 1, 1)) * 1000.0
+            if elapsed_s > 0
+            else None
+        )
     return StreamMetrics(
         ttft_ms=ttft_ms,
         tpot_ms=tpot_ms,

src/aituner/search.py

@@ -38,7 +38,10 @@ def binary_search_max_feasible(
     for _ in range(max_probes):
         if cur_high - cur_low <= tolerance and best_payload is not None:
             break
-        threshold = round((cur_low + cur_high) / 2.0, 12)
+        # Thresholds are also trace-selection boundaries.  Rounding a midpoint
+        # can cross a real sampling_u value and silently replay a different
+        # workload than the materialized window specifies.
+        threshold = (cur_low + cur_high) / 2.0
         probe = cache.get(threshold)
         if probe is None:
             probe = evaluator(threshold)

tests/test_core_flow.py

@@ -5883,6 +5883,25 @@ class CoreFlowTests(unittest.TestCase):
             [0.5, 0.25, 0.125, 0.0625, 0.03125, 0.015625],
         )
 
+    def test_binary_search_preserves_trace_selection_precision(self) -> None:
+        target = 0.016170151327299858
+        seen: list[float] = []
+
+        def evaluator(threshold: float) -> ThresholdProbe[dict[str, float]]:
+            seen.append(threshold)
+            return ThresholdProbe(threshold=threshold, feasible=True, payload={})
+
+        result = binary_search_max_feasible(
+            low=target - 1e-15,
+            high=target + 1e-15,
+            tolerance=1e-18,
+            max_probes=1,
+            evaluator=evaluator,
+        )
+        self.assertIsNotNone(result.best_feasible_payload)
+        self.assertEqual(seen, [target])
+        self.assertGreater(seen[0], 0.016170151327)
+
     def test_trace_max_requests_uses_window_wide_downsample(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             tmp_path = Path(tmp)
@@ -8794,6 +8813,36 @@ class CoreFlowTests(unittest.TestCase):
         self.assertIsNone(metrics.completion_tokens)
         self.assertEqual(metrics.completion_tokens_source, "none")
 
+    def test_stream_chat_completion_marks_same_instant_multitoken_tpot_unmeasurable(self) -> None:
+        class FakeResponse:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, traceback):
+                return False
+
+            def __iter__(self):
+                return iter(
+                    [
+                        b'data: {"choices":[{"delta":{"content":"a"}}]}\n',
+                        b'data: {"choices":[{"delta":{"content":"b"}}]}\n',
+                        b'data: {"usage":{"completion_tokens":2},"choices":[]}\n',
+                        b"data: [DONE]\n",
+                    ]
+                )
+
+        with mock.patch("aituner.http_client._urlopen", return_value=FakeResponse()):
+            with mock.patch("aituner.http_client.time.monotonic", side_effect=[1.0, 2.0, 2.0]):
+                metrics = stream_chat_completion(
+                    base_url="http://127.0.0.1:8000",
+                    body={"model": "m", "messages": [{"role": "user", "content": "x"}]},
+                    timeout_s=1.0,
+                )
+
+        self.assertEqual(metrics.ttft_ms, 1000.0)
+        self.assertIsNone(metrics.tpot_ms)
+        self.assertEqual(metrics.completion_tokens, 2)
+
     def test_loopback_urls_bypass_proxy(self) -> None:
         self.assertTrue(_should_bypass_proxy("http://127.0.0.1:8000/v1/models"))
         self.assertTrue(_should_bypass_proxy("http://localhost:8000/health"))