feat(qa): latency-rollup token/cache ledger + slab A/B comparator (Phase 3 harness)

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 <dir>/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)

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