fix(checkpoint): recheck native binding without mutating capture witness

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:

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

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

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"},

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