fix(turn): prepare managed Todo claims before builtin Codex delivery

loopx/cli_commands/turn.py

@@ -89,6 +89,46 @@ def handle_turn_command(
             registry_path=registry_path,
             runtime_root_override=runtime_root_arg,
         )
+        managed_cadence = None
+        if args.turn_command == "run-once" and bool(getattr(args, "execute", False)):
+            # Resolve the start owner through the run-once adapter so one
+            # process-wide seam binds admission, confirmation and receipts.
+            from . import turn_run_once as turn_run_once_module
+
+            managed_cadence = turn_run_once_module.managed_cadence_start(
+                runtime_root=runtime_root.resolve(),
+                goal_id=args.goal_id,
+                agent_id=args.agent_id,
+                automation_id=args.automation_id,
+                manual_reason=getattr(args, "manual_interval_bypass_reason", None),
+            )
+        execution_preparer = native_execution_preparer
+        preparation_error = "native task preparation did not make the same Todo ready"
+        if (
+            execution_preparer is None
+            and args.turn_command == "run-once"
+            and args.host == "codex-cli"
+            and bool(getattr(args, "execute", False))
+        ):
+            from ..control_plane.turn_driver.codex_execution_preparation import (
+                builtin_execution_preparation_claim,
+            )
+
+            project_cwd = Path(args.project).expanduser().resolve()
+            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=managed_cadence.admit if managed_cadence else None,
+                )
+
         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 +311,14 @@ 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:
+                from .turn_host_request import validate_turn_host_inputs
+
+                # Transport refusals land before the Turn receipt reservation.
+                # The owner interval is admitted atomically inside the same
+                # preparation boundary, ahead of every receipt and claim write.
+                validate_turn_host_inputs(args)
+                prepared_todo = execution_preparer(decision)
                 if prepared_todo is not None:
                     # Host preparation changes canonical lease state. Plan and
                     # settle against a fresh read of the same Turn and Todo.
@@ -284,7 +330,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"}
@@ -405,6 +451,7 @@ def read_decision_owner():
                 native_session_binding_resolver=native_session_binding_resolver,
                 native_completion_lease_resolver=native_completion_lease_resolver,
                 native_task_validator=native_task_validator,
+                managed_cadence=managed_cadence,
             )
         else:
             raise ValueError("turn requires the `plan` or `run-once` subcommand")

loopx/cli_commands/turn_cadence.py

@@ -3,22 +3,28 @@
 import time
 from collections.abc import Callable, Mapping
 from pathlib import Path
-from typing import Any, NamedTuple
+from dataclasses import dataclass
+from typing import Any
 
 from ..control_plane.effect_runtime import effect_runtime_result
 
 
-class ManagedCadenceStart(NamedTuple):
+@dataclass(slots=True)
+class ManagedCadenceStart:
     """Owner-cadence callbacks for one managed Turn start.
 
     `admit` reserves (or resumes) the interval slot before any host or journal
     attempt; `confirm` marks that reservation as a real host attempt once the
     Turn journal is durable. Keeping them separate means a crash in between
     leaves a resumable reservation rather than a permanently rejected Turn.
+    `accepted` returns the most recent admission so a caller that admits ahead
+    of the ordinary claim can hand the same reservation to the driver.
     """
 
     admit: Callable[[Mapping[str, Any]], dict[str, Any]]
     confirm: Callable[[], None]
