fix(todos): bind route-replan facts to typed succession evidence

docs/reference/todo-work-counts.md

@@ -138,6 +138,15 @@ A source already marked partial cannot regain `source_proof` or
 rows. Query scope and source completeness are independent conditions. These
 proofs remain read-only observations, not permission to settle a Goal.
 
+Route-replan facts now participate in the same full-source succession evidence.
+Changing a typed route flag or a historical handoff label after evaluation
+rejects filtered readback; it cannot reuse the old evidence to change a closure
+obligation. An explicit boolean, including `false`, takes precedence. The
+historical “stale handoff closeout” prose hint remains a bounded compatibility
+advisory in the TS succession owner, never gate clearance, successor evidence
+or execution permission. Supported old writers and their migration readers
+remain intact; retiring this Python decision does not force a Goal upgrade.
+
 Quota validates retained closure witnesses through the existing
 `todo.quota_planning.project` batch, alongside selection and resume planning.
 The TS succession owner now validates the source and terminal proofs as well as
@@ -169,6 +178,10 @@ Python 继续负责旧数据解码、公开字段筛选、隐私处理与文本
 微秒，不再把较晚编辑误作较晚完成。缺失／非法时间仍计入已完成总数和历史，但不进入
 最近完成列表。后继缺口警告仍按最后更新时间排序，未知时间靠后，不丢弃警告。
 已有 partial 来源不会因为再次筛选命中所有可见行，就重新获得整个来源的收尾证明。
+route-replan 的类型化标记及旧 handoff 文本提示现在绑定同一份完整来源证据；求值后
+修改这些事实，筛选读取会拒绝旧证据。显式 `false` 优先于旧提示，提示仍只是重规划
+建议，不能解除 gate、证明后继或授予执行权。此次只退役 Python 重复决策，不删除
+仍有调用方的旧 writer／迁移 reader，也不强制升级既有 Goal。
 
 quota 在原有 `todo.quota_planning.project` 批次中同时验证来源／收尾证明，不再由
 Python 独立判断。TS succession owner 负责生成和验证；Python 只传紧凑事实并还原展示。

loopx/control_plane/todos/handoff_gate.py

@@ -41,27 +41,12 @@ def _todo_text(item: dict[str, Any]) -> str:
     return str(item.get("text") or "").strip()
 
 
