diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 5774fa005..414749992 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -53,7 +53,11 @@
     render_scheduler_execution_args,
 )
 from ..control_plane.todos.contract import normalize_todo_id
-from ..control_plane.work_items.action_selection_contract import apply_action_selection_recovery
+from ..control_plane.work_items.action_selection_contract import (
+    bind_action_selection_recovery_command,
+    build_action_selection_recovery_fields,
+    current_action_selection_admission,
+)
 from ..presentation.renderers.quota_event_markdown import (
     render_quota_monitor_poll_markdown,
     render_quota_slot_preview_markdown,
@@ -203,7 +207,7 @@ def _heartbeat_quota_action_selection_bindings(
     )
 
 
-def _apply_requested_quota_action_selection_preflight(
+def _requested_quota_action_selection_preflight(
     payload: dict[str, object],
     *,
     requested_todo_id: str | None,
@@ -211,102 +215,50 @@ def _apply_requested_quota_action_selection_preflight(
     receipt_bound_replan_obligation_id: str | None,
     receipt_pending_action_todo_id: str | None = None,
     receipt_identity_upgraded: bool = False,
-) -> bool:
+) -> dict[str, object] | None:
     if not requested_todo_id:
-        return False
+        return None
     if receipt_bound_todo_id:
         if requested_todo_id != receipt_bound_todo_id:
             raise HeartbeatReceiptIdentityConflictError(
                 "heartbeat receipt settlement identity conflicts with the "
                 "current selected Todo: explicitly requested Todo differs"
             )
-        return False
-    selected_todo = payload.get("selected_todo")
-    selected_todo_id = (
-        normalize_todo_id(selected_todo.get("todo_id"))
-        if isinstance(selected_todo, Mapping)
+        return None
+    interaction = payload.get("interaction_contract")
+    agent_channel = (
+        interaction.get("agent_channel")
+        if isinstance(interaction, Mapping)
         else None
     )
-    qualification_value = payload.get("action_selection_qualification")
-    qualification: Mapping[str, object] = (
-        qualification_value if isinstance(qualification_value, Mapping) else {}
+    agent_channel = agent_channel if isinstance(agent_channel, Mapping) else {}
+    selected_todo_id, admitted = current_action_selection_admission(
+        payload,
+        requested_todo_id=requested_todo_id,
+        agent_must_attempt=agent_channel.get("must_attempt") is True,
+        agent_delivery_refused=agent_channel.get("delivery_allowed") is False,
     )
-    if selected_todo_id is None and str(qualification.get("state") or "") == (
-        "qualified"
-    ):
-        # An unsettled-host-turn recovery decision carries no top-level
-        # `selected_todo`: its qualification names the Todo that prior Turn
-        # has to settle, and binding the guard to that Todo is the documented
-        # closeout path rather than a conflict with the projection.
-        qualification_selected = qualification.get("selected_todo")
-        selected_todo_id = (
-            normalize_todo_id(qualification_selected.get("todo_id"))
-            if isinstance(qualification_selected, Mapping)
-            else None
-        )
     if receipt_bound_replan_obligation_id:
         if not receipt_identity_upgraded:
             # A Turn that started directly in autonomous replan has no Todo
             # selection authority to replace.  A later same-Turn --todo-id is
             # therefore a harmless settled replay, not successor delivery.
-            return False
+            return None
         if requested_todo_id == receipt_pending_action_todo_id:
-            return False
+            return None
         raise QuotaActionSelectionConflictError(
             QuotaActionSelectionConflictKind.CONFLICT,
             requested_todo_id=requested_todo_id,
             selected_todo_id=receipt_pending_action_todo_id,
             qualification_state="retained_selection",
         )
-    selection_binding = (
-        selected_todo.get("selection_binding")
-        if isinstance(selected_todo, Mapping)
-        else None
-    )
-    execution_obligation_value = payload.get("execution_obligation")
-    execution_obligation: Mapping[str, object] = (
-        execution_obligation_value
-        if isinstance(execution_obligation_value, Mapping)
-        else {}
-    )
-    interaction_value = payload.get("interaction_contract")
-    interaction: Mapping[str, object] = (
-        interaction_value if isinstance(interaction_value, Mapping) else {}
-    )
-    agent_channel_value = interaction.get("agent_channel")
-    agent_channel: Mapping[str, object] = (
-        agent_channel_value if isinstance(agent_channel_value, Mapping) else {}
-    )
-    pending_selection_delivery_qualified = (
-        selection_binding == "pending_action_selection"
-        and payload.get("normal_delivery_allowed") is True
-    )
-    pending_selection_workspace_repair_qualified = (
-        selection_binding == "pending_action_selection"
-        and payload.get("workspace_repair_allowed") is True
-        and payload.get("effective_action") == EffectiveAction.AGENT_WORKSPACE_REPAIR.value
-        and execution_obligation.get("kind") == "agent_workspace_repair"
-        and execution_obligation.get("must_attempt_work") is True
-        and agent_channel.get("must_attempt") is True
-        and agent_channel.get("delivery_allowed") is False
-    )
-    exact_current_obligation_qualified = (
-        selection_binding != "pending_action_selection"
-        and execution_obligation.get("must_attempt_work") is True
-        and agent_channel.get("must_attempt") is True
-    )
-    if (
-        selected_todo_id == requested_todo_id
-        and payload.get("ok") is True
-        and payload.get("should_run") is True
-        and (
-            pending_selection_delivery_qualified
-            or pending_selection_workspace_repair_qualified
-            or exact_current_obligation_qualified
-        )
-    ):
-        return False
+    if admitted:
+        return None
 
+    qualification_value = payload.get("action_selection_qualification")
+    qualification: Mapping[str, object] = (
+        qualification_value if isinstance(qualification_value, Mapping) else {}
+    )
     if not isinstance(qualification_value, Mapping):
         raise QuotaActionSelectionConflictError(
             QuotaActionSelectionConflictKind.UNQUALIFIED,
@@ -321,52 +273,7 @@ def _apply_requested_quota_action_selection_preflight(
             selected_todo_id=selected_todo_id,
             qualification_state=qualification_state,
         )
-    qualification_reason = str(
-        qualification.get("reason") or "candidate_not_currently_eligible"
-    )
-    deferred = qualification_state == "deferred"
-    auxiliary_monitor = (
-        qualification_reason
-        == "auxiliary_monitor_not_selectable_in_advancement_lane"
-    )
-    error_code = (
-        "quota_action_selection_deferred"
-        if deferred
-        else "quota_action_selection_rejected"
-    )
-    payload.update(
-        {
-            "ok": False,
-            "decision": "skip",
-            "should_run": False,
-            "effective_action": EffectiveAction.QUOTA_SKIP.value,
-            "state": error_code,
-            "waiting_on": "codex",
-            "status": error_code,
-            "error_code": error_code,
-            "reason": (
-                "explicit action selection was deferred by the current "
-                f"delivery frontier: {qualification_reason}"
-                if deferred
-                else "explicit action selection is not currently eligible: "
-                f"{qualification_reason}"
-            ),
-            "recommended_action": (
-                "handle the current delivery preemption, then rerun quota "
-                "should-run with the same --turn-instance-id; omit --todo-id "
-                "first when a refreshed action portfolio is needed"
-                if deferred
-                else "the due monitor is visible as auxiliary context, not an "
-                "independently selectable action in the current advancement lane; "
-                "choose a current advancement Todo, or rerun after the monitor "
-                "becomes the hard lane"
-                if auxiliary_monitor
-                else "rerun quota should-run with the same --turn-instance-id "
-                "without --todo-id, then choose a currently eligible Todo"
-            ),
-        }
-    )
-    return True
+    return build_action_selection_recovery_fields(payload)
 
 
 def _reconcile_requested_quota_action_selection(
@@ -380,27 +287,29 @@ def _reconcile_requested_quota_action_selection(
     receipt_pending_action_todo_id: str | None,
     receipt_identity_upgraded: bool,
 ) -> bool:
-    rejected = _apply_requested_quota_action_selection_preflight(
+    recovery = _requested_quota_action_selection_preflight(
         payload, requested_todo_id=_requested_quota_action_todo_id(args),
         receipt_bound_todo_id=receipt_bound_todo_id,
         receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
         receipt_pending_action_todo_id=receipt_pending_action_todo_id,
         receipt_identity_upgraded=receipt_identity_upgraded,
     )
-    if rejected:
-        apply_action_selection_recovery(
-            payload, registry_path=str(registry_path), runtime_root=str(context.runtime_root),
-            goal_id=args.goal_id, agent_id=args.agent_id,
-            turn_instance_id=context.heartbeat_turn_id,
-            available_capabilities=args.available_capabilities,
-            scheduler_args=render_scheduler_execution_args(
-                scheduler_execution_context=context.scheduler_context),
-        )
-        obligation = payload.get("execution_obligation")
-        if isinstance(obligation, dict):
-            obligation.update(must_attempt_work=False, delivery_allowed=False,
-                              reason=payload["recommended_action"])
-    return rejected
+    if recovery is None:
+        return False
+    payload.update(recovery)
+    bind_action_selection_recovery_command(
+        payload,
+        registry_path=str(registry_path),
+        runtime_root=str(context.runtime_root),
+        goal_id=args.goal_id,
+        agent_id=args.agent_id,
+        turn_instance_id=context.heartbeat_turn_id,
+        available_capabilities=args.available_capabilities,
+        scheduler_args=render_scheduler_execution_args(
+            scheduler_execution_context=context.scheduler_context
+        ),
+    )
+    return True
 
 
 def _attach_uncommitted_action_selection_receipt(
diff --git a/loopx/control_plane/quota/heartbeat_recommendation.py b/loopx/control_plane/quota/heartbeat_recommendation.py
index 219194375..50f633042 100644
--- a/loopx/control_plane/quota/heartbeat_recommendation.py
+++ b/loopx/control_plane/quota/heartbeat_recommendation.py
@@ -264,6 +264,23 @@ def _recommendation(*parts: dict[str, Any]) -> dict[str, Any]:
     return payload
 
 
+def build_action_selection_recovery_recommendation(
+    *, reason: str,
+) -> dict[str, Any]:
+    """Project closed heartbeat guidance after a typed selection refusal."""
+
+    return _recommendation(
+        {
+            "source": "action_selection_recovery",
+            "recommended_mode": "quota_skip",
+            "notify": "DONT_NOTIFY",
+            "spend_policy": "no quota spend until an eligible Todo is selected",
+            "reason": reason,
+            "agent_must_attempt": False,
+        }
+    )
+
+
 def _stall_self_repair_rule(
     facts: _HeartbeatRecommendationFacts,
 ) -> dict[str, Any] | None:
diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py
index e7788b811..1d536cc2e 100644
--- a/loopx/control_plane/quota/should_run_packet.py
+++ b/loopx/control_plane/quota/should_run_packet.py
@@ -39,6 +39,7 @@
     quota_execution_profile_summary as _quota_execution_profile_summary,
 )
 from ..quota.heartbeat_recommendation import (
+    build_action_selection_recovery_recommendation,
     build_heartbeat_recommendation,
     refine_heartbeat_recommendation,
 )
@@ -102,9 +103,13 @@
     qualify_action_selection_from_inventory,
 )
 from ..work_items.execution_obligation import build_execution_obligation
+from ..work_items.action_selection_contract import (
+    apply_action_selection_recovery_projection,
+)
 from ..work_items.goal_route_hint import build_goal_route_hint
 from ..work_items.interaction_contract import (
     build_interaction_contract,
+    unadmitted_action_selection,
     build_protocol_action_packet,
     finalize_user_gate_notification_cooldown,
 )
@@ -366,6 +371,40 @@ def _apply_agent_monitor_only_precedence(
     clear_quota_action_projections(payload)
 
 
+def _apply_unadmitted_action_selection_precedence(
+    payload: dict[str, Any],
+    *,
+    replay_phase: ReceiptBoundReplayPhase | None,
+) -> None:
+    """Finalize one closed selection recovery before shared projections."""
+
+    if replay_phase is ReceiptBoundReplayPhase.SETTLED or not (
+        unadmitted_action_selection(payload)
+    ):
+        return
+    for field in (
+        "agent_lane_next_action",
+        "agent_scope_frontier",
+        "autonomous_replan_obligation",
+        "execution_profile",
+        "goal_route_hint",
+        "handoff_readiness",
+        "replan_action_packet",
+        "scoped_user_gate_fallback",
+        "selected_todo",
+        "task_orchestration_contract",
+        "todo_id",
+        "todo_write_hint",
+        "work_lane_contract",
+        "workspace_guard",
+    ):
+        payload.pop(field, None)
+    apply_action_selection_recovery_projection(payload)
+    payload["heartbeat_recommendation"] = (
+        build_action_selection_recovery_recommendation(
+            reason=str(payload.get("reason") or "")
+        )
+    )
 def _delivery_preemptions_for_route(
     prepared: _QuotaDecisionPreparation,
     *,
@@ -1310,12 +1349,8 @@ def _build_active_quota_payload(
         next_action_warning=route.next_action_warning,
         replan_obligation=prepared.replan_obligation,
     )
-    bounded_research_frontier = (
-        prepared.status_payload.get("bounded_research_frontier")
-        if isinstance(
-            prepared.status_payload.get("bounded_research_frontier"), dict
-        )
-        else None
+    bounded_research_frontier = _dict_field(
+        prepared.status_payload, "bounded_research_frontier"
     )
     _attach_truthy_fields(
         payload,
@@ -1326,7 +1361,13 @@ def _build_active_quota_payload(
         monitor_only=prepared.agent_monitor_only,
         inbox_priority_due=prepared.inbox_priority_due,
     )
-    if isinstance(payload.get("autonomous_replan_obligation"), dict):
+    _apply_unadmitted_action_selection_precedence(
+        payload,
+        replay_phase=prepared.receipt_bound_replay_phase,
+    )
+    if isinstance(
+        payload.get("autonomous_replan_obligation"), dict
+    ) and not unadmitted_action_selection(payload):
         payload["replan_action_packet"] = build_replan_action_packet(
             payload["autonomous_replan_obligation"],
             goal_id=prepared.safe_goal_id,
diff --git a/loopx/control_plane/work_items/action_selection_contract.py b/loopx/control_plane/work_items/action_selection_contract.py
index 18352d8ad..e4622c48d 100644
--- a/loopx/control_plane/work_items/action_selection_contract.py
+++ b/loopx/control_plane/work_items/action_selection_contract.py
@@ -5,6 +5,8 @@
 from typing import Any
 
 from ..agents.capability_gate import runtime_capabilities_for_cli_projection
+from ..todos.contract import normalize_todo_id
+from ..quota.effective_action import EffectiveAction
 
 
 def render_cli_command_prefix(*, runtime_root: str | None = None) -> str:
@@ -157,42 +159,246 @@ def delivery_spend_allowed(
     )
 
 
-def apply_action_selection_recovery(
-    payload: dict[str, Any],
-    *,
-    registry_path: str,
-    runtime_root: str,
-    goal_id: str,
-    agent_id: str,
-    turn_instance_id: str | None,
-    scheduler_args: str,
-    available_capabilities: Any = None,
-) -> None:
-    """Render typed selection recovery before offering any settlement effects."""
-    qualification = payload.get("action_selection_qualification") or {}
+def current_action_selection_admission(
+    payload: Mapping[str, Any], *, requested_todo_id: str,
+    agent_must_attempt: bool, agent_delivery_refused: bool,
+) -> tuple[str | None, bool]:
+    """Read current obligation admission, including repair and monitor positives."""
+    selected_todo = payload.get("selected_todo")
+    selected_todo_id = (
+        normalize_todo_id(selected_todo.get("todo_id"))
+        if isinstance(selected_todo, Mapping)
+        else None
+    )
+    qualification_value = payload.get("action_selection_qualification")
+    qualification: Mapping[str, object] = (
+        qualification_value if isinstance(qualification_value, Mapping) else {}
+    )
+    if selected_todo_id is None and str(qualification.get("state") or "") == (
+        "qualified"
+    ):
+        # An unsettled-host-turn recovery decision carries no top-level
+        # `selected_todo`: its qualification names the Todo that prior Turn
+        # has to settle, and binding the guard to that Todo is the documented
+        # closeout path rather than a conflict with the projection.
+        qualification_selected = qualification.get("selected_todo")
+        selected_todo_id = (
+            normalize_todo_id(qualification_selected.get("todo_id"))
+            if isinstance(qualification_selected, Mapping)
+            else None
+        )
+    selection_binding = (
+        selected_todo.get("selection_binding")
+        if isinstance(selected_todo, Mapping)
+        else None
+    )
+    execution_obligation_value = payload.get("execution_obligation")
+    execution_obligation: Mapping[str, object] = (
+        execution_obligation_value
+        if isinstance(execution_obligation_value, Mapping)
+        else {}
+    )
+    pending_selection_delivery_qualified = (
+        selection_binding == "pending_action_selection"
+        and payload.get("normal_delivery_allowed") is True
+    )
+    pending_selection_workspace_repair_qualified = (
+        selection_binding == "pending_action_selection"
+        and payload.get("workspace_repair_allowed") is True
+        and payload.get("effective_action") == EffectiveAction.AGENT_WORKSPACE_REPAIR.value
+        and execution_obligation.get("kind") == "agent_workspace_repair"
+        and execution_obligation.get("must_attempt_work") is True
+        and agent_must_attempt
+        and agent_delivery_refused
+    )
+    exact_current_obligation_qualified = (
+        selection_binding != "pending_action_selection"
+        and execution_obligation.get("must_attempt_work") is True
+        and agent_must_attempt
+    )
+    if (
+        selected_todo_id == requested_todo_id
+        and payload.get("ok") is True
+        and payload.get("should_run") is True
+        and (
+            pending_selection_delivery_qualified
+            or pending_selection_workspace_repair_qualified
+            or exact_current_obligation_qualified
+        )
+    ):
+        return selected_todo_id, True
+
+    return selected_todo_id, False
+
+
+def action_selection_needs_recovery(
+    payload: Mapping[str, Any], *, agent_must_attempt: bool = False,
+    agent_delivery_refused: bool = False,
+) -> bool:
+    """Read the typed qualifier's result without reimplementing admission."""
+    qualification = payload.get("action_selection_qualification")
+    if not isinstance(qualification, Mapping) or qualification.get("state") not in {"deferred", "rejected"}:
+        return False
+    # A committed binding is reconciled by the receipt owner, not by pending
+    # selection recovery. Preserve the same exemption as CLI preflight.
+    selected = payload.get("selected_todo")
+    replan = payload.get("autonomous_replan_obligation")
+    if any(isinstance(value, Mapping) and value.get("selection_binding") == "heartbeat_receipt"
+           for value in (selected, replan)):
+        return False
+    _, admitted = current_action_selection_admission(
+        payload, requested_todo_id=str(qualification.get("requested_todo_id") or ""),
+        agent_must_attempt=agent_must_attempt, agent_delivery_refused=agent_delivery_refused,
+    )
+    if admitted:
+        return False
     if qualification.get("recovery_action") != "reenter_guard_without_selection":
         raise RuntimeError("rejected action selection omitted its typed recovery action")
-    argv = ["loopx", "--registry", registry_path, "--runtime-root", runtime_root,
-            "--format", "json", "quota", "should-run", "--goal-id", goal_id,
-            "--agent-id", agent_id]
+    return True
+
+
+def action_selection_recovery_fields(
+    recovery: dict[str, Any],
+) -> dict[str, Any]:
+    """Complete the one preflight result in the owning projection module."""
+    recovery["execution_obligation"] = {
+        "must_attempt_work": False,
+        "kind": EffectiveAction.QUOTA_SKIP.value,
+        "delivery_allowed": False,
+        "notify_is_execution_gate": False,
+        "reason": recovery["recommended_action"],
+        "spend_policy": "no quota spend until an eligible Todo is selected",
+    }
+    return recovery
+
+
+def build_action_selection_recovery_fields(
+    payload: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Construct the closed root facts for one typed selection refusal."""
+
+    qualification = payload.get("action_selection_qualification")
+    if not isinstance(qualification, Mapping):
+        raise RuntimeError("selection recovery requires a typed qualification")
+    qualification_state = str(qualification.get("state") or "")
+    if qualification_state not in {"deferred", "rejected"}:
+        raise RuntimeError("selection recovery requires a deferred or rejected state")
+    qualification_reason = str(
+        qualification.get("reason") or "candidate_not_currently_eligible"
+    )
+    deferred = qualification_state == "deferred"
+    auxiliary_monitor = (
+        qualification_reason
+        == "auxiliary_monitor_not_selectable_in_advancement_lane"
+    )
+    error_code = (
+        "quota_action_selection_deferred"
+        if deferred
+        else "quota_action_selection_rejected"
+    )
+    recovery = {
+        "ok": False,
+        "spend_allowed_now": False,
+        "spend_after_validation": False,
+        "decision": "skip",
+        "should_run": False,
+        "normal_delivery_allowed": False,
+        "recovery_delivery_allowed": False,
+        "self_repair_allowed": False,
+        "capability_repair_allowed": False,
+        "workspace_repair_allowed": False,
+        "actionable_by_codex": False,
+        # An unadmitted selection grants no safe bypass either: the heartbeat
+        # task body reads safe_bypass_allowed as permission to run one bounded
+        # step and spend, so the refusal must close that authority too.
+        "safe_bypass_allowed": False,
+        "safe_bypass_kind": None,
+        "safe_bypass_policy": None,
+        "effective_action": EffectiveAction.QUOTA_SKIP.value,
+        "state": error_code,
+        "waiting_on": "codex",
+        "status": error_code,
+        "error_code": error_code,
+        "reason": (
+            "explicit action selection was deferred by the current "
+            f"delivery frontier: {qualification_reason}"
+            if deferred
+            else "explicit action selection is not currently eligible: "
+            f"{qualification_reason}"
+        ),
+        "recommended_action": (
+            "handle the current delivery preemption, then rerun quota "
+            "should-run with the same --turn-instance-id; omit --todo-id "
+            "first when a refreshed action portfolio is needed"
+            if deferred
+            else "the due monitor is visible as auxiliary context, not an "
+            "independently selectable action in the current advancement lane; "
+            "choose a current advancement Todo, or rerun after the monitor "
+            "becomes the hard lane"
+            if auxiliary_monitor
+            else "rerun quota should-run with the same --turn-instance-id "
+            "without --todo-id, then choose a currently eligible Todo"
+        ),
+    }
+    return action_selection_recovery_fields(recovery)
+
+
+def apply_action_selection_recovery_projection(payload: dict[str, Any]) -> bool:
+    """Replace an unadmitted action with its closed pre-finalization facts."""
+
+    qualification = payload.get("action_selection_qualification")
+    if not isinstance(qualification, Mapping) or qualification.get("state") not in {
+        "deferred",
+        "rejected",
+    }:
+        return False
+    if qualification.get("recovery_action") != "reenter_guard_without_selection":
+        raise RuntimeError("rejected action selection omitted its typed recovery action")
+    payload.update(build_action_selection_recovery_fields(payload))
+    return True
+
+
+def action_selection_recovery_command(
+    *, registry_path: str | None = None, runtime_root: str | None = None,
+    goal_id: str, agent_id: str | None, turn_instance_id: str | None,
+    scheduler_args: str, available_capabilities: Any = None,
+) -> str:
+    argv = ["loopx"]
+    if registry_path:
+        argv.extend(["--registry", registry_path])
+    if runtime_root:
+        argv.extend(["--runtime-root", runtime_root])
+    argv.extend(["--format", "json", "quota", "should-run", "--goal-id", goal_id])
+    if agent_id:
+        argv.extend(["--agent-id", agent_id])
     if turn_instance_id:
         argv.extend(["--turn-instance-id", turn_instance_id])
     for capability in runtime_capabilities_for_cli_projection(available_capabilities):
         argv.extend(["--available-capability", capability])
-    command = shlex.join(argv) + scheduler_args
-    payload["spend_allowed_now"] = False
-    payload["spend_after_validation"] = False
-    # The current replan has not been admitted for this turn. Keeping its
-    # action packet would replace recovery in the compact TurnEnvelope.
-    payload.pop("replan_action_packet", None)
-    interaction = payload.get("interaction_contract") or {}
-    agent = interaction.get("agent_channel") or {}
-    agent.update(must_attempt=False, delivery_allowed=False, primary_action=command)
-    cli = interaction.get("cli_channel") or {}
-    for field in ("settlement_plan", "replan_settlement_contract", "selection_command", "selection_policy_ref"):
-        cli.pop(field, None)
-    cli.update(next_cli_actions=[command], selection_required=False,
-               spend_allowed_now=False, spend_after_validation=False,
-               spend_policy="rerun this turn's guard before delivery or settlement")
-    interaction.update(agent_channel=agent, cli_channel=cli)
-    payload["interaction_contract"] = interaction
+    return shlex.join(argv) + scheduler_args
+
+
+def action_selection_recovery_cli_channel(command: str) -> dict[str, Any]:
+    return {
+        "next_cli_actions": [command], "selection_required": False,
+        "spend_allowed_now": False, "spend_after_validation": False,
+        "spend_policy": "rerun this turn's guard before delivery or settlement",
+    }
+
+
+def bind_action_selection_recovery_command(
+    payload: dict[str, Any], *, registry_path: str, runtime_root: str,
+    goal_id: str, agent_id: str | None, turn_instance_id: str | None,
+    scheduler_args: str, available_capabilities: Any = None,
+) -> None:
+    """Bind the existing recovery projection to the invoking CLI's exact argv."""
+    if not action_selection_needs_recovery(payload):
+        raise RuntimeError("selection recovery binding requires a typed recovery result")
+    command = action_selection_recovery_command(
+        registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id,
+        agent_id=agent_id, turn_instance_id=turn_instance_id,
+        scheduler_args=scheduler_args, available_capabilities=available_capabilities,
+    )
+    interaction = payload["interaction_contract"]
+    interaction["agent_channel"]["primary_action"] = command
+    interaction["cli_channel"]["next_cli_actions"] = [command]
diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py
index 06a13e102..8f91d8471 100644
--- a/loopx/control_plane/work_items/interaction_contract.py
+++ b/loopx/control_plane/work_items/interaction_contract.py
@@ -44,6 +44,8 @@
 from . import runtime_capability_reentry as capability_reentry_adapter
 from .primary_action import (
     build_primary_action_projection,
+    interaction_execution_flags,
+    interaction_quiet_noop_allowed,
     protocol_action_label as _protocol_action_label,
     protocol_action_text,
     protocol_first_candidate_action as _protocol_first_candidate_action,
@@ -667,6 +669,20 @@ def _terminal_cli_actions(
     return ["no quota spend without validated transition/blocker writeback"]
 
 
+def _selection_recovery_command(
+    payload: dict[str, Any], *, available_capabilities: Any,
+    scheduler_execution_context: Mapping[str, Any] | SchedulerExecutionContextResolution | None,
+    turn_instance_id: str | None, runtime_root: str | None,
+) -> str:
+    identity = payload.get("agent_identity") if isinstance(payload.get("agent_identity"), dict) else {}
+    return selection.action_selection_recovery_command(
+        goal_id=str(payload.get("goal_id") or "<GOAL_ID>"),
+        agent_id=identity.get("agent_id"), runtime_root=runtime_root,
+        turn_instance_id=turn_instance_id, available_capabilities=available_capabilities,
+        scheduler_args=render_scheduler_execution_args(scheduler_execution_context=scheduler_execution_context),
+    )
+
+
 def interaction_next_cli_actions(
     payload: dict[str, Any],
     *,
@@ -681,6 +697,12 @@ def interaction_next_cli_actions(
     turn_instance_id: str | None = None,
     runtime_root: str | None = None,
 ) -> list[str]:
+    if unadmitted_action_selection(payload):
+        return [_selection_recovery_command(
+            payload, available_capabilities=available_capabilities,
+            scheduler_execution_context=scheduler_execution_context,
+            turn_instance_id=turn_instance_id, runtime_root=runtime_root,
+        )]
     goal_id = str(payload.get("goal_id") or "<GOAL_ID>")
     command_prefix = selection.render_cli_command_prefix(runtime_root=runtime_root)
     agent_identity = payload.get("agent_identity") if isinstance(payload.get("agent_identity"), dict) else {}
@@ -1049,72 +1071,6 @@ def _blocked_priority_fallback_user_reason(payload: dict[str, Any]) -> str | Non
     return blocked_priority_fallback_owner_reason(fallback)
 
 
-def _interaction_must_attempt(
-    execution_obligation: dict[str, Any],
-    *,
-    mode: str,
-    user_required: bool,
-    scoped_user_gate_fallback: bool,
-    bounded_delivery_with_user_notice: bool,
-) -> bool:
-    if mode == "governed_capability_intent":
-        return bool(execution_obligation.get("must_attempt_work"))
-    if user_required and not (
-        scoped_user_gate_fallback or bounded_delivery_with_user_notice
-    ):
-        return False
-    return bool(execution_obligation.get("must_attempt_work"))
-
-
-def _interaction_delivery_allowed(
-    payload: dict[str, Any],
-    execution_obligation: dict[str, Any],
-    *,
-    mode: str,
-    user_required: bool,
-    scoped_user_gate_fallback: bool,
-    bounded_delivery_with_user_notice: bool,
-) -> bool:
-    if mode == "governed_capability_intent":
-        return bool(execution_obligation.get("must_attempt_work"))
-    if mode == "mapped_noop_if_unchanged":
-        return False
-    if user_required and not (
-        scoped_user_gate_fallback or bounded_delivery_with_user_notice
-    ):
-        return False
-    return bool(
-        execution_obligation.get(
-            "delivery_allowed",
-            payload.get("normal_delivery_allowed")
-            or payload.get("recovery_delivery_allowed")
-            or payload.get("self_repair_allowed")
-            or payload.get("should_run"),
-        )
-    )
-
-
-def _interaction_quiet_noop_allowed(
-    *,
-    mode: str,
-    user_required: bool,
-    must_attempt: bool,
-) -> bool:
-    if user_required or must_attempt:
-        return False
-    return _agent_scope_frontier_action(mode) is not None or mode in {
-        "monitor_quiet_skip",
-        "mapped_noop_if_unchanged",
-        "quota_throttled",
-        "blocked_wait",
-        "user_gate_cooldown_wait",
-        "terminal_no_followup",
-        "peer_coordination_blocked",
-        "agent_monitor_only",
-        "skip",
-    }
-
-
 def _interaction_spend_after_validation(mode: str) -> bool:
     return mode in {
         "bounded_delivery",
@@ -1291,6 +1247,12 @@ def _build_interaction_cli_channel(
     turn_instance_id: str | None = None,
     runtime_root: str | None = None,
 ) -> dict[str, Any]:
+    if unadmitted_action_selection(payload):
+        return selection.action_selection_recovery_cli_channel(_selection_recovery_command(
+            payload, available_capabilities=available_capabilities,
+            scheduler_execution_context=scheduler_execution_context,
+            turn_instance_id=turn_instance_id, runtime_root=runtime_root,
+        ))
     spend_after_selection = selection.delivery_spend_allowed(payload, spend_after_validation)
     settlement_plan, replan_settlement_contract = (
         _turn_scoped_cli_settlement_context(
@@ -1525,6 +1487,22 @@ def _interaction_fallback_policy_required(payload: dict[str, Any], *, mode: str)
     } or bool(payload.get("blocked_priority_fallback"))
 
 
+def unadmitted_action_selection(payload: dict[str, Any]) -> bool:
+    """Use the same current-obligation facts as interaction and CLI preflight."""
+    qualification = payload.get("action_selection_qualification")
+    if not isinstance(qualification, Mapping) or qualification.get("state") not in {"deferred", "rejected"}:
+        return False
+    mode = _interaction_mode(payload)
+    user_required = False if payload.get("agent_work_mode") == "monitor_only" else user_channel_action_required(payload)
+    must_attempt, delivery_allowed = interaction_execution_flags(
+        payload, mode=mode, user_required=user_required,
+        blocked_successor_wait_observation=_blocked_successor_wait_observation_required(payload),
+    )
+    return selection.action_selection_needs_recovery(
+        payload, agent_must_attempt=must_attempt, agent_delivery_refused=delivery_allowed is False,
+    )
+
+
 def build_interaction_contract(
     payload: dict[str, Any],
     *,
@@ -1548,28 +1526,11 @@ def build_interaction_contract(
     mode = _interaction_mode(payload)
     monitor_only = payload.get("agent_work_mode") == "monitor_only"
     user_required = False if monitor_only else user_channel_action_required(payload)
-    scoped_user_gate_fallback = mode == "scoped_user_gate_fallback"
-    bounded_delivery_with_user_notice = mode == "bounded_delivery_with_user_notice"
-    must_attempt = _interaction_must_attempt(
-        execution_obligation,
-        mode=mode,
-        user_required=user_required,
-        scoped_user_gate_fallback=scoped_user_gate_fallback,
-        bounded_delivery_with_user_notice=bounded_delivery_with_user_notice,
+    must_attempt, delivery_allowed = interaction_execution_flags(
+        payload, mode=mode, user_required=user_required,
+        blocked_successor_wait_observation=_blocked_successor_wait_observation_required(payload),
     )
-    if mode == "automation_prompt_upgrade":
-        must_attempt = True
-    if _blocked_successor_wait_observation_required(payload):
-        must_attempt = True
-    delivery_allowed = _interaction_delivery_allowed(
-        payload,
-        execution_obligation,
-        mode=mode,
-        user_required=user_required,
-        scoped_user_gate_fallback=scoped_user_gate_fallback,
-        bounded_delivery_with_user_notice=bounded_delivery_with_user_notice,
-    )
-    quiet_noop_allowed = _interaction_quiet_noop_allowed(
+    quiet_noop_allowed = interaction_quiet_noop_allowed(
         mode=mode,
         user_required=user_required,
         must_attempt=must_attempt,
@@ -1598,14 +1559,25 @@ def build_interaction_contract(
             "notify": "DONT_NOTIFY",
             "reason": payload.get("reason"),
         }
-    agent_channel = _build_interaction_agent_channel(
-        payload,
-        mode=mode,
-        must_attempt=must_attempt,
-        delivery_allowed=delivery_allowed,
-        quiet_noop_allowed=quiet_noop_allowed,
-        capability_reentry=capability_reentry,
-    )
+    if unadmitted_action_selection(payload):
+        agent_channel = {
+            "must_attempt": False, "delivery_allowed": False,
+            "quiet_noop_allowed": quiet_noop_allowed,
+            "primary_action": _selection_recovery_command(
+                payload, available_capabilities=available_capabilities,
+                scheduler_execution_context=scheduler_execution_context,
+                turn_instance_id=turn_instance_id, runtime_root=runtime_root,
+            ),
+        }
+    else:
+        agent_channel = _build_interaction_agent_channel(
+            payload,
+            mode=mode,
+            must_attempt=must_attempt,
+            delivery_allowed=delivery_allowed,
+            quiet_noop_allowed=quiet_noop_allowed,
+            capability_reentry=capability_reentry,
+        )
     contract: dict[str, Any] = {
         "schema_version": INTERACTION_CONTRACT_SCHEMA_VERSION,
         "mode": mode,
diff --git a/loopx/control_plane/work_items/primary_action.py b/loopx/control_plane/work_items/primary_action.py
index a4c53af5f..1fc5a2080 100644
--- a/loopx/control_plane/work_items/primary_action.py
+++ b/loopx/control_plane/work_items/primary_action.py
@@ -343,3 +343,98 @@ def build_primary_action_projection(payload: dict[str, Any], *, mode: str) -> di
     if resolution_trace:
         projection["resolution_trace"] = resolution_trace
     return projection
+
+
+def _interaction_must_attempt(
+    execution_obligation: dict[str, Any],
+    *,
+    mode: str,
+    user_required: bool,
+    scoped_user_gate_fallback: bool,
+    bounded_delivery_with_user_notice: bool,
+) -> bool:
+    if mode == "governed_capability_intent":
+        return bool(execution_obligation.get("must_attempt_work"))
+    if user_required and not (
+        scoped_user_gate_fallback or bounded_delivery_with_user_notice
+    ):
+        return False
+    return bool(execution_obligation.get("must_attempt_work"))
+
+
+def _interaction_delivery_allowed(
+    payload: dict[str, Any],
+    execution_obligation: dict[str, Any],
+    *,
+    mode: str,
+    user_required: bool,
+    scoped_user_gate_fallback: bool,
+    bounded_delivery_with_user_notice: bool,
+) -> bool:
+    if mode == "governed_capability_intent":
+        return bool(execution_obligation.get("must_attempt_work"))
+    if mode == "mapped_noop_if_unchanged":
+        return False
+    if user_required and not (
+        scoped_user_gate_fallback or bounded_delivery_with_user_notice
+    ):
+        return False
+    return bool(
+        execution_obligation.get(
+            "delivery_allowed",
+            payload.get("normal_delivery_allowed")
+            or payload.get("recovery_delivery_allowed")
+            or payload.get("self_repair_allowed")
+            or payload.get("should_run"),
+        )
+    )
+
+
+def interaction_quiet_noop_allowed(
+    *,
+    mode: str,
+    user_required: bool,
+    must_attempt: bool,
+) -> bool:
+    if user_required or must_attempt:
+        return False
+    return _agent_scope_frontier_action(mode) is not None or mode in {
+        "monitor_quiet_skip",
+        "mapped_noop_if_unchanged",
+        "quota_throttled",
+        "blocked_wait",
+        "user_gate_cooldown_wait",
+        "terminal_no_followup",
+        "peer_coordination_blocked",
+        "agent_monitor_only",
+        "skip",
+    }
+
+
+def interaction_execution_flags(
+    payload: dict[str, Any], *, mode: str, user_required: bool,
+    blocked_successor_wait_observation: bool,
+) -> tuple[bool, bool]:
+    execution_obligation = payload.get("execution_obligation") if isinstance(payload.get("execution_obligation"), dict) else {}
+    scoped_user_gate_fallback = mode == "scoped_user_gate_fallback"
+    bounded_delivery_with_user_notice = mode == "bounded_delivery_with_user_notice"
+    must_attempt = _interaction_must_attempt(
+        execution_obligation,
+        mode=mode,
+        user_required=user_required,
+        scoped_user_gate_fallback=scoped_user_gate_fallback,
+        bounded_delivery_with_user_notice=bounded_delivery_with_user_notice,
+    )
+    if mode == "automation_prompt_upgrade":
+        must_attempt = True
+    if blocked_successor_wait_observation:
+        must_attempt = True
+    delivery_allowed = _interaction_delivery_allowed(
+        payload,
+        execution_obligation,
+        mode=mode,
+        user_required=user_required,
+        scoped_user_gate_fallback=scoped_user_gate_fallback,
+        bounded_delivery_with_user_notice=bounded_delivery_with_user_notice,
+    )
+    return must_attempt, delivery_allowed
diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json
index 054688550..038ee2649 100644
--- a/loopx/semantics/vocabulary_v0.json
+++ b/loopx/semantics/vocabulary_v0.json
@@ -496,7 +496,7 @@
         "unsettled_host_turn_recovery": "A host Turn is unsettled and must be recovered before anything else: the selected Todo and action portfolio are dropped, should_run is set, and normal, recovery and self-repair delivery are all refused."
       },
       "producers": [
-        "loopx/cli_commands/quota.py::_apply_requested_quota_action_selection_preflight",
+        "loopx/control_plane/work_items/action_selection_contract.py::build_action_selection_recovery_fields",
         "loopx/control_plane/quota/decision_summary.py::_task_orchestration_effective_action",
         "loopx/control_plane/quota/decision_summary.py::quota_effective_action",
         "loopx/control_plane/quota/decision_summary.py::resolve_quota_run_decision",
diff --git a/tests/control_plane/test_quota_action_selection_conflict.py b/tests/control_plane/test_quota_action_selection_conflict.py
index 0018475fc..edaa578cd 100644
--- a/tests/control_plane/test_quota_action_selection_conflict.py
+++ b/tests/control_plane/test_quota_action_selection_conflict.py
@@ -7,7 +7,7 @@
 
 import pytest
 
-from loopx.cli_commands.quota import _apply_requested_quota_action_selection_preflight
+from loopx.cli_commands.quota import _requested_quota_action_selection_preflight
 from loopx.cli_commands.quota_failure_report import quota_failure_payload
 from loopx.control_plane.quota.error_codes import (
     QuotaActionSelectionConflictError,
@@ -35,7 +35,7 @@ def _payload(**overrides: object) -> dict[str, object]:
 
 def _raise(payload: dict[str, object]) -> QuotaActionSelectionConflictError:
     with pytest.raises(QuotaActionSelectionConflictError) as raised:
-        _apply_requested_quota_action_selection_preflight(
+        _requested_quota_action_selection_preflight(
             payload,
             requested_todo_id=REQUESTED_TODO_ID,
             receipt_bound_todo_id=None,
@@ -77,7 +77,7 @@ def test_missing_qualification_is_typed_rather_than_unexplained() -> None:
 
 
 def test_a_qualified_selection_for_the_requested_todo_is_not_a_conflict() -> None:
-    is_conflict = _apply_requested_quota_action_selection_preflight(
+    is_conflict = _requested_quota_action_selection_preflight(
         _payload(action_selection_qualification=_qualified_for(REQUESTED_TODO_ID)),
         requested_todo_id=REQUESTED_TODO_ID,
         receipt_bound_todo_id=None,
@@ -86,7 +86,7 @@ def test_a_qualified_selection_for_the_requested_todo_is_not_a_conflict() -> Non
         receipt_identity_upgraded=False,
     )
 
-    assert is_conflict is False
+    assert is_conflict is None
 
 
 def test_failure_payload_reports_the_conflict_instead_of_collection_failure() -> None:
diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index 4d115b95b..f562f2c50 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -36,6 +36,46 @@
 SELECTED_REPLAN_TODO_ID = "todo_chain_000000000000"
 
 
+def _assert_action_selection_recovery_projections(payload: dict[str, Any]) -> None:
+    from loopx.control_plane.quota.turn_envelope import build_turn_envelope
+    from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority
+
+    interaction = payload["interaction_contract"]
+    assert interaction["mode"] == "skip"
+    assert interaction["agent_channel"]["must_attempt"] is False
+    assert interaction["agent_channel"]["delivery_allowed"] is False
+    assert interaction["cli_channel"]["spend_allowed_now"] is False
+    assert interaction["cli_channel"]["spend_after_validation"] is False
+    assert payload["execution_obligation"]["kind"] == "quota_skip"
+    assert payload["execution_obligation"]["must_attempt_work"] is False
+    assert payload["automation_liveness"]["automation_action"] == "keep_active"
+    assert payload["scheduler_hint"]["action"] == "backoff_until_state_change"
+    protocol_summary = payload["protocol_action_packet"]["summary"]
+    assert "agent_action_required=false" in protocol_summary
+    assert "agent_action_required=true" not in protocol_summary
+    assert "execute_bounded_work" not in protocol_summary
+    for field in (
+        "autonomous_replan_obligation",
+        "replan_action_packet",
+        "selected_todo",
+        "work_lane_contract",
+    ):
+        assert field not in payload
+
+    envelope = build_turn_envelope(payload)
+    assert envelope["contract_capsule"]["interaction_contract"]["mode"] == "skip"
+    assert envelope["contract_capsule"]["execution_obligation"][
+        "must_attempt_work"
+    ] is False
+    assert envelope["writeback"]["spend_allowed_now"] is False
+    assert envelope["writeback"]["spend_after_validation"] is False
+    authority = extract_turn_authority({"turn_envelope": envelope})
+    assert authority["primary_action"] == interaction["agent_channel"][
+        "primary_action"
+    ]
+    assert authority["write_scope"] == []
+
+
 def _write_fixture(
     root: Path,
     *,
@@ -3030,6 +3070,7 @@ def test_agent_selection_rejects_unprojected_todo(tmp_path: Path) -> None:
         "requested_todo_id": "todo_not_projected",
         "reason": "candidate_not_currently_eligible",
     }
+    _assert_action_selection_recovery_projections(invalid)
     assert invalid["heartbeat_receipt"]["status"] == "replayed"
     assert invalid["rollout_event"]["appended"] is False
     assert _heartbeat_receipt_count(runtime, turn_instance_id) == 1
@@ -3110,6 +3151,7 @@ def test_first_call_rejected_selection_does_not_commit_a_false_receipt(
 
     assert rejected_rc == 1, rejected
     assert rejected["error_code"] == "quota_action_selection_rejected"
+    _assert_action_selection_recovery_projections(rejected)
     assert rejected["heartbeat_receipt"] == {
         "schema_version": "heartbeat_quota_receipt_v0",
         "turn_instance_id": turn_instance_id,
@@ -3528,6 +3570,7 @@ def test_pending_action_selection_reports_autonomous_replan_preemption(
         "reason": "autonomous_replan",
         "delivery_preemptions": ["autonomous_replan", "delivery_not_allowed"],
     }
+    _assert_action_selection_recovery_projections(selected)
     assert selected["heartbeat_receipt"]["status"] == "selection_retained"
     assert selected["heartbeat_receipt"]["pending_action_selection"]["todo_id"] == (
         ALTERNATIVE_TODO_ID
@@ -4371,7 +4414,10 @@ def test_todoless_blocked_replan_settles_read_only_external_evidence_without_wor
 
     assert replay_rc == 0, replay
     assert replay["effective_action"] == "heartbeat_settled_skip"
+    assert replay["interaction_contract"]["mode"] == "heartbeat_settled_skip"
+    assert replay["execution_obligation"]["kind"] == "heartbeat_settled_skip"
     assert replay["should_run"] is False
+    assert replay.get("error_code") is None
     assert replay.get("selected_todo") is None
     assert replay.get("unsettled_host_turn_recovery") is None
 
diff --git a/tests/control_plane/test_unadmitted_selection_construction.py b/tests/control_plane/test_unadmitted_selection_construction.py
new file mode 100644
index 000000000..b3fc90d42
--- /dev/null
+++ b/tests/control_plane/test_unadmitted_selection_construction.py
@@ -0,0 +1,222 @@
+"""Typed rejection builds recovery before any settlement capability exists."""
+from __future__ import annotations
+
+from copy import deepcopy
+import shlex
+
+import pytest
+
+from loopx.cli_commands.quota import _requested_quota_action_selection_preflight
+from loopx.control_plane.work_items import interaction_contract
+from loopx.control_plane.work_items.action_selection_contract import (
+    action_selection_needs_recovery,
+    bind_action_selection_recovery_command,
+)
+
+
+def _source(state: str) -> dict:
+    return {
+        "goal_id": "selection-fixture", "agent_identity": {"agent_id": "fixture-agent"},
+        "ok": True, "should_run": True, "effective_action": "autonomous_replan",
+        "recommended_action": "Handle the preempting replan",
+        "execution_obligation": {"must_attempt_work": True, "delivery_allowed": True},
+        "action_selection_qualification": {
+            "schema_version": "action_selection_qualification_v0", "state": state,
+            "requested_todo_id": "todo_pending", "reason": "current_delivery_gate",
+            "recovery_action": "reenter_guard_without_selection",
+        },
+    }
+
+
+@pytest.mark.parametrize("state", ["deferred", "rejected"])
+def test_recovery_never_constructs_settlement_or_executable_primary_action(monkeypatch, state):
+    def unexpected(*args, **kwargs):
+        pytest.fail("unadmitted selection constructed an executable projection")
+
+    monkeypatch.setattr(interaction_contract, "_turn_scoped_cli_settlement_context", unexpected)
+    monkeypatch.setattr(interaction_contract, "build_primary_action_projection", unexpected)
+    source = _source(state)
+    contract = interaction_contract.build_interaction_contract(
+        source, turn_instance_id="selection-turn", runtime_root="/runtime with spaces",
+        available_capabilities=["shell"],
+    )
+    source["interaction_contract"] = contract
+    assert contract["agent_channel"]["must_attempt"] is False
+    assert contract["agent_channel"]["delivery_allowed"] is False
+    cli = contract["cli_channel"]
+    assert cli["spend_after_validation"] is False
+    assert cli["selection_required"] is False
+    for field in ("settlement_plan", "replan_settlement_contract", "selection_command", "selection_policy_ref"):
+        assert field not in cli
+    bind_action_selection_recovery_command(
+        source, registry_path="/registry with spaces.json", runtime_root="/runtime with spaces",
+        goal_id="selection-fixture", agent_id="fixture-agent", turn_instance_id="selection-turn",
+        scheduler_args=" --codex-app", available_capabilities=["shell"],
+    )
+    [command] = cli["next_cli_actions"]
+    assert contract["agent_channel"]["primary_action"] == command
+    argv = shlex.split(command)
+    for flag, value in {"--registry": "/registry with spaces.json", "--runtime-root": "/runtime with spaces",
+                        "--goal-id": "selection-fixture", "--agent-id": "fixture-agent",
+                        "--turn-instance-id": "selection-turn", "--available-capability": "shell"}.items():
+        assert argv[argv.index(flag) + 1] == value
+    assert "--codex-app" in argv
+    assert "--todo-id" not in argv and "--replan-obligation-id" not in argv
+
+
+@pytest.mark.parametrize("state", ["deferred", "rejected"])
+def test_preflight_returns_one_result_without_mutating_source(state):
+    source = _source(state)
+    before = deepcopy(source)
+    result = _requested_quota_action_selection_preflight(
+        source, requested_todo_id="todo_pending", receipt_bound_todo_id=None,
+        receipt_bound_replan_obligation_id=None,
+    )
+    assert source == before
+    assert result["error_code"] == f"quota_action_selection_{state}"
+    assert result["should_run"] is False
+    assert result["execution_obligation"]["kind"] == "quota_skip"
+    assert result["execution_obligation"]["must_attempt_work"] is False
+    for flag in ("normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed",
+                 "capability_repair_allowed", "workspace_repair_allowed", "actionable_by_codex"):
+        assert result[flag] is False
+    assert result["spend_after_validation"] is False
+
+
+@pytest.mark.parametrize("bound_field", ["selected_todo", "autonomous_replan_obligation"])
+def test_committed_binding_is_owned_by_receipt_reconciliation(bound_field):
+    source = _source("deferred")
+    source[bound_field] = {"selection_binding": "heartbeat_receipt"}
+    assert action_selection_needs_recovery(source) is False
+
+
+def test_qualified_selection_is_not_recovery():
+    assert action_selection_needs_recovery(_source("qualified")) is False
+
+
+@pytest.mark.parametrize("delivery_allowed", [False, True, None])
+def test_pending_workspace_repair_requires_explicit_delivery_refusal(delivery_allowed):
+    from loopx.control_plane.quota.error_codes import QuotaActionSelectionConflictError
+
+    source = _source("qualified")
+    source.update(
+        effective_action="agent_workspace_repair", workspace_repair_allowed=True,
+        selected_todo={"todo_id": "todo_pending", "selection_binding": "pending_action_selection"},
+        execution_obligation={"kind": "agent_workspace_repair", "must_attempt_work": True},
+    )
+    agent = {"must_attempt": True}
+    if delivery_allowed is not None:
+        agent["delivery_allowed"] = delivery_allowed
+    source["interaction_contract"] = {"agent_channel": agent}
+    kwargs = dict(requested_todo_id="todo_pending", receipt_bound_todo_id=None,
+                  receipt_bound_replan_obligation_id=None)
+    if delivery_allowed is False:
+        assert _requested_quota_action_selection_preflight(source, **kwargs) is None
+    else:
+        with pytest.raises(QuotaActionSelectionConflictError):
+            _requested_quota_action_selection_preflight(source, **kwargs)
+
+
+def _scoped_gate_status():
+    """Reuse the real scoped user-gate fallback producer fixture."""
+    import sys
+
+    sys.path.insert(0, "tests/control_plane")
+    from test_user_gate_lane_progress import APP_CONTEXT, AGENT_ID, GOAL_ID, _status_payload
+
+    return _status_payload(gate_action_kind="approve_product_first_screen"), {
+        "goal_id": GOAL_ID, "agent_id": AGENT_ID,
+        "scheduler_execution_context": APP_CONTEXT,
+    }
+
+
+def test_scoped_fallback_cannot_execute_under_an_unadmitted_selection():
+    """A runnable scoped fallback does not survive a refused selection.
+
+    The fallback readback grants safe_bypass_allowed, and the shipped heartbeat
+    task body reads that under should_run=false as permission for one bounded
+    step plus a spend. The selection refusal has to close that authority too,
+    or the same packet says both "no work" and "do one step and spend".
+    """
+    from loopx.control_plane.quota.turn_envelope import build_turn_envelope
+    from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority
+    from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown
+    from loopx.quota import build_quota_should_run
+
+    status, kwargs = _scoped_gate_status()
+    payload = build_quota_should_run(
+        status, requested_action_todo_id="todo_not_projected", **kwargs,
+    )
+
+    assert payload["effective_action"] == "quota_skip"
+    assert payload["state"] == "quota_action_selection_rejected"
+    for flag in (
+        "ok", "should_run", "actionable_by_codex", "safe_bypass_allowed",
+        "spend_allowed_now", "spend_after_validation", "normal_delivery_allowed",
+        "recovery_delivery_allowed", "self_repair_allowed",
+        "capability_repair_allowed", "workspace_repair_allowed",
+    ):
+        assert payload[flag] is False, flag
+    assert payload["safe_bypass_kind"] is None
+    assert payload["safe_bypass_policy"] is None
+    assert "scoped_user_gate_fallback" not in payload
+
+    obligation = payload["execution_obligation"]
+    assert obligation["kind"] == "quota_skip"
+    assert obligation["must_attempt_work"] is False
+    assert obligation["delivery_allowed"] is False
+    assert "no quota spend" in obligation["spend_policy"]
+
+    interaction = payload["interaction_contract"]
+    assert interaction["mode"] != "scoped_user_gate_fallback"
+    assert interaction["agent_channel"]["must_attempt"] is False
+    assert interaction["agent_channel"]["delivery_allowed"] is False
+    assert interaction["cli_channel"]["spend_allowed_now"] is False
+    assert interaction["cli_channel"]["spend_after_validation"] is False
+
+    summary = payload["protocol_action_packet"]["summary"]
+    assert "agent_action_required=false" in summary
+    assert "agent_action_required=true" not in summary
+
+    envelope = build_turn_envelope(payload)
+    assert envelope["writeback"]["spend_allowed_now"] is False
+    assert envelope["writeback"]["spend_after_validation"] is False
+    capsule = envelope["contract_capsule"]
+    assert capsule["interaction_contract"]["mode"] != "scoped_user_gate_fallback"
+    assert capsule["execution_obligation"]["must_attempt_work"] is False
+    assert extract_turn_authority({"turn_envelope": envelope})["write_scope"] == []
+
+    guidance = render_quota_should_run_markdown(payload)
+    assert "safe_bypass" not in guidance
+    assert "spend only after validated writeback" not in guidance
+
+
+def test_scoped_fallback_still_runs_without_a_refused_selection():
+    """The refusal closes the grant; it does not retire the fallback path."""
+    from loopx.quota import build_quota_should_run
+
+    status, kwargs = _scoped_gate_status()
+    payload = build_quota_should_run(status, **kwargs)
+
+    assert payload["should_run"] is True
+    assert payload["safe_bypass_allowed"] is True
+    assert payload["safe_bypass_kind"] == "scoped_user_gate_fallback"
+    assert payload["interaction_contract"]["mode"] == "scoped_user_gate_fallback"
+
+
+@pytest.mark.parametrize("state", ["deferred", "rejected"])
+def test_recovery_fields_close_the_safe_bypass_grant(state):
+    """Both refusal states share one owner, so both close the same authority."""
+    from loopx.control_plane.work_items.action_selection_contract import (
+        build_action_selection_recovery_fields,
+    )
+
+    source = _source(state)
+    source.update(
+        safe_bypass_allowed=True, safe_bypass_kind="scoped_user_gate_fallback",
+        safe_bypass_policy="advance the fallback; spend only after validated writeback",
+    )
+    recovery = build_action_selection_recovery_fields(source)
+    assert recovery["safe_bypass_allowed"] is False
+    assert recovery["safe_bypass_kind"] is None
+    assert recovery["safe_bypass_policy"] is None