+    accepted: Callable[[], dict[str, Any] | None]
+    set_on_admitted: Callable[[Callable[[], None] | None], None]
 
 
 def managed_cadence_start(
@@ -33,37 +39,70 @@ def managed_cadence_start(
     """Bind one managed Turn start to the TypeScript owner-cadence store."""
 
     admitted_request: dict[str, Any] = {}
+    latest_admission: dict[str, Any] | None = None
+    admission_hook: list[Callable[[], None] | None] = [on_admitted]
+    hook_pending = [False]
+
+    def set_on_admitted(callback: Callable[[], None] | None) -> None:
+        admission_hook[0] = callback
+        if hook_pending[0] and callback is not None:
+            # A reservation admitted ahead of the surrounding Turn preparation
+            # delivers its receipt writer as soon as the Turn adopts it.
+            hook_pending[0] = False
+            callback()
 
     def admit(identity: Mapping[str, Any]) -> dict[str, Any]:
+        nonlocal latest_admission
         now_ms = time.time_ns() // 1_000_000
-        request_id = f"{identity['turn_key']}:{identity['attempt']}"
-        admission = effect_runtime_result(
-            "quota.automation_cadence.admit",
-            {
-                "runtime_root": str(runtime_root),
-                "goal_id": goal_id,
-                "agent_id": agent_id,
-                "automation_id": automation_id,
-                "request_id": request_id,
-                "trigger_at_ms": now_ms,
-                "now_ms": now_ms,
-                "manual_reason": manual_reason,
-            },
-            retry_safe=False,
-        )
+        base_request_id = f"{identity['turn_key']}:{identity['attempt']}"
+
+        def request_admission(request_id: str) -> dict[str, Any]:
+            return effect_runtime_result(
+                "quota.automation_cadence.admit",
+                {
+                    "runtime_root": str(runtime_root),
+                    "goal_id": goal_id,
+                    "agent_id": agent_id,
+                    "automation_id": automation_id,
+                    "request_id": request_id,
+                    "trigger_at_ms": now_ms,
+                    "now_ms": now_ms,
+                    "manual_reason": manual_reason,
+                },
+                retry_safe=False,
+            )
+
+        request_id = base_request_id
+        admission = request_admission(request_id)
+        if admission.get("admitted") is not True and (
+            admission.get("reason") == "duplicate_or_stale_trigger"
+        ):
+            # A durable earlier attempt keeps its own request identity
+            # fail-closed.  Once the owner interval has elapsed, this Turn may
+            # start again under a fresh attempt identity, exactly as the
+            # attempt counter did before admission moved ahead of the claim.
+            request_id = f"{base_request_id}:{now_ms}"
+            admission = request_admission(request_id)
         admitted_request.clear()
         if admission.get("admitted") is True:
             admitted_request["request_id"] = request_id
             admitted_request["reserved"] = admission.get("reserved") is True
-            if on_admitted is not None:
-                on_admitted()
-        return {
+            hook = admission_hook[0]
+            if hook is not None:
+                hook()
+            else:
+                hook_pending[0] = True
+        latest_admission = {
             key: admission.get(key)
             for key in (
                 "admitted", "reserved", "resumed", "reason", "next_eligible_at_ms",
                 "min_interval_minutes", "pre_model_admission",
             )
         }
+        return dict(latest_admission)
+
+    def accepted() -> dict[str, Any] | None:
+        return dict(latest_admission) if latest_admission is not None else None
 
     def confirm() -> None:
         request_id = admitted_request.get("request_id")
@@ -85,4 +124,7 @@ def confirm() -> None:
                 "managed Turn start could not be confirmed against the owner cadence store"
             )
 
-    return ManagedCadenceStart(admit=admit, confirm=confirm)
+    return ManagedCadenceStart(
+        admit=admit, confirm=confirm, accepted=accepted,
+        set_on_admitted=set_on_admitted,
+    )

loopx/cli_commands/turn_host_request.py

@@ -14,27 +14,11 @@
 from ..control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission
 
 
-def prepare_turn_host_request(
-    args: argparse.Namespace, payload: Mapping[str, Any], *,
-    native_host_runner: Callable[..., dict[str, Any]] | None,
-    native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None,
-    native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None,
-    validator_builder: Callable[..., Callable[..., Mapping[str, Any]]],
-) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None, dict[str, Any]]:
-    project = Path(args.project).expanduser().resolve()
-    planned_host = (
-        cast(dict[str, Any], payload.get("host")) if isinstance(payload.get("host"), dict) else {}
-    )
-    if planned_host.get("kind") != args.host:
-        raise ValueError("--host must match the journaled LoopX Turn plan")
-    if args.host == "codex-app-server" and (
-        native_host_runner is None or native_session_binding_resolver is None
-    ):
-        raise ValueError("app-server execution requires its original in-process Chat connection")
-    if args.host != "codex-app-server" and (
-        native_host_runner is not None or native_session_binding_resolver is not None
-    ):
-        raise ValueError("a native Chat connection cannot replace a selected CLI host")
+def validate_turn_host_inputs(
+    args: argparse.Namespace,
+) -> tuple[list[str] | None, list[str] | None]:
+    """Validate transport input without reading a Turn or writing state."""
+
     if args.host == "generic-cli":
         if not args.host_command_json:
             raise ValueError(
@@ -54,9 +38,7 @@ def prepare_turn_host_request(
         if getattr(args, "host_adapter_request_schema", None):
             raise ValueError("--host-adapter-request-schema is only for generic-cli")
         raw_argv = None
-    if native_task_validator is not None and args.host != "codex-app-server":
-        raise ValueError("a native task validator cannot replace a selected CLI validator")
-    task_validator = native_task_validator
+    raw_validation_argv = None
     if args.validation_command_json:
         raw_validation_argv = json.loads(args.validation_command_json)
         if not isinstance(raw_validation_argv, list) or not all(
@@ -65,6 +47,35 @@ def prepare_turn_host_request(
             raise ValueError(
                 "--validation-command-json must be a JSON string array"
             )
+    return raw_argv, raw_validation_argv
+
+
+def prepare_turn_host_request(
+    args: argparse.Namespace, payload: Mapping[str, Any], *,
+    native_host_runner: Callable[..., dict[str, Any]] | None,
+    native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None,
+    native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None,
+    validator_builder: Callable[..., Callable[..., Mapping[str, Any]]],
+) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None, dict[str, Any]]:
+    project = Path(args.project).expanduser().resolve()
+    planned_host = (
+        cast(dict[str, Any], payload.get("host")) if isinstance(payload.get("host"), dict) else {}
+    )
+    if planned_host.get("kind") != args.host:
+        raise ValueError("--host must match the journaled LoopX Turn plan")
+    if args.host == "codex-app-server" and (
+        native_host_runner is None or native_session_binding_resolver is None
+    ):
+        raise ValueError("app-server execution requires its original in-process Chat connection")
+    if args.host != "codex-app-server" and (
+        native_host_runner is not None or native_session_binding_resolver is not None
+    ):
+        raise ValueError("a native Chat connection cannot replace a selected CLI host")
+    raw_argv, raw_validation_argv = validate_turn_host_inputs(args)
+    if native_task_validator is not None and args.host != "codex-app-server":
+        raise ValueError("a native task validator cannot replace a selected CLI validator")
+    task_validator = native_task_validator
+    if raw_validation_argv is not None:
         if task_validator is None:
             task_validator = validator_builder(
                 raw_validation_argv,

loopx/cli_commands/turn_run_once.py

@@ -92,6 +92,7 @@ def execute_turn_run_once(
     native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None,
     native_completion_lease_resolver: Callable[[], Mapping[str, Any]] | None = None,
     native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None = None,
+    managed_cadence: Any | None = None,
 ) -> dict[str, Any]:
     """Adapt one prepared CLI Turn plan to the bounded Turn driver."""
     execution_started = False
@@ -108,6 +109,33 @@ def execute_turn_run_once(
             native_task_validator=native_task_validator,
             validator_builder=build_loopx_turn_command_validator,
         )
+        preadmitted = managed_cadence.accepted() if managed_cadence is not None else None
+        if preadmitted is not None and preadmitted.get("admitted") is not True:
+            # The owner refused the interval before this Turn wrote its receipt
+            # or claimed a Todo. Report that same admission without reserving,
+            # claiming or journaling anything.
+            from ..control_plane.turn_driver.execution_readback import execution_payload
+
+            planned_host = payload.get("host")
+            waiting = {
+                "status": "interval_wait",
+                "host": planned_host if isinstance(planned_host, Mapping) else {},
+                "completed_phases": [],
+                "reason": str(preadmitted.get("reason") or "automatic start not admitted"),
+                "admission": preadmitted,
+            }
+            return execution_payload(
+                payload,
+                waiting,
+                execute=True,
+                replayed=False,
+                effects={
+                    "host_invoked": False,
+                    "state_written": False,
+                    "quota_spent": False,
+                    "scheduler_acknowledged": False,
+                },
+            )
         selected_todo = execution_context["selected_todo"]
         expected_todo_text_sha256 = execution_context["sha256"]
         transaction = (
@@ -872,14 +900,19 @@ def on_managed_start_admitted() -> None:
                 goal_ref=goal_ref,
             )
 
-        managed_cadence = managed_cadence_start(
-            runtime_root=runtime_root,
-            goal_id=args.goal_id,
-            agent_id=args.agent_id,
-            automation_id=args.automation_id,
-            manual_reason=args.manual_interval_bypass_reason,
-            on_admitted=on_managed_start_admitted,
-        )
+        if managed_cadence is None:
+            managed_cadence = managed_cadence_start(
+                runtime_root=runtime_root,
+                goal_id=args.goal_id,
+                agent_id=args.agent_id,
+                automation_id=args.automation_id,
+                manual_reason=args.manual_interval_bypass_reason,
+                on_admitted=on_managed_start_admitted,
+            )
+        else:
+            # A start admitted ahead of the claim keeps this Turn's receipt
+            # writer attached to the same owner reservation.
+            managed_cadence.set_on_admitted(on_managed_start_admitted)
 
         execution_started = bool(args.execute)
         result = run_loopx_turn_once(
@@ -912,7 +945,12 @@ def on_managed_start_admitted() -> None:
             post_settlement=(configured_post_settlement(
                 payload, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id,
             ) if args.execute else None),
-            admit_start=managed_cadence.admit if args.execute else None,
+            admit_start=(
+                managed_cadence.admit
+                if args.execute and preadmitted is None
+                else None
+            ),
+            admission=preadmitted if args.execute else None,
             confirm_start=managed_cadence.confirm if args.execute else None,
             goal_admission=goal_admission,
         )