fix: route global Todo reads through each Goal source

loopx/chat_manager_details.py

@@ -11,6 +11,7 @@
 from .presentation.public_safety import redact_public_text, scan_public_boundary_text
 from .control_plane.todos.list_readback import list_goal_todos
 from .control_plane.effect_runtime import effect_runtime_result
+from .control_plane.runtime.runtime_projection_route import resolve_goal_source_runtime_route
 
 
 def _text(value: object, limit: int = 420) -> str:
@@ -36,10 +37,13 @@ def read_manager_goal_details(
     """Use Core's canonical-first read; never parse a private project document."""
     observed_at = datetime.now(timezone.utc).isoformat()
     try:
+        route = resolve_goal_source_runtime_route(registry_path=registry_path, goal_id=goal_id)
         result = list_goal_todos(
-            registry_path=registry_path, runtime_root_arg=str(runtime_root),
-            goal_id=goal_id,
-            **({"todo_id": todo_id} if todo_id else {}),
+            registry_path=registry_path,
+            # The Manager service root is a default carrier, not a caller's
+            # explicit diagnostic override of a declared source registry.
+            runtime_root_arg=None if route["routed_to_source_registry"] else str(runtime_root),
+            goal_id=goal_id, todo_id=todo_id,
         )
         if result.get("ok") is not True:
             raise ValueError("Todo authority unavailable or conflicting")

loopx/contract.py

@@ -30,14 +30,15 @@
     index_identity,
 )
 from .control_plane.runtime.file_reads import iter_utf8_file_reads
+from .control_plane.runtime.runtime_projection_route import resolve_goal_source_runtime_route
+from .control_plane.projects.registry_codec import load_registry
 from .control_plane.todos.active_state_editing import COMPLETED_WORK_ARCHIVE_HEADING
 from .control_plane.todos.active_state_todo_parser import active_state_incomplete_warning
 from .control_plane.todos.authoring_scope import todo_contract_diagnostics
 from .history import (
     RunHistoryAudit,
     build_run_history_audit,
     collect_history,
-    load_registry,
 )
 from .paths import DEFAULT_RUNTIME_ROOT, LEGACY_RUNTIME_ROOT, rel_or_abs, resolve_runtime_root
 from .registry import inspect_registry, inspect_registry_boundary, registry_goals, resolve_state_file
@@ -467,13 +468,15 @@ def _index_duplicate_warning(
 def _todo_contract_diagnostics(
     registry: dict[str, Any],
     *,
-    runtime_root: Path,
+    registry_path: Path,
+    runtime_root_override: str | None,
     goal_id_filter: str | None = None,
     activation_state_filter: GoalActivationState | str | None = None,
     todo_snapshot: CanonicalTodoSnapshot | None = None,
 ) -> tuple[list[dict[str, Any]], int]:
     diagnostics: list[dict[str, Any]] = []
     checked = 0
+    source_snapshots: dict[Path, bytes | None] = {}
     for goal in registry_goals(registry):
         goal_id = str(goal.get("id") or "")
         if goal_id_filter and goal_id != goal_id_filter:
@@ -494,13 +497,33 @@ def add_error(code: str, message: str) -> None:
                 )
             )
 
+        try:
+            route = resolve_goal_source_runtime_route(
+                registry_path=registry_path, goal_id=goal_id, registry=registry,
+                runtime_root_override=runtime_root_override,
+                source_snapshots=source_snapshots,
+            )
+            if route["routed_to_source_registry"]:
+                source_registry = load_registry(
+                    Path(route["source_registry"]), source_snapshots=source_snapshots,
+                )
+                goal = next(item for item in registry_goals(source_registry)
+                            if str(item.get("id") or "") == goal_id)
+            goal_runtime = Path(route["source_runtime_root"])
+        except (OSError, ValueError, StopIteration) as exc:
+            add_error(
+                "canonical_todo_contract_diagnostics_unavailable",
+                f"{goal_id}: canonical Todo source route unavailable: {exc}",
+            )
+            continue
+
         # The durable fence selects the authority for diagnostics as well as
         # Todo display. Reuse the TS read-model/record validator: the Markdown
         # copy cannot invalidate or rescue a promoted collection.
         try:
             canonical_reader = todo_snapshot.read if todo_snapshot is not None else read_canonical_todos_if_promoted
             canonical = canonical_reader(
-                runtime_root=runtime_root, goal_id=goal_id,
+                runtime_root=goal_runtime, goal_id=goal_id,
             )
         except LocalCoordinationAuthorityUnavailable as exc:
             add_error(exc.code, f"{goal_id}: canonical Todo contract unavailable: {exc}")
