diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py
index 71dd9798fb..4aeed6017f 100644
--- a/loopx/cli_commands/turn.py
+++ b/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")
diff --git a/loopx/cli_commands/turn_cadence.py b/loopx/cli_commands/turn_cadence.py
index f5f965e1e7..987e61947f 100644
--- a/loopx/cli_commands/turn_cadence.py
+++ b/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,
+    )
diff --git a/loopx/cli_commands/turn_host_request.py b/loopx/cli_commands/turn_host_request.py
index 6c83d483dc..e5edd2aa23 100644
--- a/loopx/cli_commands/turn_host_request.py
+++ b/loopx/cli_commands/turn_host_request.py
@@ -14,27 +14,11 @@
 from ..control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission
 
 
-def prepare_turn_host_request(
-    args: argparse.Namespace, payload: Mapping[str, Any], *,
-    native_host_runner: Callable[..., dict[str, Any]] | None,
-    native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None,
-    native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None,
-    validator_builder: Callable[..., Callable[..., Mapping[str, Any]]],
-) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None, dict[str, Any]]:
-    project = Path(args.project).expanduser().resolve()
-    planned_host = (
-        cast(dict[str, Any], payload.get("host")) if isinstance(payload.get("host"), dict) else {}
-    )
-    if planned_host.get("kind") != args.host:
-        raise ValueError("--host must match the journaled LoopX Turn plan")
-    if args.host == "codex-app-server" and (
-        native_host_runner is None or native_session_binding_resolver is None
-    ):
-        raise ValueError("app-server execution requires its original in-process Chat connection")
-    if args.host != "codex-app-server" and (
-        native_host_runner is not None or native_session_binding_resolver is not None
-    ):
-        raise ValueError("a native Chat connection cannot replace a selected CLI host")
+def validate_turn_host_inputs(
+    args: argparse.Namespace,
+) -> tuple[list[str] | None, list[str] | None]:
+    """Validate transport input without reading a Turn or writing state."""
+
     if args.host == "generic-cli":
         if not args.host_command_json:
             raise ValueError(
@@ -54,9 +38,7 @@ def prepare_turn_host_request(
         if getattr(args, "host_adapter_request_schema", None):
             raise ValueError("--host-adapter-request-schema is only for generic-cli")
         raw_argv = None
-    if native_task_validator is not None and args.host != "codex-app-server":
-        raise ValueError("a native task validator cannot replace a selected CLI validator")
-    task_validator = native_task_validator
+    raw_validation_argv = None
     if args.validation_command_json:
         raw_validation_argv = json.loads(args.validation_command_json)
         if not isinstance(raw_validation_argv, list) or not all(
@@ -65,6 +47,35 @@ def prepare_turn_host_request(
             raise ValueError(
                 "--validation-command-json must be a JSON string array"
             )
+    return raw_argv, raw_validation_argv
+
+
+def prepare_turn_host_request(
+    args: argparse.Namespace, payload: Mapping[str, Any], *,
+    native_host_runner: Callable[..., dict[str, Any]] | None,
+    native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None,
+    native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None,
+    validator_builder: Callable[..., Callable[..., Mapping[str, Any]]],
+) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None, dict[str, Any]]:
+    project = Path(args.project).expanduser().resolve()
+    planned_host = (
+        cast(dict[str, Any], payload.get("host")) if isinstance(payload.get("host"), dict) else {}
+    )
+    if planned_host.get("kind") != args.host:
+        raise ValueError("--host must match the journaled LoopX Turn plan")
+    if args.host == "codex-app-server" and (
+        native_host_runner is None or native_session_binding_resolver is None
+    ):
+        raise ValueError("app-server execution requires its original in-process Chat connection")
+    if args.host != "codex-app-server" and (
+        native_host_runner is not None or native_session_binding_resolver is not None
+    ):
+        raise ValueError("a native Chat connection cannot replace a selected CLI host")
+    raw_argv, raw_validation_argv = validate_turn_host_inputs(args)
+    if native_task_validator is not None and args.host != "codex-app-server":
+        raise ValueError("a native task validator cannot replace a selected CLI validator")
+    task_validator = native_task_validator
+    if raw_validation_argv is not None:
         if task_validator is None:
             task_validator = validator_builder(
                 raw_validation_argv,
diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py
index 9df16f9641..cd29ba79c2 100644
--- a/loopx/cli_commands/turn_run_once.py
+++ b/loopx/cli_commands/turn_run_once.py
@@ -92,6 +92,7 @@ def execute_turn_run_once(
     native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None,
     native_completion_lease_resolver: Callable[[], Mapping[str, Any]] | None = None,
     native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None = None,
+    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,
         )
diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py
index 7c109f832e..998a82578c 100644
--- a/loopx/control_plane/turn_driver/codex_execution_preparation.py
+++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py
@@ -13,7 +13,8 @@
 from pathlib import Path
 from typing import Any
 
-from ...command_invocation import current_loopx_cli_argv
+from ...command_invocation import command_argv, current_loopx_cli_argv
+from ...turn_identity import normalize_turn_instance_id
 from ..content_digest import BARE_SHA256_PATTERN
 
 
@@ -100,11 +101,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 +128,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 +168,177 @@ def prepared_capture_reference(payload: Mapping[str, Any], digest: str) -> dict[
     }
 
 
+class ExecutionPreparationClaimFailed(ValueError):
+    """The decision's advertised managed claim was refused before delivery."""
+
+    code = "turn_execution_preparation_claim_failed"
+
+
+def _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,
+    admit_start: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None,
+    runner: CommandRunner | None = None,
+    timeout: float = 30,
+) -> str | None:
+    """Execute the exact ordinary claim a decision advertised for a Host.
+
+    The decision stays the owner of readiness: this adapter runs only the
+    claim the decision already published as Host-executable, then the caller
+    re-reads the same decision and must observe the same Todo ready.  A
+    preparation this adapter does not recognise is left untouched so its
+    original typed wait keeps its meaning.
+    """
+
+    preparation = decision.get("execution_preparation")
+    selected = decision.get("selected_todo")
+    if not isinstance(preparation, Mapping) or not isinstance(selected, Mapping):
+        return None
+    if preparation.get("state") != "claim_required":
+        return None
+    claim = preparation.get("claim")
+    if not isinstance(claim, Mapping) or claim.get("host_execution_allowed") is not True:
+        return None
+    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 admit_start is not None:
+        # Reserve the owner interval before this Turn writes its receipt or
+        # claim. A refusal therefore leaves no lease, receipt or journal
+        # behind; an accepted reservation is carried to the driver unchanged.
+        admission = admit_start(
+            {
+                "turn_key": exact_turn_instance_id
+                or f"{goal_id}:{caller_agent_id}:{todo_id}",
+                "attempt": 1,
+            }
+        )
+        if admission.get("admitted") is not True:
+            return None
+    if exact_turn_instance_id is not None:
+        # The receipt lookup and the ordinary claim are separate operations.
+        # Reserve the exact Turn binding before the claim can create a lease for
+        # 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")
+        # The receipt owner serializes this reservation with every later guard
+        # binding.  Reserving the Turn before the ordinary claim closes the
+        # check-to-claim race: a conflicting Todo either wins here or is refused
+        # here, so the losing caller cannot acquire a lease first.
+        ensure_turn_heartbeat_settlement_receipt(
+            runtime_root,
+            settlement_identity,
+            semantic_replan_guard_scoped=False,
+            semantic_replan_obligation_id=None,
+            goal_ref=goal_ref,
+        )
+    try:
+        _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
+    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/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index 3ca076c661..2522a2b9cb 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -1539,6 +1539,7 @@ def run_loopx_turn_once(
     post_settlement: PostSettlement | None = None,
     settlement_admission: Callable[[], AbstractContextManager] | None = None,
     admit_start: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None,
+    admission: Mapping[str, Any] | None = None,
     confirm_start: Callable[[], None] | None = None,
     goal_admission: FirstPartyHostGoalAdmission | None = None,
 ) -> dict[str, Any]:
@@ -1701,9 +1702,11 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                 build_loopx_turn_host_request(capture.plan) if capture is not None else request,
                 operation="host_dispatch_context",
             )
-        admission = None
-        if needs_host and admit_start is not None:
-            admission = admit_start(
+        # A caller that already admitted the start ahead of the ordinary claim
+        # hands that exact reservation down, so no second interval is reserved.
+        admitted = dict(admission) if needs_host and admission is not None else None
+        if needs_host and admitted is None and admit_start is not None:
+            admitted = admit_start(
                 {
                     "turn_key": turn_key,
                     "attempt": int(journal.get("host_attempt_count") or 0) + 1
@@ -1711,6 +1714,8 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                     else 1,
                 }
             )
+        admission = admitted
+        if needs_host and admission is not None:
             if admission.get("admitted") is not True:
                 waiting = {
                     "status": "interval_wait",
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index a394b4748e..ed4aca5c33 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -26,6 +26,42 @@
 AGENT = "agent-a"
 
 
+def test_default_runner_keeps_a_valid_receipt_when_stderr_is_undecodable(
+    tmp_path: Path,
+) -> None:
+    from loopx.control_plane.turn_driver import codex_execution_preparation
+
+    completed = codex_execution_preparation._default_runner(
+        [
+            sys.executable,
+            "-c",
+            "import sys; sys.stdout.write('{\"ok\": true}'); "
+            "sys.stderr.buffer.write(b'\\xff\\n')",
+        ],
+        cwd=tmp_path,
+        timeout=30,
+    )
+
+    assert completed.returncode == 0
+    assert json.loads(completed.stdout) == {"ok": True}
+    assert "\ufffd" in completed.stderr
+
+
+def test_default_runner_keeps_receipt_stdout_strict(tmp_path: Path) -> None:
+    from loopx.control_plane.turn_driver import codex_execution_preparation
+
+    with pytest.raises(UnicodeDecodeError):
+        codex_execution_preparation._default_runner(
+            [
+                sys.executable,
+                "-c",
+                "import sys; sys.stdout.buffer.write(b'\\xff')",
+            ],
+            cwd=tmp_path,
+            timeout=30,
+        )
+
+
 @pytest.mark.parametrize("status", ["open", "deferred"])
 def test_explicit_non_delivery_obligation_does_not_prepare_selected_execution(monkeypatch, status):
     from loopx.control_plane.quota import execution_preparation
diff --git a/tests/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..5c976b0c1e
--- /dev/null
+++ b/tests/control_plane/test_turn_run_once_claim_preparation.py
@@ -0,0 +1,294 @@
+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,
+    ]
diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py
index b748707010..1c614478bb 100644
--- a/tests/test_loopx_turn_driver.py
+++ b/tests/test_loopx_turn_driver.py
@@ -46,6 +46,7 @@
 )
 from loopx.control_plane.turn_driver.executor import BuiltInHostError
 from loopx.control_plane.quota.live_decision import bind_scheduler_followup_cli_routes
+from loopx.rollout_event_log import rollout_event_log_path
 
 
 def _envelope(
@@ -1126,6 +1127,553 @@ 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.",
+            "  <!-- loopx:todo todo_id=todo_fixture0002 status=open "
+            "task_class=advancement_task action_kind=fixture claimed_by=codex-fixture "
+            "priority=P1 -->",
+        ),
+    )
+    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.",
+            "  <!-- loopx:todo todo_id=todo_fixture0002 status=open "
+            "task_class=advancement_task action_kind=fixture claimed_by=codex-fixture "
+            "priority=P1 -->",
+        ),
+    )
+    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"])
+@pytest.mark.parametrize(
+    ("validation_json", "expected_error"),
+    [
+        ("not-json", "Expecting value"),
+        ('{"not": "an-array"}', "must be a JSON string array"),
+    ],
+    ids=["malformed", "wrong-shape"],
+)
+def test_run_once_rejects_invalid_validator_json_before_claiming(
+    tmp_path, monkeypatch, provider, validation_json, expected_error,
+):
+    from tests.test_loopx_turn_codex_cli import _fake_codex
+
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_live_fixture(
+        tmp_path, provider=provider, handoff_mode="hard_lease",
+    )
+    binary, log = _fake_codex(tmp_path)
+    monkeypatch.setenv("FAKE_CODEX_LOG", str(log))
+    home = tmp_path / "codex-home"
+    home.mkdir()
+    monkeypatch.setenv("CODEX_HOME", str(home))
+    monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1]))
+    rollout_log = rollout_event_log_path(runtime, "loopx-turn-fixture")
+    before_log = rollout_log.read_bytes() if rollout_log.exists() else b""
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        exit_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture",
+            "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary),
+            "--codex-model", "fixture-model", "--codex-reasoning-effort", "high",
+            "--codex-sandbox", "read-only", "--no-global-sync",
+            "--validation-command-json", validation_json,
+            "--execute",
+        ])
+    refused = json.loads(output.getvalue())
+    assert exit_code == 1, refused
+    assert expected_error in refused["error"]
+    assert not log.exists()
+    assert (rollout_log.read_bytes() if rollout_log.exists() else b"") == before_log
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        lease_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "task-lease", "inspect", "--goal-id", "loopx-turn-fixture",
+            "--todo-id", "todo_fixture0001",
+        ])
+    lease = read_cli_policy_payload(output.getvalue(), returncode=lease_code)
+    assert lease_code == 0, lease
+    assert lease["active"] is False, lease
+    assert lease["lease"] is None, lease
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_run_once_interval_wait_does_not_claim_the_next_todo(
+    tmp_path, monkeypatch, provider,
+):
+    from tests.test_loopx_turn_codex_cli import _fake_codex
+
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_live_fixture(
+        tmp_path,
+        provider=provider,
+        handoff_mode="hard_lease",
+        extra_agent_todo_lines=(
+            "- [ ] [P1] Complete the interval successor fixture.",
+            "  <!-- loopx:todo todo_id=todo_fixture0002 status=open "
+            "task_class=advancement_task action_kind=fixture claimed_by=codex-fixture "
+            "priority=P1 -->",
+        ),
+    )
+    binary, log = _fake_codex(tmp_path)
+    monkeypatch.setenv("FAKE_CODEX_LOG", str(log))
+    home = tmp_path / "codex-home"
+    home.mkdir()
+    monkeypatch.setenv("CODEX_HOME", str(home))
+    monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1]))
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        read_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "automation-cadence", "--goal-id", "loopx-turn-fixture",
+            "--agent-id", "codex-fixture",
+        ])
+    policy = read_cli_policy_payload(output.getvalue(), returncode=read_code)
+    assert read_code == 0, policy
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        configure_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "automation-cadence", "--goal-id", "loopx-turn-fixture",
+            "--agent-id", "codex-fixture", "--min-interval-minutes", "3",
+            "--expected-revision", str(policy["configuration_revision"]),
+            "--owner-reference", "interval fixture", "--execute",
+        ])
+    configured = read_cli_policy_payload(output.getvalue(), returncode=configure_code)
+    assert configure_code == 0 and configured["written"] is True, configured
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        first_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture",
+            "--turn-instance-id", "interval-first", "--host", "codex-cli",
+            "--project", str(project), "--codex-bin", str(binary),
+            "--codex-model", "fixture-model", "--codex-reasoning-effort", "high",
+            "--codex-sandbox", "read-only", "--no-global-sync",
+            "--validation-command-json",
+            json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]),
+            "--execute",
+        ])
+    first = json.loads(output.getvalue())
+    assert first_code == 0 and first["status"] == "committed", first
+    calls = [json.loads(line) for line in log.read_text().splitlines()]
+    assert len(calls) == 1, calls
+    rollout_log = rollout_event_log_path(runtime, "loopx-turn-fixture")
+    before_log = rollout_log.read_bytes() if rollout_log.exists() else b""
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        waiting_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture",
+            "--todo-id", "todo_fixture0002", "--turn-instance-id", "interval-second",
+            "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary),
+            "--codex-model", "fixture-model", "--codex-reasoning-effort", "high",
+            "--codex-sandbox", "read-only", "--no-global-sync",
+            "--validation-command-json",
+            json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]),
+            "--execute",
+        ])
+    waiting = json.loads(output.getvalue())
+    print("DEBUGWAIT", json.dumps(waiting, ensure_ascii=False)[:2000])
+    assert waiting_code == 1, waiting
+    assert waiting["status"] == "interval_wait", waiting
+    assert waiting["effects"]["host_invoked"] is False, waiting
+    assert waiting["effects"]["state_written"] is False, waiting
+    calls = [json.loads(line) for line in log.read_text().splitlines()]
+    assert len(calls) == 1, calls
+    assert (rollout_log.read_bytes() if rollout_log.exists() else b"") == before_log
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        lease_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "task-lease", "inspect", "--goal-id", "loopx-turn-fixture",
+            "--todo-id", "todo_fixture0002",
+        ])
+    lease = read_cli_policy_payload(output.getvalue(), returncode=lease_code)
+    assert lease_code == 0, lease
+    assert lease["active"] is False, lease
+    assert lease["lease"] is None, lease
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_run_once_manual_interval_bypass_starts_the_next_todo(
+    tmp_path, monkeypatch, provider,
+):
+    from tests.test_loopx_turn_codex_cli import _fake_codex
+
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_live_fixture(
+        tmp_path,
+        provider=provider,
+        handoff_mode="hard_lease",
+        extra_agent_todo_lines=(
+            "- [ ] [P1] Complete the manual bypass fixture.",
+            "  <!-- loopx:todo todo_id=todo_fixture0002 status=open "
+            "task_class=advancement_task action_kind=fixture claimed_by=codex-fixture "
+            "priority=P1 -->",
+        ),
+    )
+    binary, log = _fake_codex(tmp_path)
+    monkeypatch.setenv("FAKE_CODEX_LOG", str(log))
+    home = tmp_path / "codex-home"
+    home.mkdir()
+    monkeypatch.setenv("CODEX_HOME", str(home))
+    monkeypatch.setenv("LOOPX_RELEASE_ROOT", str(Path(__file__).resolve().parents[1]))
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        read_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "automation-cadence", "--goal-id", "loopx-turn-fixture",
+            "--agent-id", "codex-fixture",
+        ])
+    policy = read_cli_policy_payload(output.getvalue(), returncode=read_code)
+    assert read_code == 0, policy
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        configure_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "automation-cadence", "--goal-id", "loopx-turn-fixture",
+            "--agent-id", "codex-fixture", "--min-interval-minutes", "3",
+            "--expected-revision", str(policy["configuration_revision"]),
+            "--owner-reference", "manual bypass fixture", "--execute",
+        ])
+    configured = read_cli_policy_payload(output.getvalue(), returncode=configure_code)
+    assert configure_code == 0 and configured["written"] is True, configured
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        first_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture",
+            "--turn-instance-id", "manual-first", "--host", "codex-cli",
+            "--project", str(project), "--codex-bin", str(binary),
+            "--codex-model", "fixture-model", "--codex-reasoning-effort", "high",
+            "--codex-sandbox", "read-only", "--no-global-sync",
+            "--validation-command-json",
+            json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]),
+            "--execute",
+        ])
+    first = json.loads(output.getvalue())
+    assert first_code == 0 and first["status"] == "committed", first
+    assert len([json.loads(line) for line in log.read_text().splitlines()]) == 1
+
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        manual_code = cli_main([
+            "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+            "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture",
+            "--todo-id", "todo_fixture0002", "--turn-instance-id", "manual-second",
+            "--manual-interval-bypass-reason", "explicit operator confirmation",
+            "--host", "codex-cli", "--project", str(project), "--codex-bin", str(binary),
+            "--codex-model", "fixture-model", "--codex-reasoning-effort", "high",
+            "--codex-sandbox", "read-only", "--no-global-sync",
+            "--validation-command-json",
+            json.dumps([sys.executable, "-c", "import sys; sys.stdin.read()"]),
+            "--execute",
+        ])
+    manual = json.loads(output.getvalue())
+    assert manual_code == 0, manual
+    assert manual["status"] == "committed", manual
+    assert manual["effects"]["host_invoked"] is True, manual
+    calls = [json.loads(line) for line in log.read_text().splitlines()]
+    assert len(calls) == 2, calls
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_run_once_reserves_turn_before_claim_under_competing_guard(
+    tmp_path, monkeypatch, provider,
+):
+    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.",
+            "  <!-- loopx:todo todo_id=todo_fixture0002 status=open "
+            "task_class=advancement_task action_kind=fixture claimed_by=codex-fixture "
+            "priority=P1 -->",
+        ),
+    )
+    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 +2042,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 +2123,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(
@@ -2508,11 +3057,14 @@ def checkpoint(path, journal):
 
     def fresh_cadence(**kwargs):
         cadence = make_cadence(**kwargs)
+        latest: dict[str, Any] = {}
 
         def admit(identity):
             attempts.append(dict(identity))
             admission = cadence.admit(identity)
             admissions.append(dict(admission))
+            latest.clear()
+            latest.update(admission)
             return admission
 
         def confirm():
@@ -2525,7 +3077,15 @@ def confirm():
             if not interrupted and fault == "confirm_after":
                 interrupt()
 
-        return turn_cadence.ManagedCadenceStart(admit, confirm)
+        def accepted():
+            return dict(latest) if latest else None
+
+        def set_on_admitted(callback):
+            cadence.set_on_admitted(callback)
+
+        return turn_cadence.ManagedCadenceStart(
+            admit, confirm, accepted, set_on_admitted,
+        )
 
     monkeypatch.setattr(executor, "_write_journal", checkpoint)
     monkeypatch.setattr(turn_run_once_command, "managed_cadence_start", fresh_cadence)