From 96f54989e8a35bbd038a2ee507e8a86c5ea67020 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:39:54 +0800 Subject: [PATCH] fix(checkpoint): recheck captured session without writing launch locks Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/checkpoint.py | 13 +-- benchmark/runtime/checkpoint_wake.py | 12 +-- .../test_checkpoint_binding_observation.py | 79 +++++++++++++++++++ benchmark/tests/test_checkpoint_handoff.py | 8 +- loopx/hosts/research_checkpoint.py | 37 +++++++-- 5 files changed, 121 insertions(+), 28 deletions(-) create mode 100644 benchmark/tests/test_checkpoint_binding_observation.py diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py index 4fa3a93299..ebfbe0c41d 100644 --- a/benchmark/runtime/checkpoint.py +++ b/benchmark/runtime/checkpoint.py @@ -18,7 +18,8 @@ from loopx.control_plane.effect_runtime import effect_runtime_result from loopx.control_plane.quota.settlement import read_heartbeat_settlement from loopx.hosts.research_checkpoint import ( - _participant_id, native_binding_participant, research_participants, verify_research_participants, + _participant_id, native_binding_participant, verify_native_binding_participant, + research_participants, verify_research_participants, ) from .checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS @@ -335,11 +336,9 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, if request.get("native_session"): from loopx.hosts.codex_checkpoint import native_checkpoint_reader, native_session_checkpoint from .checkpoint_continuation import restored_capture_history - from .codex import Execution - from .sessions import BenchmarkSessionWake session = receipt.get("session") or {} - components, source_wake = native_binding_participant(env, components, metadata) + components, source_wake, registry_bytes = native_binding_participant(env, components, metadata) native_options = restored_capture_history(env, restore_gate, session_id=session.get("session_id", "")) reader = readers.enter_context(native_checkpoint_reader(env, timeout=max(.1, deadline - time.time()))) @@ -374,11 +373,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, **native_options) if confirmed != native: raise ValueError("native checkpoint cursor or participant basis moved before sealing") - confirmed_wake = BenchmarkSessionWake(env, Execution(context="resume", - sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access")), {}) - if ((confirmed_wake.session_id, confirmed_wake.goal_ref, confirmed_wake.digest) != - (source_wake.session_id, source_wake.goal_ref, source_wake.digest)): - raise ValueError("native checkpoint source session binding moved before sealing") + verify_native_binding_participant(env, source_wake, registry_bytes) if request.get("loopx_state"): verify_research_participants(env, components, metadata, research, probe["source_witness"], deadline) if opportunity is not None and opportunity["native_incumbent"] is not None: diff --git a/benchmark/runtime/checkpoint_wake.py b/benchmark/runtime/checkpoint_wake.py index 966d8913f4..3548c2ba01 100644 --- a/benchmark/runtime/checkpoint_wake.py +++ b/benchmark/runtime/checkpoint_wake.py @@ -125,8 +125,8 @@ def _capture(self, held, deadline): if self.request.get("native_session"): from loopx.hosts.codex_checkpoint import native_checkpoint_reader, native_session_checkpoint from .checkpoint_continuation import restored_capture_history - from .checkpoint import native_binding_participant - components, source_wake = native_binding_participant(self.env, components, metadata) + from .checkpoint import native_binding_participant, verify_native_binding_participant + components, source_wake, registry_bytes = native_binding_participant(self.env, components, metadata) native_options = restored_capture_history(self.env, self.receipt.get("checkpoint_restore"), session_id=self.receipt["session"]["session_id"]) guard = {"cli": self.env["LOOPX_CLI"], "workspace": self.env["LOOPX_PROJECT"], @@ -160,13 +160,7 @@ def _capture(self, held, deadline): confirmed = native_session_checkpoint(self.env, self.receipt["session"]["session_id"], timeout=max(.1, window - time.time()), interrupted_guard=guard, transport=reader, **native_options) - from .sessions import BenchmarkSessionWake - from .codex import Execution - confirmed_wake = BenchmarkSessionWake(self.env, Execution(context="resume", - sandbox=self.env.get("LOOPX_CODEX_SANDBOX", "danger-full-access")), {}) - if ((confirmed_wake.session_id, confirmed_wake.goal_ref, confirmed_wake.digest) != - (source_wake.session_id, source_wake.goal_ref, source_wake.digest)): - raise ValueError("interrupted source session binding moved before sealing") + verify_native_binding_participant(self.env, source_wake, registry_bytes) if confirmed != native: raise ValueError("interrupted native cursor moved before sealing") if self.request.get("loopx_state"): diff --git a/benchmark/tests/test_checkpoint_binding_observation.py b/benchmark/tests/test_checkpoint_binding_observation.py new file mode 100644 index 0000000000..47ec1ce7d1 --- /dev/null +++ b/benchmark/tests/test_checkpoint_binding_observation.py @@ -0,0 +1,79 @@ +"""A capture recheck must not mutate its source, or overlook a real change.""" +import json +import sys + +import pytest + + +@pytest.fixture +def source(tmp_path, capsys): + from loopx.cli import main + from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession + from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session + from loopx.hosts.research_checkpoint import native_binding_participant + + project, runtime, home = (tmp_path / name for name in ("project", "runtime", "home")) + for path in (project, runtime, home): + path.mkdir() + registry = project / ".loopx/registry.json" + (home / "config.toml").write_text("[features]\ngoals=false\n") + prefix = ["--format", "json", "--registry", str(registry), "--runtime-root", str(runtime)] + for argv in (["bootstrap", "--project", str(project), "--goal-id", "source", + "--objective", "Preserve a synthetic candidate.", "--no-global-sync"], + ["configure-goal", "--goal-id", "source", "--registered-agent", "agent", "--execute"]): + code = main([*prefix, *argv]) + assert code == 0, capsys.readouterr().out + env = {"LOOPX_REGISTRY": str(registry), "LOOPX_RUNTIME_ROOT": str(runtime), + "LOOPX_PROJECT": str(project), "LOOPX_GOAL_ID": "source", "LOOPX_AGENT_ID": "agent", + "CODEX_BIN": sys.executable, "CODEX_HOME": str(home), "MODEL_NAME": "synthetic", + "REASONING_EFFORT": "high", "LOOPX_CODEX_SANDBOX": "read-only"} + session = CodexAgentSession(env, context="resume", sandbox="read-only", receipt={}) + _store_codex_cli_session(runtime, lineage=session.lineage, session_scope="agent", + session_id="source-session", goal_ref=session.goal_ref, session_profile_digest=session.digest) + components, session, raw = native_binding_participant(env, + [{"id": "project", "source_path": str(project)}], {"session": {"session_id": "source-session"}}) + return env, components, session, raw + + +@pytest.mark.parametrize("fault", [None, "registry", "session", "profile", "provenance", "config", "missing"]) +def test_binding_observation_preserves_witness_and_rejects_changes(source, tmp_path, fault): + from pathlib import Path + from loopx.control_plane.effect_runtime import effect_runtime_result + from loopx.control_plane.turn_driver.codex_sessions import _session_path + from loopx.hosts.research_checkpoint import verify_native_binding_participant + + env, components, session, raw = source + before = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components}) + binding_path = _session_path(session.root, session.lineage, "agent") + if fault == "registry": + registry = Path(env["LOOPX_REGISTRY"]) + registry.write_bytes(registry.read_bytes() + b"\n") + elif fault == "config": + (Path(env["CODEX_HOME"]) / "config.toml").write_text("model = 'changed'\n") + elif fault == "missing": + binding_path.unlink() + elif fault: + value = json.loads(binding_path.read_bytes()) + value[{"session": "session_id", "profile": "session_profile_digest", + "provenance": "checkpoint_artifact_sha256"}[fault]] = "changed" + binding_path.write_text(json.dumps(value)) + if fault: + with pytest.raises(ValueError, match="registry moved|binding or profile moved"): + verify_native_binding_participant(env, session, raw) + else: + verify_native_binding_participant(env, session, raw) + verify_native_binding_participant(env, session, raw) + after = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components}) + assert before == after + sealed = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", + "components": components, "destination": str(tmp_path / "point"), "point_metadata": {}, + "expected_source_sha256": before["source_sha256"]}) + assert sealed["sha256"] + # Even advisory lock bytes remain part of the artifact fence. This fix + # must not introduce a wildcard exclusion for lock-named task files. + lock = Path(env["LOOPX_REGISTRY"] + ".lock") + lock.write_text("changed after preparation") + with pytest.raises(Exception, match="changed|moved|witness"): + effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", + "components": components, "destination": str(tmp_path / "later"), "point_metadata": {}, + "expected_source_sha256": before["source_sha256"]}) diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py index 40ffb5cae7..7a8bf5fb25 100644 --- a/benchmark/tests/test_checkpoint_handoff.py +++ b/benchmark/tests/test_checkpoint_handoff.py @@ -130,8 +130,8 @@ def test_capture_preparation_uses_one_window_without_renewing_phase(tmp_path, mo binding = SimpleNamespace(session_id="native-session", goal_ref="source", digest="profile") monkeypatch.setattr(checkpoint, "research_participants", lambda env, components, *a: (components, {})) monkeypatch.setattr(checkpoint, "verify_research_participants", lambda *a: None) - monkeypatch.setattr(checkpoint, "native_binding_participant", lambda env, components, *a: (components, binding)) - monkeypatch.setattr("benchmark.runtime.sessions.BenchmarkSessionWake", lambda *a, **k: binding) + monkeypatch.setattr(checkpoint, "native_binding_participant", lambda env, components, *a: (components, binding, b"registry")) + monkeypatch.setattr(checkpoint, "verify_native_binding_participant", lambda *a: None) monkeypatch.setattr("benchmark.runtime.checkpoint_continuation.restored_capture_history", lambda *a, **k: {}) @contextmanager def reader(*a, **k): @@ -361,8 +361,8 @@ def test_interrupted_capture_keeps_reader_through_publication_and_cleans_up(tmp_ held["path"].write_text("original guard") binding = SimpleNamespace(session_id="session", goal_ref="goal", digest="profile") monkeypatch.setattr("benchmark.runtime.checkpoint.native_binding_participant", - lambda env, components, metadata: (components, binding)) - monkeypatch.setattr("benchmark.runtime.sessions.BenchmarkSessionWake", lambda *a, **k: binding) + lambda env, components, metadata: (components, binding, b"registry")) + monkeypatch.setattr("benchmark.runtime.checkpoint.verify_native_binding_participant", lambda *a: None) events = [] token = object() native_options = {"retained_history": {"turn_ids": ["source"], "visible_history_sha256": "source-digest"}, diff --git a/loopx/hosts/research_checkpoint.py b/loopx/hosts/research_checkpoint.py index 62e3332eea..d1fb7a7b62 100644 --- a/loopx/hosts/research_checkpoint.py +++ b/loopx/hosts/research_checkpoint.py @@ -412,8 +412,11 @@ def _loopx_state_components(env, components, output_dir): def native_binding_participant(env, components, metadata): from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession + registry_bytes = Path(env["LOOPX_REGISTRY"]).read_bytes() source_wake = CodexAgentSession(env, context="resume", sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access"), receipt={}) + if Path(env["LOOPX_REGISTRY"]).read_bytes() != registry_bytes: + raise ValueError("checkpoint registry moved during session observation") if source_wake.session_id != metadata["session"].get("session_id"): raise ValueError("native checkpoint differs from the current source session binding") # This sideband record belongs to the existing session owner. It @@ -427,7 +430,32 @@ def native_binding_participant(env, components, metadata): metadata["session"] = {**metadata["session"], "session_profile_digest": source_wake.digest} if source_wake.binding.get("checkpoint_context_refresh") is not None: metadata["session"]["checkpoint_context_refresh"] = source_wake.binding["checkpoint_context_refresh"] - return components, source_wake + return components, source_wake, registry_bytes + + +def verify_native_binding_participant(env, source_wake, registry_bytes): + """Recheck the accepted binding without creating another launch admission. + + The initial session owner selects the binding and Goal lifetime. After the + byte probe, constructing that owner again would write its advisory locks + and invalidate the snapshot. Compare the complete binding, profile and + unchanged registry instead; this observation grants no launch authority. + """ + import shutil + from loopx.control_plane.turn_driver.codex_sessions import ( + codex_session_profile_digest, select_codex_cli_session, + ) + if Path(env["LOOPX_REGISTRY"]).read_bytes() != registry_bytes: + raise ValueError("checkpoint registry moved before sealing") + binding = select_codex_cli_session(source_wake.root, lineage=source_wake.lineage, session_scope="agent") + binary = shutil.which(env["CODEX_BIN"]) + if binary is None: + raise ValueError("checkpoint native binary is unavailable") + digest = codex_session_profile_digest(project=Path(env["LOOPX_PROJECT"]), codex_bin=binary, + home=Path(env["CODEX_HOME"]), model=env["MODEL_NAME"], reasoning_effort=env["REASONING_EFFORT"], + sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access")) + if binding != source_wake.binding or digest != source_wake.digest: + raise ValueError("checkpoint source session binding or profile moved before sealing") def research_participants(env, components, metadata, destination, evidence_path, deadline): @@ -517,7 +545,7 @@ def settled(): destination, Path(temporary) / "research-context.json", deadline) if research is None: raise ValueError("research capture requires canonical source coordination") - components, observed_session = native_binding_participant(env, components, metadata) + components, observed_session, registry_bytes = native_binding_participant(env, components, metadata) # Reopening the native reader after probing can checkpoint its history # database and invalidate our own byte witness. Keep one inspection # process open; the original logical reread and byte fence both remain. @@ -535,10 +563,7 @@ def settled(): timeout=max(.1, deadline-time.time()), retry_safe=False, large_local_snapshot=True) if native_session_checkpoint(env, session.session_id, timeout=max(.1, deadline-time.time()), transport=reader) != native: raise ValueError("native checkpoint source moved before sealing") - _, confirmed = native_binding_participant(env, components, metadata) - if ((confirmed.session_id, confirmed.goal_ref, confirmed.digest) != - (observed_session.session_id, observed_session.goal_ref, observed_session.digest)): - raise ValueError("research checkpoint session binding moved before sealing") + verify_native_binding_participant(env, observed_session, registry_bytes) verify_research_participants(env, components, metadata, research, probe["source_witness"], deadline) if settled().identity.value != original.identity.value: raise ValueError("research checkpoint source settlement moved")