From 353587646b01a17dacaddac2643fa5ace90a988d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:09:57 +0800 Subject: [PATCH 1/2] Bind checkpoint capture to the active benchmark phase clock Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 5 +++++ benchmark/edgebench/checkpoint_supervisor.py | 14 ++++++++++---- benchmark/edgebench/terminal_capture.py | 9 +++++---- benchmark/runtime/layout.py | 1 + benchmark/tests/test_checkpoint_worker.py | 18 ++++++++++++++---- benchmark/tests/test_terminal_preservation.py | 16 ++++++++++++++-- 6 files changed, 49 insertions(+), 14 deletions(-) diff --git a/benchmark/README.md b/benchmark/README.md index 1fb7f74167..a9420407af 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1533,6 +1533,11 @@ including any later failure; it cannot clear the failure or extend its deadline. A later explicit repair must bind that new failure separately. Existing operator and account authority remain required; a digest is provenance, not permission. +For in-place continuation, bind the checkpoint supervisor's `phase_deadline_path` +to the same new immutable clock supplied to the resumed scheduler. Both held +and terminal captures use that clock. Preserve the stopped phase's clock at +its original path. Fresh workers retain the standard path. + The explicit failed-phase path additionally uses the installed optional native Codex provider's existing availability policy (`loopx-codex-provider-routing`). It reads `account/read(refreshToken=false)` and `account/rateLimits/read` in a diff --git a/benchmark/edgebench/checkpoint_supervisor.py b/benchmark/edgebench/checkpoint_supervisor.py index 4341ac3132..55d060310e 100644 --- a/benchmark/edgebench/checkpoint_supervisor.py +++ b/benchmark/edgebench/checkpoint_supervisor.py @@ -12,15 +12,18 @@ from pathlib import Path from loopx.control_plane.effect_runtime import EffectRuntimeRemoteError +from benchmark.runtime.layout import _PHASE_DEADLINE from .checkpoint_controller import _controller -from .checkpoint_worker import _archive, capture_worker, qualify_export_storage +from .checkpoint_worker import _archive, _guest_path, _validate_request, capture_worker, qualify_export_storage class CheckpointSupervisor: def __init__(self, controller_file: Path, destination: Path, retention_policy: Path | None = None, *, retention_enabled: bool = True, terminal_components: list[dict[str, str]] | None = None, - terminal_receipt: Path | None = None): + terminal_receipt: Path | None = None, phase_deadline_path: str = _PHASE_DEADLINE): + _guest_path(phase_deadline_path) + self.phase_deadline_path = phase_deadline_path _controller(controller_file) # Fail before native registration. qualify_export_storage(destination) if not retention_enabled and retention_policy is not None: @@ -31,7 +34,10 @@ def __init__(self, controller_file: Path, destination: Path, retention_policy: P from .terminal_capture import FailurePreservation if terminal_receipt is None or terminal_receipt.exists(): raise ValueError("Terminal preservation requires a fresh private receipt") - self.preservation = FailurePreservation(self, terminal_components, terminal_receipt) + _validate_request({"container_id": "0" * 64, "components": terminal_components, + "phase_deadline_path": phase_deadline_path}) + self.preservation = FailurePreservation(self, terminal_components, terminal_receipt, + phase_deadline_path=phase_deadline_path) self.token = None self.stop = threading.Event() self.thread = None @@ -171,7 +177,7 @@ def _run(self, container): request = stage / "request.json" request.write_text(json.dumps({"container_id": container.id, "components": value["components"], - "phase_deadline_path": "/opt/loopx-benchmark/control/phase-deadline", + "phase_deadline_path": self.phase_deadline_path, "handoff": {"source_path": remote, "sha256": digest}})) remaining = value["deadline_epoch"] - time.time() if remaining <= 0: diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py index 5d6f62f959..3e87b517d3 100644 --- a/benchmark/edgebench/terminal_capture.py +++ b/benchmark/edgebench/terminal_capture.py @@ -11,7 +11,7 @@ from pathlib import PurePosixPath from loopx.file_lock import write_json_atomic -from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _LOOPX_RUNTIME, _REGISTRY +from benchmark.runtime.layout import _CODEX_HOME, _CONTROL, _LOOPX_RUNTIME, _PHASE_DEADLINE, _REGISTRY def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str]]: @@ -43,16 +43,17 @@ def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str # than silently dropping an explicit external research participant. from benchmark.edgebench.checkpoint_worker import _validate_request _validate_request({"container_id": "0" * 64, "components": components, - "phase_deadline_path": f"{_CONTROL}/phase-deadline"}) + "phase_deadline_path": _PHASE_DEADLINE}) return components class FailurePreservation: """One terminal attempt; a stopped source remains until explicit disposal.""" - def __init__(self, supervisor, components, receipt_path): + def __init__(self, supervisor, components, receipt_path, *, phase_deadline_path=_PHASE_DEADLINE): self.supervisor = supervisor self.components = components self.receipt_path = receipt_path + self.phase_deadline_path = phase_deadline_path self.receipt = None self.retain_source = False @@ -106,7 +107,7 @@ def finish(self, execution): self._persist_receipt() request = self.supervisor.destination / "terminal-request.json" write_json_atomic(request, {"container_id": self.supervisor.container.id, - "components": self.components, "phase_deadline_path": f"{_CONTROL}/phase-deadline"}, + "components": self.components, "phase_deadline_path": self.phase_deadline_path}, mode=0o600, fsync_directory=True) point = capture_worker(supervisor.controller_file, token=supervisor.token, capture_id=self.receipt["capture_id"], diff --git a/benchmark/runtime/layout.py b/benchmark/runtime/layout.py index f8dec8ceee..66df49713f 100644 --- a/benchmark/runtime/layout.py +++ b/benchmark/runtime/layout.py @@ -3,6 +3,7 @@ _ROOT = "/opt/loopx-benchmark" _SRC = f"{_ROOT}/source" _CONTROL = f"{_ROOT}/control" +_PHASE_DEADLINE = f"{_CONTROL}/phase-deadline" _CODEX_HOME = f"{_ROOT}/codex-home" _REGISTRY = f"{_CONTROL}/registry.json" _LOOPX_RUNTIME = f"{_ROOT}/state/runtime" diff --git a/benchmark/tests/test_checkpoint_worker.py b/benchmark/tests/test_checkpoint_worker.py index d0160a92da..2d540c1855 100644 --- a/benchmark/tests/test_checkpoint_worker.py +++ b/benchmark/tests/test_checkpoint_worker.py @@ -115,7 +115,7 @@ def test_archive_keeps_private_directory_modes_without_special_bits(writer, tmp_ @pytest.mark.parametrize("fault", [None, "source_moved", "expired", "read_race", "late_native", - "retention", "retention_default", "retention_off", "retention_pin", "retention_failure"]) + "retention", "retention_default", "retention_off", "retention_pin", "retention_failure", "continued_clock"]) def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writer, tmp_path, monkeypatch, fault): from loopx.control_plane.effect_runtime import effect_runtime_result from benchmark.edgebench.checkpoint_worker import _archive @@ -130,6 +130,11 @@ def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writ "Path('/logs/agent/checkpoint-controller').mkdir(parents=True); " "Path('/opt/loopx-benchmark/control').mkdir(parents=True); " "Path('/opt/loopx-benchmark/control/phase-deadline').write_bytes(Path('/phase').read_bytes())"]).exit_code == 0 + continued_clock = "/opt/loopx-benchmark/control/continued-phase-deadline" + if fault == "continued_clock": + assert container.exec_run(["python", "-c", "from pathlib import Path; " + f"Path({continued_clock!r}).write_bytes(Path('/phase').read_bytes()); " + "Path('/opt/loopx-benchmark/control/phase-deadline').write_text('1')"]).exit_code == 0 prepared = _archive(container, "/candidate", tmp_path / "prepared") probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": [{"id": "code", "source_path": str(prepared)}]}) @@ -186,7 +191,8 @@ def with_old_point(*args, **kwargs): return original_capture(*args, **kwargs) monkeypatch.setattr(checkpoint_supervisor, "capture_worker", with_old_point) supervisor = CheckpointSupervisor(files[0], tmp_path / "points", retention_policy=policy_file, - retention_enabled=fault != "retention_off") + retention_enabled=fault != "retention_off", + **({"phase_deadline_path": continued_clock} if fault == "continued_clock" else {})) supervisor.bind({"SFORGE_TOKEN": token}) supervisor.start(container) try: @@ -196,15 +202,19 @@ def with_old_point(*args, **kwargs): supervisor.close() response = json.loads((tmp_path / "points" / (nonce + ".json")).read_bytes()) assert response["ok"] is (fault in {None, "read_race", "retention", "retention_default", - "retention_off", "retention_pin", "retention_failure"}) + "retention_off", "retention_pin", "retention_failure", "continued_clock"}) unit = tmp_path / "points" / nonce - if fault in {None, "read_race", "retention", "retention_default", "retention_off", "retention_pin", "retention_failure"}: + if response["ok"]: manifest = json.loads((unit / "checkpoint.json").read_bytes()) metadata = manifest["point_metadata"] assert metadata["synthetic_boundary"] is True assert metadata["worker_file_snapshot"]["prepared_source_verified"] is True assert (unit / "components/code").read_text() == "retained" assert response["capture"]["research_point_qualified"] is False + if fault == "continued_clock": + capsule = unit / "components" / metadata["worker_file_snapshot"]["native_capsule_component_id"] + assert (capsule / "components/host_phase_deadline").read_bytes() == container.exec_run(["cat", "/phase"]).output + assert container.exec_run(["cat", "/opt/loopx-benchmark/control/phase-deadline"]).output == b"1" if fault in {"retention", "retention_default"}: assert not (tmp_path / "points/old").exists() receipt = json.loads((tmp_path / "points/retention-last.json").read_bytes()) diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py index 99fa0df04b..2227f4838b 100644 --- a/benchmark/tests/test_terminal_preservation.py +++ b/benchmark/tests/test_terminal_preservation.py @@ -2,7 +2,9 @@ from __future__ import annotations import errno +import io import json +import tarfile import os from types import SimpleNamespace @@ -176,6 +178,12 @@ def test_receipt_failure_never_masks_an_unconfirmed_stop(tmp_path, monkeypatch): assert preservation.receipt["stop_error_kind"] == "RuntimeError" +def _read_retained_clock(container, name): + with io.BytesIO(b"".join(container.get_archive("/opt/loopx-benchmark/control/" + name)[0])) as raw: + with tarfile.open(fileobj=raw) as archive: + return archive.extractfile(name).read() + + @pytest.mark.parametrize("profile", ["official", "native-goal", "heartbeat-explore"]) def test_cli_preservation_requires_full_native_controller_before_trial(tmp_path, profile): pytest.importorskip("sforge") @@ -208,7 +216,8 @@ def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, w script = '''from pathlib import Path import sqlite3,time root=Path('/opt/loopx-benchmark'); control=root/'control'; control.mkdir(parents=True) -(control/'phase-deadline').write_text(str(int(time.time())+600)) +(control/'phase-deadline').write_text('1') +(control/'continued-phase-deadline').write_text(str(int(time.time())+600)) (control/'registry.json').write_text('{"fixture":true}') (control/'scheduler-state.json').write_text('{"host_attempt":{"status":"failed"}}') (root/'state/runtime').mkdir(parents=True) @@ -229,7 +238,8 @@ def test_real_failed_solver_preserves_tail_and_native_opportunity(controllers, w components.append({"id": "missing", "source_path": "/missing-participant"}) supervisor = CheckpointSupervisor(files[0], tmp_path / "terminal-points", retention_enabled=False, terminal_components=components, - terminal_receipt=tmp_path / "terminal-preservation.json") + terminal_receipt=tmp_path / "terminal-preservation.json", + phase_deadline_path="/opt/loopx-benchmark/control/continued-phase-deadline") if receipt_failure: def fail_receipt(): raise OSError(errno.ENOSPC, "synthetic receipt disk full") @@ -292,6 +302,8 @@ def registered(token): assert metadata["native_cursor_qualified"] is False assert (point / "components/terminal_workspace/candidate").read_text() == 'newer than the last graded candidate' capsule = point / "components" / metadata["native_capsule_component_id"] + assert (capsule / "components/host_phase_deadline").read_bytes() == _read_retained_clock(container, "continued-phase-deadline") + assert _read_retained_clock(container, "phase-deadline") == b"1" native = json.loads((capsule / "checkpoint.json").read_text())["point_metadata"]["native_reference"] assert native["opportunity"]["next_agent"] == 2 assert native["opportunity"]["remaining_cooldown_seconds"] > 3500 From f0625c3bd16a91e466df2cc14b7370946b809d37 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 10 Nov 2026 14:00:03 +0800 Subject: [PATCH 2/2] Publish admitted continuation clock to both automatic checkpoint capture paths Signed-off-by: huangruiteng --- benchmark/README.md | 13 ++- benchmark/edgebench/checkpoint_supervisor.py | 21 ++++- benchmark/edgebench/terminal_capture.py | 5 +- benchmark/runtime/layout.py | 2 + benchmark/runtime/scheduler.py | 24 +++-- .../tests/test_scheduler_phase_resume.py | 88 ++++++++++++++++++- benchmark/tests/test_terminal_preservation.py | 3 +- 7 files changed, 137 insertions(+), 19 deletions(-) diff --git a/benchmark/README.md b/benchmark/README.md index 649d3e54c5..cfd6c71a1a 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1555,10 +1555,15 @@ including any later failure; it cannot clear the failure or extend its deadline. A later explicit repair must bind that new failure separately. Existing operator and account authority remain required; a digest is provenance, not permission. -For in-place continuation, bind the checkpoint supervisor's `phase_deadline_path` -to the same new immutable clock supplied to the resumed scheduler. Both held -and terminal captures use that clock. Preserve the stopped phase's clock at -its original path. Fresh workers retain the standard path. +After in-place continuation is admitted, the scheduler atomically publishes +`active-phase-deadline` beside its state, rounded down from that phase's existing +deadline. Replaying the same request republishes the same deadline; a refused +continuation does not replace it. Both held and terminal captures automatically +select this physical clock, including when the container is paused or stopped. +They preserve the original `phase-deadline` file as a recovery witness. Fresh +workers without a continuation clock keep the standard path. An explicitly +configured supervisor `phase_deadline_path` still takes precedence; unreadable or +invalid clocks fail capture instead of silently falling back or granting time. The explicit failed-phase path additionally uses the installed optional native Codex provider's existing availability policy (`loopx-codex-provider-routing`). diff --git a/benchmark/edgebench/checkpoint_supervisor.py b/benchmark/edgebench/checkpoint_supervisor.py index 55d060310e..3bbe88134e 100644 --- a/benchmark/edgebench/checkpoint_supervisor.py +++ b/benchmark/edgebench/checkpoint_supervisor.py @@ -12,7 +12,7 @@ from pathlib import Path from loopx.control_plane.effect_runtime import EffectRuntimeRemoteError -from benchmark.runtime.layout import _PHASE_DEADLINE +from benchmark.runtime.layout import _ACTIVE_PHASE_DEADLINE, _PHASE_DEADLINE from .checkpoint_controller import _controller from .checkpoint_worker import _archive, _guest_path, _validate_request, capture_worker, qualify_export_storage @@ -36,8 +36,7 @@ def __init__(self, controller_file: Path, destination: Path, retention_policy: P raise ValueError("Terminal preservation requires a fresh private receipt") _validate_request({"container_id": "0" * 64, "components": terminal_components, "phase_deadline_path": phase_deadline_path}) - self.preservation = FailurePreservation(self, terminal_components, terminal_receipt, - phase_deadline_path=phase_deadline_path) + self.preservation = FailurePreservation(self, terminal_components, terminal_receipt) self.token = None self.stop = threading.Event() self.thread = None @@ -57,6 +56,20 @@ def bind(self, environment): raise ValueError("Checkpoint controller requires one native worker registration") self.token = environment["SFORGE_TOKEN"] + def capture_phase_deadline_path(self): + """Use a published continuation clock; fresh workers keep their default.""" + if self.phase_deadline_path != _PHASE_DEADLINE: + return self.phase_deadline_path + from docker.errors import NotFound + + with tempfile.TemporaryDirectory(prefix=".phase-clock-", dir=self.destination) as temporary: + try: + _archive(self.container, _ACTIVE_PHASE_DEADLINE, Path(temporary) / "clock", + deadline=time.time() + 5) + except NotFound: + return _PHASE_DEADLINE + return _ACTIVE_PHASE_DEADLINE + def start(self, container): if self.thread is not None or self.token is None: raise ValueError("Checkpoint supervisor requires one bound worker execution") @@ -177,7 +190,7 @@ def _run(self, container): request = stage / "request.json" request.write_text(json.dumps({"container_id": container.id, "components": value["components"], - "phase_deadline_path": self.phase_deadline_path, + "phase_deadline_path": self.capture_phase_deadline_path(), "handoff": {"source_path": remote, "sha256": digest}})) remaining = value["deadline_epoch"] - time.time() if remaining <= 0: diff --git a/benchmark/edgebench/terminal_capture.py b/benchmark/edgebench/terminal_capture.py index db7e133eb3..c320b02c09 100644 --- a/benchmark/edgebench/terminal_capture.py +++ b/benchmark/edgebench/terminal_capture.py @@ -49,11 +49,10 @@ def terminal_components(declaration: dict, workspace: str) -> list[dict[str, str class FailurePreservation: """One terminal attempt; a stopped source remains until explicit disposal.""" - def __init__(self, supervisor, components, receipt_path, *, phase_deadline_path=_PHASE_DEADLINE): + def __init__(self, supervisor, components, receipt_path): self.supervisor = supervisor self.components = components self.receipt_path = receipt_path - self.phase_deadline_path = phase_deadline_path self.receipt = None self.retain_source = False @@ -107,7 +106,7 @@ def finish(self, execution): self._persist_receipt() request = self.supervisor.destination / "terminal-request.json" write_json_atomic(request, {"container_id": self.supervisor.container.id, - "components": self.components, "phase_deadline_path": self.phase_deadline_path}, + "components": self.components, "phase_deadline_path": supervisor.capture_phase_deadline_path()}, mode=0o600, fsync_directory=True) point = capture_worker(supervisor.controller_file, token=supervisor.token, capture_id=self.receipt["capture_id"], diff --git a/benchmark/runtime/layout.py b/benchmark/runtime/layout.py index 66df49713f..d946333206 100644 --- a/benchmark/runtime/layout.py +++ b/benchmark/runtime/layout.py @@ -4,6 +4,8 @@ _SRC = f"{_ROOT}/source" _CONTROL = f"{_ROOT}/control" _PHASE_DEADLINE = f"{_CONTROL}/phase-deadline" +_ACTIVE_PHASE_DEADLINE_NAME = "active-phase-deadline" +_ACTIVE_PHASE_DEADLINE = f"{_CONTROL}/{_ACTIVE_PHASE_DEADLINE_NAME}" _CODEX_HOME = f"{_ROOT}/codex-home" _REGISTRY = f"{_CONTROL}/registry.json" _LOOPX_RUNTIME = f"{_ROOT}/state/runtime" diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index e661c489c7..16878547ff 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -185,6 +185,18 @@ def hold(state: Path, reason: str) -> str: raise ValueError(f'Explicit idle handoff {reason}') return str(state) + def activate(state: Path) -> str: + # The retained clock is an immutable recovery witness. Publish only + # this already-admitted phase for both Host capture paths, with no + # rounding up or new scheduler/native allowance. + from .layout import _ACTIVE_PHASE_DEADLINE_NAME + from loopx.file_lock import write_bytes_atomic + + clock = _physical_state(path.with_name(_ACTIVE_PHASE_DEADLINE_NAME)) + write_bytes_atomic(clock, f"{math.floor(phase_deadline)}\n".encode(), + mode=0o600, fsync_directory=True) + return str(state) + with ExitStack() as locks: # The root serializes phase selection. Locking the existing phase files # as well keeps a running/failed phase from moving under that decision. @@ -229,7 +241,7 @@ def hold(state: Path, reason: str) -> str: != retained.get('source_proof_sha256')): raise ValueError('Explicit idle handoff replay differs from its original phase') verify_idle_host_replay(parent, retained, env=env) - return str(predecessor) # Keep any later work or failure in this same phase. + return activate(predecessor) # Keep any later work or failure in this same phase. if (same_phase or hashlib.sha256(predecessor.read_bytes()).hexdigest() != idle_host_state_sha256): raise ValueError('Explicit idle handoff source changed; inspect the retained phase') from .interruption_recovery import read_idle_host @@ -241,7 +253,7 @@ def hold(state: Path, reason: str) -> str: recovered.update(source_state_sha256=idle_host_state_sha256, phase_deadline_epoch=phase_deadline) _create_resume_state(target, predecessor, recovered=recovered, replan_attempts=state.get('replan_attempts'), handoff=handoff, idle_recovery=True) - return str(target) + return activate(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")) @@ -261,7 +273,7 @@ def hold(state: Path, reason: str) -> str: _verify_failed_phase_replay(predecessor, state, failed_phase, states=states, digest=retry_failed_state_sha256, env=env, wake_log_dir=wake_log_dir) - return str(predecessor) # Replay never erases a later failure in this phase. + return activate(predecessor) # Replay never erases a later failure in this phase. if not explicit_failed_phase: raise ValueError('Explicit failed continuation source changed; inspect the retained phase') if not isinstance(prior, dict): @@ -302,7 +314,7 @@ def hold(state: Path, reason: str) -> str: _create_resume_state(target, predecessor, recovered=recovered, replan_attempts=state.get("replan_attempts"), handoff=handoff, failed_recovery=failed_recovery) - return str(target) + return activate(target) if (prior.get("status") != "stopped" or prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE or "host_failure" in prior or _host_pending(predecessor) is None @@ -316,7 +328,7 @@ def hold(state: Path, reason: str) -> str: if target.is_symlink(): raise ValueError("Resumed scheduler state must not be a symlink") if target.exists(): - return str(target) # Never replace a retry/failure in this phase. + return activate(target) # Never replace a retry/failure in this phase. # Only a stop before native dispatch admits this shortcut. An # interrupted native Turn still needs its original recovery/settlement. # Restrict legacy receipt discovery to the source scheduler snapshot; @@ -327,7 +339,7 @@ def hold(state: Path, reason: str) -> str: or result.get("stage") != "execute"): return str(predecessor) _create_resume_state(target, predecessor, handoff=handoff) - return str(target) + return activate(target) def worker_command(env: dict[str, str], *, python: str, source: str, diff --git a/benchmark/tests/test_scheduler_phase_resume.py b/benchmark/tests/test_scheduler_phase_resume.py index 52d1410241..3329e97019 100644 --- a/benchmark/tests/test_scheduler_phase_resume.py +++ b/benchmark/tests/test_scheduler_phase_resume.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +import os from pathlib import Path import shlex import sys @@ -12,6 +13,83 @@ from benchmark.runtime.scheduler import resume_state_file, worker_command +@pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", + reason="Explicit isolated Docker routing qualification") +@pytest.mark.parametrize("continued", [False, True]) +def test_scheduler_continuation_clock_reaches_both_host_capture_paths(tmp_path, monkeypatch, continued): + """Actual guest scheduler + archive IO; only native sealing is substituted.""" + import docker + from benchmark.edgebench import checkpoint_supervisor as host + from benchmark.edgebench.checkpoint_worker import _archive + from benchmark.runtime.layout import _PHASE_DEADLINE + + client = docker.from_env() + image = os.environ.get("LOOPX_CHECKPOINT_WRITER_IMAGE", "python:3.12-slim") + client.images.get(image) # Dependency is provisioned explicitly, never pulled here. + container = client.containers.run(image, ["sleep", "120"], detach=True, + network_disabled=True, mem_limit="256m", nano_cpus=500_000_000, + volumes={str(Path(__file__).resolve().parents[2]): {"bind": "/source", "mode": "ro"}}, + working_dir="/source") + script = r''' +import json, os, sys, time +from pathlib import Path +from benchmark.runtime import scheduler +root=Path('/opt/loopx-benchmark/control'); root.mkdir(parents=True) +continued=CONTINUED +original=1 if continued else int(time.time())+400 +(root/'phase-deadline').write_text(str(original)) +state=root/'scheduler.json' +state.write_text(json.dumps({'host_attempt':{'status':'stopped','scope':'a'*64, + 'attempt':1,'stop_exit_code':75},'last_wake_rc':75,'last_wake_status':'wake_requested_stop'})) +state.with_name(state.name+'.host-attempt').write_text('{"pending":false}') +receipt=root/'wakes/scheduler/receipt.json';receipt.parent.mkdir(parents=True) +receipt.write_text(json.dumps({'budget_exhausted':True,'host_invoked':False,'mode':'heartbeat','stage':'execute'})) +if not continued:state.unlink() +os.environ.update(LOOPX_PHASE_DEADLINE_EPOCH=str(int(time.time())+800 if continued else original),LOOPX_AGENT_ID='agent', + LOOPX_WAKE_LOG_DIR=str(root/'wakes'),LOOPX_EXECUTION_MODE='heartbeat',LOOPX_CLI='/bin/true', + LOOPX_REGISTRY='/registry.json',LOOPX_RUNTIME_ROOT='/runtime',LOOPX_GOAL_ID='fixture') +sys.argv=['scheduler','--python',sys.executable,'--source','/source','--state-file',str(state),'--host-timeout','30'] +os.execvpe=lambda *args:None +scheduler.main() +mailbox=Path('/logs/agent/checkpoint-controller');mailbox.mkdir(parents=True) +(mailbox/'ready.json').write_text(json.dumps({'capture_id':'b'*32,'deadline_epoch':time.time()+90, + 'components':[{'id':'workspace','source_path':'/task'}]})) +''' + try: + result = container.exec_run(["python", "-c", script.replace("CONTINUED", repr(continued))]) + assert result.exit_code == 0, result.output.decode() + original_bytes = _archive(container, _PHASE_DEADLINE, tmp_path / "before").read_bytes() + observed = [] + destination = tmp_path / "points" + monkeypatch.setattr(host, "_controller", lambda _: {}) + supervisor = host.CheckpointSupervisor(tmp_path / "controller.json", destination, + retention_enabled=False, terminal_components=[{"id": "workspace", "source_path": "/task"}], + terminal_receipt=tmp_path / "terminal.json") + destination.mkdir() + supervisor.container, supervisor.token = container, "fixture-token" + + def capture(*args, **kwargs): + request = json.loads(kwargs["request_file"].read_text()) + path = _archive(container, request["phase_deadline_path"], tmp_path / f"read-{len(observed)}") + observed.append((request["phase_deadline_path"], float(path.read_text()) - time.time())) + supervisor.stop.set() + return {"research_point_qualified": False} + + monkeypatch.setattr(host, "capture_worker", capture) + monkeypatch.setattr("benchmark.edgebench.checkpoint_worker.capture_worker", capture) + supervisor._run(container) + supervisor.finish_solver({"execution_started": True, "exit_code": 2}) + expected_path = "/opt/loopx-benchmark/control/active-phase-deadline" if continued else _PHASE_DEADLINE + assert [path for path, _ in observed] == [expected_path, expected_path] + limit = 800 if continued else 400 + assert all(limit - 100 < remaining <= limit for _, remaining in observed) + original = _archive(container, _PHASE_DEADLINE, tmp_path / "original") + assert original.read_bytes() == original_bytes + finally: + container.remove(force=True) + client.close() + + def _stopped(path): receipt = path.parent / "wakes" / path.stem / "receipt.json" receipt.parent.mkdir(parents=True, exist_ok=True) @@ -37,6 +115,9 @@ def test_budget_phase_retries_keep_the_same_state_and_original_receipts(tmp_path end = time.time() + 1000 new = resume_state_file(str(old), phase_deadline=end, agent_id="agent", wake_log_dir=str(tmp_path / "wakes")) assert new != str(old) and Path(new).parent == old.parent + clock = tmp_path / "active-phase-deadline" + admitted_clock = clock.read_bytes() + assert float(admitted_clock) <= end < float(admitted_clock) + 1 Path(new).write_text('{"host_attempt":{"status":"failed"}}') assert resume_state_file(str(old), phase_deadline=end, agent_id="agent", wake_log_dir=str(tmp_path / "wakes")) == new state = json.loads(old.read_bytes()) @@ -46,6 +127,7 @@ def test_budget_phase_retries_keep_the_same_state_and_original_receipts(tmp_path assert json.loads(before)["host_attempt"] == json.loads(old.read_bytes())["host_attempt"] assert old.with_name(old.name + ".host-attempt").read_bytes() == witness assert json.loads(Path(new).read_bytes())["host_attempt"]["status"] == "failed" + assert clock.read_bytes() == admitted_clock def test_repeated_budget_stops_advance_from_the_actual_predecessor(tmp_path): @@ -65,6 +147,8 @@ def test_later_phase_cannot_skip_the_latest_host_attempt(tmp_path, change): root = tmp_path / "scheduler.json" _stopped(root) second = _resume(root, time.time() + 1000) + clock = tmp_path / "active-phase-deadline" + admitted_clock = clock.read_bytes() _stopped(second) state = json.loads(second.read_text()) if change == "hard_failure": @@ -97,6 +181,7 @@ def test_later_phase_cannot_skip_the_latest_host_attempt(tmp_path, change): before = second.read_bytes() later = _resume(root, time.time() + 2000) assert later == second + assert clock.read_bytes() == admitted_clock assert second.read_bytes() == before assert not list(tmp_path.glob("scheduler.phase-*.json"))[1:] @@ -228,4 +313,5 @@ def select(): assert select() == phase brief.write_text(json.dumps({'ok': True, 'handoff_text': 'Different material.'})) env['LOOPX_RECOVERY_HANDOFF_SHA256'] = hashlib.sha256(brief.read_bytes()).hexdigest() - with pytest.raises(ValueError, match='immutable'): select() + with pytest.raises(ValueError, match='immutable'): + select() diff --git a/benchmark/tests/test_terminal_preservation.py b/benchmark/tests/test_terminal_preservation.py index d50b849f15..ec9e177361 100644 --- a/benchmark/tests/test_terminal_preservation.py +++ b/benchmark/tests/test_terminal_preservation.py @@ -27,7 +27,8 @@ def close(): container = SimpleNamespace(id="a" * 64, attrs={"State": { "Running": False, "Paused": False, "Restarting": False, "Pid": 0}}, reload=lambda: None) supervisor = SimpleNamespace(destination=tmp_path, controller_file=tmp_path / "controller.json", - token="dummy-token", container=container, close=close) + token="dummy-token", container=container, close=close, + capture_phase_deadline_path=lambda: "/opt/loopx-benchmark/control/phase-deadline") def capture(*args, **kwargs): calls.append("capture") if fault == "capture":