refactor(control-plane): retire dead local snapshot and standing-decision wrapper

docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md

@@ -109,7 +109,7 @@ default-entry adoption.
 | --- | --- | --- |
 | `authority_core.py` acquire/renew/transfer/release, owner-eligibility and handoff-transition command facades | Only the old core tests; real lease and handoff adapters already use whole native transactions | No persisted command format or public CLI schema changes. Retain independent native generation, replay, conflict, cleanup and quiescence tests; exercise real File/SQLite entrypoints. |
 | `task_lease.acquire.decide`, `task_lease.lifecycle.decide`, `coordination.handoff_mode.plan` RPC registrations | Only those retired facades / handler tests; native transactions call the same typed rules directly | Obsolete private RPCs now reject unsupported methods. Keep `task_lease.owner_eligibility` and write-scope overlap: actual Python callers remain. |
-| Lease-only `local_snapshot.py` normalization and error projection | No remaining caller; native executors own lease facts and errors | Keep `todo_snapshot_from_mapping`, used by live Todo mutation authorization. No store, receipt, backup or migration reader is removed. |
+| Lease-only `local_snapshot.py` normalization and error projection | No remaining caller; native executors own lease facts and errors | After #6396 removed the last caller, the module and its remaining `todo_snapshot_from_mapping` adapter are now removed. No store, receipt, backup or migration reader is removed. |
 
 `authority_core.py` remains a live Todo bridge. At the #5395 boundary,
 `LeaseAction` / `LeaseModeGateCommand` remained registered until M4 review.

docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md

@@ -94,7 +94,7 @@ Python 退役收益。调用方清单如下：
 | --- | --- | --- |
 | `authority_core.py` 的 acquire／renew／transfer／release、owner eligibility、handoff transition command facade | 只剩旧 core 测试；真实 lease／handoff adapter 已直接使用完整 native 事务 | 不改持久化命令格式或公共 CLI schema。保留独立的原生 generation、重放、冲突、清理、静止规则测试，并走真实 File／SQLite 入口。 |
 | `task_lease.acquire.decide`、`task_lease.lifecycle.decide`、`coordination.handoff_mode.plan` RPC 注册 | 只剩这些旧 facade／handler 测试；原生事务直接复用同一 TS 规则 | 废弃私有 RPC 明确拒绝；保留仍有 Python 调用方的 `task_lease.owner_eligibility` 和 write-scope overlap。 |
-| `local_snapshot.py` 中仅供 lease 的规范化和错误投影 | 已无调用方；原生执行器拥有 lease 事实与错误 | 保留真实 Todo mutation authorization 使用的 `todo_snapshot_from_mapping`；不删 store、回执、备份或迁移 reader。 |
+| `local_snapshot.py` 中仅供 lease 的规范化和错误投影 | 已无调用方；原生执行器拥有 lease 事实与错误 | #6396 移除最后一个调用方后，现已删除该模块及其剩余的 `todo_snapshot_from_mapping` 适配器；不删 store、回执、备份或迁移 reader。 |
 
 `authority_core.py` 仍是活跃 Todo bridge；#5395 当时将
 `LeaseAction`／`LeaseModeGateCommand` 保留到 M4 评审。有界 M4 包现退役无人

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

@@ -1493,9 +1493,12 @@ migrate all T3 consumers or grant amendment commit/whole-Goal promotion authorit
 
 Standing-decision consumer closure: `todos/standing_decision.ts` owns reusable
 receipt eligibility and chronology, shared by status/quota reads and archive
-selection. Python decodes legacy metadata and submits one batch; its old receipt
-selector and the archive-local TS eligibility copy are removed. Canonical reads
-use the complete Todo snapshot, including retained archived decisions, before
+selection. Python decodes legacy metadata for the composed `todo.read` batch;
+its old receipt selector and the archive-local TS eligibility copy are removed.
+The unused `build_standing_decision_authority` wrapper and its standalone
+`todo.standing_decision.project` handler/request decoder are now removed; tests
+and the example use the composed read, while the shared TS owner remains.
+Canonical reads use the complete Todo snapshot, including retained archived decisions, before
 assigning display indexes. Later rejection/cancellation is ordered by decision
 time, not Todo ID; contradictory unresolved chronology yields a diagnostic and
 no active receipt. Explicit user-gate metadata replaces notification heuristics

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

@@ -1077,7 +1077,10 @@ T3 完成，也不授予 amendment commit／整 Goal promotion 权限。
 
 Standing decision consumer 收口：`todos/standing_decision.ts` 统一可复用决策的
 资格与先后关系，供 status/quota 读取和 archive selection 共用。Python 只解码旧
