MB5 PD ablation v2 results: concurrency axis + reuse 3-way writeup

- fig3_conc32k.json + fig3_concurrency_axis.png: agentic-corner concurrency
  sweep (in=32768, reuse=0.984, out=128), N 8->128, PD capped 600s / colo
  uncapped. colo completes 100% at every N (graceful, E2E 2.4s->81s); every
  static PD split collapses, earlier as N rises (viable only N<=16; <27% by N=32).
- analysis/mb5_pd_ablation/README.md: self-contained v2 writeup. Reuse axis
  3-way (A=d2048/o256, C=d2048/o128, B=d1024/o128) decomposes shape: output
  ~negligible, delta (real prefill/turn) dominant; crossover to colo at reuse
  ~90-95% robust. Run on dash2 (dash0 NICs faulted for Mooncake).

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

RESULTS_SUMMARY.md

@@ -97,6 +97,18 @@ dash1 GPU 0 单 instance（无 kv_connector），chunked-prefill 默认开启，
 - MB1 + MB2 的合计 cost-benefit 在 phase isolation 维度上 PD-disagg 是赢的，**但这件事被容量天花板压倒**。
 - Paper §3.2 论证应该聚焦"D 池装不下"，MB1/MB2 数据用作 supporting context（per-request transfer charge 量化、phase isolation benefit 量化）而不是 main argument。
 
+> ✅ **2026-05-30 更新 — 干净栈三轴 ablation 证实本节、并加 regime 细化。**
+> 本节的容量论点（D 池容量天花板 / decode 减半）在修复 `e13391e` 污染后的 clean stack
+> 上**得到确认**：concurrency 轴 N=64 时 PD 倾覆，**producer APC 从 71% 崩到 1.4%、TPS −30%**，
+> 而 colo 线性 scale（Fig 3）。**细化**：PD 并非"在 agentic 上一律失败"——它在
+> *低负载 / decode-heavy / 低复用* 区间**赢** colo；真正失败的是 agentic corner（高复用 +
+> 短输出 + 大上下文 + 高并发）——静态 P:D split 无法同时给出复用所需的 producer 容量
+> *和* decode 容量，而 colo 的弹性池两者兼得。
+> **另注**：旧 MB5 文档（`PD_DISAGG_RESULTS.md` §6）"session-affinity 救不了 PD / PD 复用=0%"
+> 的结论是 `e13391e`（producer 每次 KV 传输后 evict prefix）的**污染产物，已撤回**；
+> 干净栈上 session 路由的 producer APC 与 colo 持平（71–82%）。
+> 图：[`figs/mb5_pd_ablation/`](figs/mb5_pd_ablation/)。
+
 ## 5. EAR 设计的实证状态（§4）
 
 | Pillar | 已实证 | 待实证 |

analysis/mb5_pd_ablation/README.md

@@ -0,0 +1,107 @@
+# PD-disagg vs colocation — controlled reuse & concurrency axes (v2)
+
+Self-contained results for the **controlled-variable** redo of the MB5 PD-vs-colo
+ablation. Supersedes the confounded first cut (held input fixed and sliced the
+prefix, so "more reuse" was entangled with "less prefill"). All arms route through
+the proxy at fair **APC parity** (session-routed producers reach the same prefix-cache
+hit rate as colo), so PD loses on *structure*, not on broken cache.
+
+- **Config arms:** `colo` = 8×kv_both (8C-proxy, session-affinity); PD = `6P+2D / 4P+4D / 2P+6D`.
+- **Driver:** closed-loop N (`REPLAY_MAX_INFLIGHT`) + fixed think-time; `gen_synthetic_trace.py --mode regular`.
+- **PD-arm wall-cap:** collapsed PD arms drain pathologically slowly, so PD arms run with a
+  wall-deadline (`REPLAY_MAX_DURATION`; un-run turns counted as failures → honest completion%);
+  **colo is uncapped** so the reference is always fully measured.
+- **Hardware:** run on **dash2** (8×H20). dash0's RDMA NICs were faulted for Mooncake during
+  this work (could not init the transfer engine; needs an admin reset — no sudo); dash2's NICs
+  are healthy. cpfs/venv/data are shared across the boxes.
+
+---
+
+## 1. Reuse / APC axis — fixed real prefill, vary cached prefix
+
+N=8. Hold the **real new-prefill work per turn constant** (`--delta-len`) and grow the cached
+prefix → reuse = prefix/(prefix+delta). Three shapes isolate output vs delta:
+
+| | delta (real prefill/turn) | output | role |
+|---|---|---|---|
+| **A** | 2048 | 256 | original |
+| **C** | 2048 | 128 | A vs C = pure **output** 256→128 |
+| **B** | 1024 | 128 | C vs B = pure **delta** 2048→1024 |
+
+**PD-best advantage** = colo E2E-p90 / best-PD E2E-p90 (>1 ⇒ PD wins):
+
+| reuse% | A d2048/o256 | C d2048/o128 | B d1024/o128 |
+|---|---|---|---|
+| 20 | 1.34 | 1.41 | — |
+| 50 | 1.36 | 1.37 | — |
+| 67 | **1.47** | **1.49** | **1.27** |
+| 80 | 1.31 | 1.23 | 1.25 |
+| 90 | 1.15 | 1.01 | — |
+| 95 | 0.87 | 0.89 | 0.89 |
+
+![reuse 3-way](../../figs/mb5_pd_ablation/reuse_compare_ABC.png)
+
+**Findings:**
+1. **Output length is ~negligible.** A and C (same delta) track each other across the whole
+   range — halving output barely moves PD's advantage.
+2. **Delta (real prefill/turn) is the dominant shape factor.** B (delta=1024) sits clearly
+   below A/C at mid reuse (67%: 1.27 vs ~1.48). More real prefill per turn → bigger PD win,
+   because PD-disagg's benefit is isolating prefill from decode — more prefill to isolate.
+3. **Crossover to colo at reuse ~90–95% is robust** across all three shapes: PD always loses
+   the high-reuse / large-resident-context corner (it must KV-transfer the whole resident
+   context every turn for a few hundred new tokens; colo keeps it local).
+
+*Caveat:* the clean, uncapped, 100%-completion comparison region is reuse **20–80%** (carries
+findings 1–2). At reuse 90/95% the PD arms collapse and C's points are capped-completion, while
+A/B are full-drain — comparable in direction, not in exact PD completion%.
+
+Data: `fig1_reuse_fixed.json` (A), `fig1_reuse_d2048_o128.json` (C), `fig1_reuse_d1024_o128.json` (B).
+
+---
+
+## 2. Concurrency axis — agentic corner, sweep N
+
+in=32768 (prefix 32256 + delta 512, **reuse 0.984**), out=128; closed-loop N ∈ {8,16,32,48,64,96,128};
+PD arms capped 600s, colo uncapped.
+
+| N | **colo** completion · E2E-mean · TPS | best PD-arm completion |
+|---|---|---|
+| 8 | **256/256** · 2.4s · 326 | 6P+2D 256/256 |
+| 16 | **512/512** · 3.5s · 462 | 6P+2D 439/512 (86%) |
+| 32 | **1024/1024** · 13.3s · 190 | all PD **<27%** |
+| 48 | **1536/1536** · 24.9s · 168 | all PD <32% |
+| 64 | **2048/2048** · 38.4s · 166 | all PD <31% |
+| 96 | **3072/3072** · 60.0s · 171 | PD **2–7%** |
+| 128 | **4096/4096** · 80.8s · 181 | 4P+4D 6%, 2P+6D <1% |
+
+![concurrency](../../figs/mb5_pd_ablation/fig3_concurrency_axis.png)
+
+**Finding:** **colo completes 100% of requests at every concurrency level** — it degrades
+*gracefully* (latency rises 2.4s→81s, nothing dropped). **Every static PD split collapses, and
+progressively earlier as N rises**: PD is viable only at N≤8–16; by N≥32 it drops 70–99% of
+requests while its prefix-cache hit-rate craters to ~0%. colo's elastic pool absorbs the
+time-varying P/D demand; the static partition + per-turn 32k KV-transfer cannot. (Latency
+percentiles count successes only, so they *understate* PD — read them with the completion column.)
+
+Data: `fig3_conc32k.json`.
+
+*Caveat:* N=128 6P+2D is missing (one transient vLLM/Mooncake startup flake at the end); does
+not change the picture (all PD arms are already collapsed by N=128). The SLO auto-stop in the
+driver is a no-op (a stdout-capture bug), so the full grid ran — more points, not fewer.
+
+---
+
+## 3. Reproduce
+
+```bash
+# on a box with healthy Mooncake/RDMA NICs (dash2), cpfs mounted:
+R=/home/admin/cpfs/wjh/agentic-kv-fresh
+# reuse axis (three shapes): DELTA/OL pick the shape; tag carries _o${OL}
+ssh dash2 "cd $R && DELTA=2048 OL=256 bash microbench/fresh_setup/run_reuse_fixed.sh"
+ssh dash2 "cd $R && DELTA=2048 OL=128 bash microbench/fresh_setup/run_reuse_fixed.sh"
+ssh dash2 "cd $R && DELTA=1024 OL=128 bash microbench/fresh_setup/run_reuse_fixed.sh"
+# concurrency axis (capped):
+ssh dash2 "cd $R && NLIST='8 16 32 48 64 96 128' CONC_PD_MAXDUR=600 bash microbench/fresh_setup/run_conc.sh"
+# render (reads the *.json in this dir):
+python microbench/fresh_setup/plot_pd_crossover.py
+```

analysis/pd_separation_analysis.md

@@ -23,6 +23,22 @@ Per-request breakdown shows **87.7% of TTFT** is spent waiting for KV cache memo
 > Earlier cross-machine comparison (commit `1e86285`) was invalidated — baseline used warm instances. See REPORT.md §3.5.
 | **Delta** | **-45%** | **-36%** | **-44%** | **+30pp** |
 
+> ✅⚠️ **2026-05-30 — confirmed + refined by the clean MB5 ablation; one caveat.**
+> A producer-side contamination (`e13391e`: evicts a producer's prefix-cache on every
+> KV transfer) was found in the *agentic-kv-fresh* MB5 stack and gated off; everything
+> was re-run clean.
+> - **Confirmed:** this doc's central thesis — PD's failure is a **decode-side KV memory
+>   wall**, not transfer/prefill cost — is reproduced on the clean stack (concurrency
+>   axis: at N=64 the split collapses, APC 71%→1.4%, TPS −30%; colo scales). Fig 3.
+> - **Refined:** "PD separation is net negative" is **regime-dependent**, not universal.
+>   Clean ablation shows PD *wins* at low load / decode-heavy / low-reuse and loses the
+>   **agentic corner** (high reuse + short output + large context + high concurrency).
+> - **Caveat (cross-check):** if this study's runs used the fork vLLM that contains
+>   `e13391e`, any **producer prefix-cache / APC reuse** figures here (e.g. §5.3.1) may be
+>   understated. The decode-memory-wall result is *not* reuse-dependent and is unaffected.
+>   On the clean stack, session-routed producers reach APC parity with colo (71–82%).
+> Figures: [`figs/mb5_pd_ablation/`](../figs/mb5_pd_ablation/).
+
 ---
 
 ## 1. Workload Characterization

analysis/v2/PD_DISAGG_LMETRIC.md

