From fb02cc2b50dfba1ed9479e76434bc9d8a1414fb6 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Tue, 22 Sep 2026 09:43:42 +0800 Subject: [PATCH 1/2] fix(collaboration): paginate pending receiver requests Signed-off-by: duanjialing.777 --- loopx/cli_commands/manager_inbox.py | 7 +- loopx/collaboration_mcp.py | 10 +- loopx/control_plane/collaboration/inbox.py | 34 +++- loopx/control_plane/collaboration/peers.py | 6 +- tests/test_inbox_pagination.py | 214 +++++++++++++++++++++ 5 files changed, 261 insertions(+), 10 deletions(-) create mode 100644 tests/test_inbox_pagination.py diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py index 0ad1f7746e..d2a3918d0f 100644 --- a/loopx/cli_commands/manager_inbox.py +++ b/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." ) diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py index f2b0be232c..9c7ce8465b 100644 --- a/loopx/collaboration_mcp.py +++ b/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( diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py index 2e29b49e24..580a1a6e77 100644 --- a/loopx/control_plane/collaboration/inbox.py +++ b/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." + ), } diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py index 6d14a7eeb6..76560c3502 100644 --- a/loopx/control_plane/collaboration/peers.py +++ b/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 diff --git a/tests/test_inbox_pagination.py b/tests/test_inbox_pagination.py new file mode 100644 index 0000000000..efcb664e8b --- /dev/null +++ b/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()) From 8506dbc8a1666c09228f3379686681b1579e5cc7 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Tue, 22 Sep 2026 09:44:11 +0800 Subject: [PATCH 2/2] docs(collaboration): document receiver recovery and qualification limits Signed-off-by: duanjialing.777 --- .../capable-manager-semantic-handoff-v0.md | 8 ++++++++ loopx/capabilities/manager_context/README.md | 18 ++++++++++++++++++ loopx/control_plane/collaboration/README.md | 8 ++++++++ 3 files changed, 34 insertions(+) diff --git a/docs/architecture/rfcs/capable-manager-semantic-handoff-v0.md b/docs/architecture/rfcs/capable-manager-semantic-handoff-v0.md index 7d57b6d1ab..649e200b91 100644 --- a/docs/architecture/rfcs/capable-manager-semantic-handoff-v0.md +++ b/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: diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index a5bef95a4e..6ba82df77b 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/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 +``` + +Use the same registry, runtime root, Goal and Agent for every page. The scoped +MCP equivalent is `read_context(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: diff --git a/loopx/control_plane/collaboration/README.md b/loopx/control_plane/collaboration/README.md index 4184ccca37..160617fb3a 100644 --- a/loopx/control_plane/collaboration/README.md +++ b/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.