Working-set sizing tool + GLM-5.1-FP8/B300 result

Configurable KV working-set analyzer (GPU model x TP/PP/EP x model
config.json with MLA/GQA auto x KV/weight dtype). Computes Denning W(T),
oracle [first,last], and retain-forever footprints vs a per-replica KV
pool, plus the APC captured at each retention window.

GLM-5.1-FP8 (MLA, 43.9 KiB/token) on 1x B300 node (1528 GB KV pool):
live KV fits trivially (~533 GB), but the full 80.4% APC ceiling needs
~14 nodes (oracle) -> long-tail reuse motivates DRAM offload, not HBM.

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

analysis/working_set/README.md

@@ -0,0 +1,67 @@
+# KV-cache Working-Set Sizing — GLM-5.1-FP8 · TP=8 · 1× B300 node
+
+工具：`scripts/working_set_analysis.py`（可配置 GPU 型号 / 并行度 TP·PP·EP / 模型 config.json /
+KV dtype / 权重大小）。图：`figs/working_set/glm5_fp8_tp8_b300.png`。
+
+## 复现
+
+```bash
+.venv/bin/python scripts/working_set_analysis.py \
+  /home/gahow/phd/kvcache-simulator/bailian-traces/glm_coder_blksz_512_040915-040917.jsonl \
+  --model-config /home/gahow/phd/kvcache-simulator/models/GLM-5/config.json \
+  --gpu B300 --tp 8 --ep 8 --kv-dtype-bytes 1 --weight-gb 744 --min-ts 0 \
+  --out figs/working_set/glm5_fp8_tp8_b300.png
+```
+
+## 方法
+
+`hash_ids` 是全局内容寻址 block id（同内容=同 id，复用=同 id 再现）。vLLM prefix cache 是
+block 级，所以**集群级 KV footprint = 任一时刻必须常驻的 distinct block 数**，与 placement 无关
+（affinity 只搬运 block，不改总量）。三种 working set：
+- `W_all` 永不淘汰（真上界）
+- `W_oracle` 每 block 只在 `[首次, 末次复用]` 常驻（Belady 完美预知 → 满 APC 上界的最小 HBM）
+- `W_denning(T)` 滑窗 T 内被访问的 distinct block（现实 TTL-LRU）
+
+KV/token：MLA → `L·(kv_lora_rank+qk_rope_head_dim)·dtype`；GQA → `2·L·kv_heads·head_dim·dtype`
+（与 `kvcache-simulator/src/config.rs::kv_block_bytes` 一致）。
+
+## 配置
+
+| 项 | 值 |
+|---|---|
+| 模型 | GLM-5.1-FP8（MLA, L=78, kv_lora=512+rope=64） |
+| KV/token · KV/block(512) | **43.9 KiB** · **23.0 MB**（≈ Qwen3 GQA 96 KiB 的一半） |
+| 硬件 | 8× B300 (288 GB) = 2304 GB HBM/replica |
+| 预算 | FP8 权重 744 GB + act 32 GB → **KV pool = 1528 GB/node** |
+| trace | dash0 glm_coder，475k req，**1.25h active @ 106 QPS**，~40k tok/req（剔除 77 条负 ts 暖机） |
+| APC 上界 | **80.4%** |
+
+## 结果
+
+| 保留窗口 T | peak footprint | = 节点 (GPU) | APC@T |
+|---:|---:|---:|---:|
+| 2s（在飞下限）| 533 GB | 0.3 (3) | 1.7% |
+| 10s | 2,068 GB | 1.4 (11) | 15% |
+| 30s | 4,906 GB | 3.2 (26) | 42% |
+| 60s | 7,698 GB | 5.0 (40) | 56% |
+| 300s | 21,960 GB | 14.4 (115) | 74% |
+| **oracle（满 80.4%）** | **21,399 GB** | **14.0 (112)** | 80.4% |
+| retain-forever | 167,018 GB | 109 (874) | — |
+
+## 结论
+
+1. **Serving：1 节点绰绰有余。** 在飞 KV（τ≈2-5s）仅 533–1157 GB ≪ 单节点 1528 GB。
+   MLA + B300 大 HBM 让 live footprint 微不足道——跑起来根本不缺显存。
+2. **缓存全部复用（80.4%）：1 节点差 ~14×。** oracle 下限 21.4 TB = 14 节点（112 GPU），
+   真实 LRU ~2× → ~28 节点。单节点（1528 GB）只能 hold ~10s 窗口 → cache 侧 APC 仅 ~10-15%。
+   要 ~56% 需 5 节点，~74% 需 ~14 节点。
+3. **瓶颈在长尾，不在 live。** 把 APC 50%→80% 装进 GPU HBM 要 5→14 节点，极不经济
+   → offload/migration 到 CPU DRAM（每节点 ~1.5 TB）是定量动机。与 Qwen 结论方向一致。
+
+## 注意
+
+- footprint 是 TTL-LRU（最浪费）+ shared-cache 下限：真实 capacity-LRU 同容量下 APC 更高，
+  但分区/affinity 不均衡又抬高需求；oracle / retain-forever 给出下/上界。
+- GLM trace mean ~40k tok/req，是 Qwen trace（11k）的 ~3.5×（tokenizer + 抽取不同），
+  **绝对 GB 不可跨模型横比**，方法与定性结论可比。
+- EP 不改变 KV 总量（只影响 expert 权重分布），`--ep` 仅作标注。

figs/mb5/mb5_summary.csv

@@ -0,0 +1,5 @@
+config,rep,n_requests,n_success,wall_clock_s,peak_pool_frac,steady_pool_frac,p_pool_peak_frac,p_pool_steady_frac,d_pool_peak_frac,d_pool_steady_frac,peak_waiting,latency_p50_s,latency_p90_s,latency_p99_s,ttft_p50_s,ttft_p90_s,ttft_p99_s,prefix_cache_hit_ratio
+8C,1,1214,1214,2994.218414353032,0.7174957362137578,0.3439702956225128,,,,,29,10.82550932947197,83.34998885790122,194.10265863158946,6.967104309005663,53.12018221841427,114.12611859919207,0.1937163528742694
+6P+2D,1,1214,1214,3419.065942236979,0.7726478112563957,0.42145750426378625,0.743272692817889,0.3082291074474133,0.9959636156907333,0.7434906196702672,128,44.48975181748392,91.82252187062406,147.70196208347772,40.95952733900049,86.68752026481089,142.84028979733685,0.0
+4P+4D,1,1214,1214,4170.666486939997,0.6997939169982945,0.45876918703808983,0.6438459351904491,0.28540363843092664,0.9753411028993746,0.5977686185332576,152,59.52004547297838,157.08703426021387,224.03997302683115,56.419772224500775,153.07864206891392,219.73412787001706,0.0
+2P+6D,1,1214,109,5761.816568834998,0.9698692438885731,0.9435119386014781,0.9969869243888573,0.9198408186469585,0.9620238772029562,0.9494504453287853,872,26.293884326005355,499.3484142678091,577.7122636228032,23.580788671970367,498.0334587502061,576.5306194114453,0.0

microbench/fresh_setup/PD_DISAGG_INVESTIGATION.md

@@ -10,6 +10,27 @@ Living TODO 文档。记 H1/H2/H3/H4 四条假设的实验状态，以及每个
 
 **核心 gap**：我们只看到 "PD-disagg 表面性能差 10×" 的 headline 数字，但**没有 system-level breakdown 告诉 reviewer "为什么差，差在哪个组件"**。需要的是 D-pool occupancy / scheduler queue depth / KV transfer queue / GPU SM utilization 等系统指标的时间序列，能直接指出 bottleneck。
 
+### Progress（2026-05-27 16:50）
+
+**Phase 0 done**：MB5 pipeline 全部 standing up on dash1 fresh-venv:
+
+- `mb5_launch.sh`：8C / 6P+2D / 4P+4D / 2P+6D 单一 launcher；stop_all 包含 stale-port 守卫
+- `mb5_pd_proxy.py`：vendored copy of vLLM 官方 `mooncake_connector_proxy.py`，patch 了 `min_tokens` 在 prefill leg 上的兼容性 bug
+- `instrument_kv_snapshot.py`：patch V1 scheduler 暴露 `schedule()` 结束时的 per-request KV block 分配 + 修复 vLLM 0.18.1 `MooncakeConnectorWorker.bootstrap_server` 在 kv_consumer 模式下未初始化的 AttributeError
+- `plot_kv_pool_timeline.py`：per-instance KV pool 时间线（stacked-area）
+- `aggregate_mb5.py`：跨 config / 跨 rep 聚合，输出 4 张对比图 + 1 张 CSV
+
+**PD-disagg smoke (4P+4D × 20 reqs)**: 20/20 success, mean latency 3.9s, p99 17s, 8 PIDs 都写 snapshot（601 total）。
+对比 8C × 20 reqs 同样数据点会在 sweep 完成后给出。
+
+**Phase 1 在跑**：
+- RUN_TAG=`20260527_164040`
+- CONFIGS=`8C 6P+2D 4P+4D 2P+6D` × REPS=3
+- TRACE=`w600_r0.0015_st30.jsonl` (~13 min/rep)
+- ETA ~3 h
+
+**Pending**：sweep 结果 → 跑 `aggregate_mb5.py` → 写 Phase 2 system analysis。
+
 ---
 
 ## 4 条独立失败假设
@@ -30,52 +51,30 @@ Living TODO 文档。记 H1/H2/H3/H4 四条假设的实验状态，以及每个
 **结论**：用 vLLM 仓库 ship 的官方 example
 `third_party/vllm/examples/online_serving/disaggregated_serving/mooncake_connector/`：
 
-- `run_mooncake_connector.sh` —— 参数化 P/D GPU 列表、ports、bootstrap，**直接支持任意 P:D 比例**（e.g., `PREFILL_GPUS=0,1,2,3 DECODE_GPUS=4,5,6,7` 起 4P+4D）
-- `mooncake_connector_proxy.py` —— 官方 FastAPI proxy，round-robin P + round-robin D，每个请求 fire-and-forget 到 P 做 `do_remote_decode={transfer_id}`，并行用 P 的 (bootstrap_addr, engine_id) 触发 D 做 `do_remote_prefill={remote_bootstrap_addr, remote_engine_id, transfer_id}`
+- `run_mooncake_connector.sh` —— 参数化 P/D GPU 列表、ports、bootstrap，**直接支持任意 P:D 比例**
+- `mooncake_connector_proxy.py` —— 官方 FastAPI proxy，round-robin P + round-robin D；vendored 到 `microbench/fresh_setup/mb5_pd_proxy.py`，加 `min_tokens=1` 修复
 
-部署形态：P 实例用 `kv_role:kv_producer`（带 bootstrap_server），D 实例用 `kv_role:kv_consumer`（**无 bootstrap_server**，正是之前我们撞到 `AttributeError` 的那个 mode）。
+部署形态：P 实例用 `kv_role:kv_producer`（带 bootstrap_server），D 实例用 `kv_role:kv_consumer`。后者的 `bootstrap_server` AttributeError 通过 `instrument_kv_snapshot.py` patch 修复。
 
-- [ ] 修改 `run_mooncake_connector.sh` 适配我们环境：模型路径 `/home/admin/cpfs/wjh/models/Qwen/Qwen3-Coder-30B-A3B-Instruct`，TP=1，所需 vLLM 启动 flags（`--max-model-len 200000` 等）
-- [ ] 重新验证 `kv_consumer` 模式：之前我们碰到的 AttributeError 可能是 kv_transfer_params 不对（missing transfer_id 或字段名错），MB2 调试后我们已经 nail down 正确 handshake；官方 proxy 用的就是同样的 handshake，所以这条很可能直接 work
-- [ ] 如果 kv_consumer 仍然报 AttributeError，加一个 minimal patch 到 vLLM（之前看过 `self.bootstrap_server` 在 consumer 模式没被初始化）—— 单行修复
+### TODO 0.2: 包装官方 launcher ✅ DONE
 
