diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.md index 74abb2b6c7..33838e0114 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.md @@ -3637,3 +3637,43 @@ installed protocol's CI coverage gap without selecting D2/D5 or a minimal preset. Real Host initial/revised/compacted/lost-ACK adoption, complete V1–V5/V10, R1/M3, Standard/workspace publication and original-goal regression remain open. Preserve the earlier failures, frozen budgets and unchanged governance owners. + +### 2026-11-13 — Managed Codex preparation respects start admission + +The existing built-in Codex CLI route can prepare an eligible hard-lease Todo +through the exact ordinary claim advertised by its typed decision. The caller +then rereads the same Todo and requires ready before Host launch. This removes +the separate manual claim step; it does not select a new host or grant authority. +The prior proposal at `d4348db14d51952a74247e3003768a2f969c7216` still wrote a +lease and Turn receipt before invalid validator input or cadence waiting was +rejected. The repaired route validates the existing Host arguments and obtains +start admission from the same TypeScript cadence owner before those writes, +then reserves the exact Turn binding before the ordinary claim. Start admission +runs inside the existing cross-runtime Turn binding lock, so a competing caller +cannot reserve one Todo's interval and then let a different binding win. + +Preparation and execution reuse the settlement effect identity for the start +reservation. A claim acknowledgement lost before journal creation can therefore +recover its already confirmed lease and original unstarted reservation. The +existing transaction-key reservation is retained as an exact recovery fallback +for historical Codex journals; a started request cannot use that fallback to +run again. An uncertain claim reports unknown state effects instead of asserting +that nothing was written. A successful claim receipt remains visible if the +subsequent Host step fails. The existing spend precheck refuses a fresh claim +for a settled Turn even after its lease is released, directing recovery to the +recorded journal. No new persisted schema or governance owner is added. + +Real source CLI checks on isolated File and SQLite reject malformed validator +JSON/argv and scheduled waits with unchanged leases, Turn receipts, journals +and Host calls. The same independent oracles fail on the prior proposal. +Interruption after the ordinary claim recovers with one Host attempt and one +settlement; recorded-key replay has no effects. Installed core qualification +uses the same existing package lane and synthetic Codex transport, with an +independent artifact validator and result readback. + +This is an executable-start prerequisite for V2/V3/V4, not whole R1 acceptance. +The Host is synthetic: actual model adoption, complete V1–V5/V10, default +closure, paired complete costs, progressive product control, R1/M3 and original +Goal regression remain unqualified. Keep the nine original requirements, +previous failed attempts and frozen budget failures; no minimal preset, +provider default, package publication or paid operation is promoted here. diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md index 0d13a0f615..a8ca47279c 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md @@ -2836,3 +2836,33 @@ named-import 子进程对照通过；早期不完整观察器与本次核验后 不选择 D2/D5 或最小 preset。真实 Host 的 initial/revised/compacted/lost-ACK 采用、完整 V1–V5/V10、R1/M3、Standard/workspace 发布及原始目标回归继续开放。 保留早期失败、冻结预算和未变的治理 owner。 + +### 2026-11-13 — 托管 Codex 准备遵守启动准入 + +现有内置 Codex CLI 路径可对符合条件的 hard-lease Todo 执行 typed 决策公布的 +精确普通领取，随后重读同一 Todo，确认 ready 才启动 Host。这消除了调用方单独 +手工领取的步骤，没有选择新的 Host，也没有授予额外权限。此前提案 +`d4348db14d51952a74247e3003768a2f969c7216` 在拒绝无效 validator 输入或调度 +等待之前仍会写入 lease 和 Turn 回执。修复路径先校验现有 Host 参数，通过同一 +TypeScript cadence owner 取得启动准入，再原子保留精确 Turn 绑定并执行普通领取。 +启动准入回调位于既有跨 runtime Turn 绑定锁内；竞争请求不能先占用另一 Todo 的 +启动预约，再让不同绑定获胜。 + +准备与执行使用 settlement effect identity 复用同一启动预约。因此，普通领取的 +ACK 在 journal 建立前丢失时，可恢复已确认的 lease 和原有未启动预约。历史 Codex +journal 使用的 transaction-key 预约保留精确恢复兼容分支；已启动请求不能借该 +分支重复执行。领取结果不确定时，状态副作用报告为 unknown，不能宣称没有写入；领取成功但 +后续 Host 步骤失败时，结果保留已确认的领取写入。 +已结算的原 Turn 即使 lease 后来被释放，也通过现有 spend precheck +拒绝再次领取，并引导原 journal 回读。没有新增持久化 schema 或治理 owner。 + +隔离 File 和 SQLite 的真实源码 CLI 检查表明，无效 validator JSON/argv 和调度 +等待均保持 lease、Turn 回执、journal 与 Host 调用不变；同一独立 oracle 在此前 +提案上失败。普通领取后中断可恢复为一次 Host 执行、一次结算，原 journal key +重放没有副作用。独立 core 安装验证复用现有 package lane 与合成 Codex transport， +使用独立产物 validator 和结果回读。 + +这仅交付 V2/V3/V4 的可执行启动前置，不等于完整 R1 验收。Host 为合成 transport， +实际模型采用、完整 V1–V5/V10、默认闭包、配对完整成本、渐进产品控制、R1/M3 与 +原始 Goal 回归仍未验收。保留原九项要求、此前失败尝试和冻结预算失败；本次没有 +推广 minimal preset、provider 默认、包发布或付费操作。 diff --git a/docs/product/runtimes/codex-cli/loopx-turn-codex-cli-quickstart.md b/docs/product/runtimes/codex-cli/loopx-turn-codex-cli-quickstart.md index c2f4168f01..7e076763a1 100644 --- a/docs/product/runtimes/codex-cli/loopx-turn-codex-cli-quickstart.md +++ b/docs/product/runtimes/codex-cli/loopx-turn-codex-cli-quickstart.md @@ -41,6 +41,16 @@ loopx turn run-once \ --execute ``` +For an eligible task that requires preparation, this command now performs the +ordinary managed claim automatically, rechecks the same task, then starts Codex. +Previously the caller had to claim it separately. Invalid validator arguments +and scheduled waits do not claim the task or write its Turn receipt. A rejected +claim may have committed before its acknowledgement was lost: inspect the task +lease and retry the original Turn identity before starting new work. Confirmed +results recover through their recorded `--resume-turn-key`. A settled Turn +refuses a fresh claim even if its task lease was later released; use +`--resume-turn-instance-id` for the original recorded execution. + The built-in Codex adapter means there is no adapter program to write for this path. A different Agent CLI uses the same Turn contract through a thin `generic-cli` adapter that reads one JSON request from stdin and writes one JSON diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 71dd9798fb..7bbf6e0205 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -74,6 +74,12 @@ def handle_turn_command( ) payload: dict[str, Any] = {} turn_start_hook_dispatch: dict[str, Any] = {} + preparation_state_written: bool | None = False + + def project_preparation_effects(value: dict[str, Any]) -> None: + effects = value.setdefault("effects", {}) + if preparation_state_written is not False and effects.get("state_written") is not True: + effects["state_written"] = preparation_state_written try: if (native_host_runner is not None or native_session_binding_resolver is not None) and ( args.host != "codex-app-server" or native_host_runner is None or native_session_binding_resolver is None @@ -89,6 +95,61 @@ def handle_turn_command( registry_path=registry_path, runtime_root_override=runtime_root_arg, ) + preparation_admission: dict[str, Any] | None = None + 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" + and args.host == "codex-cli" + and bool(getattr(args, "execute", False)) + ): + from ..control_plane.turn_driver.codex_execution_preparation import ( + builtin_execution_preparation_claim, + ) + + from .turn_host_request import prepare_turn_host_arguments + from ..control_plane.turn_driver import build_loopx_turn_command_validator + from .turn_cadence import managed_cadence_start + + project_cwd = Path(args.project).expanduser().resolve() + + def admit_preparation(identity): + nonlocal preparation_admission + prepare_turn_host_arguments( + args, native_host_runner=None, native_session_binding_resolver=None, + native_task_validator=None, validator_builder=build_loopx_turn_command_validator, + ) + args.turn_instance_id = identity.turn_instance_id + 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, + request_id=identity.effect_id, + ) + preparation_admission = cadence.admit({"attempt": 1}) + return preparation_admission.get("admitted") is True + + def record_claim(receipt: Mapping[str, Any]) -> None: + nonlocal preparation_state_written + # Echo the canonical mutation receipt. A missing/unchanged + # claim field cannot prove the earlier binding wrote nothing. + preparation_state_written = True if receipt.get("changed") is True else None + + 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, + turn_instance_id=args.turn_instance_id, + goal_ref=goal_ref, + admit_start=admit_preparation, + on_claim=record_claim, + ) + 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 +332,24 @@ 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 preparation_admission is not None and preparation_admission.get("admitted") is not True: + from ..control_plane.turn_driver.execution_readback import execution_payload + waiting_plan = build_loopx_turn_plan( + decision, host=args.host, execution_mode=args.execution_mode, + scheduler_owner=args.scheduler_owner, session_binding=session_binding, + turn_instance_id=args.turn_instance_id, iteration_context_policy=args.iteration_context, + session_scope=args.session_scope, goal_ref=goal_ref, + ) + waiting = execution_payload( + waiting_plan, {"status": "interval_wait", "host": waiting_plan["host"], + "completed_phases": [], "reason": preparation_admission.get("reason"), + "admission": preparation_admission}, + execute=True, replayed=False, effects=waiting_plan["effects"], + ) + print_payload(waiting, output_format(args), _render_loopx_turn_execution_markdown) + return 1 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 +361,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(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"} @@ -384,6 +461,7 @@ def read_decision_owner(): execute=bool(args.execute), executed=False) from ..capabilities.turn_start import attach_turn_start_effects attach_turn_start_effects(payload, turn_start_hook_dispatch) + project_preparation_effects(payload) print_payload(payload, output_format(args), _render_loopx_turn_plan_markdown) return 0 @@ -417,6 +495,7 @@ def read_decision_owner(): ) from ..capabilities.turn_start import attach_turn_start_effects attach_turn_start_effects(payload, turn_start_hook_dispatch) + project_preparation_effects(payload) renderer = ( _render_loopx_turn_execution_markdown if args.turn_command == "run-once" diff --git a/loopx/cli_commands/turn_cadence.py b/loopx/cli_commands/turn_cadence.py index f5f965e1e7..d5e3befd8d 100644 --- a/loopx/cli_commands/turn_cadence.py +++ b/loopx/cli_commands/turn_cadence.py @@ -29,6 +29,7 @@ def managed_cadence_start( automation_id: str | None, manual_reason: str | None, on_admitted: Callable[[], None] | None = None, + request_id: str | None = None, ) -> ManagedCadenceStart: """Bind one managed Turn start to the TypeScript owner-cadence store.""" @@ -36,24 +37,36 @@ def managed_cadence_start( def admit(identity: Mapping[str, Any]) -> dict[str, Any]: 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, - ) + start_request_id = f"{request_id or identity['turn_key']}:{identity['attempt']}" + def reserve(start_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": start_id, + "trigger_at_ms": now_ms, + "now_ms": now_ms, + "manual_reason": manual_reason, + }, + retry_safe=False, + ) + admission = reserve(start_request_id) + legacy_request_id = f"{identity['turn_key']}:{identity['attempt']}" if "turn_key" in identity else None + if (request_id is not None and legacy_request_id is not None + and start_request_id != legacy_request_id + and admission.get("admitted") is not True + and admission.get("reason") == "minimum_interval_wait"): + # Historical Codex journals reserved by transaction key. Retry only + # this exact identity, after a no-write wait, so their unstarted + # reservations remain recoverable; never bypass a started request. + admission = reserve(legacy_request_id) + start_request_id = legacy_request_id admitted_request.clear() if admission.get("admitted") is True: - admitted_request["request_id"] = request_id + admitted_request["request_id"] = start_request_id admitted_request["reserved"] = admission.get("reserved") is True if on_admitted is not None: on_admitted() diff --git a/loopx/cli_commands/turn_host_request.py b/loopx/cli_commands/turn_host_request.py index 6c83d483dc..34852557a3 100644 --- a/loopx/cli_commands/turn_host_request.py +++ b/loopx/cli_commands/turn_host_request.py @@ -14,19 +14,15 @@ from ..control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission -def prepare_turn_host_request( - args: argparse.Namespace, payload: Mapping[str, Any], *, +def prepare_turn_host_arguments( + args: argparse.Namespace, *, 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]]: +) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None]: + """Validate transport arguments before preparation can write authority state.""" 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 ): @@ -72,6 +68,26 @@ def prepare_turn_host_request( timeout_seconds=args.validation_timeout_seconds, failure_recovery_kind=args.validation_failure_kind, ) + return project, raw_argv, task_validator + + +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]]: + 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") + project, raw_argv, task_validator = prepare_turn_host_arguments( + args, native_host_runner=native_host_runner, + native_session_binding_resolver=native_session_binding_resolver, + native_task_validator=native_task_validator, validator_builder=validator_builder, + ) from ..control_plane.turn_driver.driver import read_turn_execution_context return project, raw_argv, task_validator, read_turn_execution_context(payload) diff --git a/loopx/cli_commands/turn_rendering.py b/loopx/cli_commands/turn_rendering.py index 8877fc80a9..ef39efc376 100644 --- a/loopx/cli_commands/turn_rendering.py +++ b/loopx/cli_commands/turn_rendering.py @@ -82,6 +82,9 @@ def build_turn_error_payload( key: True if planned_effects.get(key) is True else None if execution_started else False for key in ("host_invoked", "state_written", "scheduler_acknowledged", "quota_spent") } + preparation_effects = getattr(exc, "execution_effects", None) + if isinstance(preparation_effects, Mapping): + effects.update({key: preparation_effects[key] for key in effects if key in preparation_effects}) return { **({"error_code": error_code, **(error_payload if isinstance(error_payload, Mapping) else {})} if isinstance(error_code, str) else {}), diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index d5af537fde..b93a5df82f 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -882,6 +882,7 @@ def on_managed_start_admitted() -> None: automation_id=args.automation_id, manual_reason=args.manual_interval_bypass_reason, on_admitted=on_managed_start_admitted, + request_id=settlement_identity.effect_id if args.host == "codex-cli" else None, ) execution_started = bool(args.execute) diff --git a/loopx/control_plane/quota/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py index 2c3d7b36d4..8eaf717a38 100644 --- a/loopx/control_plane/quota/heartbeat_receipt.py +++ b/loopx/control_plane/quota/heartbeat_receipt.py @@ -1,7 +1,7 @@ from __future__ import annotations from .effective_action import EffectiveAction -from collections.abc import Mapping +from collections.abc import Callable, Mapping from pathlib import Path from ...file_lock import exclusive_cross_runtime_file_lock @@ -205,6 +205,7 @@ def ensure_turn_heartbeat_settlement_receipt( semantic_replan_obligation_id: str | None, semantic_replan_capability_guard: Mapping[str, object] | None = None, goal_ref: Mapping[str, object] | None = None, + before_bind: Callable[[], None] | None = None, ) -> dict[str, object]: """Idempotently bind a Turn-created quota guard to its settlement identity. @@ -242,13 +243,18 @@ def ensure_turn_heartbeat_settlement_receipt( identity.binding_id, identity.effect_id, ) + observed = _receipt_settlement_identity(effective) if effective is not None else None + if observed is not None and observed != expected: + raise HeartbeatReceiptIdentityConflictError( + "Turn heartbeat receipt belongs to another settlement identity" + ) + if before_bind is not None: + # The existing cross-runtime binding lock serializes the competing + # identity before transport-specific start admission. Its owner may + # refuse admission without writing a receipt or claiming work. + before_bind() if effective is not None: - observed = _receipt_settlement_identity(effective) if observed is not None: - if observed != expected: - raise HeartbeatReceiptIdentityConflictError( - "Turn heartbeat receipt belongs to another settlement identity" - ) effective_details = effective.get("details") effective_details = ( effective_details diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index 7c109f832e..8648adaefb 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -11,9 +11,13 @@ import subprocess from collections.abc import Callable, Mapping, Sequence from pathlib import Path -from typing import Any +from typing import Any, TYPE_CHECKING -from ...command_invocation import current_loopx_cli_argv +if TYPE_CHECKING: + from ..quota.effect_program import SettlementIdentity + +from ...command_invocation import command_argv, current_loopx_cli_argv +from ...turn_identity import mint_turn_instance_id, normalize_turn_instance_id from ..content_digest import BARE_SHA256_PATTERN @@ -100,11 +104,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="replace", 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( @@ -116,13 +131,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 "{}") @@ -152,6 +171,195 @@ 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" + execution_effects = { + "host_invoked": False, "state_written": None, + "scheduler_acknowledged": False, "quota_spent": False, + } + + +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, + turn_instance_id: str | None = None, + goal_ref: Mapping[str, object] | None = None, + runner: CommandRunner | None = None, + timeout: float = 30, + admit_start: Callable[[SettlementIdentity], bool] | None = None, + on_claim: Callable[[Mapping[str, Any]], None] | None = None, +) -> 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 + 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", 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 goal_id + or not caller_agent_id + or not decision_agent_id + or not preparation_agent_id + or not 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() + or not cwd.is_absolute() + ): + return None + heartbeat = decision.get("heartbeat_receipt") + # 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): + 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 None and admit_start is not None: + exact_turn_instance_id = mint_turn_instance_id(prefix="turn") + 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 + # 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( + 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, + ) + ) + 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") + class StartDeferred(Exception): + pass + + def before_bind() -> None: + from ..quota.settlement import read_quota_spend_precheck + + original = read_quota_spend_precheck( + runtime_root, goal_id=goal_id, agent_id=caller_agent_id, todo_id=todo_id, + turn_instance_id=exact_turn_instance_id, allow_unbound_binding=True, + registry_path=registry_path, goal_ref=goal_ref, + ) + if original is not None and original.spend_run_present: + raise ValueError( + "This Turn is already settled; use --resume-turn-instance-id " + f"{exact_turn_instance_id} to read its confirmed journal before new work" + ) + if admit_start is not None and not admit_start(settlement_identity): + raise StartDeferred + # 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. + try: + ensure_turn_heartbeat_settlement_receipt( + runtime_root, + settlement_identity, + semantic_replan_guard_scoped=False, + semantic_replan_obligation_id=None, + goal_ref=goal_ref, + **({"before_bind": before_bind} if admit_start is not None else {}), + ) + except StartDeferred: + return None + try: + claim_receipt = _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, UnicodeDecodeError) as exc: + raise ExecutionPreparationClaimFailed( + f"managed claim for Todo {todo_id} was refused before delivery: {exc}" + ) from exc + if on_claim is not None: + on_claim(claim_receipt) + 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_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index 2c01787ca4..d52d284c7d 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/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..47692a920e --- /dev/null +++ b/tests/control_plane/test_turn_run_once_claim_preparation.py @@ -0,0 +1,318 @@ +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, agent_id=AGENT, 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, + ] + + +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)], +) +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, agent_id=AGENT, 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, agent_id=AGENT, 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, agent_id=AGENT, 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, agent_id=AGENT, 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, 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, + ] + + +def test_a_legacy_unstarted_cadence_reservation_recovers_without_a_second_start(tmp_path: Path) -> None: + from loopx.cli_commands.turn_cadence import managed_cadence_start + from loopx.control_plane.effect_runtime import effect_runtime_result + + runtime = tmp_path / "cadence-runtime" + effect_runtime_result("quota.automation_cadence.manage", { + "runtime_root": str(runtime), "goal_id": GOAL, "agent_id": AGENT, + "operation": "configure", "min_interval_minutes": 1440, "expected_revision": 0, + "owner_reference": "fixture-owner", "execute": True, + }) + old = managed_cadence_start(runtime_root=runtime, goal_id=GOAL, agent_id=AGENT, + automation_id=None, manual_reason=None) + attempt = {"turn_key": "sha256:" + "a" * 64, "attempt": 1} + first = old.admit(attempt) + assert first["admitted"] is True and first["reserved"] is True + migrated = managed_cadence_start(runtime_root=runtime, goal_id=GOAL, agent_id=AGENT, + automation_id=None, manual_reason=None, request_id=f"{GOAL}:{AGENT}:{TODO}:original-turn") + recovered = migrated.admit(attempt) + assert recovered["admitted"] is True and recovered["resumed"] is True + migrated.confirm() + duplicate = migrated.admit(attempt) + assert duplicate["admitted"] is False and duplicate["reason"] == "duplicate_or_stale_trigger" diff --git a/tests/test_core_distribution.py b/tests/test_core_distribution.py index 65f047f428..ffc3b0ad5a 100644 --- a/tests/test_core_distribution.py +++ b/tests/test_core_distribution.py @@ -444,3 +444,46 @@ def test_installed_core_rejects_unverified_artifact_without_settlement(core_pyth assert result["effects"]["quota_spent"] is False assert _spends(runtime) == 0 assert (workspace / "host-invocations").read_text().splitlines() == ["called"] + + +def test_installed_core_codex_prepares_only_an_admitted_valid_request(core_python, task, tmp_path, monkeypatch): + from tests.test_loopx_turn_codex_cli import _fake_codex + + workspace, runtime, registry, _ = task + prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + todo = json.loads(_run(core_python, [*prefix, "todo", "list", "--goal-id", GOAL], workspace).stdout) + todo_id = todo["todos"][0]["todo_id"] + lease = json.loads(_run(core_python, [*prefix, "task-lease", "inspect", "--goal-id", GOAL, + "--todo-id", todo_id], workspace).stdout)["lease"] + _run(core_python, [*prefix, "task-lease", "release", "--goal-id", GOAL, "--todo-id", todo_id, + "--owner", AGENT, "--idempotency-key", lease["idempotency_key"], + "--expected-version", str(lease["version"])], workspace) + binary, log = _fake_codex(tmp_path) + binary.write_text(binary.read_text().replace('output_path = ', + 'pathlib.Path("squares.json").write_text(json.dumps([n*n for n in range(8)]))\noutput_path = ')) + home = tmp_path / "codex-home" + home.mkdir() + environment = {"FAKE_CODEX_LOG": str(log), "CODEX_HOME": str(home)} + command = [*prefix, "turn", "run-once", "--host", "codex-cli", "--goal-id", GOAL, + "--agent-id", AGENT, "--todo-id", todo_id, "--project", str(workspace), + "--codex-bin", str(binary), "--codex-model", "fixture-model", "--codex-reasoning-effort", "high", + "--codex-sandbox", "read-only", "--no-global-sync", "--execute"] + refused = json.loads(_run(core_python, [*command, "--turn-instance-id", "invalid-request", + "--validation-command-json", "{"], workspace, environment=environment, expected=1).stdout) + assert all(value is False for value in refused["effects"].values()) + held = json.loads(_run(core_python, [*prefix, "task-lease", "inspect", "--goal-id", GOAL, + "--todo-id", todo_id], workspace).stdout) + assert held["active"] is False and not log.exists() and _spends(runtime) == 0 + result = json.loads(_run(core_python, [*command, "--turn-instance-id", "admitted-request", + "--validation-command-json", json.dumps([core_python, "-I", "-c", VALIDATOR_SOURCE])], + workspace, environment=environment).stdout) + assert result["status"] == "committed" and result["validation"]["status"] == "passed", result + assert json.loads((workspace / "squares.json").read_text()) == [0, 1, 4, 9, 16, 25, 36, 49] + assert len(log.read_text().splitlines()) == 1 and _spends(runtime) == 1 + resume_command = [item for index, item in enumerate(command) + if item != "--todo-id" and (index == 0 or command[index - 1] != "--todo-id")] + replay = json.loads(_run(core_python, [*resume_command, "--resume-turn-key", result["resume_turn_key"], + "--validation-command-json", json.dumps([core_python, "-I", "-c", VALIDATOR_SOURCE])], + workspace, environment=environment).stdout) + assert replay["replayed"] is True and all(value is False for value in replay["effects"].values()) + assert len(log.read_text().splitlines()) == 1 and _spends(runtime) == 1 diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index b748707010..b5b0b3464d 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1126,6 +1126,306 @@ def test_default_agent_session_reuses_planning_across_completed_todos(tmp_path, } +@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 + + 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])) + + 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 + + +@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 + + +@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 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(), @@ -1494,6 +1794,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 +1875,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( @@ -4334,3 +4635,317 @@ def fake_codex_host( # selected Todo has already completed and disappeared from the live route. assert session_binding_calls == 2 assert session_sources[0] == session_sources[1] + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("refusal", ["invalid_json", "invalid_argv", "interval_wait"]) +def test_builtin_claim_preflight_refusal_preserves_lease_receipts_and_journal( + tmp_path, monkeypatch, provider, refusal, +): + from tests.test_loopx_turn_codex_cli import _fake_codex + from loopx.rollout_event_log import rollout_event_log_path + + 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] Advance the next independent 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])) + route = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + + def invoke(arguments): + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main([*route, *arguments]) + return code, json.loads(output.getvalue()) + + command = ["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", "--execute"] + validator = json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]) + if refusal == "interval_wait": + code, policy = invoke(["automation-cadence", "--goal-id", "loopx-turn-fixture", + "--agent-id", "codex-fixture", "--min-interval-minutes", "1440", "--expected-revision", "0", + "--owner-reference", "fixture-owner", "--execute"]) + assert code == 0, policy + code, first = invoke([*command, "--todo-id", "todo_fixture0001", "--turn-instance-id", "first-start", + "--validation-command-json", validator]) + assert code == 0 and first["status"] == "committed", first + rejected_validator = validator + else: + rejected_validator = "{" if refusal == "invalid_json" else json.dumps([1]) + lease_command = ["task-lease", "inspect", "--goal-id", "loopx-turn-fixture", "--todo-id", "todo_fixture0002"] + _, before_lease = invoke(lease_command) + assert before_lease["lease"] is None and before_lease["active"] is False + event_path = rollout_event_log_path(runtime, "loopx-turn-fixture") + before_events = event_path.read_bytes() if event_path.exists() else b"" + before_host = log.read_bytes() if log.exists() else b"" + journal_root = runtime / "goals" / "loopx-turn-fixture" / "turns" + before_journals = {p.name: p.read_bytes() for p in journal_root.glob("*.json")} + code, result = invoke([*command, "--todo-id", "todo_fixture0002", "--turn-instance-id", "refused-start", + "--validation-command-json", rejected_validator]) + assert code == 1, result + if refusal == "interval_wait": + assert result["status"] == "interval_wait", result + assert result["admission"]["next_eligible_at_ms"] > 0 + assert all(value is False for value in result["effects"].values()), result + _, after_lease = invoke(lease_command) + assert after_lease["lease"] is None and after_lease["active"] is False + assert (event_path.read_bytes() if event_path.exists() else b"") == before_events + assert (log.read_bytes() if log.exists() else b"") == before_host + assert {p.name: p.read_bytes() for p in journal_root.glob("*.json")} == before_journals + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_claim_ack_loss_before_journal_resumes_the_original_cadence_reservation( + tmp_path, monkeypatch, provider, +): + from tests.test_loopx_turn_codex_cli import _fake_codex + from loopx.control_plane.turn_driver import codex_execution_preparation + from loopx.rollout_event_log import rollout_event_log_path + + 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])) + route = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + + def invoke(arguments): + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main([*route, *arguments]) + return code, json.loads(output.getvalue()) + + code, policy = invoke(["automation-cadence", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--min-interval-minutes", "1440", "--expected-revision", "0", "--owner-reference", "fixture-owner", "--execute"]) + assert code == 0, policy + command = ["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", "--execute", "--turn-instance-id", "interrupted-preparation", + "--validation-command-json", json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"])] + claim = codex_execution_preparation._run_json + claims = 0 + + def lose_claim_ack(*args, **kwargs): + nonlocal claims + claim(*args, **kwargs) + claims += 1 + raise RuntimeError("injected lost acknowledgement after the ordinary claim") + + monkeypatch.setattr(codex_execution_preparation, "_run_json", lose_claim_ack) + code, interrupted = invoke(command) + assert code == 1 and interrupted["error_code"] == "turn_execution_preparation_claim_failed", interrupted + assert interrupted["effects"] == { + "host_invoked": False, "state_written": None, "quota_spent": False, "scheduler_acknowledged": False, + } + assert not log.exists() + assert not list((runtime / "goals/loopx-turn-fixture/turns").glob("*.json")) + lease_command = ["task-lease", "inspect", "--goal-id", "loopx-turn-fixture", "--todo-id", "todo_fixture0001"] + _, lease = invoke(lease_command) + assert lease["active"] is True + monkeypatch.setattr(codex_execution_preparation, "_run_json", claim) + code, recovered = invoke(command) + assert code == 0 and recovered["status"] == "committed", recovered + assert claims == 1 and len(log.read_text().splitlines()) == 1 + _, recovered_lease = invoke(lease_command) + assert recovered_lease["lease"] == lease["lease"] + rows = [json.loads(line) for line in rollout_event_log_path(runtime, "loopx-turn-fixture").read_text().splitlines()] + assert sum(row.get("classification") == "quota_slot_spent" for row in rows) == 1 + # Fresh planning is not journal recovery. It may refuse the settled Turn, + # but cannot repeat its confirmed Host; resume uses the recorded key. + invoke(command) + assert len(log.read_text().splitlines()) == 1 + resume_command = [item for index, item in enumerate(command) + if item != "--turn-instance-id" and (index == 0 or command[index - 1] != "--turn-instance-id")] + code, replay = invoke([*resume_command, "--resume-turn-key", recovered["resume_turn_key"]]) + assert code == 0 and replay["replayed"] is True, replay + assert all(value is False for value in replay["effects"].values()), replay + assert len(log.read_text().splitlines()) == 1 + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("guard_entry", ["loopx.entrypoint", "loopx.cli"]) +def test_concurrent_guard_cannot_reserve_the_other_todos_start( + tmp_path, monkeypatch, provider, guard_entry, +): + import subprocess + import time + from tests.test_loopx_turn_codex_cli import _fake_codex + from loopx.cli_commands import turn_cadence + + 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 concurrent second fixture.", + " ", + )) + binary, log = _fake_codex(tmp_path) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + + def invoke(arguments): + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main([*prefix, *arguments]) + return code, json.loads(output.getvalue()) + + code, policy = invoke(["automation-cadence", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--min-interval-minutes", "1440", "--expected-revision", "0", "--owner-reference", "fixture-owner", "--execute"]) + assert code == 0, policy + make = turn_cadence.managed_cadence_start + guard = None + guard_reply = None + ready = tmp_path / "guard-ready" + + def with_competing_guard(**kwargs): + cadence = make(**kwargs) + + def admit(identity): + nonlocal guard, guard_reply + admission = cadence.admit(identity) + if guard is None and admission.get("admitted") is True: + arguments = [*prefix, "quota", "should-run", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--todo-id", "todo_fixture0001", "--turn-instance-id", "competing-start", "--context-delivery", "off"] + script = (f"import json,sys;from pathlib import Path;from {guard_entry} import main;" + "Path(sys.argv[1]).write_text('ready');raise SystemExit(main(json.loads(sys.argv[2])))") + guard = subprocess.Popen([sys.executable, "-c", script, str(ready), json.dumps(arguments)], + cwd=project, env=os.environ.copy(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + deadline = time.monotonic() + 10 + while not ready.exists() and time.monotonic() < deadline: + time.sleep(0.01) + assert ready.exists(), "the competing CLI did not reach its entrypoint" + try: + # The old unlocked ordering lets A commit here. The binding + # lock holds A until B has reserved its matching identity. + guard_reply = guard.communicate(timeout=3) + except subprocess.TimeoutExpired: + pass + return admission + + return turn_cadence.ManagedCadenceStart(admit, cadence.confirm) + + monkeypatch.setattr(turn_cadence, "managed_cadence_start", with_competing_guard) + command = ["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", "--execute", "--turn-instance-id", "competing-start", + "--validation-command-json", json.dumps([sys.executable, "-c", "import sys;sys.stdin.read()"])] + try: + code_b, result_b = invoke([*command, "--todo-id", "todo_fixture0002"]) + assert guard is not None + if guard_reply is None: + guard_reply = guard.communicate(timeout=20) + if code_b == 0: + assert result_b["status"] == "committed", result_b + # Guard readback is not execution authority. A conflicting start + # must still refuse, including a guard's existing-receipt replay. + code_a, result_a = invoke([*command, "--todo-id", "todo_fixture0001"]) + assert code_a == 1, result_a + unclaimed_todo = "todo_fixture0001" + else: + assert guard.returncode == 0, guard_reply + code_a, result_a = invoke([*command, "--todo-id", "todo_fixture0001"]) + assert code_a == 0 and result_a["status"] == "committed", { + "loser": result_b, "winner_status": result_a.get("status"), "winner_admission": result_a.get("admission")} + unclaimed_todo = "todo_fixture0002" + _, unclaimed = invoke(["task-lease", "inspect", "--goal-id", "loopx-turn-fixture", "--todo-id", unclaimed_todo]) + assert unclaimed["active"] is False and unclaimed["lease"] is None + assert len(log.read_text().splitlines()) == 1 + finally: + if guard is not None and guard.poll() is None: + guard.kill() + guard.communicate(timeout=5) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_settled_turn_does_not_reclaim_a_released_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") + binary, log = _fake_codex(tmp_path) + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("FAKE_CODEX_LOG", str(log)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + + def invoke(arguments): + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main([*prefix, *arguments]) + return code, json.loads(output.getvalue()) + + command = ["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", "--execute", "--turn-instance-id", "settled-preparation", + "--validation-command-json", json.dumps([sys.executable, "-c", "import sys;sys.stdin.read()"])] + code, first = invoke(command) + assert code == 0 and first["status"] == "committed", first + inspection = ["task-lease", "inspect", "--goal-id", "loopx-turn-fixture", "--todo-id", "todo_fixture0001"] + _, lease = invoke(inspection) + code, released = invoke(["task-lease", "release", "--goal-id", "loopx-turn-fixture", "--todo-id", "todo_fixture0001", + "--owner", "codex-fixture", "--idempotency-key", lease["lease"]["idempotency_key"], + "--expected-version", str(lease["lease"]["version"])]) + assert code == 0, released + _, before = invoke(inspection) + code, refused = invoke(command) + assert code == 1 and "--resume-turn-instance-id" in refused["error"], refused + _, after = invoke(inspection) + assert after["active"] is False and after["lease"] == before["lease"] + assert len(log.read_text().splitlines()) == 1 + resume = [item for index, item in enumerate(command) + if item != "--turn-instance-id" and (index == 0 or command[index - 1] != "--turn-instance-id")] + code, replay = invoke([*resume, "--resume-turn-key", first["resume_turn_key"]]) + assert code == 0 and replay["replayed"] is True, replay + assert all(value is False for value in replay["effects"].values()) + assert len(log.read_text().splitlines()) == 1 + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_claim_effect_remains_visible_when_codex_is_unavailable(tmp_path, monkeypatch, provider): + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture(tmp_path, provider=provider, handoff_mode="hard_lease") + home = tmp_path / "codex-home" + home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1])) + prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main([*prefix, "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", + "--host", "codex-cli", "--project", str(project), "--codex-bin", str(tmp_path / "missing-codex"), + "--codex-model", "fixture-model", "--no-global-sync", "--execute", "--turn-instance-id", "missing-host", + "--validation-command-json", json.dumps([sys.executable, "-c", "import sys;sys.stdin.read()"])]) + result = json.loads(output.getvalue()) + assert code == 1 and result["effects"]["state_written"] is True, result + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = cli_main([*prefix, "task-lease", "inspect", "--goal-id", "loopx-turn-fixture", "--todo-id", "todo_fixture0001"]) + assert code == 0 and json.loads(output.getvalue())["active"] is True + assert result["effects"]["quota_spent"] is False