fix(checkpoint): avoid nested registry lock during session admission

benchmark/tests/test_checkpoint_binding_lock.py

@@ -0,0 +1,102 @@
+"""Checkpoint binding keeps the real registry fence without nesting its owner."""
+
+import hashlib
+import json
+import sys
+
+import pytest
+
+
+@pytest.mark.parametrize("replay", [False, True])
+@pytest.mark.parametrize("change_goal", [False, True])
+def test_checkpoint_binding_and_readback_recheck_destination(
+    tmp_path, monkeypatch, capsys, replay, change_goal,
+):
+    from loopx.cli import main
+    from loopx.control_plane import effect_runtime
+    from loopx.control_plane.turn_driver import codex_agent_session
+    from loopx.control_plane.turn_driver.codex_sessions import (
+        _store_codex_cli_session, select_codex_cli_session,
+    )
+    from loopx.hosts import codex_checkpoint
+
+    registry, runtime = tmp_path / "registry.json", tmp_path / "runtime"
+    workspace, home, unit = (tmp_path / name for name in ("workspace", "home", "point"))
+    for path in (workspace, home, unit):
+        path.mkdir()
+    config = home / "config.toml"
+    config.write_text("[features]\ngoals=false\n")
+    prefix = ["--format", "json", "--registry", str(registry), "--runtime-root", str(runtime)]
+
+    def cli(*args):
+        code = main([*prefix, *args])
+        output = capsys.readouterr().out
+        assert code == 0, output
+
+    cli("bootstrap", "--project", str(workspace), "--goal-id", "destination",
+        "--objective", "Verify a synthetic retained candidate.",
+        "--adapter-kind", "read_only_project_map_v0", "--adapter-status", "connected-read-only",
+        "--write-scope", "**", "--no-global-sync")
+    cli("configure-goal", "--goal-id", "destination", "--registered-agent", "receiver",
+        "--agent-work-mode", "receiver=active", "--execute")
+    env = {"LOOPX_REGISTRY": str(registry), "LOOPX_RUNTIME_ROOT": str(runtime),
+           "LOOPX_PROJECT": str(workspace), "LOOPX_GOAL_ID": "destination",
+           "LOOPX_AGENT_ID": "receiver", "CODEX_BIN": sys.executable,
+           "CODEX_HOME": str(home), "MODEL_NAME": "synthetic", "REASONING_EFFORT": "high"}
+    session_type = codex_agent_session.CodexAgentSession
+    prepared = session_type(env, context="resume", sandbox="read-only", receipt={})
+    digest = "a" * 64
+    native = {"session_id": "source-session", "completed_turn_id": "retained-turn",
+              "turn_ids": ["retained-turn"], "visible_history_sha256": "b" * 64,
+              "host_version": "fixture", "guest_home": str(home), "guest_workspace": str(workspace),
+              "config_sha256": hashlib.sha256(config.read_bytes()).hexdigest()}
+    (unit / "checkpoint.json").write_text(json.dumps({"point_metadata": {
+        "session": {"session_profile_digest": prepared.digest}, "native_session": native,
+        "loopx_state": {"goal_id": "source", "agent_id": "source-agent"}}}))
+    if replay:
+        _store_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent",
+            session_id="destination-session", goal_ref=prepared.goal_ref,
+            session_profile_digest=prepared.digest, checkpoint_artifact_sha256=digest,
+            checkpoint_artifact_root=str(unit))
+    before = select_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent")
+
+    # Native storage/inference is outside this lock regression. Registry,
+    # launch-authority capture and session binding use their production owners.
+    real_effect = effect_runtime.effect_runtime_result
+
+    def verify(name, payload, **kwargs):
+        if name == "runtime.checkpoint_artifacts":
+            assert payload == {"action": "verify", "root": str(unit), "expected_sha256": digest}
+            return {"verified": True}
+        return real_effect(name, payload, **kwargs)
+
+    forks = []
+
+    def fork(*args, **kwargs):
+        forks.append(True)
+        return {"session_id": "destination-session", "model_calls": 0}
+
+    def construct(*args, **kwargs):
+        session = session_type(*args, **kwargs)
+        if change_goal:
+            cli("configure-goal", "--goal-id", "destination",
+                "--agent-work-mode", "receiver=monitor_only", "--execute")
+        return session
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", verify)
+    monkeypatch.setattr(codex_checkpoint, "fork_native_checkpoint", fork)
+    monkeypatch.setattr(codex_checkpoint, "native_session_checkpoint", lambda *a, **k: {"metadata": native})
+    monkeypatch.setattr(codex_agent_session, "CodexAgentSession", construct)
+    if change_goal:
+        with pytest.raises(ValueError, match="Goal configuration changed"):
+            codex_checkpoint.admit_native_checkpoint_session(
+                env, unit, digest, config, sandbox="read-only", timeout=30)
+        assert select_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent") == before
+    else:
+        result = codex_checkpoint.admit_native_checkpoint_session(
+            env, unit, digest, config, sandbox="read-only", timeout=30)
+        assert result["session_binding_admitted"] and result["session_id"] == "destination-session"
+        binding = select_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent")
+        assert binding["checkpoint_artifact_sha256"] == digest
+        assert binding["checkpoint_artifact_root"] == str(unit)
+    assert len(forks) == (0 if replay else 1)