-### TODO 0.2: 包装官方 launcher
+`microbench/fresh_setup/mb5_launch.sh` 单一 launcher，支持 `8C / 6P+2D / 4P+4D / 2P+6D`；
+配套 `mb5_run.sh` orchestrator（CONFIG × REP 迭代，含 launch/replay/teardown）。
 
-直接基于 `run_mooncake_connector.sh` 改：
+### TODO 0.3: System-level instrumentation ✅ DONE (per-request KV)
 
-- [ ] `microbench/fresh_setup/mb5_pd_launch.sh` —— 直接 sourceable，参数 `PREFILL_GPUS` / `DECODE_GPUS` / `PREFILL_PORTS` / `BOOTSTRAP_PORTS` / `DECODE_PORTS` / `PROXY_PORT`，启动 P + D + 官方 proxy
-- [ ] 注入正确模型路径 + vLLM flags（`--max-model-len 200000 --gpu-memory-utilization 0.9 --enable-prefix-caching --max-num-batched-tokens 8192`）
-- [ ] 配置覆盖：先做 4 个 — `8C colo` + `6P+2D` + `4P+4D` + `2P+6D`
-- [ ] 注：colo baseline `8C` 不走这个 launcher，沿用 dash0 现有 8-instance unified setup（最公平的对照）
+选了比 `/metrics` 更深的层面：**patch V1 scheduler 直接 dump 每个 `schedule()` 回合的 per-request KV block 分配**（10 Hz throttle）。
 
-### TODO 0.3: System-level instrumentation
+- `instrument_kv_snapshot.py` 输出 schema: `{t_unix, step, total_blocks, free_blocks, used_blocks, running:[{req_id, n_blocks, n_computed, n_prompt, n_tokens, status}], waiting:[...]}`
+- 每个 EngineCore PID 一份 jsonl，集中写入 `MB5_LOG_DIR`
+- 跟 prometheus `/metrics` 比：(a) 不需要轮询，(b) 拿到 per-request 而不只是 aggregate，(c) 可以反推 D-pool 在某时刻被谁占着
 
-不止看 latency / success rate；要记 system 行为时间序列才能 attribute bottleneck。
+如后续需要 prometheus `/metrics`（admission denial 事件之类），可以再加一个 sampler；目前的 per-request 数据已经能撑住 Phase 1 + Phase 2 分析。
 
-**必须采集的 metrics**（每秒一次，每实例）：
+### TODO 0.4: D 池 occupancy timeline 可视化 ✅ DONE
 
-- [ ] `vllm:gpu_cache_usage_perc` — KV pool 占用率（核心 H1 证据）
-- [ ] `vllm:num_requests_running` — 并发 in-flight 数
-- [ ] `vllm:num_requests_waiting` — scheduler 排队数
-- [ ] `vllm:time_to_first_token_seconds` (histogram) — 每 instance TTFT 分布
-- [ ] `vllm:time_per_output_token_seconds` (histogram) — TPOT 分布
-- [ ] D 池请求 admission control 拒绝事件（vllm 拒收新请求时的事件）—— 看 vllm 是否暴露 metric
-- [ ] Mooncake 侧：`send_blocks` 事件（MB2 instrument 已存在）；可选 transfer queue depth
-
-**实现**：
-
-- [ ] 写 `metrics_sampler.py`：周期性 GET `/metrics` 解析 prometheus 文本，输出 jsonl
-- [ ] 每个 instance 一个采样进程 / 或者一个集中采样进程拉所有 instance
-- [ ] 输出 schema: `{t_unix, instance_id, role, kv_pool_perc, num_running, num_waiting, ...}`
-
-### TODO 0.4: D 池 occupancy timeline 可视化
-
-- [ ] 写 `plot_d_pool_timeline.py`：heatmap 或 stacked area
-  - x 轴：trace replay 时间
-  - y 轴：每个 D-instance（heatmap）或 KV pool 总占用比（stacked area）
-  - 色彩：占用率 0–100%
-  - 标 90% 红线（"vllm stops admitting new requests" 阈值，参考 colleague 旧数据）
-- [ ] Output: 每个 config 一张图，stacked 起来对比
+- `plot_kv_pool_timeline.py` —— per-instance 视图（stacked-area: 时间 × 块数 × per-request 色块；底下 waiting queue depth 子图）
+- `aggregate_mb5.py` —— 跨 config / 跨 rep 聚合视图（cluster-wide KV 时间线、peak 占用率 bar、latency p50/p90/p99 bar、summary CSV）
 
 ---
 
@@ -106,14 +105,13 @@ reps = 3
 - [ ] **TTFT per request scatter**（colored by which-instance）— 看 D 池满了的时候 TTFT 是不是直接挂掉
 - [ ] **Admission denial events**（如果 vllm 暴露）
 
-### Output figures
+### Output figures (`aggregate_mb5.py` will write all of these)
 
-- [ ] `mb5_latency_bars.png` — config × TTFT/TPOT/E2E p90 bar
-- [ ] `mb5_success.png` — success rate per config
-- [ ] `mb5_wallclock.png` — 实测 trace replay 时间 vs 8C colo
-- [ ] `mb5_d_pool_timeline.png` — 4 configs × 8 instances heatmap
-- [ ] `mb5_queue_depth_timeline.png` — 同上结构
-- [ ] `mb5_diagnostic_summary.png` — 1 张总图把上面 5 个塞进去给 paper 用
+- [ ] `figs/mb5/mb5_kv_timeline.png` — 4 panels (one per config), cluster-wide KV % 时间线，faint per-rep line + bold median
+- [ ] `figs/mb5/mb5_peak_utilization.png` — bar chart peak vs steady KV per config，含 ±std error bars
+- [ ] `figs/mb5/mb5_latency_compare.png` — bar chart p50/p90/p99 e2e latency per config
+- [ ] `figs/mb5/mb5_summary.csv` — flat per-(config, rep) 表（latency, KV, prefix cache, success rate）
+- [ ] （manual）`figs/mb5/mb5_per_instance_timeline.png` — pick 1 rep per config, plot per-instance stacked-area via `plot_kv_pool_timeline.py`，给 paper §3.x system breakdown 用
 
 ### 判定标准（H1 何时被 confirmed）
 

microbench/fresh_setup/PD_DISAGG_RESULTS.md