-def _stale_handoff_closeout_replan_required(gate: dict[str, Any]) -> bool:
-    # Legacy compatibility only: old authors did not write the typed route flag.
-    # Do not use this prose hint for successor existence, gate state or permission.
-    # Retire it after the remaining route-closeout writers emit the typed flag.
-    if isinstance(gate.get("route_continuation_replan_required"), bool):
-        return gate["route_continuation_replan_required"] is True
-    if _todo_done(gate):
-        return False
-    label = " ".join(
-        str(gate.get(key) or "")
-        for key in ("action_kind", "title", "text")
-        if str(gate.get(key) or "").strip()
-    ).lower()
-    return "stale" in label and "handoff" in label and "closeout" in label
-
-
 def _compact_handoff_gate(
     gate: dict[str, Any],
     *,
     state: HandoffGateState,
     successor_ids: list[str],
+    route_replan_required: bool,
 ) -> dict[str, Any]:
     payload: dict[str, Any] = {
         "schema_version": TODO_HANDOFF_GATE_SCHEMA_VERSION,
@@ -104,7 +89,7 @@ def _compact_handoff_gate(
     superseded_by = normalize_todo_id(payload.get("superseded_by"))
     if superseded_by:
         payload["superseded_by"] = superseded_by
-    if _stale_handoff_closeout_replan_required(gate):
+    if route_replan_required:
         payload["route_continuation_replan_required"] = True
         payload.setdefault(
             "route_continuation_reason",
@@ -125,7 +110,8 @@ def build_todo_handoff_gate_states(
     decisions = evaluations if evaluations is not None else project_succession(todo_items)
     gates = [
         _compact_handoff_gate(item, state=HandoffGateState(decision["handoff_state"]),
-            successor_ids=decision["successor_todo_ids"])
+            successor_ids=decision["successor_todo_ids"],
+            route_replan_required=decision["route_continuation_replan_required"])
         for item, decision in zip(todo_items, decisions, strict=True)
         if decision["handoff_state"] is not None
     ]

loopx/control_plane/todos/succession.ts

@@ -19,6 +19,7 @@ interface Row {
   active: boolean; advancement: boolean; noFollowup: boolean; tracked: boolean;
   successors: string[]; supersededBy: string | null; unblocks: string | null;
   resumes: string | null; handoff: boolean;
+  done: boolean; routeFlag: boolean | null; legacyRouteLabel: string;
 }
 function id(value: unknown): string | null {
   if (value === null) return null;
@@ -34,6 +35,9 @@ function decode(value: unknown): Row {
   const status = requireStringLiteral(facts.status, ["open", "blocked", "done", "deferred"], "status");
   const active = requireBoolean(facts.active, "active");
   const advancement = requireBoolean(facts.advancement, "advancement");
+  if (typeof facts.legacy_route_label !== "string") {
+    throw new EffectRuntimeRequestError("legacy route label must be text");
+  }
   if (!Array.isArray(facts.successors) || !Array.isArray(facts.context_fields) ||
       facts.context_fields.some(field => typeof field !== "string")) {
     throw new EffectRuntimeRequestError("succession lists must be arrays");
@@ -47,7 +51,10 @@ function decode(value: unknown): Row {
       if (!target) throw new EffectRuntimeRequestError("successor identity cannot be null");
       return target;
     }), supersededBy: id(facts.superseded_by), unblocks: id(facts.unblocks),
-    resumes: id(facts.resumes), handoff: requireBoolean(facts.handoff, "handoff")};
+    resumes: id(facts.resumes), handoff: requireBoolean(facts.handoff, "handoff"),
+    done: requireBoolean(facts.done, "done"),
+    routeFlag: facts.route_flag === null ? null : requireBoolean(facts.route_flag, "route_flag"),
+    legacyRouteLabel: facts.legacy_route_label};
 }
 function handoffState(row: Row, successors: readonly string[]): HandoffState | null {
   if (!row.active || !row.handoff) return null;
@@ -58,6 +65,16 @@ function handoffState(row: Row, successors: readonly string[]): HandoffState | n
   return successors.length ? "cleared_with_successor" : "cleared_without_successor";
 }
 
+/** Retain the historical prose hint only as a replan advisory. An explicit
+ * boolean wins, including false; this never clears a gate or grants work.
+ * Retire the hint when the supported route-closeout writers emit typed flags. */
+function routeReplanRequired(row: Row): boolean {
+  if (row.routeFlag !== null) return row.routeFlag;
+  if (!row.active || !row.handoff || row.done || row.status === "done" || row.status === "deferred") return false;
+  const label = row.legacyRouteLabel.toLowerCase();
+  return label.includes("stale") && label.includes("handoff") && label.includes("closeout");
+}
+
 /** The same edge index drives live readback and bounded archive capture. */
 export function indexInferredSuccessors(rows: readonly Pick<Row, "id" | "advancement" | "unblocks" | "resumes">[]): Map<string, string[]> {
   const inferred = new Map<string, string[]>();
@@ -96,7 +113,8 @@ export function evaluateTodoSuccession(values: readonly unknown[]): JsonObject[]
     return {schema_version: EVALUATION_SCHEMA, item_sha256: canonicalAuthoritySha256(row.facts),
       successor_todo_ids: successors, unresolved_successor_ids: declared.filter(target => !resolved.includes(target)),
       tracked_completion: row.tracked, successor_gap: row.tracked && !row.noFollowup && successors.length === 0,
-      handoff_state: handoffState(row, successors)};
+      handoff_state: handoffState(row, successors),
+      route_continuation_replan_required: routeReplanRequired(row)};
   });
 }
 
@@ -112,11 +130,13 @@ export function validateTodoSuccession(facts: unknown, value: unknown): JsonObje
   }
   requireBoolean(evaluation.tracked_completion, "tracked_completion");
   requireBoolean(evaluation.successor_gap, "successor_gap");
+  requireBoolean(evaluation.route_continuation_replan_required, "route_continuation_replan_required");
   if (evaluation.handoff_state !== null) requireStringLiteral(evaluation.handoff_state, HANDOFF_STATES, "handoff_state");
   const successors = evaluation.successor_todo_ids as string[];
   if (evaluation.tracked_completion !== row.tracked ||
       evaluation.successor_gap !== (row.tracked && !row.noFollowup && successors.length === 0) ||
       evaluation.handoff_state !== handoffState(row, successors) ||
+      evaluation.route_continuation_replan_required !== routeReplanRequired(row) ||
       new Set(successors).size !== successors.length || successors.includes(row.id ?? "")) {
     throw new EffectRuntimeRequestError("inconsistent Todo succession evaluation");
   }

loopx/control_plane/todos/succession_warning.py

@@ -158,6 +158,9 @@ def __init__(self, item: dict[str, Any], evaluation: dict[str, Any]) -> None:
 
 def succession_facts(item: dict[str, Any]) -> dict[str, Any]:
     resume = normalize_todo_resume_when(item.get("resume_when")) or ""
