fix(quota): allow explicit independent work under scoped approval

docs/quota-allocation.md

@@ -868,6 +868,21 @@ including `generic_cli`, `outer_controller`, and native CLI profiles. An explici
 asking for the selection it just accepted. This does not enable `--begin-turn`
 for those hosts or change legacy calls without a Turn identity.
 
+Hosts that already support `--begin-turn` may combine it with `--todo-id` to
+qualify a known choice on the initial guard. Reuse the returned Turn identity
+for recovery; beginning a Turn does not claim a lease or bypass admission.
+With an open scoped user gate, an explicitly selected eligible action is checked
+against that gate directly, including choices outside the compact display.
+The default fallback ranking is not an authority boundary. Covered actions,
+budget/health stops, capability requirements and committed receipt identities
+retain their existing checks.
+
+已支持 `--begin-turn` 的宿主可以同时传入 `--todo-id`，在第一次 guard 中校验
+已选工作，减少一次选择调用。恢复时复用返回的 Turn 身份；开始 Turn 不等于
+取得租约或绕过准入。存在限定范围的用户审批时，显式选择的合格工作直接按该
+审批范围校验，即使它不在精简展示中。默认推荐顺序不构成权限边界；受审批
+约束的工作、预算和健康停止条件、能力要求及已绑定回执仍保留原有校验。
+
 A single already-claimed candidate keeps the direct execution path without an
 extra selection round trip. A sole unclaimed candidate instead projects the
 existing explicit selection contract: recommendation alone must not create a

loopx/cli_commands/quota_request.py

@@ -161,9 +161,9 @@ def validate_quota_command_request(args: argparse.Namespace) -> None:
         raise QuotaCommandValidationError(
             "`loopx quota void-slot` requires --void-generated-at"
         )
