From f56ecad64d6ff721fe868f5ef6dde1bc2361c95f Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Mon, 13 Jul 2026 17:17:30 +0800 Subject: [PATCH] Harden trace replay measurement integrity --- .../collectivespec/summarize_fixed_grid.py | 82 +++++++++++++++++-- src/aituner/http_client.py | 10 ++- src/aituner/search.py | 5 +- tests/test_core_flow.py | 49 +++++++++++ 4 files changed, 138 insertions(+), 8 deletions(-) diff --git a/scripts/collectivespec/summarize_fixed_grid.py b/scripts/collectivespec/summarize_fixed_grid.py index 5bb665f..cd819dc 100644 --- a/scripts/collectivespec/summarize_fixed_grid.py +++ b/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) diff --git a/src/aituner/http_client.py b/src/aituner/http_client.py index 746bba4..186cf0d 100644 --- a/src/aituner/http_client.py +++ b/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, diff --git a/src/aituner/search.py b/src/aituner/search.py index 7b1d2d1..a4f9808 100644 --- a/src/aituner/search.py +++ b/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) diff --git a/tests/test_core_flow.py b/tests/test_core_flow.py index 48c52b7..e665c81 100644 --- a/tests/test_core_flow.py +++ b/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"))