loopx/hosts/codex_checkpoint.py

@@ -492,12 +492,15 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh
         # The original session binding is also the admission receipt. Only a
         # source-bound committed admission can recover a lost controller reply;
         # an unrelated binding or a partial native fork is never replaced.
+        # Session construction captures launch authority through the registry
+        # owner too. Construct before taking the non-reentrant binding fence;
+        # recheck current Goal and Agent authority inside it before readback.
+        wake = CodexAgentSession(env, context="resume", sandbox=sandbox, receipt={})
         with project_registry_transaction(registry, operation="checkpoint_session_binding", agent_id=lineage["agent_id"]):
             current = load_goal_from_registry(registry, lineage["goal_id"])
             require_registered_agent_id(registry_path=registry, goal_id=lineage["goal_id"], agent_id=lineage["agent_id"])
             if current != goal:
                 raise ValueError("destination Goal configuration changed during native checkpoint readback")
-            wake = CodexAgentSession(env, context="resume", sandbox=sandbox, receipt={})
             binding = select_codex_cli_session(wake.root, lineage=lineage, session_scope="agent",
                                               goal_admission=wake.admission)
             if not binding or binding.get("checkpoint_artifact_sha256") != expected_sha256:
@@ -522,15 +525,15 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh
                 "session_binding_admitted": True, "session_profile_digest": wake.digest}
     result = fork_native_checkpoint(env, unit, expected_sha256, config_path,
                                     timeout=max(.1, deadline - time.monotonic()))
+    receipt: dict[str, Any] = {}
+    wake = CodexAgentSession(env, context="resume", sandbox=sandbox, receipt=receipt)
     # Hold the existing registry owner's fence through the session write. This
     # does not mint a lease, task or work admission from historical state.
     with project_registry_transaction(registry, operation="checkpoint_session_binding", agent_id=lineage["agent_id"]):
         current = load_goal_from_registry(registry, lineage["goal_id"])
         require_registered_agent_id(registry_path=registry, goal_id=lineage["goal_id"], agent_id=lineage["agent_id"])
         if current != goal:
             raise ValueError("destination Goal configuration changed during native checkpoint reconstruction")
-        receipt: dict[str, Any] = {}
-        wake = CodexAgentSession(env, context="resume", sandbox=sandbox, receipt=receipt)
         wake._bind_verified_fork(result["session_id"], source_profile, expected_sha256, str(unit))
     return {**result, "destination_goal_ref": wake.goal_ref, "destination_agent_id": lineage["agent_id"],
             "session_binding_admitted": True, "session_profile_digest": wake.digest}