diff --git a/qa/latency_rollup.py b/qa/latency_rollup.py index bd71eeff..a029c6e1 100644 --- a/qa/latency_rollup.py +++ b/qa/latency_rollup.py @@ -100,6 +100,56 @@ def _api_seconds(res: dict) -> Optional[float]: return max(0.0, float(ms) / 1000.0) +# Token/cache fields read off a beat's terminal ``result`` event for the tool-schema-slab A/B +# (slab decision, Phase 3). The slab is a per-request PREFILL tax, so its effect shows up in the +# token ledger, not in duration_api_ms (which is generation-bound). The load-bearing A/B signals: +# * cache_creation_input_tokens — tokens WRITTEN to the prompt cache. On a healthy CACHED run the +# continuing beats sit on a stable prefix, so routine cache_creation ≈ 0; a tiering change that +# DENTS the cache (re-creates the changed tool block) shows up as elevated routine cache_creation. +# * cache_read_input_tokens — tokens served from cache (cheap). * input_tokens — uncached prefill. +# * ttft_ms — time-to-first-token; a smaller pinned tool block should lower it (prior art). +_USAGE_TOKEN_FIELDS = ( + "input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens", +) + + +def _usage(res: dict) -> dict[str, Optional[float]]: + """Per-beat token usage + time-to-first-token from a ``result`` event's ``usage`` block. + Absent fields are None (tolerant of older transcripts that predate the field).""" + u = res.get("usage") if isinstance(res.get("usage"), dict) else {} + out: dict[str, Optional[float]] = {} + for k in _USAGE_TOKEN_FIELDS: + v = u.get(k) + out[k] = float(v) if isinstance(v, (int, float)) else None + ttft = res.get("ttft_ms") + out["ttft_ms"] = float(ttft) if isinstance(ttft, (int, float)) else None + return out + + +def _mean(xs: Iterable[Any]) -> Optional[float]: + vals = [float(x) for x in xs if isinstance(x, (int, float))] + return round(sum(vals) / len(vals), 1) if vals else None + + +def _token_aggregates(rows: list[dict]) -> Optional[dict]: + """Split per-beat usage rows (beat order, cold-open first) into cold-open vs routine means.""" + if not rows: + return None + + def block(rs: list[dict]) -> Optional[dict]: + if not rs: + return None + return { + "cache_creation": _mean(r.get("cache_creation_input_tokens") for r in rs), + "cache_read": _mean(r.get("cache_read_input_tokens") for r in rs), + "input": _mean(r.get("input_tokens") for r in rs), + "output": _mean(r.get("output_tokens") for r in rs), + "ttft_ms": _mean(r.get("ttft_ms") for r in rs), + } + + return {"coldopen": block(rows[:1]), "routine": block(rows[1:])} + + def beat_files(transcript_dir: str | Path, run_id: str) -> list[str]: """The run's per-beat DM transcripts, ordered cold-open-FIRST (ascending nanos).""" pat = os.path.join(str(transcript_dir), f"{run_id}.dm.*.jsonl") @@ -117,6 +167,7 @@ def rollup_files(files: list[str]) -> dict[str, Any]: than a misleading 0.""" api_s: list[float] = [] # successful-beat generation seconds, in beat order turns: list[float] = [] # successful-beat num_turns, in beat order + usage_rows: list[dict] = [] # successful-beat token usage + ttft, in beat order failed = 0 for path in files: res = _final_result(path) @@ -131,6 +182,7 @@ def rollup_files(files: list[str]) -> dict[str, Any]: continue api_s.append(secs) turns.append(float(n) if isinstance(n, (int, float)) else 0.0) + usage_rows.append(_usage(res)) out: dict[str, Any] = { "s_per_beat": None, @@ -139,6 +191,9 @@ def rollup_files(files: list[str]) -> dict[str, Any]: "beats": len(api_s), # successful beats counted "cold_open_turns": None, "failed_beats": failed, + # Additive (slab decision, Phase 3): per-beat token/cache ledger for the tiering A/B. + # cold-open vs routine means of cache_creation / cache_read / input / output / ttft_ms. + "tokens": _token_aggregates(usage_rows), } if not api_s: return out @@ -187,6 +242,82 @@ def stamp_sidecars(rollup: dict[str, Any], run_dirs: Iterable[str | Path]) -> li return written +# --- Tool-schema-slab tiering A/B (slab decision, Phase 3) ----------------------------------- +# Compare two SAME-SHA / SAME-SEED duo rollups — arm A = baseline (WORLDOS_ENGINE_ALWAYSLOAD=1, +# whole-server pin) vs arm B = tiered (=0, per-tool PINNED_ALLOWLIST). The merge gate for the +# tiering PR (per the FPAD decision) is: cold-open NOT worse, routine NOT worse, and the prompt +# cache NOT dented — PLUS a chance-corrected tool-SELECTION check that lives outside this module +# (selection is about which tool the DM picks, not latency). Thresholds are deliberately explicit +# and tunable; this encodes the gate so a sweep yields a PASS/FAIL, not a wall of numbers. +AB_LATENCY_TOLERANCE = 0.05 # a 5% regression on a wall-clock metric counts as "worse" +AB_CACHE_CREATION_ABS_TOL = 2_000 # routine cache re-creation tokens tolerated as cache noise + + +def _delta(a: Any, b: Any) -> dict: + """A->B delta cell. None-safe (missing metric => null delta, never a crash).""" + if not isinstance(a, (int, float)) or not isinstance(b, (int, float)): + return {"a": a, "b": b, "delta": None, "pct": None} + d = b - a + return {"a": a, "b": b, "delta": round(d, 1), "pct": (round(100.0 * d / a, 1) if a else None)} + + +def compare_arms(arm_a: dict, arm_b: dict) -> dict: + """Compare a baseline rollup (arm_a) against a tiered rollup (arm_b) on the slab-A/B gate. + + Returns ``{verdict, checks, metrics, note}``. ``verdict`` is PASS / FAIL / INSUFFICIENT_DATA + (the last when a required metric is missing on either arm). ``checks`` are the hard gates; + ``metrics`` carries every A/B delta (incl. the expected token-mass WIN) for the reviewer.""" + ca = (arm_a.get("tokens") or {}).get("coldopen") or {} + cb = (arm_b.get("tokens") or {}).get("coldopen") or {} + ra = (arm_a.get("tokens") or {}).get("routine") or {} + rb = (arm_b.get("tokens") or {}).get("routine") or {} + + metrics = { + "coldopen_s": _delta(arm_a.get("coldopen_s"), arm_b.get("coldopen_s")), + "coldopen_cache_creation": _delta(ca.get("cache_creation"), cb.get("cache_creation")), + "coldopen_ttft_ms": _delta(ca.get("ttft_ms"), cb.get("ttft_ms")), + "s_per_beat": _delta(arm_a.get("s_per_beat"), arm_b.get("s_per_beat")), + "routine_cache_creation": _delta(ra.get("cache_creation"), rb.get("cache_creation")), + "routine_input": _delta(ra.get("input"), rb.get("input")), # the expected WIN (down) + "routine_ttft_ms": _delta(ra.get("ttft_ms"), rb.get("ttft_ms")), + } + + checks: dict[str, bool] = {} + a_co, b_co = arm_a.get("coldopen_s"), arm_b.get("coldopen_s") + if isinstance(a_co, (int, float)) and isinstance(b_co, (int, float)): + checks["cold_open_not_worse"] = b_co <= a_co * (1 + AB_LATENCY_TOLERANCE) + a_sb, b_sb = arm_a.get("s_per_beat"), arm_b.get("s_per_beat") + if isinstance(a_sb, (int, float)) and isinstance(b_sb, (int, float)): + checks["routine_not_worse"] = b_sb <= a_sb * (1 + AB_LATENCY_TOLERANCE) + a_cc, b_cc = ra.get("cache_creation"), rb.get("cache_creation") + if isinstance(a_cc, (int, float)) and isinstance(b_cc, (int, float)): + checks["cache_not_dented"] = b_cc <= a_cc + AB_CACHE_CREATION_ABS_TOL + # Informational (not a safety gate): the tiered arm SHOULD inject less uncached prefill. + a_in, b_in = ra.get("input"), rb.get("input") + if isinstance(a_in, (int, float)) and isinstance(b_in, (int, float)): + checks["input_mass_down"] = b_in < a_in + + required = ("cold_open_not_worse", "routine_not_worse", "cache_not_dented") + if any(k not in checks for k in required): + verdict = "INSUFFICIENT_DATA" + elif all(checks[k] for k in required): + verdict = "PASS" + else: + verdict = "FAIL" + + return { + "verdict": verdict, + "checks": checks, + "metrics": metrics, + "note": ( + "tiering A/B: arm_a=baseline (WORLDOS_ENGINE_ALWAYSLOAD=1), arm_b=tiered (=0). " + "Hard gate = cold_open_not_worse AND routine_not_worse AND cache_not_dented. " + "input_mass_down is the expected WIN (informational). The chance-corrected tool-" + "SELECTION check is gated separately. Run paired same-SHA/same-seed duos for signal." + ), + } + + def _main(argv: Optional[list[str]] = None) -> int: ap = argparse.ArgumentParser(description="Derive the F13-4 latency ledger from DM beat transcripts.") ap.add_argument("--dir", help="transcript directory ($T) — used with --run") @@ -195,9 +326,20 @@ def _main(argv: Optional[list[str]] = None) -> int: ap.add_argument("--stamp-into", default="", help="comma-separated PERSONA run dirs to stamp the " "rollup into as /latency.json (the shape release_readiness.read_latency reads) " "— this is what ACTIVATES the additive RRI latency gate on a real sweep") + ap.add_argument("--compare", nargs=2, metavar=("BASELINE_JSON", "TIERED_JSON"), + help="slab-A/B (Phase 3): compare two rollup JSON files (arm A=baseline " + "WORLDOS_ENGINE_ALWAYSLOAD=1, arm B=tiered =0) on the cold-open / routine / " + "cache-dent gate; prints {verdict, checks, metrics}, exits non-zero on FAIL") ap.add_argument("files", nargs="*", help="explicit beat transcript paths (overrides --dir/--run)") args = ap.parse_args(argv) + if args.compare: + arm_a = json.loads(Path(args.compare[0]).read_text()) + arm_b = json.loads(Path(args.compare[1]).read_text()) + cmp = compare_arms(arm_a, arm_b) + print(json.dumps(cmp, indent=2)) + return 0 if cmp["verdict"] == "PASS" else 1 + if args.files: files = sorted(args.files, key=lambda p: int(m.group(1)) if (m := _DM_RE.search(p)) else 0) result = rollup_files(files) diff --git a/qa/test_latency_rollup.py b/qa/test_latency_rollup.py index 7f80c41e..72caf5b6 100644 --- a/qa/test_latency_rollup.py +++ b/qa/test_latency_rollup.py @@ -20,15 +20,25 @@ import latency_rollup # noqa: E402 +def _usage(*, input_t, cache_creation, cache_read, output_t): + return {"input_tokens": input_t, "cache_creation_input_tokens": cache_creation, + "cache_read_input_tokens": cache_read, "output_tokens": output_t} + + def _write_beat(d: Path, run: str, nanos: int, *, api_ms, num_turns=1, - is_error=False, api_error_status=None, with_field=True): + is_error=False, api_error_status=None, with_field=True, + usage=None, ttft_ms=None): """Write a minimal stream-json transcript whose terminal result event carries the - given duration_api_ms / num_turns (the only fields the rollup reads).""" + given duration_api_ms / num_turns (+ optional usage token block / ttft_ms).""" res = {"type": "result", "subtype": "success", "is_error": is_error, "api_error_status": api_error_status, "duration_ms": (api_ms or 0) + 3000, "num_turns": num_turns, "result": "prose"} if with_field and api_ms is not None: res["duration_api_ms"] = api_ms + if usage is not None: + res["usage"] = usage + if ttft_ms is not None: + res["ttft_ms"] = ttft_ms path = d / f"{run}.dm.{nanos}.jsonl" lines = [ json.dumps({"type": "system", "subtype": "init"}), @@ -186,3 +196,84 @@ def test_cli_stamp_into_writes_sidecars(tmp_path): for d in (d1, d2): sidecar = json.loads((d / "latency.json").read_text()) assert sidecar["coldopen_s"] == 240.0 and sidecar["s_per_beat"] == 130.0 + + +# ── tool-schema-slab A/B (Phase 3): token/cache ledger + arm comparator ───────────── +# The slab is a per-request PREFILL tax, so the A/B reads the token ledger (cache_creation/ +# cache_read/input + ttft), and the gate is cold-open-not-worse + routine-not-worse + cache-not- +# dented. These verify the parsing and the PASS/FAIL/INSUFFICIENT_DATA verdict logic. + +def test_token_aggregates_cold_open_vs_routine(tmp_path): + run = "duo-tok" + # cold open: heavy cache CREATION (building the prefix); routine: cache_creation ~0 (cached). + _write_beat(tmp_path, run, 1000, api_ms=240000, num_turns=18, ttft_ms=8000, + usage=_usage(input_t=20000, cache_creation=150000, cache_read=1400000, output_t=12000)) + _write_beat(tmp_path, run, 2000, api_ms=100000, num_turns=4, ttft_ms=2000, + usage=_usage(input_t=18000, cache_creation=0, cache_read=1500000, output_t=9000)) + _write_beat(tmp_path, run, 3000, api_ms=120000, num_turns=5, ttft_ms=2200, + usage=_usage(input_t=20000, cache_creation=0, cache_read=1520000, output_t=11000)) + tok = latency_rollup.rollup_run(tmp_path, run)["tokens"] + assert tok["coldopen"]["cache_creation"] == 150000.0 + assert tok["coldopen"]["ttft_ms"] == 8000.0 + assert tok["routine"]["cache_creation"] == 0.0 # mean(0, 0) — a healthy cached prefix + assert tok["routine"]["input"] == 19000.0 # mean(18000, 20000) + assert tok["routine"]["ttft_ms"] == 2100.0 # mean(2000, 2200) + + +def _arm(coldopen_s, s_per_beat, *, co_cc, ro_cc, ro_input, ro_ttft=2000.0): + """A minimal rollup shaped like rollup_files() output, for comparator tests.""" + return { + "coldopen_s": coldopen_s, "s_per_beat": s_per_beat, "turns_per_beat": 4.0, "beats": 4, + "tokens": { + "coldopen": {"cache_creation": co_cc, "cache_read": 1_400_000.0, "input": 20000.0, + "output": 12000.0, "ttft_ms": 8000.0}, + "routine": {"cache_creation": ro_cc, "cache_read": 1_500_000.0, "input": ro_input, + "output": 9000.0, "ttft_ms": ro_ttft}, + }, + } + + +def test_compare_arms_pass_when_tiered_not_worse_and_cache_clean(): + base = _arm(240.0, 100.0, co_cc=160000.0, ro_cc=0.0, ro_input=40000.0) + tiered = _arm(236.0, 98.0, co_cc=150000.0, ro_cc=0.0, ro_input=26000.0) # faster, leaner, clean + cmp = latency_rollup.compare_arms(base, tiered) + assert cmp["verdict"] == "PASS" + assert cmp["checks"]["cold_open_not_worse"] and cmp["checks"]["routine_not_worse"] + assert cmp["checks"]["cache_not_dented"] and cmp["checks"]["input_mass_down"] + # the WIN is surfaced in metrics (input prefill mass down) + assert cmp["metrics"]["routine_input"]["delta"] == -14000.0 + + +def test_compare_arms_fail_on_cache_dent(): + base = _arm(240.0, 100.0, co_cc=160000.0, ro_cc=0.0, ro_input=40000.0) + # tiered re-creates the prompt cache on routine beats (15k > 0 + 2k tol) -> DENT -> FAIL + tiered = _arm(236.0, 98.0, co_cc=150000.0, ro_cc=15000.0, ro_input=26000.0) + cmp = latency_rollup.compare_arms(base, tiered) + assert cmp["verdict"] == "FAIL" + assert cmp["checks"]["cache_not_dented"] is False + + +def test_compare_arms_fail_on_cold_open_regression(): + base = _arm(240.0, 100.0, co_cc=160000.0, ro_cc=0.0, ro_input=40000.0) + tiered = _arm(300.0, 98.0, co_cc=150000.0, ro_cc=0.0, ro_input=26000.0) # cold open +25% -> worse + cmp = latency_rollup.compare_arms(base, tiered) + assert cmp["verdict"] == "FAIL" + assert cmp["checks"]["cold_open_not_worse"] is False + + +def test_compare_arms_insufficient_data_when_metric_missing(): + base = _arm(240.0, 100.0, co_cc=160000.0, ro_cc=0.0, ro_input=40000.0) + thin = {"coldopen_s": 236.0, "s_per_beat": 98.0, "tokens": {"coldopen": {}, "routine": None}} + cmp = latency_rollup.compare_arms(base, thin) + assert cmp["verdict"] == "INSUFFICIENT_DATA" # routine cache_creation absent on arm B + + +def test_cli_compare_exit_codes(tmp_path): + base = tmp_path / "a.json" + tiered_ok = tmp_path / "b_ok.json" + tiered_bad = tmp_path / "b_bad.json" + base.write_text(json.dumps(_arm(240.0, 100.0, co_cc=160000.0, ro_cc=0.0, ro_input=40000.0))) + tiered_ok.write_text(json.dumps(_arm(236.0, 98.0, co_cc=150000.0, ro_cc=0.0, ro_input=26000.0))) + tiered_bad.write_text(json.dumps(_arm(236.0, 98.0, co_cc=150000.0, ro_cc=15000.0, ro_input=26000.0))) + assert latency_rollup._main(["--compare", str(base), str(tiered_ok)]) == 0 + assert latency_rollup._main(["--compare", str(base), str(tiered_bad)]) == 1