From 2e3a4a749927c99731f48833e8634f351ac4f5cc Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Sat, 12 Sep 2026 12:14:06 +0800
Subject: [PATCH 1/2] refactor(work-items): share typed planning relations with
 task graph topology

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 .../task-graph-projection-fixture-smoke.py    |  16 +-
 .../control_plane/effect_runtime_handlers.ts  |   2 +
 .../work_items/planning_horizon.ts            |   4 +-
 .../work_items/planning_inventory.ts          |  84 +------
 .../work_items/planning_relations.ts          |  98 ++++++++
 loopx/control_plane/work_items/task_graph.py  | 233 ++++++------------
 loopx/control_plane/work_items/task_graph.ts  | 128 ++++++++++
 .../control_plane/test_task_graph_topology.py | 123 +++++++++
 .../control_plane_ts/planning_horizon.test.ts |  12 +
 tests/control_plane_ts/task_graph.test.ts     | 109 ++++++++
 10 files changed, 556 insertions(+), 253 deletions(-)
 create mode 100644 loopx/control_plane/work_items/planning_relations.ts
 create mode 100644 loopx/control_plane/work_items/task_graph.ts
 create mode 100644 tests/control_plane/test_task_graph_topology.py
 create mode 100644 tests/control_plane_ts/task_graph.test.ts

diff --git a/examples/control_plane/task-graph-projection-fixture-smoke.py b/examples/control_plane/task-graph-projection-fixture-smoke.py
index 4b87f29544..f94df4388f 100644
--- a/examples/control_plane/task-graph-projection-fixture-smoke.py
+++ b/examples/control_plane/task-graph-projection-fixture-smoke.py
@@ -482,10 +482,10 @@ def assert_diamond_dag_predecessor_edges() -> None:
     a_id = node_ids["todo_a"]
     b_id = node_ids["todo_b"]
     shared_id = node_ids["todo_shared"]
-    assert (root_id, a_id, "depends_on") in edge_pairs, edge_pairs
-    assert (root_id, b_id, "depends_on") in edge_pairs, edge_pairs
-    assert (a_id, shared_id, "depends_on") in edge_pairs, edge_pairs
-    assert (b_id, shared_id, "depends_on") in edge_pairs, edge_pairs
+    assert (root_id, a_id, "continues") in edge_pairs, edge_pairs
+    assert (root_id, b_id, "continues") in edge_pairs, edge_pairs
+    assert (a_id, shared_id, "continues") in edge_pairs, edge_pairs
+    assert (b_id, shared_id, "continues") in edge_pairs, edge_pairs
     assert projection["limits"]["predecessor_truncated"] is False
 
 
@@ -700,15 +700,15 @@ def assert_cycle_predecessor_safety() -> None:
     # Should not have exploded; should have exactly 3 deliverable nodes
     deliverable_nodes = [n for n in projection["nodes"] if n["kind"] == "deliverable"]
     assert len(deliverable_nodes) == 3, deliverable_nodes
-    # root->a and a->b edges should exist; b->a is correctly skipped due to cycle detection
+    # All lineage edges survive; visited nodes, not edges, bound cycle traversal.
     edge_pairs = {(e["from_node_id"], e["to_node_id"], e["relation"]) for e in projection["edges"]}
     node_ids = {n["refs"]["todo_ids"][0]: n["node_id"] for n in deliverable_nodes}
     root_id = node_ids["todo_cycle_root"]
     a_id = node_ids["todo_cycle_a"]
     b_id = node_ids["todo_cycle_b"]
-    assert (root_id, a_id, "depends_on") in edge_pairs
-    assert (a_id, b_id, "depends_on") in edge_pairs
-    assert (b_id, a_id, "depends_on") in edge_pairs
+    assert (root_id, a_id, "continues") in edge_pairs
+    assert (a_id, b_id, "continues") in edge_pairs
+    assert (b_id, a_id, "continues") in edge_pairs
 
 
 def main() -> int:
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 0122154cee..f8e7b6df88 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -101,6 +101,7 @@ import {
   qualifyActionSelection,
 } from "./work_items/action_portfolio.ts";
 import { projectQuotaPlanningHorizon } from "./work_items/planning_horizon.ts";
