PD-sep matrix results: C2/C3/C4 figures + empirical mechanism refined

Captures 5 runs from the experiment matrix (combined-ca x3 seeds,
pdsep-4p4d seed1, pdsep-6p2d seed1) on traces/w600_r0.0015_st30.jsonl
with cuda graphs enabled. The headline:

  combined-ca:  TTFT p50 0.91s   success 99.5%
  pdsep-4p4d:   TTFT p50 62.8s   success 52%   (69x worse, half dropped)
  pdsep-6p2d:   TTFT p50 51.1s   success 68%   (56x worse, third dropped)

C2 (fig_c2): headline bars per config with error bars.
C3 (fig_c3): per-instance KV utilization time-series. Both PD-sep
  splits hit the memory wall, but the side differs by P:D ratio --
  4P+4D pins the P-side, 6P+2D pins both sides (D-side back-pressures
  P-side).
C4 (fig_c4): TTFT stacked breakdown. 99% of PD-sep TTFT is P-side
  prefill compute; D-side wait + first token is <=1.2s. The bottleneck
  is P-side prefill queueing, not D-side decode wait as the original
  analytical model assumed.

system_analysis.md gains a Layer 5b that reconciles the analytical
KV-wall model (which considered D-side only) with the empirical
finding that the wall hits whichever side has fewer GPUs, and
co-saturates both at extreme splits via D-side back-pressure.

