diff --git a/loopx/chat_manager_details.py b/loopx/chat_manager_details.py index 38edf6d224..f43881f18a 100644 --- a/loopx/chat_manager_details.py +++ b/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") diff --git a/loopx/contract.py b/loopx/contract.py index 1f9f53f5d3..7910965ea2 100644 --- a/loopx/contract.py +++ b/loopx/contract.py @@ -30,6 +30,8 @@ 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 @@ -37,7 +39,6 @@ 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, ) diff --git a/loopx/control_plane/todos/list_readback.py b/loopx/control_plane/todos/list_readback.py index a0aec0f462..e52b836eff 100644 --- a/loopx/control_plane/todos/list_readback.py +++ b/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 diff --git a/tests/control_plane/test_todo_source_runtime_route.py b/tests/control_plane/test_todo_source_runtime_route.py new file mode 100644 index 0000000000..de9b3394d2 --- /dev/null +++ b/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"