-metadata 并批量调用，删除旧 receipt selector 与 TS archive 内的重复资格判断。
+metadata 并交给组合后的 `todo.read` 批量调用，删除旧 receipt selector 与 TS archive
+内的重复资格判断。现已删除无调用方的 `build_standing_decision_authority` 包装及其
+独立 `todo.standing_decision.project` handler／请求解码器；测试与示例改用组合读取，
+共享 TS owner 保留。
 Canonical 读取在生成展示 index 前使用完整 Todo 快照，包括保留的归档决策。
 后续拒绝／取消按决策时间覆盖旧批准，不再依赖 Todo ID；矛盾历史无法定序时给出
 诊断且不提供 active receipt。此授权面必须有显式 user-gate metadata，不再借用

examples/control_plane/todo-standing-decision-authority-smoke.py

@@ -26,7 +26,7 @@
     standing_decision_authority_for_agent,
 )
 from loopx.control_plane.todos.standing_decision import (  # noqa: E402
-    build_standing_decision_authority,
+    standing_decision_source_items,
 )
 from loopx.quota import build_quota_should_run  # noqa: E402
 
@@ -36,6 +36,17 @@
 SCOPE = "write_scope:goal:benchmark_pr_self_merge"
 
 
+def project_standing_authority(items, *, legacy_source_order=True, canonical_records=False):
+    """Exercise the production source decoder and composed Todo read transaction."""
+    from loopx.control_plane.todos.goal_todo_projection import source_todo_fields
+
+    return source_todo_fields(
+        [], lineage=[],
+        standing_items=standing_decision_source_items(items, canonical_records=canonical_records),
+        legacy_source_order=legacy_source_order and not canonical_records,
+    ).get("standing_decision_authority")
+
+
 def user_receipt(
     todo_id: str,
     *,
@@ -122,7 +133,7 @@ def assert_receipt_is_agent_scoped_and_revocable(authority: dict) -> None:
     assert consistency["ok"] is True, consistency
     assert consistency["standing_authority_match_count"] == 1, consistency
 
-    revoked = build_standing_decision_authority(
+    revoked = project_standing_authority(
         [
             user_receipt("todo_standing_approve"),
             user_receipt("todo_standing_reject", outcome="reject"),
@@ -145,7 +156,7 @@ def assert_receipt_is_agent_scoped_and_revocable(authority: dict) -> None:
 
 
 def assert_one_shot_and_non_done_gates_are_not_standing() -> None:
-    authority = build_standing_decision_authority(
+    authority = project_standing_authority(
         [
             user_receipt(
                 "todo_linked_approve",

loopx/control_plane/effect_runtime_handlers.ts

@@ -271,7 +271,6 @@ export function createEffectRuntimeHandlers(
     ["turn_journal.observed_capabilities", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({readTurnJournalCapabilities}) => readTurnJournalCapabilities)],
     ["todo.completion_fence.evaluate", lazyHandler(() => import("./todos/completion_fence.ts"), ({evaluateTodoCompletionFence}) => evaluateTodoCompletionFence)],
     ["todo.priority.plan", lazyHandler(() => import("./todos/priority.ts"), ({evaluateTodoPriority}) => evaluateTodoPriority)],
-    ["todo.standing_decision.project", lazyHandler(() => import("./todos/standing_decision.ts"), ({evaluateStandingDecisionProjection}) => evaluateStandingDecisionProjection)],
     ["todo.read", lazyHandler(() => import("./todos/read_transport.ts"), ({readTodoRuntime}) => readTodoRuntime)],
     ["capabilities.periodic_report.progress.select", lazyHandler(() => import("./capabilities/periodic_report_progress.ts"), ({selectPeriodicReportProgress}) => selectPeriodicReportProgress)],
     ["capabilities.periodic_report.approval_retry.select", lazyHandler(() => import("./capabilities/periodic_report_progress.ts"), ({selectPeriodicReportApprovalRetry}) => selectPeriodicReportApprovalRetry)],

loopx/control_plane/todos/standing_decision.py

@@ -1,10 +1,7 @@
-"""Legacy metadata decoding and one batch call to the typed receipt owner."""
+"""Decode source metadata for the composed typed Todo read projection."""
 
 from __future__ import annotations
 
-from typing import Any
-
-from ..effect_runtime import effect_runtime_result
 from .contract import (
     normalize_todo_blocks_agent,
     normalize_todo_decision_outcome,
@@ -15,36 +12,6 @@
 )
 
 
-def build_standing_decision_authority(
-    user_items: list[dict[str, Any]] | None,
-    *,
-    legacy_source_order: bool = True,
-    canonical_records: bool = False,
-) -> dict[str, Any] | None:
-    """Decode legacy metadata; canonical records are already decoded.
-
-    Canonical IDs obey their provider contract, not the Markdown token codec.
-    Both sources use the same typed eligibility and chronology owner.
-    """
-    items = standing_decision_source_items(user_items, canonical_records=canonical_records)
-    if not items:
-        return None
-    result = effect_runtime_result(
-        "todo.standing_decision.project",
-        {
-            "schema_version": "standing_decision_projection_request_v0",
-            "items": items,
-            "legacy_source_order": legacy_source_order and not canonical_records,
-        },
-    )
-    if result is not None and (
-        not isinstance(result, dict)
-        or result.get("schema_version") != "standing_decision_authority_v0"
-    ):
-        raise ValueError("invalid typed standing decision projection")
-    return result
-
-
 def standing_decision_source_items(user_items, *, canonical_records=False):
     items = []
     for item in user_items or []:

loopx/control_plane/todos/standing_decision.ts

@@ -1,12 +1,10 @@
 /** Read-only standing decisions. Storage order is not decision chronology;
  * this projection neither grants a lease nor commits an owner decision. */
 import type {JsonObject} from "../effect_program.ts";
-import {requireJsonObject} from "../runtime_decode.ts";
 import {parseTodoTimestampMicros} from "../runtime_timestamp.ts";
 import {authorityUnicodeCompare} from "../coordination/authority_store_codec.ts";
 import {todoPresentationMetadata} from "../coordination/todo_presentation.ts";
 
-export const STANDING_DECISION_REQUEST_SCHEMA = "standing_decision_projection_request_v0";
 const SCOPE_KINDS = new Set(["private_read", "write_scope", "resource", "production", "public_claim", "direction", "other"]);
 const BROAD_SCOPES = new Set(["goal", "project", "global"]);
 // The existing decision_scope_v0 wire grammar; not a prose classifier.
@@ -110,10 +108,3 @@ export function projectStandingDecisions(items: readonly JsonObject[], legacySou
     inactive_count: entries.filter(e => !e.active).length, entries,
     ...(conflicts.length ? {conflict_count: conflicts.length, conflicts} : {})};
 }
-
-export function evaluateStandingDecisionProjection(value: unknown): JsonObject | null {
-  const request = requireJsonObject(value, "standing decision projection request");
-  if (request.schema_version !== STANDING_DECISION_REQUEST_SCHEMA || !Array.isArray(request.items) ||
-      typeof request.legacy_source_order !== "boolean") throw new TypeError("invalid standing decision projection request");
-  return projectStandingDecisions(request.items.map(item => requireJsonObject(item, "standing decision item")), request.legacy_source_order);
-}

tests/control_plane/test_standing_decision_order.py

@@ -28,10 +28,21 @@
     build_required_decision_scope_repair_hint,
 )
 from loopx.control_plane.todos.standing_decision import (
-    build_standing_decision_authority,
+    standing_decision_source_items,
 )
 
 
+def project_standing_authority(items, *, legacy_source_order=True, canonical_records=False):
+    """Exercise the production source decoder and composed Todo read transaction."""
+    from loopx.control_plane.todos.goal_todo_projection import source_todo_fields
+
+    return source_todo_fields(
+        [], lineage=[],
+        standing_items=standing_decision_source_items(items, canonical_records=canonical_records),
+        legacy_source_order=legacy_source_order and not canonical_records,
+    ).get("standing_decision_authority")
+
+
 def decision(todo_id: str, outcome: str, completed_at: str) -> dict:
     return {
         "todo_id": todo_id,
@@ -52,7 +63,7 @@ def test_later_rejection_wins_even_when_older_approval_sorts_last() -> None:
         decision("todo_zzz_approve", "approve", "2026-09-10T01:00:00Z"),
     ]
     for ordered in [items, list(reversed(items))]:
-        result = build_standing_decision_authority(ordered)
+        result = project_standing_authority(ordered)
         assert result["active_count"] == 0
         assert result["entries"][0]["source_todo_id"] == "todo_aaa_reject"
 
@@ -66,7 +77,7 @@ def test_conflicts_survive_lane_filtering_without_granting_cross_agent_authority
     ]
     for item in items:
         item.update(global_gate=False, blocks_agent="agent-a")
-    result = build_standing_decision_authority(items, legacy_source_order=False)
+    result = project_standing_authority(items, legacy_source_order=False)
     scoped = standing_decision_authority_for_agent(result, agent_id="agent-a")
     assert scoped["active_count"] == 0
     assert scoped["entries"] == []
@@ -113,7 +124,7 @@ def test_canonical_identifiers_are_not_redecoded_as_markdown_tokens() -> None:
                 "scope_key": "release",
             },
         )
-    result = build_standing_decision_authority(items, canonical_records=True)
+    result = project_standing_authority(items, canonical_records=True)
     assert result["active_count"] == 0
     assert result["entries"][0]["source_todo_id"] == "receipt-revoke"
 

tests/control_plane/test_todo_decision_scope_consistency.py

@@ -12,7 +12,7 @@
     build_required_decision_scope_repair_hint,
     standing_decision_authority_for_agent,
 )
-from loopx.control_plane.todos.standing_decision import build_standing_decision_authority
+from loopx.control_plane.todos.standing_decision import standing_decision_source_items
 from loopx.quota import build_quota_should_run
 
 AGENT_ID = "codex-quality-qualification"
@@ -24,6 +24,17 @@
 }
 
 
+def project_standing_authority(items, *, legacy_source_order=True, canonical_records=False):
+    """Exercise the production source decoder and composed Todo read transaction."""
+    from loopx.control_plane.todos.goal_todo_projection import source_todo_fields
+
+    return source_todo_fields(
+        [], lineage=[],
+        standing_items=standing_decision_source_items(items, canonical_records=canonical_records),
+        legacy_source_order=legacy_source_order and not canonical_records,
+    ).get("standing_decision_authority")
+
+
 def test_explicit_gate_scope_is_not_cancelled_by_another_claimant() -> None:
     gate = {"todo_id": "todo_scoped_gate", "status": "open", "task_class": "user_gate",
             "blocks_agent": AGENT_ID, "claimed_by": "agent-b", "decision_scope": SCOPE}
@@ -291,7 +302,7 @@ def test_standing_approval_satisfies_matching_agent_scope() -> None:
         "granularity": "goal",
         "scope_key": "benchmark_pr_self_merge",
     }