plot_pd_matrix.py ingests outputs/pd_matrix/* into all four figures.
bench.sh gained AGENTIC_STEP_LOG_DIR hooks for future runs (set during
this work but not used by the current matrix's data).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

analysis/characterization/joined_analysis.py

@@ -0,0 +1,531 @@
+"""A5: joined-record analysis from instrumented runs.
+
+Inputs (all optional; functions degrade gracefully when missing):
+
+- replayer metrics.jsonl with A1 fields
+  (t_dispatch_unix, t_first_token_unix, t_finish_unix, proxy_request_id,
+  endpoint_url, trace_hash_ids)
+- proxy breakdown.json with A2 fields
+  (session_id, candidate_scores, chosen_score_*, t_first_token_unix,
+  t_done_unix, t_decision_unix)
+- proxy worker_state.jsonl with A2 schema (one row per route decision)
+- vLLM scheduler engine_state JSONLs from A3
+  (one per engine, env AGENTIC_STEP_LOG_PATH)
+
+Outputs under <out_dir>/:
+
+- joined.jsonl           — per-request join across all sources
+- reuse_decomposition.json
+- interference_index.json
+- hotspot_index.json
+- failure_label.jsonl
+- window_summary.json    — when run_meta.json (from SRR loadgen) is present
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import statistics
+from collections import defaultdict
+from pathlib import Path
+from typing import Any, Iterable
+
+JsonDict = dict[str, Any]
+
+
+# ---------- I/O ---------------------------------------------------------
+
+def load_jsonl(path: Path) -> list[JsonDict]:
+    if not path.exists():
+        return []
+    out: list[JsonDict] = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            out.append(json.loads(line))
+        except json.JSONDecodeError:
+            continue
+    return out
+
+
+def load_json(path: Path) -> Any:
+    if not path.exists():
+        return None
+    text = path.read_text(encoding="utf-8").strip()
+    if not text:
+        return None
+    return json.loads(text)
+
+
+def write_json(path: Path, data: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, indent=2, sort_keys=True))
+
+
+def write_jsonl(path: Path, rows: Iterable[JsonDict]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as fh:
+        for row in rows:
+            fh.write(json.dumps(row, sort_keys=True) + "\n")
+
+
+# ---------- Joining -----------------------------------------------------
+
+def build_joined_records(
+    metrics: list[JsonDict],
+    breakdown: list[JsonDict],
+    worker_state: list[JsonDict],
+) -> list[JsonDict]:
+    """Join metrics + breakdown + worker_state by request_id.
+
+    Returns one row per metrics record (the load-generator's view of truth).
+    Missing sources leave the corresponding columns as None.
+    """
+    bk_by_id = {str(r.get("request_id")): r for r in breakdown if r.get("request_id")}
+    ws_by_id = {str(r.get("request_id")): r for r in worker_state if r.get("request_id")}
+
+    joined: list[JsonDict] = []
+    for m in metrics:
+        rid = str(m.get("request_id") or m.get("proxy_request_id") or "")
+        bk = bk_by_id.get(rid)
+        ws = ws_by_id.get(rid)
+        row: JsonDict = {
+            "request_id": rid,
+            "session_id": m.get("session_id"),
+            "turn_id": m.get("turn_id"),
+            "trace_timestamp_s": m.get("trace_timestamp_s"),
+            "input_length": m.get("input_length"),
+            "output_length": m.get("output_length"),
+            "cached_tokens": m.get("cached_tokens"),
+            "actual_output_tokens": m.get("actual_output_tokens"),
+            "latency_s": m.get("latency_s"),
+            "ttft_s": m.get("ttft_s"),
+            "tpot_s": m.get("tpot_s"),
+            "t_dispatch_unix": m.get("t_dispatch_unix"),
+            "t_first_token_unix": m.get("t_first_token_unix"),
+            "t_finish_unix": m.get("t_finish_unix"),
+            "endpoint_url": m.get("endpoint_url"),
+            "trace_hash_ids": m.get("trace_hash_ids") or [],
+            "error": m.get("error"),
+        }
+        if bk:
+            row["policy"] = bk.get("policy")
+            row["route_class"] = bk.get("route_class")
+            row["routed_to"] = bk.get("routed_to")
+            row["chosen_idx"] = bk.get("chosen_idx")
+            row["chosen_score_linear"] = bk.get("chosen_score_linear")
+            row["chosen_score_lmetric"] = bk.get("chosen_score_lmetric")
+            row["estimated_new_tokens"] = bk.get("estimated_new_tokens")
+            row["cache_hit_proxy"] = bk.get("cache_hit")
+            row["proxy_t_decision_unix"] = bk.get("t_decision_unix")
+            row["proxy_t_first_token_unix"] = bk.get("t_first_token_unix")
+            row["proxy_t_done_unix"] = bk.get("t_done_unix")
+        if ws:
+            row["worker_state_at_decision"] = ws.get("workers")
+        joined.append(row)
+    return joined
+
+
+# ---------- Reuse decomposition (real) ----------------------------------
+
+def reuse_decomposition(records: list[JsonDict], block_size: int = 16) -> JsonDict:
+    """Real intra/cross/shared decomposition keyed on (session, hash blocks)."""
+    if not records:
+        return {"status": "unavailable", "reason": "no joined records"}
+
+    # block first-seen index: hash_id -> (session_id, first_seen_seq)
+    first_seen: dict[int, tuple[str, int]] = {}
+    block_sessions: dict[int, set[str]] = defaultdict(set)
+    seq = 0
+    for r in sorted(records, key=lambda x: x.get("t_dispatch_unix") or 0.0):
+        sid = str(r.get("session_id"))
+        for h in r.get("trace_hash_ids") or []:
+            h_int = int(h)
+            if h_int not in first_seen:
+                first_seen[h_int] = (sid, seq)
+            block_sessions[h_int].add(sid)
+        seq += 1
+
+    total_cached = 0
+    intra = cross = shared = unclassified = 0
+    for r in records:
+        cached = r.get("cached_tokens") or 0
+        if not cached:
+            continue
+        total_cached += cached
+        sid = str(r.get("session_id"))
+        hashes = [int(h) for h in (r.get("trace_hash_ids") or [])]
+        if not hashes:
+            unclassified += cached
+            continue
+        # Approximate: classify the cached tokens by the first non-current
+        # owner of any hash block we've seen before.
+        block_tokens = max(cached // max(len(hashes), 1), 1)
+        for h in hashes:
+            first_sid, _ = first_seen.get(h, (None, None))
+            if first_sid is None:
+                unclassified += min(block_tokens, cached)
+            elif first_sid == sid:
+                intra += min(block_tokens, cached)
+            elif len(block_sessions[h]) >= 8:
+                shared += min(block_tokens, cached)
+            else:
+                cross += min(block_tokens, cached)
+            cached -= block_tokens
+            if cached <= 0:
+                break
+
+    return {
+        "status": "supported",
+        "total_cached_tokens": total_cached,
+        "intra_session_tokens": intra,
+        "cross_session_tokens": cross,
+        "shared_prefix_tokens": shared,
+        "unclassified_tokens": unclassified,
+        "fractions": _fractions(intra, cross, shared, unclassified),
+        "shared_prefix_min_sessions": 8,
+    }
+
+
+def _fractions(*parts: int) -> JsonDict:
+    total = sum(parts)
+    if total == 0:
+        return {"intra": 0.0, "cross": 0.0, "shared": 0.0, "unclassified": 0.0}
+    labels = ["intra", "cross", "shared", "unclassified"]
+    return {label: parts[i] / total for i, label in enumerate(labels)}
+
+
+# ---------- Interference index (B2) -------------------------------------
+
+def interference_index(
+    joined: list[JsonDict],
+    engine_state_by_worker: dict[str, list[JsonDict]],
+) -> JsonDict:
+    """Label each completed request's decode period as overlap / no-overlap.
+
+    A request 'overlaps same-worker prefill' if any scheduler step on the
+    chosen worker between (t_first_token_unix, t_finish_unix) had
+    prefill_tokens > 0 from a request other than this one.
+    """
+    if not joined or not engine_state_by_worker:
+        return {"status": "unavailable",
+                "reason": "missing joined records or engine state"}
+
+    tpot_overlap: list[float] = []
+    tpot_clean: list[float] = []
+    for r in joined:
+        rid = r.get("request_id")
+        worker = _normalize_worker(r.get("endpoint_url") or r.get("routed_to"))
+        steps = engine_state_by_worker.get(worker)
+        if not steps:
+            continue
+        t0 = r.get("t_first_token_unix")
+        t1 = r.get("t_finish_unix")
+        tpot = r.get("tpot_s")
+        if t0 is None or t1 is None or tpot is None or r.get("error"):
+            continue
+        overlap = False
+        for s in steps:
+            t = s.get("t_unix")
+            if t is None or t < t0 or t > t1:
+                continue
+            if not s.get("prefill_tokens"):
+                continue
+            # If the only prefill belongs to *this* request, that's still
+            # this request's own prefill warming up, not interference.
+            other_prefill = False
+            for pr in s.get("per_req", []) or []:
+                if pr.get("phase") == "prefill" and pr.get("rid") != rid:
+                    other_prefill = True
+                    break
+            if other_prefill:
+                overlap = True
+                break
+        (tpot_overlap if overlap else tpot_clean).append(float(tpot))
+
+    p90_overlap = _percentile(tpot_overlap, 0.90) if tpot_overlap else None
+    p90_clean = _percentile(tpot_clean, 0.90) if tpot_clean else None
+    idx = None
+    if p90_overlap is not None and p90_clean and p90_clean > 0:
+        idx = p90_overlap / p90_clean
+
+    return {
+        "status": "supported" if idx is not None else "partial",
+        "n_overlap_requests": len(tpot_overlap),
+        "n_clean_requests": len(tpot_clean),
+        "tpot_p90_overlap_s": p90_overlap,
+        "tpot_p90_clean_s": p90_clean,
+        "interference_index": idx,
+    }
+
+
+def _normalize_worker(url_or_id: str | None) -> str | None:
+    """Map endpoint URLs to AGENTIC_WORKER_ID convention engine_{i}."""
+    if not url_or_id:
+        return None
+    if url_or_id.startswith("engine_"):
+        return url_or_id
+    # Endpoint URLs look like http://host:8000; map by port offset against 8000.
+    try:
+        port = int(url_or_id.rsplit(":", 1)[1].split("/")[0])
+        return f"engine_{port - 8000}"
+    except (ValueError, IndexError):
+        return url_or_id
+
+
+# ---------- Hotspot index (B3) ------------------------------------------
+
+def hotspot_index(joined: list[JsonDict]) -> JsonDict:
+    """max/median per-worker queue-delay p90 across completed requests."""
+    if not joined:
+        return {"status": "unavailable"}
+
+    by_worker_queue: dict[str, list[float]] = defaultdict(list)
+    by_worker_latency: dict[str, list[float]] = defaultdict(list)
+    for r in joined:
+        if r.get("error"):
+            continue
+        worker = r.get("routed_to") or r.get("endpoint_url")
+        if not worker:
+            continue
+        # queue delay proxy: (t_first_token - t_dispatch) - prefill estimate
+        # is fragile; use raw TTFT as the queue-stressing signal.
+        ttft = r.get("ttft_s")
+        lat = r.get("latency_s")
+        if ttft is not None:
+            by_worker_queue[worker].append(float(ttft))
+        if lat is not None:
+            by_worker_latency[worker].append(float(lat))
+
+    worker_p90_q: dict[str, float] = {
+        w: _percentile(v, 0.90) for w, v in by_worker_queue.items() if v
+    }
+    worker_p90_lat: dict[str, float] = {
+        w: _percentile(v, 0.90) for w, v in by_worker_latency.items() if v
+    }
+    p90s_q = sorted(worker_p90_q.values())
+    idx = None
+    if len(p90s_q) >= 2:
+        median = p90s_q[len(p90s_q) // 2]
+        if median > 0:
+            idx = max(p90s_q) / median
+
+    return {
+        "status": "supported" if idx is not None else "partial",
+        "per_worker_ttft_p90_s": worker_p90_q,
+        "per_worker_latency_p90_s": worker_p90_lat,
+        "hotspot_index_ttft_p90": idx,
+    }
+
+
+# ---------- Failure label (B5) ------------------------------------------
+
+DEFAULT_SLO = {
+    "ttft_p90_s": 2.0,
+    "tpot_p90_s": 0.15,
+}
+
+
+def label_slow_requests(
+    joined: list[JsonDict],
+    engine_state_by_worker: dict[str, list[JsonDict]],
+    slo: JsonDict | None = None,
+    slow_ttft_factor: float = 2.0,
+) -> list[JsonDict]:
+    slo = slo or DEFAULT_SLO
+    ttft_threshold = float(slo["ttft_p90_s"]) * slow_ttft_factor
+
+    # Per-worker queue p90 to flag hot workers
+    by_worker_ttft: dict[str, list[float]] = defaultdict(list)
+    for r in joined:
+        if r.get("ttft_s") is not None:
+            by_worker_ttft[r.get("routed_to") or ""].append(float(r["ttft_s"]))
+    worker_p90 = {w: _percentile(v, 0.90) for w, v in by_worker_ttft.items() if v}
+    global_p90 = _percentile(
+        [v for vs in by_worker_ttft.values() for v in vs], 0.90,
+    ) if by_worker_ttft else None
+    hot_workers = {w for w, p in worker_p90.items()
+                   if global_p90 and p > global_p90 * 1.2}
+
+    labels: list[JsonDict] = []
+    for r in joined:
+        ttft = r.get("ttft_s")
+        if ttft is None or r.get("error"):
+            continue
+        if ttft <= ttft_threshold:
+            continue
+        label = _assign_label(r, hot_workers, engine_state_by_worker)
+        labels.append({
+            "request_id": r.get("request_id"),
+            "session_id": r.get("session_id"),
+            "routed_to": r.get("routed_to"),
+            "ttft_s": ttft,
+            "latency_s": r.get("latency_s"),
+            "input_length": r.get("input_length"),
+            "cached_tokens": r.get("cached_tokens"),
+            "estimated_new_tokens": r.get("estimated_new_tokens"),
+            "label": label,
+        })
+    return labels
+
+
+def _assign_label(
+    r: JsonDict,
+    hot_workers: set[str],
+    engine_state_by_worker: dict[str, list[JsonDict]],
+) -> str:
+    worker = _normalize_worker(r.get("endpoint_url") or r.get("routed_to"))
+    rid = r.get("request_id")
+    steps = engine_state_by_worker.get(worker, [])
+    t0 = r.get("t_first_token_unix")
+    t1 = r.get("t_finish_unix")
+    if steps and t0 and t1:
+        for s in steps:
+            t = s.get("t_unix")
+            if t is None or t < t0 or t > t1:
+                continue
+            for pr in s.get("per_req", []) or []:
+                if pr.get("phase") == "prefill" and pr.get("rid") != rid:
+                    return "same_worker_prefill_overlap"
+    if (r.get("routed_to") or "") in hot_workers:
+        return "hot_worker_queue"
+    est = r.get("estimated_new_tokens") or 0
+    inp = r.get("input_length") or 0
+    if est and inp and est >= 0.5 * inp:
+        return "cache_miss_large_append"
+    snap = r.get("worker_state_at_decision") or []
+    if snap:
+        chosen_idx = r.get("chosen_idx")
+        if isinstance(chosen_idx, int) and 0 <= chosen_idx < len(snap):
+            cached = snap[chosen_idx].get("cached_blocks", 0)
+            if cached and cached > 50_000:
+                return "high_kv_occupancy"
+    return "unknown"
+
+
+# ---------- Window summary (B4) -----------------------------------------
+
+def window_summary(joined: list[JsonDict], run_meta: JsonDict | None) -> JsonDict:
+    if not run_meta:
+        return {"status": "unavailable", "reason": "no run_meta"}
+    warmup_end = float(run_meta["warmup_end_unix"])
+    steady_end = float(run_meta["steady_end_unix"])
+
+    buckets: dict[str, list[JsonDict]] = {"warmup": [], "steady": [], "drain": []}
+    for r in joined:
+        t = r.get("t_dispatch_unix")
+        if t is None:
+            continue
+        if t < warmup_end:
+            buckets["warmup"].append(r)
+        elif t < steady_end:
+            buckets["steady"].append(r)
+        else:
+            buckets["drain"].append(r)
+
+    out: JsonDict = {"run_meta": run_meta, "windows": {}}
+    for name, rows in buckets.items():
+        ttft = [r["ttft_s"] for r in rows if r.get("ttft_s") is not None]
+        tpot = [r["tpot_s"] for r in rows if r.get("tpot_s") is not None]
+        e2e = [r["latency_s"] for r in rows if r.get("latency_s") is not None]
+        errs = sum(1 for r in rows if r.get("error"))
+        out["windows"][name] = {
+            "attempted": len(rows),
+            "completed": len(rows) - errs,
+            "errored": errs,
+            "ttft_p50_s": _percentile(ttft, 0.50) if ttft else None,
+            "ttft_p90_s": _percentile(ttft, 0.90) if ttft else None,
+            "tpot_p50_s": _percentile(tpot, 0.50) if tpot else None,
+            "tpot_p90_s": _percentile(tpot, 0.90) if tpot else None,
+            "e2e_p50_s": _percentile(e2e, 0.50) if e2e else None,
+            "e2e_p90_s": _percentile(e2e, 0.90) if e2e else None,
+        }
+    return out
+
+
+# ---------- helpers -----------------------------------------------------
+
+def _percentile(values: list[float], pct: float) -> float | None:
+    if not values:
+        return None
+    sorted_vals = sorted(values)
+    if len(sorted_vals) == 1:
+        return sorted_vals[0]
+    rank = pct * (len(sorted_vals) - 1)
+    lo = int(rank)
+    hi = min(lo + 1, len(sorted_vals) - 1)
+    frac = rank - lo
+    return sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac
+
+
+def load_engine_state(dir_path: Path) -> dict[str, list[JsonDict]]:
+    """Load all engine_*.jsonl files from a directory; key by worker id."""
+    if not dir_path.exists() or not dir_path.is_dir():
+        return {}
+    by_worker: dict[str, list[JsonDict]] = {}
+    for p in sorted(dir_path.glob("engine_*.jsonl")):
+        worker_id = p.stem  # 'engine_0', etc.
+        rows = load_jsonl(p)
+        # Sort steps by time for binary-search-friendly access.
+        rows.sort(key=lambda r: r.get("t_unix") or 0.0)
+        by_worker[worker_id] = rows
+    return by_worker
+
+
+# ---------- CLI ---------------------------------------------------------
+
+def main(argv: list[str] | None = None) -> None:
+    p = argparse.ArgumentParser(description="A5 joined analysis")
+    p.add_argument("--metrics", type=Path, required=True)
+    p.add_argument("--breakdown", type=Path, default=None)
+    p.add_argument("--worker-state", type=Path, default=None)
+    p.add_argument("--engine-state-dir", type=Path, default=None,
+                   help="Directory containing engine_*.jsonl from A3 patch")
+    p.add_argument("--run-meta", type=Path, default=None,
+                   help="run_meta or window_summary.json from SRR loadgen")
+    p.add_argument("--out-dir", type=Path, required=True)
+    p.add_argument("--slow-ttft-factor", type=float, default=2.0)
+    args = p.parse_args(argv)
+
+    metrics = load_jsonl(args.metrics)
+    breakdown_raw = load_json(args.breakdown) if args.breakdown else []
+    if isinstance(breakdown_raw, dict):
+        breakdown_raw = breakdown_raw.get("records", [breakdown_raw])
+    breakdown = list(breakdown_raw or [])
+    worker_state_raw = load_json(args.worker_state) if args.worker_state else []
+    if isinstance(worker_state_raw, dict):
+        worker_state_raw = worker_state_raw.get("records", [worker_state_raw])
+    worker_state = list(worker_state_raw or [])
+    engine_state = (
+        load_engine_state(args.engine_state_dir) if args.engine_state_dir else {}
+    )
+    run_meta = load_json(args.run_meta) if args.run_meta else None
+
+    joined = build_joined_records(metrics, breakdown, worker_state)
+
+    args.out_dir.mkdir(parents=True, exist_ok=True)
+    write_jsonl(args.out_dir / "joined.jsonl", joined)
+    write_json(args.out_dir / "reuse_decomposition.json",
+                reuse_decomposition(joined))
+    write_json(args.out_dir / "interference_index.json",
+                interference_index(joined, engine_state))
+    write_json(args.out_dir / "hotspot_index.json",
+                hotspot_index(joined))
+    labels = label_slow_requests(joined, engine_state,
+                                  slow_ttft_factor=args.slow_ttft_factor)
+    write_jsonl(args.out_dir / "failure_label.jsonl", labels)
+    counts: dict[str, int] = defaultdict(int)
+    for L in labels:
+        counts[L["label"]] += 1
+    write_json(args.out_dir / "failure_breakdown.json",
+                {"counts": dict(counts), "n_slow": len(labels)})
+    write_json(args.out_dir / "window_summary.json",
+                window_summary(joined, run_meta))
+
+
+if __name__ == "__main__":
+    main()

analysis/characterization_todo_for_interns.md

@@ -2,6 +2,35 @@
 
 Status: execution checklist for interns
 Date: 2026-05-25
+Last progress audit: 2026-05-25
+
+## Progress Snapshot (2026-05-25)
+
+| Batch | State | Evidence |
+|---|---|---|
+| B0 Substrate audit | tool DONE, legacy runs partial | `analysis/characterization/analyze.py` implements per-session concurrency/arrival/inter-turn analyzer; legacy `metrics.jsonl` lacks dispatch/finish timestamps so actual sequentiality cannot be proven on old runs (correctly labeled in `current_results/`) |
+| B1 Workload characterization | trace-shape DONE, reuse pending | `current_results/full_trace_summary.json` covers 2.11M req / 1.31M sessions from `051315-051317.jsonl`; KV-footprint and reuse decomposition still require `--kv-bytes-per-token` rerun and cached_tokens+hash_ids joined records |
+| B2 PD interference | protocol DONE, run pending | `analysis/characterization/protocols.md` Batch 2 section ready; needs fresh GPU run with decode-step + prefill-chunk timestamps |
+| B3 Hot-spot imbalance | partial; needs new instrumentation | Legacy `gpu_util.csv` shows imbalance but lacks per-worker queue delay and session→worker map |
+| B4 SRR sweep | NOT DONE | No arrival-rate sweep artifacts; depends on session-causal open-loop loadgen |
+| B5 Failure attribution | NOT DONE | Depends on B2/B4 outputs |
+| B6 Audit package | scaffold DONE | `current_results/{characterization_claim_matrix.md, all_figures_index.md, reviewer_risk_register.md, main_claim_allowed_runs.md, reproduction_commands.sh}` + 6 figures committed |
+
+Reusable assets already in repo:
+
+- `analysis/characterization/analyze.py` — B0+B1 CPU-only analyzer
+- `analysis/characterization/summarize_runs.py` — existing-run audit producing the B6 scaffold
+- `analysis/characterization/plot_current_results.py` — figure regeneration script
+- `analysis/characterization/protocols.md` — B2–B6 protocol with required instrumentation, sweep, pass condition
+- `analysis/characterization/current_results/` — current audit package (claim matrix + risk register + allowed-runs gate + 6 PNG figures)
+
+Hard gates still blocking main claims:
+
+1. Replayer/proxy must emit per-request dispatch + finish/error wall-clock timestamps (blocks B0 actual sequentiality, B4 SRR validity).
+2. Per-request record must carry `session_id` + `hash_ids` + `cached_tokens` jointly (blocks B1 reuse decomposition).
+3. Engine/proxy must log decode-step and prefill-chunk timestamps with worker id (blocks B2 interference index).
+4. Proxy must log route decision, chosen worker, candidate scores, per-worker queue/KV/APC snapshot (blocks B3 hot-spot proof).
+
 
 ## 0. Purpose
 
@@ -84,6 +113,8 @@ saved alongside the artifact.
 
 ## 2. Batch 0: Benchmark Substrate Audit
 
+Status: analyzer DONE (`analyze.py`); legacy-run sequentiality claim BLOCKED by missing dispatch/finish timestamps in `metrics.jsonl`. New replayer must add those fields before any `online_realistic` classification is allowed.
+
 ### Goal
 
 Prove the load generator and trace replay are valid before trusting any
@@ -147,6 +178,8 @@ The `audit.md` must answer:
 
 ## 3. Batch 1: Workload Characterization
 
+Status: trace-shape items (1, 2, 3, 6, 8) DONE on full 7200 s GLM-5.1 trace; recorded in `current_results/full_trace_summary.json`. Items 4 (KV footprint), 5 (reuse decomposition), 7 (uncached append delta) are PENDING because they need `--kv-bytes-per-token` for the production model and joinable `cached_tokens`+`hash_ids` per request.
+
 ### Goal
 
 Establish agentic workload facts independent of any proposed system.
@@ -225,6 +258,8 @@ links and plotted figures.
 
 ## 4. Batch 2: PD-Colo Prefill-Decode Interference Proof
 
+Status: protocol DONE (`analysis/characterization/protocols.md` §"Batch 2 Protocol"); execution NOT STARTED — needs new engine instrumentation for decode-step and prefill-chunk timestamps.
+
 ### Goal
 
 Prove that PD-colocation can suffer from prefill-decode interference under
@@ -302,6 +337,8 @@ The `audit.md` must answer:
 
 ## 5. Batch 3: Session Hot-Spot Residual Imbalance Proof
 
+Status: protocol DONE; partial signal from legacy `gpu_util.csv` (GPU-util imbalance visible) but causal proof NOT STARTED — needs per-worker queue/KV/APC and session→worker map from instrumented proxy.
+
 ### Goal
 
 Prove that cache-aware/LMetric is a strong baseline but still leaves residual
@@ -398,6 +435,8 @@ The `audit.md` must answer:
 
 ## 6. Batch 4: Sustainable Request Rate Sweep
 
+Status: protocol DONE; execution NOT STARTED — requires open-loop session-causal loadgen and policy-comparable arrival process.
+
 ### Goal
 
 Connect interference and hot-spot mechanisms to the final metric:
@@ -476,6 +515,8 @@ The `audit.md` must answer:
 
 ## 7. Batch 5: Failure Attribution Near SRR Boundary
 
+Status: protocol DONE; execution NOT STARTED — depends on B2 instrumentation and B4 SRR boundary.
+
 ### Goal
 
 At and around the PD-colo/LMetric failure point, determine whether SLO
@@ -542,6 +583,8 @@ The batch must answer:
 
 ## 8. Batch 6: Audit Package
 
+Status: scaffold DONE — all five final artifacts exist under `analysis/characterization/current_results/` and are regenerated by `summarize_runs.py` + `plot_current_results.py`. Future B2–B5 outputs must be merged into the same package by re-running `summarize_runs.py` after new runs.
+
 ### Goal
 
 Make the whole characterization package reviewable by a strict systems

analysis/claude_characterization_work_plan.md

@@ -0,0 +1,360 @@
+# Claude Characterization Work Plan
+
+Status: planning, awaiting dash0 idle
+Date: 2026-05-25
+Owner: Claude (not interns)
+Source of requirements: `analysis/characterization_todo_for_interns.md`
+
+## Scope
+
+This plan covers the four hard gates and the B2–B5 GPU experiments that the
+intern TODO marks as `NOT DONE` / `protocol DONE`. The B0 analyzer, the
+B1 trace-shape statistics, and the B6 audit scaffold are already done; this
+plan does **not** re-do them, only refreshes their inputs.
+
+The work is split into:
+
+- **Phase A (CPU-only)** — instrumentation + analyzer extensions. Can run
+  on the local dev box; does **not** need dash0. Must finish before any
+  GPU run.
+- **Phase B (dash0 GPU)** — controlled microbench + routing sweep + SRR
+  sweep + failure attribution.
+- **Phase C (CPU-only)** — final audit package refresh.
+
+## Phase A: Instrumentation + Analyzer (CPU-only, before dash0)
+
+### A1. Replayer instrumentation — close Gate 1 + Gate 2
+
+File: `replayer/metrics.py`, `replayer/replay.py`
+
+Add these fields to `RequestMetrics`:
+
+```text
+t_dispatch_unix     float   # absolute wall-clock when POST starts
+t_first_token_unix  float   # absolute wall-clock at first stream chunk
+t_finish_unix       float   # absolute wall-clock at stream done or error
+proxy_request_id    str     # value sent in X-Request-Id (matches breakdown)
+endpoint_url        str     # which proxy/instance the request hit
+trace_hash_ids      list[int]  # carried from trace for reuse joins
+```
+
+Change `_dispatch_request` to:
+
+- send a deterministic `X-Request-Id: <session_id>:<turn_id>` header (so
+  proxy breakdown can be joined to metrics by exact key);
+- record `time.time()` (unix) at dispatch, first token, finish; keep
+  `perf_counter` for the latency arithmetic.
+
+Acceptance: a 30-request smoke run produces `metrics.jsonl` where every
+row has those fields; `breakdown.json` rows from the proxy have the same
+`request_id` keys.
+
+Effort: 1 small PR. Pure CPU.
+
+### A2. Proxy instrumentation — close Gate 1 + Gate 3 + Gate 4
+
+File: `scripts/cache_aware_proxy.py`
+
+Changes:
+
+1. Honor incoming `X-Request-Id`: if header present, use it instead of
+   generating a new uuid. Falls back to uuid otherwise.
+2. Record on every breakdown row:
+   - `session_id` (already on header, not currently stored)
+   - `input_length`
+   - `estimated_new_tokens` (already produced by router)
+   - `candidate_scores` (list of `{url, p_tokens_score, cache_score, bs,
+     occupancy}`)
+   - `chosen_score`
+3. At route decision time, snapshot per-worker state:
+   - `pending_prefill_tokens` per worker
+   - `running_decode_requests` per worker
+   - `kv_blocks_used` / `kv_blocks_total` per worker
+   - `apc_hits` / `apc_queries` cumulative per worker
+   Write to a separate `worker_state.jsonl` (one line per route decision)
+   with `(t_decision_unix, request_id, per_worker_state)`.
+4. New endpoint `GET /worker_state` returns the latest snapshot per worker
+   (for sanity / live debugging).
+
+Acceptance: smoke run produces `breakdown.json` with new fields and a
+non-empty `worker_state.jsonl` that joins to breakdown by `request_id`.
+
+Effort: 1 medium PR. Pure CPU + light proxy work.
+
+### A3. Engine-side step timestamps — close Gate 3 for B2
+
+vLLM 0.18.1 already exposes:
+
+- `vllm:request_prefill_time_seconds` (histogram, per-request)
+- `vllm:request_decode_time_seconds`
+- `vllm:time_per_output_token_seconds`
+- step-level scheduler stats via `engine.async_step` logging
+
+For B2 we need decode-step and prefill-chunk timestamps with worker id.
+Plan:
+
+1. Inspect whether the vLLM proxy can be polled at high rate (e.g.
+   100 Hz) for per-engine scheduler counters
+   (`num_running`, `num_waiting`, `gpu_cache_usage`,
+   `prefix_cache_queries`, `prefix_cache_hits`). If yes, sample
+   into `engine_state.jsonl` during runs.
+2. If finer step-level data is needed, patch one vLLM file
+   (`vllm/engine/async_llm_engine.py` step loop or
+   `vllm/v1/core/sched/scheduler.py`) to emit a JSONL line per
+   scheduler step with `(t_unix, worker_id, num_prefill_tokens_scheduled,
+   num_decode_steps, running_request_ids)`. Patch goes under `patches/`
+   so it can be applied/reverted cleanly.
+3. Worker id mapping: when running TP1xDP8 or similar, each engine
+   listens on a distinct port; `worker_id == endpoint_url`.
+
+Acceptance: a single 10-minute run produces `engine_state.jsonl` from
+which a decode step at time T on worker W can be classified as
+"overlapping a same-worker prefill chunk" or not.
+
+Effort: 1 medium investigation (decide poll vs patch) + 1 medium PR.
+
+### A4. Open-loop session-causal loadgen for B4
+
+File: `replayer/replay.py` (new mode) or new `replayer/srr_loadgen.py`
+
+Current replayer dispatches by trace timestamps. SRR sweep needs:
+
+- pool of session templates (each = ordered list of turns from the
+  trace);
+- Poisson arrivals of new sessions at rate `lambda`;
+- within a session: strict sequentiality (turn N+1 waits for turn N
+  finish);
+- per-run warmup window (e.g. 60s) + steady-state window (e.g. 300s);
+- attempted / completed / error counters per window.
+
+Add a new mode `--mode srr --arrival-rate <lambda>
+--warmup-s 60 --steady-s 300 --session-pool-size N`. The trace
+file becomes the pool; sessions are drawn with replacement.
+
+Acceptance: at `lambda = 0.5 sess/s`, the run shows exponential inter-
+arrival times and per-session sequentiality in `metrics.jsonl`. A
+`window_summary.json` lists warmup vs steady-state attempted/completed.
+
+Effort: 1 medium PR.
+
+### A5. Analyzer extensions
+
+File: `analysis/characterization/analyze.py` (extend, do not rewrite)
+
+Add:
+
+1. **Joined-record builder.** Given `--metrics metrics.jsonl
+   --breakdown breakdown.json --worker-state worker_state.jsonl
+   --engine-state engine_state.jsonl`, produce
+   `joined.jsonl` keyed on `request_id` with all fields merged.
+2. **Reuse decomposition (real).** Using joined records that carry
+   `session_id` + `hash_ids` + `cached_tokens`, compute
+   `intra_session` / `cross_session` / `shared_prefix` /
+   `unclassified` cached-token mass. Replaces the current
+   `status: unavailable` placeholder when fields are present.
+3. **Interference index.** Per decode step, label "overlap same-
+   worker prefill" using `engine_state.jsonl`. Compute
+   `TPOT_p90(overlap) / TPOT_p90(no_overlap)`.
+4. **Hotspot index.** Per worker queue delay p90, output
+   `max_worker_q_p90 / median_worker_q_p90`.
+5. **Failure label.** For each slow / SLO-violating request, assign
+   one of: `same_worker_prefill_overlap`, `hot_worker_queue`,
+   `high_kv_occupancy`, `cache_miss_large_append`, `transfer_wait`,
+   `p_queue_wait`, `d_admission_wait`, `unknown`.
+6. **Window summary.** For SRR runs, compute attempted/completed/
+   error/goodput plus latency percentiles on the steady-state
+   window only.
+
+Acceptance: re-run analyzer on smoke output and confirm `reuse_decomposition`
+no longer says `unavailable`; `interference_index.json` produced when
+engine state present; `failure_breakdown.json` populated when
+labels assigned.
+
+Effort: 1 large PR. CPU-only.
+
+## Phase B: GPU experiments (needs dash0)
+
+### B1' Workload characterization closure
+
+Inputs: instrumented replayer + small smoke trace (≤500 req).
+
+Steps:
+
+1. Pick `kv_bytes_per_token` for the production model. For
+   Qwen3-Coder TP1 the value depends on layer/head config; compute
+   from `vllm.config` once at run start and record in manifest.
+2. Re-run analyzer on full GLM-5.1 trace with `--kv-bytes-per-token`.
+   Output: KV footprint p50/p90/p99 in `kv_footprint_summary.json`.
+3. Run a 1k-request session-causal smoke replay with instrumented
+   proxy. Use the joined records to populate real reuse decomposition
+   for the small sample. (Full-trace replay is too expensive; sample
+   is acceptable for the decomposition claim.)
+
+Wall-clock: ~30 min GPU. Produces 2 figures: KV footprint CDF, reuse
+decomposition stacked bar.
+
+### B2 PD-colo interference microbench
+
+Setup: 1 combined instance on TP1. Two synthetic load generators:
+
+1. **Decode-only steady load** — short-prompt sessions at fixed
+   per-second arrival, designed to saturate decode without prefill
+   contention.
+2. **Prefill injector** — single-shot long-prompt requests at
+   controlled cadence; same worker (target the decode worker) vs
+   different worker (route to a paired idle instance).
+
+Sweep `uncached_prefill_tokens ∈ {2k, 8k, 16k, 32k, 64k}` × `{same,
+different} worker`.
+
+Outputs: `interference_microbench_summary.json`,
+`decode_step_timeseries.csv` (from `engine_state.jsonl`),
+`prefill_overlap_events.jsonl`, `interference_index.json`,
+TPOT-with-overlay figure, interference-index-vs-prefill-size figure.
+
+Wall-clock: ~2–3 h GPU including warm-up between sweeps.
+
+### B3 Routing sweep on session-causal trace
+
+Setup: 8 combined instances (TP1 × DP8) with the cache-aware proxy.
+
+Run the same session-causal trace (e.g. r=0.0015 st=30 850-req config
+from auto-mem `feedback-bench-config.md`) under five policies:
+
+1. corrected LMetric / cache-aware (`--policy lmetric`)
+2. load-only (new policy `--policy load_only` — picks min running)
+3. hard sticky (new policy `--policy sticky` — once a session lands
+   on a worker, never moves)
+4. current Unified hybrid (`--policy unified`)
+5. session-mass capped replay (filter the trace so no session exceeds
+   `cap_turns` or `cap_input_tokens`; rerun policy 1)
+
+Per run, collect: replayer metrics, proxy breakdown, worker_state,
+engine_state. Compute per-worker queue delay, GPU util, KV occupancy,
+APC, session-to-worker map.
+
+Outputs: `worker_balance_summary.json`, `session_to_worker_map.json`,
+`session_mass_summary.json`, `routing_policy_comparison.json`,
+`hotspot_index.json`, `capped_session_replay_summary.json`,
+8 figures from the TODO list (§5.figures).
+
+Wall-clock: 5 runs × ~13 min ≈ 1.5 h GPU.
+
+Implementation note: `load_only` and `sticky` are small additions to
+`scripts/cache_aware_proxy.py` — they reuse existing affinity / score
+machinery.
+
+### B4 Sustainable Request Rate sweep
+
+Setup: same 8 instances. Use Phase-A `--mode srr` loadgen.
+
+SLO (locked per-class):
+
+```text
+TTFT_p90 <= 2.0 s
+TPOT_p90 <= 0.15 s
+error_rate <= 0.5%
+queue length stable (no monotone growth over steady window)
+KV occupancy stable
+E2E_p90 <= T_class[c]  for each output-length decile c
+```
+
+`T_class[c]` is derived from a low-load reference run as
+`E2E_p90_low_load(c) * 2` (factor configurable). The reference run
+is done once and cached as `analysis/characterization/srr/slo_classes.json`.
+
+Per policy sweep `lambda` from low (clearly safe) to high (clearly
+broken) using a bisection-ish search:
+
+```
+λ_low  = 0.1 sess/s
+λ_high = doubling until first SLO violation
+binary-search λ_low .. λ_high for max sustainable λ
+```
+
+Policies covered: LMetric, static PD-disagg, Unified, hard sticky,
+load-only.
+
+Outputs: `srr_curve.json`, `lambda_runs/<lambda>/summary.json`,
+`slo_violation_reason.json`, `goodput_vs_arrival_rate.json`,
+`stability_summary.json`, all 8 figures from §6.figures.
+
+Wall-clock: this is the most expensive batch. With binary search,
+~6 lambda points × 5 policies × ~8 min (warmup + steady) ≈ 4 h GPU.
+
+### B5 Failure attribution near SRR boundary
+
+For each policy: pick `λ ∈ {0.9, 1.0, 1.1} × SRR`, run with full
+instrumentation, then run the analyzer's failure-label step.
+
+Outputs: `slow_request_attribution.jsonl`, `failure_breakdown.json`,
+`case_studies.md`, `worker_failure_windows.json`, 5 figures from §7.
+
+Wall-clock: 3 lambdas × 5 policies × 8 min ≈ 2 h GPU.
+
+## Phase C: Audit package refresh (CPU)
+
+Re-run `summarize_runs.py` and `plot_current_results.py` after each
+GPU batch. Final pass after B5: refresh `claim_matrix`, `risk_register`,
+`allowed_runs`, regenerate all figures, update
+`reproduction_commands.sh`.
+
+Effort: ~1 h CPU.
+
+## Sequencing & rough timeline
+
+```text
+Phase A (CPU, before dash0):
+  A1 + A2     (parallel)               ~half day CPU
+  A3 patch    (scheduler.py)           ~half day CPU
+  A4 SRR loadgen                       ~half day CPU
+  A5 analyzer extensions               ~1 day   CPU
+
+Window 1 on dash0 (B2 + B3 only, ~5 h GPU):
+  smoke validation of A1–A4            ~30 min GPU
+  B1' KV footprint + reuse decomp      ~30 min GPU
+  B2 interference microbench           ~3 h    GPU
+  B3 routing sweep (5 policies)        ~1.5 h  GPU
+  Phase C partial refresh              ~30 min CPU
+  ── HARD STOP, hand results back ──
+
+Window 2 on dash0 (B4 + B5, ~6 h GPU, only after review):
+  B4 SRR sweep (5 policies × bisect)   ~4 h    GPU
+  B5 failure attribution               ~2 h    GPU
+  Phase C final refresh                ~1 h    CPU
+```
+
+## Decisions (locked 2026-05-25)
+
+1. **Target model**: Qwen3-Coder-30B-A3B. Compute
+   `kv_bytes_per_token` from this model's config at manifest time.
+2. **GPU topology**: TP1 × 8 vLLM instances (DP8). All proxies and
+   sweeps assume 8 worker endpoints.
+3. **Trace for B3/B4**: `traces/w600_r0.0015_st30.jsonl` (~850
+   requests). No resampling.
+4. **E2E SLO**: per-class. Split requests by `requested_output_tokens`
+   decile, set separate E2E thresholds per class. No normalized-E2E
+   headline.
+5. **vLLM scheduler patch**: accepted. Step-level JSONL log goes
+   through a patch under `patches/`. Polling falls back to per-engine
+   `/metrics` for sanity only.
+6. **GPU phasing**: hard stop after B2 and B3. Hand results back for
+   review before committing to B4 SRR sweep or B5 attribution.
+
+## What stays with the interns
+
+- Re-running `summarize_runs.py` after each GPU batch (mechanical).
+- Reviewing the auto-generated `current_results.md` for typos.
+- Maintaining `main_claim_allowed_runs.md` if new traces are added.
+- Anything reading the audit package — not extending it.
+
+## Out of scope for this plan
+
+- New routing policy design (Unified-v2 / PUSH variants).
+- Production-grade KV transfer engineering.
+- Any change to the production paper figures in
+  `analysis/pd_sep_paper_section/`.
+- vLLM upstream contributions.
+
+These are downstream of characterization; once B2/B3/B5 attribution is
+in, we decide separately.

analysis/pd_sep_paper_section/README.md

@@ -30,10 +30,10 @@ analysis/pd_sep_paper_section/
 |---|---|---|
 | C1a: agentic input distribution (p50=33.5k, p90=101k, p99=132k); I/O = 142x | `figures/fig_c1a_io_cdf.pdf` | **rendered** |
 | C1b: 79% intra-session reuse + 0.8% cross-session | `figures/fig_c1b_reuse.pdf` | **rendered** |
-| C2: PD-sep vs Combined headline numbers | (not yet) | **needs re-run without --enforce-eager on `traces/w600_r0.0015_st30.jsonl`** |
-| C3: decode KV cache memory wall (time-series) | (not yet) | needs step-level vLLM telemetry during PD-sep run |
-| C4: TTFT stacked breakdown (prefill / KV pull / decode wait) | (not yet) | needs per-request breakdown.json from PD-sep run |
-| C5: cuda-graph ablation (eager vs cudagraph × Combined vs PD-sep) | (not yet) | needs the 2×2 matrix |
+| C2: PD-sep vs Combined headline (TTFT 69× worse, success 52%) | `figures/fig_c2_pdsep_vs_combined.pdf` | **rendered** (N=3 combined-ca, N=1 each PD-sep config) |
+| C3: KV cache time-series — both PD-sep splits hit the wall | `figures/fig_c3_kv_timeseries.pdf` | **rendered** |
+| C4: TTFT decomposition — 99% is P-side prefill compute | `figures/fig_c4_ttft_stacked.pdf` | **rendered** |
+| C5: cuda-graph ablation (eager vs cudagraph × Combined vs PD-sep) | (not yet) | needs `--with-eager` re-run |
 | C6: prefill stays compute-bound at 95% reuse | `figures/fig_c6_roofline.pdf` | **rendered** |
 | C7: cache-aware routing is a larger lever than PD-sep | `figures/fig_c7_routing_lever.pdf` | **rendered** (legacy data, footer caveat) |
 | KV-WALL: per-D-instance KV demand vs PD layout (system mechanism) | `figures/fig_kv_memory_wall.pdf` | **rendered** (analytical, audit constants in script) |

analysis/pd_sep_paper_section/scripts/bench_pd_matrix.sh

@@ -41,6 +41,7 @@ WITH_RR=false
 WITH_EAGER=false
 DRY_RUN=false
 TAG_PREFIX="pd_matrix"
+ONLY=""             # comma-separated list of tags to run (subset of the matrix)
 
 while [[ $# -gt 0 ]]; do
     case "$1" in
@@ -50,6 +51,7 @@ while [[ $# -gt 0 ]]; do
         --with-rr) WITH_RR=true; shift ;;
         --with-eager) WITH_EAGER=true; shift ;;
         --tag-prefix) TAG_PREFIX="$2"; shift 2 ;;
+        --only) ONLY="$2"; shift 2 ;;
         --dry-run) DRY_RUN=true; shift ;;
         -h|--help)
             sed -n '2,30p' "$0"; exit 0 ;;
@@ -57,6 +59,20 @@ while [[ $# -gt 0 ]]; do
     esac
 done
 
+# Build set of allowed tags from --only (if provided).
+declare -A ALLOWED
+if [ -n "$ONLY" ]; then
+    IFS=',' read -ra _tags <<< "$ONLY"
+    for t in "${_tags[@]}"; do
+        ALLOWED["$(echo "$t" | xargs)"]=1
+    done
+fi
+is_allowed() {
+    # if ONLY not set, everything is allowed
+    [ -z "$ONLY" ] && return 0
+    [ -n "${ALLOWED[$1]:-}" ]
+}
+
 if [ ! -f "$TRACE" ]; then
     echo "[ERROR] trace not found: $TRACE"
     exit 1
@@ -141,13 +157,17 @@ for c in "${CONFIGS[@]}"; do
     for m in "${MODES[@]}"; do
         mode_name="${m%%|*}"; mode_args="${m##*|}"
         for s in $(seq 1 $SEEDS); do
+            tag_name="${cfg_name}_${mode_name}_seed${s}"
+            if ! is_allowed "$tag_name"; then
+                continue
+            fi
             if run_one "$cfg_name" "$cfg_args" "$mode_name" "$mode_args" "$s"; then
                 N_DONE=$((N_DONE + 1))
             else
                 N_FAIL=$((N_FAIL + 1))
             fi
             ELAPSED=$(( $(date +%s) - START_TS ))
-            echo "[progress] $N_DONE done, $N_FAIL failed, $((N_TOTAL - N_DONE - N_FAIL)) remaining, ${ELAPSED}s elapsed"
+            echo "[progress] $N_DONE done, $N_FAIL failed, ${ELAPSED}s elapsed"
         done
     done
 done

analysis/pd_sep_paper_section/scripts/plot_pd_matrix.py

@@ -0,0 +1,490 @@
+"""Render C2/C3/C4/C5 from outputs/pd_matrix/.
+
+Reads each completed run in outputs/pd_matrix/<config>_<mode>_seed<N>/ and
+produces:
+
+  C2: figures/fig_c2_pdsep_vs_combined.pdf
+      Bar chart with mean ± stderr (over seeds) for
+      TTFT p50, TTFT p90, TPOT p90, E2E p50.
+      Bars per config (combined-ca / pdsep-4p4d / pdsep-6p2d),
+      grouped by cuda-graph mode if eager runs present.
+
+  C3: figures/fig_c3_kv_timeseries.pdf
+      Per-instance GPU KV cache usage time-series mined from
+      vllm_inst_*.log "Engine 000: ... GPU KV cache usage: X%" lines.
+      One panel per config; D-instances in PD-sep configs highlighted.
+      Memory-wall threshold (90%) drawn.
+
+  C4: figures/fig_c4_ttft_stacked.pdf
+      Stacked TTFT bar per config showing per-stage time:
+        Combined: just TTFT (single segment, no stage decomposition).
+        PD-sep:   proxy_recv -> prefill_sent (queue on P)
+                  prefill_sent -> prefill_done (prefill compute on P)
+                  prefill_done -> decode_sent (proxy hop)
+                  decode_sent -> first_token (KV pull + decode wait on D)
+
+  C5: figures/fig_c5_cudagraph_ablation.pdf  (only if eager runs exist)
+      Cuda-graph on vs off, per config. Captures the "PD-sep would benefit
+      from D-only graphs" claim quantitatively.
+
+Usage:
+  .venv/bin/python analysis/pd_sep_paper_section/scripts/plot_pd_matrix.py
+"""
+import argparse
+import json
+import re
+import statistics
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import numpy as np
+
+CONFIG_ORDER = ["combined-ca", "combined-rr", "pdsep-4p4d", "pdsep-6p2d"]
+CONFIG_COLOR = {
+    "combined-ca":  "#2ca02c",
+    "combined-rr":  "#7f7f7f",
+    "pdsep-4p4d":   "#ff7f0e",
+    "pdsep-6p2d":   "#d62728",
+}
+
+# Engine log line: timestamps interleaved with usage metrics.
+# Example:
+#   (APIServer pid=...) INFO 05-22 18:29:55 [loggers.py:259] Engine 000:
+#     Avg prompt throughput: ... Running: N reqs, Waiting: M reqs,
+#     GPU KV cache usage: Z%, Prefix cache hit rate: P%
+KV_LOG_RE = re.compile(
+    r"INFO (\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*Engine 000: "
+    r".*Running: (\d+) reqs, Waiting: (\d+) reqs, "
+    r"GPU KV cache usage: ([\d.]+)%"
+)
+
+
+@dataclass
+class Run:
+    tag: str
+    config: str
+    mode: str       # cudagraph | eager
+    seed: int
+    summary: dict
+    breakdown: list = field(default_factory=list)
+    kv_series: dict = field(default_factory=dict)  # inst_idx -> [(t_sec, usage%), ...]
+
+    @property
+    def is_pdsep(self) -> bool:
+        return self.config.startswith("pdsep")
+
+    @property
+    def is_combined(self) -> bool:
+        return self.config.startswith("combined")
+
+
+def parse_tag(name: str) -> tuple[str, str, int] | None:
+    """combined-ca_cudagraph_seed1 -> ("combined-ca", "cudagraph", 1)"""
+    m = re.match(r"(combined-ca|combined-rr|pdsep-4p4d|pdsep-6p2d)_(cudagraph|eager)_seed(\d+)", name)
+    if not m:
+        return None
+    return m.group(1), m.group(2), int(m.group(3))
+
+
+def mine_kv_series(log_path: Path, start_epoch: float | None = None) -> list[tuple[float, float]]:
+    """Return (seconds_since_first_log, kv_usage_percent) pairs."""
+    out: list[tuple[float, float]] = []
+    first_t: float | None = None
+    for line in log_path.read_text(errors="ignore").splitlines():
+        m = KV_LOG_RE.search(line)
+        if not m:
+            continue
+        # "05-22 18:29:55" -> seconds since first log of this file
+        # We can't recover absolute epoch without a year, but relative is enough.
+        ts_str = m.group(1)
+        # parse MM-DD HH:MM:SS into a comparable scalar (mins since 0)
+        try:
+            mm, dd = map(int, ts_str.split(" ")[0].split("-"))
+            hh, mi, ss = map(int, ts_str.split(" ")[1].split(":"))
+        except ValueError:
+            continue
+        t_abs = ((mm - 1) * 31 + (dd - 1)) * 86400 + hh * 3600 + mi * 60 + ss
+        if first_t is None:
+            first_t = t_abs
+        usage = float(m.group(4))
+        out.append((t_abs - first_t, usage))
+    return out
+
+
+def load_runs(matrix_dir: Path) -> list[Run]:
+    runs: list[Run] = []
+    if not matrix_dir.exists():
+        return runs
+    for run_dir in sorted(matrix_dir.iterdir()):
+        if not run_dir.is_dir():
+            continue
+        parsed = parse_tag(run_dir.name)
+        if parsed is None:
+            continue
+        config, mode, seed = parsed
+        summary_p = run_dir / "metrics.summary.json"
+        if not summary_p.exists():
+            continue  # in-flight or failed
+        summary = json.loads(summary_p.read_text())
+        breakdown_p = run_dir / "breakdown.json"
+        breakdown = json.loads(breakdown_p.read_text()) if breakdown_p.exists() else []
+        kv_series: dict = {}
+        for log in sorted(run_dir.glob("vllm_inst_*.log")):
+            m = re.match(r"vllm_inst_(\d+)\.log", log.name)
+            if not m:
+                continue
+            kv_series[int(m.group(1))] = mine_kv_series(log)
+        runs.append(Run(tag=run_dir.name, config=config, mode=mode, seed=seed,
+                        summary=summary, breakdown=breakdown, kv_series=kv_series))
+    return runs
+
+
+# ---------- C2: headline bars with error bars ----------
+
+C2_METRICS = [
+    ("TTFT p50 (s)",  "ttft_stats_s", "p50"),
+    ("TTFT p90 (s)",  "ttft_stats_s", "p90"),
+    ("TPOT p90 (s)",  "tpot_stats_s", "p90"),
+    ("E2E p50 (s)",   "latency_stats_s", "p50"),
+]
+
+
+def aggregate_seeds(runs: list[Run]) -> dict:
+    """Group by (config, mode); for each metric, return mean and stderr across seeds."""
+    grouped: dict[tuple[str, str], list[Run]] = {}
+    for r in runs:
+        grouped.setdefault((r.config, r.mode), []).append(r)
+    out: dict[tuple[str, str], dict] = {}
+    for key, rs in grouped.items():
+        agg: dict = {"n_seeds": len(rs)}
+        for label, family, percentile in C2_METRICS:
+            vals = []
+            for r in rs:
+                v = r.summary.get(family, {}).get(percentile)
+                if v is not None:
+                    vals.append(float(v))
+            if not vals:
+                agg[label] = (float("nan"), 0.0)
+            elif len(vals) == 1:
+                agg[label] = (vals[0], 0.0)
+            else:
+                agg[label] = (statistics.mean(vals), statistics.stdev(vals) / (len(vals) ** 0.5))
+        out[key] = agg
+    return out
+
+
+def plot_c2(runs: list[Run], out_path: Path):
+    agg = aggregate_seeds(runs)
+    if not agg:
+        print("[C2] no runs available; skipped")
+        return
+    modes_present = sorted({k[1] for k in agg})
+    configs_present = [c for c in CONFIG_ORDER if any(k[0] == c for k in agg)]
+
+    fig, axes = plt.subplots(1, len(C2_METRICS), figsize=(3.0 * len(C2_METRICS), 3.6))
+    if len(C2_METRICS) == 1:
+        axes = [axes]
+
+    bar_w = 0.8 / max(1, len(modes_present))
+    for ax, (label, _, _) in zip(axes, C2_METRICS):
+        for mi, mode in enumerate(modes_present):
+            xs, ys, errs, colors = [], [], [], []
+            for ci, cfg in enumerate(configs_present):
+                if (cfg, mode) not in agg:
+                    continue
+                mean, sem = agg[(cfg, mode)][label]
+                xs.append(ci + (mi - (len(modes_present) - 1) / 2) * bar_w)
+                ys.append(mean)
+                errs.append(sem)
+                colors.append(CONFIG_COLOR.get(cfg, "#444"))
+            ax.bar(xs, ys, width=bar_w * 0.9, color=colors, yerr=errs,
+                   capsize=3, edgecolor="black", linewidth=0.5,
+                   label=mode if mi >= 0 else None,
+                   hatch=("" if mode == "cudagraph" else "//"))
+        ax.set_xticks(range(len(configs_present)))
+        labels_with_n = [
+            f"{c}\n(N={agg[(c, modes_present[0])]['n_seeds']})"
+            for c in configs_present
+        ]
+        ax.set_xticklabels(labels_with_n, fontsize=8.5)
+        ax.set_title(label, fontsize=10)
+        ax.grid(True, axis="y", alpha=0.25)
+        ax.set_ylim(bottom=0)
+
+    handles = []
+    for mode in modes_present:
+        handles.append(plt.Rectangle((0, 0), 1, 1, fc="#aaa",
+                                     hatch=("" if mode == "cudagraph" else "//"),
+                                     edgecolor="black"))
+    if len(modes_present) > 1:
+        fig.legend(handles, modes_present, loc="upper right", fontsize=9,
+                   bbox_to_anchor=(0.99, 0.99))
+
+    fig.suptitle(
+        "PD-sep vs Combined headline latency  "
+        "(trace=w600_r0.0015_st30, 850 reqs; per-config N shown under labels)",
+        fontsize=10, y=1.02,
+    )
+    fig.tight_layout()
+    fig.savefig(out_path, bbox_inches="tight")
+    plt.close(fig)
+    print(f"[C2] wrote {out_path}")
+    for key, v in sorted(agg.items()):
+        print(f"  {key[0]:13s} {key[1]:10s} n={v['n_seeds']}  "
+              + "  ".join(f"{lbl.split(' ')[0].lower()}={v[lbl][0]:.3f}" for lbl, _, _ in C2_METRICS))
+
+
+# ---------- C3: KV cache utilization time-series ----------
+
+def plot_c3(runs: list[Run], out_path: Path):
+    # Show seed=1 cudagraph runs only, one panel per config, all instances overlaid.
+    selected = [r for r in runs if r.mode == "cudagraph" and r.seed == 1 and r.kv_series]
+    if not selected:
+        print("[C3] no kv timeseries data; skipped")
+        return
+    selected.sort(key=lambda r: CONFIG_ORDER.index(r.config) if r.config in CONFIG_ORDER else 99)
+
+    fig, axes = plt.subplots(1, len(selected), figsize=(4.2 * len(selected), 3.6),
+                             sharey=True)
+    if len(selected) == 1:
+        axes = [axes]
+
+    for ax, r in zip(axes, selected):
+        n_p, n_d = pd_split(r.config)
+        # First pass: P (or combined) lines under D lines so D is on top.
+        p_label_done = False; d_label_done = False
+        for inst_idx, series in sorted(r.kv_series.items()):
+            if not series:
+                continue
+            xs = [t for t, _ in series]
+            ys = [u for _, u in series]
+            if r.is_combined:
+                color = "#1f77b4"; lw = 0.9; zorder = 2
+                label = "combined GPU" if not p_label_done else None
+                p_label_done = True
+            elif inst_idx < n_p:  # P-instance in pdsep
+                color = "#ff7f0e"; lw = 1.4; zorder = 3
+                label = f"P (inst 0..{n_p-1})" if not p_label_done else None
+                p_label_done = True
+            else:  # D-instance
+                color = "#d62728"; lw = 1.4; zorder = 4
+                label = f"D (inst {n_p}..{n_p+n_d-1})" if not d_label_done else None
+                d_label_done = True
+            ax.plot(xs, ys, color=color, lw=lw, alpha=0.85, zorder=zorder, label=label)
+        ax.axhline(90, color="#888", ls=":", lw=1)
+        ax.text(ax.get_xlim()[1] * 0.98, 92, "memory wall (90%)",
+                fontsize=8, color="#666", ha="right")
+        # peak summary in title
+        peaks = [max(u for _, u in s) if s else 0 for s in r.kv_series.values()]
+        peak_summary = f"peaks {min(peaks):.0f}..{max(peaks):.0f}%"
+        ax.set_title(f"{r.config}\n{peak_summary}", fontsize=10)
+        ax.set_xlabel("time since first engine log (s)")
+        ax.set_ylim(0, 105)
+        ax.grid(True, alpha=0.25)
+        ax.legend(loc="lower right", fontsize=8, framealpha=0.9)
+    axes[0].set_ylabel("GPU KV cache usage (%)")
+    fig.suptitle(
+        "KV cache utilization: PD-sep concentrates pressure on whichever side has fewer GPUs",
+        fontsize=10, y=1.02,
+    )
+    fig.tight_layout()
+    fig.savefig(out_path, bbox_inches="tight")
+    plt.close(fig)
+    print(f"[C3] wrote {out_path}")
+    for r in selected:
+        peaks = [max(u for _, u in s) if s else 0 for s in r.kv_series.values()]
+        print(f"  {r.config:13s}  peak KV per inst: {[f'{p:.0f}%' for p in peaks]}")
+
+
+def pd_split(config: str) -> tuple[int, int]:
+    """Return (N_P, N_D) for the given config name. Combined = (8, 0)."""
+    if config == "pdsep-4p4d":
+        return 4, 4
+    if config == "pdsep-6p2d":
+        return 6, 2
+    return 8, 0  # combined: all instances do both
+
+
+# ---------- C4: TTFT stacked breakdown ----------
+
+def stages_for_record(rec: dict, is_pdsep: bool) -> dict | None:
+    t0 = rec.get("t_proxy_recv")
+    t_ft = rec.get("t_first_token")
+    if t0 is None or t_ft is None:
+        return None
+    if not is_pdsep:
+        return {"TTFT (combined)": t_ft - t0}
+    t_ps = rec.get("t_prefill_sent")
+    t_pd = rec.get("t_prefill_done")
+    t_ds = rec.get("t_decode_sent")
+    if any(x is None for x in (t_ps, t_pd, t_ds)):
+        return None
+    return {
+        "proxy->P queue":    max(0.0, t_ps - t0),
+        "P prefill compute": max(0.0, t_pd - t_ps),
+        "P->D handoff":      max(0.0, t_ds - t_pd),
+        "D wait + first tok": max(0.0, t_ft - t_ds),
+    }
+
+
+def plot_c4(runs: list[Run], out_path: Path):
+    """Stacked TTFT for each (config, seed=1). Combined gets a single-segment bar
+    so the comparison against PD-sep is direct, even though Combined has no
+    stage decomposition."""
+    # Pool all cudagraph seeds that have breakdown data, then compute per-stage p50.
+    by_config: dict[str, list[Run]] = {}
+    for r in runs:
+        if r.mode != "cudagraph" or not r.breakdown:
+            continue
+        by_config.setdefault(r.config, []).append(r)
+    if not by_config:
+        print("[C4] no breakdown data; skipped")
+        return
+
+    bars = []
+    for config in CONFIG_ORDER:
+        if config not in by_config:
+            continue
+        per_stage: dict[str, list[float]] = {}
+        is_pdsep = config.startswith("pdsep")
+        for r in by_config[config]:
+            for rec in r.breakdown:
+                s = stages_for_record(rec, is_pdsep)
+                if not s:
+                    continue
+                for k, v in s.items():
+                    per_stage.setdefault(k, []).append(v)
+        p50 = {k: float(np.median(v)) for k, v in per_stage.items() if v}
+        bars.append((config, p50, len(by_config[config])))
+
+    fig, ax = plt.subplots(figsize=(7.0, 4.2))
+    width = 0.55
+    stage_colors = {
+        "TTFT (combined)":       "#2ca02c",
+        "proxy->P queue":        "#1f77b4",
+        "P prefill compute":     "#9467bd",
+        "P->D handoff":          "#8c564b",
+        "D wait + first tok":    "#d62728",
+    }
+    # consistent stage order
+    stage_order = ["TTFT (combined)", "proxy->P queue", "P prefill compute",
+                   "P->D handoff", "D wait + first tok"]
+    x = list(range(len(bars)))
+    legend_seen: set[str] = set()
+    for i, (config, stages, n_seeds) in enumerate(bars):
+        bottom = 0.0
+        for stage in stage_order:
+            if stage not in stages:
+                continue
+            val = stages[stage]
+            color = stage_colors.get(stage, "#444")
+            label = stage if stage not in legend_seen else None
+            legend_seen.add(stage)
+            ax.bar(i, val, bottom=bottom, width=width, color=color,
+                   edgecolor="white", linewidth=0.5, label=label)
+            if val > 1.0:
+                ax.text(i, bottom + val / 2, f"{val:.2f}s",
+                        ha="center", va="center", fontsize=9, color="white",
+                        fontweight="bold")
+            bottom += val
+        ax.text(i, bottom + 1.5, f"TTFT p50\n{bottom:.2f}s",
+                ha="center", va="bottom", fontsize=9, color="#222",
+                fontweight="bold")
+
+    ax.set_xticks(x)
+    ax.set_xticklabels([f"{b[0]}\n(N={b[2]})" for b in bars],
+                       rotation=0, ha="center", fontsize=9)
+    ax.set_ylabel("TTFT p50 (s, stacked)")
+    ax.set_title(
+        "TTFT decomposition: PD-sep's TTFT is dominated by P-side prefill queueing",
+        fontsize=10,
+    )
+    if legend_seen:
+        ax.legend(loc="upper left", fontsize=8, frameon=False)
+    ax.grid(True, axis="y", alpha=0.25)
+    fig.tight_layout()
+    fig.savefig(out_path, bbox_inches="tight")
+    plt.close(fig)
+    print(f"[C4] wrote {out_path}")
+    for config, stages, n_seeds in bars:
+        tot = sum(stages.values())
+        print(f"  {config:13s} (N={n_seeds}) TTFT p50 = {tot:.3f}s  "
+              + "  ".join(f"{k}={v:.3f}" for k, v in stages.items()))
+
+
+# ---------- C5: cuda-graph ablation 2x2 ----------
+
+def plot_c5(runs: list[Run], out_path: Path):
+    modes = {r.mode for r in runs}
+    if "eager" not in modes:
+        print("[C5] skipped (no --with-eager runs)")
+        return
+    agg = aggregate_seeds(runs)
+    configs_present = [c for c in CONFIG_ORDER if any(k[0] == c for k in agg)]
+
+    metrics = [("TTFT p50 (s)", "ttft_stats_s", "p50"),
+               ("TPOT p90 (s)", "tpot_stats_s", "p90")]
+    fig, axes = plt.subplots(1, len(metrics), figsize=(3.5 * len(metrics), 3.6))
+    if len(metrics) == 1:
+        axes = [axes]
+
+    for ax, (label, _, _) in zip(axes, metrics):
+        xs = np.arange(len(configs_present))
+        w = 0.38
+        for offset, mode in zip([-w / 2, w / 2], ["eager", "cudagraph"]):
+            ys, errs = [], []
+            for cfg in configs_present:
+                k = (cfg, mode)
+                if k in agg:
+                    m, e = agg[k][label]
+                else:
+                    m, e = float("nan"), 0.0
+                ys.append(m); errs.append(e)
+            ax.bar(xs + offset, ys, w, yerr=errs, capsize=3,
+                   label=mode, edgecolor="black", linewidth=0.5)
+        ax.set_xticks(xs)
+        ax.set_xticklabels(configs_present, rotation=20, ha="right", fontsize=8.5)
+        ax.set_title(label, fontsize=10)
+        ax.legend(fontsize=8)
+        ax.grid(True, axis="y", alpha=0.25)
+
+    fig.suptitle("Cuda-graph ablation: PD-sep's D-only graphs are the structural advantage that --enforce-eager suppressed",
+                 fontsize=10, y=1.02)
+    fig.tight_layout()
+    fig.savefig(out_path, bbox_inches="tight")
+    plt.close(fig)
+    print(f"[C5] wrote {out_path}")
+
+
+# ---------- entrypoint ----------
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--matrix-dir", default="outputs/pd_matrix")
+    ap.add_argument("--outdir", default="analysis/pd_sep_paper_section/figures")
+    args = ap.parse_args()
+
+    matrix_dir = Path(args.matrix_dir)
+    outdir = Path(args.outdir)
+    outdir.mkdir(parents=True, exist_ok=True)
+
+    runs = load_runs(matrix_dir)
+    print(f"loaded {len(runs)} completed runs from {matrix_dir}")
+    for r in runs:
+        print(f"  - {r.tag}")
+
+    if not runs:
+        print("no runs yet; nothing to plot.")
+        return
+
+    plot_c2(runs, outdir / "fig_c2_pdsep_vs_combined.pdf")
+    plot_c3(runs, outdir / "fig_c3_kv_timeseries.pdf")
+    plot_c4(runs, outdir / "fig_c4_ttft_stacked.pdf")
+    plot_c5(runs, outdir / "fig_c5_cudagraph_ablation.pdf")
+
+
+if __name__ == "__main__":
+    main()

analysis/pd_sep_paper_section/system_analysis.md

@@ -160,6 +160,97 @@ The empirical KV occupancy on the 6P+2D run was 97 %
 (`analysis/pd_separation_analysis.md` §3.3) — the model and the
 measurement agree to within the resolution of the steady-state assumption.
 
+## Layer 5b: empirical refinement — the bottleneck side depends on the P:D split
+
+The model above predicts D-side saturation. The 6P+2D run in
+`analysis/pd_separation_analysis.md` §3.3 is consistent with it. But the
+new 4P+4D run (`outputs/pd_matrix/pdsep-4p4d_cudagraph_seed1/`, captured
+during this section's experiment matrix) tells a richer story.
+
+Empirical numbers (combined-ca vs both PD-sep splits, same trace, all
+cudagraph, `figures/fig_c2_pdsep_vs_combined.pdf`):
+
+| metric | combined-ca (N=3) | pdsep-4p4d (N=1) | pdsep-6p2d (N=1) |
+|---|---|---|---|
+| success | 99.5 % | **52 %** (444/850) | **68 %** (574/850) |
+| TTFT p50 | 0.91 s | **62.8 s** (69×) | **51.1 s** (56×) |
+| TTFT p90 | 12.7 s | **491 s** (39×) | **400 s** (31×) |
+| TPOT p90 | 0.027 s | 0.013 s (-52 %) | 0.020 s (-26 %) |
+| E2E p50 | 2.5 s | **65.1 s** (26×) | **53.4 s** (21×) |
+| wall clock | 944 s | 7558 s (8×) | 3693 s (3.9×) |
+
+The per-stage TTFT decomposition (`figures/fig_c4_ttft_stacked.pdf`)
+shows that for both PD-sep splits **>97 % of TTFT is P-side prefill
+compute** (65.6 s / 66.2 s in 4p4d; 43.1 s / 44.3 s in 6p2d). D-side
+wait + first token is at most 1.2 s in either config.
+
+The KV-utilization time-series (`figures/fig_c3_kv_timeseries.pdf`)
+tells the full story:
+
+- **combined-ca**: 8 GPUs oscillate 0–98 %, peaks bursty and short
+- **pdsep-4p4d**: P-instances (orange) pinned at 85–100 % the entire
+  2-hour run; D-instances (red) bounce between 10 % and 50 % — only
+  P side hits the wall
+- **pdsep-6p2d**: **both** sides pinned near 100 % the entire run
+  (per-instance peaks 99–100 % across all 8). P-side fills because D
+  back-pressures (D can't free KV slots fast enough → P can't
+  hand off → P-side KV accumulates).
+
+This refines Layer 5: PD separation hits a memory wall on whichever
+side has fewer GPUs, and at extreme splits it co-saturates both sides
+through D-back-pressure.
+
+### Why P-side fills in 4P+4D
+
+Two effects combine on P:
+
+1. **Compute concentration.** Combined spreads prefill across 8 GPUs;
+   4P+4D over 4. Per-P-GPU prefill load is 2× the per-Combined-GPU load.
+   With chunked prefill, multiple in-flight prefills compete for the
+   scheduler.
+
+2. **KV residency on P.** Mooncake does block-by-block transfer *after*
+   the full prefill completes. Until D pulls and acknowledges every
+   block, the completed-but-not-yet-transferred KV sits in P's pool —
+   on top of all the partially-prefilled in-flight KV. Many concurrent
+   33–132 k contexts overwhelm a single 28 GB pool.
+
+This is the same memory-wall mechanism Layer 5 described, but on the
+*prefill* side. The Layer 5 analytical model in
+`figures/fig_kv_memory_wall.pdf` accounted only for *decode-side* KV
+demand. The full model is:
+
+```
+P-side occupancy = (in_flight_prefills × KV_per_req) / (N_P × pool)
+D-side occupancy = (concurrent_decode × KV_per_req) / (N_D × pool)
+```
+
+Whichever side hits the wall first becomes the back-pressure source.
+4P+4D's P-side fills first because 4 GPUs are doing 8 GPUs' worth of
+prefill. 6P+2D's D-side hits the wall (4× concentration), which then
+back-pressures P (no slots to hand off into) until P also fills. In
+either case, *some* side blocks — which is why PD separation regresses
+across the P:D ratios we tested.
+
+### Updated falsifiable condition
+
+The condition for PD separation to *not* regress is now two-sided:
+
+```
+max(
+  in_flight_prefills × KV_per_req / (N_P × pool),
+  concurrent_decode  × KV_per_req / (N_D × pool)
+) < 1
+```
+
+For chatbot workloads (KV/req ≈ 200 MB), this holds easily on either
+side. For agentic with KV/req ≈ 3.3 GB on average and 10–13 GB at the
+tail, both terms cross 1 well below the chosen N_P or N_D.
+
+Followups: re-render `fig_kv_memory_wall.pdf` to show both P and D
+curves once the 6P+2D run lands, with the empirical P-side and D-side
+peaks marked.
+
 ## Layer 6: the DistServe / Splitwise assumption that silently breaks
 
 To formalize: the regime in which PD separation pays is bounded by both a

replayer/metrics.py

@@ -28,6 +28,12 @@ class RequestMetrics:
     requested_output_tokens: int | None = None
     finish_reason: str | None = None
     error: str | None = None
+    t_dispatch_unix: float | None = None
+    t_first_token_unix: float | None = None
+    t_finish_unix: float | None = None
+    proxy_request_id: str | None = None
+    endpoint_url: str | None = None
+    trace_hash_ids: tuple[int, ...] = ()
 
 
 class IncrementalMetricSink:

replayer/replay.py

@@ -125,6 +125,8 @@ async def _dispatch_request(
     }
 
     start = time.perf_counter()
+    t_dispatch_unix = time.time()
+    t_first_token_unix: float | None = None
     ttft_s = None
     n_output = 0
     cached_tokens = 0
@@ -133,7 +135,10 @@ async def _dispatch_request(
     token_times: list[float] = []
     output_token_ids: list[int] = []
 
-    req_headers = {"X-Session-Id": req.session_id}
+    req_headers = {
+        "X-Session-Id": req.session_id,
+        "X-Request-Id": req.request_id,
+    }
 
     async with sem:
         try:
@@ -169,11 +174,13 @@ async def _dispatch_request(
                             if clean_ids:
                                 if ttft_s is None:
                                     ttft_s = now - start
+                                    t_first_token_unix = time.time()
                                 output_token_ids.extend(clean_ids)
                                 token_times.extend([now] * len(clean_ids))
                         elif delta:
                             if ttft_s is None:
                                 ttft_s = now - start
+                                t_first_token_unix = time.time()
                             token_times.append(now)
                         fr = choices[0].get("finish_reason")
                         if fr:
@@ -187,6 +194,7 @@ async def _dispatch_request(
             err = repr(exc)[:300]
 
     end = time.perf_counter()
+    t_finish_unix = time.time()
     e2e = end - start
     if output_token_ids:
         n_output = len(output_token_ids)
@@ -222,6 +230,12 @@ async def _dispatch_request(
             requested_output_tokens=req.output_length,
             finish_reason=finish_reason,
             error=err,
+            t_dispatch_unix=t_dispatch_unix,
+            t_first_token_unix=t_first_token_unix,
+            t_finish_unix=t_finish_unix,
+            proxy_request_id=req.request_id,
+            endpoint_url=endpoint,
+            trace_hash_ids=req.hash_ids,
         ),
         output_token_ids=output_token_ids,
     )

replayer/srr.py

@@ -0,0 +1,320 @@
+"""Open-loop session-causal SRR (Sustainable Request Rate) loadgen.
+
+Differs from `replayer.replay` in three ways:
+
+- Sessions arrive at Poisson rate `lambda` (sessions/s) independent of the
+  trace timestamps. The trace is treated as a *pool* of session templates;
+  each arrival picks one and replays its turns sequentially.
+- A warmup window and a steady-state window are explicit; per-window
+  attempted / completed / error counters are written to
+  `window_summary.json` so latency tails are computable on steady state
+  only.
+- Each arrival gets a fresh `session_id` and per-turn `request_id`,
+  so APC and session affinity are not contaminated by repeated draws.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import random
+import time
+from collections import defaultdict
+from dataclasses import dataclass, replace
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from .metrics import IncrementalMetricSink, RequestMetrics, write_summary_json
+from .replay import (
+    ReplayConfig,
+    _apply_realized_prefix,
+    _build_prompt_token_ids,
+    _dispatch_request,
+)
+from .trace import TraceRequest, load_trace
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class SrrConfig:
+    trace_path: Path
+    output_path: Path
+    endpoint_url: str
+    arrival_rate: float  # sessions per second
+    warmup_s: float = 60.0
+    steady_s: float = 300.0
+    drain_s: float = 60.0
+    concurrency_limit: int = 2000
+    request_timeout_s: float = 600.0
+    model_name: str = "default"
+    session_pool_size: int | None = None  # cap on distinct session templates
+    rng_seed: int = 42
+    request_limit: int | None = None  # cap on raw trace lines loaded
+
+
+def _build_session_pool(
+    requests: list[TraceRequest],
+    pool_cap: int | None,
+) -> list[list[TraceRequest]]:
+    """Group trace requests by session id and keep turn order."""
+    by_session: dict[str, list[TraceRequest]] = defaultdict(list)
+    for r in requests:
+        by_session[r.session_id].append(r)
+    sessions = []
+    for sid, turns in by_session.items():
+        turns.sort(key=lambda r: (r.turn_id, r.timestamp_s))
+        sessions.append(turns)
+    if pool_cap is not None and pool_cap < len(sessions):
+        sessions = sessions[:pool_cap]
+    return sessions
+
+
+def _window_for(t_unix: float, warmup_end: float, steady_end: float) -> str:
+    if t_unix < warmup_end:
+        return "warmup"
+    if t_unix < steady_end:
+        return "steady"
+    return "drain"
+
+
+def _clone_session_for_arrival(
+    template: list[TraceRequest],
+    arrival_idx: int,
+) -> list[TraceRequest]:
+    """Return a copy of the session turns with fresh ids for this arrival.
+
+    The replayer uses request.request_id as X-Request-Id and request.session_id
+    as X-Session-Id. Reusing a template's ids across arrivals would alias on
+    both server side and proxy. Stamp `srr<arrival_idx>` to disambiguate.
+    """
+    new_session_id = f"srr{arrival_idx}_{template[0].session_id}"
+    cloned: list[TraceRequest] = []
+    for t in template:
+        cloned.append(replace(
+            t,
+            session_id=new_session_id,
+            request_id=f"{new_session_id}:{t.turn_id}:{t.chat_id}",
+        ))
+    return cloned
+
+
+async def _run_one_session(
+    *,
+    turns: list[TraceRequest],
+    config: SrrConfig,
+    replay_cfg: ReplayConfig,
+    client: httpx.AsyncClient,
+    request_sem: asyncio.Semaphore,
+    sink: IncrementalMetricSink,
+    counters: dict[str, dict[str, int]],
+    window_for_now,
+    deadline_unix: float,
+) -> None:
+    realized_context: list[int] = []
+    for req in turns:
+        t_dispatch_unix = time.time()
+        if t_dispatch_unix > deadline_unix:
+            return
+        window = window_for_now(t_dispatch_unix)
+        counters["attempted"][window] += 1
+
+        token_ids = _apply_realized_prefix(
+            _build_prompt_token_ids(req),
+            realized_context,
+        )
+        result = await _dispatch_request(
+            client=client,
+            config=replay_cfg,
+            req=req,
+            prompt_token_ids=token_ids,
+            sem=request_sem,
+        )
+        metric = result.metric
+        await sink.append(metric)
+        if metric.error is None:
+            counters["completed"][window] += 1
+            realized_context = token_ids + result.output_token_ids
+        else:
+            counters["errored"][window] += 1
+
+
+async def run_srr(config: SrrConfig) -> dict[str, Any]:
+    requests = load_trace(config.trace_path, request_limit=config.request_limit)
+    pool = _build_session_pool(requests, config.session_pool_size)
+    if not pool:
+        raise ValueError(f"empty session pool from trace {config.trace_path}")
+
+    rng = random.Random(config.rng_seed)
+    sink = IncrementalMetricSink(config.output_path)
+    request_sem = asyncio.Semaphore(config.concurrency_limit)
+
+    replay_cfg = ReplayConfig(
+        trace_path=config.trace_path,
+        output_path=config.output_path,
+        endpoint_url=config.endpoint_url,
+        model_name=config.model_name,
+        concurrency_limit=config.concurrency_limit,
+        request_timeout_s=config.request_timeout_s,
+    )
+
+    run_start_unix = time.time()
+    warmup_end_unix = run_start_unix + config.warmup_s
+    steady_end_unix = warmup_end_unix + config.steady_s
+    drain_end_unix = steady_end_unix + config.drain_s
+
+    def _window(t: float) -> str:
+        return _window_for(t, warmup_end_unix, steady_end_unix)
+
+    counters: dict[str, dict[str, int]] = {
+        "attempted": defaultdict(int),
+        "completed": defaultdict(int),
+        "errored": defaultdict(int),
+    }
+    inter_arrivals: list[float] = []
+    arrival_unix: list[float] = []
+
+    limits = httpx.Limits(
+        max_connections=2000,
+        max_keepalive_connections=500,
+        keepalive_expiry=30.0,
+    )
+
+    logger.info(
+        "SRR start: pool=%d lambda=%.4f sess/s warmup=%.0fs steady=%.0fs drain=%.0fs",
+        len(pool), config.arrival_rate,
+        config.warmup_s, config.steady_s, config.drain_s,
+    )
+
+    arrival_idx = 0
+    session_tasks: list[asyncio.Task] = []
+    try:
+        async with httpx.AsyncClient(
+            timeout=config.request_timeout_s,
+            trust_env=False,
+            limits=limits,
+        ) as client:
+            while True:
+                now = time.time()
+                if now >= steady_end_unix:
+                    break
+                dt = rng.expovariate(config.arrival_rate)
+                inter_arrivals.append(dt)
+                await asyncio.sleep(dt)
+                if time.time() >= steady_end_unix:
+                    break
+                template = rng.choice(pool)
+                cloned = _clone_session_for_arrival(template, arrival_idx)
+                arrival_idx += 1
+                arrival_unix.append(time.time())
+                task = asyncio.create_task(_run_one_session(
+                    turns=cloned,
+                    config=config,
+                    replay_cfg=replay_cfg,
+                    client=client,
+                    request_sem=request_sem,
+                    sink=sink,
+                    counters=counters,
+                    window_for_now=_window,
+                    deadline_unix=drain_end_unix,
+                ))
+                session_tasks.append(task)
+
+            # Drain phase: stop accepting new arrivals, wait for in-flight.
+            drain_timeout = max(0.0, drain_end_unix - time.time())
+            logger.info("SRR drain: waiting up to %.1fs for %d in-flight sessions",
+                        drain_timeout, sum(1 for t in session_tasks if not t.done()))
+            if session_tasks:
+                done, pending = await asyncio.wait(
+                    session_tasks, timeout=drain_timeout,
+                )
+                for t in pending:
+                    t.cancel()
+    finally:
+        sink.close()
+
+    summary = {
+        "run_start_unix": run_start_unix,
+        "warmup_end_unix": warmup_end_unix,
+        "steady_end_unix": steady_end_unix,
+        "drain_end_unix": drain_end_unix,
+        "arrival_rate": config.arrival_rate,
+        "session_pool_size": len(pool),
+        "sessions_arrived": arrival_idx,
+        "inter_arrival_mean_s": (
+            sum(inter_arrivals) / len(inter_arrivals)
+            if inter_arrivals else None
+        ),
+        "attempted": dict(counters["attempted"]),
+        "completed": dict(counters["completed"]),
+        "errored": dict(counters["errored"]),
+        "rng_seed": config.rng_seed,
+    }
+    summary_path = config.output_path.with_name(
+        config.output_path.stem + ".window_summary.json"
+    )
+    summary_path.write_text(json.dumps(summary, indent=2, sort_keys=True))
+    logger.info(
+        "SRR done: arrived=%d attempted=%s completed=%s errored=%s",
+        arrival_idx,
+        dict(counters["attempted"]),
+        dict(counters["completed"]),
+        dict(counters["errored"]),
+    )
+    return summary
+
+
+def _parse_args(argv: list[str] | None = None):
+    import argparse
+
+    p = argparse.ArgumentParser(
+        description="Open-loop Poisson SRR loadgen with session-causal turns",
+    )
+    p.add_argument("--trace", type=Path, required=True)
+    p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--endpoint", type=str, required=True)
+    p.add_argument("--model", type=str, default="default")
+    p.add_argument("--arrival-rate", type=float, required=True,
+                   help="Sessions per second (Poisson)")
+    p.add_argument("--warmup-s", type=float, default=60.0)
+    p.add_argument("--steady-s", type=float, default=300.0)
+    p.add_argument("--drain-s", type=float, default=60.0)
+    p.add_argument("--concurrency-limit", type=int, default=2000)
+    p.add_argument("--request-timeout", type=float, default=600.0)
+    p.add_argument("--session-pool-size", type=int, default=None)
+    p.add_argument("--request-limit", type=int, default=None)
+    p.add_argument("--rng-seed", type=int, default=42)
+    p.add_argument("-v", "--verbose", action="store_true")
+    return p.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> None:
+    args = _parse_args(argv)
+    logging.basicConfig(
+        level=logging.DEBUG if args.verbose else logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    config = SrrConfig(
+        trace_path=args.trace,
+        output_path=args.output,
+        endpoint_url=args.endpoint.rstrip("/"),
+        model_name=args.model,
+        arrival_rate=args.arrival_rate,
+        warmup_s=args.warmup_s,
+        steady_s=args.steady_s,
+        drain_s=args.drain_s,
+        concurrency_limit=args.concurrency_limit,
+        request_timeout_s=args.request_timeout,
+        session_pool_size=args.session_pool_size,
+        request_limit=args.request_limit,
+        rng_seed=args.rng_seed,
+    )
+    summary = asyncio.run(run_srr(config))
+    print(json.dumps(summary, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

scripts/bench.sh

@@ -161,12 +161,24 @@ launch_instances() {
         use_mooncake=true
     fi
 
+    # Optional: when AGENTIC_STEP_LOG_DIR is exported, point each engine at its
+    # own JSONL file so the patched scheduler emits per-step records.
+    local step_log_dir="${AGENTIC_STEP_LOG_DIR:-}"
+    if [ -n "$step_log_dir" ]; then
+        mkdir -p "$step_log_dir"
+    fi
+
     for i in $(seq 0 $((N_INSTANCES - 1))); do
         local port=$((BASE_PORT + i))
         local master=$((29500 + i))
         local logfile="$OUTDIR/vllm_inst_${i}.log"
+        local step_env=""
+        if [ -n "$step_log_dir" ]; then
+            step_env="AGENTIC_STEP_LOG_PATH=$step_log_dir/engine_${i}.jsonl AGENTIC_WORKER_ID=engine_${i}"
+        fi
 
         if [ "$use_mooncake" = "true" ]; then
+            env $step_env \
             PYTHONHASHSEED=42 \
             VLLM_MOONCAKE_BOOTSTRAP_PORT=$((8998 + i)) \
             MASTER_PORT=$master \
@@ -180,6 +192,7 @@ launch_instances() {
                 $vllm_extra_args \
                 > "$logfile" 2>&1 &
         else
+            env $step_env \
             MASTER_PORT=$master \
             CUDA_VISIBLE_DEVICES=$i \
             $VLLM serve "$MODEL" \

scripts/cache_aware_proxy.py

@@ -107,6 +107,42 @@ def _p_offload_penalty(inst: InstanceState) -> int:
     return inst.active_p_offloads * SETTINGS.heavy_threshold
 
 
+def snapshot_workers(
+    instances: list[InstanceState],
+    token_ids: list[int] | None = None,
+    input_length: int = 0,
+) -> list[dict]:
+    """Per-worker state at route-decision time.
+
+    All routing-relevant counters plus the score each policy would
+    have produced for `input_length` if it were dispatched now. Cheap
+    enough to call on every request; B3 hot-spot analysis depends on
+    this being captured per decision.
+    """
+    snap: list[dict] = []
+    for i, inst in enumerate(instances):
+        cache_hit = inst.estimate_cache_hit(token_ids) if token_ids else 0
+        new_prefill = max(0, input_length - cache_hit)
+        snap.append({
+            "idx": i,
+            "url": inst.url,
+            "ongoing_tokens": inst.ongoing_tokens,
+            "ongoing_decode_tokens": inst.ongoing_decode_tokens,
+            "pending_prefill_tokens": inst.pending_prefill_tokens,
+            "num_requests": inst.num_requests,
+            "active_p_offloads": inst.active_p_offloads,
+            "cached_blocks": len(inst.cached_blocks),
+            "cache_hit": cache_hit,
+            "new_prefill": new_prefill,
+            "score_linear": (inst.ongoing_tokens
+                              + _p_offload_penalty(inst)
+                              - CACHE_HIT_ALPHA * cache_hit),
+            "score_lmetric": (inst.pending_prefill_tokens + new_prefill)
+                              * inst.num_requests,
+        })
+    return snap
+
+
 def pick_instance(instances: list[InstanceState], token_ids: list[int] | None,
                   session_id: str | None, input_length: int,
                   affinity: dict[str, int]) -> tuple[InstanceState, int]:
@@ -308,6 +344,7 @@ session_affinity_prefill: dict[str, int] = {}
 session_affinity = session_affinity_combined
 is_pd_sep = False
 _breakdown_log: list[dict] = []
+_worker_state_log: list[dict] = []
 
 
 async def init_prefill_bootstrap(instances: list[InstanceState], ready: asyncio.Event):
@@ -445,7 +482,8 @@ async def _handle(request: Request, api: str):
         raise HTTPException(status_code=503, detail="Service Unavailable")
 
     req_data = await request.json()
-    request_id = str(uuid.uuid4())
+    incoming_rid = request.headers.get("X-Request-Id")
+    request_id = incoming_rid or str(uuid.uuid4())
     prompt = req_data.get("prompt")
     token_ids = prompt if isinstance(prompt, list) else None
     input_length = len(token_ids) if token_ids else 0
@@ -490,6 +528,7 @@ async def _handle_local_request(api, req_data, headers, token_ids, input_length,
                                 chosen.pending_prefill_tokens -= estimated_new
                                 chosen.ongoing_decode_tokens += input_length
                                 breakdown["t_first_token"] = _time.monotonic()
+                                breakdown["t_first_token_unix"] = _time.time()
                                 prefill_done = True
                             yield chunk
                     chosen.record_prefix(
@@ -507,6 +546,7 @@ async def _handle_local_request(api, req_data, headers, token_ids, input_length,
             chosen.ongoing_tokens -= input_length
             chosen.num_requests -= 1
             breakdown["t_done"] = _time.monotonic()
+            breakdown["t_done_unix"] = _time.time()
             _breakdown_log.append(breakdown)
 
     return StreamingResponse(generate(), media_type="text/event-stream")
@@ -527,13 +567,20 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h
     both regressed E2E tail). Re-enabling requires a new transfer mechanism.
     """
     policy = getattr(global_args, 'policy', 'linear')
+    t_decision_unix = _time.time()
+    request_id = headers.get("X-Request-Id", "")
     breakdown: dict = {
-        "request_id": headers.get("X-Request-Id", ""),
+        "request_id": request_id,
+        "session_id": session_id,
         "input_length": input_length,
         "t_proxy_recv": _time.monotonic(),
+        "t_decision_unix": t_decision_unix,
         "policy": policy,
     }
 
+    pre_decision_workers = snapshot_workers(
+        combined_instances, token_ids, input_length)
+
     if policy == "lmetric":
         chosen, best_idx = pick_instance_lmetric(
             combined_instances, token_ids, session_id, input_length,
@@ -550,14 +597,29 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h
             combined_instances, token_ids, session_id, input_length,
             session_affinity_combined)
 
-    cache_hit = chosen.estimate_cache_hit(token_ids)
-    estimated_new = max(0, input_length - cache_hit)
+    chosen_snap = pre_decision_workers[best_idx]
+    cache_hit = chosen_snap["cache_hit"]
+    estimated_new = chosen_snap["new_prefill"]
     breakdown.update({
         "cache_hit": cache_hit,
         "estimated_new_tokens": estimated_new,
         "route_class": "LOCAL",
         "routed_to": chosen.url,
+        "chosen_idx": best_idx,
+        "candidate_scores": pre_decision_workers,
+        "chosen_score_linear": chosen_snap["score_linear"],
+        "chosen_score_lmetric": chosen_snap["score_lmetric"],
     })
+
+    _worker_state_log.append({
+        "t_decision_unix": t_decision_unix,
+        "request_id": request_id,
+        "session_id": session_id,
+        "policy": policy,
+        "chosen_idx": best_idx,
+        "workers": pre_decision_workers,
+    })
+
     return await _handle_local_request(
         api, req_data, headers, token_ids, input_length,
         chosen, estimated_new, breakdown)
@@ -566,17 +628,41 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h
 async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length,
                           session_id, headers):
     """PD-Sep mode with per-stage breakdown profiling."""
+    t_decision_unix = _time.time()
     breakdown = {
         "request_id": request_id,
+        "session_id": session_id,
         "input_length": input_length,
         "t_proxy_recv": _time.monotonic(),
+        "t_decision_unix": t_decision_unix,
+        "policy": "pd_sep",
     }
 
-    p_inst, _ = pick_instance(prefill_instances, token_ids, session_id,
-                               input_length, session_affinity_prefill)
-    d_inst = min(decode_instances, key=lambda x: x.ongoing_tokens)
+    pre_decision_p = snapshot_workers(prefill_instances, token_ids, input_length)
+    pre_decision_d = snapshot_workers(decode_instances, token_ids, input_length)
+
+    p_inst, p_idx = pick_instance(prefill_instances, token_ids, session_id,
+                                   input_length, session_affinity_prefill)
+    d_idx = min(range(len(decode_instances)),
+                 key=lambda i: decode_instances[i].ongoing_tokens)
+    d_inst = decode_instances[d_idx]
     breakdown["p_inst"] = p_inst.url
     breakdown["d_inst"] = d_inst.url
+    breakdown["candidate_scores_prefill"] = pre_decision_p
+    breakdown["candidate_scores_decode"] = pre_decision_d
+    breakdown["chosen_p_idx"] = p_idx
+    breakdown["chosen_d_idx"] = d_idx
+
+    _worker_state_log.append({
+        "t_decision_unix": t_decision_unix,
+        "request_id": request_id,
+        "session_id": session_id,
+        "policy": "pd_sep",
+        "chosen_p_idx": p_idx,
+        "chosen_d_idx": d_idx,
+        "workers_prefill": pre_decision_p,
+        "workers_decode": pre_decision_d,
+    })
 
     prefill_data = req_data.copy()
     prefill_data["kv_transfer_params"] = {
@@ -592,15 +678,18 @@ async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length,
 
     p_inst.ongoing_tokens += input_length
     breakdown["t_prefill_sent"] = _time.monotonic()
+    breakdown["t_prefill_sent_unix"] = _time.time()
 
     try:
         resp = await p_inst.client.post(api, json=prefill_data, headers=p_headers)
         breakdown["t_prefill_done"] = _time.monotonic()
+        breakdown["t_prefill_done_unix"] = _time.time()
         resp.raise_for_status()
         await resp.aclose()
         p_inst.record_prefix(token_ids)
     except Exception as e:
         breakdown["t_prefill_done"] = _time.monotonic()
+        breakdown["t_prefill_done_unix"] = _time.time()
         breakdown["prefill_error"] = True
         _breakdown_log.append(breakdown)
         raise HTTPException(status_code=502, detail=f"Prefill failed: {e}")
@@ -621,6 +710,7 @@ async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length,
     }
 
     breakdown["t_decode_sent"] = _time.monotonic()
+    breakdown["t_decode_sent_unix"] = _time.time()
 
     async def generate():
         first_token = True
@@ -635,11 +725,13 @@ async def _handle_pd_sep(api, req_data, request_id, token_ids, input_length,
                     output_token_ids.extend(new_output_ids)
                     if first_token:
                         breakdown["t_first_token"] = _time.monotonic()
+                        breakdown["t_first_token_unix"] = _time.time()
                         first_token = False
                     yield chunk
             d_inst.record_prefix(_realized_tokens(token_ids, output_token_ids))
         finally:
             breakdown["t_done"] = _time.monotonic()
+            breakdown["t_done_unix"] = _time.time()
             d_inst.ongoing_tokens -= input_length
             _breakdown_log.append(breakdown)
 
@@ -652,6 +744,29 @@ async def get_breakdown():
     return _breakdown_log
 
 
+@app.get("/worker_state")
+async def get_worker_state():
+    """Return per-decision worker-state snapshot log (one entry per route decision)."""
+    return _worker_state_log
+
+
+@app.get("/worker_state/latest")
+async def get_worker_state_latest():
+    """Return current per-worker state snapshot without recording it."""
+    if combined_instances:
+        return {
+            "t_unix": _time.time(),
+            "mode": "combined",
+            "workers": snapshot_workers(combined_instances),
+        }
+    return {
+        "t_unix": _time.time(),
+        "mode": "pd_sep",
+        "workers_prefill": snapshot_workers(prefill_instances),
+        "workers_decode": snapshot_workers(decode_instances),
+    }
+
+
 @app.get("/stats")
 async def get_stats():
     """Return per-instance live state for debugging."""

tests/test_joined_analysis.py

@@ -0,0 +1,170 @@
+"""Tests for A5 joined analysis: join + indices + labels."""
+
+from __future__ import annotations
+
+from analysis.characterization.joined_analysis import (
+    build_joined_records,
+    hotspot_index,
+    interference_index,
+    label_slow_requests,
+    reuse_decomposition,
+    window_summary,
+    _normalize_worker,
+    _percentile,
+)
+
+
+def _mk_metric(rid, **kw):
+    base = {
+        "request_id": rid, "session_id": "s1", "turn_id": 0,
+        "trace_timestamp_s": 1.0, "input_length": 1000, "output_length": 50,
+        "cached_tokens": 0, "actual_output_tokens": 50,
+        "latency_s": 1.0, "ttft_s": 0.5, "tpot_s": 0.04,
+        "t_dispatch_unix": 1000.0, "t_first_token_unix": 1000.5,
+        "t_finish_unix": 1001.0, "endpoint_url": "http://h:8000",
+        "trace_hash_ids": [], "error": None,
+    }
+    base.update(kw)
+    return base
+
+
+def test_build_joined_records_merges_by_request_id():
+    metrics = [_mk_metric("r1"), _mk_metric("r2")]
+    breakdown = [{"request_id": "r1", "policy": "lmetric", "chosen_idx": 3,
+                  "estimated_new_tokens": 500, "routed_to": "http://h:8000"}]
+    worker_state = [{"request_id": "r2", "workers": [{"idx": 0, "url": "x"}]}]
+
+    joined = build_joined_records(metrics, breakdown, worker_state)
+    assert len(joined) == 2
+    j_by_id = {r["request_id"]: r for r in joined}
+    assert j_by_id["r1"]["policy"] == "lmetric"
+    assert j_by_id["r1"]["chosen_idx"] == 3
+    assert j_by_id["r1"]["estimated_new_tokens"] == 500
+    assert j_by_id["r2"]["worker_state_at_decision"][0]["url"] == "x"
+    assert j_by_id["r2"].get("policy") is None  # no breakdown for r2
+
+
+def test_reuse_decomposition_classifies_intra_and_cross():
+    records = [
+        _mk_metric("r1", session_id="A", trace_hash_ids=[11],
+                   cached_tokens=0, t_dispatch_unix=1.0),
+        _mk_metric("r2", session_id="A", trace_hash_ids=[11],
+                   cached_tokens=100, t_dispatch_unix=2.0),
+        _mk_metric("r3", session_id="B", trace_hash_ids=[11],
+                   cached_tokens=100, t_dispatch_unix=3.0),
+    ]
+    out = reuse_decomposition(records)
+    assert out["status"] == "supported"
+    assert out["intra_session_tokens"] > 0
+    assert out["cross_session_tokens"] > 0
+    fr = out["fractions"]
+    assert abs(sum(fr.values()) - 1.0) < 1e-9
+
+
+def test_normalize_worker_maps_port_to_engine_id():
+    assert _normalize_worker("http://node:8000") == "engine_0"
+    assert _normalize_worker("http://node:8005/foo") == "engine_5"
+    assert _normalize_worker("engine_3") == "engine_3"
+    assert _normalize_worker(None) is None
+
+
+def test_interference_index_marks_overlap_when_other_request_prefilling():
+    metrics = [
+        _mk_metric("decode_target", endpoint_url="http://h:8000",
+                   t_first_token_unix=10.0, t_finish_unix=11.0,
+                   tpot_s=0.10),
+        _mk_metric("decode_clean", endpoint_url="http://h:8001",
+                   t_first_token_unix=20.0, t_finish_unix=21.0,
+                   tpot_s=0.04),
+    ]
+    joined = build_joined_records(metrics, [], [])
+    engine_state = {
+        "engine_0": [
+            {"t_unix": 10.5, "prefill_tokens": 8000,
+             "per_req": [{"rid": "other", "phase": "prefill"}]},
+        ],
+        "engine_1": [
+            {"t_unix": 20.5, "prefill_tokens": 0,
+             "per_req": [{"rid": "decode_clean", "phase": "decode"}]},
+        ],
+    }
+    out = interference_index(joined, engine_state)
+    assert out["status"] == "supported"
+    assert out["n_overlap_requests"] == 1
+    assert out["n_clean_requests"] == 1
+    assert out["interference_index"] is not None
+    # Overlap p90 = 0.10; clean p90 = 0.04; ratio > 2
+    assert out["interference_index"] > 2.0
+
+
+def test_hotspot_index_max_over_median_p90():
+    """One hot worker with TTFT 10x the others should drive a >1 index."""
+    rows = []
+    for i in range(3):
+        for _ in range(10):
+            rows.append({
+                "request_id": f"x{i}", "routed_to": f"http://h:800{i}",
+                "endpoint_url": f"http://h:800{i}",
+                "ttft_s": 0.5 if i < 2 else 5.0, "latency_s": 1.0,
+                "error": None,
+            })
+    out = hotspot_index(rows)
+    assert out["status"] == "supported"
+    assert out["hotspot_index_ttft_p90"] > 5.0
+
+
+def test_label_slow_requests_flags_overlap_and_hot_worker():
+    metrics = [
+        _mk_metric("slow_overlap", ttft_s=10.0,
+                   t_first_token_unix=10.0, t_finish_unix=11.0),
+        _mk_metric("slow_no_overlap", ttft_s=10.0,
+                   endpoint_url="http://h:8005",
+                   t_first_token_unix=20.0, t_finish_unix=21.0),
+        _mk_metric("fast", ttft_s=0.5,
+                   t_first_token_unix=15.0, t_finish_unix=16.0),
+    ]
+    metrics[0]["routed_to"] = "http://h:8000"
+    metrics[1]["routed_to"] = "http://h:8005"
+    metrics[2]["routed_to"] = "http://h:8000"
+    bk = [
+        {"request_id": "slow_overlap", "routed_to": "http://h:8000"},
+        {"request_id": "slow_no_overlap", "routed_to": "http://h:8005"},
+        {"request_id": "fast", "routed_to": "http://h:8000"},
+    ]
+    joined = build_joined_records(metrics, bk, [])
+    engine_state = {
+        "engine_0": [{"t_unix": 10.5, "prefill_tokens": 5000,
+                       "per_req": [{"rid": "other", "phase": "prefill"}]}],
+    }
+    labels = label_slow_requests(joined, engine_state, slow_ttft_factor=2.0)
+    by_id = {L["request_id"]: L["label"] for L in labels}
+    assert by_id.get("slow_overlap") == "same_worker_prefill_overlap"
+    assert "fast" not in by_id
+    assert "slow_no_overlap" in by_id
+
+
+def test_window_summary_buckets_by_dispatch_unix():
+    run_meta = {
+        "run_start_unix": 1000.0,
+        "warmup_end_unix": 1010.0,
+        "steady_end_unix": 1030.0,
+        "drain_end_unix": 1040.0,
+    }
+    joined = [
+        _mk_metric("w", t_dispatch_unix=1005.0, ttft_s=0.5, latency_s=1.0,
+                   tpot_s=0.04),
+        _mk_metric("s", t_dispatch_unix=1020.0, ttft_s=0.6, latency_s=1.5,
+                   tpot_s=0.05),
+        _mk_metric("d", t_dispatch_unix=1035.0, ttft_s=0.7, latency_s=2.0,
+                   tpot_s=0.06),
+    ]
+    out = window_summary(joined, run_meta)
+    assert out["windows"]["warmup"]["attempted"] == 1
+    assert out["windows"]["steady"]["attempted"] == 1
+    assert out["windows"]["drain"]["attempted"] == 1
+    assert out["windows"]["steady"]["ttft_p90_s"] is not None
+
+
+def test_percentile_helper_handles_singleton():
+    assert _percentile([5.0], 0.99) == 5.0
+    assert _percentile([], 0.50) is None

tests/test_metrics_instrumentation.py

@@ -0,0 +1,75 @@
+"""Tests for A1 replayer instrumentation: new timestamp + join fields."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict
+
+from replayer.metrics import RequestMetrics
+
+
+def _make_metric(**overrides) -> RequestMetrics:
+    base = dict(
+        request_id="sess1:0:42:0",
+        session_id="sess1",
+        turn_id=0,
+        trace_timestamp_s=1.0,
+        input_length=1000,
+        output_length=50,
+        request_type="user",
+        effective_input_length=1000,
+        cached_tokens=512,
+        latency_s=2.5,
+        ttft_s=0.5,
+        tpot_s=0.04,
+    )
+    base.update(overrides)
+    return RequestMetrics(**base)
+
+
+def test_request_metrics_has_new_join_fields():
+    m = _make_metric()
+    fields = asdict(m)
+    for key in (
+        "t_dispatch_unix",
+        "t_first_token_unix",
+        "t_finish_unix",
+        "proxy_request_id",
+        "endpoint_url",
+        "trace_hash_ids",
+    ):
+        assert key in fields, f"missing {key}"
+
+
+def test_request_metrics_defaults_when_unset():
+    m = _make_metric()
+    assert m.t_dispatch_unix is None
+    assert m.t_first_token_unix is None
+    assert m.t_finish_unix is None
+    assert m.proxy_request_id is None
+    assert m.endpoint_url is None
+    assert m.trace_hash_ids == ()
+
+
+def test_request_metrics_round_trips_through_json():
+    m = _make_metric(
+        t_dispatch_unix=1700000000.123,
+        t_first_token_unix=1700000000.5,
+        t_finish_unix=1700000002.0,
+        proxy_request_id="sess1:0:42:0",
+        endpoint_url="http://node:8000",
+        trace_hash_ids=(11, 22, 33),
+    )
+    payload = json.dumps(asdict(m), sort_keys=True)
+    parsed = json.loads(payload)
+    assert parsed["t_dispatch_unix"] == 1700000000.123
+    assert parsed["t_first_token_unix"] == 1700000000.5
+    assert parsed["t_finish_unix"] == 1700000002.0
+    assert parsed["proxy_request_id"] == "sess1:0:42:0"
+    assert parsed["endpoint_url"] == "http://node:8000"
+    assert parsed["trace_hash_ids"] == [11, 22, 33]
+
+
+def test_proxy_request_id_matches_replayer_request_id():
+    m = _make_metric(proxy_request_id="sess1:0:42:0")
+    assert m.proxy_request_id == m.request_id

tests/test_proxy_instrumentation.py

@@ -0,0 +1,145 @@
+"""Tests for A2 proxy instrumentation: worker snapshot + request_id passthrough."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+PROXY_PATH = Path(__file__).resolve().parent.parent / "scripts" / "cache_aware_proxy.py"
+
+
+def _install_stub_modules() -> None:
+    if "uvicorn" not in sys.modules:
+        sys.modules["uvicorn"] = types.ModuleType("uvicorn")
+
+    if "fastapi" not in sys.modules:
+        fastapi_mod = types.ModuleType("fastapi")
+
+        class _FastAPI:
+            def __init__(self, *a, **kw):
+                self.state = types.SimpleNamespace()
+
+            def post(self, *a, **kw):
+                def deco(fn): return fn
+                return deco
+
+            def get(self, *a, **kw):
+                def deco(fn): return fn
+                return deco
+
+        class _HTTPException(Exception):
+            def __init__(self, status_code=500, detail=""):
+                self.status_code = status_code
+                self.detail = detail
+
+        class _Request:
+            pass
+
+        fastapi_mod.FastAPI = _FastAPI
+        fastapi_mod.HTTPException = _HTTPException
+        fastapi_mod.Request = _Request
+        sys.modules["fastapi"] = fastapi_mod
+
+        responses_mod = types.ModuleType("fastapi.responses")
+
+        class _StreamingResponse:
+            def __init__(self, *a, **kw): pass
+
+        responses_mod.StreamingResponse = _StreamingResponse
+        sys.modules["fastapi.responses"] = responses_mod
+
+    if "httpx" not in sys.modules:
+        httpx_mod = types.ModuleType("httpx")
+
+        class _AsyncClient:
+            def __init__(self, *a, **kw): pass
+            async def aclose(self): pass
+
+        class _Limits:
+            def __init__(self, *a, **kw): pass
+
+        httpx_mod.AsyncClient = _AsyncClient
+        httpx_mod.Limits = _Limits
+        sys.modules["httpx"] = httpx_mod
+
+
+@pytest.fixture(scope="module")
+def proxy():
+    _install_stub_modules()
+    spec = importlib.util.spec_from_file_location("cache_aware_proxy", PROXY_PATH)
+    if spec is None or spec.loader is None:
+        pytest.skip(f"cannot load proxy module at {PROXY_PATH}")
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["cache_aware_proxy"] = mod
+    try:
+        spec.loader.exec_module(mod)
+    except ModuleNotFoundError as exc:
+        pytest.skip(f"proxy dependency missing: {exc}")
+    return mod
+
+
+def _make_inst(proxy, url, **kw):
+    inst = proxy.InstanceState(url)
+    for k, v in kw.items():
+        setattr(inst, k, v)
+    return inst
+
+
+def test_snapshot_workers_includes_all_required_fields(proxy):
+    insts = [
+        _make_inst(proxy, "http://a", ongoing_tokens=100, num_requests=2,
+                   pending_prefill_tokens=500),
+        _make_inst(proxy, "http://b", ongoing_tokens=2000, num_requests=10,
+                   pending_prefill_tokens=8000),
+    ]
+    snap = proxy.snapshot_workers(insts, None, 1000)
+    assert len(snap) == 2
+    required = {
+        "idx", "url", "ongoing_tokens", "ongoing_decode_tokens",
+        "pending_prefill_tokens", "num_requests", "active_p_offloads",
+        "cached_blocks", "cache_hit", "new_prefill",
+        "score_linear", "score_lmetric",
+    }
+    for entry in snap:
+        assert required.issubset(entry.keys()), f"missing fields in {entry}"
+    assert snap[0]["url"] == "http://a"
+    assert snap[1]["url"] == "http://b"
+
+
+def test_snapshot_workers_lmetric_score_reflects_p_tokens_times_bs(proxy):
+    insts = [
+        _make_inst(proxy, "http://a", pending_prefill_tokens=0, num_requests=0),
+        _make_inst(proxy, "http://b", pending_prefill_tokens=4000, num_requests=5),
+    ]
+    snap = proxy.snapshot_workers(insts, None, 1000)
+    assert snap[0]["score_lmetric"] == 0
+    assert snap[1]["score_lmetric"] == (4000 + 1000) * 5
+
+
+def test_snapshot_workers_cache_hit_propagates(proxy):
+    """When token_ids carry a cached prefix, snapshot must record the hit."""
+    insts = [_make_inst(proxy, "http://a"), _make_inst(proxy, "http://b")]
+    block_size = proxy.BLOCK_SIZE
+    prefix = [42] * block_size * 2
+    insts[1].record_prefix(prefix)
+    snap = proxy.snapshot_workers(insts, prefix, len(prefix))
+    assert snap[0]["cache_hit"] == 0
+    assert snap[1]["cache_hit"] == block_size * 2
+    assert snap[0]["new_prefill"] == len(prefix)
+    assert snap[1]["new_prefill"] == 0
+
+
+def test_worker_state_log_is_initially_empty_and_appendable(proxy):
+    """The proxy module exposes a global _worker_state_log list."""
+    assert hasattr(proxy, "_worker_state_log")
+    assert isinstance(proxy._worker_state_log, list)
+    snapshot_count_before = len(proxy._worker_state_log)
+    proxy._worker_state_log.append({"sentinel": True})
+    try:
+        assert len(proxy._worker_state_log) == snapshot_count_before + 1
+    finally:
+        proxy._worker_state_log.pop()

tests/test_scheduler_step_log_patch.py

@@ -0,0 +1,64 @@
+"""Tests for A3 vLLM scheduler patch: per-step JSONL log hooks.
+
+The vendored vLLM at third_party/vllm/vllm/v1/core/sched/scheduler.py must
+contain the agentic-kv hooks. We test by inspecting the file on disk because
+importing vllm.Scheduler requires the full GPU runtime — those imports are
+expensive and not portable to CI. Hooks are anchor strings we control.
+"""
+
+from __future__ import annotations
+
+import py_compile
+from pathlib import Path
+
+SCHEDULER_PATH = (
+    Path(__file__).resolve().parent.parent
+    / "third_party" / "vllm" / "vllm" / "v1" / "core" / "sched" / "scheduler.py"
+)
+
+
+def test_scheduler_file_exists():
+    assert SCHEDULER_PATH.exists(), f"missing {SCHEDULER_PATH}"
+
+
+def test_scheduler_syntactically_valid():
+    py_compile.compile(str(SCHEDULER_PATH), doraise=True)
+
+
+def test_scheduler_has_agentic_step_log_init_hook():
+    src = SCHEDULER_PATH.read_text()
+    assert "AGENTIC_STEP_LOG_PATH" in src
+    assert "AGENTIC_WORKER_ID" in src
+    assert "self._agentic_step_log_fh" in src
+    assert "self._agentic_worker_id" in src
+
+
+def test_scheduler_has_step_emit_helper():
+    src = SCHEDULER_PATH.read_text()
+    assert "def _agentic_emit_step_log(" in src
+    assert "prefill_tokens" in src
+    assert "decode_tokens" in src
+    assert "n_running_total" in src
+    assert "n_waiting" in src
+    assert "per_req" in src
+
+
+def test_scheduler_emit_is_invoked_in_schedule_return_path():
+    """The emit call must sit between _update_after_schedule and return."""
+    src = SCHEDULER_PATH.read_text()
+    assert "self._agentic_emit_step_log(" in src
+    update_idx = src.index("self._update_after_schedule(scheduler_output)")
+    emit_idx = src.index("self._agentic_emit_step_log(")
+    return_idx = src.index("return scheduler_output", update_idx)
+    assert update_idx < emit_idx < return_idx, (
+        "emit hook must be after update_after_schedule and before return"
+    )
+
+
+def test_bench_script_threads_step_log_env():
+    bench = (
+        Path(__file__).resolve().parent.parent / "scripts" / "bench.sh"
+    ).read_text()
+    assert "AGENTIC_STEP_LOG_DIR" in bench
+    assert "AGENTIC_STEP_LOG_PATH" in bench
+    assert "AGENTIC_WORKER_ID" in bench

tests/test_srr_loadgen.py

@@ -0,0 +1,100 @@
+"""Tests for A4 SRR loadgen helpers (no network I/O)."""
+
+from __future__ import annotations
+
+import random
+
+from replayer.srr import (
+    _build_session_pool,
+    _clone_session_for_arrival,
+    _window_for,
+)
+from replayer.trace import TraceRequest
+
+
+def _mk_req(session_id: str, turn: int, chat_id: int, ts: float) -> TraceRequest:
+    return TraceRequest(
+        request_id=f"{session_id}:{turn}:{chat_id}:{turn}",
+        session_id=session_id,
+        chat_id=chat_id,
+        parent_chat_id=-1 if turn == 0 else chat_id - 1,
+        timestamp_s=ts,
+        input_length=100,
+        output_length=10,
+        request_type="user",
+        turn_id=turn,
+        hash_ids=(chat_id,),
+    )
+
+
+def test_build_session_pool_groups_and_orders_turns():
+    reqs = [
+        _mk_req("s1", 1, 11, 2.0),
+        _mk_req("s2", 0, 20, 1.5),
+        _mk_req("s1", 0, 10, 1.0),
+    ]
+    pool = _build_session_pool(reqs, pool_cap=None)
+    assert len(pool) == 2
+    by_sid = {turns[0].session_id: turns for turns in pool}
+    assert [t.turn_id for t in by_sid["s1"]] == [0, 1]
+    assert [t.turn_id for t in by_sid["s2"]] == [0]
+
+
+def test_build_session_pool_honors_pool_cap():
+    reqs = [_mk_req(f"s{i}", 0, i, float(i)) for i in range(5)]
+    pool = _build_session_pool(reqs, pool_cap=2)
+    assert len(pool) == 2
+
+
+def test_window_for_classifies_correctly():
+    warmup_end = 100.0
+    steady_end = 400.0
+    assert _window_for(50.0, warmup_end, steady_end) == "warmup"
+    assert _window_for(100.0, warmup_end, steady_end) == "steady"
+    assert _window_for(399.999, warmup_end, steady_end) == "steady"
+    assert _window_for(400.0, warmup_end, steady_end) == "drain"
+    assert _window_for(500.0, warmup_end, steady_end) == "drain"
+
+
+def test_clone_session_uses_fresh_ids_so_arrivals_do_not_alias():
+    template = [_mk_req("orig", 0, 100, 1.0), _mk_req("orig", 1, 101, 2.0)]
+    clone_a = _clone_session_for_arrival(template, arrival_idx=7)
+    clone_b = _clone_session_for_arrival(template, arrival_idx=8)
+
+    for c in (clone_a, clone_b):
+        assert c[0].session_id == c[1].session_id  # within an arrival
+        assert c[0].session_id.startswith("srr")
+        assert c[0].request_id != template[0].request_id
+
+    assert clone_a[0].session_id != clone_b[0].session_id
+    assert clone_a[0].request_id != clone_b[0].request_id
+
+
+def test_clone_session_preserves_token_payload_fields():
+    template = [_mk_req("orig", 0, 100, 1.0)]
+    template = [TraceRequest(
+        request_id=template[0].request_id,
+        session_id=template[0].session_id,
+        chat_id=template[0].chat_id,
+        parent_chat_id=template[0].parent_chat_id,
+        timestamp_s=template[0].timestamp_s,
+        input_length=4000,
+        output_length=300,
+        request_type="user",
+        turn_id=0,
+        hash_ids=(1, 2, 3, 4, 5),
+    )]
+    cloned = _clone_session_for_arrival(template, arrival_idx=1)
+    assert cloned[0].input_length == 4000
+    assert cloned[0].output_length == 300
+    assert cloned[0].hash_ids == (1, 2, 3, 4, 5)
+    assert cloned[0].turn_id == 0
+
+
+def test_poisson_inter_arrival_mean_matches_rate():
+    """Sanity check on the exponential RNG used for arrivals."""
+    rng = random.Random(0)
+    rate = 5.0  # 5 sess/s -> mean inter-arrival ~ 0.2 s
+    samples = [rng.expovariate(rate) for _ in range(20000)]
+    mean = sum(samples) / len(samples)
+    assert abs(mean - 1.0 / rate) < 0.01

third_party/vllm/vllm/v1/core/sched/scheduler.py

@@ -287,6 +287,29 @@ class Scheduler(SchedulerInterface):
 
         self._pause_state: PauseState = PauseState.UNPAUSED
 
+        # agentic-kv: optional step-level JSONL logger for B2 interference
+        # analysis. Activated by env vars; no-op otherwise.
+        self._agentic_step_log_fh = None
+        self._agentic_step_id = 0
+        self._agentic_worker_id = None
+        import os as _os
+        _step_path = _os.environ.get("AGENTIC_STEP_LOG_PATH")
+        if _step_path:
+            try:
+                _os.makedirs(_os.path.dirname(_step_path) or ".", exist_ok=True)
+                self._agentic_step_log_fh = open(_step_path, "a", buffering=1)
+                self._agentic_worker_id = _os.environ.get(
+                    "AGENTIC_WORKER_ID",
+                    f"dp{self.parallel_config.data_parallel_rank}",
+                )
+                logger.info(
+                    "agentic-kv step log enabled: path=%s worker_id=%s",
+                    _step_path, self._agentic_worker_id,
+                )
+            except Exception as _exc:
+                logger.warning("agentic-kv step log disabled (%r)", _exc)
+                self._agentic_step_log_fh = None
+
     def _mamba_block_aligned_split(
         self,
         request: Request,
@@ -926,8 +949,83 @@ class Scheduler(SchedulerInterface):
 
         with record_function_or_nullcontext("schedule: update_after_schedule"):
             self._update_after_schedule(scheduler_output)
+
+        if self._agentic_step_log_fh is not None:
+            self._agentic_emit_step_log(
+                scheduled_timestamp=scheduled_timestamp,
+                num_scheduled_tokens=num_scheduled_tokens,
+                total_num_scheduled_tokens=total_num_scheduled_tokens,
+                scheduled_new_reqs=scheduled_new_reqs,
+                scheduled_resumed_reqs=scheduled_resumed_reqs,
+                scheduled_running_reqs=scheduled_running_reqs,
+                preempted_reqs=preempted_reqs,
+            )
+
         return scheduler_output
 
+    def _agentic_emit_step_log(
+        self,
+        scheduled_timestamp: float,
+        num_scheduled_tokens: dict[str, int],
+        total_num_scheduled_tokens: int,
+        scheduled_new_reqs: list[Request],
+        scheduled_resumed_reqs: list[Request],
+        scheduled_running_reqs: list[Request],
+        preempted_reqs: list[Request],
+    ) -> None:
+        """Emit one JSONL line per scheduler step for agentic-kv B2 analysis.
+
+        Cheap when enabled (a few dozen dict lookups + one write). When the
+        env var AGENTIC_STEP_LOG_PATH is unset the caller does not invoke
+        this method at all.
+        """
+        import json as _json
+        import time as _time
+        new_ids = {r.request_id for r in scheduled_new_reqs}
+        resumed_ids = {r.request_id for r in scheduled_resumed_reqs}
+        running_ids = {r.request_id for r in scheduled_running_reqs}
+        per_req: list[dict[str, Any]] = []
+        prefill_tokens = 0
+        decode_tokens = 0
+        for rid, n in num_scheduled_tokens.items():
+            is_new = rid in new_ids
+            is_resumed = rid in resumed_ids
+            # Heuristic: any step touching a new/resumed request is prefill;
+            # otherwise per-step tokens >1 is chunked prefill, ==1 is decode.
+            if is_new or is_resumed or n > 1:
+                prefill_tokens += n
+                phase = "prefill"
+            else:
+                decode_tokens += n
+                phase = "decode"
+            per_req.append({
+                "rid": rid, "n": n, "phase": phase,
+                "is_new": is_new, "is_resumed": is_resumed,
+            })
+        record = {
+            "t_unix": _time.time(),
+            "t_monotonic": scheduled_timestamp,
+            "step_id": self._agentic_step_id,
+            "worker_id": self._agentic_worker_id,
+            "total_scheduled_tokens": total_num_scheduled_tokens,
+            "prefill_tokens": prefill_tokens,
+            "decode_tokens": decode_tokens,
+            "n_new": len(scheduled_new_reqs),
+            "n_resumed": len(scheduled_resumed_reqs),
+            "n_running_scheduled": len(scheduled_running_reqs),
+            "n_running_total": len(self.running),
+            "n_waiting": len(self.waiting),
+            "n_preempted": len(preempted_reqs),
+            "preempted_ids": [r.request_id for r in preempted_reqs],
+            "per_req": per_req,
+        }
+        try:
+            self._agentic_step_log_fh.write(_json.dumps(record) + "\n")
+        except Exception as _exc:
+            logger.warning("agentic-kv step log write failed (%r)", _exc)
+            self._agentic_step_log_fh = None
+        self._agentic_step_id += 1
+
     def _preempt_request(self, request: Request, timestamp: float) -> None:
         """Preempt a request and put it back to the waiting queue.