fix(quota): recover undeclared capabilities before fallback selection

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -636,6 +636,18 @@ read or resume evaluation is added. Remaining T3 work includes consumers that
 reconstruct diagnostics from compact summaries; do not call those migrated.
 This does not close T1/T2, all T3 consumers, or any durability/promotion hold.
 
+Runtime capability re-entry now uses that same TS owner for verification-target selection,
+owner-authority exclusion, advisory versus bound Turn handling, and the recovery contract
+without durable grants. Python removes the former target lookup/filter rules and only adapts
+host/scheduler facts, calls one typed reducer, and renders shell argv. One interaction packet
+reuses the result; a healthy path adds no runtime call. The intentional correction is that an
+eligible fallback recommendation cannot hide a blocked task's real capability check before
+explicit selection. Success re-enters the same Turn; failure still allows explicit fallback
+selection. A committed receipt's Todo remains bound. This is a host-local read plan under
+section 3 of the shared-authority RFC, not a capability lease, shared grant, or provider write.
+CLI/managed Turn reuse the existing re-entry fields; the generated `/loopx` skill requires
+checking missing declarations. No frontend configuration or second UI state is introduced.
+
 Advancement-frontier checkpoint closure: `todos/frontier_revision.ts` now owns
 agent selection, completeness, material hashing, long-chain thresholds and exact
 ACK/rearm classification. Python retains the v0 field manifest and legacy JSON/

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -486,6 +486,16 @@ Todo，同一 Todo 的不同展示不重复计算，权威空 backlog 不再复
 target capability 是修复产出，不是安装或授权。没有新 provider／inventory／enablement／
 promotion；压缩候选来源的上限和其余 T3 consumer 仍需分别闭合。
 
+运行时能力重入现在复用同一个 TS owner：验证目标选择、owner 权限排除、推荐与已绑定
+Turn 的区分，以及无持久授权的恢复合同由 `agents/capability_gate.ts` 负责。Python 删除
+旧目标查找与过滤规则，仅适配 host/scheduler 参数、调用一次 typed reducer 并渲染 shell
+argv；同一 interaction packet 复用结果，健康路径不增加 runtime 调用。修正行为是：显式
+选择前，可执行的低优先级推荐不能隐藏受阻任务的真实能力验证；验证成功在原 Turn 重入，
+失败后仍可显式选择其他工作。已提交 receipt 的 Todo 不变。该观察属于 host-local read
+plan，遵循 shared-authority RFC 第 3 节边界；不增加 capability lease、共享 grant 或
+provider 写入。CLI/managed Turn 复用已有重入字段，生成的 `/loopx` skill 明确要求核对
+缺失声明；没有新增前端配置或第二份 UI 状态。
+
 Advancement-frontier checkpoint 闭合：`todos/frontier_revision.ts` 现统一 Agent
 选择、完整度、实质内容哈希、长链阈值与精确 ACK/rearm 分类。Python 保留 v0 字段清单
 与 legacy JSON/metadata codec，使合法且未变化的 frontier 保持已有指纹；删除旧 Python

loopx/control_plane/agents/capability_gate.ts

@@ -1,6 +1,6 @@
 /** Read policy only: requirements are not enablement, credentials or write authority. */
 import type {JsonObject} from "../effect_program.ts";
-import {requireJsonObject, requireStringArray, requireInteger, requireNonEmptyString} from "../runtime_decode.ts";
+import {requireJsonObject, requireStringArray, requireInteger, requireNonEmptyString, requireBoolean, optionalNonEmptyString} from "../runtime_decode.ts";
 
 const DEFAULT_AVAILABLE = ["shell", "filesystem_read", "filesystem_write"];
 const OWNER_HELD = new Set(["credentials", "production_access"]);
@@ -112,10 +112,76 @@
     reason: "all visible executable todo candidates require unavailable capabilities"};
 }
 
