diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md
index 8dbb6c0f50..77b2a4cb8d 100644
--- a/loopx/capabilities/manager_context/README.md
+++ b/loopx/capabilities/manager_context/README.md
@@ -284,6 +284,17 @@ The local Inbox read transaction now lives in
 Goal-instance, request-revision and agenda owners inside one runtime request.
 The Turn hook remains read-only. CLI/MCP reads record the supplied requests and
 peer-return read receipts; they do not adopt requests or consume peer results.
+If the runtime response is lost, a read may already have published some or all
+of those receipts. The CLI returns JSON with `ok=false`, `read_complete=false`
+and exit status 1; MCP returns a tool error containing the same recovery object.
+It retains the original Goal/Agent (and exact Goal instance when bound), read
+arguments, known request revisions and offered peer-result keys. Unknown source
+lists are `null`, distinct from a verified empty list. It never retries, advances
+a cursor or reports an empty Inbox to hide that ambiguity. Inspect those exact
+durable read/delivery receipts using the existing Inbox/status readers, recheck
+the original source and Goal lifetime, then recover the same page with its
+original arguments. An absent receipt does not prove the timed-out writer has
+stopped. This recovery grants no work replay, result consumption or quota spend.
 Python releases the initial Goal lifetime guard for provider enrichment.
 Core Todo and workspace inputs retain their actual IO adapters and provider
 fences. The Host reacquires the Goal guard before native finalization.
diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py
index 2652620293..e2497389ca 100644
--- a/loopx/cli_commands/manager_inbox.py
+++ b/loopx/cli_commands/manager_inbox.py
@@ -10,6 +10,7 @@
     configure_evidence_scope,
 )
 from ..control_plane.projects.registry_codec import load_project_registry
+from ..control_plane.collaboration.peers import InboxReadResponseAmbiguous
 from ..presentation.answer_instruction import collaboration_answer_instruction
 
 
@@ -338,6 +339,8 @@ def handle_manager_inbox(args, registry_path, runtime_root):
                 request_revision=getattr(args, "request_revision", None),
                 request_revision_digest=getattr(args, "request_digest", None),
             )
