fix(periodic-report): close canonical progress and approval retry consumers

docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md

@@ -3036,6 +3036,18 @@ source paths, authorize monitor writeback, or change provider/promotion holds.
 
 **D1 — qualify permanent projection delivery; may overlap T1/T2.**
 
+Periodic-report staging, live editorial fallback and approval retry now share
+one canonical-first Todo source. Frontier and progress selection reuse the same
+complete evaluated snapshot; missing/stale display cannot invent or hide work.
+`capabilities/periodic_report_progress.ts` owns report selection and rejection
+retry ordering, retiring Python selection/sorting loops. Offset-aware instants
+retain microseconds, canonical archived rejection records remain effective, and
+explicit runtime-root applies to both intent and Todo IO. Frozen editorial
+requests retain their original basis. See [operation and boundaries](../../../loopx/capabilities/periodic_report/README.md#todo-authority-and-report-retries).
+This closes that T3/L5 consumer family, not D1 permanent display freshness,
+D2 durability, D3 whole-Goal qualification or default-provider selection. The
+conditional 5–8 remaining delivery-package estimate is unchanged.
+
 Summary/work-lane counts now remain independent of display limits and retain incomplete-source knowledge through Agent scoping; canonical list acceptance holds match status. This closes one L5 read consumer, not permanent projection freshness or D1–D3. See [count semantics](../../reference/todo-work-counts.md).
 
 The Goal Channel ownership observation consumes one complete provider revision before bounding display. It never repairs Markdown or revives old local leases; provider failures and truncation stay visible. This is a T3 read closure with shared TS interpretation, not D1/D2 qualification or D3 cutover. See [coordination observation](../../reference/coordination-observation.md).

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

@@ -789,6 +789,18 @@ all T2 commands or authorize whole-Goal promotion.
 
 **T3 — close remaining structured consumers, then remove their old reads.**
 
+Periodic-report staging, live editorial fallback and approval retry now share
+one canonical-first Todo source. Frontier and progress selection reuse the same
+complete evaluated snapshot; missing/stale display cannot invent or hide work.
+`capabilities/periodic_report_progress.ts` owns report selection and rejection
+retry ordering, retiring Python selection/sorting loops. Offset-aware instants
+retain microseconds, canonical archived rejection records remain effective, and
+explicit runtime-root applies to both intent and Todo IO. Frozen editorial
+requests retain their original basis. See [operation and boundaries](../../../loopx/capabilities/periodic_report/README.md#todo-authority-and-report-retries).
+This closes that T3/L5 consumer family, not D1 permanent display freshness,
+D2 durability, D3 whole-Goal qualification or default-provider selection. The
+conditional 5–8 remaining delivery-package estimate is unchanged.
+
 Todo summary lanes and pre-limit work counts now share `todos/summary_lanes.ts`.
 Python's lane classification and hidden-work inference loops are removed; quota
 recomputes counts after scope selection and carries incomplete source knowledge

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

@@ -599,6 +599,15 @@ delivery pending；这不代表全部 T2 命令或整 Goal promotion 已完成
 
 **T3 — 闭合剩余 structured consumer，删除各自旧读路径。**
 
+Periodic-report 的阶段判断、实时编辑输入回退与审批重试现共用 canonical-first
+Todo 来源；frontier 和报告事实复用同一完整已求值快照。展示缺失、过期或损坏不再
+隐藏／复活工作。`capabilities/periodic_report_progress.ts` 拥有报告选择及拒绝重试
+排序，删除 Python 对应循环；时间按带偏移的实际时刻比较并保留微秒，canonical
+归档拒绝记录仍有效，显式 runtime-root 同时约束 intent 和 Todo IO。已冻结的编辑
+请求沿用原始依据，不因重试刷新。见[操作边界](../../../loopx/capabilities/periodic_report/README.md#todo-authority-and-report-retries)。
+这闭合一组 T3/L5 消费者，不代表 D1 永久展示新鲜度、D2 耐久性、D3 整 Goal
+资格或默认 provider 已完成；条件性的 5–8 个后续完整交付批次估算保持不变。
+
 Todo 摘要 lane 与裁剪前工作计数现共用 `todos/summary_lanes.ts`，删除 Python 的
 lane 分类和隐藏任务推断循环。quota 在作用域筛选后重新计数，不完整来源状态贯穿
 压缩与重复投影；公开 canonical Todo 列表保留同版本 acceptance 限制。见

loopx/capabilities/periodic_report/README.md

@@ -95,6 +95,45 @@ delivery-Todo pipeline. It acknowledges the provider source only after
 `delivery_ready` durability; failed ACKs become settlement-only retries and do
 not duplicate delivery work.
 
+## Todo authority and report retries
+
+Stage-boundary frontier evaluation and project-progress selection share one
+complete evaluated Todo snapshot. After Goal promotion, File/SQLite authority
+owns those facts even when the Markdown display is stale, absent or malformed.
+An empty canonical graph is empty; a failed canonical read remains a failed
+observation and never falls back to Markdown. Reading does not repair the display.
+Before promotion, the existing Markdown source remains in use.
+
+The TS report selector consumes bounded decision fields, not report prose. It
+keeps the reporter's outcomes before peer outcomes, excludes unowned work, and
+uses the existing resume/acceptance evaluation for the next action. Report
+consumption and all report-repair actions are excluded from next-action facts.
+Publication-history filtering still runs before the six-outcome cap so already
+published facts do not hide an unseen seventh outcome.
+
+Completion and retry ordering compares offset-aware instants at microsecond
+precision, rather than timestamp strings. Invalid/naive completion timestamps
+are not reportable; an invalid stage timestamp is an error. A retained canonical
+User rejection/cancellation can supersede an approval-pending receipt after
+archival. It must match the exact decision scope and addressed Agent; this is
+permission to reconsider that pending attempt, not approval to publish. Existing
+retry-key bytes remain unchanged for the same selected decision.
+
+Use the existing local command to consume pending work:
+
+```bash
+loopx periodic-report consume-pending --goal-id <goal-id> --agent-id <agent-id> --execute --format json
+```
+
+An explicit global `--runtime-root` routes both intent state and Todo reads.
+When the result is `editorial_required`, inspect its local editorial-request
+artifact: facts identify their source Todos. An identical retry reuses the
+frozen editorial input; it does not rebuild it from newer Todo state. Provider
+outages leave the intent for retry after authority recovery. Keep existing
+editorial artifacts and receipts when rolling back; do not delete them to force
+re-generation. No provider default, subscription, scheduler, external-delivery
+permission or publication cursor is changed by this read-path refactor.
+
 ## Customize or schedule
 
 The capability remains **inactive for background work and external writes by

loopx/capabilities/periodic_report/pending_intent.py

@@ -15,12 +15,12 @@
     POST_WRITEBACK_HOOK_RECEIPT_SCHEMA_VERSION,
     InteractionProjectionHookRegistration,
 )
-from ...control_plane.todos.active_state_todo_parser import parse_active_state_todos
+from ...control_plane.effect_runtime import effect_runtime_result
+from .todo_source import read_report_todo_source
 from ...registry import (
     atomic_write_json,
     find_registry_goal,
     read_json,
-    resolve_state_file,
 )
 from ...todos import add_goal_todo
 from ...file_lock import LockAcquisitionPolicy, exclusive_file_lock
@@ -177,57 +177,45 @@ def _editorial_response_path(
     )
 
 
-def _decision_scope_text(value: object) -> str:
-    if isinstance(value, Mapping):
-        return ":".join(
-            str(value.get(field) or "")
-            for field in ("kind", "granularity", "scope_key")
-        )
-    return str(value or "")
-
-
 def _superseding_approval_revision(
     *,
     registry_path: Path,
     goal_id: str,
     agent_id: str,
     receipt: Mapping[str, Any],
+    runtime_root: Path | None = None,
 ) -> str | None:
     approval_scope = str(receipt.get("approval_scope") or "")
     if receipt.get("status") != "approval_pending" or not approval_scope:
         return None
-    registry = read_json(registry_path)
-    goal = find_registry_goal(registry, goal_id)
-    if not isinstance(goal, Mapping):
-        return None
-    repo = Path(str(goal.get("repo") or "")).expanduser()
-    state_path = resolve_state_file(repo, str(goal.get("state_file") or ""))
-    if state_path is None or not state_path.is_file():
-        return None
-    parsed = parse_active_state_todos(
-        state_path.read_text(encoding="utf-8"),
-        goal=dict(goal),
-        state_path=state_path,
-        item_limit=None,
+    _, items = read_report_todo_source(
+        registry_path=registry_path, goal_id=goal_id, runtime_root=runtime_root
     )
-    user_summary = parsed.get("user_todos")
-    items = user_summary.get("items") if isinstance(user_summary, Mapping) else []
-    superseding = [
-        item
-        for item in items or []
-        if isinstance(item, Mapping)
-        and item.get("status") == "done"
-        and item.get("action_kind")
-        in {"approve_periodic_report_payload", "cancel_periodic_report_payload"}
-        and item.get("decision_outcome") in {"reject", "cancel"}
-        and _decision_scope_text(item.get("decision_scope")) == approval_scope
-        and str(item.get("bound_agent") or item.get("blocks_agent") or "") == agent_id
-    ]
-    if not superseding:
-        return None
-    latest = max(superseding, key=lambda item: str(item.get("updated_at") or ""))
-    revision = f"{latest.get('todo_id')}:{latest.get('updated_at')}"
-    return hashlib.sha256(revision.encode("utf-8")).hexdigest()[:16]
+    keys = (
+        "todo_id",
+        "status",
+        "action_kind",
+        "decision_outcome",
+        "decision_scope",
+        "bound_agent",
+        "blocks_agent",
+        "updated_at",
+    )
+    result = effect_runtime_result(
+        "capabilities.periodic_report.approval_retry.select",
+        {
+            "schema_version": "periodic_report_approval_retry_request_v0",
+            "agent_id": agent_id,
+            "approval_scope": approval_scope,
+            "items": [{key: item.get(key) for key in keys} for item in items],
+        },
+    )
+    if (
+        not isinstance(result, dict)
+        or result.get("schema_version") != "periodic_report_approval_retry_result_v0"
+    ):
+        raise ValueError("periodic-report approval retry selection result mismatch")
+    return result["revision"]
 
 
 def _load_consumption_receipt(
@@ -292,7 +280,7 @@ def _next_attempt_revision(
     candidate_revisions: list[str] = []
     for receipt in receipts:
         revision = _superseding_approval_revision(
-            registry_path=registry_path,
+            runtime_root=runtime_root, registry_path=registry_path,
             goal_id=goal_id,
             agent_id=agent_id,
             receipt=receipt,
@@ -540,7 +528,7 @@ def _progress_facts(
     from ...rollout_event_log import load_rollout_events, rollout_event_log_path
 
     snapshot = build_project_progress_snapshot(
-        registry_path=registry_path,
+        runtime_root=runtime_root, registry_path=registry_path,
         goal_id=goal_id,
         agent_id=agent_id,
         completed_at=completed_at,

loopx/capabilities/periodic_report/post_writeback_hook.py

@@ -14,7 +14,7 @@
 from ...control_plane.goals.goal_frontier import (
     build_goal_frontier_projection_from_summaries,
 )
-from ...control_plane.todos.active_state_todo_parser import parse_active_state_todos
+from .todo_source import read_report_todo_source
 from ...control_plane.todos.quota_summary import summarize_user_todos_for_quota
 from ...control_plane.todos.todo_index import MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL
 from ...history import collect_history, load_registry
@@ -24,7 +24,7 @@
 from .stage_completion import STAGE_COMPLETION_RECEIPT_SCHEMA
 from .stage_completion import derive_periodic_report_stage_completion_from_runs
 from .presets import build_periodic_report_preset_activation
-from .project_progress_snapshot import build_project_progress_snapshot_from_state
+from .project_progress_snapshot import build_project_progress_snapshot_from_fields
 from .incremental import read_periodic_report_goal_publication_cursors
 from .machine_defaults import resolve_goal_periodic_report_subscription
 from .machine_store import read_periodic_report_machine_defaults
@@ -144,18 +144,10 @@ def periodic_report_post_writeback_hooks_for_goal(
 
 def _frontier_projection(
     *,
-    state_text: str,
-    goal: Mapping[str, Any],
-    state_path: Path,
+    todos: Mapping[str, Any],
     goal_id: str,
     agent_id: str,
 ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
-    todos = parse_active_state_todos(
-        state_text,
-        goal=dict(goal),
-        state_path=state_path,
-        item_limit=None,
-    )
     raw_user_summary = (
         dict(todos.get("user_todos"))
         if isinstance(todos.get("user_todos"), Mapping)
@@ -197,29 +189,31 @@ def build_periodic_report_post_writeback_projection(
     """Reduce private runtime state to one bounded public-safe stage receipt."""
 
     normalized_agent_id = str(agent_id or "").strip()
+    if not normalized_agent_id:
+        return {}
+    available_capabilities = payload.get("available_capabilities")
+    if available_capabilities is None and isinstance(payload.get("turn"), Mapping):
+        available_capabilities = payload["turn"].get("available_capabilities")
+    events = load_rollout_events(
+        rollout_event_log_path(runtime_root, goal_id),
+        limit=MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL,
+    )
     state = payload.get("state")
-    state_path_value = (
+    state_path = (
         state.get("path") if isinstance(state, Mapping) else None
     ) or payload.get("state_file")
-    state_path = Path(str(state_path_value or "")).expanduser()
-    if not normalized_agent_id or not state_path.is_file():
-        return {}
-    registry = load_registry(registry_path)
-    goal = next(
-        (
-            item
-            for item in registry_goals(registry)
-            if str(item.get("id") or "").strip() == str(goal_id or "").strip()
-        ),
-        {},
+    fields, _ = read_report_todo_source(
+        registry_path=registry_path,
+        runtime_root=runtime_root,
+        goal_id=goal_id,
+        state_path=Path(state_path)
+        if isinstance(state_path, str) and state_path
+        else None,
+        rollout_events=events,
+        available_capabilities=available_capabilities,
     )
-    state_text = state_path.read_text(encoding="utf-8")
     projection, _user_summary, _agent_summary = _frontier_projection(
-        state_text=state_text,
-        goal=goal,
-        state_path=state_path,
-        goal_id=goal_id,
-        agent_id=normalized_agent_id,
+        todos=fields, goal_id=goal_id, agent_id=normalized_agent_id
     )
     history = collect_history(
         registry_path=registry_path,
@@ -289,23 +283,13 @@ def build_periodic_report_post_writeback_projection(
     publication_cursor = next(
         (c for c in goal_cursors if c["agent_id"] == normalized_agent_id), None
     )
-    available_capabilities = payload.get("available_capabilities")
-    if available_capabilities is None and isinstance(payload.get("turn"), Mapping):
-        available_capabilities = payload["turn"].get("available_capabilities")
-    project_progress = build_project_progress_snapshot_from_state(
-        state_text=state_text,
-        goal=goal,
-        state_path=state_path,
+    project_progress = build_project_progress_snapshot_from_fields(
+        fields=fields,
         goal_id=goal_id,
         agent_id=normalized_agent_id,
         completed_at=str(receipt["completed_at"]),
         publication_cursor=publication_cursor,
         goal_cursors=goal_cursors,
-        available_capabilities=available_capabilities,
-        rollout_events=load_rollout_events(
-            rollout_event_log_path(runtime_root, goal_id),
-            limit=MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL,
-        ),
     )
     if publication_cursor is not None and project_progress is None:
         return {}