+    handoff = bool(normalize_todo_excluded_agents(item.get("excluded_agents"))) \
+        and bool(normalize_todo_id(item.get("unblocks_todo_id")))
+    route_flag = item.get("route_continuation_replan_required")
     return {
         "todo_id": normalize_todo_id(item.get("todo_id")),
         "status": normalize_todo_status(item.get("status")) or ("done" if item.get("done") else "open"),
@@ -169,8 +172,15 @@ def succession_facts(item: dict[str, Any]) -> dict[str, Any]:
         "superseded_by": normalize_todo_id(item.get("superseded_by")),
         "unblocks": normalize_todo_id(item.get("unblocks_todo_id")),
         "resumes": normalize_todo_id(resume.partition(":")[2]) if resume.startswith("todo_done:") else None,
-        "handoff": bool(normalize_todo_excluded_agents(item.get("excluded_agents")))
-            and bool(normalize_todo_id(item.get("unblocks_todo_id"))),
+        "handoff": handoff,
+        "done": item.get("done") is True,
+        "route_flag": route_flag if isinstance(route_flag, bool) else None,
+        # Carry prose only for historical handoff rows without a typed flag.
+        # The existing TS owner decides whether it is a replan advisory.
+        "legacy_route_label": " ".join(str(item.get(key) or "")
+            for key in ("action_kind", "title", "text")
+            if str(item.get(key) or "").strip())
+            if handoff and not isinstance(route_flag, bool) else "",
         "context_fields": sorted(key for key, value in item.items()
             if value is not None and key != "succession_evaluation"),
     }

loopx/control_plane/todos/succession_wire_v1.json

@@ -12,6 +12,9 @@
     "unblocks",
     "resumes",
     "handoff",
+    "done",
+    "route_flag",
+    "legacy_route_label",
     "context_fields"
   ],
   "evaluation_columns": [
@@ -21,6 +24,7 @@
     "unresolved_successor_ids",
     "tracked_completion",
     "successor_gap",
-    "handoff_state"
+    "handoff_state",
+    "route_continuation_replan_required"
   ]
 }

loopx/control_plane/todos/summary_projection.ts

@@ -68,7 +68,8 @@ function validateSummarySuccession(request: JsonObject, rows: readonly JsonObjec
     const evaluation = Object.fromEntries(SUCCESSION_EVALUATION_COLUMNS.map((name, column) => [name, evaluations[index][column]]));
     if (row.todo_id !== fact.todo_id || row.status !== fact.status || row.no_followup !== fact.no_followup ||
         (row.task_class === "advancement_task") !== fact.advancement ||
-        row.successor_gap !== evaluation.successor_gap || row.handoff_state !== evaluation.handoff_state) {
+        row.successor_gap !== evaluation.successor_gap || row.handoff_state !== evaluation.handoff_state ||
+        row.replan !== evaluation.route_continuation_replan_required) {
       throw new EffectRuntimeRequestError("summary facts disagree with validated succession evidence");
     }
   }

loopx/control_plane/todos/todo_summary.py

@@ -840,9 +840,6 @@ def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] |
     # the same full-source hashes/semantics before using any summary fact,
     # avoiding a separate RPC for every role/filter (including empty roles).
     succession = succession_evaluations(items)
-    handoff_gates = build_todo_handoff_gate_states(items, evaluations=succession)
-    replan_gates = {gate.get("todo_id") for gate in handoff_gates
-        if gate.get("route_continuation_replan_required") is True}
     rows = []
     for item, evaluation in zip(items, succession, strict=True):
         resume = normalize_todo_resume_when(item.get("resume_when"))