+    except InboxReadResponseAmbiguous as exc:
+        result = exc.result
     except (OSError, ValueError) as exc:
         result = {"ok": False, "error": str(exc)}
     print(json.dumps(result, ensure_ascii=False, indent=2))
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 7005b93eba..7387cdf169 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -68,6 +68,7 @@
 )
 from .control_plane.collaboration.source_policy_fence import source_policy_fence
 from .control_plane.collaboration.peers import (
+    InboxReadResponseAmbiguous,
     _goal,
     consume_return,
     read_inbox,
@@ -237,15 +238,20 @@ def read_context(cursor: str | None = None) -> dict:
         Follow next_cursor for later requests. Omit cursor to start a fresh scan.
         Pages are live; reading all pages does not complete outstanding work.
         """
-        result = read_inbox(
-            root,
-            registry,
-            goal_id,
-            agent_id,
-            workspace=workspace,
-            cursor=cursor,
-            caller_goal_ref=caller_goal_ref,
-        )
+        try:
+            result = read_inbox(
+                root,
+                registry,
+                goal_id,
+                agent_id,
+                workspace=workspace,
+                cursor=cursor,
+                caller_goal_ref=caller_goal_ref,
+            )
+        except InboxReadResponseAmbiguous as exc:
+            from mcp.server.fastmcp.exceptions import ToolError
+
+            raise ToolError(json.dumps(exc.result, ensure_ascii=False)) from exc
         result["instruction"] += " " + collaboration_answer_instruction()
         return result
 
diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py
index 7d31fdca5e..188f069230 100644
--- a/loopx/control_plane/collaboration/peers.py
+++ b/loopx/control_plane/collaboration/peers.py
@@ -12,6 +12,10 @@
 import stat
 from contextlib import nullcontext
 from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from ..effect_runtime import EffectRuntimeResponseAmbiguous
 
 from .inbox import (
     ENTRY_SCHEMA,
@@ -50,6 +54,35 @@
 )
 
 
+class InboxReadResponseAmbiguous(RuntimeError):
+    """Preserve the read's original receipt addresses without replaying it."""
+
+    def __init__(self, error: EffectRuntimeResponseAmbiguous, *, stage: str,
+                 scope: dict[str, Any], read_arguments: dict[str, Any],
+                 items: list[dict[str, Any]] | None = None,
+                 deliveries: list[dict[str, Any]] | None = None) -> None:
+        super().__init__(str(error))
+        self.result: dict[str, Any] = {
+            "ok": False, "error": str(error), "error_code": error.diagnostic_code,
+            "read_complete": False,
+            "recovery": {
+                "stage": stage, "scope": scope, "read_arguments": read_arguments,
+                "requests": None if items is None else [{key: item[key] for key in (
+                    "goal_id", "agent_id", "goal_ref", "request_id", "request_revision", "request_revision_digest",
+                ) if key in item} for item in items],
+                "peer_returns": None if deliveries is None else [
+                    {"request": item["request"], "result_key": item["key"]} for item in deliveries],
+                "automatic_retry_performed": False,
+                "instruction": (
+                    "Inspect the exact request read receipts and peer result delivery receipts through the existing "
+                    "Inbox/status readers before retrying. A missing receipt does not prove the timed-out writer stopped. "
+                    "Revalidate the same Goal lifetime, source and original read arguments; never advance a cursor to skip recovery. "
+                    "Unknown request/result lists are null, not an empty Inbox. No work, result consumption or quota retry is authorized."
+                ),
+            },
+        }
+
+
 def _goal(registry, goal_id, *agents, require_active=False):
     goal = next(
         (
@@ -576,17 +609,24 @@ def read_inbox(
     from .inbox import read_inbox_transaction
     from .turn_context_observation import delegation_read_command
     from .links import receiver_work_observations
+    from ..effect_runtime import EffectRuntimeResponseAmbiguous
 
     read_command = delegation_read_command(root, registry, goal_id, agent_id)
+    read_arguments = {"cursor": cursor, "operation_cursor": operation_cursor, "request_id": request_id}
     with collaboration_goal_scope(
         registry, goal_id=goal_id, agents=(agent_id,), caller_goal_ref=caller_goal_ref,
     ) as goal_scope:
-        snapshot = read_inbox_transaction(root, {
-            "operation": "receiver_read", "goal_id": goal_id, "agent_id": agent_id,
-            "cursor": cursor, "operation_cursor": operation_cursor, "request_id": request_id,
-            "read_command": read_command,
-            "defer_recording": operation_cursor is not None or (root / "chat" / "actions" / "actions.json").is_file(),
-        }, scope=goal_scope)
+        read_scope = {"goal_id": goal_id, "agent_id": agent_id,
+            "goal_ref": dict(goal_scope.caller_goal_ref or {}) if goal_scope.exact else None}
+        try:
+            snapshot = read_inbox_transaction(root, {
+                "operation": "receiver_read", "goal_id": goal_id, "agent_id": agent_id,
+                **read_arguments, "read_command": read_command,
+                "defer_recording": operation_cursor is not None or (root / "chat" / "actions" / "actions.json").is_file(),
+            }, scope=goal_scope)
+        except EffectRuntimeResponseAmbiguous as exc:
+            raise InboxReadResponseAmbiguous(exc, stage="receiver_read", scope=read_scope,
+                read_arguments=read_arguments) from exc
         snapshot.update(goal_id=goal_id, agent_id=agent_id)
         result = _pending_result(root, snapshot, goal_scope, operation_cursor)
         observed_goal = dict(goal_scope.goal)
@@ -603,12 +643,16 @@ def read_inbox(
         with collaboration_goal_scope(
             registry, goal_id=goal_id, agents=(agent_id,), caller_goal_ref=observed_goal_ref,
         ) as goal_scope:
-            final = read_inbox_transaction(root, {
-                "operation": "record_read", "goal_id": goal_id, "agent_id": agent_id,
-                "items": result["items"],
-                "deliveries": snapshot["deliveries"], "work_observations": work,
-                "include_context": request_id is None, "read_command": read_command,
-            }, scope=goal_scope)
+            try:
+                final = read_inbox_transaction(root, {
+                    "operation": "record_read", "goal_id": goal_id, "agent_id": agent_id,
+                    "items": result["items"],
+                    "deliveries": snapshot["deliveries"], "work_observations": work,
+                    "include_context": request_id is None, "read_command": read_command,
+                }, scope=goal_scope)
+            except EffectRuntimeResponseAmbiguous as exc:
+                raise InboxReadResponseAmbiguous(exc, stage="record_read", scope=read_scope,
+                    read_arguments=read_arguments, items=result["items"], deliveries=snapshot["deliveries"]) from exc
     else:
         final = snapshot
         for item in result["items"]:
diff --git a/tests/control_plane/test_inbox_read_recovery.py b/tests/control_plane/test_inbox_read_recovery.py
index c94ab44b7d..26150b7c38 100644
--- a/tests/control_plane/test_inbox_read_recovery.py
+++ b/tests/control_plane/test_inbox_read_recovery.py
@@ -4,6 +4,8 @@
 import time
 from pathlib import Path
 
+import pytest
+
 from loopx.control_plane.collaboration.goal_instance_scope import collaboration_goal_scope
 from loopx.control_plane.collaboration.inbox import _root, pending
 from loopx.control_plane.collaboration.peers import request
@@ -12,6 +14,71 @@
 from loopx.file_lock import exclusive_cross_runtime_file_lock
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("loss", ["source", "partial", "committed"])
+def test_native_cli_keeps_exact_receipt_recovery_after_lost_read_response(tmp_path, monkeypatch, capsys, provider, loss):
+    from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+    from loopx.cli import main
+    from loopx.control_plane.collaboration import inbox
+    from loopx.control_plane.effect_runtime import EffectRuntimeResponseAmbiguous
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, root, _ = promoted_create_fixture(tmp_path, provider=provider, registered_agents=["worker", "source"])
+    brief = {"schema_version": "collaboration_brief_v0", "purpose": "Inspect the artifact",
+        "context": "Private source body must not enter error metadata", "constraints": [], "inputs": [],
+        "acceptance": ["Return checked findings"], "return_requirement": "Reply to the source"}
+    ids = {request(root, registry, "goal-a", "source", "worker", operation, brief)["request_id"]
+        for operation in ("one", "two")}
+    transport = inbox.read_inbox_transaction
+    attempts = []
+
+    def lost_reply(runtime, query, **kwargs):
+        stage = "receiver_read" if loss == "source" else "record_read"
+        if query["operation"] == stage and not attempts:
+            attempts.append(query)
+            if loss != "source":
+                commit = query if loss == "committed" else {**query, "items": query["items"][:1],
+                    "deliveries": [], "work_observations": query["work_observations"][:1], "include_context": False}
+                transport(runtime, commit, **kwargs)
+            raise EffectRuntimeResponseAmbiguous("collaboration.inbox.read", timeout=10)
+        return transport(runtime, query, **kwargs)
+
+    monkeypatch.setattr(inbox, "read_inbox_transaction", lost_reply)
+    route = ["--registry", str(registry), "--runtime-root", str(root), "--format", "json", "manager-inbox"]
+    assert main([*route, "read", "--goal-id", "goal-a", "--agent-id", "worker"]) == 1
+    error = json.loads(capsys.readouterr().out)
+    assert error["ok"] is False and error["error_code"] == "runtime_response_ambiguous"
+    assert error["read_complete"] is False
+    recovery = error["recovery"]
+    assert recovery["scope"] == {"goal_id": "goal-a", "agent_id": "worker", "goal_ref": None}
+    assert recovery["read_arguments"] == {"cursor": None, "operation_cursor": None, "request_id": None}
+    assert recovery["automatic_retry_performed"] is False and len(attempts) == 1
+    if loss == "source":
+        assert recovery["stage"] == "receiver_read"
+        assert recovery["requests"] is None and recovery["peer_returns"] is None
+    else:
+        assert recovery["stage"] == "record_read"
+        assert {row["request_id"] for row in recovery["requests"]} == ids
+        assert recovery["peer_returns"] == []
+    assert brief["context"] not in json.dumps(error)
+    assert str(tmp_path) not in json.dumps(error)
+    before = {p: p.read_bytes() for p in _root(root).rglob("*.json")}
+    recorded = []
+    for rid in ids:
+        assert main([*route, "status", "--goal-id", "goal-a", "--agent-id", "worker", "--request-id", rid]) == 0
+        status = json.loads(capsys.readouterr().out)["rows"][0]
+        recorded.append(status["read"]["status"] == "supplied_to_receiver")
+    assert sum(recorded) == {"source": 0, "partial": 1, "committed": 2}[loss]
+    assert {p: p.read_bytes() for p in _root(root).rglob("*.json")} == before
+    # Recover only after reading the exact durable statuses. The same request
+    # page is supplied without replacing earlier receipts or consuming results.
+    assert main([*route, "read", "--goal-id", "goal-a", "--agent-id", "worker"]) == 0
+    assert {row["request_id"] for row in json.loads(capsys.readouterr().out)["items"]} == ids
+    assert all(path.read_bytes() == value for path, value in before.items())
+    assert len(list((_root(root) / "reads").glob("*.json"))) == 2
+    assert not list((_root(root) / "replies").rglob("*.consumed.json"))
+
+
 def test_interrupted_page_preserves_first_receipt_and_finishes_remaining_reads(tmp_path):
     registry = tmp_path / "registry.json"
     with source_session_registry_transaction(registry, operation="create_inbox_fixture", create=lambda: {
diff --git a/tests/test_collaboration_mcp.py b/tests/test_collaboration_mcp.py
index 724af737bb..af722da30a 100644
--- a/tests/test_collaboration_mcp.py
+++ b/tests/test_collaboration_mcp.py
@@ -21,6 +21,103 @@ def test_serving_mcp_still_requires_an_explicit_workspace(tmp_path):
     assert "--workspace is required when serving MCP" in result.stderr
 
 
+def test_stdio_lost_read_response_preserves_original_page_and_result_receipts(tmp_path, monkeypatch):
+    from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+    from loopx.control_plane.collaboration.inbox import _receipt, _root, acknowledge, pending
+    from loopx.control_plane.collaboration.goal_instance_scope import collaboration_goal_scope
+    from loopx.control_plane.collaboration.peers import request, return_result
+    from loopx.control_plane.projects.registry_codec import source_session_registry_transaction
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry = tmp_path / "registry.json"
+    ref = {"goal_id": "delivery", "goal_instance_id": "ginst_" + "a" * 32}
+    with source_session_registry_transaction(registry, operation="create_inbox_fixture", create=lambda: {
+        "profile_id": "source_session_v1", "goals": [{"id": "delivery", "repo": str(tmp_path),
+        "status": "active", "goal_instance_id": ref["goal_instance_id"],
+        "coordination": {"registered_agents": ["worker", "source"]}}],
+        "session_bindings": [], "session_receipts": [], "lifetime_receipts": [],
+    }) as transaction:
+        transaction.commit(transaction.payload_copy())
+    brief = {"schema_version": "collaboration_brief_v0", "purpose": "Inspect the artifact",
+        "context": "Private source body", "constraints": [], "inputs": [],
+        "acceptance": ["Return checked findings"], "return_requirement": "Reply to the source"}
+    for number in range(22):
+        request(tmp_path, registry, "delivery", "source", "worker", f"check-{number}", brief)
+    # Only the transport reply is faulted: this launches the real stdio server,
+    # with its native request/receipt owner and existing tool registration.
+    launcher = tmp_path / "lost_response.py"
+    launcher.write_text('''from loopx.control_plane.collaboration import inbox
+from loopx.control_plane.effect_runtime import EffectRuntimeResponseAmbiguous
+from loopx.collaboration_mcp import main
+transport = inbox.read_inbox_transaction
+records = 0
+def interrupted(root, query, **kwargs):
+    global records
+    if query["operation"] == "record_read":
+        records += 1
+        if records == 2:
+            transport(root, {**query, "items": query["items"][:1], "deliveries": [],
+                "work_observations": query["work_observations"][:1], "include_context": False}, **kwargs)
+            raise EffectRuntimeResponseAmbiguous("collaboration.inbox.read", timeout=10)
+    return transport(root, query, **kwargs)
+inbox.read_inbox_transaction = interrupted
+main()
+''')
+
+    async def exercise():
+        params = StdioServerParameters(command=sys.executable, args=[str(launcher),
+            "--runtime-root", str(tmp_path), "--registry", str(registry), "--goal-id", "delivery",
+            "--agent-id", "worker", "--workspace", str(tmp_path)])
+        async with stdio_client(params) as (read, write), ClientSession(read, write) as session:
+            await session.initialize()
+            first = await session.call_tool("read_context", {})
+            assert not first.isError
+            page = json.loads(first.content[0].text)
+            assert len(page["items"]) == 20 and page["next_cursor"]
+            cursor = page["next_cursor"]
+            rid = request(tmp_path, registry, "delivery", "worker", "source", "peer-check", brief)["request_id"]
+            acknowledge(tmp_path, "delivery", "source", rid, "adopt", "Check the artifact", registry=registry)
+            return_result(tmp_path, "delivery", "source", rid, "Private peer conclusion", registry=registry)
+            with collaboration_goal_scope(registry, goal_id="delivery", agents=("worker",)) as scope:
+                tail = pending(tmp_path, "delivery", "worker", cursor=cursor, scope=scope)["items"]
+            assert len(tail) == 2
+            failed = await session.call_tool("read_context", {"cursor": cursor})
+            assert failed.isError
+            # FastMCP retains its standard error prefix; the bounded recovery
+            # object is JSON rather than a traceback or successful empty page.
+            error = json.loads(failed.content[0].text.split(": ", 1)[1])
+            assert error["ok"] is False and error["read_complete"] is False
+            recovery = error["recovery"]
+            assert recovery["scope"] == {"goal_id": "delivery", "agent_id": "worker", "goal_ref": ref}
+            assert recovery["read_arguments"] == {"cursor": cursor, "operation_cursor": None, "request_id": None}
+            assert {row["request_id"] for row in recovery["requests"]} == {row["request_id"] for row in tail}
+            assert recovery["peer_returns"] == [{"request": {"request_id": rid, "goal_id": "delivery",
+                "agent_id": "source", "goal_ref": ref}, "result_key": "conclusion"}]
+            assert recovery["automatic_retry_performed"] is False
+            assert "Private source body" not in str(error) and "Private peer conclusion" not in str(error)
+            assert "instruction" not in error and str(tmp_path) not in str(error)
+            before = {path: path.read_bytes() for path in _root(tmp_path).rglob("*.json")}
+            receipts = [_receipt(tmp_path, "reads", row)[0] for row in tail]
+            assert sum(bool(row) for row in receipts) == 1
+            assert {path: path.read_bytes() for path in before} == before
+            assert not (_root(tmp_path) / "replies" / rid / "conclusion.delivery.json").exists()
+            recovered = await session.call_tool("read_context", {"cursor": cursor})
+            assert not recovered.isError
+            assert {row["request_id"] for row in json.loads(recovered.content[0].text)["items"]} == {row["request_id"] for row in tail}
+            assert all(path.read_bytes() == data for path, data in before.items())
+            assert (_root(tmp_path) / "replies" / rid / "conclusion.delivery.json").is_file()
+            assert not list((_root(tmp_path) / "replies").rglob("*.consumed.json"))
+            with source_session_registry_transaction(registry, operation="revoke_fixture_agent") as transaction:
+                changed = transaction.payload_copy()
+                changed["goals"][0]["coordination"]["registered_agents"] = ["source"]
+                transaction.commit(changed)
+            revoked = await session.call_tool("read_context", {"cursor": cursor})
+            assert revoked.isError and "requires registered Agents" in revoked.content[0].text
+            assert '"recovery"' not in revoked.content[0].text
+
+    asyncio.run(exercise())
+
+
 def test_scoped_stdio_tools_do_not_offer_shell_or_sender_override(tmp_path):
     registry = tmp_path / "registry.json"
     config = {