"""Jev-candidate bench (2026-09-30): latency and capacity at OUR shape, through one backend. Shapes (binary yes/no criteria, single ordering, as in semif-serve's acceptance step 5): short1 one decision over authored144 row 0 (~130 input tokens) crit21 21 binary criteria over authored144 row 0's state in ONE request (semif: /decide/shared; systemone: one request with 21 questions, split into chunks of --max-questions when the runtime caps questions per request) long1 one binary criterion over the ~3,900-token state long16 16 binary criteria over the ~3,900-token state (SemIf's 12 GiB capacity at that size) Each shape: 2 warm-ups, then RUNS runs x PER requests; every request's end-to-end ms and the server-reported ms are kept. Capacity (--capacity): rows = binary criteria in one request, stepped up at the short and the long state until the first non-200. The VRAM cap is whatever the server process was started under. Phase markers (unix time) are written so the nvidia-smi poll can be split into rest / load. """ from __future__ import annotations import argparse import json import statistics as st import time from pathlib import Path from bench_sets import DATA, post YESNO = [{"id": "yes", "description": "Yes"}, {"id": "no", "description": "No"}] def criteria(n: int) -> list[dict]: return [{"id": f"c{i}", "question": f"Does the evidence mention item number {i}?", "options": YESNO} for i in range(n)] def server_ms(j: dict) -> float | None: if "timing" in j and isinstance(j["timing"], dict) and "total_seconds" in j["timing"]: return j["timing"]["total_seconds"] * 1000 # semif /decide/shared if "total_seconds" in j: return j["total_seconds"] * 1000 # semif /decide if "latency_ms" in j: return j["latency_ms"] # jevk5-serve u = j.get("usage") or {} if "total_ms" in u: return u["total_ms"] # imajev playground server t = j.get("timing") or {} if "server_ms" in t: return t["server_ms"] # our Intern-Decision wrapper return None class Semif: def __init__(self, url, token): self.url, self.h = url.rstrip("/"), {"Authorization": f"Bearer {token}"} def request(self, state, crits): if len(crits) == 1: c = crits[0] s, j, ms = post(f"{self.url}/decide", {"id": c["id"], "state": state, "question": c["question"], "options": c["options"]}, self.h) return s, ms, server_ms(j) if s == 200 else None, j.get("input_tokens") if s == 200 else None, j s, j, ms = post(f"{self.url}/decide/shared", {"state": state, "decisions": crits}, self.h) tokens = j["timing"].get("prefix_tokens") if s == 200 else None return s, ms, server_ms(j) if s == 200 else None, tokens, j class SystemOne: def __init__(self, url, max_questions): self.url, self.maxq = url.rstrip("/"), max_questions def request(self, state, crits): total_ms, total_srv, tokens, status, last = 0.0, 0.0, 0, 200, {} for k in range(0, len(crits), self.maxq): chunk = crits[k:k + self.maxq] body = {"state": state, "questions": {c["id"]: {"type": "choice", "instructions": c["question"], "criteria": {o["id"]: o["description"] for o in c["options"]}} for c in chunk}} s, j, ms = post(f"{self.url}/v1/systemone", body, {}) total_ms += ms last = j if s != 200: return s, total_ms, None, None, j srv = server_ms(j) total_srv = None if srv is None or total_srv is None else total_srv + srv tokens += (j.get("usage") or {}).get("input_tokens") or 0 return status, total_ms, total_srv, tokens, last def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--backend", choices=("semif", "systemone"), required=True) ap.add_argument("--url", required=True) ap.add_argument("--token-file") ap.add_argument("--max-questions", type=int, default=10_000) ap.add_argument("--long-state-file", required=True) ap.add_argument("--label", required=True) ap.add_argument("--out", required=True) ap.add_argument("--runs", type=int, default=3) ap.add_argument("--per", type=int, default=20) ap.add_argument("--capacity", action="store_true") ap.add_argument("--short-steps", default="16,32,48,56,60,63,64,96,128") ap.add_argument("--long-steps", default="1,4,8,12,14,16,18,20,24,32,48,64") args = ap.parse_args() b = (Semif(args.url, Path(args.token_file).read_text().strip()) if args.backend == "semif" else SystemOne(args.url, args.max_questions)) short_state = json.loads((DATA / "authored144.jsonl").read_text().splitlines()[0])["state"] long_state = Path(args.long_state_file).read_text() shapes = {"short1": (short_state, criteria(1)), "crit21": (short_state, criteria(21)), "long1": (long_state, criteria(1)), "long16": (long_state, criteria(16))} report = {"label": args.label, "backend": args.backend, "runs": args.runs, "per": args.per, "max_questions": args.max_questions, "events": [["start", time.time()]], "shapes": {}} for name, (state, crits) in shapes.items(): for _ in range(2): b.request(state, crits) report["events"].append([f"{name}:measure", time.time()]) runs = [] for _ in range(args.runs): e2e, srv, statuses, tokens = [], [], [], None for _ in range(args.per): s, ms, sm, tk, _ = b.request(state, crits) statuses.append(s) e2e.append(ms) if sm is not None: srv.append(sm) tokens = tk if tk is not None else tokens runs.append({"e2e_ms": e2e, "server_ms": srv, "statuses": statuses, "tokens": tokens, "median_e2e": st.median(e2e), "median_server": st.median(srv) if srv else None}) all_e2e = [x for r in runs for x in r["e2e_ms"]] report["shapes"][name] = { "rows": len(crits), "runs": runs, "tokens": runs[-1]["tokens"], "median_e2e_ms": round(st.median(all_e2e), 1), "run_medians_e2e_ms": [round(r["median_e2e"], 1) for r in runs], "median_server_ms": (round(st.median([x for r in runs for x in r["server_ms"]]), 1) if all(r["server_ms"] for r in runs) else None), "non_200": sum(s != 200 for r in runs for s in r["statuses"])} print(f"[{args.label}] {name}: rows={len(crits)} tokens={runs[-1]['tokens']} " f"e2e median {report['shapes'][name]['median_e2e_ms']} ms (runs {report['shapes'][name]['run_medians_e2e_ms']}) " f"server {report['shapes'][name]['median_server_ms']} non200={report['shapes'][name]['non_200']}", flush=True) report["events"].append(["shapes:done", time.time()]) if args.capacity: report["capacity"] = {} for label, state, steps in (("short", short_state, args.short_steps), ("long", long_state, args.long_steps)): series = [] for n in [int(x) for x in steps.split(",")]: report["events"].append([f"capacity:{label}:{n}", time.time()]) s, ms, sm, tk, j = b.request(state, criteria(n)) err = None if s == 200 else (j.get("error") if isinstance(j, dict) else str(j)) series.append({"rows": n, "status": s, "e2e_ms": round(ms, 1), "tokens": tk, "error": str(err)[:300] if err else None}) print(f"[{args.label}] capacity {label} rows={n}: {s} {round(ms)} ms tokens={tk} {str(err)[:120] if err else ''}", flush=True) if s != 200: break report["capacity"][label] = series report["events"].append(["capacity:done", time.time()]) report["events"].append(["end", time.time()]) Path(args.out).write_text(json.dumps(report)) return 0 if __name__ == "__main__": raise SystemExit(main())