diff --git a/benchmark/edgebench/managed_run.py b/benchmark/edgebench/managed_run.py index 70b2da0a21..dcee10cfbb 100644 --- a/benchmark/edgebench/managed_run.py +++ b/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: diff --git a/tests/benchmark/test_edgebench_managed_run.py b/tests/benchmark/test_edgebench_managed_run.py index 973d5d13dd..989aeb4403 100644 --- a/tests/benchmark/test_edgebench_managed_run.py +++ b/tests/benchmark/test_edgebench_managed_run.py @@ -78,7 +78,8 @@ 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" @@ -86,20 +87,22 @@ def entry(tmp_path, monkeypatch): "--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)