Spaces:
Sleeping
Sleeping
| """End-to-end smoke test: hit the real API with every gallery note and | |
| summarize latency + the invariants that S1/S2 depend on. | |
| Runs against a live server (default http://127.0.0.1:7860). Start the API | |
| first with `uv run hcc-meat serve` or `make api` in another terminal, then: | |
| uv run python scripts/smoke_test.py | |
| What it checks: | |
| 1. /api/healthz returns ok. | |
| 2. /api/gallery returns a non-empty list. | |
| 3. For each gallery note: | |
| - POST /api/code succeeds. | |
| - Every MEAT quote in the response is a byte-exact substring of the | |
| input note text (S1). | |
| - Latency per note is recorded. | |
| 4. Aggregate latency p50/p95/max is computed against the 8s S2 target. | |
| 5. POST /api/benchmark returns consistent numbers with the per-note pass. | |
| Exits non-zero if any invariant fails. Suitable as a pre-push gate. | |
| """ | |
| from __future__ import annotations | |
| import statistics | |
| import sys | |
| from dataclasses import dataclass | |
| import httpx | |
| import typer | |
| from rich.console import Console | |
| from rich.table import Table | |
| app = typer.Typer(help=__doc__, no_args_is_help=False, invoke_without_command=True) | |
| console = Console() | |
| class CaseReport: | |
| case_id: str | |
| chars: int | |
| api_ms: float | |
| pipeline_ms: float | |
| candidates: int | |
| accepted: int | |
| verifier_drops: int | |
| fabrication: bool | |
| def run( | |
| base_url: str = typer.Option("http://127.0.0.1:7860", help="API base URL."), | |
| timeout_s: float = typer.Option(30.0, help="Per-request timeout in seconds."), | |
| target_p95_ms: float = typer.Option(8000.0, help="S2 target: p95 latency ceiling."), | |
| ) -> None: | |
| client = httpx.Client(base_url=base_url, timeout=timeout_s) | |
| # 1. health | |
| try: | |
| r = client.get("/api/healthz") | |
| r.raise_for_status() | |
| assert r.json().get("status") == "ok", r.json() | |
| except Exception as e: | |
| console.print(f"[red]health check failed: {e}[/red]") | |
| raise typer.Exit(1) | |
| # 2. gallery | |
| gal_resp = client.get("/api/gallery") | |
| gal_resp.raise_for_status() | |
| gallery = gal_resp.json() | |
| if not gallery: | |
| console.print("[red]gallery is empty[/red]") | |
| raise typer.Exit(1) | |
| console.print(f"gallery: {len(gallery)} cases") | |
| # 3. per-note pass | |
| reports: list[CaseReport] = [] | |
| fabrications: list[str] = [] | |
| for item in gallery: | |
| text = item["text"] | |
| case_id = item["case_id"] | |
| r = client.post("/api/code", json={"text": text, "note_id": case_id}) | |
| elapsed_ms = r.elapsed.total_seconds() * 1000.0 if r.elapsed else 0.0 | |
| r.raise_for_status() | |
| body = r.json() | |
| # Faithfulness check — S1 invariant. | |
| fab_here = False | |
| for c in body["candidates"]: | |
| for ev in c["meat_evidence"]: | |
| if ev["quote"] not in text: | |
| fabrications.append( | |
| f"{case_id}:{c['hcc']}:{ev['category']} → {ev['quote']!r}" | |
| ) | |
| fab_here = True | |
| reports.append( | |
| CaseReport( | |
| case_id=case_id, | |
| chars=len(text), | |
| api_ms=elapsed_ms, | |
| pipeline_ms=float(body.get("latency_ms") or 0.0), | |
| candidates=len(body["candidates"]), | |
| accepted=sum(1 for c in body["candidates"] if not c["hierarchy"]["trumped"]), | |
| verifier_drops=int(body.get("verifier_drops", 0)), | |
| fabrication=fab_here, | |
| ) | |
| ) | |
| # 4. latency rollup | |
| api_latencies = sorted(r.api_ms for r in reports) | |
| pipe_latencies = sorted(r.pipeline_ms for r in reports) | |
| table = Table(title="Smoke test — per-case") | |
| for col in ("case", "chars", "api ms", "pipeline ms", "cand", "accept", "drops", "S1"): | |
| table.add_column(col) | |
| for r in reports: | |
| table.add_row( | |
| r.case_id, | |
| str(r.chars), | |
| f"{r.api_ms:.0f}", | |
| f"{r.pipeline_ms:.0f}", | |
| str(r.candidates), | |
| str(r.accepted), | |
| str(r.verifier_drops), | |
| "[red]FAIL[/red]" if r.fabrication else "[green]OK[/green]", | |
| ) | |
| console.print(table) | |
| def _p(values: list[float], p: float) -> float: | |
| if not values: | |
| return 0.0 | |
| return values[min(len(values) - 1, int(p * (len(values) - 1)))] | |
| console.print( | |
| f"\nAPI latency median={statistics.median(api_latencies):.0f}ms " | |
| f"p95={_p(api_latencies, 0.95):.0f}ms max={max(api_latencies):.0f}ms" | |
| ) | |
| console.print( | |
| f"Pipeline median={statistics.median(pipe_latencies):.0f}ms " | |
| f"p95={_p(pipe_latencies, 0.95):.0f}ms max={max(pipe_latencies):.0f}ms" | |
| ) | |
| # 5. benchmark endpoint cross-check | |
| bench = client.post("/api/benchmark").json() | |
| console.print( | |
| f"Benchmark EP median={bench['median_ms']}ms " | |
| f"p95={bench['p95_ms']}ms max={bench['max_ms']}ms" | |
| ) | |
| # Report card. | |
| p95_api = _p(api_latencies, 0.95) | |
| fails: list[str] = [] | |
| if fabrications: | |
| fails.append(f"S1 fabrications: {len(fabrications)} items") | |
| for f in fabrications[:10]: | |
| console.print(f" [red]{f}[/red]") | |
| if p95_api > target_p95_ms: | |
| fails.append(f"S2 p95 API latency {p95_api:.0f}ms > {target_p95_ms:.0f}ms target") | |
| if fails: | |
| console.print("[red]SMOKE FAIL:[/red] " + "; ".join(fails)) | |
| raise typer.Exit(2) | |
| console.print("[green]SMOKE OK[/green]") | |
| if __name__ == "__main__": | |
| try: | |
| app() | |
| except httpx.HTTPError as e: | |
| console.print(f"[red]HTTP error: {e}[/red]") | |
| sys.exit(1) | |