diff --git a/benchmark/README.md b/benchmark/README.md index 1fb7f74167..35613cc5f7 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1493,6 +1493,13 @@ Recovery settles process transport only. Preserved pending semantic replan attempts still require their original settlement owner; an interrupted replan is not silently declared complete or cleared to force dispatch. +For a terminal capture after an in-place continuation, pass the admitted phase +deadlines to `terminal_components(scheduler_phase_deadlines=...)` in predecessor +order. The capture keeps the original scheduler state and those exact phase +files, not the whole control directory. An absent or duplicate phase does not +fall back to the original state. The stopped worker is still required for +recovery and its pending Host witness. + ### Explicitly continue after native account allowance is repaired A hard native quota failure is still held by default. Repair or replenish the diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py index 5d6f62f959..77c414759f 100644 --- a/benchmark/edgebench/terminal_capture.py +++ b/benchmark/edgebench/terminal_capture.py @@ -14,8 +14,10 @@ from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _LOOPX_RUNTIME, _REGISTRY -def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str]]: - """Explicit full-capture participants, without the credential-bearing home.""" +def terminal_components( + declaration: dict, workspace: str, *, scheduler_phase_deadlines: tuple[float, ...] = (), +) -> list[dict[str, str]]: + """Allowlisted capture, retaining any already-admitted in-place phases.""" from benchmark.runtime.research_artifacts import validate_research_scope validate_research_scope(workspace, ["."]) @@ -39,6 +41,15 @@ def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str ("terminal_native_history", f"{_CODEX_HOME}/thread_history_1.sqlite"), ("terminal_native_config", f"{_CODEX_HOME}/config.toml"), )) + # Do not archive the control directory. It can hold unrelated state and + # private transport material. The scheduler clock names each phase file. + from benchmark.runtime.scheduler import phase_state_name + + root = PurePosixPath(_CONTROL, "scheduler-state.json") + components.extend({ + "id": f"terminal_scheduler_phase_{index}", + "source_path": str(root.with_name(phase_state_name(root, deadline))), + } for index, deadline in enumerate(scheduler_phase_deadlines, start=1)) # Let the shared transport reject aliases/overlaps before launch, rather # than silently dropping an explicit external research participant. from benchmark.edgebench.checkpoint_worker import _validate_request diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 266becff97..d0e29de00a 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -6,7 +6,7 @@ import json import math import os -from pathlib import Path +from pathlib import Path, PurePath import re import shlex import sys @@ -23,6 +23,15 @@ def phase_work_reserve(mode: str) -> int: return 0 if mode == "heartbeat" else PHASE_WORK_RESERVE_SECONDS +def phase_state_name(root: PurePath, phase_deadline: float) -> str: + """Name a scheduler phase by the finite clock that owns its allowance.""" + if (isinstance(phase_deadline, bool) or not isinstance(phase_deadline, (int, float)) + or not math.isfinite(phase_deadline) or phase_deadline <= 0): + raise ValueError("Scheduler phase clock must be finite and positive") + suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20] + return f"{root.stem}.phase-{suffix}{root.suffix}" + + def _physical_state(path: Path) -> Path: if path.is_symlink() or path.resolve() != path: raise ValueError("Scheduler phase state must be a physical path") @@ -199,8 +208,8 @@ def hold(state: Path, reason: str) -> str: recovery = state.get("interruption_recovery") or state.get('failed_host_recovery') can_continue_stopped_phase = (isinstance(prior, dict) and prior.get("status") == "stopped" and prior.get("stop_exit_code") == BUDGET_EXHAUSTED_EXIT_CODE) - requested_suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20] - same_phase = predecessor.name == f"{path.stem}.phase-{requested_suffix}{path.suffix}" + target = path.with_name(phase_state_name(path, phase_deadline)) + same_phase = predecessor == target explicit_failed_phase = (retry_failed_state_sha256 is not None and not same_phase and hashlib.sha256(predecessor.read_bytes()).hexdigest() == retry_failed_state_sha256) if ((recovery is not None or state.get("resume_from")) @@ -254,8 +263,6 @@ def hold(state: Path, reason: str) -> str: if recovered is not None: if failed_recovery and phase_deadline <= time.time(): raise ValueError('Explicit failed continuation exhausted its phase during qualification') - suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20] - target = path.with_name(f"{path.stem}.phase-{suffix}{path.suffix}") recovered.update(source_state_sha256=hashlib.sha256(predecessor.read_bytes()).hexdigest(), phase_deadline_epoch=phase_deadline) _create_resume_state(target, predecessor, recovered=recovered, @@ -270,8 +277,6 @@ def hold(state: Path, reason: str) -> str: return str(predecessor) # Only the immutable phase clock names the new state. Todo/reset-token # changes cannot clear a failure already recorded in this new phase. - suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20] - target = path.with_name(f"{path.stem}.phase-{suffix}{path.suffix}") if target.is_symlink(): raise ValueError("Resumed scheduler state must not be a symlink") if target.exists(): diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py index 99fa0df04b..411cb4c31f 100644 --- a/benchmark/tests/test_terminal_preservation.py +++ b/benchmark/tests/test_terminal_preservation.py @@ -190,9 +190,13 @@ def test_cli_preservation_requires_full_native_controller_before_trial(tmp_path, @pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", reason="Requires isolated Docker and native HTTP checkpoint provider") -@pytest.mark.parametrize("capture_failure", [False, True]) -@pytest.mark.parametrize("receipt_failure", [False, True]) -def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, writer, tmp_path, monkeypatch, capture_failure, receipt_failure): +@pytest.mark.parametrize("capture_failure,receipt_failure,continued", [ + (False, False, False), (False, True, False), (True, False, False), (True, True, False), + (False, False, True), +]) +def test_real_failed_solver_preserves_tail_and_native_opportunity( + controllers, writer, tmp_path, monkeypatch, capture_failure, receipt_failure, continued, +): from sforge.harness.backend.docker_backend import DockerContainerHandle from benchmark.runtime.sforge_backend import RecordingDockerBackend from benchmark.edgebench.checkpoint_supervisor import CheckpointSupervisor @@ -224,7 +228,32 @@ def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, w Path('/data/candidate').write_text('newer than the last graded candidate') ''' assert container.exec_run(["python", "-c", script]).exit_code == 0 - components = terminal_components({"components": []}, "/data") + phase_deadlines = () + if continued: + from benchmark.tests.test_scheduler_phase_resume import _resume, _stopped + from pathlib import Path + import time + + # Keep both actual scheduler-selected names, including the phase that + # failed after the second resume. The root alone has different evidence. + root = tmp_path / "scheduler-state.json" + _stopped(root) + phase_one = time.time() + 500 + first_phase = _resume(root, phase_one) + _stopped(first_phase) + phase_two = time.time() + 600 + second_phase = _resume(root, phase_two) + assert len({root, first_phase, second_phase}) == 3 + control = Path("/opt/loopx-benchmark/control") + for path, content in [ + (first_phase, {"resume_from": str(control / root.name), "phase": "prior"}), + (second_phase, {"resume_from": str(control / first_phase.name), "phase": "failed latest"}), + (tmp_path / "unselected-private.json", {"unselected": True}), + ]: + assert container.exec_run(["python", "-c", "from pathlib import Path; " + f"Path({str(control / path.name)!r}).write_text({json.dumps(content)!r})"]).exit_code == 0 + phase_deadlines = (phase_one, phase_two) + components = terminal_components({"components": []}, "/data", scheduler_phase_deadlines=phase_deadlines) if capture_failure: components.append({"id": "missing", "source_path": "/missing-participant"}) supervisor = CheckpointSupervisor(files[0], tmp_path / "terminal-points", @@ -296,3 +325,9 @@ def registered(token): assert native["opportunity"]["next_agent"] == 2 assert native["opportunity"]["remaining_cooldown_seconds"] > 3500 assert not list(point.rglob("auth.json")) + originals = json.loads((point / "components/terminal_scheduler").read_text()) + assert originals == {"host_attempt": {"status": "failed"}} + captured_phases = sorted((point / "components").glob("terminal_scheduler_phase_*")) + assert [json.loads(path.read_text())["phase"] for path in captured_phases] == ( + ["prior", "failed latest"] if continued else []) + assert not list(point.rglob("unselected-private.json"))