fix(quota): make deferred selection recovery executable before settlement

docs/quota-allocation.md

@@ -486,7 +486,15 @@ bounded suggestions. That request is only a pending selection: the second guard
 re-runs current lane arbitration and eligibility checks before upgrading the
 receipt. A newly due hard-priority monitor, blocking user gate, or other current
 preemption defers the request and leaves the receipt identity-less. Delivery and
-quota spend remain disabled until binding succeeds. A single-candidate response
+quota spend remain disabled until binding succeeds. Deferred/rejected selections
+return the TS-owned `recovery_action=reenter_guard_without_selection`: execute
+the single command in `interaction_contract.cli_channel.next_cli_actions`, with
+the same turn id and no Todo/replan argument. That guard either binds the current
+hard lane or returns a refreshed portfolio. No settlement plan is exposed before
+reentry, and a previously bound receipt cannot be retargeted. `recommended_action`
+retains the human-readable rejection or deferral guidance; the executable recovery
+command lives in `next_cli_actions` and `agent_channel.primary_action`.
+A single-candidate response
 keeps the direct execution path and does not add an extra selection round trip.
 
 When the selected Todo has meaningful strategic context, the same default

docs/reference/protocols/turn-envelope-v0.md

@@ -78,8 +78,12 @@ settlement-identity conflict. The full quota response preserves the TypeScript
 the exact current preemption or eligibility reason. An existing identity-less
 receipt is replayed without mutation; a first-call rejection reports
 `heartbeat_receipt.status=not_committed` and writes no receipt event. The agent
-can therefore refresh the current portfolio with the same Turn id and re-enter
-deterministically. A receipt already bound to a different Todo or autonomous
+receives `recovery_action=reenter_guard_without_selection` and one executable
+same-Turn guard in the full decision's `cli_channel.next_cli_actions`; the compact
+envelope preserves the recovery in its action and writeback preview. The failed
+selection exposes no settlement plan, spend command, or unadmitted replan action
+packet. Execute that guard without
+a Todo/replan argument before following the resulting binding or portfolio. A receipt already bound to a different Todo or autonomous
 replan obligation remains a hard `heartbeat_receipt_identity_conflict`.
 When a due monitor is visible only as auxiliary context for an advancement lane,
 the typed reason is

loopx/cli_commands/quota.py

