From d6c91cf7f1415f8312605f2c4788c4f23e223f1f Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 12 Nov 2026 13:35:07 +0000 Subject: [PATCH 1/8] fix(turn): prepare managed Todo claims before Codex delivery Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/cli_commands/turn.py | 27 ++- .../codex_execution_preparation.py | 76 ++++++++ .../test_turn_run_once_claim_preparation.py | 183 ++++++++++++++++++ tests/test_loopx_turn_driver.py | 49 ++++- 4 files changed, 331 insertions(+), 4 deletions(-) create mode 100644 tests/control_plane/test_turn_run_once_claim_preparation.py diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 71dd9798fb..6ed3eaeca8 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -89,6 +89,27 @@ def handle_turn_command( registry_path=registry_path, runtime_root_override=runtime_root_arg, ) + execution_preparer = native_execution_preparer + if ( + execution_preparer is None + and args.turn_command == "run-once" + and args.host == "codex-cli" + and bool(getattr(args, "execute", False)) + ): + from ..control_plane.turn_driver.codex_execution_preparation import ( + builtin_execution_preparation_claim, + ) + + project_cwd = Path(args.project).expanduser().resolve() + + def execution_preparer(decision: Mapping[str, Any]) -> str | None: + return builtin_execution_preparation_claim( + decision, + registry_path=registry_path.expanduser().resolve(), + runtime_root=runtime_root.resolve(), + cwd=project_cwd, + ) + resume_instance = getattr(args, "resume_turn_instance_id", None) if resume_instance is not None: from ..control_plane.turn_driver.journal_store import find_loopx_turn_key_by_instance @@ -271,8 +292,8 @@ def read_decision_owner(): scheduler_context = decision_owner.scheduler_execution_context decision = compose_turn_decision(decision_owner.resolve(), scheduler_execution_context=scheduler_context) resume_requested, session_binding = resolve_turn_resume_session_binding(args) - if native_execution_preparer is not None and args.turn_command == "run-once" and args.execute: - prepared_todo = native_execution_preparer(decision) + if execution_preparer is not None and args.turn_command == "run-once" and args.execute: + prepared_todo = execution_preparer(decision) if prepared_todo is not None: # Host preparation changes canonical lease state. Plan and # settle against a fresh read of the same Turn and Todo. @@ -284,7 +305,7 @@ def read_decision_owner(): if (preparation.get("state") != "ready" or preparation.get("todo_id") != prepared_todo or (decision.get("selected_todo") or {}).get("todo_id") != prepared_todo or decision.get("normal_delivery_allowed") is not True): - raise ValueError("native task preparation did not make the same Todo ready") + raise ValueError("execution preparation did not make the same Todo ready") # Operation-equipped app-server sessions retain their Todo-bound # approval/handoff contract; this option controls exec conversation reuse. session_scope = (args.session_scope if args.host in {"codex-cli", "codex-app-server"} diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index 7c109f832e..bbf6901132 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -152,6 +152,82 @@ def prepared_capture_reference(payload: Mapping[str, Any], digest: str) -> dict[ } +class ExecutionPreparationClaimFailed(ValueError): + """The decision's advertised managed claim was refused before delivery.""" + + code = "turn_execution_preparation_claim_failed" + + +def builtin_execution_preparation_claim( + decision: Mapping[str, Any], + *, + registry_path: Path, + runtime_root: Path, + cwd: Path, + runner: CommandRunner | None = None, + timeout: float = 30, +) -> str | None: + """Execute the exact ordinary claim a decision advertised for a Host. + + The decision stays the owner of readiness: this adapter runs only the + claim the decision already published as Host-executable, then the caller + re-reads the same decision and must observe the same Todo ready. A + preparation this adapter does not recognise is left untouched so its + original typed wait keeps its meaning. + """ + + preparation = decision.get("execution_preparation") + selected = decision.get("selected_todo") + if not isinstance(preparation, Mapping) or not isinstance(selected, Mapping): + return None + if preparation.get("state") != "claim_required": + return None + claim = preparation.get("claim") + if not isinstance(claim, Mapping) or claim.get("host_execution_allowed") is not True: + return None + goal_id = str(decision.get("goal_id") or "").strip() + decision_agent_id = str( + decision.get("agent_id") or selected.get("agent_id") or "" + ).strip() + agent_id = str(preparation.get("agent_id") or "").strip() + todo_id = str(preparation.get("todo_id") or "").strip() + expected_claim = [ + "todo", "claim", "--goal-id", goal_id, "--todo-id", todo_id, + "--agent-id", agent_id, "--claimed-by", agent_id, + ] + if ( + preparation.get("schema_version") != "execution_preparation_v0" + or preparation.get("delivery_allowed") is not False + or decision.get("normal_delivery_allowed") is not False + or decision.get("should_run") is not True + or not decision_agent_id + or decision_agent_id != agent_id + or not goal_id + or not agent_id + or not todo_id + or str(selected.get("todo_id") or "").strip() != todo_id + or claim.get("cli_args") != expected_claim + or not registry_path.is_absolute() + or not runtime_root.is_absolute() + or not cwd.is_absolute() + ): + return None + try: + _run_json( + expected_claim, + registry_path=registry_path, + runtime_root=runtime_root, + cwd=cwd, + timeout=timeout, + runner=runner or _default_runner, + ) + except (OSError, subprocess.SubprocessError, RuntimeError) as exc: + raise ExecutionPreparationClaimFailed( + f"managed claim for Todo {todo_id} was refused before delivery: {exc}" + ) from exc + return todo_id + + def _native_claim_operation_id(request: Mapping[str, Any]) -> str: """Reuse the captured managed claim identity on an authorized retry.""" digest = request.get("claim_source_sha256") diff --git a/tests/control_plane/test_turn_run_once_claim_preparation.py b/tests/control_plane/test_turn_run_once_claim_preparation.py new file mode 100644 index 0000000000..2601479848 --- /dev/null +++ b/tests/control_plane/test_turn_run_once_claim_preparation.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +from loopx.control_plane.turn_driver.codex_execution_preparation import ( + ExecutionPreparationClaimFailed, + builtin_execution_preparation_claim, +) + + +GOAL = "goal-claim-preparation" +TODO = "todo_claim_preparation" +AGENT = "agent-a" + + +def _decision(*, state: str = "claim_required", host_execution_allowed: bool = True, + cli_args: list[str] | None = None, should_run: bool = True, + todo_id: str = TODO, normal_delivery_allowed: bool | None = None, + decision_agent_id: str = AGENT) -> dict[str, object]: + advertised = ["todo", "claim", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", AGENT, "--claimed-by", AGENT] + return { + "ok": True, + "schema_version": "quota_should_run_decision_v0", + "goal_id": GOAL, + "agent_id": decision_agent_id, + "should_run": should_run, + "normal_delivery_allowed": ( + state == "ready" if normal_delivery_allowed is None + else normal_delivery_allowed + ), + "selected_todo": {"todo_id": TODO}, + "execution_preparation": { + "schema_version": "execution_preparation_v0", + "todo_id": todo_id, + "agent_id": AGENT, + "handoff_mode": "hard_lease", + "state": state, + "delivery_allowed": state == "ready", + "reason_code": "task_lease_missing", + "claim": { + "command": "loopx todo claim", + "cli_args": advertised if cli_args is None else cli_args, + "host_execution_allowed": host_execution_allowed, + "postcondition": ( + "rerun quota should-run and require " + "execution_preparation.state=ready" + ), + }, + }, + } + + +class _Completed: + returncode = 0 + + def __init__(self, payload: dict[str, object]) -> None: + self.stdout = json.dumps(payload, ensure_ascii=False) + self.stderr = "" + + +def test_the_advertised_claim_runs_once_before_delivery(tmp_path: Path) -> None: + registry = tmp_path / "registry.json" + runtime_root = tmp_path / "runtime" + project = tmp_path / "project" + project.mkdir() + calls: list[tuple[list[str], Path]] = [] + + def run(command, *, cwd, timeout): + calls.append((list(command), cwd)) + return _Completed({"ok": True}) + + prepared = builtin_execution_preparation_claim( + _decision(), registry_path=registry, runtime_root=runtime_root, + cwd=project, runner=run, + ) + + assert prepared == TODO + assert len(calls) == 1 + command, cwd = calls[0] + assert cwd == project + assert command[-16:] == [ + "--registry", str(registry), "--runtime-root", str(runtime_root), + "--format", "json", "todo", "claim", "--goal-id", GOAL, + "--todo-id", TODO, "--agent-id", AGENT, "--claimed-by", AGENT, + ] + + +@pytest.mark.parametrize( + "state, host_execution_allowed", + [("ready", True), ("claim_required", False), ("selection_required", True)], +) +def test_preparation_this_adapter_cannot_own_stays_a_typed_wait( + tmp_path: Path, state: str, host_execution_allowed: bool, +) -> None: + def run(command, *, cwd, timeout): # pragma: no cover - must not run + raise AssertionError(f"unexpected claim: {command}") + + assert builtin_execution_preparation_claim( + _decision(state=state, host_execution_allowed=host_execution_allowed), + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", + cwd=tmp_path, runner=run, + ) is None + + +@pytest.mark.parametrize( + "cli_args, should_run, todo_id", + [ + (["todo", "claim", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", "agent-b", "--claimed-by", "agent-b"], True, TODO), + (["todo", "claim"], True, TODO), + (None, False, TODO), + (None, True, "todo_other"), + ], +) +def test_only_the_exact_advertised_claim_is_executed( + tmp_path: Path, cli_args: list[str] | None, should_run: bool, todo_id: str, +) -> None: + def run(command, *, cwd, timeout): # pragma: no cover - must not run + raise AssertionError(f"unexpected claim: {command}") + + assert builtin_execution_preparation_claim( + _decision(cli_args=cli_args, should_run=should_run, todo_id=todo_id), + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", + cwd=tmp_path, runner=run, + ) is None + + +@pytest.mark.parametrize( + "normal_delivery_allowed, decision_agent_id", + [(True, AGENT), (False, "agent-b")], +) +def test_inconsistent_decision_identity_is_not_claimed( + tmp_path: Path, normal_delivery_allowed: bool, decision_agent_id: str, +) -> None: + def run(command, *, cwd, timeout): # pragma: no cover - must not run + raise AssertionError(f"unexpected claim: {command}") + + assert builtin_execution_preparation_claim( + _decision( + normal_delivery_allowed=normal_delivery_allowed, + decision_agent_id=decision_agent_id, + ), + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", + cwd=tmp_path, runner=run, + ) is None + + +def test_a_timed_out_claim_surfaces_as_preparation_failure(tmp_path: Path) -> None: + def run(command, *, cwd, timeout): + raise subprocess.TimeoutExpired(command, timeout) + + with pytest.raises(ExecutionPreparationClaimFailed) as refused: + builtin_execution_preparation_claim( + _decision(), registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", cwd=tmp_path, runner=run, + ) + + assert refused.value.code == "turn_execution_preparation_claim_failed" + assert TODO in str(refused.value) + + +def test_a_refused_claim_surfaces_the_atomic_authority_reason(tmp_path: Path) -> None: + def run(command, *, cwd, timeout): + return _Completed({ + "ok": False, + "reason": "Todo claimed_by does not grant this actor update authority", + "reason_code": "update_owner_mismatch", + }) + + with pytest.raises(ExecutionPreparationClaimFailed) as refused: + builtin_execution_preparation_claim( + _decision(), registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", cwd=tmp_path, runner=run, + ) + + assert refused.value.code == "turn_execution_preparation_claim_failed" + assert "Todo claimed_by does not grant this actor update authority" in str(refused.value) + assert TODO in str(refused.value) diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index b748707010..4d122de14c 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1126,6 +1126,52 @@ def test_default_agent_session_reuses_planning_across_completed_todos(tmp_path, } +def test_run_once_claims_zero_lease_before_codex_host(tmp_path, monkeypatch): + from tests.test_loopx_turn_codex_cli import _fake_codex + + project, runtime, registry = _write_live_fixture(tmp_path, handoff_mode="hard_lease") + binary, log = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + + output = io.StringIO() + with contextlib.redirect_stdout(output): + guard_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "quota", "should-run", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--context-delivery", "off", + ]) + before = read_cli_policy_payload(output.getvalue(), returncode=guard_code) + assert guard_code == 0, before + preparation = before["execution_preparation"] + assert preparation["state"] == "claim_required", before + assert preparation["reason_code"] == "task_lease_missing", before + assert preparation["claim"]["host_execution_allowed"] is True, before + assert before["normal_delivery_allowed"] is False, before + assert not log.exists() + + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + result = json.loads(output.getvalue()) + assert exit_code == 0, result + assert result["status"] == "committed", result + calls = [json.loads(line) for line in log.read_text().splitlines()] + assert len(calls) == 1, calls + + def test_turn_plan_transaction_key_is_stable_and_todo_scoped() -> None: first = build_loopx_turn_plan( _envelope(), @@ -1494,6 +1540,7 @@ def _write_live_fixture( todo_metadata_extra: str = "", extra_agent_todo_lines: tuple[str, ...] = (), provider: str = "file", + handoff_mode: str = "soft_claim", ) -> tuple[Path, Path, Path]: project = root / "project" runtime = root / "runtime" @@ -1574,7 +1621,7 @@ def _write_live_fixture( ) projection = build_todo_runtime_shadow_projection( goal_id="loopx-turn-fixture", - handoff_mode="soft_claim", + handoff_mode=handoff_mode, todos=parsed["agent_todos"]["items"], ) initialize_canonical_authority( From c576f64c17c2bcb12d527479f8be3e659aa2d0d8 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 12 Nov 2026 14:02:21 +0000 Subject: [PATCH 2/8] fix(turn): keep the advertised claim and caller Agent exact The builtin preparation adapter rebuilt the advertised claim from whitespace-trimmed identities and accepted a decision Agent that only matched the projected preparation. Rebuild the same argv from exact decision strings, require the caller, decision, preparation, selected Todo and identity contract to agree, and refuse anything else back to its typed wait. Cover the PowerShell launcher route, the sqlite provider, and the e2e claim-before-host path. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/cli_commands/turn.py | 5 +- .../codex_execution_preparation.py | 64 ++++++--- .../test_turn_run_once_claim_preparation.py | 127 ++++++++++++++++-- tests/test_loopx_turn_driver.py | 11 +- 4 files changed, 177 insertions(+), 30 deletions(-) diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 6ed3eaeca8..3ba5edac91 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -90,6 +90,7 @@ def handle_turn_command( runtime_root_override=runtime_root_arg, ) execution_preparer = native_execution_preparer + preparation_error = "native task preparation did not make the same Todo ready" if ( execution_preparer is None and args.turn_command == "run-once" @@ -101,10 +102,12 @@ def handle_turn_command( ) project_cwd = Path(args.project).expanduser().resolve() + preparation_error = "execution preparation did not make the same Todo ready" def execution_preparer(decision: Mapping[str, Any]) -> str | None: return builtin_execution_preparation_claim( decision, + agent_id=args.agent_id, registry_path=registry_path.expanduser().resolve(), runtime_root=runtime_root.resolve(), cwd=project_cwd, @@ -305,7 +308,7 @@ def read_decision_owner(): if (preparation.get("state") != "ready" or preparation.get("todo_id") != prepared_todo or (decision.get("selected_todo") or {}).get("todo_id") != prepared_todo or decision.get("normal_delivery_allowed") is not True): - raise ValueError("execution preparation did not make the same Todo ready") + raise ValueError(preparation_error) # Operation-equipped app-server sessions retain their Todo-bound # approval/handoff contract; this option controls exec conversation reuse. session_scope = (args.session_scope if args.host in {"codex-cli", "codex-app-server"} diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index bbf6901132..4c2d12d4d5 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -13,7 +13,7 @@ from pathlib import Path from typing import Any -from ...command_invocation import current_loopx_cli_argv +from ...command_invocation import command_argv, current_loopx_cli_argv from ..content_digest import BARE_SHA256_PATTERN @@ -116,13 +116,17 @@ def _run_json( timeout: float, runner: CommandRunner, ) -> dict[str, Any]: - command = [ - *current_loopx_cli_argv(), - "--registry", str(registry_path), - *(["--runtime-root", str(runtime_root)] if runtime_root is not None else []), - "--format", "json", - *arguments, - ] + route = current_loopx_cli_argv() + command = command_argv( + Path(route[0]), + [ + *route[1:], + "--registry", str(registry_path), + *(["--runtime-root", str(runtime_root)] if runtime_root is not None else []), + "--format", "json", + *arguments, + ], + ) completed = runner(command, cwd=cwd, timeout=timeout) try: payload = json.loads(completed.stdout or "{}") @@ -158,9 +162,17 @@ class ExecutionPreparationClaimFailed(ValueError): code = "turn_execution_preparation_claim_failed" +def _exact_identity(value: object) -> str | None: + """Keep a decision identity exact; display normalization cannot stand in for it.""" + if isinstance(value, str) and value and value == value.strip(): + return value + return None + + def builtin_execution_preparation_claim( decision: Mapping[str, Any], *, + agent_id: str, registry_path: Path, runtime_root: Path, cwd: Path, @@ -185,27 +197,41 @@ def builtin_execution_preparation_claim( claim = preparation.get("claim") if not isinstance(claim, Mapping) or claim.get("host_execution_allowed") is not True: return None - goal_id = str(decision.get("goal_id") or "").strip() - decision_agent_id = str( - decision.get("agent_id") or selected.get("agent_id") or "" - ).strip() - agent_id = str(preparation.get("agent_id") or "").strip() - todo_id = str(preparation.get("todo_id") or "").strip() + identity = decision.get("agent_identity") + declared_agent = decision.get("agent_id") + selected_agent = selected.get("agent_id") + identity_agent = identity.get("agent_id") if isinstance(identity, Mapping) else None + goal_id = _exact_identity(decision.get("goal_id")) + caller_agent_id = _exact_identity(agent_id) + preparation_agent_id = _exact_identity(preparation.get("agent_id")) + todo_id = _exact_identity(preparation.get("todo_id")) + selected_todo_id = _exact_identity(selected.get("todo_id")) + selected_agent_id = _exact_identity(selected_agent) + identity_agent_id = _exact_identity(identity_agent) + # An absent declaration may fall back to the composed identity contract. + # A present declaration that needed trimming is not the same identity. + decision_agent_id = ( + _exact_identity(declared_agent) if declared_agent is not None else identity_agent_id + ) expected_claim = [ "todo", "claim", "--goal-id", goal_id, "--todo-id", todo_id, - "--agent-id", agent_id, "--claimed-by", agent_id, + "--agent-id", caller_agent_id, "--claimed-by", caller_agent_id, ] if ( preparation.get("schema_version") != "execution_preparation_v0" or preparation.get("delivery_allowed") is not False or decision.get("normal_delivery_allowed") is not False or decision.get("should_run") is not True - or not decision_agent_id - or decision_agent_id != agent_id or not goal_id - or not agent_id + or not caller_agent_id + or not decision_agent_id + or not preparation_agent_id or not todo_id - or str(selected.get("todo_id") or "").strip() != todo_id + or selected_todo_id != todo_id + or decision_agent_id != caller_agent_id + or preparation_agent_id != caller_agent_id + or (selected_agent is not None and selected_agent_id != caller_agent_id) + or (identity_agent is not None and identity_agent_id != caller_agent_id) or claim.get("cli_args") != expected_claim or not registry_path.is_absolute() or not runtime_root.is_absolute() diff --git a/tests/control_plane/test_turn_run_once_claim_preparation.py b/tests/control_plane/test_turn_run_once_claim_preparation.py index 2601479848..5c976b0c1e 100644 --- a/tests/control_plane/test_turn_run_once_claim_preparation.py +++ b/tests/control_plane/test_turn_run_once_claim_preparation.py @@ -76,7 +76,7 @@ def run(command, *, cwd, timeout): prepared = builtin_execution_preparation_claim( _decision(), registry_path=registry, runtime_root=runtime_root, - cwd=project, runner=run, + cwd=project, agent_id=AGENT, runner=run, ) assert prepared == TODO @@ -90,6 +90,43 @@ def run(command, *, cwd, timeout): ] +def test_powershell_launcher_is_invoked_through_pwsh( + tmp_path: Path, monkeypatch, +) -> None: + from loopx.command_invocation import shutil as invocation_shutil + from loopx.control_plane.turn_driver import codex_execution_preparation + + launcher = tmp_path / "loopx.ps1" + monkeypatch.setattr( + codex_execution_preparation, + "current_loopx_cli_argv", + lambda: [str(launcher), "--cli-bin", str(launcher)], + ) + monkeypatch.setattr( + invocation_shutil, + "which", + lambda command: "pwsh" if command == "pwsh" else None, + ) + calls = [] + + def run(command, *, cwd, timeout): + calls.append(list(command)) + return _Completed({"ok": True}) + + assert builtin_execution_preparation_claim( + _decision(), agent_id=AGENT, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", cwd=tmp_path, runner=run, + ) == TODO + assert calls == [[ + "pwsh", "-NoLogo", "-NoProfile", "-File", str(launcher), + "--cli-bin", str(launcher), "--registry", str(tmp_path / "registry.json"), + "--runtime-root", str(tmp_path / "runtime"), "--format", "json", + "todo", "claim", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", AGENT, "--claimed-by", AGENT, + ]] + + @pytest.mark.parametrize( "state, host_execution_allowed", [("ready", True), ("claim_required", False), ("selection_required", True)], @@ -103,7 +140,7 @@ def run(command, *, cwd, timeout): # pragma: no cover - must not run assert builtin_execution_preparation_claim( _decision(state=state, host_execution_allowed=host_execution_allowed), registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", - cwd=tmp_path, runner=run, + cwd=tmp_path, agent_id=AGENT, runner=run, ) is None @@ -126,7 +163,7 @@ def run(command, *, cwd, timeout): # pragma: no cover - must not run assert builtin_execution_preparation_claim( _decision(cli_args=cli_args, should_run=should_run, todo_id=todo_id), registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", - cwd=tmp_path, runner=run, + cwd=tmp_path, agent_id=AGENT, runner=run, ) is None @@ -146,7 +183,7 @@ def run(command, *, cwd, timeout): # pragma: no cover - must not run decision_agent_id=decision_agent_id, ), registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", - cwd=tmp_path, runner=run, + cwd=tmp_path, agent_id=AGENT, runner=run, ) is None @@ -156,8 +193,8 @@ def run(command, *, cwd, timeout): with pytest.raises(ExecutionPreparationClaimFailed) as refused: builtin_execution_preparation_claim( - _decision(), registry_path=tmp_path / "registry.json", - runtime_root=tmp_path / "runtime", cwd=tmp_path, runner=run, + _decision(), registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", cwd=tmp_path, agent_id=AGENT, runner=run, ) assert refused.value.code == "turn_execution_preparation_claim_failed" @@ -174,10 +211,84 @@ def run(command, *, cwd, timeout): with pytest.raises(ExecutionPreparationClaimFailed) as refused: builtin_execution_preparation_claim( - _decision(), registry_path=tmp_path / "registry.json", - runtime_root=tmp_path / "runtime", cwd=tmp_path, runner=run, + _decision(), registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", cwd=tmp_path, agent_id=AGENT, runner=run, ) assert refused.value.code == "turn_execution_preparation_claim_failed" assert "Todo claimed_by does not grant this actor update authority" in str(refused.value) assert TODO in str(refused.value) + + +def _identity_decision(**changes: object) -> dict[str, object]: + decision = _decision() + preparation = decision["execution_preparation"] + selected = decision["selected_todo"] + assert isinstance(preparation, dict) and isinstance(selected, dict) + decision["goal_id"] = changes.get("goal_id", GOAL) + decision["agent_id"] = changes.get("decision_agent_id", AGENT) + preparation["agent_id"] = changes.get("preparation_agent_id", AGENT) + preparation["todo_id"] = changes.get("todo_id", TODO) + selected["todo_id"] = changes.get("selected_todo_id", TODO) + if "selected_agent_id" in changes: + selected["agent_id"] = changes["selected_agent_id"] + if "identity_agent_id" in changes: + decision["agent_identity"] = {"agent_id": changes["identity_agent_id"]} + return decision + + +@pytest.mark.parametrize( + "changes", + [ + {"goal_id": f" {GOAL}"}, + {"preparation_agent_id": f" {AGENT}"}, + {"todo_id": f" {TODO}"}, + {"selected_todo_id": f"{TODO} "}, + {"decision_agent_id": f" {AGENT}"}, + {"selected_agent_id": f"{AGENT} "}, + {"identity_agent_id": "agent-b"}, + {"decision_agent_id": "agent-b"}, + {"goal_id": 7}, + {"todo_id": None}, + ], +) +def test_an_identity_that_would_change_the_claim_is_not_normalized( + tmp_path: Path, changes: dict[str, object], +) -> None: + def run(command, *, cwd, timeout): # pragma: no cover - must not run + raise AssertionError(f"unexpected claim: {command}") + + assert builtin_execution_preparation_claim( + _identity_decision(**changes), agent_id=AGENT, + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", + cwd=tmp_path, runner=run, + ) is None + + +def test_a_call_for_another_agent_does_not_claim(tmp_path: Path) -> None: + def run(command, *, cwd, timeout): # pragma: no cover - must not run + raise AssertionError(f"unexpected claim: {command}") + + assert builtin_execution_preparation_claim( + _decision(), agent_id="agent-b", + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", + cwd=tmp_path, runner=run, + ) is None + + +def test_matching_identity_contracts_still_run_the_exact_claim(tmp_path: Path) -> None: + calls: list[list[str]] = [] + + def run(command, *, cwd, timeout): + calls.append(list(command)) + return _Completed({"ok": True}) + + assert builtin_execution_preparation_claim( + _identity_decision(selected_agent_id=AGENT, identity_agent_id=AGENT), + agent_id=AGENT, registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", cwd=tmp_path, runner=run, + ) == TODO + assert calls[0][-10:] == [ + "todo", "claim", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", AGENT, "--claimed-by", AGENT, + ] diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 4d122de14c..6b3c6ca76c 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1126,10 +1126,17 @@ def test_default_agent_session_reuses_planning_across_completed_todos(tmp_path, } -def test_run_once_claims_zero_lease_before_codex_host(tmp_path, monkeypatch): +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_run_once_claims_zero_lease_before_codex_host( + tmp_path, monkeypatch, provider, +): from tests.test_loopx_turn_codex_cli import _fake_codex - project, runtime, registry = _write_live_fixture(tmp_path, handoff_mode="hard_lease") + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, provider=provider, handoff_mode="hard_lease", + ) binary, log = _fake_codex(tmp_path) monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) home = tmp_path / "codex-home" From d39a090226a8f63d9b9c4a21d0bd73f36d8f60ab Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 12 Nov 2026 14:05:29 +0000 Subject: [PATCH 3/8] fix(turn): keep the advertised claim and caller Agent exact The builtin preparation adapter rebuilt the advertised claim from whitespace-trimmed identities and accepted a decision Agent that only matched the projected preparation. Rebuild the same argv from exact decision strings, require the caller, decision, preparation, selected Todo and identity contract to agree, and refuse anything else back to its typed wait. Cover the PowerShell launcher route, the sqlite provider, and the e2e claim-before-host path. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- .../control_plane/turn_driver/codex_execution_preparation.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index 4c2d12d4d5..738849ba93 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -102,7 +102,7 @@ def _default_runner( ) -> subprocess.CompletedProcess[str]: return subprocess.run( list(command), cwd=str(cwd), capture_output=True, text=True, - encoding="utf-8", errors="replace", timeout=timeout, + encoding="utf-8", errors="strict", timeout=timeout, check=False, ) @@ -247,7 +247,7 @@ def builtin_execution_preparation_claim( timeout=timeout, runner=runner or _default_runner, ) - except (OSError, subprocess.SubprocessError, RuntimeError) as exc: + except (OSError, subprocess.SubprocessError, RuntimeError, UnicodeDecodeError) as exc: raise ExecutionPreparationClaimFailed( f"managed claim for Todo {todo_id} was refused before delivery: {exc}" ) from exc From 55164368afe46372b7ca2c6d9dd2b502ab489bde Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:02:42 +0000 Subject: [PATCH 4/8] fix(turn): check receipt identity before managed claim Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/cli_commands/turn.py | 2 + .../control_plane/quota/heartbeat_receipt.py | 31 ++++++++ .../codex_execution_preparation.py | 56 ++++++++++++++- .../test_quota_execution_preparation.py | 36 ++++++++++ tests/test_loopx_turn_driver.py | 71 +++++++++++++++++++ 5 files changed, 193 insertions(+), 3 deletions(-) diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 3ba5edac91..15375b5754 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -111,6 +111,8 @@ def execution_preparer(decision: Mapping[str, Any]) -> str | None: registry_path=registry_path.expanduser().resolve(), runtime_root=runtime_root.resolve(), cwd=project_cwd, + turn_instance_id=args.turn_instance_id, + goal_ref=goal_ref, ) resume_instance = getattr(args, "resume_turn_instance_id", None) diff --git a/loopx/control_plane/quota/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py index 2c3d7b36d4..2f1d03523c 100644 --- a/loopx/control_plane/quota/heartbeat_receipt.py +++ b/loopx/control_plane/quota/heartbeat_receipt.py @@ -197,6 +197,37 @@ def find_heartbeat_receipt( ) +def assert_turn_heartbeat_settlement_receipt_identity( + runtime_root: Path, + identity: SettlementIdentity, + *, + goal_ref: Mapping[str, object] | None = None, +) -> None: + """Reject a conflicting settlement identity without mutating the receipt log.""" + + receipt = find_heartbeat_receipt( + runtime_root, + goal_id=identity.goal_id, + agent_id=identity.agent_id, + turn_instance_id=identity.turn_instance_id, + goal_ref=goal_ref, + ) + if receipt is None: + return + observed = _receipt_settlement_identity(receipt) + if observed is None: + return + expected = ( + identity.binding_kind, + identity.binding_id, + identity.effect_id, + ) + if observed != expected: + raise HeartbeatReceiptIdentityConflictError( + "Turn heartbeat receipt belongs to another settlement identity" + ) + + def ensure_turn_heartbeat_settlement_receipt( runtime_root: Path, identity: SettlementIdentity, diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index 738849ba93..401709b911 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -100,11 +100,22 @@ def compact_execution_preparation_request( def _default_runner( command: Sequence[str], *, cwd: Path, timeout: float, ) -> subprocess.CompletedProcess[str]: - return subprocess.run( - list(command), cwd=str(cwd), capture_output=True, text=True, - encoding="utf-8", errors="strict", timeout=timeout, + completed = subprocess.run( + list(command), cwd=str(cwd), capture_output=True, text=False, + timeout=timeout, check=False, ) + # The receipt is protocol data and must decode exactly. Stderr is only + # diagnostic output from the launcher; an unrelated invalid byte there + # must not turn a valid receipt into a failed preparation. + stdout = (completed.stdout or b"").decode("utf-8", errors="strict") + stderr = (completed.stderr or b"").decode("utf-8", errors="replace") + return subprocess.CompletedProcess( + completed.args, + completed.returncode, + stdout, + stderr, + ) def _run_json( @@ -176,6 +187,8 @@ def builtin_execution_preparation_claim( registry_path: Path, runtime_root: Path, cwd: Path, + turn_instance_id: str | None = None, + goal_ref: Mapping[str, object] | None = None, runner: CommandRunner | None = None, timeout: float = 30, ) -> str | None: @@ -238,6 +251,43 @@ def builtin_execution_preparation_claim( or not cwd.is_absolute() ): return None + heartbeat = decision.get("heartbeat_receipt") + exact_turn_instance_id = _exact_identity(turn_instance_id) + if exact_turn_instance_id is None and isinstance(heartbeat, Mapping): + exact_turn_instance_id = _exact_identity(heartbeat.get("turn_instance_id")) + if exact_turn_instance_id is not None: + # The receipt lookup and the ordinary claim are separate operations. + # Refuse a conflicting binding before the claim can create a lease for + # a different Todo; planning also re-reads this identity after the + # claim, so this check remains side-effect free. + from ..quota.effect_program import SettlementIdentity + from ..quota.heartbeat_receipt import ( + assert_turn_heartbeat_settlement_receipt_identity, + ) + from ..quota.turn_envelope import turn_envelope_scoped_settlement_identity + + raw_identity = turn_envelope_scoped_settlement_identity( + decision, turn_instance_id=exact_turn_instance_id, + ) + # A blocked pre-claim decision has no completed transaction to project. + # The ordinary Todo claim already fixes the remaining identity fields, + # so derive the same settlement binding directly and compare it with + # the receipt before the claim can create a lease. + settlement_identity = ( + SettlementIdentity.from_runtime_payload(raw_identity) + if raw_identity is not None + else SettlementIdentity( + goal_id=goal_id, + agent_id=caller_agent_id, + todo_id=todo_id, + turn_instance_id=exact_turn_instance_id, + ) + ) + assert_turn_heartbeat_settlement_receipt_identity( + runtime_root, + settlement_identity, + goal_ref=goal_ref, + ) try: _run_json( expected_claim, diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index a394b4748e..ed4aca5c33 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -26,6 +26,42 @@ AGENT = "agent-a" +def test_default_runner_keeps_a_valid_receipt_when_stderr_is_undecodable( + tmp_path: Path, +) -> None: + from loopx.control_plane.turn_driver import codex_execution_preparation + + completed = codex_execution_preparation._default_runner( + [ + sys.executable, + "-c", + "import sys; sys.stdout.write('{\"ok\": true}'); " + "sys.stderr.buffer.write(b'\\xff\\n')", + ], + cwd=tmp_path, + timeout=30, + ) + + assert completed.returncode == 0 + assert json.loads(completed.stdout) == {"ok": True} + assert "\ufffd" in completed.stderr + + +def test_default_runner_keeps_receipt_stdout_strict(tmp_path: Path) -> None: + from loopx.control_plane.turn_driver import codex_execution_preparation + + with pytest.raises(UnicodeDecodeError): + codex_execution_preparation._default_runner( + [ + sys.executable, + "-c", + "import sys; sys.stdout.buffer.write(b'\\xff')", + ], + cwd=tmp_path, + timeout=30, + ) + + @pytest.mark.parametrize("status", ["open", "deferred"]) def test_explicit_non_delivery_obligation_does_not_prepare_selected_execution(monkeypatch, status): from loopx.control_plane.quota import execution_preparation diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 6b3c6ca76c..9141e31dfc 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1179,6 +1179,77 @@ def test_run_once_claims_zero_lease_before_codex_host( assert len(calls) == 1, calls +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_run_once_conflicting_turn_identity_does_not_claim_another_todo( + tmp_path, monkeypatch, provider, +): + from tests.test_loopx_turn_codex_cli import _fake_codex + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, + provider=provider, + handoff_mode="hard_lease", + extra_agent_todo_lines=( + "- [ ] [P1] Complete the conflicting successor fixture.", + " ", + ), + ) + binary, log = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + + output = io.StringIO() + with contextlib.redirect_stdout(output): + guard_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "quota", "should-run", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--todo-id", "todo_fixture0001", + "--turn-instance-id", "bound-preparation-turn", "--context-delivery", "off", + ]) + bound = read_cli_policy_payload(output.getvalue(), returncode=guard_code) + assert guard_code == 0, bound + assert bound["heartbeat_receipt"]["turn_instance_id"] == "bound-preparation-turn" + assert bound["heartbeat_receipt"]["settlement_identity"]["todo_id"] == "todo_fixture0001" + + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--todo-id", "todo_fixture0002", + "--turn-instance-id", "bound-preparation-turn", + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + refused = json.loads(output.getvalue()) + assert exit_code == 1, refused + assert "belongs to another settlement identity" in refused["error"] + assert not log.exists() + + output = io.StringIO() + with contextlib.redirect_stdout(output): + lease_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "task-lease", "inspect", "--goal-id", "loopx-turn-fixture", + "--todo-id", "todo_fixture0002", + ]) + lease = read_cli_policy_payload(output.getvalue(), returncode=lease_code) + assert lease_code == 0, lease + assert lease["active"] is False, lease + assert lease["lease"] is None, lease + + def test_turn_plan_transaction_key_is_stable_and_todo_scoped() -> None: first = build_loopx_turn_plan( _envelope(), From 474307ba234ac52eafc6c992dbfce43e28825f04 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:44:54 +0000 Subject: [PATCH 5/8] fix(turn): reserve identity before managed claim Bind the Turn receipt before the ordinary claim so a competing guard cannot acquire a lease for a different Todo. Cover the check-to-claim interleaving on File and SQLite providers. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- .../control_plane/quota/heartbeat_receipt.py | 31 ------ .../codex_execution_preparation.py | 28 ++++-- tests/test_loopx_turn_driver.py | 98 +++++++++++++++++++ 3 files changed, 118 insertions(+), 39 deletions(-) diff --git a/loopx/control_plane/quota/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py index 2f1d03523c..2c3d7b36d4 100644 --- a/loopx/control_plane/quota/heartbeat_receipt.py +++ b/loopx/control_plane/quota/heartbeat_receipt.py @@ -197,37 +197,6 @@ def find_heartbeat_receipt( ) -def assert_turn_heartbeat_settlement_receipt_identity( - runtime_root: Path, - identity: SettlementIdentity, - *, - goal_ref: Mapping[str, object] | None = None, -) -> None: - """Reject a conflicting settlement identity without mutating the receipt log.""" - - receipt = find_heartbeat_receipt( - runtime_root, - goal_id=identity.goal_id, - agent_id=identity.agent_id, - turn_instance_id=identity.turn_instance_id, - goal_ref=goal_ref, - ) - if receipt is None: - return - observed = _receipt_settlement_identity(receipt) - if observed is None: - return - expected = ( - identity.binding_kind, - identity.binding_id, - identity.effect_id, - ) - if observed != expected: - raise HeartbeatReceiptIdentityConflictError( - "Turn heartbeat receipt belongs to another settlement identity" - ) - - def ensure_turn_heartbeat_settlement_receipt( runtime_root: Path, identity: SettlementIdentity, diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index 401709b911..daedebe7b2 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -257,13 +257,11 @@ def builtin_execution_preparation_claim( exact_turn_instance_id = _exact_identity(heartbeat.get("turn_instance_id")) if exact_turn_instance_id is not None: # The receipt lookup and the ordinary claim are separate operations. - # Refuse a conflicting binding before the claim can create a lease for - # a different Todo; planning also re-reads this identity after the - # claim, so this check remains side-effect free. - from ..quota.effect_program import SettlementIdentity - from ..quota.heartbeat_receipt import ( - assert_turn_heartbeat_settlement_receipt_identity, - ) + # Reserve the exact Turn binding before the claim can create a lease for + # a different Todo. The reservation owner serializes this write with every + # later guard, and planning re-reads the same identity after the claim. + from ..quota.effect_program import SettlementBindingKind, SettlementIdentity + from ..quota.heartbeat_receipt import ensure_turn_heartbeat_settlement_receipt from ..quota.turn_envelope import turn_envelope_scoped_settlement_identity raw_identity = turn_envelope_scoped_settlement_identity( @@ -283,9 +281,23 @@ def builtin_execution_preparation_claim( turn_instance_id=exact_turn_instance_id, ) ) - assert_turn_heartbeat_settlement_receipt_identity( + if ( + settlement_identity.goal_id != goal_id + or settlement_identity.agent_id != caller_agent_id + or settlement_identity.turn_instance_id != exact_turn_instance_id + or settlement_identity.binding_kind is not SettlementBindingKind.TODO + or settlement_identity.binding_id != todo_id + ): + raise ValueError("execution preparation identity does not match its ordinary claim") + # The receipt owner serializes this reservation with every later guard + # binding. Reserving the Turn before the ordinary claim closes the + # check-to-claim race: a conflicting Todo either wins here or is refused + # here, so the losing caller cannot acquire a lease first. + ensure_turn_heartbeat_settlement_receipt( runtime_root, settlement_identity, + semantic_replan_guard_scoped=False, + semantic_replan_obligation_id=None, goal_ref=goal_ref, ) try: diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 9141e31dfc..c30074ec36 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1250,6 +1250,104 @@ def test_run_once_conflicting_turn_identity_does_not_claim_another_todo( assert lease["lease"] is None, lease +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_run_once_reserves_turn_before_claim_under_competing_guard( + tmp_path, monkeypatch, provider, +): + from loopx.control_plane.turn_driver import codex_execution_preparation + from tests.test_loopx_turn_codex_cli import _fake_codex + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, + provider=provider, + handoff_mode="hard_lease", + extra_agent_todo_lines=( + "- [ ] [P1] Compete for the same Turn fixture.", + " ", + ), + ) + binary, log = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + + original_runner = codex_execution_preparation._default_runner + competing_guard: dict[str, object] = {} + + def interleaving_runner(command, *, cwd, timeout): + # This runs after the Turn reservation and immediately before the + # ordinary claim subprocess, the exact window that previously let a + # competing guard bind the same Turn to a different Todo first. + if "claim" in command and "todo" in command: + output = io.StringIO() + with contextlib.redirect_stdout(output): + guard_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", "quota", "should-run", + "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--todo-id", "todo_fixture0001", + "--turn-instance-id", "reserved-preparation-turn", + "--context-delivery", "off", + ]) + competing_guard["code"] = guard_code + competing_guard["payload"] = json.loads(output.getvalue()) + return original_runner(command, cwd=cwd, timeout=timeout) + + monkeypatch.setattr(codex_execution_preparation, "_default_runner", interleaving_runner) + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", "turn", "run-once", + "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--todo-id", "todo_fixture0002", + "--turn-instance-id", "reserved-preparation-turn", + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + result = json.loads(output.getvalue()) + assert exit_code == 0, result + assert result["status"] == "committed", result + assert competing_guard["code"] == 1, competing_guard + assert ( + competing_guard["payload"]["error_code"] + == "heartbeat_receipt_identity_conflict" + ), competing_guard + assert ( + "settlement identity conflicts with the current selected Todo" + in competing_guard["payload"]["reason"] + ), competing_guard + calls = [json.loads(line) for line in log.read_text().splitlines()] + assert len(calls) == 1, calls + + for todo_id, expected_active in ( + ("todo_fixture0001", False), + ("todo_fixture0002", True), + ): + output = io.StringIO() + with contextlib.redirect_stdout(output): + lease_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", "task-lease", "inspect", + "--goal-id", "loopx-turn-fixture", "--todo-id", todo_id, + ]) + lease = read_cli_policy_payload(output.getvalue(), returncode=lease_code) + assert lease_code == 0, lease + assert lease["active"] is expected_active, lease + if not expected_active: + assert lease["lease"] is None, lease + + def test_turn_plan_transaction_key_is_stable_and_todo_scoped() -> None: first = build_loopx_turn_plan( _envelope(), From d4348db14d51952a74247e3003768a2f969c7216 Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Fri, 13 Nov 2026 02:53:17 +0000 Subject: [PATCH 6/8] fix(turn): normalize the Turn identity before the managed claim Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- .../codex_execution_preparation.py | 13 +++- tests/test_loopx_turn_driver.py | 78 +++++++++++++++++++ 2 files changed, 89 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index daedebe7b2..b2d23bceb9 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -14,6 +14,7 @@ from typing import Any from ...command_invocation import command_argv, current_loopx_cli_argv +from ...turn_identity import normalize_turn_instance_id from ..content_digest import BARE_SHA256_PATTERN @@ -252,9 +253,17 @@ def builtin_execution_preparation_claim( ): return None heartbeat = decision.get("heartbeat_receipt") - exact_turn_instance_id = _exact_identity(turn_instance_id) + # Resolve one Turn identity before reading a receipt or running the claim. + # The transaction plan applies the same normalization, so a padded or + # invalid identity must be refused here; otherwise the claim could lease a + # different Todo before planning rejects the mismatch. + exact_turn_instance_id = normalize_turn_instance_id(turn_instance_id) if exact_turn_instance_id is None and isinstance(heartbeat, Mapping): - exact_turn_instance_id = _exact_identity(heartbeat.get("turn_instance_id")) + receipt_turn_instance_id = heartbeat.get("turn_instance_id") + if isinstance(receipt_turn_instance_id, str): + exact_turn_instance_id = normalize_turn_instance_id( + receipt_turn_instance_id + ) if exact_turn_instance_id is not None: # The receipt lookup and the ordinary claim are separate operations. # Reserve the exact Turn binding before the claim can create a lease for diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index c30074ec36..cc3f51e02a 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1250,6 +1250,84 @@ def test_run_once_conflicting_turn_identity_does_not_claim_another_todo( assert lease["lease"] is None, lease +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize( + ("requested_turn", "expected_error"), + [ + (" bound-preparation-turn ", "belongs to another settlement identity"), + ("bound preparation turn", "turn_instance_id must be"), + ], + ids=["padded-valid", "explicit-invalid"], +) +def test_run_once_validates_the_turn_identity_before_claiming( + tmp_path, monkeypatch, provider, requested_turn, expected_error, +): + from tests.test_loopx_turn_codex_cli import _fake_codex + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, + provider=provider, + handoff_mode="hard_lease", + extra_agent_todo_lines=( + "- [ ] [P1] Complete the conflicting successor fixture.", + " ", + ), + ) + binary, log = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + + output = io.StringIO() + with contextlib.redirect_stdout(output): + guard_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "quota", "should-run", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--todo-id", "todo_fixture0001", + "--turn-instance-id", "bound-preparation-turn", "--context-delivery", "off", + ]) + bound = read_cli_policy_payload(output.getvalue(), returncode=guard_code) + assert guard_code == 0, bound + assert bound["heartbeat_receipt"]["settlement_identity"]["todo_id"] == "todo_fixture0001" + + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--todo-id", "todo_fixture0002", + "--turn-instance-id", requested_turn, + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + refused = json.loads(output.getvalue()) + assert exit_code == 1, refused + assert expected_error in refused["error"] + assert not log.exists() + + output = io.StringIO() + with contextlib.redirect_stdout(output): + lease_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "task-lease", "inspect", "--goal-id", "loopx-turn-fixture", + "--todo-id", "todo_fixture0002", + ]) + lease = read_cli_policy_payload(output.getvalue(), returncode=lease_code) + assert lease_code == 0, lease + assert lease["active"] is False, lease + assert lease["lease"] is None, lease + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_run_once_reserves_turn_before_claim_under_competing_guard( tmp_path, monkeypatch, provider, From a1842895bbfde1523b8bdafb658ba3086ed3beed Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Fri, 13 Nov 2026 04:00:35 +0000 Subject: [PATCH 7/8] fix(turn): refuse transport and interval waits before the managed claim Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/cli_commands/turn.py | 20 ++- loopx/cli_commands/turn_cadence.py | 46 +++++++ loopx/cli_commands/turn_host_request.py | 59 +++++---- loopx/cli_commands/turn_run_once.py | 30 +++++ tests/test_loopx_turn_driver.py | 161 ++++++++++++++++++++++++ 5 files changed, 291 insertions(+), 25 deletions(-) diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 15375b5754..bd8bf1644f 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -190,6 +190,7 @@ def execution_preparer(decision: Mapping[str, Any]) -> str | None: "--operation-source-route-json requires --codex-operation-tools for codex-cli or --dsh-operation-module for dsh" ) decision: dict[str, Any] | None = None + preclaim_cadence_wait: dict[str, Any] | None = None # A resumed Turn already chose its transaction. New work and context # belong to the next Turn, not to recovery of this journal. if args.turn_command == "run-once" and args.resume_turn_key: @@ -298,7 +299,23 @@ def read_decision_owner(): decision = compose_turn_decision(decision_owner.resolve(), scheduler_execution_context=scheduler_context) resume_requested, session_binding = resolve_turn_resume_session_binding(args) if execution_preparer is not None and args.turn_command == "run-once" and args.execute: - prepared_todo = execution_preparer(decision) + from .turn_cadence import managed_cadence_wait_projection + from .turn_host_request import validate_turn_host_inputs + + # Transport and interval refusals land before the Turn receipt + # reservation and the ordinary claim can write a lease. + validate_turn_host_inputs(args) + preclaim_cadence_wait = managed_cadence_wait_projection( + runtime_root, + goal_id=args.goal_id, + agent_id=args.agent_id, + automation_id=getattr(args, "automation_id", None), + ) + prepared_todo = ( + None + if preclaim_cadence_wait is not None + else execution_preparer(decision) + ) if prepared_todo is not None: # Host preparation changes canonical lease state. Plan and # settle against a fresh read of the same Turn and Todo. @@ -431,6 +448,7 @@ def read_decision_owner(): native_session_binding_resolver=native_session_binding_resolver, native_completion_lease_resolver=native_completion_lease_resolver, native_task_validator=native_task_validator, + preclaim_cadence_wait=preclaim_cadence_wait, ) else: raise ValueError("turn requires the `plan` or `run-once` subcommand") diff --git a/loopx/cli_commands/turn_cadence.py b/loopx/cli_commands/turn_cadence.py index f5f965e1e7..cee1abe980 100644 --- a/loopx/cli_commands/turn_cadence.py +++ b/loopx/cli_commands/turn_cadence.py @@ -21,6 +21,52 @@ class ManagedCadenceStart(NamedTuple): confirm: Callable[[], None] +def managed_cadence_wait_projection( + runtime_root: Path, + *, + goal_id: str, + agent_id: str | None, + automation_id: str | None, +) -> dict[str, Any] | None: + """Read the owner interval before any Turn write and report a wait. + + The authoritative reservation stays with the driver; this read only keeps + an interval refusal from claiming a Todo or appending a Turn receipt first. + """ + + if not agent_id: + return None + from ..control_plane.quota.automation_cadence_readback import ( + automation_cadence_readback, + ) + + readback = automation_cadence_readback( + effect_runtime_result( + "quota.automation_cadence.manage", + { + "operation": "read", + "runtime_root": str(runtime_root), + "goal_id": goal_id, + "agent_id": agent_id, + "automation_id": automation_id, + }, + ), + expected_goal_id=goal_id, + expected_agent_id=agent_id, + expected_automation_id=automation_id, + ) + eligibility = readback["eligibility"] + if eligibility.get("state") != "waiting": + return None + return { + "admitted": False, + "reason": eligibility["reason"], + "next_eligible_at_ms": eligibility["next_eligible_at_ms"], + "min_interval_minutes": readback["min_interval_minutes"], + "pre_model_admission": "managed_turn_only", + } + + def managed_cadence_start( *, runtime_root: Path, diff --git a/loopx/cli_commands/turn_host_request.py b/loopx/cli_commands/turn_host_request.py index 6c83d483dc..e5edd2aa23 100644 --- a/loopx/cli_commands/turn_host_request.py +++ b/loopx/cli_commands/turn_host_request.py @@ -14,27 +14,11 @@ from ..control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission -def prepare_turn_host_request( - args: argparse.Namespace, payload: Mapping[str, Any], *, - native_host_runner: Callable[..., dict[str, Any]] | None, - native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None, - native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None, - validator_builder: Callable[..., Callable[..., Mapping[str, Any]]], -) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None, dict[str, Any]]: - project = Path(args.project).expanduser().resolve() - planned_host = ( - cast(dict[str, Any], payload.get("host")) if isinstance(payload.get("host"), dict) else {} - ) - if planned_host.get("kind") != args.host: - raise ValueError("--host must match the journaled LoopX Turn plan") - if args.host == "codex-app-server" and ( - native_host_runner is None or native_session_binding_resolver is None - ): - raise ValueError("app-server execution requires its original in-process Chat connection") - if args.host != "codex-app-server" and ( - native_host_runner is not None or native_session_binding_resolver is not None - ): - raise ValueError("a native Chat connection cannot replace a selected CLI host") +def validate_turn_host_inputs( + args: argparse.Namespace, +) -> tuple[list[str] | None, list[str] | None]: + """Validate transport input without reading a Turn or writing state.""" + if args.host == "generic-cli": if not args.host_command_json: raise ValueError( @@ -54,9 +38,7 @@ def prepare_turn_host_request( if getattr(args, "host_adapter_request_schema", None): raise ValueError("--host-adapter-request-schema is only for generic-cli") raw_argv = None - if native_task_validator is not None and args.host != "codex-app-server": - raise ValueError("a native task validator cannot replace a selected CLI validator") - task_validator = native_task_validator + raw_validation_argv = None if args.validation_command_json: raw_validation_argv = json.loads(args.validation_command_json) if not isinstance(raw_validation_argv, list) or not all( @@ -65,6 +47,35 @@ def prepare_turn_host_request( raise ValueError( "--validation-command-json must be a JSON string array" ) + return raw_argv, raw_validation_argv + + +def prepare_turn_host_request( + args: argparse.Namespace, payload: Mapping[str, Any], *, + native_host_runner: Callable[..., dict[str, Any]] | None, + native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None, + native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None, + validator_builder: Callable[..., Callable[..., Mapping[str, Any]]], +) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None, dict[str, Any]]: + project = Path(args.project).expanduser().resolve() + planned_host = ( + cast(dict[str, Any], payload.get("host")) if isinstance(payload.get("host"), dict) else {} + ) + if planned_host.get("kind") != args.host: + raise ValueError("--host must match the journaled LoopX Turn plan") + if args.host == "codex-app-server" and ( + native_host_runner is None or native_session_binding_resolver is None + ): + raise ValueError("app-server execution requires its original in-process Chat connection") + if args.host != "codex-app-server" and ( + native_host_runner is not None or native_session_binding_resolver is not None + ): + raise ValueError("a native Chat connection cannot replace a selected CLI host") + raw_argv, raw_validation_argv = validate_turn_host_inputs(args) + if native_task_validator is not None and args.host != "codex-app-server": + raise ValueError("a native task validator cannot replace a selected CLI validator") + task_validator = native_task_validator + if raw_validation_argv is not None: if task_validator is None: task_validator = validator_builder( raw_validation_argv, diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index 9df16f9641..0cb2d6cc90 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -92,6 +92,7 @@ def execute_turn_run_once( native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None, native_completion_lease_resolver: Callable[[], Mapping[str, Any]] | None = None, native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None = None, + preclaim_cadence_wait: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Adapt one prepared CLI Turn plan to the bounded Turn driver.""" execution_started = False @@ -108,6 +109,35 @@ def execute_turn_run_once( native_task_validator=native_task_validator, validator_builder=build_loopx_turn_command_validator, ) + if preclaim_cadence_wait is not None: + # The owner's interval read refused the start before any Turn write. + # Return its wait through the normal execution projection without + # reserving an interval, claiming a Todo, or journaling a Turn. + from ..control_plane.turn_driver.execution_readback import execution_payload + + planned_host = payload.get("host") + waiting = { + "status": "interval_wait", + "host": planned_host if isinstance(planned_host, Mapping) else {}, + "completed_phases": [], + "reason": str( + preclaim_cadence_wait.get("reason") + or "automatic start not admitted" + ), + "admission": dict(preclaim_cadence_wait), + } + return execution_payload( + payload, + waiting, + execute=True, + replayed=False, + effects={ + "host_invoked": False, + "state_written": False, + "quota_spent": False, + "scheduler_acknowledged": False, + }, + ) selected_todo = execution_context["selected_todo"] expected_todo_text_sha256 = execution_context["sha256"] transaction = ( diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index cc3f51e02a..45f11933c5 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -46,6 +46,7 @@ ) from loopx.control_plane.turn_driver.executor import BuiltInHostError from loopx.control_plane.quota.live_decision import bind_scheduler_followup_cli_routes +from loopx.rollout_event_log import rollout_event_log_path def _envelope( @@ -1328,6 +1329,166 @@ def test_run_once_validates_the_turn_identity_before_claiming( assert lease["lease"] is None, lease +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize( + ("validation_json", "expected_error"), + [ + ("not-json", "Expecting value"), + ('{"not": "an-array"}', "must be a JSON string array"), + ], + ids=["malformed", "wrong-shape"], +) +def test_run_once_rejects_invalid_validator_json_before_claiming( + tmp_path, monkeypatch, provider, validation_json, expected_error, +): + from tests.test_loopx_turn_codex_cli import _fake_codex + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, provider=provider, handoff_mode="hard_lease", + ) + binary, log = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + rollout_log = rollout_event_log_path(runtime, "loopx-turn-fixture") + before_log = rollout_log.read_bytes() if rollout_log.exists() else b"" + + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", validation_json, + "--execute", + ]) + refused = json.loads(output.getvalue()) + assert exit_code == 1, refused + assert expected_error in refused["error"] + assert not log.exists() + assert (rollout_log.read_bytes() if rollout_log.exists() else b"") == before_log + + output = io.StringIO() + with contextlib.redirect_stdout(output): + lease_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "task-lease", "inspect", "--goal-id", "loopx-turn-fixture", + "--todo-id", "todo_fixture0001", + ]) + lease = read_cli_policy_payload(output.getvalue(), returncode=lease_code) + assert lease_code == 0, lease + assert lease["active"] is False, lease + assert lease["lease"] is None, lease + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_run_once_interval_wait_does_not_claim_the_next_todo( + tmp_path, monkeypatch, provider, +): + from tests.test_loopx_turn_codex_cli import _fake_codex + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, + provider=provider, + handoff_mode="hard_lease", + extra_agent_todo_lines=( + "- [ ] [P1] Complete the interval successor fixture.", + " ", + ), + ) + binary, log = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + + output = io.StringIO() + with contextlib.redirect_stdout(output): + read_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "automation-cadence", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", + ]) + policy = read_cli_policy_payload(output.getvalue(), returncode=read_code) + assert read_code == 0, policy + output = io.StringIO() + with contextlib.redirect_stdout(output): + configure_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "automation-cadence", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--min-interval-minutes", "3", + "--expected-revision", str(policy["configuration_revision"]), + "--owner-reference", "interval fixture", "--execute", + ]) + configured = read_cli_policy_payload(output.getvalue(), returncode=configure_code) + assert configure_code == 0 and configured["written"] is True, configured + + output = io.StringIO() + with contextlib.redirect_stdout(output): + first_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--turn-instance-id", "interval-first", "--host", "codex-cli", + "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + first = json.loads(output.getvalue()) + assert first_code == 0 and first["status"] == "committed", first + calls = [json.loads(line) for line in log.read_text().splitlines()] + assert len(calls) == 1, calls + rollout_log = rollout_event_log_path(runtime, "loopx-turn-fixture") + before_log = rollout_log.read_bytes() if rollout_log.exists() else b"" + + output = io.StringIO() + with contextlib.redirect_stdout(output): + waiting_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--todo-id", "todo_fixture0002", "--turn-instance-id", "interval-second", + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + waiting = json.loads(output.getvalue()) + assert waiting_code == 1, waiting + assert waiting["status"] == "interval_wait", waiting + assert waiting["effects"]["host_invoked"] is False, waiting + assert waiting["effects"]["state_written"] is False, waiting + calls = [json.loads(line) for line in log.read_text().splitlines()] + assert len(calls) == 1, calls + assert (rollout_log.read_bytes() if rollout_log.exists() else b"") == before_log + + output = io.StringIO() + with contextlib.redirect_stdout(output): + lease_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "task-lease", "inspect", "--goal-id", "loopx-turn-fixture", + "--todo-id", "todo_fixture0002", + ]) + lease = read_cli_policy_payload(output.getvalue(), returncode=lease_code) + assert lease_code == 0, lease + assert lease["active"] is False, lease + assert lease["lease"] is None, lease + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_run_once_reserves_turn_before_claim_under_competing_guard( tmp_path, monkeypatch, provider, From 51e72fcf4ad572ff67b2421cd2d68a8d34d17b1f Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Fri, 13 Nov 2026 06:17:14 +0000 Subject: [PATCH 8/8] fix(turn): admit the owner interval before the managed claim Reserve or resume the owner cadence slot inside the execution-preparation boundary, ahead of the Turn receipt and the ordinary claim, and hand that same admission to the driver so a losing request leaves no lease, receipt or journal behind. The manual interval bypass reason now reaches the same atomic admission instead of a read-only projection that could not see it. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- loopx/cli_commands/turn.py | 35 ++--- loopx/cli_commands/turn_cadence.py | 130 +++++++++--------- loopx/cli_commands/turn_run_once.py | 46 ++++--- .../codex_execution_preparation.py | 14 ++ loopx/control_plane/turn_driver/executor.py | 11 +- tests/test_loopx_turn_driver.py | 100 +++++++++++++- 6 files changed, 230 insertions(+), 106 deletions(-) diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index bd8bf1644f..4aeed6017f 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -89,6 +89,19 @@ def handle_turn_command( registry_path=registry_path, runtime_root_override=runtime_root_arg, ) + managed_cadence = None + if args.turn_command == "run-once" and bool(getattr(args, "execute", False)): + # Resolve the start owner through the run-once adapter so one + # process-wide seam binds admission, confirmation and receipts. + from . import turn_run_once as turn_run_once_module + + managed_cadence = turn_run_once_module.managed_cadence_start( + runtime_root=runtime_root.resolve(), + goal_id=args.goal_id, + agent_id=args.agent_id, + automation_id=args.automation_id, + manual_reason=getattr(args, "manual_interval_bypass_reason", None), + ) execution_preparer = native_execution_preparer preparation_error = "native task preparation did not make the same Todo ready" if ( @@ -113,6 +126,7 @@ def execution_preparer(decision: Mapping[str, Any]) -> str | None: cwd=project_cwd, turn_instance_id=args.turn_instance_id, goal_ref=goal_ref, + admit_start=managed_cadence.admit if managed_cadence else None, ) resume_instance = getattr(args, "resume_turn_instance_id", None) @@ -190,7 +204,6 @@ def execution_preparer(decision: Mapping[str, Any]) -> str | None: "--operation-source-route-json requires --codex-operation-tools for codex-cli or --dsh-operation-module for dsh" ) decision: dict[str, Any] | None = None - preclaim_cadence_wait: dict[str, Any] | None = None # A resumed Turn already chose its transaction. New work and context # belong to the next Turn, not to recovery of this journal. if args.turn_command == "run-once" and args.resume_turn_key: @@ -299,23 +312,13 @@ def read_decision_owner(): decision = compose_turn_decision(decision_owner.resolve(), scheduler_execution_context=scheduler_context) resume_requested, session_binding = resolve_turn_resume_session_binding(args) if execution_preparer is not None and args.turn_command == "run-once" and args.execute: - from .turn_cadence import managed_cadence_wait_projection from .turn_host_request import validate_turn_host_inputs - # Transport and interval refusals land before the Turn receipt - # reservation and the ordinary claim can write a lease. + # Transport refusals land before the Turn receipt reservation. + # The owner interval is admitted atomically inside the same + # preparation boundary, ahead of every receipt and claim write. validate_turn_host_inputs(args) - preclaim_cadence_wait = managed_cadence_wait_projection( - runtime_root, - goal_id=args.goal_id, - agent_id=args.agent_id, - automation_id=getattr(args, "automation_id", None), - ) - prepared_todo = ( - None - if preclaim_cadence_wait is not None - else execution_preparer(decision) - ) + prepared_todo = execution_preparer(decision) if prepared_todo is not None: # Host preparation changes canonical lease state. Plan and # settle against a fresh read of the same Turn and Todo. @@ -448,7 +451,7 @@ def read_decision_owner(): native_session_binding_resolver=native_session_binding_resolver, native_completion_lease_resolver=native_completion_lease_resolver, native_task_validator=native_task_validator, - preclaim_cadence_wait=preclaim_cadence_wait, + managed_cadence=managed_cadence, ) else: raise ValueError("turn requires the `plan` or `run-once` subcommand") diff --git a/loopx/cli_commands/turn_cadence.py b/loopx/cli_commands/turn_cadence.py index cee1abe980..987e61947f 100644 --- a/loopx/cli_commands/turn_cadence.py +++ b/loopx/cli_commands/turn_cadence.py @@ -3,68 +3,28 @@ import time from collections.abc import Callable, Mapping from pathlib import Path -from typing import Any, NamedTuple +from dataclasses import dataclass +from typing import Any from ..control_plane.effect_runtime import effect_runtime_result -class ManagedCadenceStart(NamedTuple): +@dataclass(slots=True) +class ManagedCadenceStart: """Owner-cadence callbacks for one managed Turn start. `admit` reserves (or resumes) the interval slot before any host or journal attempt; `confirm` marks that reservation as a real host attempt once the Turn journal is durable. Keeping them separate means a crash in between leaves a resumable reservation rather than a permanently rejected Turn. + `accepted` returns the most recent admission so a caller that admits ahead + of the ordinary claim can hand the same reservation to the driver. """ admit: Callable[[Mapping[str, Any]], dict[str, Any]] confirm: Callable[[], None] - - -def managed_cadence_wait_projection( - runtime_root: Path, - *, - goal_id: str, - agent_id: str | None, - automation_id: str | None, -) -> dict[str, Any] | None: - """Read the owner interval before any Turn write and report a wait. - - The authoritative reservation stays with the driver; this read only keeps - an interval refusal from claiming a Todo or appending a Turn receipt first. - """ - - if not agent_id: - return None - from ..control_plane.quota.automation_cadence_readback import ( - automation_cadence_readback, - ) - - readback = automation_cadence_readback( - effect_runtime_result( - "quota.automation_cadence.manage", - { - "operation": "read", - "runtime_root": str(runtime_root), - "goal_id": goal_id, - "agent_id": agent_id, - "automation_id": automation_id, - }, - ), - expected_goal_id=goal_id, - expected_agent_id=agent_id, - expected_automation_id=automation_id, - ) - eligibility = readback["eligibility"] - if eligibility.get("state") != "waiting": - return None - return { - "admitted": False, - "reason": eligibility["reason"], - "next_eligible_at_ms": eligibility["next_eligible_at_ms"], - "min_interval_minutes": readback["min_interval_minutes"], - "pre_model_admission": "managed_turn_only", - } + accepted: Callable[[], dict[str, Any] | None] + set_on_admitted: Callable[[Callable[[], None] | None], None] def managed_cadence_start( @@ -79,37 +39,70 @@ def managed_cadence_start( """Bind one managed Turn start to the TypeScript owner-cadence store.""" admitted_request: dict[str, Any] = {} + latest_admission: dict[str, Any] | None = None + admission_hook: list[Callable[[], None] | None] = [on_admitted] + hook_pending = [False] + + def set_on_admitted(callback: Callable[[], None] | None) -> None: + admission_hook[0] = callback + if hook_pending[0] and callback is not None: + # A reservation admitted ahead of the surrounding Turn preparation + # delivers its receipt writer as soon as the Turn adopts it. + hook_pending[0] = False + callback() def admit(identity: Mapping[str, Any]) -> dict[str, Any]: + nonlocal latest_admission now_ms = time.time_ns() // 1_000_000 - request_id = f"{identity['turn_key']}:{identity['attempt']}" - admission = effect_runtime_result( - "quota.automation_cadence.admit", - { - "runtime_root": str(runtime_root), - "goal_id": goal_id, - "agent_id": agent_id, - "automation_id": automation_id, - "request_id": request_id, - "trigger_at_ms": now_ms, - "now_ms": now_ms, - "manual_reason": manual_reason, - }, - retry_safe=False, - ) + base_request_id = f"{identity['turn_key']}:{identity['attempt']}" + + def request_admission(request_id: str) -> dict[str, Any]: + return effect_runtime_result( + "quota.automation_cadence.admit", + { + "runtime_root": str(runtime_root), + "goal_id": goal_id, + "agent_id": agent_id, + "automation_id": automation_id, + "request_id": request_id, + "trigger_at_ms": now_ms, + "now_ms": now_ms, + "manual_reason": manual_reason, + }, + retry_safe=False, + ) + + request_id = base_request_id + admission = request_admission(request_id) + if admission.get("admitted") is not True and ( + admission.get("reason") == "duplicate_or_stale_trigger" + ): + # A durable earlier attempt keeps its own request identity + # fail-closed. Once the owner interval has elapsed, this Turn may + # start again under a fresh attempt identity, exactly as the + # attempt counter did before admission moved ahead of the claim. + request_id = f"{base_request_id}:{now_ms}" + admission = request_admission(request_id) admitted_request.clear() if admission.get("admitted") is True: admitted_request["request_id"] = request_id admitted_request["reserved"] = admission.get("reserved") is True - if on_admitted is not None: - on_admitted() - return { + hook = admission_hook[0] + if hook is not None: + hook() + else: + hook_pending[0] = True + latest_admission = { key: admission.get(key) for key in ( "admitted", "reserved", "resumed", "reason", "next_eligible_at_ms", "min_interval_minutes", "pre_model_admission", ) } + return dict(latest_admission) + + def accepted() -> dict[str, Any] | None: + return dict(latest_admission) if latest_admission is not None else None def confirm() -> None: request_id = admitted_request.get("request_id") @@ -131,4 +124,7 @@ def confirm() -> None: "managed Turn start could not be confirmed against the owner cadence store" ) - return ManagedCadenceStart(admit=admit, confirm=confirm) + return ManagedCadenceStart( + admit=admit, confirm=confirm, accepted=accepted, + set_on_admitted=set_on_admitted, + ) diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index 0cb2d6cc90..cd29ba79c2 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -92,7 +92,7 @@ def execute_turn_run_once( native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None, native_completion_lease_resolver: Callable[[], Mapping[str, Any]] | None = None, native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None = None, - preclaim_cadence_wait: Mapping[str, Any] | None = None, + managed_cadence: Any | None = None, ) -> dict[str, Any]: """Adapt one prepared CLI Turn plan to the bounded Turn driver.""" execution_started = False @@ -109,10 +109,11 @@ def execute_turn_run_once( native_task_validator=native_task_validator, validator_builder=build_loopx_turn_command_validator, ) - if preclaim_cadence_wait is not None: - # The owner's interval read refused the start before any Turn write. - # Return its wait through the normal execution projection without - # reserving an interval, claiming a Todo, or journaling a Turn. + preadmitted = managed_cadence.accepted() if managed_cadence is not None else None + if preadmitted is not None and preadmitted.get("admitted") is not True: + # The owner refused the interval before this Turn wrote its receipt + # or claimed a Todo. Report that same admission without reserving, + # claiming or journaling anything. from ..control_plane.turn_driver.execution_readback import execution_payload planned_host = payload.get("host") @@ -120,11 +121,8 @@ def execute_turn_run_once( "status": "interval_wait", "host": planned_host if isinstance(planned_host, Mapping) else {}, "completed_phases": [], - "reason": str( - preclaim_cadence_wait.get("reason") - or "automatic start not admitted" - ), - "admission": dict(preclaim_cadence_wait), + "reason": str(preadmitted.get("reason") or "automatic start not admitted"), + "admission": preadmitted, } return execution_payload( payload, @@ -902,14 +900,19 @@ def on_managed_start_admitted() -> None: goal_ref=goal_ref, ) - managed_cadence = managed_cadence_start( - runtime_root=runtime_root, - goal_id=args.goal_id, - agent_id=args.agent_id, - automation_id=args.automation_id, - manual_reason=args.manual_interval_bypass_reason, - on_admitted=on_managed_start_admitted, - ) + if managed_cadence is None: + managed_cadence = managed_cadence_start( + runtime_root=runtime_root, + goal_id=args.goal_id, + agent_id=args.agent_id, + automation_id=args.automation_id, + manual_reason=args.manual_interval_bypass_reason, + on_admitted=on_managed_start_admitted, + ) + else: + # A start admitted ahead of the claim keeps this Turn's receipt + # writer attached to the same owner reservation. + managed_cadence.set_on_admitted(on_managed_start_admitted) execution_started = bool(args.execute) result = run_loopx_turn_once( @@ -942,7 +945,12 @@ def on_managed_start_admitted() -> None: post_settlement=(configured_post_settlement( payload, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id, ) if args.execute else None), - admit_start=managed_cadence.admit if args.execute else None, + admit_start=( + managed_cadence.admit + if args.execute and preadmitted is None + else None + ), + admission=preadmitted if args.execute else None, confirm_start=managed_cadence.confirm if args.execute else None, goal_admission=goal_admission, ) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index b2d23bceb9..998a82578c 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -190,6 +190,7 @@ def builtin_execution_preparation_claim( cwd: Path, turn_instance_id: str | None = None, goal_ref: Mapping[str, object] | None = None, + admit_start: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None, runner: CommandRunner | None = None, timeout: float = 30, ) -> str | None: @@ -264,6 +265,19 @@ def builtin_execution_preparation_claim( exact_turn_instance_id = normalize_turn_instance_id( receipt_turn_instance_id ) + if admit_start is not None: + # Reserve the owner interval before this Turn writes its receipt or + # claim. A refusal therefore leaves no lease, receipt or journal + # behind; an accepted reservation is carried to the driver unchanged. + admission = admit_start( + { + "turn_key": exact_turn_instance_id + or f"{goal_id}:{caller_agent_id}:{todo_id}", + "attempt": 1, + } + ) + if admission.get("admitted") is not True: + return None if exact_turn_instance_id is not None: # The receipt lookup and the ordinary claim are separate operations. # Reserve the exact Turn binding before the claim can create a lease for diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index 3ca076c661..2522a2b9cb 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/loopx/control_plane/turn_driver/executor.py @@ -1539,6 +1539,7 @@ def run_loopx_turn_once( post_settlement: PostSettlement | None = None, settlement_admission: Callable[[], AbstractContextManager] | None = None, admit_start: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None, + admission: Mapping[str, Any] | None = None, confirm_start: Callable[[], None] | None = None, goal_admission: FirstPartyHostGoalAdmission | None = None, ) -> dict[str, Any]: @@ -1701,9 +1702,11 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None: build_loopx_turn_host_request(capture.plan) if capture is not None else request, operation="host_dispatch_context", ) - admission = None - if needs_host and admit_start is not None: - admission = admit_start( + # A caller that already admitted the start ahead of the ordinary claim + # hands that exact reservation down, so no second interval is reserved. + admitted = dict(admission) if needs_host and admission is not None else None + if needs_host and admitted is None and admit_start is not None: + admitted = admit_start( { "turn_key": turn_key, "attempt": int(journal.get("host_attempt_count") or 0) + 1 @@ -1711,6 +1714,8 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None: else 1, } ) + admission = admitted + if needs_host and admission is not None: if admission.get("admitted") is not True: waiting = { "status": "interval_wait", diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 45f11933c5..1c614478bb 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1468,6 +1468,7 @@ def test_run_once_interval_wait_does_not_claim_the_next_todo( "--execute", ]) waiting = json.loads(output.getvalue()) + print("DEBUGWAIT", json.dumps(waiting, ensure_ascii=False)[:2000]) assert waiting_code == 1, waiting assert waiting["status"] == "interval_wait", waiting assert waiting["effects"]["host_invoked"] is False, waiting @@ -1489,6 +1490,92 @@ def test_run_once_interval_wait_does_not_claim_the_next_todo( assert lease["lease"] is None, lease +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_run_once_manual_interval_bypass_starts_the_next_todo( + tmp_path, monkeypatch, provider, +): + from tests.test_loopx_turn_codex_cli import _fake_codex + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture( + tmp_path, + provider=provider, + handoff_mode="hard_lease", + extra_agent_todo_lines=( + "- [ ] [P1] Complete the manual bypass fixture.", + " ", + ), + ) + binary, log = _fake_codex(tmp_path) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + + output = io.StringIO() + with contextlib.redirect_stdout(output): + read_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "automation-cadence", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", + ]) + policy = read_cli_policy_payload(output.getvalue(), returncode=read_code) + assert read_code == 0, policy + output = io.StringIO() + with contextlib.redirect_stdout(output): + configure_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "automation-cadence", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--min-interval-minutes", "3", + "--expected-revision", str(policy["configuration_revision"]), + "--owner-reference", "manual bypass fixture", "--execute", + ]) + configured = read_cli_policy_payload(output.getvalue(), returncode=configure_code) + assert configure_code == 0 and configured["written"] is True, configured + + output = io.StringIO() + with contextlib.redirect_stdout(output): + first_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--turn-instance-id", "manual-first", "--host", "codex-cli", + "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + first = json.loads(output.getvalue()) + assert first_code == 0 and first["status"] == "committed", first + assert len([json.loads(line) for line in log.read_text().splitlines()]) == 1 + + output = io.StringIO() + with contextlib.redirect_stdout(output): + manual_code = cli_main([ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--todo-id", "todo_fixture0002", "--turn-instance-id", "manual-second", + "--manual-interval-bypass-reason", "explicit operator confirmation", + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary), + "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", + "--validation-command-json", + json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]), + "--execute", + ]) + manual = json.loads(output.getvalue()) + assert manual_code == 0, manual + assert manual["status"] == "committed", manual + assert manual["effects"]["host_invoked"] is True, manual + calls = [json.loads(line) for line in log.read_text().splitlines()] + assert len(calls) == 2, calls + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_run_once_reserves_turn_before_claim_under_competing_guard( tmp_path, monkeypatch, provider, @@ -2970,11 +3057,14 @@ def checkpoint(path, journal): def fresh_cadence(**kwargs): cadence = make_cadence(**kwargs) + latest: dict[str, Any] = {} def admit(identity): attempts.append(dict(identity)) admission = cadence.admit(identity) admissions.append(dict(admission)) + latest.clear() + latest.update(admission) return admission def confirm(): @@ -2987,7 +3077,15 @@ def confirm(): if not interrupted and fault == "confirm_after": interrupt() - return turn_cadence.ManagedCadenceStart(admit, confirm) + def accepted(): + return dict(latest) if latest else None + + def set_on_admitted(callback): + cadence.set_on_admitted(callback) + + return turn_cadence.ManagedCadenceStart( + admit, confirm, accepted, set_on_admitted, + ) monkeypatch.setattr(executor, "_write_journal", checkpoint) monkeypatch.setattr(turn_run_once_command, "managed_cadence_start", fresh_cadence)