fix(quota): preserve explicit Todo selection gates and retry

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

@@ -71,6 +71,33 @@ only a qualified request upgrades the identity-less receipt. A newly due hard
 lane leaves the receipt unbound, and only the resulting receipt-bound envelope
 is a delivery contract.
 
+An explicit selection that is not admitted returns the shared TypeScript
+`action_selection_qualification_v0` result as `action_selection` in the quota
+packet. `--turn-envelope` also retains this typed failed-preflight packet,
+including its exact re-entry command, instead of rendering an executable Turn.
+`quota_action_selection_rejected`
+means the requested candidate is not currently eligible;
+`quota_action_selection_deferred` preserves the current preemption reason,
+such as `autonomous_replan`. A displayed runnable Todo alone does not override
+`normal_delivery_allowed=false`. These are preflight outcomes, not heartbeat
+receipt identity conflicts: no receipt is created or upgraded, and no quota is
+spent. The returned `recommended_action` and `next_cli_actions[0]` re-enter the
+current guard with the same registry, runtime, Goal, Agent, Turn and scheduler
+context, without the refused `--todo-id`. Follow that guard's actual obligation
+before retrying selection. Repeated unsuccessful preflight leaves existing
+receipts unchanged; a genuinely conflicting committed identity still fails
+with `heartbeat_receipt_identity_conflict`.
+
+显式选择未获准时，quota 的 `action_selection` 保留 TypeScript 类型化结果；
+`--turn-envelope` 同样返回这个失败预检载荷，保留完整重入命令，不渲染可执行 Turn。
+其中 `quota_action_selection_rejected` 表示当前候选不满足资格；`quota_action_selection_deferred` 保留 `autonomous_replan` 等
+真实抢占原因。列表中显示可执行 Todo，并不能覆盖 `normal_delivery_allowed=false`。
+此时不创建或升级回执、不扣额度，也不伪报回执身份冲突。返回命令保留 registry、
+runtime、Goal、Agent、Turn 和调度上下文，去掉被拒的 `--todo-id`，先重新进入 guard
+并处理其真实义务，再重试选择。重复失败不改变已有回执；已提交身份的真实冲突仍
+返回 `heartbeat_receipt_identity_conflict`。能力验证通过后的重试仍需重新声明实际
+可用的能力，不因错误诊断或列表显示获得额外权限。
+
 Portfolio v2 preserves v1's selection policy, candidate ordering, and
 settlement rules, and adds an optional `continuation_hint` to each suggested
 action. The default quota producer and Turn controller now require v2. The

loopx/cli_commands/quota.py

@@ -20,6 +20,7 @@
 from ..control_plane.quota.effect_program import SettlementIdentity
 from ..control_plane.quota.error_codes import (
     HeartbeatReceiptIdentityConflictError,
+    QuotaActionSelectionNotAdmitted,
     QuotaCommandValidationError,
     QuotaIdentityPreconditionError,
     quota_error_code,
@@ -77,6 +78,11 @@
     build_lark_operator_inbox_urgency_projector,
     dispatch_goal_lark_turn_start_hooks,
 )