@@ -0,0 +1,165 @@
+# PD-colo vs PD-disagg on the real agentic trace — LMetric (cache-aware) clean-stack anchor
+
+**Figure:** [`figs/v2/fig4_lmetric_pd_vs_colo.png`](../../figs/v2/fig4_lmetric_pd_vs_colo.png)
+**Data:**   [`analysis/v2/fig4l_lmetric.json`](fig4l_lmetric.json)
+**Date:**   2026-05-31 · Hardware: dash1, 8×H20 · Model: Qwen3-Coder-30B-A3B-Instruct
+· vLLM 0.18.1 (V1, chunked-prefill on, `e13391e` eviction gated **off**)
+· Mooncake 0.3.11 · Routing: cache-aware proxy with **`--policy lmetric`**
+· Replayer per-request timeout 600 s.
+
+## TL;DR
+
+On the production agentic trace with the *right* routing baseline (LMetric, cache-aware),
+**PD-colo (8× kv_both) keeps 100 % completion on both traces** and matches the daily-bench
+expectation (~17 min for the high-load first600s, ~50 min for the full trace, with E2E p50
+~3 s and TTFT p50 ~1 s — **3.5–7× better than the original §3 round-robin baseline at the
+same wall-clock**). Every static **PD-disagg ratio fails** (14–65 % completion), and the
+failure mode rotates predictably with the split — **no static partition has a working
+operating point on this workload**. LMetric improves colo dramatically; it does *not*
+rescue PD-disagg, confirming the bottleneck is structural (D-pool admission capacity +
+multi-turn KV accumulation), not routing. A follow-up linear-policy run with PD-disagg
+**wall-capped at 2× the colo wall** (see end of doc) hits the **identical** success-rate
+ceiling — confirming the cap is structural, not policy-driven.
+
+## Setup
+
+- Trace: `w600_r0.0015_st30.jsonl` (1214 reqs, 274 sessions, agentic multi-turn,
+  contexts up to ~112 k tokens; "first600s" variant = same heavy sessions compressed
+  into 600 s → 807 reqs at 3.2× higher arrival rate).
+- 8 instances on 8 GPUs.
+- `--mode baseline` for colo (plain vLLM); `--mode pdsep --pd-ratio P:D` for the three PD
+  splits, all with Mooncake KV transfer.
+- Cache-aware proxy with LMetric scoring (`P_tokens × num_requests`) + session affinity
+  for multi-turn (the colleague's canonical baseline).
+
+## Results
+
+### first600s (1.35 req/s, high-load stress)
+
+| arm | success | E2E mean / p50 / p90 / p99 | TTFT p90 | TPOT p99 | TPS | wall |
+|---|---|---|---|---|---|---|
+| **colo (8C)** | **807/807 = 100 %** | 11.1 / 3.27 / 28.6 / 95.9 s | 14.5 s | 388 ms | 226 | 17.0 min |
+| pd6 (6:2) | 474/807 = **58.7 %** | 83.2 / 6.75 / 382 / 542 s | 380 s | 19 ms | 40 | 55 min |
+| pd4 (4:4) | 348/807 = **43.1 %** | 203 / 215 / 477 / 575 s | 475 s | 25 ms | 15 | 114 min |
+| pd2 (2:6) | 180/807 = **22.3 %** | 380 / 536 / 579 / 602 s | 577 s | 18 ms | 34 | 321 min* |
+
+### Full trace (0.42 req/s, original §3 anchor load)
+
+| arm | success | E2E mean / p50 / p90 / p99 | TTFT p90 | TPOT p99 | TPS | wall |
+|---|---|---|---|---|---|---|
+| **colo (8C)** | **1214/1214 = 100 %** | 10.9 / 3.13 / 29.6 / 93.7 s | 16.9 s | 254 ms | 125 | 49.9 min |
+| pd6 (6:2) | 793/1214 = **65.3 %** | 61.9 / 3.70 / 307 / 477 s | 300 s | 18 ms | 46 | 94 min |
+| pd4 (4:4) | 533/1214 = **43.9 %** | 131 / 8.22 / 468 / 531 s | 467 s | 21 ms | 13 | 231 min |
+| pd2 (2:6) | 169/1214 = **13.9 %** | 195 / 6.82 / 552 / 593 s | 549 s | 13 ms | 1 | 563 min |
+
+\* The pd2 wall-clock is dominated by per-request timeouts (`request_timeout=600 s`)
+draining concurrently behind the multi-turn session causality.
+
+## Five clean findings
+
+1. **LMetric+colo is the right baseline.** Full-trace colo wall **49.9 min ≈ the original
+   §3 RR's 49.9 min**, but E2E p50 **3.13 s vs §3's 10.8 s (3.5×)** and TTFT p50
+   **1.02 s vs §3's 7.0 s (7×)**. Same throughput envelope, far better latency — by virtue
+   of cache-aware routing concentrating each session's turns onto one instance for
+   prefix-cache reuse. The original §3 RR was an *unfairly weak* colo baseline.
+
+2. **Every static PD-disagg ratio fails on the agentic workload.** Completion drops to
+   14–65 %, on *both* traces. The drop is not a high-load artifact (it holds at the
+   original §3 arrival rate of 0.42 req/s); it is structural.
+
+3. **Failure mode rotates predictably with the P:D split:**
+   - **pd2 (2 producers)** → prefill-bound → 78–86 % TTFT timeouts.
+   - **pd6 (2 decode)** → decode-admission-bound → 35–41 % TTFT timeouts.
+   - **pd4 (4P+4D)** → both bottlenecks hit → 57 % TTFT timeouts.
+   - **No static ratio works.** Colo's elastic 8-GPU pool absorbs whichever phase is
+     hot at the moment.
+
+4. **Decode isolation works, but doesn't matter under failure.** TPOT p99 on every PD
+   arm is **13–25 ms** — an order of magnitude better than colo's 254–388 ms — but the
+   win applies only to the 14–65 % of requests that get admitted. The other 35–86 %
+   time out before ever seeing a first token, so the TPOT win is invisible to the end user.
+
+5. **The §3 RR "100 % PD completion" was a measurement artifact.** Original §3
+   (contaminated stack, RR routing) reported 100 % completion for pd6/pd4. LMetric on
+   the clean stack shows 44–65 %. Most plausible mechanism: `e13391e`'s eviction of
+   producer KV on every transfer acted as a **relief valve**, reducing producer-pool
+   pressure and letting more requests squeeze under the 600 s timeout — at the (uncosted)
+   price of cross-turn re-prefill. With the eviction off, producers retain prefix
+   correctly → cache works on PD too → but the cache itself contends for producer
+   pool capacity, and the decode-pool admission ceiling tips earlier. **PD-disagg is
+   worse on agentic than §3 advertised, not better.**
+
+## Linear-policy + wall-cap follow-up (2026-06-01) — the success ceiling is policy-invariant
+
+To check whether the LMetric routing was secretly handicapping PD-disagg, we re-ran
+first600s with the **default `--policy linear`** (cache-aware load score + sticky
+session affinity — the original baseline of the cache_aware_proxy stack) and
+**wall-capped each PD-disagg arm at 2 × the colo wall** (kill bench.sh + cleanup
+GPUs once cap is exceeded, record `records_at_cap`).
+
+| arm | linear success | linear wall | linear @-cap? | lmetric success | lmetric wall |
+|---|---|---|---|---|---|
+| **colo** | 807/807 = **100 %** | 964 s | — | 807/807 = **100 %** | 1021 s |
+| **pd6 (6:2)** | **472/807 = 58 %** | 2280 s | ⊗ cap (706 dispatched) | 474/807 = 59 % | 3325 s |
+| **pd4 (4:4)** | **349/807 = 43 %** | 2281 s | ⊗ cap (577 dispatched) | 348/807 = 43 % | 6850 s |
+| **pd2 (2:6)** | **176/807 = 22 %** | 2280 s | ⊗ cap (521 dispatched) | 180/807 = 22 % | 19275 s |
+
+→ Figure: [`figs/v2/fig4_linear_vs_lmetric.png`](../../figs/v2/fig4_linear_vs_lmetric.png) ·
+data: [`fig4r_linear.json`](fig4r_linear.json)
+
+**Three clean conclusions from the wall-cap experiment:**
+
+1. **The success-rate ceiling is structural, not a routing artifact.** Linear and
+   LMetric — two very different scoring policies (one session-sticky cache-aware,
+   the other non-sticky pure load) — converge on **identical success rates**
+   (58 / 43 / 22 %) for every PD-disagg ratio. Routing has *zero* effect on the
+   completion ceiling. The bottleneck is the static P:D split itself.
+
+2. **LMetric's longer wall was wall *wasted on requests that will never succeed*.**
+   When the cap is enforced, linear hits the same ceiling in 2280 s as LMetric did
+   in 3300–19000 s — the extra wall just slowly times out the unreachable
+   requests at 600 s each.
+
+3. **The wall-cap is the right way to bench PD-disagg.** Reporting "completion %"
+   without a wall budget is misleading (the bench eventually completes if you wait
+   forever, but only by counting timeouts as failures over hours). The honest
+   metric is **success-in-2×-colo-wall**, which gives the same answer for both
+   routings and matches what an end user would observe on a real SLO.
+
+This **strengthens** the §5 D-pool capacity-ceiling thesis: even with
+session-affinity routing serving every request to a warm prefix cache (which
+*should* maximise PD's throughput), the static D-pool can't admit more than
+~22 / 43 / 58 % of the agentic trace within 2× the colo budget. Colo wins not
+because routing is smarter, but because its **elastic pool** absorbs whichever
+phase is hot — there's no cap to hit.
+
+---
+
+## Reproduce
+
+```bash
+# On dash1, from the main repo /home/admin/cpfs/wjh/agentic-kv:
+for TR in w600_r0.0015_st30.jsonl w600_r0.0015_st30_first600s.jsonl; do
+  TRACE=traces/$TR bash scripts/bench.sh --tag fig4l_lmetric_colo_${TR%.*} \
+    --mode baseline --policy lmetric
+  for r in 6:2 4:4 2:6; do
+    TRACE=traces/$TR bash scripts/bench.sh --tag fig4l_lmetric_${r/:/p}_${TR%.*} \
+      --mode pdsep --pd-ratio $r --policy lmetric
+  done
+done
+
+python microbench/fresh_setup/plot_fig4l_lmetric.py
+python microbench/fresh_setup/plot_fig4_linear_vs_lmetric.py
+```
+
+For the linear + 2× wall-cap variant, run colo first to get `wall_clock_s`,
+compute `CAP=2*wall`, then launch each PD-disagg arm in the background and
+`SIGTERM` it (so bench.sh's cleanup trap fires) once `date +%s` minus the
+arm's start time exceeds `CAP`. The capped runs lack `metrics.summary.json`
+(replayer was killed before it could write); compute the summary directly from
+`metrics.jsonl` (see the inline collector used to build
+`analysis/v2/fig4r_linear.json`).
+
+Source `bench.sh` cleans GPUs before each arm and writes `metrics.jsonl` +
+`metrics.summary.json` per tag. Aggregation script: see the inline JSON dump used
+to build `analysis/v2/fig4l_lmetric.json`.

analysis/v2/fig4l_lmetric.json

@@ -0,0 +1 @@
+[{"tag": "fig4l_lmetric_colo_first600s", "arm": "colo", "trace": "first600s", "n": 807, "req": 807, "e2e": {"count": 807.0, "mean": 11.066699584425269, "p50": 3.27055042097345, "p90": 28.745733462180937, "p99": 97.40008939541167}, "ttft": {"count": 807.0, "mean": 5.119651803458883, "p50": 1.2114678020589054, "p90": 14.777630288852365, "p99": 50.68302261995841}, "tpot": {"count": 807.0, "mean": 0.03004899278845205, "p50": 0.009643197803618922, "p90": 0.042092699501536976, "p99": 0.3919741264067197}, "wall": 1020.5351374909515, "tps": 226.12940164644368}, {"tag": "fig4l_lmetric_colo_full", "arm": "colo", "trace": "full", "n": 1214, "req": 1214, "e2e": {"count": 1214.0, "mean": 10.928977524270508, "p50": 3.1279119075043127, "p90": 30.011970606888667, "p99": 94.77313101590481}, "ttft": {"count": 1214.0, "mean": 5.533819193267678, "p50": 1.017395684029907, "p90": 17.36427243486981, "p99": 51.49416554694993}, "tpot": {"count": 1214.0, "mean": 0.02049970290344434, "p50": 0.009544484575988867, "p90": 0.032480608771520716, "p99": 0.26057810739537074}, "wall": 2993.276069591986, "tps": 125.38402448497122}, {"tag": "fig4l_lmetric_pd2_first600s", "arm": "2P+6D", "trace": "first600s", "n": 180, "req": 807, "e2e": {"count": 180.0, "mean": 380.2505690135715, "p50": 535.6594606440049, "p90": 579.5011055286858, "p99": 601.5567972306756}, "ttft": {"count": 180.0, "mean": 378.7133691522933, "p50": 534.4269686369807, "p90": 577.3534130641376, "p99": 596.404559875431}, "tpot": {"count": 180.0, "mean": 0.007975266077679418, "p50": 0.007166497974743372, "p90": 0.012511071875514153, "p99": 0.017508981961061446}, "wall": 19275.367093455978, "tps": 1.8895100582735462}, {"tag": "fig4l_lmetric_pd2_full", "arm": "2P+6D", "trace": "full", "n": 169, "req": 1214, "e2e": {"count": 169.0, "mean": 194.88523891245458, "p50": 6.817620265996084, "p90": 552.1569225640735, "p99": 595.3934216396092}, "ttft": {"count": 169.0, "mean": 193.4153314989016, "p50": 5.60239192598965, "p90": 549.3611521873856, "p99": 582.4436428000824}, "tpot": {"count": 169.0, "mean": 0.007747395842651413, "p50": 0.007691574401794991, "p90": 0.011201243427351017, "p99": 0.013311375577245894}, "wall": 33770.57413210906, "tps": 0.9869539045920406}, {"tag": "fig4l_lmetric_pd4_first600s", "arm": "4P+4D", "trace": "first600s", "n": 348, "req": 807, "e2e": {"count": 348.0, "mean": 202.63302869595395, "p50": 214.03008900902933, "p90": 477.40967412578175, "p99": 576.6393926549597}, "ttft": {"count": 348.0, "mean": 199.96385804087797, "p50": 213.50966987549327, "p90": 475.7766476540827, "p99": 559.6153268160638}, "tpot": {"count": 348.0, "mean": 0.008873619369764751, "p50": 0.007645836479973812, "p90": 0.013845969236959285, "p99": 0.02567216653158788}, "wall": 6850.181333696004, "tps": 15.00296050477674}, {"tag": "fig4l_lmetric_pd4_full", "arm": "4P+4D", "trace": "full", "n": 533, "req": 1214, "e2e": {"count": 533.0, "mean": 130.94711188977982, "p50": 8.219856544979848, "p90": 473.44134307731883, "p99": 533.2597587251009}, "ttft": {"count": 533.0, "mean": 127.83193208824007, "p50": 4.8246813879814, "p90": 467.54664219671395, "p99": 528.8304683346115}, "tpot": {"count": 533.0, "mean": 0.008886429490232585, "p50": 0.007981476340708988, "p90": 0.013570741891233497, "p99": 0.023050950961825044}, "wall": 13884.384965199977, "tps": 12.621372890425038}, {"tag": "fig4l_lmetric_pd6_first600s", "arm": "6P+2D", "trace": "first600s", "n": 474, "req": 807, "e2e": {"count": 474.0, "mean": 83.15809065495806, "p50": 6.7270191764691845, "p90": 391.6558471220078, "p99": 544.7372293809171}, "ttft": {"count": 474.0, "mean": 80.70155321074382, "p50": 4.1273433425230905, "p90": 390.00296151017517, "p99": 539.0574236416071}, "tpot": {"count": 474.0, "mean": 0.008519881756330928, "p50": 0.00803907146806204, "p90": 0.012583933303093976, "p99": 0.018606097790947705}, "wall": 3325.2749515309697, "tps": 39.705588838364164}, {"tag": "fig4l_lmetric_pd6_full", "arm": "6P+2D", "trace": "full", "n": 793, "req": 1214, "e2e": {"count": 793.0, "mean": 61.907526705667, "p50": 3.69814173609484, "p90": 308.2633092067672, "p99": 477.48038318102715}, "ttft": {"count": 793.0, "mean": 59.25069201986225, "p50": 1.402295546955429, "p90": 302.5604081378088, "p99": 475.3738951798529}, "tpot": {"count": 793.0, "mean": 0.009137289999448822, "p50": 0.008635683270933276, "p90": 0.013065757584108427, "p99": 0.01816783740464599}, "wall": 5662.029295974993, "tps": 39.24494000021532}]

analysis/v2/fig4r_linear.json

@@ -0,0 +1 @@
+[{"tag": "fig4r_linear_colo_first600s", "arm": "colo", "trace": "first600s", "policy": "linear", "n": 807, "req": 807, "dispatched": 807, "e2e": {"count": 807.0, "mean": 8.436370009274967, "p50": 2.5224755640374497, "p90": 22.65510415879542, "p99": 75.54369598095519}, "ttft": {"count": 807.0, "mean": 4.2332503390957195, "p50": 0.8872958200518042, "p90": 11.684667797433207, "p99": 44.98891795879462}, "tpot": {"count": 807.0, "mean": 0.020958194728517718, "p50": 0.00851320761584622, "p90": 0.026440129078245465, "p99": 0.30344440533287176}, "wall": 963.6191155100241, "tps": 239.4857016486815, "capped": false}, {"tag": "fig4r_linear_pd2_first600s", "arm": "2P+6D", "trace": "first600s", "policy": "linear", "n": 176, "req": 807, "dispatched": 521, "e2e": {"count": 176, "mean": 378.5561210460834, "p50": 536.7719694490079, "p90": 583.832092280034, "p99": 601.3415494390065}, "ttft": {"count": 176, "mean": 377.12570991374446, "p50": 536.1157373189926, "p90": 580.3465002350276, "p99": 598.0943597999867}, "tpot": {"count": 176, "mean": 0.007864906140929698, "p50": 0.007212154543958604, "p90": 0.011962352272927423, "p99": 0.017870794738764347}, "wall": 2280, "tps": 14.419736842105262, "capped": true}, {"tag": "fig4r_linear_pd4_first600s", "arm": "4P+4D", "trace": "first600s", "policy": "linear", "n": 349, "req": 807, "dispatched": 577, "e2e": {"count": 349, "mean": 264.8537863784421, "p50": 306.6853819829412, "p90": 488.64622142596636, "p99": 596.5830293919425}, "ttft": {"count": 349, "mean": 262.3163347712099, "p50": 299.75751709297765, "p90": 485.475125996978, "p99": 596.4081599479541}, "tpot": {"count": 349, "mean": 0.010442244895290958, "p50": 0.008213572105774598, "p90": 0.019443845545703716, "p99": 0.028178529054794}, "wall": 2281, "tps": 38.306882946076286, "capped": true}, {"tag": "fig4r_linear_pd6_first600s", "arm": "6P+2D", "trace": "first600s", "policy": "linear", "n": 472, "req": 807, "dispatched": 706, "e2e": {"count": 472, "mean": 118.632779156234, "p50": 12.702161715948023, "p90": 458.1609142010566, "p99": 526.5488834320568}, "ttft": {"count": 472, "mean": 115.80202843308507, "p50": 9.745031949947588, "p90": 455.81679951993283, "p99": 516.5850186559837}, "tpot": {"count": 472, "mean": 0.00950947083585719, "p50": 0.008435572332624966, "p90": 0.015233499645638644, "p99": 0.023447183093280886}, "wall": 2280, "tps": 61.69210526315789, "capped": true}]

figs/mb5/CORRECTION.md

@@ -0,0 +1,19 @@
+# ⚠️ Correction notice for figs/mb5/ (2026-05-30)
+
+These figures back `microbench/fresh_setup/PD_DISAGG_RESULTS.md`. A producer-side
+contamination was later found in the stack that produced them: commit **`e13391e`**
+(deployed over the "fresh" pip vLLM by `scripts/deploy_vllm_patches.sh`) evicts a
+producer's prefix-cache blocks on every KV transfer, so a disaggregated producer
+could never keep a session's prefix warm. It is now gated behind
+`VLLM_EVICT_SENT_BLOCKS` (default off) and everything was re-run clean.
+
+| figure | section | status |
+|---|---|---|
+| `mb5_producer_hotspot.png` | §6.3 session-affinity hot-pinning | 🛑 **RETRACTED** — pure `e13391e` artifact. On the clean stack, session-routed producers reach APC parity with colo (71–82%); there is no 0%-util stall / hot-pin pathology. |
+| `mb5_latency_compare.png` | §3 round-robin headline | ✅ stands — RR's ~0% prefix-hit is a *routing* artifact (consecutive turns → different producers), not the eviction bug; reproduced clean. |
+| `mb5_kv_timeline.png`, `mb5_role_split.png`, `mb5_peak_utilization.png` | §5 per-role KV pool occupancy | ✅ D-pool capacity-ceiling mechanism stands (decode pegs while prefill strands). P-pool occupancy may read slightly low under eviction; the qualitative split is unaffected. |
+| `mb5_summary.csv` | aggregate | mixed — §3/§5 rows valid; any session-affinity rows superseded. |
+
+**Superseded by the corrected three-axis ablation:** [`../mb5_pd_ablation/`](../mb5_pd_ablation/)
+(reuse / shape / concurrency), data in [`../../analysis/mb5_pd_ablation/`](../../analysis/mb5_pd_ablation/).
+Raw §6 data `analysis/mb5/session_prod.json` is contaminated; `analysis/mb5/rr_prod.json` (round-robin) stands.

microbench/fresh_setup/PD_DISAGG_RESULTS.md

@@ -10,6 +10,51 @@ Date: 2026-05-28 · Hardware: dash1, 8×GPU · Model: Qwen3-Coder-30B-A3B-Instru
 
 ---
 
+## ⚠️ CORRECTION (2026-05-30) — read before §6
+
+A contamination was found in the "fresh" vLLM used for the runs below.
+`scripts/deploy_vllm_patches.sh` had copied our fork commit **`e13391e`** over the
+pip-installed release; that commit calls `evict_blocks(sent_block_ids)` on
+`finished_sending`, i.e. it **evicts a producer's prefix-cache blocks on every KV
+transfer**. So a disaggregated producer could never keep a session's prefix warm,
+*regardless of routing*. We have since gated that behind `VLLM_EVICT_SENT_BLOCKS`
+(default off) and re-run everything on the corrected stack.
+
+**Retracted (was a pure artifact of `e13391e`):**
+- **All of §6** ("smarter routing does not save PD" / "session-affinity is
+  *strictly worse*" / "GPUs at ~0%" / "producer hot-pinning" / "producer prefix-hit
+  ~0.2%"). On the corrected stack, **session-affinity recovers producer reuse to
+  full parity with colo (APC 71–82%)** — the collapse was the eviction bug starving
+  the very cache session-affinity exists to fill, not a routing pathology.
+- The framing that PD reuse is "0% / fundamentally broken." PD reuses prefix
+  *exactly as well as colo* once routing is session-sticky.
+
+**Still stands (independent of `e13391e`):**
+- **§3 round-robin** numbers — RR sends consecutive turns to *different* producers,
+  so its ~0% prefix-hit is a **routing** artifact (not the eviction bug) and is
+  reproduced on the clean stack; RR PD still loses to 8C.
+- **§4** PD wins TPOT (decode isolation) — robust.
+- **§5.1** the consumer counter-underflow crash — a real, separate vLLM 0.18.1 bug.
+- **§5** the D-pool capacity-ceiling mechanism (decode side pegs while prefill
+  strands) — real.
+
+**Corrected verdict (the real reason PD loses on agentic).** It is *not* "routing
+can't help." On the clean stack PD is **regime-dependent**: it *wins* at low
+load / decode-heavy / low-reuse, and loses the **agentic corner** (high reuse +
+short output + large context + high concurrency) through a structural crossover —
+its static P:D split cannot simultaneously provide the prefix-cache capacity
+(needs many producers) *and* the decode capacity (needs many decoders) that
+agentic demands at once, while colo's elastic pool provides both. See the
+three-axis ablation: **reuse** erodes the edge (1.57×→1.10×), **shape** rotates the
+best ratio and is catastrophic at the prefill extreme, and **concurrency** tips PD
+at N=64 (APC craters 71%→1.4%, TPS −30%) while colo scales cleanly.
+
+→ Figures: [`figs/mb5_pd_ablation/`](../../figs/mb5_pd_ablation/) ·
+data: [`analysis/mb5_pd_ablation/`](../../analysis/mb5_pd_ablation/) ·
+the clean re-run of *this exact* w600 experiment (ratio-swept) is the Fig 4 anchor.
+
+---
+
 ## TL;DR (verdict)
 
 **No static prefill/decode split beats 8-way colocation (8C) on this agentic
@@ -205,6 +250,15 @@ single failed request, which is required to compare routing arms fairly in §6.
 
 ## 6. The routing handicap — and whether smarter routing rescues PD
 
+> 🛑 **RETRACTED (2026-05-30) — this entire section is an artifact of `e13391e`.**
+> The session-affinity runs below were starved by the producer-eviction bug, so
+> they could never collect prefix-cache reuse. On the corrected stack
+> session-affinity reaches **APC parity with colo (71–82%)** and does *not* stall
+> at 0% GPU util. The real mechanism is the capacity/concurrency crossover, not a
+> routing pathology — see the CORRECTION banner at the top and
+> [`figs/mb5_pd_ablation/`](../../figs/mb5_pd_ablation/). Text kept below for the
+> record only.
+
 Every PD config above shows **prefix-cache hit = 0%**, versus 8C's 19%. That
 is not fundamental to disaggregation — it is the stock proxy round-robining
 the **prefill** side: consecutive turns of one agentic session land on

microbench/fresh_setup/backfill_d2048o128.sh

@@ -0,0 +1,35 @@
+#!/usr/bin/env bash
+# Backfill the d2048/o128 reuse arms that vLLM startup-flaked out (transient
+# "Engine core initialization failed", intermittent). Retry up to 4x each with a
+# clean teardown between attempts; HEALTH_MAX_TRIES=180 so a crashed launch fails
+# in ~6min (not 10) before retrying. Then re-aggregate the figure JSON.
+cd /home/admin/cpfs/wjh/agentic-kv-fresh
+export MB5_VENV=$PWD/.venv_dash0
+export HEALTH_MAX_TRIES=180
+VPY=$MB5_VENV/bin/python
+DELTA=2048; OL=128; N=8; THINK=0.5; TURNS=8; NSESS=48
+MISS="${MISS:-4096:6P+2D 18432:6P+2D 38912:8C-proxy 38912:6P+2D}"
+echo "=== BACKFILL START $(date) miss='$MISS' ==="
+for pc in $MISS; do
+  pfx=${pc%%:*}; cfg=${pc##*:}
+  tag="reuse_p${pfx}_d${DELTA}_o${OL}"; trace="traces_synth/${tag}.jsonl"
+  $VPY scripts/gen_synthetic_trace.py --out "$trace" --mode regular --qps "$NSESS" --duration-s 1 \
+     --turns "$TURNS" --prefix-len "$pfx" --delta-len "$DELTA" --output-len "$OL" --seed 42 >/dev/null 2>&1
+  dur=""; [ "$cfg" != "8C-proxy" ] && dur=500
+  ok=0
+  for attempt in 1 2 3 4; do
+    echo "[backfill] $tag $cfg attempt=$attempt $(date +%T)"
+    MB5_P_ROUTING=session MB5_COLO_ROUTING=session \
+      REPLAY_MAX_INFLIGHT=$N REPLAY_INTER_TURN_THINK_S=$THINK REPLAY_NO_REALIZED_PREFIX=1 REPLAY_MAX_DURATION="$dur" \
+      CONFIGS="$cfg" REPS=1 TRACE="$trace" RUN_TAG="$tag" \
+      bash scripts/mb5_run_gpu.sh >/dev/null 2>&1
+    if [ -f "mb5_runs/${tag}_${cfg}_rep1/replay_metrics.summary.json" ]; then
+      echo "  OK $cfg pfx=$pfx attempt=$attempt"; ok=1; break; fi
+    echo "  FAILED attempt=$attempt; cleanup+retry"
+    MB5_VENV=$PWD/.venv_dash0 bash scripts/mb5_launch.sh stop >/dev/null 2>&1; sleep 5
+  done
+  [ $ok = 0 ] && echo "[backfill] GAVE UP $tag $cfg"
+done
+dirs=(); for d in mb5_runs/reuse_*_d2048_o128_*_rep1; do [ -f "$d/replay_metrics.summary.json" ] && dirs+=("$d"); done
+$VPY scripts/fig_agg.py --json "${dirs[@]}" > analysis/mb5_pd_ablation/fig1_reuse_d2048_o128.json
+echo "=== BACKFILL DONE dirs=${#dirs[@]}/24 $(date) ==="

microbench/fresh_setup/fig_agg.py

@@ -0,0 +1,142 @@
+"""Aggregate a set of MB5 run dirs into one comparison table.
+
+Pulls the three core metrics the analysis cares about, per run:
+  - E2E latency  (from replay_metrics.summary.json: latency_stats_s)
+  - TPS          (output tokens / wall_clock_s)
+  - GPU util by workers (gpu_util.csv over run_window, split prefill/decode by role)
+plus honest reuse (producer-side APC from instance_apc.txt) and TTFT/TPOT for logs.
+
+Arm + GPU role split + producer APC ports are inferred from the dir name:
+  *_colo_*  -> 8 kv_both     ; apc ports 8000-8007 (all keep prefix)
+  *_pd6_*   -> 6P+2D P0-5/D6-7 ; apc 8000-8005
+  *_pd_*    -> 4P+4D P0-3/D4-7 ; apc 8000-8003   (note: "pd" not "pd4")
+  *_pd2_*   -> 2P+6D P0-1/D2-7 ; apc 8000-8001
+
+Usage: fig_agg.py <run_dir> [<run_dir> ...]
+"""
+from __future__ import annotations
+
+import csv
+import json
+import re
+import statistics
+import sys
+from pathlib import Path
+
+
+def arm_of(name: str):
+    # New driver naming (run_conc.sh / run_reuse_fixed.sh): "..._<CONFIG>_rep<r>".
+    if "8C-proxy" in name:
+        return "colo", list(range(8)), [], list(range(8000, 8008))
+    if "6P+2D" in name:
+        return "6P+2D", [0, 1, 2, 3, 4, 5], [6, 7], list(range(8000, 8006))
+    if "2P+6D" in name:
+        return "2P+6D", [0, 1], [2, 3, 4, 5, 6, 7], list(range(8000, 8002))
+    if "4P+4D" in name:
+        return "4P+4D", [0, 1, 2, 3], [4, 5, 6, 7], list(range(8000, 8004))
+    # Legacy naming (original May-30 corrected runs).
+    if "_colo_" in name or name.endswith("_colo"):
+        return "colo", list(range(8)), [], list(range(8000, 8008))
+    if "_pd6_" in name:
+        return "6P+2D", [0, 1, 2, 3, 4, 5], [6, 7], list(range(8000, 8006))
+    if "_pd2_" in name:
+        return "2P+6D", [0, 1], [2, 3, 4, 5, 6, 7], list(range(8000, 8002))
+    if "_pd4_" in name or "_pd_" in name:
+        return "4P+4D", [0, 1, 2, 3], [4, 5, 6, 7], list(range(8000, 8004))
+    return "?", list(range(8)), [], list(range(8000, 8008))
+
+
+def util_split(run: Path, pgpus, dgpus):
+    win = {}
+    wp = run / "run_window.json"
+    if wp.exists():
+        win = json.load(open(wp))
+    t0, t1 = win.get("t_start_unix"), win.get("t_end_unix")
+    csvp = run / "gpu_util.csv"
+    if not csvp.exists():
+        return None, None
+    by = {}
+    for row in csv.DictReader(open(csvp)):
+        try:
+            ts = float(row["timestamp"]); g = int(row["gpu"]); u = float(row["util_pct"])
+        except (ValueError, KeyError):
+            continue
+        if t0 and not (t0 <= ts <= t1):
+            continue
+        by.setdefault(g, []).append(u)
+    pm = [v for g in pgpus for v in by.get(g, [])]
+    dm = [v for g in dgpus for v in by.get(g, [])]
+    return (statistics.fmean(pm) if pm else None,
+            statistics.fmean(dm) if dm else None)
+
+
+def apc(run: Path, ports):
+    f = run / "instance_apc.txt"
+    if not f.exists():
+        return None
+    q = h = 0
+    for line in open(f):
+        m = dict(re.findall(r"(\w+)=(\S+)", line))
+        try:
+            p = int(m.get("port", -1))
+        except ValueError:
+            continue
+        if p in ports:
+            q += float(m.get("queries", 0)); h += float(m.get("hits", 0))
+    return (h / q) if q else None
+
+
+def main():
+    args = sys.argv[1:]
+    as_json = False
+    if "--json" in args:
+        as_json = True
+        args = [a for a in args if a != "--json"]
+    rows = []
+    for d in args:
+        run = Path(d)
+        sp = run / "replay_metrics.summary.json"
+        if not sp.exists():
+            continue
+        s = json.load(open(sp))
+        arm, pg, dg, ports = arm_of(run.name)
+        # `or {}` because a fully-collapsed arm (0 successes) writes these as null,
+        # and dict.get(k, {}) returns null (not {}) when the key exists with value null.
+        lat = s.get("latency_stats_s") or {}
+        ttft = s.get("ttft_stats_s") or {}
+        tpot = s.get("tpot_stats_s") or {}
+        wall = s.get("wall_clock_s") or 1.0
+        out = s.get("actual_output_tokens_stats") or {}
+        n = s.get("success_count", 0); req = s.get("request_count", 0)
+        tot_out = out.get("count", 0) * out.get("mean", 0)
+        tps = tot_out / wall
+        pu, du = util_split(run, pg, dg)
+        a = apc(run, ports)
+        rows.append({
+            "name": run.name, "arm": arm, "n": n, "req": req,
+            "e2e_p50": lat.get("p50"), "e2e_p90": lat.get("p90"), "e2e_p99": lat.get("p99"),
+            "e2e_mean": lat.get("mean"),
+            "ttft_p90": ttft.get("p90"), "tpot_p99": tpot.get("p99"),
+            "tps": tps, "wall": wall, "pu": pu, "du": du, "apc": a,
+        })
+
+    if as_json:
+        print(json.dumps(rows))
+        return
+
+    def f(x, w=7, p=1):
+        return f"{x:>{w}.{p}f}" if isinstance(x, (int, float)) else f"{'-':>{w}}"
+
+    hdr = (f"{'run':<34}{'arm':>7}{'ok/req':>9}{'E2Ep50':>8}{'E2Ep90':>8}{'E2Ep99':>8}"
+           f"{'TPS':>8}{'Putil':>7}{'Dutil':>7}{'APC%':>7}{'TTFTp90':>9}{'TPOTp99ms':>10}")
+    print(hdr); print("-" * len(hdr))
+    for r in sorted(rows, key=lambda r: r["name"]):
+        print(f"{r['name']:<34}{r['arm']:>7}{str(r['n'])+'/'+str(r['req']):>9}"
+              f"{f(r['e2e_p50'])}{f(r['e2e_p90'])}{f(r['e2e_p99'])}"
+              f"{f(r['tps'],8,1)}{f(r['pu'])}{f(r['du'])}"
+              f"{f((r['apc'] or 0)*100)}{f(r['ttft_p90'],9,2)}"
+              f"{f((r['tpot_p99'] or 0)*1000,10,1)}")
+
+
+if __name__ == "__main__":
+    main()

microbench/fresh_setup/gpu_monitor.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+# Sample GPU utilization every 5s, output CSV
+# Usage: bash gpu_monitor.sh <output_file> [interval_s]
+#   Runs until killed (Ctrl+C or kill)
+
+OUT="${1:-/tmp/gpu_util.csv}"
+INTERVAL="${2:-5}"
+
+echo "timestamp,gpu,util_pct,mem_used_mb,mem_total_mb,power_w" > "$OUT"
+
+while true; do
+    TS=$(date +%s.%N)
+    nvidia-smi --query-gpu=index,utilization.gpu,memory.used,memory.total,power.draw \
+        --format=csv,noheader,nounits 2>/dev/null | while IFS=', ' read -r idx util mem_used mem_total power; do
+        echo "$TS,$idx,$util,$mem_used,$mem_total,$power"
+    done >> "$OUT"
+    sleep "$INTERVAL"
+done

microbench/fresh_setup/gpu_util_report.py

@@ -0,0 +1,71 @@
+"""Per-GPU utilization report from gpu_util.csv (companion to bench_report.py).
+
+bench_report's per-worker GPU util needs request routing (breakdown.json), which
+the MB5 proxy doesn't log. But worker == GPU by index, and the prefill/decode role
+split is fixed by config, so per-GPU util from gpu_util.csv directly answers
+"GPU utils by workers" — and for PD it exposes the key signal: are the prefill-side
+GPUs saturated while the decode-side idles (or vice versa, or stalled at ~0)?
+
+Usage:
+  gpu_util_report.py <run_dir> [--prefill-gpus 0,1,2,3 --decode-gpus 4,5,6,7]
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import statistics
+from pathlib import Path
+
+
+def pct(xs, p):
+    xs = sorted(xs)
+    return xs[max(0, min(len(xs) - 1, int(round(p / 100 * (len(xs) - 1)))))] if xs else None
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("run_dir", type=Path)
+    ap.add_argument("--prefill-gpus", default="")
+    ap.add_argument("--decode-gpus", default="")
+    a = ap.parse_args()
+
+    win = {}
+    wp = a.run_dir / "run_window.json"
+    if wp.exists():
+        win = json.load(open(wp))
+    t0, t1 = win.get("t_start_unix"), win.get("t_end_unix")
+
+    csvp = a.run_dir / "gpu_util.csv"
+    if not csvp.exists():
+        print(f"{a.run_dir.name}: gpu_util.csv absent"); return
+    by_gpu = {}
+    for row in csv.DictReader(open(csvp)):
+        try:
+            ts = float(row["timestamp"]); g = int(row["gpu"]); u = float(row["util_pct"]); m = float(row["mem_used_mb"])
+        except (ValueError, KeyError):
+            continue
+        if t0 and not (t0 <= ts <= t1):
+            continue
+        by_gpu.setdefault(g, {"u": [], "m": []})
+        by_gpu[g]["u"].append(u); by_gpu[g]["m"].append(m)
+
+    print(f"=== {a.run_dir.name}: per-GPU util over replay window ({sum(len(d['u']) for d in by_gpu.values())} samples) ===")
+    print(f"{'gpu':>4}{'util_mean':>11}{'util_p90':>10}{'util_max':>10}{'mem_max_GB':>12}")
+    for g in sorted(by_gpu):
+        u, m = by_gpu[g]["u"], by_gpu[g]["m"]
+        print(f"{g:>4}{statistics.fmean(u):>11.1f}{pct(u,90):>10.1f}{max(u):>10.1f}{max(m)/1024:>12.1f}")
+
+    def agg(gpus, label):
+        gpus = [int(x) for x in gpus.split(",") if x != ""]
+        us = [v for g in gpus for v in by_gpu.get(g, {}).get("u", [])]
+        if us:
+            print(f"  {label:<14} gpus={gpus} util mean={statistics.fmean(us):.1f}% p90={pct(us,90):.1f}% max={max(us):.1f}%")
+    if a.prefill_gpus:
+        agg(a.prefill_gpus, "prefill-side")
+    if a.decode_gpus:
+        agg(a.decode_gpus, "decode-side")
+
+
+if __name__ == "__main__":
+    main()

microbench/fresh_setup/mb5_run.sh

@@ -69,6 +69,13 @@ run_one() {
     source "${VENV}/bin/activate"
     local replay_out="${rundir}/replay_metrics.jsonl"
     mkdir -p "$(dirname "${replay_out}")"
+    # bench_report.py inputs: worker->gpu map (worker i == gpu i for every config;
+    # for PD, workers 0-3 are producers on gpu0-3, 4-7 consumers on gpu4-7).
+    printf '{"base_port":8000,"n_instances":8,"gpu_indices":[0,1,2,3,4,5,6,7]}\n' \
+        > "${rundir}/bench_config.json"
+    # per-GPU utilization timeseries over the replay window (2s sampling)
+    bash "${SCRIPT_DIR}/gpu_monitor.sh" "${rundir}/gpu_util.csv" 2 >/dev/null 2>&1 &
+    local GPU_MON=$!
     local t0
     t0=$(date +%s.%N)
     if ! PYTHONPATH="${FRESH_ROOT}" python -m replayer \
@@ -82,6 +89,7 @@ run_one() {
         t1=$(date +%s.%N)
         local wall=$(python -c "print(${t1} - ${t0})")
         echo "[mb5-run] REPLAY FAILED after ${wall} s; see ${OUT_ROOT}/${config}_rep${rep}_replay.log"
+        kill "${GPU_MON}" 2>/dev/null || true
         bash "${LAUNCH}" stop > /dev/null 2>&1 || true
         return 1
     fi
@@ -91,6 +99,9 @@ run_one() {
     wall_clock_s=$(python -c "print(${t1} - ${t0})")
     echo "[mb5-run] replay done in ${wall_clock_s}s"
     echo "${wall_clock_s}" > "${rundir}/wall_clock_s.txt"
+    kill "${GPU_MON}" 2>/dev/null || true
+    printf '{"t_start_unix":%s,"t_end_unix":%s}\n' "${t0}" "${t1}" > "${rundir}/run_window.json"
+    cp -f "${replay_out}" "${rundir}/metrics.jsonl"   # bench_report.py expects metrics.jsonl
 
     # Per-instance prefix-cache counters, scraped from each backend BEFORE
     # teardown. For PD this is the only honest reuse signal: producer ports

microbench/fresh_setup/mb5_run_gpu.sh

@@ -0,0 +1,144 @@
+#!/usr/bin/env bash
+# Orchestrator for MB5: for each CONFIG × rep, bring up the stack, run a
+# trace replay against it, collect KV snapshots and replayer metrics,
+# tear down.
+#
+# Designed to be run on dash1 (or any host with cpfs mounted at
+# /home/admin/cpfs/wjh/).
+#
+# Env vars (with defaults):
+#   CONFIGS    : space-separated MB5 configs (default: "8C 6P+2D 4P+4D 2P+6D")
+#   REPS       : reps per config (default: 3)
+#   TRACE      : trace JSONL path
+#                (default: /home/admin/cpfs/wjh/agentic-kv/traces/w600_r0.0015_st30.jsonl)
+#   RUN_TAG    : output root tag (default: $(date +%Y%m%d_%H%M%S))
+#   REQUEST_LIMIT : optional, cap replay requests (default: none)
+
+set -eo pipefail
+
+FRESH_ROOT="/home/admin/cpfs/wjh/agentic-kv-fresh"
+# MB5_VENV lets a second host use an isolated venv clone (see mb5_launch.sh).
+VENV="${MB5_VENV:-${FRESH_ROOT}/.venv}"
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+LAUNCH="${SCRIPT_DIR}/mb5_launch.sh"
+REPLAYER_DIR="${FRESH_ROOT}/replayer"
+
+CONFIGS="${CONFIGS:-8C 6P+2D 4P+4D 2P+6D}"
+REPS="${REPS:-3}"
+TRACE="${TRACE:-/home/admin/cpfs/wjh/agentic-kv/traces/w600_r0.0015_st30.jsonl}"
+RUN_TAG="${RUN_TAG:-$(date +%Y%m%d_%H%M%S)}"
+MODEL_NAME="${MODEL_NAME:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-Coder-30B-A3B-Instruct}"
+REQUEST_LIMIT_ARG=""
+if [ -n "${REQUEST_LIMIT:-}" ]; then
+    REQUEST_LIMIT_ARG="--request-limit ${REQUEST_LIMIT}"
+fi
+
+OUT_ROOT="${FRESH_ROOT}/mb5_runs/${RUN_TAG}"
+mkdir -p "${OUT_ROOT}"
+echo "[mb5-run] RUN_TAG=${RUN_TAG}"
+echo "[mb5-run] OUT_ROOT=${OUT_ROOT}"
+echo "[mb5-run] CONFIGS=${CONFIGS}"
+echo "[mb5-run] REPS=${REPS}"
+echo "[mb5-run] TRACE=${TRACE}"
+
+run_one() {
+    local config="$1" rep="$2"
+    local label="${RUN_TAG}_${config}_rep${rep}"
+    local rundir="${FRESH_ROOT}/mb5_runs/${label}"
+    echo ""
+    echo "======== ${config} rep${rep} ========"
+
+    # Launch
+    if ! CONFIG="${config}" RUN_LABEL="${RUN_TAG}_${config}_rep${rep}" \
+            bash "${LAUNCH}" start > "${OUT_ROOT}/${config}_rep${rep}_launch.log" 2>&1; then
+        echo "[mb5-run] LAUNCH FAILED for ${config} rep${rep}; see ${OUT_ROOT}/${config}_rep${rep}_launch.log"
+        return 1
+    fi
+
+    # Extract ENDPOINTS line emitted by mb5_launch.sh
+    local endpoints
+    endpoints=$(grep "^ENDPOINTS=" "${OUT_ROOT}/${config}_rep${rep}_launch.log" | tail -1 | cut -d= -f2-)
+    if [ -z "${endpoints}" ]; then
+        echo "[mb5-run] ERROR: no ENDPOINTS in launch log"
+        bash "${LAUNCH}" stop > /dev/null 2>&1 || true
+        return 1
+    fi
+    echo "[mb5-run] endpoints: ${endpoints}"
+
+    # Replay
+    source "${VENV}/bin/activate"
+    local replay_out="${rundir}/replay_metrics.jsonl"
+    mkdir -p "$(dirname "${replay_out}")"
+    # per-GPU utilization timeseries over the replay window (2s sampling)
+    bash "${FRESH_ROOT}/microbench/fresh_setup/gpu_monitor.sh" "${rundir}/gpu_util.csv" 2 >/dev/null 2>&1 &
+    local GPU_MON=$!
+    local t0
+    t0=$(date +%s.%N)
+    if ! PYTHONPATH="${FRESH_ROOT}" python -m replayer \
+            --endpoint "${endpoints}" \
+            --trace "${TRACE}" \
+            --output "${replay_out}" \
+            --model "${MODEL_NAME}" \
+            ${REQUEST_LIMIT_ARG} \
+            > "${OUT_ROOT}/${config}_rep${rep}_replay.log" 2>&1; then
+        local t1
+        t1=$(date +%s.%N)
+        local wall=$(python -c "print(${t1} - ${t0})")
+        echo "[mb5-run] REPLAY FAILED after ${wall} s; see ${OUT_ROOT}/${config}_rep${rep}_replay.log"
+        kill "${GPU_MON}" 2>/dev/null || true
+        bash "${LAUNCH}" stop > /dev/null 2>&1 || true
+        return 1
+    fi
+    local t1
+    t1=$(date +%s.%N)
+    local wall_clock_s
+    wall_clock_s=$(python -c "print(${t1} - ${t0})")
+    echo "[mb5-run] replay done in ${wall_clock_s}s"
+    echo "${wall_clock_s}" > "${rundir}/wall_clock_s.txt"
+    kill "${GPU_MON}" 2>/dev/null || true
+    printf '{"t_start_unix":%s,"t_end_unix":%s}\n' "${t0}" "${t1}" > "${rundir}/run_window.json"
+
+    # Per-instance prefix-cache counters, scraped from each backend BEFORE
+    # teardown. For PD this is the only honest reuse signal: producer ports
+    # (the low ones) show cross-turn prefix-cache hits; the consumer's
+    # per-request cached_tokens is meaningless (it counts transferred KV).
+    {
+        for p in 8000 8001 8002 8003 8004 8005 8006 8007; do
+            m=$(curl -s --noproxy '*' "http://127.0.0.1:${p}/metrics" 2>/dev/null) || continue
+            q=$(printf '%s' "$m" | awk '/^vllm:prefix_cache_queries_total/{print $2; exit}')
+            h=$(printf '%s' "$m" | awk '/^vllm:prefix_cache_hits_total/{print $2; exit}')
+            [ -n "${q}" ] && echo "port=${p} queries=${q} hits=${h}"
+        done
+    } > "${rundir}/instance_apc.txt" 2>/dev/null || true
+
+    # Stop launch (cleans up vllm + proxy; reverts patch on last call)
+    CONFIG="${config}" RUN_LABEL="${RUN_TAG}_${config}_rep${rep}" \
+        bash "${LAUNCH}" stop > "${OUT_ROOT}/${config}_rep${rep}_stop.log" 2>&1 || true
+
+    sleep 10  # cooldown so GPUs settle before next config
+    echo "[mb5-run] DONE ${config} rep${rep}"
+}
+
+# Quick check that the launch script and replayer are reachable
+if [ ! -f "${LAUNCH}" ]; then echo "missing ${LAUNCH}"; exit 1; fi
+if [ ! -d "${REPLAYER_DIR}" ]; then echo "missing ${REPLAYER_DIR}"; exit 1; fi
+if [ ! -f "${TRACE}" ]; then echo "missing trace ${TRACE}"; exit 1; fi
+
+# Iterate
+failures=0
+for config in ${CONFIGS}; do
+    for ((rep=1; rep<=REPS; rep++)); do
+        if ! run_one "${config}" "${rep}"; then
+            failures=$((failures+1))
+        fi
+    done
+done
+
+# Final patch revert (defensive — mb5_launch.sh stop also reverts)
+python "${SCRIPT_DIR}/instrument_kv_snapshot.py" --revert --venv "${VENV}" 2>/dev/null || true
+
+echo ""
+echo "======== ALL CONFIGS DONE ========"
+echo "failures: ${failures}"
+echo "results under: ${FRESH_ROOT}/mb5_runs/${RUN_TAG}_*"
+echo "to plot:  python plot_kv_pool_timeline.py --run-tag ${RUN_TAG}"

microbench/fresh_setup/partial_summary.py

@@ -0,0 +1,98 @@
+"""Compute a per-run summary directly from replay_metrics.jsonl (for partial / in-flight runs).
+
+Used when the replayer hasn't completed (so replay_metrics.summary.json doesn't exist
+yet) but enough records have streamed to disk to read out the per-arm result.
+
+Also accepts a finished run's directory and prints the same one-line summary for
+apples-to-apples comparison.
+"""
+from __future__ import annotations
+
+import json
+import re
+import statistics
+import sys
+from pathlib import Path
+
+
+def stats(xs):
+    xs = sorted(xs)
+    n = len(xs)
+    if n == 0:
+        return None
+    return {
+        "n": n,
+        "mean": statistics.fmean(xs),
+        "p50": xs[n // 2],
+        "p90": xs[int(0.9 * (n - 1))],
+        "p99": xs[int(0.99 * (n - 1))],
+    }
+
+
+def apc(run: Path, producer_ports):
+    f = run / "instance_apc.txt"
+    if not f.exists():
+        return None
+    q = h = 0.0
+    for line in open(f):
+        m = dict(re.findall(r"(\w+)=(\S+)", line))
+        try:
+            p = int(m.get("port", -1))
+        except ValueError:
+            continue
+        if p in producer_ports:
+            q += float(m.get("queries", 0))
+            h += float(m.get("hits", 0))
+    return (h / q) if q else None
+
+
+def main():
+    for d in sys.argv[1:]:
+        run = Path(d)
+        # prefer the live replay_metrics.jsonl (so partials work); fall back to metrics.jsonl
+        for fn in ("replay_metrics.partial.jsonl", "replay_metrics.jsonl", "metrics.jsonl"):
+            p = run / fn
+            if p.exists():
+                rec_path = p
+                break
+        else:
+            print(f"{run.name}: no records"); continue
+        recs = [json.loads(l) for l in open(rec_path)]
+        oks = [r for r in recs if r.get("error") is None]
+        lat = stats([r["latency_s"] for r in oks if "latency_s" in r])
+        ttft = stats([r["ttft_s"] for r in oks if "ttft_s" in r])
+        tpot = stats([r["tpot_s"] for r in oks if "tpot_s" in r])
+        out = sum(r.get("actual_output_tokens", r.get("output_length", 0)) for r in oks)
+        ts = [r["t_dispatch_unix"] for r in oks if "t_dispatch_unix" in r]
+        tf = [r["t_finish_unix"] for r in oks if "t_finish_unix" in r]
+        span = max(tf) - min(ts) if ts and tf else 0
+        tps = out / span if span else 0
+
+        # producer ports by arm tag in dirname
+        n = run.name
+        if "_colo_" in n:
+            ports = list(range(8000, 8008))
+        elif "_pd6_" in n:
+            ports = list(range(8000, 8006))
+        elif "_pd2_" in n:
+            ports = list(range(8000, 8002))
+        else:
+            ports = list(range(8000, 8004))
+        a = apc(run, ports)
+
+        print(f"{run.name}")
+        print(f"  n_ok={len(oks)}/{len(recs)}"
+              + (f"  (target=1214 -> {len(oks)*100/1214:.1f}%)" if len(recs) < 1214 else ""))
+        if lat:
+            print(f"  E2E   mean={lat['mean']:.2f}  p50={lat['p50']:.2f}  p90={lat['p90']:.2f}  p99={lat['p99']:.2f}")
+        if ttft:
+            print(f"  TTFT  mean={ttft['mean']:.2f}  p50={ttft['p50']:.2f}  p90={ttft['p90']:.2f}  p99={ttft['p99']:.2f}")
+        if tpot:
+            print(f"  TPOT  mean={tpot['mean']*1000:.1f}ms  p90={tpot['p90']*1000:.1f}ms  p99={tpot['p99']*1000:.1f}ms")
+        print(f"  output_tokens={out:.0f}  span={span:.0f}s  TPS={tps:.0f}")
+        if a is not None:
+            print(f"  producer APC={a*100:.1f}%")
+
+
+if __name__ == "__main__":
+    main()

microbench/fresh_setup/plot_fig4_linear_vs_lmetric.py

@@ -0,0 +1,104 @@
+"""Linear vs LMetric routing on the real agentic trace (first600s).
+
+Visualizes the wall-cap finding: with the 2x-colo-wall cap on PD-disagg arms,
+linear and LMetric reach the *same* success-rate ceiling -- the static P:D
+split has a structural completion ceiling that does not depend on the routing
+policy or on how long you keep retrying.  Routing affects only how much wall
+time is wasted on requests that will never succeed.
+
+Inputs : analysis/v2/fig4l_lmetric.json   (8 arms, both traces; we use first600s)
+         analysis/v2/fig4r_linear.json    (4 arms, first600s, PD wall-capped)
+Output : figs/v2/fig4_linear_vs_lmetric.png
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import numpy as np
+
+ROOT = Path(__file__).resolve().parents[2]
+DATA = ROOT / "analysis" / "v2"
+OUT = ROOT / "figs" / "v2" / "fig4_linear_vs_lmetric.png"
+
+ARMS = ["colo", "6P+2D", "4P+4D", "2P+6D"]
+POLICY_COLOR = {"linear": "#9467bd", "lmetric": "#2ca02c"}
+POLICY_LABEL = {"linear": "linear (cache-aware + session-affinity)",
+                "lmetric": "LMetric (P_tokens × BS)"}
+
+
+def pick(rows, arm, trace="first600s"):
+    for r in rows:
+        if r["arm"] == arm and r["trace"] == trace:
+            return r
+    return None
+
+
+def main():
+    lin = json.load(open(DATA / "fig4r_linear.json"))
+    lme = json.load(open(DATA / "fig4l_lmetric.json"))
+
+    # colo wall (linear) sets the 2x cap reference
+    colo_lin_wall = pick(lin, "colo")["wall"]
+    cap = 2 * colo_lin_wall
+
+    fig, axes = plt.subplots(1, 3, figsize=(15, 4.5))
+    x = np.arange(len(ARMS))
+    w = 0.38
+
+    # (a) success rate
+    ax = axes[0]
+    for i, (pol, rows) in enumerate([("linear", lin), ("lmetric", lme)]):
+        vals = [pick(rows, a)["n"] / pick(rows, a)["req"] * 100 for a in ARMS]
+        bars = ax.bar(x + (i - 0.5) * w, vals, w, color=POLICY_COLOR[pol], label=POLICY_LABEL[pol])
+        for bx, bv in zip(x + (i - 0.5) * w, vals):
+            ax.annotate(f"{bv:.0f}%", (bx, bv + 1.5), ha="center", fontsize=8)
+    ax.axhline(100, color="grey", ls=":", lw=1)
+    ax.set_xticks(x); ax.set_xticklabels(ARMS)
+    ax.set_ylabel("success rate (% of trace)"); ax.set_ylim(0, 115)
+    ax.set_title("(a) success ceiling is policy-invariant")
+    ax.legend(fontsize=8, loc="upper right"); ax.grid(alpha=.3, axis="y")
+
+    # (b) wall (log y) with cap line
+    ax = axes[1]
+    for i, (pol, rows) in enumerate([("linear", lin), ("lmetric", lme)]):
+        vals = [pick(rows, a)["wall"] for a in ARMS]
+        bars = ax.bar(x + (i - 0.5) * w, vals, w, color=POLICY_COLOR[pol],
+                      label=POLICY_LABEL[pol])
+        for bx, bv, r in zip(x + (i - 0.5) * w, vals,
+                              [pick(rows, a) for a in ARMS]):
+            mark = " ⊗" if r.get("capped") else ""
+            ax.annotate(f"{bv:.0f}s{mark}", (bx, bv * 1.05), ha="center", fontsize=7)
+    ax.axhline(cap, color="red", ls="--", lw=1.5,
+               label=f"2× colo wall cap = {cap:.0f}s")
+    ax.set_xticks(x); ax.set_xticklabels(ARMS)
+    ax.set_ylabel("wall-clock (s, log)"); ax.set_yscale("log")
+    ax.set_title("(b) linear w/ cap vs lmetric w/o cap — ⊗ = cap-killed")
+    ax.legend(fontsize=8, loc="upper left"); ax.grid(alpha=.3, which="both", axis="y")
+
+    # (c) goodput per minute of wall (success rate / wall × 60)
+    ax = axes[2]
+    for i, (pol, rows) in enumerate([("linear", lin), ("lmetric", lme)]):
+        vals = [pick(rows, a)["n"] / pick(rows, a)["wall"] * 60 for a in ARMS]
+        bars = ax.bar(x + (i - 0.5) * w, vals, w, color=POLICY_COLOR[pol], label=POLICY_LABEL[pol])
+        for bx, bv in zip(x + (i - 0.5) * w, vals):
+            ax.annotate(f"{bv:.1f}", (bx, bv + max(vals) * 0.02),
+                        ha="center", fontsize=8)
+    ax.set_xticks(x); ax.set_xticklabels(ARMS)
+    ax.set_ylabel("goodput (successful req / min)")
+    ax.set_title("(c) linear+cap is 1.5–17× more wall-efficient on PD")
+    ax.legend(fontsize=8, loc="upper right"); ax.grid(alpha=.3, axis="y")
+
+    fig.suptitle("Fig 4r — Linear vs LMetric on the real agentic trace (first600s, "
+                 "PD-disagg wall-capped at 2× colo)",
+                 fontsize=12, y=1.0)
+    fig.tight_layout()
+    fig.savefig(OUT, dpi=130, bbox_inches="tight")
+    print(f"wrote {OUT}")
+
+
+if __name__ == "__main__":
+    main()

microbench/fresh_setup/plot_fig4l_lmetric.py

@@ -0,0 +1,113 @@
+"""Render the LMetric PD-colo vs PD-disagg figure on the real agentic trace.
+
+Input  : analysis/v2/fig4l_lmetric.json     (8 arms = 4 ratios x 2 traces)
+Output : figs/v2/fig4_lmetric_pd_vs_colo.png
+
+Four panels x four ratios x two traces:
+  (a) completion rate %
+  (b) E2E latency (mean / p50 / p90)
+  (c) throughput (output tokens / second)
+  (d) bench wall-clock seconds
+
+The thesis the figure visualizes: with LMetric routing,
+  - colo (elastic 8-GPU pool) holds 100% completion on both traces
+  - every PD-disagg ratio fails (completion 14-65%), and the failure mode
+    rotates with the split (pd2 = prefill-bound, pd6 = decode-bound)
+  - routing policy does not rescue PD-disagg; the bottleneck is structural.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import numpy as np
+
+ROOT = Path(__file__).resolve().parents[2]
+DATA = ROOT / "analysis" / "v2" / "fig4l_lmetric.json"
+OUT = ROOT / "figs" / "v2" / "fig4_lmetric_pd_vs_colo.png"
+OUT.parent.mkdir(parents=True, exist_ok=True)
+
+ARMS = ["colo", "6P+2D", "4P+4D", "2P+6D"]   # decode-rich -> prefill-rich
+TRACES = ["first600s", "full"]
+TRACE_LABEL = {"first600s": "first600s (1.35 req/s, high load)",
+               "full": "full w600 (0.42 req/s, original §3)"}
+COLOR = {"first600s": "#1f77b4", "full": "#ff7f0e"}
+
+
+def pick(rows, trace, arm):
+    for r in rows:
+        if r["trace"] == trace and r["arm"] == arm:
+            return r
+    return None
+
+
+def main():
+    rows = json.load(open(DATA))
+    fig, axes = plt.subplots(2, 2, figsize=(12, 8))
+    width = 0.38
+    x = np.arange(len(ARMS))
+
+    # (a) completion %
+    ax = axes[0, 0]
+    for i, tr in enumerate(TRACES):
+        vals = [pick(rows, tr, a)["n"] / pick(rows, tr, a)["req"] * 100 for a in ARMS]
+        bars = ax.bar(x + (i - 0.5) * width, vals, width, color=COLOR[tr], label=TRACE_LABEL[tr])
+        for bx, bv in zip(x + (i - 0.5) * width, vals):
+            ax.annotate(f"{bv:.0f}%", (bx, bv + 1.5), ha="center", fontsize=8)
+    ax.axhline(100, color="grey", ls=":", lw=1)
+    ax.set_xticks(x); ax.set_xticklabels(ARMS)
+    ax.set_ylabel("completion (%)"); ax.set_ylim(0, 115)
+    ax.set_title("(a) request completion — colo holds 100%, all PD ratios fail")
+    ax.legend(fontsize=8); ax.grid(alpha=.3, axis="y")
+
+    # (b) E2E percentiles
+    ax = axes[0, 1]
+    for i, tr in enumerate(TRACES):
+        p50 = [pick(rows, tr, a)["e2e"]["p50"] for a in ARMS]
+        p90 = [pick(rows, tr, a)["e2e"]["p90"] for a in ARMS]
+        off = (i - 0.5) * width
+        ax.bar(x + off, p90, width, color=COLOR[tr], alpha=0.55, label=f"{tr} p90")
+        ax.bar(x + off, p50, width, color=COLOR[tr], alpha=1.0, label=f"{tr} p50")
+    ax.axhline(600, color="red", ls=":", lw=1, label="600 s request timeout")
+    ax.set_xticks(x); ax.set_xticklabels(ARMS)
+    ax.set_ylabel("E2E latency (s, log)"); ax.set_yscale("log")
+    ax.set_title("(b) E2E p50 (solid) + p90 (faded) — PD pegs at the timeout")
+    ax.legend(fontsize=7, ncol=2); ax.grid(alpha=.3, which="both", axis="y")
+
+    # (c) TPS
+    ax = axes[1, 0]
+    for i, tr in enumerate(TRACES):
+        vals = [pick(rows, tr, a)["tps"] for a in ARMS]
+        ax.bar(x + (i - 0.5) * width, vals, width, color=COLOR[tr], label=TRACE_LABEL[tr])
+        for bx, bv in zip(x + (i - 0.5) * width, vals):
+            ax.annotate(f"{bv:.0f}", (bx, bv + 4), ha="center", fontsize=8)
+    ax.set_xticks(x); ax.set_xticklabels(ARMS)
+    ax.set_ylabel("throughput (output tokens/s)")
+    ax.set_title("(c) throughput — PD throughput crashes 5–100×")
+    ax.legend(fontsize=8); ax.grid(alpha=.3, axis="y")
+
+    # (d) wall (min)
+    ax = axes[1, 1]
+    for i, tr in enumerate(TRACES):
+        vals = [pick(rows, tr, a)["wall"] / 60 for a in ARMS]
+        ax.bar(x + (i - 0.5) * width, vals, width, color=COLOR[tr], label=TRACE_LABEL[tr])
+        for bx, bv in zip(x + (i - 0.5) * width, vals):
+            ax.annotate(f"{bv:.0f}m", (bx, bv * 1.05), ha="center", fontsize=8)
+    ax.set_xticks(x); ax.set_xticklabels(ARMS)
+    ax.set_ylabel("bench wall-clock (min, log)"); ax.set_yscale("log")
+    ax.set_title("(d) wall-clock — PD drain dilates the run")
+    ax.legend(fontsize=8); ax.grid(alpha=.3, which="both", axis="y")
+
+    fig.suptitle("Fig 4 (LMetric) — PD-colo vs PD-disagg on the real agentic trace "
+                 "(`w600_r0.0015_st30`), clean stack, cache-aware LMetric routing",
+                 fontsize=12, y=1.0)
+    fig.tight_layout()
+    fig.savefig(OUT, dpi=130, bbox_inches="tight")
+    print(f"wrote {OUT}")
+
+
+if __name__ == "__main__":
+    main()

microbench/fresh_setup/plot_pd_crossover.py

@@ -0,0 +1,189 @@
+"""Render the three PD-vs-colo crossover figures from fig_agg JSON dumps.
+
+Inputs (produced by `fig_agg.py --json`):
+  analysis/mb5_pd_ablation/fig1_reuse_fixed.json  reuse axis (N=8, FIXED real
+        prefill delta=2048; vary cached prefix -> reuse = pfx/(pfx+delta).
+        Controlled-variable: real new-prefill work is constant across the sweep,
+        only the cached fraction (and total context) grows. Supersedes the old
+        fig1.json, which held input=8192 and sliced prefix out of it so delta
+        shrank 15x as reuse rose — a confound, not a pure reuse axis.)
+  analysis/mb5_pd_ablation/fig2.json   shape axis   (N=8, reuse~70%)
+  analysis/mb5_pd_ablation/fig3_conc32k.json  concurrency (in32768/out128,
+        reuse~0.984 = 32256 resident + 512 real new-prefill per turn; retuned
+        2026-05-31 to the agentic corner so PD pays the full-context per-turn
+        KV-transfer tax while colo keeps it resident; vary N by step 8 up to the
+        mean-E2E<=10s SLO ceiling)
+
+Each figure overlays colo + the three PD ratios and marks the PD-best advantage.
+All three share the corrected (uncontaminated, e13391e-gated-off) stack.
+"""
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+
+ROOT = Path(__file__).resolve().parents[2]
+DATA = ROOT / "analysis" / "mb5_pd_ablation"
+OUT = ROOT / "figs" / "mb5_pd_ablation"
+OUT.mkdir(parents=True, exist_ok=True)
+
+PD_ARMS = ["2P+6D", "4P+4D", "6P+2D"]
+STYLE = {
+    "colo":  dict(color="k",        marker="o", lw=2.4, ls="-",  label="colo (8×kv_both)"),
+    "2P+6D": dict(color="#1f77b4",  marker="s", lw=1.6, ls="--", label="PD 2P+6D"),
+    "4P+4D": dict(color="#2ca02c",  marker="^", lw=1.6, ls="--", label="PD 4P+4D"),
+    "6P+2D": dict(color="#ff7f0e",  marker="v", lw=1.6, ls="--", label="PD 6P+2D"),
+}
+
+
+def load(name):
+    return json.load(open(DATA / name))
+
+
+def by_axis(rows, keyfn):
+    """Group rows -> {axis_val: {arm: row}}."""
+    out = {}
+    for r in rows:
+        k = keyfn(r["name"])
+        if k is None:
+            continue
+        out.setdefault(k, {})[r["arm"]] = r
+    return out
+
+
+def pd_best(armmap, metric="e2e_p90"):
+    vals = [(a, armmap[a][metric]) for a in PD_ARMS
+            if a in armmap and armmap[a].get(metric) is not None]
+    return min(vals, key=lambda t: t[1]) if vals else (None, None)
+
+
+def series(grp, xs, arm, metric):
+    return [grp[x][arm].get(metric) if arm in grp[x] else None for x in xs]
+
+
+# ---------- Fig 1: reuse axis ----------
+def _reuse_pct(name):
+    """Reuse % from a `reuse_p{pfx}_d{delta}_{arm}` run name: pfx/(pfx+delta)."""
+    m = re.search(r"_p(\d+)_d(\d+)", name)
+    if not m:
+        return None
+    pfx, delta = int(m.group(1)), int(m.group(2))
+    return round(pfx / (pfx + delta) * 100)
+
+
+def fig_reuse():
+    g = by_axis(load("fig1_reuse_fixed.json"), _reuse_pct)
+    xs = sorted(g)
+    reuse = xs
+    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 4.2))
+    for arm in ["colo", *PD_ARMS]:
+        ax1.plot(reuse, series(g, xs, arm, "e2e_p90"), **STYLE[arm])
+    ax1.set_xlabel("intra-session KV reuse (%)  [fixed real prefill, delta=2048]")
+    ax1.set_ylabel("E2E latency p90 (s)")
+    ax1.set_title("(a) E2E-p90 vs reuse  (N=8, delta=2048/out256)")
+    ax1.legend(fontsize=8); ax1.grid(alpha=.3)
+
+    adv, putil = [], []
+    for x in xs:
+        co = g[x]["colo"]["e2e_p90"]; _, b = pd_best(g[x])
+        adv.append(co / b if b else None)
+        a = pd_best(g[x])[0]
+        putil.append(g[x][a].get("pu") if a else None)
+    ax2.plot(reuse, adv, color="purple", marker="D", lw=2, label="PD-best advantage (colo/PD)")
+    ax2.axhline(1.0, color="grey", ls=":", lw=1)
+    ax2.set_xlabel("intra-session KV reuse (%)"); ax2.set_ylabel("advantage  (>1 = PD wins)")
+    ax2b = ax2.twinx()
+    ax2b.plot(reuse, putil, color="brown", marker="x", lw=1.4, ls="-.", label="PD-best prefill-GPU util")
+    ax2b.set_ylabel("prefill-GPU util (%)", color="brown"); ax2b.tick_params(axis="y", colors="brown")
+    ax2.set_title("(b) advantage erodes; prefill GPUs go idle")
+    l1, la1 = ax2.get_legend_handles_labels(); l2, la2 = ax2b.get_legend_handles_labels()
+    ax2.legend(l1 + l2, la1 + la2, fontsize=8, loc="center right"); ax2.grid(alpha=.3)
+    fig.suptitle("Fig 1 — Reuse axis (fixed real prefill delta=2048): PD's edge vs rising cache reuse",
+                 fontsize=11, y=1.02)
+    fig.tight_layout(); p = OUT / "fig1_reuse_axis.png"; fig.savefig(p, dpi=130, bbox_inches="tight")
+    print("wrote", p)
+
+
+# ---------- Fig 2: shape axis ----------
+def fig_shape():
+    g = by_axis(load("fig2.json"),
+                lambda n: ((int(m.group(1)), int(m.group(2)))
+                           if (m := re.search(r"_in(\d+)_out(\d+)_", n)) else None))
+    xs = sorted(g, key=lambda t: t[0])  # ascending input
+    labels = [f"in{i}\nout{o}" for i, o in xs]
+    xi = list(range(len(xs)))
+    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 4.2))
+    for arm in ["colo", *PD_ARMS]:
+        ax1.plot(xi, series(g, xs, arm, "e2e_p90"), **STYLE[arm])
+    ax1.set_xticks(xi); ax1.set_xticklabels(labels, fontsize=7)
+    ax1.set_xlabel("shape  (decode-heavy → prefill-heavy)"); ax1.set_ylabel("E2E latency p90 (s)")
+    ax1.set_title("(a) E2E-p90 vs shape  (N=8, reuse~70%)")
+    ax1.legend(fontsize=8); ax1.grid(alpha=.3)
+
+    adv, comp = [], []
+    for x in xs:
+        co = g[x]["colo"]["e2e_p90"]; a, b = pd_best(g[x])
+        adv.append(co / b if b else None)
+        # completion of the worst PD arm (exposes catastrophic ratio)
+        worst = min((g[x][arm]["n"] / g[x][arm]["req"]) for arm in PD_ARMS if arm in g[x])
+        comp.append(worst * 100)
+    ax2.plot(xi, adv, color="purple", marker="D", lw=2, label="PD-best advantage (colo/PD)")
+    ax2.axhline(1.0, color="grey", ls=":", lw=1)
+    ax2.set_xticks(xi); ax2.set_xticklabels(labels, fontsize=7)
+    ax2.set_xlabel("shape"); ax2.set_ylabel("advantage  (>1 = PD wins)")
+    ax2b = ax2.twinx()
+    ax2b.plot(xi, comp, color="red", marker="x", lw=1.4, ls="-.", label="worst-PD-arm completion %")
+    ax2b.set_ylabel("worst PD completion (%)", color="red"); ax2b.tick_params(axis="y", colors="red")
+    ax2b.set_ylim(80, 101)
+    ax2.set_title("(b) advantage peaks mid-sweep; wrong ratio catastrophic at prefill extreme")
+    l1, la1 = ax2.get_legend_handles_labels(); l2, la2 = ax2b.get_legend_handles_labels()
+    ax2.legend(l1 + l2, la1 + la2, fontsize=8, loc="lower left"); ax2.grid(alpha=.3)
+    fig.suptitle("Fig 2 — Shape axis: PD wins decode-heavy, ties prefill-heavy; optimal ratio rotates",
+                 fontsize=11, y=1.02)
+    fig.tight_layout(); p = OUT / "fig2_shape_axis.png"; fig.savefig(p, dpi=130, bbox_inches="tight")
+    print("wrote", p)
+
+
+# ---------- Fig 3: concurrency axis ----------
+def fig_conc():
+    g = by_axis(load("fig3_conc32k.json"),
+                lambda n: (int(m.group(1)) if (m := re.search(r"_N(\d+)_", n)) else None))
+    xs = sorted(g)
+    fig, (ax1, ax2, ax3) = plt.subplots(1, 3, figsize=(15, 4.2))
+
+    # (a) request completion % — the headline (latency percentiles count successes
+    # only, so they understate PD; completion is the honest collapse signal).
+    for arm in ["colo", *PD_ARMS]:
+        comp = [(g[x][arm]["n"] / g[x][arm]["req"] * 100) if arm in g[x] else None for x in xs]
+        ax1.plot(xs, comp, **STYLE[arm])
+    ax1.axhline(100, color="grey", ls=":", lw=1)
+    ax1.set_xticks(xs); ax1.set_xticklabels(xs, fontsize=7)
+    ax1.set_xlabel("concurrent sessions N"); ax1.set_ylabel("request completion (%)")
+    ax1.set_title("(a) completion: colo 100%, PD collapses"); ax1.legend(fontsize=8); ax1.grid(alpha=.3)
+
+    for arm in ["colo", *PD_ARMS]:
+        ax2.plot(xs, series(g, xs, arm, "e2e_mean"), **STYLE[arm])
+    ax2.axhline(10.0, color="red", ls=":", lw=1, label="SLO 10s")
+    ax2.set_yscale("log"); ax2.set_xticks(xs); ax2.set_xticklabels(xs, fontsize=7)
+    ax2.set_xlabel("concurrent sessions N"); ax2.set_ylabel("E2E latency mean (s, log)")
+    ax2.set_title("(b) mean-E2E (successes only)"); ax2.legend(fontsize=8); ax2.grid(alpha=.3, which="both")
+
+    for arm in ["colo", *PD_ARMS]:
+        ax3.plot(xs, series(g, xs, arm, "tps"), **STYLE[arm])
+    ax3.set_xticks(xs); ax3.set_xticklabels(xs, fontsize=7)
+    ax3.set_xlabel("concurrent sessions N"); ax3.set_ylabel("throughput (tok/s)")
+    ax3.set_title("(c) TPS"); ax3.legend(fontsize=8); ax3.grid(alpha=.3)
+    fig.suptitle("Fig 3 — Concurrency axis (in32768/out128, reuse~0.984, PD capped 600s / colo uncapped): "
+                 "colo degrades gracefully (100% completion), PD collapses earlier as N rises",
+                 fontsize=10, y=1.02)
+    fig.tight_layout(); p = OUT / "fig3_concurrency_axis.png"; fig.savefig(p, dpi=130, bbox_inches="tight")
+    print("wrote", p)
+
+
+if __name__ == "__main__":
+    fig_reuse(); fig_shape(); fig_conc()

microbench/fresh_setup/run_campaign.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+# Unattended serial PD-ablation campaign: reuse sweep -> conc sweep.
+# STRICTLY one driver at a time (the hard lesson): each inner driver brings up and
+# tears down its own vLLM per config via scripts/mb5_run_gpu.sh, and the two sweeps
+# run sequentially (reuse fully finishes + tears down before conc starts). We verify
+# GPUs are clear between sweeps. NO set -e here: a sub-sweep nonzero must NOT skip the
+# other sweep; rc is captured and reported. Detached launch writes a DONE marker.
+cd /home/admin/cpfs/wjh/agentic-kv-fresh
+export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
+FS=microbench/fresh_setup
+
+echo "=== CAMPAIGN START $(date) ==="
+
+echo "=== [1/2] REUSE SWEEP (fixed real prefill delta=2048, out=256, reuse 20-95%, N=8) $(date) ==="
+bash "$FS/run_reuse_fixed.sh"; rc_reuse=$?
+echo "=== reuse sweep rc=$rc_reuse $(date) ==="
+
+sleep 15
+echo "--- GPU mem after reuse sweep (expect ~0 before conc) ---"
+nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
+
+echo "=== [2/2] CONC SWEEP (in=32768 reuse=0.984, balanced N grid 8 16 32 48 64 96 128) $(date) ==="
+NLIST="8 16 32 48 64 96 128" bash "$FS/run_conc.sh"; rc_conc=$?
+echo "=== conc sweep rc=$rc_conc $(date) ==="
+
+echo "=== CAMPAIGN DONE reuse_rc=$rc_reuse conc_rc=$rc_conc $(date) ==="

microbench/fresh_setup/run_campaign2.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+# Campaign 2 (2026-05-31): two extra reuse sweeps at out=128 (user request:
+# delta=1024/out=128 and delta=2048/out=128), then the capped conc restart.
+# STRICTLY one driver at a time; reuse sweeps run uncapped (mild collapse, matches
+# the existing d2048/o256 sweep), conc runs with the PD-arm wall-cap. NO set -e.
+cd /home/admin/cpfs/wjh/agentic-kv-fresh
+export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
+FS=microbench/fresh_setup
+
+echo "=== CAMPAIGN2 START $(date) ==="
+
+echo "=== [1/3] REUSE delta=1024 out=128 (reuse 0.33-0.97) $(date) ==="
+DELTA=1024 OL=128 bash "$FS/run_reuse_fixed.sh"; rc1=$?
+echo "=== reuse d1024 o128 rc=$rc1 $(date) ==="
+sleep 12; nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
+
+echo "=== [2/3] REUSE delta=2048 out=128 (reuse 0.20-0.95) $(date) ==="
+DELTA=2048 OL=128 bash "$FS/run_reuse_fixed.sh"; rc2=$?
+echo "=== reuse d2048 o128 rc=$rc2 $(date) ==="
+sleep 12; nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
+
+echo "=== [3/3] CONC capped (PD wall=${CONC_PD_MAXDUR:-600}s, colo uncapped), N 8..128 $(date) ==="
+NLIST="8 16 32 48 64 96 128" bash "$FS/run_conc.sh"; rc3=$?
+echo "=== conc rc=$rc3 $(date) ==="
+
+echo "=== CAMPAIGN2 DONE reuse_d1024_o128=$rc1 reuse_d2048_o128=$rc2 conc=$rc3 $(date) ==="

microbench/fresh_setup/run_campaign3.sh

@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+# Campaign 3 (2026-05-31): the uncapped d2048/o128 reuse sweep stalled on a
+# collapse-draining high-reuse PD arm (4P+4D @ reuse 0.90, ~1 req/several-min).
+# Finish it by re-running ONLY the high-reuse points (0.90, 0.95) WITH the PD
+# wall-cap (low-reuse arms already completed and are cap-insensitive). Then run
+# the capped conc sweep. STRICTLY serial. NO set -e.
+cd /home/admin/cpfs/wjh/agentic-kv-fresh
+export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
+FS=microbench/fresh_setup
+echo "=== CAMPAIGN3 START $(date) ==="
+
+echo "=== [1/2] finish reuse d2048/o128: re-run pts pfx=18432,38912 (PD capped 500s) $(date) ==="
+DELTA=2048 OL=128 PFXS="18432 38912" REUSE_PD_MAXDUR=500 bash "$FS/run_reuse_fixed.sh"; rc1=$?
+echo "=== reuse d2048 o128 finish rc=$rc1 $(date) ==="
+sleep 12; nvidia-smi --query-gpu=index,memory.used --format=csv,noheader | head -8
+
+echo "=== [2/2] CONC capped (PD wall=600s, colo uncapped), N 8..128 $(date) ==="
+NLIST="8 16 32 48 64 96 128" CONC_PD_MAXDUR=600 bash "$FS/run_conc.sh"; rc2=$?
+echo "=== conc rc=$rc2 $(date) ==="
+echo "=== CAMPAIGN3 DONE reuse_finish=$rc1 conc=$rc2 $(date) ==="

microbench/fresh_setup/run_conc.sh

@@ -0,0 +1,70 @@
+#!/usr/bin/env bash
+# Concurrency axis, agentic-corner config. Supersedes old fig3 (in~8192/out256).
+# RETUNED 2026-05-31 for realism (C2): hold total context in=32768 but shrink the
+# real per-turn new-prefill to delta=512 and push reuse to 0.984 (real agentic
+# reuse ->99.6%). prefix 32256 + delta 512. out=128. This is the corner that
+# exposes PD's structural tax: colo keeps the 32k resident KV local, but PD must
+# KV-transfer the whole 32k context every turn even though only 512 tokens are new
+# (C2 PD-tax ~250-450x). Sweep closed-loop N by step 8 up to mean-E2E<=SLO ceiling.
+# Wiring per memory project-mb5-pd-ablation-wiring: .venv_dash0, traces_synth/,
+# CONFIG 8C-proxy + PD, MB5_P_ROUTING=session + MB5_COLO_ROUTING=session,
+# N=REPLAY_MAX_INFLIGHT closed loop + REPLAY_INTER_TURN_THINK_S,
+# REPLAY_NO_REALIZED_PREFIX=1. RUN ONLY ONE DRIVER AT A TIME (shared GPUs/ports).
+set -eo pipefail
+cd /home/admin/cpfs/wjh/agentic-kv-fresh
+export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
+VPY="$MB5_VENV/bin/python"
+
+PFX="${PFX:-32256}"; DELTA="${DELTA:-512}"; OL="${OL:-128}"   # reuse=0.984, in=32768
+THINK="${THINK:-0.5}"; TURNS="${TURNS:-8}"
+NSTART="${NSTART:-8}"; NSTEP="${NSTEP:-8}"; NMAX="${NMAX:-128}"
+NLIST="${NLIST:-}"            # explicit N grid (overrides NSTART/STEP/MAX), e.g. "8 16 32 48 64 96 128"
+CONC_PD_MAXDUR="${CONC_PD_MAXDUR:-600}"   # wall-deadline (s) for PD arms only; bounds collapsed-arm
+                                          # drain (un-run turns = failures). colo (8C-proxy) runs UNCAPPED
+                                          # so the headline reference is always fully measured.
+SLO="${SLO:-10.0}"
+SESS_PER_N="${SESS_PER_N:-4}"
+CFGS="${CFGS:-8C-proxy 2P+6D 4P+4D 6P+2D}"
+ONLY_N="${ONLY_N:-}"
+
+run_N() {
+    local N="$1"; local sess=$(( SESS_PER_N * N ))
+    local tag="conc32k_N${N}"; local trace="traces_synth/${tag}.jsonl"
+    "$VPY" scripts/gen_synthetic_trace.py --out "$trace" --mode regular \
+        --qps "$sess" --duration-s 1 --turns "$TURNS" \
+        --prefix-len "$PFX" --delta-len "$DELTA" --output-len "$OL" --seed 42 >/dev/null
+    echo "[conc32k] N=$N sess=$sess in=$((PFX+DELTA)) out=$OL -> $trace"
+    for cfg in $CFGS; do
+        echo "  -> $cfg"
+        local dur=""; [ "$cfg" != "8C-proxy" ] && dur="$CONC_PD_MAXDUR"   # colo uncapped
+        MB5_P_ROUTING=session MB5_COLO_ROUTING=session \
+            REPLAY_MAX_INFLIGHT="$N" REPLAY_INTER_TURN_THINK_S="$THINK" REPLAY_NO_REALIZED_PREFIX=1 \
+            REPLAY_MAX_DURATION="$dur" \
+            CONFIGS="$cfg" REPS=1 TRACE="$trace" RUN_TAG="$tag" \
+            bash scripts/mb5_run_gpu.sh >/dev/null 2>&1 || echo "    [warn] ${tag}_${cfg} failed" >&2
+    done
+    local d="mb5_runs/${tag}_8C-proxy_rep1"
+    if [ -f "$d/replay_metrics.summary.json" ]; then
+        "$VPY" scripts/fig_agg.py --json "$d" 2>/dev/null \
+          | "$VPY" -c "import sys,json;r=json.load(sys.stdin);print(r[0].get('e2e_mean') if r else 'nan')"
+    else echo nan; fi
+}
+
+if [ -n "$ONLY_N" ]; then
+    echo "[conc32k] SMOKE N=$ONLY_N cfgs='$CFGS'"
+    t0=$(date +%s); m=$(run_N "$ONLY_N"); t1=$(date +%s)
+    echo "[conc32k] SMOKE N=$ONLY_N colo mean-E2E=${m}s wall=$(( t1 - t0 ))s; compare:"
+    "$VPY" scripts/fig_agg.py mb5_runs/conc32k_N${ONLY_N}_*_rep1 2>&1
+    exit 0
+fi
+
+if [ -n "$NLIST" ]; then NSEQ="$NLIST"; else NSEQ=$(seq "$NSTART" "$NSTEP" "$NMAX"); fi
+for N in $NSEQ; do
+    echo "[conc32k] === N=$N ==="
+    m=$(run_N "$N"); echo "[conc32k] N=$N colo mean-E2E=${m}s"
+    over=$("$VPY" -c "print(1 if float('${m}')>${SLO} else 0)" 2>/dev/null || echo 0)
+    [ "$over" = "1" ] && { echo "[conc32k] colo crossed SLO ${SLO}s at N=$N -> stop"; break; }
+done
+dirs=(); for d in mb5_runs/conc32k_N*_rep1; do [ -d "$d" ] && dirs+=("$d"); done
+"$VPY" scripts/fig_agg.py --json "${dirs[@]}" > analysis/mb5_pd_ablation/fig3_conc32k.json
+echo "[conc32k] done -> analysis/mb5_pd_ablation/fig3_conc32k.json (${#dirs[@]} dirs)"

microbench/fresh_setup/run_reuse_fixed.sh

@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+# Reuse axis, DONE RIGHT (controlled variable). Supersedes old fig1.
+#   Hold REAL (uncached) prefill work constant: --delta-len = U fixed.
+#   Vary only --prefix-len = C -> reuse = C/(C+U). Context grows with reuse but
+#   the tokens that must actually be prefilled each turn stays = U.
+# Old fig1 held input=8192 and sliced prefix out of it, so delta shrank 15x as
+# reuse rose -> confounded "more reuse" with "less prefill". This fixes that.
+#
+# Wiring matches the corrected MB5 stack (see memory project-mb5-pd-ablation-wiring):
+#   .venv_dash0, traces_synth/, CONFIG 8C-proxy + PD, MB5_P_ROUTING=session,
+#   N injected via REPLAY_MAX_INFLIGHT (closed loop) + REPLAY_INTER_TURN_THINK_S,
+#   REPLAY_NO_REALIZED_PREFIX=1 (reuse governed by hash_ids, required for this sweep).
+set -eo pipefail
+cd /home/admin/cpfs/wjh/agentic-kv-fresh
+export MB5_VENV="${MB5_VENV:-/home/admin/cpfs/wjh/agentic-kv-fresh/.venv_dash0}"
+VPY="$MB5_VENV/bin/python"
+
+DELTA="${DELTA:-2048}"        # fixed real prefill per turn (USER-CHOSEN)
+OL="${OL:-256}"
+N="${N:-8}"
+THINK="${THINK:-0.5}"
+TURNS="${TURNS:-8}"
+NSESS="${NSESS:-48}"          # number of sessions (closed-loop: arrival rate is
+                             # irrelevant, only the count matters; ~6 waves at N=8)
+PFXS="${PFXS:-512 2048 4096 8192 18432 38912}"   # reuse .20 .50 .67 .80 .90 .95
+CFGS="${CFGS:-8C-proxy 2P+6D 4P+4D 6P+2D}"
+REUSE_PD_MAXDUR="${REUSE_PD_MAXDUR:-500}"   # wall-deadline (s) for PD arms only (colo uncapped):
+                                            # bounds the collapse-drain that stalls high-reuse PD arms
+                                            # (un-run turns = failures, honest completion%). 0/empty = off.
+ONLY_PFX="${ONLY_PFX:-}"     # smoke a single prefix then exit
+
+run_point() {  # <pfx>
+    local pfx="$1"
+    local reuse; reuse=$(python3 -c "print(f'{$pfx/($pfx+$DELTA):.3f}')")
+    local tag="reuse_p${pfx}_d${DELTA}_o${OL}"   # _o${OL} so different output lens don't collide
+    local trace="traces_synth/${tag}.jsonl"
+    # Closed-loop: pass NSESS as qps with duration 1 so n_sessions = NSESS
+    # exactly (gen_regular: n_sessions = int(duration_s * session_qps)).
+    "$VPY" scripts/gen_synthetic_trace.py --out "$trace" --mode regular \
+        --qps "$NSESS" --duration-s 1 --turns "$TURNS" \
+        --prefix-len "$pfx" --delta-len "$DELTA" --output-len "$OL" --seed 42 >/dev/null
+    echo "[reuse] pfx=$pfx delta=$DELTA reuse=$reuse in=$((pfx+DELTA)) -> $trace"
+    for cfg in $CFGS; do
+        echo "  -> $cfg"
+        # Both routings set to session so BOTH colo (kv_both) and PD producers
+        # pin a session's turns to one instance and reuse its prefix cache — the
+        # fair cache-aware comparison. P_ROUTING is ignored by colo, COLO_ROUTING
+        # by PD, so setting both is harmless and symmetric.
+        local dur=""; [ "$cfg" != "8C-proxy" ] && dur="$REUSE_PD_MAXDUR"   # colo uncapped
+        MB5_P_ROUTING=session MB5_COLO_ROUTING=session \
+            REPLAY_MAX_INFLIGHT="$N" REPLAY_INTER_TURN_THINK_S="$THINK" \
+            REPLAY_NO_REALIZED_PREFIX=1 REPLAY_MAX_DURATION="$dur" \
+            CONFIGS="$cfg" REPS=1 TRACE="$trace" RUN_TAG="$tag" \
+            bash scripts/mb5_run_gpu.sh >/dev/null 2>&1 || echo "    [warn] $cfg failed" >&2
+    done
+}
+
+if [ -n "$ONLY_PFX" ]; then
+    echo "[reuse] SMOKE pfx=$ONLY_PFX cfgs='$CFGS'"
+    t0=$(date +%s); run_point "$ONLY_PFX"; t1=$(date +%s)
+    echo "[reuse] SMOKE done wall=$(( t1 - t0 ))s; compare:"
+    "$VPY" scripts/fig_agg.py mb5_runs/reuse_p${ONLY_PFX}_d${DELTA}_o${OL}_*_rep1
+    exit 0
+fi
+
+for pfx in $PFXS; do run_point "$pfx"; done
+# Aggregate ONLY this sweep's dirs (matched by delta+output) so the three
+# reuse figures (d2048/o256, d1024/o128, d2048/o128) never cross-contaminate.
+dirs=(); for d in mb5_runs/reuse_*_d${DELTA}_o${OL}_*_rep1; do [ -d "$d" ] && dirs+=("$d"); done
+OUTJSON="analysis/mb5_pd_ablation/fig1_reuse_d${DELTA}_o${OL}.json"
+"$VPY" scripts/fig_agg.py --json "${dirs[@]}" > "$OUTJSON"
+echo "[reuse] done -> $OUTJSON (${#dirs[@]} dirs)"

paper/data/f2a_mixture_sweep.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""
+f2a sensitivity: how does the intra/cross reuse split move as we change the
+single-turn session fraction?  (Tests whether the old 93%-intra sample vs 54.6%
+full-trace gap is just session-mixture selection bias.)
+
+Keep ALL multi-turn sessions; downsample single-turn sessions to hit each target
+single-turn fraction f.  Re-run the LRU (last-touched), reuse-hits-only
+classification on the filtered request stream.
+
+    python3 f2a_mixture_sweep.py ~/ali-trace/.../051315-051317.jsonl /tmp/f2a_sweep.json
+"""
+import sys, json, time, random
+from collections import Counter, defaultdict
+
+PATH = sys.argv[1]
+OUT = sys.argv[2] if len(sys.argv) > 2 else "/tmp/f2a_sweep.json"
+random.seed(0)
+
+t0 = time.time()
+chat_parent = {}
+records = []
+with open(PATH) as f:
+    for line in f:
+        d = json.loads(line)
+        cid = d["chat_id"]; pc = d.get("parent_chat_id")
+        chat_parent[cid] = 0 if pc is None else pc
+        records.append((d.get("timestamp", 0.0), cid, d.get("hash_ids") or []))
+sys.stderr.write(f"[{time.time()-t0:.0f}s] loaded {len(records)}\n")
+
+root_cache = {}
+def resolve_root(cid):
+    chain = []; cur = cid
+    while True:
+        if cur in root_cache:
+            r = root_cache[cur]; break
+        p = chat_parent.get(cur, 0)
+        if p == 0 or p not in chat_parent:
+            r = cur; break
+        chain.append(cur); cur = p
+        if len(chain) > 100000:
+            r = cur; break
+    for nd in chain:
+        root_cache[nd] = r
+    root_cache[cid] = r
+    return r
+
+records.sort(key=lambda x: x[0])
+roots = [resolve_root(cid) for _, cid, _ in records]
+req_per_root = Counter(roots)
+single_roots = [r for r, c in req_per_root.items() if c == 1]
+multi_roots = [r for r, c in req_per_root.items() if c >= 2]
+M = len(multi_roots)
+sys.stderr.write(f"[{time.time()-t0:.0f}s] roots: single={len(single_roots)} multi={M}\n")
+
+GAP_EDGES = [1, 10, 60, 300, 1800, 3600, float("inf")]
+def gbucket(g):
+    for i, e in enumerate(GAP_EDGES):
+        if g < e:
+            return i
+    return len(GAP_EDGES) - 1
+
+def classify(kept):  # kept=None -> keep all
+    last_root = {}; last_ts = {}
+    intra = cross = new = 0
+    rec_i = [0] * len(GAP_EDGES); rec_c = [0] * len(GAP_EDGES)
+    for (ts, cid, hs), r in zip(records, roots):
+        if kept is not None and r not in kept:
+            continue
+        for h in hs:
+            lr = last_root.get(h)
+            if lr is None:
+                new += 1
+            else:
+                gb = gbucket(max(0.0, ts - last_ts[h]))
+                if lr == r:
+                    intra += 1; rec_i[gb] += 1
+                else:
+                    cross += 1; rec_c[gb] += 1
+            last_root[h] = r; last_ts[h] = ts
+    return intra, cross, new, rec_i, rec_c
+
+def cum_le(rec, idx):   # cumulative fraction with gap-bucket <= idx
+    tot = sum(rec) or 1
+    return sum(rec[: idx + 1]) / tot
+
+targets = [("full", None), (0.75, None), (0.50, None),
+           (0.25, None), (0.10, None), (0.00, None)]
+rows = []
+for label, _ in targets:
+    if label == "full":
+        kept = None
+        f_actual = len(single_roots) / (len(single_roots) + M)
+    else:
+        f = float(label)
+        S = min(len(single_roots), int(round(M * f / (1 - f)))) if f < 1 else len(single_roots)
+        keep_single = set(random.sample(single_roots, S)) if S < len(single_roots) else set(single_roots)
+        kept = set(multi_roots) | keep_single
+        f_actual = S / (S + M)
+    intra, cross, new, rec_i, rec_c = classify(kept)
+    reuse = intra + cross
+    n_sess = (len(single_roots) + M) if kept is None else len(kept)
+    row = {
+        "target": label, "single_turn_frac": round(f_actual, 4), "n_sessions": n_sess,
+        "new": new, "intra": intra, "cross": cross, "reuse": reuse,
+        "intra_frac_of_reuse": round(intra / reuse, 4),
+        "cross_frac_of_reuse": round(cross / reuse, 4),
+        "intra_le60s": round(cum_le(rec_i, 2), 4),
+        "cross_le60s": round(cum_le(rec_c, 2), 4),
+    }
+    rows.append(row)
+    sys.stderr.write(f"[{time.time()-t0:.0f}s] f={row['single_turn_frac']}: "
+                     f"intra={row['intra_frac_of_reuse']} cross={row['cross_frac_of_reuse']}\n")
+
+json.dump({"rows": rows, "n_single": len(single_roots), "n_multi": M}, open(OUT, "w"), indent=2)
+print(f"{'single-turn%':>12} {'sessions':>10} {'intra%':>8} {'cross%':>8} {'intra<=60s':>11} {'cross<=60s':>11}")
+for r in rows:
+    print(f"{r['single_turn_frac']*100:>11.1f}% {r['n_sessions']:>10} "
+          f"{r['intra_frac_of_reuse']*100:>7.1f}% {r['cross_frac_of_reuse']*100:>7.1f}% "
+          f"{r['intra_le60s']*100:>10.1f}% {r['cross_le60s']*100:>10.1f}%")

paper/data/f2a_mixture_sweep_result.json

@@ -0,0 +1,84 @@
+{
+  "rows": [
+    {
+      "target": "full",
+      "single_turn_frac": 0.9026,
+      "n_sessions": 1307276,
+      "new": 20650883,
+      "intra": 65166144,
+      "cross": 54134925,
+      "reuse": 119301069,
+      "intra_frac_of_reuse": 0.5462,
+      "cross_frac_of_reuse": 0.4538,
+      "intra_le60s": 0.8865,
+      "cross_le60s": 0.8706
+    },
+    {
+      "target": 0.75,
+      "single_turn_frac": 0.75,
+      "n_sessions": 509144,
+      "new": 15446415,
+      "intra": 66081759,
+      "cross": 26932604,
+      "reuse": 93014363,
+      "intra_frac_of_reuse": 0.7104,
+      "cross_frac_of_reuse": 0.2896,
+      "intra_le60s": 0.8844,
+      "cross_le60s": 0.8568
+    },
+    {
+      "target": 0.5,
+      "single_turn_frac": 0.5,
+      "n_sessions": 254572,
+      "new": 12843712,
+      "intra": 66548474,
+      "cross": 18990485,
+      "reuse": 85538959,
+      "intra_frac_of_reuse": 0.778,
+      "cross_frac_of_reuse": 0.222,
+      "intra_le60s": 0.8832,
+      "cross_le60s": 0.8881
+    },
+    {
+      "target": 0.25,
+      "single_turn_frac": 0.25,
+      "n_sessions": 169715,
+      "new": 11553493,
+      "intra": 66732961,
+      "cross": 16726772,
+      "reuse": 83459733,
+      "intra_frac_of_reuse": 0.7996,
+      "cross_frac_of_reuse": 0.2004,
+      "intra_le60s": 0.8827,
+      "cross_le60s": 0.9087
+    },
+    {
+      "target": 0.1,
+      "single_turn_frac": 0.1,
+      "n_sessions": 141429,
+      "new": 11036894,
+      "intra": 66798704,
+      "cross": 16084035,
+      "reuse": 82882739,
+      "intra_frac_of_reuse": 0.8059,
+      "cross_frac_of_reuse": 0.1941,
+      "intra_le60s": 0.8826,
+      "cross_le60s": 0.9152
+    },
+    {
+      "target": 0.0,
+      "single_turn_frac": 0.0,
+      "n_sessions": 127286,
+      "new": 10724167,
+      "intra": 66834552,
+      "cross": 15799085,
+      "reuse": 82633637,
+      "intra_frac_of_reuse": 0.8088,
+      "cross_frac_of_reuse": 0.1912,
+      "intra_le60s": 0.8825,
+      "cross_le60s": 0.9184
+    }
+  ],
+  "n_single": 1179990,
+  "n_multi": 127286
+}

paper/data/f2a_reuse_topology_analyze.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""
+f2a reuse topology — full-trace, infinite-KV-cache decomposition (LRU semantics).
+
+Question: on the real 2h cluster trace, assuming an *infinite* KV cache (nothing
+ever evicted), where do prefix-cache REUSE HITS come from?
+
+We classify only reuse hits (the 1st occurrence of a block is `new` = irreducible
+prefill; it is reported only as context for the APC ceiling, not in the split).
+
+A block (content-addressed `hash_id`) processed in timestamp order. For each hit we
+look at the block's **most recent prior holder** (last computed OR used = LRU):
+
+  intra : last touch was the SAME session (parent_chat_id chain)
+  cross : last touch was a DIFFERENT session
+
+After classifying, the block's last-holder / last-time are updated to the current
+request (LRU refresh). The reuse "recency" is the **LRU reuse distance** = time since
+the block was last touched (what a finite TTL/LRU cache would need to retain).
+
+`cross` is further resolved by *block popularity* = number of distinct sessions that
+ever touch the block: a handful of hugely-popular blocks are the shared system/tool
+prefix; low-popularity cross blocks are genuine cross-session content.
+
+Run on dash2 (trace lives there):
+    python3 f2a_reuse_topology_analyze.py \
+        ~/ali-trace/trace-glm5.1-formatted/051315-051317.jsonl /tmp/f2a_result.json
+"""
+import sys, json, time
+from collections import defaultdict
+
+PATH = sys.argv[1]
+OUT = sys.argv[2] if len(sys.argv) > 2 else "/tmp/f2a_result.json"
+POP_CAP = 4096  # cap per-block root set; >= this is "very shared", buckets unaffected
+
+t0 = time.time()
+chat_parent = {}
+records = []          # (ts, chat_id, hash_ids)
+total_input_tokens = 0
+total_blocks = 0
+turn1 = 0
+n = 0
+with open(PATH) as f:
+    for line in f:
+        d = json.loads(line)
+        cid = d["chat_id"]
+        pc = d.get("parent_chat_id")
+        chat_parent[cid] = 0 if pc is None else pc
+        hs = d.get("hash_ids") or []
+        records.append((d.get("timestamp", 0.0), cid, hs))
+        total_input_tokens += d.get("input_length", 0) or 0
+        total_blocks += len(hs)
+        if (d.get("turn", 1) or 1) == 1:
+            turn1 += 1
+        n += 1
+sys.stderr.write(f"[{time.time()-t0:.0f}s] loaded {n} reqs, {total_blocks} block-occ\n")
+
+# resolve session root by following parent_chat_id to turn-1 / out-of-window head
+root_cache = {}
+def resolve_root(cid):
+    chain = []
+    cur = cid
+    while True:
+        if cur in root_cache:
+            r = root_cache[cur]; break
+        p = chat_parent.get(cur, 0)
+        if p == 0 or p not in chat_parent:
+            r = cur; break
+        chain.append(cur); cur = p
+        if len(chain) > 100000:
+            r = cur; break
+    for nd in chain:
+        root_cache[nd] = r
+    root_cache[cid] = r
+    return r
+
+records.sort(key=lambda r: r[0])
+sys.stderr.write(f"[{time.time()-t0:.0f}s] sorted by ts\n")
+
+last_root = {}                  # block -> root of MOST RECENT holder (LRU)
+last_ts = {}                    # block -> ts of most recent touch (LRU)
+roots_of = defaultdict(set)     # block -> set of distinct roots (capped) = popularity
+intra_cnt = defaultdict(int)    # block -> intra reuse hits
+cross_cnt = defaultdict(int)    # block -> cross reuse hits
+new = intra = cross = 0
+
+# LRU reuse distance of each hit: gap = consumer_ts - last_touch_ts
+GAP_EDGES = [1, 10, 60, 300, 1800, 3600, float("inf")]   # seconds
+GAP_LABELS = ["<1s", "1-10s", "10-60s", "1-5min", "5-30min", "30-60min", ">60min"]
+rec_intra = [0] * len(GAP_EDGES)
+rec_cross = [0] * len(GAP_EDGES)
+def gap_bucket(g):
+    for i, e in enumerate(GAP_EDGES):
+        if g < e:
+            return i
+    return len(GAP_EDGES) - 1
+
+for ts, cid, hs in records:
+    if not hs:
+        continue
+    r = resolve_root(cid)
+    for h in hs:
+        lr = last_root.get(h)
+        if lr is None:
+            new += 1                       # first compute: not a hit
+        else:
+            gb = gap_bucket(max(0.0, ts - last_ts[h]))
+            if lr == r:
+                intra += 1; intra_cnt[h] += 1; rec_intra[gb] += 1
+            else:
+                cross += 1; cross_cnt[h] += 1; rec_cross[gb] += 1
+        last_root[h] = r                   # LRU refresh: now held by current session
+        last_ts[h] = ts
+        s = roots_of[h]
+        if len(s) < POP_CAP:
+            s.add(r)
+sys.stderr.write(f"[{time.time()-t0:.0f}s] classified: new={new} intra={intra} cross={cross}\n")
+
+# popularity buckets: distinct sessions touching a block
+POP_EDGES = [2, 10, 100, 1000, float("inf")]
+POP_LABELS = ["1 (private)", "2-9", "10-99", "100-999", ">=1000"]
+def pop_bucket(p):
+    if p <= 1:
+        return 0
+    for i, e in enumerate(POP_EDGES[1:], start=1):
+        if p < e:
+            return i
+    return len(POP_LABELS) - 1
+pop_blocks = [0] * len(POP_LABELS)
+pop_intra = [0] * len(POP_LABELS)
+pop_cross = [0] * len(POP_LABELS)
+for h in last_root:
+    p = len(roots_of[h])
+    b = pop_bucket(p)
+    pop_blocks[b] += 1
+    pop_intra[b] += intra_cnt.get(h, 0)
+    pop_cross[b] += cross_cnt.get(h, 0)
+
+eff_blk = total_input_tokens / total_blocks if total_blocks else 0.0
+total_occ = new + intra + cross
+reuse = intra + cross
+result = {
+    "trace": PATH,
+    "semantics": "LRU last-touched; reuse-hits only (new excluded from split)",
+    "n_requests": n,
+    "n_sessions": len(set(resolve_root(c) for c in chat_parent)),
+    "turn1_frac": turn1 / n,
+    "block_size_tokens_eff": eff_blk,
+    "total_input_tokens": total_input_tokens,
+    "total_block_occ": total_occ,
+    "distinct_blocks": len(last_root),
+    "new_occ": new,                         # context only
+    "apc_ceiling": reuse / total_occ,       # context only
+    # REUSE-ONLY decomposition (the headline)
+    "reuse_total": reuse,
+    "reuse": {"intra": intra, "cross": cross},
+    "reuse_frac": {"intra": intra / reuse, "cross": cross / reuse},
+    # cross resolved by popularity (over reuse hits)
+    "pop_labels": POP_LABELS,
+    "pop_blocks": pop_blocks,
+    "pop_intra": pop_intra,
+    "pop_cross": pop_cross,
+    # LRU reuse-distance recency (over reuse hits)
+    "gap_labels": GAP_LABELS,
+    "rec_intra": rec_intra,
+    "rec_cross": rec_cross,
+}
+with open(OUT, "w") as f:
+    json.dump(result, f, indent=2)
+sys.stderr.write(f"[{time.time()-t0:.0f}s] wrote {OUT}\n")
+
+# human summary
+print(json.dumps({k: result[k] for k in
+      ("n_requests","n_sessions","distinct_blocks","reuse_total",
+       "reuse_frac","apc_ceiling")}, indent=2))
+print(f"new(context)={new}  intra={intra}  cross={cross}")
+print("popularity   blocks / intra-hits / cross-hits:")
+for i, lab in enumerate(POP_LABELS):
+    print(f"  {lab:>12}: {pop_blocks[i]:>10} | {pop_intra[i]:>11} | {pop_cross[i]:>11}")
+print("LRU reuse-distance   intra / cross:")
+for i, lab in enumerate(GAP_LABELS):
+    print(f"  {lab:>8}: {rec_intra[i]:>11} | {rec_cross[i]:>11}")

paper/data/f2a_reuse_topology_result.json

@@ -0,0 +1,77 @@
+{
+  "trace": "051315-051317.jsonl",
+  "semantics": "LRU last-touched; reuse-hits only (new excluded from split)",
+  "n_requests": 2114220,
+  "n_sessions": 1307276,
+  "turn1_frac": 0.6183254344391785,
+  "block_size_tokens_eff": 508.1517503092776,
+  "total_input_tokens": 71116829368,
+  "total_block_occ": 139951952,
+  "distinct_blocks": 20650883,
+  "new_occ": 20650883,
+  "apc_ceiling": 0.8524430513123532,
+  "reuse_total": 119301069,
+  "reuse": {
+    "intra": 65166144,
+    "cross": 54134925
+  },
+  "reuse_frac": {
+    "intra": 0.5462326913432771,
+    "cross": 0.45376730865672293
+  },
+  "pop_labels": [
+    "1 (private)",
+    "2-9",
+    "10-99",
+    "100-999",
+    ">=1000"
+  ],
+  "pop_blocks": [
+    14581108,
+    5535433,
+    517069,
+    16153,
+    1120
+  ],
+  "pop_intra": [
+    44515497,
+    14288480,
+    5421050,
+    924419,
+    16698
+  ],
+  "pop_cross": [
+    0,
+    20230912,
+    13750153,
+    7689338,
+    12464522
+  ],
+  "gap_labels": [
+    "<1s",
+    "1-10s",
+    "10-60s",
+    "1-5min",
+    "5-30min",
+    "30-60min",
+    ">60min"
+  ],
+  "rec_intra": [
+    390952,
+    26060293,
+    31317556,
+    5877221,
+    1384772,
+    109673,
+    25677
+  ],
+  "rec_cross": [
+    13222875,
+    22254795,
+    11653445,
+    4965765,
+    1747487,
+    220816,
+    69742
+  ]
+}

replayer/__main__.py

@@ -30,12 +30,23 @@ def main() -> None:
                    default=float(_env_think) if _env_think else None,
                    help="Closed-loop think-time (s) after each turn completes; "
                         "ignore absolute trace schedule. Env: REPLAY_INTER_TURN_THINK_S")
+    p.add_argument("--no-realized-prefix",
+                   action="store_true",
+                   default=bool(os.environ.get("REPLAY_NO_REALIZED_PREFIX")),
+                   help="Controlled-reuse mode: prompt = hash-built tokens only "
+                        "(reuse set by hash_ids). Env: REPLAY_NO_REALIZED_PREFIX")
     p.add_argument("--dispatch-mode", choices=["tracets", "thinktime"],
                    default=os.environ.get("REPLAY_DISPATCH_MODE", "tracets"),
                    help="tracets (Mode 1): absolute trace ts = max(prev_finished, ts). "
                         "thinktime (Mode 2): turn-k at prev_finished + "
                         "time_to_parent_chat. Env: REPLAY_DISPATCH_MODE")
     p.add_argument("--request-timeout", type=float, default=600.0)
+    _env_maxdur = os.environ.get("REPLAY_MAX_DURATION")
+    p.add_argument("--max-duration", type=float,
+                   default=float(_env_maxdur) if _env_maxdur else None,
+                   help="Overall wall-clock deadline (s): cancel in-flight + write "
+                        "summary (un-run turns counted as failures) to bound a "
+                        "collapsed config's drain. Env: REPLAY_MAX_DURATION")
     p.add_argument("--request-limit", type=int, default=None,
                    help="Limit number of requests to replay")
     p.add_argument("-v", "--verbose", action="store_true")
@@ -56,7 +67,9 @@ def main() -> None:
         request_limit=args.request_limit,
         max_inflight_sessions=args.max_inflight_sessions,
         inter_turn_think_s=args.inter_turn_think,
+        no_realized_prefix=args.no_realized_prefix,
         dispatch_mode=args.dispatch_mode,
+        max_duration_s=args.max_duration,
     )
 
     results = asyncio.run(replay_trace(config))

replayer/replay.py

@@ -66,6 +66,13 @@ class ReplayConfig:
     # max_inflight_sessions=N this is a stable N-user closed-loop (no open-loop
     # runaway), so it removes the "immediate retrigger under load" artifact.
     inter_turn_think_s: float | None = None
+    # Controlled-reuse mode: skip _apply_realized_prefix so each turn's prompt is
+    # exactly the hash-built tokens. Then prefix-cache reuse is governed solely by
+    # the generated hash_ids (shared prefix blocks hit, fresh delta blocks miss) —
+    # required for the reuse-fraction sweep, where realized-prefix would otherwise
+    # force every fixed-length turn to ≈ the prior turn (≈100% reuse regardless).
+    # Keep OFF (realized-prefix ON) for the real agentic trace.
+    no_realized_prefix: bool = False
     # Dispatch timing for intra-session turns:
     #  "tracets"   (Mode 1): fire at absolute trace timestamp -> effectively
     #              max(prev_finished, trace_ts); collapses think-time to 0 when
@@ -73,6 +80,25 @@ class ReplayConfig:
     #  "thinktime" (Mode 2): turn-1 at trace arrival; turn-k at
     #              prev_finished + time_to_parent_chat (real production gap).
     dispatch_mode: str = "tracets"
+    # Overall wall-clock deadline for the whole replay (seconds). When exceeded,
+    # stop awaiting in-flight sessions, cancel them, and write the summary over
+    # whatever completed — un-run turns are counted as failures so completion%
+    # stays honest (request_count == full trace). None = no deadline (default,
+    # original behavior unchanged). Used to bound the slow drain of a collapsed
+    # config in a sweep. Env: REPLAY_MAX_DURATION.
+    max_duration_s: float | None = None
+
+
+def _skipped_metric() -> "RequestMetrics":
+    """Placeholder failure row for a turn never run due to a max_duration cutoff.
+    Only its error (non-None) matters: it counts toward request/error totals but
+    is excluded from latency/ttft/tpot percentiles (successes only)."""
+    return RequestMetrics(
+        request_id="deadline_skipped", session_id="", turn_id=-1,
+        trace_timestamp_s=0.0, input_length=0, output_length=0,
+        request_type="skipped", effective_input_length=None, cached_tokens=0,
+        latency_s=None, ttft_s=None, tpot_s=None, error="deadline_skipped",
+    )
 
 
 def _build_prompt_token_ids(req: TraceRequest) -> list[int]:
@@ -318,10 +344,9 @@ async def _run_session(
                 if elapsed < target_wall:
                     await asyncio.sleep(target_wall - elapsed)
 
-            token_ids = _apply_realized_prefix(
-                _build_prompt_token_ids(req),
-                realized_context,
-            )
+            token_ids = _build_prompt_token_ids(req)
+            if not config.no_realized_prefix:
+                token_ids = _apply_realized_prefix(token_ids, realized_context)
             result = await _dispatch_request(
                 client=client, config=config, req=req,
                 prompt_token_ids=token_ids, sem=request_sem,
@@ -410,25 +435,44 @@ async def replay_trace(config: ReplayConfig) -> list[RequestMetrics]:
             trust_env=False,
             limits=limits,
         ) as client:
+            states = [_SessionState(session_id=sid, turns=turns)
+                      for sid, turns in sessions]
             tasks = [
                 asyncio.create_task(_run_session(
-                    state=_SessionState(session_id=sid, turns=turns),
-                    config=config, client=client,
+                    state=st, config=config, client=client,
                     request_sem=request_sem,
                     earliest_ts=earliest_ts, sweep_start=sweep_start,
                     sink=sink,
                     session_sem=session_sem,
                 ))
-                for sid, turns in sessions
+                for st in states
             ]
-            all_results = await asyncio.gather(*tasks)
+            if config.max_duration_s and config.max_duration_s > 0:
+                _done, pending = await asyncio.wait(
+                    tasks, timeout=config.max_duration_s)
+                if pending:
+                    logger.warning(
+                        "max_duration %.0fs reached: cancelling %d in-flight "
+                        "session(s); un-run turns counted as failures",
+                        config.max_duration_s, len(pending))
+                    for t in pending:
+                        t.cancel()
+                    await asyncio.gather(*pending, return_exceptions=True)
+            else:
+                await asyncio.gather(*tasks)
     finally:
         sink.close()
 
     sweep_elapsed = time.perf_counter() - sweep_start
     post_metrics = await _snapshot_prefix_cache_metrics(config.endpoint_url)
 
-    flat = [m for group in all_results for m in group]
+    # Build from the session states (identical to the gather return in the
+    # uncapped path) so partially-completed (cancelled) sessions still contribute
+    # their finished turns; pad un-run turns as failures so request_count == trace.
+    flat = [m for st in states for m in st.metrics]
+    missing = n_requests - len(flat)
+    if missing > 0:
+        flat.extend(_skipped_metric() for _ in range(missing))
     summary_path = config.output_path.with_suffix(".summary.json")
     write_summary_json(summary_path, flat)
 

v2/exp_d_policy_dispatch/README.md

@@ -27,6 +27,13 @@ Analyzer: `scripts/bench_report.py` (summaries in `results/`).
 
 ## Result (ms; `figs/exp_d_policy_dispatch.png`)
 
+The figure has 6 panels — TTFT mean/p90, E2E mean/p90, decode TPS (output
+goodput), and the per-worker GPU-util box (8 workers/arm). Decode TPS is the
+honest throughput metric (total/prefill TPS is inflated by cache-miss recompute,
+e.g. LMetric); thinktime ≥ tracets on it everywhere (the system drains faster
+with real think-time). The GPU-util box shows LPWL also keeps the tightest
+worker balance.
+
 | policy | mode | TTFT p90 | E2E mean | E2E p90 | E2E p99 | TPOT p90 | APC | req-bal |
 |---|---|---:|---:|---:|---:|---:|---:|---:|
 | **LPWL** | tracets | 11099 | 9827 | 25366 | 93929 | 33 | 0.650 | **1.49×** |

v2/exp_d_policy_dispatch/plot.py

@@ -1,8 +1,9 @@
 """exp (d): 5-policy routing under tracets vs thinktime dispatch.
 
-Shows the ranking FLIP: under the faithful `thinktime` load the parameter-free
-LPWL (leastwork) is the clear winner, but under `tracets` (think-collapse bursts)
-its advantage disappears (it ties unified_ab on TTFT p90 and *loses* on E2E mean).
+Six panels: TTFT mean/p90, E2E mean/p90, decode-TPS (output goodput), and the
+per-worker GPU-util distribution. Shows the ranking FLIP — under faithful
+`thinktime` the parameter-free LPWL (leastwork) is the clear winner; under
+`tracets` (think-collapse bursts) its advantage disappears.
 
 Reads the two bench_report summaries; writes v2/figs/exp_d_policy_dispatch.png.
 Usage: python v2/exp_d_policy_dispatch/plot.py
@@ -13,56 +14,78 @@ import os
 import matplotlib
 matplotlib.use("Agg")
 import matplotlib.pyplot as plt
+from matplotlib.patches import Patch
 
 HERE = os.path.dirname(__file__)
 TC = json.load(open(os.path.join(HERE, "results/tracets.json")))
 TT = json.load(open(os.path.join(HERE, "results/thinktime.json")))
 
-# canonical order: LPWL first; pretty labels
 ARMS = ["leastwork", "unified_ab", "unified_def", "lmetric", "sticky"]
 LABEL = {"leastwork": "LPWL\n(leastwork)", "unified_ab": "unified\n+A+B",
          "unified_def": "unified\ndefault", "lmetric": "LMetric", "sticky": "sticky"}
 C_TC, C_TT = "#d62728", "#2ca02c"  # tracets red / thinktime green (match exp_c)
+W = 0.38
 
 
-def panel(ax, key, sub, title, ylab):
-    tc = [TC[a][key][sub] / 1000.0 for a in ARMS]   # ms -> s
-    tt = [TT[a][key][sub] / 1000.0 for a in ARMS]
+def bar_panel(ax, tc, tt, title, ylab, fmt="{:.1f}", higher_better=False):
     x = range(len(ARMS))
-    w = 0.38
-    b1 = ax.bar([i - w / 2 for i in x], tc, w, label="tracets (burst)", color=C_TC)
-    b2 = ax.bar([i + w / 2 for i in x], tt, w, label="thinktime (faithful)", color=C_TT)
+    b1 = ax.bar([i - W / 2 for i in x], tc, W, color=C_TC)
+    b2 = ax.bar([i + W / 2 for i in x], tt, W, color=C_TT)
     for bars in (b1, b2):
         for r in bars:
             ax.text(r.get_x() + r.get_width() / 2, r.get_height(),
-                    f"{r.get_height():.1f}", ha="center", va="bottom", fontsize=8)
-    ax.set_xticks(list(x)); ax.set_xticklabels([LABEL[a] for a in ARMS], fontsize=9)
-    ax.set_ylabel(ylab); ax.set_title(title, fontsize=11)
+                    fmt.format(r.get_height()), ha="center", va="bottom", fontsize=7.5)
+    ax.set_xticks(list(x)); ax.set_xticklabels([LABEL[a] for a in ARMS], fontsize=8.5)
+    arrow = "higher = better" if higher_better else "lower = better"
+    ax.set_ylabel(ylab); ax.set_title(f"{title}  ({arrow})", fontsize=10.5)
+    ax.grid(axis="y", alpha=.3); ax.set_ylim(0, max(tc + tt) * 1.18)
+
+
+def gpu_panel(ax):
+    """Per-worker gpu_util_mean distribution: tracets vs thinktime box per policy."""
+    def utils(D, a):
+        pw = D[a]["per_worker"]
+        return [pw[w]["gpu_util_mean"] for w in sorted(pw, key=int)
+                if pw[w].get("gpu_util_mean") is not None]
+    for i, a in enumerate(ARMS):
+        for D, off, c in [(TC, -W / 2, C_TC), (TT, +W / 2, C_TT)]:
+            bp = ax.boxplot([utils(D, a)], positions=[i + off], widths=0.30,
+                            patch_artist=True, showfliers=False,
+                            medianprops=dict(color="black"))
+            bp["boxes"][0].set(facecolor=c, alpha=.65)
+    ax.set_xticks(range(len(ARMS)))
+    ax.set_xticklabels([LABEL[a] for a in ARMS], fontsize=8.5)
+    ax.set_ylabel("per-worker GPU util %"); ax.set_ylim(0, 100)
+    ax.set_title("Per-worker GPU util (box = 8 workers; tighter = balanced)", fontsize=10.5)
     ax.grid(axis="y", alpha=.3)
-    ax.set_ylim(0, max(tc + tt) * 1.18)
-    # mark LPWL-thinktime as the winner (lowest green) in each panel
-    ax.annotate("LPWL wins\nunder thinktime", xy=(0 + w / 2, tt[0]),
-                xytext=(0.9, max(tc + tt) * 0.86), fontsize=8.5, color=C_TT,
-                ha="left", arrowprops=dict(arrowstyle="->", color=C_TT, lw=1.3))
-    return b1, b2
 
 
-fig, (axL, axR) = plt.subplots(1, 2, figsize=(11.2, 4.6))
-panel(axL, "ttft_ms", "p90", "TTFT p90 (lower = better)", "TTFT p90 (s)")
-panel(axR, "e2e_ms", "mean", "E2E mean (lower = better)", "E2E mean (s)")
-axL.legend(loc="upper left", fontsize=9)
-fig.suptitle("5-policy routing: dispatch mode flips the ranking — "
-             "LPWL is best under faithful thinktime, only ties/loses under tracets bursts",
-             fontsize=11.5)
-fig.tight_layout(rect=(0, 0, 1, 0.95))
+def col(D, key, sub, scale=1.0):
+    return [D[a][key][sub] * scale for a in ARMS]
+
+
+fig, ax = plt.subplots(2, 3, figsize=(15.5, 8.6))
+bar_panel(ax[0, 0], col(TC, "ttft_ms", "mean", 1e-3), col(TT, "ttft_ms", "mean", 1e-3),
+          "TTFT mean", "s")
+bar_panel(ax[0, 1], col(TC, "ttft_ms", "p90", 1e-3), col(TT, "ttft_ms", "p90", 1e-3),
+          "TTFT p90", "s")
+bar_panel(ax[0, 2],
+          [TC[a]["throughput"]["decode_tps"] for a in ARMS],
+          [TT[a]["throughput"]["decode_tps"] for a in ARMS],
+          "Decode TPS (output goodput)", "tok/s", fmt="{:.0f}", higher_better=True)
+bar_panel(ax[1, 0], col(TC, "e2e_ms", "mean", 1e-3), col(TT, "e2e_ms", "mean", 1e-3),
+          "E2E mean", "s")
+bar_panel(ax[1, 1], col(TC, "e2e_ms", "p90", 1e-3), col(TT, "e2e_ms", "p90", 1e-3),
+          "E2E p90", "s")
+gpu_panel(ax[1, 2])
+
+fig.legend(handles=[Patch(facecolor=C_TC, label="tracets (burst artifact)"),
+                    Patch(facecolor=C_TT, label="thinktime (faithful load)")],
+           loc="lower center", ncol=2, fontsize=10.5, bbox_to_anchor=(0.5, 0.0))
+fig.suptitle("5-policy routing: tracets vs thinktime (807 reqs, dash0 8xH20) — "
+             "LPWL wins across the board under faithful thinktime",
+             fontsize=12.5)
+fig.tight_layout(rect=(0, 0.035, 1, 0.96))
 out = os.path.join(HERE, "..", "figs", "exp_d_policy_dispatch.png")
 fig.savefig(out, dpi=140)
 print("wrote", os.path.normpath(out))
-
-# also print the deltas the README cites
-print("\npolicy        TTFTp90 tc->tt    E2Emean tc->tt")
-for a in ARMS:
-    t1, t2 = TC[a]["ttft_ms"]["p90"], TT[a]["ttft_ms"]["p90"]
-    e1, e2 = TC[a]["e2e_ms"]["mean"], TT[a]["e2e_ms"]["mean"]
-    print(f"{a:<13} {t1/1000:5.1f}->{t2/1000:4.1f}s ({(t2-t1)/t1:+.0%})   "
-          f"{e1/1000:5.1f}->{e2/1000:4.1f}s ({(e2-e1)/e1:+.0%})")