@@ -46,8 +46,10 @@
 from ..control_plane.quota.turn_envelope import build_turn_envelope
 from ..control_plane.scheduler.execution_context import (
     GUIDED_START_TURN_RUNTIME_PROFILES,
+    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 ..presentation.renderers.quota_event_markdown import (
     render_quota_monitor_poll_markdown,
     render_quota_slot_preview_markdown,
@@ -331,6 +333,36 @@ def _apply_requested_quota_action_selection_preflight(
     return True
 
 
+def _reconcile_requested_quota_action_selection(
+    payload: dict[str, object],
+    args: argparse.Namespace,
+    *,
+    registry_path: Path,
+    context: QuotaCommandContext,
+    receipt_bound_todo_id: str | None,
+    receipt_bound_replan_obligation_id: str | None,
+) -> bool:
+    rejected = _apply_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,
+    )
+    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
+
+
 def _attach_uncommitted_action_selection_receipt(
     payload: dict[str, object],
     *,
@@ -552,15 +584,10 @@ def handle_quota_command(
                 turn_start_hook_dispatch=turn_start_hook_dispatch,
             )
             _attach_turn_start_hook_dispatch(payload, turn_start_hook_dispatch)
-            action_selection_preflight_failed = (
-                _apply_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
-                    ),
-                )
+            action_selection_preflight_failed = _reconcile_requested_quota_action_selection(
+                payload, args, registry_path=registry_path, context=context,
+                receipt_bound_todo_id=receipt_bound_todo_id,
+                receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id,
             )
             if heartbeat_turn_id:
                 if action_selection_preflight_failed:

loopx/control_plane/work_items/action_portfolio.ts

@@ -330,7 +330,12 @@ export function projectQuotaActionPortfolio(value: unknown): JsonObject | null {
  * reducer alone decides whether that pending choice may become the settlement
  * candidate.  Committed receipt replay remains a separate state.
  */
-export function qualifyActionSelection(value: unknown): JsonObject {
+type ActionSelectionQualification = JsonObject & (
+  | {state: "qualified"; selected_todo: JsonObject; recovery_action?: never}
+  | {state: "deferred" | "rejected"; recovery_action: "reenter_guard_without_selection"; selected_todo?: never}
+);
+
+export function qualifyActionSelection(value: unknown): ActionSelectionQualification {
   const request = requireJsonObject(value, "action_selection_qualification_request");
   if (
     request.schema_version !==
@@ -366,6 +371,7 @@ export function qualifyActionSelection(value: unknown): JsonObject {
     return {
       schema_version: ACTION_SELECTION_QUALIFICATION_SCHEMA_VERSION,
       state: "rejected",
+      recovery_action: "reenter_guard_without_selection",
       requested_todo_id: requestedTodoId,
       reason: requestedTaskClass === "continuous_monitor"
         ? "auxiliary_monitor_not_selectable_in_advancement_lane"
@@ -389,6 +395,7 @@ export function qualifyActionSelection(value: unknown): JsonObject {
     return {
       schema_version: ACTION_SELECTION_QUALIFICATION_SCHEMA_VERSION,
       state: "deferred",
+      recovery_action: "reenter_guard_without_selection",
       requested_todo_id: requestedTodoId,
       reason: preemptions[0] ?? "current_delivery_gate",
       delivery_preemptions: preemptions,

loopx/control_plane/work_items/action_selection_contract.py

@@ -4,6 +4,8 @@
 from collections.abc import Mapping
 from typing import Any
 
+from ..agents.capability_gate import runtime_capabilities_for_cli_projection
+
 
 def render_cli_command_prefix(*, runtime_root: str | None = None) -> str:
     return (
@@ -153,3 +155,44 @@ def delivery_spend_allowed(
     return spend_after_validation and not action_portfolio_requires_explicit_selection(
         payload
     )
+
+
+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 {}
+    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]
+    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

tests/control_plane/test_quota_settlement_cli.py

@@ -3020,6 +3020,7 @@ def test_agent_selection_rejects_unprojected_todo(tmp_path: Path) -> None:
     assert invalid["action_selection_qualification"] == {
         "schema_version": "action_selection_qualification_v0",
         "state": "rejected",
+        "recovery_action": "reenter_guard_without_selection",
         "requested_todo_id": "todo_not_projected",
         "reason": "candidate_not_currently_eligible",
     }
@@ -3508,6 +3509,7 @@ def test_pending_action_selection_reports_autonomous_replan_preemption(
     assert selected["action_selection_qualification"] == {
         "schema_version": "action_selection_qualification_v0",
         "state": "deferred",
+        "recovery_action": "reenter_guard_without_selection",
         "requested_todo_id": ALTERNATIVE_TODO_ID,
         "reason": "autonomous_replan",
         "delivery_preemptions": ["autonomous_replan", "delivery_not_allowed"],

tests/control_plane/test_selection_replan_reentry.py

@@ -0,0 +1,94 @@
+"""A deferred selection must expose a runnable recovery before settlement."""
+import shlex
+
+import pytest
+
+from test_quota_settlement_cli import (
+    AGENT_ID, GOAL_ID, SELECTED_REPLAN_TODO_ID, TODO_ID,
+    _configure_autonomous_replan_fixture, _configure_selected_todo_replan_fixture,
+    _configure_selectable_alternative, _heartbeat_receipt_count, _projected_cli_args,
+    _run_cli, _run_generated_cli, _spend_run_count, _write_fixture,
+)
+
+
+@pytest.mark.parametrize("binding", ["todo", "autonomous_replan"])
+def test_deferred_selection_recovers_same_turn_and_settles_once(tmp_path, binding):
+    project, runtime, registry = _write_fixture(tmp_path)
+    _configure_selectable_alternative(project)
+    turn = "turn-selection-preempted"
+    guard = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
+             "--agent-id", AGENT_ID, "--turn-instance-id", turn, "--scan-path", str(project))
+    rc, first = _run_cli(registry, runtime, *guard)
+    assert rc == 0 and first["decision"] == "run"
+    assert first["interaction_contract"]["cli_channel"]["selection_required"]
+    assert "settlement_identity" not in first["heartbeat_receipt"]
+    if binding == "todo":
+        _configure_selected_todo_replan_fixture(project, registry)
+        selected_id = SELECTED_REPLAN_TODO_ID
+    else:
+        _configure_autonomous_replan_fixture(project, runtime, registry)
+        selected_id = TODO_ID
+    rc, deferred = _run_cli(registry, runtime, *guard, "--todo-id", selected_id)
+    assert rc == 1 and not deferred["should_run"]
+    assert deferred["heartbeat_receipt"]["event_id"] == first["heartbeat_receipt"]["event_id"]
+    assert _heartbeat_receipt_count(runtime, turn) == 1
+    channel = deferred["interaction_contract"]["cli_channel"]
+    assert "settlement_plan" not in channel
+    assert "replan_settlement_contract" not in channel
+    [command] = channel["next_cli_actions"]
+    assert "rerun quota should-run" in deferred["recommended_action"]
+    assert deferred["execution_obligation"]["reason"] == deferred["recommended_action"]
+    assert deferred["interaction_contract"]["agent_channel"]["primary_action"] == command
+    rc, envelope = _run_cli(
+        registry, runtime, *guard, "--todo-id", selected_id, "--turn-envelope"
+    )
+    assert rc == 1, envelope
+    assert envelope["replan_action_packet"] is None
+    assert envelope["action"]["primary_action"].startswith("loopx ")
+    assert not envelope["action"]["must_attempt"]
+    assert not envelope["action"]["delivery_allowed"]
+    [recovery_preview] = envelope["writeback"]["next_cli_actions"]
+    assert recovery_preview.startswith("loopx ")
+    assert not envelope["writeback"]["spend_after_validation"]
+    assert _heartbeat_receipt_count(runtime, turn) == 1
+    argv = shlex.split(command)
+    assert argv[argv.index("--turn-instance-id") + 1] == turn
+    assert "--todo-id" not in argv and "--replan-obligation-id" not in argv
+    assert "should-run" in argv and "--codex-app" in argv
+    assert not deferred["interaction_contract"]["agent_channel"]["must_attempt"]
+    rc, resumed = _run_generated_cli(command, registry_path=registry)
+    assert rc == 0, resumed
+    receipt = resumed["heartbeat_receipt"]
+    assert receipt["status"] == "upgraded"
+    identity = receipt["settlement_identity"]
+    assert identity["turn_instance_id"] == turn
+    assert ("todo_id" in identity) == (binding == "todo")
+    assert _heartbeat_receipt_count(runtime, turn) == 2
+    cli = resumed["interaction_contract"]["cli_channel"]
+    assert cli["settlement_plan"]["identity"] == identity
+    refresh = next(c for c in cli["next_cli_actions"] if "refresh-state" in c)
+    for key, value in {"<advanced|blocked|exploration_exhausted|no_followup>": "advanced",
+                       "<surface-id>": "accepted-artifact", "<hypothesis-id>": "adoption",
+                       "<probe-kind>": "acceptance", "<evidence-id>": "evidence:readback"}.items():
+        refresh = refresh.replace(key, value)
+    rc, result = _run_cli(registry, runtime, *_projected_cli_args(refresh, turn_instance_id=turn))
+    assert rc == 0, result
+    assert result["settlement_result"]["ok"]
+    spend = next(c for c in cli["next_cli_actions"] if "spend-slot" in c)
+    spend_args = _projected_cli_args(spend, turn_instance_id=turn)
+    for replay in (False, True):
+        rc, result = _run_cli(registry, runtime, *spend_args, "--scan-path", str(project))
+        assert rc == 0, result
+        assert result["settlement_result"]["ok"]
+        if replay:
+            assert result["idempotent_replay"] and not result["appended"]
+    assert _spend_run_count(runtime) == 1
+    rc, settled = _run_cli(registry, runtime, *guard)
+    assert rc == 0
+    if binding == "autonomous_replan":
+        assert settled["effective_action"] == "heartbeat_settled_skip"
+    assert settled["heartbeat_receipt"]["settlement_identity"] == identity
+    rc, conflict = _run_cli(registry, runtime, *guard, "--todo-id", "todo_another_selection")
+    assert rc == 1 and conflict["ok"] is False
+    assert _heartbeat_receipt_count(runtime, turn) == 2
+    assert _spend_run_count(runtime) == 1

tests/control_plane_ts/action_portfolio.test.ts

@@ -351,6 +351,7 @@ test("pending selection qualifies only after current hard-lane arbitration", ()
   assert.deepEqual(deferred, {
     schema_version: "action_selection_qualification_v0",
     state: "deferred",
+    recovery_action: "reenter_guard_without_selection",
     requested_todo_id: successor.todo_id,
     reason: "blocking_work_lane",
     delivery_preemptions: ["blocking_work_lane"],
@@ -368,6 +369,7 @@ test("pending selection rejects a Todo absent from the current eligible set", ()
   }), {
     schema_version: "action_selection_qualification_v0",
     state: "rejected",
+    recovery_action: "reenter_guard_without_selection",
     requested_todo_id: "todo_missing001",
     reason: "candidate_not_currently_eligible",
   });
@@ -385,6 +387,7 @@ test("pending selection explains an auxiliary monitor outside the advancement la
   }), {
     schema_version: "action_selection_qualification_v0",
     state: "rejected",
+    recovery_action: "reenter_guard_without_selection",
     requested_todo_id: "todo_monitor001",
     reason: "auxiliary_monitor_not_selectable_in_advancement_lane",
   });