diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b58edf3..77a7d56 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -108,3 +108,40 @@ jobs: # The Chromium revision that playwright-core inside the runtime's pinned @playwright/mcp expects. - run: npx -y -p "$(uv run --no-sync python -c 'import s1a.entry; from s1a.config import DEFAULT_PLAYWRIGHT_MCP_ARGS as a; print(a.split()[-1])')" playwright-core install --with-deps chromium - run: S1A_BROWSER_TESTS=1 uv run pytest -q tests/system + + benchmarks: + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + JEVBENCH_ROOT: ${{ github.workspace }}/_reference/jevbench + OMNI_BENCH_SOURCE: ${{ github.workspace }}/_reference/omni/benchmarks/bench.py + steps: + - uses: actions/checkout@v7 + - uses: actions/checkout@v7 + with: + repository: fstandhartinger/jevbench + ref: 7ce310c7262ed49cc85853339a8a42459298e3f3 + path: _reference/jevbench + persist-credentials: false + - uses: actions/checkout@v7 + with: + repository: ThinkFlowLab/system1-omni + ref: 4b9cccbde841272ff5ace6aee3cf51610fe1324f + path: _reference/omni + persist-credentials: false + - uses: astral-sh/setup-uv@v10.2.0 + with: + python-version: "3.11" + enable-cache: true + - run: uv sync --extra dev + - name: Required benchmark tests + run: | + uv run pytest tests/benchmarks -q --junitxml="${RUNNER_TEMP}/benchmark-tests.xml" + uv run python - <<'PY' + import os + import xml.etree.ElementTree as ET + from pathlib import Path + + report = Path(os.environ["RUNNER_TEMP"]) / "benchmark-tests.xml" + raise SystemExit(bool(ET.parse(report).findall(".//skipped"))) + PY diff --git a/CHANGELOG.md b/CHANGELOG.md index 92bc76c..67f0f5c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); ver ### Added +- Offline quality consumer for shared Omni public231 raw, retaining planned + denominators and original scoring. + +- Public231 decision-model benchmark with raw HTTP evidence, pinned upstream scoring, + complete-denominator quality metrics and per-round service latency. + - Optional local Cua-S1 4B inference through the `cua-four-b` extra and `CUA_S1_VARIANT=4b`, supporting text and multimodal input. Desktop `--pixel-target` offers named screenshot points; clicks remain bound to the observed window and capture. diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 0000000..2f392b6 --- /dev/null +++ b/benchmarks/README.md @@ -0,0 +1,144 @@ +# Decision model benchmarks + +Compare an existing decision service on JevBench's 231 public questions: easy48, +original72 and hard111. The collector saves every request and response; a separate +command re-scores them with the pinned upstream scorer. + +[H800 results](results/public231-h800.md) compare five open checkpoints. This is the +public231 set, not the article's 248-question or 23-model leaderboard. + +The [accuracy leaderboard](accuracy-leaderboard.md) ranks that frozen run and +defines how to add comparable results. Request latency has a separate +[System1-Omni leaderboard](https://github.com/ThinkFlowLab/system1-omni/pull/131). + +## Run + +Use the repository's development environment and a running service. Clone +[JevBench](https://github.com/fstandhartinger/jevbench) at +`7ce310c7262ed49cc85853339a8a42459298e3f3`; the tool verifies the content hashes of +all three datasets and every scorer module it imports. No dataset is vendored. + +```bash +export JEVBENCH_ROOT=/absolute/path/to/jevbench +python -m benchmarks.legs.leg1_public231 predict \ + --endpoint http://127.0.0.1:8000 --path /v1/systemone \ + --model open-jev-9b --thinking off --warmup 5 --rounds 3 \ + --run-meta /absolute/path/to/environment.json \ + --out /absolute/path/to/results/open-jev-9b +python -m benchmarks.legs.leg1_public231 aggregate \ + --run /absolute/path/to/results/open-jev-9b \ + --out /absolute/path/to/results/open-jev-9b +``` + +Omni native Open-Jev uses `/v1/systemone`; Intern's official service uses +`/v1/decisions`. `--model` is optional. The collector preserves state, questions and +option order, sends `thinking.enabled=false`, and never sends targets or provenance. +It makes one request at a time, without retries. It does not start services or choose GPUs. + +`environment.json` describes the **actual service**, without credentials: + +```json +{ + "checkpoint": {"repo": "owner/model", "revision": "full-checkpoint-commit"}, + "backend": {"name": "serving-backend", "version": "full-source-commit"}, + "sampling": {"temperature": 1.0}, + "hardware": {"gpu": "GPU model"} +} +``` + +Also record dtype, tokenizer/base/adapter/head revisions, loaded libraries, context +limits, cache settings and job identity. The file is copied byte-for-byte and hashed; +all original fields must match the run metadata. Omit `stop_reason` and conflicting +collector fields. Temperature is recorded, not configured by the client. Authenticate +outside this tool if needed; it does not accept or save API keys. + +## Reading the results + +- Each round's accuracy is **correct / 231**. Failures and unattempted questions stay + in the denominator. Three repeated rounds provide timing samples, not 693 unique questions. +- Upstream schema validity, Brier and ECE are reported separately. Missing probabilities + are not replaced with one-hot predictions. Native `noul` scalars become `{no: 1-p, yes: p}`. +- Per-round and pooled p50/p95 cover HTTP POST through full response receipt. All attempts, + successes and failures have separate counts; empty latency sets are null. Warmups are excluded. +- `sufficient=true` requires five warmups and three complete 231-question rounds. + `--limit`, fewer rounds or missing attempts produce partial observations. +- HTTP 401/403/429, interruption or three consecutive infrastructure failures stop collection. + HTTP 422 counts as a failed question and does not trigger that stop counter. + +Evidence goes in a fresh directory **outside the repository**. Each round contains +`results.jsonl` and `raw/.json`; warmups are separate. Raw records retain +the request, response text, decoded response, HTTP status, error, timestamp and latency. +`run.meta.json` and `completion.json` record the plan and any stop reason. + +Aggregation checks source hashes, the environment, input order, raw hashes and record +consistency before re-scoring. Orphan or missing raw files are errors. It never overwrites +an existing summary; use a new output directory for another recomputation. These checks +detect inconsistent evidence, not coordinated rewriting of an unsigned archive. + +## Tests + +```bash +JEVBENCH_ROOT=/absolute/path/to/jevbench python -m pytest tests/benchmarks -q --confcutdir=tests/benchmarks +ruff check benchmarks tests/benchmarks +ruff format --check benchmarks tests/benchmarks +``` + +Reference-dependent tests skip if `JEVBENCH_ROOT` is absent. Mock and loopback tests +check collection and scoring behavior; model quality and latency come from the linked real runs. + +## Offline quality from shared raw + +This consumer re-scores saved Omni requests without another model run. +The existing `predict` and `aggregate` commands and their +default timing basis are unchanged. + +For public231, provide the frozen input manifest, the independently frozen Omni +`bench.py` interface (`5aa20a2c…`), and one `--run` path per planned round in order: + +```bash +export JEVBENCH_ROOT=/absolute/path/to/pinned/jevbench +python -m benchmarks.quality_public231 \ + --input-manifest /absolute/path/to/input-r1/manifest.json \ + --collector /absolute/path/to/omni-raw-interface-v2/bench.py \ + --model-config-id open-9b \ + --run /absolute/path/to/round-1 --run /absolute/path/to/round-2 \ + --run /absolute/path/to/round-3 \ + --out /absolute/path/to/new-quality-output +``` + +Each raw directory must contain the complete frozen `requests.jsonl`, `config.json`, +`completion.json` and saved `responses.jsonl` when measurement started. The Omni +interface checks file/response hashes, IDs, decoding and its saved wire verdict. +The quality consumer checks the pinned inputs and request order, then uses the +original JevBench scorer on the raw HTTP/typed probabilities. An Omni +`invalid_response` can still be a valid relaxed JevBench distribution. Wire validity, +distribution validity, strict validity, renormalization and label correctness remain +separate. Choice/score hard labels are never substituted for missing probabilities; +noul scalars use `{no: 1-p, yes: p}`. Accuracy uses the original argmax rule, including +lexical tie breaking; ordinal expected value is used for MAE. + +Missing round directories and started requests without a saved terminal remain +in the plan. Three rounds have 231 unique tasks and 693 planned observations. +The ordered CLI paths define the round plan; the producer saves one directory per +round. A saved `metadata.round`, when present, must match that ordered round; +the remaining model/configuration metadata must agree between available rounds. +The resulting `summary.json` and full-plan `records.jsonl` carry original raw +pointers and hashes. Quality output uses null timestamps/latencies; it does not +report speed. The raw clock is POST through body decoding, JSON and schema validation, +and the saved response is decoded HTTPX text, not compressed wire bytes. + +Use a fresh output directory outside the repository. The tool checks archive +consistency; it does not authenticate a coordinated rewrite of unsigned evidence. +CPU fixtures exercise the reader and scorer, not model quality or GPU execution. + +The required benchmark CI job fetches the pinned JevBench reference and Omni +collector, generates the public231 fixture, and rejects skipped tests. To run the +same suite locally, set `JEVBENCH_ROOT` and `OMNI_BENCH_SOURCE` to those checkouts: + +```bash +python -m pytest tests/benchmarks -q --confcutdir=tests/benchmarks +``` + +The reader accepts the original collector (`5aa20a2c…`). Each archive must match +the exact collector supplied with `--collector`; the probability rules, input +bytes and timing basis are unchanged. diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..391ce6c --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Offline decision benchmark tools, separate from agent runtime.""" diff --git a/benchmarks/accuracy-leaderboard.md b/benchmarks/accuracy-leaderboard.md new file mode 100644 index 0000000..a473840 --- /dev/null +++ b/benchmarks/accuracy-leaderboard.md @@ -0,0 +1,73 @@ +# Public231 accuracy leaderboard + +This board ranks five deployed decision models on 231 public JevBench tasks: +easy48, original72 and hard111. It is a fixed evaluation group, not the article's +248-question leaderboard or a claim about general model quality. + +## H800 group: Slurm 415076, 2026-10-07 UTC + +Rank uses correct / 231 in the first measured round. All three rounds +produced identical predictions and probability distributions for each deployment. +The repeated rounds add timing samples, not unique quality tasks. Equal correct +counts share a rank; calibration does not break ties. + +| Rank | Deployment | Correct / planned | Accuracy | Validity | Brier | ECE | +|---:|---|---:|---:|---:|---:|---:| +| 1 | Intern-Decision-4B, official HF | 200 / 231 | 86.5801% | 100% | 0.189601 | 0.067214 | +| 2 | Open-Jev-27B-v1.1, Omni native | 198 / 231 | 85.7143% | 100% | 0.240493 | 0.130333 | +| 3 | Intern-Decision-2B, official HF | 181 / 231 | 78.3550% | 100% | 0.306049 | 0.115420 | +| 4 | Open-Jev-9B, Omni native | 179 / 231 | 77.4892% | 100% | 0.320752 | 0.086786 | +| 5 | Intern-Decision-0.8B, official HF | 164 / 231 | 70.9957% | 100% | 0.369906 | 0.111077 | + +Each deployment completed 693 / 693 measured requests without failed or invalid +responses. Validity, Brier and 10-bin top-label ECE use those 693 distributions; +their repetition does not increase the independent sample size. Brier and ECE +are secondary calibration measures, with lower values better. + +## Source and controls + +- Dataset and scorer: [JevBench 7ce310c7](https://github.com/fstandhartinger/jevbench/tree/7ce310c7262ed49cc85853339a8a42459298e3f3). + The collector checks the pinned dataset and imported scorer file hashes. +- Canonical dataset SHA-256 (`dataset_hash` of the pinned tasks, also recorded + as `canonical_dataset_sha256` in the numeric attachment's source manifest): + `dc3995d8ae1e2fc8e81ce38431add509eb8bb39b85aadfd0c7c32079382dde51`. +- One H800, BF16, concurrency 1, thinking off, no retries, fixed task and option + order; five client warmups per deployment followed by three measured traversals. +- Intern used its official HF service at T=1, with optional fast paths unavailable. + Open-Jev used native Rust/CUDA, merged LoRA/head and its checkpoint temperature; + graph and prefix cache were disabled. These are different service configurations. +- The [immutable run report](https://github.com/linear3735/system1-agents/blob/40bf742e2b84ad6157bf2990dd5e4bab73571702/benchmarks/results/public231-h800.md) + records exact checkpoint, backend and dependency revisions, all round results + and limitations. The companion [latency contribution](https://github.com/ThinkFlowLab/system1-omni/pull/131) + keeps request timing separate from quality. + +The [public summary bundle](https://github.com/user-attachments/files/33304077/public231-benchmark-output.zip) +contains this campaign's five aggregate summaries, run metadata and collector replay +receipt. It omits per-question predictions, probabilities and raw responses, so it +does not support independent numeric recomputation. The separate per-question +numeric audit attachment remains unpublished; full response replay also requires +the original raw responses. The public bundle is collector evidence from PR #52, +not a replay of this PR's `quality_public231.py` consumer. + +## Add a result + +Use the [collection and aggregation commands](README.md#run) against an existing +decision endpoint. Keep the dataset/scorer version, task order, precision and +sampling policy fixed; record exact model/backend revisions and any required +configuration differences. Labels and provenance must never enter model requests. + +Publish planned, attempted, valid, failed, incomplete and unattempted counts. +Failures and missing tasks stay in the planned accuracy denominator. Score the +original probabilities with the pinned scorer; do not replace them with one-hot +labels. Missing calibration inputs remain missing, rather than becoming zero error. + +Predeclare the primary round and repeat plan. This group uses the first measured +round for ranking; publish every round and any prediction disagreement. Never +choose the best round after seeing results. A partial run remains an unranked +observation with its coverage shown. + +Add a dated campaign with a linked report and auditable numeric evidence; preserve +prior campaigns. Changed datasets, scorer rules or sampling policies need a separate +group. Compare new deployments with a reference in the same campaign. Ticket-routing +completion, screenshot/VL tasks and other suites also need separate groups and +their own fixed inputs; they cannot inherit this text board's rank. diff --git a/benchmarks/client.py b/benchmarks/client.py new file mode 100644 index 0000000..38d6f00 --- /dev/null +++ b/benchmarks/client.py @@ -0,0 +1,68 @@ +"""One synchronous POST per decision request; no retries or model inference.""" + +import json +import math +import time +from urllib.parse import urlsplit + +import httpx + + +def decode(text): + def invalid_constant(value): + raise ValueError(f"Non-JSON numeric constant: {value}") + + def finite_float(value): + result = float(value) + if not math.isfinite(result): + raise ValueError(f"Non-finite JSON number: {value}") + return result + + return json.loads(text, parse_constant=invalid_constant, parse_float=finite_float) + + +class DecisionClient: + def __init__(self, endpoint, *, path, model=None, timeout_s=120, transport=None): + if not math.isfinite(timeout_s) or timeout_s <= 0: + raise ValueError("timeout_s must be finite and positive") + url = urlsplit(endpoint) + if ( + url.scheme not in ("http", "https") + or not url.netloc + or url.username + or url.password + or url.query + or url.fragment + ): + raise ValueError("Use an HTTP endpoint without credentials or query parameters") + if not path.startswith("/") or "?" in path or "#" in path: + raise ValueError("path must be an absolute API path without query parameters") + self.url, self.model = endpoint.rstrip("/") + path, model + self.http = httpx.Client(timeout=timeout_s, transport=transport, follow_redirects=False, trust_env=False) + + def request(self, body): + request = dict(body) + if self.model is not None: + request.setdefault("model", self.model) + request = decode(json.dumps(request, ensure_ascii=False, allow_nan=False)) + raw = dict(request=request, response_body=None, response=None, http_status=None, client_error=None) + raw["ts"] = time.time() + started = time.perf_counter() + try: + response = self.http.post(self.url, json=request) + raw.update(http_status=response.status_code, response_body=response.text) + except httpx.HTTPError as exc: + raw["client_error"] = f"{type(exc).__name__}: {exc}" + except KeyboardInterrupt: + raw["client_error"] = "KeyboardInterrupt" + finally: + raw["latency_s"] = time.perf_counter() - started + if raw["response_body"] is not None: + try: + raw["response"] = decode(raw["response_body"]) + except ValueError: + pass # The original body remains available for offline diagnosis. + return raw + + def close(self): + self.http.close() diff --git a/benchmarks/evidence.py b/benchmarks/evidence.py new file mode 100644 index 0000000..79174a6 --- /dev/null +++ b/benchmarks/evidence.py @@ -0,0 +1,174 @@ +"""Pinned upstream scoring and reproducible public231 evidence.""" + +import hashlib +import importlib +import json +import os +import sys +from pathlib import Path +from types import SimpleNamespace + +from benchmarks.client import decode + +PIN = "7ce310c7262ed49cc85853339a8a42459298e3f3" +HASHES = { + "jevbench/__init__.py": "64c28dc1534502c429a3b67eee79ab7909fab995b7d116c9b089f0f3bb8652f7", + "jevbench/tasks.py": "b9c2f8ba9a7301519656d79d47c31b8ab0542a5545498ce3b339d5e93b976812", + "jevbench/scoring.py": "6aa17d212ea20a3eeb6876149367842e8462c11dcb54d73df062fb406368331e", + "jevbench/summarize.py": "aa5f24da407d004cd80bad1cd957b6987aba152591084c096d1de51dd9775cbe", + "jevbench/metrics.py": "4038c8423ad1cf28601afb6babe777b71f08e379b9ca125b09b52c0ac7d60515", + "datasets/public/easy.jsonl": "231df3c2c8e88a1a8c137ebe85de96ba70fabd330849098ac7b3c52c70b7172b", + "datasets/public/original.jsonl": "5c2414edb3006b8bfcb70fda433f0f9ca015759433849f8d3104328a1f7c4180", + "datasets/public/hard.jsonl": "89e9e6becb33ed88c1de7d42dcc87531b2fb64cfaef4e1986faf7c37b3f80ebb", +} + + +def digest(data): + return hashlib.sha256(data).hexdigest() + + +def dump(value): + return json.dumps(value, ensure_ascii=False, allow_nan=False) + + +def write_json(path, value): + with Path(path).open("x", encoding="utf-8") as stream: + stream.write(dump(value) + "\n") + + +def outside_repo(path): + path = Path(path).resolve() + if path.is_relative_to(Path(__file__).resolve().parents[1]): + raise ValueError("Evidence must be outside the source repository") + return path + + +def load_reference(): + root = Path(os.environ["JEVBENCH_ROOT"]).resolve() + for name, expected in HASHES.items(): + if digest((root / name).read_bytes()) != expected: + raise ValueError(f"Pinned reference hash mismatch: {name}") + for name, module in tuple(sys.modules.items()): + if name == "jevbench" or name.startswith("jevbench."): + if not Path(module.__file__).resolve().is_relative_to(root): + raise ValueError("Another JevBench root is already imported") + sys.path.insert(0, str(root)) + upstream = SimpleNamespace( + **{name: importlib.import_module(f"jevbench.{name}") for name in ("tasks", "scoring", "summarize", "metrics")} + ) + items = [] + for tier in ("easy", "original", "hard"): + for line, text in enumerate( + (root / f"datasets/public/{tier}.jsonl").read_text(encoding="utf-8").split("\n")[:-1], 1 + ): + task = upstream.tasks.Task.from_dict(decode(text)) + identity = dict( + suite=f"jevbench-{tier}", + task_id=task.id, + source_file=f"{tier}.jsonl", + source_line=line, + family=task.family, + tier=tier, + split=task.split, + ) + items.append((identity, task)) + dataset = dict( + jevbench_pin=PIN, + root=str(root), + hashes=HASHES, + dataset_hash=upstream.tasks.dataset_hash([task for _, task in items]), + ) + return upstream, items, dataset + + +def body_for(task, model): + body = {"state": task.state, "questions": {"decision": task.question}, "thinking": {"enabled": False}} + if model is not None: + body["model"] = model + return body + + +def score_record( + upstream, + identity, + task, + raw, + model, + round_number, + kind, + raw_hash, + *, + latency_basis="client_single_request_wall_seconds", +): + status, error, probs, source = "ok", None, None, "unavailable" + response = raw["response"] + code = raw["http_status"] + if raw["client_error"] is not None: + status, error = "client_error", raw["client_error"] + elif code is None or not 200 <= code < 300: + status, error = ("refused" if code == 422 else "http_error"), raw["response_body"] + else: + answers = response.get("answers") if isinstance(response, dict) else None + answer = answers.get("decision") if isinstance(answers, dict) else None + if not isinstance(answer, dict) or answer.get("type") != task.question["type"]: + status, error = "parse_error", "Missing or mismatched typed decision answer" + else: + probs = answer.get("probabilities") + source = "response_probabilities" if probs is not None else "unavailable" + if probs is None and task.question["type"] == "noul": + value = answer.get("noul") + if isinstance(value, (float, int)) and not isinstance(value, bool): + probs, source = {"no": 1 - value, "yes": value}, "native_noul_scalar" + try: + scored = upstream.scoring.score_task(probs, task) + except OverflowError: + scored = upstream.scoring.score_task(None, task) + scored["error"] = "Probability exceeds the scorer's numeric range" + return dict( + identity, + round=round_number, + kind=kind, + ts=raw["ts"], + attempted=True, + ok=status == "ok", + status=status, + status_code=code, + error=error or scored.get("error"), + latency_s=raw["latency_s"], + latency_basis=latency_basis, + **{key: scored.get(key) for key in ("valid", "strict_valid", "renormalized", "correct", "predicted", "probs")}, + expected=task.expected, + candidate_count=len(task.labels), + probs_as_returned=probs, + probs_source=source, + score_source="jevbench_7ce310c7", + model_configured=model, + model=response.get("model") if isinstance(response, dict) else None, + usage=response.get("usage") if isinstance(response, dict) else None, + raw_sha256=raw_hash, + ) + + +def summarize(upstream, tasks, records): + # Upstream accuracy is attempted-only; keep it explicitly diagnostic. + projections = [dict(row, model=str(row["model"]) if row["model"] is not None else "") for row in records] + original = upstream.summarize.summarize(tasks, projections) + correct = sum(bool(row["correct"]) for row in records) + latency = { + name: upstream.metrics.latency_summary([row["latency_s"] for row in rows]) + for name, rows in ( + ("all_attempts", records), + ("success", [r for r in records if r["ok"]]), + ("failure", [r for r in records if not r["ok"]]), + ) + } + return dict( + planned=len(tasks), + attempted=len(records), + valid=sum(r["valid"] for r in records), + correct=correct, + accuracy=correct / len(tasks), + coverage=len(records) / len(tasks), + latency=latency, + upstream_attempted_metrics=original, + ) diff --git a/benchmarks/legs/__init__.py b/benchmarks/legs/__init__.py new file mode 100644 index 0000000..56d7447 --- /dev/null +++ b/benchmarks/legs/__init__.py @@ -0,0 +1 @@ +"""Benchmark collection and aggregation command lines.""" diff --git a/benchmarks/legs/leg1_public231.py b/benchmarks/legs/leg1_public231.py new file mode 100644 index 0000000..97e88cc --- /dev/null +++ b/benchmarks/legs/leg1_public231.py @@ -0,0 +1,250 @@ +"""Collect serial public231 rounds or validate and aggregate saved evidence.""" + +import argparse +import copy +from contextlib import closing +import math +import os +import sys +import time +from pathlib import Path + +from benchmarks.client import DecisionClient, decode +from benchmarks.evidence import ( + body_for, + digest, + dump, + load_reference, + outside_repo, + score_record, + summarize, + write_json, +) + + +def validate_environment(meta): + for group, fields in (("checkpoint", ("repo", "revision")), ("backend", ("name", "version"))): + if any(not meta.get(group, {}).get(field) for field in fields): + raise ValueError(f"run-meta requires actual {group}: {fields}") + value = meta.get("sampling", {}).get("temperature") + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0: + raise ValueError("run-meta requires actual positive sampling.temperature") + if not meta.get("hardware"): + raise ValueError("run-meta requires actual hardware") + + +def predict(args): + upstream, items, dataset = load_reference() + environment = args.run_meta.read_bytes() + meta = decode(environment) + validate_environment(meta) + if "stop_reason" in meta: + raise ValueError("stop_reason is collector-owned state; omit it from the supplied environment") + if args.rounds < 1 or not 0 <= args.warmup <= 231 or not 1 <= args.limit <= 231: + raise ValueError("Invalid rounds, warmup or limit") + output = outside_repo(args.out) + plan = dict( + warmup=args.warmup, + rounds=args.rounds, + tasks=231, + limit=args.limit, + order="dataset sequential", + identities=[identity for identity, _ in items], + ) + collector = dict( + run_id=output.name, + argv=sys.argv, + endpoint=args.endpoint, + path=args.path, + model_configured=args.model, + response_parse="typed", + thinking=False, + dataset=dataset, + plan=plan, + env_file=str(args.run_meta.resolve()), + env_sha256=digest(environment), + stop_reason=None, + ts=time.time(), + ) + if any(key in meta and meta[key] != value for key, value in collector.items()): + raise ValueError("Supplied environment conflicts with collector metadata") + meta.update(collector) + with closing(DecisionClient(args.endpoint, path=args.path, model=args.model, timeout_s=args.timeout)) as client: + output.mkdir(parents=True, exist_ok=False) + with (output / "environment.json").open("xb") as stream: + stream.write(environment) + write_json(output / "run.meta.json", meta) + stop, errors = None, 0 + try: + for number in range(args.rounds + 1): + kind = "warmup" if number == 0 else "measured" + directory = output / ("warmup" if number == 0 else f"round-{number}") + (directory / "raw").mkdir(parents=True) + selected = items[: args.warmup if number == 0 else args.limit] + write_json( + directory / "predict.meta.json", dict(round=number, planned=args.warmup if number == 0 else 231) + ) + with (directory / ("records.jsonl" if number == 0 else "results.jsonl")).open( + "x", encoding="utf-8" + ) as stream: + for identity, task in selected: + raw = client.request(body_for(task, args.model)) + raw_path = directory / "raw" / (digest(task.id.encode()) + ".json") + write_json(raw_path, raw) + record = score_record( + upstream, identity, task, raw, args.model, number, kind, digest(raw_path.read_bytes()) + ) + stream.write(dump(record) + "\n") + stream.flush() + os.fsync(stream.fileno()) + if raw["client_error"] == "KeyboardInterrupt": + stop = "interrupted" + break + errors = errors + 1 if not record["ok"] and record["status_code"] != 422 else 0 + if record["status_code"] in (401, 403, 429) or errors >= 3: + stop = "access/rate limit or three consecutive infrastructure errors" + break + if stop: + break + except KeyboardInterrupt: + stop = "interrupted" + except Exception as exc: + stop = f"{type(exc).__name__}: {exc}" + raise + finally: + write_json(output / "completion.json", dict(stop_reason=stop, ts=time.time())) + meta["stop_reason"] = stop + write_json(output / "run.meta.final.json", meta) + (output / "run.meta.final.json").replace(output / "run.meta.json") + return output + + +def read_round(directory, expected, upstream, model, number): + path = directory / ("records.jsonl" if number == 0 else "results.jsonl") + records = [] + if path.exists(): + with path.open(encoding="utf-8") as stream: + records = [decode(line) for line in stream] + if len(records) > len(expected): + raise ValueError("Too many or duplicate records") + expected_raw = {digest(task.id.encode()) + ".json" for _, task in expected[: len(records)]} + actual_raw = {path.name for path in (directory / "raw").glob("*")} + if expected_raw != actual_raw: + raise ValueError( + f"Raw/record mismatch in {directory}: missing={sorted(expected_raw - actual_raw)}, " + f"orphan={sorted(actual_raw - expected_raw)}" + ) + for row, (identity, task) in zip(records, expected): + raw_path = directory / "raw" / (digest(task.id.encode()) + ".json") + data = raw_path.read_bytes() + raw = decode(data) + if dump(raw["request"]) != dump(body_for(task, model)): + raise ValueError("Raw request differs from frozen task") + parsed = None + if raw["response_body"] is not None: + try: + parsed = decode(raw["response_body"]) + except ValueError: + pass + if raw["response"] != parsed: + raise ValueError("Decoded response differs from original body") + if ( + not isinstance(raw["latency_s"], (int, float)) + or not math.isfinite(raw["latency_s"]) + or raw["latency_s"] < 0 + ): + raise ValueError("Invalid measured latency") + recomputed = score_record( + upstream, identity, task, raw, model, number, "warmup" if number == 0 else "measured", digest(data) + ) + if row != recomputed: + raise ValueError("Record differs from raw evidence or planned identity/order") + return records + + +def aggregate(args): + upstream, items, dataset = load_reference() + run = outside_repo(args.run) + meta = decode((run / "run.meta.json").read_text(encoding="utf-8")) + validate_environment(meta) + environment = (run / "environment.json").read_bytes() + if digest(environment) != meta["env_sha256"]: + raise ValueError("Environment source hash mismatch") + env = decode(environment) + if any(key not in meta or meta[key] != value for key, value in env.items()): + raise ValueError("Run metadata differs from supplied environment") + plan = meta["plan"] + if ( + any(meta["dataset"][key] != dataset[key] for key in ("jevbench_pin", "hashes", "dataset_hash")) + or plan["identities"] != [identity for identity, _ in items] + or plan["tasks"] != 231 + ): + raise ValueError("Frozen plan/reference mismatch") + if plan["rounds"] < 1 or not 0 <= plan["warmup"] <= 231 or not 1 <= plan["limit"] <= 231: + raise ValueError("Invalid frozen run plan") + expected_dirs = {f"round-{i}" for i in range(1, plan["rounds"] + 1)} + if {p.name for p in run.glob("round-*")} - expected_dirs: + raise ValueError("Unknown round directory") + warmup = read_round(run / "warmup", items[: plan["warmup"]], upstream, meta["model_configured"], 0) + rounds, pooled_tasks, pooled_records = [], [], [] + for number in range(1, plan["rounds"] + 1): + records = read_round( + run / f"round-{number}", items[: plan["limit"]], upstream, meta["model_configured"], number + ) + tasks = [task for _, task in items] + rounds.append(dict(round=number, **summarize(upstream, tasks, records))) + # Give repeated tasks distinct IDs only for upstream duplicate checking. + for task in tasks: + clone = copy.copy(task) + clone.id = f"{number}:{task.id}" + clone.group = f"{number}:{task.group}" if task.group else None + pooled_tasks.append(clone) + pooled_records.extend(dict(row, task_id=f"{number}:{row['task_id']}") for row in records) + sufficient = ( + plan["rounds"] == 3 + and plan["warmup"] == 5 + and plan["limit"] == 231 + and len(warmup) == 5 + and all(r["attempted"] == 231 for r in rounds) + ) + result = dict( + sufficient=sufficient, + status="sufficient" if sufficient else "insufficient", + warmup_attempted=len(warmup), + rounds=rounds, + pooled=summarize(upstream, pooled_tasks, pooled_records), + distinct_tasks=231, + cost_usd=None, + cost_basis="unmetered is not free", + latency_success_basis="HTTP success with a typed decision structure; schema validity is separate", + ) + output = outside_repo(args.out) + output.mkdir(parents=True, exist_ok=True) + write_json(output / "summary.json", result) + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + collector = commands.add_parser("predict") + collector.add_argument("--endpoint", required=True) + collector.add_argument("--path", required=True, choices=("/v1/systemone", "/v1/decisions")) + collector.add_argument("--model") + collector.add_argument("--thinking", choices=("off",), default="off") + collector.add_argument("--warmup", type=int, default=5) + collector.add_argument("--rounds", type=int, default=3) + collector.add_argument("--limit", type=int, default=231) + collector.add_argument("--timeout", type=float, default=120) + collector.add_argument("--run-meta", type=Path, required=True) + collector.add_argument("--out", type=Path, required=True) + merger = commands.add_parser("aggregate") + merger.add_argument("--run", type=Path, required=True) + merger.add_argument("--out", type=Path, required=True) + args = parser.parse_args() + result = predict(args) if args.command == "predict" else aggregate(args) + print(result if isinstance(result, Path) else dump(result)) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/quality_public231.py b/benchmarks/quality_public231.py new file mode 100644 index 0000000..3a19387 --- /dev/null +++ b/benchmarks/quality_public231.py @@ -0,0 +1,218 @@ +"""Score the same frozen Omni raw offline with the pinned public231 scorer.""" + +import argparse +import copy +import importlib.util +from pathlib import Path + +from benchmarks.client import decode +from benchmarks.evidence import ( + body_for, + digest, + dump, + load_reference, + outside_repo, + score_record, + summarize, + write_json, +) + +COLLECTOR_SHA256 = "5aa20a2c612b7da4cd2f1c1442d37e77feb9d6b9ada4375b59a08f4cada63044" +RAW_CLOCK = "post_to_body_json_and_schema_validation" + + +def load_collector(path): + # Only this reviewed interface is executable; never import a file selected by raw metadata. + source_hash = digest(path.read_bytes()) + if source_hash != COLLECTOR_SHA256: + raise ValueError("Frozen Omni collector hash mismatch") + spec = importlib.util.spec_from_file_location("public231_omni_raw", path) + collector = importlib.util.module_from_spec(spec) + spec.loader.exec_module(collector) + collector.SOURCE_SHA256 = source_hash + return collector + + +def adapt_record(upstream, identity, task, saved, model, number, raw_file): + response = None + if saved["response"] is not None: + try: + response = decode(saved["response"]) + except ValueError: + pass + # invalid_response is the Omni wire verdict, not the pinned scorer's probability verdict. + client_error = saved["error"] if saved["error_kind"] in ("timeout", "transport", "interrupted") else None + raw = dict( + response=response, + response_body=saved["response"], + http_status=saved["status"], + client_error=client_error, + ts=None, + latency_s=None, + ) + row = score_record( + upstream, identity, task, raw, model, number, "measured", saved["response_sha256"], latency_basis=None + ) + row.update( + wire_valid=saved["error_kind"] is None, + wire_error_kind=saved["error_kind"], + raw_record_state=saved["state"], + raw_file=raw_file, + raw_latency_basis=RAW_CLOCK, + ) + return row + + +def verify_inputs(manifest_path, items, dataset): + manifest = decode(manifest_path.read_text(encoding="utf-8")) + path = manifest_path.parent / manifest["output"]["path"] + if path.resolve() != manifest_path.with_name("public231.jsonl").resolve(): + raise ValueError("Frozen input path must be public231.jsonl beside its manifest") + data = path.read_bytes() + if digest(data) != manifest["output"]["sha256"]: + raise ValueError("Frozen input hash mismatch") + if ( + manifest["sources"]["jevbench_pin"] != dataset["jevbench_pin"] + or manifest["sources"]["reference_dataset_hash"] != dataset["dataset_hash"] + ): + raise ValueError("Frozen input reference mismatch") + expected = [] + for identity, task in items: + target = task.expected == "yes" if task.question["type"] == "noul" else task.expected + expected.append( + dict( + id=task.id, + request=body_for(task, None), + expected={"decision": target}, + source={**identity, "jevbench_pin": dataset["jevbench_pin"]}, + labels=task.labels, + provenance=task.provenance, + group=task.group, + reference_expected=task.expected, + ) + ) + if data != "".join(dump(row) + "\n" for row in expected).encode("utf-8"): + raise ValueError("Input order, requests or evaluator fields differ from the pinned dataset") + return data + + +def aggregate(runs, input_manifest, collector, model_config_id): + if not runs or len({path.resolve() for path in runs}) != len(runs) or not model_config_id.strip(): + raise ValueError("Provide unique planned round paths and a model_config_id") + upstream, items, dataset = load_reference() + frozen = verify_inputs(input_manifest, items, dataset) + tasks = [task for _, task in items] + rounds, all_rows, pooled_tasks, attempted_pooled = [], [], [], [] + binding = None + for number, run in enumerate(runs, 1): + records, source, config, started_ids = {}, None, None, set() + if run.exists(): + if (run / "requests.jsonl").read_bytes() != frozen: + raise ValueError("Omni requests differ from the complete frozen 231 plan") + config = decode((run / "config.json").read_text(encoding="utf-8")) + if config["runner_sha256"] != collector.SOURCE_SHA256 or config["latency_basis"] != RAW_CLOCK: + raise ValueError("Omni runner or clock differs from the frozen interface") + identity = {key: value for key, value in config.items() if key not in ("python", "httpx")} + metadata = dict(config.get("metadata", {})) + recorded_round = metadata.pop("round", None) + if "round" in config.get("metadata", {}) and (type(recorded_round) is not int or recorded_round != number): + raise ValueError("Recorded round differs from the ordered planned round") + identity["metadata"] = metadata + if binding is not None and identity != binding: + raise ValueError("Model configuration differs between planned rounds") + binding = identity + collector.summarize_saved(run) # raw/config/response hashes, typed decode and complete plan associations + completion = decode((run / "completion.json").read_text(encoding="utf-8")) + started_ids = set(completion["attempted_ids"]) + source = dict( + run=str(run.resolve()), + config_sha256=completion["config_sha256"], + manifest_sha256=completion["manifest_sha256"], + responses_sha256=completion["responses_sha256"], + stop_reason=completion["stop_reason"], + phase=config["phase"], + recorded_round=recorded_round, + ) + if (run / "responses.jsonl").exists(): + records = { + row["id"]: row + for row in ( + decode(line) for line in (run / "responses.jsonl").read_text(encoding="utf-8").splitlines() + ) + } + rows, attempted = [], [] + for identity, task in items: + if task.id in records: + row = adapt_record( + upstream, identity, task, records[task.id], config["model"], number, str(run / "responses.jsonl") + ) + attempted.append(row) + else: + started = task.id in started_ids + row = dict( + identity, + round=number, + attempted=started, + ok=False, + valid=False, + strict_valid=False, + renormalized=False, + correct=False, + predicted=None, + probs=None, + status="started_without_saved_terminal" if started else "unexecuted", + expected=task.expected, + model=None, + model_configured=config["model"] if config else None, + probs_source="unavailable", + raw_file=None, + raw_sha256=None, + latency_s=None, + latency_basis=None, + ts=None, + ) + if started: + attempted.append(row) + rows.append(dict(row, model_config_id=model_config_id, planned_run=str(run.resolve()))) + clone = copy.copy(task) + clone.id = f"{number}:{task.id}" + clone.group = f"{number}:{task.group}" if task.group else None + pooled_tasks.append(clone) + all_rows.extend(rows) + attempted_pooled.extend(dict(row, task_id=f"{number}:{row['task_id']}") for row in attempted) + rounds.append(dict(round=number, source=source, **summarize(upstream, tasks, attempted))) + result = dict( + schema="public231-omni-offline-quality-v1", + model_config_id=model_config_id, + distinct_tasks=231, + rounds=rounds, + pooled=summarize(upstream, pooled_tasks, attempted_pooled), + dataset=dataset, + input_manifest_sha256=digest(input_manifest.read_bytes()), + model_binding=binding, + identity_basis="saved metadata; build ancestry not independently proven", + latency_reported=False, + raw_latency_basis=RAW_CLOCK, + evidence_complete=all(row["source"] and row["source"]["stop_reason"] == "completed" for row in rounds), + ) + return result, all_rows + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run", type=Path, action="append", required=True, help="one path for each planned round") + parser.add_argument("--input-manifest", type=Path, required=True) + parser.add_argument("--collector", type=Path, required=True, help="frozen Omni bench.py; only imported offline") + parser.add_argument("--model-config-id", required=True) + parser.add_argument("--out", type=Path, required=True) + args = parser.parse_args() + result, records = aggregate(args.run, args.input_manifest, load_collector(args.collector), args.model_config_id) + output = outside_repo(args.out) + output.mkdir(parents=True, exist_ok=False) + write_json(output / "summary.json", result) + (output / "records.jsonl").write_text("".join(dump(row) + "\n" for row in records), encoding="utf-8") + print(dump(result)) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/results/public231-h800.md b/benchmarks/results/public231-h800.md new file mode 100644 index 0000000..c08a993 --- /dev/null +++ b/benchmarks/results/public231-h800.md @@ -0,0 +1,92 @@ +# Public231 on one H800 + +Five open checkpoints, one NVIDIA H800, one HTTP request at a time. Each service +received five warmups followed by the same 231 questions in the same order for three +rounds. Latency covers the complete HTTP request and response. No profiler was active. + +Open-Jev used System1-Omni's native Rust/CUDA service; Intern-Decision used its +upstream HF service. These results compare deployed configurations. Their ratios +include different architectures, templates and execution paths, not just model size +or kernel speed. + +## Results + +| Model | Service | Correct / 231 | Accuracy | Validity | Brier | ECE | p50 / p95 ms | +|---|---|---:|---:|---:|---:|---:|---:| +| Intern-Decision-0.8B | HF / T=1 | 164/231 | 70.9957% | 100% | 0.369906 | 0.111077 | 80.493 / 175.237 | +| Intern-Decision-2B | HF / T=1 | 181/231 | 78.3550% | 100% | 0.306049 | 0.115420 | 79.054 / 175.276 | +| Intern-Decision-4B | HF / T=1 | 200/231 | 86.5801% | 100% | 0.189601 | 0.067214 | 104.552 / 235.291 | +| Open-Jev-9B | Omni native | 179/231 | 77.4892% | 100% | 0.320752 | 0.086786 | 42.688 / 327.057 | +| Open-Jev-27B-v1.1 | Omni native | 198/231 | 85.7143% | 100% | 0.240493 | 0.130333 | 68.877 / 1068.817 | + +All five configurations completed 693/693 measured requests, with no failed or invalid +responses. Predictions and probabilities matched across the three rounds. Accuracy +therefore has 231 unique questions; the repeated rounds are timing repetitions. +Brier/ECE use all 693 valid distributions. Failure latency has n=0 and null quantiles. + +Open-Jev 9B had the lowest median latency. Intern 4B answered 200/231 correctly versus +Open-Jev 27B's 198/231, and had a lower p95; Open-Jev 27B had a lower median. This small +set does not establish a general quality ranking or a uniform 2–3× speed advantage. + +## Per-round latency + +| Model | Round 1 p50 / p95 ms | Round 2 p50 / p95 ms | Round 3 p50 / p95 ms | +|---|---:|---:|---:| +| Intern-Decision-0.8B | 84.534 / 180.725 | 73.301 / 173.430 | 73.485 / 172.817 | +| Intern-Decision-2B | 81.454 / 180.338 | 74.251 / 173.638 | 74.154 / 173.411 | +| Intern-Decision-4B | 107.711 / 240.246 | 98.628 / 232.829 | 98.065 / 232.709 | +| Open-Jev-9B | 43.166 / 317.867 | 42.624 / 317.644 | 42.525 / 317.579 | +| Open-Jev-27B-v1.1 | 68.925 / 1030.605 | 68.892 / 1034.555 | 68.821 / 1033.953 | + +Pooled quantiles above are computed from all raw timings, not from these three quantiles. + +## Configuration + +- BF16; client concurrency 1; no retries; thinking disabled. +- Intern: official source `3572c8a68b5df5dafe02d0e093989ba8ec0183bc`, + PyTorch 2.9.1+cu128, torchvision 0.24.1+cu128, Transformers 5.14.1, SDPA, + temperature 1, no calibration artifact, no KV cache. Optional FLA/causal-conv1d + fast paths were unavailable, so the service used the upstream Torch fallback. +- Open-Jev: Omni `4a79980d8a75fd063cb3f8247e06215e288b18ac`, CUDA 13, + merged LoRA and decision head, Graph and prefix cache disabled. 9B evaluates + candidates sequentially; 27B packs candidates. A single service request is not + necessarily one candidate sequence. +- Context limits: Intern 8192 tokens for the complete chat; Open-Jev 16384 per + candidate. Native usage counts tokens summed over candidates, so the two services' + token counts are not directly comparable. + +| Checkpoint | Revision | Temperature | Spawn-to-ready s | +|---|---|---:|---:| +| [Intern 0.8B](https://huggingface.co/internlm/Intern-Decision-0.8B) | `85a0cc5a99d67ea8d56dfe98115689212867171d` | 1 | 16.084 | +| [Intern 2B](https://huggingface.co/internlm/Intern-Decision-2B) | `8797836c65fc91a2435b1fb6850b5f0aabd75cc3` | 1 | 14.045 | +| [Intern 4B](https://huggingface.co/internlm/Intern-Decision-4B) | `0e5e6aa7d6d750e2b1504ba11a8136cb58aeb3cd` | 1 | 16.048 | +| [Open-Jev 9B](https://huggingface.co/ZefanCai/Open-Jev-9B) | `47e966881e489511c0c7f5633a9e1960a676a551` | 1.8969118766 | 8.015 | +| [Open-Jev 27B v1.1](https://huggingface.co/ZefanCai/Open-Jev-27B-v1.1) | `28cf73067d5b337860bbef3c85b8b82ba8730956` | 2.5343690298 | 20.029 | + +Cold startup is excluded from request latency and includes native internal warmup. +Open 9B's base is `c202236235762e1c871ad0ccb60c8ee5ba337b9a`; Open 27B's base is +`1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0`. + +## Inputs and verification + +[JevBench](https://github.com/fstandhartinger/jevbench/tree/7ce310c7262ed49cc85853339a8a42459298e3f3) +provides easy48, original72 and hard111 and the scorer. Source and data hashes are +pinned in `benchmarks/evidence.py`. The article's complete 248-question set was not +available; this is not a reproduction of its 23-model leaderboard. + +The run used Agents base `9790b52ef4d78a2215e8dd5a526d3a8daea9875a`, dated +2026-10-07 UTC, Slurm job 415076. All 3490 raw requests, including warmups, were +checked for source identity, order, response hash and absence of target leakage. +A non-author independently re-scored the saved responses with the pinned scorer +and recomputed latency from raw samples. The final collector's offline aggregation +matched counts exactly and floating-point metrics within 1e-12. + +Collection preceded fixes to metadata validation, URL validation before file writes, +and JSONL physical-line reading. These fixes preserve the HTTP and scoring paths. +Verification replays the saved responses; it is not a second model run. + +Actual loaded-library maps and the weight hashes were recorded. The native binary +SHA-256 was `a0694633308ea0ad0ca2a302bfb16cbb73fd3367cd36cd142596761a71d23afc`; +CUDA library SHA-256 was `a0db4905df4fe2fb5d006acaaa09b5b51dbd3933cc8e9b20ec8114f23def8b20`. +Use the [collection and replay commands](../README.md#run) with the fixed inputs and +record the actual serving environment. Large raw traces and weights stay outside Git. diff --git a/tests/benchmarks/test_public231.py b/tests/benchmarks/test_public231.py new file mode 100644 index 0000000..16da0e4 --- /dev/null +++ b/tests/benchmarks/test_public231.py @@ -0,0 +1,569 @@ +"""CPU protocol fixtures only; no model service or network connection.""" + +import copy +import json +import os +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import httpx +import pytest + +from benchmarks.client import DecisionClient +from benchmarks.evidence import body_for, digest, dump, load_reference, score_record, summarize +from benchmarks.legs import leg1_public231 as cli + + +@pytest.fixture +def reference(): + if not os.environ.get("JEVBENCH_ROOT"): + pytest.skip("Set JEVBENCH_ROOT to the pinned external reference") + return load_reference() + + +def test_multifield_single_post_preserves_input(): + body = { + "state": [{"context": "structured"}], + "questions": { + "second": {"type": "choice", "criteria": {"z": "last", "a": "first"}}, + "first": {"type": "noul", "instructions": "accept?"}, + }, + "thinking": {"enabled": False}, + } + before, calls = copy.deepcopy(body), [] + + def handler(request): + calls.append(json.loads(request.content)) + assert str(request.url) == "http://fixture.invalid/v1/decisions" + return httpx.Response(200, text=' {"answers": {"second": {}, "first": {}}, "model":"fixture"}\n') + + client = DecisionClient("http://fixture.invalid", path="/v1/decisions", transport=httpx.MockTransport(handler)) + raw = client.request(body) + client.close() + assert body == before and calls == [body] + assert list(calls[0]["questions"]) == ["second", "first"] + assert list(calls[0]["questions"]["second"]["criteria"]) == ["z", "a"] + assert raw["response_body"].startswith(" ") and raw["response_body"].endswith("\n") + assert set(raw["response"]["answers"]) == {"second", "first"} + assert raw["http_status"] == 200 and raw["latency_s"] > 0 + + +@pytest.mark.parametrize("status,text", [(422, "too long"), (500, "failure"), (200, "broken json")]) +def test_raw_bad_responses_survive(status, text): + client = DecisionClient( + "http://fixture.invalid", + path="/v1/systemone", + transport=httpx.MockTransport(lambda _: httpx.Response(status, text=text)), + ) + raw = client.request({"questions": {}}) + client.close() + assert (raw["http_status"], raw["response_body"], raw["response"]) == (status, text, None) + assert raw["latency_s"] > 0 + + +@pytest.mark.parametrize("error", [httpx.ReadTimeout("fixture timeout"), KeyboardInterrupt()]) +def test_network_error_has_real_elapsed(error): + def handler(_): + raise error + + client = DecisionClient("http://fixture.invalid", path="/v1/systemone", transport=httpx.MockTransport(handler)) + raw = client.request({"state": "fixture"}) + client.close() + assert raw["http_status"] is None and raw["response_body"] is None + assert raw["client_error"] and raw["latency_s"] > 0 + + +@pytest.mark.parametrize("url", ["http://user:secret@fixture.invalid", "http://fixture.invalid?key=x"]) +def test_credentials_cannot_enter_endpoint_evidence(url): + with pytest.raises(ValueError): + DecisionClient(url, path="/v1/systemone") + + +def raw_answer(answer, *, status=200, seconds=1): + response = {"answers": {"decision": answer}, "model": "fixture", "usage": {"decision_count": 1}} + return dict( + request={}, + response=response, + response_body=dump(response), + http_status=status, + client_error=None, + latency_s=seconds, + ts=123.0, + ) + + +@pytest.mark.parametrize( + "probs,valid,renormalized", + [ + ({"no": 0.4, "yes": 0.595}, True, True), + ({"no": 0.4, "yes": 0.5}, False, False), + ({"no": 0.4}, False, False), + ({"no": 0.4, "yes": 0.6, "extra": 0}, False, False), + ({"no": -0.1, "yes": 1.1}, False, False), + ], +) +def test_upstream_alone_decides_probability_band(reference, probs, valid, renormalized): + upstream, items, _ = reference + identity, task = next(item for item in items if item[1].question["type"] == "noul") + row = score_record( + upstream, + identity, + task, + raw_answer({"type": "noul", "probabilities": probs}), + "configured", + 1, + "measured", + "fixture-hash", + ) + assert row["ok"] and row["valid"] is valid and row["renormalized"] is renormalized + assert row["probs_as_returned"] == probs + if renormalized: + assert row["strict_valid"] is False and sum(row["probs"].values()) == pytest.approx(1) + if not valid: + assert row["correct"] is False and row["probs"] is None + + +def test_native_scalar_and_missing_probabilities(reference): + upstream, items, _ = reference + for kind in ("noul", "choice", "score"): + identity, task = next(item for item in items if item[1].question["type"] == kind) + answer = {"type": kind, "noul": 0.6} if kind == "noul" else {"type": kind, kind: task.expected} + row = score_record(upstream, identity, task, raw_answer(answer), None, 1, "measured", "h") + assert row["valid"] is (kind == "noul") + assert row["model_configured"] is None and row["model"] == "fixture" + if kind == "noul": + assert row["probs"] == {"no": 0.4, "yes": 0.6} + else: + assert row["probs"] is None and row["probs_as_returned"] is None + + +@pytest.mark.parametrize("answers", [None, [], "bad", {"decision": None}, {"decision": {"type": "wrong"}}]) +def test_malformed_typed_structure_is_recorded(reference, answers): + upstream, items, _ = reference + identity, task = items[0] + raw = raw_answer({}) + raw["response"]["answers"] = answers + row = score_record(upstream, identity, task, raw, None, 1, "measured", "h") + assert row["status"] == "parse_error" and not row["ok"] and not row["valid"] + + +def test_nonfinite_response_is_not_json(): + client = DecisionClient( + "http://fixture.invalid", + path="/v1/systemone", + transport=httpx.MockTransport( + lambda _: httpx.Response( + 200, text='{"answers":{"decision":{"type":"noul","probabilities":{"no":NaN,"yes":1}}}}' + ) + ), + ) + raw = client.request({}) + client.close() + assert raw["response"] is None and "NaN" in raw["response_body"] + + +@pytest.fixture +def collect(reference, monkeypatch, tmp_path): + upstream, items, _ = reference + env = tmp_path / "actual-environment.json" + env.write_text( + dump( + dict( + checkpoint={"repo": "CPU-fixture", "revision": "fixture-only"}, + backend={"name": "httpx.MockTransport", "version": httpx.__version__}, + sampling={"temperature": 1.0}, + hardware="CPU protocol fixture, no inference", + dtype="fixture-original", + tokenizer={"revision": "fixture-tokenizer"}, + base={"revision": "fixture-base"}, + adapter={"revision": "fixture-adapter"}, + head={"revision": "fixture-head"}, + context_limit=16384, + cache={"enabled": False}, + job_id="CPU-fixture", + ) + ) + ) + lookup = {dump(body_for(task, "configured")): task for _, task in items} + + def run(*, warmup=5, rounds=3, limit=231, statuses=None, response_text=None): + calls = [] + + def handler(request): + body = json.loads(request.content) + assert "targets" not in body and "provenance" not in body and "expected" not in body + assert body["thinking"] == {"enabled": False} + task = lookup[dump(body)] + calls.append(body) + status = statuses[min(len(calls) - 1, len(statuses) - 1)] if statuses else 200 + if response_text is not None: + return httpx.Response(status, text=response_text) + answer = { + "type": task.question["type"], + "probabilities": {label: float(label == str(task.expected)) for label in task.labels}, + } + return httpx.Response(status, json={"answers": {"decision": answer}, "model": "fixture-response"}) + + class FixtureClient(DecisionClient): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs, transport=httpx.MockTransport(handler)) + + def request(self, body): + raw = super().request(body) + raw["latency_s"] = 999 if len(calls) <= warmup else (1 if raw["http_status"] == 200 else 120) + return raw + + monkeypatch.setattr(cli, "DecisionClient", FixtureClient) + output = tmp_path / f"run-{len(list(tmp_path.glob('run-*')))}" + args = SimpleNamespace( + endpoint="http://fixture.invalid", + path="/v1/decisions", + model="configured", + timeout=10, + run_meta=env, + out=output, + warmup=warmup, + rounds=rounds, + limit=limit, + ) + cli.predict(args) + return output, calls, args + + return run + + +def aggregate(run, out=None): + return cli.aggregate(SimpleNamespace(run=run, out=out or run)) + + +@pytest.mark.parametrize( + "endpoint", ["http://user:DUMMY_SECRET@fixture.invalid", "http://fixture.invalid?key=DUMMY_SECRET"] +) +def test_predict_cli_rejects_credentials_before_output(collect, monkeypatch, endpoint): + run, calls, args = collect(warmup=0, rounds=1, limit=1) + args.out = run.parent / "rejected-endpoint" + monkeypatch.setattr( + "sys.argv", + [ + "public231", + "predict", + "--endpoint", + endpoint, + "--path", + args.path, + "--run-meta", + str(args.run_meta), + "--out", + str(args.out), + ], + ) + with pytest.raises(ValueError, match="without credentials"): + cli.main() + assert not args.out.exists() and len(calls) == 1 + assert all("DUMMY_SECRET" not in path.read_text() for path in run.parent.rglob("*") if path.is_file()) + + +@pytest.mark.parametrize("failure", ["mkdir", "environment.json", "run.meta.json"]) +def test_predict_closes_client_when_output_setup_fails(collect, monkeypatch, failure): + run, calls, args = collect(warmup=0, rounds=1, limit=1) + args.out = run.parent / "failed-output" + created, original_client, original_open = [], cli.DecisionClient, Path.open + + def client(*args, **kwargs): + result = original_client(*args, **kwargs) + created.append(result) + return result + + def open_file(path, *options, **kwargs): + if path == args.out / failure: + raise OSError("fixture output failure") + return original_open(path, *options, **kwargs) + + if failure == "mkdir": + args.out.mkdir() + monkeypatch.setattr(cli, "DecisionClient", client) + monkeypatch.setattr(Path, "open", open_file) + with pytest.raises(OSError): + cli.predict(args) + assert len(created) == 1 and created[0].http.is_closed + assert len(calls) == 1 + + +@pytest.mark.parametrize("separator", ["\u2028", "\u0085", "\u2029"]) +def test_collect_and_aggregate_preserve_unicode_in_error_body(collect, separator): + run, calls, _ = collect( + warmup=0, rounds=1, limit=1, statuses=[422], response_text=dump({"detail": f"bad{separator}input"}) + ) + with (run / "round-1/results.jsonl").open() as stream: + records = [json.loads(line) for line in stream] + assert len(calls) == len(records) == 1 and records[0]["status"] == "refused" + assert separator in records[0]["error"] + result = aggregate(run) + assert result["pooled"]["planned"] == 231 and result["pooled"]["attempted"] == 1 + assert result["pooled"]["correct"] == result["pooled"]["accuracy"] == 0 + assert not result["sufficient"] + + +@pytest.mark.parametrize("line", ["\n", "{broken}\n"]) +def test_aggregate_rejects_blank_or_malformed_physical_line(collect, line): + run, _, _ = collect(warmup=0, rounds=1, limit=1) + (run / "round-1/results.jsonl").write_text(line) + with pytest.raises(ValueError): + aggregate(run) + + +def test_complete_5_plus_3_rounds_and_no_warmup_in_latency(collect): + run, calls, _ = collect() + result = aggregate(run) + assert len(calls) == 698 and result["sufficient"] + assert result["pooled"]["planned"] == result["pooled"]["correct"] == 693 + assert result["pooled"]["latency"]["all_attempts"] == {"n": 693, "p50_s": 1, "p95_s": 1} + assert result["pooled"]["latency"]["failure"] == {"n": 0, "p50_s": None, "p95_s": None} + assert all(round_["attempted"] == 231 for round_ in result["rounds"]) + assert len(list((run / "warmup/raw").glob("*.json"))) == 5 + second = aggregate(run, run / "offline-recompute") + assert result == second + with pytest.raises(FileExistsError): + aggregate(run) + + +@pytest.mark.parametrize("options", [{"limit": 10}, {"warmup": 0}, {"rounds": 1}]) +def test_debug_run_is_insufficient_with_fixed_denominator(collect, options): + run, _, _ = collect(**options) + result = aggregate(run) + assert not result["sufficient"] + if "limit" in options: + assert result["pooled"]["accuracy"] == 30 / 693 + assert result["rounds"][0]["accuracy"] == 10 / 231 + + +@pytest.mark.parametrize( + "statuses,attempted", [([401], 1), ([403], 1), ([429], 1), ([500], 3), ([500, 500, 422, 500, 500, 500], 6)] +) +def test_stop_policy_leaves_unattempted_in_plan(collect, statuses, attempted): + run, calls, _ = collect(warmup=0, statuses=statuses) + result = aggregate(run) + assert len(calls) == attempted + assert result["pooled"]["accuracy"] == 0 and result["pooled"]["planned"] == 693 + assert result["pooled"]["attempted"] == attempted and not result["sufficient"] + assert json.loads((run / "run.meta.json").read_text())["stop_reason"] + + +def test_failures_and_partial_denominator_with_two_latencies(collect): + run, _, _ = collect(warmup=0, limit=2, rounds=1, statuses=[200, 500]) + result = aggregate(run)["rounds"][0] + assert result["accuracy"] == 1 / 231 + assert result["latency"]["success"] == {"n": 1, "p50_s": 1, "p95_s": 1} + assert result["latency"]["failure"] == {"n": 1, "p50_s": 120, "p95_s": 120} + assert result["latency"]["all_attempts"]["p50_s"] == 60.5 + assert result["latency"]["all_attempts"]["p95_s"] == pytest.approx(114.05) + + +@pytest.mark.parametrize("change", ["duplicate", "unknown", "score", "raw", "environment", "order"]) +def test_offline_tampering_rejected(collect, change): + run, _, _ = collect(warmup=0, rounds=1, limit=2) + path = run / "round-1/results.jsonl" + rows = [json.loads(line) for line in path.read_text().splitlines()] + if change == "duplicate": + rows[1] = rows[0] + elif change == "unknown": + rows[0]["task_id"] = "unknown" + elif change == "score": + rows[0]["correct"] = False + elif change == "order": + rows.reverse() + elif change == "raw": + raw = run / "round-1/raw" / (digest(rows[0]["task_id"].encode()) + ".json") + raw.write_text(raw.read_text() + " ") + else: + path_env = run / "environment.json" + path_env.write_text(path_env.read_text() + " ") + path.write_text("".join(dump(row) + "\n" for row in rows)) + with pytest.raises(ValueError): + aggregate(run) + + +def test_output_and_environment_boundaries(collect, tmp_path): + run, _, args = collect(warmup=0, limit=1, rounds=1) + with pytest.raises(FileExistsError): + cli.predict(args) + args.out = Path(__file__).resolve().parents[2] / "forbidden-evidence" + with pytest.raises(ValueError, match="outside"): + cli.predict(args) + env = json.loads(args.run_meta.read_text()) + env["sampling"]["temperature"] = 0 + args.run_meta.write_text(dump(env)) + args.out = tmp_path / "invalid-temperature" + with pytest.raises(ValueError, match="temperature"): + cli.predict(args) + assert not args.out.exists() + + +def test_reference_hash_tampering_rejected(reference, monkeypatch, tmp_path): + _, _, dataset = reference + from benchmarks.evidence import HASHES + + root = tmp_path / "reference" + for name in HASHES: + destination = root / name + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes((Path(dataset["root"]) / name).read_bytes()) + (root / "jevbench/scoring.py").write_text("raise RuntimeError('must not execute')\n") + monkeypatch.setenv("JEVBENCH_ROOT", str(root)) + with pytest.raises(ValueError, match="hash mismatch"): + load_reference() + + +def test_empty_summary_has_null_latency(reference): + upstream, items, _ = reference + result = summarize(upstream, [task for _, task in items], []) + assert result["accuracy"] == 0 and result["coverage"] == 0 + assert all(value == {"n": 0, "p50_s": None, "p95_s": None} for value in result["latency"].values()) + + +@pytest.mark.parametrize("literal", ["1e309", "NaN", "Infinity"]) +def test_nonfinite_wire_numbers_are_retained_as_parse_failures(literal): + response = '{"answers":{"decision":{"type":"noul","noul":' + literal + "}}}" + client = DecisionClient( + "http://fixture.invalid", + path="/v1/systemone", + transport=httpx.MockTransport(lambda _: httpx.Response(200, text=response)), + ) + raw = client.request({}) + client.close() + assert raw["response"] is None and raw["response_body"] == response + + +def test_scorer_numeric_overflow_counts_wrong(reference): + upstream, items, _ = reference + identity, task = next(item for item in items if item[1].question["type"] == "noul") + probs = {"no": 10**400, "yes": 0} + row = score_record( + upstream, identity, task, raw_answer({"type": "noul", "probabilities": probs}), "configured", 1, "measured", "h" + ) + assert row["ok"] and not row["valid"] and not row["correct"] + assert row["probs_as_returned"] == probs + + +@pytest.mark.parametrize("directory,filename", [("warmup", "records.jsonl"), ("round-1", "results.jsonl")]) +@pytest.mark.parametrize("change", ["missing_row", "missing_records_file", "missing_raw", "unknown_raw"]) +def test_raw_and_records_must_correspond(collect, directory, filename, change): + run, _, _ = collect(warmup=2, rounds=1, limit=2) + directory = run / directory + path = directory / filename + rows = path.read_text().splitlines() + row = json.loads(rows[-1]) + raw = directory / "raw" / (digest(row["task_id"].encode()) + ".json") + changed = raw + if change == "missing_row": + path.write_text(rows[0] + "\n") + elif change == "missing_records_file": + path.unlink() + elif change == "missing_raw": + raw.unlink() + else: + changed = directory / "raw/unknown.json" + changed.write_bytes(raw.read_bytes()) + with pytest.raises(ValueError, match="Raw/record mismatch") as error: + aggregate(run) + assert changed.name in str(error.value) + assert not (run / "summary.json").exists() + + +@pytest.mark.parametrize("field", ["dtype", "tokenizer", "base", "adapter", "head", "context_limit", "cache", "job_id"]) +@pytest.mark.parametrize("change", ["different", "missing"]) +def test_all_environment_fields_must_match(collect, field, change): + run, _, _ = collect(warmup=0, rounds=1, limit=1) + path = run / "run.meta.json" + meta = json.loads(path.read_text()) + if change == "different": + meta[field] = "fixture-contradiction" + else: + del meta[field] + path.write_text(dump(meta)) + with pytest.raises(ValueError, match="supplied environment"): + aggregate(run) + assert not (run / "summary.json").exists() + + +def test_conflicting_environment_controls_rejected_before_collection(collect): + run, calls, args = collect(warmup=0, rounds=1, limit=1) + env = json.loads(args.run_meta.read_text()) + env["endpoint"] = "http://conflicting.invalid" + args.run_meta.write_text(dump(env)) + args.out = run.parent / "conflicting-run" + with pytest.raises(ValueError, match="collector"): + cli.predict(args) + assert len(calls) == 1 and not args.out.exists() + + +def test_matching_environment_controls_are_compatible(collect): + run, _, args = collect(warmup=0, rounds=1, limit=1) + env = json.loads(args.run_meta.read_text()) + env.update(endpoint=args.endpoint, path=args.path, thinking=False) + args.run_meta.write_text(dump(env)) + args.out = run.parent / "matching-run" + cli.predict(args) + assert aggregate(args.out)["pooled"]["attempted"] == 1 + + +@pytest.mark.parametrize("stop_reason", [None, "stale stop reason"]) +def test_environment_stop_reason_rejected_before_request(collect, stop_reason): + run, calls, args = collect(warmup=0, rounds=1, limit=1, statuses=[401]) + assert aggregate(run)["pooled"]["attempted"] == 1 + env = json.loads(args.run_meta.read_text()) + env["stop_reason"] = stop_reason + args.run_meta.write_text(dump(env)) + args.out = run.parent / "invalid-stop-reason" + with pytest.raises(ValueError, match="stop_reason"): + cli.predict(args) + assert len(calls) == 1 and not args.out.exists() + + +def test_cli_uses_utf8_independent_of_locale(reference, tmp_path): + script = r""" +import json +import sys +from pathlib import Path +import httpx +from benchmarks.client import DecisionClient +from benchmarks.evidence import write_json +from benchmarks.legs import leg1_public231 as cli + +root = Path(sys.argv[1]) +environment = root / "environment.json" +write_json(environment, dict( + checkpoint=dict(repo="fixture", revision="fixture"), + backend=dict(name="MockTransport", version=httpx.__version__), + sampling=dict(temperature=1), hardware="CPU \u6d4b\u8bd5", +)) +class Client(DecisionClient): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs, transport=httpx.MockTransport( + lambda _: httpx.Response(422, text="refused\u2028request") + )) +cli.DecisionClient = Client +run = root / "run" +sys.argv = ["public231", "predict", "--endpoint", "http://fixture.invalid", + "--path", "/v1/decisions", "--run-meta", str(environment), + "--out", str(run), "--warmup", "0", "--rounds", "1", "--limit", "2"] +cli.main() +sys.argv = ["public231", "aggregate", "--run", str(run), "--out", str(root / "summary")] +cli.main() +with (run / "round-1/results.jsonl").open(encoding="utf-8") as stream: + records = [json.loads(line) for line in stream] +assert len(records) == 2 and all(row["error"] == "refused\u2028request" for row in records) +summary = json.loads((root / "summary/summary.json").read_text(encoding="utf-8")) +assert summary["pooled"]["attempted"] == 2 and summary["pooled"]["planned"] == 231 +""" + result = subprocess.run( + [sys.executable, "-c", script, str(tmp_path)], + env={**os.environ, "PYTHONUTF8": "0", "PYTHONCOERCECLOCALE": "0", "LC_ALL": "C", "PYTHONIOENCODING": "utf-8"}, + capture_output=True, + text=True, + encoding="utf-8", + timeout=30, + ) + assert result.returncode == 0, result.stderr diff --git a/tests/benchmarks/test_quality_public231.py b/tests/benchmarks/test_quality_public231.py new file mode 100644 index 0000000..77fb4f9 --- /dev/null +++ b/tests/benchmarks/test_quality_public231.py @@ -0,0 +1,255 @@ +"""Deterministic saved Omni responses, scored by the pinned public231 oracle.""" + +import copy +import json +import os +from pathlib import Path + +import pytest + +from benchmarks.evidence import body_for, digest, dump, load_reference +from benchmarks.quality_public231 import adapt_record, aggregate, load_collector + + +@pytest.fixture +def reference(): + if not os.environ.get("JEVBENCH_ROOT"): + pytest.skip("Set JEVBENCH_ROOT to the pinned reference") + return load_reference() + + +@pytest.fixture +def collector(): + source = os.environ.get("OMNI_BENCH_SOURCE") + if not source: + pytest.skip("Set OMNI_BENCH_SOURCE to the frozen raw collector") + return load_collector(Path(source)) + + +@pytest.fixture +def input_manifest(tmp_path, reference): + _, items, dataset = reference + rows = [ + dict( + id=task.id, + request=body_for(task, None), + expected={"decision": task.expected == "yes" if task.question["type"] == "noul" else task.expected}, + source={**identity, "jevbench_pin": dataset["jevbench_pin"]}, + labels=task.labels, + provenance=task.provenance, + group=task.group, + reference_expected=task.expected, + ) + for identity, task in items + ] + data = "".join(dump(row) + "\n" for row in rows).encode("utf-8") + assert digest(data) == "00d1f10ca1142499a598b5f1c6602d329cce1cef04c961c46fc0318daa2c67d3" + (tmp_path / "public231.jsonl").write_bytes(data) + manifest = tmp_path / "manifest.json" + manifest.write_text( + dump( + dict( + output={"path": "public231.jsonl", "sha256": digest(data)}, + sources={"jevbench_pin": dataset["jevbench_pin"], "reference_dataset_hash": dataset["dataset_hash"]}, + ) + ), + encoding="utf-8", + ) + return manifest + + +def response_record(answer, *, kind="invalid_response", status=200): + text = dump({"answers": {"decision": answer}, "model": "configured"}) + return dict( + id="unused", + state="completed", + status=status, + response=text, + response_sha256=digest(text.encode()), + answers={}, + error_kind=kind, + error="wire schema rejected" if kind else None, + latency_ms=12.0, + ) + + +def test_wire_rejection_does_not_discard_pinned_distribution(reference): + upstream, items, _ = reference + identity, task = next(row for row in items if row[1].question["type"] == "noul") + raw = response_record({"type": "noul", "probabilities": {"no": 0.5, "yes": 0.505}}) + row = adapt_record(upstream, identity, task, raw, "configured", 1, "fixture") + assert row["valid"] and row["renormalized"] and not row["strict_valid"] + assert row["wire_valid"] is False and row["ok"] is True + assert row["predicted"] == "yes" + assert row["latency_s"] is None and row["latency_basis"] is None and row["ts"] is None + + +def test_scalar_tie_and_label_only(reference): + upstream, items, _ = reference + for value, expected in ((0.5, "no"), (0.8, "yes")): + identity, task = next(row for row in items if row[1].question["type"] == "noul") + row = adapt_record( + upstream, identity, task, response_record({"type": "noul", "noul": value}), "configured", 1, "fixture" + ) + assert row["predicted"] == expected and row["probs_source"] == "native_noul_scalar" + identity, task = next(row for row in items if row[1].question["type"] == "choice") + row = adapt_record( + upstream, + identity, + task, + response_record({"type": "choice", "choice": task.expected}), + "configured", + 1, + "fixture", + ) + assert not row["valid"] and row["probs"] is None + + +def test_ordinal_accuracy_uses_argmax(reference): + upstream, items, _ = reference + identity, original = next(row for row in items if row[1].question["type"] == "score") + task = copy.copy(original) + task.labels, task.expected = ["0", "1", "2"], 0 + probabilities = {"0": 0.4, "1": 0.35, "2": 0.25} + row = adapt_record( + upstream, + identity, + task, + response_record({"type": "score", "score": 0.85, "probabilities": probabilities}), + "configured", + 1, + "fixture", + ) + assert row["predicted"] == "0" and row["correct"] is (task.expected == 0) + + +@pytest.mark.parametrize("kind,status", [("http_422", 422), ("timeout", None)]) +def test_failures_are_scored_as_invalid(reference, kind, status): + upstream, items, _ = reference + identity, task = items[0] + raw = response_record({"type": "choice", "probabilities": {}}, kind=kind, status=status) + if status is None: + raw.update(response=None, response_sha256=None) + row = adapt_record(upstream, identity, task, raw, "configured", 1, "fixture") + assert not row["valid"] and not row["correct"] and row["attempted"] + + +def saved_run(tmp_path, collector, input_path, *, count=231): + cases = collector.load_cases(input_path) + run = tmp_path / "round-1" + run.mkdir() + (run / "requests.jsonl").write_bytes(input_path.read_bytes()) + config = dict( + manifest_sha256=digest(input_path.read_bytes()), + runner_sha256=collector.SOURCE_SHA256, + model="configured", + metadata={"model_revision": "fixture"}, + phase="measured", + warmup_manifest_sha256=None, + latency_basis="post_to_body_json_and_schema_validation", + ) + (run / "config.json").write_text(dump(config)) + records = [] + for case in cases[:count]: + q = case["request"]["questions"]["decision"] + answer = {"type": q["type"]} + if q["type"] == "noul": + answer["noul"] = float(case["expected"]["decision"]) + else: + labels = list(q["criteria"]) if q["type"] == "choice" else list(map(str, range(len(q["criteria"])))) + answer["probabilities"] = {key: float(key == str(case["expected"]["decision"])) for key in labels} + answer[q["type"]] = case["expected"]["decision"] + raw = response_record(answer, kind=None) + raw.update(id=case["id"], answers=collector.read_answers(case, json.loads(raw["response"]))) + records.append(raw) + (run / "responses.jsonl").write_text("".join(dump(row) + "\n" for row in records)) + state = dict( + manifest_sha256=config["manifest_sha256"], + config_sha256=digest((run / "config.json").read_bytes()), + responses_sha256=digest((run / "responses.jsonl").read_bytes()), + attempted_ids=[r["id"] for r in records], + completed_ids=[r["id"] for r in records], + active_ids=[], + started_at=1.0, + finished_at=2.0, + wall_seconds=1.0, + stop_reason="completed" if count == 231 else "interrupted", + ) + (run / "completion.json").write_text(dump(state)) + return run + + +def test_full_plan_missing_round_and_raw_integrity(tmp_path, collector, input_manifest): + manifest = input_manifest + inputs = manifest.with_name("public231.jsonl") + run = saved_run(tmp_path, collector, inputs, count=230) + result, records = aggregate( + [run, tmp_path / "missing", tmp_path / "missing-3"], manifest, collector, "fixture-config" + ) + assert result["distinct_tasks"] == 231 and result["pooled"]["planned"] == 693 + assert result["rounds"][0]["accuracy"] == 230 / 231 + assert result["rounds"][1]["accuracy"] == 0 and result["rounds"][1]["coverage"] == 0 + assert len(records) == 693 and sum(row["attempted"] for row in records) == 230 + (run / "responses.jsonl").write_text("{}\n") + with pytest.raises(ValueError, match="checksum"): + aggregate([run], manifest, collector, "fixture-config") + + +def test_source_hash_is_required(tmp_path): + source = tmp_path / "bench.py" + source.write_text("raise AssertionError('must not execute')") + with pytest.raises(ValueError, match="collector hash"): + load_collector(source) + + +def test_recorded_round_identity_is_separate_from_stable_model_config(tmp_path, collector, input_manifest): + import shutil + + manifest = input_manifest + first = saved_run(tmp_path, collector, manifest.with_name("public231.jsonl")) + runs = [first, tmp_path / "round-2", tmp_path / "round-3"] + for path in runs[1:]: + shutil.copytree(first, path) + + def set_metadata(path, **values): + config = json.loads((path / "config.json").read_text()) + config["metadata"].update(values) + (path / "config.json").write_text(dump(config)) + completion = json.loads((path / "completion.json").read_text()) + completion["config_sha256"] = digest((path / "config.json").read_bytes()) + (path / "completion.json").write_text(dump(completion)) + + for number, run in enumerate(runs, 1): + set_metadata(run, round=number) + result, records = aggregate(runs, manifest, collector, "fixture-config") + assert result["distinct_tasks"] == 231 and result["pooled"]["planned"] == 693 + assert len(records) == 693 and result["evidence_complete"] + assert [row["source"]["recorded_round"] for row in result["rounds"]] == [1, 2, 3] + with pytest.raises(ValueError, match="Recorded round"): + aggregate([runs[1], runs[0], runs[2]], manifest, collector, "fixture-config") + set_metadata(runs[1], round=True) + with pytest.raises(ValueError, match="Recorded round"): + aggregate(runs, manifest, collector, "fixture-config") + set_metadata(runs[1], round=2, model_revision="other") + with pytest.raises(ValueError, match="Model configuration"): + aggregate(runs, manifest, collector, "fixture-config") + + +def test_started_without_terminal_keeps_attempt_and_full_denominator(tmp_path, collector, reference, input_manifest): + manifest = input_manifest + run = saved_run(tmp_path, collector, manifest.with_name("public231.jsonl"), count=0) + _, items, _ = reference + state = json.loads((run / "completion.json").read_text()) + state.update( + attempted_ids=[items[0][1].id], + active_ids=[items[0][1].id], + finished_at=None, + wall_seconds=None, + stop_reason=None, + ) + (run / "completion.json").write_text(dump(state)) + result, rows = aggregate([run], manifest, collector, "fixture-config") + assert rows[0]["attempted"] and rows[0]["status"] == "started_without_saved_terminal" + assert result["rounds"][0]["attempted"] == 1 and result["rounds"][0]["planned"] == 231 + assert result["rounds"][0]["upstream_attempted_metrics"]["brier_mean"] is None + assert not result["evidence_complete"]