-    if command == "should-run" and args.todo_id and not args.turn_instance_id:
+    if command == "should-run" and args.todo_id and not (args.turn_instance_id or begin_turn):
         raise QuotaCommandValidationError(
-            "`loopx quota should-run --todo-id` requires --turn-instance-id"
+            "`loopx quota should-run --todo-id` requires --turn-instance-id or --begin-turn"
         )
     if begin_turn and command != "should-run":
         raise QuotaCommandValidationError(
@@ -173,11 +173,6 @@ def validate_quota_command_request(args: argparse.Namespace) -> None:
         raise QuotaCommandValidationError(
             "--begin-turn cannot be combined with --turn-instance-id"
         )
-    if begin_turn and args.todo_id:
-        raise QuotaCommandValidationError(
-            "--begin-turn cannot select --todo-id; reuse the exact "
-            "--turn-instance-id returned by the initial guard"
-        )
     if begin_turn and not args.agent_id:
         raise QuotaCommandValidationError(
             "--begin-turn requires exact --agent-id identity"

loopx/control_plane/agents/agent_scope.py

@@ -217,6 +217,8 @@ def _scoped_user_gate_fallback(
     capability_gate: dict[str, Any] | None = None,
     allow_unrelated_gate: bool = False,
     monitor_debt_backoff_active: bool = False,
+    requested_todo_id: str | None = None,
+    requested_candidate: dict[str, Any] | None = None,
 ) -> dict[str, Any] | None:
     gates = _open_user_gate_todo_items(user_todo_summary)
     if not gates or not isinstance(agent_todo_summary, dict):
@@ -258,6 +260,10 @@ def _scoped_user_gate_fallback(
             *ready_deferred_candidates,
         ]
     executable_items = [item for item in executable_items if isinstance(item, dict)]
+    # Exact selection was qualified from the complete inventory. Display lanes
+    # may omit it; transport the original row so gate metadata is not lost.
+    if requested_candidate is not None:
+        executable_items = [requested_candidate, *executable_items]
     claim_scope = (
         agent_todo_summary.get("claim_scope")
         if isinstance(agent_todo_summary.get("claim_scope"), dict)
@@ -266,6 +272,7 @@ def _scoped_user_gate_fallback(
     selection = select_scoped_gate_fallback(
         gates, executable_items, agent_id=claim_scope.get("agent_id") if claim_scope else None,
         allow_unrelated_gate=allow_unrelated_gate, monitor_debt_backoff_active=monitor_debt_backoff_active,
+        requested_todo_id=requested_todo_id,
     )
     if selection is None:
         return None

loopx/control_plane/quota/should_run_prepare.py

@@ -740,12 +740,30 @@ def _prepare_quota_should_run_item(
     if task_orchestration_contract_is_actionable(task_orchestration_contract):
         capability_monitor_contract = capability_monitor_fallback = None
     work_lane_contract = capability_monitor_contract or work_lane_contract
+    # Resolve the exact choice before selecting a gate-independent route. The
+    # ranked fallback is advice, not an additional permission boundary.
+    requested_action_candidate = (
+        build_explicit_advancement_next_action(
+            agent_identity=agent_identity,
+            agent_todo_items=agent_todo_planning_source_items,
+            available_capabilities=effective_available_capabilities,
+            todo_id=requested_action_todo_id,
+            selection_binding="pending_action_selection",
+        )
+        if requested_action_todo_id
+        else None
+    )
     scoped_user_gate_fallback = _scoped_user_gate_fallback(
         user_todo_summary,
         agent_todo_summary,
         capability_gate=capability_gate,
         allow_unrelated_gate=bool(quota.get("safe_bypass_allowed")),
         monitor_debt_backoff_active=bool(monitor_debt_arbitration.get("active")),
+        requested_todo_id=requested_action_todo_id,
+        requested_candidate=next((
+            item for item in agent_todo_planning_source_items
+            if item.get("todo_id") == requested_action_todo_id
+        ), None) if requested_action_candidate is not None else None,
     )
     work_lane_contract = (
         scoped_user_gate_due_monitor_contract(
@@ -912,19 +930,6 @@ def _prepare_quota_should_run_item(
         recovery_allowed = False
         reason = str(projection_gap_repair.get("reason") or reason)
     boundary_projection_repair = None
-    # Resolve exact identity before Agent/display compaction. Live callers supply
-    # the complete source; pure status callers use only their supplied snapshot.
-    requested_action_candidate = (
-        build_explicit_advancement_next_action(
-            agent_identity=agent_identity,
-            agent_todo_items=agent_todo_planning_source_items,
-            available_capabilities=effective_available_capabilities,
-            todo_id=requested_action_todo_id,
-            selection_binding="pending_action_selection",
-        )
-        if requested_action_todo_id
-        else None
-    )
     state_path = goal_state_path(registry_goal)
     return _QuotaDecisionPreparation(
         status_payload=status_payload,

loopx/control_plane/todos/decision_scope.py

@@ -317,7 +317,8 @@ def todo_gate_relations(gates: list[dict[str, Any]], items: list[dict[str, Any]]
 
 def select_scoped_gate_fallback(gates: list[dict[str, Any]], items: list[dict[str, Any]], *,
                                agent_id: str | None, allow_unrelated_gate: bool,
-                               monitor_debt_backoff_active: bool) -> dict[str, Any] | None:
+                               monitor_debt_backoff_active: bool,
+                               requested_todo_id: str | None = None) -> dict[str, Any] | None:
     """Decode legacy facts; the typed owner returns positions, never display rows."""
     if not gates or not items:
         return None
@@ -336,6 +337,7 @@ def facts(item: dict[str, Any]) -> dict[str, Any]:
         "fallback",
         _evaluate("fallback", gates=[facts(gate) for gate in gates], candidates=[facts(item) for item in items],
                   agent_id=normalize_todo_claimed_by(agent_id), allow_unrelated_gate=allow_unrelated_gate,
+                  requested_todo_id=normalize_todo_id(requested_todo_id),
                   monitor_debt_backoff_active=monitor_debt_backoff_active),
         schema_versions=frozenset({"scoped_gate_fallback_selection_v0"}),
         nullable=True,

loopx/control_plane/todos/decision_scope.ts

@@ -97,6 +97,7 @@ export function fallbackGateRelation(gate: JsonObject, item: JsonObject): JsonOb
  * Return source positions, leaving display compaction and wording to adapters. */
 export function selectScopedGateFallback(request: JsonObject): JsonObject | null {
   const agent = optionalNonEmptyString(request.agent_id, "agent_id");
+  const requested = optionalNonEmptyString(request.requested_todo_id, "requested_todo_id");
   const debt = requireBoolean(request.monitor_debt_backoff_active, "monitor_debt_backoff_active");
   const allowUnrelated = requireBoolean(request.allow_unrelated_gate, "allow_unrelated_gate");
   const gates = rows(request.gates).map((gate, index) => ({gate, index})).filter(({gate}) =>
@@ -129,7 +130,7 @@ export function selectScopedGateFallback(request: JsonObject): JsonObject | null
     if (match) {
       blockingGate ??= match;
       blocked.push({candidate_index: candidate.index, gate_index: match.index, relation: match.relation});
-    } else selected ??= candidate;
+    } else if (!requested || candidate.item.todo_id === requested) selected ??= candidate;
   }
   if (!selected || (!blockingGate && !allowUnrelated)) return null;
   const surface = blockingGate ?? gates[0]!;

tests/control_plane/test_quota_settlement_cli.py

@@ -6313,6 +6313,7 @@ def test_pending_action_selection_does_not_commit_after_new_user_gate(
         "--priority", "P0", "--task-class", "user_gate",
         "--action-kind", "approve_delivery", "--blocks-agent", AGENT_ID,
         "--agent-id", AGENT_ID, "--decision-scope", "direction:action:fixture",
+        "--unblocks-todo-id", ALTERNATIVE_TODO_ID,
     )
     assert added_rc == 0, added
     selected_rc, selected = _run_cli(