From 2d47d037ba6b4722cdcb81309003ce526a7bbc67 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 14:49:01 +0800 Subject: [PATCH] fix(checkpoint): prepare session before the registry binding fence Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../tests/test_checkpoint_binding_lock.py | 102 ++++++++++++++++++ loopx/hosts/codex_checkpoint.py | 9 +- 2 files changed, 108 insertions(+), 3 deletions(-) create mode 100644 benchmark/tests/test_checkpoint_binding_lock.py diff --git a/benchmark/tests/test_checkpoint_binding_lock.py b/benchmark/tests/test_checkpoint_binding_lock.py new file mode 100644 index 0000000000..014ccb2042 --- /dev/null +++ b/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) diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index 4692ab1027..76475ea0a2 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/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,6 +525,8 @@ 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"]): @@ -529,8 +534,6 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh 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}