Admit managed Codex preparation before claiming work

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.

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 默认、包发布或付费操作。

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

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"

loopx/cli_commands/turn_cadence.py

@@ -29,31 +29,44 @@ 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."""
 
     admitted_request: dict[str, Any] = {}
 
     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()

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)