fix(status): restore review material links from canonical Todo text

examples/user-todo-review-material-smoke.py

@@ -151,7 +151,7 @@ def write_fixture(root: Path) -> tuple[Path, Path, Path]:
     registry.write_text(
         json.dumps(
             {
-                "runtime_root": str(runtime_root),
+                "common_runtime_root": str(runtime_root),
                 "goals": [
                     {
                         "id": GOAL_ID,
@@ -169,6 +169,8 @@ def write_fixture(root: Path) -> tuple[Path, Path, Path]:
         + "\n",
         encoding="utf-8",
     )
+    from loopx.control_plane.testing.canary_harness import import_fixture_authority
+    import_fixture_authority(registry, GOAL_ID, provider="sqlite")
     return registry, runtime_root, review_doc
 
 

loopx/control_plane/todos/active_state_todos.py

@@ -14,6 +14,7 @@
 from .succession_warning import public_todo_summary
 from .resume_condition import TodoResumeRolloutEvents
 from ...agent_registry import registered_agent_ids_for_goal
+from ...materials import extract_review_materials
 from ..work_items.recommendation_source_io import (
     load_recommendation_source_goal, recommendation_source_context,
     project_agent_next_actions, recommendation_runs,
@@ -41,6 +42,10 @@ def _redacted_status_todo_fields(fields: dict[str, Any]) -> dict[str, Any]:
                         continue
                     material_copy = dict(material)
                     material_copy.pop("resolved_path", None)
+                    if any(isinstance(value, str) and value
+                           and public_safe_compact_text(value, limit=len(value)) is None
+                           for value in material_copy.values()):
+                        continue
                     redacted_materials.append(material_copy)
                 item_copy["review_materials"] = redacted_materials
             items.append(item_copy)
@@ -106,6 +111,19 @@ def active_state_todo_fields(
     from .goal_todo_projection import observe_todo_acceptance
     fields = observe_todo_acceptance(canonical["_todo_fields"], canonical,
         registry_path=source_registry, goal_id=goal_id)
+    # Material existence is a local read projection, not persisted Todo state.
+    # Derive it from the canonical text so stale display links cannot take over.
+    for key in ("user_todos", "agent_todos"):
+        summary = fields.get(key)
+        if not isinstance(summary, dict):
+            continue
+        for todo in summary.get("items") or []:
+            if not isinstance(todo, dict):
+                continue
+            materials = extract_review_materials(str(todo.get("text") or ""),
+                goal=goal, state_path=state_path)
+            if materials:
+                todo["review_materials"] = materials
     # An explicit canonical objective revision owns current intent. A scoped
     # check objective still cannot substitute for Goal intent. Keep this status
     # field as display content; admission uses the native revision and guard.

loopx/control_plane/todos/goal_todo_projection.py

@@ -64,7 +64,10 @@ def summary_items(fields: dict[str, Any], role: str) -> list[dict[str, Any]]:
     return [item for item in summary.get("items") or [] if isinstance(item, dict)]
 
 
-def observe_todo_acceptance(fields, canonical, *, registry_path, goal_id):
+def observe_todo_acceptance(
+    fields: dict[str, Any], canonical: dict[str, Any], *,
+    registry_path: Path | None, goal_id: str,
+) -> dict[str, Any]:
     from ..goals.acceptance import observe_canonical_goal_acceptance
 
     acceptance = observe_canonical_goal_acceptance(

tests/control_plane/test_canonical_status_todos.py

@@ -132,6 +132,47 @@ def test_unavailable_provider_does_not_restore_stale_display_tasks(promoted_goal
     assert "todo_stale" in state.read_text()
 
 
+def test_review_materials_follow_canonical_text_without_rewriting_authority(
+    tmp_path, provider, record_format,
+):
+    from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+
+    docs = tmp_path / "docs"
+    docs.mkdir()
+    material = docs / "review.md"
+    material.write_text("# Review\n\n## Proof\nSynthetic evidence.\n", encoding="utf-8")
+    state = tmp_path / "ACTIVE_GOAL_STATE.md"
+    state.write_text("## User Todo\n- [ ] Read `docs/stale.md`.\n", encoding="utf-8")
+    goal = {"id": "goal-a", "repo": str(tmp_path), "state_file": str(state)}
+    records = [{
+        "schema_version": "todo_item_v0", "todo_id": "todo_review", "index": 1,
+        "role": "user", "status": "open", "done": False, "archive_state": "active",
+        "source_section": "User Todo", "task_class": "user_action",
+        "text": ("Read [Review](docs/review.md#proof) and `../outside.md`. "
+                 f"Private [reference]({tmp_path / 'private.md'})."),
+    }]
+    runtime = tmp_path / "runtime"
+    initialize_canonical_authority(runtime, goal["id"], _projection(goal["id"], records, record_format),
+        state_path=state, provider=provider)
+    # A valid stale projection still must not supply current links.
+    state.write_text("## User Todo\n- [ ] Read `docs/stale.md`.\n", encoding="utf-8")
+    before_state = state.read_bytes()
+    before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal["id"])
+    for exists in (True, False):
+        if not exists:
+            material.unlink()
+        fields = active_state_todo_fields(goal, runtime_root=runtime)
+        [todo] = fields["user_todos"]["items"]
+        review, outside = todo["review_materials"]
+        assert review == {"label": "Review", "path": "docs/review.md#proof", "anchor": "proof", "exists": exists}
+        assert outside["exists"] is False
+        assert "resolved_path" not in json.dumps(fields)
+        assert "docs/stale.md" not in json.dumps(fields)
+        assert str(tmp_path) not in json.dumps(todo["review_materials"])
+        assert state.read_bytes() == before_state
+        assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal["id"]) == before
+
+
 def test_empty_canonical_collection_is_an_explicit_empty_read_model(tmp_path):
     state = tmp_path / "state.md"
     state.write_text("# Goal\n\n## Agent Todo\n\n- [ ] Stale work\n")