fix(collaboration): recover archived linked work in request readback

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,15 +56,17 @@ 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
             ) as scope:
                 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")

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"

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"], [])