+from .quota_action_selection import (
+    commit_requested_action_selection,
+    reject_action_selection,
+    require_requested_quota_action_selection,
+)
 from .quota_context import (
     QuotaCommandContext,
     prepare_quota_command_context,
@@ -379,79 +385,6 @@ def _heartbeat_quota_action_selection_bindings(
     return existing, todo_id, replan_obligation_id
 
 
-def _require_requested_quota_action_selection(
-    payload: Mapping[str, object],
-    *,
-    requested_todo_id: str | None,
-    receipt_bound_todo_id: str | None,
-    receipt_bound_replan_obligation_id: str | None,
-) -> None:
-    if not requested_todo_id or (
-        receipt_bound_todo_id or receipt_bound_replan_obligation_id
-    ):
-        return
-    selected_todo = payload.get("selected_todo")
-    selected_todo_id = (
-        normalize_todo_id(selected_todo.get("todo_id"))
-        if isinstance(selected_todo, 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 {}
-    )
-    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_qualified = (
-        selection_binding == "pending_action_selection"
-        and payload.get("normal_delivery_allowed") is True
-    )
-    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
-        or payload.get("ok") is not True
-        or payload.get("should_run") is not True
-        or not (pending_selection_qualified or exact_current_obligation_qualified)
-    ):
-        raise HeartbeatReceiptIdentityConflictError(
-            "explicit action selection must name one currently projected "
-            "agent-scoped, capability-ready Todo"
-        )
-
-
-def _commit_requested_action_selection(
-    payload: Mapping[str, object],
-    *,
-    requested_todo_id: str | None,
-) -> None:
-    """Project the exact requested selection only after receipt reconciliation."""
-
-    selected_todo = payload.get("selected_todo")
-    if (
-        requested_todo_id
-        and isinstance(selected_todo, dict)
-        and normalize_todo_id(selected_todo.get("todo_id")) == requested_todo_id
-    ):
-        selected_todo["selection_binding"] = "heartbeat_receipt"
-
-
 def _dispatch_quota_turn_start_hooks(
     args: argparse.Namespace,
     *,
@@ -509,7 +442,6 @@ def _attach_turn_start_hook_dispatch(
         payload["turn_start_capability_hook_dispatch"] = dict(dispatch)
 
 
-
 def _render_turn_envelope_payload(
     payload: dict[str, object],
     scheduler_context: object,
@@ -521,6 +453,13 @@ def _render_turn_envelope_payload(
     renderer rejection keeps the typed diagnostic itself (with the skip reason)
     instead of masking it with a crash (issue #3687).
     """
+    if payload.get("error_code") in {
+        "quota_action_selection_rejected", "quota_action_selection_deferred",
+    }:
+        # This is a failed preflight, not an executable Turn. Keep its typed
+        # reason and exact re-entry command rather than truncating the command
+        # or replacing it with the unbound replan's settlement actions.
+        return payload
     try:
         return build_turn_envelope(
             payload,
@@ -531,6 +470,7 @@ def _render_turn_envelope_payload(
         degraded["turn_envelope_skipped"] = str(envelope_error)[:200]
         return degraded
 
+
 def handle_quota_command(
     args: argparse.Namespace,
     *,
@@ -547,6 +487,7 @@ def handle_quota_command(
     heartbeat_stall_observation = "not_evaluated"
     detail_sections: frozenset[str] = frozenset()
     context: QuotaCommandContext | None = None
+    selection_not_admitted = False
     try:
         turn_start_hook_dispatch, turn_start_mutated = _dispatch_quota_turn_start_hooks(
             args,
@@ -636,7 +577,7 @@ def handle_quota_command(
                 turn_start_hook_dispatch=turn_start_hook_dispatch,
             )
             _attach_turn_start_hook_dispatch(payload, turn_start_hook_dispatch)
-            _require_requested_quota_action_selection(
+            require_requested_quota_action_selection(
                 payload,
                 requested_todo_id=_requested_quota_action_todo_id(args),
                 receipt_bound_todo_id=receipt_bound_todo_id,
@@ -797,6 +738,13 @@ def handle_quota_command(
             payload = build_quota_plan(status_payload, mode=args.quota_command)
         if cache_metadata:
             payload["status_projection_cache"] = cache_metadata
+    except QuotaActionSelectionNotAdmitted:
+        # The preflight did not accept any Todo or replan settlement identity.
+        selection_not_admitted = True
+        assert context is not None
+        payload = reject_action_selection(
+            payload, args=args, registry_path=registry_path, context=context,
+        )
     except QuotaCommandValidationError as exc:
         # Only typed CLI validation diagnostics are public-safe by contract.
         payload = _quota_validation_failure_payload(
@@ -812,7 +760,7 @@ def handle_quota_command(
             runtime_root_arg=runtime_root_arg,
             error=exc,
         )
-    if _should_log_quota(args.quota_command, payload):
+    if not selection_not_admitted and _should_log_quota(args.quota_command, payload):
         spend_turn_instance_id = _effective_spend_turn_instance_id(
             payload,
             heartbeat_turn_id=heartbeat_turn_id,
@@ -846,7 +794,7 @@ def handle_quota_command(
                     status=heartbeat_receipt_existing_status,
                     appended=heartbeat_receipt_existing_appended,
                 )
-                _commit_requested_action_selection(
+                commit_requested_action_selection(
                     payload,
                     requested_todo_id=_requested_quota_action_todo_id(args),
                 )
@@ -913,7 +861,7 @@ def handle_quota_command(
                         if rollout_event.get("appended")
                         else "replayed",
                     )
-                    _commit_requested_action_selection(
+                    commit_requested_action_selection(
                         payload,
                         requested_todo_id=_requested_quota_action_todo_id(args),
                     )

loopx/cli_commands/quota_action_selection.py

@@ -0,0 +1,161 @@
+"""CLI transport for typed explicit selection admission and receipt recovery."""
+
+from __future__ import annotations
+
+import argparse
+import shlex
+from collections.abc import Mapping
+from pathlib import Path
+
+from ..control_plane.quota.error_codes import (
+    HeartbeatReceiptIdentityConflictError,
+    QuotaActionSelectionNotAdmitted,
+)
+from ..control_plane.scheduler.execution_context import render_scheduler_execution_args
+from ..control_plane.todos.contract import normalize_todo_id
+from .quota_context import QuotaCommandContext
+
+
+def require_requested_quota_action_selection(
+    payload: dict[str, object],
+    *,
+    requested_todo_id: str | None,
+    receipt_bound_todo_id: str | None,
+    receipt_bound_replan_obligation_id: str | None,
+) -> None:
+    if not requested_todo_id or (
+        receipt_bound_todo_id or receipt_bound_replan_obligation_id
+    ):
+        return
+    qualification = payload.get("action_selection")
+    selected_todo = payload.get("selected_todo")
+    selected_todo_id = (
+        normalize_todo_id(selected_todo.get("todo_id"))
+        if isinstance(selected_todo, 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 {}
+    )
+    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_qualified = (
+        selection_binding == "pending_action_selection"
+        and payload.get("normal_delivery_allowed") is True
+    )
+    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
+        or payload.get("ok") is not True
+        or payload.get("should_run") is not True
+        or not (pending_selection_qualified or exact_current_obligation_qualified)
+    ):
+        if isinstance(qualification, Mapping) and qualification.get("state") in {
+            "rejected",
+            "deferred",
+        }:
+            raise QuotaActionSelectionNotAdmitted(str(qualification["reason"]))
+        raise HeartbeatReceiptIdentityConflictError(
+            "explicit action selection must name one currently projected "
+            "agent-scoped, capability-ready Todo"
+        )
+    if exact_current_obligation_qualified:
+        # Exact due-monitor selection uses the existing obligation route,
+        # rather than the advancement-only candidate qualifier.
+        payload.pop("action_selection", None)
+
+
+def commit_requested_action_selection(
+    payload: Mapping[str, object],
+    *,
+    requested_todo_id: str | None,
+) -> None:
+    """Project the exact requested selection only after receipt reconciliation."""
+
+    selected_todo = payload.get("selected_todo")
+    if (
+        requested_todo_id
+        and isinstance(selected_todo, dict)
+        and normalize_todo_id(selected_todo.get("todo_id")) == requested_todo_id
+    ):
+        selected_todo["selection_binding"] = "heartbeat_receipt"
+
+
+def reject_action_selection(
+    payload: dict[str, object],
+    *,
+    args: argparse.Namespace,
+    registry_path: Path,
+    context: QuotaCommandContext,
+) -> dict[str, object]:
+    """Keep the failed preflight typed and re-enter before any settlement."""
+    qualification = payload["action_selection"]
+    payload.update(
+        ok=False,
+        should_run=False,
+        normal_delivery_allowed=False,
+        error_code=f"quota_action_selection_{qualification['state']}",
+        reason=str(qualification["reason"]),
+    )
+    command = shlex.join(
+        [
+            "loopx",
+            "--registry",
+            str(registry_path),
+            "--runtime-root",
+            str(context.runtime_root),
+            "--format",
+            "json",
+            "quota",
+            "should-run",
+            "--goal-id",
+            args.goal_id,
+            "--agent-id",
+            args.agent_id,
+            *(
+                ["--turn-instance-id", context.heartbeat_turn_id]
+                if context.heartbeat_turn_id
+                else []
+            ),
+            *[
+                token
+                for capability in (args.available_capabilities or [])
+                for token in ("--available-capability", capability)
+            ],
+        ]
+    ) + render_scheduler_execution_args(
+        scheduler_execution_context=context.scheduler_context
+    )
+    payload["recommended_action"] = command
+    payload.pop("selected_todo", None)
+    payload.pop("action_portfolio", None)
+    payload["interaction_contract"]["cli_channel"] = {
+        "next_cli_actions": [command],
+        "spend_allowed_now": False,
+        "spend_after_validation": False,
+        "spend_policy": "re-enter the current guard before delivery or settlement",
+    }
+    payload["interaction_contract"]["agent_channel"].update(
+        delivery_allowed=False,
+        selection_required=False,
+        primary_action="re-enter the current guard before selecting work",
+    )
+    return payload

loopx/control_plane/quota/error_codes.py

@@ -12,6 +12,10 @@ class HeartbeatReceiptIdentityConflictError(ValueError):
     """Public-safe diagnostic for a same-turn settlement identity conflict."""
 
 
+class QuotaActionSelectionNotAdmitted(ValueError):
+    """A typed selection preflight result, before any receipt is committed."""
+
+
 class QuotaIdentityPrecondition(StrEnum):
     """Typed identity admission preconditions for scoped quota decisions."""