refactor(frontier): unify typed revision and long-chain replan policy

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

@@ -2655,6 +2655,14 @@ and retired Python selectors. Real FileAuthorityStore CLI tests cover missing an
 stale display without writing it back. This is consumer-rule consolidation, not
 a transaction/store change, provider qualification or whole-Goal cutover.
 
+Long-chain checkpoint reads now use one typed frontier revision/ACK policy across
+legacy and canonical sources (TS RFC T3). The index is built before display limits;
+excluded work cannot spuriously rearm another Agent, and an incomplete or ambiguous
+checkpoint cannot acknowledge the chain. Python keeps the persisted v0 codec, not
+a second revision/threshold policy. This is a consumer change: it adds no provider,
+commit receipt, promotion route or Markdown writer. Existing CAS/replay, permanent
+projection and D1–D3 qualification remain unchanged.
+
 The original direction remains; execution cards expand these stages rather than cancel them:
 
 1. **Close TS transactions and consumers.** Follow [T0–T3](typescript-control-plane-migration-v0.md#execution-cards-after-the-current-stack) to consolidate rules and delete duplicate decisions.

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

@@ -2102,6 +2102,13 @@ selector 见 TS RFC 的 T3 卡。真实 FileAuthorityStore CLI 测试覆盖展
 不写回。这是 consumer 规则收拢，不是 transaction/store 改造、provider 资格化或
 整 Goal cutover。
 
+长链 checkpoint 读取现由同一 typed frontier revision/ACK 策略处理 legacy 与 canonical
+来源（TS RFC T3）。Index 在展示限制之前生成；被排除工作不能误触发该 Agent，
+不完整或有歧义的 checkpoint 不能确认长链已处理。Python 保留持久 v0 codec，
+不再持有第二套 revision/threshold 策略。这是 consumer 改造，不新增 provider、
+commit receipt、promotion 路由或 Markdown writer。既有 CAS/replay、永久投影与
+D1–D3 资格化要求保持不变。
+
 以下规划保留原有方向；执行卡是它们的展开，不是替代或取消：
 
 1. **闭合 TS 事务与 consumer。** 按 [T0–T3](typescript-control-plane-migration-v0.zh-CN.md#当前-stack-合入后的执行卡) 收口规则并删除重复决策。

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

@@ -588,6 +588,30 @@ 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.
 
+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/
+metadata codecs so unchanged legal frontiers retain their persisted fingerprints;
+the old Python revision builder, index selector and two-step long-chain decision
+are retired. Terminal advancement rows still affect material identity, while
+timestamp-only maintenance does not rearm it. Thresholds remain 15 advancement
+Todos or 20 selectable open Todos with advancement work. Excluded unclaimed work
+no longer changes that Agent's checkpoint, including Agents with no claimed rows;
+removing the exclusion makes that work relevant again. Duplicate identities in a
+selected frontier, duplicate matching index lanes and incomplete timestamps cannot
+provide a complete checkpoint or suppress replanning. These are explicit read
+corrections, not new execution permissions. The existing canonical source feeds
+the index before display truncation. Complex-fixture tests replay accepted ACKs,
+excluded/eligible edits and newly available work through a real provider with
+stale/missing display; a read-only private-snapshot comparison remains private.
+This closes one T3 rule group, not the remaining consumers or T1/T2/D1–D3.
+
+Large source facts use lossless deflate/base64 transport above 512 KiB, retaining
+the exact v0 material bytes and the shared 2 MiB request boundary. The TS decoder
+rejects malformed payloads and inflation beyond 64 MiB; it never truncates rows
+or silently falls back to Python decisions. Real completed-history HTTP reads
+and complete-checkpoint tail edits guard against transport-size regressions.
+
 The list-filter consumer now uses `compact_evaluated_todo_group` instead of
 re-running resume evaluation on active-only rows. Initial parsing/canonical reads
 still evaluate against the full source through the TS owner; filtering requires

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

@@ -449,6 +449,24 @@ Todo，同一 Todo 的不同展示不重复计算，权威空 backlog 不再复
 target capability 是修复产出，不是安装或授权。没有新 provider／inventory／enablement／
 promotion；压缩候选来源的上限和其余 T3 consumer 仍需分别闭合。
 
+Advancement-frontier checkpoint 闭合：`todos/frontier_revision.ts` 现统一 Agent
+选择、完整度、实质内容哈希、长链阈值与精确 ACK/rearm 分类。Python 保留 v0 字段清单
+与 legacy JSON/metadata codec，使合法且未变化的 frontier 保持已有指纹；删除旧 Python
+revision builder、index selector 和分两步执行的长链决策。终态 advancement 仍影响
+实质身份；仅更新时间不重新触发。阈值仍是 15 项 advancement，或存在 advancement
+时的 20 项可选 open Todo。被排除的 unclaimed 工作不再改变该 Agent 的 checkpoint，
+包括没有 claimed Todo 的 Agent；取消 exclusion 后，该工作重新相关。选中 frontier
+中的重复 ID、重复匹配的 index lane 与不完整时间不能提供完整 checkpoint 或压制
+replan。这些是明确的只读语义修正，不是执行授权。既有 canonical source 在展示截断前
+生成 index。复杂 fixture 经真实 provider 验证 accepted ACK、excluded/eligible 修改、
+新可用工作及陈旧／缺失展示；私有快照只读对照结果不公开原始数据。
+本批闭合一个 T3 规则组，不代表其余 consumer 或 T1/T2/D1–D3 完成。
+
+来源 facts 超过 512 KiB 时使用无损 deflate/base64 传输，保留精确 v0 内容和共享
+2 MiB 请求边界。TS 拒绝畸形载荷及解压超过 64 MiB 的输入，不截断 Todo，也不
+静默退回 Python 决策。真实 completed-history HTTP 和完整 checkpoint 尾项变更
+回归保护传输容量语义。
+
 列表过滤现改用 `compact_evaluated_todo_group`，不再用仅活动项重算 resume。
 初始解析／canonical 读取仍通过 TS owner 在完整来源上求值；过滤要求匹配的已求值
 条件，不能把归档中的已完成依赖变成丢失。共享合成 fixture 增补“有 scope 无 outcome”

loopx/control_plane/effect_runtime_handlers.ts

@@ -149,6 +149,7 @@ import {evaluateStandingDecisionProjection} from "./todos/standing_decision.ts";
 import {evaluateDecisionScope} from "./todos/decision_scope.ts";
 import {evaluateCapabilityGate} from "./agents/capability_gate.ts";
 import {captureArchivedTodoDependencies} from "./todos/archive_capture.ts";
+import {projectAdvancementFrontier, evaluateLongTodoChain} from "./todos/frontier_revision.ts";
 import { evaluateCoordinationTodoSuccessorDerivation } from "./coordination/todo_successor_derivation.ts";
 import {
   checkLegacyCoordinationWriteAllowed,
@@ -417,6 +418,8 @@ export function createEffectRuntimeHandlers(
     ["todo.resume_condition.evaluate", evaluateTodoResumeConditions],
     ["todo.resume_planning.project", projectTodoResumePlanning],
     ["todo.quota_planning.project", projectTodoQuotaPlanning],
+    ["todo.frontier_revision.project", projectAdvancementFrontier],
+    ["goal.long_todo_chain.evaluate", evaluateLongTodoChain],
     ["todo.external_wait.plan", planTodoExternalWaitTransition],
     ["scheduler.state_transition.evaluate", evaluateSchedulerStateTransition],
     ["scheduler.state.evaluate", evaluateSchedulerStateOperation],

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -62,8 +62,7 @@
 )
 from .long_todo_chain import (
     LONG_TODO_CHAIN_TRIGGER,
-    classify_long_todo_chain_ack,
-    observe_long_todo_chain,
+    evaluate_long_todo_chain,
 )
 from .replan_rules import (
     GoalFrontierReplanFacts,
@@ -983,20 +982,13 @@ def derive_goal_frontier_replan_obligation_from_summaries(
         agent_todo_summary,
         agent_id=agent_id,
     )
-    long_chain_observation = observe_long_todo_chain(
+    long_chain_observation, long_chain_ack_decision = evaluate_long_todo_chain(
         agent_todo_summary=agent_todo_summary,
         agent_counts=agent_counts,
         frontier_counts=frontier_counts,
         agent_id=agent_id,
         agent_todo_source_items=agent_todo_source_items,
-    )
-    long_chain_ack_decision = (
-        classify_long_todo_chain_ack(
-            long_chain_observation,
-            current_transition_replan_ack or latest_replan_ack,
-        )
-        if long_chain_observation is not None
-        else None
+        latest_replan_ack=current_transition_replan_ack or latest_replan_ack,
     )
     replan_rule = select_goal_frontier_replan_rule(
         GoalFrontierReplanFacts(

loopx/control_plane/goals/goal_frontier/long_todo_chain.py

@@ -11,43 +11,17 @@
     advancement_frontier_revision_from_index,
     selectable_advancement_frontier_revision,
 )
-from ...todos.contract import normalize_todo_replan_obligation_id
+from ...effect_runtime import effect_runtime_result
+from ...todos.frontier_revision import frontier_source_facts
 
 
 LONG_TODO_CHAIN_TRIGGER = "long_todo_chain"
-LONG_TODO_CHAIN_ADVANCEMENT_THRESHOLD = 15
-LONG_TODO_CHAIN_OPEN_THRESHOLD = 20
 TODO_TASK_CLASS_ADVANCEMENT = "advancement_task"
 LONG_TODO_CHAIN_FRONTIER_REVISION_SCHEMA_VERSION = (
     TODO_FRONTIER_REVISION_SCHEMA_VERSION
 )
 
 
-def _safe_non_negative_int(value: Any) -> int:
-    try:
-        return max(0, int(value or 0))
-    except (TypeError, ValueError):
-        return 0
-
-
-def _selectable_advancement_frontier_revision(
-    source_items: list[dict[str, Any]] | None,
-    *,
-    agent_id: str | None,
-) -> tuple[str | None, str | None, bool]:
-    """Return a complete material revision for one selectable agent lane.
-
-    Terminal advancement rows remain relevant because completion and pruning
-    mutate the frontier. Incomplete source revisions fail closed so a legacy
-    projection cannot silently suppress an obligation.
-    """
-
-    return selectable_advancement_frontier_revision(
-        source_items,
-        agent_id=agent_id,
-    )
-
-
 @dataclass(frozen=True)
 class LongTodoChainObservation:
     trigger_count: int
@@ -100,7 +74,7 @@ def long_todo_chain_source_checkpoint(
     frontier_revision, frontier_updated_at, revision_complete = (
         projected
         if projected is not None
-        else _selectable_advancement_frontier_revision(
+        else selectable_advancement_frontier_revision(
             source_items,
             agent_id=agent_id,
         )
@@ -116,119 +90,28 @@ def long_todo_chain_source_checkpoint(
     )
 
 
-def observe_long_todo_chain(
-    *,
-    agent_todo_summary: dict[str, Any] | None,
-    agent_counts: dict[str, int],
-    frontier_counts: dict[str, int],
-    agent_id: str | None,
-    agent_todo_source_items: list[dict[str, Any]] | None = None,
-) -> LongTodoChainObservation | None:
-    """Observe one agent-scoped long chain without inferring from prose."""
-
-    current_advancement = frontier_counts.get(
-        "current_agent_claimed_advancement_count", 0
-    )
-    unclaimed_advancement = frontier_counts.get("unclaimed_advancement_count", 0)
-    selectable_advancement = current_advancement + unclaimed_advancement
-    if isinstance(agent_todo_summary, dict):
-        current_open = _safe_non_negative_int(
-            agent_todo_summary.get("current_agent_claimed_open_count")
-        )
-        unclaimed_open = _safe_non_negative_int(
-            agent_todo_summary.get("unclaimed_open_count")
-        )
-        selectable_open = max(
-            current_open + unclaimed_open,
-            selectable_advancement,
-        )
-    else:
-        selectable_open = max(agent_counts.get("open", 0), selectable_advancement)
-    threshold: int | None = None
-    trigger_count = 0
-    count_kind = ""
-    if selectable_advancement >= LONG_TODO_CHAIN_ADVANCEMENT_THRESHOLD:
-        threshold = LONG_TODO_CHAIN_ADVANCEMENT_THRESHOLD
-        trigger_count = selectable_advancement
-        count_kind = "selectable_advancement_todos"
-    elif (
-        selectable_open >= LONG_TODO_CHAIN_OPEN_THRESHOLD
-        and selectable_advancement > 0
-    ):
-        threshold = LONG_TODO_CHAIN_OPEN_THRESHOLD
-        trigger_count = selectable_open
-        count_kind = "selectable_open_todos"
-    if threshold is None:
-        return None
-    projected = advancement_frontier_revision_from_index(
-        (agent_todo_summary or {}).get("advancement_frontier_revision_index"),
-        agent_id=agent_id,
-    )
-    frontier_revision, _, revision_complete = (
-        projected
-        if projected is not None
-        else _selectable_advancement_frontier_revision(
-            agent_todo_source_items,
-            agent_id=agent_id,
-        )
-    )
-    return LongTodoChainObservation(
-        trigger_count=trigger_count,
-        count_kind=count_kind,
-        selectable_open_count=selectable_open,
-        selectable_advancement_count=selectable_advancement,
-        current_agent_claimed_advancement_count=current_advancement,
-        unclaimed_advancement_count=unclaimed_advancement,
-        threshold=threshold,
-        agent_id=agent_id,
-        frontier_revision=frontier_revision,
-        frontier_revision_complete=revision_complete,
-    )
-
-
-def classify_long_todo_chain_ack(
-    observation: LongTodoChainObservation,
-    latest_replan_ack: dict[str, Any] | None,
-) -> LongTodoChainAckDecision:
-    """Classify an accepted checkpoint against the current frontier revision."""
-
-    if (
-        not isinstance(latest_replan_ack, dict)
-        or latest_replan_ack.get("recorded") is not True
-    ):
-        return LongTodoChainAckDecision(acknowledged=False)
-    semantic_delta_value = latest_replan_ack.get("semantic_delta")
-    semantic_delta: dict[str, Any] = (
-        semantic_delta_value if isinstance(semantic_delta_value, dict) else {}
-    )
-    trigger_kinds = {
-        str(value or "").strip()
-        for value in semantic_delta.get("trigger_kinds") or []
-        if str(value or "").strip()
-    }
-    obligation_id = normalize_todo_replan_obligation_id(
-        semantic_delta.get("obligation_id")
-    )
-    if (
-        semantic_delta.get("accepted") is not True
-        or LONG_TODO_CHAIN_TRIGGER not in trigger_kinds
-        or not obligation_id
-    ):
-        return LongTodoChainAckDecision(acknowledged=False)
-    if not observation.frontier_revision_complete or not observation.frontier_revision:
-        return LongTodoChainAckDecision(acknowledged=False)
-    checkpoint_matches = any(
-        isinstance(checkpoint, dict)
-        and str(checkpoint.get("kind") or "").strip() == LONG_TODO_CHAIN_TRIGGER
-        and str(checkpoint.get("frontier_revision") or "").strip()
-        == observation.frontier_revision
-        for checkpoint in semantic_delta.get("trigger_checkpoints") or []
-    )
-    if checkpoint_matches:
-        return LongTodoChainAckDecision(acknowledged=True)
-    return LongTodoChainAckDecision(
-        acknowledged=False,
-        rearmed_after_obligation_id=obligation_id,
+def evaluate_long_todo_chain(
+    *, agent_todo_summary: dict[str, Any] | None,
+    agent_counts: dict[str, int], frontier_counts: dict[str, int],
+    agent_id: str | None, agent_todo_source_items: list[dict[str, Any]] | None = None,
+    latest_replan_ack: dict[str, Any] | None = None,
+) -> tuple[LongTodoChainObservation | None, LongTodoChainAckDecision | None]:
+    """One typed observation + checkpoint qualification, not two rule RPCs."""
+    result = effect_runtime_result("goal.long_todo_chain.evaluate", {
+        "schema_version": "long_todo_chain_request_v0", "operation": "observe",
+        "summary": agent_todo_summary, "agent_counts": agent_counts,
+        "frontier_counts": frontier_counts, "agent_id": agent_id,
+        "rows": (
+            None if isinstance((agent_todo_summary or {}).get("advancement_frontier_revision_index"), dict)
+            else frontier_source_facts(agent_todo_source_items)
+        ),
+        "ack": latest_replan_ack,
+    })
+    observation = result["observation"]
+    decision = result["decision"]
+    return (
+        LongTodoChainObservation(**observation) if observation is not None else None,
+        LongTodoChainAckDecision(**decision) if decision is not None else None,
     )