From 316d2b0bfc79666beaa4f8dff1318b70f06bcd6a Mon Sep 17 00:00:00 2001 From: Xiwei Pan Date: Mon, 6 Jul 2026 23:16:36 +0800 Subject: [PATCH 1/2] feat(runner): zero-trust cost re-metering + crash-safe whole-repo output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Meter cost from token usage × declared price (not a self-reported total), and make whole-repo runs durable so an early-stop/crash keeps the found bugs. - cost: submissions carry usage_totals (4 token buckets) + prices; the backend recomputes total_cost_usd/efficiency from them and ignores the self-reported figure (mirrors the zero-trust bug re-verification). whole-repo now surfaces its session-level 4-bucket usage (was dropped). - whole-repo durability: the agent appends each certificate to TRAJECTORY_DIR/certs.txt the moment it finds it; the trajectory is persisted every step (output_path); the runner harvests certs from BOTH the trajectory and the disk log (deduped). - fix trajectory loss: output_path was unset and trajectory_dir was never passed, so the --rm'd container dropped everything. Both are now wired, with TRAJECTORY_DIR exposed as a config knob. - versioned output: submission.json stays the stable "latest" pointer; a submission--.json archive is written beside it so runs don't clobber each other. - envelope self-describes: adds usage_totals, prices, agent_mode. - docs (submission.env.example, skill field table + SKILL.md), schemas, and tests updated. Co-Authored-By: Claude Opus 4.8 (1M context) --- .claude/skills/run-benchmark/SKILL.md | 8 +- .../references/env-and-troubleshoot.md | 6 +- benchmark/config_repo.yaml | 17 +++- benchmark/cost.py | 40 ++++++++ benchmark/results.schema.json | 28 +++++- benchmark/run_mini.py | 41 +++++++- benchmark/run_submission.py | 96 +++++++++++++++++-- benchmark/submission.schema.json | 25 ++++- benchmark/tests/test_cost.py | 28 ++++++ benchmark/tests/test_run_submission.py | 39 ++++++++ benchmark/tests/test_verify_submission.py | 31 ++++++ benchmark/tests/test_whole_repo.py | 25 +++++ benchmark/verify_submission.py | 29 +++++- submission.env.example | 9 ++ 14 files changed, 394 insertions(+), 28 deletions(-) diff --git a/.claude/skills/run-benchmark/SKILL.md b/.claude/skills/run-benchmark/SKILL.md index b7b2944..da0ba0b 100644 --- a/.claude/skills/run-benchmark/SKILL.md +++ b/.claude/skills/run-benchmark/SKILL.md @@ -89,7 +89,13 @@ must keep `BUDGET_USD=20` and omit `MAX_RULES`. per rule with the budget split evenly; `whole-repo` runs ONE session over the whole library and lets the agent enumerate and triage the rules itself under a single budget. Both produce the same `out/submission.json` and are scored identically — set `AGENT_MODE=whole-repo` to try -it. (`MAX_RULES` only applies to `per-rule`.) +it. (`MAX_RULES` only applies to `per-rule`.) In `whole-repo`, the agent also writes each +certificate to `TRAJECTORY_DIR/certs.txt` (default `/out`) as it finds it, and the trajectory +is persisted every step — so an early-stop/crash still leaves the found bugs on disk. + +**Output is versioned.** `out/submission.json` is the stable "latest" pointer; every run ALSO +writes a versioned archive `out/submission--.json` beside it, so runs don't +overwrite each other. **Confirm the experiment parameters with the user — don't silently default them.** These shape the result and the spend, so state the resolved set and get an explicit OK before diff --git a/.claude/skills/run-benchmark/references/env-and-troubleshoot.md b/.claude/skills/run-benchmark/references/env-and-troubleshoot.md index a6ab610..744286a 100644 --- a/.claude/skills/run-benchmark/references/env-and-troubleshoot.md +++ b/.claude/skills/run-benchmark/references/env-and-troubleshoot.md @@ -25,12 +25,14 @@ these as env vars (CLI flags would override, but the skill uses the env-file). | `PER_RULE_BUDGET` | 0.5 | per-rule cost cap | | `SAFETY_MARGIN` | 1.0 | USD held back so the budget-crossing call stays under cap | | `MAX_TOKENS` | 8192 | per-call output ceiling | -| `MAX_RULES` | all | cap rules attempted — **smoke runs only**; omit for a ranked run | +| `MAX_RULES` | all | cap rules attempted — **smoke runs only**; omit for a ranked run (per-rule only) | +| `AGENT_MODE` | `per-rule` | `per-rule` (isolated session/rule, budget split evenly) or `whole-repo` (ONE session, the agent triages the rules itself) | +| `TRAJECTORY_DIR` | `OUTPUT`'s dir (`/out`) | where **whole-repo** persists the trajectory + the durable incremental cert log (`certs.txt`); the agent writes each certificate here the moment it finds it, so an early-stop/crash still leaves the found bugs on disk | | `AGENT_CONFIG` / `AGENT_STRATEGY_FILE` | bundled | bring-your-own prompt; the files must be **mounted** into the container (`-v "$PWD/cfg:/cfg"`) and the path given as a container path | | `SUBMITTED_BY` | — | your handle, recorded in the envelope | | `EXPECTED_PRED_VERSION` / `EXPECTED_PRED_COMMIT` | baked | debugging only; `EXPECTED_PRED_VERSION=""` disables the version check | -`REPO_DIR` and `OUTPUT` are container-internal and already baked — don't set them. +`OUTPUT` (default `/out/submission.json`) is the stable "latest" pointer `prb submit` reads; every run **also** writes a versioned archive `submission--.json` beside it, so history isn't clobbered. `REPO_DIR` is container-internal and baked — don't set it. Non-standard endpoint example: ```ini diff --git a/benchmark/config_repo.yaml b/benchmark/config_repo.yaml index 03f5145..258cc64 100644 --- a/benchmark/config_repo.yaml +++ b/benchmark/config_repo.yaml @@ -59,6 +59,16 @@ agent: } CERTIFICATE_END + THEN, immediately, append the SAME block to the durable log so the bug survives even if + the run is cut off (budget/crash) — one command, append only, never overwrite: + cat >> {{certs_file}} <<'CERT' + CERTIFICATE_START + { ...the identical certificate JSON... } + CERTIFICATE_END + CERT + Do this every time, the moment you find a bug, before moving to the next rule. The file + is your safety net: anything written there is recoverable even if you never reach submit. + Only `rule`, `source`, and the target type (from `bundle`) are required; `target_config` is optional. The verifier re-derives everything from `source` with pred and never trusts your claim — a wrong or non-minimal certificate is simply rejected, so report only what @@ -74,9 +84,10 @@ agent: Enumerate the rules with `pred list --rules --json`, triage by suspicion, and test the round-trip correctness of each with pred. Emit a CERTIFICATE_START...CERTIFICATE_END - block for every bug — many are expected; do not stop after the first. You have a limited - budget — be efficient and spend it on the most suspicious rules first. When done, run: - echo COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT + block for every bug — many are expected; do not stop after the first — AND append the + same block to {{certs_file}} as you go, so no bug is lost if the run stops early. You + have a limited budget — be efficient and spend it on the most suspicious rules first. + When done, run: echo COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT step_limit: 300 diff --git a/benchmark/cost.py b/benchmark/cost.py index 9795280..465a90f 100644 --- a/benchmark/cost.py +++ b/benchmark/cost.py @@ -109,3 +109,43 @@ def extract_usage(messages: list) -> Usage: if resp is not None and _get(resp, "usage", None) is not None: total = total + usage_from_response(_get(resp, "usage", None)) return total + + +# ── (de)serialization: the 4-bucket token totals + the declared price snapshot ────── +# These travel in the submission so the backend can RE-METER cost as tokens × price +# (zero-trust, mirroring the bug re-verification) instead of trusting a self-reported +# dollar total. Tokens are the reproducible primitive; the declared price (dated by the +# submission's created_at) is a snapshot anyone can swap to recompute under other prices. + +def usage_as_dict(u: Usage) -> dict: + """Serialize a Usage to the 4-bucket dict shape used on rows and the envelope.""" + return {"input": u.input_tokens, "output": u.output_tokens, + "cache_read": u.cache_read_tokens, "cache_write": u.cache_write_tokens} + + +def usage_from_dict(d) -> Usage: + """Parse a 4-bucket dict (or None/missing → all zeros) back into a Usage.""" + return Usage( + input_tokens=int(_get(d, "input")), + output_tokens=int(_get(d, "output")), + cache_read_tokens=int(_get(d, "cache_read")), + cache_write_tokens=int(_get(d, "cache_write")), + ) + + +def price_as_dict(p: Price) -> dict: + """Serialize a Price to a plain dict (USD per 1M tokens, per bucket).""" + return {"input": p.input, "output": p.output, + "cache_read": p.cache_read, "cache_write": p.cache_write} + + +def price_from_dict(d) -> Price | None: + """Parse a price dict back into a Price, or None if absent (legacy submission).""" + if not d: + return None + return Price( + input=float(_get(d, "input")), + output=float(_get(d, "output")), + cache_read=float(_get(d, "cache_read")), + cache_write=float(_get(d, "cache_write")), + ) diff --git a/benchmark/results.schema.json b/benchmark/results.schema.json index 2fba61f..d0f6cc3 100644 --- a/benchmark/results.schema.json +++ b/benchmark/results.schema.json @@ -19,15 +19,39 @@ "minimum": 0, "description": "Number of verified, novel bugs found" }, + "test": { + "type": "boolean", + "description": "Carried through from the submission: scored + stored privately but excluded from the public leaderboard." + }, "total_cost_usd": { "type": "number", "minimum": 0, - "description": "Total spend in USD" + "description": "Total spend in USD — re-metered by the backend as usage_totals × prices (falls back to the self-reported figure only for legacy submissions without those)." }, "total_tokens_k": { "type": "number", "minimum": 0, - "description": "Total tokens used (in thousands)" + "description": "Total tokens used (in thousands); the sum of usage_totals / 1000." + }, + "usage_totals": { + "type": ["object", "null"], + "description": "Aggregate token usage in the four billed buckets (input/output/cache_read/cache_write) the cost was metered from. The reproducible primitive — lets any reader recompute cost under other prices.", + "properties": { + "input": {"type": "integer", "minimum": 0}, + "output": {"type": "integer", "minimum": 0}, + "cache_read": {"type": "integer", "minimum": 0}, + "cache_write": {"type": "integer", "minimum": 0} + } + }, + "prices": { + "type": ["object", "null"], + "description": "Declared per-token price snapshot (USD per 1M tokens) the cost was metered from; dated by the submission's created_at. null for legacy/FAKE submissions.", + "properties": { + "input": {"type": "number", "minimum": 0}, + "output": {"type": "number", "minimum": 0}, + "cache_read": {"type": "number", "minimum": 0}, + "cache_write": {"type": "number", "minimum": 0} + } }, "efficiency_bugs_per_ktok": { "type": "number", diff --git a/benchmark/run_mini.py b/benchmark/run_mini.py index 5f384e3..b067d55 100644 --- a/benchmark/run_mini.py +++ b/benchmark/run_mini.py @@ -253,6 +253,7 @@ def run_repo_session( *, api_base: str | None = None, trajectory_dir: Path | None = None, + certs_path: Path | None = None, price: Price | None = None, max_tokens: int = DEFAULT_MAX_TOKENS, step_limit: int | None = None, @@ -266,6 +267,12 @@ def run_repo_session( per bug. Returns ``{"rows": [...], "cost": float, "tokens_k": float}`` — one result row per distinct emitted certificate, each re-verified with pred and carrying the shared session trajectory for provenance. Contrast with ``run_one`` (one isolated rule/session). + + Durability: the agent is prompted to append each certificate to ``certs_path`` the moment + it finds it (the {{certs_file}} slot), so bugs survive an early stop; certificates are + harvested from BOTH the trajectory and that file (deduped). ``output_path`` is pointed at + ``trajectory_dir`` so mini-swe-agent persists the full trajectory to disk after every + step (crash-proof) rather than only at the end. """ from minisweagent.agents.default import DefaultAgent from minisweagent.environments.local import LocalEnvironment @@ -280,6 +287,21 @@ def run_repo_session( if step_limit is not None: agent_cfg["step_limit"] = step_limit + safe_model = model_name.replace("/", "_").replace(":", "_") + # Per-step trajectory persistence (mini-swe-agent saves config.output_path after every + # step): point it at trajectory_dir so a mid-run crash still leaves the latest trajectory + # on disk. Was previously unset (None → nothing written). + if trajectory_dir is not None: + Path(trajectory_dir).mkdir(parents=True, exist_ok=True) + agent_cfg["output_path"] = str(Path(trajectory_dir) / f"{safe_model}_whole-repo.step.json") + + # The durable incremental cert log the prompt appends to — start each run from empty so it + # only ever holds THIS run's bugs. + if certs_path is not None: + certs_path = Path(certs_path) + certs_path.parent.mkdir(parents=True, exist_ok=True) + certs_path.write_text("", encoding="utf-8") + agent = DefaultAgent( _build_model(model_name, api_base, max_tokens, price, model_kwargs=model_kwargs, api_key=api_key), @@ -291,19 +313,28 @@ def run_repo_session( "commit_hash": ctx.commit_hash[:7], "cost_limit": cost_limit, "strategy": strategy, + "certs_file": str(certs_path) if certs_path is not None else "/tmp/certs.txt", } agent.run(task="find-bugs") - cost, tokens_k, _usage = _session_cost(agent, price) + cost, tokens_k, usage = _session_cost(agent, price) trajectory = _trajectory(agent) if trajectory_dir is not None: - safe_model = model_name.replace("/", "_").replace(":", "_") save_trajectory(agent.messages, Path(trajectory_dir) / f"{safe_model}_whole-repo.jsonl") - rows = _rows_from_certificates(parse_all_certificates(agent.messages)) + # Harvest from the trajectory AND the durable incremental log (parse_all_certificates + # dedups by rule+source across both), so a cert the agent wrote to disk still counts even + # if it never made it into the parsed messages. + sources = list(agent.messages) + if certs_path is not None and certs_path.exists(): + sources.append({"content": certs_path.read_text(encoding="utf-8")}) + rows = _rows_from_certificates(parse_all_certificates(sources)) # The whole session is ONE trajectory shared by every bug — return it once for the - # envelope (build_submission) instead of copying it onto each row. - return {"rows": rows, "cost": cost, "tokens_k": tokens_k, "trajectory": trajectory} + # envelope (build_submission) instead of copying it onto each row. ``usage`` is the + # session-level 4-bucket token total (per-rule rows carry their own; whole-repo rows + # don't), so the envelope can re-price it — see build_submission. + return {"rows": rows, "cost": cost, "tokens_k": tokens_k, "trajectory": trajectory, + "usage": usage} def _rows_from_certificates(certs: list[dict]) -> list[dict]: diff --git a/benchmark/run_submission.py b/benchmark/run_submission.py index 68dae70..4aac465 100644 --- a/benchmark/run_submission.py +++ b/benchmark/run_submission.py @@ -26,9 +26,11 @@ import argparse import json import os +import re import tempfile from pathlib import Path +from benchmark.cost import Usage, price_as_dict, usage_as_dict, usage_from_dict from benchmark.env_context import EnvContext from benchmark.env_setup import find_pred_binary, pinned_commit, verify_pred_version from benchmark.run_mini import list_rules @@ -53,6 +55,9 @@ def build_submission( total_tokens_k: float | None = None, trajectory: list[dict] | None = None, pred_version: str = "", + price=None, + usage_totals=None, + agent_mode: str | None = None, ) -> dict: """Assemble the submission envelope from the runner's result rows. @@ -60,14 +65,35 @@ def build_submission( don't count as "reached"). For per-rule that is the rules attempted; for whole-repo it is the distinct rules the agent emitted a certificate for — a floor, since rules the agent probed but found clean aren't represented as rows. ``bugs_found`` is distinct - rules with a confirmed bug. ``total_cost_usd`` / ``total_tokens_k`` default to the sum - of row values; pass explicit session totals (per-rule: the scheduler's tracked spend; - whole-repo: the one session's cost/tokens, which don't live on individual rows). + rules with a confirmed bug. + + Cost is metered from tokens, not hand-reported: when ``price`` (the submitter's declared + per-token rate) is given, ``total_cost_usd`` is DERIVED as ``price × token_usage`` — the + same authoritative figure the backend recomputes (benchmark/verify_submission.py), so the + self-reported total is never trusted. The 4-bucket token total is either passed in + (``usage_totals`` — whole-repo, one session) or summed from each row's ``usage`` block + (per-rule). It rides on the envelope as ``usage_totals`` (the reproducible primitive) next + to the ``prices`` snapshot, so any reader can recompute spend under other prices. Without a + ``price`` (e.g. FAKE mode) it falls back to explicit session totals or the row-sum. """ attempted = [r for r in rows if r.get("result") != "skipped_budget"] bugs = count_bugs(rows) - cost = total_cost_usd if total_cost_usd is not None else sum(r.get("cost", 0.0) for r in rows) - tokens_k = total_tokens_k if total_tokens_k is not None else sum(r.get("tokens_k", 0.0) for r in rows) + + # 4-bucket token usage: the reproducible primitive the backend re-prices. + if usage_totals is None: + usage_totals = Usage() + for r in rows: + usage_totals = usage_totals + usage_from_dict(r.get("usage")) + elif isinstance(usage_totals, dict): + usage_totals = usage_from_dict(usage_totals) + + if price is not None: + cost = price.cost(usage_totals) + tokens_k = usage_totals.total_tokens / 1000 + else: + cost = total_cost_usd if total_cost_usd is not None else sum(r.get("cost", 0.0) for r in rows) + tokens_k = total_tokens_k if total_tokens_k is not None else sum(r.get("tokens_k", 0.0) for r in rows) + envelope = { "schema_version": SCHEMA_VERSION, "model": model, @@ -79,9 +105,17 @@ def build_submission( "efficiency_bugs_per_ktok": round(bugs / tokens_k, 4) if tokens_k else 0, "efficiency_bugs_per_dollar": round(bugs / cost, 4) if cost else 0, "rules_tested": len({r.get("rule") for r in attempted}), + # Aggregate token totals + the declared price snapshot — the backend re-meters cost + # from these (zero-trust); dated by created_at, recomputable under any price table. + "usage_totals": usage_as_dict(usage_totals), + "prices": price_as_dict(price) if price is not None else None, "results": rows, + # Version/provenance stamp so a produced file self-identifies its run (no more + # "everything overwrites one submission.json"). agent_mode + created_at + the + # runner/pred/library pins together pin down exactly what produced this file. "runner_version": runner_version, "pred_version": pred_version, + "agent_mode": agent_mode, "created_at": created_at, "submitted_by": submitted_by, } @@ -114,9 +148,15 @@ def run( model_kwargs: dict | None = None, api_key: str | None = None, mode: str = "per-rule", + trajectory_dir: str | Path | None = None, ) -> dict: """Run the full budgeted session for one model and return the submission dict. + ``trajectory_dir`` is where the whole-repo agent's trajectory + the durable incremental + cert log are persisted (default: the output file's directory). Beside the stable + ``output`` a versioned archive (``submission--.json``) is also written + so successive runs don't all overwrite one submission.json. + ``mode`` selects the runner: ``per-rule`` (default) schedules one isolated agent session per rule under a shared budget; ``whole-repo`` runs ONE session over the whole library — the agent enumerates and triages the rules itself and emits a certificate per bug. @@ -149,16 +189,27 @@ def run( config_path=config_path, strategy=strategy, model_kwargs=model_kwargs, api_key=api_key) + # Where to persist the whole-repo trajectory + durable cert log. An explicit + # TRAJECTORY_DIR wins; otherwise default to the output file's directory (the mounted + # /out). Exposed as a parameter so it shows up in the runner's config surface, not + # hardcoded — a crash/early-stop then still leaves the found bugs on disk. + traj_dir = Path(trajectory_dir) if trajectory_dir else ( + Path(output).parent if output is not None else None) + if mode == "whole-repo" and not fake: from benchmark.run_mini import run_repo_session + out_dir = traj_dir session = run_repo_session( model, ctx, cost_limit=max(budget - safety_margin, 0.0), api_base=api_base, price=price, max_tokens=max_tokens, + trajectory_dir=out_dir, + certs_path=(out_dir / "certs.txt") if out_dir is not None else None, config_path=config_path, strategy=strategy, model_kwargs=model_kwargs, api_key=api_key, ) rows, total_cost, total_tokens = session["rows"], session["cost"], session["tokens_k"] session_trajectory = session["trajectory"] + session_usage = session.get("usage") # session-level 4-bucket total to re-price else: rules = list_rules(str(repo)) if max_rules is not None: @@ -181,25 +232,46 @@ def run( completed = scheduler.run_all() spent = scheduler._spent.get(model) # per-rule totals: cost is the scheduler's tracked spend; tokens sum from the rows. - # Per-rule rows carry their own trajectories, so there is no envelope-level one. + # Per-rule rows carry their own trajectories AND their own 4-bucket ``usage``, so the + # envelope aggregates usage from the rows (session_usage=None). rows, total_cost, total_tokens = completed[model], spent, None session_trajectory = None + session_usage = None sub = build_submission( model, rows, budget_cap=budget, library_commit=commit, created_at=created_at, submitted_by=submitted_by, total_cost_usd=total_cost, total_tokens_k=total_tokens, trajectory=session_trajectory, pred_version=getattr(ctx, "pred_version", ""), + price=price, usage_totals=session_usage, + agent_mode=mode, ) if output is not None: output = Path(output) output.parent.mkdir(parents=True, exist_ok=True) - output.write_text(json.dumps(sub, indent=2), encoding="utf-8") + blob = json.dumps(sub, indent=2) + output.write_text(blob, encoding="utf-8") + # ALSO write a versioned archive copy so runs don't clobber each other: the stable + # `output` is the "latest" pointer (what `prb submit` reads); the archive keeps history. + archive = output.with_name(_versioned_name(output, model, created_at)) + if archive.name != output.name: + archive.write_text(blob, encoding="utf-8") return sub +def _versioned_name(output: Path, model: str, created_at: str | None) -> str: + """Archive filename that encodes the run: ``--``. + + ``model`` is made filesystem-safe; ``timestamp`` is the compact UTC created_at + (digits + 'T'). Keeps each run's output as a distinct, self-identifying file next to + the stable ``output`` pointer.""" + label = model.replace("/", "_").replace(":", "_") + stamp = re.sub(r"[^0-9T]", "", created_at)[:15] if created_at else "unknown" + return f"{output.stem}-{label}-{stamp}{output.suffix}" + + def _env(name: str, default: str | None = None) -> str | None: return os.environ.get(name, default) @@ -219,7 +291,11 @@ def main() -> None: parser.add_argument("--repo-dir", default=_env("REPO_DIR", "/app/pr-src"), help="problem-reductions source tree (env REPO_DIR)") parser.add_argument("--output", default=_env("OUTPUT", "/out/submission.json"), - help="Where to write submission.json (env OUTPUT)") + help="Stable 'latest' submission path (env OUTPUT). A versioned archive " + "copy (submission-