search-with-proof: per-search fair-share pilot budget (stop one search starving the rest)

--ceiling-usd was one shared SpendBudget passed to every search, so an expensive
search drained the pool and starved later ones: in the last 2-pair/$2 run the first
search spent $1.60 and the second hit the ceiling at $0.34 after finishing its analysis.
The driver now allocates a dynamic fair share per search (remaining_ceiling /
searches_not_yet_started), rolls unused settled budget forward, caps each search at its
share, and keeps total settled <= ceiling. A search now gets room to author and verify
after it finds the path.

Driver-only change (benchmarks/search/pilot_run.py) plus tests. tests/test_search_pilot_run.py
39 passed; ruff clean; mypy unchanged (pre-existing benchmarks errors only).

Co-Authored-By: Claude Opus 4.8 <[email protected]>

benchmarks/search/pilot_run.py

@@ -419,7 +419,7 @@ def main(argv=None):
     args = parser.parse_args(argv)
     if not math.isfinite(args.ceiling_usd) or args.ceiling_usd < 0:
         parser.error("ceiling must be finite and nonnegative")
-    run_budget = SpendBudget(args.ceiling_usd)
+    remaining_ceiling = args.ceiling_usd
     if not re.fullmatch(r"\d+(,\d+)*", args.pairs) or args.max_tokens < 1:
         parser.error("pairs must be comma-separated nonnegative indices; max tokens must be positive")
     indices = [int(i) for i in args.pairs.split(",")]
@@ -522,17 +522,20 @@ def main(argv=None):
                 )
         except Exception as exc:
             aborted_reason = exception_reason(exc)
-        for index in indices if aborted_reason is None else []:
+        for position, index in enumerate(indices if aborted_reason is None else []):
             local = root / str(index)
             local.mkdir()
             if args.dry_run:
                 sides, demos = materialise(local / "probe", "path")
                 client = scripted(load_demo(demos["real"]))
-            pilot = Pilot(args, rows[index], index, local, store, run_budget, client, prices, lane)
+            searches_remaining = len(indices) - position
+            per_search = remaining_ceiling / searches_remaining if searches_remaining > 0 and remaining_ceiling > 0 else 0.0
+            pilot = Pilot(args, rows[index], index, local, store, SpendBudget(per_search), client, prices, lane)
             if args.dry_run:
                 pilot.sides = sides
                 pilot.record["input_bytes"] = len((sides[0 if args.side == "vulnerable" else 1].checkout / "app.py").read_bytes())
             record = pilot.run()
+            remaining_ceiling = max(0.0, remaining_ceiling - record["spend"]["settled"])
             records.append(record)
             output.write(json.dumps(record, allow_nan=False) + "\n")
             output.flush()

tests/test_search_pilot_run.py

@@ -67,6 +67,79 @@ def test_ceiling_refuses_before_stub_call(tmp_path):
     assert record["task_metrics"][0]["end"] == "run_ceiling"
 
 
+@pytest.mark.parametrize(
+    "ceiling,first_spend,allocations",
+    [(2.0, 0.25, [1.0, 1.75]), (2.0, 1.0, [1.0, 1.0]), (0.0, 0.0, [0.0, 0.0])],
+)
+def test_driver_allocates_fair_share_with_settled_rollover(tmp_path, monkeypatch, ceiling, first_spend, allocations):
+    from types import SimpleNamespace
+
+    from benchmarks.search import pilot_run
+    from openultrasast.plane.budget import BudgetExhausted
+
+    monkeypatch.chdir(tmp_path)
+    root = Path("benchmarks/search/private")
+    root.mkdir(parents=True)
+    row = dict(
+        repo="https://github.com/private/project",
+        vulnerable="a" * 40,
+        fixed="b" * 40,
+        family="path",
+        language="python",
+        file="app.py",
+        function="main",
+    )
+    (root / "pilot.json").write_text(json.dumps({"pairs": [row] * 3}))
+    monkeypatch.setattr(pilot_run, "load_dotenv", lambda: None)
+    monkeypatch.setenv(pilot_run.DEEPSEEK_KEY_ENV, "fake-test-key")
+    monkeypatch.setattr(pilot_run, "OpenRouterChatClient", lambda **kw: SimpleNamespace(max_attempts=1))
+    monkeypatch.setattr(pilot_run, "AXLane", lambda *args: object())
+    budgets = []
+
+    def run(pilot):
+        budget = pilot.worker.run_budget
+        share = allocations[len(budgets)]
+        assert all(budget is not previous for previous in budgets)
+        assert budget.spent == budget.reserved == 0
+        # Prove the cap through admission, then settle less than reserved to test rollover.
+        with pytest.raises(BudgetExhausted):
+            budget.reserve(share + 0.01)
+        reservation = budget.reserve(share)
+        reservation.settle(first_spend if not budgets else share)
+        budgets.append(budget)
+        pilot.record.update(spend={"reserved": share, "settled": budget.spent}, model_calls=0, tokens_in=0, tokens_out=0)
+        return pilot.record
+
+    monkeypatch.setattr(pilot_run.Pilot, "run", run)
+    output = tmp_path / "records.jsonl"
+    assert (
+        main(
+            [
+                "--side",
+                "vulnerable",
+                "--pairs",
+                "2,0",
+                "--ceiling-usd",
+                str(ceiling),
+                "--verify-lane",
+                "ax",
+                "--image",
+                "image@sha256:" + "a" * 64,
+                "--out",
+                str(output),
+            ]
+        )
+        == 0
+    )
+    first, second, summary = [json.loads(line) for line in output.read_text().splitlines()]
+    assert [first["pair_index"], second["pair_index"]] == [2, 0]
+    assert len(budgets) == summary["searches"] == 2
+    assert second["spend"]["settled"] == pytest.approx(allocations[1])
+    total = sum(record["spend"]["settled"] for record in (first, second))
+    assert summary["spend"]["settled"] == total <= ceiling
+    assert total == pytest.approx(ceiling)
+
+
 def make_pilot(tmp_path, *, side="fixed", dry_run=False):
     from argparse import Namespace