@@ -864,7 +861,7 @@ def _project_summary(items: list[dict[str, Any]], preferred_todo_ids: set[str] |
             "linked_user_action": bool(normalize_todo_id(item.get("unblocks_todo_id"))),
             "no_followup": normalize_todo_no_followup(item.get("no_followup")) is True,
             "successor_gap": evaluation["successor_gap"], "handoff_state": evaluation["handoff_state"],
-            "replan": item.get("route_continuation_replan_required") is True or item.get("todo_id") in replan_gates,
+            "replan": evaluation["route_continuation_replan_required"],
             **{"todo_id": normalize_todo_id(item.get("todo_id")),
                 "claim": normalize_todo_claimed_by(item.get("claimed_by")),
                 "bound": normalize_todo_bound_agent(item.get("bound_agent")),

tests/control_plane/test_succession_provider_readback.py

@@ -31,6 +31,40 @@ def cli(registry: Path, *args: str):
     return json.loads(child.stdout)
 
 
+@pytest.mark.parametrize('provider', ['legacy', 'file', 'sqlite'])
+def test_real_cli_historical_route_advisory_preserves_old_source_and_provider(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    runtime, state, registry = tmp_path / 'runtime', tmp_path / 'state.md', tmp_path / 'registry.json'
+    gate = {'schema_version': 'todo_item_v0', 'archive_state': 'active', 'source_section': 'Agent Todo',
+            'role': 'agent', 'text': 'stale handoff closeout', 'status': 'open', 'done': False,
+            'task_class': 'advancement_task', 'excluded_agents': ['agent-b'], 'unblocks_todo_id': 'todo_work'}
+    records = [{**gate, 'todo_id': 'todo_legacy'},
+               {**gate, 'todo_id': 'todo_ordinary', 'text': 'Review the route'},
+               {**gate, 'todo_id': 'todo_closed', 'status': 'done', 'done': True, 'no_followup': True}]
+    projection = build_todo_runtime_shadow_projection(goal_id='goal-a', todos=records, handoff_mode='soft_claim')
+    state.write_text(render_canonical_todo_sections('# Goal\n\n## Agent Todo\n', projection['todos'],
+        provider_revision='fixture-source').markdown)
+    registry.write_text(json.dumps({'common_runtime_root': str(runtime), 'goals': [{
+        'id': 'goal-a', 'repo': str(tmp_path), 'state_file': state.name, 'status': 'active',
+        'coordination': {'registered_agents': ['agent-a', 'agent-b']},
+    }]}))
+    if provider != 'legacy':
+        initialize_canonical_authority(runtime, 'goal-a', projection, state_path=state, provider=provider)
+        state.unlink()
+    source = state.read_bytes() if state.exists() else None
+    before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a')
+    for todo_id, expected in [('todo_legacy', True), ('todo_ordinary', False), ('todo_closed', False)]:
+        result = cli(registry, '--todo-id', todo_id)
+        summary = result['agent_todos']
+        handoff = summary['handoff_gates'][0]
+        assert (handoff.get('route_continuation_replan_required') is True) is expected
+        assert handoff['gate_state'] == ('cleared_no_followup' if todo_id == 'todo_closed' else 'blocking')
+        assert 'terminal_closure_proof' not in summary or todo_id == 'todo_closed'
+        assert result['todo']['excluded_agents'] == ['agent-b']
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') == before
+    assert (state.read_bytes() if state.exists() else None) == source
+
+
 @pytest.mark.parametrize('provider', ['legacy', 'file', 'sqlite'])
 def test_real_cli_graph_selection_history_and_read_only_manager(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)

tests/control_plane/test_todo_succession_read_model.py

@@ -125,6 +125,40 @@ def test_explicit_route_flag_is_not_overridden_by_legacy_prose_hint():
     assert result["gate_state"] == "blocking"
 
 
+@pytest.mark.parametrize("status,done,flag,label,expected", [
+    ("open", False, None, "STALE handoff closeout", True),
+    ("open", False, False, "stale handoff closeout", False),
+    ("done", True, True, "", True),
+    ("done", True, None, "stale handoff closeout", False),
+    ("deferred", False, None, "stale handoff closeout", False),
+    ("open", True, None, "stale handoff closeout", False),
+    ("open", False, None, "handoff closeout", False),
+])
+def test_route_replan_retains_explicit_and_legacy_advisory_semantics(status, done, flag, label, expected):
+    gate = work("todo_gate", status=status, done=done, text=label,
+                excluded_agents=["agent-b"], unblocks_todo_id="todo_work")
+    if flag is not None:
+        gate["route_continuation_replan_required"] = flag
+    result = build_todo_handoff_gate_states([gate])[0]
+    assert (result.get("route_continuation_replan_required") is True) is expected
+    # A replan hint never resolves ownership or the handoff's successor.
+    assert result["successor_count"] == 0
+    assert result["excluded_agents"] == ["agent-b"]
+
+
+@pytest.mark.parametrize("field,value", [
+    ("route_continuation_replan_required", False),
+    ("text", "handoff closeout"),
+])
+def test_filtered_summary_rejects_changed_route_facts(field, value):
+    gate = work("todo_gate", status="open", done=False, text="stale handoff closeout",
+                excluded_agents=["agent-b"], unblocks_todo_id="todo_work")
+    source = summary([gate])
+    source["items"][0][field] = value
+    with pytest.raises(Exception, match="matching full-source"):
+        filtered_todo_summary(source, role="agent", todo_id="todo_gate")
+
+
 def test_archived_identity_can_be_recreated_but_duplicate_active_authority_rejects():
     source = work("todo_source", no_followup=True)
     result = summary([source], resume_source_items=[source, {**source, "archive_state": "archive"}])