From 64b5dc824a188f2403f8e04bb6017090c31101a1 Mon Sep 17 00:00:00 2001 From: Xiwei Pan Date: Mon, 20 Jul 2026 04:11:26 +0800 Subject: [PATCH] feat: enforce per-rule evidence budgets --- benchmark/agent_environment.py | 90 +++ benchmark/agent_pred.py | 53 ++ benchmark/agent_spool.py | 57 ++ benchmark/agent_submit.py | 62 +- benchmark/backend_score.py | 9 +- .../design/top50-evidence-budget-benchmark.md | 504 +++++++++++++ .../design/top50-evidence-budget-issues.md | 494 +++++++++++++ benchmark/env_setup.py | 10 +- benchmark/evidence_budget.py | 690 ++++++++++++++++++ benchmark/preflight.py | 4 +- benchmark/privilege_smoke.py | 79 ++ benchmark/process_control.py | 173 +++++ benchmark/submit_session.py | 133 +++- benchmark/tests/test_env_setup.py | 3 +- benchmark/tests/test_evidence_budget.py | 341 +++++++++ benchmark/tests/test_preflight.py | 1 + docker/Dockerfile | 29 +- 17 files changed, 2658 insertions(+), 74 deletions(-) create mode 100644 benchmark/agent_environment.py create mode 100644 benchmark/agent_pred.py create mode 100644 benchmark/agent_spool.py create mode 100644 benchmark/docs/design/top50-evidence-budget-benchmark.md create mode 100644 benchmark/docs/design/top50-evidence-budget-issues.md create mode 100644 benchmark/evidence_budget.py create mode 100644 benchmark/privilege_smoke.py create mode 100644 benchmark/process_control.py create mode 100644 benchmark/tests/test_evidence_budget.py diff --git a/benchmark/agent_environment.py b/benchmark/agent_environment.py new file mode 100644 index 0000000..c2ce4f5 --- /dev/null +++ b/benchmark/agent_environment.py @@ -0,0 +1,90 @@ +"""Run mini-swe shell actions as the dedicated unprivileged agent identity.""" +from __future__ import annotations + +import os +import subprocess + +from benchmark.process_control import ProcessLimits, run_capped_process + + +def run_as_agent( + command: str, + *, + cwd: str, + env: dict[str, str], + timeout: int, + uid: int, + gid: int, + extra_groups: tuple[int, ...] = (), + max_output_chars: int = 10_000, +) -> subprocess.CompletedProcess[str]: + """Execute one shell action after irreversibly dropping child privileges.""" + if os.name != "posix": + raise RuntimeError("the rankable agent environment requires POSIX privilege separation") + if os.geteuid() != 0 and (uid != os.getuid() or gid != os.getgid()): + raise PermissionError("runner must be root to launch a different agent identity") + + result = run_capped_process( + command, + shell=True, + cwd=cwd, + env=env, + timeout=timeout, + max_output_chars=max_output_chars, + combine_stderr=True, + uid=uid, + gid=gid, + extra_groups=extra_groups, + limits=ProcessLimits(cpu_seconds=max(1, timeout), memory_bytes=2 * 1024 ** 3, + file_bytes=64 * 1024 ** 2), + ) + if result.timed_out: + raise subprocess.TimeoutExpired(command, timeout, output=result.stdout) + return subprocess.CompletedProcess(command, result.returncode, stdout=result.stdout) + + +def make_agent_environment(session, *, uid: int, gid: int, + extra_groups: tuple[int, ...] = (), timeout: int = 300): + """Construct a pinned mini-swe LocalEnvironment with a privilege-dropping execute path.""" + from minisweagent.environments.local import LocalEnvironment + + class EvidenceLocalEnvironment(LocalEnvironment): + def execute(self, action: dict, cwd: str = "", *, timeout: int | None = None) -> dict: + command = action.get("command", "") + action_cwd = cwd or self.config.cwd or str(session.workdir) + if not session.admit_shell_action(command): + output = { + "output": "shell action budget exhausted\n", + "returncode": 75, + "exception_info": "", + } + self._check_finished(output) + return output + try: + result = run_as_agent( + command, + cwd=action_cwd, + env=os.environ | self.config.env, + timeout=timeout or self.config.timeout, + uid=uid, + gid=gid, + extra_groups=extra_groups, + max_output_chars=session.budget.max_output_chars, + ) + output = {"output": result.stdout, "returncode": result.returncode, + "exception_info": ""} + except Exception as error: + raw_output = getattr(error, "output", None) + raw_output = (raw_output.decode("utf-8", errors="replace") + if isinstance(raw_output, bytes) else (raw_output or "")) + output = { + "output": raw_output, + "returncode": -1, + "exception_info": f"An error occurred while executing the command: {error}", + "extra": {"exception_type": type(error).__name__, + "exception": str(error)}, + } + self._check_finished(output) + return output + + return EvidenceLocalEnvironment(cwd=str(session.workdir), timeout=timeout) diff --git a/benchmark/agent_pred.py b/benchmark/agent_pred.py new file mode 100644 index 0000000..9ca69f3 --- /dev/null +++ b/benchmark/agent_pred.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""Thin agent-facing client for the evaluation-owned pred gateway.""" +from __future__ import annotations + +import json +import os +import sys +from benchmark.agent_spool import spool_request + +DEFAULT_REQUEST_TIMEOUT_SECONDS = 300 + + +def _request(payload: dict, request_id: str | None = None) -> dict: + return spool_request( + payload, + channel_env="PRB_PRED_DIR", + timeout_env="PRB_PRED_TIMEOUT", + default_timeout_seconds=DEFAULT_REQUEST_TIMEOUT_SECONDS, + service_name="pred gateway", + request_id=request_id, + ) + + +def _format_status(budget: dict) -> str: + pred = budget["pred_calls"] + solve = budget["solve_calls"] + return (f"pred calls: {pred['used']}/{pred['limit']} used, {pred['remaining']} remaining; " + f"solve calls: {solve['used']}/{solve['limit']} used, " + f"{solve['remaining']} remaining") + + +def main() -> None: + args = sys.argv[1:] + try: + if args == ["--budget-status"]: + response = _request({"op": "status", "cwd": os.getcwd()}) + print(_format_status(response["budget"])) + return + response = _request({"op": "pred", "args": args, "cwd": os.getcwd()}) + except (OSError, ValueError, json.JSONDecodeError, RuntimeError, TimeoutError) as error: + print(f"pred unavailable: {error}", file=sys.stderr) + raise SystemExit(2) + + if response.get("stdout"): + print(response["stdout"], end="" if response["stdout"].endswith("\n") else "\n") + if response.get("stderr"): + print(response["stderr"], end="" if response["stderr"].endswith("\n") else "\n", + file=sys.stderr) + raise SystemExit(int(response.get("returncode", 2))) + + +if __name__ == "__main__": + main() diff --git a/benchmark/agent_spool.py b/benchmark/agent_spool.py new file mode 100644 index 0000000..21dc20b --- /dev/null +++ b/benchmark/agent_spool.py @@ -0,0 +1,57 @@ +"""Shared thin-client transport for evaluation-owned agent commands.""" +from __future__ import annotations + +import json +import os +import time +import uuid +from pathlib import Path + +POLL_INTERVAL_SECONDS = 0.02 + + +def spool_request( + payload: dict, + *, + channel_env: str, + timeout_env: str, + default_timeout_seconds: int, + service_name: str, + request_id: str | None = None, +) -> dict: + """Publish one idempotent request and await its atomic response.""" + channel = os.environ.get(channel_env) + if not channel: + raise RuntimeError( + f"{service_name} is only available inside an active benchmark evaluation") + request_id = request_id or os.environ.get("PRB_REQUEST_ID") or uuid.uuid4().hex + if len(request_id) != 32 or any(char not in "0123456789abcdef" for char in request_id): + raise ValueError("request id must be 32 lowercase hexadecimal characters") + + channel_dir = Path(channel) + inbox = channel_dir / "inbox" + outbox = channel_dir / "outbox" + request_path = inbox / f"{request_id}.json" + response_path = outbox / f"{request_id}.json" + envelope = {"request_id": request_id, **payload} + + timeout = int(os.environ.get(timeout_env, default_timeout_seconds)) + attempts = int(os.environ.get("PRB_CLIENT_RETRIES", "1")) + 1 + for attempt in range(attempts): + temporary = inbox / f".{request_id}.{os.getpid()}.{attempt}.tmp" + temporary.write_text(json.dumps(envelope), encoding="utf-8") + os.replace(temporary, request_path) + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + raw = response_path.read_text(encoding="utf-8") + except FileNotFoundError: + time.sleep(POLL_INTERVAL_SECONDS) + continue + response = json.loads(raw) + try: + response_path.unlink() + except FileNotFoundError: + pass + return response + raise TimeoutError(f"{service_name} did not respond within {timeout}s") diff --git a/benchmark/agent_submit.py b/benchmark/agent_submit.py index b3edef5..3170942 100644 --- a/benchmark/agent_submit.py +++ b/benchmark/agent_submit.py @@ -9,15 +9,13 @@ import argparse import json -import os import sys -import time -import uuid from pathlib import Path +from benchmark.agent_spool import spool_request + MAX_CERTIFICATE_BYTES = 1024 * 1024 DEFAULT_REQUEST_TIMEOUT_SECONDS = 300 -POLL_INTERVAL_SECONDS = 0.02 def _read_certificate(path: str) -> str: @@ -32,54 +30,33 @@ def _read_certificate(path: str) -> str: return raw.decode("utf-8") -def _request(payload: dict) -> dict: - channel = os.environ.get("PRB_SUBMIT_DIR") - if not channel: - raise RuntimeError("submit is only available inside an active benchmark evaluation") - channel_dir = Path(channel) - request_id = uuid.uuid4().hex - inbox = channel_dir / "inbox" - outbox = channel_dir / "outbox" - request_path = inbox / f"{request_id}.json" - temp_path = inbox / f".{request_id}.{os.getpid()}.tmp" - response_path = outbox / f"{request_id}.json" - - envelope = {"request_id": request_id, **payload} - temp_path.write_text(json.dumps(envelope), encoding="utf-8") - os.replace(temp_path, request_path) - - timeout = int(os.environ.get("PRB_SUBMIT_TIMEOUT", DEFAULT_REQUEST_TIMEOUT_SECONDS)) - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - try: - raw = response_path.read_text(encoding="utf-8") - except FileNotFoundError: - time.sleep(POLL_INTERVAL_SECONDS) - continue - response = json.loads(raw) - try: - response_path.unlink() - except FileNotFoundError: - pass - return response - raise TimeoutError(f"submit service did not respond within {timeout}s") +def _request(payload: dict, request_id: str | None = None) -> dict: + return spool_request( + payload, + channel_env="PRB_SUBMIT_DIR", + timeout_env="PRB_SUBMIT_TIMEOUT", + default_timeout_seconds=DEFAULT_REQUEST_TIMEOUT_SECONDS, + service_name="submit service", + request_id=request_id, + ) def main() -> None: parser = argparse.ArgumentParser( prog="submit", description="Submit one rule counterexample to the benchmark verifier. Every " - "submission, accepted or rejected, consumes one run-wide attempt.", + "submission, accepted or rejected, consumes one rule-scoped attempt.", ) parser.add_argument("certificate", nargs="?", help="Certificate JSON file, or - for stdin") parser.add_argument("--status", action="store_true", help="Show remaining attempts (free)") + parser.add_argument("--request-id", help=argparse.SUPPRESS) args = parser.parse_args() try: if args.status: if args.certificate: parser.error("--status does not take a certificate") - response = _request({"op": "status"}) + response = _request({"op": "status"}, args.request_id) print(f"submit budget: {response['used']}/{response['limit']} used, " f"{response['remaining']} remaining") return @@ -87,12 +64,12 @@ def main() -> None: parser.error("a certificate JSON file is required (use - for stdin)") try: text = _read_certificate(args.certificate) - response = _request({"op": "submit", "certificate_text": text}) + response = _request({"op": "submit", "certificate_text": text}, args.request_id) except (OSError, UnicodeError, ValueError) as e: # File/encoding/size failures are still submission attempts; let the service # account for them instead of failing locally without consuming budget. - response = _request({"op": "submit", "client_error": str(e)}) - except (OSError, ValueError, json.JSONDecodeError, RuntimeError) as e: + response = _request({"op": "submit", "client_error": str(e)}, args.request_id) + except (OSError, ValueError, json.JSONDecodeError, RuntimeError, TimeoutError) as e: print(f"submit unavailable: {e}", file=sys.stderr) raise SystemExit(2) @@ -102,9 +79,10 @@ def main() -> None: return reason = response.get("reason") or response.get("error") or "rejected" - label = "BUDGET_EXHAUSTED" if response.get("exhausted") else "REJECTED" + label = ("INFRASTRUCTURE_ERROR" if response.get("infrastructure_error") else + "BUDGET_EXHAUSTED" if response.get("exhausted") else "REJECTED") print(f"{label} attempt {used}/{limit} ({remaining} remaining): {reason}", file=sys.stderr) - raise SystemExit(2 if response.get("exhausted") else 1) + raise SystemExit(2 if response.get("exhausted") or response.get("infrastructure_error") else 1) if __name__ == "__main__": diff --git a/benchmark/backend_score.py b/benchmark/backend_score.py index 2a4d503..78bbb08 100644 --- a/benchmark/backend_score.py +++ b/benchmark/backend_score.py @@ -19,7 +19,6 @@ import hashlib import json import re -import shutil import sys from pathlib import Path @@ -123,10 +122,12 @@ def _assert_pred_version() -> None: """The backend is the authoritative verifier, so its pred must be the pinned version. Skip when no pred is on PATH (pred-free unit tests never verify real certificates) or when EXPECTED_PRED_VERSION is set empty; otherwise hard-fail on a mismatch.""" - if not shutil.which("pred"): + from benchmark.env_setup import find_pred_binary, verify_pred_version + try: + pred_binary = find_pred_binary() + except RuntimeError: return - from benchmark.env_setup import verify_pred_version - verify_pred_version("pred") # raises ValueError on mismatch + verify_pred_version(pred_binary) # raises ValueError on mismatch # ── status helpers ──────────────────────────────────────────────────────────── diff --git a/benchmark/docs/design/top50-evidence-budget-benchmark.md b/benchmark/docs/design/top50-evidence-budget-benchmark.md new file mode 100644 index 0000000..36d4585 --- /dev/null +++ b/benchmark/docs/design/top50-evidence-budget-benchmark.md @@ -0,0 +1,504 @@ +# Top-50 Evidence-Budget Benchmark — Product Design + +Status: proposed for the next benchmark version +Last updated: 2026-07-20 + +## Need + +The current benchmark gives a model the complete `problem-reductions` repository and lets +it decide how broadly and deeply to search and when to stop. The runner limits submissions, +but it does not limit model turns, shell actions, or calls to `pred`. Consequently, two runs +may receive radically different amounts of feedback from the target system. Wall-clock time +is not a suitable correction: the model cannot budget against it reliably, and it also +contains provider latency, retries, network delay, host load, and harness overhead. + +The next benchmark version must compare models under the same **logical opportunities to +obtain evidence**, rather than under the same elapsed time. A model first selects the 50 +rules it considers most likely to contain bugs, then investigates every selected rule with +the same non-transferable per-rule budget. The primary result remains the number of distinct +rules for which the evaluation-owned verifier accepts a counterexample. + +The benchmark is for researchers and developers comparing base models. It has one +standardized Model Track, using one reference harness and one frozen evaluation contract. +It intentionally does not compare rapidly changing coding-agent products or custom agent +systems. + +Success means that: + +- a longer-running session cannot obtain more target-system feedback than a shorter one; +- shell scripts, loops, and parallel subprocesses cannot bypass the `pred` or submission + budgets; +- all 50 selected rules receive the same investigation allowance; +- infrastructure failures are distinguishable from model failures and never become clean + zero scores; +- the score has a narrow, defensible interpretation: verified bug-finding yield under a + fixed evidence budget. + +## Prior art and landscape + +The design follows a common pattern in agent evaluation: bound interaction opportunities +directly and retain timeouts only as safety mechanisms. + +- [METR's time-horizon methodology](https://metr.org/time-horizons/) notes that exact agent + runtime varies with the inference provider and agent setup and is not itself the reported + capability measure. +- [Inspect](https://inspect.aisi.org.uk/agent-custom.html) exposes separate limits for + messages, tokens, turns, time, working time, and cost. Its working-time accounting excludes + delays such as rate-limit waiting, illustrating why resource dimensions should not be + collapsed into elapsed time. +- [InterCode](https://proceedings.neurips.cc/paper_files/paper/2023/file/4b175d846fb008d540d233c188379ff9-Paper-Datasets_and_Benchmarks.pdf) + bounds interaction turns and observation size. [OSWorld](https://github.com/xlang-ai/OSWorld) + and [OpenCUA](https://github.com/xlang-ai/OpenCUA) likewise publish results at explicit + action-step budgets. +- [tau2-bench](https://github.com/sierra-research/tau2-bench/blob/main/docs/cli-reference.md) + separates maximum steps, maximum errors, retries, trials, and an optional wall-clock + timeout instead of treating time as the sole budget. +- [SWE-agent](https://swe-agent.com/latest/usage/competitive_runs/) recommends a per-instance + cost or turn limit because otherwise individual instances can consume unbounded resources. +- [AI Agents That Matter](https://arxiv.org/abs/2407.01502) argues that agent results should + expose capability and resource use jointly, including Pareto-style comparisons, instead + of hiding unequal evaluation cost. +- Systems such as [AlphaCode 2](https://deepmind.google/AlphaCode2_Tech_Report.pdf) separate + large internal search from a small external submission allowance. This motivates separate + ledgers for `pred` feedback and formal bug submissions. + +Existing repository components provide useful building blocks: + +- `SubmissionSession` already owns an evaluation-side, atomic, append-only submission + ledger and an agent-facing file-spool transport. +- `verify.py` already applies payload, CPU, memory, output, and solver watchdog limits. +- API and CLI runners already share prompts and normalized result packaging. + +The repository does not yet provide an evaluation-owned `pred` gateway, per-rule budgets, +isolated per-rule episodes, shortlist freezing, or a distinction between verifier +infrastructure errors and rejected model submissions. Those gaps belong to this benchmark. + +## Features + +### Selected for the first release + +| Feature | User value | Rough implementation effort | +|---|---|---:| +| Static Top-50 triage and shortlist freezing | Measures risk prioritization without allowing dynamic experiments before commitment | 1–2 days | +| Fifty isolated per-rule investigation episodes | Gives every selected rule the same opportunity and removes order and long-context effects | 3–5 days | +| Evaluation-owned `pred`/`solve` gateway | Makes all target-system feedback countable and prevents script/parallelism bypasses | 2–4 days | +| Two submissions **per selected rule** | Allows one formal attempt and one feedback-driven correction without creating an unbounded proof loop | 1–2 days | +| Visible remaining-budget counters | Lets the model plan in logical units it can observe | about 1 day | +| Single-run scoring and audit schema | Produces one comparable, re-verifiable result under a named budget contract | 2–3 days | +| Internal budget-calibration pilot | Selects defensible turn and `pred` limits before the public contract is frozen | 2–4 days plus inference cost | + +The estimates are implementation-order estimates, not elapsed calendar commitments. The +existing submission and verification machinery is reused where possible. + +### Deferred + +#### Multiple seeds + +The first release records any available seed and sampling parameters but does not require +repeated runs, compute means or confidence intervals, or rank aggregate results. This can be +added after the single-run contract and its operating cost are understood. + +#### Fixed Top-50 diagnostic + +A future diagnostic may give every model the same 50 rules to isolate investigation ability +from risk-ranking ability. It is not required for the main benchmark, which deliberately +measures the combined workflow, and is deferred to keep the first release small. + +### Dropped + +#### System Track + +There is no leaderboard for Codex CLI, Claude Code, custom prompts, memory systems, +subagents, or other agent products. These systems change too quickly, attract less interest +for this benchmark, and confound model capability with orchestration. Existing CLI adapters +may remain useful for development, but their runs are not rankable in this track. + +#### Wall-clock, token, cost, or best-of-N ranking + +Wall-clock time remains a watchdog and an auxiliary measurement. Tokens and monetary cost +are reported when available. None is a primary score or tie-breaker, and ties in verified +bug count remain ties. The first release has no best-of-N selection. + +## Evaluation contract + +### Track + +The benchmark has one rankable track: **Standardized Model Track**. Rankable runs use the +same: + +- reference API harness and harness version; +- system and task prompts; +- pinned library repository and `pred` version; +- rule inventory; +- tool surface and network policy; +- triage and per-rule budgets; +- maximum model output and observation sizes; +- reasoning/sampling configuration, to the extent the model API exposes it; and +- submission verifier. + +The result identifies the model, provider/endpoint class, runner version, library commit, +budget-contract version, and exposed inference parameters. A custom coding-agent harness is +outside this track even if it uses the same underlying model. + +Rankable runs are produced and submitted only by the trusted internal team. The first release +does not accept untrusted external runs and therefore does not build remote execution or +cryptographic runner attestation. The backend still re-verifies bugs and ledger consistency, +but the operator trust boundary is explicit. + +### Phase 0: preflight + +Before any scored model generation, the runner checks the pinned repository, rule inventory, +`pred` gateway, and submission judge. Status and version probes are cached and free. If any +required service is unavailable, the run ends as `run_error` and is not rankable. + +### Phase 1: static triage + +The model receives the pinned repository and a canonical, cached inventory of all runnable +rules. It may inspect source code with ordinary read-only shell tools under a fixed triage +generation/action/observation budget. Dynamic `pred` operations and `submit` are unavailable +during this phase. + +Before the triage budget expires, the model must commit an ordered list of exactly 50 unique, +valid rule identifiers. It may attach a short, size-capped risk hypothesis for each rule. +The runner validates and freezes this artifact. The list cannot change after investigation +begins. + +Failure to produce 50 valid unique rules is an invalid run, not a 0-bug result. This avoids +silently comparing a 50-rule run with a smaller task. + +The order is retained for audit and future diagnostics such as Bugs@10 and Bugs@25, but the +first-release headline score uses all 50 rules. + +### Phase 2: isolated investigation + +The runner creates one fresh episode for each frozen rule. Every episode has: + +- a fresh model conversation and writable scratch directory; +- the same pinned read-only repository; +- the current rule identifier and its capped triage hypothesis; +- the same per-rule model-generation, shell-action, `pred`, `solve`, observation, and + submission limits; and +- a submission judge that accepts certificates only for the current rule. + +No conversation, scratch files, remaining budget, or model-authored memory transfers between +rules. Unused allowance is not transferable. The episodes may execute sequentially or in +parallel as an implementation detail, provided the model and infrastructure configuration is +identical and each ledger remains isolated. + +An accepted certificate ends that rule's episode immediately. Exhausting a counter leaves +the episode as completed without an accepted bug; it does not borrow from another rule. + +### Logical resource vector + +The frozen public contract is named by a vector rather than a time allowance: + +`Top50[T, E_t, M, E, P, P_solve, S=2, O]` + +where: + +- `T` is the number of usable model generations in static triage; +- `E_t` is the number of executed triage shell actions; +- `M` is the number of usable model generations per selected rule; +- `E` is the number of executed shell actions per selected rule; +- `P` is the number of agent-initiated dynamic `pred` invocations per selected rule; +- `P_solve` is the subset of `P` that may invoke `pred solve`; +- `S` is the number of submission opportunities per selected rule, fixed initially at 2; + and +- `O` contains fixed observation, command-output, and payload-size ceilings. + +`S=2` is a per-rule ledger. With 50 selected rules, a run can therefore make at most 100 +charged submissions, but there is no run-wide pool from which one rule can take another +rule's unused attempts. Once a rule is accepted, its unused submission opportunity expires. + +The exact values of `T`, `E_t`, `M`, `E`, `P`, `P_solve`, and `O` are selected by the internal +calibration pilot and then frozen with the benchmark version. + +Every observation includes a compact machine-generated status block, for example: + +```text +rule 17/50: maximum_independent_set_to_clique +model generations: 6/10 +shell actions: 6/10 +pred calls: 13/24 +solve calls: 5/10 +submit attempts: 1/2 +``` + +This is the model's planning signal. Wall-clock time is not. + +## Modules + +### 1. Inventory and Triage Controller + +Purpose: provide a canonical rule inventory, run static analysis, validate the shortlist, +and freeze the ordered Top50 artifact. + +Interface: + +- input: pinned repository metadata and triage budget contract; +- output: an immutable ordered list of 50 rule IDs plus capped hypotheses and a triage + ledger; +- dependencies: reference model harness, read-only repository, budget controller. + +### 2. Episode Runner + +Purpose: create 50 fresh, equal-budget investigation episodes and combine their immutable +ledgers without sharing model state or resources. + +Interface: + +- input: frozen Top50 artifact and per-rule budget contract; +- output: 50 episode records with completion, accepted bug, or infrastructure-error status; +- dependencies: model harness, `PredGateway`, `SubmitJudge`, observation renderer. + +### 3. Pred Gateway + +Purpose: be the only executable route to dynamic `pred` behavior and atomically enforce `P` +and `P_solve` for the current episode. + +Interface: + +- agent command: a `pred`-compatible shim; +- runner API: status, execute, ledger, and health operations; +- output: idempotent response plus updated per-rule counters; +- dependencies: hidden real `pred` binary, pinned execution environment, file-spool or RPC + transport. + +### 4. Submit Judge + +Purpose: enforce exactly two non-transferable formal attempts for the current rule and +retain the verifier-authoritative certificate ledger. + +Interface: + +- agent command: `submit ` and free `submit --status`; +- runner API: status, verify, ledger, and health operations; +- output: accepted, rejected, exhausted, or infrastructure-error response; +- dependencies: existing certificate verifier and hidden internal `pred` access. + +### 5. Scoring and Calibration + +Purpose: validate rankability, compute the single-run headline score, expose audit metrics, +and determine the fixed public budgets before release. + +Interface: + +- input: triage ledger, 50 episode ledgers, inference usage, and run metadata; +- output: versioned submission artifact and internal pilot analysis; +- dependencies: schema validator and backend re-verifier. + +## Technical approaches + +### Inventory and triage + +**Chosen: one bounded static-analysis session followed by an atomic shortlist commit.** This +directly measures the desired prioritization behavior while preventing the model from using +dynamic fuzzing to decide what belongs in the Top50. + +Alternatives considered: + +- Ask the model to score every rule independently. This is easier to normalize per rule but + converts strategic prioritization into 290+ expensive microtasks and loses cross-rule + comparison in one context. +- Use a benchmark-authored heuristic to prefilter candidates. This is cheaper but removes a + central model ability from the task and risks encoding maintainer bias. + +### Episode execution + +**Chosen: 50 fresh, equal-budget, per-rule episodes.** This ensures that all shortlisted +rules receive the same opportunity and that later rules are not disadvantaged by context +growth or earlier spending. + +Alternatives considered: + +- One global investigation session with a run-wide pool. This measures adaptive allocation, + but recreates unequal rule coverage and makes scores difficult to interpret. +- A global pool plus a per-rule ceiling. This blocks extreme concentration but still makes a + miss ambiguous: the model may be weak on the rule or simply have allocated less budget. + +### Pred enforcement + +**Chosen: an evaluation-owned shim and service modeled on `SubmissionSession`.** The real +binary is not placed directly on the agent's `PATH`; the shim atomically reserves budget +before spawning it. All agent shells, Python programs, loops, and parallel subprocesses +share the same per-episode ledger. + +Alternatives considered: + +- Infer calls from shell commands or trajectories. One shell command can start hundreds of + subprocesses, so this is neither complete nor enforceable. +- Expose the real binary and ask the agent to self-report. This is unauditable and permits + accidental bypasses. + +Simple invocation counting is preferred over weighted command costs. Each agent-initiated +`create`, `reduce`, `solve`, `evaluate`, or `extract` consumes one `P`; `solve` also consumes +one `P_solve`. Cached inventory, version, help, and status operations are free. `pred` calls +made internally by the submission verifier do not consume the agent's investigation budget. + +Model-authored automation remains allowed: writing a fuzzer or batch script is useful agent +behavior. Its actual gateway invocations are charged individually, so automation is an +ability rather than a loophole. + +### Submission enforcement + +**Chosen: adapt the existing evaluation-owned ledger from run-wide accounting to one ledger +per rule.** A rejected, malformed, duplicate, wrong-rule, or model-caused verifier timeout +consumes one of that rule's two opportunities. `submit --status` is free. Acceptance closes +the episode. + +Pure verifier, gateway, container, or transport failures do not consume an opportunity. +Requests carry stable idempotency keys; retrying the same request after a lost response +returns the stored result instead of charging again. An unrecoverable judge failure marks the +run unrankable rather than converting the episode into a rejection. + +Alternatives considered: + +- Keep one shared 100-attempt run pool. Although its total matches 50 × 2, it lets one rule + consume attempts intended for another and violates the equal-opportunity contract. +- Allow only one attempt. This measures first-shot rigor, but provides no bounded opportunity + to use verifier feedback to correct a nearly valid certificate. +- Allow attempts until acceptance. This rewards persistence with unequal external feedback + and recreates the original bias. + +### Scoring + +**Chosen: count distinct accepted rules under the complete frozen budget vector.** The +headline metric is: + +`VerifiedBugs@Top50[T,E_t,M,E,P,P_solve,S=2,O]` + +It is the number from 0 to 50 of selected rules with at least one certificate accepted by the +evaluation-owned verifier. It is reported for one run in the first release. Equal counts are +ties. + +Audit and efficiency fields include: + +- ordered Top50 and hypotheses; +- Bugs@10, Bugs@25, and Bugs@50 along the frozen shortlist order; +- accepted on submission 1 versus submission 2; +- `pred` and `solve` calls per accepted bug; +- cap-hit counts for each logical resource; +- token usage, monetary cost when available, and elapsed time; and +- all infrastructure retries and errors. + +These fields explain a run but do not override the headline ranking. + +Because there is no exhaustive ground truth for which rules are buggy, the benchmark must +not call unaccepted rules clean or report Top50 precision/recall. It measures verified bug +discovery yield, not complete bug prevalence. + +### Internal budget calibration + +**Chosen: run a small pre-release budget grid on development data, then freeze one contract.** +Candidate values may start around `M ∈ {6, 10, 14}` and `P ∈ {8, 16, 32}`, with a separate +`P_solve` ceiling. These are pilot coordinates, not proposed public values. + +The pilot uses an older library snapshot or other non-ranking development target and a small +set of available models. It examines: + +- verified bugs as each budget increases; +- the fraction of episodes hitting each cap; +- marginal bugs gained per additional generation or `pred` call; +- infrastructure failure rate; and +- whether one counter consistently binds long before the others. + +The selected contract should sit near the useful-yield knee: large enough that most episodes +can conduct a meaningful experiment, but before extra calls mostly add cost. The chosen +values and rationale are published and versioned. Formal leaderboard runs execute only that +one frozen budget, not the grid. The pilot does not introduce a multi-seed requirement. + +## Counter and error semantics + +Counters are charged for opportunities controlled by the model and retried for failures +controlled by the evaluation infrastructure. + +| Event | Counter effect | Result semantics | +|---|---|---| +| Provider returns a usable model response | consume one `T` or `M` | response may lead to an action | +| Model emits invalid action/format | consume one `T` or `M`; no shell action if none executes | model failure | +| API 429/5xx/disconnect before any usable response | no model-generation charge; fixed infrastructure retry policy | unrecoverable exhaustion makes run unrankable | +| Shell command starts | consume one `E_t` or `E` | nonzero exit and model-caused timeout still consume | +| Dynamic `pred` request is admitted | atomically consume one `P`; `solve` also consumes `P_solve` | invalid arguments, bad model-authored input, nonzero exit, and target timeout consume | +| Gateway fails before admitting the request | no charge; retry idempotently | unrecoverable failure makes run unrankable | +| Submission is admitted for verification | atomically consume one of the current rule's two `S` | malformed, rejected, duplicate, and wrong-rule attempts consume | +| Judge infrastructure fails | no `S` charge; retry the same request idempotently | unrecoverable failure makes run unrankable | +| Status/help/version/cached inventory probe | free | rate-limited separately if abuse protection is needed | + +A reservation is written before the underlying process starts so parallel requests cannot +overspend. Every ledger record includes episode ID, rule ID, request ID, command class, +counter values before and after, outcome class, and timestamps for auditing. Timestamps are +not part of the score. + +## Quality requirements + +### Fairness and reproducibility + +- Budget values, prompts, schemas, repository commit, `pred`, harness, and verifier are + immutable within a benchmark version. +- All 50 per-rule episodes receive identical non-transferable limits. +- The real `pred` binary and verifier-only interfaces are inaccessible to the agent. +- Network access from the agent environment is disabled; provider access remains in the + runner. +- The execution image, host class, watchdogs, memory limits, and file/output ceilings are + fixed for official runs and recorded where relevant. + +### Reliability + +- Budget and result ledgers are evaluation-owned, append-only, thread-safe, and idempotent. +- Crashes preserve accepted certificates and audit logs, but any incomplete infrastructure + run is marked unrankable rather than published as a clean partial score. +- Free preflight probes must succeed before scored generations begin. +- Watchdogs remain on model calls, commands, `pred`, verification, and the complete run to + terminate wedged processes; they protect the service and do not define capability. + +### Compatibility + +- The agent-facing `pred` shim preserves the documented CLI shape wherever possible so + existing prompts and model-authored scripts continue to work. +- Result schema changes are tied to a benchmark-version bump. Older whole-repository + submissions remain readable as historical results but are not mixed into the new ranking. + +## What the benchmark can claim + +Under a standardized harness and fixed logical budget, the score reflects a combination of: + +- static risk ranking across a large rule library; +- source-code understanding; +- high-information experiment design; +- counterexample construction and minimization; +- interpretation of bounded `pred` and verifier feedback; +- disciplined use of limited formal submissions; and +- end-to-end reliability in producing verifier-accepted evidence. + +It does not isolate pure model reasoning from all provider implementation details, equalize +hidden inference FLOPs, measure complete recall over all bugs, compare custom agent systems, +or rank network and runtime speed. + +## Acceptance criteria + +The first release is ready when automated tests demonstrate that: + +1. triage cannot call dynamic `pred` or `submit`, and only an exact valid Top50 can start + investigation; +2. all 50 episodes start with identical counters and isolated model/scratch state; +3. direct shell use, Python subprocesses, loops, and concurrent requests cannot exceed `P` + or `P_solve`; +4. no rule can make more than two charged submissions or use another rule's unused attempt; +5. accepted certificates close their episode and score at most one bug per rule; +6. model-caused invalid operations charge the documented counters; +7. injected provider, gateway, transport, and verifier infrastructure failures follow the + documented no-charge/idempotent behavior and make unrecoverable runs unrankable; +8. every observation exposes authoritative remaining counters; +9. the backend recomputes the headline score only from accepted evaluation-owned ledgers; + and +10. the published result identifies the complete frozen budget contract and necessary + provenance. + +## Open questions before implementation + +These are intentionally resolved by implementation discovery or the calibration pilot, not +by arbitrary values in the design: + +- final numeric values for `T`, `E_t`, `M`, `E`, `P`, `P_solve`, and `O`; +- the smallest stable inference-parameter contract shared by supported model APIs; +- migration and display treatment for historical unlimited whole-repository results. diff --git a/benchmark/docs/design/top50-evidence-budget-issues.md b/benchmark/docs/design/top50-evidence-budget-issues.md new file mode 100644 index 0000000..4bb1b22 --- /dev/null +++ b/benchmark/docs/design/top50-evidence-budget-issues.md @@ -0,0 +1,494 @@ +# Top-50 Evidence-Budget Benchmark — Implementation Issue Drafts + +Status: filed as GitHub issues #65–#68; implementation in progress +Source design: [Top-50 Evidence-Budget Benchmark](top50-evidence-budget-benchmark.md) + +## Agreed boundary + +This batch implements one trusted-team, standardized Model API track in which a model commits +to its own Top50 and investigates every selected rule under identical logical budgets. It does +not implement multiple seeds, a Fixed Top50 diagnostic, a coding-agent/System Track, external +submission attestation, or ranking by elapsed time, tokens, or cost. + +The four issues are deliberately coarse. Each delivers a complete component with one compound +behavioural acceptance suite; internal classes and substeps are not separate issues. + +## Milestones and dependency graph + +```text +M1 Enforceable evidence budget +└── I1 Evidence Budget Service + └── M2 End-to-end Top50 benchmark + └── I2 Self-selected Top50 Runner + └── I3 vNext result and ranking contract + └── M3 Calibration and release + └── I4 Calibrate and publish the frozen contract +``` + +I3 depends on I2 because its schema and scorer validate the actual Top50 run artifact. I4 +depends on I1–I3 because calibration must exercise the same implementation that will be +released and must publish the selected values through the versioned result contract. + +--- + +## I1 — Build the evaluation-owned Evidence Budget Service + +### Background + +The current benchmark exposes the real `pred` executable to the model and limits only a +run-wide submission count. One shell action can start hundreds of `pred` subprocesses, so turns +and shell commands do not bound how much target-system feedback a model receives. The +[Top-50 design](top50-evidence-budget-benchmark.md) instead gives every selected rule an +independent allowance for model generations, shell actions, `pred` calls, `solve` calls, and +two formal submissions. + +The runner image currently executes as root and installs the real binary at +`/usr/local/bin/pred`. Replacing `PATH` alone is therefore not enforcement: a model-authored +script could find and execute the binary by absolute path. This issue owns the complete trusted +service and OS boundary that make the logical budget enforceable. + +### Objective + +Build one evaluation-owned Evidence Budget Service that atomically accounts for a single rule +episode, exposes `pred`-compatible and `submit` shims to an unprivileged agent shell, and keeps +the real `pred`, verifier, counters, and accepted-certificate ledger outside the agent's access. + +Each rule receives exactly two non-transferable submission opportunities. Rejected, malformed, +duplicate, wrong-rule, and other model-caused attempts consume one; pure gateway, transport, +container, or verifier infrastructure errors do not. Retrying the same request after a lost +response must be idempotent. + +### Interface (Input → Output) + +- **Input:** episode ID, current rule ID, frozen budget contract + `{M, E, P, P_solve, S=2, O}`, pinned real `pred` path, and certificate verifier. +- **Agent surface:** an unprivileged writable scratch directory; a `pred` shim supporting the + documented CLI shape; `submit `; and free status/help/version/cached-inventory + operations. +- **Output:** immutable per-episode ledgers for model generations, shell actions, admitted + `pred`/`solve` requests, submission attempts, idempotent responses, accepted certificate, and + classified infrastructure errors. + +### Technical recommendations + +- Reuse the atomic file-spool, directory-FD, and append-only ledger patterns in + `benchmark/submit_session.py` instead of introducing a network service. +- Add separate non-root container identities for model-authored shell actions and admitted + `pred` subprocesses. The real binary is executable only by the oracle identity; both identities + share only the episode scratch group. The privileged Python parent owns counters and spawning, + but never runs model-parameterized `pred` as root. +- Reserve a counter under a lock before spawning a process. A dynamic `create`, `reduce`, + `solve`, `evaluate`, or `extract` request consumes one `P`; `solve` additionally consumes one + `P_solve`. Nonzero exits, invalid arguments, pathological model input, and target timeouts + consume the reservation. +- Give requests caller-stable idempotency keys and retain completed responses. Do not generate + a new logical request ID merely because the client lost the first response. +- Refactor `SubmissionSession` into, or adapt it behind, a rule-scoped judge. Treat verifier + exceptions as `infrastructure_error`, not ordinary rejection. Acceptance closes the rule and + expires its unused submission attempt. +- Keep verifier-internal `pred` calls outside the agent's `P`/`P_solve` ledger. + +These are recommendations; an alternative implementation is acceptable if it passes the same +privilege, concurrency, charging, and idempotency checks. + +### Verification + +Add one integration acceptance suite and run: + +```bash +pytest -v benchmark/tests/test_evidence_budget.py +``` + +It must pass named cases demonstrating all of the following observable behaviour: + +1. With `P=4` and `P_solve=1`, a model-authored loop that starts 20 sequential or concurrent + `pred` processes receives exactly four admitted dynamic calls, at most one admitted solve, + and `budget_exhausted` for the rest; the authoritative ledger contains the same counts. +2. Running the real binary's absolute path from the agent shell fails with a permission error, + while the `pred` shim succeeds. This is the negative control proving `PATH` replacement is + not the only protection. +3. Invalid JSON, invalid CLI arguments, nonzero `pred`, and a model-caused `pred` timeout consume + admitted `P`; an injected gateway failure before admission consumes nothing. +4. Rule A can make two charged submissions and no third. Rule B still has two, proving the + allowance is per rule rather than a shared run-wide pool. Acceptance on Rule B's first attempt + closes Rule B and expires its second attempt. +5. Malformed, rejected, duplicate, and wrong-rule submissions consume `S`; an injected verifier + exception consumes nothing and returns `infrastructure_error`. +6. Replaying one completed request ID after simulating a lost response returns the identical + stored response without changing any counter. Two genuinely different concurrent request IDs + can never overspend a counter. +7. Status/help/version/cached inventory probes do not change any charged counter. + +As a regression check, also run: + +```bash +pytest -v benchmark/tests/test_submit_session.py benchmark/tests/test_verify.py +``` + +Existing certificate acceptance/rejection behaviour must remain green. A test fixture that makes +the real binary executable by the agent must make the absolute-path negative control fail, proving +that the acceptance suite detects a broken privilege boundary. + +### Out of scope + +- Top50 selection and episode orchestration; +- choosing the public numeric values of `M`, `E`, `P`, `P_solve`, or output ceilings; +- protecting the runner from a malicious trusted team member or cryptographically attesting a + remotely uploaded run; and +- changing `problem-reductions` or the `pred` CLI upstream. + +--- + +## I2 — Implement the Self-selected Top50 Runner + +### Background + +The existing runner launches one unlimited whole-repository conversation. Models can inspect +different numbers of rules, spend different amounts of target-system feedback on each one, and +carry discoveries and accumulated context throughout the session. The +[Top-50 design](top50-evidence-budget-benchmark.md) replaces that with a bounded static triage +followed by 50 fresh, equal-budget rule episodes. + +This issue owns the complete run workflow. A Top50 result is meaningful only if the shortlist is +frozen before dynamic experimentation, contains exactly 50 valid unique rules, and every rule +receives a clean context, clean scratch directory, and new Evidence Budget Service from I1. + +### Objective + +Replace the unlimited whole-repository Model API session with a two-phase `Top50Runner`: + +1. a bounded source-only triage conversation that atomically commits an ordered Top50 with + capped hypotheses; and +2. 50 sequential, isolated investigation episodes with identical non-transferable budgets. + +Automatically inject authoritative remaining counters into every observation. Preserve partial +audit artifacts after a crash, but mark any unrecoverable provider, gateway, judge, or container +failure as `run_error` so it cannot be ranked as a clean completion. + +### Interface (Input → Output) + +- **Input:** pinned repository and canonical rule inventory; standardized model client and + inference settings; triage contract `{T, E_t, O_t}`; per-rule contract + `{M, E, P, P_solve, S=2, O}`; and the I1 service factory. +- **Triage output:** immutable ordered JSON artifact containing exactly 50 unique canonical rule + IDs and optional size-capped hypotheses, plus its generation/action ledger. +- **Run output:** one Top50 run record containing the shortlist, triage ledger, exactly 50 ordered + episode records, usage metadata, completion status, and any classified infrastructure error. + +### Technical recommendations + +- Keep `mini-swe-agent`/LiteLLM as the sole rankable reference harness. Remove the forced + unlimited `step_limit=0` behaviour and account for usable responses and executed shell actions + explicitly. +- Provide triage with a runner-cached rule inventory and read-only source tree. Do not expose the + dynamic `pred` or `submit` shims until after the shortlist passes validation. +- Commit the shortlist through an evaluation-owned command or spool rather than scraping final + prose. Reject lists with the wrong count, duplicates, unknown IDs, oversized hypotheses, or a + second commit attempt. +- Start each rule with a fresh model message history, fresh scratch directory, and fresh I1 + service. Pass only the current rule ID and its capped triage hypothesis; transfer no generated + files or model-authored memory. +- Execute the 50 episodes sequentially in the first release. This avoids provider throttling and + host contention becoming rule-order-dependent. Because budgets are isolated, parallelism may be + revisited later without changing the public contract. +- Provider 429/5xx/disconnects before a usable response follow one frozen retry policy and do not + consume `T`/`M`. A usable but malformed model action consumes its generation. An executed shell + action consumes `E_t`/`E`, including nonzero exit and model-caused timeout. +- End a rule immediately after acceptance or budget exhaustion. Do not transfer unused allowance. + +### Verification + +Add a deterministic fake-model end-to-end acceptance suite and run: + +```bash +pytest -v benchmark/tests/test_top50_runner.py +``` + +The suite must pass named scenarios demonstrating: + +1. A fake triage response commits 50 known unique rule IDs; the runner then creates exactly 50 + episode records in the frozen order, each starting with identical full counters and an empty + conversation/scratch state. +2. The fake model writes a sentinel file and message in episode 1; episode 2 cannot observe + either. Spending all `pred` and submit allowance in episode 1 leaves episode 2's counters full. +3. Dynamic `pred` and `submit` attempts during triage are unavailable and do not produce target + feedback. A shortlist with 49 entries, 51 entries, a duplicate, an unknown rule, or an + oversized hypothesis is rejected before any investigation episode starts. These are the + negative controls for shortlist enforcement. +4. Every model observation contains authoritative `rule i/50`, generation, shell, `pred`, + `solve`, and `submit` usage/limit values matching the underlying ledger. +5. Acceptance in an episode stops that episode without a further model call. Exhausting any hard + counter ends only that rule and does not start an extra action. +6. An injected 429 before a usable response retries without consuming a model generation; a + malformed usable response consumes one. Exhausted infrastructure retries produce `run_error` + and an unrankable record rather than a clean 0-bug run. +7. The rankable execution path rejects `claude-code`, `codex`, a custom strategy file, or a custom + prompt config. The same adapters may remain callable in an explicitly unranked development + mode. + +Also run the existing no-API runner tests: + +```bash +pytest -v benchmark/tests/test_run_mini.py benchmark/tests/test_run_submission.py benchmark/tests/test_whole_repo.py +``` + +Update or replace whole-repository expectations so they assert the versioned historical path, +not the new rankable path. A fake model that tries to return a second shortlist or reuse one +episode's workspace must make the new suite fail. + +### Dependencies + +Depends on I1, the evaluation-owned Evidence Budget Service. + +### Out of scope + +- Fixed Top50 evaluation; +- multiple seeds or aggregate confidence intervals; +- parallel rule episodes; +- custom agent systems and coding-agent CLI ranking; and +- selecting final public budget values. + +--- + +## I3 — Define and enforce the vNext result and ranking contract + +### Background + +The current submission schema describes one whole-repository session with a shared +`submit_limit`, a flat `submit_log`, `agent_mode: whole-repo`, and token efficiency as a +tie-breaker. Those fields cannot prove that a valid Top50 was frozen or that 50 rules received +identical non-transferable budgets. + +The new private submission artifact must carry the complete evaluation-owned triage and episode +ledgers so the backend can recompute rankability and score. The public leaderboard must continue +to expose aggregate data only; rule identities, hypotheses, certificates, and trajectories remain +private answer-key material. + +### Objective + +Introduce a versioned Top50 submission and scoring contract, make the backend independently +validate its shortlist and all 50 episode ledgers, re-verify every accepted certificate, and rank +only clean Standardized Model Track results by distinct accepted rules. + +Historical unlimited whole-repository and coding-agent results must remain viewable under their +original contract but must never be mixed into the Top50 ranking. Equal verified bug counts remain +ties; elapsed time, token usage, cost, and bugs/Ktok are informational only. + +### Interface (Input → Output) + +- **Input:** I2 Top50 run record, evaluation-owned ledgers, model/provider and exposed inference + parameters, pinned repository/`pred`/runner identifiers, and accepted certificates. +- **Private output:** schema-valid versioned submission containing the complete budget vector, + ordered Top50, triage ledger, exactly 50 rule-bound episode ledgers, usage, errors, and + certificates. +- **Scored output:** backend-derived `VerifiedBugs@Top50[T,E_t,M,E,P,P_solve,S=2,O]`, Bugs@10/25/50, + first-versus-second-attempt acceptance, calls per accepted bug, cap-hit diagnostics, and + rankability verdict. +- **Public output:** aggregate leaderboard entry with model and contract provenance, headline + verified bug count, and non-ranking efficiency fields; no answer-key data. + +### Technical recommendations + +- Add an explicit benchmark-contract identifier instead of inferring semantics from optional + legacy fields. Keep a deliberate legacy parser for historical data rather than making the new + schema accept ambiguous mixtures. +- Validate that the shortlist has exactly 50 canonical unique IDs, episode order matches it, + every configured limit is identical, each usage count is within its limit, `P_solve <= P`, + `S=2` per rule, and request IDs/attempt numbering are internally consistent. +- Derive result rows only from evaluation-owned accepted submit records for the matching current + rule. Re-run `verify()` on every accepted certificate; ignore self-reported `bugs_found`. +- Treat missing episodes, infrastructure errors, CLI/custom-harness provenance, custom prompts, + or inconsistent ledgers as unrankable. Preserve the private artifact for debugging. +- Version the public board by benchmark contract. Remove token efficiency from sort keys for the + new contract while leaving old-board rendering intact. +- The operators are a trusted internal team. Do not add remote-run attestation, signatures, or + external-user authentication to this issue. + +### Verification + +Run the contract and scorer acceptance suites: + +```bash +pytest -v benchmark/tests/test_top50_submission.py benchmark/tests/test_verify_submission.py benchmark/tests/test_backend_score.py +``` + +They must use a schema-valid synthetic Top50 artifact and demonstrate: + +1. The valid artifact has exactly 50 frozen rules and 50 matching episode ledgers; the backend + ignores a deliberately false self-reported score and returns the distinct count implied by + accepted, re-verified certificates. +2. Two accepted certificates for one rule count once. Accepted rules at shortlist positions 7, + 18, and 41 produce Bugs@10 = 1, Bugs@25 = 2, and Bugs@50 = 3. +3. A certificate for a rule outside the shortlist, a certificate submitted from another rule's + episode, a third attempt, a shared `100`-attempt pool, duplicate/missing episode, over-limit + counter, changed budget, or forged request ordering makes the run unrankable. These are the + negative controls proving the scorer does not trust the envelope. +4. A pure `run_error` or unrecoverable infrastructure episode is preserved privately but excluded + from ranking. A rejected model submission remains a clean charged attempt and does not make the + run an infrastructure failure. +5. A `codex`/`claude-code` agent mode or custom prompt provenance is excluded from the Top50 board. + A canonical API harness run is accepted. +6. Public output contains no rule ID, hypothesis, certificate, source instance, target config, + submit reason, or trajectory. The aggregate guard fails if any such answer-key field is added. +7. For equal verified bug counts, changing tokens, cost, or elapsed time does not change rank + order; the UI represents the entries as tied. A larger verified count ranks first. +8. One legacy v0.6 whole-repository fixture remains readable in its historical view but cannot + enter or replace an entry in the Top50 board. + +Also run: + +```bash +python .github/scripts/check_aggregate.py site/results.json +``` + +against a generated development board. It must print no errors for the aggregate-only fixture; +injecting one `rule` or `certificate` field must make the guard exit nonzero. + +### Dependencies + +Depends on I2 so the contract validates the real `Top50Runner` artifact rather than a parallel +invented format. + +### Out of scope + +- cryptographic attestation or accepting untrusted external runs; +- multiple-seed aggregation, best-of-N, or confidence intervals; +- Fixed Top50 scores; +- ranking custom agent systems; and +- declaring unaccepted rules bug-free or reporting Top50 precision/recall. + +--- + +## I4 — Calibrate and publish the first frozen Top50 budget contract + +### Background + +I1–I3 make logical budgets enforceable and scoreable, but the public values for triage turns, +per-rule model turns, shell actions, `pred`, `solve`, and output ceilings should not be chosen by +intuition. The [Top-50 design](top50-evidence-budget-benchmark.md) calls for one internal +pre-release calibration over development data, followed by one frozen budget for all formal runs. + +The calibration grid is a development procedure, not a public multi-budget leaderboard and not a +multiple-seed requirement. The released benchmark runs each model once at the selected contract. + +### Objective + +Provide a reproducible internal calibration workflow, execute it on an explicitly non-ranking +development target, publish the measured trade-offs and chosen knee, then freeze those values +through configuration, prompts, preflight, schema, documentation, CI, and leaderboard display for +the next benchmark version. + +This issue is the release gate: after it lands, no rankable run can request unlimited or custom +budgets, and all user-facing documentation tells the same Self-selected Top50 story. + +### Interface (Input → Output) + +- **Input:** development target identifier; a small named set of internal model runs; candidate + contracts such as `M in {6,10,14}` and `P in {8,16,32}`; I1–I3 artifacts; and measured token, + cost, elapsed-time, cap-hit, retry, and accepted-bug data. +- **Calibration output:** machine-readable observations plus a checked-in Markdown report showing + yield, cap-hit rates, marginal gain, infrastructure errors, and the reason for selecting each + public limit. +- **Release output:** one immutable named budget contract referenced by runner defaults, prompt, + result schema, scorer, docs, preflight, image/version metadata, CI, and leaderboard. + +### Technical recommendations + +- Add a small analysis command that consumes completed development artifacts. Do not embed a + supposedly objective automatic knee detector; show the raw table/curves and record the team's + explicit selection rationale. +- Use an older library snapshot or mark every pilot artifact non-ranking. A pilot model set and + candidate grid may be small; do not introduce a repeated-seed requirement. +- Start from the design's candidate coordinates, then adjust only when cap-hit or failure data + shows that one counter binds pathologically. Record every tested contract, including failed + runs, instead of publishing only the winner. +- Replace current `step_limit: 0`, `SUBMIT_LIMIT=100`, whole-repository prompt language, custom + strategy hooks on the rankable path, and token-efficiency tie-breaking. Keep process/pred/model + watchdogs as fixed safety controls. +- Update `README.md`, `CONTRIBUTING.md`, `submission.env.example`, `benchmark/config.yaml`, runner + help, preflight, Docker image metadata, schema docs, and the public site together. +- After release, reconcile stale issues #2, #4, and #42 by linking the new implementation and + stating precisely which old recommendations were implemented, deferred, or dropped. Closing or + editing those issues remains a maintainer action after the implementation is merged. + +### Verification + +The implementation PR must include the calibration inputs/report and pass: + +```bash +pytest -v -m "not integration" +``` + +followed in the pinned runner image by: + +```bash +python -m benchmark.verify --calibrate +``` + +A reviewer must also be able to check the committed evidence offline in a few seconds: + +```bash +python -m benchmark.calibrate_budget \ + --check benchmark/docs/budget-calibration.json +``` + +It must print `PASS: calibration evidence matches ` and exit 0. Changing any +selected limit, deleting a tested candidate, or making the Markdown report disagree with the +machine-readable evidence must make the same command exit nonzero and identify the mismatch. + +The release evidence must additionally demonstrate: + +1. The calibration report lists every tested contract and, for each, model/target provenance, + verified bugs, cap-hit counts, usage, retries, infrastructure failures, token/cost/time + references, and marginal yield. The chosen values exactly match the checked-in named contract. +2. At least one smaller and one larger candidate surround the selected `M` and `P` values, so the + report demonstrates a trade-off rather than merely documenting a predetermined number. +3. Starting a rankable run with any unlimited counter, `S != 2`, custom budget, custom prompt, + custom strategy, or coding-agent backend fails preflight before a scored model generation. + These are the release negative controls. +4. A standard fake Top50 run completes with exactly 50 episodes and emits a schema-valid artifact + carrying the frozen contract ID. Changing one embedded limit makes backend validation fail. +5. A forced hung model call and a forced hung `pred` child are killed by watchdogs and recorded as + infrastructure/model outcomes according to the design; changing the watchdog duration does not + change the named logical budget or score formula. +6. The README and leaderboard describe the primary metric as verified distinct bugs at the frozen + Top50 evidence budget. They do not present time, cost, tokens, Fixed Top50, multiple seeds, or a + System Track as primary ranking dimensions. +7. Historical whole-repository results remain visible under their old contract and cannot be + compared or sorted into the new Top50 table. + +Before merging the release, run the repository's required CI and calibration workflows. A PR must +not be merged until CI is green and the user has explicitly approved the merge. + +### Dependencies + +Depends on I1, I2, and I3. + +### Out of scope + +- a public budget grid or bugs-versus-budget leaderboard; +- multiple seeds and confidence intervals; +- Fixed Top50 diagnostics; +- System Track/coding-agent ranking; +- external contributor submission; and +- choosing future benchmark versions' budgets without a new calibration and version bump. + +## Coverage scan + +| Design area | Covered by | +|---|---| +| Logical resource model and charging semantics | I1 | +| Privilege boundary and `pred` bypass prevention | I1 | +| Per-rule two-attempt submission judge | I1 | +| Static triage and atomic Top50 freeze | I2 | +| Fresh equal-budget rule episodes and counter visibility | I2 | +| Provider/infrastructure failure semantics | I1, I2, I3 | +| Versioned private result and backend re-verification | I3 | +| Headline score, diagnostics, privacy, and historical separation | I3 | +| Internal pilot and public budget freeze | I4 | +| Prompt, preflight, docs, CI, and release migration | I4 | +| Multiple seeds, Fixed Top50, System Track | explicitly out of scope in all relevant issues | + +The final scan found no uncovered first-release area from the source design. Cross-cutting +security, privacy, migration, observability, reliability, negative controls, and stale-issue +reconciliation are assigned above rather than split into additional issues. diff --git a/benchmark/env_setup.py b/benchmark/env_setup.py index aad77be..c21c6e5 100644 --- a/benchmark/env_setup.py +++ b/benchmark/env_setup.py @@ -51,14 +51,18 @@ def pinned_pred_version() -> str: def find_pred_binary() -> Path: - """Find pred binary in PATH.""" - pred_path = shutil.which("pred") + """Find the trusted pred binary from PRED_BINARY or PATH.""" + configured = os.environ.get("PRED_BINARY") + pred_path = configured or shutil.which("pred") if not pred_path: raise RuntimeError( "pred binary not found in PATH. " "Install with: cargo install --git https://github.com/CodingThrust/problem-reductions problemreductions-cli" ) - return Path(pred_path) + path = Path(pred_path).expanduser().resolve() + if not path.is_file() or not os.access(path, os.X_OK): + raise RuntimeError(f"pred binary is not executable: {path}") + return path def pred_version(binary: str | Path = "pred") -> str: diff --git a/benchmark/evidence_budget.py b/benchmark/evidence_budget.py new file mode 100644 index 0000000..12ccc44 --- /dev/null +++ b/benchmark/evidence_budget.py @@ -0,0 +1,690 @@ +"""Evaluation-owned logical budgets for one rule investigation episode. + +The model sees thin ``pred`` and ``submit`` commands in its scratch environment. The +authoritative counters, the real pred binary, and certificate verification remain in the +runner process. This module intentionally uses the same atomic file-spool shape as +``SubmissionSession`` because localhost sockets are unavailable in several supported +agent sandboxes. +""" +from __future__ import annotations + +import copy +import errno +import hashlib +import json +import os +import shlex +import stat +import sys +import threading +from concurrent.futures import Future, ThreadPoolExecutor +from contextlib import ExitStack +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Callable + +from benchmark.submit_session import SubmissionSession +from benchmark.process_control import ProcessLimits, run_capped_process, terminate_process_group +from benchmark.verify import (CPU_LIMIT_SECONDS, FSIZE_LIMIT_BYTES, MEM_LIMIT_BYTES, + Verdict, verify) + +MAX_REQUEST_BYTES = 2 * 1024 * 1024 +DEFAULT_PRED_TIMEOUT_SECONDS = 300 +DEFAULT_MAX_OUTPUT_CHARS = 10_000 +_DYNAMIC_COMMANDS = {"create", "reduce", "solve", "evaluate", "extract"} + + +@dataclass(frozen=True) +class EvidenceBudget: + """The non-transferable logical allowance for one selected rule.""" + + model_generations: int + shell_actions: int + pred_calls: int + solve_calls: int + submit_attempts: int = 2 + max_output_chars: int = DEFAULT_MAX_OUTPUT_CHARS + pred_timeout_seconds: int = DEFAULT_PRED_TIMEOUT_SECONDS + + def __post_init__(self) -> None: + for name, value in asdict(self).items(): + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ValueError(f"{name} must be a non-negative integer") + if self.solve_calls > self.pred_calls: + raise ValueError("solve_calls must be <= pred_calls") + if self.submit_attempts != 2: + raise ValueError("submit_attempts must be exactly 2 for the Top50 contract") + if self.max_output_chars == 0: + raise ValueError("max_output_chars must be > 0") + if self.pred_timeout_seconds == 0: + raise ValueError("pred_timeout_seconds must be > 0") + + +class EvidenceBudgetState: + """Atomically reserve, release, and report the counters for one episode.""" + + def __init__(self, budget: EvidenceBudget): + self.budget = budget + self._limits = { + "model_generations": budget.model_generations, + "shell_actions": budget.shell_actions, + "pred_calls": budget.pred_calls, + "solve_calls": budget.solve_calls, + } + self._used = {name: 0 for name in self._limits} + self._lock = threading.RLock() + + def reserve(self, *names: str) -> dict | None: + """Reserve all named counters or none of them.""" + with self._lock: + unknown = [name for name in names if name not in self._limits] + if unknown: + raise KeyError(f"unknown evidence counters: {unknown}") + if any(self._used[name] >= self._limits[name] for name in names): + return None + before = dict(self._used) + for name in names: + self._used[name] += 1 + return {"before": before, "after": dict(self._used)} + + def release(self, *names: str) -> None: + """Release an infrastructure-failed reservation.""" + with self._lock: + for name in names: + if self._used[name] <= 0: + raise RuntimeError(f"cannot release unreserved counter {name}") + for name in names: + self._used[name] -= 1 + + def status(self) -> dict: + with self._lock: + return { + name: { + "used": self._used[name], + "limit": limit, + "remaining": max(0, limit - self._used[name]), + } + for name, limit in self._limits.items() + } + + +class PredGatewaySession: + """Own the only agent-facing route to a real pred process.""" + + def __init__( + self, + *, + pred_binary: str | Path, + workdir: str | Path, + budget_state: EvidenceBudgetState, + oracle_uid: int | None = None, + oracle_gid: int | None = None, + oracle_extra_groups: tuple[int, ...] = (), + ): + if (oracle_uid is None) != (oracle_gid is None): + raise ValueError("oracle_uid and oracle_gid must be provided together") + self.pred_binary = Path(pred_binary).resolve() + self.workdir = Path(workdir).resolve() + self.budget_state = budget_state + self.oracle_uid = oracle_uid + self.oracle_gid = oracle_gid + self.oracle_extra_groups = oracle_extra_groups + self._ledger: list[dict] = [] + self._responses: dict[str, tuple[str, dict]] = {} + self._cache: dict[tuple[str, ...], dict] = {} + self._lock = threading.RLock() + self._thread: threading.Thread | None = None + self._executor: ThreadPoolExecutor | None = None + self._futures: set[Future] = set() + self._active_processes: set = set() + self._channel: Path | None = None + self._inbox_fd: int | None = None + self._processing_fd: int | None = None + self._outbox_fd: int | None = None + self._old_channel_env: str | None = None + self._stopping = threading.Event() + + @property + def ledger(self) -> list[dict]: + with self._lock: + return copy.deepcopy(self._ledger) + + @property + def channel(self) -> Path: + if self._channel is None: + raise RuntimeError("pred gateway is not active") + return self._channel + + def __enter__(self) -> "PredGatewaySession": + if not self.pred_binary.is_file() or not os.access(self.pred_binary, os.X_OK): + raise RuntimeError(f"pred gateway binary is unavailable: {self.pred_binary}") + + bin_dir = self.workdir.parent / ".prb-bin" + self._channel = self.workdir.parent / ".prb-pred" + inbox = self._channel / "inbox" + processing = self._channel / "processing" + outbox = self._channel / "outbox" + for directory in (bin_dir, inbox, processing, outbox): + directory.mkdir(parents=True, exist_ok=True) + + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + self._inbox_fd = os.open(inbox, directory_flags) + self._processing_fd = os.open(processing, directory_flags) + self._outbox_fd = os.open(outbox, directory_flags) + + shim = bin_dir / "pred" + package_root = Path(__file__).resolve().parent.parent + shim.write_text( + f"#!/bin/sh\nPYTHONPATH={shlex.quote(str(package_root))}${{PYTHONPATH:+:$PYTHONPATH}} " + f"exec {shlex.quote(sys.executable)} -m benchmark.agent_pred \"$@\"\n", + encoding="utf-8", + ) + shim.chmod(0o755) + + self._thread = threading.Thread(target=self._serve, name="prb-pred", daemon=True) + worker_count = max(8, min(64, self.budget_state.budget.pred_calls + 16)) + self._executor = ThreadPoolExecutor(max_workers=worker_count, + thread_name_prefix="prb-pred-call") + self._thread.start() + self._old_channel_env = os.environ.get("PRB_PRED_DIR") + os.environ["PRB_PRED_DIR"] = str(self._channel) + return self + + def __exit__(self, exc_type, exc, tb) -> None: + if self._old_channel_env is None: + os.environ.pop("PRB_PRED_DIR", None) + else: + os.environ["PRB_PRED_DIR"] = self._old_channel_env + self._stopping.set() + if self._thread is not None: + self._thread.join() + with self._lock: + active = list(self._active_processes) + for process in active: + terminate_process_group(process) + if self._executor is not None: + self._executor.shutdown(wait=True, cancel_futures=True) + for fd in (self._inbox_fd, self._processing_fd, self._outbox_fd): + if fd is not None: + os.close(fd) + + def prepare_agent_access(self, uid: int, gid: int) -> None: + """Give the unprivileged agent access only to request/response directories.""" + for name in ("inbox", "outbox"): + directory = self.channel / name + os.chown(directory, uid, gid) + directory.chmod(0o700) + + def _serve(self) -> None: + assert self._inbox_fd is not None + assert self._processing_fd is not None + idle_wait = 0.02 + while not self._stopping.is_set(): + handled = False + for name in os.listdir(self._inbox_fd): + if not _valid_request_filename(name): + continue + handled = True + try: + os.replace(name, name, src_dir_fd=self._inbox_fd, + dst_dir_fd=self._processing_fd) + except FileNotFoundError: + continue + request_id = name[:-5] + assert self._executor is not None + future = self._executor.submit(self._process_request_file, name, request_id) + with self._lock: + self._futures.add(future) + future.add_done_callback(self._forget_future) + if handled: + idle_wait = 0.02 + else: + self._stopping.wait(idle_wait) + idle_wait = min(idle_wait * 2, 0.5) + + def _forget_future(self, future: Future) -> None: + with self._lock: + self._futures.discard(future) + + def _process_request_file(self, name: str, request_id: str) -> None: + assert self._processing_fd is not None + raw = b"" + try: + raw = _read_request_bytes(self._processing_fd, name) + request = json.loads(raw.decode("utf-8")) + if not isinstance(request, dict) or request.get("request_id") != request_id: + raise ValueError("request id does not match filename") + response = self._handle_idempotent(request_id, request) + except Exception as error: + response = self._handle_invalid_envelope(request_id, raw, error) + self._write_response(request_id, response) + try: + os.unlink(name, dir_fd=self._processing_fd) + except FileNotFoundError: + pass + + def _handle_idempotent(self, request_id: str, request: dict) -> dict: + fingerprint = json.dumps(request, sort_keys=True, separators=(",", ":")) + with self._lock: + old = self._responses.get(request_id) + if old is not None: + old_fingerprint, old_response = old + if old_fingerprint != fingerprint: + return { + "infrastructure_error": True, + "reason": "request id was reused with a different payload", + "returncode": 2, + "stdout": "", + "stderr": "pred gateway request-id collision\n", + "budget": self.budget_state.status(), + } + return copy.deepcopy(old_response) + + response = self._handle(request_id, request) + with self._lock: + self._responses[request_id] = (fingerprint, copy.deepcopy(response)) + return response + + def _handle_invalid_envelope(self, request_id: str, raw: bytes, error: Exception) -> dict: + fingerprint = f"invalid:{hashlib.sha256(raw).hexdigest()}" + with self._lock: + old = self._responses.get(request_id) + if old is not None: + old_fingerprint, old_response = old + if old_fingerprint != fingerprint: + return { + "infrastructure_error": True, + "reason": "request id was reused with a different payload", + "returncode": 2, + "stdout": "", + "stderr": "pred gateway request-id collision\n", + "budget": self.budget_state.status(), + } + return copy.deepcopy(old_response) + response = self._handle_invalid_args(request_id) + response["reason"] = f"invalid pred gateway request: {type(error).__name__}: {error}" + with self._lock: + self._responses[request_id] = (fingerprint, copy.deepcopy(response)) + return response + + def _handle(self, request_id: str, request: dict) -> dict: + op = request.get("op") + if op == "status": + return {"status": "ok", "returncode": 0, "stdout": "", "stderr": "", + "budget": self.budget_state.status()} + if op != "pred": + return {"returncode": 2, "stdout": "", "stderr": "unknown pred gateway operation\n", + "reason": "unknown operation", "budget": self.budget_state.status()} + + args = request.get("args") + if (not isinstance(args, list) or not all(isinstance(arg, str) for arg in args) + or sum(len(arg) for arg in args) > MAX_REQUEST_BYTES): + return self._handle_invalid_args(request_id) + cwd = self._validated_cwd(request.get("cwd")) + command = _command_name(args) + free = _is_free_command(args, command) + cache_key = tuple(args) if free else None + with self._lock: + if cache_key in self._cache: + return copy.deepcopy(self._cache[cache_key]) + + counters = () if free else (("pred_calls", "solve_calls") + if command == "solve" else ("pred_calls",)) + reservation = self.budget_state.reserve(*counters) + if counters and reservation is None: + response = { + "returncode": 75, + "stdout": "", + "stderr": "pred evidence budget exhausted\n", + "exhausted": True, + "budget": self.budget_state.status(), + } + self._append_pred_record(request_id, args, command, False, "budget_exhausted", + response["budget"]) + return response + + record_index = self._append_pred_record( + request_id, args, command, bool(counters), "running", self.budget_state.status()) + try: + result = self._run_pred(args, cwd) + except OSError as error: + model_caused = error.errno == errno.E2BIG + if counters and not model_caused: + self.budget_state.release(*counters) + response = { + "returncode": 2, + "stdout": "", + "stderr": (f"pred invocation rejected: {error}\n" if model_caused else + f"pred gateway infrastructure error: {error}\n"), + "infrastructure_error": not model_caused, + "reason": f"{type(error).__name__}: {error}", + "budget": self.budget_state.status(), + } + self._finish_pred_record( + record_index, "model_error" if model_caused else "infrastructure_error", response) + return response + + response = {**result, "budget": self.budget_state.status()} + outcome = "timeout" if result.get("timed_out") else ( + "completed" if result["returncode"] == 0 else "nonzero_exit") + self._finish_pred_record(record_index, outcome, response) + if cache_key is not None and result["returncode"] == 0: + with self._lock: + self._cache[cache_key] = copy.deepcopy(response) + return response + + def _handle_invalid_args(self, request_id: str) -> dict: + reservation = self.budget_state.reserve("pred_calls") + if reservation is None: + return {"returncode": 75, "stdout": "", + "stderr": "pred evidence budget exhausted\n", "exhausted": True, + "budget": self.budget_state.status()} + response = { + "returncode": 2, + "stdout": "", + "stderr": "pred request arguments must be a bounded list of strings\n", + "model_error": True, + "budget": self.budget_state.status(), + } + index = self._append_pred_record( + request_id, [], None, True, "model_error", response["budget"]) + self._finish_pred_record(index, "model_error", response) + return response + + def _validated_cwd(self, raw: object) -> Path: + cwd = Path(raw) if isinstance(raw, str) and raw else self.workdir + cwd = cwd.resolve() + if not cwd.is_relative_to(self.workdir): + # The action is model-authored, so keep it inside the episode scratch area. + return self.workdir + return cwd + + def _run_pred(self, args: list[str], cwd: Path) -> dict: + result = run_capped_process( + [str(self.pred_binary), *args], + shell=False, + cwd=cwd, + env=None, + timeout=self.budget_state.budget.pred_timeout_seconds, + max_output_chars=self.budget_state.budget.max_output_chars, + uid=self.oracle_uid, + gid=self.oracle_gid, + extra_groups=self.oracle_extra_groups, + limits=ProcessLimits( + cpu_seconds=CPU_LIMIT_SECONDS, + memory_bytes=MEM_LIMIT_BYTES, + file_bytes=FSIZE_LIMIT_BYTES, + ), + on_start=self._track_process, + on_finish=self._untrack_process, + ) + stderr = result.stderr + if result.timed_out: + stderr += ("\n" if stderr and not stderr.endswith("\n") else "") + stderr += "pred process timed out\n" + return { + "returncode": result.returncode, + "stdout": result.stdout, + "stderr": stderr, + "timed_out": result.timed_out, + } + + def _track_process(self, process) -> None: + with self._lock: + self._active_processes.add(process) + + def _untrack_process(self, process) -> None: + with self._lock: + self._active_processes.discard(process) + + def _append_pred_record(self, request_id: str, args: list[str], command: str | None, + charged: bool, outcome: str, budget: dict) -> int: + with self._lock: + self._ledger.append({ + "sequence": len(self._ledger) + 1, + "request_id": request_id, + "args": copy.deepcopy(args), + "command": command, + "charged": charged, + "outcome": outcome, + "budget": copy.deepcopy(budget), + }) + return len(self._ledger) - 1 + + def _finish_pred_record(self, index: int, outcome: str, response: dict) -> None: + with self._lock: + self._ledger[index]["outcome"] = outcome + self._ledger[index]["returncode"] = response["returncode"] + self._ledger[index]["budget"] = copy.deepcopy(response["budget"]) + + def _write_response(self, request_id: str, response: dict) -> None: + assert self._outbox_fd is not None + destination = f"{request_id}.json" + temporary = f".{request_id}.{threading.get_ident()}.tmp" + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + response_fd = os.open(temporary, flags, 0o644, dir_fd=self._outbox_fd) + try: + with os.fdopen(response_fd, "w", encoding="utf-8", closefd=False) as handle: + json.dump(response, handle) + handle.flush() + finally: + os.close(response_fd) + os.replace(temporary, destination, src_dir_fd=self._outbox_fd, + dst_dir_fd=self._outbox_fd) + + +class EvidenceBudgetSession: + """Compose the pred gateway and a two-attempt, rule-scoped submit judge.""" + + def __init__( + self, + *, + rule: str, + budget: EvidenceBudget, + pred_binary: str | Path, + verifier: Callable[[dict], Verdict] = verify, + agent_uid: int | None = None, + agent_gid: int | None = None, + oracle_uid: int | None = None, + oracle_gid: int | None = None, + evidence_gid: int | None = None, + ): + if not rule: + raise ValueError("rule must be non-empty") + if (agent_uid is None) != (agent_gid is None): + raise ValueError("agent_uid and agent_gid must be provided together") + if (oracle_uid is None) != (oracle_gid is None): + raise ValueError("oracle_uid and oracle_gid must be provided together") + if evidence_gid is not None and (agent_uid is None or oracle_uid is None): + raise ValueError("evidence_gid requires both agent and oracle identities") + self.rule = rule + self.budget = budget + self.state = EvidenceBudgetState(budget) + self.pred_binary = Path(pred_binary) + self.verifier = verifier + self.agent_uid = agent_uid + self.agent_gid = agent_gid + self.oracle_uid = oracle_uid + self.oracle_gid = oracle_gid + self.evidence_gid = evidence_gid + self.submit: SubmissionSession | None = None + self.pred: PredGatewaySession | None = None + self._scratch: Path | None = None + self._stack: ExitStack | None = None + self._event_lock = threading.RLock() + self._model_events: list[dict] = [] + self._shell_events: list[dict] = [] + + @property + def workdir(self) -> Path: + if self._scratch is None: + raise RuntimeError("evidence budget session is not active") + return self._scratch + + def __enter__(self) -> "EvidenceBudgetSession": + stack = ExitStack() + try: + self.submit = stack.enter_context(SubmissionSession( + limit=self.budget.submit_attempts, + verifier=self.verifier, + expected_rule=self.rule, + )) + self._scratch = self.submit.workdir / "scratch" + self._scratch.mkdir(mode=0o700) + self.pred = stack.enter_context(PredGatewaySession( + pred_binary=self.pred_binary, + workdir=self._scratch, + budget_state=self.state, + oracle_uid=self.oracle_uid, + oracle_gid=self.oracle_gid, + oracle_extra_groups=((self.evidence_gid,) if self.evidence_gid is not None else ()), + )) + if self.agent_uid is not None and self.agent_gid is not None: + self._prepare_agent_access(self.agent_uid, self.agent_gid) + self._stack = stack + return self + except BaseException: + stack.__exit__(*sys.exc_info()) + raise + + def __exit__(self, exc_type, exc, tb) -> None: + if self._stack is not None: + self._stack.__exit__(exc_type, exc, tb) + + def _prepare_agent_access(self, uid: int, gid: int) -> None: + assert self.submit is not None and self.pred is not None and self._scratch is not None + scratch_gid = self.evidence_gid if self.evidence_gid is not None else gid + os.chown(self._scratch, uid, scratch_gid) + self._scratch.chmod(0o2770 if self.evidence_gid is not None else 0o700) + self.submit.prepare_agent_access(uid, gid) + self.pred.prepare_agent_access(uid, gid) + + def admit_shell_action(self, command: str) -> bool: + """Atomically debit one model-authored shell action and append its audit event.""" + reservation = self.state.reserve("shell_actions") + with self._event_lock: + self._shell_events.append({ + "sequence": len(self._shell_events) + 1, + "command": command, + "charged": reservation is not None, + "outcome": "admitted" if reservation is not None else "budget_exhausted", + "budget": self.state.status(), + }) + return reservation is not None + + def record_model_generation(self, *, outcome: str = "completed", + infrastructure_error: bool = False) -> bool: + """Record a provider generation; infrastructure failures do not debit the model.""" + reservation = None if infrastructure_error else self.state.reserve("model_generations") + admitted = infrastructure_error or reservation is not None + with self._event_lock: + self._model_events.append({ + "sequence": len(self._model_events) + 1, + "charged": reservation is not None, + "outcome": outcome if admitted else "budget_exhausted", + "budget": self.state.status(), + }) + return admitted + + def status(self) -> dict: + if self.submit is None: + raise RuntimeError("evidence budget session is not active") + return { + **self.state.status(), + "submit_attempts": { + "used": self.submit.used, + "limit": self.submit.limit, + "remaining": self.submit.remaining, + }, + } + + def ledger(self) -> dict: + if self.submit is None or self.pred is None: + raise RuntimeError("evidence budget session is not active") + with self._event_lock: + model_events = copy.deepcopy(self._model_events) + shell_events = copy.deepcopy(self._shell_events) + return { + "rule": self.rule, + "budget": asdict(self.budget), + "status": self.status(), + "pred": self.pred.ledger, + "submit": self.submit.attempts, + "model_generations": model_events, + "shell_actions": shell_events, + } + + +def _valid_request_filename(name: str) -> bool: + return (len(name) == 37 and name.endswith(".json") + and all(char in "0123456789abcdef" for char in name[:-5])) + + +def _read_request_bytes(directory_fd: int, name: str) -> bytes: + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + request_fd = os.open(name, flags, dir_fd=directory_fd) + try: + info = os.fstat(request_fd) + if not stat.S_ISREG(info.st_mode): + raise ValueError("request is not a regular file") + if info.st_size > MAX_REQUEST_BYTES: + raise ValueError(f"request exceeds {MAX_REQUEST_BYTES} bytes") + with os.fdopen(request_fd, "rb", closefd=False) as request_file: + raw = request_file.read(MAX_REQUEST_BYTES + 1) + finally: + os.close(request_fd) + if len(raw) > MAX_REQUEST_BYTES: + raise ValueError(f"request exceeds {MAX_REQUEST_BYTES} bytes") + return raw + + +def _command_name(args: list[str]) -> str | None: + skip_value = False + for arg in args: + if skip_value: + skip_value = False + continue + if arg in {"-o", "--output"}: + skip_value = True + continue + if arg.startswith("--output=") or (arg.startswith("-o") and arg != "-o"): + continue + if arg in {"-q", "--quiet", "--json"}: + continue + if arg in _DYNAMIC_COMMANDS or arg in {"list", "help"}: + return arg + if not arg.startswith("-"): + return arg + return None + + +def _is_free_command(args: list[str], command: str | None) -> bool: + if not args: + return True # invoking pred without args only displays help + if any(arg in {"--help", "-h", "--version", "-V"} for arg in args): + return True + if command == "help": + return True + return command == "list" and "--rules" in args + + +def _drain_capped(stream, limit: int, destination: dict[str, str], key: str) -> None: + """Drain a child pipe completely while retaining at most ``limit`` characters.""" + chunks: list[str] = [] + kept = 0 + total = 0 + while True: + chunk = stream.read(64 * 1024) + if not chunk: + break + total += len(chunk) + if kept < limit: + piece = chunk[:limit - kept] + chunks.append(piece) + kept += len(piece) + text = "".join(chunks) + if total > kept: + marker = f"\n... {total - kept} characters elided ...\n" + prefix_limit = max(0, limit - len(marker)) + text = text[:prefix_limit] + marker[:limit] + destination[key] = text diff --git a/benchmark/preflight.py b/benchmark/preflight.py index 4f712f7..d48e6d4 100644 --- a/benchmark/preflight.py +++ b/benchmark/preflight.py @@ -16,7 +16,7 @@ """ from __future__ import annotations -from benchmark.env_setup import verify_pred_version +from benchmark.env_setup import find_pred_binary, verify_pred_version from benchmark.run_mini import DEFAULT_MAX_TOKENS, _build_model, list_rules from benchmark.usage import usage_from_response @@ -39,7 +39,7 @@ def run_checks( # 1. pred binary + version (no API). try: - ver = verify_pred_version("pred") + ver = verify_pred_version(find_pred_binary()) results.append(("pred binary", True, f"version {ver}")) except Exception as e: results.append(("pred binary", False, str(e))) diff --git a/benchmark/privilege_smoke.py b/benchmark/privilege_smoke.py new file mode 100644 index 0000000..ef8ccb8 --- /dev/null +++ b/benchmark/privilege_smoke.py @@ -0,0 +1,79 @@ +"""Build-time smoke check for the runner/oracle privilege boundary.""" +from __future__ import annotations + +import os +import shlex + +from benchmark.agent_environment import run_as_agent +from benchmark.env_setup import find_pred_binary +from benchmark.evidence_budget import EvidenceBudget, EvidenceBudgetSession +from benchmark.verify import Verdict + + +def main() -> None: + uid = int(os.environ["PRB_AGENT_UID"]) + gid = int(os.environ["PRB_AGENT_GID"]) + oracle_uid = int(os.environ["PRB_ORACLE_UID"]) + oracle_gid = int(os.environ["PRB_ORACLE_GID"]) + evidence_gid = int(os.environ["PRB_EVIDENCE_GID"]) + pred_binary = find_pred_binary() + budget = EvidenceBudget( + model_generations=1, + shell_actions=2, + pred_calls=1, + solve_calls=0, + submit_attempts=2, + ) + with EvidenceBudgetSession( + rule="privilege-smoke", + budget=budget, + pred_binary=pred_binary, + verifier=lambda cert: Verdict(False, "not used"), + agent_uid=uid, + agent_gid=gid, + oracle_uid=oracle_uid, + oracle_gid=oracle_gid, + evidence_gid=evidence_gid, + ) as session: + env = dict(os.environ) + through_gateway = run_as_agent( + "pred --version", + cwd=str(session.workdir), + env=env, + timeout=10, + uid=uid, + gid=gid, + extra_groups=(evidence_gid,), + ) + if through_gateway.returncode != 0: + raise SystemExit(f"pred gateway failed for agent: {through_gateway.stdout}") + + direct = run_as_agent( + f"{shlex.quote(str(pred_binary))} --version", + cwd=str(session.workdir), + env=env, + timeout=10, + uid=uid, + gid=gid, + extra_groups=(evidence_gid,), + ) + if direct.returncode == 0: + raise SystemExit("agent unexpectedly executed the real pred oracle") + + submit_status = run_as_agent( + "submit --status", + cwd=str(session.workdir), + env=env, + timeout=10, + uid=uid, + gid=gid, + extra_groups=(evidence_gid,), + ) + if submit_status.returncode != 0: + raise SystemExit(f"submit gateway failed for agent: {submit_status.stdout}") + + print("PASS: unprivileged agent can use gateways but cannot execute the pred oracle") + + +if __name__ == "__main__": + main() diff --git a/benchmark/process_control.py b/benchmark/process_control.py new file mode 100644 index 0000000..76b55d9 --- /dev/null +++ b/benchmark/process_control.py @@ -0,0 +1,173 @@ +"""Bounded subprocess execution shared by agent actions and pred gateway calls.""" +from __future__ import annotations + +import os +import signal +import subprocess +import threading +from dataclasses import dataclass +from typing import Callable, Sequence + +try: # POSIX-only; official rankable runs use Linux. + import resource +except ImportError: # pragma: no cover - Windows is not a rankable platform + resource = None + + +@dataclass(frozen=True) +class ProcessLimits: + cpu_seconds: int + memory_bytes: int + file_bytes: int + processes: int = 64 + + +@dataclass(frozen=True) +class ProcessResult: + returncode: int + stdout: str + stderr: str + timed_out: bool + + +class _CombinedCapture: + def __init__(self, limit: int): + self.limit = limit + self.remaining = limit + self.total = 0 + self.totals = {"stdout": 0, "stderr": 0} + self.parts = {"stdout": [], "stderr": []} + self._lock = threading.Lock() + + def drain(self, stream, key: str) -> None: + while True: + chunk = stream.read(64 * 1024) + if not chunk: + break + with self._lock: + self.total += len(chunk) + self.totals[key] += len(chunk) + if self.remaining: + piece = chunk[:self.remaining] + self.parts[key].append(piece) + self.remaining -= len(piece) + + def result(self) -> tuple[str, str]: + stdout = "".join(self.parts["stdout"]) + stderr = "".join(self.parts["stderr"]) + retained = len(stdout) + len(stderr) + if self.total > retained: + marker = f"\n... {self.total - retained} characters elided ...\n" + keep = max(0, self.limit - len(marker)) + if self.totals["stdout"] > len(stdout): + stdout = stdout[:keep] + marker[:self.limit - min(len(stdout), keep)] + stderr = "" + else: + stdout = stdout[:keep] + stderr_keep = max(0, keep - len(stdout)) + stderr = stderr[:stderr_keep] + marker[:self.limit - len(stdout) - stderr_keep] + return stdout, stderr + + +def run_capped_process( + command: str | Sequence[str], + *, + shell: bool, + cwd: str, + env: dict[str, str] | None, + timeout: int, + max_output_chars: int, + combine_stderr: bool = False, + uid: int | None = None, + gid: int | None = None, + extra_groups: Sequence[int] = (), + limits: ProcessLimits | None = None, + on_start: Callable[[subprocess.Popen], None] | None = None, + on_finish: Callable[[subprocess.Popen], None] | None = None, +) -> ProcessResult: + """Run one process with a combined observation cap and kill its whole group on timeout.""" + privilege_args = {} + if uid is not None or gid is not None: + if uid is None or gid is None: + raise ValueError("uid and gid must be provided together") + if os.name != "posix": + raise RuntimeError("rankable privilege separation requires POSIX") + if os.geteuid() == 0: + privilege_args = {"user": uid, "group": gid, "extra_groups": list(extra_groups)} + elif uid != os.getuid() or gid != os.getgid(): + raise PermissionError("runner must be root to launch a different identity") + + process = subprocess.Popen( + command, + shell=shell, + text=True, + cwd=cwd, + env=env, + encoding="utf-8", + errors="replace", + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT if combine_stderr else subprocess.PIPE, + start_new_session=True, + **privilege_args, + ) + if on_start is not None: + on_start(process) + try: + _apply_limits(process.pid, limits) + capture = _CombinedCapture(max_output_chars) + assert process.stdout is not None + readers = [threading.Thread(target=capture.drain, args=(process.stdout, "stdout"), + daemon=True)] + if not combine_stderr: + assert process.stderr is not None + readers.append(threading.Thread(target=capture.drain, + args=(process.stderr, "stderr"), daemon=True)) + for reader in readers: + reader.start() + + timed_out = False + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + terminate_process_group(process) + process.wait() + for reader in readers: + reader.join() + stdout, stderr = capture.result() + return ProcessResult( + returncode=124 if timed_out else process.returncode, + stdout=stdout, + stderr=stderr, + timed_out=timed_out, + ) + finally: + if on_finish is not None: + on_finish(process) + + +def terminate_process_group(process: subprocess.Popen) -> None: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + +def _apply_limits(pid: int, limits: ProcessLimits | None) -> None: + if limits is None or resource is None or not hasattr(resource, "prlimit"): + return + for name, value in ( + ("RLIMIT_CPU", limits.cpu_seconds), + ("RLIMIT_AS", limits.memory_bytes), + ("RLIMIT_FSIZE", limits.file_bytes), + ("RLIMIT_NPROC", limits.processes), + ): + resource_id = getattr(resource, name, None) + if resource_id is None: + continue + try: + resource.prlimit(pid, resource_id, (value, value)) + except (PermissionError, ProcessLookupError, ValueError, OSError): + # macOS lacks prlimit and some kernels reject individual limits. Official + # Linux runs additionally carry fixed container-level resource limits. + continue diff --git a/benchmark/submit_session.py b/benchmark/submit_session.py index aa1dffe..47659db 100644 --- a/benchmark/submit_session.py +++ b/benchmark/submit_session.py @@ -9,6 +9,7 @@ from __future__ import annotations import copy +import hashlib import json import os import shutil @@ -26,14 +27,24 @@ class SubmissionSession: - """Own one run-wide submit budget and its append-only in-memory ledger.""" + """Own one submit budget and its append-only in-memory ledger. - def __init__(self, limit: int = 100, verifier: Callable[[dict], Verdict] = verify): + Historical callers may still use a run-wide limit. Evidence-budget episodes pass an + ``expected_rule`` and create one independent two-attempt session per rule. + """ + + def __init__(self, limit: int = 100, verifier: Callable[[dict], Verdict] = verify, + expected_rule: str | None = None): if limit < 0: raise ValueError("submit limit must be >= 0") + if expected_rule is not None and not expected_rule: + raise ValueError("expected_rule must be non-empty when provided") self.limit = limit self._verifier = verifier + self.expected_rule = expected_rule self._attempts: list[dict] = [] + self._responses: dict[str, tuple[str, dict]] = {} + self._closed = False self._lock = threading.RLock() self._thread: threading.Thread | None = None self._tmpdir: Path | None = None @@ -56,8 +67,15 @@ def used(self) -> int: @property def remaining(self) -> int: + if self.closed: + return 0 return max(0, self.limit - self.used) + @property + def closed(self) -> bool: + with self._lock: + return self._closed + @property def attempts(self) -> list[dict]: with self._lock: @@ -95,9 +113,10 @@ def __enter__(self) -> "SubmissionSession": self._artifact_fd = os.open(artifact_dir, directory_flags) shim = bin_dir / "submit" - client = Path(__file__).with_name("agent_submit.py") + package_root = Path(__file__).resolve().parent.parent shim.write_text( - f"#!/bin/sh\nexec {shlex.quote(sys.executable)} {shlex.quote(str(client))} \"$@\"\n", + f"#!/bin/sh\nPYTHONPATH={shlex.quote(str(package_root))}${{PYTHONPATH:+:$PYTHONPATH}} " + f"exec {shlex.quote(sys.executable)} -m benchmark.agent_submit \"$@\"\n", encoding="utf-8", ) shim.chmod(0o755) @@ -129,7 +148,7 @@ def __exit__(self, exc_type, exc, tb) -> None: self._stopping.set() if self._thread is not None: - self._thread.join(timeout=1) + self._thread.join() for fd in (self._workdir_fd, self._inbox_fd, self._processing_fd, self._outbox_fd, self._artifact_fd): if fd is not None: @@ -171,6 +190,21 @@ def preserve_artifacts(self, destination: str | Path) -> list[Path]: copied.append(target) return copied + def prepare_agent_access(self, uid: int, gid: int) -> None: + """Give a non-root agent access to its request queues and artifact directory.""" + if self._tmpdir is None: + raise RuntimeError("submission session is not active") + self._tmpdir.chmod(0o711) + (self._tmpdir / "work").chmod(0o711) + paths = ( + self._tmpdir / "work" / ".prb-submit" / "inbox", + self._tmpdir / "work" / ".prb-submit" / "outbox", + self._tmpdir / "work" / "artifacts", + ) + for path in paths: + os.chown(path, uid, gid) + path.chmod(0o700) + def result_rows(self) -> list[dict]: """Collapse attempts to one authoritative result per named rule. @@ -208,6 +242,7 @@ def _serve(self) -> None: except FileNotFoundError: continue request_id = name[:-5] + raw = b"" try: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) request_fd = os.open(name, flags, dir_fd=self._processing_fd) @@ -226,9 +261,9 @@ def _serve(self) -> None: request = json.loads(raw.decode("utf-8")) if request.get("request_id") != request_id: raise ValueError("request id does not match filename") - response = self._handle(request) + response = self._handle_idempotent(request_id, request) except Exception as e: # malformed requests do not kill the service - response = self._record_rejection(f"invalid request: {e}") + response = self._handle_malformed_idempotent(request_id, raw, e) self._write_response(request_id, response) try: os.unlink(name, dir_fd=self._processing_fd) @@ -246,7 +281,8 @@ def _write_response(self, request_id: str, response: dict) -> None: temporary = f".{request_id}.{threading.get_ident()}.tmp" flags = (os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)) - response_fd = os.open(temporary, flags, 0o600, dir_fd=self._outbox_fd) + # The service may run as root while the agent client is deliberately unprivileged. + response_fd = os.open(temporary, flags, 0o644, dir_fd=self._outbox_fd) try: with os.fdopen(response_fd, "w", encoding="utf-8", closefd=False) as response_file: json.dump(response, response_file) @@ -256,12 +292,49 @@ def _write_response(self, request_id: str, response: dict) -> None: os.replace(temporary, destination, src_dir_fd=self._outbox_fd, dst_dir_fd=self._outbox_fd) - def _handle(self, request: dict) -> dict: + def _handle_idempotent(self, request_id: str, request: dict) -> dict: + fingerprint = json.dumps(request, sort_keys=True, separators=(",", ":")) + with self._lock: + old = self._responses.get(request_id) + if old is not None: + old_fingerprint, old_response = old + if old_fingerprint != fingerprint: + return {"accepted": False, "infrastructure_error": True, + "reason": "request id was reused with a different payload", + "used": self.used, "limit": self.limit, + "remaining": self.remaining} + return copy.deepcopy(old_response) + response = self._handle(request, request_id=request_id) + with self._lock: + self._responses[request_id] = (fingerprint, copy.deepcopy(response)) + return response + + def _handle_malformed_idempotent( + self, request_id: str, raw: bytes, error: Exception + ) -> dict: + fingerprint = f"invalid:{hashlib.sha256(raw).hexdigest()}" + with self._lock: + old = self._responses.get(request_id) + if old is not None: + old_fingerprint, old_response = old + if old_fingerprint != fingerprint: + return {"accepted": False, "infrastructure_error": True, + "reason": "request id was reused with a different payload", + "used": self.used, "limit": self.limit, + "remaining": self.remaining} + return copy.deepcopy(old_response) + response = self._record_rejection( + f"invalid request: {type(error).__name__}: {error}", request_id=request_id) + self._responses[request_id] = (fingerprint, copy.deepcopy(response)) + return response + + def _handle(self, request: dict, *, request_id: str | None = None) -> dict: if request.get("op") == "status": with self._lock: self._status_checks += 1 return {"status": "ok", "used": self.used, "limit": self.limit, - "remaining": self.remaining} + "remaining": self.remaining, "closed": self.closed, + "expected_rule": self.expected_rule} if request.get("op") != "submit": return {"accepted": False, "error": "unknown operation", "used": self.used, "limit": self.limit, "remaining": self.remaining} @@ -269,6 +342,10 @@ def _handle(self, request: dict) -> dict: # Verification is serialized under this lock. The budget check + append is atomic, # including if a future runner enables parallel agent sessions. with self._lock: + if self._closed: + return {"accepted": False, "exhausted": True, "closed": True, + "reason": f"rule episode already accepted: {self.expected_rule}", + "used": len(self._attempts), "limit": self.limit, "remaining": 0} if len(self._attempts) >= self.limit: return {"accepted": False, "exhausted": True, "reason": f"submit budget exhausted ({self.limit}/{self.limit})", @@ -290,25 +367,42 @@ def _handle(self, request: dict) -> dict: else: if not isinstance(cert, dict): reason = "certificate must be a JSON object" + elif (self.expected_rule is not None + and cert.get("rule") != self.expected_rule): + reason = (f"certificate rule {cert.get('rule')!r} does not match " + f"this episode's rule {self.expected_rule!r}") else: try: verdict = self._verifier(cert) accepted = bool(verdict.accepted) reason = verdict.reason details = verdict.details - except Exception as e: # one bad certificate must not kill the budget server - reason = f"verifier error: {type(e).__name__}: {e}" + except Exception as e: # infrastructure errors never debit the model + return { + "accepted": False, + "infrastructure_error": True, + "reason": f"verifier error: {type(e).__name__}: {e}", + "used": len(self._attempts), + "limit": self.limit, + "remaining": self.remaining, + } - return self._append_attempt(cert, accepted, reason, details) + return self._append_attempt(cert, accepted, reason, details, + request_id=request_id) - def _record_rejection(self, reason: str) -> dict: + def _record_rejection(self, reason: str, *, request_id: str | None = None) -> dict: with self._lock: + if self._closed: + return {"accepted": False, "exhausted": True, "closed": True, + "reason": f"rule episode already accepted: {self.expected_rule}", + "used": len(self._attempts), "limit": self.limit, "remaining": 0} if len(self._attempts) >= self.limit: return {"accepted": False, "exhausted": True, "reason": reason, "used": self.limit, "limit": self.limit, "remaining": 0} - return self._append_attempt(None, False, reason) + return self._append_attempt(None, False, reason, request_id=request_id) - def _append_attempt(self, cert, accepted: bool, reason: str, details=None) -> dict: + def _append_attempt(self, cert, accepted: bool, reason: str, details=None, + request_id: str | None = None) -> dict: attempt_no = len(self._attempts) + 1 record = { "attempt": attempt_no, @@ -317,11 +411,16 @@ def _append_attempt(self, cert, accepted: bool, reason: str, details=None) -> di "reason": reason, "certificate": cert, } + if request_id is not None: + record["request_id"] = request_id if details is not None: record["verify_details"] = details self._attempts.append(record) + if accepted and self.expected_rule is not None: + self._closed = True return {**record, "used": attempt_no, "limit": self.limit, - "remaining": self.limit - attempt_no} + "remaining": 0 if self._closed else self.limit - attempt_no, + "closed": self._closed} def _attempt_to_row(attempt: dict) -> dict: diff --git a/benchmark/tests/test_env_setup.py b/benchmark/tests/test_env_setup.py index 725906d..0001e06 100644 --- a/benchmark/tests/test_env_setup.py +++ b/benchmark/tests/test_env_setup.py @@ -17,7 +17,7 @@ import subprocess import pytest from pathlib import Path -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock from benchmark.env_setup import ( PINNED_COMMIT, @@ -137,6 +137,7 @@ def test_pred_not_in_path_raises(self, monkeypatch): def test_pred_found_returns_path(self, monkeypatch, tmp_path): fake_pred = tmp_path / "pred" fake_pred.touch() + fake_pred.chmod(0o755) monkeypatch.setattr("shutil.which", lambda name: str(fake_pred)) result = find_pred_binary() assert result == fake_pred diff --git a/benchmark/tests/test_evidence_budget.py b/benchmark/tests/test_evidence_budget.py new file mode 100644 index 0000000..16ec992 --- /dev/null +++ b/benchmark/tests/test_evidence_budget.py @@ -0,0 +1,341 @@ +"""Acceptance tests for the evaluation-owned per-rule evidence budget.""" +from __future__ import annotations + +import json +import os +import subprocess +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from benchmark.agent_pred import _request as pred_request +from benchmark.agent_submit import _request as submit_request +from benchmark.agent_environment import run_as_agent +from benchmark.evidence_budget import EvidenceBudget, EvidenceBudgetSession +from benchmark.verify import Verdict + + +def _fake_pred(tmp_path: Path) -> Path: + script = tmp_path / "real-pred" + script.write_text( + "#!/bin/sh\n" + "case \"$1\" in\n" + " --version) echo 'pred 0.0.0'; exit 0 ;;\n" + " --help) echo 'help'; exit 0 ;;\n" + " list) echo '[\"r1\",\"r2\"]'; exit 0 ;;\n" + " fail) echo 'bad args' >&2; exit 7 ;;\n" + " sleep) sleep 2; exit 0 ;;\n" + " spam) python3 -c 'print(\"x\" * 5000)'; exit 0 ;;\n" + " *) echo \"ran:$*\"; exit 0 ;;\n" + "esac\n", + encoding="utf-8", + ) + script.chmod(0o700) + return script + + +def _budget(*, pred=4, solve=1, timeout=1) -> EvidenceBudget: + return EvidenceBudget( + model_generations=10, + shell_actions=10, + pred_calls=pred, + solve_calls=solve, + submit_attempts=2, + max_output_chars=1024, + pred_timeout_seconds=timeout, + ) + + +def _cert(rule: str) -> dict: + return {"rule": rule, "source": {"type": "Example"}, + "bundle": {"target": {"type": "Target"}}} + + +def _run(*args: str, cwd: Path) -> subprocess.CompletedProcess[str]: + return subprocess.run(list(args), cwd=cwd, capture_output=True, text=True) + + +def _raw_spool(channel: Path, request_id: str, raw: bytes) -> dict: + request = channel / "inbox" / f"{request_id}.json" + response = channel / "outbox" / f"{request_id}.json" + request.write_bytes(raw) + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + try: + payload = json.loads(response.read_text()) + except FileNotFoundError: + time.sleep(0.01) + continue + response.unlink() + return payload + raise AssertionError("spool service did not respond") + + +def test_pred_and_solve_caps_apply_to_real_invocations(tmp_path): + with EvidenceBudgetSession( + rule="r1", budget=_budget(), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + assert _run("pred", "solve", "input.json", cwd=session.workdir).returncode == 0 + assert _run("pred", "solve", "input.json", cwd=session.workdir).returncode == 75 + results = [_run("pred", "create", "X", cwd=session.workdir) for _ in range(5)] + + assert sum(result.returncode == 0 for result in results) == 3 + assert sum(result.returncode == 75 for result in results) == 2 + status = session.status() + assert status["pred_calls"]["used"] == 4 + assert status["solve_calls"]["used"] == 1 + assert sum(record["charged"] for record in session.pred.ledger) == 4 + + +def test_global_output_option_cannot_bypass_solve_budget(tmp_path): + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=2, solve=1), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + first = _run("pred", "-o", "out.json", "solve", "input.json", cwd=session.workdir) + second = _run("pred", "--output=other.json", "solve", "input.json", + cwd=session.workdir) + + assert first.returncode == 0 + assert second.returncode == 75 + assert session.status()["pred_calls"]["used"] == 1 + assert session.status()["solve_calls"]["used"] == 1 + + +def test_concurrent_pred_clients_cannot_overspend(tmp_path): + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=4, solve=0), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + with ThreadPoolExecutor(max_workers=20) as pool: + results = list(pool.map( + lambda _: _run("pred", "create", "X", cwd=session.workdir), range(20))) + + assert sum(result.returncode == 0 for result in results) == 4 + assert sum(result.returncode == 75 for result in results) == 16 + assert session.status()["pred_calls"] == {"used": 4, "limit": 4, "remaining": 0} + + +def test_free_commands_are_cached_and_do_not_charge(tmp_path): + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=1, solve=0), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + assert _run("pred", "--budget-status", cwd=session.workdir).returncode == 0 + assert _run("pred", "--version", cwd=session.workdir).returncode == 0 + assert _run("pred", "--version", cwd=session.workdir).returncode == 0 + assert _run("pred", "list", "--rules", "--json", cwd=session.workdir).returncode == 0 + assert _run("pred", "list", "--rules", "--json", cwd=session.workdir).returncode == 0 + + assert session.status()["pred_calls"]["used"] == 0 + # Cache hits return without another real process or ledger entry. + assert len(session.pred.ledger) == 2 + + +def test_nonzero_exit_and_timeout_consume_pred_budget(tmp_path): + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=2, solve=0), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + failed = _run("pred", "fail", cwd=session.workdir) + timed_out = _run("pred", "sleep", cwd=session.workdir) + exhausted = _run("pred", "create", "X", cwd=session.workdir) + + assert failed.returncode == 7 + assert timed_out.returncode == 124 and "timed out" in timed_out.stderr + assert exhausted.returncode == 75 + assert session.status()["pred_calls"]["used"] == 2 + assert [record["outcome"] for record in session.pred.ledger[:2]] == [ + "nonzero_exit", "timeout"] + + +def test_gateway_infrastructure_failure_releases_reservation(tmp_path): + real_pred = _fake_pred(tmp_path) + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=1, solve=0), pred_binary=real_pred, + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + real_pred.unlink() # force spawn failure after the gateway health check + result = _run("pred", "create", "X", cwd=session.workdir) + assert result.returncode == 2 + assert "infrastructure error" in result.stderr + assert session.status()["pred_calls"]["used"] == 0 + assert session.pred.ledger[0]["outcome"] == "infrastructure_error" + + +def test_pred_output_is_drained_but_observation_is_capped(tmp_path): + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=1, solve=0), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + result = _run("pred", "spam", cwd=session.workdir) + assert result.returncode == 0 + assert len(result.stdout) <= 1024 + assert "characters elided" in result.stdout + + +def test_submit_budget_is_two_per_rule_and_acceptance_closes_episode(tmp_path): + pred = _fake_pred(tmp_path) + rejected = tmp_path / "rejected.json" + accepted = tmp_path / "accepted.json" + rejected.write_text(json.dumps(_cert("r1"))) + accepted.write_text(json.dumps(_cert("r2"))) + + with EvidenceBudgetSession( + rule="r1", budget=_budget(), pred_binary=pred, + verifier=lambda cert: Verdict(False, "no bug"), + ) as first: + assert _run("submit", str(rejected), cwd=first.workdir).returncode == 1 + assert _run("submit", str(rejected), cwd=first.workdir).returncode == 1 + assert _run("submit", str(rejected), cwd=first.workdir).returncode == 2 + assert first.status()["submit_attempts"] == {"used": 2, "limit": 2, "remaining": 0} + + with EvidenceBudgetSession( + rule="r2", budget=_budget(), pred_binary=pred, + verifier=lambda cert: Verdict(True, "confirmed"), + ) as second: + assert second.status()["submit_attempts"]["remaining"] == 2 + assert _run("submit", str(accepted), cwd=second.workdir).returncode == 0 + closed = _run("submit", str(accepted), cwd=second.workdir) + assert closed.returncode == 2 and "BUDGET_EXHAUSTED" in closed.stderr + assert second.submit.used == 1 + assert second.submit.remaining == 0 + + +def test_wrong_rule_and_malformed_submissions_consume(tmp_path): + malformed = tmp_path / "bad.json" + wrong = tmp_path / "wrong.json" + malformed.write_text("{bad") + wrong.write_text(json.dumps(_cert("r2"))) + with EvidenceBudgetSession( + rule="r1", budget=_budget(), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(True, "confirmed"), + ) as session: + assert _run("submit", str(malformed), cwd=session.workdir).returncode == 1 + result = _run("submit", str(wrong), cwd=session.workdir) + assert result.returncode == 1 and "does not match" in result.stderr + assert session.submit.used == 2 + + +def test_verifier_infrastructure_error_does_not_consume(tmp_path): + certificate = tmp_path / "certificate.json" + certificate.write_text(json.dumps(_cert("r1"))) + + def broken_verifier(cert): + raise RuntimeError("oracle unavailable") + + with EvidenceBudgetSession( + rule="r1", budget=_budget(), pred_binary=_fake_pred(tmp_path), + verifier=broken_verifier, + ) as session: + result = _run("submit", str(certificate), cwd=session.workdir) + assert result.returncode == 2 + assert "INFRASTRUCTURE_ERROR" in result.stderr + assert session.submit.used == 0 + assert session.submit.remaining == 2 + + +def test_pred_request_replay_is_idempotent(tmp_path, monkeypatch): + request_id = "a" * 32 + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=1, solve=0), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + monkeypatch.chdir(session.workdir) + first = pred_request({"op": "pred", "args": ["create", "X"], + "cwd": str(session.workdir)}, request_id) + second = pred_request({"op": "pred", "args": ["create", "X"], + "cwd": str(session.workdir)}, request_id) + assert first == second + assert session.status()["pred_calls"]["used"] == 1 + assert len(session.pred.ledger) == 1 + + +def test_invalid_pred_envelope_is_idempotent_and_charged(tmp_path): + request_id = "c" * 32 + with EvidenceBudgetSession( + rule="r1", budget=_budget(pred=1, solve=0), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + first = _raw_spool(session.pred.channel, request_id, b"{bad") + second = _raw_spool(session.pred.channel, request_id, b"{bad") + + assert first == second + assert first["model_error"] is True + assert session.status()["pred_calls"]["used"] == 1 + assert len(session.pred.ledger) == 1 + + +def test_submit_request_replay_is_idempotent(tmp_path): + request_id = "b" * 32 + payload = {"op": "submit", "certificate_text": json.dumps(_cert("r1"))} + with EvidenceBudgetSession( + rule="r1", budget=_budget(), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + first = submit_request(payload, request_id) + second = submit_request(payload, request_id) + assert first == second + assert session.submit.used == 1 + assert len(session.submit.attempts) == 1 + + +def test_malformed_submit_envelope_is_idempotent(tmp_path): + request_id = "d" * 32 + with EvidenceBudgetSession( + rule="r1", budget=_budget(), pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + channel = Path(os.environ["PRB_SUBMIT_DIR"]) + first = _raw_spool(channel, request_id, b"{bad") + second = _raw_spool(channel, request_id, b"{bad") + + assert first == second + assert session.submit.used == 1 + assert session.submit.attempts[0]["request_id"] == request_id + + +def test_model_and_shell_budgets_have_auditable_events(tmp_path): + budget = EvidenceBudget(1, 1, 0, 0, submit_attempts=2) + with EvidenceBudgetSession( + rule="r1", budget=budget, pred_binary=_fake_pred(tmp_path), + verifier=lambda cert: Verdict(False, "no bug"), + ) as session: + assert session.record_model_generation(outcome="completed") is True + assert session.record_model_generation(outcome="completed") is False + assert session.record_model_generation( + outcome="provider_error", infrastructure_error=True) is True + assert session.admit_shell_action("pwd") is True + assert session.admit_shell_action("pwd") is False + + ledger = session.ledger() + assert [event["charged"] for event in ledger["model_generations"]] == [True, False, False] + assert [event["charged"] for event in ledger["shell_actions"]] == [True, False] + assert session.status()["model_generations"]["used"] == 1 + assert session.status()["shell_actions"]["used"] == 1 + + +def test_agent_shell_output_has_one_combined_cap(tmp_path): + result = run_as_agent( + "python3 -c 'import sys; print(\"x\" * 5000); print(\"y\" * 5000, file=sys.stderr)'", + cwd=str(tmp_path), + env=dict(os.environ), + timeout=5, + uid=os.getuid(), + gid=os.getgid(), + max_output_chars=512, + ) + + assert result.returncode == 0 + assert len(result.stdout) <= 512 + assert "characters elided" in result.stdout + + +def test_budget_contract_rejects_invalid_values(): + with pytest.raises(ValueError, match="exactly 2"): + EvidenceBudget(1, 1, 1, 1, submit_attempts=100) + with pytest.raises(ValueError, match="solve_calls"): + EvidenceBudget(1, 1, 1, 2) diff --git a/benchmark/tests/test_preflight.py b/benchmark/tests/test_preflight.py index 405ac3c..a6b89a5 100644 --- a/benchmark/tests/test_preflight.py +++ b/benchmark/tests/test_preflight.py @@ -35,6 +35,7 @@ def _calculate_cost(self, response): def _patch(monkeypatch, *, ver="0.6.0", rules=("a", "b"), model=None, build_exc=None): + monkeypatch.setattr(pf, "find_pred_binary", lambda: "/fake/pred") monkeypatch.setattr(pf, "verify_pred_version", lambda *a, **k: ver) monkeypatch.setattr(pf, "list_rules", lambda repo: list(rules)) diff --git a/docker/Dockerfile b/docker/Dockerfile index 5c84bf6..0639dee 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -46,7 +46,7 @@ RUN CARGO_PROFILE_RELEASE_LTO=off \ FROM python:3.12-slim-bookworm AS runtime # Runtime libs pred may dynamically link (HiGHS uses OpenMP / libstdc++) — ref-independent, # so keep before ARG to preserve the cache across ref bumps. -RUN apt-get update && apt-get install -y --no-install-recommends libgomp1 libstdc++6 \ +RUN apt-get update && apt-get install -y --no-install-recommends libgomp1 libstdc++6 passwd \ && rm -rf /var/lib/apt/lists/* ARG PR_REF LABEL org.opencontainers.image.title="problem-reductions-benchmark" \ @@ -54,18 +54,37 @@ LABEL org.opencontainers.image.title="problem-reductions-benchmark" \ org.opencontainers.image.licenses="MIT" \ problem_reductions_ref="${PR_REF}" -COPY --from=pred-build /src/target/release/pred /usr/local/bin/pred -ENV PRED_BINARY=pred PYTHONUNBUFFERED=1 +RUN groupadd --gid 10003 prb-evidence \ + && groupadd --gid 10001 prb-agent \ + && groupadd --gid 10002 prb-oracle \ + && useradd --uid 10001 --gid 10001 --no-create-home --shell /usr/sbin/nologin prb-agent \ + && useradd --uid 10002 --gid 10002 --no-create-home --shell /usr/sbin/nologin prb-oracle \ + && usermod -a -G prb-evidence prb-agent \ + && usermod -a -G prb-evidence prb-oracle \ + && install -d -o root -g prb-oracle -m 0750 /usr/local/libexec/prb +COPY --from=pred-build /src/target/release/pred /usr/local/libexec/prb/pred +RUN chown root:prb-oracle /usr/local/libexec/prb/pred \ + && chmod 0750 /usr/local/libexec/prb/pred +ENV PRED_BINARY=/usr/local/libexec/prb/pred \ + PRB_AGENT_UID=10001 \ + PRB_AGENT_GID=10001 \ + PRB_ORACLE_UID=10002 \ + PRB_ORACLE_GID=10002 \ + PRB_EVIDENCE_GID=10003 \ + PYTHONUNBUFFERED=1 WORKDIR /app COPY benchmark/ ./benchmark/ # Fail the build if pred can't run (catches a missing runtime lib early). -RUN pred --version +RUN /usr/local/libexec/prb/pred --version # Bake the target version pins so benchmark/env_setup.py reads the *built* commit/version # rather than a frozen Python literal (the target tracks PR_REF, not a constant). COPY --from=pred-build /src/PINNED_COMMIT /app/benchmark/PINNED_COMMIT -RUN pred --version | grep -oE '[0-9]+\.[0-9]+\.[0-9]+' | head -n1 > /app/benchmark/PINNED_VERSION +RUN /usr/local/libexec/prb/pred --version | grep -oE '[0-9]+\.[0-9]+\.[0-9]+' | head -n1 > /app/benchmark/PINNED_VERSION +# Prove the complete privilege boundary, not just a file mode: the dedicated agent user can +# call both shims while an absolute-path oracle invocation is denied. +RUN python -m benchmark.privilege_smoke # Default: run the verifier calibration (fixtures) as a self-test. # Override CMD to verify a certificate, e.g. (run with NO network — the verifier only needs