From 91008a1cd42358dfac7965781f5d5c73f9527010 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 23:02:27 +0530 Subject: [PATCH] fix(collaboration): retain archived linked work in receiver readback Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/collaboration/links.py | 42 ++++++++---- tests/test_collaboration_mcp.py | 18 ++++- tests/test_manager_context_tracking.py | 76 ++++++++++++++++++++++ 3 files changed, 121 insertions(+), 15 deletions(-) diff --git a/loopx/control_plane/collaboration/links.py b/loopx/control_plane/collaboration/links.py index d6b26e56b1..fd993103e1 100644 --- a/loopx/control_plane/collaboration/links.py +++ b/loopx/control_plane/collaboration/links.py @@ -14,13 +14,29 @@ from .goal_instance_scope import collaboration_goal_scope, decide_collaboration_lifecycle -def _core_todos(registry_path, root, goal_id): - result = list_goal_todos( - registry_path=registry_path, runtime_root_arg=str(root), goal_id=goal_id - ) - if result.get("ok") is not True: - raise ValueError("Core Todo authority unavailable") - return {r["todo_id"]: r for r in result.get("todos", []) if r.get("todo_id")} +def _core_todos(registry_path, root, goal_id, *, todo_ids=(), cached_todos=None): + if cached_todos is None: + result = list_goal_todos( + registry_path=registry_path, runtime_root_arg=str(root), goal_id=goal_id + ) + if result.get("ok") is not True: + raise ValueError("Core Todo authority unavailable") + cached_todos = {r["todo_id"]: r for r in result.get("todos", []) if r.get("todo_id")} + for tid in todo_ids: + if tid in cached_todos: + continue + # The queue omits completed history. Resolve only the explicit missing + # links through Core's existing exact read, without exporting history. + try: + result = list_goal_todos( + registry_path=registry_path, runtime_root_arg=str(root), + goal_id=goal_id, todo_id=tid, + ) + todo = result.get("todo") if result.get("ok") is True else None + cached_todos[tid] = todo if isinstance(todo, dict) and todo.get("todo_id") == tid else None + except (OSError, ValueError, RuntimeError): + cached_todos[tid] = None + return cached_todos def read_linked_work(root, registry_path, item, todos_cache, *, include_core_details): @@ -40,7 +56,9 @@ def read_linked_work(root, registry_path, item, todos_cache, *, include_core_det gid, aid = item["goal_id"], item["agent_id"] goal_ref = item.get("goal_ref") cache_key = (gid, (goal_ref or {}).get("goal_instance_id")) - if tids and cache_key not in todos_cache: + core = todos_cache.get(cache_key) + if tids and (cache_key not in todos_cache or + (core is not None and any(tid not in core for tid in tids))): try: with collaboration_goal_scope( registry_path, goal_id=gid, agents=(), caller_goal_ref=goal_ref @@ -48,7 +66,7 @@ def read_linked_work(root, registry_path, item, todos_cache, *, include_core_det admission = decide_collaboration_lifecycle(scope, operation="inbox_observe", record=item) todos_cache[cache_key] = ( None if admission.get("kind") == "omit" - else _core_todos(registry_path, root, gid) + else _core_todos(registry_path, root, gid, todo_ids=tids, cached_todos=core) ) except (OSError, ValueError, RuntimeError): todos_cache[cache_key] = None @@ -72,7 +90,7 @@ def read_linked_work(root, registry_path, item, todos_cache, *, include_core_det def receiver_work_observations(root, registry_path, items): - """Read the actual linked Core provider once per Goal instance.""" + """Cache the active queue and exact linked reads within each Goal instance.""" todos_cache, observations = {}, [] for item in items: work = read_linked_work(root, registry_path, item, todos_cache, include_core_details=True) @@ -130,9 +148,9 @@ def link( record=row, ) if todo_ids: - rows = _core_todos(registry_path, root, goal_id) + rows = _core_todos(registry_path, root, goal_id, todo_ids=todo_ids) for tid in todo_ids: - todo = rows.get(tid, {}) + todo = rows.get(tid) or {} if todo.get("claimed_by") != agent_id and todo.get("bound_agent") != agent_id: raise ValueError("linked Todo must belong to the receiving Agent") path = _root(root) / "links" / (request_id + ".json") diff --git a/tests/test_collaboration_mcp.py b/tests/test_collaboration_mcp.py index caaf80438f..724af737bb 100644 --- a/tests/test_collaboration_mcp.py +++ b/tests/test_collaboration_mcp.py @@ -5,6 +5,8 @@ import subprocess import sys +import pytest + from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client @@ -139,13 +141,14 @@ async def exercise(): asyncio.run(exercise()) -def test_stdio_receiver_links_existing_work_and_returns_without_duplicate_tasks(tmp_path, monkeypatch): +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_stdio_receiver_links_existing_work_and_returns_without_duplicate_tasks(tmp_path, monkeypatch, provider): from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture from loopx.control_plane.collaboration.peers import request, read_inbox from loopx.control_plane.collaboration.inbox import _root isolate_sqlite_runtime(tmp_path, monkeypatch) - registry, root, _ = promoted_create_fixture(tmp_path) + registry, root, _ = promoted_create_fixture(tmp_path, provider=provider) config = json.loads(registry.read_text()) config["goals"][0]["coordination"]["registered_agents"].append("agent-b") registry.write_text(json.dumps(config)) @@ -208,6 +211,14 @@ async def call(name, args): assert [(t["todo_id"], t["title"], t["status"]) for t in resumed["linked_todos"]] == [ (own, "[P1] Current accepted work", "open") ] + cli("todo", "complete", "--goal-id", "goal-a", "--todo-id", own, + "--agent-id", "agent-a", "--evidence", "Verified result", "--no-follow-up") + cli("todo", "archive-completed", "--goal-id", "goal-a", + "--max-active-done", "0", "--execute") + resumed = (await call("read_context", {}))["items"][0]["receiver_followthrough"] + assert resumed["step"] == "return_answer" and resumed["answer_owed"] + assert [(t["todo_id"], t["status"]) for t in resumed["linked_todos"]] == [(own, "done")] + assert path.read_bytes() == original await call("return_result", {"request_id": rid, "text": conclusion}) assert (await call("read_context", {}))["items"] == [] # A short factual answer requires neither a fabricated task nor a link. @@ -230,4 +241,5 @@ async def call(name, args): received = read_inbox(root, registry, "goal-a", "agent-b")["peer_returns"]["items"] assert len(received) == 2 assert next(item for item in received if item["request_id"] == rid)["text"] == conclusion - assert len(cli("todo", "list", "--goal-id", "goal-a")["todos"]) == 2 + assert [row["todo_id"] for row in cli("todo", "list", "--goal-id", "goal-a")["todos"]] == [foreign] + assert cli("todo", "list", "--goal-id", "goal-a", "--todo-id", own)["todo"]["archive_state"] == "archive" diff --git a/tests/test_manager_context_tracking.py b/tests/test_manager_context_tracking.py index 60eebf1b39..ee95dcfb10 100644 --- a/tests/test_manager_context_tracking.py +++ b/tests/test_manager_context_tracking.py @@ -334,6 +334,26 @@ def add(owner, goal="goal-a"): cli("manager-inbox", "link", *inbox, "--evidence-id", ref) follow = cli("manager-inbox", "read", "--goal-id", "goal-a", "--agent-id", "agent-a")["items"][0]["receiver_followthrough"] assert follow["evidence_refs"] == [ref] + + # Archiving completed work changes its storage lane, not the request's + # explicit link or the receiver's need to return an answer. + link_bytes = path.read_bytes() + cli("todo", "complete", "--goal-id", "goal-a", "--todo-id", tid, + "--agent-id", "agent-a", "--evidence", "Verified isolated result", "--no-follow-up") + cli("todo", "archive-completed", "--goal-id", "goal-a", + "--max-active-done", "0", "--execute") + archived = cli("todo", "list", "--goal-id", "goal-a", "--todo-id", tid) + assert archived["todo"]["archive_state"] == "archive" + assert archived["todo"]["status"] == "done" + follow = cli("manager-inbox", "read", "--goal-id", "goal-a", "--agent-id", "agent-a")["items"][0]["receiver_followthrough"] + assert follow["step"] == "return_answer" + assert [(t["todo_id"], t["title"], t["status"]) for t in follow["linked_todos"]] == [ + (tid, "[P1] Verified current work", "done") + ] + assert follow["recorded_decision"] == "adopt" and follow["answer_owed"] is True + assert cli("manager-inbox", "status", *inbox)["rows"][0]["linked_todos"] == follow["linked_todos"] + assert path.read_bytes() == link_bytes + assert cli("todo", "list", "--goal-id", "goal-a", "--todo-id", tid)["authority_read"]["provider_revision"] == archived["authority_read"]["provider_revision"] assert state.exists() @@ -430,3 +450,59 @@ def core_read(**kwargs): assert page[rid]["warnings"] == ["links_unreadable_or_conflicting"] assert page[other]["receiver_followthrough"]["step"] == "assess_request" assert path.read_text() == "{damaged" + + +def test_history_links_are_scoped_and_do_not_hide_other_work(fixture, monkeypatch): + from loopx.control_plane.collaboration.peers import read_inbox + import loopx.control_plane.collaboration.links as links + + root, registry, session, turn, target = fixture + active = {"todo_id": "todo_active", "claimed_by": "worker", "status": "open", "text": "Current work"} + history = {"todo_id": "todo_history", "claimed_by": "worker", "status": "done", "text": "Verified history"} + reads, unavailable = [], False + + def core_read(**kwargs): + tid = kwargs.get("todo_id") + reads.append(tid) + if tid is None: + return {"ok": True, "todos": [active]} + assert tid == "todo_history" + if unavailable: + raise ValueError("Exact source unavailable") + return {"ok": True, "todo": history} + + monkeypatch.setattr(links, "list_goal_todos", core_read) + ids = [] + for key, tid in [("active", "todo_active"), ("history-1", "todo_history"), ("history-2", "todo_history")]: + rid = deliver(root, registry, session=session, + turn=dict(turn, client_turn_id=key), request=target)["request_id"] + acknowledge(root, "research", "worker", rid, "adopt", "Use existing work") + link(root, registry, "research", "worker", rid, [tid], []) + ids.append(rid) + items = {item["request_id"]: item for item in read_inbox(root, registry, "research", "worker")["items"]} + + def observe(): + reads.clear() + rows = links.receiver_work_observations(root, registry, [items[rid] for rid in ids]) + # A later request can need history after the Goal queue is cached; + # repeated links need no second exact lookup or complete history export. + assert reads == [None, "todo_history"] + assert rows[0]["linked_todos"][0]["status"] == "open" + assert rows[0]["evidence_unavailable"] is False + return rows + + assert [row["linked_todos"][0]["status"] for row in observe()] == ["open", "done", "done"] + unavailable = True + missing = observe()[1:] + assert all(row["evidence_unavailable"] for row in missing) + assert all(row["linked_todos"][0]["status"] == "unknown" for row in missing) + assert all(row["linked_todos"][0]["title"] is None for row in missing) + unavailable = False + assert [row["linked_todos"][0]["status"] for row in observe()] == ["open", "done", "done"] + history["claimed_by"] = "peer" + history["text"] = "Other receiver's private work" + foreign = observe()[1:] + assert all(row["linked_todos"][0]["status"] == "unknown" for row in foreign) + assert "Other receiver's private work" not in json.dumps(foreign) + with pytest.raises(ValueError, match="must belong"): + link(root, registry, "research", "worker", ids[1], ["todo_history"], [])