feat(bench): add three-phase prefill fleet benchmark

docs/ops/distributed-prefill-kv-network.md

@@ -275,6 +275,33 @@ publish failures, fallbacks, or `/tmp/kakeya-cache-fill.stop`, and never expects
 resident fleet usage to exceed the configured 1+8 GiB ceiling. Churn is accepted
 when `bytes_evicted` increases while resident bytes remain bounded.
 
+## Three-phase architecture benchmark
+
+Run from Primary; services are started only if missing and remain running:
+
+```bash
+bash scripts/run_prefill_architecture_benchmark.sh \
+  --output-tokens 32 \
+  --report /tmp/kakeya-prefill-benchmark.json
+```
+
+The task runs `remote_compute`, `primary_hot_hit`, and
+`allens_cold_restore`, recording client-side append latency, TTFT, decode
+latency/tokens-per-second, E2E throughput, and server-side hit/promotion deltas.
+Reports never persist prompts, token IDs, cache keys, raw addresses, or user
+paths.
+
+Benchmark APIs are public reads and API-key writes:
+
+```bash
+curl -fsS https://kakeya.ai/v1/network/benchmarks
+curl -fsS https://kakeya.ai/v1/network/benchmarks/live
+curl -fsS https://kakeya.ai/v1/network/benchmarks/<run-id>
+```
+
+The `Benchmarks` dashboard tab shows live progress, phase comparison, history,
+and complete redacted stage details.
+
 ## Rollback
 
 The cache is an optimization; inference correctness does not depend on it.

inference_engine/bench/prefill_fleet_report.py

@@ -0,0 +1,99 @@
+"""Canonical report schema and aggregation for the two-Mac prefill benchmark."""
+from __future__ import annotations
+
+import statistics
+from typing import Any, Sequence
+
+PHASES = ("remote_compute", "primary_hot_hit", "allens_cold_restore")
+HIT_SOURCES = ("remote_worker", "primary_hot", "allens_offload", "unknown")
+_PRIVATE_KEYS = {
+    "prompt",
+    "token_ids",
+    "cache_key",
+    "block_hash",
+    "payload_sha256",
+    "peer_address",
+    "source_path",
+}
+
+
+def normalize_stage(stage: dict[str, Any]) -> dict[str, Any]:
+    name = stage.get("name")
+    if name not in PHASES:
+        raise ValueError(f"unknown benchmark phase {name!r}")
+    hit_source = stage.get("hit_source", "unknown")
+    if hit_source not in HIT_SOURCES:
+        raise ValueError(f"unknown hit_source {hit_source!r}")
+    output_tokens = int(stage.get("output_tokens", 0))
+    prefix_tokens = int(stage.get("prefix_tokens", 0))
+    append_s = float(stage.get("append_s", 0.0))
+    decode_s = float(stage.get("decode_s", 0.0))
+    e2e_s = float(stage.get("e2e_s", 0.0))
+    if min(output_tokens, prefix_tokens) < 0 or min(append_s, decode_s, e2e_s) < 0:
+        raise ValueError("benchmark token and duration values must be non-negative")
+    normalized = dict(stage)
+    normalized.update({
+        "name": name,
+        "hit_source": hit_source,
+        "prefix_tokens": prefix_tokens,
+        "output_tokens": output_tokens,
+        "append_s": append_s,
+        "ttft_s": float(stage.get("ttft_s", 0.0)),
+        "decode_s": decode_s,
+        "e2e_s": e2e_s,
+        "prefill_or_restore_tok_s": (
+            prefix_tokens / append_s if append_s > 0 else 0.0
+        ),
+        "decode_tok_s": output_tokens / decode_s if decode_s > 0 else 0.0,
+        "generation_latency_ms_per_token": (
+            decode_s / output_tokens * 1000.0 if output_tokens > 0 else 0.0
+        ),
+        "e2e_tok_s": output_tokens / e2e_s if e2e_s > 0 else 0.0,
+    })
+    assert_public_safe(normalized)
+    return normalized
+
+
+def summarize_stages(stages: Sequence[dict[str, Any]]) -> dict[str, Any]:
+    normalized = [normalize_stage(stage) for stage in stages]
+    sources = {source: 0 for source in HIT_SOURCES}
+    for stage in normalized:
+        sources[stage["hit_source"]] += 1
+    decode = [stage["decode_tok_s"] for stage in normalized]
+    return {
+        "stages_total": len(normalized),
+        "stages_failed": sum(not stage.get("ok", False) for stage in normalized),
+        "ttft_p50_s": _median(stage["ttft_s"] for stage in normalized),
+        "prefill_tok_s_p50": _median(
+            stage["prefill_or_restore_tok_s"] for stage in normalized
+        ),
+        "decode_tok_s_p50": statistics.median(decode) if decode else 0.0,
+        "e2e_tok_s_p50": _median(stage["e2e_tok_s"] for stage in normalized),
+        "generation_latency_ms_p50": _median(
+            stage["generation_latency_ms_per_token"] for stage in normalized
+        ),
+        "bytes_received": sum(
+            int(stage.get("delta", {}).get("bytes_received", 0))
+            for stage in normalized
+        ),
+        "hit_source_counts": sources,
+    }
+
+
+def assert_public_safe(value: Any) -> None:
+    if isinstance(value, dict):
+        for key, child in value.items():
+            if key in _PRIVATE_KEYS:
+                raise ValueError(f"private benchmark field {key!r} is forbidden")
+            assert_public_safe(child)
+    elif isinstance(value, (list, tuple)):
+        for child in value:
+            assert_public_safe(child)
+    elif isinstance(value, str):
+        if "/Users/" in value or "169.254." in value:
+            raise ValueError("benchmark report contains a private path or address")
+
+
+def _median(values) -> float:
+    items = list(values)
+    return float(statistics.median(items)) if items else 0.0

