feat: enforce per-rule evidence budgets

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)

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()

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")

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,67 +30,46 @@ 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
         if not args.certificate:
             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__":

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 ────────────────────────────────────────────────────────────