+import { projectTaskGraphTopology } from "./work_items/task_graph.ts";
 import { projectDeliveryHistory, projectDeliveryResponse } from "./work_items/delivery_history.ts";
 import { validateDeliveryClaim } from "./work_items/delivery_outcome.ts";
 import {
@@ -426,6 +427,7 @@ export function createEffectRuntimeHandlers(
     ["work_item.action_portfolio.project", projectQuotaActionPortfolio],
     ["work_item.action_selection.qualify", qualifyActionSelection],
     ["work_item.planning_horizon.project", projectQuotaPlanningHorizon],
+    ["work_item.task_graph.topology", projectTaskGraphTopology],
     ["work_item.planning_inventory.project", projectTodoPlanningInventory],
     ["work_item.planning_inventory.detail", projectTodoPlanningInventoryDetail],
     ["work_item.refresh_recommendation.resolve", resolveRefreshRecommendation],
diff --git a/loopx/control_plane/work_items/planning_horizon.ts b/loopx/control_plane/work_items/planning_horizon.ts
index e36bf06167..aa42eaf56d 100644
--- a/loopx/control_plane/work_items/planning_horizon.ts
+++ b/loopx/control_plane/work_items/planning_horizon.ts
@@ -6,12 +6,10 @@ import {
 } from "../runtime_decode.ts";
 import {
   decodeTodoPlanningInventory,
-  relationKey,
-  todoRef,
   type PlanningInventoryItem,
-  type PlanningInventoryRelation,
   type PlanningState,
 } from "./planning_inventory.ts";
+import { relationKey, todoRef, type PlanningInventoryRelation } from "./planning_relations.ts";
 
 import type { JsonObject } from "../effect_program.ts";
 
diff --git a/loopx/control_plane/work_items/planning_inventory.ts b/loopx/control_plane/work_items/planning_inventory.ts
index d4426ba834..3bd254bc8d 100644
--- a/loopx/control_plane/work_items/planning_inventory.ts
+++ b/loopx/control_plane/work_items/planning_inventory.ts
@@ -18,7 +18,6 @@ export const TODO_PLANNING_INVENTORY_DETAIL_SCHEMA_VERSION =
 
 const MAX_REQUEST_ITEMS_PER_LANE = 128;
 const MAX_INVENTORY_ITEMS = 64;
-const TODO_ID = /^todo_[A-Za-z0-9_-]{3,80}$/;
 
 export type PlanningState =
   | "selected"
@@ -30,24 +29,10 @@ export type PlanningState =
 
 export type ClaimState = "current_agent" | "unclaimed" | "other_agent";
 
-const ENFORCEMENT_BY_RELATION = {
-  successor: "lineage_only",
-  unblocks: "typed_lifecycle",
-  resumes_when: "typed_condition",
-  superseded_by: "lineage_only",
-  routes_via: "read_only_context",
-} as const;
-
-export type PlanningRelationKind = keyof typeof ENFORCEMENT_BY_RELATION;
-
-export type PlanningInventoryRelation = {
-  [Kind in PlanningRelationKind]: JsonObject & {
-    from_todo_id: string;
-    to_ref: string;
-    relation: Kind;
-    enforcement: (typeof ENFORCEMENT_BY_RELATION)[Kind];
-  };
-}[PlanningRelationKind];
+import {
+  projectRelations, relation, ENFORCEMENT_BY_RELATION, PLANNING_TODO_ID as TODO_ID,
+  type PlanningInventoryRelation, type PlanningRelationKind,
+} from "./planning_relations.ts";
 
 interface Candidate extends JsonObject {
   todo_id: string;
@@ -211,67 +196,6 @@ function planningState(
   return "context";
 }
 
-function relation<Kind extends PlanningRelationKind>(
-  fromTodoId: string,
-  toRef: string,
-  kind: Kind,
-): Extract<PlanningInventoryRelation, { relation: Kind }> {
-  return {
-    from_todo_id: fromTodoId,
-    to_ref: toRef,
-    relation: kind,
-    enforcement: ENFORCEMENT_BY_RELATION[kind],
-  } as Extract<PlanningInventoryRelation, { relation: Kind }>;
-}
-
-function candidateRelations(item: Candidate): PlanningInventoryRelation[] {
-  const projected: PlanningInventoryRelation[] = [];
-  for (const successor of new Set(item.successor_todo_ids)) {
-    if (successor !== item.todo_id) {
-      projected.push(relation(item.todo_id, successor, "successor"));
-    }
-  }
-  if (item.unblocks_todo_id && item.unblocks_todo_id !== item.todo_id) {
-    projected.push(relation(item.todo_id, item.unblocks_todo_id, "unblocks"));
-  }
-  if (item.resume_when) {
-    projected.push(relation(item.todo_id, item.resume_when, "resumes_when"));
-  }
-  if (item.superseded_by && item.superseded_by !== item.todo_id) {
-    projected.push(relation(item.todo_id, item.superseded_by, "superseded_by"));
-  }
-  const routeRef = item.route_id || item.route_key;
-  if (routeRef) {
-    projected.push(relation(item.todo_id, `route:${routeRef}`, "routes_via"));
-  }
-  return projected;
-}
-
-export function relationKey(value: PlanningInventoryRelation): string {
-  return `${value.from_todo_id}\u0000${value.relation}\u0000${value.to_ref}`;
-}
-
-function projectRelations(items: readonly Candidate[]): PlanningInventoryRelation[] {
-  const projected: PlanningInventoryRelation[] = [];
-  const seen = new Set<string>();
-  for (const item of items) {
-    for (const value of candidateRelations(item)) {
-      const key = relationKey(value);
-      if (seen.has(key)) continue;
-      seen.add(key);
-      projected.push(value);
-    }
-  }
-  return projected;
-}
-
-export function todoRef(value: string): string | null {
-  if (TODO_ID.test(value)) return value;
-  const separator = value.indexOf(":");
-  if (separator < 0) return null;
-  const suffix = value.slice(separator + 1);
-  return TODO_ID.test(suffix) ? suffix : null;
-}
 
 function inventoryItem(
   item: Candidate,
diff --git a/loopx/control_plane/work_items/planning_relations.ts b/loopx/control_plane/work_items/planning_relations.ts
new file mode 100644
index 0000000000..9440073f4c
--- /dev/null
+++ b/loopx/control_plane/work_items/planning_relations.ts
@@ -0,0 +1,98 @@
+import type { JsonObject } from "../effect_program.ts";
+import { normalizeTodoResumeWhen, TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION } from "../todos/resume_condition.ts";
+
+export const PLANNING_TODO_ID = /^todo_[A-Za-z0-9_-]{3,80}$/;
+
+export const ENFORCEMENT_BY_RELATION = {
+  successor: "lineage_only",
+  unblocks: "typed_lifecycle",
+  resumes_when: "typed_condition",
+  superseded_by: "lineage_only",
+  routes_via: "read_only_context",
+} as const;
+
+export type PlanningRelationKind = keyof typeof ENFORCEMENT_BY_RELATION;
+
+export type PlanningInventoryRelation = {
+  [Kind in PlanningRelationKind]: JsonObject & {
+    from_todo_id: string;
+    to_ref: string;
+    relation: Kind;
+    enforcement: (typeof ENFORCEMENT_BY_RELATION)[Kind];
+  };
+}[PlanningRelationKind];
+
+export function relation<Kind extends PlanningRelationKind>(
+  fromTodoId: string,
+  toRef: string,
+  kind: Kind,
+): Extract<PlanningInventoryRelation, { relation: Kind }> {
+  return {
+    from_todo_id: fromTodoId,
+    to_ref: toRef,
+    relation: kind,
+    enforcement: ENFORCEMENT_BY_RELATION[kind],
+  } as Extract<PlanningInventoryRelation, { relation: Kind }>;
+}
+
+function candidateRelations(item: PlanningRelationSource): PlanningInventoryRelation[] {
+  const projected: PlanningInventoryRelation[] = [];
+  for (const successor of new Set(item.successor_todo_ids)) {
+    if (successor !== item.todo_id) {
+      projected.push(relation(item.todo_id, successor, "successor"));
+    }
+  }
+  if (item.unblocks_todo_id && item.unblocks_todo_id !== item.todo_id) {
+    projected.push(relation(item.todo_id, item.unblocks_todo_id, "unblocks"));
+  }
+  if (item.resume_when) {
+    projected.push(relation(item.todo_id, item.resume_when, "resumes_when"));
+  }
+  if (item.superseded_by && item.superseded_by !== item.todo_id) {
+    projected.push(relation(item.todo_id, item.superseded_by, "superseded_by"));
+  }
+  const routeRef = item.route_id || item.route_key;
+  if (routeRef) {
+    projected.push(relation(item.todo_id, `route:${routeRef}`, "routes_via"));
+  }
+  return projected;
+}
+
+export function relationKey(value: PlanningInventoryRelation): string {
+  return `${value.from_todo_id}\u0000${value.relation}\u0000${value.to_ref}`;
+}
+
+export function projectRelations(items: readonly PlanningRelationSource[]): PlanningInventoryRelation[] {
+  const projected: PlanningInventoryRelation[] = [];
+  const seen = new Set<string>();
+  for (const item of items) {
+    for (const value of candidateRelations(item)) {
+      const key = relationKey(value);
+      if (seen.has(key)) continue;
+      seen.add(key);
+      projected.push(value);
+    }
+  }
+  return projected;
+}
+
+export function todoRef(value: string): string | null {
+  if (PLANNING_TODO_ID.test(value)) return value;
+  const condition = normalizeTodoResumeWhen({
+    schema_version: TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION, resume_when: value,
+  });
+  if (!condition) return null;
+  const [kind, target] = condition.split(":");
+  return kind === "todo_done" || kind === "monitor_changed" ? target : null;
+}
+
+export interface PlanningRelationSource {
+  todo_id: string;
+  successor_todo_ids: string[];
+  unblocks_todo_id?: string;
+  resume_when?: string;
+  superseded_by?: string;
+  route_id?: string;
+  route_key?: string;
+}
+
diff --git a/loopx/control_plane/work_items/task_graph.py b/loopx/control_plane/work_items/task_graph.py
index c153d4e654..661ad01b90 100644
--- a/loopx/control_plane/work_items/task_graph.py
+++ b/loopx/control_plane/work_items/task_graph.py
@@ -5,6 +5,7 @@
 from typing import Any, Callable
 
 from ..runtime.time import now_utc_iso
+from ..effect_runtime import effect_runtime_result
 from ..todos.summary_item import todo_planning_source_items
 
 TASK_GRAPH_PROJECTION_SCHEMA_VERSION = "task_graph_projection_v0"
@@ -394,52 +395,6 @@ def add_edge(
         self.edges.append(edge)
 
 
-def _task_graph_build_predecessor_indexes(
-    all_todos_by_id: dict[str, dict[str, Any]],
-    *,
-    public_safe_compact_text: Callable[..., str | None],
-) -> tuple[dict[str, list[str]], dict[str, list[str]]]:
-    predecessors_by_successor: dict[str, list[str]] = {}
-    predecessors_by_supersedes: dict[str, list[str]] = {}
-    for tid, todo_item in all_todos_by_id.items():
-        if not isinstance(todo_item, dict):
-            continue
-        successor_ids = todo_item.get("successor_todo_ids")
-        if isinstance(successor_ids, list):
-            for sid in successor_ids:
-                sid_str = public_safe_compact_text(sid, limit=120)
-                if sid_str:
-                    predecessors_by_successor.setdefault(sid_str, []).append(tid)
-        superseded_by = public_safe_compact_text(todo_item.get("superseded_by"), limit=120)
-        if superseded_by:
-            predecessors_by_supersedes.setdefault(superseded_by, []).append(tid)
-    return predecessors_by_successor, predecessors_by_supersedes
-
-
-def _task_graph_resolve_direct_predecessors(
-    current_tid: str,
-    current_todo: dict[str, Any],
-    *,
-    predecessors_by_successor: dict[str, list[str]],
-    predecessors_by_supersedes: dict[str, list[str]],
-    public_safe_compact_text: Callable[..., str | None],
-) -> list[tuple[str, str | None]]:
-    pred_ids: list[tuple[str, str | None]] = []
-    for pred_tid in predecessors_by_successor.get(current_tid, []):
-        if pred_tid != current_tid:
-            pred_ids.append((pred_tid, None))
-    for pred_tid in predecessors_by_supersedes.get(current_tid, []):
-        if pred_tid != current_tid and not any(p[0] == pred_tid for p in pred_ids):
-            pred_ids.append((pred_tid, "supersedes"))
-    unblocks_parent = public_safe_compact_text(current_todo.get("unblocks_todo_id"), limit=120)
-    if unblocks_parent and unblocks_parent != current_tid and not any(p[0] == unblocks_parent for p in pred_ids):
-        pred_ids.append((unblocks_parent, None))
-    resume_when = str(current_todo.get("resume_when") or "")
-    if resume_when.startswith("todo_done:"):
-        resume_tid = public_safe_compact_text(resume_when.split(":", 1)[1], limit=120)
-        if resume_tid and resume_tid != current_tid and not any(p[0] == resume_tid for p in pred_ids):
-            pred_ids.append((resume_tid, None))
-    return sorted(pred_ids, key=lambda item: (item[0], item[1] or ""))
 
 
 def _task_graph_attach_handoff(
@@ -578,7 +533,6 @@ def _task_graph_build_predecessor_chain(
     *,
     selected_todo_id: str,
     selected_node_id: str,
-    selected_todo: dict[str, Any],
     all_todos_by_id: dict[str, dict[str, Any]],
     builder: _TaskGraphProjectionBuilder,
     public_safe_compact_text: Callable[..., str | None],
@@ -588,128 +542,81 @@ def _task_graph_build_predecessor_chain(
     max_predecessor_nodes: int,
     source_truncated: bool,
 ) -> dict[str, Any]:
-    predecessors_by_successor, predecessors_by_supersedes = _task_graph_build_predecessor_indexes(
-        all_todos_by_id,
-        public_safe_compact_text=public_safe_compact_text,
-    )
-    visited: set[str] = set()
-    queue: list[tuple[str, str | None, str | None]] = [(selected_todo_id, None, None)]
-    emitted_count = 0
-    truncated = False
-
-    while queue:
-        current_tid, successor_nid, edge_rel_hint = queue.pop(0)
-        already_visited = current_tid in visited
-        current_todo = all_todos_by_id.get(current_tid)
-        if not isinstance(current_todo, dict):
+    # Python owns public-safe rendering; TypeScript owns relation discovery and
+    # bounded traversal. Never serialize evidence, notes, or private source text.
+    renderable_todos: dict[str, dict[str, Any]] = {}
+    rows = []
+    for tid, value in all_todos_by_id.items():
+        if not public_safe_compact_text(value.get("title") or value.get("text"), limit=160):
             continue
-        is_root = current_tid == selected_todo_id
-        is_done = bool(current_todo.get("done")) or todo_done_for_status(
-            str(normalize_todo_status(current_todo.get("status")) or todo_status_open)
+        renderable_todos[tid] = value
+        state = _task_graph_todo_state(
+            value,
+            normalize_todo_status=normalize_todo_status,
+            todo_done_for_status=todo_done_for_status,
+            todo_status_open=todo_status_open, waiting_default=tid != selected_todo_id,
         )
-
-        if not is_root and not already_visited and emitted_count >= max_predecessor_nodes:
-            truncated = True
-            break
-
-        current_nid: str | None
-        if is_root:
-            current_nid = selected_node_id
-        else:
-            current_nid = builder.add_node(
-                _task_graph_deliverable_node(
-                    todo=current_todo,
-                    public_safe_compact_text=public_safe_compact_text,
-                    normalize_todo_status=normalize_todo_status,
-                    todo_done_for_status=todo_done_for_status,
-                    todo_status_open=todo_status_open,
-                    waiting_default=True,
-                )
-            )
-            if current_nid and successor_nid:
-                if edge_rel_hint == "supersedes":
-                    edge_rel = "supersedes"
-                    edge_reason = f"Successor supersedes completed todo {current_tid}."
-                else:
-                    edge_rel = "depends_on"
-                    edge_reason = f"Work depends on predecessor todo {current_tid}."
-                builder.add_edge(
-                    edge_id=_task_graph_node_id(
-                        f"edge_{edge_rel}",
-                        f"{successor_nid}:{current_nid}",
-                        public_safe_compact_text=public_safe_compact_text,
-                    ),
-                    from_node_id=successor_nid,
-                    to_node_id=current_nid,
-                    relation=edge_rel,
-                    reason=edge_reason,
-                    refs=_task_graph_refs(
-                        "todo_ids",
-                        current_tid,
-                        public_safe_compact_text=public_safe_compact_text,
-                    ),
-                )
-
-        if not current_nid:
-            continue
-
-        if already_visited:
-            continue
-
-        visited.add(current_tid)
-
-        if not is_root:
-            emitted_count += 1
-
-        if is_done:
+        row = {"todo_id": tid, "done": state == "done",
+               "successor_todo_ids": []}
+        for field in ("unblocks_todo_id", "superseded_by", "resume_when"):
+            text = public_safe_compact_text(value.get(field), limit=240)
+            if text:
+                row[field] = text
+        successors = value.get("successor_todo_ids")
+        if isinstance(successors, list):
+            row["successor_todo_ids"] = [
+                text for raw in successors
+                if (text := public_safe_compact_text(raw, limit=120))
+            ]
+        rows.append(row)
+    result = effect_runtime_result("work_item.task_graph.topology", {
+        "schema_version": "task_graph_topology_request_v0",
+        "selected_todo_id": selected_todo_id, "items": rows,
+        "predecessor_limit": max_predecessor_nodes,
+        "source_truncated": source_truncated,
+    })
+    if not isinstance(result, dict) or result.get("schema_version") != "task_graph_topology_result_v0":
+        raise RuntimeError("TypeScript task graph topology shape mismatch")
+    node_ids = {selected_todo_id: selected_node_id}
+    for tid in result["predecessor_todo_ids"]:
+        node_ids[tid] = builder.add_node(_task_graph_deliverable_node(
+            todo=renderable_todos[tid], public_safe_compact_text=public_safe_compact_text,
+            normalize_todo_status=normalize_todo_status,
+            todo_done_for_status=todo_done_for_status,
+            todo_status_open=todo_status_open, waiting_default=True,
+        ))
+    for edge in result["edges"]:
+        source, target = edge["from_todo_id"], edge["to_todo_id"]
+        builder.add_edge(
+            edge_id=_task_graph_node_id(
+                f"edge_{edge['source_relation']}",
+                f"{node_ids[source]}:{node_ids[target]}",
+                public_safe_compact_text=public_safe_compact_text,
+            ),
+            from_node_id=node_ids[source], to_node_id=node_ids[target],
+            relation=edge["relation"], reason=edge["reason"],
+            refs=_task_graph_refs("todo_ids", target,
+                                  public_safe_compact_text=public_safe_compact_text),
+        )
+    for tid in [selected_todo_id, *result["predecessor_todo_ids"]]:
+        value, nid = all_todos_by_id[tid], node_ids[tid]
+        if _task_graph_todo_state(value, normalize_todo_status=normalize_todo_status,
+                todo_done_for_status=todo_done_for_status, todo_status_open=todo_status_open) == "done":
             _task_graph_attach_evidence(
-                current_todo=current_todo,
-                current_tid=current_tid,
-                current_nid=current_nid,
-                builder=builder,
+                current_todo=value, current_tid=tid, current_nid=nid, builder=builder,
                 public_safe_compact_text=public_safe_compact_text,
             )
-
-        if not is_root:
+        if tid != selected_todo_id:
+            # A shared ancestor may have several edges. Handoff presentation
+            # keeps its historical single attachment to the first discovered one.
+            successor = next(e["from_todo_id"] for e in result["edges"]
+                             if e["to_todo_id"] == tid)
             _task_graph_attach_handoff(
-                current_todo=current_todo,
-                current_tid=current_tid,
-                current_nid=current_nid,
-                successor_nid=successor_nid,
-                builder=builder,
-                public_safe_compact_text=public_safe_compact_text,
-            )
-
-        if not is_done:
-            if not is_root:
-                continue
-            pred_list = _task_graph_resolve_direct_predecessors(
-                current_tid,
-                current_todo,
-                predecessors_by_successor=predecessors_by_successor,
-                predecessors_by_supersedes=predecessors_by_supersedes,
+                current_todo=value, current_tid=tid, current_nid=nid,
+                successor_nid=node_ids[successor], builder=builder,
                 public_safe_compact_text=public_safe_compact_text,
             )
-            for pred_id, rel_hint in pred_list:
-                queue.append((pred_id, current_nid, rel_hint))
-            continue
-
-        pred_list = _task_graph_resolve_direct_predecessors(
-            current_tid,
-            current_todo,
-            predecessors_by_successor=predecessors_by_successor,
-            predecessors_by_supersedes=predecessors_by_supersedes,
-            public_safe_compact_text=public_safe_compact_text,
-        )
-        for pred_id, rel_hint in pred_list:
-            queue.append((pred_id, current_nid, rel_hint))
-
-    return {
-        "emitted_predecessor_count": emitted_count,
-        "predecessor_limit": max_predecessor_nodes,
-        "predecessor_truncated": truncated,
-        "source_truncated": source_truncated,
-    }
+    return result["completeness"]
 
 
 def build_task_graph_projection(
@@ -811,10 +718,10 @@ def build_task_graph_projection(
     predecessor_metrics: dict[str, Any] = {}
     if selected_todo_id and selected_node_id and isinstance(selected_todo, dict):
         all_todos_by_id = {**agent_todos_by_id, **user_todos_by_id}
+        all_todos_by_id.setdefault(selected_todo_id, selected_todo)
         predecessor_metrics = _task_graph_build_predecessor_chain(
             selected_todo_id=selected_todo_id,
             selected_node_id=selected_node_id,
-            selected_todo=selected_todo,
             all_todos_by_id=all_todos_by_id,
             builder=builder,
             public_safe_compact_text=public_safe_compact_text,
@@ -1019,6 +926,8 @@ def build_task_graph_projection(
         limits["emitted_predecessor_count"] = predecessor_metrics["emitted_predecessor_count"]
         limits["predecessor_truncated"] = predecessor_metrics["predecessor_truncated"]
         limits["source_truncated"] = predecessor_metrics["source_truncated"]
+        limits["missing_predecessor_count"] = predecessor_metrics["missing_predecessor_count"]
+        limits["topology_complete"] = predecessor_metrics["topology_complete"]
     return {
         "schema_version": TASK_GRAPH_PROJECTION_SCHEMA_VERSION,
         "mode": "read_only",
diff --git a/loopx/control_plane/work_items/task_graph.ts b/loopx/control_plane/work_items/task_graph.ts
new file mode 100644
index 0000000000..1cd9f124de
--- /dev/null
+++ b/loopx/control_plane/work_items/task_graph.ts
@@ -0,0 +1,128 @@
+import type { JsonObject } from "../effect_program.ts";
+import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import { requireBoolean, requireInteger, requireJsonObject, requireNonEmptyString,
+  requireStringArray } from "../runtime_decode.ts";
+import { projectRelations, todoRef, type PlanningInventoryRelation,
+  type PlanningRelationSource } from "./planning_relations.ts";
+
+export const TASK_GRAPH_TOPOLOGY_REQUEST = "task_graph_topology_request_v0";
+export const TASK_GRAPH_TOPOLOGY_RESULT = "task_graph_topology_result_v0";
+
+interface GraphRow extends PlanningRelationSource { done: boolean }
+type GraphRelation = "depends_on" | "continues" | "supersedes";
+interface GraphEdge extends JsonObject {
+  from_todo_id: string;
+  to_todo_id: string;
+  relation: GraphRelation;
+  source_relation: PlanningInventoryRelation["relation"];
+  enforcement: PlanningInventoryRelation["enforcement"];
+  reason: string;
+}
+
+function row(value: unknown): GraphRow {
+  const raw = requireJsonObject(value, "graph row");
+  const result: GraphRow = {
+    todo_id: requireNonEmptyString(raw.todo_id, "graph row.todo_id"),
+    done: requireBoolean(raw.done, "graph row.done"),
+    successor_todo_ids: requireStringArray(raw.successor_todo_ids, "graph row.successor_todo_ids"),
+  };
+  for (const key of ["unblocks_todo_id", "superseded_by", "resume_when"] as const) {
+    if (raw[key] !== undefined && raw[key] !== null && raw[key] !== "") {
+      result[key] = requireNonEmptyString(raw[key], `graph row.${key}`);
+    }
+  }
+  return result;
+}
+
+/** Display direction is work -> prerequisite/ancestor, never transition authority. */
+function graphEdge(value: PlanningInventoryRelation): GraphEdge | null {
+  const source = value.from_todo_id;
+  // Direct persisted lineage IDs are opaque (legacy fixtures include short IDs).
+  // Only a condition needs decoding; never infer a Todo from an arbitrary suffix.
+  const target = value.relation === "resumes_when" ? todoRef(value.to_ref) : value.to_ref;
+  if (!target || source === target) return null;
+  const common = { source_relation: value.relation, enforcement: value.enforcement };
+  switch (value.relation) {
+    case "successor":
+      return { ...common, from_todo_id: target, to_todo_id: source, relation: "continues",
+        reason: "Work continues predecessor lineage; this is not a completion prerequisite." };
+    case "superseded_by":
+      return { ...common, from_todo_id: target, to_todo_id: source, relation: "supersedes",
+        reason: "Work supersedes predecessor lineage; this does not authorize a transition." };
+    case "unblocks":
+      return { ...common, from_todo_id: target, to_todo_id: source, relation: "depends_on",
+        reason: "Work has a typed lifecycle dependency on the linked unblocking Todo." };
+    case "resumes_when":
+      return { ...common, from_todo_id: source, to_todo_id: target, relation: "depends_on",
+        reason: value.to_ref.trim().toLowerCase().startsWith("monitor_changed:")
+          ? "Work waits for a Monitor generation change, not Monitor completion."
+          : "Work has a Todo completion condition; readiness is decided by the resume evaluator." };
+    case "routes_via": return null;
+  }
+}
+
+function edgeKey(edge: GraphEdge): string {
+  return [edge.from_todo_id, edge.to_todo_id, edge.source_relation].join("\0");
+}
+
+/** Bounded read lens over one supplied snapshot. No provider reads or writes. */
+export function projectTaskGraphTopology(value: unknown): JsonObject {
+  const request = requireJsonObject(value, "task graph topology request");
+  if (request.schema_version !== TASK_GRAPH_TOPOLOGY_REQUEST) {
+    throw new EffectRuntimeRequestError("Task graph topology request schema mismatch");
+  }
+  const selected = requireNonEmptyString(request.selected_todo_id, "selected_todo_id");
+  const limit = requireInteger(request.predecessor_limit, "predecessor_limit");
+  if (limit < 0 || limit > 32) throw new EffectRuntimeRequestError("predecessor_limit must be in 0..32");
+  const sourceTruncated = requireBoolean(request.source_truncated, "source_truncated");
+  if (!Array.isArray(request.items)) throw new EffectRuntimeRequestError("graph items must be an array");
+  const rows = request.items.map(row);
+  const byId = new Map(rows.map(item => [item.todo_id, item]));
+  if (byId.size !== rows.length) throw new EffectRuntimeRequestError("Duplicate task graph Todo id");
+  const adjacency = new Map<string, GraphEdge[]>();
+  for (const relation of projectRelations(rows)) {
+    const edge = graphEdge(relation);
+    if (!edge) continue;
+    const neighbors = adjacency.get(edge.from_todo_id) ?? [];
+    neighbors.push(edge);
+    adjacency.set(edge.from_todo_id, neighbors);
+  }
+  for (const edges of adjacency.values()) {
+    edges.sort((a, b) => edgeKey(a) < edgeKey(b) ? -1 : edgeKey(a) > edgeKey(b) ? 1 : 0);
+  }
+  const emitted = new Set<string>();
+  const missing = new Set<string>();
+  const omitted = new Set<string>();
+  const candidates = new Map<string, GraphEdge>();
+  const queue: string[] = [];
+  if (byId.has(selected)) { emitted.add(selected); queue.push(selected); }
+  else missing.add(selected);
+  // Continue scanning admitted vertices after the cap: a diamond's second edge
+  // must not disappear just because an earlier neighbor would exceed the cap.
+  for (let cursor = 0; cursor < queue.length; cursor++) {
+    const current = queue[cursor];
+    if (current !== selected && byId.get(current)?.done !== true) continue;
+    for (const edge of adjacency.get(current) ?? []) {
+      candidates.set(edgeKey(edge), edge);
+      const target = edge.to_todo_id;
+      if (!byId.has(target)) { missing.add(target); continue; }
+      if (emitted.has(target)) continue;
+      if (emitted.size - 1 >= limit) { omitted.add(target); continue; }
+      emitted.add(target);
+      queue.push(target);
+    }
+  }
+  const edges = [...candidates.values()].filter(edge => emitted.has(edge.to_todo_id));
+  const predecessors = queue.filter(id => id !== selected);
+  return {
+    schema_version: TASK_GRAPH_TOPOLOGY_RESULT,
+    predecessor_todo_ids: predecessors,
+    edges,
+    completeness: {
+      predecessor_limit: limit, emitted_predecessor_count: predecessors.length,
+      predecessor_truncated: omitted.size > 0, source_truncated: sourceTruncated,
+      missing_predecessor_count: missing.size,
+      topology_complete: !sourceTruncated && missing.size === 0 && omitted.size === 0,
+    },
+  };
+}
diff --git a/tests/control_plane/test_task_graph_topology.py b/tests/control_plane/test_task_graph_topology.py
new file mode 100644
index 0000000000..186f4ba80e
--- /dev/null
+++ b/tests/control_plane/test_task_graph_topology.py
@@ -0,0 +1,123 @@
+"""Public graph semantics: lineage is not execution authority."""
+import json
+import subprocess
+import sys
+
+import pytest
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+
+from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.todos.active_state_todo_parser import parse_todo_source
+from loopx.control_plane.todos.contract import format_todo_metadata_line
+from loopx.control_plane.todos.todo_summary import canonical_todo_read_record, structured_todo_item
+from loopx.status import active_state_todo_fields
+from loopx.status import build_task_graph_projection
+
+
+def todo(todo_id, **fields):
+    return {"todo_id": todo_id, "text": todo_id, "status": "open", **fields}
+
+
+def graph(rows, total=None):
+    return build_task_graph_projection(
+        {"goal_id": "graph-goal", "agent_todos": {
+            "items": rows, "total_count": len(rows) if total is None else total,
+            "open_count": sum(row["status"] != "done" for row in rows),
+        }}, goal={"id": "graph-goal"},
+    )
+
+
+def edges_by_todo(result):
+    ids = {n["node_id"]: n["refs"]["todo_ids"][0]
+           for n in result["nodes"] if n["kind"] == "deliverable"}
+    return {(ids[e["from_node_id"]], ids[e["to_node_id"]], e["relation"])
+            for e in result["edges"]
+            if e["from_node_id"] in ids and e["to_node_id"] in ids}
+
+
+def test_successor_lineage_is_not_a_completion_dependency():
+    result = graph([todo("todo_root"), todo("todo_parent", status="done",
+                    successor_todo_ids=["todo_root"])])
+    assert edges_by_todo(result) == {("todo_root", "todo_parent", "continues")}
+
+
+def test_monitor_condition_is_visible_without_claiming_monitor_completion():
+    result = graph([todo("todo_root", resume_when="monitor_changed:todo_monitor"),
+                    todo("todo_monitor", task_class="continuous_monitor")])
+    assert ("todo_root", "todo_monitor", "depends_on") in edges_by_todo(result)
+    edge = next(e for e in result["edges"] if e["relation"] == "depends_on")
+    assert "generation" in edge["reason"].lower()
+
+
+def test_unblocks_child_is_prerequisite_of_parent_not_the_reverse():
+    result = graph([todo("todo_root"), todo("todo_child", unblocks_todo_id="todo_root")])
+    assert ("todo_root", "todo_child", "depends_on") in edges_by_todo(result)
+    reversed_root = graph([todo("todo_child", unblocks_todo_id="todo_root"), todo("todo_root")])
+    assert ("todo_child", "todo_root", "depends_on") not in edges_by_todo(reversed_root)
+
+
+def test_missing_relation_target_is_not_reported_as_complete():
+    result = graph([todo("todo_root", resume_when="todo_done:todo_missing")])
+    assert result["limits"]["missing_predecessor_count"] == 1
+    assert result["limits"]["topology_complete"] is False
+
+
+def test_lineage_and_condition_between_same_pair_are_both_retained():
+    result = graph([todo("todo_root", resume_when="todo_done:todo_parent"),
+                    todo("todo_parent", status="done", successor_todo_ids=["todo_root"])])
+    assert edges_by_todo(result) == {
+        ("todo_root", "todo_parent", "continues"),
+        ("todo_root", "todo_parent", "depends_on"),
+    }
+
+
+@pytest.mark.parametrize("provider,display", [
+    ("legacy", "current"), ("file", "current"), ("sqlite", "current"),
+    ("file", "missing"), ("sqlite", "missing"),
+])
+def test_real_status_reader_uses_provider_snapshot_without_repair_writes(tmp_path, monkeypatch, provider, display):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    state = tmp_path / "STATE.md"
+    source = "\n".join([
+        "# Goal", "## Agent Todo", "- [ ] Deliver selected work",
+        format_todo_metadata_line(todo_id="todo_root", status="open", task_class="advancement_task",
+                                  resume_when="monitor_changed:todo_monitor"),
+        "- [ ] Observe changes",
+        format_todo_metadata_line(todo_id="todo_monitor", status="open", task_class="continuous_monitor",
+                                  material_change_generation=2),
+        "- [x] Earlier delivery",
+        format_todo_metadata_line(todo_id="todo_parent", status="done", successor_todo_ids=["todo_root"]),
+        "## User Todo", "",
+    ])
+    state.write_text(source)
+    runtime = tmp_path / "runtime"
+    goal = {"id": "graph-goal", "repo": str(tmp_path), "state_file": str(state), "status": "active",
+            "domain": "software", "adapter": {"kind": "manual"}}
+    if provider != "legacy":
+        groups, _, _ = parse_todo_source(source)
+        records = [canonical_todo_read_record(structured_todo_item(item, role=role,
+                   source_section=item.get("source_section")))
+                   for role, rows in groups.items() for item in rows]
+        initialize_canonical_authority(runtime, goal["id"],
+            build_todo_runtime_shadow_projection(goal_id=goal["id"], todos=records),
+            state_path=state, provider=provider)
+        if display == "missing":
+            state.unlink()
+    before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal["id"])
+    fields = active_state_todo_fields(goal, runtime_root=runtime)
+    result = build_task_graph_projection({"goal_id": goal["id"], **fields}, goal=goal)
+    assert ("todo_root", "todo_monitor", "depends_on") in edges_by_todo(result)
+    assert ("todo_root", "todo_parent", "continues") in edges_by_todo(result)
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal["id"]) == before
+    assert (state.read_text() if state.exists() else None) == (None if display == "missing" else source)
+
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [goal]}))
+    process = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
+        "--format", "json", "status", "--goal-id", goal["id"]],
+        capture_output=True, text=True, timeout=60)
+    assert process.returncode == 0, (process.stderr, json.loads(process.stdout).get("contract_errors"))
+    assert "graph-goal" in process.stdout
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal["id"]) == before
+    assert (state.read_text() if state.exists() else None) == (None if display == "missing" else source)
diff --git a/tests/control_plane_ts/planning_horizon.test.ts b/tests/control_plane_ts/planning_horizon.test.ts
index c56fe00e40..d744288462 100644
--- a/tests/control_plane_ts/planning_horizon.test.ts
+++ b/tests/control_plane_ts/planning_horizon.test.ts
@@ -272,6 +272,18 @@ test("planning horizon preserves claim requirements for unclaimed runnable conte
   assert.equal(unclaimedProjection?.claim_required_before_work, true);
 });
 
+test("opaque route refs cannot invent Todo proximity in the horizon", () => {
+  const selected = todo("todo_selected001", 1, "P1", {route_id: "todo_unrelated001"});
+  const unrelated = todo("todo_unrelated001", 2, "P0", {status: "deferred"});
+  const result = projectQuotaPlanningHorizon(request({
+    selected_todo: selected, candidates: [selected, unrelated], source_context_todo_count: 2,
+  }));
+  const item = (result?.work_items as Array<Record<string, unknown>>)
+    .find(value => value.todo_id === unrelated.todo_id);
+  assert.ok(item);
+  assert.equal((item.context_reasons as string[]).includes("related_to_selected"), false);
+});
+
 test("planning inventory detail reuses Todo rows without duplicating their payload", () => {
   const selected = todo("todo_selected001", 2, "P1", {
     required_capabilities: ["network"],
diff --git a/tests/control_plane_ts/task_graph.test.ts b/tests/control_plane_ts/task_graph.test.ts
new file mode 100644
index 0000000000..bbc042f9aa
--- /dev/null
+++ b/tests/control_plane_ts/task_graph.test.ts
@@ -0,0 +1,109 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { projectTaskGraphTopology, TASK_GRAPH_TOPOLOGY_REQUEST } from "../../loopx/control_plane/work_items/task_graph.ts";
+import { projectRelations, todoRef } from "../../loopx/control_plane/work_items/planning_relations.ts";
+import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
+import { productionScaleCoordinationFixture } from "./production_scale_coordination_fixture.ts";
+
+const row = (todo_id: string, extra: JsonObject = {}): JsonObject =>
+  ({ todo_id, done: true, successor_todo_ids: [], ...extra });
+function graph(items: JsonObject[], extra: JsonObject = {}): JsonObject {
+  return projectTaskGraphTopology({ schema_version: TASK_GRAPH_TOPOLOGY_REQUEST,
+    selected_todo_id: "todo_root", predecessor_limit: 4, source_truncated: false, items, ...extra });
+}
+
+test("known Todo conditions only: opaque route/capability refs cannot become graph edges", () => {
+  for (const value of ["route:todo_fake", "capacity_available:todo_fake", "unknown:todo_fake", "pr_merged:todo_fake"]) {
+    assert.equal(todoRef(value), null);
+  }
+  assert.equal(todoRef("todo_real"), "todo_real");
+  assert.equal(todoRef("monitor_changed:todo_real"), "todo_real");
+  assert.equal(todoRef("todo_done:todo_real"), "todo_real");
+});
+
+test("shared catalog preserves lineage, lifecycle and condition as distinct knowledge", () => {
+  const relations = projectRelations([{todo_id: "todo_root", successor_todo_ids: ["todo_child", "todo_child"],
+    unblocks_todo_id: "todo_parent", resume_when: "monitor_changed:todo_monitor"}]);
+  assert.deepEqual(relations.map(r => [r.relation, r.enforcement]), [
+    ["successor", "lineage_only"], ["unblocks", "typed_lifecycle"], ["resumes_when", "typed_condition"],
+  ]);
+});
+
+test("budget saturation retains every edge between admitted diamond vertices", () => {
+  const rows = [row("todo_root", {done: false}),
+    row("todo_a", {successor_todo_ids: ["todo_root"]}),
+    row("todo_b", {successor_todo_ids: ["todo_root"]}),
+    row("todo_shared", {successor_todo_ids: ["todo_a", "todo_b"]}),
+    row("todo_aaa_overflow", {successor_todo_ids: ["todo_b"]})];
+  const result = graph(rows, {predecessor_limit: 3});
+  assert.deepEqual(result.predecessor_todo_ids, ["todo_a", "todo_b", "todo_shared"]);
+  assert.deepEqual((result.edges as JsonObject[]).map(e => [e.from_todo_id, e.to_todo_id]), [
+    ["todo_root", "todo_a"], ["todo_root", "todo_b"],
+    ["todo_a", "todo_shared"], ["todo_b", "todo_shared"],
+  ]);
+  assert.equal((result.completeness as JsonObject).predecessor_truncated, true);
+  assert.deepEqual(graph([...rows].reverse(), {predecessor_limit: 3}), result);
+});
+
+test("cycles and parallel semantic edges terminate without duplicating nodes", () => {
+  const result = graph([row("todo_root", {resume_when: "todo_done:todo_parent", successor_todo_ids: ["todo_parent"]}),
+    row("todo_parent", {successor_todo_ids: ["todo_root"]})]);
+  assert.deepEqual(result.predecessor_todo_ids, ["todo_parent"]);
+  assert.equal((result.edges as JsonObject[]).length, 3);
+  assert.equal((result.completeness as JsonObject).topology_complete, true);
+});
+
+test("missing, source truncation and display truncation are independent", () => {
+  const result = graph([row("todo_root", {resume_when: "todo_done:todo_missing"})]);
+  assert.deepEqual(result.completeness, {predecessor_limit: 4, emitted_predecessor_count: 0,
+    predecessor_truncated: false, source_truncated: false, missing_predecessor_count: 1, topology_complete: false});
+  assert.equal((graph([row("todo_root")], {source_truncated: true}).completeness as JsonObject).topology_complete, false);
+  const omitted = graph([row("todo_root"), row("todo_parent", {successor_todo_ids: ["todo_root"]})], {predecessor_limit: 0});
+  assert.equal((omitted.completeness as JsonObject).predecessor_truncated, true);
+  assert.equal((omitted.completeness as JsonObject).missing_predecessor_count, 0);
+});
+
+test("open predecessor remains an expansion boundary, but not a dropped node", () => {
+  const result = graph([row("todo_root"), row("todo_parent", {done: false, successor_todo_ids: ["todo_root"]}),
+    row("todo_ancestor", {successor_todo_ids: ["todo_parent"]})]);
+  assert.deepEqual(result.predecessor_todo_ids, ["todo_parent"]);
+  assert.equal((result.completeness as JsonObject).topology_complete, true);
+});
+
+test("thousands of unrelated rows and deep ancestry keep projection bounded", () => {
+  const rows = [row("todo_root"), ...Array.from({length: 4096}, (_, i) => row(`todo_unrelated_${i}`)),
+    ...Array.from({length: 512}, (_, i) => row(`todo_chain_${i}`, {
+      successor_todo_ids: [i === 0 ? "todo_root" : `todo_chain_${i - 1}`],
+    }))];
+  const result = graph(rows);
+  assert.equal((result.predecessor_todo_ids as string[]).length, 4);
+  assert.equal((result.edges as JsonObject[]).length, 4);
+  assert.equal((result.completeness as JsonObject).predecessor_truncated, true);
+});
+
+test("invalid wire input fails at the typed boundary", () => {
+  assert.throws(() => graph([row("todo_root"), row("todo_root")]), /Duplicate/);
+  for (const predecessor_limit of [-1, 33, 0.5]) assert.throws(() => graph([], {predecessor_limit}));
+  assert.throws(() => graph([row("todo_root", {done: "false"})]), /boolean/);
+  assert.throws(() => graph([], {schema_version: "unknown"}), /schema/);
+});
+
+test("production-scale canonical fixture supports mixed ancestry without changing authority", () => {
+  const fixture = productionScaleCoordinationFixture("graph-goal");
+  const before = JSON.stringify(fixture.projection);
+  const records = fixture.projection.todos as JsonObject[];
+  assert.equal(records.length, fixture.expected_initial_todo_count);
+  // Add a small, explicit relationship overlay to the existing mixed status,
+  // claim, Monitor and User-gate fixture; do not replace it with a small mock.
+  const rows = records.map(r => row(r.todo_id as string, {done: r.done === true}));
+  const root = fixture.completion_todo_id;
+  const parent = rows[0].todo_id as string;
+  rows[0].successor_todo_ids = [root];
+  rows[1].successor_todo_ids = [parent];
+  rows.find(r => r.todo_id === root)!.resume_when = `monitor_changed:${rows[2].todo_id}`;
+  const result = graph(rows, {selected_todo_id: root});
+  assert.equal((result.predecessor_todo_ids as string[]).length, 3);
+  assert.deepEqual(new Set((result.edges as JsonObject[]).map(e => e.enforcement)),
+    new Set(["lineage_only", "typed_condition"]));
+  assert.equal(JSON.stringify(fixture.projection), before);
+});

From 4ffaa63fc1e92d4fe047161bc5bc7c74dd84aba5 Mon Sep 17 00:00:00 2001
From: huangruiteng <huangrt01@163.com>
Date: Sat, 12 Sep 2026 12:14:06 +0800
Subject: [PATCH 2/2] docs(rfcs): record typed topology scope and read-only
 semantic corrections

Signed-off-by: huangruiteng <huangrt01@163.com>
---
 ...shared-goal-authority-state-provider-v0.md |  7 ++++
 ...-goal-authority-state-provider-v0.zh-CN.md |  6 +++
 .../typescript-control-plane-migration-v0.md  | 12 ++++++
 ...script-control-plane-migration-v0.zh-CN.md |  9 ++++
 .../protocols/task-graph-projection-v0.md     | 41 ++++++++++++++++++-
 .../work_items/planning_relations.ts          |  1 -
 6 files changed, 74 insertions(+), 2 deletions(-)

diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
index 877b15a7e9..3697f0af76 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -2664,6 +2664,13 @@ The original direction remains; execution cards expand these stages rather than
 
 #### Durability execution cards
 
+The task graph's T3 topology consumer now shares the inventory/horizon relation
+catalog and consumes one supplied status snapshot. Its missing/truncated metrics
+describe read completeness, not canonical validity or promotion qualification.
+File/SQLite reader replay with a missing Markdown display must remain read-only;
+the graph never repairs display or changes authority. This retires duplicate
+Python relationship/traversal knowledge without changing the D1–D3 gates below.
+
 Capability-gap consumers now share the TS requirement/resolution owner across
 legacy and canonical inputs, including quota's Monitor capability partition.
 The old Python missing-set and owner/repair decision builders are removed;
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
index b918713a4c..7ad11c0226 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
@@ -2111,6 +2111,12 @@ selector 见 TS RFC 的 T3 卡。真实 FileAuthorityStore CLI 测试覆盖展
 
 #### 持久化执行卡
 
+Task graph 的 T3 topology consumer 现共用 inventory/horizon 关系目录，消费
+一次已提供的 status 快照。缺失/截断指标描述读取完整度，不代表 canonical
+有效性或 promotion 资格。File/SQLite 在 Markdown 展示缺失时的 reader 回放
+必须只读：图不修复展示，也不改变 authority。本批删除 Python 重复关系与
+遍历知识，不改变以下 D1–D3 门禁。
+
 命令清单、update/monitor 事务和 consumer 删除统一按
 [TS 执行卡](typescript-control-plane-migration-v0.zh-CN.md#当前-stack-合入后的执行卡)
 推进，不在这里复制第二套实现路线，也不把 read-policy PR 合并视为存储就绪。
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index 6fb4f9665c..cda4f203aa 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -508,6 +508,18 @@ delivery. This does not finish all T2 commands or authorize whole-Goal promotion
 
 **T3 — close remaining structured consumers, then remove their old reads.**
 
+Task-graph topology now shares `work_items/planning_relations.ts` with inventory
+and horizon. One pure TS request owns relationship discovery, deterministic
+bounded traversal, edge deduplication and missing/truncated completeness; the
+Python predecessor indexes, condition parser and traversal are retired. Python
+retains status source adaptation and public-safe node/evidence/handoff rendering.
+This intentionally distinguishes successor lineage from completion dependencies,
+corrects unblocks direction, includes Monitor generation conditions and preserves
+parallel/diamond edges at the node cap. See the [graph contract](../../reference/protocols/task-graph-projection-v0.md#typed-todo-topology).
+It does not change lifecycle admission, claim/lease semantics or default provider.
+The status source can still be incomplete: this closes one T3 interpretation
+boundary, not all graph source delivery or the remaining T1–T4 work.
+
 Capability resolution now shares `agents/capability_gate.ts`: missing prerequisites,
 repair outputs, owner/agent resolution and blocked-Todo bindings have one typed
 owner. Quota planning v1 passes normalized requirements, not Python-computed
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index a38c3b3d87..1c54248fb0 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -390,6 +390,15 @@ promotion 已完成。
 
 **T3 — 闭合剩余 structured consumer，删除各自旧读路径。**
 
+Task graph topology 与 inventory/horizon 共用 `work_items/planning_relations.ts`。
+一轮纯 TS 请求拥有关系发现、稳定有界遍历、边去重与缺失/截断完整度；删除
+Python 的前驱索引、条件拆解和遍历。Python 保留 status 来源适配及节点、
+evidence/handoff 的脱敏展示。明确的语义修正：successor 谱系不再冒充完成
+依赖，unblocks 方向修正，补 Monitor generation 条件，上限处保留平行关系
+和菱形汇合边。详见[图协议](../../reference/protocols/task-graph-projection-v0.md#typed-todo-topology)。
+不改变生命周期准入、claim/lease 或默认 provider。来源仍可能不完整：本批
+闭合一个 T3 解释边界，不宣称所有图来源交付或 T1–T4 已完成。
+
 Quota 的 scope/claim 消费者现通过每个 source 一次 `todo.quota_planning.project`，
 组合选择、有限展示与既有 resume planner。`quota_selection.ts` 替代 Python
 claim-visibility 模块及 Agent-scope 中独立的 User gate/action 过滤器。Python
diff --git a/docs/reference/protocols/task-graph-projection-v0.md b/docs/reference/protocols/task-graph-projection-v0.md
index b9cd47c149..8f94ed83f9 100644
--- a/docs/reference/protocols/task-graph-projection-v0.md
+++ b/docs/reference/protocols/task-graph-projection-v0.md
@@ -131,7 +131,46 @@ compact blocker or validation writebacks:
 - `audits` says compact run-history evidence reviews, checks, or bounds a
   selected work lane.
 - `continues` says compact run-history evidence is a continuation of a selected
-  work lane.
+  work lane, or a Todo follows a predecessor through explicit successor lineage.
+
+### Typed Todo topology
+
+Planning inventory, horizon and task graph share the TS `planning_relations`
+catalog. The graph is a different **read lens**, not another lifecycle reducer:
+
+| Persisted relation | Graph direction and label | Meaning |
+| --- | --- | --- |
+| A has successor B | B → A, `continues` | Lineage only; does not require A to complete |
+| A is superseded by B | B → A, `supersedes` | Lineage only; does not authorize a transition |
+| A unblocks B | B → A, `depends_on` | Typed lifecycle link, not the reverse dependency |
+| A resumes when B is done | A → B, `depends_on` | Completion condition; the resume evaluator owns readiness |
+| A resumes when Monitor M changes | A → M, `depends_on` | Generation-change condition, not Monitor completion |
+
+These are intentional corrections to the old graph, which collapsed successor
+lineage into dependencies, reversed unblocks discovery, and omitted Monitor
+conditions. Parallel lineage and condition edges are retained. Opaque route,
+capability and unknown-condition suffixes must not be interpreted as Todo IDs
+by either graph or horizon. Existing read-only node kinds, status normalization,
+claim presentation and evidence/handoff renderers are unchanged.
+
+The predecessor lens expands the selected Todo and completed predecessors;
+open predecessors are visible boundaries, not recursive traversal roots.
+It admits at most four predecessor nodes, in deterministic breadth-first and
+Todo-ID order. Cycles do not duplicate nodes. Reaching the node cap must not
+discard another edge between already admitted nodes (including diamond joins).
+
+`limits.missing_predecessor_count` counts unique referenced predecessors absent
+from the supplied renderable snapshot. `source_truncated` records upstream
+omission; `predecessor_truncated` records display-limit omission. The additive
+`topology_complete` flag is true only if none of those conditions applies,
+**within this expansion policy**, not for the entire Goal graph. No missing
+target creates a phantom node, a provider read, a repair write, or an execution
+permission. No extra complete-state read is introduced: status supplies its
+existing source, and an incomplete summary stays explicitly incomplete.
+
+中文：谱系不等于依赖；Monitor 的 generation 条件不等于完成 Monitor。
+节点上限不应吞掉已展示节点间的边。完整度只针对上述有界展开策略，不能
+把缺失、上游裁剪或展示裁剪说成完整 Goal 图；图始终没有写入或准入权限。
 
 These relations may help a dashboard or reviewer explain why a work item is
 still active, stale, repaired, or safe to hand off. They must not create a graph
diff --git a/loopx/control_plane/work_items/planning_relations.ts b/loopx/control_plane/work_items/planning_relations.ts
index 9440073f4c..611000fe91 100644
--- a/loopx/control_plane/work_items/planning_relations.ts
+++ b/loopx/control_plane/work_items/planning_relations.ts
@@ -95,4 +95,3 @@ export interface PlanningRelationSource {
   route_id?: string;
   route_key?: string;
 }
-