@@ -1265,7 +1288,8 @@ def add_global_error(code: str, message: str) -> None:
         _todo_contract_diagnostics(
             registry,
             todo_snapshot=todo_snapshot,
-            runtime_root=runtime_root,
+            registry_path=registry_path,
+            runtime_root_override=runtime_root_override,
             goal_id_filter=goal_id_filter,
             activation_state_filter=activation_state_filter,
         )

loopx/control_plane/todos/list_readback.py

@@ -9,7 +9,7 @@
 from typing import Any, Literal
 
 from ..projects.registry_codec import load_registry
-from ...paths import resolve_runtime_root
+from ..runtime.runtime_projection_route import resolve_goal_source_runtime_route
 from ...rollout_event_log import load_rollout_events, rollout_event_log_path
 from ..goals.state_resolution import resolve_goal_state
 from .contract import normalize_todo_claimed_by, normalize_todo_id, normalize_todo_status
@@ -49,7 +49,16 @@ def list_goal_todos(
         raise ValueError("agent_id must be a public-safe agent token such as codex-main-control")
     if limit is not None and limit < 1:
         raise ValueError("todo list --limit must be at least 1")
-    registry = load_registry(registry_path)
+    source_snapshots: dict[Path, bytes | None] = {}
+    registry = load_registry(registry_path, source_snapshots=source_snapshots)
+    route = resolve_goal_source_runtime_route(
+        registry_path=registry_path, goal_id=goal_id,
+        runtime_root_override=runtime_root_arg, registry=registry,
+        source_snapshots=source_snapshots,
+    )
+    source_registry = Path(route["source_registry"])
+    if route["routed_to_source_registry"]:
+        registry = load_registry(source_registry, source_snapshots=source_snapshots)
     goal, resolved_project, resolved_state_file = resolve_goal_state(
         registry=registry,
         goal_id=goal_id,
@@ -59,11 +68,7 @@ def list_goal_todos(
     if goal is None:
         raise ValueError(f"goal {goal_id!r} is not present in the registry")
 
-    runtime_root = resolve_runtime_root(
-        registry,
-        runtime_root_arg,
-        registry_path=registry_path,
-    )
+    runtime_root = Path(route["source_runtime_root"])
     rollout_events = load_rollout_events(
         rollout_event_log_path(runtime_root, goal_id),
         limit=MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL,
@@ -97,5 +102,5 @@ def list_goal_todos(
     if "canonical_acceptance" in result:
         from .goal_todo_projection import observe_todo_acceptance
         payload = observe_todo_acceptance(payload, result["canonical_acceptance"],
-            registry_path=registry_path, goal_id=goal_id)
+            registry_path=source_registry, goal_id=goal_id)
     return payload

tests/control_plane/test_todo_source_runtime_route.py

@@ -0,0 +1,150 @@
+"""Global read entrypoints follow the declared source, without importing it."""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from canonical_authority_fixture import (
+    assert_todo_read_unchanged,
+    initialize_canonical_authority,
+    isolate_sqlite_runtime,
+)
+from loopx.chat_manager_details import read_manager_goal_details
+from loopx.contract import check_contract
+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.testing.canary_harness import run_json_cli_result
+from loopx.control_plane.todos.list_readback import list_goal_todos
+
+
+GOAL = "source-route-fixture"
+LONG_TEXT = "[P1] Retain the complete original requirement. " + "正文🙂 " * 240 + "Keep the final constraint."
+
+
+def seed(runtime, state, provider, *, prefix="source"):
+    records = [{"schema_version": "todo_item_v0", "todo_id": f"todo_{prefix}_{i}",
+        "role": "agent", "task_class": "advancement_task", "claimed_by": "agent-a",
+        "text": LONG_TEXT if i == 0 else f"Keep {prefix} result {i}.",
+        "note": "Retain the original completion evidence.",
+        "status": "open" if i == 0 else "done", "done": i != 0,
+        "archive_state": "archive" if i == 5 else "active",
+        "source_section": "Completed Work Archive" if i == 5 else "Agent Todo", "index": i + 1}
+        for i in range(6)]
+    initialize_canonical_authority(runtime, GOAL,
+        build_todo_runtime_shadow_projection(goal_id=GOAL, todos=records, handoff_mode="soft_claim"),
+        state_path=state, provider=provider)
+
+
+@pytest.fixture(params=["file", "sqlite"])
+def routed(tmp_path, monkeypatch, request):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project = tmp_path / "project"
+    source = project / ".loopx" / "registry.json"
+    source.parent.mkdir(parents=True)
+    state = project / "state.md"
+    state.write_text("# Goal\n\nKeep original work.\n\n## User Todo\n\n## Agent Todo\n\n## Completed Work Archive\n")
+    goal = {"id": GOAL, "repo": str(project), "state_file": state.name,
+            "domain": "generic", "adapter": {"kind": "generic"},
+            "coordination": {"registered_agents": ["agent-a"]}}
+    source.write_text(json.dumps({"schema_version": 1, "common_runtime_root": ".loopx/runtime", "goals": [goal]}))
+    runtime, shared = project / ".loopx/runtime", tmp_path / "shared"
+    global_registry = shared / "registry.global.json"
+    shared.mkdir()
+    global_registry.write_text(json.dumps({"schema_version": 1, "registry_role": "global-local",
+        "common_runtime_root": str(shared), "goals": [{**goal, "source_registry": str(source),
+            "coordination": {"registered_agents": ["stale-agent"]}}]}))
+    seed(runtime, state, request.param)
+    return source, runtime, global_registry, shared, state, request.param
+
+
+def diagnostics(registry, *, override=None):
+    return check_contract(registry_path=registry, runtime_root_override=override,
+        scan_roots=[], limit=50, goal_id_filter=GOAL, include_public_boundary_scan=False)
+
+
+def test_global_cli_check_and_manager_read_the_original_provider(routed):
+    source, runtime, registry, shared, state, _ = routed
+    before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL)
+    original = state.read_bytes(), source.read_bytes(), registry.read_bytes()
+    expected = list_goal_todos(registry_path=source, goal_id=GOAL, json_references=True)
+    code, actual = run_json_cli_result("todo", "list", "--goal-id", GOAL, registry_path=registry)
+    assert code == 0, actual
+    assert actual.pop("_returncode") == 0  # Test harness metadata, outside the CLI payload.
+    assert_todo_read_unchanged(actual, expected)
+    assert actual["todo_count"] == 5
+    assert {row["todo_id"] for row in actual["todos"]} == {f"todo_source_{i}" for i in range(5)}
+    code, exact = run_json_cli_result("todo", "list", "--goal-id", GOAL,
+        "--todo-id", "todo_source_0", registry_path=registry)
+    assert code == 0 and exact["todo"]["text"] == LONG_TEXT
+    assert diagnostics(registry)["error_diagnostics"] == []
+    manager = read_manager_goal_details(registry, shared, GOAL, owner_scope=True, todo_id="todo_source_0")
+    assert manager["status"] == "read" and manager["authority_revision"] == before["provider_revision"]
+    assert manager["todos"][0]["title"] == LONG_TEXT
+    assert manager["todos"][0]["continuation"] == "Retain the original completion evidence."
+    group = read_manager_goal_details(registry, shared, GOAL, owner_scope=False, todo_id="todo_source_0")
+    assert group["status"] == "read" and "continuation" not in group["todos"][0]
+    history = list_goal_todos(registry_path=registry, goal_id=GOAL,
+        role="agent", status="done", read_scope="completed_history", todo_id="todo_source_5")
+    assert history["todo"]["text"] == "Keep source result 5."
+    assert (state.read_bytes(), source.read_bytes(), registry.read_bytes()) == original
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) == before
+    assert not (shared / "goals" / GOAL).exists(), "Reading must not create a second authority"
+
+
+def test_explicit_runtime_override_retains_its_diagnostic_meaning(routed):
+    source, _, registry, shared, state, provider = routed
+    seed(shared, state, provider, prefix="explicit")
+    override = str(shared)
+    expected = list_goal_todos(registry_path=registry, goal_id=GOAL, runtime_root_arg=override)
+    assert expected["todos"][0]["todo_id"] == "todo_explicit_0"
+    code, actual = run_json_cli_result("todo", "list", "--goal-id", GOAL,
+        registry_path=registry, runtime_root=shared)
+    assert code == 0 and actual["todos"][0]["todo_id"] == "todo_explicit_0"
+    assert list_goal_todos(registry_path=registry, goal_id=GOAL)["todos"][0]["todo_id"] == "todo_source_0"
+    assert list_goal_todos(registry_path=source, goal_id=GOAL)["todos"][0]["todo_id"] == "todo_source_0"
+    assert diagnostics(registry, override=override)["error_diagnostics"] == []
+
+
+@pytest.mark.parametrize("failure", ["missing", "unreadable", "wrong_goal"])
+def test_bad_source_never_falls_back_to_a_valid_shared_authority(routed, failure):
+    source, _, registry, shared, state, provider = routed
+    seed(shared, state, provider, prefix="decoy")
+    if failure == "wrong_goal":
+        payload = json.loads(source.read_text())
+        payload["goals"][0]["id"] = "unrelated-goal"
+        source.write_text(json.dumps(payload))
+    else:
+        source.unlink()
+        if failure == "unreadable":
+            source.mkdir()  # A real IO refusal, independent of the test user's privilege.
+    with pytest.raises((OSError, ValueError), match="source_registry"):
+        list_goal_todos(registry_path=registry, goal_id=GOAL)
+    code, result = run_json_cli_result("todo", "list", "--goal-id", GOAL, registry_path=registry)
+    assert code == 1 and not result["ok"] and "source_registry" in result["error"]
+    errors = diagnostics(registry)["error_diagnostics"]
+    assert len(errors) == 1 and errors[0]["code"] == "canonical_todo_contract_diagnostics_unavailable"
+    assert "source_registry" in errors[0]["message"]
+    assert read_manager_goal_details(registry, shared, GOAL, owner_scope=True)["status"] == "unavailable"
+    assert list_goal_todos(registry_path=registry, goal_id=GOAL,
+        runtime_root_arg=str(shared))["todos"][0]["todo_id"] == "todo_decoy_0"
+
+
+def test_each_read_rechecks_the_source_registration(routed):
+    source, _, registry, shared, state, provider = routed
+    assert list_goal_todos(registry_path=registry, goal_id=GOAL)["todos"][0]["todo_id"] == "todo_source_0"
+    replacement = state.parent / "replacement-runtime"
+    seed(replacement, state, provider, prefix="replacement")
+    payload = json.loads(source.read_text())
+    payload["common_runtime_root"] = str(replacement)
+    source.write_text(json.dumps(payload))
+    assert list_goal_todos(registry_path=registry, goal_id=GOAL)["todos"][0]["todo_id"] == "todo_replacement_0"
+    assert diagnostics(registry)["error_diagnostics"] == []
+    manager = read_manager_goal_details(registry, shared, GOAL, owner_scope=True, todo_id="todo_replacement_1")
+    assert manager["status"] == "read" and manager["todos"][0]["title"] == "Keep replacement result 1."
+    payload["goals"] = []
+    source.write_text(json.dumps(payload))
+    with pytest.raises(ValueError, match="absent from source_registry"):
+        list_goal_todos(registry_path=registry, goal_id=GOAL)
+    assert "absent from source_registry" in diagnostics(registry)["error_diagnostics"][0]["message"]
+    assert read_manager_goal_details(registry, shared, GOAL, owner_scope=True)["status"] == "unavailable"