fix(collaboration): recover pending requests beyond the first inbox page

docs/architecture/rfcs/capable-manager-semantic-handoff-v0.md

@@ -81,6 +81,14 @@ transfer or shared-intent mutation. This is a bounded R3 slice, not M2–M4/G1
 completion; external-audience peer forwarding and cross-host continuation remain
 with their existing roadmap owners.
 
+Receiver recovery now exposes later pending requests through the existing CLI
+and identity-scoped MCP `next_cursor`. Deferred requests awaiting conclusions no
+longer prevent access beyond the first 20. `tests/test_inbox_pagination.py`
+qualifies 45 durable requests, process restart, scope rejection, live registration
+revocation and independent result return. This covers the local receiver paging
+part of A20 and restart evidence for A5/A13. It does not qualify cross-day managed
+execution, live Lark, frontend journey completion or the remaining M1–M4 work.
+
 ## 4. Current-system contract: audited facts
 
 The baseline already has substantial reusable machinery:

loopx/capabilities/manager_context/README.md

@@ -201,6 +201,24 @@ The receiving Agent still owns relevance and priority; normal context delivery
 never changes its Todos or interrupts its current work.
 
 `manager-inbox read` records the first provision of context to the receiver.
+Each response returns at most 20 pending requests and now includes `next_cursor`.
+When `has_more` is true, pass that cursor to read later requests without first
+concluding the earlier ones:
+
+```sh
+loopx manager-inbox read --goal-id research --agent-id worker --cursor <next_cursor>
+```
+
+Use the same registry, runtime root, Goal and Agent for every page. The scoped
+MCP equivalent is `read_context(cursor=<next_cursor>)`; a call without arguments
+still reads the first page. A cursor survives process restart and removal or
+completion of its anchor request. It is a navigation position, not a grant.
+Each call checks current registration and records reads only for returned requests.
+`next_cursor: null` ends this scan, not the outstanding work. Pages are live,
+ordered by request id; restart without a cursor to find new requests sorted
+before the last position. Peer results retain their separate consumption flow
+and are not paginated by this cursor. `status --offset/--limit` remains separate.
+
 After `acknowledge`, the request remains in the turn-start hook until the worker
 publishes a conclusion. The worker uses `link` for canonical Todo/evidence lineage
 and `report` to publish the answer intended for the original audience:

loopx/cli_commands/manager_inbox.py

@@ -49,12 +49,16 @@ def register_manager_inbox(subparsers, add_format):
     parser.add_argument("--evidence-id", action="append", default=[])
     parser.add_argument("--offset", type=int, default=0)
     parser.add_argument("--limit", type=int, default=8)
+    parser.add_argument("--cursor", help="For read: continue with the previous page's next_cursor.")
     parser.add_argument("--decision", choices=("adopt", "defer", "reject", "no_change"))
     parser.add_argument("--reason")
 
 
 def handle_manager_inbox(args, registry_path, runtime_root):
     try:
+        cursor = getattr(args, "cursor", None)
+        if cursor is not None and args.manager_inbox_action != "read":
+            raise ValueError("--cursor is only supported for read")
         if args.manager_inbox_action == "configure-ssh-read-scope":
             from ..capabilities.manager_context.ssh_evidence import configure
             result = configure(runtime_root, channel=args.channel_id or "", host=args.ssh_host,
@@ -92,7 +96,8 @@ def handle_manager_inbox(args, registry_path, runtime_root):
             result = consume_return(runtime_root, args.goal_id, args.agent_id, args.request_id)
         elif args.manager_inbox_action == "read":
             from ..control_plane.collaboration.peers import read_inbox
-            result = read_inbox(runtime_root, registry_path, args.goal_id, args.agent_id, workspace=Path.cwd())
+            result = read_inbox(runtime_root, registry_path, args.goal_id, args.agent_id,
+                                workspace=Path.cwd(), cursor=cursor)
             result["followthrough"] = (
                 "After reading and deciding, associate Core work with manager-inbox link. Then use manager-inbox report --phase conclusion --reply-text to return this request's concrete result, replan decision, or explicit blocker/defer reason to its original audience automatically. Use optional --phase decision only for meaningful interim news during longer work. Adoption/linking alone is not a completed exchange. Do not wait for the owner to ask again. Write audience-ready text, not private deliberation."
             )

loopx/collaboration_mcp.py

@@ -66,9 +66,13 @@ def check_scope():
         _goal(registry, goal_id, agent_id)
 
     @server.tool()
-    def read_context() -> dict:
-        """Read pending requests, material version checks and unconsumed peer results."""
-        return read_inbox(root, registry, goal_id, agent_id, workspace=workspace)
+    def read_context(cursor: str | None = None) -> dict:
+        """Read pending requests, material version checks and unconsumed peer results.
+
+        Follow next_cursor for later requests. Omit cursor to start a fresh scan.
+        Pages are live; reading all pages does not complete outstanding work.
+        """
+        return read_inbox(root, registry, goal_id, agent_id, workspace=workspace, cursor=cursor)
 
     @server.tool()
     def assess_request(

loopx/control_plane/collaboration/README.md

@@ -36,3 +36,11 @@ request amendment/cancellation, cross-Goal/host delegation, dynamic Agent creati
 and lifecycle supervision remain owned by their existing roadmap contracts.
 The nested-coordinator regression and the [managed delivery demo](../../../examples/collaboration-delivery/README.md)
 exercise this boundary without imposing a maximum tree depth or a manager hop.
+
+Pending request pagination also belongs to `inbox.py`. Its stateless cursor
+binds the resolved runtime root, Goal and receiving Agent to a request-id
+position. CLI and MCP use the same live 20-request pages. Invalid cursors and
+unreadable entry directories fail before request read receipts are written.
+No scan index, receipt migration or additional authority store is required.
+See [receiver pagination](../../capabilities/manager_context/README.md#a-delegation-returns-automatically)
+for restart and concurrent-arrival behavior.

loopx/control_plane/collaboration/inbox.py

@@ -72,14 +72,36 @@ def normalize_request(value: Any) -> dict | None:
         raise ValueError(str(exc)) from exc
 
 
-def pending(runtime_root: Path, goal_id: str, agent_id: str) -> dict:
+def pending(
+    runtime_root: Path, goal_id: str, agent_id: str, *, cursor: str | None = None
+) -> dict:
+    scope = _hash(["pending_requests_v1", str(runtime_root.resolve()), goal_id, agent_id])
+    after = ""
+    if cursor is not None:
+        if not isinstance(cursor, str) or not re.fullmatch(
+            r"1:[a-f0-9]{64}:[a-f0-9]{64}", cursor
+        ):
+            raise ValueError("invalid pending request cursor")
+        _, cursor_scope, after = cursor.split(":")
+        if cursor_scope != scope:
+            raise ValueError("pending request cursor scope mismatch")
     folder = (
         _root(runtime_root)
         / "entries"
         / _hash(dict(goal_id=goal_id, agent_id=agent_id))
     )
+    try:
+        paths = sorted(folder.iterdir())
+    except FileNotFoundError:
+        paths = []
     items = []
-    for path in sorted(folder.glob("*.json")):
+    for path in paths:
+        if path.suffix != ".json":
+            continue
+        if not re.fullmatch(r"[a-f0-9]{64}", path.stem):
+            raise ValueError("invalid context request filename")
+        if path.stem <= after:
+            continue
         decided = (_root(runtime_root) / "decisions" / path.name).exists()
         if decided and not needs_conclusion(runtime_root, path.stem):
             continue
@@ -88,6 +110,7 @@ def pending(runtime_root: Path, goal_id: str, agent_id: str) -> dict:
             item.get("schema_version") != ENTRY_SCHEMA
             or item.get("goal_id") != goal_id
             or item.get("agent_id") != agent_id
+            or item.get("request_id") != path.stem
         ):
             raise ValueError("context inbox scope mismatch")
         if decided:
@@ -107,7 +130,12 @@ def pending(runtime_root: Path, goal_id: str, agent_id: str) -> dict:
         **({"peer_returns": peer_returns} if peer_returns["items"] else {}),
         "items": items[:20],
         "has_more": len(items) > 20,
-        "instruction": REQUEST_TRIAGE_INSTRUCTION,
+        "next_cursor": f"1:{scope}:{items[19]['request_id']}" if len(items) > 20 else None,
+        "instruction": (
+            REQUEST_TRIAGE_INSTRUCTION
+            + " Follow next_cursor to read later pending requests. Restart without a cursor "
+            "to discover new requests before it. The end of a page sequence is not work completion."
+        ),
     }
 
 

loopx/control_plane/collaboration/peers.py

@@ -344,13 +344,12 @@ def input_readiness(registry, goal_id, brief, *, workspace=None):
     return result
 
 
-def read_inbox(root, registry, goal_id, agent_id, *, workspace=None):
+def read_inbox(root, registry, goal_id, agent_id, *, workspace=None, cursor=None):
     from .inbox import pending
     from .inbox import record_read
 
     _goal(registry, goal_id, agent_id)
-    result = pending(root, goal_id, agent_id)
-    record_read(root, result["items"])
+    result = pending(root, goal_id, agent_id, cursor=cursor)
     for item in result["items"]:
         if item.get("brief"):
             item["input_readiness"] = input_readiness(
@@ -366,6 +365,7 @@ def read_inbox(root, registry, goal_id, agent_id, *, workspace=None):
         "Finish the original request with return_result, including evidence and remaining gaps. "
         "Adoption, file hashes and returned opinions are not independent acceptance or Todo completion."
     )
+    record_read(root, result["items"])
     return result
 
 

tests/test_inbox_pagination.py

@@ -0,0 +1,214 @@
+"""Receiver recovery through real CLI processes and identity-bound MCP stdio."""
+
+import asyncio
+import json
+import subprocess
+import sys
+
+import pytest
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from loopx.control_plane.collaboration.inbox import acknowledge
+from loopx.control_plane.collaboration.peers import request
+
+
+@pytest.fixture
+def inbox(tmp_path):
+    registry = tmp_path / "registry.json"
+    config = {"goals": [
+        {"id": goal, "repo": str(tmp_path),
+         "coordination": {"registered_agents": ["sender", "receiver", "other"]}}
+        for goal in ("delivery", "another")
+    ]}
+    registry.write_text(json.dumps(config))
+    brief = {
+        "schema_version": "collaboration_brief_v0",
+        "purpose": "Review the corrected allocation",
+        "context": "Proportional rounding was rejected. Reserve two units.",
+        "constraints": ["Do not place orders"],
+        "inputs": [],
+        "acceptance": ["Check the reserve and capacity"],
+        "return_requirement": "Return findings and remaining gaps",
+    }
+
+    def seed(count):
+        return sorted(
+            request(tmp_path, registry, "delivery", "sender", "receiver",
+                    f"review-{number}", brief)["request_id"]
+            for number in range(count)
+        )
+
+    return tmp_path, registry, seed, brief
+
+
+def cli(root, registry, action="read", *args, agent="receiver", goal="delivery", ok=True):
+    result = subprocess.run([
+        sys.executable, "-m", "loopx.cli",
+        "--runtime-root", str(root), "--registry", str(registry),
+        "manager-inbox", action, "--goal-id", goal, "--agent-id", agent, *args,
+    ], capture_output=True, text=True, timeout=30)
+    assert result.returncode == (0 if ok else 1), (result.stdout, result.stderr)
+    return json.loads(result.stdout)
+
+
+def read_ids(root):
+    return {path.stem for path in (root / ".local/manager-context/reads").glob("*.json")}
+
+
+def ids(page):
+    return [row["request_id"] for row in page["items"]]
+
+
+def test_cli_recovers_later_context_without_concluding_deferred_requests(inbox):
+    root, registry, seed, brief = inbox
+    expected = seed(45)
+    for request_id in expected[:20]:
+        acknowledge(root, "delivery", "receiver", request_id, "defer", "Await input.")
+    first = cli(root, registry)
+    assert ids(first) == expected[:20] and first["has_more"]
+    assert all(row["receiver_decision_recorded"] for row in first["items"])
+    assert read_ids(root) == set(expected[:20])
+    assert first.get("next_cursor"), "pending request 21 has no recovery cursor"
+
+    second = cli(root, registry, "read", "--cursor", first["next_cursor"])
+    assert ids(second) == expected[20:40] and second["has_more"]
+    assert second["items"][0]["brief"] == brief
+    assert read_ids(root) == set(expected[:40])
+    # Every CLI invocation starts a fresh process; replay never consumes a page.
+    assert cli(root, registry, "read", "--cursor", first["next_cursor"]) == second
+    third = cli(root, registry, "read", "--cursor", second["next_cursor"])
+    assert ids(third) == expected[40:]
+    assert not third["has_more"] and third["next_cursor"] is None
+    assert read_ids(root) == set(expected)
+
+    chosen = expected[20]
+    cli(root, registry, "acknowledge", "--request-id", chosen,
+        "--decision", "adopt", "--reason", "Check the corrected reserve.")
+    cli(root, registry, "report", "--request-id", chosen,
+        "--reply-text", "Reserve check passed; capacity remains unverified.")
+    returned = cli(root, registry, agent="sender")["peer_returns"]["items"]
+    assert [(row["request_id"], row["text"]) for row in returned] == [
+        (chosen, "Reserve check passed; capacity remains unverified.")]
+    cli(root, registry, "acknowledge-return", "--request-id", chosen, agent="sender")
+    assert "peer_returns" not in cli(root, registry, agent="sender")
+    assert ids(cli(root, registry)) == expected[:20]
+    assert chosen not in ids(cli(root, registry, "read", "--cursor", first["next_cursor"]))
+
+    # Neither a removed anchor nor completed earlier rows shifts the continuation.
+    anchor = next((root / ".local/manager-context/entries").glob(f"*/{expected[19]}.json"))
+    anchor.unlink()
+    cli(root, registry, "report", "--request-id", expected[0], "--reply-text", "Still blocked.")
+    resumed = cli(root, registry, "read", "--cursor", first["next_cursor"])
+    assert ids(resumed) == expected[21:41]
+
+
+@pytest.mark.parametrize("count", [0, 20, 21])
+def test_cli_page_boundaries(inbox, count):
+    root, registry, seed, _ = inbox
+    expected = seed(count)
+    page = cli(root, registry)
+    assert ids(page) == expected[:20]
+    assert page["has_more"] is (count > 20)
+    assert bool(page["next_cursor"]) is (count > 20)
+    assert read_ids(root) == set(expected[:20])
+
+
+def test_cli_rejects_wrong_scope_and_malformed_cursors_before_receipts(inbox):
+    root, registry, seed, _ = inbox
+    expected = seed(21)
+    cursor = cli(root, registry)["next_cursor"]
+    for token in ("", "../entry", "2:" + "a" * 64 + ":" + "b" * 64, cursor + "x"):
+        result = cli(root, registry, "read", "--cursor", token, ok=False)
+        assert "cursor" in result["error"]
+    for overrides in ({"agent": "other"}, {"goal": "another"}):
+        assert "scope" in cli(root, registry, "read", "--cursor", cursor,
+                              ok=False, **overrides)["error"]
+    other_root = root / "other-runtime"
+    assert "scope" in cli(other_root, registry, "read", "--cursor", cursor, ok=False)["error"]
+    assert not other_root.exists()
+    assert "read" in cli(root, registry, "status", "--cursor", cursor, ok=False)["error"]
+    assert read_ids(root) == set(expected[:20])
+
+
+def test_new_arrivals_before_cursor_are_found_by_fresh_scan(inbox):
+    root, registry, seed, _ = inbox
+    expected = seed(45)
+    entry = next((root / ".local/manager-context/entries").glob(f"*/{expected[0]}.json"))
+    original = entry.read_bytes()
+    entry.unlink()
+    first = cli(root, registry)
+    assert ids(first) == expected[1:21]
+    entry.write_bytes(original)
+    second = cli(root, registry, "read", "--cursor", first["next_cursor"])
+    assert ids(second) == expected[21:41]
+    assert expected[0] in ids(cli(root, registry))
+
+
+@pytest.mark.parametrize("damage", ["directory", "identity", "schema", "filename"])
+def test_unreadable_or_conflicting_entries_fail_without_read_receipts(inbox, damage):
+    root, registry, seed, _ = inbox
+    expected = seed(1)
+    entry = next((root / ".local/manager-context/entries").glob(f"*/{expected[0]}.json"))
+    if damage == "directory":
+        folder = entry.parent
+        entry.unlink()
+        folder.rmdir()
+        folder.write_text("not a directory")
+    elif damage == "filename":
+        entry.rename(entry.with_name("invalid.json"))
+    else:
+        row = json.loads(entry.read_text())
+        row["request_id" if damage == "identity" else "schema_version"] = "invalid"
+        entry.write_text(json.dumps(row))
+    assert not cli(root, registry, ok=False)["ok"]
+    assert read_ids(root) == set()
+
+
+def test_mcp_continuation_survives_restart_and_rechecks_registration(inbox):
+    root, registry, seed, brief = inbox
+    expected = seed(45)
+    params = StdioServerParameters(command=sys.executable, args=[
+        "-m", "loopx.collaboration_mcp", "--runtime-root", str(root),
+        "--registry", str(registry), "--goal-id", "delivery",
+        "--agent-id", "receiver", "--workspace", str(root),
+    ])
+
+    async def exercise():
+        async with stdio_client(params) as (read, write), ClientSession(read, write) as session:
+            await session.initialize()
+            tools = await session.list_tools()
+            read_tool = next(tool for tool in tools.tools if tool.name == "read_context")
+            assert set(read_tool.inputSchema["properties"]) == {"cursor"}
+            result = await session.call_tool("read_context", {})
+            assert not result.isError
+            first = json.loads(result.content[0].text)
+            assert ids(first) == expected[:20]
+            assert read_ids(root) == set(expected[:20])
+        async with stdio_client(params) as (read, write), ClientSession(read, write) as session:
+            await session.initialize()
+            result = await session.call_tool("read_context", {"cursor": first["next_cursor"]})
+            assert not result.isError
+            second = json.loads(result.content[0].text)
+            assert ids(second) == expected[20:40]
+            assert second["items"][0]["brief"] == brief
+            assert read_ids(root) == set(expected[:40])
+            result = await session.call_tool("read_context", {"cursor": "../invalid"})
+            assert result.isError
+            # Revocation affects the same running server and an otherwise valid cursor.
+            config = json.loads(registry.read_text())
+            config["goals"][0]["coordination"]["registered_agents"] = ["sender"]
+            registry.write_text(json.dumps(config))
+            result = await session.call_tool("read_context", {"cursor": second["next_cursor"]})
+            assert result.isError
+            assert read_ids(root) == set(expected[:40])
+            config["goals"][0]["coordination"]["registered_agents"].append("receiver")
+            config["goals"][0]["status"] = "stopped"
+            registry.write_text(json.dumps(config))
+            result = await session.call_tool("read_context", {"cursor": second["next_cursor"]})
+            assert not result.isError
+            third = json.loads(result.content[0].text)
+            assert ids(third) == expected[40:]
+            assert third["next_cursor"] is None
+
+    asyncio.run(exercise())