B3 policies: pseudocode reference for the five-policy sweep

Documents each pick_instance_* function from cache_aware_proxy.py in
pseudocode so the policy semantics can be cited without re-reading
implementation details. Covers lmetric (main baseline), load_only
(no cache / no affinity control), sticky (hard affinity control),
unified (gated affinity + LMetric fallback), and capped (lmetric on
a per-session turn-capped trace).

Includes a decision matrix that maps each policy to whether it uses
session affinity, cache awareness, load awareness, and overload
break, plus a one-liner per control explaining what comparison
isolates which factor.

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

analysis/characterization/b2_sweep_analysis.py

@@ -0,0 +1,117 @@
+"""Aggregate B2 microbench cells into a single interference-index sweep summary.
+
+Per cell (variant × prefill_size):
+- read metrics.jsonl + run_window.json
+- slice the shared engine_*.jsonl by run window
+- run interference_index() against the slice
+- record (variant, prefill_size, n_overlap, n_clean, tpot_p90_*, idx)
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+from analysis.characterization.joined_analysis import (
+    _percentile,
+    _vllm_rid_matches,
+    interference_index,
+    load_engine_state,
+    load_jsonl,
+    write_json,
+)
+
+
+def _slice_engine_state(
+    engine_state_by_worker: dict[str, list[dict]],
+    t_start: float,
+    t_end: float,
+) -> dict[str, list[dict]]:
+    sliced: dict[str, list[dict]] = {}
+    for worker, steps in engine_state_by_worker.items():
+        sliced[worker] = [s for s in steps
+                          if t_start <= (s.get("t_unix") or 0.0) <= t_end]
+    return sliced
+
+
+def _to_joined_shape(metrics_rows: list[dict], variant: str) -> list[dict]:
+    """Adapt B2 metric rows to what interference_index expects."""
+    joined: list[dict] = []
+    for r in metrics_rows:
+        if r.get("workload") != "decode":
+            continue
+        joined.append({
+            "request_id": r["request_id"],
+            "tpot_s": r.get("tpot_s"),
+            "ttft_s": r.get("ttft_s"),
+            "latency_s": r.get("latency_s"),
+            "endpoint_url": r.get("endpoint"),
+            "routed_to": r.get("endpoint"),
+            "t_first_token_unix": (
+                (r["t_dispatch_unix"] + r["ttft_s"])
+                if r.get("ttft_s") is not None
+                and r.get("t_dispatch_unix") is not None else None
+            ),
+            "t_finish_unix": r.get("t_finish_unix"),
+            "error": r.get("error"),
+        })
+    return joined
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="B2 sweep aggregation")
+    p.add_argument("--sweep-dir", type=Path, required=True,
+                   help="Top-level dir produced by scripts/b2_interference.py")
+    p.add_argument("--engine-state-dir", type=Path, required=True)
+    p.add_argument("--worker-map", type=str, required=True,
+                   help="URL=worker_id pairs, comma-separated")
+    p.add_argument("--out", type=Path, default=None)
+    args = p.parse_args()
+
+    worker_map = {}
+    for entry in args.worker_map.split(","):
+        url, _, wid = entry.strip().partition("=")
+        if url and wid:
+            worker_map[url] = wid
+
+    engine_state = load_engine_state(args.engine_state_dir)
+    rows: list[dict] = []
+    for variant_dir in sorted(args.sweep_dir.glob("*/")):
+        if variant_dir.name in ("logs",):
+            continue
+        for cell_dir in sorted(variant_dir.glob("p*/")):
+            window_path = cell_dir / "run_window.json"
+            metrics_path = cell_dir / "metrics.jsonl"
+            if not window_path.exists() or not metrics_path.exists():
+                continue
+            window = json.loads(window_path.read_text())
+            metrics_rows = load_jsonl(metrics_path)
+            joined = _to_joined_shape(metrics_rows, variant_dir.name)
+            sliced = _slice_engine_state(
+                engine_state, window["t_start_unix"], window["t_end_unix"],
+            )
+            idx = interference_index(joined, sliced, worker_map)
+            rows.append({
+                "variant": variant_dir.name,
+                "prefill_size": int(window["prefill_size"]),
+                "decode_endpoint": window["decode_endpoint"],
+                "prefill_endpoint": window["prefill_endpoint"],
+                "n_decode_requests": sum(1 for r in metrics_rows
+                                          if r.get("workload") == "decode"
+                                          and r.get("error") is None),
+                "n_prefill_injections": sum(1 for r in metrics_rows
+                                             if r.get("workload") == "prefill"
+                                             and r.get("error") is None),
+                **idx,
+            })
+    summary = {"rows": rows}
+    out_path = args.out or args.sweep_dir / "b2_sweep_summary.json"
+    write_json(out_path, summary)
+    print(json.dumps(rows, indent=2))
+
+
+if __name__ == "__main__":
+    main()

analysis/characterization/b3_policies_pseudocode.md

@@ -0,0 +1,167 @@
+# B3 Routing Policies — Pseudocode
+
+Reference: `scripts/cache_aware_proxy.py`. All five policies share the
+same per-worker state machine; only the per-request `pick_instance_*`
+function differs.
+
+## Shared per-instance state
+
+```text
+inst.url
+inst.ongoing_tokens        # sum of input_length across in-flight reqs
+inst.pending_prefill_tokens
+inst.ongoing_decode_tokens
+inst.num_requests          # waiting + running
+inst.cached_blocks         # LRU set of 512-token block hashes
+inst.estimate_cache_hit(tokens) -> int
+                           # longest prefix of `tokens` (in BLOCK_SIZE
+                           # chunks) currently in cached_blocks
+```
+
+Each pick is one round-trip on every routing decision; counters are
+mutated when a request starts/finishes, not inside the picker.
+
+## 1. `lmetric` — main baseline
+
+Pure per-request LMetric scoring. No session affinity, no
+overload-break logic.
+
+```text
+def pick_lmetric(instances, token_ids, input_length):
+    best, best_score = None, +inf
+    for inst in instances:
+        cache_hit   = inst.estimate_cache_hit(token_ids)
+        new_prefill = max(0, input_length - cache_hit)
+        p_tokens    = inst.pending_prefill_tokens + new_prefill
+        bs          = inst.num_requests
+        score       = p_tokens * bs
+        if score < best_score:
+            best, best_score = inst, score
+    return best
+```
+
+Intuition: prefer the instance where the expected new prefill cost
+times the running batch size is smallest. Cache hit reduces
+`new_prefill`, so cache-warm workers win at equal load.
+
+## 2. `load_only` — B3 control (no cache, no affinity)
+
+```text
+def pick_load_only(instances):
+    return min(instances, key=lambda inst: inst.num_requests)
+```
+
+Ties: Python `min` returns the first-seen, so when `num_requests` is
+equal across all instances (e.g. fresh start), pick always lands on
+`instances[0]`. This produces unintentional stickiness at low
+concurrency — the B3 lmetric/load_only comparison reads APC=54.1%
+for load_only partly because of that.
+
+## 3. `sticky` — B3 control (hard affinity)
+
+Once a session is bound, never break the binding under any load.
+
+```text
+def pick_sticky(instances, session_id, affinity):
+    if session_id in affinity:
+        return instances[affinity[session_id]]  # unconditional
+    chosen = min(instances, key=lambda i: i.num_requests)
+    affinity[session_id] = index_of(chosen)
+    return chosen
+```
+
+This is the upper bound on locality and the worst case on hot-spot
+behavior — a single heavy session pins one worker forever.
+
+## 4. `unified` — hybrid affinity + LMetric fallback
+
+Sticks to the affinity worker only when the cache is genuinely warm
+and the affinity worker is not overloaded; otherwise falls back to
+LMetric with a 4-key tie-breaker.
+
+```text
+def pick_unified(instances, token_ids, input_length, session_id, affinity):
+    avg_reqs = max(mean(inst.num_requests for inst in instances), 1.0)
+
+    # Affinity gate (both must hold)
+    if session_id in affinity:
+        a = instances[affinity[session_id]]
+        a_hit_ratio = a.estimate_cache_hit(token_ids) / max(input_length, 1)
+        if a_hit_ratio > 0.5 \
+                and a.num_requests <= avg_reqs * OVERLOAD_FACTOR:
+            return a   # stick
+
+    # LMetric fallback with multi-key tie-break
+    keys = []
+    for inst in instances:
+        cache_hit   = inst.estimate_cache_hit(token_ids)
+        new_prefill = max(0, input_length - cache_hit)
+        p_tokens    = inst.pending_prefill_tokens + new_prefill
+        bs          = inst.num_requests
+        score       = p_tokens * bs
+        keys.append((score, new_prefill, bs, idx_of(inst)))
+
+    best_3tuple = min(k[:3] for k in keys)
+    tied = [k for k in keys if k[:3] == best_3tuple]
+    if len(tied) > 1:
+        # Round-robin among ties so brand-new traffic doesn't pin
+        # instance 0 when BS=0 across the board.
+        winner = tied[_rr_counter % len(tied)]
+        _rr_counter += 1
+    else:
+        winner = tied[0]
+    return instances[winner.idx]
+```
+
+Tie-break ordering: `score` (LMetric primary), then `new_prefill`
+(prefer the most cache-warm at equal score), then `num_requests`
+(prefer least-loaded), then a global round-robin counter.
+
+`OVERLOAD_FACTOR` defaults to 2.0; when the affinity worker is
+above 2× average load, the picker treats it as overloaded and steers
+away.
+
+## 5. `capped` — `lmetric` on a session-mass-capped trace
+
+Not a new picker. The picker is the same `pick_lmetric` from §1; the
+input trace is preprocessed.
+
+```text
+def build_capped_trace(input_path, output_path, MAX_TURNS=8):
+    by_session = group_by_session_id(load(input_path))
+    capped = []
+    for sid, turns in by_session.items():
+        turns.sort_by(lambda t: (t.turn_id, t.timestamp))
+        capped.extend(turns[:MAX_TURNS])
+    capped.sort_by(timestamp)        # restore wall-clock order
+    write_jsonl(capped, output_path)
+
+# At run time:
+trace      = build_capped_trace("w600_r0.0015_st30.jsonl")
+picker     = pick_lmetric
+```
+
+For this trace `MAX_TURNS=8` truncates the heavy-tail sessions (full
+trace turns/session p90=1, p99=18, max=3091). The intent is to
+isolate "what does LMetric look like when no session is heavy
+enough to hot-spot a worker?" — comparing capped vs lmetric is the
+session-mass ablation.
+
+## Decision matrix
+
+|  | session affinity | cache aware | load aware | overload break |
+|---|---|---|---|---|
+| `lmetric` | ✗ | ✓ (via `cache_hit` → `new_prefill`) | ✓ (`num_requests` BS factor) | n/a |
+| `load_only` | ✗ | ✗ | ✓ (`num_requests` only) | n/a |
+| `sticky` | ✓ (hard) | ✗ (relies on physical hits, not scoring) | only on first turn | **never** |
+| `unified` | ✓ (gated) | ✓ | ✓ | gate: `cache_ratio>0.5` AND `num_req ≤ 2× avg` |
+| `capped` | same as `lmetric`; the trace itself is truncated | | | |
+
+## What each control isolates
+
+- `lmetric` vs `load_only` → contribution of cache awareness alone.
+- `lmetric` vs `sticky` → contribution of session affinity vs
+  per-request LMetric scoring at the cost of hot-spot.
+- `lmetric` vs `unified` → did adding gated session affinity help?
+- `lmetric` vs `capped` → how much of the residual hot-spot in
+  `lmetric` is driven by heavy-tail sessions specifically?

analysis/characterization/joined_analysis.py

@@ -204,6 +204,7 @@ def _fractions(*parts: int) -> JsonDict:
 def interference_index(
     joined: list[JsonDict],
     engine_state_by_worker: dict[str, list[JsonDict]],
+    worker_map: dict[str, str] | None = None,
 ) -> JsonDict:
     """Label each completed request's decode period as overlap / no-overlap.
 