@@ -0,0 +1,260 @@
+# PD-disaggregation under an agentic workload — does it work?
+
+**Consolidated results doc.** Self-contained writeup of every PD-disagg
+argument and experiment, with figures inline. For the live experiment TODO
+list see [PD_DISAGG_INVESTIGATION.md](PD_DISAGG_INVESTIGATION.md).
+
+Date: 2026-05-28 · Hardware: dash1, 8×GPU · Model: Qwen3-Coder-30B-A3B-Instruct
+· vLLM 0.18.1 (V1, chunked-prefill on) · Mooncake 0.3.11 · Trace:
+`w600_r0.0015_st30.jsonl` (1214 requests, agentic multi-turn).
+
+---
+
+## TL;DR (verdict)
+
+**No static prefill/decode split beats 8-way colocation (8C) on this agentic
+workload.** Every disaggregated ratio we tried is dominated by 8C on the
+metric the user actually feels (TTFT, end-to-end latency, request
+completion), and the failure *moves* with the ratio:
+
+- **D-heavy bottleneck** (6P+2D, 4P+4D): the decode pool saturates (peak
+  **99.6% / 97.5%**) while the prefill pool sits at **~30%** — half the
+  cluster's KV is stranded on the wrong side.
+- **P-heavy bottleneck** (2P+6D): the 2 prefill instances can't keep up,
+  the prefill pool jams at **99.7%**, **872 requests** pile up in the queue
+  and **91% of requests never complete**.
+- **8C** keeps a single elastic pool that absorbs whichever phase is hot at
+  the moment → steady utilization **34%**, **100% completion**, fastest
+  wall-clock, best p50/p90 latency.
+
+PD-disagg *does* deliver the phase-isolation win we predicted in MB1 — its
+**TPOT is 10–35× cleaner** — but that win is swamped by TTFT inflation,
+request loss, and a total collapse of prefix-cache reuse under the stock
+round-robin router.
+
+This is the empirical backing for the paper's claim: **agentic workloads
+have time-varying P:D demand that no static partition can track; colocation
+wins because its pool is elastic.** (H1 *and* H2 from the investigation doc,
+unified by one mechanism.)
+
+---
+
+## 1. Why this experiment exists
+
+Earlier cost accounting (MB1 phase-interference, MB2 KV-transfer cost) showed
+that on the **phase-isolation axis alone**, PD-disagg actually *wins*: it
+removes prefill→decode interference, and the transfer cost is small relative
+to the interference it avoids. So "PD-disagg is bad for agentic" could not be
+argued from phase isolation — we needed a system-level experiment that
+measures the whole picture (queueing, pool capacity, cache reuse), not just
+the isolated phase cost.
+
+See [analysis/mb1](../../analysis/mb1) and [analysis/mb2](../../analysis/mb2)
+for that accounting. This doc is the system-level answer.
+
+---
+
+## 2. Setup
+
+| | |
+|---|---|
+| Configs | `8C` (8× kv_both colo), `6P+2D`, `4P+4D`, `2P+6D` (prefill+decode split) |
+| PD routing | stock **round-robin** on both P and D (vLLM official `mooncake_connector_proxy`) |
+| Trace | `w600_r0.0015_st30.jsonl`, 1214 requests, agentic multi-turn |
+| Reps | 1 (rep1) for this analysis; the 3-rep sweep confirmed run-to-run consistency before we converged on rep1 for iteration speed |
+| KV instrumentation | V1 scheduler patched to dump per-request KV block allocation every 100 ms per EngineCore (see `instrument_kv_snapshot.py`) |
+
+8C is the fair baseline: 8 colocated instances, replayer round-robins across
+them directly (no proxy). PD configs route through the proxy.
+
+---
+
+## 3. Headline result — no PD ratio beats 8C
+
+All numbers are rep1.
+
+| Metric | **8C** | 6P+2D | 4P+4D | 2P+6D |
+|---|---|---|---|---|
+| **completion** | **100%** | 100% | 100% | **9%** 💀 |
+| wall-clock (drain trace) | **2994 s** | 3419 s | 4171 s | 5762 s |
+| prefix-cache hit | **19.4%** | 0% | 0% | 0% |
+| TTFT mean | **18.0 s** | 44.8 s | 70.0 s | 106.8 s |
+| TTFT p50 | **7.0 s** | 41.0 s | 56.4 s | 23.6 s |
+| TTFT p90 | **53.1 s** | 86.7 s | 153.1 s | 498 s |
+| E2E p50 | **10.8 s** | 44.5 s | 59.5 s | 26.3 s |
+| E2E p90 | **83.3 s** | 91.8 s | 157.1 s | 499 s |
+
+![e2e latency by config](../../figs/mb5/mb5_latency_compare.png)
+
+> ⚠️ **Read the percentiles with the completion rate.** Latency percentiles
+> are computed over *successful* requests only. 2P+6D's "p99 = 577 s" covers
+> just the 9% that finished — the other 91% never returned, so its real
+> experience is far worse than any latency bar suggests.
+
+8C wins p50 by **4×** and p90 decisively. The only metric where a PD config
+edges 8C is E2E **p99** (6P+2D 148 s vs 8C 194 s) — and that is the flip side
+of the next result.
+
+---
+
+## 4. The duality — PD wins TPOT, loses TTFT
+
+PD-disagg delivers exactly the phase-isolation benefit MB1 predicted: with no
+prefill stealing decode steps, **inter-token latency is dramatically cleaner.**
+
+| TPOT | **8C** | 6P+2D | 4P+4D | 2P+6D |
+|---|---|---|---|---|
+| mean | 87 ms | 11 ms | 9 ms | 6 ms |
+| p90 | 230 ms | 18 ms | 14 ms | 8 ms |
+| p99 | **1129 ms** | **26 ms** | **20 ms** | **12 ms** |
+
+PD's TPOT p99 is **10–35× lower** — once a request reaches a dedicated decode
+instance it streams without interruption. 8C's 1.1 s TPOT p99 *is* the
+chunked-prefill interference tax (decode steps occasionally stalled behind an
+8k-token prefill chunk), consistent with MB1.
+
+**But the win is local.** TTFT inflates 2.5–6× because every request now pays
+P→D handoff + admission into a smaller, saturated decode pool. For this
+workload's modest output lengths, TTFT dominates total time, so the TPOT win
+never pays for itself. This is the cost/benefit imbalance made concrete:
+phase isolation is real, but it is the wrong thing to optimize when the pool
+is the binding constraint.
+
+---
+
+## 5. Root cause — per-role KV pool occupancy (the kill shot)
+
+The cluster-average KV utilization is *misleading* and nearly hid the result:
+
+![cluster KV timeline](../../figs/mb5/mb5_kv_timeline.png)
+
+6P+2D and 4P+4D look only ~42–46% utilized on cluster average — yet they have
+128–152 requests queued. The average hides that **one pool is pegged while
+the other idles.** Splitting the KV pool by role exposes it:
+
+![per-role KV pool: P-pool vs D-pool](../../figs/mb5/mb5_role_split.png)
+
+| Config | P-pool steady | D-pool steady | D-pool **peak** | binding side |
+|---|---|---|---|---|
+| 8C | — single shared pool — | 34% | 72% | none (elastic) |
+| 6P+2D | 31% | **74%** | **99.6%** | **decode** |
+| 4P+4D | 29% | **60%** | **97.5%** | **decode** |
+| 2P+6D | **92%** | 95% | 96% | **prefill** (P jams first) |
+
+![peak vs steady utilization](../../figs/mb5/mb5_peak_utilization.png)
+
+**The mechanism, unified:**
+
+- A static P:D split fixes the KV capacity on each side at deploy time.
+- The agentic workload's instantaneous P:D demand *drifts* (bursts of new
+  sessions = prefill-heavy; long tool-call-driven turns = decode-heavy).
+- Whichever side is undersized *for the current phase* saturates and
+  back-pressures the whole pipeline, while the other side's KV sits stranded.
+  - 6P+2D / 4P+4D → decode side too small → D-pool hits ~100%, prefilled
+    requests queue for a decode slot → TTFT explodes (this is **H1**).
+  - 2P+6D → prefill side too small → P-pool hits ~100%, requests can't even
+    start → 872 queued, 91% dropped.
+- **8C colocation has no partition**: prefill and decode share one pool, so
+  the pool elastically reallocates to whichever phase is hot. Steady
+  utilization stays at 34% with 100% completion.
+
+This is **H1 (D-pool capacity ceiling)** and **H2 (static-partition
+mismatch)** turning out to be the *same* phenomenon seen from two ratios.
+
+### 5.1 The same pressure crashes consumers (a vLLM 0.18.1 fragility)
+
+D-pool saturation doesn't just slow things down — under this workload it
+**crashes the decode instances**. The exact chain, from the 6P+2D consumer
+logs:
+
+1. D-pool fills to **97.2%** (the capacity ceiling above).
+2. A large request needs its KV pulled to the consumer, but the transfer
+   fails: `Mooncake transfer engine returned -1` (observed on a **112,793-token**
+   request — agentic sessions have very long multi-turn contexts, and the
+   pool had no room).
+3. `kv_load_failure_policy=fail` fails that request — by itself recoverable.
+4. **But** the failure path computes `PromptTokenStats.local_cache_hit =
+   num_cached + recomputed − num_external_computed`, which goes **negative**
+   when the external transfer exceeded the scheduler's cached count.
+5. `loggers.record()` calls `Counter.inc(negative)` → prometheus_client raises
+   *"Counters can only be incremented by non-negative amounts"* → the
+   **EngineCore dies**.
+6. Once the consumer's engine is dead, **every** subsequent request fails.
+
+The signature is a cliff, not a slope: in the session-routing 6P+2D run, all
+80 successes landed in the first ~110 s, then **zero** of the next ~2,800 s.
+This same intermittent consumer death is almost certainly why the
+round-robin 6P+2D reps varied so wildly (100% / 56% / 80%) — the consumer
+crashed at different points in each rep.
+
+**Two takeaways:** (a) PD-disagg under agentic context lengths hits KV-transfer
+failures that colocation never does (8C never transfers — it prefills and
+decodes in the same pool); (b) vLLM 0.18.1's failure handling amplifies one
+failed request into a total collapse. We patched the counter underflow
+(`instrument_kv_snapshot.py`, clamp to ≥ 0) so a transfer failure stays a
+single failed request, which is required to compare routing arms fairly in §6.
+
+---
+
+## 6. The routing handicap — and whether smarter routing rescues PD
+
+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
+*different* producers, so each turn re-prefills the whole conversation from
+scratch. That both inflates TTFT and piles extra load on the prefill pool
+(directly worsening the 2P+6D collapse).
+
+The correct PD scheduling policy (as the design argues): **P should be chosen
+by session affinity** (reuse the producer's prefix cache) while **D is chosen
+by load balance** (decode KV is freshly transferred per turn, so D gains
+nothing from affinity). We added this as an env-gated mode in the proxy
+(`MB5_P_ROUTING=session`, consistent hash on `X-Session-Id`; D stays
+round-robin) and re-ran the best-performing disaggregated config, **6P+2D**.
+
+> **Status: session-affinity 6P+2D run in progress.** Results below will be
+> filled in when it completes; the question it answers is *how much of the
+> gap to 8C does restoring prefix-cache reuse close.*
+
+<!-- SESSION_AFFINITY_RESULTS -->
+*(pending)*
+
+---
+
+## 7. Caveats / honesty
+
+- **Single rep** for this analysis. The earlier 3-rep sweep showed 8C and
+  4P+4D are tight run-to-run, but 6P+2D completion varied (rep1 100% vs rep2
+  56% vs rep3 80%) — i.e. the D-pool sits right at the cliff edge, so 6P+2D's
+  "100% rep1" is optimistic. The qualitative ranking is robust; exact numbers
+  on the marginal configs are not.
+- **Latency percentiles count successes only** (see §3 warning). For failing
+  configs the latency bars *understate* the damage.
+- **Round-robin baseline.** §6 addresses the routing fairness concern head-on
+  with a session-affinity re-run.
+- Trace is a single agentic workload; conclusions are about *this* class of
+  workload (sub-second tool-call cadence, multi-turn sessions), not all LLM
+  serving.
+
+---
+
+## 8. Reproduce
+
+```bash
+# from repo root, after microbench/fresh_setup/deploy.sh dash1
+# 1. round-robin baseline sweep (1 rep)
+ssh dash1 'CONFIGS="8C 6P+2D 4P+4D 2P+6D" REPS=1 RUN_TAG=<tag> \
+    bash /home/admin/cpfs/wjh/agentic-kv-fresh/scripts/mb5_run.sh'
+
+# 2. reduce on dash1 (numpy-only; handles the multi-GB snapshot dirs)
+ssh dash1 '.venv/bin/python scripts/aggregate_mb5.py --sweep-root mb5_runs \
+    --tag <tag> --configs "8C 6P+2D 4P+4D 2P+6D" --reps 1 \
+    --reduce-to mb5_runs/reduced_<tag>.json'
+
+# 3. pull the compact JSON, render figures locally
+scp dash1:.../mb5_runs/reduced_<tag>.json analysis/mb5/
+.venv/bin/python microbench/fresh_setup/aggregate_mb5.py \
+    --from-reduced analysis/mb5/reduced_<tag>.json --out-dir figs/mb5
+
+# session-affinity arm: prefix the run with MB5_P_ROUTING=session
+```

microbench/fresh_setup/aggregate_mb5.py

@@ -0,0 +1,481 @@
+#!/usr/bin/env python3
+"""Aggregate MB5 sweep data into cross-config comparison figures.
+
+Reads a sweep root (e.g. /home/admin/.../mb5_runs/) and a tag
+(e.g. "20260527_164040"). For each (config, rep) tuple, loads:
+
+  ${tag}_${config}_rep${N}/replay_metrics.summary.json     -> aggregate stats
+  ${tag}_${config}_rep${N}/replay_metrics.jsonl            -> per-request latency
+  ${tag}_${config}_rep${N}_${config}/kv_snapshots/         -> per-instance KV state
+
+Produces, in --out-dir:
+  mb5_kv_timeline.png       — 4 panels, cluster-wide KV utilization over time
+                              (1 faint line per rep + bold median across reps)
+  mb5_peak_utilization.png  — bar chart: peak / steady KV util per config
+                              (mean across reps + error bars)
+  mb5_latency_compare.png   — bar chart: p50 / p90 / p99 e2e latency per config
+  mb5_summary.csv           — flat table for the writeup
+
+Use case:
+  python aggregate_mb5.py --sweep-root /home/.../mb5_runs \\
+                           --tag 20260527_164040 \\
+                           --configs "8C 6P+2D 4P+4D 2P+6D" \\
+                           --reps 3 \\
+                           --out-dir figs/mb5
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+from collections import defaultdict
+from pathlib import Path
+
+import numpy as np
+
+# matplotlib is imported lazily inside the plot functions so the --reduce
+# path (numpy-only) can run on a serving host without matplotlib installed.
+
+
+def load_snapshots_for_run(snap_dir: Path) -> list[dict]:
+    """Merge all per-PID snapshot files in snap_dir, tag with pid, sort by t_unix."""
+    out = []
+    for f in sorted(snap_dir.glob("mb5_kv_snapshot_pid*.jsonl")):
+        pid = int(f.stem.replace("mb5_kv_snapshot_pid", ""))
+        with f.open() as fh:
+            for line in fh:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    d = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                d["pid"] = pid
+                out.append(d)
+    out.sort(key=lambda d: d["t_unix"])
+    return out
+
+
+def load_pid_roles(logs_dir: Path) -> dict[int, str]:
+    """Map EngineCore PID -> 'P' | 'D' | 'C' by parsing vllm_logs filenames.
+
+    File names look like vllm_idx{i}_gpu{g}_kv_{producer|consumer|both}.log and
+    each contains '(EngineCore pid=NNNN)'. Returns {} if no logs found.
+    """
+    role_map = {"producer": "P", "consumer": "D", "both": "C"}
+    out: dict[int, str] = {}
+    if not logs_dir.is_dir():
+        return out
+    for f in logs_dir.glob("vllm_idx*_kv_*.log"):
+        role = None
+        for key, short in role_map.items():
+            if f.name.endswith(f"kv_{key}.log"):
+                role = short
+                break
+        if role is None:
+            continue
+        with f.open(errors="ignore") as fh:
+            for line in fh:
+                if "EngineCore pid=" in line:
+                    try:
+                        pid = int(line.split("EngineCore pid=")[1].split(")")[0].split()[0])
+                        out[pid] = role
+                        break
+                    except (ValueError, IndexError):
+                        continue
+    return out
+
+
+def cluster_timeline(snaps: list[dict], bin_size_s: float = 1.0,
+                     keep_pids: set | None = None,
+                     t0: float | None = None,
+                     n_bins: int | None = None) -> tuple[np.ndarray, ...]:
+    """Bin per-PID snapshots into a cluster-wide timeline.
+
+    For each time bin, sum used_blocks across PIDs that emitted a snapshot
+    in that bin. PIDs without a sample in a bin carry their previous value
+    forward (so a quiet PID doesn't artificially drop the total).
+
+    If keep_pids is given, only those PIDs are counted (used for per-role
+    P-pool / D-pool splits); the pool ceiling is summed over the same subset.
+    Pass a shared t0/n_bins so role-splits land on the same time grid.
+    """
+    if keep_pids is not None:
+        snaps = [s for s in snaps if s["pid"] in keep_pids]
+    if not snaps:
+        empty = np.array([], dtype=float)
+        return empty, empty, empty, empty, empty
+    if t0 is None:
+        t0 = snaps[0]["t_unix"]
+    if n_bins is None:
+        t_end = snaps[-1]["t_unix"]
+        n_bins = max(1, int(np.ceil((t_end - t0) / bin_size_s)) + 1)
+    times = np.arange(n_bins) * bin_size_s
+
+    pids = sorted({s["pid"] for s in snaps})
+    pid_to_idx = {pid: i for i, pid in enumerate(pids)}
+
+    # last-known used/total/waiting per PID at each bin (carry-forward)
+    used = np.zeros((len(pids), n_bins), dtype=np.int64)
+    waiting = np.zeros((len(pids), n_bins), dtype=np.int64)
+    running = np.zeros((len(pids), n_bins), dtype=np.int64)
+    total_per_pid = np.zeros(len(pids), dtype=np.int64)
+
+    last_used = [0] * len(pids)
+    last_waiting = [0] * len(pids)
+    last_running = [0] * len(pids)
+
+    snap_iter = iter(snaps)
+    next_snap = next(snap_iter, None)
+
+    for b in range(n_bins):
+        t_lo = t0 + b * bin_size_s
+        t_hi = t_lo + bin_size_s
+        while next_snap is not None and next_snap["t_unix"] < t_hi:
+            i = pid_to_idx[next_snap["pid"]]
+            last_used[i] = next_snap.get("used_blocks", 0)
+            last_waiting[i] = len(next_snap.get("waiting", []))
+            last_running[i] = len(next_snap.get("running", []))
+            total_per_pid[i] = next_snap.get("total_blocks", 0)
+            next_snap = next(snap_iter, None)
+        for i in range(len(pids)):
+            used[i, b] = last_used[i]
+            waiting[i, b] = last_waiting[i]
+            running[i, b] = last_running[i]
+
+    total_used = used.sum(axis=0)
+    total_pool = int(total_per_pid.sum())
+    total_waiting = waiting.sum(axis=0)
+    total_running = running.sum(axis=0)
+    pool_frac = total_used / max(total_pool, 1)
+    return times, total_used, pool_frac, total_waiting, total_running
+
+
+def load_summary(rundir: Path) -> dict | None:
+    p = rundir / "replay_metrics.summary.json"
+    if not p.is_file():
+        return None
+    return json.loads(p.read_text())
+
+
+def _steady_median(arr: np.ndarray) -> float:
+    n = len(arr)
+    if n == 0:
+        return 0.0
+    if n >= 10:
+        return float(np.median(arr[int(n * 0.1):int(n * 0.9)]))
+    return float(np.median(arr))
+
+
+def per_run_metrics(snaps_dir: Path, rundir: Path) -> dict:
+    snaps = load_snapshots_for_run(snaps_dir)
+    summary = load_summary(rundir) or {}
+
+    # Establish a shared time grid (global t0 / n_bins) so the overall and
+    # per-role timelines all line up on the same x axis.
+    if snaps:
+        t0 = snaps[0]["t_unix"]
+        t_end = snaps[-1]["t_unix"]
+        n_bins = max(1, int(np.ceil(t_end - t0)) + 1)
+    else:
+        t0, n_bins = None, None
+
+    times, total_used, pool_frac, total_waiting, total_running = cluster_timeline(
+        snaps, t0=t0, n_bins=n_bins
+    )
+    n = len(times)
+
+    out = {
+        "times": times.tolist(),
+        "total_used": total_used.tolist(),
+        "pool_frac": pool_frac.tolist(),
+        "total_waiting": total_waiting.tolist(),
+        "total_running": total_running.tolist(),
+        "peak_pool_frac": float(pool_frac.max()) if n else 0.0,
+        "steady_pool_frac": _steady_median(pool_frac),
+        "peak_waiting": int(total_waiting.max()) if n else 0,
+        "summary": summary,
+    }
+
+    # Per-role (P-pool vs D-pool) split for PD configs.
+    roles = load_pid_roles(snaps_dir.parent / "vllm_logs")
+    p_pids = {pid for pid, r in roles.items() if r == "P"}
+    d_pids = {pid for pid, r in roles.items() if r == "D"}
+    if p_pids and d_pids:
+        for tag, subset in (("p", p_pids), ("d", d_pids)):
+            _, _, frac, _, run = cluster_timeline(
+                snaps, keep_pids=subset, t0=t0, n_bins=n_bins
+            )
+            out[f"{tag}_pool_frac"] = frac.tolist()
+            out[f"{tag}_running"] = run.tolist()
+            out[f"{tag}_peak_frac"] = float(frac.max()) if len(frac) else 0.0
+            out[f"{tag}_steady_frac"] = _steady_median(frac)
+    return out
+
+
+def collect_sweep(sweep_root: Path, tag: str, configs: list[str], reps: int) -> dict:
+    """Returns {config: [run_record_per_rep]}."""
+    out: dict[str, list[dict]] = defaultdict(list)
+    for config in configs:
+        for rep in range(1, reps + 1):
+            rundir = sweep_root / f"{tag}_{config}_rep{rep}"
+            snap_dir = sweep_root / f"{tag}_{config}_rep{rep}_{config}/kv_snapshots"
+            if not snap_dir.is_dir():
+                print(f"[agg] MISSING: {snap_dir}")
+                continue
+            metrics = per_run_metrics(snap_dir, rundir)
+            metrics["rep"] = rep
+            out[config].append(metrics)
+            print(
+                f"[agg] {config} rep{rep}: peak={metrics['peak_pool_frac']:.1%} "
+                f"steady={metrics['steady_pool_frac']:.1%} "
+                f"peak_wait={metrics['peak_waiting']}"
+            )
+    return out
+
+
+def plot_kv_timeline(sweep: dict, out: Path) -> None:
+    import matplotlib
+    matplotlib.use("Agg")
+    import matplotlib.pyplot as plt
+
+    n_configs = len(sweep)
+    if n_configs == 0:
+        return
+    fig, axes = plt.subplots(n_configs, 1, figsize=(14, 2.5 * n_configs), sharex=True)
+    if n_configs == 1:
+        axes = [axes]
+    for ax, (config, reps) in zip(axes, sweep.items()):
+        for rep_data in reps:
+            t = np.asarray(rep_data["times"])
+            ax.plot(t, np.asarray(rep_data["pool_frac"]) * 100, alpha=0.4, lw=1.0,
+                    label=f"rep{rep_data['rep']}")
+        # bold median across reps (need to align times — use longest series)
+        if reps:
+            max_len = max(len(r["times"]) for r in reps)
+            arr = np.full((len(reps), max_len), np.nan)
+            for i, r in enumerate(reps):
+                arr[i, :len(r["pool_frac"])] = r["pool_frac"]
+            median = np.nanmedian(arr, axis=0) * 100
+            ax.plot(np.arange(max_len), median, color="#222", lw=2.0, label="median")
+        ax.axhline(90, color="#c44e52", ls="--", alpha=0.6, lw=1, label="90%")
+        ax.set_ylim(0, 105)
+        ax.set_ylabel(f"{config}\ncluster KV (%)")
+        ax.grid(True, alpha=0.3)
+        ax.legend(loc="upper right", fontsize=8)
+    axes[-1].set_xlabel("wall-clock since first snapshot (s)")
+    fig.suptitle("MB5: cluster-wide KV pool utilization over time", fontsize=12)
+    fig.tight_layout()
+    out.parent.mkdir(parents=True, exist_ok=True)
+    fig.savefig(out, dpi=120)
+    plt.close(fig)
+    print(f"wrote {out}")
+
+
+def plot_peak_utilization(sweep: dict, out: Path) -> None:
+    import matplotlib
+    matplotlib.use("Agg")
+    import matplotlib.pyplot as plt
+
+    configs = list(sweep.keys())
+    peaks = [[r["peak_pool_frac"] * 100 for r in sweep[c]] for c in configs]
+    steady = [[r["steady_pool_frac"] * 100 for r in sweep[c]] for c in configs]
+    peak_means = [np.mean(p) if p else 0 for p in peaks]
+    peak_std = [np.std(p) if len(p) > 1 else 0 for p in peaks]
+    steady_means = [np.mean(s) if s else 0 for s in steady]
+    steady_std = [np.std(s) if len(s) > 1 else 0 for s in steady]
+
+    x = np.arange(len(configs))
+    width = 0.35
+
+    fig, ax = plt.subplots(figsize=(9, 4.5))
+    ax.bar(x - width/2, peak_means, width, yerr=peak_std, label="peak",
+           color="#c44e52", capsize=4)
+    ax.bar(x + width/2, steady_means, width, yerr=steady_std, label="steady (10–90%)",
+           color="#4c72b0", capsize=4)
+    ax.axhline(90, color="#444", ls="--", alpha=0.5, lw=1, label="90% red line")
+    ax.set_xticks(x)
+    ax.set_xticklabels(configs)
+    ax.set_ylabel("Cluster KV pool utilization (%)")
+    ax.set_ylim(0, 105)
+    ax.set_title("MB5: KV pool pressure — peak vs steady-state")
+    ax.legend(loc="upper left", fontsize=9)
+    ax.grid(True, axis="y", alpha=0.3)
+    fig.tight_layout()
+    fig.savefig(out, dpi=120)
+    plt.close(fig)
+    print(f"wrote {out}")
+
+
+def plot_latency_compare(sweep: dict, out: Path) -> None:
+    import matplotlib
+    matplotlib.use("Agg")
+    import matplotlib.pyplot as plt
+
+    configs = list(sweep.keys())
+    metrics = ["p50", "p90", "p99"]
+    data = {m: [] for m in metrics}
+    for c in configs:
+        for m in metrics:
+            vals = []
+            for r in sweep[c]:
+                s = r["summary"].get("latency_stats_s")
+                if s and s.get(m) is not None:
+                    vals.append(s[m])
+            data[m].append(np.mean(vals) if vals else 0.0)
+
+    x = np.arange(len(configs))
+    width = 0.25
+    colors = {"p50": "#4c72b0", "p90": "#dd8452", "p99": "#c44e52"}
+    fig, ax = plt.subplots(figsize=(9, 4.5))
+    for i, m in enumerate(metrics):
+        ax.bar(x + (i - 1) * width, data[m], width, label=m, color=colors[m])
+    ax.set_xticks(x)
+    ax.set_xticklabels(configs)
+    ax.set_ylabel("End-to-end latency (s)")
+    ax.set_title("MB5: e2e latency by PD configuration")
+    ax.legend()
+    ax.grid(True, axis="y", alpha=0.3)
+    fig.tight_layout()
+    fig.savefig(out, dpi=120)
+    plt.close(fig)
+    print(f"wrote {out}")
+
+
+def plot_role_split(sweep: dict, out: Path) -> None:
+    """For PD configs, show P-pool vs D-pool KV % over time (rep1) — exposes
+    the imbalance that the cluster average hides. 8C (no role split) shows
+    the overall cluster line for reference."""
+    import matplotlib
+    matplotlib.use("Agg")
+    import matplotlib.pyplot as plt
+
+    n_configs = len(sweep)
+    if n_configs == 0:
+        return
+    fig, axes = plt.subplots(n_configs, 1, figsize=(14, 2.6 * n_configs), sharex=True)
+    if n_configs == 1:
+        axes = [axes]
+    for ax, (config, reps) in zip(axes, sweep.items()):
+        if not reps:
+            continue
+        r = reps[0]  # rep1
+        t = np.asarray(r["times"])
+        if "p_pool_frac" in r and "d_pool_frac" in r:
+            ax.plot(t, np.asarray(r["p_pool_frac"]) * 100, color="#4c72b0",
+                    lw=1.5, label="P-pool (prefill)")
+            ax.plot(t, np.asarray(r["d_pool_frac"]) * 100, color="#c44e52",
+                    lw=1.5, label="D-pool (decode)")
+            ax.plot(t, np.asarray(r["pool_frac"]) * 100, color="#999",
+                    lw=1.0, ls=":", label="cluster avg")
+        else:
+            ax.plot(t, np.asarray(r["pool_frac"]) * 100, color="#222",
+                    lw=1.5, label="cluster (kv_both)")
+        ax.axhline(90, color="#444", ls="--", alpha=0.5, lw=1)
+        ax.set_ylim(0, 105)
+        ax.set_ylabel(f"{config}\nKV pool (%)")
+        ax.grid(True, alpha=0.3)
+        ax.legend(loc="upper right", fontsize=8, ncol=3)
+    axes[-1].set_xlabel("wall-clock since first snapshot (s)")
+    fig.suptitle("MB5: per-role KV pool utilization (P-pool vs D-pool), rep1",
+                 fontsize=12)
+    fig.tight_layout()
+    out.parent.mkdir(parents=True, exist_ok=True)
+    fig.savefig(out, dpi=120)
+    plt.close(fig)
+    print(f"wrote {out}")
+
+
+def write_summary_csv(sweep: dict, out: Path) -> None:
+    rows = []
+    for config, reps in sweep.items():
+        for r in reps:
+            s = r["summary"]
+            lat = s.get("latency_stats_s") or {}
+            ttft = s.get("ttft_stats_s") or {}
+            rows.append({
+                "config": config,
+                "rep": r["rep"],
+                "n_requests": s.get("request_count"),
+                "n_success": s.get("success_count"),
+                "wall_clock_s": s.get("wall_clock_s"),
+                "peak_pool_frac": r["peak_pool_frac"],
+                "steady_pool_frac": r["steady_pool_frac"],
+                "p_pool_peak_frac": r.get("p_peak_frac"),
+                "p_pool_steady_frac": r.get("p_steady_frac"),
+                "d_pool_peak_frac": r.get("d_peak_frac"),
+                "d_pool_steady_frac": r.get("d_steady_frac"),
+                "peak_waiting": r["peak_waiting"],
+                "latency_p50_s": lat.get("p50"),
+                "latency_p90_s": lat.get("p90"),
+                "latency_p99_s": lat.get("p99"),
+                "ttft_p50_s": ttft.get("p50"),
+                "ttft_p90_s": ttft.get("p90"),
+                "ttft_p99_s": ttft.get("p99"),
+                "prefix_cache_hit_ratio": s.get("prefix_cache_hit_ratio"),
+            })
+    if not rows:
+        print("[agg] no rows; skipping CSV")
+        return
+    out.parent.mkdir(parents=True, exist_ok=True)
+    with out.open("w", newline="") as fh:
+        w = csv.DictWriter(fh, fieldnames=list(rows[0].keys()))
+        w.writeheader()
+        w.writerows(rows)
+    print(f"wrote {out} ({len(rows)} rows)")
+
+
+def render_all(sweep: dict, out_dir: Path) -> None:
+    plot_kv_timeline(sweep, out_dir / "mb5_kv_timeline.png")
+    plot_role_split(sweep, out_dir / "mb5_role_split.png")
+    plot_peak_utilization(sweep, out_dir / "mb5_peak_utilization.png")
+    plot_latency_compare(sweep, out_dir / "mb5_latency_compare.png")
+    write_summary_csv(sweep, out_dir / "mb5_summary.csv")
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(
+        description="MB5 aggregate. Two-stage: --reduce (numpy-only, runs on "
+                    "a serving host) dumps a compact JSON; --from-reduced "
+                    "(needs matplotlib) renders figures locally. Or run "
+                    "directly (raw snapshots -> figures) when both the data "
+                    "and matplotlib are local."
+    )
+    p.add_argument("--sweep-root", type=Path,
+                   help="dir containing ${tag}_${config}_rep${N}/ subdirs")
+    p.add_argument("--tag")
+    p.add_argument("--configs", default="8C 6P+2D 4P+4D 2P+6D",
+                   help="space-separated config names")
+    p.add_argument("--reps", type=int, default=3)
+    p.add_argument("--out-dir", type=Path, default=Path("figs/mb5"))
+    p.add_argument("--reduce-to", type=Path,
+                   help="numpy-only: write reduced sweep JSON here and exit "
+                        "(no plotting, no matplotlib needed)")
+    p.add_argument("--from-reduced", type=Path,
+                   help="load a reduced sweep JSON (from --reduce-to) and "
+                        "render figures into --out-dir")
+    args = p.parse_args()
+
+    if args.from_reduced:
+        sweep = json.loads(args.from_reduced.read_text())
+        render_all(sweep, args.out_dir)
+        return
+
+    if not (args.sweep_root and args.tag):
+        p.error("--sweep-root and --tag are required unless --from-reduced is given")
+
+    configs = args.configs.split()
+    sweep = collect_sweep(args.sweep_root, args.tag, configs, args.reps)
+
+    if args.reduce_to:
+        args.reduce_to.parent.mkdir(parents=True, exist_ok=True)
+        args.reduce_to.write_text(json.dumps(sweep))
+        print(f"wrote reduced sweep -> {args.reduce_to}")
+        return
+
+    render_all(sweep, args.out_dir)
+
+
+if __name__ == "__main__":
+    main()

microbench/fresh_setup/instrument_kv_snapshot.py

@@ -32,6 +32,11 @@ from pathlib import Path
 
 DEFAULT_VENV = Path("/home/admin/cpfs/wjh/agentic-kv-fresh/.venv")
 TARGET_REL = "lib/python3.12/site-packages/vllm/v1/core/sched/scheduler.py"
+MOONCAKE_REL = (
+    "lib/python3.12/site-packages/vllm/distributed/kv_transfer/"
+    "kv_connector/v1/mooncake/mooncake_connector.py"
+)
+LOGGERS_REL = "lib/python3.12/site-packages/vllm/v1/metrics/loggers.py"
 
 START_MARK = "# MB5_INSTRUMENT_START"
 END_MARK = "# MB5_INSTRUMENT_END"
@@ -165,45 +170,94 @@ SCHED_RET_REPLACE = f"""        {START_MARK}
     def _agentic_emit_step_log("""
 
 
-PATCHES = [
+SCHED_PATCHES = [
     ("header",            HEADER_ANCHOR,     HEADER_ANCHOR + HEADER_INSERT),
     ("schedule() return", SCHED_RET_TARGET,  SCHED_RET_REPLACE),
 ]
 
+# ---------- Patch 3: vLLM 0.18.1 kv_consumer AttributeError fix --------------
+# In MooncakeConnectorWorker.__init__, `self.bootstrap_server` is only assigned
+# inside the `is_kv_producer` branch (around line 615). For kv_consumer roles
+# the attribute is never set, but later code paths (e.g. line ~1060) check
+# `if self.bootstrap_server is not None:` and AttributeError. We initialize it
+# unconditionally just before the role-conditional branch.
+MOONCAKE_ANCHOR = "        self.reqs_need_send: dict[TransferId, SendBlockMeta] = {}\n"
+MOONCAKE_INSERT = (
+    f"        {START_MARK}\n"
+    f"        self.bootstrap_server = None  # vLLM 0.18.1 kv_consumer fix\n"
+    f"        {END_MARK}\n"
+)
 
-def find_target(venv_or_path: Path) -> Path:
-    candidates = [venv_or_path, DEFAULT_VENV / TARGET_REL]
+MOONCAKE_PATCHES = [
+    ("kv_consumer bootstrap_server init", MOONCAKE_ANCHOR,
+     MOONCAKE_ANCHOR + MOONCAKE_INSERT),
+]
+
+# ---------- Patch 4: vLLM 0.18.1 PD-consumer metrics counter underflow ------
+# In PromptTokenStats.update_from_output, local_cache_hit is computed as
+# (num_cached_tokens + recomputed - num_external_computed_tokens). On a
+# kv_consumer, a remote KV transfer can report more external-computed tokens
+# than the scheduler's cached count (esp. on a KV-load failure for a large
+# request), driving local_cache_hit negative. loggers.record() then calls
+# Counter.inc() with that negative value and prometheus_client raises
+# "Counters can only be incremented by non-negative amounts.", which kills the
+# EngineCore — turning one failed request into a total config collapse.
+# We clamp the per-source counts to >= 0 right before they are recorded.
+LOGGERS_ANCHOR = "        pts = iteration_stats.prompt_token_stats\n"
+LOGGERS_INSERT = (
+    f"        {START_MARK}\n"
+    f"        if pts.local_cache_hit < 0:\n"
+    f"            pts.local_cache_hit = 0\n"
+    f"        if pts.computed < 0:\n"
+    f"            pts.computed = 0\n"
+    f"        if pts.external_kv_transfer < 0:\n"
+    f"            pts.external_kv_transfer = 0\n"
+    f"        {END_MARK}\n"
+)
+
+LOGGERS_PATCHES = [
+    ("PD-consumer counter underflow clamp", LOGGERS_ANCHOR,
+     LOGGERS_ANCHOR + LOGGERS_INSERT),
+]
+
+PATCH_FILES = [
+    (TARGET_REL, SCHED_PATCHES),
+    (MOONCAKE_REL, MOONCAKE_PATCHES),
+    (LOGGERS_REL, LOGGERS_PATCHES),
+]
+
+
+def find_target(venv_or_path: Path, rel_path: str) -> Path:
+    candidates = [venv_or_path / rel_path, DEFAULT_VENV / rel_path]
     for c in candidates:
         if c.is_file():
             return c
-        if c.is_dir():
-            sub = c / TARGET_REL
-            if sub.is_file():
-                return sub
-    raise FileNotFoundError(f"cannot find vllm V1 scheduler at {venv_or_path}")
+    raise FileNotFoundError(
+        f"cannot find {rel_path} under {venv_or_path}"
+    )
 
 
 def is_patched(text: str) -> bool:
     return START_MARK in text
 
 
-def apply(target: Path) -> None:
+def apply_one(target: Path, patches: list) -> None:
     text = target.read_text()
     if is_patched(text):
         print(f"[mb5-instr] already patched: {target}")
         return
     new = text
-    for name, src, dst in PATCHES:
+    for name, src, dst in patches:
         if src not in new:
             raise RuntimeError(
                 f"patch {name!r}: anchor not found in {target}."
             )
         new = new.replace(src, dst, 1)
     target.write_text(new)
-    print(f"[mb5-instr] applied {len(PATCHES)} patches -> {target}")
+    print(f"[mb5-instr] applied {len(patches)} patches -> {target}")
 
 
-def revert(target: Path) -> None:
+def revert_one(target: Path) -> None:
     text = target.read_text()
     if not is_patched(text):
         print(f"[mb5-instr] not patched (nothing to revert): {target}")
@@ -225,16 +279,18 @@ def main() -> None:
     p.add_argument("--check", action="store_true")
     p.add_argument("--venv", type=Path, default=DEFAULT_VENV)
     args = p.parse_args()
-    target = find_target(args.venv)
-    if args.apply:
-        apply(target)
-    elif args.revert:
-        revert(target)
-    elif args.check:
-        text = target.read_text()
-        print(f"[mb5-instr] {'PATCHED' if is_patched(text) else 'CLEAN'}: {target}")
-    else:
-        p.error("specify --apply / --revert / --check")
+    for rel_path, patches in PATCH_FILES:
+        target = find_target(args.venv, rel_path)
+        if args.apply:
+            apply_one(target, patches)
+        elif args.revert:
+            revert_one(target)
+        elif args.check:
+            text = target.read_text()
+            state = 'PATCHED' if is_patched(text) else 'CLEAN'
+            print(f"[mb5-instr] {state}: {target}")
+        else:
+            p.error("specify --apply / --revert / --check")
 
 
 if __name__ == "__main__":

microbench/fresh_setup/mb5_launch.sh

@@ -28,7 +28,7 @@ VENV="${FRESH_ROOT}/.venv"
 MODEL="${MODEL:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-Coder-30B-A3B-Instruct}"
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 INSTRUMENT="${SCRIPT_DIR}/instrument_kv_snapshot.py"
-PROXY_SRC="${SCRIPT_DIR}/../../third_party/vllm/examples/online_serving/disaggregated_serving/mooncake_connector/mooncake_connector_proxy.py"
+PROXY_SRC="${SCRIPT_DIR}/mb5_pd_proxy.py"
 
 CONFIG="${CONFIG:-8C}"
 RUN_LABEL="${RUN_LABEL:-default}"
@@ -44,10 +44,21 @@ BASE_BP=8998
 BASE_MASTER=29500
 
 stop_all() {
+    pkill -9 -f "mb5_pd_proxy.py" 2>/dev/null || true
     pkill -9 -f "mooncake_connector_proxy.py" 2>/dev/null || true
     pkill -9 -f "vllm serve" 2>/dev/null || true
     pkill -9 -f "EngineCore" 2>/dev/null || true
     sleep 3
+    # Hard guarantee: required ports must be free before we start. If they
+    # aren't, an earlier run left a stale process holding the socket and the
+    # readiness check would (silently) probe the stale proxy.
+    for port in 8000 8001 8002 8003 8004 8005 8006 8007 "${PROXY_PORT}"; do
+        if ss -ltn 2>/dev/null | awk '{print $4}' | grep -qE "[:.]${port}\$"; then
+            echo "[mb5] FATAL port ${port} still in use after stop_all; manual cleanup needed"
+            ss -ltnp 2>/dev/null | grep -E "[:.]${port}\$" || true
+            exit 1
+        fi
+    done
 }
 
 case "${1:-start}" in
@@ -179,13 +190,17 @@ for port in "${all_ports[@]}"; do
 done
 
 if [ "${ROLES}" = "pd" ]; then
-    echo "[mb5] launching mooncake_connector_proxy on ${PROXY_PORT}"
+    P_ROUTING="${MB5_P_ROUTING:-rr}"
+    echo "[mb5] launching mooncake_connector_proxy on ${PROXY_PORT} (P routing=${P_ROUTING})"
+    MB5_P_ROUTING="${P_ROUTING}" \
     nohup python "${PROXY_SRC}" "${proxy_args[@]}" --port "${PROXY_PORT}" --host 0.0.0.0 \
         > "${LOGS_DIR}/proxy.log" 2>&1 &
     disown
-    # wait for proxy
+    # wait for proxy. Official mooncake_connector_proxy only handles
+    # /v1/completions, so /health and /v1/models return 404 — accept any
+    # HTTP response as "alive".
     tries=0
-    while ! curl -sf "http://127.0.0.1:${PROXY_PORT}/v1/models" >/dev/null 2>&1; do
+    while ! curl -s -o /dev/null -w "%{http_code}" "http://127.0.0.1:${PROXY_PORT}/" 2>/dev/null | grep -qE "^[0-9]"; do
         tries=$((tries+1))
         if [ ${tries} -gt 60 ]; then
             echo "[mb5] FATAL proxy did not come up in 2 min"
@@ -194,7 +209,7 @@ if [ "${ROLES}" = "pd" ]; then
         fi
         sleep 2
     done
-    echo "  proxy port=${PROXY_PORT} ready"
+    echo "  proxy port=${PROXY_PORT} ready (HTTP responding)"
     ENDPOINTS="http://127.0.0.1:${PROXY_PORT}"
 fi
 

microbench/fresh_setup/mb5_pd_proxy.py

@@ -0,0 +1,413 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+import argparse
+import asyncio
+import hashlib
+import ipaddress
+import itertools
+import os
+import urllib
+import uuid
+from contextlib import asynccontextmanager
+from typing import Any
+
+import httpx
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import StreamingResponse
+
+
+def maybe_wrap_ipv6_address(address: str) -> str:
+    try:
+        ipaddress.IPv6Address(address)
+        return f"[{address}]"
+    except ValueError:
+        return address
+
+
+def make_http_path(host: str, port: int) -> str:
+    return f"http://{host}:{port}"
+
+
+def prefiller_cycle(prefill_clients: list[Any]):
+    while True:
+        for prefill_client in prefill_clients:
+            for i in range(prefill_client["dp_size"]):
+                yield prefill_client, i
+
+
+async def get_prefiller_info(prefill_clients: list, ready: asyncio.Event):
+    for prefill_client in prefill_clients:
+        while True:
+            try:
+                # Wait for prefill service to be ready
+                response = await prefill_client["client"].get("/health")
+                response.raise_for_status()
+            except Exception:
+                await asyncio.sleep(1)
+                continue
+
+            response = await prefill_client["client"].get(
+                prefill_client["bootstrap_addr"] + "/query"
+            )
+            response.raise_for_status()
+            data = response.json()
+            break
+
+        for dp_rank, dp_entry in data.items():
+            prefill_client["dp_engine_id"][int(dp_rank)] = dp_entry["engine_id"]
+        dp_size = len(data)
+        prefill_client["dp_size"] = dp_size
+        print(f"Inited prefiller {prefill_client['url']} with dp_size={dp_size}")
+
+    ready.set()
+    print("All prefiller instances are ready.")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """
+    Lifespan context manager to handle startup and shutdown events.
+    """
+    # Startup: Initialize client pools for prefiller and decoder services
+    app.state.prefill_clients = []
+    app.state.decode_clients = []
+    app.state.ready = asyncio.Event()
+
+    # Create prefill clients
+    for i, (url, bootstrap_port) in enumerate(global_args.prefill):
+        parsed_url = urllib.parse.urlparse(url)
+        hostname = maybe_wrap_ipv6_address(parsed_url.hostname)
+        app.state.prefill_clients.append(
+            {
+                "client": httpx.AsyncClient(
+                    timeout=None,
+                    base_url=url,
+                    limits=httpx.Limits(
+                        max_connections=None,
+                        max_keepalive_connections=None,
+                    ),
+                ),
+                "url": url,
+                "bootstrap_addr": make_http_path(hostname, bootstrap_port or 8998),
+                "dp_engine_id": {},
+            }
+        )
+
+    # Create decode clients
+    for i, url in enumerate(global_args.decode):
+        parsed_url = urllib.parse.urlparse(url)
+        hostname = maybe_wrap_ipv6_address(parsed_url.hostname)
+        app.state.decode_clients.append(
+            {
+                "client": httpx.AsyncClient(
+                    timeout=None,
+                    base_url=url,
+                    limits=httpx.Limits(
+                        max_connections=None,
+                        max_keepalive_connections=None,
+                    ),
+                ),
+            }
+        )
+
+    asyncio.create_task(get_prefiller_info(app.state.prefill_clients, app.state.ready))
+
+    # Initialize round-robin iterators
+    app.state.prefill_iterator = prefiller_cycle(app.state.prefill_clients)
+    app.state.decode_iterator = itertools.cycle(range(len(app.state.decode_clients)))
+
+    print(
+        f"Got {len(app.state.prefill_clients)} prefill clients "
+        f"and {len(app.state.decode_clients)} decode clients."
+    )
+
+    yield
+
+    # Shutdown: Close all clients
+    for client_info in app.state.prefill_clients:
+        await client_info["client"].aclose()
+
+    for client_info in app.state.decode_clients:
+        await client_info["client"].aclose()
+
+
+# Update FastAPI app initialization to use lifespan
+app = FastAPI(lifespan=lifespan)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument("--port", type=int, default=8000)
+    # Always use 127.0.0.1 as localhost binds to IPv6 which is blocked on CI
+    parser.add_argument("--host", type=str, default="127.0.0.1")
+
+    # For prefiller instances
+    parser.add_argument(
+        "--prefill",
+        nargs="+",
+        action="append",
+        dest="prefill_raw",
+        metavar=("URL", "bootstrap_port"),
+        help=(
+            "Prefill server URL and optional bootstrap port. "
+            "Can be specified multiple times. "
+            "Format: --prefill URL [BOOTSTRAP_PORT]. "
+            "BOOTSTRAP_PORT can be a port number, "
+            "'none', or omitted (defaults to none)."
+        ),
+    )
+
+    # For decoder instances
+    parser.add_argument(
+        "--decode",
+        nargs=1,
+        action="append",
+        dest="decode_raw",
+        metavar=("URL",),
+        help="Decode server URL. Can be specified multiple times.",
+    )
+
+    args = parser.parse_args()
+    args.prefill = _parse_prefill_urls(args.prefill_raw)
+    args.decode = _parse_decode_urls(args.decode_raw)
+
+    return args
+
+
+# From sglang router_args.py
+def _parse_prefill_urls(prefill_list):
+    """Parse prefill URLs from --prefill arguments.
+
+    Format: --prefill URL [BOOTSTRAP_PORT]
+    Example:
+        --prefill http://prefill1:8080 9000  # With bootstrap port
+        --prefill http://prefill2:8080 none  # Explicitly no bootstrap port
+        --prefill http://prefill3:8080       # Defaults to no bootstrap port
+    """
+    if not prefill_list:
+        return []
+
+    prefill_urls = []
+    for prefill_args in prefill_list:
+        url = prefill_args[0]
+
+        # Handle optional bootstrap port
+        if len(prefill_args) >= 2:
+            bootstrap_port_str = prefill_args[1]
+            # Handle 'none' as None
+            if bootstrap_port_str.lower() == "none":
+                bootstrap_port = None
+            else:
+                try:
+                    bootstrap_port = int(bootstrap_port_str)
+                except ValueError as e:
+                    raise ValueError(
+                        f"Invalid bootstrap port: {bootstrap_port_str}. Must be a number or 'none'"  # noqa: E501
+                    ) from e
+        else:
+            # No bootstrap port specified, default to None
+            bootstrap_port = None
+
+        prefill_urls.append((url, bootstrap_port))
+
+    return prefill_urls
+
+
+def _parse_decode_urls(decode_list):
+    """Parse decode URLs from --decode arguments.
+
+    Format: --decode URL
+    Example: --decode http://decode1:8081 --decode http://decode2:8081
+    """
+    if not decode_list:
+        return []
+
+    # decode_list is a list of single-element lists due to nargs=1
+    return [url[0] for url in decode_list]
+
+
+# MB5: routing mode for the prefill (producer) side.
+#   "rr"      — round-robin (official upstream behavior)
+#   "session" — consistent hash on X-Session-Id, so all turns of a session
+#               land on the same producer and reuse its prefix cache.
+# Decode side stays round-robin (load balance) regardless.
+MB5_P_ROUTING = os.environ.get("MB5_P_ROUTING", "rr").lower()
+
+
+def get_prefill_by_session(app, session_id: str):
+    """Pick a (prefill_client, dp_rank) deterministically from session_id.
+
+    Uses a stable (non-PYTHONHASHSEED-dependent) hash so the mapping is
+    reproducible across processes. dp_size is usually 1 here (TP=1, no DP),
+    but we hash into the flat (client, dp_rank) slot space to stay correct
+    if a producer ever reports dp_size > 1.
+    """
+    clients = app.state.prefill_clients
+    slots = [(c, r) for c in clients for r in range(max(1, c.get("dp_size", 1)))]
+    h = int(hashlib.md5(session_id.encode()).hexdigest()[:8], 16)
+    return slots[h % len(slots)]
+
+
+def get_next_client(app, service_type: str):
+    """
+    Get the next client in round-robin fashion.
+
+    Args:
+        app: The FastAPI app instance
+        service_type: Either 'prefill' or 'decode'
+
+    Returns:
+        The next client to use
+    """
+    if service_type == "prefill":
+        return next(app.state.prefill_iterator)
+    elif service_type == "decode":
+        client_idx = next(app.state.decode_iterator)
+        return app.state.decode_clients[client_idx]
+    else:
+        raise ValueError(f"Unknown service type: {service_type}")
+
+
+async def send_request_to_service(
+    client_info: dict, dp_rank: int, endpoint: str, req_data: dict, request_id: str
+):
+    """
+    Send a request to a service using a client from the pool.
+    """
+    req_data = req_data.copy()
+    req_data["kv_transfer_params"] = {
+        "do_remote_decode": True,
+        "do_remote_prefill": False,
+        "transfer_id": f"xfer-{request_id}",
+    }
+    req_data["stream"] = False
+    req_data["max_tokens"] = 1
+    # MB5 fix: clients (our replayer) may set min_tokens to enforce a fixed
+    # output length. After the proxy caps max_tokens=1 on the prefill leg,
+    # any min_tokens > 1 violates vLLM's `min_tokens <= max_tokens` check.
+    if "min_tokens" in req_data:
+        req_data["min_tokens"] = 1
+    if "max_completion_tokens" in req_data:
+        req_data["max_completion_tokens"] = 1
+    if "stream_options" in req_data:
+        del req_data["stream_options"]
+    headers = {
+        "Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY')}",
+        "X-Request-Id": request_id,
+        "X-data-parallel-rank": str(dp_rank),
+    }
+
+    response = await client_info["client"].post(
+        endpoint, json=req_data, headers=headers
+    )
+    response.raise_for_status()
+
+    # CRITICAL: Release connection back to pool
+    await response.aclose()
+
+
+async def stream_service_response(
+    prefill_client_info: dict,
+    prefill_dp_rank: int,
+    decode_client_info: dict,
+    endpoint: str,
+    req_data: dict,
+    request_id: str,
+):
+    """
+    Asynchronously stream response from a service using a client from the pool.
+    """
+    headers = {
+        "Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY')}",
+        "X-Request-Id": request_id,
+    }
+
+    req_data["kv_transfer_params"] = {
+        "do_remote_decode": False,
+        "do_remote_prefill": True,
+        "remote_bootstrap_addr": prefill_client_info["bootstrap_addr"],
+        "remote_engine_id": prefill_client_info["dp_engine_id"][prefill_dp_rank],
+        "transfer_id": f"xfer-{request_id}",
+    }
+
+    async with decode_client_info["client"].stream(
+        "POST", endpoint, json=req_data, headers=headers
+    ) as response:
+        response.raise_for_status()
+        async for chunk in response.aiter_bytes():
+            yield chunk
+
+
+async def _handle_completions(api: str, request: Request):
+    if not app.state.ready.is_set():
+        raise HTTPException(status_code=503, detail="Service Unavailable")
+
+    try:
+        req_data = await request.json()
+        request_id = str(uuid.uuid4())
+
+        # Select the prefill (producer) client.
+        if MB5_P_ROUTING == "session":
+            session_id = request.headers.get("X-Session-Id") or request_id
+            prefill_client_info, prefill_dp_rank = get_prefill_by_session(
+                request.app, session_id
+            )
+        else:
+            # Round-robin (official upstream behavior).
+            prefill_client_info, prefill_dp_rank = get_next_client(
+                request.app, "prefill"
+            )
+
+        # Send request to prefill service
+        asyncio.create_task(
+            send_request_to_service(
+                prefill_client_info, prefill_dp_rank, api, req_data, request_id
+            )
+        )
+
+        decode_client_info = get_next_client(request.app, "decode")
+
+        # Stream response from decode service
+        async def generate_stream():
+            async for chunk in stream_service_response(
+                prefill_client_info,
+                prefill_dp_rank,
+                decode_client_info,
+                api,
+                req_data,
+                request_id=request_id,
+            ):
+                yield chunk
+
+        return StreamingResponse(generate_stream(), media_type="application/json")
+
+    except Exception as e:
+        import sys
+        import traceback
+
+        exc_info = sys.exc_info()
+        print(f"Error occurred in disagg prefill proxy server - {api} endpoint")
+        print(e)
+        print("".join(traceback.format_exception(*exc_info)))
+        raise
+
+
+@app.post("/v1/completions")
+async def handle_completions(request: Request):
+    return await _handle_completions("/v1/completions", request)
+
+
+@app.post("/v1/chat/completions")
+async def handle_chat_completions(request: Request):
+    return await _handle_completions("/v1/chat/completions", request)
+
+
+if __name__ == "__main__":
+    global global_args
+    global_args = parse_args()
+
+    import uvicorn
+
+    uvicorn.run(app, host=global_args.host, port=global_args.port)

microbench/fresh_setup/plot_kv_pool_timeline.py

@@ -70,6 +70,29 @@ def plot_one_instance(snaps: list[dict], out: Path, title: str) -> None:
     # Sort by first-seen time so the band order follows arrival
     all_req_ids.sort(key=lambda r: req_first_seen[r])
 
+    if not all_req_ids:
+        # No requests ever ran on this instance; plot a flat used_blocks line
+        # instead of the stackplot (which can't handle empty input).
+        fig, ax1 = plt.subplots(figsize=(13, 4))
+        used = [s["used_blocks"] for s in snaps]
+        ax1.plot(times, used, color="#888", lw=1.5, label="used_blocks (no running reqs sampled)")
+        ax1.axhline(total_blocks, color="#444", lw=1.5, ls="-",
+                    label=f"pool total = {total_blocks} blocks")
+        ax1.axhline(total_blocks * 0.9, color="#c44e52", lw=1.2, ls="--", alpha=0.7,
+                    label="90% capacity")
+        ax1.set_ylabel("KV blocks")
+        ax1.set_ylim(0, total_blocks * 1.05)
+        ax1.set_xlabel("wall-clock since first snapshot (s)")
+        ax1.set_title(title + "  [no per-request data; instance idle?]")
+        ax1.legend(loc="upper right", fontsize=9)
+        ax1.grid(True, alpha=0.3)
+        out.parent.mkdir(parents=True, exist_ok=True)
+        fig.tight_layout()
+        fig.savefig(out, dpi=120)
+        plt.close(fig)
+        print(f"wrote {out}  (n_snapshots={len(snaps)}, 0 running reqs ever)")
+        return
+
     matrix = np.zeros((len(all_req_ids), len(times)), dtype=np.int64)
     req_to_row = {r: i for i, r in enumerate(all_req_ids)}
     for j, s in enumerate(snaps):

scripts/working_set_analysis.py

@@ -0,0 +1,276 @@
+"""KV-cache working-set sizing for agentic traces, across GPU / model / parallelism.
+
+WHAT IT COMPUTES
+  hash_ids in these traces are global content-addressed block ids (same content
+  -> same id; reuse = repeated id). vLLM prefix cache is block-level, so the
+  cluster-wide KV footprint at any instant = the set of distinct block ids that
+  must be resident. Session/instance placement only moves blocks between GPUs;
+  it does not change this aggregate, so the analysis is placement-independent.
+
+  Three working-set notions, swept over a retention window T:
+    W_all          retain every block forever          (true upper bound)
+    W_oracle       keep block in [first_use, last_use]  (Belady foresight floor)
+    W_denning(T)   distinct blocks touched in (t-T, t]  (realistic TTL=T LRU)
+  and the APC actually captured at each T (validates vs the trie ceiling).
+
+HARDWARE MODEL
+  KV pool per serving replica =
+      gpus_per_replica * hbm_per_gpu  -  model_weights  -  activation_reserve
+  (TP/EP shard weights+KV across the replica's GPUs; the *aggregate* KV pool is
+   what we size against, so only gpus_per_replica and total weights matter.)
+
+  KV bytes / token:
+    GQA/MHA :  2 * L * kv_heads * head_dim * kv_dtype_bytes
+    MLA     :  L * (kv_lora_rank + qk_rope_head_dim) * kv_dtype_bytes
+  (matches kvcache-simulator/src/config.rs::kv_block_bytes)
+
+All sizes reported in GB = 1e9 bytes (matches the simulator's `hbm_bytes` e9
+convention).
+"""
+from __future__ import annotations
+import argparse, json
+import numpy as np
+
+GB = 1e9
+
+# Nominal HBM per GPU, in GB (decimal).
+GPU_HBM_GB = {
+    "H100": 80, "H200": 141, "H20": 96, "H20-141G": 141,
+    "A100-40G": 40, "A100-80G": 80,
+    "B200": 192, "B300": 288, "GB200": 192,
+}
+
+
+# ----------------------------------------------------------------------------- model
+def load_model(config_json: str) -> dict:
+    v = json.load(open(config_json))
+    L = int(v["num_hidden_layers"])
+    out = {"name": v.get("model_type", "?"), "L": L}
+    if "kv_lora_rank" in v:                       # MLA (DeepSeek / GLM-MoE-DSA)
+        out["mla"] = True
+        out["kv_lora_rank"] = int(v["kv_lora_rank"])
+        out["qk_rope_head_dim"] = int(v["qk_rope_head_dim"])
+    else:                                          # GQA / MHA
+        out["mla"] = False
+        H = int(v.get("num_attention_heads", 0))
+        out["kv_heads"] = int(v.get("num_key_value_heads", H) or H)
+        out["head_dim"] = int(v.get("head_dim") or (v["hidden_size"] // H))
+    return out
+
+
+def kv_bytes_per_token(model: dict, kv_dtype_bytes: int) -> int:
+    L = model["L"]
+    if model["mla"]:
+        return L * (model["kv_lora_rank"] + model["qk_rope_head_dim"]) * kv_dtype_bytes
+    return 2 * L * model["kv_heads"] * model["head_dim"] * kv_dtype_bytes
+
+
+# ----------------------------------------------------------------------------- trace
+def load_trace(path: str, min_ts=None, max_ts=None):
+    ids, ts = [], []
+    n = dropped = 0
+    with open(path) as fh:
+        for line in fh:
+            line = line.strip()
+            if not line:
+                continue
+            r = json.loads(line)
+            h = r.get("hash_ids")
+            if isinstance(h, str):
+                h = json.loads(h)
+            if not h:
+                continue
+            t = float(r.get("timestamp", 0.0))
+            if (min_ts is not None and t < min_ts) or (max_ts is not None and t > max_ts):
+                dropped += 1
+                continue
+            ids.extend(h)
+            ts.extend([t] * len(h))
+            n += 1
+    if dropped:
+        print(f"  (clipped {dropped} reqs outside [{min_ts}, {max_ts}])")
+    return n, np.asarray(ids, dtype=np.int64), np.asarray(ts, dtype=np.float64)
+
+
+def _sweep_peak(starts, ends):
+    """Peak concurrency of intervals [start, end); ends applied before starts at ties."""
+    ev = np.concatenate([starts, ends])
+    d = np.concatenate([np.ones(len(starts), np.int64), -np.ones(len(ends), np.int64)])
+    order = np.lexsort((d, ev))                    # at equal time: -1 (end) before +1 (start)
+    return int(np.cumsum(d[order]).max())
+
+
+def _series(starts, ends, grid):
+    s = np.sort(starts); e = np.sort(ends)
+    return np.searchsorted(s, grid, side="right") - np.searchsorted(e, grid, side="right")
+
+
+def compute_working_set(ids, ts, taus):
+    """Return dict with appearance stats + per-tau Denning peaks + oracle/all."""
+    A = len(ids)
+    order = np.lexsort((ts, ids))
+    ids_s, ts_s = ids[order], ts[order]
+    same_prev = np.empty(A, bool); same_prev[0] = False
+    same_prev[1:] = ids_s[1:] == ids_s[:-1]
+    same_next = np.empty(A, bool); same_next[-1] = False
+    same_next[:-1] = ids_s[:-1] == ids_s[1:]
+    prev_gap = np.full(A, np.inf); prev_gap[1:][same_prev[1:]] = (ts_s[1:] - ts_s[:-1])[same_prev[1:]]
+    next_gap = np.full(A, np.inf); next_gap[:-1][same_next[:-1]] = (ts_s[1:] - ts_s[:-1])[same_next[:-1]]
+
+    n_unique = int((~same_prev).sum())
+    grid = np.linspace(ts.min(), ts.max(), 400)
+
+    # oracle [first,last]
+    first = np.full(ids.max() + 1, np.inf); last = np.full(ids.max() + 1, -np.inf)
+    np.minimum.at(first, ids, ts); np.maximum.at(last, ids, ts)
+    seen = np.isfinite(first)
+    oracle_peak = _sweep_peak(first[seen], last[seen])
+
+    rows = []
+    for T in taus:
+        enter = ts_s[prev_gap > T]
+        exit_ = ts_s[next_gap > T] + T
+        peak = _sweep_peak(enter, exit_)
+        ser = _series(enter, exit_, grid)
+        rows.append({
+            "tau": T, "peak_blocks": peak,
+            "p99_blocks": float(np.percentile(ser, 99)),
+            "p50_blocks": float(np.percentile(ser, 50)),
+            "apc": float((prev_gap <= T).sum() / A),
+        })
+    return {
+        "A": A, "n_unique": n_unique, "n_reuse": A - n_unique,
+        "apc_ceiling": (A - n_unique) / A,
+        "oracle_peak_blocks": oracle_peak,
+        "span": float(ts.max() - ts.min()),
+        "taus": rows,
+    }
+
+
+# ----------------------------------------------------------------------------- plot
+def plot(ws, hw, block_bytes, label, out_path):
+    import matplotlib
+    matplotlib.use("Agg")
+    import matplotlib.pyplot as plt
+
+    bgb = block_bytes / GB
+    taus = [r["tau"] for r in ws["taus"]]
+    peak_gb = np.array([r["peak_blocks"] * bgb for r in ws["taus"]])
+    apc = np.array([r["apc"] * 100 for r in ws["taus"]])
+    oracle_gb = ws["oracle_peak_blocks"] * bgb
+    ceil = ws["apc_ceiling"] * 100
+    pool = hw["kv_pool_gb"]                      # per replica
+
+    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
+
+    # --- panel 1: APC vs required KV footprint ---
+    ax1.plot(peak_gb, apc, "o-", color="#1f77b4", lw=2, ms=7, label="TTL-LRU W(T)")
+    for r, x, y in zip(ws["taus"], peak_gb, apc):
+        ax1.annotate(f"{r['tau']:g}s", (x, y), fontsize=8,
+                     textcoords="offset points", xytext=(4, 5))
+    ax1.scatter([oracle_gb], [ceil], marker="*", s=320, color="#d62728", zorder=5,
+                label=f"oracle / ceiling ({ceil:.1f}%)")
+    ax1.axhline(ceil, ls=":", color="#d62728", alpha=.5)
+    for k in (1, 2, 4, 8):
+        x = pool * k
+        ax1.axvline(x, ls="--", color="#2ca02c", alpha=.55)
+        ax1.text(x, 2, f"{k} replica\n{k*hw['gpus_per_replica']} GPU",
+                 rotation=90, va="bottom", ha="right", fontsize=8, color="#2ca02c")
+    ax1.set_xscale("log")
+    ax1.set_xlabel("KV footprint that must be resident  (GB, log)")
+    ax1.set_ylabel("Achievable prefix-cache hit rate  (APC %)")
+    ax1.set_title("APC vs KV-pool budget")
+    ax1.grid(alpha=.3, which="both"); ax1.legend(loc="lower right"); ax1.set_ylim(0, 100)
+
+    # --- panel 2: footprint over time for a few T ---
+    span = ws["span"]; grid = np.linspace(0, span, 400)
+    # recompute series for a representative subset from stored peaks is not enough;
+    # show peak/p50 bars instead (compact, robust)
+    sel = [r for r in ws["taus"] if r["tau"] in (2, 30, 300, 600)]
+    xs = np.arange(len(sel)); w = 0.38
+    ax2.bar(xs - w/2, [r["peak_blocks"]*bgb for r in sel], w, label="peak", color="#1f77b4")
+    ax2.bar(xs + w/2, [r["p50_blocks"]*bgb for r in sel], w, label="median", color="#aec7e8")
+    ax2.axhline(pool, ls="--", color="#2ca02c", lw=2, label=f"1 replica KV pool ({pool:.0f} GB)")
+    ax2.axhline(oracle_gb, ls=":", color="#d62728", lw=2, label=f"oracle full-ceiling ({oracle_gb:.0f} GB)")
+    ax2.set_xticks(xs); ax2.set_xticklabels([f"T={r['tau']:g}s\nAPC={r['apc']*100:.0f}%" for r in sel])
+    ax2.set_ylabel("KV footprint (GB)")
+    ax2.set_yscale("log")
+    ax2.set_title("Footprint by retention window vs pool")
+    ax2.grid(alpha=.3, axis="y", which="both"); ax2.legend(loc="upper left", fontsize=9)
+
+    fig.suptitle(label, fontsize=13, fontweight="bold")
+    fig.tight_layout(rect=[0, 0, 1, 0.97])
+    fig.savefig(out_path, dpi=130)
+    print(f"  figure -> {out_path}")
+
+
+# ----------------------------------------------------------------------------- main
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("trace")
+    ap.add_argument("--model-config", required=True, help="path to HF config.json")
+    ap.add_argument("--gpu", required=True, choices=sorted(GPU_HBM_GB))
+    ap.add_argument("--tp", type=int, default=8)
+    ap.add_argument("--pp", type=int, default=1)
+    ap.add_argument("--ep", type=int, default=0, help="informational only (KV unchanged by EP)")
+    ap.add_argument("--kv-dtype-bytes", type=int, default=1, help="1=FP8, 2=BF16")
+    ap.add_argument("--weight-gb", type=float, required=True, help="total resident model weights, GB")
+    ap.add_argument("--activation-gb", type=float, default=32.0, help="activation+ctx reserve, GB")
+    ap.add_argument("--block-size", type=int, default=512)
+    ap.add_argument("--min-ts", type=float, default=None, help="drop reqs with timestamp < this")
+    ap.add_argument("--max-ts", type=float, default=None, help="drop reqs with timestamp > this")
+    ap.add_argument("--label", default="")
+    ap.add_argument("--out", default="figs/working_set.png")
+    a = ap.parse_args()
+
+    model = load_model(a.model_config)
+    kv_tok = kv_bytes_per_token(model, a.kv_dtype_bytes)
+    block_bytes = kv_tok * a.block_size
+
+    gpus_per_replica = a.tp * a.pp
+    total_hbm = gpus_per_replica * GPU_HBM_GB[a.gpu]
+    kv_pool_gb = total_hbm - a.weight_gb - a.activation_gb
+    hw = {"gpus_per_replica": gpus_per_replica, "kv_pool_gb": kv_pool_gb}
+
+    taus = [1, 2, 5, 10, 30, 60, 300, 600, 1800]
+    n, ids, ts = load_trace(a.trace, a.min_ts, a.max_ts)
+    ws = compute_working_set(ids, ts, taus)
+
+    label = a.label or f"{model['name']} {a.gpu} TP{a.tp}" + (f" EP{a.ep}" if a.ep else "")
+    print("=" * 84)
+    print(f"  {label}")
+    print("=" * 84)
+    print(f"  model            {model['name']}  L={model['L']}  "
+          + (f"MLA(kv_lora={model['kv_lora_rank']}+rope={model['qk_rope_head_dim']})"
+             if model["mla"] else f"GQA(kv_heads={model['kv_heads']}xhd={model['head_dim']})"))
+    print(f"  KV / token       {kv_tok:,} B ({kv_tok/1024:.1f} KiB)   KV / block({a.block_size}) {block_bytes/1e6:.1f} MB")
+    print(f"  hardware         {gpus_per_replica}x {a.gpu} ({GPU_HBM_GB[a.gpu]} GB) = {total_hbm:.0f} GB HBM/replica"
+          + (f"  EP={a.ep}" if a.ep else ""))
+    print(f"  weights {a.weight_gb:.0f} GB ({a.kv_dtype_bytes}B-KV) + act {a.activation_gb:.0f} GB"
+          f"  =>  KV pool/replica = {kv_pool_gb:.0f} GB")
+    print()
+    print(f"  trace            {n:,} reqs  span {ws['span']:.0f}s ({ws['span']/3600:.2f}h)  QPS~{n/ws['span']:.1f}")
+    print(f"  block appearances {ws['A']:,}   distinct {ws['n_unique']:,}   APC ceiling {ws['apc_ceiling']*100:.2f}%")
+    bgb = block_bytes / GB
+    print(f"  W_all (retain forever)  {ws['n_unique']*bgb:>10,.0f} GB"
+          f"  = {ws['n_unique']*bgb/kv_pool_gb:6.1f} replicas ({ws['n_unique']*bgb/kv_pool_gb*gpus_per_replica:,.0f} GPU)")
+    print(f"  W_oracle (full ceiling) {ws['oracle_peak_blocks']*bgb:>10,.0f} GB"
+          f"  = {ws['oracle_peak_blocks']*bgb/kv_pool_gb:6.1f} replicas ({ws['oracle_peak_blocks']*bgb/kv_pool_gb*gpus_per_replica:,.0f} GPU)")
+    print()
+    print(f"  {'T':>7} | {'peak GB':>9} {'p50 GB':>8} | {'replicas':>8} {'GPUs':>6} | {'APC@T':>6}")
+    print("  " + "-" * 60)
+    for r in ws["taus"]:
+        pg = r["peak_blocks"] * bgb
+        rep = pg / kv_pool_gb
+        print(f"  {r['tau']:>6g}s | {pg:>9,.0f} {r['p50_blocks']*bgb:>8,.0f} | "
+              f"{rep:>8.1f} {rep*gpus_per_replica:>6.0f} | {r['apc']*100:>5.1f}%")
+    print()
+    print(f"  [ref] 1 replica = {gpus_per_replica} GPU = {kv_pool_gb:.0f} GB KV pool")
+
+    import os
+    os.makedirs(os.path.dirname(a.out) or ".", exist_ok=True)
+    plot(ws, hw, block_bytes, label, a.out)
+
+
+if __name__ == "__main__":
+    main()