fix: read global status from canonical Goal sources

loopx/control_plane/coordination/local_authority.py

@@ -7,7 +7,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from copy import deepcopy
 from pathlib import Path
 from typing import Any, NoReturn
@@ -348,12 +348,28 @@ class CanonicalTodoSnapshot:
     keep using fresh reads and the provider's transaction checks.
     """
 
-    def __init__(self) -> None:
+    def __init__(
+        self, *, runtime_root_for_goal: Callable[[str], Path] | None = None,
+    ) -> None:
+        self._runtime_root_for_goal = runtime_root_for_goal
+        self._routes: dict[str, Path | LocalCoordinationAuthorityUnavailable] = {}
         self._reads: dict[
             tuple[Path, str], dict[str, Any] | LocalCoordinationAuthorityUnavailable
         ] = {}
 
     def read(self, *, runtime_root: Path, goal_id: str) -> dict[str, Any]:
+        if self._runtime_root_for_goal is not None:
+            if goal_id not in self._routes:
+                try:
+                    self._routes[goal_id] = self._runtime_root_for_goal(goal_id)
+                except LocalCoordinationAuthorityUnavailable as error:
+                    self._routes[goal_id] = error
+            route = self._routes[goal_id]
+            if isinstance(route, LocalCoordinationAuthorityUnavailable):
+                raise LocalCoordinationAuthorityUnavailable(
+                    str(route), code=route.code, payload=deepcopy(route.payload),
+                )
+            runtime_root = route
         key = (runtime_root.expanduser().resolve(strict=False), goal_id)
         if key not in self._reads:
             try:

loopx/control_plane/status/collection.py

@@ -9,7 +9,10 @@
 from typing import Any, Callable
 
 from ..agents.execution_facts import collect_agent_execution_facts
-from ..coordination.local_authority import CanonicalTodoSnapshot
+from ..coordination.local_authority import (
+    CanonicalTodoSnapshot,
+    LocalCoordinationAuthorityUnavailable,
+)
 from ..goals.acceptance_observation import attach_goal_acceptance_observations
 from ..goals.artifact_lifecycle import attach_goal_artifact_lifecycle_projections
 from ..goals.contract_health import project_contract_health_for_goal
@@ -20,6 +23,7 @@
 )
 from ..runtime.runtime_projection_route import (
     collect_runtime_projection_route_diagnostics,
+    resolve_goal_source_runtime_route,
 )
 from ..projection_envelope_facts import seal_projection_envelope, source_fact
 from ..quota.blocked_retry import overlay_status_turn_retries
@@ -127,10 +131,28 @@ def collect_status(
     )
     history = history_collection.status_history
     history_read_at = now_utc_iso()
-    todo_snapshot = CanonicalTodoSnapshot()
+    source_snapshots: dict[Path, bytes | None] = {}
+
+    def canonical_runtime_for_goal(current_goal_id: str) -> Path:
+        try:
+            route = resolve_goal_source_runtime_route(
+                registry_path=registry_path, goal_id=current_goal_id,
+                runtime_root_override=runtime_root_override, registry=registry,
+                source_snapshots=source_snapshots,
+            )
+        except (OSError, ValueError) as error:
+            raise LocalCoordinationAuthorityUnavailable(
+                "Canonical Todo source route is unavailable",
+                code="canonical_todo_contract_diagnostics_unavailable",
+                payload={"goal_id": current_goal_id, "write_admitted": False,
+                         "legacy_fallback_used": False},
+            ) from error
+        return Path(route["source_runtime_root"])
+
+    todo_snapshot = CanonicalTodoSnapshot(runtime_root_for_goal=canonical_runtime_for_goal)
     contract = context.check_contract(
         registry_path=registry_path,
-        runtime_root_override=str(runtime_root),
+        runtime_root_override=runtime_root_override,
         scan_roots=[],
         limit=limit,
         allow_missing_registry=allow_missing_registry,

tests/control_plane/test_canonical_status_todos.py

@@ -442,3 +442,105 @@ def remove_then_project(**kwargs):
     with pytest.raises(local_authority.LocalCoordinationAuthorityUnavailable):
         status_module.collect_status(**arguments)
     assert 'todo_stale' in state.read_text()
+
+
+@pytest.mark.parametrize("shared_authority", [False, True])
+def test_global_status_reads_source_authority_and_preserves_explicit_override(
+    promoted_goal, tmp_path, provider, record_format, shared_authority, monkeypatch,
+):
+    import loopx.status as status_module
+    from loopx.control_plane.coordination import local_authority
+
+    goal, source_runtime, state = promoted_goal
+    goal = {**goal, "coordination": {"registered_agents": ["agent-a"]}}
+    source_registry = tmp_path / "source-registry.json"
+    source_registry.write_text(json.dumps({
+        "goals": [goal], "common_runtime_root": str(source_runtime),
+    }))
+    shared_runtime = tmp_path / "shared-runtime"
+    registry = tmp_path / "global-registry.json"
+    registry.write_text(json.dumps({
+        "registry_role": "global-local", "common_runtime_root": str(shared_runtime),
+        "goals": [{**goal, "source_registry": str(source_registry)}],
+    }))
+    if shared_authority:
+        shadow = {
+            "schema_version": "todo_item_v0", "todo_id": "todo_shadow", "index": 1,
+            "role": "agent", "status": "open", "done": False,
+            "text": "Shared projection must not replace source work", "priority": "P1",
+            "archive_state": "active", "source_section": "Agent Todo",
+        }
+        initialize_canonical_authority(shared_runtime, goal["id"],
+            _projection(goal["id"], [shadow], record_format),
+            state_path=state, provider=provider)
+    before = local_authority.read_canonical_todos_if_promoted(
+        runtime_root=source_runtime, goal_id=goal["id"],
+    )
+    state_before = state.read_bytes()
+    reads = []
+    original = local_authority.read_canonical_todos_if_promoted
+
+    def observe(**kwargs):
+        reads.append((kwargs["runtime_root"], kwargs["goal_id"]))
+        return original(**kwargs)
+
+    monkeypatch.setattr(local_authority, "read_canonical_todos_if_promoted", observe)
+    arguments = dict(registry_path=registry, runtime_root_override=None,
+                     limit=5, goal_id=goal["id"])
+    payload = status_module.collect_status(**arguments)
+    work = next(item for item in payload["attention_queue"]["items"]
+                if item["goal_id"] == goal["id"] and item.get("agent_todos"))
+    assert [item["todo_id"] for item in work["agent_todos"]["items"]] == ["todo_canonical"]
+    # Contract and attention consume one request-local source read.
+    assert reads == [(source_runtime.resolve(), goal["id"])]
+    assert not any(error.get("code") == "canonical_authority_required"
+                   for error in payload.get("error_diagnostics", []))
+    if shared_authority:
+        overridden = status_module.collect_status(
+            **{**arguments, "runtime_root_override": str(shared_runtime)},
+        )
+        work = next(item for item in overridden["attention_queue"]["items"]
+                    if item["goal_id"] == goal["id"] and item.get("agent_todos"))
+        assert [item["todo_id"] for item in work["agent_todos"]["items"]] == ["todo_shadow"]
+    else:
+        with pytest.raises(local_authority.LocalCoordinationAuthorityUnavailable) as error:
+            status_module.collect_status(
+                **{**arguments, "runtime_root_override": str(shared_runtime)},
+            )
+        assert error.value.code == "canonical_authority_required"
+    after = original(runtime_root=source_runtime, goal_id=goal["id"])
+    assert after["provider_revision"] == before["provider_revision"]
+    assert after["todos"] == before["todos"]
+    assert state.read_bytes() == state_before
+
+
+@pytest.mark.parametrize("source_failure", ["missing", "invalid", "absent_goal"])
+def test_global_status_never_uses_valid_shared_shadow_when_source_route_fails(
+    promoted_goal, tmp_path, provider, record_format, source_failure,
+):
+    import loopx.status as status_module
+
+    goal, source_runtime, state = promoted_goal
+    source_registry = tmp_path / "source-registry.json"
+    if source_failure == "invalid":
+        source_registry.write_text("{invalid")
+    elif source_failure == "absent_goal":
+        source_registry.write_text(json.dumps({
+            "goals": [], "common_runtime_root": str(source_runtime),
+        }))
+    shared_runtime = tmp_path / "shared-runtime"
+    initialize_canonical_authority(shared_runtime, goal["id"],
+        _projection(goal["id"], [], record_format), state_path=state, provider=provider)
+    registry = tmp_path / "global-registry.json"
+    registry.write_text(json.dumps({
+        "registry_role": "global-local", "common_runtime_root": str(shared_runtime),
+        "goals": [{**goal, "source_registry": str(source_registry)}],
+    }))
+    state_before = state.read_bytes()
+    with pytest.raises(LocalCoordinationAuthorityUnavailable) as error:
+        status_module.collect_status(registry_path=registry, runtime_root_override=None,
+                                     limit=5, goal_id=goal["id"])
+    assert error.value.code == "canonical_todo_contract_diagnostics_unavailable"
+    assert error.value.payload["write_admitted"] is False
+    assert error.value.payload["legacy_fallback_used"] is False
+    assert state.read_bytes() == state_before