@@ -219,7 +220,12 @@ def interference_index(
     tpot_clean: list[float] = []
     for r in joined:
         rid = r.get("request_id")
-        worker = _normalize_worker(r.get("endpoint_url") or r.get("routed_to"))
+        # routed_to is the vLLM worker URL; endpoint_url is the proxy URL.
+        # For worker-id matching we want routed_to.
+        worker = _resolve_worker(
+            r.get("routed_to") or r.get("endpoint_url"),
+            worker_map,
+        )
         steps = engine_state_by_worker.get(worker)
         if not steps:
             continue
@@ -239,9 +245,15 @@ def interference_index(
             # this request's own prefill warming up, not interference.
             other_prefill = False
             for pr in s.get("per_req", []) or []:
-                if pr.get("phase") == "prefill" and pr.get("rid") != rid:
-                    other_prefill = True
-                    break
+                if pr.get("phase") != "prefill":
+                    continue
+                # vLLM rewrites rid as `cmpl-<our_id>-<idx>-<hash>`; strip
+                # the prefix and the trailing suffix so equality to our
+                # proxy request_id works.
+                if _vllm_rid_matches(pr.get("rid"), rid):
+                    continue
+                other_prefill = True
+                break
             if other_prefill:
                 overlap = True
                 break
@@ -264,12 +276,15 @@ def interference_index(
 
 
 def _normalize_worker(url_or_id: str | None) -> str | None:
-    """Map endpoint URLs to AGENTIC_WORKER_ID convention engine_{i}."""
+    """Best-effort: map URLs like http://h:8000 to engine_0 by base-port 8000.
+
+    Use `_resolve_worker(url, worker_map)` instead when you have an
+    explicit URL→worker_id map (e.g. from bench.sh).
+    """
     if not url_or_id:
         return None
     if url_or_id.startswith("engine_"):
         return url_or_id
-    # Endpoint URLs look like http://host:8000; map by port offset against 8000.
     try:
         port = int(url_or_id.rsplit(":", 1)[1].split("/")[0])
         return f"engine_{port - 8000}"
@@ -277,10 +292,36 @@ def _normalize_worker(url_or_id: str | None) -> str | None:
         return url_or_id
 
 
+def _resolve_worker(
+    url_or_id: str | None,
+    worker_map: dict[str, str] | None,
+) -> str | None:
+    if not url_or_id:
+        return None
+    if worker_map and url_or_id in worker_map:
+        return worker_map[url_or_id]
+    return _normalize_worker(url_or_id)
+
+
+def _vllm_rid_matches(vllm_rid: Any, proxy_rid: Any) -> bool:
+    """vLLM internally rewrites rid as `cmpl-<proxy_id>-<i>-<hash>`."""
+    if vllm_rid is None or proxy_rid is None:
+        return False
+    if vllm_rid == proxy_rid:
+        return True
+    s = str(vllm_rid)
+    p = str(proxy_rid)
+    return s.startswith(f"cmpl-{p}-") or s.startswith(f"chatcmpl-{p}-")
+
+
 # ---------- Hotspot index (B3) ------------------------------------------
 
 def hotspot_index(joined: list[JsonDict]) -> JsonDict:
-    """max/median per-worker queue-delay p90 across completed requests."""
+    """max/median per-worker queue-delay p90 across completed requests.
+
+    Worker key is the raw `routed_to` URL (or proxy `endpoint_url`
+    fallback), so per-worker rows match the user's mental model.
+    """
     if not joined:
         return {"status": "unavailable"}
 
@@ -289,6 +330,7 @@ def hotspot_index(joined: list[JsonDict]) -> JsonDict:
     for r in joined:
         if r.get("error"):
             continue
+        # routed_to is the vLLM worker URL; endpoint_url is the proxy URL.
         worker = r.get("routed_to") or r.get("endpoint_url")
         if not worker:
             continue
@@ -335,6 +377,7 @@ def label_slow_requests(
     engine_state_by_worker: dict[str, list[JsonDict]],
     slo: JsonDict | None = None,
     slow_ttft_factor: float = 2.0,
+    worker_map: dict[str, str] | None = None,
 ) -> list[JsonDict]:
     slo = slo or DEFAULT_SLO
     ttft_threshold = float(slo["ttft_p90_s"]) * slow_ttft_factor
@@ -358,7 +401,8 @@ def label_slow_requests(
             continue
         if ttft <= ttft_threshold:
             continue
-        label = _assign_label(r, hot_workers, engine_state_by_worker)
+        label = _assign_label(r, hot_workers, engine_state_by_worker,
+                               worker_map)
         labels.append({
             "request_id": r.get("request_id"),
             "session_id": r.get("session_id"),
@@ -377,8 +421,12 @@ def _assign_label(
     r: JsonDict,
     hot_workers: set[str],
     engine_state_by_worker: dict[str, list[JsonDict]],
+    worker_map: dict[str, str] | None = None,
 ) -> str:
-    worker = _normalize_worker(r.get("endpoint_url") or r.get("routed_to"))
+    worker = _resolve_worker(
+        r.get("routed_to") or r.get("endpoint_url"),
+        worker_map,
+    )
     rid = r.get("request_id")
     steps = engine_state_by_worker.get(worker, [])
     t0 = r.get("t_first_token_unix")
@@ -389,8 +437,11 @@ def _assign_label(
             if t is None or t < t0 or t > t1:
                 continue
             for pr in s.get("per_req", []) or []:
-                if pr.get("phase") == "prefill" and pr.get("rid") != rid:
-                    return "same_worker_prefill_overlap"
+                if pr.get("phase") != "prefill":
+                    continue
+                if _vllm_rid_matches(pr.get("rid"), rid):
+                    continue
+                return "same_worker_prefill_overlap"
     if (r.get("routed_to") or "") in hot_workers:
         return "hot_worker_queue"
     est = r.get("estimated_new_tokens") or 0
@@ -489,7 +540,17 @@ def main(argv: list[str] | None = None) -> None:
                    help="run_meta or window_summary.json from SRR loadgen")
     p.add_argument("--out-dir", type=Path, required=True)
     p.add_argument("--slow-ttft-factor", type=float, default=2.0)
+    p.add_argument("--worker-map", type=str, default=None,
+                   help="Comma-separated URL=worker_id pairs, e.g. "
+                        "http://h:9100=engine_0,http://h:9101=engine_1")
     args = p.parse_args(argv)
+    worker_map: dict[str, str] | None = None
+    if args.worker_map:
+        worker_map = {}
+        for entry in args.worker_map.split(","):
+            url, _, wid = entry.strip().partition("=")
+            if url and wid:
+                worker_map[url] = wid
 
     metrics = load_jsonl(args.metrics)
     breakdown_raw = load_json(args.breakdown) if args.breakdown else []
@@ -512,11 +573,12 @@ def main(argv: list[str] | None = None) -> None:
     write_json(args.out_dir / "reuse_decomposition.json",
                 reuse_decomposition(joined))
     write_json(args.out_dir / "interference_index.json",
-                interference_index(joined, engine_state))
+                interference_index(joined, engine_state, worker_map))
     write_json(args.out_dir / "hotspot_index.json",
                 hotspot_index(joined))
     labels = label_slow_requests(joined, engine_state,
-                                  slow_ttft_factor=args.slow_ttft_factor)
+                                  slow_ttft_factor=args.slow_ttft_factor,
+                                  worker_map=worker_map)
     write_jsonl(args.out_dir / "failure_label.jsonl", labels)
     counts: dict[str, int] = defaultdict(int)
     for L in labels:

scripts/b2_interference.py

@@ -0,0 +1,278 @@
+"""B2 PD-colo interference microbench.
+
+Concurrently issues:
+- a steady stream of short-prompt decode-heavy requests to a designated
+  "decode" instance;
+- a periodic single large-prompt one-token request that hits either the
+  same instance (same-worker) or a separate one (different-worker).
+
+The harness writes per-request metrics with a `workload` tag
+({"decode", "prefill"}) and a `variant` tag (same/different) plus
+unix timestamps so the analyzer can label same-worker overlap directly
+from the engine step JSONL.
+
+Outputs under <out-dir>/<variant>/<prefill_size>/:
+- metrics.jsonl                — per-request rows for this cell
+- run_window.json              — t_start_unix/t_end_unix for analyzer slice
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+import random
+import time
+from pathlib import Path
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+
+async def _send(
+    client: httpx.AsyncClient,
+    endpoint: str,
+    model: str,
+    prompt_ids: list[int],
+    max_tokens: int,
+    *,
+    workload: str,
+    variant: str,
+    prefill_size: int,
+    out_fh,
+    fh_lock: asyncio.Lock,
+    idx: int,
+) -> None:
+    payload = {
+        "model": model,
+        "prompt": prompt_ids,
+        "max_tokens": max_tokens,
+        "min_tokens": max_tokens,
+        "temperature": 0,
+        "stream": True,
+        "stream_options": {"include_usage": True},
+    }
+    rid = f"{workload}_{variant}_p{prefill_size}_{idx}"
+    t_dispatch = time.time()
+    ttft = None
+    finish = None
+    n_output = 0
+    err = None
+    token_times: list[float] = []
+    try:
+        async with client.stream(
+            "POST", f"{endpoint}/v1/completions", json=payload,
+            headers={"X-Request-Id": rid, "X-Session-Id": rid},
+            timeout=600.0,
+        ) as resp:
+            resp.raise_for_status()
+            async for raw_line in resp.aiter_lines():
+                if not raw_line or not raw_line.startswith("data:"):
+                    continue
+                data = raw_line[5:].strip()
+                if data == "[DONE]":
+                    break
+                try:
+                    chunk = json.loads(data)
+                except json.JSONDecodeError:
+                    continue
+                choices = chunk.get("choices", [])
+                if choices:
+                    now = time.time()
+                    token_ids = choices[0].get("token_ids")
+                    if isinstance(token_ids, list) and token_ids:
+                        if ttft is None:
+                            ttft = now - t_dispatch
+                        token_times.extend([now] * len(token_ids))
+                usage = chunk.get("usage")
+                if usage:
+                    n_output = usage.get("completion_tokens", n_output)
+        finish = time.time()
+    except Exception as exc:
+        err = repr(exc)[:300]
+        finish = time.time()
+
+    tpot = None
+    if len(token_times) > 1:
+        diffs = [token_times[i + 1] - token_times[i]
+                  for i in range(len(token_times) - 1)]
+        tpot = sum(diffs) / len(diffs)
+
+    row = {
+        "request_id": rid,
+        "workload": workload,
+        "variant": variant,
+        "prefill_size": prefill_size,
+        "endpoint": endpoint,
+        "input_length": len(prompt_ids),
+        "max_tokens": max_tokens,
+        "t_dispatch_unix": t_dispatch,
+        "t_finish_unix": finish,
+        "ttft_s": ttft,
+        "tpot_s": tpot,
+        "latency_s": (finish - t_dispatch) if finish else None,
+        "actual_output_tokens": n_output,
+        "error": err,
+    }
+    async with fh_lock:
+        out_fh.write(json.dumps(row, sort_keys=True) + "\n")
+        out_fh.flush()
+
+
+async def decode_load(
+    *, client, endpoint, model, qps, duration_s,
+    workload, variant, prefill_size, out_fh, fh_lock,
+    decode_prompt_tokens, decode_output_tokens, rng,
+) -> None:
+    period = 1.0 / qps
+    end_t = time.time() + duration_s
+    pending: list[asyncio.Task] = []
+    idx = 0
+    while time.time() < end_t:
+        prompt_ids = [rng.randint(1000, 100000) for _ in range(decode_prompt_tokens)]
+        task = asyncio.create_task(_send(
+            client, endpoint, model, prompt_ids, decode_output_tokens,
+            workload="decode", variant=variant, prefill_size=prefill_size,
+            out_fh=out_fh, fh_lock=fh_lock, idx=idx,
+        ))
+        pending.append(task)
+        idx += 1
+        await asyncio.sleep(period)
+    await asyncio.gather(*pending, return_exceptions=True)
+
+
+async def prefill_injections(
+    *, client, endpoint, model, prefill_size, n_injections, interval_s,
+    variant, out_fh, fh_lock, start_delay_s, rng,
+) -> None:
+    await asyncio.sleep(start_delay_s)
+    for i in range(n_injections):
+        prompt_ids = [rng.randint(1000, 100000) for _ in range(prefill_size)]
+        await _send(
+            client, endpoint, model, prompt_ids, max_tokens=1,
+            workload="prefill", variant=variant, prefill_size=prefill_size,
+            out_fh=out_fh, fh_lock=fh_lock, idx=i,
+        )
+        await asyncio.sleep(interval_s)
+
+
+async def run_cell(
+    *,
+    decode_endpoint, prefill_endpoint, model, prefill_size, variant,
+    qps, duration_s, n_injections, injection_interval_s, start_delay_s,
+    decode_prompt_tokens, decode_output_tokens,
+    out_dir,
+) -> dict:
+    cell_dir = out_dir / variant / f"p{prefill_size}"
+    cell_dir.mkdir(parents=True, exist_ok=True)
+    metrics_path = cell_dir / "metrics.jsonl"
+    fh_lock = asyncio.Lock()
+    rng = random.Random(42 + prefill_size + (0 if variant == "same" else 1000))
+
+    t_start = time.time()
+    logger.info("[b2] start variant=%s prefill_size=%d", variant, prefill_size)
+    with metrics_path.open("w", encoding="utf-8") as out_fh:
+        limits = httpx.Limits(max_connections=2000, max_keepalive_connections=500)
+        async with httpx.AsyncClient(timeout=600.0, limits=limits) as client:
+            await asyncio.gather(
+                decode_load(
+                    client=client, endpoint=decode_endpoint, model=model,
+                    qps=qps, duration_s=duration_s,
+                    workload="decode", variant=variant,
+                    prefill_size=prefill_size,
+                    out_fh=out_fh, fh_lock=fh_lock,
+                    decode_prompt_tokens=decode_prompt_tokens,
+                    decode_output_tokens=decode_output_tokens,
+                    rng=rng,
+                ),
+                prefill_injections(
+                    client=client, endpoint=prefill_endpoint, model=model,
+                    prefill_size=prefill_size, n_injections=n_injections,
+                    interval_s=injection_interval_s, variant=variant,
+                    out_fh=out_fh, fh_lock=fh_lock,
+                    start_delay_s=start_delay_s, rng=rng,
+                ),
+            )
+    t_end = time.time()
+    window = {
+        "variant": variant,
+        "prefill_size": prefill_size,
+        "decode_endpoint": decode_endpoint,
+        "prefill_endpoint": prefill_endpoint,
+        "qps": qps, "duration_s": duration_s,
+        "n_injections": n_injections,
+        "injection_interval_s": injection_interval_s,
+        "decode_prompt_tokens": decode_prompt_tokens,
+        "decode_output_tokens": decode_output_tokens,
+        "t_start_unix": t_start, "t_end_unix": t_end,
+    }
+    (cell_dir / "run_window.json").write_text(json.dumps(window, indent=2))
+    logger.info("[b2] done  variant=%s prefill_size=%d wall=%.1fs",
+                 variant, prefill_size, t_end - t_start)
+    return window
+
+
+async def amain(args: argparse.Namespace) -> None:
+    sizes = [int(s) for s in args.prefill_sizes.split(",")]
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    overall: list[dict] = []
+    for size in sizes:
+        for variant in args.variants.split(","):
+            variant = variant.strip()
+            if variant == "same":
+                d_ep, p_ep = args.decode_endpoint, args.decode_endpoint
+            elif variant == "different":
+                d_ep, p_ep = args.decode_endpoint, args.prefill_endpoint
+            else:
+                raise ValueError(f"unknown variant {variant!r}")
+            w = await run_cell(
+                decode_endpoint=d_ep, prefill_endpoint=p_ep,
+                model=args.model, prefill_size=size, variant=variant,
+                qps=args.decode_qps, duration_s=args.duration_s,
+                n_injections=args.injections,
+                injection_interval_s=args.injection_interval_s,
+                start_delay_s=args.start_delay_s,
+                decode_prompt_tokens=args.decode_prompt_tokens,
+                decode_output_tokens=args.decode_output_tokens,
+                out_dir=out_dir,
+            )
+            overall.append(w)
+    (out_dir / "sweep_meta.json").write_text(json.dumps(overall, indent=2))
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="B2 interference microbench")
+    p.add_argument("--decode-endpoint", required=True,
+                   help="e.g. http://127.0.0.1:8000")
+    p.add_argument("--prefill-endpoint", required=True,
+                   help="e.g. http://127.0.0.1:8001")
+    p.add_argument("--model", required=True)
+    p.add_argument("--out-dir", required=True)
+    p.add_argument("--prefill-sizes", default="2048,8192,16384,32768,65536")
+    p.add_argument("--variants", default="different,same",
+                   help="Comma-separated variants in run order")
+    p.add_argument("--decode-qps", type=float, default=4.0,
+                   help="Decode-load arrival rate (req/s)")
+    p.add_argument("--duration-s", type=float, default=60.0,
+                   help="Decode-load duration (s) per cell")
+    p.add_argument("--injections", type=int, default=4,
+                   help="Number of prefill injections per cell")
+    p.add_argument("--injection-interval-s", type=float, default=12.0)
+    p.add_argument("--start-delay-s", type=float, default=10.0,
+                   help="Warmup before first prefill injection")
+    p.add_argument("--decode-prompt-tokens", type=int, default=256)
+    p.add_argument("--decode-output-tokens", type=int, default=100)
+    p.add_argument("-v", "--verbose", action="store_true")
+    args = p.parse_args()
+    logging.basicConfig(
+        level=logging.DEBUG if args.verbose else logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    asyncio.run(amain(args))
+
+
+if __name__ == "__main__":
+    main()

scripts/b3_analyze.sh

@@ -0,0 +1,111 @@
+#!/usr/bin/env bash
+# Per-policy joined_analysis driver for a completed B3 sweep.
+#
+# For each policy directory under <SWEEP_DIR>:
+#   - slice engine_state by run_window.json
+#   - run joined_analysis.py to emit interference / hotspot / reuse
+#     / failure breakdown
+# Then emit b3_policy_comparison.json aggregating one row per policy.
+
+set -euo pipefail
+
+ROOT="${ROOT:-/home/admin/cpfs/wjh/agentic-kv}"
+VENV="$ROOT/.venv/bin"
+SWEEP_DIR="${1:?usage: $0 <sweep_dir>}"
+
+WORKER_MAP="http://127.0.0.1:8000=engine_0,http://127.0.0.1:8001=engine_1,http://127.0.0.1:8002=engine_2,http://127.0.0.1:8003=engine_3,http://127.0.0.1:8004=engine_4,http://127.0.0.1:8005=engine_5,http://127.0.0.1:8006=engine_6,http://127.0.0.1:8007=engine_7"
+
+for policy_dir in "$SWEEP_DIR"/*/; do
+    policy=$(basename "$policy_dir")
+    case "$policy" in
+        engine_state|logs|capped) ;;
+    esac
+    if [ ! -f "$policy_dir/run_window.json" ]; then
+        continue
+    fi
+    echo "=== $policy ==="
+
+    PYTHONPATH="$ROOT" "$VENV/python" \
+        "$ROOT/scripts/slice_engine_state.py" \
+        --input-dir "$SWEEP_DIR/engine_state" \
+        --output-dir "$policy_dir/engine_state" \
+        --window "$policy_dir/run_window.json"
+
+    PYTHONPATH="$ROOT" "$VENV/python" \
+        "$ROOT/analysis/characterization/joined_analysis.py" \
+        --metrics "$policy_dir/metrics.jsonl" \
+        --breakdown "$policy_dir/breakdown.json" \
+        --worker-state "$policy_dir/worker_state.json" \
+        --engine-state-dir "$policy_dir/engine_state" \
+        --worker-map "$WORKER_MAP" \
+        --out-dir "$policy_dir/joined"
+done
+
+# Also handle capped/ which is nested
+if [ -f "$SWEEP_DIR/capped/run_window.json" ]; then
+    echo "=== capped ==="
+    PYTHONPATH="$ROOT" "$VENV/python" \
+        "$ROOT/scripts/slice_engine_state.py" \
+        --input-dir "$SWEEP_DIR/engine_state" \
+        --output-dir "$SWEEP_DIR/capped/engine_state" \
+        --window "$SWEEP_DIR/capped/run_window.json"
+    PYTHONPATH="$ROOT" "$VENV/python" \
+        "$ROOT/analysis/characterization/joined_analysis.py" \
+        --metrics "$SWEEP_DIR/capped/metrics.jsonl" \
+        --breakdown "$SWEEP_DIR/capped/breakdown.json" \
+        --worker-state "$SWEEP_DIR/capped/worker_state.json" \
+        --engine-state-dir "$SWEEP_DIR/capped/engine_state" \
+        --worker-map "$WORKER_MAP" \
+        --out-dir "$SWEEP_DIR/capped/joined"
+fi
+
+# Aggregate per-policy summary
+"$VENV/python" - <<PY
+import json, os, statistics
+from pathlib import Path
+
+sweep = Path("$SWEEP_DIR")
+def pct(vals, p):
+    if not vals: return None
+    vs = sorted(vals)
+    return vs[int(p * (len(vs)-1))]
+
+rows = []
+for sub in sorted(sweep.iterdir()):
+    rw = sub / "run_window.json"
+    jd = sub / "joined"
+    if not rw.exists() or not jd.exists():
+        continue
+    policy = sub.name
+    metrics = [json.loads(l) for l in (sub / "metrics.jsonl").open()]
+    ok = [r for r in metrics if r.get("error") is None]
+    ttfts = [r["ttft_s"] for r in ok if r.get("ttft_s") is not None]
+    tpots = [r["tpot_s"] for r in ok if r.get("tpot_s") is not None]
+    e2es  = [r["latency_s"] for r in ok if r.get("latency_s") is not None]
+    total_input  = sum(r.get("input_length", 0) for r in ok)
+    total_cached = sum(r.get("cached_tokens", 0) for r in ok)
+    interf = json.loads((jd / "interference_index.json").read_text())
+    hot    = json.loads((jd / "hotspot_index.json").read_text())
+    reuse  = json.loads((jd / "reuse_decomposition.json").read_text())
+    fail   = json.loads((jd / "failure_breakdown.json").read_text())
+    rows.append({
+        "policy": policy,
+        "n_ok": len(ok), "n_total": len(metrics),
+        "ttft_p50_s": pct(ttfts, 0.5), "ttft_p90_s": pct(ttfts, 0.9),
+        "ttft_p99_s": pct(ttfts, 0.99),
+        "tpot_p50_s": pct(tpots, 0.5), "tpot_p90_s": pct(tpots, 0.9),
+        "tpot_p99_s": pct(tpots, 0.99),
+        "e2e_p50_s":  pct(e2es,  0.5), "e2e_p90_s":  pct(e2es,  0.9),
+        "e2e_p99_s":  pct(e2es,  0.99),
+        "apc_ratio":   total_cached / max(total_input, 1),
+        "interference_index": interf.get("interference_index"),
+        "hotspot_index_ttft_p90": hot.get("hotspot_index_ttft_p90"),
+        "reuse_intra_frac":  reuse.get("fractions", {}).get("intra"),
+        "reuse_cross_frac":  reuse.get("fractions", {}).get("cross"),
+        "n_slow":            fail.get("n_slow"),
+        "failure_counts":    fail.get("counts"),
+    })
+out = sweep / "b3_policy_comparison.json"
+out.write_text(json.dumps({"rows": rows}, indent=2))
+print(json.dumps(rows, indent=2))
+PY

scripts/b3_sweep.sh

@@ -0,0 +1,177 @@
+#!/usr/bin/env bash
+# B3 routing sweep: 5 policies on 8x TP1 instances with full instrumentation.
+#
+# Policies:
+#   lmetric    — cache-aware P_tokens × BS routing (main baseline)
+#   load_only  — pure min-num_requests (B3 control: no cache)
+#   sticky     — hard session affinity (B3 control: perfect locality)
+#   unified    — hybrid affinity + LMetric fallback
+#   capped     — lmetric on a per-session turn-capped trace
+#
+# Each policy run produces metrics.jsonl + breakdown.json + worker_state.json
+# + run_window.json (start/end unix timestamps so the analyzer can slice the
+# shared engine_*.jsonl by time).
+
+set -euo pipefail
+
+ROOT="${ROOT:-/home/admin/cpfs/wjh/agentic-kv}"
+VENV="$ROOT/.venv/bin"
+MODEL="${MODEL:-/home/admin/cpfs/wjh/models/Qwen/Qwen3-Coder-30B-A3B-Instruct}"
+TRACE="${TRACE:-$ROOT/traces/w600_r0.0015_st30.jsonl}"
+OUTDIR="${OUTDIR:-$ROOT/outputs/b3_sweep_$(date +%Y%m%d_%H%M%S)}"
+PROXY_PORT="${PROXY_PORT:-9300}"
+BASE_PORT="${BASE_PORT:-8000}"
+# Space-separated list of GPU indices to use, one vLLM instance per index.
+# Override via GPU_INDICES="1 2 3 4 5 6 7" when GPU 0 holds ghost memory.
+GPU_INDICES="${GPU_INDICES:-0 1 2 3 4 5 6 7}"
+POLICIES="${POLICIES:-lmetric load_only sticky unified}"
+MAX_TURNS_CAP="${MAX_TURNS_CAP:-8}"
+EXTRA_VLLM_ARGS="${EXTRA_VLLM_ARGS:---enable-prompt-tokens-details}"
+
+# Derive N_INSTANCES from GPU_INDICES
+N_INSTANCES=$(echo $GPU_INDICES | wc -w)
+
+mkdir -p "$OUTDIR/engine_state" "$OUTDIR/logs"
+echo "[b3_sweep] OUTDIR=$OUTDIR"
+
+cleanup() {
+    pkill -9 -f "vllm serve" 2>/dev/null || true
+    # vLLM spawns an EngineCore child whose process name is
+    # "VLLM::EngineCor" — pkill -f "vllm serve" misses it and leaves
+    # the GPU memory locked by a dead-but-tracked-by-driver context.
+    pkill -9 -f "EngineCore" 2>/dev/null || true
+    pkill -9 -f cache_aware_proxy 2>/dev/null || true
+    sleep 3
+}
+trap cleanup EXIT
+
+# 1) Launch one vLLM per GPU index in GPU_INDICES; each emits engine_<i>.jsonl
+launch_vllm() {
+    echo "[b3_sweep] launching $N_INSTANCES vLLM instances on GPUs $GPU_INDICES ..."
+    local i=0
+    for gpu in $GPU_INDICES; do
+        local port=$((BASE_PORT + i))
+        local master=$((29500 + i))
+        local log="$OUTDIR/logs/vllm_inst_${i}_gpu${gpu}.log"
+        AGENTIC_STEP_LOG_PATH="$OUTDIR/engine_state/engine_${i}.jsonl" \
+        AGENTIC_WORKER_ID="engine_${i}" \
+        CUDA_VISIBLE_DEVICES=$gpu \
+        MASTER_PORT=$master \
+        nohup "$VENV/vllm" serve "$MODEL" \
+            --host 0.0.0.0 --port "$port" \
+            --tensor-parallel-size 1 \
+            --trust-remote-code --enable-prefix-caching \
+            --dtype auto --gpu-memory-utilization 0.9 \
+            --max-model-len 200000 \
+            $EXTRA_VLLM_ARGS \
+            > "$log" 2>&1 &
+        disown
+        sleep 2
+        i=$((i + 1))
+    done
+
+    echo "[b3_sweep] waiting for vLLM health ..."
+    for i in $(seq 0 $((N_INSTANCES - 1))); do
+        local port=$((BASE_PORT + i))
+        local tries=0
+        while ! curl -sf "http://127.0.0.1:$port/health" >/dev/null 2>&1; do
+            tries=$((tries + 1))
+            if [ $tries -gt 90 ]; then
+                echo "[b3_sweep] FATAL: inst_$i (port $port) not healthy after 180s"
+                exit 1
+            fi
+            sleep 2
+        done
+        echo "  inst_$i ready"
+    done
+}
+
+launch_proxy() {
+    local policy="$1"
+    local logfile="$2"
+    local combined_args=""
+    for i in $(seq 0 $((N_INSTANCES - 1))); do
+        combined_args="$combined_args http://127.0.0.1:$((BASE_PORT + i))"
+    done
+    nohup "$VENV/python" "$ROOT/scripts/cache_aware_proxy.py" \
+        --port "$PROXY_PORT" \
+        --combined $combined_args \
+        --policy "$policy" \
+        > "$logfile" 2>&1 &
+    disown
+    local tries=0
+    until curl -sf "http://127.0.0.1:$PROXY_PORT/stats" >/dev/null 2>&1; do
+        tries=$((tries + 1))
+        if [ $tries -gt 30 ]; then
+            echo "[b3_sweep] FATAL: proxy did not come up in 60s"
+            tail -30 "$logfile"
+            exit 1
+        fi
+        sleep 2
+    done
+}
+
+run_policy() {
+    local policy="$1"
+    local trace="$2"
+    local rundir="$OUTDIR/$policy"
+    mkdir -p "$rundir"
+    echo "[b3_sweep] === policy=$policy trace=$(basename "$trace") ==="
+
+    pkill -9 -f cache_aware_proxy 2>/dev/null || true
+    sleep 2
+    launch_proxy "$policy" "$rundir/proxy.log"
+
+    local t_start
+    t_start=$(date +%s.%N)
+    echo "{\"policy\": \"$policy\", \"trace\": \"$trace\", \"t_start_unix\": $t_start}" \
+        > "$rundir/run_window.json.partial"
+
+    PYTHONPATH="$ROOT" "$VENV/python" -m replayer \
+        --trace "$trace" \
+        --output "$rundir/metrics.jsonl" \
+        --endpoint "http://127.0.0.1:$PROXY_PORT" \
+        --model "$MODEL" \
+        2>&1 | tee "$rundir/replayer.log" | tail -3
+
+    local t_end
+    t_end=$(date +%s.%N)
+    python3 - "$rundir" "$policy" "$trace" "$t_start" "$t_end" <<'PY'
+import json, sys
+rundir, policy, trace, t_start, t_end = sys.argv[1:]
+with open(f"{rundir}/run_window.json", "w") as f:
+    json.dump({
+        "policy": policy, "trace": trace,
+        "t_start_unix": float(t_start),
+        "t_end_unix": float(t_end),
+    }, f, indent=2)
+PY
+    rm -f "$rundir/run_window.json.partial"
+
+    curl -s "http://127.0.0.1:$PROXY_PORT/breakdown" > "$rundir/breakdown.json"
+    curl -s "http://127.0.0.1:$PROXY_PORT/worker_state" > "$rundir/worker_state.json"
+    curl -s "http://127.0.0.1:$PROXY_PORT/stats" > "$rundir/stats.json"
+    echo "[b3_sweep] $policy done: $(wc -l < "$rundir/metrics.jsonl") metric rows"
+}
+
+# 2) Run each policy
+launch_vllm
+
+for policy in $POLICIES; do
+    run_policy "$policy" "$TRACE"
+done
+
+# 3) Capped variant on lmetric
+echo "[b3_sweep] building capped trace (max_turns=$MAX_TURNS_CAP) ..."
+CAPPED_TRACE="$OUTDIR/capped/trace.jsonl"
+mkdir -p "$OUTDIR/capped"
+"$VENV/python" "$ROOT/scripts/build_capped_trace.py" \
+    --input "$TRACE" \
+    --output "$CAPPED_TRACE" \
+    --max-turns "$MAX_TURNS_CAP" | tee "$OUTDIR/capped/build.log"
+run_policy "capped" "$CAPPED_TRACE"
+
+# 4) Snapshot final engine state file sizes for the analyzer
+ls -l "$OUTDIR/engine_state/" > "$OUTDIR/engine_state_files.txt"
+
+echo "[b3_sweep] sweep complete. OUTDIR=$OUTDIR"

scripts/build_capped_trace.py

@@ -0,0 +1,80 @@
+"""Cap per-session turn count to isolate session-mass effects in B3.
+
+Input trace is grouped by session_id (or reconstructed from
+parent_chat_id chains). Sessions with more than --max-turns turns are
+truncated to keep only their first N turns in trace order. The output
+preserves the original line ordering (timestamp order).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from pathlib import Path
+
+
+def _resolve_session_id(row: dict, chat_to_session: dict[int, str]) -> str:
+    if "session_id" in row:
+        return str(row["session_id"])
+    cid = int(row["chat_id"])
+    pcid = int(row["parent_chat_id"])
+    if pcid < 0:
+        sid = str(cid)
+    else:
+        sid = chat_to_session.get(pcid, str(pcid))
+    chat_to_session[cid] = sid
+    return sid
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description="Cap per-session turn count")
+    p.add_argument("--input", type=Path, required=True)
+    p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--max-turns", type=int, default=8,
+                   help="Keep at most N earliest turns per session")
+    args = p.parse_args()
+
+    chat_to_session: dict[int, str] = {}
+    kept: dict[str, int] = defaultdict(int)
+    rows: list[tuple[str, dict]] = []
+    with args.input.open("r", encoding="utf-8") as fh:
+        for line in fh:
+            line = line.strip()
+            if not line:
+                continue
+            row = json.loads(line)
+            sid = _resolve_session_id(row, chat_to_session)
+            row["session_id"] = sid
+            rows.append((sid, row))
+
+    in_n = len(rows)
+    sessions = len({sid for sid, _ in rows})
+
+    rows.sort(key=lambda x: (x[1]["session_id"], x[1].get("turn", 0)))
+    capped_rows: list[dict] = []
+    for sid, row in rows:
+        if kept[sid] >= args.max_turns:
+            continue
+        kept[sid] += 1
+        capped_rows.append(row)
+    capped_rows.sort(key=lambda r: r.get("timestamp", 0.0))
+
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    with args.output.open("w", encoding="utf-8") as fh:
+        for r in capped_rows:
+            fh.write(json.dumps(r) + "\n")
+
+    print(f"input rows: {in_n}, sessions: {sessions}")
+    print(f"capped rows: {len(capped_rows)} (max_turns={args.max_turns})")
+    dropped = in_n - len(capped_rows)
+    print(f"dropped:    {dropped} ({100 * dropped / max(in_n, 1):.1f}%)")
+    if capped_rows:
+        from collections import Counter
+        turns_dist = Counter(kept[s] for s in kept)
+        top = sorted(turns_dist.items())[:6]
+        print(f"turns/session (capped) sample: {top}")
+
+
+if __name__ == "__main__":
+    main()

scripts/cache_aware_proxy.py

@@ -178,6 +178,48 @@ def pick_instance(instances: list[InstanceState], token_ids: list[int] | None,
     return instances[best_idx], best_idx
 
 
+def pick_instance_load_only(
+    instances: list[InstanceState],
+    token_ids: list[int] | None,
+    session_id: str | None,
+    input_length: int,
+    affinity: dict[str, int],
+) -> tuple[InstanceState, int]:
+    """Pure load balancing: pick instance with fewest in-flight requests.
+
+    Ignores cache hits and session affinity. Used as a B3 control to
+    isolate the locality contribution of cache-aware policies.
+    """
+    best_idx = min(range(len(instances)),
+                    key=lambda i: instances[i].num_requests)
+    return instances[best_idx], best_idx
+
+
+def pick_instance_sticky(
+    instances: list[InstanceState],
+    token_ids: list[int] | None,
+    session_id: str | None,
+    input_length: int,
+    affinity: dict[str, int],
+) -> tuple[InstanceState, int]:
+    """Hard session affinity: once assigned, never break.
+
+    First turn of a session picks the instance with the lowest
+    num_requests; subsequent turns always return to the same instance
+    regardless of load. Used as a B3 control to isolate the hot-spot
+    cost of perfect locality.
+    """
+    if session_id and session_id in affinity:
+        idx = affinity[session_id]
+        if idx < len(instances):
+            return instances[idx], idx
+    best_idx = min(range(len(instances)),
+                    key=lambda i: instances[i].num_requests)
+    if session_id:
+        affinity[session_id] = best_idx
+    return instances[best_idx], best_idx
+
+
 def pick_instance_lmetric(instances: list[InstanceState], token_ids: list[int] | None,
                           session_id: str | None, input_length: int,
                           affinity: dict[str, int]) -> tuple[InstanceState, int]:
@@ -585,6 +627,14 @@ async def _handle_combined(api, req_data, token_ids, input_length, session_id, h
         chosen, best_idx = pick_instance_lmetric(
             combined_instances, token_ids, session_id, input_length,
             session_affinity_combined)
+    elif policy == "load_only":
+        chosen, best_idx = pick_instance_load_only(
+            combined_instances, token_ids, session_id, input_length,
+            session_affinity_combined)
+    elif policy == "sticky":
+        chosen, best_idx = pick_instance_sticky(
+            combined_instances, token_ids, session_id, input_length,
+            session_affinity_combined)
     elif policy == "unified":
         chosen, best_idx, decision = pick_instance_unified_hybrid(
             combined_instances, token_ids, session_id, input_length,
@@ -799,8 +849,10 @@ def parse_args():
     p.add_argument("--bootstrap-ports", type=str, default="",
                    help="Comma-separated bootstrap ports for combined instances (for offload mode)")
     p.add_argument("--policy", type=str, default="linear",
-                   choices=["linear", "lmetric", "unified"],
+                   choices=["linear", "lmetric", "load_only", "sticky", "unified"],
                    help="Routing policy: linear (cache-aware), lmetric (P_tokens × BS), "
+                        "load_only (B3 control: pure min-num_requests), "
+                        "sticky (B3 control: hard session affinity), "
                         "or unified (hybrid affinity + LMetric fallback)")
     p.add_argument("--overload-factor", type=float, default=2.0,
                    help="Break session affinity when instance load > factor * avg")

scripts/render_b3_report.py

@@ -0,0 +1,151 @@
+"""Render a B3 comparison markdown from b3_policy_comparison.json.
+
+Drops rows for policies not yet present in the sweep. Designed to be
+re-run whenever a new policy finishes; the markdown lives next to
+the sweep directory so future re-runs just overwrite it.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+POLICY_ORDER = ["lmetric", "load_only", "sticky", "unified", "capped"]
+POLICY_DESCR = {
+    "lmetric":   "cache-aware P_tokens × BS (main baseline)",
+    "load_only": "control: min(num_requests), no cache, no affinity",
+    "sticky":    "control: hard session affinity (never break)",
+    "unified":   "hybrid affinity + LMetric fallback",
+    "capped":    "lmetric on per-session turn-capped trace",
+}
+
+
+def _fmt(v, kind="f2"):
+    if v is None:
+        return "—"
+    if kind == "f2":
+        return f"{v:.2f}"
+    if kind == "f3":
+        return f"{v:.3f}"
+    if kind == "pct":
+        return f"{v * 100:.1f}%"
+    if kind == "ms":
+        return f"{v * 1000:.1f}"
+    return str(v)
+
+
+def main() -> None:
+    p = argparse.ArgumentParser()
+    p.add_argument("--sweep-dir", type=Path, required=True)
+    p.add_argument("--output", type=Path, default=None,
+                   help="Markdown output; default <sweep-dir>/b3_report.md")
+    args = p.parse_args()
+
+    comp_path = args.sweep_dir / "b3_policy_comparison.json"
+    if not comp_path.exists():
+        raise SystemExit(f"missing {comp_path}; run b3_analyze.sh first")
+    data = json.loads(comp_path.read_text())
+    by_pol = {r["policy"]: r for r in data["rows"]}
+
+    lines: list[str] = []
+    lines.append(f"# B3 Routing Sweep Report")
+    lines.append("")
+    lines.append(f"Sweep dir: `{args.sweep_dir.name}`")
+    lines.append(f"Trace: w600_r0.0015_st30.jsonl (~1.2k reqs, 8 × TP1)")
+    lines.append(f"Policies present: {', '.join(p for p in POLICY_ORDER if p in by_pol)}")
+    lines.append(f"Policies pending: {', '.join(p for p in POLICY_ORDER if p not in by_pol) or '—'}")
+    lines.append("")
+
+    lines.append("## Headline latencies + APC")
+    lines.append("")
+    lines.append("| policy | ok/total | TTFT p50/p90/p99 (s) | TPOT p50/p90/p99 (ms) | E2E p50/p90/p99 (s) | APC |")
+    lines.append("|---|---:|---|---|---|---:|")
+    for pol in POLICY_ORDER:
+        r = by_pol.get(pol)
+        if not r:
+            lines.append(f"| **{pol}** | _pending_ | _pending_ | _pending_ | _pending_ | _pending_ |")
+            continue
+        lines.append(
+            f"| **{pol}** | {r['n_ok']}/{r['n_total']} | "
+            f"{_fmt(r['ttft_p50_s'])}/{_fmt(r['ttft_p90_s'])}/{_fmt(r['ttft_p99_s'])} | "
+            f"{_fmt(r['tpot_p50_s'], 'ms')}/{_fmt(r['tpot_p90_s'], 'ms')}/{_fmt(r['tpot_p99_s'], 'ms')} | "
+            f"{_fmt(r['e2e_p50_s'])}/{_fmt(r['e2e_p90_s'])}/{_fmt(r['e2e_p99_s'])} | "
+            f"{_fmt(r['apc_ratio'], 'pct')} |"
+        )
+    lines.append("")
+
+    lines.append("## Mechanism indices")
+    lines.append("")
+    lines.append("| policy | interference_index | hotspot_index (TTFT p90) | intra-session reuse | cross-session reuse | n_slow |")
+    lines.append("|---|---:|---:|---:|---:|---:|")
+    for pol in POLICY_ORDER:
+        r = by_pol.get(pol)
+        if not r:
+            lines.append(f"| **{pol}** | _pending_ | _pending_ | _pending_ | _pending_ | _pending_ |")
+            continue
+        lines.append(
+            f"| **{pol}** | {_fmt(r['interference_index'])} | "
+            f"{_fmt(r['hotspot_index_ttft_p90'])} | "
+            f"{_fmt(r['reuse_intra_frac'], 'pct')} | "
+            f"{_fmt(r['reuse_cross_frac'], 'pct')} | "
+            f"{r['n_slow']} |"
+        )
+    lines.append("")
+    lines.append("- **interference_index** = TPOT_p90(decode overlapping same-worker prefill) / TPOT_p90(clean)")
+    lines.append("- **hotspot_index** = max(worker TTFT_p90) / median(worker TTFT_p90)")
+    lines.append("")
+
+    lines.append("## Slow-request cause breakdown")
+    lines.append("")
+    lines.append("| policy | n_slow | same-worker overlap | hot worker queue | cache miss large append | high KV | unknown |")
+    lines.append("|---|---:|---:|---:|---:|---:|---:|")
+    for pol in POLICY_ORDER:
+        r = by_pol.get(pol)
+        if not r:
+            lines.append(f"| **{pol}** | _pending_ | _pending_ | _pending_ | _pending_ | _pending_ | _pending_ |")
+            continue
+        fc = r.get("failure_counts") or {}
+        def cnt(k): return fc.get(k, 0)
+        lines.append(
+            f"| **{pol}** | {r['n_slow']} | "
+            f"{cnt('same_worker_prefill_overlap')} | {cnt('hot_worker_queue')} | "
+            f"{cnt('cache_miss_large_append')} | {cnt('high_kv_occupancy')} | "
+            f"{cnt('unknown')} |"
+        )
+    lines.append("")
+
+    lines.append("## Policy notes")
+    lines.append("")
+    for pol in POLICY_ORDER:
+        lines.append(f"- **{pol}** — {POLICY_DESCR[pol]}")
+    lines.append("")
+
+    lines.append("## Per-policy per-worker TTFT p90 (s)")
+    lines.append("")
+    for pol in POLICY_ORDER:
+        r = by_pol.get(pol)
+        if not r:
+            lines.append(f"### {pol} _(pending)_")
+            lines.append("")
+            continue
+        path = args.sweep_dir / pol / "joined" / "hotspot_index.json"
+        if not path.exists():
+            continue
+        hot = json.loads(path.read_text())
+        per = hot.get("per_worker_ttft_p90_s") or {}
+        lines.append(f"### {pol}")
+        lines.append("")
+        lines.append("| worker | TTFT p90 (s) |")
+        lines.append("|---|---:|")
+        for w, v in sorted(per.items()):
+            lines.append(f"| {w} | {v:.2f} |")
+        lines.append("")
+
+    out = args.output or args.sweep_dir / "b3_report.md"
+    out.write_text("\n".join(lines))
+    print(f"wrote {out}")
+
+
+if __name__ == "__main__":
+    main()

scripts/slice_engine_state.py

@@ -0,0 +1,46 @@
+"""Slice shared engine_*.jsonl by a [t_start_unix, t_end_unix] window.
+
+Used between B3 policy runs and the analyzer so each policy gets
+its own per-window engine_state directory.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+def main() -> None:
+    p = argparse.ArgumentParser()
+    p.add_argument("--input-dir", type=Path, required=True)
+    p.add_argument("--output-dir", type=Path, required=True)
+    p.add_argument("--window", type=Path, required=True,
+                   help="run_window.json with t_start_unix / t_end_unix")
+    args = p.parse_args()
+
+    window = json.loads(args.window.read_text())
+    ts = float(window["t_start_unix"])
+    te = float(window["t_end_unix"])
+
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    print(f"window {ts:.3f} .. {te:.3f}")
+    for src in sorted(args.input_dir.glob("engine_*.jsonl")):
+        n_in = n_out = 0
+        dst = args.output_dir / src.name
+        with src.open() as fi, dst.open("w") as fo:
+            for line in fi:
+                n_in += 1
+                try:
+                    r = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                t = r.get("t_unix", 0)
+                if ts <= t <= te:
+                    fo.write(line)
+                    n_out += 1
+        print(f"  {src.name}: {n_out}/{n_in}")
+
+
+if __name__ == "__main__":
+    main()

tests/test_joined_analysis.py

@@ -11,6 +11,8 @@ from analysis.characterization.joined_analysis import (
     window_summary,
     _normalize_worker,
     _percentile,
+    _resolve_worker,
+    _vllm_rid_matches,
 )
 
 
@@ -70,18 +72,22 @@ def test_normalize_worker_maps_port_to_engine_id():
 
 def test_interference_index_marks_overlap_when_other_request_prefilling():
     metrics = [
-        _mk_metric("decode_target", endpoint_url="http://h:8000",
+        _mk_metric("decode_target",
                    t_first_token_unix=10.0, t_finish_unix=11.0,
                    tpot_s=0.10),
-        _mk_metric("decode_clean", endpoint_url="http://h:8001",
+        _mk_metric("decode_clean",
                    t_first_token_unix=20.0, t_finish_unix=21.0,
                    tpot_s=0.04),
     ]
-    joined = build_joined_records(metrics, [], [])
+    breakdown = [
+        {"request_id": "decode_target", "routed_to": "http://h:8000"},
+        {"request_id": "decode_clean", "routed_to": "http://h:8001"},
+    ]
+    joined = build_joined_records(metrics, breakdown, [])
     engine_state = {
         "engine_0": [
             {"t_unix": 10.5, "prefill_tokens": 8000,
-             "per_req": [{"rid": "other", "phase": "prefill"}]},
+             "per_req": [{"rid": "cmpl-other-0-aaaa", "phase": "prefill"}]},
         ],
         "engine_1": [
             {"t_unix": 20.5, "prefill_tokens": 0,
@@ -93,10 +99,22 @@ def test_interference_index_marks_overlap_when_other_request_prefilling():
     assert out["n_overlap_requests"] == 1
     assert out["n_clean_requests"] == 1
     assert out["interference_index"] is not None
-    # Overlap p90 = 0.10; clean p90 = 0.04; ratio > 2
     assert out["interference_index"] > 2.0
 
 
+def test_resolve_worker_prefers_explicit_map():
+    assert _resolve_worker("http://h:9100", {"http://h:9100": "engine_0"}) == "engine_0"
+    assert _resolve_worker("http://h:9100", None) == "engine_1100"
+
+
+def test_vllm_rid_matches_strips_cmpl_prefix():
+    assert _vllm_rid_matches("cmpl-1237198:1:1237198:0-0-b07fed77",
+                              "1237198:1:1237198:0")
+    assert _vllm_rid_matches("chatcmpl-abc-0-xx", "abc")
+    assert not _vllm_rid_matches("cmpl-other-0-xx", "1237198:1:1237198:0")
+    assert not _vllm_rid_matches(None, "x")
+
+
 def test_hotspot_index_max_over_median_p90():
     """One hot worker with TTFT 10x the others should drive a >1 index."""
     rows = []
@@ -118,14 +136,10 @@ def test_label_slow_requests_flags_overlap_and_hot_worker():
         _mk_metric("slow_overlap", ttft_s=10.0,
                    t_first_token_unix=10.0, t_finish_unix=11.0),
         _mk_metric("slow_no_overlap", ttft_s=10.0,
-                   endpoint_url="http://h:8005",
                    t_first_token_unix=20.0, t_finish_unix=21.0),
         _mk_metric("fast", ttft_s=0.5,
                    t_first_token_unix=15.0, t_finish_unix=16.0),
     ]
-    metrics[0]["routed_to"] = "http://h:8000"
-    metrics[1]["routed_to"] = "http://h:8005"
-    metrics[2]["routed_to"] = "http://h:8000"
     bk = [
         {"request_id": "slow_overlap", "routed_to": "http://h:8000"},
         {"request_id": "slow_no_overlap", "routed_to": "http://h:8005"},
@@ -134,7 +148,8 @@ def test_label_slow_requests_flags_overlap_and_hot_worker():
     joined = build_joined_records(metrics, bk, [])
     engine_state = {
         "engine_0": [{"t_unix": 10.5, "prefill_tokens": 5000,
-                       "per_req": [{"rid": "other", "phase": "prefill"}]}],
+                       "per_req": [{"rid": "cmpl-other-0-x",
+                                     "phase": "prefill"}]}],
     }
     labels = label_slow_requests(joined, engine_state, slow_ttft_factor=2.0)
     by_id = {L["request_id"]: L["label"] for L in labels}

tests/test_proxy_pick.py

@@ -301,6 +301,48 @@ def test_settings_has_runtime_knobs(proxy):
     s.cache_gate_ratio = saved
 
 
+def test_pick_instance_load_only_picks_min_num_requests(proxy):
+    insts = [_make_inst(proxy, "http://a"), _make_inst(proxy, "http://b"),
+             _make_inst(proxy, "http://c")]
+    insts[0].num_requests = 5
+    insts[1].num_requests = 2
+    insts[2].num_requests = 8
+    chosen, idx = proxy.pick_instance_load_only(insts, None, "sess1", 1000, {})
+    assert idx == 1 and chosen is insts[1]
+
+
+def test_pick_instance_load_only_ignores_cache_hits(proxy):
+    insts = [_make_inst(proxy, "http://a"), _make_inst(proxy, "http://b")]
+    block_size = proxy.BLOCK_SIZE
+    prefix = [123] * (block_size * 4)
+    insts[0].record_prefix(prefix)
+    insts[0].num_requests = 10
+    insts[1].num_requests = 0
+    chosen, idx = proxy.pick_instance_load_only(insts, prefix, None,
+                                                  len(prefix), {})
+    assert idx == 1, "load_only must ignore cache hit on inst[0]"
+
+
+def test_pick_instance_sticky_first_turn_picks_min_load(proxy):
+    insts = [_make_inst(proxy, "http://a"), _make_inst(proxy, "http://b")]
+    insts[0].num_requests = 10
+    insts[1].num_requests = 2
+    affinity = {}
+    chosen, idx = proxy.pick_instance_sticky(insts, None, "sess1", 1000, affinity)
+    assert idx == 1
+    assert affinity == {"sess1": 1}
+
+
+def test_pick_instance_sticky_subsequent_never_breaks(proxy):
+    """Once assigned, sticky must never re-route even under massive overload."""
+    insts = [_make_inst(proxy, "http://a"), _make_inst(proxy, "http://b")]
+    affinity = {"sess1": 0}
+    insts[0].num_requests = 1_000_000
+    insts[1].num_requests = 0
+    chosen, idx = proxy.pick_instance_sticky(insts, None, "sess1", 1000, affinity)
+    assert idx == 0, "sticky must stay even when pinned instance is saturated"
+
+
 def test_p_offload_penalty_uses_settings_heavy_threshold(proxy):
     """M2: tweaking SETTINGS.heavy_threshold changes the P-offload penalty."""
     inst = proxy.InstanceState("http://x")