inference_engine/network/api.py

@@ -40,6 +40,18 @@ class DrainCaptureRequest(BaseModel):
     max_items: int = Field(default=8, ge=1, le=64)
 
 
+class BenchmarkCreateRequest(BaseModel):
+    kind: str = Field(default="distributed_prefill_fleet_benchmark", max_length=100)
+    config: dict = Field(default_factory=dict)
+    started_at: Optional[float] = None
+
+
+class BenchmarkUpdateRequest(BaseModel):
+    stages: list[dict] = Field(default_factory=list)
+    status: Optional[str] = None
+    finished_at: Optional[float] = None
+
+
 def create_network_app(
     state: NetworkState,
     *,
@@ -124,6 +136,56 @@ def tokens():
     def prefill():
         return state.prefill_stats()
 
+    @app.get("/v1/network/benchmarks")
+    def benchmarks(limit: int = 20, status: Optional[str] = None):
+        try:
+            return state.list_benchmarks(limit=limit, status=status)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.get("/v1/network/benchmarks/live")
+    def benchmark_live():
+        return state.live_benchmark()
+
+    @app.get("/v1/network/benchmarks/{run_id}")
+    def benchmark_detail(run_id: str):
+        try:
+            return state.get_benchmark(run_id)
+        except KeyError as exc:
+            raise HTTPException(status_code=404, detail="benchmark not found") from exc
+
+    @app.get("/v1/network/benchmarks/{run_id}/stages")
+    def benchmark_stages(run_id: str, offset: int = 0, limit: int = 50):
+        if offset < 0 or not 1 <= limit <= 200:
+            raise HTTPException(status_code=400, detail="invalid stage pagination")
+        try:
+            stages = state.get_benchmark(run_id)["stages"]
+        except KeyError as exc:
+            raise HTTPException(status_code=404, detail="benchmark not found") from exc
+        return {"items": stages[offset:offset + limit], "total": len(stages)}
+
+    @app.post(
+        "/v1/network/benchmarks",
+        dependencies=[Depends(require_key)],
+    )
+    def create_benchmark(request: BenchmarkCreateRequest):
+        try:
+            return state.create_benchmark(**request.model_dump())
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.patch(
+        "/v1/network/benchmarks/{run_id}",
+        dependencies=[Depends(require_key)],
+    )
+    def update_benchmark(run_id: str, request: BenchmarkUpdateRequest):
+        try:
+            return state.update_benchmark(run_id, **request.model_dump())
+        except KeyError as exc:
+            raise HTTPException(status_code=404, detail="benchmark not found") from exc
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
     @app.get(
         "/v1/network/maintenance/capture",
         dependencies=[Depends(require_maintenance_key)],

inference_engine/network/dashboard.py

@@ -30,29 +30,32 @@ def dashboard_html() -> str:
 </head>
 <body><main class="wrap">
 <div class="top"><div><h1>Kakeya Inference Network</h1><div class="sub">P2P Prefill KV sharing across trusted inference nodes</div></div>
-<div class="tabs"><button class="active" data-tab="overview">Overview</button><button data-tab="nodes">Nodes</button><button data-tab="groups">Groups</button><button class="primary" id="registerBtn">Register node</button></div></div>
+<div class="tabs"><button class="active" data-tab="overview">Overview</button><button data-tab="nodes">Nodes</button><button data-tab="groups">Groups</button><button data-tab="benchmarks">Benchmarks</button><button class="primary" id="registerBtn">Register node</button></div></div>
 <section id="register" class="card register hidden"><h3>Register inference node</h3><div class="grid3"><label>Alias<input id="alias" placeholder="prefill-worker-tb"></label><label>Address<input id="address" placeholder="169.254.27.104:53051"></label><label>Region<input id="region" placeholder="Hong Kong"></label></div><label>Admin API key<input id="adminKey" type="password" placeholder="Required for network changes"></label><button class="primary" id="createRegistration">Create pairing token</button><code id="pairing" class="hidden"></code></section>
 <section class="stats"><div class="card stat"><b id="online">0</b><span>Online nodes</span></div><div class="card stat"><b id="groupCount">0</b><span>Inference groups</span></div><div class="card stat"><b id="tokens">0</b><span>Completed tokens</span></div><div class="card stat"><b id="hitRate">0%</b><span>KV-assisted tokens</span></div><div class="card stat"><b id="cache">0 GB</b><span>Shared cache online</span></div><div class="card stat"><b id="remoteJobs">0</b><span>Remote prefill jobs</span></div><div class="card stat"><b id="remoteHits">0</b><span>Remote KV imports</span></div><div class="card stat"><b id="reusedTokens">0</b><span>Tokens reused</span></div><div class="card stat"><b id="evictions">0</b><span>LRU evictions</span></div><div class="card stat"><b id="publishFailures">0</b><span>Publish failures</span></div></section>
 <section id="overview" class="tab">
 <div class="grid2"><div><h2>Online node distribution</h2><div class="card map" id="map"></div></div><div><h2>Live KV discovery</h2><div class="card" id="events"><div class="event"><time>live</time><div><b>Waiting for node telemetry</b><div class="muted">Capability gossip and prefix lookups appear here.</div></div></div></div><h2>Cache capacity</h2><div class="card"><span id="capacityLabel">0 / 0 GB</span><div class="bar"><i id="capacityBar" style="width:0%"></i></div></div></div></div>
 </section>
 <section id="nodes" class="tab hidden"><h2>Registered inference nodes</h2><div class="card"><table><thead><tr><th>Node</th><th>Role</th><th>Region</th><th>Model / Cache</th><th>Link</th><th>Status</th></tr></thead><tbody id="nodesBody"></tbody></table></div></section>
 <section id="groups" class="tab hidden"><h2>Paired inference groups</h2><div id="groupCards" class="grid3"></div><div class="card register"><h3>Create group</h3><label>Name<input id="groupName" placeholder="Snow Fox Studio"></label><label>Node IDs (comma separated)<input id="groupNodes"></label><button class="primary" id="createGroup">Create group</button></div></section>
+<section id="benchmarks" class="tab hidden"><h2>Live benchmark</h2><div class="card" id="benchmarkLive">No benchmark running.</div><h2>Phase comparison</h2><div class="grid3" id="benchmarkPhases"><div class="card muted">Select a benchmark run.</div></div><h2>Recent benchmark history</h2><div class="card"><table><thead><tr><th>Run</th><th>Status</th><th>Started</th><th>Hit sources</th><th>Prefill tok/s</th><th>Decode tok/s</th><th>E2E tok/s</th></tr></thead><tbody id="benchmarkHistory"></tbody></table></div><h2>Run detail</h2><div class="card" id="benchmarkDetail">Select a run from history.</div></section>
 <p class="muted">Exact IPs, raw prompt hashes and cache keys are administrator-only. Region is operator-selected and coarse.</p>
 </main>
 <script>
-const $=id=>document.getElementById(id), fmt=n=>new Intl.NumberFormat().format(n||0), gb=n=>(n/1073741824).toFixed(1);
+const $=id=>document.getElementById(id), fmt=n=>new Intl.NumberFormat().format(n||0), gb=n=>(n/1073741824).toFixed(1), num=n=>(Number(n||0)).toFixed(2), esc=s=>String(s??'').replace(/[&<>"']/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
 document.querySelectorAll('[data-tab]').forEach(b=>b.onclick=()=>{document.querySelectorAll('[data-tab]').forEach(x=>x.classList.remove('active'));b.classList.add('active');document.querySelectorAll('.tab').forEach(x=>x.classList.add('hidden'));$(b.dataset.tab).classList.remove('hidden')});
 $('registerBtn').onclick=()=>$('register').classList.toggle('hidden');
 const writeHeaders=()=>({'content-type':'application/json','X-API-Key':$('adminKey').value});
 $('createRegistration').onclick=async()=>{let r=await fetch('/v1/network/nodes/register',{method:'POST',headers:writeHeaders(),body:JSON.stringify({alias:$('alias').value,address:$('address').value,region:$('region').value,role:'hybrid'})});let j=await r.json();$('pairing').textContent=r.ok?`Pairing token: ${j.pairing_token}\nExpires: ${new Date(j.expires_at*1000).toLocaleTimeString()}`:`Error: ${j.detail||r.status}`;$('pairing').classList.remove('hidden');load()};
 $('createGroup').onclick=async()=>{await fetch('/v1/network/groups',{method:'POST',headers:writeHeaders(),body:JSON.stringify({name:$('groupName').value,node_ids:$('groupNodes').value.split(',').map(x=>x.trim()).filter(Boolean)})});load()};
 function nodePosition(i,total){let a=(i/Math.max(total,1))*Math.PI*2;return {x:50+38*Math.cos(a),y:53+35*Math.sin(a)}}
-async function load(){let [s,n,g]=await Promise.all([fetch('/v1/network/summary').then(r=>r.json()),fetch('/v1/network/nodes').then(r=>r.json()),fetch('/v1/network/groups').then(r=>r.json())]);
+function phaseCards(stages){$('benchmarkPhases').innerHTML=stages.map(x=>`<div class="card"><b>${esc(x.name)}</b><p>${esc(x.hit_source)} · ${x.ok?'PASS':'FAIL'}</p><div class="muted">TTFT ${num(x.ttft_s)}s<br>Prefill/restore ${num(x.prefill_or_restore_tok_s)} tok/s<br>Decode ${num(x.decode_tok_s)} tok/s<br>Generation ${num(x.generation_latency_ms_per_token)} ms/token<br>E2E ${num(x.e2e_tok_s)} tok/s</div></div>`).join('')||'<div class="card muted">No stages yet.</div>'}
+async function showBenchmark(id){let r=await fetch('/v1/network/benchmarks/'+encodeURIComponent(id)).then(x=>x.json());phaseCards(r.stages||[]);$('benchmarkDetail').innerHTML=`<b>${esc(r.id)} · ${esc(r.status)}</b><p class="muted">${esc(r.kind)} · ${new Date(r.started_at*1000).toLocaleString()}</p><table><thead><tr><th>Phase</th><th>Source</th><th>TTFT</th><th>Prefill/restore</th><th>Decode</th><th>Latency/token</th><th>E2E</th></tr></thead><tbody>${(r.stages||[]).map(x=>`<tr><td>${esc(x.name)}</td><td>${esc(x.hit_source)}</td><td>${num(x.ttft_s)}s</td><td>${num(x.prefill_or_restore_tok_s)}</td><td>${num(x.decode_tok_s)}</td><td>${num(x.generation_latency_ms_per_token)}ms</td><td>${num(x.e2e_tok_s)}</td></tr>`).join('')}</tbody></table>`}
+async function load(){let [s,n,g,live,runs]=await Promise.all([fetch('/v1/network/summary').then(r=>r.json()),fetch('/v1/network/nodes').then(r=>r.json()),fetch('/v1/network/groups').then(r=>r.json()),fetch('/v1/network/benchmarks/live').then(r=>r.json()),fetch('/v1/network/benchmarks?limit=20').then(r=>r.json())]);
 $('online').textContent=s.online_nodes;$('groupCount').textContent=s.groups;$('tokens').textContent=fmt(s.completed_tokens);$('hitRate').textContent=(s.kv_hit_rate*100).toFixed(0)+'%';$('cache').textContent=gb(s.cache_bytes_used+s.cache_bytes_free)+' GB';let p=s.prefill||{};$('remoteJobs').textContent=fmt(p.remote_jobs);$('remoteHits').textContent=fmt(p.remote_hits);$('reusedTokens').textContent=fmt(p.tokens_reused);$('evictions').textContent=fmt(s.cache_evictions);$('publishFailures').textContent=fmt(p.publish_failures);
 let total=s.cache_bytes_used+s.cache_bytes_free,pct=total?s.cache_bytes_used/total*100:0;$('capacityLabel').textContent=`${gb(s.cache_bytes_used)} / ${gb(total)} GB`;$('capacityBar').style.width=pct+'%';
 $('map').innerHTML=n.map((x,i)=>{let p=nodePosition(i,n.length);return `<div class="dot ${x.role.includes('head')?'head':''}" style="left:${p.x}%;top:${p.y}%"><i></i><small>${x.region}<br>${x.alias}</small></div>`}).join('');
 $('nodesBody').innerHTML=n.map(x=>`<tr><td>${x.alias}</td><td>${x.role}</td><td>${x.region}</td><td>${x.cache?(x.cache.model_id+' / '+x.cache.format):'—'}</td><td>${x.endpoint.network} ${x.endpoint.rtt_ms?x.endpoint.rtt_ms+'ms':''}</td><td class="${x.status}">${x.status}</td></tr>`).join('');
-$('groupCards').innerHTML=g.map(x=>`<div class="card"><b>${x.name}</b><p class="muted">${x.online} / ${x.node_ids.length} online</p><div>${x.node_ids.join(' · ')}</div></div>`).join('')||'<div class="card muted">Create the first inference group.</div>'}
+$('groupCards').innerHTML=g.map(x=>`<div class="card"><b>${x.name}</b><p class="muted">${x.online} / ${x.node_ids.length} online</p><div>${x.node_ids.join(' · ')}</div></div>`).join('')||'<div class="card muted">Create the first inference group.</div>';$('benchmarkLive').innerHTML=live?`<b>${esc(live.id)} · ${esc(live.status)}</b><p class="muted">${esc(live.kind)} · ${(live.stages||[]).length} stages completed</p>`:'No benchmark running.';$('benchmarkHistory').innerHTML=runs.map(r=>{let s=r.summary||{},h=s.hit_source_counts||{};return `<tr data-run="${esc(r.id)}"><td><button onclick="showBenchmark('${esc(r.id)}')">${esc(r.id)}</button></td><td>${esc(r.status)}</td><td>${new Date(r.started_at*1000).toLocaleString()}</td><td>worker ${h.remote_worker||0} · hot ${h.primary_hot||0} · offload ${h.allens_offload||0}</td><td>${num(s.prefill_tok_s_p50)}</td><td>${num(s.decode_tok_s_p50)}</td><td>${num(s.e2e_tok_s_p50)}</td></tr>`}).join('')||'<tr><td colspan="7" class="muted">No benchmark history.</td></tr>';if(live)phaseCards(live.stages||[])}
 load();setInterval(load,5000);
 </script></body></html>"""