Keep resumed scheduler state in benchmark failure checkpoints

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

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

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():

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"))