Add live failure demo and MPS training stats

README.md

@@ -36,6 +36,12 @@ Open the local dashboard:
 http://localhost:8000/dashboard
 ```
 
+Run live traffic against the dashboard:
+
+```bash
+python bench/replay_traffic.py --seconds-per-stage 8 --rps 18
+```
+
 ## Docker
 
 ```bash
@@ -56,17 +62,17 @@ Latest local run:
 
 | Scenario | RPS | p50 ms | p95 ms | Cache hits | Failures |
 | --- | ---: | ---: | ---: | ---: | ---: |
-| cache enabled | 211.37 | 5.54 | 312.52 | 71 | 0 |
-| cache disabled | 24.92 | 318.85 | 326.38 | 0 | 0 |
-| slow worker, round-robin | 25.68 | 67.3 | 1033.24 | 0 | 0 |
-| slow worker, load-aware | 128.51 | 4.94 | 425.03 | 0 | 0 |
-| dead worker recovery | 698.97 | 11.15 | 28.1 | 0 | 0 |
+| cache enabled | 728.95 | 5.01 | 52.04 | 72 | 0 |
+| cache disabled | 185.7 | 41.1 | 53.09 | 0 | 0 |
+| slow worker, round-robin | 187.36 | 57.79 | 111.91 | 0 | 0 |
+| slow worker, load-aware | 688.48 | 5.21 | 111.9 | 0 | 0 |
+| dead worker recovery | 707.98 | 11.37 | 28.93 | 0 | 0 |
 
 Full output is in `bench/results/latest.md` and `bench/results/latest.json`.
 
 ## Training
 
-The worker can load a small trained text model artifact from `worker/app/artifacts/tiny_text_model.json`.
+The worker can load a small trained text model artifact from `worker/app/artifacts/tiny_text_model.json`. The current model trains on generated service-status examples, so the point is the local training and serving path, not benchmark-grade NLP accuracy.
 
 On Apple Silicon, the training script uses PyTorch MPS. It refuses to run on CPU by default so a slow accidental CPU run fails fast.
 
@@ -77,6 +83,21 @@ python training/train_text_model.py
 
 If it is using the Mac GPU, the output includes `'device': 'mps'`.
 
+Latest training run on an M3 Pro:
+
+| Metric | Value |
+| --- | ---: |
+| Device | mps |
+| Epochs | 320 |
+| Train examples | 3,276 |
+| Test examples | 820 |
+| Train time | 0.4351 s |
+| Examples/sec | 2,409,512.37 |
+| Train accuracy | 1.0 |
+| Test accuracy | 1.0 |
+
+The full run output is in `training/runs/latest.md` and `training/runs/latest.json`.
+
 ## What is in here now
 
 - FastAPI gateway with `/predict`, `/health`, `/workers`, and `/metrics`
@@ -85,6 +106,7 @@ If it is using the Mac GPU, the output includes `'device': 'mps'`.
 - Retry on failed worker calls and basic load-aware routing
 - Scenario benchmark runner with saved results
 - Local dashboard for gateway and worker state
+- Worker delay/failure controls for live demos
 - Unit tests for routing, caching, metrics, and the model
 
 This is still intentionally small. The useful part is the system behavior around the model, not the model itself.

bench/replay_traffic.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+
+ROOT = Path(__file__).resolve().parents[1]
+RESULT_PATH = ROOT / "bench" / "results" / "replay-latest.json"
+TEXTS = [
+    "fast stable reliable service",
+    "cache hit keeps service fast",
+    "great latency stable route",
+    "slow broken worker timeout",
+    "worker crash caused retry",
+    "request completed healthy worker",
+]
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--base-url", default="http://localhost:8000")
+    parser.add_argument("--seconds-per-stage", type=float, default=8.0)
+    parser.add_argument("--rps", type=float, default=18.0)
+    parser.add_argument("--output", type=Path, default=RESULT_PATH)
+    return parser.parse_args()
+
+
+async def control_worker(client: httpx.AsyncClient, base_url: str, index: int, *, delay_ms: int, fail_rate: float) -> bool:
+    try:
+        response = await client.post(
+            f"{base_url}/workers/{index}/control",
+            json={"delay_ms": delay_ms, "fail_rate": fail_rate},
+            timeout=1.5,
+        )
+        return response.status_code == 200
+    except httpx.HTTPError:
+        return False
+
+
+async def reset_workers(client: httpx.AsyncClient, base_url: str) -> None:
+    try:
+        workers = (await client.get(f"{base_url}/workers", timeout=1.5)).json()
+    except httpx.HTTPError:
+        return
+
+    for index, _ in enumerate(workers):
+        await control_worker(client, base_url, index, delay_ms=0, fail_rate=0)
+
+
+async def snapshot(client: httpx.AsyncClient, base_url: str, stage: str) -> dict[str, Any]:
+    metrics = (await client.get(f"{base_url}/metrics")).json()
+    workers = (await client.get(f"{base_url}/workers")).json()
+    return {
+        "stage": stage,
+        "metrics": metrics,
+        "workers": workers,
+    }
+
+
+async def send_stage(client: httpx.AsyncClient, base_url: str, stage: str, seconds: float, rps: float) -> dict[str, Any]:
+    deadline = time.perf_counter() + seconds
+    interval = 1 / rps
+    sent = 0
+    ok = 0
+    started = time.perf_counter()
+
+    while time.perf_counter() < deadline:
+        text = TEXTS[sent % len(TEXTS)]
+        if stage in {"slow-worker", "flaky-worker"}:
+            text = f"{text} {stage} live {sent}"
+
+        try:
+            response = await client.post(
+                f"{base_url}/predict",
+                json={"text": text, "request_id": f"{stage}-{sent}"},
+                timeout=3.0,
+            )
+            ok += int(response.status_code == 200)
+        except httpx.HTTPError:
+            pass
+
+        sent += 1
+        elapsed = time.perf_counter() - started
+        sleep_for = max(0.0, sent * interval - elapsed)
+        await asyncio.sleep(sleep_for)
+
+    snap = await snapshot(client, base_url, stage)
+    return {
+        "stage": stage,
+        "sent": sent,
+        "ok": ok,
+        **snap,
+    }
+
+
+async def main() -> None:
+    args = parse_args()
+    base_url = args.base_url.rstrip("/")
+    stages: list[dict[str, Any]] = []
+
+    async with httpx.AsyncClient(timeout=5.0) as client:
+        await reset_workers(client, base_url)
+        stages.append(await send_stage(client, base_url, "baseline", args.seconds_per_stage, args.rps))
+
+        await control_worker(client, base_url, 1, delay_ms=350, fail_rate=0)
+        stages.append(await send_stage(client, base_url, "slow-worker", args.seconds_per_stage, args.rps))
+
+        await reset_workers(client, base_url)
+        await control_worker(client, base_url, 0, delay_ms=0, fail_rate=0.45)
+        stages.append(await send_stage(client, base_url, "flaky-worker", args.seconds_per_stage, args.rps))
+
+        await reset_workers(client, base_url)
+        stages.append(await send_stage(client, base_url, "recovery", args.seconds_per_stage, args.rps))
+
+    payload = {
+        "base_url": base_url,
+        "seconds_per_stage": args.seconds_per_stage,
+        "target_rps": args.rps,
+        "stages": stages,
+    }
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(payload, indent=2) + "\n")
+    print(json.dumps(payload, indent=2))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())