From b89150097c1cd664c486d46e0de744fb3c370fed Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Fri, 13 Nov 2026 11:51:23 +0800 Subject: [PATCH] fix(turn): admit managed Codex preparation before claiming work Preserve the existing typed ordinary claim and exact Turn binding while moving validator transport checks and start admission before preparation writes. Reuse the settlement identity for unstarted cadence recovery, retain historical transaction-key reservations, and qualify source and standalone-core rejection/recovery paths. Co-authored-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Signed-off-by: lusendong.6789 Signed-off-by: lusendong.6789 Co-authored-by: TRAE CLI --- .../rfcs/monorepo-distribution-split-v0.md | 40 ++ .../monorepo-distribution-split-v0.zh-CN.md | 30 + .../loopx-turn-codex-cli-quickstart.md | 10 + loopx/cli_commands/turn.py | 85 ++- loopx/cli_commands/turn_cadence.py | 45 +- loopx/cli_commands/turn_host_request.py | 32 +- loopx/cli_commands/turn_rendering.py | 3 + loopx/cli_commands/turn_run_once.py | 1 + .../control_plane/quota/heartbeat_receipt.py | 18 +- .../codex_execution_preparation.py | 232 ++++++- .../test_quota_execution_preparation.py | 36 + .../test_turn_run_once_claim_preparation.py | 318 +++++++++ tests/test_core_distribution.py | 43 ++ tests/test_loopx_turn_driver.py | 617 +++++++++++++++++- 14 files changed, 1464 insertions(+), 46 deletions(-) create mode 100644 tests/control_plane/test_turn_run_once_claim_preparation.py 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