fix(quota): project scoped override actions from final admission

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

@@ -1151,6 +1151,15 @@ debit. This closes the demonstrated T3 consumer gap, not D1–D3, provider
 promotion, or the remaining Python transaction adapters. See the
 [operating contract](../../quota-allocation.md#receipt-backed-settlement-progress).
 
+**Scoped gate action readback.** The final quota packet now projects the scoped
+User gate/action override through a typed quota rule after selection, capability,
+workspace, receipt and notification decisions. Its optional `selected_action`
+comes only from the final selected Todo when the interaction allows delivery;
+selection-required, repair and settled packets omit it. Admission diagnostics
+and receipt identity remain intact. This corrects misleading CLI JSON readback
+for peer-scoped gates; it does not migrate the remaining route or primary-action
+builders, reserve recommendations, or complete T3/D1–D3.
+
 **Long-history transport boundary.** Replan history still has one TS decision
 owner. Small requests retain the inline codec; larger complete fact snapshots
 travel through a private, digest-bound local file reference. The same reducer

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

@@ -838,6 +838,8 @@ T3/D1 reader，未完成全部 Todo writer、retention/compaction 或 promotion
 
 配额准入与结算消费者现在从统一 Todo reader 读取完整来源，在显示压缩前解析显式 Todo 选择。它删除直接追加 Markdown 候选的路径，保留 promote 前的事件适配；promote 后权威为空或不可读都不能复活展示行。结算进度由现有 TS 回执链归约，Python 负责完整身份命令及 JSON/Markdown 展示。现有幂等 writer 可补齐缺失的 spend 回执而不再次扣款。这关闭已复现的 T3 消费者缺口，不代表 D1–D3、provider promotion 或剩余 Python 事务适配已完成。操作语义见[结算进度契约](../../quota-allocation.md#receipt-backed-settlement-progress)。
 
+**Scoped gate 动作回读。** 最终 quota 包在选择、能力、workspace、回执及通知决策之后，通过 TS quota 规则投影 scoped User gate/action override。可选 `selected_action` 只在 interaction 允许交付时取最终选中 Todo 的文字；待选择、修复和已结算包省略该字段。准入诊断与回执身份保留。这修正 peer gate 场景的 CLI JSON 误导，不代表剩余 route／primary-action builder 已迁移、推荐已成为预留，或 T3/D1–D3 已完成。
+
 **长历史传输边界。** Replan 历史仍由一个 TS owner 决策。小请求保留 inline
 codec；较大的完整事实快照通过私有临时文件和摘要绑定的引用传递。同一 reducer
 校验全部记录、agent 作用域内的 ACK 及重试身份；RPC 预算和展示窗口都不允许截断

loopx/control_plane/effect_runtime_handlers.ts

@@ -87,6 +87,7 @@ import {
   preflightPriorHostTurnCloseout,
   reduceUnsettledHostTurnRecovery,
 } from "./quota/unsettled_host_turn_recovery.ts";
+import { projectScopedOverride } from "./quota/scoped_override.ts";
 import { evaluateTurnEnvelope } from "./quota/turn_envelope.ts";
 import { evaluateQuotaMonitorPollCommit } from "./quota/monitor_poll_commit.ts";
 import { planMonitorSuccessor, selectMonitorTodoRequest } from "./scheduler/monitor_successor.ts";
@@ -577,6 +578,7 @@ export function createEffectRuntimeHandlers(
       reduceUnsettledHostTurnRecovery,
     ],
     ["quota.turn_envelope.evaluate", evaluateTurnEnvelope],
+    ["quota.scoped_override.project", projectScopedOverride],
     ["task_lease.owner_eligibility", evaluateTaskLeaseOwnerEligibility],
     ["task_lease.acquire.decide", evaluateTaskLeaseAcquireDecision],
     ["task_lease.acquire.native", executeTaskLeaseAcquire],

loopx/control_plane/quota/scoped_override.ts

@@ -0,0 +1,18 @@
+import type { JsonObject } from "../effect_program.ts";
+import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts";
+import { decodeInteractionContract } from "../work_items/interaction_contract.ts";
+
+/** Project diagnostics after selection, guards, settlement and notification overlays.
+ * Gate admission metadata is historical context; it cannot select another Todo.
+ */
+export function projectScopedOverride(params: JsonObject): JsonObject {
+  const override = { ...requireJsonObject(params.override, "override") };
+  const interaction = decodeInteractionContract(params.interaction_contract);
+  delete override.selected_action;
+  if (interaction.agent_channel.delivery_allowed && params.selected_todo != null) {
+    const selected = requireJsonObject(params.selected_todo, "selected_todo");
+    requireNonEmptyString(selected.todo_id, "selected_todo.todo_id");
+    override.selected_action = requireNonEmptyString(selected.text, "selected_todo.text");
+  }
+  return override;
+}

loopx/control_plane/quota/should_run_packet.py

@@ -1532,6 +1532,15 @@ def _build_quota_should_run_payload(
         interaction_contract=payload.get("interaction_contract"),
         scheduler_hint=payload.get("scheduler_hint"),
     )
+    if prepared.agent_scoped_user_todo_override:
+        # Admission is decided earlier; expose an action only from the final
+        # selected Todo and interaction authority, including cooldown changes.
+        kind = str(prepared.agent_scoped_user_todo_override["kind"])
+        payload[kind] = effect_runtime_result("quota.scoped_override.project", {
+            "override": payload[kind],
+            "selected_todo": payload.get("selected_todo"),
+            "interaction_contract": payload["interaction_contract"],
+        })
     return payload
 
 

tests/control_plane/test_scoped_override_final_action.py

@@ -0,0 +1,67 @@
+"""Scoped gate diagnostics must describe only the final admitted Todo."""
+import pytest
+
+from loopx.control_plane.quota.should_run import build_quota_should_run
+from loopx.control_plane.effect_program import ReceiptBoundReplayPhase
+from loopx.control_plane.scheduler.execution_context import scheduler_execution_context_for_runtime_profile
+from loopx.control_plane.testing.quota_fixtures import quota_status_payload, quota_todo_item
+
+
+def decision(*, capabilities=("shell",), task_class="user_gate", claimed=True, **options):
+    rows = [quota_todo_item(
+        todo_id=todo_id, title=title, priority=priority,
+        claimed_by="agent-a" if claimed else None,
+        required_capabilities=required,
+    ) for todo_id, title, priority, required in (
+        ("todo_network", "Validate remote result", "P0", ["shell", "network"]),
+        ("todo_local", "Validate local result", "P1", ["shell"]),
+    )]
+    peer_gate = quota_todo_item(
+        todo_id="todo_peer", title="Choose peer destination", role="user",
+        task_class=task_class, blocks_agent="agent-b" if task_class == "user_gate" else None,
+        bound_agent="agent-b" if task_class == "user_action" else None,
+    )
+    status = quota_status_payload(
+        goal_id="scoped-action", status="active", quota_state="operator_gate",
+        recommended_action="Wait for the peer destination", waiting_on="controller",
+        agent_todo_items=rows, user_todo_items=[peer_gate],
+        coordination={"agent_model": "peer_v1", "registered_agents": ["agent-a", "agent-b"]},
+    )
+    return build_quota_should_run(status, goal_id="scoped-action", agent_id="agent-a",
+                                 available_capabilities=list(capabilities),
+                                 scheduler_execution_context=scheduler_execution_context_for_runtime_profile("codex_app_heartbeat"),
+                                 **options)
+
+
+@pytest.mark.parametrize("task_class", ["user_gate", "user_action"])
+@pytest.mark.parametrize("capabilities,todo_id", [(("shell",), "todo_local"),
+                                                 (("shell", "network"), "todo_network")])
+def test_override_uses_final_capability_eligible_todo(task_class, capabilities, todo_id):
+    payload = decision(task_class=task_class, capabilities=capabilities, requested_action_todo_id=todo_id)
+    selected = payload["selected_todo"]
+    assert selected["todo_id"] == todo_id
+    assert payload["interaction_contract"]["agent_channel"]["delivery_allowed"] is True
+    assert todo_id in payload["interaction_contract"]["agent_channel"]["primary_action"]
+    override = payload[f"agent_scoped_{task_class}_override"]
+    assert override["selected_action"] == selected["text"]
+    assert override["from_state"] == "operator_gate"
+
+
+@pytest.mark.parametrize("options", [
+    {"capabilities": ("network",)},
+    {"receipt_bound_replay_phase": ReceiptBoundReplayPhase.SETTLED},
+    {"requested_action_todo_id": "todo_absent"},
+    {"claimed": False, "capabilities": ("shell", "network")},
+])
+def test_non_delivery_packet_does_not_advertise_override_action(options):
+    payload = decision(**options)
+    assert payload["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+    assert "selected_action" not in payload["agent_scoped_user_gate_override"]
+
+
+def test_receipt_binding_preserves_identity_across_peer_gate_override():
+    payload = decision(capabilities=("shell", "network"), receipt_bound_todo_id="todo_local")
+    selected = payload["selected_todo"]
+    assert selected["todo_id"] == "todo_local"
+    assert selected["selection_binding"] == "heartbeat_receipt"
+    assert payload["agent_scoped_user_gate_override"]["selected_action"] == selected["text"]

tests/control_plane_ts/interaction_contract.test.ts

@@ -120,3 +120,22 @@ test("rejects contradictory host-facing channel states", () => {
     /quiet no-op conflicts/,
   );
 });
+
+test("scoped override preserves admission context without selecting new work", async () => {
+  const {projectScopedOverride} = await import("../../loopx/control_plane/quota/scoped_override.ts");
+  const override = {kind: "agent_scoped_user_gate_override", from_state: "operator_gate",
+    to_state: "eligible", selected_action: "old candidate"};
+  const interaction = successorReplanContract();
+  const selected = {todo_id: "todo_bound", text: "Current work", selection_binding: "heartbeat_receipt"};
+  const before = structuredClone({override, selected});
+  assert.deepEqual(projectScopedOverride({override, selected_todo: selected,
+    interaction_contract: interaction}), {kind: override.kind, from_state: "operator_gate", to_state: "eligible"});
+  interaction.agent_channel = {must_attempt: true, delivery_allowed: true, quiet_noop_allowed: false};
+  assert.equal(projectScopedOverride({override, selected_todo: selected,
+    interaction_contract: interaction}).selected_action, "Current work");
+  assert.equal(projectScopedOverride({override, selected_todo: null,
+    interaction_contract: interaction}).selected_action, undefined);
+  assert.deepEqual({override, selected}, before);
+  assert.throws(() => projectScopedOverride({override, selected_todo: {text: "No identity"},
+    interaction_contract: interaction}), /selected_todo.todo_id/);
+});