diff --git a/benchmark/agentic-rollout/README.md b/benchmark/agentic-rollout/README.md
new file mode 100644
index 000000000..be9155ce0
--- /dev/null
+++ b/benchmark/agentic-rollout/README.md
@@ -0,0 +1,102 @@
+# Rollout Simulator and Explorer
+
+Run independent conversations against an existing SGLang server, with simulated tool waits and growing history. The client runs on any machine with Python 3.11+ and network access to the server. `uv run` installs the client dependencies; it does not need SGLang, CUDA, a trainer, or a dataset locally.
+
+```text
+prompt -> generate -> tool delay -> append tool result -> repeat
+finish configured turns -> close
+```
+
+`--conversations N --turns T` runs exactly N conversations of T turns each. `--concurrency` limits in-flight generation requests; tool waits release those slots.
+
+Inputs are readable synthetic text, tokenized, repeated, and trimmed to exact lengths. A distinct conversation prefix limits cross-conversation reuse. Model outputs remain in the history as exact token IDs. The client inserts a tool step after every generation; it does not parse or execute model-produced tool calls. The seed controls tool delays and synthetic inputs, not bitwise reproducibility of GPU outputs.
+
+## Quick start
+
+Example server command (GPU required; validation status is in the evidence note):
+
+```bash
+python -m sglang.launch_server \
+ --model-path Qwen/Qwen3-0.6B --host 0.0.0.0 --port 30000 \
+ --enable-session-radix-cache --enable-hierarchical-cache \
+ --hicache-ratio 1.6 --hicache-write-policy write_through \
+ --enable-metrics --stream-interval 1
+```
+
+From the repository root (or use an absolute script path from another directory):
+
+```bash
+uv run benchmark/agentic-rollout/simulate.py \
+ --base-url http://SERVER:30000 --tokenizer Qwen/Qwen3-0.6B \
+ --mode ordinary --conversations 8 --concurrency 8 \
+ --initial-tokens 2048 --tool-tokens 256 --output-tokens 128 \
+ --turns 8 --tool-delay 1 3 --seed 1 \
+ --output-dir results/ordinary
+
+uv run benchmark/agentic-rollout/explore.py results/ordinary --output rollout.html
+
+uv run benchmark/agentic-rollout/plot.py results/ordinary \
+ --output ordinary.png
+```
+
+The tokenizer must match the server; use a fixed local tokenizer directory for reproducible runs. `--trust-remote-code` is explicit. The defaults are a small functional workload, not a guarantee of cache pressure. For a smoke test, use `--turns 2 --initial-tokens 128 --tool-tokens 32 --output-tokens 16`.
+
+## Compare features
+
+| Client mode | History sent on each continuation |
+| --- | --- |
+| `full-history` | All previous input and output token IDs, plus the new result |
+| `ordinary` | Session/request IDs and only the new result |
+| `streaming` | Streaming-session/request IDs and only the new result |
+
+All modes use HTTP streaming for timing. The server must support the selected session mode. For servers that gate streaming sessions, launch with `--enable-streaming-session` for that mode. Unsupported requests fail; the client never substitutes a different mode.
+
+Restart the server between runs. Hold the model, workload, seed, rank assignment, and all unrelated launch settings fixed. Change one feature at a time: session mode, `--enable-session-radix-cache`, `--enable-hierarchical-cache`, `--hicache-ratio`, `--hicache-write-policy`, or `--page-size`. Supported combinations depend on the serving version and attention backend. Keep the exact server command and wheel/source revision with each result. The client records resolved `/server_info`, but cannot read a remote machine's wheel checksum or environment.
+
+By default, client-side DP sticky routing assigns conversation `i` to `i % dp_size`, using the server's reported size. Every turn stays on that rank. A missing response rank is accepted only when the server reports exactly one DP worker. Use `--disable-dp-sticky-routing` to let the server choose ranks. Choose conversation and concurrency counts divisible by DP size. Tool waits release concurrency slots. The client sends `capacity_of_str_len=1000` as an unused placeholder required by older servers; it does not change engine KV capacity.
+
+A larger pressure workload:
+
+```bash
+uv run benchmark/agentic-rollout/simulate.py \
+ --base-url http://SERVER:30000 --tokenizer /path/to/matching/tokenizer \
+ --mode ordinary --conversations 256 --concurrency 256 \
+ --initial-tokens 8192 --tool-tokens 2048 --output-tokens 128 \
+ --turns 56 --tool-delay 1 3 --seed 1 \
+ --output-dir results/page1-a
+```
+
+Check measured GPU/host capacities first. Final history per conversation is `initial + (turns - 1) * (tool + output) + output`. Increase conversations until the live histories exceed GPU capacity while fitting host capacity, then freeze the workload. Verify actual CPU restores; a large configured workload alone is not evidence of cache pressure. Use fresh output directories and servers for each comparison; repeat in reverse order to check repeatability.
+
+## Read results
+
+The client writes `manifest.json`, `requests.jsonl`, `sessions.jsonl`, and raw `metrics.jsonl`. It stops on request failures, records partial evidence, and attempts session cleanup. A close HTTP response confirms submission, not synchronous release of every cache allocation. Metrics scrape errors are recorded separately; missing metrics do not invalidate successful client timing.
+
+The plotter accepts multiple directories and writes elapsed-time and per-turn plots, per-run CSV/summary files, and a configuration sidecar for reviewing differences. Failed runs are labeled and must be excluded from clean performance comparisons. It retains their successful requests for diagnosis.
+
+TTFT starts when the client submits HTTP work and ends at the first generated-token event. Local semaphore waiting and tool delays are separate. Token timing uses observed stream events; several tokens can arrive together, so it is not a kernel timing or a distribution of individual token gaps. Output rate counts token increments in their arrival windows. Cache-hit fractions are token-weighted; missing GPU/host breakdowns stay unavailable.
+
+By default, scrape the generation endpoint's `/metrics` once per second. Repeat `--metrics-url http://NODE:PORT/metrics` for **distinct** distributed exporters. Do not list the same exporter twice through aliases. Plots keep exporters separate and show maximum reported occupancy across ranks; raw labels remain intact. Only exported ranks are observed. A flat total hit rate can conceal a shift from GPU hits to more expensive CPU restores.
+
+A candidate slowdown is three consecutive 30-second windows with p95 TTFT at least twice a preceding stable period while total cache hit stays within five percentage points. Inspect context lengths and queueing before attributing it to HiCache. Compare matching turns as well as elapsed time; faster runs advance through the workload sooner. The plotter exposes measurements rather than automatically declaring a cause.
+
+## Offline Explorer
+
+Open the generated rollout.html directly in a browser. Data, CSS and JavaScript are embedded; no server, CDN, Grafana or frontend build is needed. Keep sibling source files together when running explore.py.
+
+- **Timeline:** one conversation per lane, grouped by worker. Sampling spans HTTP submission to completion/failure; tool call spans the simulated delay **before that turn**; wait spans client queue entry to submission. Select a phase or use the request picker, including failed requests. Drag to zoom, or use Window/Position.
+- **Engine Metrics:** 15 panels with client distributions in two-second windows and server samples at recorded scrape times. Select one exporter by endpoint; exporters are never combined. Click legends to hide series.
+- **L1/L2:** GPU occupied = active + evictable; CPU occupied = host used. Each usage percentage divides by its own capacity; capacities are dashed. Hit fractions use deltas of prefill_effective_tokens_total: device, host and storage hits plus uncached input form the denominator; total hits include storage.
+- Missing counters, changed label sets, resets and zero hit denominators are gaps. Explicit host-hit zero remains 0%. Occupancy aggregates matching label sets inside the selected exporter. Mamba usage shows the maximum reported rank; speculative acceptance length shows an unweighted rank mean.
+- Older recordings without exact tool/wait timestamps mark those phases unavailable. Failed runs retain measured phases, token events and available TTFT. Interrupted runs are labeled incomplete.
+
+The output contains model identifiers, endpoint labels and recorded errors. Review it before sharing. Raw recordings and generated viewers should stay outside Git.
+
+## CPU tests
+
+```bash
+uv run --no-project --with aiohttp --with transformers --with prometheus-client \
+ python -m unittest discover -s benchmark/agentic-rollout/tests -v
+```
+
+See [the spec](spec/feature-00-session-hicache-latency.md) and [validation evidence](spec/evidence/feature-00-session-hicache-latency.md).
diff --git a/benchmark/agentic-rollout/explore.py b/benchmark/agentic-rollout/explore.py
new file mode 100644
index 000000000..4ad7ce81c
--- /dev/null
+++ b/benchmark/agentic-rollout/explore.py
@@ -0,0 +1,433 @@
+# /// script
+# requires-python = ">=3.11"
+# dependencies = ["prometheus-client"]
+# ///
+"""Build an offline Rollout Simulator and Explorer HTML from a recording."""
+
+import argparse
+import json
+import math
+import statistics
+from pathlib import Path
+
+from metrics import (
+ counter_rate,
+)
+from metrics import gauge as metric_gauge
+from metrics import (
+ hit_percentages,
+ metric_values,
+ occupancy,
+ values,
+)
+
+
+def read_rows(path):
+ if not path.exists():
+ return []
+ rows = []
+ for line in path.read_text().splitlines():
+ if line.strip():
+ try:
+ rows.append(json.loads(line))
+ except json.JSONDecodeError:
+ # Interrupted writes remain visible as recording errors.
+ rows.append({"error": "Incomplete JSONL record"})
+ return rows
+
+
+def panels_for(reqs, scrapes, origin, duration):
+ series = []
+ parsed = []
+ for row in sorted(scrapes, key=lambda r: r["timestamp"]):
+ try:
+ samples = metric_values(row.get("text", "")) if not row.get("error") else {}
+ except ValueError:
+ samples = {}
+ parsed.append((row["timestamp"] - origin, samples))
+
+ def gauge(name, mult=1):
+ return [
+ [t, None if (v := metric_gauge(ss, name)) is None else v * mult]
+ for t, ss in parsed
+ ]
+
+ def rate(name):
+ return [
+ [t1, counter_rate(s0, s1, "sglang:" + name, t1 - t0)]
+ for (t0, s0), (t1, s1) in zip(parsed, parsed[1:])
+ ]
+
+ cache_hits = [[], [], []]
+ for (t0, s0), (t1, s1) in zip(parsed, parsed[1:]):
+ for points, value in zip(
+ cache_hits, hit_percentages(s0, s1) if t1 > t0 else [None] * 3
+ ):
+ points.append([t1, value])
+ l1_occupied, l1_capacity, l2_occupied, l2_capacity = [], [], [], []
+ l1_usage, l2_usage = [], []
+ for t, samples in parsed:
+ active, evictable, capacity, usage1 = occupancy(
+ samples, ("kv_used_tokens", "kv_evictable_tokens", "max_total_num_tokens")
+ )
+ host, host_capacity, usage2 = occupancy(
+ samples, ("hicache_host_used_tokens", "hicache_host_total_tokens")
+ )
+ l1_occupied.append([t, active + evictable if active is not None else None])
+ l1_capacity.append([t, metric_gauge(samples, "max_total_num_tokens")])
+ l2_occupied.append([t, metric_gauge(samples, "hicache_host_used_tokens")])
+ l2_capacity.append([t, metric_gauge(samples, "hicache_host_total_tokens")])
+ l1_usage.append([t, usage1])
+ l2_usage.append([t, usage2])
+
+ bins = [(i, min(i + 2, duration)) for i in range(0, math.ceil(duration), 2)]
+
+ def distribution(observations):
+ buckets = [[] for _ in bins]
+ for elapsed, value in observations:
+ if 0 <= elapsed < duration:
+ buckets[int(elapsed // 2)].append(value)
+ groups = [
+ ((a + b) / 2, sorted(values)) for (a, b), values in zip(bins, buckets)
+ ]
+ return [
+ {
+ "name": name,
+ "points": [
+ [
+ t,
+ (
+ (
+ statistics.mean(v)
+ if q is None
+ else v[math.ceil(q * len(v)) - 1]
+ )
+ if v
+ else None
+ ),
+ ]
+ for t, v in groups
+ ],
+ }
+ for name, q in (("avg", None), ("p50", 0.5), ("p90", 0.9), ("p99", 0.99))
+ ]
+
+ def dist(key, scale=1):
+ return distribution(
+ [
+ (
+ r.get(
+ "completed_at",
+ r.get(
+ "failed_at",
+ r.get("first_token_at", r.get("submitted_at", origin)),
+ ),
+ )
+ - origin,
+ r[key] * scale,
+ )
+ for r in reqs
+ if r.get(key) is not None
+ ]
+ )
+
+ def add(title, unit, source, items):
+ series.append({"title": title, "unit": unit, "source": source, "series": items})
+
+ def one(name, points, **style):
+ return {"name": name, "points": points, **style}
+
+ add(
+ "QPS",
+ "req/s",
+ "Successful client completions / 2-second window",
+ [
+ one(
+ "requests",
+ [
+ [
+ (a + b) / 2,
+ sum(
+ a <= r.get("completed_at", float("inf")) - origin < b
+ and not r.get("error")
+ for r in reqs
+ )
+ / (b - a),
+ ]
+ for a, b in bins
+ ],
+ )
+ ],
+ )
+ add(
+ "Running And Queued Requests",
+ "requests",
+ "Server num_running_reqs / num_queue_reqs gauges",
+ [
+ one("running", gauge("num_running_reqs")),
+ one("queued", gauge("num_queue_reqs")),
+ ],
+ )
+ add(
+ "Input Length",
+ "tokens",
+ "Client context, grouped by completion/failure time",
+ dist("context_tokens"),
+ )
+ reqs = [
+ {**r, "output_tokens": (r.get("meta_info") or {}).get("completion_tokens")}
+ for r in reqs
+ ]
+ add(
+ "Output Length",
+ "tokens",
+ "Observed output token counts; includes partial requests",
+ dist("output_tokens"),
+ )
+ add(
+ "TTFT",
+ "ms",
+ "Client submit to first output; includes partial requests with measured TTFT",
+ dist("ttft_s", 1000),
+ )
+ # Report the distribution of observed per-token stream intervals, not GPU timing.
+ itl = []
+ for r in reqs:
+ for (t0, n0), (t1, n1) in zip(r.get("events", []), r.get("events", [])[1:]):
+ if n1 > n0:
+ itl.append(
+ (r["submitted_at"] + t1 - origin, (t1 - t0) / (n1 - n0) * 1000)
+ )
+ add(
+ "ITL",
+ "ms",
+ "Client stream intervals / received token increments; not kernel timing",
+ distribution(itl),
+ )
+ add(
+ "Prompt / Input Throughput",
+ "tokens/s",
+ "Server counter deltas / scrape interval; cached sums all tiers",
+ [
+ one("prompt", rate("prompt_tokens_total")),
+ one("cached", rate("cached_tokens_total")),
+ ],
+ )
+ add(
+ "Output Throughput",
+ "tokens/s",
+ "Server generation counter deltas / scrape interval",
+ [one("total", rate("generation_tokens_total"))],
+ )
+ add(
+ "Cache Hit Rate",
+ "%",
+ "Counter deltas from prefill_effective_tokens_total; each tier / all prompt tokens. Total includes storage hits.",
+ [
+ one("L1 GPU", cache_hits[0], color=0),
+ one("L2 CPU", cache_hits[1], color=1),
+ one("Total", cache_hits[2], color=2, dashed=True),
+ ],
+ )
+ add(
+ "E2E Request Latency",
+ "ms",
+ "Client submit to completed stream",
+ dist("latency_s", 1000),
+ )
+ add(
+ "KV Usage",
+ "%",
+ "L1: (active + evictable) / GPU capacity. L2: host used / host capacity.",
+ [one("L1 GPU", l1_usage, color=0), one("L2 CPU", l2_usage, color=1)],
+ )
+ add(
+ "KV Tokens",
+ "tokens",
+ "L1 occupied = active + evictable GPU tokens. L2 occupied = host used tokens. Dashed lines show capacity.",
+ [
+ one("L1 occupied", l1_occupied, color=0),
+ one("L1 capacity", l1_capacity, color=0, dashed=True),
+ one("L2 occupied", l2_occupied, color=1),
+ one("L2 capacity", l2_capacity, color=1, dashed=True),
+ ],
+ )
+ add(
+ "Mamba Usage",
+ "%",
+ "Maximum server mamba_usage across reported ranks",
+ [
+ one(
+ "usage",
+ [
+ [
+ t,
+ (
+ None
+ if not (v := list(values(ss, "mamba_usage").values()))
+ else max(v) * 100
+ ),
+ ]
+ for t, ss in parsed
+ ],
+ )
+ ],
+ )
+ add(
+ "Mamba Tokens",
+ "tokens",
+ "Server Mamba token gauges",
+ [
+ one("used", gauge("mamba_used_tokens")),
+ one("available", gauge("mamba_available_tokens")),
+ one("evictable", gauge("mamba_evictable_tokens")),
+ ],
+ )
+ add(
+ "Spec Acceptance Length",
+ "tokens",
+ "Unweighted mean spec_accept_length across reported ranks",
+ [
+ one(
+ "accepted",
+ [
+ [
+ t,
+ (
+ statistics.mean(v)
+ if (v := list(values(ss, "spec_accept_length").values()))
+ else None
+ ),
+ ]
+ for t, ss in parsed
+ ],
+ )
+ ],
+ )
+ return series
+
+
+def load_run(directory):
+ manifest = json.loads((directory / "manifest.json").read_text())
+ requests = read_rows(directory / "requests.jsonl")
+ scrapes = read_rows(directory / "metrics.jsonl")
+ sessions = read_rows(directory / "sessions.jsonl")
+ origin = manifest["started_at"]
+ stamps = [origin] + [
+ r[k]
+ for r in requests + scrapes + sessions
+ for k in ("completed_at", "failed_at", "submitted_at", "timestamp")
+ if isinstance(r.get(k), (int, float))
+ ]
+ stamps.extend(
+ r["submitted_at"] + event[0]
+ for r in requests
+ if "submitted_at" in r
+ for event in r.get("events", [])
+ )
+ duration = max(0.001, manifest.get("finished_at", max(stamps)) - origin)
+ info = manifest.get("server_info", {})
+ args = manifest.get("arguments", {})
+ server = info.get("server_args", info)
+ rows = []
+ for r in requests:
+ meta = r.get("meta_info") or {}
+ rank = meta.get("dp_rank")
+ if rank is None:
+ rank = r.get("rank")
+ if rank is None and info.get("dp_size") == 1:
+ rank = 0
+ phases, unavailable = [], []
+ for kind, first, last in (
+ ("Tool call", "tool_started_at", "tool_completed_at"),
+ ("Wait", "client_wait_started_at", "submitted_at"),
+ (
+ "Sampling",
+ "submitted_at",
+ "completed_at" if "completed_at" in r else "failed_at",
+ ),
+ ):
+ if kind == "Tool call" and r.get("turn") == 0:
+ continue
+ if (
+ r.get(first) is not None
+ and r.get(last) is not None
+ and r[last] >= r[first]
+ ):
+ phases.append(
+ {"type": kind, "start": r[first] - origin, "end": r[last] - origin}
+ )
+ else:
+ unavailable.append(kind)
+ rows.append(
+ {
+ "conversation": r.get("conversation", "unknown"),
+ "turn": r.get("turn", "?"),
+ "worker": rank,
+ "phases": phases,
+ "unavailable": unavailable,
+ "error": r.get("error"),
+ "ttft": r.get("ttft_s"),
+ "latency": r.get("latency_s"),
+ "wait": r.get("client_wait_s"),
+ "context": r.get("context_tokens"),
+ "output": meta.get("completion_tokens"),
+ "hits": meta.get("cached_tokens_details") or {},
+ }
+ )
+ exporters = {}
+ for scrape in scrapes:
+ if "timestamp" in scrape:
+ exporters.setdefault(scrape.get("url", "unknown"), []).append(scrape)
+ if not exporters:
+ exporters["No metrics recorded"] = []
+ summary = {
+ "model": server.get("model_path", args.get("tokenizer", "Unknown model")),
+ "duration": duration,
+ "requests": len(requests),
+ "conversations": len({r["conversation"] for r in rows}),
+ "turns": args.get("turns", "?"),
+ "concurrency": args.get("concurrency", "?"),
+ "page_size": server.get("page_size", "?"),
+ "status": (
+ manifest.get("status", "incomplete")
+ if manifest.get("finished_at")
+ else "incomplete"
+ ),
+ "errors": sum(bool(r.get("error")) for r in requests + sessions + scrapes),
+ "error": manifest.get("error"),
+ }
+ return {
+ "summary": summary,
+ "rows": rows,
+ "exporters": {
+ url: panels_for(requests, records, origin, duration)
+ for url, records in exporters.items()
+ },
+ }
+
+
+def build(directory, output):
+ data = load_run(directory)
+ assets = Path(__file__).parent
+ encoded = json.dumps(data, allow_nan=False).replace("<", r"\u003c")
+ html = (assets / "viewer.html").read_text()
+ html = html.replace("__CSS__", (assets / "viewer.css").read_text())
+ html = html.replace("__JS__", (assets / "viewer.js").read_text())
+ html = html.replace("__REAL_DATA__", encoded)
+ output.write_text(html)
+ return data
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("directory", type=Path, help="Recorded run directory")
+ parser.add_argument(
+ "--output",
+ type=Path,
+ default=Path("rollout.html"),
+ help="Self-contained HTML output",
+ )
+ args = parser.parse_args()
+ build(args.directory, args.output)
+ print(args.output)
diff --git a/benchmark/agentic-rollout/metrics.py b/benchmark/agentic-rollout/metrics.py
new file mode 100644
index 000000000..6a983cf27
--- /dev/null
+++ b/benchmark/agentic-rollout/metrics.py
@@ -0,0 +1,96 @@
+"""Prometheus parsing and label-aware cache measurements."""
+
+import math
+
+from prometheus_client.parser import text_string_to_metric_families
+
+
+def metric_values(text):
+ return {
+ (sample.name, tuple(sorted(sample.labels.items()))): sample.value
+ for family in text_string_to_metric_families(text)
+ for sample in family.samples
+ if math.isfinite(sample.value)
+ }
+
+
+def counter_rate(previous, current, name, seconds):
+ keys = {key for key in previous.keys() | current.keys() if key[0] == name}
+ if not keys or seconds <= 0:
+ return None
+ # Missing series and counter resets are gaps, not zero or negative rates.
+ if any(
+ k not in previous
+ or k not in current
+ or not math.isfinite(previous[k])
+ or not math.isfinite(current[k])
+ or current[k] < previous[k]
+ for k in keys
+ ):
+ return None
+ return sum(current[k] - previous[k] for k in keys) / seconds
+
+
+def values(samples, name):
+ return {
+ labels: value
+ for (metric, labels), value in samples.items()
+ if metric == "sglang:" + name
+ }
+
+
+def gauge(samples, name):
+ items = values(samples, name)
+ return (
+ sum(items.values())
+ if items and all(math.isfinite(v) for v in items.values())
+ else None
+ )
+
+
+def percent(used, capacity):
+ return (
+ 100 * used / capacity
+ if used is not None and capacity is not None and capacity > 0
+ else None
+ )
+
+
+def occupancy(samples, names):
+ tiers = [values(samples, name) for name in names]
+ if not all(tiers) or any(t.keys() != tiers[0].keys() for t in tiers):
+ return [None] * (len(names) + 1)
+ totals = [sum(t.values()) for t in tiers]
+ if not all(math.isfinite(v) for v in totals):
+ return [None] * (len(names) + 1)
+ return [*totals, percent(sum(totals[:-1]), totals[-1])]
+
+
+def hit_percentages(before, after):
+ old = values(before, "prefill_effective_tokens_total")
+ new = values(after, "prefill_effective_tokens_total")
+ modes = ("device_hit", "host_hit", "storage_hit", "input")
+ if not old or old.keys() != new.keys():
+ return [None] * 3
+ workers = {}
+ for key, value in new.items():
+ labels = dict(key)
+ mode = labels.pop("mode", None)
+ if mode not in modes or not math.isfinite(value) or not math.isfinite(old[key]):
+ return [None] * 3
+ delta = value - old[key]
+ if delta < 0:
+ return [None] * 3
+ workers.setdefault(tuple(sorted(labels.items())), {})[mode] = delta
+ if any(set(v) != set(modes) for v in workers.values()):
+ return [None] * 3
+ deltas = {mode: sum(v[mode] for v in workers.values()) for mode in modes}
+ denominator = sum(deltas.values())
+ return [
+ percent(v, denominator)
+ for v in (
+ deltas["device_hit"],
+ deltas["host_hit"],
+ denominator - deltas["input"],
+ )
+ ]
diff --git a/benchmark/agentic-rollout/plot.py b/benchmark/agentic-rollout/plot.py
new file mode 100644
index 000000000..de6f64a18
--- /dev/null
+++ b/benchmark/agentic-rollout/plot.py
@@ -0,0 +1,277 @@
+# /// script
+# requires-python = ">=3.11"
+# dependencies = ["matplotlib", "prometheus-client"]
+# ///
+"""Plot one or more synthetic-session result directories; no running server needed."""
+
+import argparse
+import csv
+import json
+import math
+import os
+from collections import defaultdict
+from pathlib import Path
+
+from metrics import counter_rate, metric_values
+
+
+def read_rows(path):
+ with path.open() as file:
+ for line in file:
+ if line.strip():
+ yield json.loads(line)
+
+
+def percentile(values, q):
+ values = sorted(v for v in values if v is not None)
+ return values[max(0, math.ceil(q * len(values)) - 1)] if values else None
+
+
+def cache_hit(rows, source):
+ values = [
+ (
+ r["meta_info"].get("cached_tokens")
+ if source == "total"
+ else (r["meta_info"].get("cached_tokens_details") or {}).get(source)
+ )
+ for r in rows
+ ]
+ total = sum(r["meta_info"]["prompt_tokens"] for r in rows)
+ return sum(values) / total if total and all(v is not None for v in values) else None
+
+
+def summarize(rows):
+ return {
+ "requests": len(rows),
+ "ttft_p50_s": percentile([r["ttft_s"] for r in rows], 0.5),
+ "ttft_p95_s": percentile([r["ttft_s"] for r in rows], 0.95),
+ "avg_token_time_s": (
+ sum(r["avg_token_time_s"] for r in rows) / len(rows)
+ if rows and all(r["avg_token_time_s"] is not None for r in rows)
+ else None
+ ),
+ "context_tokens": (
+ sum(r["context_tokens"] for r in rows) / len(rows) if rows else None
+ ),
+ **{
+ f"{source}_hit": cache_hit(rows, source)
+ for source in ("total", "device", "host")
+ },
+ }
+
+
+def export_csv(path, rows):
+ if not rows:
+ return
+ with path.open("w") as file:
+ writer = csv.DictWriter(file, fieldnames=list(rows[0]))
+ writer.writeheader()
+ writer.writerows(rows)
+
+
+def analyze(directory, window):
+ manifest = json.loads((directory / "manifest.json").read_text())
+ all_rows = list(read_rows(directory / "requests.jsonl"))
+ rows = [r for r in all_rows if "error" not in r]
+ start, finish = manifest["started_at"], manifest["finished_at"]
+ windows, turns, rates = defaultdict(list), defaultdict(list), defaultdict(int)
+ for row in rows:
+ windows[int((row["first_token_at"] - start) // window)].append(row)
+ turns[row["turn"]].append(row)
+ # Attribute received token increments to their actual arrival windows.
+ previous_count = 0
+ for offset, count in row["events"]:
+ rates[int((row["submitted_at"] + offset - start) // window)] += (
+ count - previous_count
+ )
+ previous_count = count
+ by_time = []
+ for index in range(int((finish - start) // window) + 1):
+ duration = min(window, finish - start - index * window)
+ if duration <= 0:
+ continue
+ by_time.append(
+ {
+ "elapsed_s": index * window,
+ **summarize(windows[index]),
+ "output_tokens_s": rates[index] / duration,
+ }
+ )
+ by_turn = [{"turn": t, **summarize(group)} for t, group in sorted(turns.items())]
+ export_csv(directory / "windows.csv", by_time)
+ export_csv(directory / "turns.csv", by_turn)
+ summary = {
+ "status": manifest["status"],
+ "failed_requests": len(all_rows) - len(rows),
+ **summarize(rows),
+ "output_tokens_s": sum(rates.values()) / (finish - start),
+ }
+ (directory / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
+ return manifest, by_time, by_turn
+
+
+def plot(directories, output, window):
+ import matplotlib
+
+ matplotlib.use("Agg")
+ import matplotlib.pyplot as plt
+
+ figure, axes = plt.subplots(4, 2, figsize=(15, 15), constrained_layout=True)
+ axes = axes.ravel()
+ titles = [
+ "TTFT (seconds)",
+ "Received output tokens / second",
+ "Average time per output token (s)",
+ "Token-weighted cache hit",
+ "Running / queued requests (per exporter)",
+ "Cache occupancy (per exporter)",
+ "Cache transfer rates (tokens/s)",
+ "Mean context tokens",
+ ]
+ turn_figure, turn_axes = plt.subplots(
+ 1, 2, figsize=(13, 5), constrained_layout=True
+ )
+ configs = {}
+ common = Path(os.path.commonpath([p.resolve().parent for p in directories]))
+ colors = plt.rcParams["axes.prop_cycle"].by_key()["color"]
+ for run_index, directory in enumerate(directories):
+ color = colors[run_index % len(colors)]
+ manifest, by_time, by_turn = analyze(directory, window)
+ label = str(directory.resolve().relative_to(common)) + (
+ " [FAILED]" if manifest["status"] != "completed" else ""
+ )
+ configs[label] = {
+ "arguments": manifest["arguments"],
+ "server_info": manifest["server_info"],
+ }
+ x = [r["elapsed_s"] for r in by_time]
+ for axis, key, suffix in [
+ (0, "ttft_p50_s", "p50"),
+ (0, "ttft_p95_s", "p95"),
+ (1, "output_tokens_s", ""),
+ (2, "avg_token_time_s", ""),
+ (3, "total_hit", "total"),
+ (3, "device_hit", "GPU"),
+ (3, "host_hit", "CPU"),
+ (7, "context_tokens", ""),
+ ]:
+ values = [r[key] for r in by_time]
+ if any(v is not None for v in values):
+ axes[axis].plot(
+ x,
+ values,
+ color=color,
+ linestyle={"p50": "--", "GPU": "--", "CPU": ":"}.get(suffix, "-"),
+ marker=".",
+ label=f"{label} {suffix}",
+ )
+ for axis, key in [(0, "ttft_p95_s"), (1, "context_tokens")]:
+ turn_axes[axis].plot(
+ [r["turn"] for r in by_turn],
+ [r[key] for r in by_turn],
+ color=color,
+ marker=".",
+ label=label,
+ )
+ series = defaultdict(list)
+ previous = {}
+ for row in read_rows(directory / "metrics.jsonl"):
+ url = row["url"]
+ if "error" in row:
+ previous.pop(url, None)
+ continue
+ current = metric_values(row["text"])
+ elapsed = row["timestamp"] - manifest["started_at"]
+ for name in (
+ "sglang:num_running_reqs",
+ "sglang:num_queue_reqs",
+ "sglang:token_usage",
+ ):
+ values = [
+ value for (metric, _), value in current.items() if metric == name
+ ]
+ if values:
+ value = max(values) if "usage" in name else sum(values)
+ series[(5 if "usage" in name else 4, url, name)].append(
+ (elapsed, value)
+ )
+ host = []
+ for (name, labels), used in current.items():
+ if name == "sglang:hicache_host_used_tokens":
+ total = current.get(("sglang:hicache_host_total_tokens", labels))
+ if total is not None and total > 0:
+ host.append(used / total)
+ if host:
+ series[(5, url, "host_usage")].append((elapsed, max(host)))
+ if url in previous:
+ stamp, before = previous[url]
+ names = {
+ name
+ for name, _ in current
+ if name.endswith("_total")
+ and any(word in name for word in ("restore", "backup", "load"))
+ and "token" in name
+ }
+ for name in sorted(names):
+ value = counter_rate(
+ before, current, name, row["timestamp"] - stamp
+ )
+ series[(6, url, name)].append((elapsed, value))
+ previous[url] = row["timestamp"], current
+ urls = {
+ url: index for index, url in enumerate(sorted({key[1] for key in series}))
+ }
+ for (axis, url, name), points in series.items():
+ if any(v is not None for _, v in points):
+ axes[axis].plot(
+ [t for t, _ in points],
+ [v for _, v in points],
+ color=color,
+ linestyle=(
+ "--"
+ if name in ("sglang:num_queue_reqs", "host_usage")
+ or "backup" in name
+ else "-"
+ ),
+ marker=".",
+ label=f"{label} exporter {urls[url]} {name.removeprefix('sglang:')}",
+ )
+ for axis, title in zip(axes, titles):
+ axis.set(title=title, xlabel="Elapsed seconds")
+ axis.grid(alpha=0.2)
+ if axis.lines:
+ axis.legend(fontsize=6)
+ else:
+ axis.text(0.5, 0.5, "Unavailable", ha="center", transform=axis.transAxes)
+ for axis, title in zip(turn_axes, ("TTFT p95 (seconds)", "Mean context tokens")):
+ axis.set(title=title, xlabel="Turn")
+ axis.grid(alpha=0.2)
+ if axis.lines:
+ axis.legend(fontsize=7)
+ figure.savefig(output, dpi=160)
+ turn_figure.savefig(output.with_name(output.stem + "-turns.png"), dpi=160)
+ output.with_suffix(".configs.json").write_text(json.dumps(configs, indent=2) + "\n")
+ plt.close("all")
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument(
+ "directories", nargs="+", type=Path, help="Run result directories to compare"
+ )
+ parser.add_argument(
+ "--output",
+ type=Path,
+ default=Path("synthetic-sessions.png"),
+ help="PNG path; also writes a turn plot and configuration sidecar",
+ )
+ parser.add_argument(
+ "--window",
+ type=float,
+ default=30,
+ help="Seconds per time window for request statistics",
+ )
+ args = parser.parse_args()
+ if args.window <= 0:
+ parser.error("--window must be positive")
+ plot(args.directories, args.output, args.window)
diff --git a/benchmark/agentic-rollout/simulate.py b/benchmark/agentic-rollout/simulate.py
new file mode 100644
index 000000000..6301cdf85
--- /dev/null
+++ b/benchmark/agentic-rollout/simulate.py
@@ -0,0 +1,443 @@
+# /// script
+# requires-python = ">=3.11"
+# dependencies = ["aiohttp", "transformers"]
+# ///
+"""Synthetic multi-turn HTTP workload. See README.md for examples."""
+
+import argparse
+import asyncio
+import hashlib
+import importlib.metadata
+import json
+import random
+import sys
+import time
+import uuid
+from pathlib import Path
+
+import aiohttp
+
+
+def write_json(path, value):
+ path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
+
+
+def record(file, value):
+ file.write(json.dumps(value, sort_keys=True) + "\n")
+ file.flush()
+
+
+def synthetic_tokens(tokenizer, seed, conversation, turn, length):
+ text = (
+ f"Session {seed}/{conversation}. Tool result {turn}: "
+ "The local search returned a synthetic measurement. "
+ "Read the result and continue the investigation. "
+ )
+ ids = tokenizer.encode(text, add_special_tokens=False)
+ if not ids:
+ raise ValueError("Tokenizer produced no input tokens")
+ return (ids * ((length + len(ids) - 1) // len(ids)))[:length]
+
+
+async def sse_events(content):
+ # aiohttp iterates complete lines, even when TCP splits a UTF-8 character.
+ data = []
+ async for raw in content:
+ line = raw.decode("utf-8").rstrip("\r\n")
+ if not line:
+ if data:
+ yield "\n".join(data)
+ data.clear()
+ elif line.startswith("data:"):
+ data.append(line[5:].lstrip(" "))
+ if data:
+ raise ValueError("Truncated SSE frame")
+
+
+async def check_response(response):
+ if response.status >= 400:
+ detail = (await response.text())[:2000]
+ raise ValueError(f"HTTP {response.status} {response.url}: {detail}")
+
+
+async def json_request(http, method, url, **kwargs):
+ async with http.request(method, url, **kwargs) as response:
+ await check_response(response)
+ return await response.json()
+
+
+async def generate(http, url, payload, row, incremental, dp_size=None):
+ start = time.perf_counter()
+ row.update(submitted_at=time.time(), events=[])
+ count, ids, meta, done = 0, [], {}, False
+ async with http.post(url + "/generate", json=payload) as response:
+ await check_response(response)
+ async for event in sse_events(response.content):
+ if event == "[DONE]":
+ done = True
+ break
+ message = json.loads(event)
+ if "error" in message:
+ raise ValueError(str(message["error"]))
+ meta = message["meta_info"]
+ current = meta["completion_tokens"]
+ if current < count:
+ raise ValueError("Completion count moved backwards")
+ if current > count:
+ row["events"].append([time.perf_counter() - start, current])
+ if not count:
+ row["first_token_at"] = time.time()
+ row["ttft_s"] = row["events"][-1][0]
+ count = current
+ if "output_ids" in message:
+ if incremental:
+ ids.extend(message["output_ids"])
+ else:
+ ids = message["output_ids"]
+ row["meta_info"] = meta
+ row.update(completed_at=time.time(), latency_s=time.perf_counter() - start)
+ if (
+ not done
+ or not count
+ or (meta.get("finish_reason") or {}).get("type") != "length"
+ ):
+ raise ValueError(
+ f"Incomplete or aborted generation: {meta.get('finish_reason')}"
+ )
+ expected = payload["sampling_params"]["max_new_tokens"]
+ if count != expected or len(ids) != expected:
+ raise ValueError(
+ f"Expected {expected} tokens, got count={count}, IDs={len(ids)}"
+ )
+ if meta["prompt_tokens"] != row["context_tokens"]:
+ raise ValueError(f"Wrong context length: {meta['prompt_tokens']}")
+ if (
+ row["rank"] is not None
+ and meta.get("dp_rank") != row["rank"]
+ and not (dp_size == 1 and row["rank"] == 0 and meta.get("dp_rank") is None)
+ ):
+ raise ValueError(f"Wrong DP rank: {meta.get('dp_rank')}")
+ events = row["events"]
+ row["avg_token_time_s"] = (
+ (events[-1][0] - events[0][0]) / (count - events[0][1])
+ if count > events[0][1]
+ else None
+ )
+ row["output_sha256"] = hashlib.sha256(json.dumps(ids).encode()).hexdigest()
+ return meta["id"], ids
+
+
+async def scrape_metrics(http, url, file, stopped):
+ while not stopped.is_set():
+ started = time.monotonic()
+ row = {"timestamp": time.time(), "url": url}
+ try:
+ async with http.get(url) as response:
+ response.raise_for_status()
+ row["text"] = await response.text()
+ except (aiohttp.ClientError, OSError, asyncio.TimeoutError) as exc:
+ row["error"] = str(exc)
+ row["duration_s"] = time.monotonic() - started
+ record(file, row)
+ try:
+ await asyncio.wait_for(stopped.wait(), max(0.01, 1 - row["duration_s"]))
+ except asyncio.TimeoutError:
+ pass
+
+
+def max_context_tokens(args):
+ return (
+ args.initial_tokens
+ + (args.turns - 1) * (args.tool_tokens + args.output_tokens)
+ + args.output_tokens
+ )
+
+
+async def conversation(
+ http, tokenizer, args, info, slot, semaphore, requests, sessions
+):
+ rank = slot % info["dp_size"] if not args.disable_dp_sticky_routing else None
+ rng = random.Random(f"{args.seed}/{slot}")
+ await asyncio.sleep(args.start_spread * slot / args.conversations)
+ sid, rid, history = None, None, []
+ identity = {"conversation": slot, "rank": rank}
+ try:
+ if args.mode != "full-history":
+ sid = uuid.uuid4().hex
+ opened = await json_request(
+ http,
+ "POST",
+ args.base_url + "/open_session",
+ json={
+ "session_id": sid,
+ "streaming": args.mode == "streaming",
+ # Required by older servers, but unused by session execution.
+ "capacity_of_str_len": 1000,
+ },
+ )
+ if opened != sid:
+ raise ValueError(f"Unexpected open response: {opened}")
+ record(
+ sessions,
+ {**identity, "event": "open", "id": sid, "timestamp": time.time()},
+ )
+ for turn in range(args.turns):
+ delay = rng.uniform(*args.tool_delay) if turn else 0
+ tool_started_at = time.time()
+ delay_start = time.perf_counter()
+ await asyncio.sleep(delay)
+ actual_delay = time.perf_counter() - delay_start
+ tool_completed_at = time.time()
+ length = args.tool_tokens if turn else args.initial_tokens
+ chunk = synthetic_tokens(tokenizer, args.seed, slot, turn, length)
+ history.extend(chunk)
+ row = {
+ **identity,
+ "turn": turn,
+ "context_tokens": len(history),
+ "tool_delay_s": delay,
+ "tool_started_at": tool_started_at if turn else None,
+ "tool_completed_at": tool_completed_at if turn else None,
+ "actual_tool_delay_s": actual_delay,
+ "input_tokens": length,
+ }
+ payload = {
+ "input_ids": history if args.mode == "full-history" else chunk,
+ "stream": True,
+ "sampling_params": {
+ "temperature": 0,
+ "ignore_eos": True,
+ "max_new_tokens": args.output_tokens,
+ },
+ }
+ if sid is not None:
+ payload["session_params"] = {"id": sid, "rid": rid}
+ if rank is not None:
+ payload["routed_dp_rank"] = rank
+ row["client_wait_started_at"] = time.time()
+ waiting = time.perf_counter()
+ try:
+ async with semaphore:
+ row["client_wait_s"] = time.perf_counter() - waiting
+ rid, output = await generate(
+ http,
+ args.base_url,
+ payload,
+ row,
+ info.get("incremental_streaming_output", False),
+ dp_size=info.get("dp_size"),
+ )
+ history.extend(output)
+ except BaseException as exc:
+ row["failed_at"] = time.time()
+ row["error"] = f"{type(exc).__name__}: {exc}"
+ raise
+ finally:
+ record(requests, row)
+ finally:
+ if sid is not None:
+ event = {
+ **identity,
+ "event": "close",
+ "id": sid,
+ "timestamp": time.time(),
+ }
+ try:
+ async with http.post(
+ args.base_url + "/close_session",
+ json={"session_id": sid},
+ timeout=aiohttp.ClientTimeout(total=10),
+ ) as response:
+ response.raise_for_status()
+ except (aiohttp.ClientError, OSError, asyncio.TimeoutError) as exc:
+ event["error"] = str(exc)
+ raise
+ finally:
+ record(sessions, event)
+
+
+async def run(args, tokenizer):
+ args.output_dir.mkdir(parents=True, exist_ok=False)
+ manifest = {
+ "arguments": {
+ k: str(v) if isinstance(v, Path) else v for k, v in vars(args).items()
+ },
+ "started_at": time.time(),
+ "status": "running",
+ "python": sys.version,
+ "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+ "dependencies": {
+ n: importlib.metadata.version(n) for n in ("aiohttp", "transformers")
+ },
+ }
+ stopped = asyncio.Event()
+ manifest_path = args.output_dir / "manifest.json"
+ write_json(manifest_path, manifest)
+ try:
+ async with (
+ aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=args.timeout),
+ connector=aiohttp.TCPConnector(limit=0),
+ ) as http,
+ aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=10)
+ ) as metrics_http,
+ ):
+ info = await json_request(http, "GET", args.base_url + "/server_info")
+ manifest["server_info"] = info
+ context = max_context_tokens(args)
+ if info.get("context_length") and context > info["context_length"]:
+ raise ValueError(
+ f"Workload requires {context} tokens; server context is too small"
+ )
+ if not args.disable_dp_sticky_routing and info.get("dp_size", 0) < 1:
+ raise ValueError("DP sticky routing requires server_info.dp_size")
+ write_json(manifest_path, manifest)
+ semaphore = asyncio.Semaphore(args.concurrency)
+ with (
+ (args.output_dir / "requests.jsonl").open("w") as requests,
+ (args.output_dir / "sessions.jsonl").open("w") as sessions,
+ (args.output_dir / "metrics.jsonl").open("w") as metrics,
+ ):
+ scrapers = [
+ asyncio.create_task(
+ scrape_metrics(metrics_http, url, metrics, stopped)
+ )
+ for url in (args.metrics_url or [args.base_url + "/metrics"])
+ ]
+ try:
+ async with asyncio.TaskGroup() as group:
+ for slot in range(args.conversations):
+ group.create_task(
+ conversation(
+ http,
+ tokenizer,
+ args,
+ info,
+ slot,
+ semaphore,
+ requests,
+ sessions,
+ )
+ )
+ finally:
+ stopped.set()
+ await asyncio.gather(*scrapers)
+ manifest["status"] = "completed"
+ except BaseException as exc:
+ manifest.update(status="failed", error=repr(exc))
+ raise
+ finally:
+ manifest["finished_at"] = time.time()
+ write_json(manifest_path, manifest)
+ print(f"{manifest['status']}: {args.output_dir}", flush=True)
+
+
+def parse_args(argv=None):
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument(
+ "--base-url", default="http://127.0.0.1:30000", help="SGLang server URL"
+ )
+ parser.add_argument(
+ "--tokenizer",
+ required=True,
+ help="Model ID or tokenizer path matching the server",
+ )
+ parser.add_argument(
+ "--trust-remote-code", action="store_true", help="Allow custom tokenizer code"
+ )
+ parser.add_argument(
+ "--mode",
+ choices=("full-history", "ordinary", "streaming"),
+ default="ordinary",
+ help="Send full history or use ordinary/streaming sessions",
+ )
+ parser.add_argument(
+ "--conversations",
+ type=int,
+ default=8,
+ help="Number of independent conversations",
+ )
+ parser.add_argument(
+ "--concurrency",
+ type=int,
+ default=8,
+ help="Maximum in-flight generation requests",
+ )
+ parser.add_argument(
+ "--turns",
+ type=int,
+ default=8,
+ help="Generation turns per conversation",
+ )
+ parser.add_argument(
+ "--initial-tokens",
+ type=int,
+ default=2048,
+ help="Tokens in each initial prompt",
+ )
+ parser.add_argument(
+ "--tool-tokens",
+ type=int,
+ default=256,
+ help="Synthetic tool-result tokens appended each turn",
+ )
+ parser.add_argument(
+ "--output-tokens",
+ type=int,
+ default=128,
+ help="Exact number of tokens generated per turn",
+ )
+ parser.add_argument(
+ "--tool-delay",
+ type=float,
+ nargs=2,
+ default=[1, 3],
+ metavar=("MIN", "MAX"),
+ help="Range of simulated tool waits between turns, in seconds",
+ )
+ parser.add_argument(
+ "--start-spread",
+ type=float,
+ default=5,
+ help="Seconds over which conversation starts are staggered",
+ )
+ parser.add_argument(
+ "--seed", type=int, default=1, help="Seed for synthetic inputs and tool delays"
+ )
+ parser.add_argument(
+ "--disable-dp-sticky-routing",
+ action="store_true",
+ help="Disable default routing of conversation i to data-parallel rank i %% dp_size",
+ )
+ parser.add_argument(
+ "--timeout",
+ type=float,
+ default=600,
+ help="HTTP request timeout in seconds",
+ )
+ parser.add_argument(
+ "--metrics-url",
+ action="append",
+ help="Metrics endpoint; repeat for distinct exporters (default: BASE_URL/metrics)",
+ )
+ parser.add_argument(
+ "--output-dir",
+ type=Path,
+ required=True,
+ help="New directory for request records, metrics, and run settings",
+ )
+ args = parser.parse_args(argv)
+ args.base_url = args.base_url.rstrip("/")
+ return args
+
+
+if __name__ == "__main__":
+ from transformers import AutoTokenizer
+
+ args = parse_args()
+ tokenizer = AutoTokenizer.from_pretrained(
+ args.tokenizer,
+ trust_remote_code=args.trust_remote_code,
+ )
+ asyncio.run(run(args, tokenizer))
diff --git a/benchmark/agentic-rollout/spec/README.md b/benchmark/agentic-rollout/spec/README.md
new file mode 100644
index 000000000..6cd9c9700
--- /dev/null
+++ b/benchmark/agentic-rollout/spec/README.md
@@ -0,0 +1,12 @@
+Status: implemented
+
+# Rollout Simulator and Explorer
+
+**TL;DR:** A portable client makes growing-conversation cache experiments repeatable.
+
+- [feature-00-session-hicache-latency.md](feature-00-session-hicache-latency.md): workload, measurements, and validation contract.
+- [evidence/feature-00-session-hicache-latency.md](evidence/feature-00-session-hicache-latency.md): exact validation commands and measured results.
+
+## Boundaries
+
+This spec covers the standalone benchmark, not the serving engine.
diff --git a/benchmark/agentic-rollout/spec/evidence/feature-00-session-hicache-latency.md b/benchmark/agentic-rollout/spec/evidence/feature-00-session-hicache-latency.md
new file mode 100644
index 000000000..6b5d819ef
--- /dev/null
+++ b/benchmark/agentic-rollout/spec/evidence/feature-00-session-hicache-latency.md
@@ -0,0 +1,52 @@
+Status: implemented
+
+# Validation evidence
+
+## Packaging checks — 2026-09-17
+
+17 CPU tests pass: all three request modes and both stream formats, default DP routing, verified-single-worker omitted ranks, strict multiworker ranks, context/output validation, aborted/truncated streams, timeouts and cleanup. Measurement tests cover occupancy, per-label resets, missing/idle counters, token-weighted denominators, separate exporters, old/interrupted recordings, partial failed requests and safe self-contained HTML.
+
+~~~bash
+uv run --no-project --with aiohttp --with transformers --with prometheus-client \
+ python -m unittest discover -s benchmark/agentic-rollout/tests -v
+~~~
+
+Viewers were rebuilt from the two saved Qwen runs below, each with 16 conversations and 96 requests. Both tabs, all 16 lanes, zoom, request selection, legends and static plots were checked. Embedded-asset tests pass, and the browser renders with external resource loading blocked by the page policy. Direct file-URL navigation was blocked by the browser automation policy, so that path was not manually verified. Packaging does not rerun GPU inference. Raw recordings, generated HTML and machine-specific orchestration stay outside this PR.
+
+## Saved Qwen comparison — 2026-09-17
+
+One H200, Qwen/Qwen3-1.7B, a downstream SGLang serving build at revision 41da06adca698c0032f75000445087f0922dbe43. These saved runs are not GPU validation against current upstream main. Ordinary sessions, session-aware cache, HiCache ratio 1.6, write-through, direct I/O, layer-first layout and resolved FA3 attention. One fresh-server run per page size.
+
+Launch command, with executable and port normalized for portability:
+
+~~~bash
+python -m sglang.launch_server \
+ --model-path Qwen/Qwen3-1.7B --host 127.0.0.1 --port 30000 \
+ --enable-session-radix-cache --enable-hierarchical-cache \
+ --hicache-ratio 1.6 --hicache-write-policy write_through \
+ --hicache-io-backend direct --hicache-mem-layout layer_first \
+ --mem-fraction-static 0.5 --max-total-tokens 32768 --context-length 8192 \
+ --cuda-graph-max-bs-decode 8 --enable-metrics --stream-interval 1 \
+ --page-size "$PAGE"
+
+uv run benchmark/agentic-rollout/simulate.py \
+ --tokenizer Qwen/Qwen3-1.7B --mode ordinary \
+ --conversations 16 --concurrency 16 --turns 6 \
+ --initial-tokens 1024 --tool-tokens 256 --output-tokens 128 \
+ --tool-delay 1 3 --start-spread 1 --seed 101 \
+ --disable-dp-sticky-routing --output-dir "results/page$PAGE"
+~~~
+
+The recorded client disabled routing hints because this single-worker server omitted its response rank. The packaged client now accepts that omission with default sticky routing only for verified DP=1; this adjustment is CPU-tested, not GPU-rerun.
+
+| Measurement | Page 1 | Page 64 |
+| --- | ---: | ---: |
+| Successful requests | 96/96 | 96/96 |
+| GPU / CPU capacity (tokens) | 32,768 / 52,429 | 32,768 / 52,480 |
+| Duration | 18.54 s | 14.96 s |
+| Output throughput | 662.8 tokens/s | 821.1 tokens/s |
+| Turns 4–5 TTFT p95 | 2,502.49 ms | 20.73 ms |
+| Turns 4–5 mean time per output token | 10.56 ms | 2.33 ms |
+| Last observed eviction / restore counters | 92,200 / 76,088 | 66,112 / 49,984 |
+
+Limitations: one short run per setting, no reverse-order repeat. Client inputs, delays and lengths matched, but server-generated random seeds differed; 85/96 output hashes matched. This validates the client and exposes cache pressure; it does not establish a general page-size speedup or a kernel-level cause. Counters are last scraped values and may miss the tail.
diff --git a/benchmark/agentic-rollout/spec/feature-00-session-hicache-latency.md b/benchmark/agentic-rollout/spec/feature-00-session-hicache-latency.md
new file mode 100644
index 000000000..bec3dac8a
--- /dev/null
+++ b/benchmark/agentic-rollout/spec/feature-00-session-hicache-latency.md
@@ -0,0 +1,27 @@
+Status: implemented
+
+# Synthetic session workload
+
+**TL;DR:** A portable HTTP client compares full-history requests, ordinary sessions, and streaming sessions using independent synthetic conversations. Server launch settings control feature ablations.
+
+```text
+generate -> seeded tool delay -> append synthetic result -> repeat
+finish configured turns -> close
+```
+
+The client runs exactly `--conversations` conversations, each with `--turns` sequential generations. Each conversation uses fixed input/output lengths and a distinct prefix. Actual output token IDs remain in its history. Conversations run concurrently, each with sequential turns; tool waits release concurrency slots. Client-side DP sticky routing is enabled by default: every turn of conversation `i` routes to rank `i % dp_size`. `--disable-dp-sticky-routing` omits the routing hint. All modes stream responses for timing.
+
+The client records resolved server settings, its arguments/version, exact request timings, streamed token-count increments, output hashes, session events, and one-second raw metrics. Request errors stop the run and trigger bounded cleanup; evidence survives. Metrics errors remain visible. Missing measurements remain unavailable.
+
+Separate plots show time and turn number, TTFT p50/p95, observed output throughput, average token timing, token-weighted GPU/CPU/total cache hit, running/queued requests, occupancy, and available transfer rates. Independent metrics exporters remain separate. Several tokens arriving together are not individual token-gap samples.
+
+Cache-pressure claims require observed restores and measured capacity. Public smoke success does not establish slowdown reproduction. Exact commands and measured results belong in [evidence](evidence/feature-00-session-hicache-latency.md).
+
+## Boundaries
+
+Synthetic text and simulated tool delays only. No trainer, training-stack dependency, real tool execution, engine changes, server orchestration, or private deployment details in the shared client. Features are tested only when supported by the chosen server; no silent compatibility fallback.
+
+
+## Explorer packaging (2026-09-17)
+
+Keep the client, plotter, shared Prometheus parser, viewer assets and tests in benchmark/agentic-rollout/. The explorer embeds all assets/data without a frontend dependency. Timeline contains only sampling, tool call and client wait using exact timestamps; missing old phases remain unavailable. Engine Metrics retains 15 panels with L1/L2 in KV Usage, KV Tokens and Cache Hit Rate. Exporters remain separately selectable. Missing/reset/idle counter intervals are gaps. Preserve failed-run status and partial measurements. Verify with CPU tests and both saved 16-conversation Qwen recordings; no new GPU run is required.
diff --git a/benchmark/agentic-rollout/tests/test_explore.py b/benchmark/agentic-rollout/tests/test_explore.py
new file mode 100644
index 000000000..d38ba2a76
--- /dev/null
+++ b/benchmark/agentic-rollout/tests/test_explore.py
@@ -0,0 +1,173 @@
+"""Explorer checks use small recordings, without GPUs or external services."""
+
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import explore
+from metrics import counter_rate, hit_percentages, metric_values, occupancy
+
+
+def counters(device=0, host=0, storage=0, uncached=0, rank="0"):
+ return metric_values(
+ "\n".join(
+ f'sglang:prefill_effective_tokens_total{{mode="{mode}",dp_rank="{rank}"}} {value}'
+ for mode, value in zip(
+ ("device_hit", "host_hit", "storage_hit", "input"),
+ (device, host, storage, uncached),
+ )
+ )
+ )
+
+
+class MetricTests(unittest.TestCase):
+ def test_occupancy_includes_evictable_and_matches_labels(self):
+ samples = metric_values("""
+sglang:kv_used_tokens{rank="0"} 20
+sglang:kv_evictable_tokens{rank="0"} 30
+sglang:max_total_num_tokens{rank="0"} 100
+sglang:hicache_host_used_tokens{rank="0"} 80
+sglang:hicache_host_total_tokens{rank="0"} 200
+""")
+ names = ("kv_used_tokens", "kv_evictable_tokens", "max_total_num_tokens")
+ self.assertEqual(occupancy(samples, names), [20, 30, 100, 50])
+ self.assertEqual(
+ occupancy(
+ samples, ("hicache_host_used_tokens", "hicache_host_total_tokens")
+ ),
+ [80, 200, 40],
+ )
+ samples.update(metric_values('sglang:kv_used_tokens{rank="1"} 1'))
+ self.assertEqual(occupancy(samples, names), [None] * 4)
+
+ def test_hits_include_storage_and_uncached_denominator(self):
+ self.assertEqual(
+ hit_percentages(counters(), counters(40, 20, 10, 30)), [40, 20, 70]
+ )
+ self.assertEqual(
+ hit_percentages(counters(), counters(80, 0, 0, 20)), [80, 0, 80]
+ )
+ self.assertEqual(hit_percentages(counters(), counters()), [None] * 3)
+ missing = counters(80, 0, 0, 20)
+ missing.pop(next(k for k in missing if dict(k[1])["mode"] == "host_hit"))
+ self.assertEqual(hit_percentages(counters(), missing), [None] * 3)
+
+ def test_per_label_reset_cannot_be_hidden_by_another_worker(self):
+ before = counters(20) | counters(20, rank="1")
+ after = counters(19) | counters(100, rank="1")
+ self.assertEqual(hit_percentages(before, after), [None] * 3)
+ self.assertIsNone(
+ counter_rate(before, after, "sglang:prefill_effective_tokens_total", 1)
+ )
+ self.assertIsNone(
+ counter_rate({}, after, "sglang:prefill_effective_tokens_total", 1)
+ )
+
+ def test_scrape_error_breaks_rates(self):
+ records = [
+ {"timestamp": 0, "text": "sglang:generation_tokens_total 1"},
+ {"timestamp": 1, "error": "timeout"},
+ {"timestamp": 2, "text": "sglang:generation_tokens_total 9"},
+ ]
+ panels = explore.panels_for([], records, 0, 3)
+ rate = next(p for p in panels if p["title"] == "Output Throughput")
+ self.assertEqual(rate["series"][0]["points"], [[1, None], [2, None]])
+
+ def test_nonfinite_missing_and_zero_capacity(self):
+ self.assertEqual(metric_values("sglang:kv_used_tokens NaN"), {})
+ self.assertEqual(occupancy({}, ("a", "b")), [None] * 3)
+ self.assertEqual(
+ occupancy(metric_values("sglang:a 1\nsglang:b 0"), ("a", "b")), [1, 0, None]
+ )
+
+
+class RecordingTests(unittest.TestCase):
+ def test_old_failed_recording_and_offline_assets(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp)
+ (path / "manifest.json").write_text(
+ json.dumps(
+ {
+ "started_at": 10,
+ "status": "failed",
+ "finished_at": 15,
+ "server_info": {"dp_size": 1},
+ "arguments": {
+ "tokenizer": ""
+ },
+ }
+ )
+ )
+ (path / "requests.jsonl").write_text(
+ json.dumps(
+ {
+ "conversation": 7,
+ "turn": 2,
+ "submitted_at": 11,
+ "failed_at": 14,
+ "ttft_s": 0.5,
+ "first_token_at": 11.5,
+ "events": [[0.5, 1], [1, 2]],
+ "error": "disconnected",
+ "meta_info": {"completion_tokens": 2},
+ }
+ )
+ + "\n"
+ )
+ (path / "metrics.jsonl").write_text(
+ "\n".join(
+ json.dumps(
+ {
+ "timestamp": 12,
+ "url": url,
+ "text": f"sglang:num_running_reqs {n}",
+ }
+ )
+ for url, n in (("one", 2), ("two", 3))
+ )
+ )
+ output = path / "view.html"
+ data = explore.build(path, output)
+ row = data["rows"][0]
+ self.assertEqual(row["worker"], 0)
+ self.assertEqual(row["unavailable"], ["Tool call", "Wait"])
+ self.assertEqual(
+ row["phases"], [{"type": "Sampling", "start": 1, "end": 4}]
+ )
+ self.assertEqual(data["summary"]["errors"], 1)
+ self.assertEqual(set(data["exporters"]), {"one", "two"})
+ for url, expected in (("one", 2), ("two", 3)):
+ panels = data["exporters"][url]
+ self.assertEqual(len(panels), 15)
+ running = next(
+ p for p in panels if p["title"] == "Running And Queued Requests"
+ )
+ self.assertEqual(running["series"][0]["points"], [[2, expected]])
+ ttft = next(p for p in panels if p["title"] == "TTFT")
+ self.assertTrue(any(v == 500 for _, v in ttft["series"][0]["points"]))
+ html = output.read_text()
+ self.assertNotIn("
+
+