fix(inbox): preserve receipt recovery after lost read responses

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.

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))

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
 

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"]:

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: {