From 1ce8ce751046c49f7aab7d493871323f01535e45 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:23:10 +0800 Subject: [PATCH 1/2] test: prove global status canonical source routing boundaries Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_canonical_status_todos.py | 102 ++++++++++++++++++ 1 file changed, 102 insertions(+) diff --git a/tests/control_plane/test_canonical_status_todos.py b/tests/control_plane/test_canonical_status_todos.py index 7be47baf94..433c543a0e 100644 --- a/tests/control_plane/test_canonical_status_todos.py +++ b/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 From fab17cf63f51d53b3387054e34563d8416ad7622 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 22:57:05 +0800 Subject: [PATCH 2/2] fix: route global status snapshots to canonical Goal sources Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../coordination/local_authority.py | 20 +++++++++++-- loopx/control_plane/status/collection.py | 28 +++++++++++++++++-- 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/loopx/control_plane/coordination/local_authority.py b/loopx/control_plane/coordination/local_authority.py index 06e989e491..e5b315e237 100644 --- a/loopx/control_plane/coordination/local_authority.py +++ b/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: diff --git a/loopx/control_plane/status/collection.py b/loopx/control_plane/status/collection.py index 110dee9cec..ffe2a9ea5e 100644 --- a/loopx/control_plane/status/collection.py +++ b/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,