Allow bounded native solver histories during benchmark closeout

benchmark/edgebench/managed_run.py

@@ -35,6 +35,9 @@
 RUNTIME_FILENAME = "runtime-receipt.json"
 _MAX_JSON_BYTES = 1024 * 1024
 _MAX_HISTORY_BYTES = 16 * 1024 * 1024
+# These two native artifacts carry every solver segment, including repeated
+# resumes. Keep their budget separate from the compact managed control records.
+_MAX_NATIVE_SOLVER_EVIDENCE_BYTES = 64 * 1024 * 1024
 _IDENTITY_FIELDS = {"benchmark_id", "study_id", "case_id", "run_id", "arm_id", "arm_role", "attempt"}
 _BOARD_STATUS = {"terminal": "completed", "runner_failed": "runner_invalid", "cancelled": "cancelled"}
 _NATIVE_TERMINAL_STATUS = {"normal_exit", "nonzero_exit", "timed_out", "cancelled", "signaled"}
@@ -163,7 +166,9 @@ def retain_managed_solver(trial: Path, binding: dict, launch: dict, *, receipt:
     """Bind the exact native solver snapshot to the returned runtime receipt."""
     retained = dict(launch)
     if receipt["status"] in _BOARD_STATUS and getattr(result, "solver_execution", None) is not None:
-        execution, digest = _read_record(trial / "solver_execution.json")
+        execution, digest = _read_record(
+            trial / "solver_execution.json", limit=_MAX_NATIVE_SOLVER_EVIDENCE_BYTES,
+        )
         admitted = _timestamp(binding["admission_receipt"]["run"]["admitted_at"])
         if (execution == result.solver_execution and _solver_ended(
                 execution, admitted_at=admitted.timestamp(), observed_at=datetime.now(UTC).timestamp())):
@@ -216,7 +221,7 @@ def _grading_drained(trial: Path, *, history_file: str, run_id: str, task_id: st
 def _retained_completed_result(trial: Path, receipt: dict) -> dict | None:
     """Read the runner's completion marker when its terminal receipt did not land."""
     try:
-        result, _ = _read_record(trial / "final_result.json")
+        result, _ = _read_record(trial / "final_result.json", limit=_MAX_NATIVE_SOLVER_EVIDENCE_BYTES)
     except FileNotFoundError:
         return None
     duration = result.get("runtime_seconds")
@@ -255,7 +260,9 @@ def publish_managed_terminal(trial: Path, *, execute: bool = False) -> bool:
     history_file = retained.get("history_file")
     if history_file not in _NATIVE_HISTORY_FILES:
         raise ValueError("Retained managed grading history path differs")
-    execution, solver_digest = _read_record(trial / "solver_execution.json")
+    execution, solver_digest = _read_record(
+        trial / "solver_execution.json", limit=_MAX_NATIVE_SOLVER_EVIDENCE_BYTES,
+    )
     if recovered_completion is None:
         solver_matches = retained.get("solver_execution_sha256") == solver_digest
     else:

tests/benchmark/test_edgebench_managed_run.py

@@ -78,28 +78,31 @@ def entry(tmp_path, monkeypatch):
         internet=False, game_mode=False, submit_paths=["solver.py"],
         work=SimpleNamespace(agent_query="Synthetic task"),
         judge=SimpleNamespace(selection="score_first", score_direction="maximize")))
-    backend = SimpleNamespace(image_exists=lambda image: True, native_goal_receipt=None,
+    backend = SimpleNamespace(observe_local_images=lambda **keys: {},
+                              registration_options=lambda provider: {}, native_goal_receipt=None,
                               execution_receipt={"exit_code": 0, "execution_started": True})
     monkeypatch.setattr(run, "RecordingDockerBackend", lambda **kwargs: backend)
     trial = tmp_path / "runs" / "attempt" / "task"
     argv = ["--task", "task", "--tasks-dir", str(tmp_path), "--log-dir", str(tmp_path),
             "--run-id", "attempt", "--worker", "official", "--m", "fixture",
             "--effort", "high", "--judge-url", "http://localhost:9999"]
 
-    def invoke(binding: Path | None = None, *, evidence: str = "drained") -> int:
+    def invoke(binding: Path | None = None, *, evidence: str = "drained", segment_count: int = 1) -> int:
         def finish(**kwargs):
             assert kwargs["run_id"] == "attempt" and kwargs["task_spec"].task_id == "task"
             assert "managed_run" not in kwargs and "admission_receipt" not in kwargs
             if evidence == "exception":
                 raise RuntimeError("native provider failed")
             (trial / "started_at").write_text("2026-01-01T00:00:00Z\n", encoding="utf-8")
-            now = time.time()
-            segment = {
-                "index": 1, "status": "normal_exit", "finished_at": time.time(),
-                "started_at": now - 100 if evidence == "predates_admission" else now,
-                "terminal_observed": evidence != "unconfirmed_solver", "native_exit_code": 0,
-            }
-            execution = {"schema_version": 1, "status": "normal_exit", "segments": [segment]}
+            segments = []
+            for index in range(1, segment_count + 1):
+                now = time.time()
+                segments.append({
+                    "index": index, "status": "normal_exit", "finished_at": time.time(),
+                    "started_at": now - 100 if evidence == "predates_admission" else now,
+                    "terminal_observed": evidence != "unconfirmed_solver", "native_exit_code": 0,
+                })
+            execution = {"schema_version": 1, "status": "normal_exit", "segments": segments}
             native._write_solver_execution(trial, execution)
             history = {"run_id": "attempt", "entries": []}
             if evidence == "pending":
@@ -140,7 +143,7 @@ def _replay(trial: Path, *, execute: bool = False) -> subprocess.CompletedProces
                           capture_output=True, text=True, timeout=30)
 
 
-@pytest.mark.parametrize("stage", ["drained", "pending", "write_failure",
+@pytest.mark.parametrize("stage", ["drained", "large_solver_history", "pending", "write_failure",
     "receipt_before_replace", "receipt_after_replace", "receipt_persistent"])
 def test_managed_terminal_flows_from_runner_to_controller(tmp_path, entry, capsys, monkeypatch, stage):
     trial, invoke, drain = entry
@@ -195,7 +198,10 @@ def fail_terminal_receipt_once(path, payload, **kwargs):
                 return write_json_atomic(path, payload, **kwargs)
 
             patch.setattr(managed_run, "write_json_atomic", fail_terminal_receipt_once)
-        assert invoke(_binding_file(tmp_path, admission), evidence=stage) == 0
+        segment_count = 5500 if stage in {"large_solver_history", "receipt_persistent"} else 1
+        assert invoke(_binding_file(tmp_path, admission), evidence=stage, segment_count=segment_count) == 0
+    if segment_count > 1:
+        assert all((trial / name).stat().st_size > 1 << 20 for name in ("solver_execution.json", "final_result.json"))
     runtime = json.loads(capsys.readouterr().out)
     assert runtime["status"] == "terminal" and runtime["best_score"] == 4
     assert GOAL not in json.dumps(runtime) and str(project) not in json.dumps(runtime)