-    authority = build_standing_decision_authority(
+    authority = project_standing_authority(
         [
             {
                 "todo_id": "todo_standing_approval",
@@ -338,7 +349,7 @@ def test_latest_rejection_revokes_exact_standing_approval() -> None:
         "blocks_agent": AGENT_ID,
         "decision_scope": "write_scope:goal:benchmark_pr_self_merge",
     }
-    authority = build_standing_decision_authority(
+    authority = project_standing_authority(
         [
             {
                 **base,
@@ -393,7 +404,7 @@ def test_latest_rejection_revokes_exact_standing_approval() -> None:
 def test_non_authoritative_completed_items_do_not_become_standing(
     invalid_item: dict,
 ) -> None:
-    assert build_standing_decision_authority([invalid_item]) is None
+    assert project_standing_authority([invalid_item]) is None
 
 
 @pytest.mark.parametrize("decision_outcome", ["reject", "cancel"])

tests/control_plane_ts/effect_runtime_fairness.test.ts

@@ -78,8 +78,8 @@ test("a real runtime serves lightweight work during cold File history verificati
   await delay(100);
   const binding = {id: "review", agent_id: "reviewer", todo_id: "todo_review", workspace: root,
     requesters: ["coordinator"], host_args: ["--host", "dsh"], timeout_seconds: 60, output_refs: ["output.json"]};
-  const [ping, decisions, scope, ownership, selectedBinding] = await Promise.all([rpc("runtime.ping"), rpc("todo.standing_decision.project", {
-    schema_version: "standing_decision_projection_request_v0", items: [], legacy_source_order: false,
+  const [ping, decisions, scope, ownership, selectedBinding] = await Promise.all([rpc("runtime.ping"), rpc("todo.read", {
+    schema_version: "todo_read_request_v0", kind: "source_fields", groups: [], standing_items: [], legacy_source_order: false,
   }), rpc("todo.authoring_scope.plan", {
     schema_version: "todo_authoring_scope_request_v0", command: "class", role: "agent",
     todo: {task_class: "continuous_monitor"}, intent: {},
@@ -93,7 +93,7 @@ test("a real runtime serves lightweight work during cold File history verificati
   assert.equal(scope.schema_version, "todo_authoring_scope_result_v0");
   assert.equal(ownership.ownership_gate, "require_holder");
   assert.ok(ping.pid);
-  assert.equal(decisions, null, "no standing decisions means no authority projection");
+  assert.equal((decisions.fields as Record<string, unknown>).standing_decision_authority, undefined, "no standing decisions means no authority projection");
   assert.equal(heavyFinished, false, "a full history proof must not monopolize other requests");
   const receipt = await heavy;
   assert.equal(receipt.status, "found");