+/** Host-local read plan: never writes a grant or changes a committed Turn binding. */
+export function projectRuntimeCapabilityReentry(request: JsonObject): JsonObject | null {
+  const gate = requireJsonObject(request.gate, "gate");
+  const observed = unique(requireStringArray(request.available, "available").filter(c => !OWNER_HELD.has(c)));
+  const selectionRequired = requireBoolean(request.selection_required, "selection_required");
+  const selectedId = optionalNonEmptyString(request.selected_todo_id, "selected_todo_id");
+  const receiptId = optionalNonEmptyString(request.receipt_todo_id, "receipt_todo_id");
+  // A receipt remains binding even if an inconsistent advisory flag accompanies it.
+  const boundId = receiptId ?? (selectionRequired ? null : selectedId);
+  const baseArgs = requireStringArray(request.command_prefix, "command_prefix");
+  const schedulerArgs = requireStringArray(request.scheduler_args, "scheduler_args");
+  const missing = requireStringArray(gate.repair_missing ?? [], "repair_missing")
+    .filter(c => !OWNER_HELD.has(c) && !observed.includes(c));
+  const resolutions = list(gate.resolution_bindings ?? []).map(value => {
+    const row = requireJsonObject(value, "resolution binding");
+    return {
+      owner: requireNonEmptyString(row.owner, "owner"),
+      capability: requireNonEmptyString(row.capability, "capability"),
+      primary: optionalNonEmptyString(row.primary_blocked_todo_id, "primary_blocked_todo_id"),
+      ids: requireStringArray(row.blocked_todo_ids ?? [], "blocked_todo_ids"),
+    };
+  });
+  const blocked = list(gate.blocked_candidates ?? []).map(value => {
+    const row = requireJsonObject(value, "blocked candidate");
+    return {
+      id: optionalNonEmptyString(row.todo_id, "todo_id"),
+      instruction: optionalNonEmptyString(row.text, "text"),
+      action: optionalNonEmptyString(row.action_kind, "action_kind") ?? "unspecified",
+      target: optionalNonEmptyString(row.target_key, "target_key"),
+      required: requireStringArray(row.required_capabilities ?? [], "required_capabilities"),
+    };
+  });
+  if (!schedulerArgs.length) return null;
+  const candidates: JsonObject[] = [];
+  for (const capability of unique(missing)) {
+    const bindings = resolutions.filter(row => row.owner === "agent" && row.capability === capability);
+    const ids = new Set(bindings.flatMap(row => row.ids.length ? row.ids : row.primary ? [row.primary] : []));
+    const eligible = blocked.filter(row => row.id && ids.has(row.id) && row.instruction &&
+      row.required.includes(capability) && (!boundId || row.id === boundId));
+    // Prefer the gate's highest-priority target, not incidental display order.
+    const target = eligible.find(row => bindings.some(binding => binding.primary === row.id)) ?? eligible[0];
+    if (!target) continue;
+    candidates.push({
+      capability, verification_required: "successful_real_callsite_observation",
+      verification_target: {todo_id: target.id!, action_kind: target.action, instruction: target.instruction!,
+        ...(target.target ? {target_ref: target.target} : {})},
+      command_argv: [...baseArgs, ...observed.flatMap(c => ["--available-capability", c]),
+        "--available-capability", capability, ...schedulerArgs],
+    });
+  }
+  if (!candidates.length) return null;
+  return {
+    schema_version: "runtime_capability_reentry_v0", state: "verification_required",
+    source: "quota_should_run.capability_gate.repair_missing", candidates,
+    verification_contract: {scope: "real_task_facing_callsite_for_blocked_todo", ordinary_delivery_allowed: false,
+      advancement_checkpoint: false, settles_turn: false,
+      on_success: "rerun_quota_in_same_turn_then_continue_if_allowed",
+      on_failure: "record_exact_blocker_without_capability_flag"},
+    inheritance_contract: {source_invocation: "verified quota should-run reentry",
+      propagates_to: ["interaction_contract.cli_channel.next_cli_actions", "quota spend-slot", "quota monitor-poll"],
+      session_scoped: true, durable_grant_written: false},
+    failure_policy: "Do not add the capability flag when the real callsite check fails; continue the capability repair or record the concrete blocker.",
+  };
+}
+
 export function evaluateCapabilityGate(value: unknown): JsonObject {
   const request = requireJsonObject(value, "capability gate request");
   if (request.schema_version !== "capability_gate_request_v0") throw new TypeError("capability gate request schema mismatch");
   if (request.operation === "project") return {schema_version: "capability_gate_result_v0", result: projectCapabilityGate(request)};
+  if (request.operation === "reentry") return {schema_version: "capability_gate_result_v0", result: projectRuntimeCapabilityReentry(request)};
   if (request.operation === "missing") {
     const available = requireStringArray(request.available, "available");
     return {schema_version: "capability_gate_result_v0", result: list(request.items).map(item => {

loopx/control_plane/work_items/interaction_contract.py

@@ -40,6 +40,7 @@
 )
 from .accountable_settlement import build_accountable_work_item_settlement_plan
 from . import action_selection_contract as selection
+from . import runtime_capability_reentry as capability_reentry_adapter
 from .primary_action import (
     build_primary_action_projection,
     protocol_action_label as _protocol_action_label,
@@ -48,7 +49,6 @@
     protocol_monitor_action as _protocol_monitor_action,
 )
 from .replan_settlement import project_replan_settlement_contract
-from .runtime_capability_reentry import build_runtime_capability_reentry_packet
 from .user_action_frontier import user_action_owns_empty_agent_lane
 
 INTERACTION_CONTRACT_SCHEMA_VERSION = "loopx_interaction_contract_v0"
@@ -667,6 +667,7 @@ def interaction_next_cli_actions(
         Mapping[str, Any] | SchedulerExecutionContextResolution | None
     ) = None,
     capability_reentry: dict[str, Any] | None = None,
+    capability_reentry_resolved: bool = False,
     settlement_plan: Mapping[str, Any] | None = None,
     turn_instance_id: str | None = None,
     runtime_root: str | None = None,
@@ -763,8 +764,8 @@ def interaction_next_cli_actions(
             else None
         ),
     )
-    if capability_reentry is None:
-        capability_reentry = build_runtime_capability_reentry_packet(
+    if capability_reentry is None and not capability_reentry_resolved:
+        capability_reentry = capability_reentry_adapter.build_runtime_capability_reentry_packet(
             payload,
             available_capabilities=available_capabilities,
             scheduler_execution_context=scheduler_execution_context,
@@ -1172,22 +1173,11 @@ def _build_interaction_agent_channel(
         channel["action_portfolio_ref"] = "$.action_portfolio"
     selection.apply_action_selection_agent_gate(channel, payload)
     if capability_reentry is not None:
-        candidate = capability_reentry["candidates"][0]
-        target = candidate["verification_target"]
-        channel["next_task_action"] = {
-            "kind": "capability_verification",
-            "capability": candidate["capability"],
-            "todo_id": target["todo_id"],
-            "action_kind": target["action_kind"],
-            "operation": target["action_kind"],
-            "instruction": target["instruction"],
-            "preflight_allowed": False,
-            "advancement_checkpoint": False,
-            "settles_turn": False,
-            "continuation_cli_action_index": 0,
-        }
-        if target.get("target_ref"):
-            channel["next_task_action"]["target_ref"] = target["target_ref"]
+        capability_reentry_adapter.apply_agent_channel_projection(
+            channel,
+            capability_reentry,
+            selection_required=selection.action_portfolio_requires_explicit_selection(payload),
+        )
     if _blocked_successor_wait_observation_required(payload):
         channel["primary_action"] = (
             "record one no-spend blocked-successor wait observation, rerun quota, "
@@ -1269,14 +1259,6 @@ def _build_interaction_cli_channel(
     runtime_root: str | None = None,
 ) -> dict[str, Any]:
     spend_after_selection = selection.delivery_spend_allowed(payload, spend_after_validation)
-    if capability_reentry is None:
-        capability_reentry = build_runtime_capability_reentry_packet(
-            payload,
-            available_capabilities=available_capabilities,
-            scheduler_execution_context=scheduler_execution_context,
-            turn_instance_id=turn_instance_id,
-            runtime_root=runtime_root,
-        )
     settlement_plan, replan_settlement_contract = (
         _turn_scoped_cli_settlement_context(
             payload,
@@ -1293,6 +1275,7 @@ def _build_interaction_cli_channel(
             available_capabilities=available_capabilities,
             scheduler_execution_context=scheduler_execution_context,
             capability_reentry=capability_reentry,
+            capability_reentry_resolved=True,
             settlement_plan=settlement_plan,
             turn_instance_id=turn_instance_id,
             runtime_root=runtime_root,
@@ -1312,7 +1295,11 @@ def _build_interaction_cli_channel(
     if settlement_plan is not None and replan_settlement_contract is not None:
         channel["replan_settlement_contract"] = replan_settlement_contract
     if capability_reentry is not None:
-        channel["runtime_capability_reentry"] = capability_reentry
+        capability_reentry_adapter.apply_cli_channel_projection(
+            channel,
+            capability_reentry,
+            selection_required=selection.action_portfolio_requires_explicit_selection(payload),
+        )
     selected_todo = (
         payload.get("selected_todo")
         if isinstance(payload.get("selected_todo"), Mapping)
@@ -1481,7 +1468,7 @@ def build_interaction_contract(
         and todo_lifecycle_settlement_obligation(payload) is None
     )
     required_reads = _interaction_required_reads(payload)
-    capability_reentry = build_runtime_capability_reentry_packet(
+    capability_reentry = capability_reentry_adapter.build_runtime_capability_reentry_packet(
         payload,
         available_capabilities=available_capabilities,
         scheduler_execution_context=scheduler_execution_context,