fix(inbox): honor exact source reads without draining pending work

loopx/capabilities/manager_context/README.md

@@ -691,6 +691,24 @@ loopx manager-inbox request --goal-id allocation --agent-id builder \
   --brief-file review.json --parent-request-id <received-request-id>
 ```
 
+To inspect an original source by its saved request identity, including a concluded
+request, use the same registered Goal and receiving Agent:
+
+```sh
+loopx manager-inbox read --goal-id allocation --agent-id builder --request-id <received-request-id>
+```
+
+`--request-id` now selects that exact request instead of being silently ignored.
+It retains the full current brief, original ancestry and typed receipt state;
+reading a conclusion does not reopen its work or repeat its result. The read
+records only that request's receipt and does not deliver unrelated peer results.
+An unknown identity, another recipient or Goal lifetime fails without falling
+back to a pending page. Exact reads cannot use inbox cursors. Omit the option for
+the existing paged pending inbox, peer results and operation handoffs. These are
+trusted local inspection commands, not new source grants or execution authority.
+App/project dispatch, receiver adoption and original-conversation return retain
+their separate existing entrypoints and acceptance.
+
 In App and Goal conversations, the existing read tools expose the same bounded
 local observation without shell access: search `view=agents`, then read
 `{"view":"agent_route","goal_id":"allocation","agent_id":"reviewer"}` through

loopx/cli_commands/manager_inbox.py

@@ -69,7 +69,7 @@ def register_manager_inbox(subparsers, add_format):
     parser.add_argument("--local-delivery-scope", choices=("all_registered", "selected"))
     parser.add_argument("--sender-id", help="For configuring a new delivery source: its independently verified sender.")
     parser.add_argument("--execute", action="store_true")
-    parser.add_argument("--request-id")
+    parser.add_argument("--request-id", help="For read: inspect this exact current or concluded request instead of the pending inbox.")
     parser.add_argument(
         "--phase", choices=("decision", "conclusion"), default="conclusion"
     )
@@ -237,14 +237,17 @@ def handle_manager_inbox(args, registry_path, runtime_root):
             )
         elif args.manager_inbox_action == "read":
             from ..control_plane.collaboration.peers import read_inbox
+            request_id = getattr(args, "request_id", None)
             result = read_inbox(runtime_root, registry_path, args.goal_id, args.agent_id,
-                                workspace=Path.cwd(), cursor=cursor, operation_cursor=operation_cursor)
+                                workspace=Path.cwd(), cursor=cursor, operation_cursor=operation_cursor,
+                                request_id=request_id)
             result["instruction"] += " " + collaboration_answer_instruction()
-            result["followthrough"] += (
-                " CLI: record assessment with manager-inbox acknowledge; optionally associate existing work "
-                "with manager-inbox link. Return audience-ready results with manager-inbox report "
-                "--phase conclusion --reply-text; use --phase decision for meaningful interim news."
-            )
+            if request_id is None:
+                result["followthrough"] += (
+                    " CLI: record assessment with manager-inbox acknowledge; optionally associate existing work "
+                    "with manager-inbox link. Return audience-ready results with manager-inbox report "
+                    "--phase conclusion --reply-text; use --phase decision for meaningful interim news."
+                )
             if any(item.get("request_revision") is not None for item in result.get("items", [])):
                 result["followthrough"] += (
                     " When request_revision is supplied, acknowledge with --request-revision and --request-digest, "

loopx/control_plane/collaboration/inbox.py

@@ -145,15 +145,59 @@ def plan_request_publication(request, *, entry, operation=None):
         raise ValueError(str(exc)) from exc
 
 
+def _inbox_item(row: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
+    """Present the existing typed receipt state without changing its lifecycle."""
+    item = {**row, "inbox_state": state["kind"]}
+    item.pop("request_control", None)
+    if state.get("request_control_view") is not None:
+        view = state["request_control_view"]
+        item.update(original_brief=row["brief"], brief=view["brief"], request_control_view=view,
+            request_revision=view["revision"], request_revision_digest=view["revision_digest"])
+        item["source_request"] = {"request_id": row["request_id"], "request_revision": view["revision"],
+            "request_digest": view["revision_digest"]}
+    item["recorded_decision"] = state["recorded_decision"]
+    if state["recorded_decision"] is not None:
+        item["receiver_decision_recorded"] = True
+        item["next_action"] = "Review this request's recorded decision and linked work before returning its actual result. A recorded adoption is not a completed task; preserve the current owner and do not reprioritize unrelated work."
+    if state["warnings"]:
+        item["warnings"] = state["warnings"]
+        item["next_action"] = "Receiver receipt readback is unavailable or conflicting; recover the original receipt before continuing this request. Do not repeat the decision, result or execution. Other requests can continue."
+    elif state["kind"] == "settled":
+        item["next_action"] = "Inspect the concluded request as historical source. Reading it does not reopen work, repeat its result or grant execution authority."
+    elif state.get("request_control_view") is not None:
+        item["next_action"] = state["request_control_view"]["receiver_instruction"]
+    return item
+
+
 def pending(
     runtime_root: Path,
     goal_id: str,
     agent_id: str,
     *,
     cursor: str | None = None,
     operation_cursor: str | None = None,
+    request_id: str | None = None,
     scope: CollaborationGoalScope | None = None,
 ) -> dict:
+    if request_id is not None:
+        if cursor is not None or operation_cursor is not None:
+            raise ValueError("an exact request read cannot use inbox cursors")
+        try:
+            row = _entry(runtime_root, goal_id, agent_id, request_id, scope=scope)
+        except FileNotFoundError as exc:
+            raise ValueError("context request not found for this Goal and Agent") from exc
+        if row.get("schema_version") != (EXACT_ENTRY_SCHEMA if scope is not None and scope.exact else ENTRY_SCHEMA):
+            raise ValueError("context inbox scope mismatch")
+        if scope is not None:
+            from .goal_instance_scope import decide_collaboration_lifecycle
+
+            lifecycle = decide_collaboration_lifecycle(scope, operation="inbox_observe", record=row)
+            if lifecycle.get("kind") == "omit":
+                raise ValueError("context request is unavailable in this Goal lifetime")
+        state = _pending_receipt_states(runtime_root, [row])[0]
+        return {"ok": True, "request_id": request_id, "items": [_inbox_item(row, state)],
+            "has_more": False, "next_cursor": None,
+            "instruction": REQUEST_TRIAGE_INSTRUCTION + " This is an exact source read, including concluded requests; it does not reopen their work."}
     cursor_scope = _hash(
         (
             [
@@ -206,24 +250,7 @@ def append_batch():
                 )
                 if lifecycle.get("kind") == "omit":
                     continue
-            item = {**row, "inbox_state": state["kind"]}
-            item.pop("request_control", None)
-            if state.get("request_control_view") is not None:
-                view = state["request_control_view"]
-                item.update(original_brief=row["brief"], brief=view["brief"], request_control_view=view,
-                    request_revision=view["revision"], request_revision_digest=view["revision_digest"])
-                item["source_request"] = {"request_id": row["request_id"], "request_revision": view["revision"],
-                    "request_digest": view["revision_digest"]}
-            item["recorded_decision"] = state["recorded_decision"]
-            if state["recorded_decision"] is not None:
-                item["receiver_decision_recorded"] = True
-                item["next_action"] = "Review this request's recorded decision and linked work before returning its actual result. A recorded adoption is not a completed task; preserve the current owner and do not reprioritize unrelated work."
-            if state["warnings"]:
-                item["warnings"] = state["warnings"]
-                item["next_action"] = "Receiver receipt readback is unavailable or conflicting; recover the original receipt before continuing this request. Do not repeat the decision, result or execution. Other requests can continue."
-            elif state.get("request_control_view") is not None:
-                item["next_action"] = state["request_control_view"]["receiver_instruction"]
-            items.append(item)
+            items.append(_inbox_item(row, state))
             if len(items) == 21:
                 break
         batch.clear()

loopx/control_plane/collaboration/peers.py

@@ -607,6 +607,7 @@ def read_inbox(
     workspace=None,
     cursor=None,
     operation_cursor=None,
+    request_id=None,
     caller_goal_ref=None,
 ):
     from .inbox import pending
@@ -624,14 +625,18 @@ def read_inbox(
             agent_id,
             cursor=cursor,
             operation_cursor=operation_cursor,
+            request_id=request_id,
             scope=goal_scope,
         )
-        peer_returns, return_deliveries = _collect_returns(
-            root,
-            goal_id,
-            agent_id,
-            scope=goal_scope,
-        )
+        if request_id is None:
+            peer_returns, return_deliveries = _collect_returns(
+                root,
+                goal_id,
+                agent_id,
+                scope=goal_scope,
+            )
+        else:
+            peer_returns, return_deliveries = {"items": []}, []
         if peer_returns["items"]:
             result["peer_returns"] = peer_returns
         else:
@@ -673,8 +678,9 @@ def read_inbox(
             _record_return_reads(root, return_deliveries, scope=goal_scope)
     from .turn_context_observation import read_delegation_turn_context
 
-    result["unfinished_delegations"] = read_delegation_turn_context(
-        root, registry, goal_id, agent_id)["unfinished_delegations"]
+    if request_id is None:
+        result["unfinished_delegations"] = read_delegation_turn_context(
+            root, registry, goal_id, agent_id)["unfinished_delegations"]
     result["followthrough"] = (
         "Use each request's receiver_followthrough to reconcile it with actual Core work. "
         "Record an explicit assessment even when continuing other work; a read is not a decision. "
@@ -688,6 +694,12 @@ def read_inbox(
         "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."
     )
+    if request_id is not None:
+        result["followthrough"] = (
+            "Inspect this exact request's current source, revisions and receipts. "
+            "A historical read does not reopen concluded work or deliver other peer results. "
+            "For unfinished work, follow its existing receiver_followthrough and current authority."
+        )
     return result
 
 

tests/control_plane/test_inbox_exact_request.py

@@ -0,0 +1,148 @@
+"""Named source reads must not substitute or acknowledge another inbox page."""
+
+import json
+import os
+import subprocess
+import sys
+
+import pytest
+
+from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+from loopx.control_plane.collaboration.inbox import _root, acknowledge
+from loopx.control_plane.collaboration.peers import read_inbox, request, return_result
+
+
+@pytest.fixture(params=["file", "sqlite"])
+def inbox(tmp_path, monkeypatch, request):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, root, _ = promoted_create_fixture(tmp_path, provider=request.param)
+    data = json.loads(registry.read_text())
+    data["goals"][0]["coordination"]["registered_agents"] += ["reviewer", "third"]
+    registry.write_text(json.dumps(data))
+    return registry, root
+
+
+def brief(purpose):
+    return {
+        "schema_version": "collaboration_brief_v0",
+        "purpose": purpose,
+        "context": "\n".join(["完整原始来源：保留约束与结论。"] * 300),
+        "constraints": ["Read only; do not launch work or change its owner."],
+        "inputs": [],
+        "acceptance": ["Inspect the exact original request."],
+        "return_requirement": "Keep the original return route.",
+    }
+
+
+def create(inbox, operation, *, source="agent-a", target="reviewer"):
+    registry, root = inbox
+    content = brief(operation)
+    receipt = request(root, registry, "goal-a", source, target, operation, content)
+    return receipt["request_id"], content
+
+
+def conclude(inbox, rid, *, agent="reviewer"):
+    registry, root = inbox
+    read_inbox(root, registry, "goal-a", agent)
+    acknowledge(root, "goal-a", agent, rid, "adopt", "Inspected the original request", registry=registry)
+    return_result(root, "goal-a", agent, rid, "The original inspection is complete.", registry=registry)
+
+
+def cli(inbox, *extra, agent="reviewer"):
+    registry, root = inbox
+    process = subprocess.run(
+        [os.environ.get("LOOPX_TEST_CLI_PYTHON", sys.executable), "-I", "-m", "loopx.cli", "--registry", str(registry),
+         "--runtime-root", str(root), "--format", "json", "manager-inbox", "read",
+         "--goal-id", "goal-a", "--agent-id", agent, *extra],
+        cwd=registry.parent, text=True, capture_output=True, timeout=30,
+    )
+    return process.returncode, json.loads(process.stdout)
+
+
+def test_exact_cli_recovers_concluded_source_without_reopening_or_reading_pending(inbox):
+    _, root = inbox
+    original, content = create(inbox, "original-source")
+    conclude(inbox, original)
+    other, _ = create(inbox, "unrelated-pending")
+    owner = _root(root)
+    retained = {p: p.read_bytes() for p in owner.rglob("*.json")
+                if "entries" in p.relative_to(owner).parts
+                or p.parent.name == "decisions" or p.name == "conclusion.json"}
+
+    code, result = cli(inbox, "--request-id", original)
+
+    assert code == 0, result
+    assert [item["request_id"] for item in result["items"]] == [original]
+    item = result["items"][0]
+    assert item["brief"] == content and item["inbox_state"] == "settled"
+    assert "does not reopen" in item["next_action"]
+    assert result["has_more"] is False and result["next_cursor"] is None
+    assert not (owner / "reads" / f"{other}.json").exists()
+    assert {p: p.read_bytes() for p in retained} == retained
+    code, pending = cli(inbox)
+    assert code == 0 and [item["request_id"] for item in pending["items"]] == [other]
+
+
+@pytest.mark.parametrize("extra,agent", [
+    (["--request-id", "f" * 64], "reviewer"),
+    (["--request-id", "../invalid"], "reviewer"),
+    (["--request-id", "{original}"], "third"),
+    (["--request-id", "{original}", "--cursor", "invalid"], "reviewer"),
+    (["--request-id", "{original}", "--operation-cursor", "invalid"], "reviewer"),
+])
+def test_invalid_or_ambiguous_exact_read_never_falls_back_to_pending(inbox, extra, agent):
+    _, root = inbox
+    original, _ = create(inbox, "pending-source")
+    before = {p: p.read_bytes() for p in _root(root).rglob("*.json")}
+    code, result = cli(inbox, *(value.replace("{original}", original) for value in extra), agent=agent)
+    assert code == 1 and result["ok"] is False, result
+    assert "items" not in result
+    assert {p: p.read_bytes() for p in _root(root).rglob("*.json")} == before
+    assert not (_root(root) / "reads" / f"{original}.json").exists()
+
+
+def test_exact_read_does_not_deliver_unrelated_peer_results(inbox):
+    _, root = inbox
+    outgoing, _ = create(inbox, "awaiting-peer-result")
+    conclude(inbox, outgoing)
+    incoming, content = create(inbox, "named-incoming", source="third", target="agent-a")
+    before = {p: p.read_bytes() for p in _root(root).rglob("*.delivery.json")}
+
+    code, result = cli(inbox, "--request-id", incoming, agent="agent-a")
+
+    assert code == 0, result
+    assert [item["request_id"] for item in result["items"]] == [incoming]
+    assert result["items"][0]["brief"] == content
+    assert "peer_returns" not in result and "operation_handoffs" not in result
+    assert {p: p.read_bytes() for p in _root(root).rglob("*.delivery.json")} == before
+    code, pending = cli(inbox, agent="agent-a")
+    assert code == 0 and pending["peer_returns"]["items"][0]["request_id"] == outgoing
+
+
+def test_exact_read_keeps_source_goal_lifetime(tmp_path, monkeypatch):
+    from loopx.control_plane.projects.registry_codec import source_session_registry_transaction
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, root = tmp_path / "registry.json", tmp_path / "runtime"
+    with source_session_registry_transaction(registry, operation="create_fixture", create=lambda: {
+        "profile_id": "source_session_v1",
+        "goals": [{"id": "goal-a", "repo": str(tmp_path), "status": "active",
+                   "goal_instance_id": "ginst_" + "a" * 32,
+                   "coordination": {"registered_agents": ["agent-a", "reviewer"]}}],
+        "session_bindings": [], "session_receipts": [], "lifetime_receipts": [],
+    }) as transaction:
+        transaction.commit(transaction.payload_copy())
+    original, content = create((registry, root), "source-lifetime")
+    code, observed = cli((registry, root), "--request-id", original)
+    assert code == 0 and observed["items"][0]["brief"] == content
+    assert observed["items"][0]["goal_ref"] == {
+        "goal_id": "goal-a", "goal_instance_id": "ginst_" + "a" * 32,
+    }
+    with source_session_registry_transaction(registry, operation="replace_fixture") as transaction:
+        payload = transaction.payload_copy()
+        payload["goals"][0]["goal_instance_id"] = "ginst_" + "b" * 32
+        transaction.commit(payload)
+    before = {p: p.read_bytes() for p in _root(root).rglob("*.json")}
+    code, rejected = cli((registry, root), "--request-id", original)
+    assert code == 1 and rejected["ok"] is False and "items" not in rejected
+    assert {p: p.read_bytes() for p in _root(root).rglob("*.json")} == before

tests/test_collaboration_goal_instance.py

@@ -115,12 +115,18 @@ def _create_source_registry(root: Path) -> Path:
 
 def _http_snapshot(root, registry, session_id):
     import http.client
+    from types import SimpleNamespace
     from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
 
     server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
     server.verbose = False
     server.registry_path, server.runtime_root = registry, root
     server.chat_store = ChatSessionStore(root)
+    # These fixtures script Host liveness; the real HTTP/store projection below
+    # tests collaboration lifetime and receipts, not exited-process recovery.
+    server.runtime_controller = SimpleNamespace(
+        recover_exited_turn=lambda _session_id: None, close=lambda: None,
+    )
     worker = threading.Thread(target=server.serve_forever, daemon=True)
     worker.start()
     connection = http.client.HTTPConnection(*server.server_address, timeout=10)