diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index e3b01a16a5..333b977a27 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -4,7 +4,7 @@
 - Supersedes / closes: none
 - Proposed by: LoopX maintainers
 - Date: 2026-08-15
-- Last revised: 2026-10-16
+- Last revised: 2026-10-19
 - Scope: an incremental, replacement-first migration of the LoopX control-plane
   core from Python to TypeScript without maintaining two semantic
   implementations
@@ -17,6 +17,25 @@
 
 ### Product conversations and asynchronous inbox
 
+**Receiver Inbox read transaction (2026-10-19).**
+`loopx/control_plane/collaboration/inbox_read.ts` now composes local source scans, pagination,
+Goal-instance admission, receiver and peer-return read receipts, and unfinished
+delegation context through the existing typed owners. Turn-start observation
+uses a read-only request. Receiver reads without a pending Host provider commit
+their local read receipts in that request. Actual Core Todo, workspace-input and
+operation-ticket providers remain explicit Python IO; finalization reacquires
+the Goal lifetime guard and rereads entries under the shared request fence.
+Reading still grants neither adoption, consumption nor execution authority.
+The old Python scan/composition loops and five fine-grained RPC registrations
+are removed. The legacy Inbox read imports and tracking/presentation receipt
+views retain thin native facades; their exit is migration of those remaining
+Python consumers. Existing return-display file IO retains its Python adapter
+and the shared persisted delivery vocabulary. Large pages reuse the shared
+private snapshot transport. Normal runtime IPC, installed
+HTTP caller parity and complete default-SQLite CLI performance remain separate
+qualification gates. This source cutover does not certify R1–R3 or the complete
+asynchronous collaboration lifecycle.
+
 R1–R3 also consume the typed kernel through the App, not only CLI settlement.
 The [App/inbox integration plan](app-conversation-and-async-inbox-v0.md#ts-and-generic-async-inbox-migrate-with-the-user-path)
 orders caller fixes, characterization, one complete async lifecycle cutover,
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index eb238403b9..27abbd7992 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -4,7 +4,7 @@
 - 替代 / 关闭：无
 - Proposed by：LoopX maintainers
 - Date：2026-08-15
-- Last revised：2026-10-16
+- Last revised：2026-10-19
 - Scope：LoopX 控制面核心从 Python 到 TypeScript 的增量、replacement-first
   迁移；不长期维护两份语义实现
 - Tracking issue：[#3225](https://github.com/loopx-project/loopx/issues/3225)
@@ -17,6 +17,21 @@
 
 ### 产品对话与通用异步 inbox
 
+**接收方 Inbox 读取事务（2026-10-19）。**
+`loopx/control_plane/collaboration/inbox_read.ts` 现在通过既有 typed owner 组合本地源扫描、分页、
+Goal-instance 准入、接收方及 peer-return 读取回执和未完成委派上下文。
+Turn-start 观察使用只读请求。没有待处理 Host provider 的接收读取，在同一个请求中
+提交本地读取回执。实际 Core Todo、工作区输入和操作票据 provider 仍是显式 Python IO；
+finalize 重新取得 Goal lifetime guard，并在共享 request fence 内重读 entry。
+读取仍不授予 adoption、consumption 或执行权限。
+旧 Python 扫描与组合循环及五个细粒度 RPC 注册已删除。
+旧 Inbox 读取 import 和 tracking/presentation 回执视图保留薄 native facade；
+这些层在剩余 Python 消费者迁移后退出。既有 return 展示文件 IO 保留 Python adapter，
+复用同一份持久化 delivery 词汇。大页面复用共享私有 snapshot 传输。
+正常 runtime IPC、已安装 HTTP 调用者一致性、
+完整默认 SQLite CLI 性能仍是独立资格门。本次源码切换不证明 R1–R3 或完整异步协作
+生命周期已通过验收。
+
 R1–R3 的 TS 消费者包括 App 产品路径，不只 CLI 结算。
 [App/inbox 融合设计](app-conversation-and-async-inbox-v0.md#ts-and-generic-async-inbox-migrate-with-the-user-path)
 按调用者修复、行为刻画、完整异步生命周期切换、真实 adapter 恢复与旧代码删除推进。
diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md
index b621a77a66..dd5a359fd6 100644
--- a/loopx/capabilities/manager_context/README.md
+++ b/loopx/capabilities/manager_context/README.md
@@ -164,6 +164,20 @@ LoopX quota/heartbeat Turn-start path; arbitrary host Turns that do not invoke
 that path are outside this integration. Frontend/Lark conversations retain their
 existing request and return views; no new screen or setting is introduced.
 
+The local Inbox read transaction now lives in
+`loopx/control_plane/collaboration/inbox_read.ts`. It reuses the typed receipt,
+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.
+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.
+Operation-ticket observations retain their existing provider. Complete large
+pages use the shared private snapshot transport. Existing return-display file
+IO remains in its Python adapter. Existing storage addresses, source records and public
+response shapes remain compatible. Normal runtime IPC, installed HTTP journeys
+and complete CLI performance require separate qualification.
+
 Read and record a decision through the installed CLI:
 
 ```sh
diff --git a/loopx/capabilities/manager_context/__init__.py b/loopx/capabilities/manager_context/__init__.py
index 04bd9ec2c7..218d437dd3 100644
--- a/loopx/capabilities/manager_context/__init__.py
+++ b/loopx/capabilities/manager_context/__init__.py
@@ -238,22 +238,20 @@ def turn_start_hook(
     def produce():
         nonlocal snapshot
         try:
-            from ...control_plane.collaboration.turn_context_observation import read_delegation_turn_context
+            from ...control_plane.collaboration.inbox_read import read_inbox_transaction
+            from ...control_plane.collaboration.inbox import _pending_result
 
-            snapshot = read_delegation_turn_context(runtime_root, registry_path, goal_id, agent_id,
-                read_command=command)
-            agenda = snapshot["unfinished_delegations"]
             with collaboration_goal_scope(
-                registry_path,
-                goal_id=goal_id,
-                agents=(agent_id,),
+                registry_path, goal_id=goal_id, agents=(agent_id,),
             ) as goal_scope:
-                inbox = pending(
-                    runtime_root,
-                    goal_id,
-                    agent_id,
-                    scope=goal_scope,
-                )
+                observed = read_inbox_transaction(runtime_root, {
+                    "operation": "turn_start", "goal_id": goal_id, "agent_id": agent_id,
+                    "read_command": command,
+                }, scope=goal_scope)
+                observed.update(goal_id=goal_id, agent_id=agent_id)
+                inbox = _pending_result(runtime_root, observed, goal_scope, None)
+            snapshot = observed["context"]
+            agenda = snapshot["unfinished_delegations"]
             count = (
                 max(len(inbox["items"]), agenda["incoming_count"])
                 + max(len(inbox.get("peer_returns", {}).get("items", [])), agenda["outgoing_count"])
diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py
index 47485b269e..3aa08c7368 100644
--- a/loopx/control_plane/collaboration/inbox.py
+++ b/loopx/control_plane/collaboration/inbox.py
@@ -15,7 +15,7 @@
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
-from ...file_lock import exclusive_file_lock
+from ...file_lock import exclusive_cross_runtime_file_lock
 from ..runtime.file_paths import windows_extended_path
 from ..content_digest import BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN
 from ..todos.contract import TODO_ID_PATTERN
@@ -104,7 +104,7 @@ def _request_lock(
         if scope is not None and scope.exact
         else legacy_path
     )
-    with exclusive_file_lock(lock):
+    with exclusive_cross_runtime_file_lock(lock):
         yield
 
 
@@ -145,234 +145,50 @@ 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,
+    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(
-        (
-            [
-                "pending_requests_v1",
-                str(runtime_root.resolve()),
-                _target(goal_id, agent_id, scope),
-            ]
-            if scope is not None and scope.exact
-            else [
-                "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")
-        _, supplied_scope, after = cursor.split(":")
-        if supplied_scope != cursor_scope:
-            raise ValueError("pending request cursor scope mismatch")
-    folder = (
-        _root(runtime_root)
-        / "entries"
-        / _hash(_target(goal_id, agent_id, scope))
-    )
-    try:
-        paths = sorted(folder.iterdir())
-    except FileNotFoundError:
-        paths = []
-    items, batch = [], []
-
-    def append_batch():
-        classified = iter(_pending_receipt_states(runtime_root, batch))
-        for row in batch:
-            if isinstance(row, Exception):
-                raise row
-            state = next(classified)
-            if state["kind"] == "settled":
-                continue
-            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":
-                    continue
-            items.append(_inbox_item(row, state))
-            if len(items) == 21:
-                break
-        batch.clear()
-
-    for path in paths:
-        if path.suffix != ".json" or not BARE_SHA256_PATTERN.fullmatch(path.stem):
-            continue  # Ignore lock holder sidecars and other non-entry files.
-        if path.stem <= after:
-            continue
-        try:
-            batch.append(_pending_entry(path, goal_id, agent_id, scope))
-        except (OSError, ValueError) as exc:
-            # Evaluate errors in scan order; batch lookahead cannot let a later
-            # page's damaged entry block an already complete current page.
-            batch.append(exc)
-        if len(batch) == 128:
-            append_batch()
-            if len(items) == 21:
-                break
-    if batch:
-        append_batch()
-    from .peers import returns
-
-    peer_returns = returns(runtime_root, goal_id, agent_id, scope=scope)
+    from .inbox_read import read_inbox_transaction
+
+    snapshot = read_inbox_transaction(runtime_root, {
+        "operation": "pending", "goal_id": goal_id, "agent_id": agent_id,
+        "cursor": cursor, "operation_cursor": operation_cursor, "request_id": request_id,
+    }, scope=scope)
+    snapshot.update(goal_id=goal_id, agent_id=agent_id)
+    return _pending_result(runtime_root, snapshot, scope, operation_cursor)
+
+
+def _pending_result(root, snapshot, scope, operation_cursor):
+    result = snapshot["result"]
+    if "request_id" in result or snapshot.get("completed"):
+        return result
     from .operation_handoff import pending_operation_handoffs
 
-    operation_handoffs = pending_operation_handoffs(
-        runtime_root,
-        goal_id,
-        agent_id,
-        registry_path=scope.registry_path if scope is not None else None,
-        scope=scope,
-        cursor=operation_cursor,
-        cursor_scope=cursor_scope,
+    handoffs = pending_operation_handoffs(
+        root, snapshot["goal_id"],
+        snapshot["agent_id"], registry_path=scope.registry_path if scope else None,
+        scope=scope, cursor=operation_cursor, cursor_scope=snapshot["cursor_scope"],
     )
-    return {
-        "ok": True,
-        **(
-            {
-                "operation_handoffs": operation_handoffs["items"],
-                "operation_handoff_pending_count": operation_handoffs["pending_count"],
-                "operation_handoff_overflow": operation_handoffs["overflow"],
-                "operation_handoff_next_cursor": operation_handoffs["next_cursor"],
-            }
-            if operation_handoffs["items"] or operation_cursor is not None
-            else {}
-        ),
-        **({"peer_returns": peer_returns} if peer_returns["items"] else {}),
-        "items": items[:20],
-        "has_more": len(items) > 20,
-        "next_cursor": (
-            f"1:{cursor_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."
-        ),
-    }
+    if handoffs["items"] or operation_cursor is not None:
+        result.update(operation_handoffs=handoffs["items"],
+            operation_handoff_pending_count=handoffs["pending_count"],
+            operation_handoff_overflow=handoffs["overflow"],
+            operation_handoff_next_cursor=handoffs["next_cursor"])
+    return result
 
 
-def _pending_entry(path, goal_id, agent_id, scope):
-    if not BARE_SHA256_PATTERN.fullmatch(path.stem):
-        raise ValueError("invalid context request filename")
-    item = _read(path)
-    if (
-        item.get("schema_version")
-        != (EXACT_ENTRY_SCHEMA if scope is not None and scope.exact else 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")
-    return item
 
 
 def _pending_receipt_states(runtime_root, rows):
-    """Observe private files, then batch the typed read model by wire bytes."""
-    from ..effect_runtime import effect_runtime_result
+    from .inbox_read import read_inbox_transaction
 
-    root = _root(runtime_root)
-    observations, states, encoded_bytes = [], [], 0
-    for row in rows:
-        if isinstance(row, Exception):
-            continue
-        request_id = row["request_id"]
-        observation = {
-            "request": {key: row[key] for key in (
-                "request_id", "goal_id", "agent_id", "source_id", "goal_ref"
-            ) if key in row},
-            "route_present": (root / "roundtrips" / (request_id + ".json")).exists(),
-            "decision": _observe_receipt(root / "decisions" / (request_id + ".json")),
-            "conclusion": _observe_receipt(root / "replies" / request_id / "conclusion.json"),
-        }
-        if "request_control" in row:
-            observation["request"] = row
-            observation["read"] = _observe_receipt(root / "reads" / (request_id + ".json"))
-            results = []
-            for path in result_paths(root / "replies" / request_id):
-                try:
-                    results.append(_result_observation(path.stem, _read(path)))
-                except (OSError, ValueError, TypeError):
-                    results.append({"unavailable": True})
-            observation["results"] = results
-        size = len(json.dumps(observation, separators=(",", ":")).encode())
-        # The existing bridge serializes ASCII-escaped JSON and admits 2 MiB.
-        # Leave room for its envelope; legal Unicode replies must not turn
-        # a bounded page into another oversized history request.
-        if observations and encoded_bytes + size > 1_000_000:
-            states.extend(effect_runtime_result(
-                "collaboration.inbox.inspect_receipts", {"observations": observations}
-            )["items"])
-            observations, encoded_bytes = [], 0
-        observations.append(observation)
-        encoded_bytes += size
-    if observations:
-        states.extend(effect_runtime_result(
-            "collaboration.inbox.inspect_receipts", {"observations": observations}
-        )["items"])
-    return states
+    return read_inbox_transaction(runtime_root, {
+        "operation": "receipt_states", "rows": rows,
+    })["items"]
 
 
 def controlled_request_readback(root: Path, row: dict[str, Any]) -> dict[str, Any] | None:
@@ -493,63 +309,19 @@ def _entry(root: Path, goal_id: str, agent_id: str, request_id: str, *,
     return row
 
 
-def record_read(
-    root: Path,
-    items: list[dict],
-    *,
-    scope: CollaborationGoalScope | None = None,
-) -> None:
-    """CLI supplied these messages to the receiver; not proof of comprehension."""
-    for row in items:
-        _entry(
-            root,
-            row["goal_id"],
-            row["agent_id"],
-            row["request_id"],
-            scope=scope,
-        )
-        if scope is not None:
-            from .goal_instance_scope import decide_collaboration_lifecycle
+def record_read(root: Path, items: list[dict], *, scope: CollaborationGoalScope | None = None) -> None:
+    """Record supplied source without certifying comprehension or adoption."""
+    if not items:
+        return
+    from .inbox_read import read_inbox_transaction
 
-            decide_collaboration_lifecycle(
-                scope,
-                operation="read_record",
-                record=row,
-            )
-        path = _root(root) / "reads" / (row["request_id"] + ".json")
-        with _request_lock(
-            root,
-            row["request_id"],
-            scope,
-            path.with_suffix(".lock"),
-        ):
-            stored = _entry(root, row["goal_id"], row["agent_id"], row["request_id"], scope=scope)
-            if "request_control" in stored:
-                from ..effect_runtime import effect_runtime_result
-
-                decision, error = _receipt(root, "decisions", stored)
-                plan = effect_runtime_result("collaboration.request.record_read", {
-                    "entry": stored, "actor_agent_id": row["agent_id"], "observed_at": _now(),
-                    "revision": row.get("request_revision"), "revision_digest": row.get("request_revision_digest"),
-                    "legacy_assessment": decision or None, "legacy_assessment_unavailable": bool(error)})
-                if plan["changed"]:
-                    entry_path = _root(root) / "entries" / _hash(_target(row["goal_id"], row["agent_id"], scope)) / (row["request_id"] + ".json")
-                    _write(entry_path, plan["entry"])
-                row["request_read_receipt"] = plan["readback"]
-                continue
-            if not path.exists():
-                _write(
-                    path,
-                    {
-                        k: row[k]
-                        for k in ("request_id", "goal_id", "agent_id", "goal_ref")
-                        if k in row
-                    }
-                    | {
-                        "read_at": _now(),
-                        "kind": "receiver_cli_read",
-                    },
-                )
+    result = read_inbox_transaction(root, {
+        "operation": "record_read", "goal_id": items[0]["goal_id"],
+        "agent_id": items[0]["agent_id"], "items": items, "deliveries": [],
+    }, scope=scope)
+    for row, receipt in zip(items, result["read_receipts"], strict=True):
+        if receipt is not None:
+            row["request_read_receipt"] = receipt
 
 
 def _receipt(root: Path, lane: str, row: dict[str, Any]) -> tuple[dict[str, Any], str | None]:
@@ -719,31 +491,11 @@ def iter_result_paths(folder: Path):
             yield from result_paths(request_folder)
 
 
-DELIVERY_STATUSES = {
-    "admitted",
-    "queued",
-    "retry_pending",
-    "verification_required",
-    "delivered",
-    "superseded",
-    "explicit_unverified",
-}
-DELIVERY_ERRORS = {
-    "provider_delivery_unverified",
-    "provider_locator_unavailable",
-    "provider_verifier_unavailable",
-    "provider_verification_unavailable",
-    "provider_delivery_intent_conflict",
-    "provider_message_missing",
-    "provider_delivery_mismatch",
-    "return_authorization_unavailable",
-    "original_route_unavailable",
-    "initial_delivery_receipt_unavailable",
-    "original_route_or_return_delivery_unavailable",
-    "delivery_state_unreadable",
-    "return_transport_unavailable",
-    "manager_return_payload_conflict",
-}
+# Retained public re-exports share the native delivery contract.
+_delivery_contract = json.loads(Path(__file__).with_name("return_delivery_contract.json").read_text(encoding="utf-8"))
+DELIVERY_STATUSES = set(_delivery_contract["delivery_statuses"])
+DELIVERY_ERRORS = set(_delivery_contract["delivery_errors"])
+
 
 def reply_status(root, row):
     result = []
diff --git a/loopx/control_plane/collaboration/inbox_read.py b/loopx/control_plane/collaboration/inbox_read.py
new file mode 100644
index 0000000000..0a712b58fd
--- /dev/null
+++ b/loopx/control_plane/collaboration/inbox_read.py
@@ -0,0 +1,42 @@
+"""Transport for the native Inbox owner; callers retain their Goal lifetime fence."""
+from pathlib import Path
+
+from ..effect_runtime import (
+    EffectRuntimeIOError,
+    EffectRuntimeRejected,
+    effect_runtime_result,
+)
+
+
+def read_inbox_transaction(root: Path, request: dict, *, scope=None) -> dict:
+    from .inbox import _now
+
+    facts = {
+        "runtime_root": str(Path(root).resolve()),
+        "scope": {
+            "profile_id": scope.profile_id if scope is not None else None,
+            "caller_goal_ref": scope.caller_goal_ref if scope is not None else None,
+            "current_goal_ref": scope.current_goal_ref if scope is not None else None,
+        },
+        "observed_at": _now(),
+        **request,
+    }
+    if request.get("operation") == "record_read":
+        facts["items"] = [{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 request["items"]]
+    # Full source pages may exceed the ordinary wire even when every record is legal.
+    snapshot = request["operation"] in {"pending", "turn_start", "receiver_read", "returns"}
+    snapshot = snapshot or len(request.get("rows", [])) > 1
+    try:
+        result = effect_runtime_result("collaboration.inbox.read", facts, large_local_snapshot=snapshot)
+    except EffectRuntimeRejected as exc:
+        raise ValueError(str(exc)) from exc
+    except EffectRuntimeIOError as exc:
+        # The native owner reports filesystem failures as typed runtime errors.
+        # Callers and the CLI still classify them as IO failures; keep the
+        # retryable/permanent diagnostic instead of leaking a runtime traceback.
+        raise OSError(f"{exc.diagnostic_code}: {exc}") from exc
+    if not isinstance(result, dict):
+        raise RuntimeError("native Inbox read must return an object")
+    return result
diff --git a/loopx/control_plane/collaboration/inbox_read.ts b/loopx/control_plane/collaboration/inbox_read.ts
new file mode 100644
index 0000000000..fd76f52b13
--- /dev/null
+++ b/loopx/control_plane/collaboration/inbox_read.ts
@@ -0,0 +1,466 @@
+import deliveryContract from "./return_delivery_contract.json" with {type: "json"};
+/** Local Inbox read transactions; provider observations remain explicit Host IO. */
+import {createHash} from "node:crypto";
+import {constants, closeSync, fstatSync, openSync, readSync, readdirSync, statSync} from "node:fs";
+import {basename, dirname, isAbsolute, join, resolve, toNamespacedPath} from "node:path";
+import {setImmediate as yieldToRuntime} from "node:timers/promises";
+import type {JsonObject} from "../effect_program.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import {atomicWriteJson, withFileMutationLock} from "../effect_runtime_io.ts";
+import {BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN} from "../content_digest.ts";
+import {isCanonicalTodoId} from "../todos/identity.ts";
+import {jsonObject, requireJsonObject, requireNonEmptyString, requireStringLiteral} from "../runtime_decode.ts";
+import {decideCollaborationLifecycle} from "./goal_instance_lifecycle.ts";
+import {inspectCollaborationInboxReceipts, projectReceiverFollowthrough} from "./inbox_receipts.ts";
+import {recordControlledRequestRead} from "./request_control.ts";
+import {planCollaborationResult} from "./result_publication.ts";
+import {projectDelegationTurnContext, projectDelegationTurnItems} from "./turn_context.ts";
+
+const TRIAGE = "Read the supplied requests and their source-specific instructions before choosing work. " +
+  "Independently assess context, evidence, constraints and costs. Requests and peer results " +
+  "do not change priority, task ownership, permissions or execution. Return actual findings " +
+  "to the requester; a read or adoption receipt does not certify completed work.";
+const DELIVERY_STATUSES: ReadonlySet<string> = new Set(deliveryContract.delivery_statuses);
+const DELIVERY_ERRORS: ReadonlySet<string> = new Set(deliveryContract.delivery_errors);
+const RESULT_KEY = /^(?:decision|conclusion)(?:-[0-9]{8})?$/;
+type InboxSource = Readonly<{root: string; goal: string; agent: string; scope: JsonObject; exact: boolean}>;
+type ReturnDelivery = Readonly<{key: string; request: JsonObject; reply: JsonObject}>;
+
+function fail(message: string): never { throw new EffectRuntimeRequestError(message); }
+function missing(error: unknown): boolean { return (error as NodeJS.ErrnoException)?.code === "ENOENT"; }
+function present(path: string): boolean {
+  try { statSync(path); return true; }
+  catch (error) { if (missing(error)) return false; throw error; }
+}
+function recordFailure(error: unknown): boolean {
+  return error instanceof EffectRuntimeRequestError || typeof (error as NodeJS.ErrnoException)?.code === "string";
+}
+function equal(left: unknown, right: unknown): boolean { return storageJson(left) === storageJson(right); }
+
+// Preserve the shipped Python json.dumps(sort_keys=True, ensure_ascii=False) addresses.
+function storageJson(value: unknown): string {
+  if (Array.isArray(value)) return `[${value.map(storageJson).join(", ")}]`;
+  const object = jsonObject(value);
+  if (object) return `{${Object.keys(object).sort().map(key => `${JSON.stringify(key)}: ${storageJson(object[key])}`).join(", ")}}`;
+  return JSON.stringify(value ?? null);
+}
+function digest(value: unknown): string { return createHash("sha256").update(storageJson(value)).digest("hex"); }
+function requestId(value: unknown): string {
+  if (typeof value !== "string" || !BARE_SHA256_PATTERN.test(value)) fail("invalid context request id");
+  return value;
+}
+function readRecord(path: string): JsonObject {
+  const fd = openSync(toNamespacedPath(path), constants.O_RDONLY | (constants.O_NONBLOCK ?? 0));
+  try {
+    const stat = fstatSync(fd);
+    if (!stat.isFile()) fail("invalid manager context record");
+    if (stat.size > 128_000) fail("manager context record too large");
+    const buffer = Buffer.alloc(128_001);
+    let size = 0, count: number;
+    while (size < buffer.length && (count = readSync(fd, buffer, size, buffer.length - size, null)) > 0) size += count;
+    if (size > 128_000) fail("manager context record too large");
+    const bytes = buffer.subarray(0, size);
+    let parsed: unknown;
+    try {
+      parsed = JSON.parse(new TextDecoder("utf-8", {fatal: true, ignoreBOM: true}).decode(bytes), (_key, value) => {
+        if (typeof value === "number" && !Number.isFinite(value)) throw new SyntaxError("non-finite record value");
+        return value;
+      });
+    }
+    catch (error) {
+      if (!(error instanceof SyntaxError || error instanceof TypeError)) throw error;
+      fail("invalid manager context record");
+    }
+    return requireJsonObject(parsed, "manager context record");
+  } finally { closeSync(fd); }
+}
+function names(folder: string): string[] {
+  try { return readdirSync(toNamespacedPath(folder)).sort(); }
+  catch (error) { if (missing(error)) return []; throw error; }
+}
+function target(source: InboxSource, agent = source.agent): JsonObject {
+  return source.exact ? {goal_ref: source.scope.caller_goal_ref, agent_id: agent}
+    : {goal_id: source.goal, agent_id: agent};
+}
+function identity(source: InboxSource, agent = source.agent): JsonObject {
+  return {goal_id: source.goal, agent_id: agent,
+    ...(source.exact ? {goal_ref: source.scope.caller_goal_ref} : {})};
+}
+function entryPath(source: InboxSource, id: string, agent = source.agent): string {
+  return join(source.root, "entries", digest(target(source, agent)), requestId(id) + ".json");
+}
+function sourceRecord(row: JsonObject): JsonObject {
+  if (Object.hasOwn(row, "message") && typeof row.message !== "string"
+      || row.brief != null && !jsonObject(row.brief)) fail("invalid Inbox source content");
+  return row;
+}
+function entry(source: InboxSource, id: string, agent = source.agent): JsonObject {
+  const row = readRecord(entryPath(source, id, agent));
+  if (Object.entries({...identity(source, agent), request_id: id}).some(([key, value]) => !equal(row[key], value))) {
+    fail("context receipt scope mismatch");
+  }
+  return sourceRecord(row);
+}
+function pendingEntry(source: InboxSource, path: string): JsonObject {
+  const id = basename(path, ".json"), row = readRecord(path);
+  if (row.schema_version !== (source.exact ? "loopx_manager_context_entry_v2" : "loopx_manager_context_entry_v1")
+      || row.goal_id !== source.goal || row.agent_id !== source.agent || row.request_id !== id) {
+    fail("context inbox scope mismatch");
+  }
+  return sourceRecord(row);
+}
+function lifecycle(source: InboxSource, operation: string, row: JsonObject, route: JsonObject = {}): boolean {
+  const decision = decideCollaborationLifecycle({...source.scope, operation,
+    record_goal_ref: row.goal_ref ?? null, route_goal_ref: route.goal_ref ?? null, initial_delivery: null});
+  if (decision.kind === "reject") fail(`collaboration lifecycle rejected: ${decision.code}`);
+  return decision.kind !== "omit";
+}
+function observe(path: string): JsonObject {
+  try { return {state: "read", value: readRecord(path)}; }
+  catch (error) {
+    if (missing(error)) return {state: "absent"};
+    if (!recordFailure(error)) throw error;
+    return {state: "unavailable"};
+  }
+}
+function resultKeys(folder: string): string[] {
+  return names(folder).filter(name => name.endsWith(".json") && RESULT_KEY.test(name.slice(0, -5)))
+    .map(name => name.slice(0, -5)).sort((left, right) => {
+      const rank = (key: string) => key === "decision" ? 0 : key === "conclusion" ? 1 : 2;
+      return rank(left) - rank(right) || (left < right ? -1 : left > right ? 1 : 0);
+    });
+}
+function resultObservation(key: string, reply: JsonObject): JsonObject {
+  const text = reply.text;
+  if (typeof text !== "string" || !text.trim()) fail("bounded reply text is required");
+  // UTF-8 hashing must refuse unpaired surrogates rather than hash replacement bytes.
+  if (/[\uD800-\uDFFF]/u.test(text)) fail("invalid reply text encoding");
+  return {key, value: Object.fromEntries(["request_id", "goal_id", "agent_id", "source_id", "goal_ref", "phase",
+    "decision", "result_key", "previous_result_key", "update_id", "attachments", "request_revision",
+    "request_revision_digest"].filter(field => Object.hasOwn(reply, field)).map(field => [field, reply[field]])),
+  text_chars: [...text].length, text_sha256: createHash("sha256").update(text).digest("hex")};
+}
+function receiptState(root: string, row: JsonObject): JsonObject {
+  const id = requestId(row.request_id), folder = join(root, "replies", id);
+  const observation: JsonObject = {request: Object.fromEntries(["request_id", "goal_id", "agent_id", "source_id", "goal_ref"]
+    .filter(key => Object.hasOwn(row, key)).map(key => [key, row[key]])),
+  route_present: present(join(root, "roundtrips", id + ".json")),
+  decision: observe(join(root, "decisions", id + ".json")), conclusion: observe(join(folder, "conclusion.json"))};
+  if (Object.hasOwn(row, "request_control")) {
+    observation.request = row;
+    observation.read = observe(join(root, "reads", id + ".json"));
+    observation.results = resultKeys(folder).map(key => {
+      try { return resultObservation(key, readRecord(join(folder, key + ".json"))); }
+      catch (error) { if (!recordFailure(error)) throw error; return {unavailable: true}; }
+    });
+  }
+  return (inspectCollaborationInboxReceipts({observations: [observation]}).items as JsonObject[])[0];
+}
+function inboxItem(row: JsonObject, state: JsonObject): JsonObject {
+  const item: JsonObject = {...row, inbox_state: state.kind};
+  delete item.request_control;
+  const view = jsonObject(state.request_control_view);
+  if (view) Object.assign(item, {original_brief: row.brief, brief: view.brief, request_control_view: view,
+    request_revision: view.revision, request_revision_digest: view.revision_digest,
+    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 !== null) Object.assign(item, {receiver_decision_recorded: true,
+    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 as unknown[]).length) Object.assign(item, {warnings: state.warnings,
+    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."});
+  else if (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.";
+  else if (view) item.next_action = view.receiver_instruction;
+  return item;
+}
+async function pendingPage(source: InboxSource, params: JsonObject): Promise<JsonObject> {
+  if (params.request_id != null) {
+    if (params.cursor != null || params.operation_cursor != null) fail("an exact request read cannot use inbox cursors");
+    let row: JsonObject;
+    try { row = entry(source, requestId(params.request_id)); }
+    catch (error) { if (missing(error)) fail("context request not found for this Goal and Agent"); throw error; }
+    if (row.schema_version !== (source.exact ? "loopx_manager_context_entry_v2" : "loopx_manager_context_entry_v1")) fail("context inbox scope mismatch");
+    if (!lifecycle(source, "inbox_observe", row)) fail("context request is unavailable in this Goal lifetime");
+    return {ok: true, request_id: params.request_id, items: [inboxItem(row, receiptState(source.root, row))],
+      has_more: false, next_cursor: null, instruction: TRIAGE + " This is an exact source read, including concluded requests; it does not reopen their work."};
+  }
+  const scope = digest(source.exact ? ["pending_requests_v1", params.runtime_root, target(source)]
+    : ["pending_requests_v1", params.runtime_root, source.goal, source.agent]);
+  let after = "";
+  if (params.cursor != null) {
+    if (typeof params.cursor !== "string" || !/^1:[a-f0-9]{64}:[a-f0-9]{64}$/.test(params.cursor)) fail("invalid pending request cursor");
+    const [, suppliedScope, suppliedAfter] = params.cursor.split(":");
+    if (suppliedScope !== scope) fail("pending request cursor scope mismatch");
+    after = suppliedAfter;
+  }
+  const folder = join(source.root, "entries", digest(target(source))), items: JsonObject[] = [];
+  for (const [index, name] of names(folder).entries()) {
+    if (index > 0 && index % 128 === 0) await yieldToRuntime();
+    if (!name.endsWith(".json") || !BARE_SHA256_PATTERN.test(name.slice(0, -5)) || name.slice(0, -5) <= after) continue;
+    const row = pendingEntry(source, join(folder, name)), state = receiptState(source.root, row);
+    if (state.kind === "settled" || !lifecycle(source, "inbox_observe", row)) continue;
+    items.push(inboxItem(row, state));
+    if (items.length === 21) break;
+  }
+  return {ok: true, items: items.slice(0, 20), has_more: items.length > 20,
+    next_cursor: items.length > 20 ? `1:${scope}:${items[19].request_id}` : null, _cursor_scope: scope,
+    instruction: TRIAGE + " 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."};
+}
+function inspectResult(root: string, row: JsonObject, key: string, reply: JsonObject): JsonObject | null {
+  const folder = join(root, "replies", requestId(row.request_id));
+  try {
+    const results = reply.update_id != null || reply.phase === "decision" ? resultKeys(folder).map(candidate =>
+      resultObservation(candidate, readRecord(join(folder, candidate + ".json")))) : [resultObservation(key, reply)];
+    const legacy = Object.hasOwn(row, "request_control") || Object.hasOwn(reply, "request_revision")
+      ? observe(join(root, "decisions", String(row.request_id) + ".json")) : {};
+    const inspected = planCollaborationResult({request: row, phase: reply.phase, text: reply.text,
+      decision: reply.decision, update_id: reply.update_id, attachments: reply.attachments,
+      request_revision: reply.request_revision, legacy_assessment: legacy.value, results});
+    return (reply.result_key ?? reply.phase) === key ? inspected : null;
+  } catch (error) { if (!recordFailure(error)) throw error; return null; }
+}
+function returnConsumed(root: string, key: string, row: JsonObject, reply: JsonObject, agent: unknown): boolean {
+  const path = join(root, "replies", requestId(row.request_id), key + ".consumed.json");
+  if (!present(path)) return false;
+  const saved = readRecord(path), expected: JsonObject = {request_id: row.request_id, goal_id: row.goal_id,
+    agent_id: agent, ...(Object.hasOwn(reply, "result_key") ? {result_key: key} : {}),
+    ...(Object.hasOwn(row, "goal_ref") ? {goal_ref: row.goal_ref} : {})};
+  if (Object.entries(expected).some(([field, value]) => !equal(saved[field], value))) fail("peer consumption receipt scope mismatch");
+  return true;
+}
+async function collectReturns(source: InboxSource): Promise<{view: JsonObject; deliveries: ReturnDelivery[]}> {
+  const items: JsonObject[] = [], deliveries: ReturnDelivery[] = [];
+  const folder = join(source.root, "peer-operations", digest(target(source)));
+  for (const [index, name] of names(folder).entries()) {
+    if (index > 0 && index % 128 === 0) await yieldToRuntime();
+    if (!name.endsWith(".json") || !BARE_SHA256_PATTERN.test(name.slice(0, -5))) continue;
+    const operation = readRecord(join(folder, name));
+    if (operation.goal_id !== source.goal || operation.source_agent_id !== source.agent
+        || source.exact && !equal(operation.goal_ref, source.scope.caller_goal_ref)) fail("peer return scope mismatch");
+    let row: JsonObject;
+    try { row = entry(source, requestId(operation.request_id), requireNonEmptyString(operation.agent_id, "peer Agent")); }
+    catch (error) { if (missing(error)) continue; throw error; }
+    if (row.source_agent_id !== source.agent || row.source_id !== operation.source_id) fail("peer return scope mismatch");
+    const replies = join(source.root, "replies", String(row.request_id));
+    for (const [resultIndex, key] of resultKeys(replies).entries()) {
+      if (resultIndex > 0 && resultIndex % 128 === 0) await yieldToRuntime();
+      if (key === "decision") continue;
+      const reply = readRecord(join(replies, key + ".json")), inspected = inspectResult(source.root, row, key, reply);
+      if (["request_id", "goal_id", "agent_id", "source_id", "goal_ref"].some(field => Object.hasOwn(row, field) && !equal(reply[field], row[field]))
+          || reply.phase !== "conclusion" || !inspected) fail("peer reply identity conflict");
+      const route = readRecord(join(source.root, "roundtrips", String(row.request_id) + ".json"));
+      if (!lifecycle(source, "peer_return_observe", row, route)
+          || returnConsumed(source.root, key, row, reply, source.agent)) continue;
+      items.push({request_id: row.request_id, ...(key !== "conclusion" ? {result_key: key} : {}),
+        agent_id: row.agent_id, parent_request_id: row.parent_request_id ?? null,
+        brief: inspected.request_brief ?? row.brief,
+        ...(Object.hasOwn(inspected, "request_revision") ? {request_revision: inspected.request_revision,
+          current_request_revision: inspected.current_request_revision} : {}),
+        text: reply.text, decision: reply.decision, created_at: reply.created_at});
+      if (items.length > 20) break;
+      deliveries.push({key, request: Object.fromEntries(["request_id", "goal_id", "agent_id", "goal_ref"]
+        .filter(field => Object.hasOwn(row, field)).map(field => [field, row[field]])),
+      reply: Object.hasOwn(reply, "result_key") ? {result_key: reply.result_key} : {}});
+    }
+    if (items.length > 20) break;
+  }
+  return {view: {items: items.slice(0, 20), has_more: items.length > 20}, deliveries};
+}
+function replyStatuses(root: string, row: JsonObject): JsonObject[] {
+  const folder = join(root, "replies", requestId(row.request_id));
+  return resultKeys(folder).map(key => {
+    const phase = key === "decision" || key.startsWith("decision-") ? "decision" : "conclusion";
+    const identity = {phase, ...(key !== phase ? {result_key: key} : {})};
+    let reply: JsonObject, state: JsonObject;
+    try {
+      reply = readRecord(join(folder, key + ".json"));
+      const path = join(folder, key + ".delivery.json");
+      state = present(path) ? readRecord(path) : {};
+    } catch (error) {
+      if (!recordFailure(error)) throw error;
+      return {...identity, status: "explicit_unverified", created_at: null, delivered_at: null, error: "delivery_state_unreadable"};
+    }
+    let status = Object.hasOwn(state, "status") ? state.status : "queued", error = state.error ?? null;
+    if (jsonObject(status) || Array.isArray(status) || jsonObject(error) || Array.isArray(error)) {
+      fail("invalid return delivery state");
+    }
+    if (typeof status !== "string" || !DELIVERY_STATUSES.has(status)
+        || error !== null && (typeof error !== "string" || !DELIVERY_ERRORS.has(error))) {
+      status = "explicit_unverified"; error = "delivery_state_unreadable";
+    }
+    return {...identity, status, created_at: reply.created_at ?? null,
+      delivered_at: typeof state.delivered_at === "string" && [...state.delivered_at].length <= 80 ? state.delivered_at : null,
+      error, ...(state.verification === "reconciled_after_restart" ? {verification: state.verification} : {})};
+  });
+}
+function agendaResults(root: string, row: JsonObject): JsonObject[] {
+  const folder = join(root, "replies", requestId(row.request_id));
+  const statuses = new Map(replyStatuses(root, row).map(status => [status.result_key ?? status.phase, status.status]));
+  return resultKeys(folder).flatMap(key => {
+    const reply = readRecord(join(folder, key + ".json"));
+    if (reply.phase !== "conclusion") return [];
+    if (!inspectResult(root, row, key, reply)) fail("delegation result identity unavailable");
+    return [{result_key: key, request_revision: reply.request_revision ?? 1, status: statuses.get(key),
+      consumed: row.source_kind === "peer" && returnConsumed(root, key, row, reply, row.source_agent_id)}];
+  });
+}
+function linksReceipt(root: string, row: JsonObject): {value: JsonObject; unavailable: boolean} {
+  const observation = observe(join(root, "links", requestId(row.request_id) + ".json"));
+  if (observation.state === "absent") return {value: {}, unavailable: false};
+  const value = jsonObject(observation.value);
+  const validRefs = (key: string, valid: (ref: string) => boolean) => Array.isArray(value?.[key]) && (value![key] as unknown[]).length <= 16
+    && (value![key] as unknown[]).every(ref => typeof ref === "string" && valid(ref));
+  const valid = value && ["goal_id", "agent_id", "request_id", "goal_ref"].every(field => !Object.hasOwn(row, field) || equal(value[field], row[field]))
+    && validRefs("todo_ids", isCanonicalTodoId) && validRefs("evidence_ids", ref => ENVELOPED_SHA256_PATTERN.test(ref));
+  return {value: valid ? value : {}, unavailable: !valid};
+}
+async function turnContext(source: InboxSource, params: JsonObject): Promise<JsonObject> {
+  let complete = true, unavailable = 0;
+  const batches: JsonObject[] = [], observations: JsonObject[] = [];
+  const flush = () => { if (observations.length) batches.push(projectDelegationTurnItems({observations: observations.splice(0)})); };
+  for (const direction of ["incoming", "outgoing"] as const) {
+    const folder = join(source.root, direction === "incoming" ? "entries" : "peer-operations", digest(target(source)));
+    const paths = names(folder).filter(name => name.endsWith(".json") && BARE_SHA256_PATTERN.test(name.slice(0, -5)));
+    complete &&= paths.length <= 2000;
+    for (const [index, name] of paths.slice(0, 2000).entries()) {
+      if (index > 0 && index % 128 === 0) await yieldToRuntime();
+      let row: JsonObject;
+      try {
+        if (direction === "incoming") row = pendingEntry(source, join(folder, name));
+        else {
+          const operation = readRecord(join(folder, name));
+          if (operation.goal_id !== source.goal || operation.source_agent_id !== source.agent
+              || source.exact && !equal(operation.goal_ref, source.scope.caller_goal_ref)) fail("delegation operation scope mismatch");
+          row = entry(source, requestId(operation.request_id), requireNonEmptyString(operation.agent_id, "peer Agent"));
+          if (row.source_agent_id !== source.agent || row.source_id !== operation.source_id) fail("delegation request scope mismatch");
+        }
+        if (!lifecycle(source, "inbox_observe", row)) continue;
+      } catch (error) { if (!recordFailure(error)) throw error; unavailable++; continue; }
+      const state = receiptState(source.root, row), links = linksReceipt(source.root, row);
+      let results: JsonObject[] = [], evidenceUnavailable = links.unavailable;
+      try { results = agendaResults(source.root, row); }
+      catch (error) { if (!recordFailure(error)) throw error; evidenceUnavailable = true; }
+      const view = jsonObject(state.request_control_view);
+      const receipt: JsonObject = {kind: state.kind, recorded_decision: state.recorded_decision};
+      if (view) receipt.request_control_view = {control_state: view.control_state, active_reservation: view.active_reservation,
+        revision: view.revision, revision_digest: view.revision_digest, brief: {purpose: (view.brief as JsonObject).purpose}};
+      observations.push({direction, request: {...Object.fromEntries(["request_id", "agent_id", "source_kind", "source_agent_id", "goal_ref"]
+        .filter(key => Object.hasOwn(row, key)).map(key => [key, row[key]])),
+      purpose: (jsonObject(row.brief)?.purpose || [...(row.message as string ?? "")].slice(0, 2000).join(""))},
+      receipt, todo_ids: links.value.todo_ids ?? [], evidence_unavailable: evidenceUnavailable, results});
+      if (observations.length === 128) flush();
+    }
+  }
+  flush();
+  return projectDelegationTurnContext({batches, scan_complete: complete, unavailable_count: unavailable,
+    observed_at: params.observed_at, read_command: params.read_command});
+}
+async function recordReads(source: InboxSource, params: JsonObject): Promise<JsonObject> {
+  if (!Array.isArray(params.items) || params.items.length > 20
+      || !Array.isArray(params.deliveries) || params.deliveries.length > 20) fail("bounded inbox read observations are required");
+  const followthrough = params.work_observations == null ? []
+    : projectReceiverFollowthrough({observations: params.work_observations}).items;
+  const receipts: (JsonObject | null)[] = [];
+  for (const raw of params.items) {
+    const row = requireJsonObject(raw, "supplied request"), id = requestId(row.request_id);
+    if (row.goal_id !== source.goal || row.agent_id !== source.agent) fail("context receipt scope mismatch");
+    entry(source, id);
+    lifecycle(source, "read_record", row);
+    const readPath = join(source.root, "reads", id + ".json");
+    const lock = source.exact ? join(source.root, "request-locks", id) : readPath.slice(0, -5) + ".lock";
+    receipts.push(await withFileMutationLock(lock, async () => {
+      const stored = entry(source, id);
+      if (Object.hasOwn(stored, "request_control")) {
+        const observation = observe(join(source.root, "decisions", id + ".json"));
+        const plan = recordControlledRequestRead({entry: stored, actor_agent_id: row.agent_id,
+          observed_at: params.observed_at, revision: row.request_revision, revision_digest: row.request_revision_digest,
+          legacy_assessment: observation.value ?? null, legacy_assessment_unavailable: observation.state === "unavailable"});
+        if (plan.changed) await atomicWriteJson(entryPath(source, id), plan.entry as JsonObject);
+        return plan.readback as JsonObject;
+      }
+      if (!present(readPath)) await atomicWriteJson(readPath,
+        {...Object.fromEntries(["request_id", "goal_id", "agent_id", "goal_ref"].filter(key => Object.hasOwn(row, key)).map(key => [key, row[key]])),
+          read_at: params.observed_at, kind: "receiver_cli_read"});
+      return null;
+    }));
+  }
+  for (const raw of params.deliveries) {
+    const delivered = requireJsonObject(raw, "supplied peer result"), key = requireNonEmptyString(delivered.key, "result key");
+    if (!RESULT_KEY.test(key)) fail("invalid peer result key");
+    const row = requireJsonObject(delivered.request, "peer request"), reply = requireJsonObject(delivered.reply, "peer result");
+    if (row.goal_id !== source.goal || Object.hasOwn(reply, "result_key") && reply.result_key !== key) fail("peer return scope mismatch");
+    const path = join(source.root, "replies", requestId(row.request_id), key + ".delivery.json");
+    const lock = source.exact ? join(source.root, "request-locks", String(row.request_id)) : join(dirname(path), key + ".lock");
+    await withFileMutationLock(lock, async () => {
+      const stored = entry(source, String(row.request_id), requireNonEmptyString(row.agent_id, "peer Agent"));
+      if (stored.source_agent_id !== source.agent) fail("peer return scope mismatch");
+      lifecycle(source, "read_record", stored);
+      if (!present(path)) await atomicWriteJson(path, {status: "delivered",
+        ...(Object.hasOwn(reply, "result_key") ? {result_key: key} : {}), delivered_at: params.observed_at,
+        kind: "requester_cli_read", ...(Object.hasOwn(stored, "goal_ref") ? {goal_ref: stored.goal_ref} : {})});
+    });
+  }
+  return {read_receipts: receipts, followthrough,
+    ...(params.include_context === true ? {context: await turnContext(source, params)} : {})};
+}
+
+function receiverWork(source: InboxSource, items: JsonObject[]): {observations: JsonObject[]; needsProvider: boolean} {
+  let needsProvider = false;
+  const observations = items.map(item => {
+    const links = linksReceipt(source.root, item), tids = (links.value.todo_ids ?? []) as unknown[];
+    // The existing Host reader also resolves the workspace/alias for an empty input list.
+    needsProvider ||= tids.length > 0 || Object.hasOwn(item, "brief");
+    return {kind: item.inbox_state, recorded_decision: item.recorded_decision,
+      linked_todos: [], evidence_refs: links.value.evidence_ids ?? [], evidence_unavailable: links.unavailable};
+  });
+  return {observations, needsProvider};
+}
+
+export async function readCollaborationInbox(params: JsonObject): Promise<JsonObject> {
+  const runtime = requireNonEmptyString(params.runtime_root, "runtime root");
+  if (!isAbsolute(runtime)) fail("inbox runtime root must be absolute");
+  const root = toNamespacedPath(join(runtime, ".local", "manager-context"));
+  const operation = requireStringLiteral(params.operation, ["pending", "turn_context", "turn_start", "receiver_read",
+    "record_read", "returns", "receipt_states"], "inbox read operation");
+  if (operation === "receipt_states") {
+    if (!Array.isArray(params.rows) || params.rows.length > 128) fail("bounded inbox requests are required");
+    return {items: params.rows.map(row => receiptState(root, requireJsonObject(row, "inbox request")))};
+  }
+  const scope = requireJsonObject(params.scope, "Goal scope");
+  if (scope.profile_id !== null && scope.profile_id !== "source_session_v1") fail("unsupported collaboration profile");
+  const source: InboxSource = {root, goal: requireNonEmptyString(params.goal_id, "Goal"),
+    agent: requireNonEmptyString(params.agent_id, "Agent"), scope, exact: scope.profile_id === "source_session_v1"};
+  const observedAt = requireNonEmptyString(params.observed_at, "Inbox observation time");
+  if ([...observedAt].length > 80) fail("Inbox observation time is too long");
+  if (operation === "turn_context" || operation === "turn_start" || operation === "receiver_read" || params.include_context === true) {
+    requireNonEmptyString(params.read_command, "Inbox read command");
+  }
+  if (operation === "record_read") return recordReads(source, params);
+  if (operation === "returns") {
+    const offered = await collectReturns(source);
+    if (params.mark_read === true) await recordReads(source, {...params, items: [], deliveries: offered.deliveries});
+    return offered.view;
+  }
+  if (operation === "turn_context") return turnContext(source, params);
+  const page = await pendingPage(source, {...params, runtime_root: resolve(runtime)});
+  const cursorScope = page._cursor_scope;
+  delete page._cursor_scope;
+  const returns = params.request_id == null ? await collectReturns(source) : {view: {items: []}, deliveries: []};
+  if ((returns.view.items as unknown[]).length) page.peer_returns = returns.view;
+  if (operation === "turn_start") return {result: page, cursor_scope: cursorScope, context: await turnContext(source, params)};
+  if (operation === "receiver_read") {
+    const work = receiverWork(source, page.items as JsonObject[]);
+    if (params.defer_recording !== true && !work.needsProvider) {
+      for (const [index, observation] of work.observations.entries()) {
+        if (observation.evidence_unavailable) {
+          const item = (page.items as JsonObject[])[index];
+          item.warnings = [...(item.warnings as unknown[] ?? []), "links_unreadable_or_conflicting"];
+        }
+      }
+      const final = await recordReads(source, {...params, items: page.items, deliveries: returns.deliveries,
+        work_observations: work.observations, include_context: params.request_id == null});
+      return {result: page, cursor_scope: cursorScope, completed: true, ...final};
+    }
+  }
+  return {result: page, cursor_scope: cursorScope, ...(operation === "receiver_read" ? {deliveries: returns.deliveries} : {})};
+}
diff --git a/loopx/control_plane/collaboration/links.py b/loopx/control_plane/collaboration/links.py
index 61e4346c23..d6b26e56b1 100644
--- a/loopx/control_plane/collaboration/links.py
+++ b/loopx/control_plane/collaboration/links.py
@@ -9,7 +9,6 @@
 from ...control_plane.todos.list_readback import list_goal_todos
 from ..runtime.public_safety import public_safe_compact_text
 from ..todos.contract import TODO_ID_PATTERN
-from ..effect_runtime import effect_runtime_result
 from ..content_digest import ENVELOPED_SHA256_PATTERN
 from .inbox import _entry, _now, _receipt, _request_lock, _root, _write
 from .goal_instance_scope import collaboration_goal_scope, decide_collaboration_lifecycle
@@ -72,31 +71,19 @@ def read_linked_work(root, registry_path, item, todos_cache, *, include_core_det
     }
 
 
-def receiver_followthrough(root, registry_path, items):
-    """Compose one bounded read model for CLI and scoped MCP receivers.
-
-    Does not select work, create a Todo, infer links from prose or write a
-    decision on behalf of another Agent. The typed owner supplies advice.
-    """
+def receiver_work_observations(root, registry_path, items):
+    """Read the actual linked Core provider once per Goal instance."""
     todos_cache, observations = {}, []
     for item in items:
         work = read_linked_work(root, registry_path, item, todos_cache, include_core_details=True)
         observations.append({
-            "kind": item["inbox_state"],
-            "recorded_decision": item["recorded_decision"],
-            "linked_todos": work["linked_todos"],
-            "evidence_refs": work["evidence_refs"],
+            "kind": item["inbox_state"], "recorded_decision": item["recorded_decision"],
+            "linked_todos": work["linked_todos"], "evidence_refs": work["evidence_refs"],
             "evidence_unavailable": work["evidence_unavailable"],
         })
         if work["warnings"]:
             item.setdefault("warnings", []).extend(work["warnings"])
-    if items:
-        views = effect_runtime_result(
-            "collaboration.inbox.receiver_followthrough", {"observations": observations}
-        )["items"]
-        for item, view in zip(items, views, strict=True):
-            item["receiver_followthrough"] = view
-    return items
+    return observations
 
 
 def link(
diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py
index e891d5784d..14e3ce2742 100644
--- a/loopx/control_plane/collaboration/peers.py
+++ b/loopx/control_plane/collaboration/peers.py
@@ -234,150 +234,18 @@ def request(
         }
 
 
-def _return_consumed(path, row, reply, *, agent_id):
-    """One readback identity for peer consumption and the per-Turn agenda."""
-    consumed = path.with_name(path.stem + ".consumed.json")
-    if not consumed.exists():
-        return False
-    value = _read(consumed)
-    expected = {"request_id": row["request_id"], "goal_id": row["goal_id"],
-        "agent_id": agent_id,
-        **({"result_key": path.stem} if "result_key" in reply else {}),
-        **({"goal_ref": row["goal_ref"]} if "goal_ref" in row else {})}
-    if any(value.get(key) != item for key, item in expected.items()):
-        raise ValueError("peer consumption receipt scope mismatch")
-    return True
-
-
-def _collect_returns(root, goal_id, agent_id, *, scope=None):
-    items = []
-    deliveries = []
-    folder = (
-        _root(root)
-        / "peer-operations"
-        / _hash(_target(goal_id, agent_id, scope))
-    )
-    for operation_path in sorted(folder.glob("*.json")):
-        if not BARE_SHA256_PATTERN.fullmatch(operation_path.stem):
-            continue  # Lock holder sidecars are not peer operation records.
-        operation = _read(operation_path)
-        if (
-            operation.get("goal_id") != goal_id
-            or operation.get("source_agent_id") != agent_id
-            or (
-                scope is not None
-                and scope.exact
-                and operation.get("goal_ref") != scope.caller_goal_ref
-            )
-        ):
-            raise ValueError("peer return scope mismatch")
-        try:
-            row = _entry(
-                root,
-                goal_id,
-                operation["agent_id"],
-                operation["request_id"],
-                scope=scope,
-            )
-        except FileNotFoundError:
-            continue  # A reserved send without an entry is repaired by its exact retry.
-        if row.get("source_agent_id") != agent_id or row.get(
-            "source_id"
-        ) != operation.get("source_id"):
-            raise ValueError("peer return scope mismatch")
-        from .inbox import result_paths, inspect_result_identity
-
-        for path in result_paths(_root(root) / "replies" / row["request_id"]):
-            if path.stem == "decision":
-                continue
-            reply = _read(path)
-            inspected = inspect_result_identity(reply, row, path)
-            if (
-                any(
-                    reply.get(k) != row.get(k)
-                    for k in (
-                        "request_id",
-                        "goal_id",
-                        "agent_id",
-                        "source_id",
-                        "goal_ref",
-                    )
-                    if k in row
-                )
-                or reply.get("phase") != "conclusion"
-                or inspected is None
-            ):
-                raise ValueError("peer reply identity conflict")
-            route = _read(_root(root) / "roundtrips" / (row["request_id"] + ".json"))
-            if scope is not None:
-                lifecycle = decide_collaboration_lifecycle(
-                    scope,
-                    operation="peer_return_observe",
-                    record=row,
-                    route=route,
-                )
-                if lifecycle.get("kind") == "omit":
-                    continue
-            if _return_consumed(path, row, reply, agent_id=agent_id):
-                continue
-            items.append(
-                {
-                    "request_id": row["request_id"],
-                    **({"result_key": path.stem} if path.stem != "conclusion" else {}),
-                    "agent_id": row["agent_id"],
-                    "parent_request_id": row.get("parent_request_id"),
-                    "brief": inspected.get("request_brief", row["brief"]),
-                    **({key: inspected[key] for key in ("request_revision", "current_request_revision")} if "request_revision" in inspected else {}),
-                    "text": reply["text"],
-                    "decision": reply["decision"],
-                    "created_at": reply["created_at"],
-                }
-            )
-            if len(items) > 20:
-                break
-            deliveries.append((path, row, reply))
-        if len(items) > 20:
-            break
-    return {"items": items[:20], "has_more": len(items) > 20}, deliveries
 
 
-def _record_return_reads(root, deliveries, *, scope=None):
-    for path, row, reply in deliveries:
-        state = path.with_name(path.stem + ".delivery.json")
-        with _request_lock(
-            root,
-            row["request_id"],
-            scope,
-            path.with_suffix(".lock"),
-        ):
-            if not state.exists():
-                _write(
-                    state,
-                    {
-                        "status": "delivered",
-                        **({"result_key": path.stem} if "result_key" in reply else {}),
-                        "delivered_at": _now(),
-                        "kind": "requester_cli_read",
-                        **(
-                            {"goal_ref": row["goal_ref"]}
-                            if "goal_ref" in row
-                            else {}
-                        ),
-                    },
-                )
+
+
 
 
 def returns(root, goal_id, agent_id, *, mark_read=False, scope=None):
-    """Re-offer results until the requester explicitly acknowledges consumption."""
-    result, deliveries = _collect_returns(
-        root,
-        goal_id,
-        agent_id,
-        scope=scope,
-    )
-    if mark_read:
-        _record_return_reads(root, deliveries, scope=scope)
-    return result
+    from .inbox_read import read_inbox_transaction
+
+    return read_inbox_transaction(root, {
+        "operation": "returns", "goal_id": goal_id, "agent_id": agent_id, "mark_read": mark_read,
+    }, scope=scope)
 
 
 def consume_return(
@@ -617,77 +485,54 @@ def read_inbox(
     request_id=None,
     caller_goal_ref=None,
 ):
-    from .inbox import pending
-    from .inbox import record_read
+    from .inbox import _pending_result
+    from .inbox_read import read_inbox_transaction
+    from .turn_context_observation import delegation_read_command
+    from .links import receiver_work_observations
 
+    read_command = delegation_read_command(root, registry, goal_id, agent_id)
     with collaboration_goal_scope(
-        registry,
-        goal_id=goal_id,
-        agents=(agent_id,),
-        caller_goal_ref=caller_goal_ref,
+        registry, goal_id=goal_id, agents=(agent_id,), caller_goal_ref=caller_goal_ref,
     ) as goal_scope:
-        result = pending(
-            root,
-            goal_id,
-            agent_id,
-            cursor=cursor,
-            operation_cursor=operation_cursor,
-            request_id=request_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:
-            result.pop("peer_returns", None)
+        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)
+        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)
-        observed_goal_ref = (
-            dict(goal_scope.caller_goal_ref or {}) if goal_scope.exact else None
-        )
-
-    from .links import receiver_followthrough
-
-    receiver_followthrough(root, registry, result["items"])
-    # Input hashing can touch arbitrary workspace files and does not participate
-    # in Goal lifetime admission.
-    for item in result["items"]:
-        if item.get("brief"):
-            item["input_readiness"] = _input_readiness_for_goal(
-                registry,
-                goal_id,
-                observed_goal,
-                item["brief"],
-                workspace=workspace,
-            )
-    if result["items"] or return_deliveries:
+        observed_goal_ref = dict(goal_scope.caller_goal_ref or {}) if goal_scope.exact else None
+
+    if not snapshot.get("completed"):
+        work = receiver_work_observations(root, registry, result["items"])
+        # Input hashing remains outside Goal lifetime admission.
+        for item in result["items"]:
+            if item.get("brief"):
+                item["input_readiness"] = _input_readiness_for_goal(
+                    registry, goal_id, observed_goal, item["brief"], workspace=workspace,
+                )
         with collaboration_goal_scope(
-            registry,
-            goal_id=goal_id,
-            agents=(agent_id,),
-            caller_goal_ref=observed_goal_ref,
+            registry, goal_id=goal_id, agents=(agent_id,), caller_goal_ref=observed_goal_ref,
         ) as goal_scope:
-            if result["items"]:
-                record_read(root, result["items"], scope=goal_scope)
-            else:
-                decide_collaboration_lifecycle(
-                    goal_scope,
-                    operation="read_record",
-                    record=return_deliveries[0][1],
-                )
-            _record_return_reads(root, return_deliveries, scope=goal_scope)
-    from .turn_context_observation import read_delegation_turn_context
-
+            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)
+    else:
+        final = snapshot
+        for item in result["items"]:
+            if item.get("brief"):
+                item["input_readiness"] = []
+    for item, receipt, advice in zip(result["items"], final["read_receipts"], final["followthrough"], strict=True):
+        if receipt is not None:
+            item["request_read_receipt"] = receipt
+        item["receiver_followthrough"] = advice
     if request_id is None:
-        result["unfinished_delegations"] = read_delegation_turn_context(
-            root, registry, goal_id, agent_id)["unfinished_delegations"]
+        result["unfinished_delegations"] = final["context"]["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. "
diff --git a/loopx/control_plane/collaboration/return_delivery_contract.json b/loopx/control_plane/collaboration/return_delivery_contract.json
new file mode 100644
index 0000000000..db0839aeed
--- /dev/null
+++ b/loopx/control_plane/collaboration/return_delivery_contract.json
@@ -0,0 +1,27 @@
+{
+  "delivery_statuses": [
+    "admitted",
+    "delivered",
+    "explicit_unverified",
+    "queued",
+    "retry_pending",
+    "superseded",
+    "verification_required"
+  ],
+  "delivery_errors": [
+    "delivery_state_unreadable",
+    "initial_delivery_receipt_unavailable",
+    "manager_return_payload_conflict",
+    "original_route_or_return_delivery_unavailable",
+    "original_route_unavailable",
+    "provider_delivery_intent_conflict",
+    "provider_delivery_mismatch",
+    "provider_delivery_unverified",
+    "provider_locator_unavailable",
+    "provider_message_missing",
+    "provider_verification_unavailable",
+    "provider_verifier_unavailable",
+    "return_authorization_unavailable",
+    "return_transport_unavailable"
+  ]
+}
diff --git a/loopx/control_plane/collaboration/turn_context_observation.py b/loopx/control_plane/collaboration/turn_context_observation.py
index 37873412b1..2aec38e178 100644
--- a/loopx/control_plane/collaboration/turn_context_observation.py
+++ b/loopx/control_plane/collaboration/turn_context_observation.py
@@ -1,114 +1,20 @@
-"""Read-only filesystem adapter for the TS-owned unfinished delegation agenda."""
-from __future__ import annotations
-
-import json
+"""Host command formatting and Goal admission for native Inbox observations."""
 import shlex
 from pathlib import Path
 
-from ..content_digest import BARE_SHA256_PATTERN
-from ..effect_runtime import effect_runtime_result
-from .goal_instance_scope import collaboration_goal_scope, decide_collaboration_lifecycle
-from .inbox import (
-    _entry, _hash, _now, _pending_entry, _pending_receipt_states, _read,
-    _receipt, _root, _target, inspect_result_identity, result_paths, reply_status,
-)
-
-
-def _paths(folder):
-    # Same exact-Agent folders as inbox/peer requests; never scan other Agents' inboxes.
-    try:
-        paths = folder.iterdir()
-        return sorted(path for path in paths if path.suffix == ".json" and BARE_SHA256_PATTERN.fullmatch(path.stem))
-    except FileNotFoundError:
-        return []
-
+from .goal_instance_scope import collaboration_goal_scope
+from .inbox_read import read_inbox_transaction
 
-def _results(root, row):
-    statuses = {r.get("result_key", r["phase"]): r for r in reply_status(root, row)}
-    results = []
-    for path in result_paths(_root(root) / "replies" / row["request_id"]):
-        value = _read(path)
-        if value.get("phase") != "conclusion":
-            continue
-        if inspect_result_identity(value, row, path) is None:
-            raise ValueError("delegation result identity unavailable")
-        from .peers import _return_consumed
 
-        consumed = row.get("source_kind") == "peer" and _return_consumed(
-            path, row, value, agent_id=row["source_agent_id"])
-        results.append({"result_key": path.stem, "request_revision": value.get("request_revision", 1),
-            "status": statuses[path.stem]["status"], "consumed": consumed})
-    return results
+def delegation_read_command(root, registry, goal_id, agent_id):
+    return shlex.join(["loopx", "--registry", str(registry), "--runtime-root", str(root),
+        "manager-inbox", "read", "--goal-id", goal_id, "--agent-id", agent_id])
 
 
 def read_delegation_turn_context(root: Path, registry: Path, goal_id: str, agent_id: str,
                                  *, read_command: str | None = None) -> dict:
-    read_command = read_command or shlex.join(["loopx", "--registry", str(registry), "--runtime-root",
-        str(root), "manager-inbox", "read", "--goal-id", goal_id, "--agent-id", agent_id])
-    rows, complete, unavailable = [], True, 0
     with collaboration_goal_scope(registry, goal_id=goal_id, agents=(agent_id,)) as scope:
-        base = _root(root)
-        folders = (("incoming", base / "entries" / _hash(_target(goal_id, agent_id, scope))),
-            ("outgoing", base / "peer-operations" / _hash(_target(goal_id, agent_id, scope))))
-        for direction, folder in folders:
-            paths = _paths(folder)
-            complete = complete and len(paths) <= 2000
-            for path in paths[:2000]:
-                try:
-                    if direction == "incoming":
-                        row = _pending_entry(path, goal_id, agent_id, scope)
-                    else:
-                        operation = _read(path)
-                        if (operation.get("goal_id") != goal_id or operation.get("source_agent_id") != agent_id
-                            or scope.exact and operation.get("goal_ref") != scope.caller_goal_ref):
-                            raise ValueError("delegation operation scope mismatch")
-                        row = _entry(root, goal_id, operation["agent_id"], operation["request_id"], scope=scope)
-                        if row.get("source_agent_id") != agent_id or row.get("source_id") != operation.get("source_id"):
-                            raise ValueError("delegation request scope mismatch")
-                    if decide_collaboration_lifecycle(scope, operation="inbox_observe", record=row).get("kind") == "omit":
-                        continue
-                    rows.append((direction, row))
-                except (OSError, ValueError, KeyError, TypeError):
-                    unavailable += 1
-        batches = []
-        # The existing receipt reader batches full controlled requests by RPC wire bytes.
-        for start in range(0, len(rows), 128):
-            batch = rows[start:start + 128]
-            states = _pending_receipt_states(root, [row for _, row in batch])
-            observations = []
-            for (direction, row), state in zip(batch, states, strict=True):
-                links, error = _receipt(root, "links", row)
-                results = []
-                try:
-                    results = _results(root, row)
-                except (OSError, ValueError, KeyError, TypeError):
-                    error = "delegation_result_unavailable"
-                # Only the selected current brief and receipt facts cross this boundary;
-                # source messages, transcripts, result text and private rationale do not.
-                control = state.get("request_control_view")
-                state = {"kind": state["kind"], "recorded_decision": state["recorded_decision"]}
-                if control is not None:
-                    state["request_control_view"] = {key: control.get(key) for key in (
-                        "control_state", "active_reservation", "revision", "revision_digest")}
-                    state["request_control_view"]["brief"] = {"purpose": control["brief"]["purpose"]}
-                observations.append({"direction": direction,
-                    "request": {**{key: row[key] for key in ("request_id", "agent_id", "source_kind",
-                        "source_agent_id", "goal_ref") if key in row},
-                        "purpose": (row.get("brief") or {}).get("purpose") or row.get("message", "")[:2000]},
-                    "receipt": state, "todo_ids": links.get("todo_ids", []),
-                    "evidence_unavailable": bool(error), "results": results})
-            # Large legal Unicode briefs must not overflow the ordinary local RPC.
-            pending = []
-            size = 0
-            for observation in observations:
-                encoded = len(json.dumps(observation, separators=(",", ":")).encode())
-                if pending and size + encoded > 1_500_000:
-                    batches.append(effect_runtime_result("collaboration.inbox.turn_items", {"observations": pending}))
-                    pending, size = [], 0
-                pending.append(observation)
-                size += encoded
-            if pending:
-                batches.append(effect_runtime_result("collaboration.inbox.turn_items", {"observations": pending}))
-    return effect_runtime_result("collaboration.inbox.turn_context", {"batches": batches,
-        "scan_complete": complete, "unavailable_count": unavailable,
-        "observed_at": _now(), "read_command": read_command})
+        return read_inbox_transaction(root, {
+            "operation": "turn_context", "goal_id": goal_id, "agent_id": agent_id,
+            "read_command": read_command or delegation_read_command(root, registry, goal_id, agent_id),
+        }, scope=scope)
diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py
index 7e18939128..22e4966316 100644
--- a/loopx/control_plane/effect_runtime.py
+++ b/loopx/control_plane/effect_runtime.py
@@ -42,6 +42,7 @@
 MAX_REQUEST_BYTES = 2 * 1024 * 1024
 MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024
 LOCAL_SNAPSHOT_METHODS = frozenset({
+    "collaboration.inbox.read",
     "collaboration.delegation.inventory_page",
     "benchmark.trajectory.codex-usage",
     "coordination.local_authority.todo_source",
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 809f5ee933..71edd1cc47 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -653,14 +653,10 @@ export function createEffectRuntimeHandlers(
     ["collaboration.source_context.normalize", lazyHandler(() => import("./collaboration/semantic_request.ts"), ({normalizeCollaborationSourceContext}) => normalizeCollaborationSourceContext)],
     ["collaboration.request.plan_publication", lazyHandler(() => import("./collaboration/request_control.ts"), ({planControlledRequestPublication}) => planControlledRequestPublication)],
     ["collaboration.request.control", lazyHandler(() => import("./collaboration/request_control.ts"), ({planRequestControl}) => planRequestControl)],
-    ["collaboration.request.record_read", lazyHandler(() => import("./collaboration/request_control.ts"), ({recordControlledRequestRead}) => recordControlledRequestRead)],
     ["collaboration.request.assess", lazyHandler(() => import("./collaboration/request_control.ts"), ({recordControlledRequestAssessment}) => recordControlledRequestAssessment)],
     ["collaboration.request.amendment_readback", lazyHandler(() => import("./collaboration/request_control.ts"), ({requestAmendmentReadback}) => requestAmendmentReadback)],
     ["goal.amendment.request_receipt", lazyHandler(() => import("./goals/goal_amendment_authority.ts"), ({projectRequestAmendmentReceipt}) => projectRequestAmendmentReceipt)],
-    ["collaboration.inbox.inspect_receipts", lazyHandler(() => import("./collaboration/inbox_receipts.ts"), ({inspectCollaborationInboxReceipts}) => inspectCollaborationInboxReceipts)],
-    ["collaboration.inbox.turn_items", lazyHandler(() => import("./collaboration/turn_context.ts"), ({projectDelegationTurnItems}) => projectDelegationTurnItems)],
-    ["collaboration.inbox.turn_context", lazyHandler(() => import("./collaboration/turn_context.ts"), ({projectDelegationTurnContext}) => projectDelegationTurnContext)],
-    ["collaboration.inbox.receiver_followthrough", lazyHandler(() => import("./collaboration/inbox_receipts.ts"), ({projectReceiverFollowthrough}) => projectReceiverFollowthrough)],
+    ["collaboration.inbox.read", lazyHandler(() => import("./collaboration/inbox_read.ts"), ({readCollaborationInbox}) => readCollaborationInbox)],
     ["collaboration.result.plan_publication", lazyHandler(() => import("./collaboration/result_publication.ts"), ({planCollaborationResult}) => planCollaborationResult)],
     ["collaboration.result.attachment_refs", lazyHandler(() => import("./collaboration/result_publication.ts"), ({resultAttachmentRefs}) => resultAttachmentRefs)],
     ["collaboration.result.delivery_ready", lazyHandler(() => import("./collaboration/result_publication.ts"), ({collaborationResultDeliveryReady}) => collaborationResultDeliveryReady)],
diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts
index 57852c25d1..9e869b39df 100644
--- a/loopx/control_plane/effect_runtime_server.ts
+++ b/loopx/control_plane/effect_runtime_server.ts
@@ -28,6 +28,7 @@ const MAX_REQUEST_BYTES = 2 * 1024 * 1024;
 const MAX_INLINE_RESPONSE_BYTES = 2 * 1024 * 1024;
 // Explicit opt-in: ordinary effects retain the 2 MiB request/response wire.
 const LOCAL_SNAPSHOT_METHODS = new Set([
+  "collaboration.inbox.read",
   "collaboration.delegation.inventory_page",
   "benchmark.trajectory.codex-usage",
   "coordination.local_authority.todo_source",
diff --git a/pyproject.toml b/pyproject.toml
index dbe072b74b..c6123f2965 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -72,7 +72,7 @@ include = ["loopx*"]
 "loopx.semantics" = ["*.json"]
 "loopx.control_plane.agents" = ["*.ts"]
 "loopx.control_plane.capabilities" = ["*.ts"]
-"loopx.control_plane.collaboration" = ["*.ts"]
+"loopx.control_plane.collaboration" = ["*.ts", "*.json"]
 "loopx.control_plane.coordination" = ["*.ts", "*.json"]
 "loopx.control_plane.goals" = ["*.ts"]
 "loopx.control_plane.goals.goal_frontier" = ["*.ts"]
diff --git a/tests/control_plane/test_inbox_read_recovery.py b/tests/control_plane/test_inbox_read_recovery.py
new file mode 100644
index 0000000000..ee78226ac4
--- /dev/null
+++ b/tests/control_plane/test_inbox_read_recovery.py
@@ -0,0 +1,140 @@
+"""A terminated native reader can replay a partially supplied page."""
+import json
+import subprocess
+import time
+from pathlib import Path
+
+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
+from loopx.control_plane.projects.registry_codec import source_session_registry_transaction
+from loopx.control_plane.effect_runtime import _node_executable
+from loopx.file_lock import exclusive_cross_runtime_file_lock
+
+
+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: {
+        "profile_id": "source_session_v1", "goals": [{"id": "research", "repo": str(tmp_path),
+        "status": "active", "goal_instance_id": "ginst_" + "a" * 32,
+        "coordination": {"registered_agents": ["source", "worker"]}}],
+        "session_bindings": [], "session_receipts": [], "lifetime_receipts": [],
+    }) as transaction:
+        transaction.commit(transaction.payload_copy())
+    brief = {"schema_version": "collaboration_brief_v0", "purpose": "Inspect the current artifact",
+        "context": "Keep existing authority", "constraints": [], "inputs": [],
+        "acceptance": ["Return the findings"], "return_requirement": "Reply to the requester"}
+    for operation in ("first", "second"):
+        request(tmp_path, registry, "research", "source", "worker", operation, brief)
+    owner = _root(tmp_path)
+    before = {p: p.read_bytes() for p in (owner / "entries").rglob("*.json")}
+    module = (Path(__file__).resolve().parents[2] / "loopx/control_plane/collaboration/inbox_read.ts").as_uri()
+    script = f"""import {{readCollaborationInbox}} from {json.dumps(module)};
+let input = ''; for await (const chunk of process.stdin) input += chunk;
+process.stdout.write(JSON.stringify(await readCollaborationInbox(JSON.parse(input))));"""
+    command = [_node_executable(), "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script]
+    with collaboration_goal_scope(registry, goal_id="research", agents=("worker",)) as scope:
+        items = pending(tmp_path, "research", "worker", scope=scope)["items"]
+        first = owner / "reads" / (items[0]["request_id"] + ".json")
+        second = owner / "reads" / (items[1]["request_id"] + ".json")
+        payload = {"operation": "record_read", "runtime_root": str(tmp_path),
+            "goal_id": "research", "agent_id": "worker", "items": items, "deliveries": [],
+            "scope": {"profile_id": scope.profile_id, "caller_goal_ref": scope.caller_goal_ref,
+                "current_goal_ref": scope.current_goal_ref}, "observed_at": "2026-10-19T00:00:00+00:00"}
+        with exclusive_cross_runtime_file_lock(owner / "request-locks" / items[1]["request_id"]):
+            child = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE, text=True, encoding="utf-8")
+            try:
+                child.stdin.write(json.dumps(payload))
+                child.stdin.close()
+                deadline = time.monotonic() + 15
+                while not first.exists() and child.poll() is None and time.monotonic() < deadline:
+                    time.sleep(0.02)
+                assert first.exists(), child.stderr.read() if child.poll() is not None else "first read was not committed"
+                assert not second.exists()
+                committed = first.read_bytes()
+            finally:
+                if child.poll() is None:
+                    child.terminate()
+                child.wait(timeout=15)
+                child.stdout.close()
+                child.stderr.close()
+        payload["observed_at"] = "2026-10-19T00:00:01+00:00"
+        replay = subprocess.run(command, input=json.dumps(payload), capture_output=True,
+            text=True, encoding="utf-8", timeout=30)
+        assert replay.returncode == 0, replay.stderr
+        assert json.loads(replay.stdout)["read_receipts"] == [None, None]
+        assert first.read_bytes() == committed
+        assert json.loads(second.read_text(encoding="utf-8"))["read_at"] == payload["observed_at"]
+    assert {p: p.read_bytes() for p in before} == before
+
+
+def test_large_legal_inbox_page_keeps_complete_source_and_records_supplied_reads(tmp_path):
+    from loopx.control_plane.collaboration.inbox import (
+        ENTRY_SCHEMA, _hash, _write, normalize_request, normalize_source_context, plan_request_publication,
+    )
+    from loopx.control_plane.collaboration.peers import read_inbox
+    from loopx.control_plane.effect_runtime import MAX_RESPONSE_BYTES
+
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"goals": [{"id": "research", "repo": str(tmp_path),
+        "coordination": {"registered_agents": ["worker"]}}]}), encoding="utf-8")
+    message = normalize_source_context("😀" * 24_000)
+    target = {"goal_id": "research", "agent_id": "worker"}
+    normalized = normalize_request({**target, "brief": {"schema_version": "collaboration_brief_v0",
+        "purpose": "p" * 500, "context": "c" * 6000, "constraints": ["x" * 1000] * 8,
+        "inputs": [], "acceptance": ["a" * 1000], "return_requirement": "Return the findings"}})
+    folder = _root(tmp_path) / "entries" / _hash(target)
+    for index in range(20):
+        source_id = f"synthetic-source:{index}"
+        rid = _hash([source_id, target])
+        source = {"schema_version": ENTRY_SCHEMA, "request_id": rid, "source_id": source_id,
+            **normalized, "message": message}
+        publication = plan_request_publication(source, entry=None)
+        _write(folder / (rid + ".json"), publication["entry"])
+    before = {p: p.read_bytes() for p in folder.glob("*.json")}
+
+    response = read_inbox(tmp_path, registry, "research", "worker", workspace=tmp_path)
+
+    assert len(response["items"]) == 20
+    assert len(json.dumps(response, ensure_ascii=False).encode()) > MAX_RESPONSE_BYTES
+    assert all(item["message"] == message and item["brief"] == normalized["brief"] for item in response["items"])
+    assert all(item["recorded_decision"] is None for item in response["items"])
+    assert {p: p.read_bytes() for p in before} == before
+    assert len(list((_root(tmp_path) / "reads").glob("*.json"))) == 20
+
+
+def test_operation_source_appearing_after_committed_read_is_observed_on_next_read(tmp_path, monkeypatch):
+    import pytest
+    from loopx.chat_store import ChatSessionStore
+    from loopx.capabilities.manager_context import deliver
+    from loopx.control_plane.collaboration import inbox_read
+    from loopx.control_plane.collaboration.peers import read_inbox
+
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"goals": [{"id": "research", "repo": str(tmp_path),
+        "coordination": {"registered_agents": ["worker"]}}]}), encoding="utf-8")
+    store = ChatSessionStore(tmp_path)
+    session = store.create_session(goal_id="loopx-manager", agent_id="codex", adapter_kind="codex_app_server",
+        upstream_thread_id="synthetic", channel_id="manager")
+    turn, _ = store.create_turn(session["session_id"], client_turn_id="source", message="Inspect the artifact", origin="web")
+    receipt = deliver(tmp_path, registry, session=session, turn=turn,
+        request={"goal_id": "research", "agent_id": "worker"})
+    transport = inbox_read.read_inbox_transaction
+
+    def source_appears(root, query, **kwargs):
+        response = transport(root, query, **kwargs)
+        if query["operation"] == "receiver_read":
+            path = root / "chat" / "actions" / "actions.json"
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text("{broken", encoding="utf-8")
+        return response
+
+    monkeypatch.setattr(inbox_read, "read_inbox_transaction", source_appears)
+    observed = read_inbox(tmp_path, registry, "research", "worker")
+    assert [item["request_id"] for item in observed["items"]] == [receipt["request_id"]]
+    reads = {p: p.read_bytes() for p in (_root(tmp_path) / "reads").glob("*.json")}
+    monkeypatch.setattr(inbox_read, "read_inbox_transaction", transport)
+    with pytest.raises(ValueError):
+        read_inbox(tmp_path, registry, "research", "worker")
+    assert {p: p.read_bytes() for p in reads} == reads
diff --git a/tests/control_plane_ts/inbox_read_transaction.test.ts b/tests/control_plane_ts/inbox_read_transaction.test.ts
new file mode 100644
index 0000000000..6aeeee0ce3
--- /dev/null
+++ b/tests/control_plane_ts/inbox_read_transaction.test.ts
@@ -0,0 +1,222 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {mkdtempSync, mkdirSync, readFileSync, readdirSync, rmSync, writeFileSync} from "node:fs";
+import {join} from "node:path";
+import {tmpdir} from "node:os";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+import {readCollaborationInbox} from "../../loopx/control_plane/collaboration/inbox_read.ts";
+
+const ref = {goal_id: "research", goal_instance_id: "ginst_" + "a".repeat(32)};
+// Addresses captured from the shipped Python storage codec, including its spaces.
+const targets = {
+  legacy: "433a92de158f221e8f912110b3736ae8aca7f605fc96be50019e32311407e69b",
+  exact: "f1f1d6040c09e2d54ce7ba2d6ec93659c1bfdf365a2b072bf40d38da0a8e52d8",
+};
+function fixture(t: test.TestContext, profile: keyof typeof targets = "exact") {
+  const root = mkdtempSync(join(tmpdir(), "loopx-inbox-来源-"));
+  t.after(() => rmSync(root, {recursive: true, force: true}));
+  const base = join(root, ".local", "manager-context");
+  const folder = join(base, "entries", targets[profile]);
+  mkdirSync(folder, {recursive: true});
+  const scope = {profile_id: profile === "exact" ? "source_session_v1" : null,
+    caller_goal_ref: profile === "exact" ? ref : null, current_goal_ref: profile === "exact" ? ref : null};
+  const row = (index: number, extra: JsonObject = {}) => ({
+    schema_version: profile === "exact" ? "loopx_manager_context_entry_v2" : "loopx_manager_context_entry_v1",
+    goal_id: "research", agent_id: "worker", request_id: index.toString(16).padStart(64, "0"),
+    ...(profile === "exact" ? {goal_ref: ref} : {}), source_id: "source:" + index,
+    message: "Inspect the current artifact. " + "x".repeat(300) + "PRIVATE_LONG_SOURCE",
+    ...extra,
+  });
+  const publish = (index: number, extra: JsonObject = {}) => {
+    const request = row(index, extra);
+    writeFileSync(join(folder, request.request_id + ".json"), JSON.stringify(request));
+    return request;
+  };
+  const call = (operation: string, extra: JsonObject = {}) => readCollaborationInbox({runtime_root: root,
+    scope, goal_id: "research", agent_id: "worker", operation,
+    observed_at: "2026-10-19T00:00:00.000000+00:00", read_command: "loopx manager-inbox read", ...extra});
+  return {root, base, folder, row, publish, call, scope};
+}
+function sourceBytes(folder: string): [string, string][] {
+  return readdirSync(folder).sort().map(name => [name, readFileSync(join(folder, name)).toString("hex")]);
+}
+
+for (const profile of ["legacy", "exact"] as const) {
+  test(`${profile}: Turn observation retains full counts without writing or exposing long source`, async t => {
+    const f = fixture(t, profile);
+    for (let n = 1; n <= 128; n++) f.publish(n);
+    const before = sourceBytes(f.folder), observed = await f.call("turn_start");
+    const page = observed.result as JsonObject;
+    assert.equal((page.items as unknown[]).length, 20);
+    assert.equal(page.has_more, true);
+    const context = (observed.context as JsonObject).unfinished_delegations as JsonObject;
+    assert.equal(context.incoming_count, 128);
+    assert.equal(context.omitted_count, 108);
+    assert.equal(context.scan_complete, true);
+    assert.equal(JSON.stringify(context).includes("PRIVATE_LONG_SOURCE"), false);
+    assert.deepEqual(sourceBytes(f.folder), before);
+    assert.deepEqual(readdirSync(f.base).sort(), ["entries"]);
+  });
+
+  test(`${profile}: a completed receiver page records only its supplied requests and replays without adoption`, async t => {
+    const f = fixture(t, profile);
+    for (let n = 1; n <= 26; n++) f.publish(n);
+    const source = sourceBytes(f.folder), first = await f.call("receiver_read");
+    assert.equal(first.completed, true);
+    assert.equal((first.followthrough as JsonObject[])[0].step, "assess_request");
+    const reads = join(f.base, "reads"), before = sourceBytes(reads).filter(([name]) => name.endsWith(".json"));
+    assert.equal(before.length, 20);
+    for (const [name] of before) assert.equal(JSON.parse(readFileSync(join(reads, name), "utf8")).kind, "receiver_cli_read");
+    const repeated = await f.call("receiver_read");
+    assert.equal((repeated.result as JsonObject).has_more, true);
+    assert.deepEqual(sourceBytes(reads).filter(([name]) => name.endsWith(".json")), before);
+    assert.deepEqual(sourceBytes(f.folder), source);
+    assert.equal((repeated.context as JsonObject).read_complete, false);
+  });
+}
+
+test("a later damaged entry cannot block the complete current page", async t => {
+  const f = fixture(t);
+  for (let n = 1; n <= 21; n++) f.publish(n);
+  writeFileSync(join(f.folder, "f".repeat(64) + ".json"), "{broken");
+  const first = (await f.call("pending")).result as JsonObject;
+  assert.equal((first.items as unknown[]).length, 20);
+  await assert.rejects(f.call("pending", {cursor: first.next_cursor}), /invalid manager context record/);
+  const context = (await f.call("turn_context")).unfinished_delegations as JsonObject;
+  assert.equal(context.incoming_count, 21);
+  assert.equal(context.unavailable_count, 1);
+  assert.equal(context.scan_complete, false);
+});
+
+test("named reads never fall back to another page or record unrelated requests", async t => {
+  const f = fixture(t), original = f.publish(1), other = f.publish(2);
+  const result = await f.call("receiver_read", {request_id: original.request_id});
+  assert.equal((result.result as JsonObject).request_id, original.request_id);
+  assert.equal((result.result as JsonObject).peer_returns, undefined);
+  assert.equal(result.context, undefined);
+  assert.deepEqual(readdirSync(join(f.base, "reads")).filter(name => name.endsWith(".json")), [original.request_id + ".json"]);
+  await assert.rejects(f.call("receiver_read", {request_id: other.request_id, cursor: "invalid"}), /cannot use inbox cursors/);
+  await assert.rejects(f.call("receiver_read", {request_id: "f".repeat(64)}), /not found/);
+  assert.deepEqual(readdirSync(join(f.base, "reads")).filter(name => name.endsWith(".json")), [original.request_id + ".json"]);
+});
+
+test("a stale Goal instance cannot acknowledge an old request", async t => {
+  const f = fixture(t), request = f.publish(1), before = sourceBytes(f.folder);
+  await assert.rejects(f.call("record_read", {items: [request], deliveries: [],
+    scope: {...f.scope, current_goal_ref: {...ref, goal_instance_id: "ginst_" + "b".repeat(32)}}}), /lifecycle rejected/);
+  assert.deepEqual(sourceBytes(f.folder), before);
+  assert.deepEqual(readdirSync(f.base), ["entries"]);
+});
+
+for (const field of ["goal_id", "agent_id"]) {
+  test(`a supplied request cannot change its ${field} while reusing an existing id`, async t => {
+    const f = fixture(t, "legacy"), row = f.publish(1);
+    await assert.rejects(f.call("record_read", {items: [{...row, [field]: "other"}], deliveries: []}), /scope mismatch/);
+    assert.deepEqual(readdirSync(f.base), ["entries"]);
+  });
+}
+
+test("input-provider work defers receipts until finalization", async t => {
+  const f = fixture(t), request = f.publish(1, {brief: {inputs: [{ref: "artifact.txt"}]}});
+  const prepared = await f.call("receiver_read");
+  assert.equal(prepared.completed, undefined);
+  assert.deepEqual(readdirSync(f.base), ["entries"]);
+  const final = await f.call("record_read", {items: [request], deliveries: prepared.deliveries,
+    work_observations: [{kind: "pending", recorded_decision: null, linked_todos: [], evidence_refs: [], evidence_unavailable: false}],
+    include_context: true});
+  assert.equal((final.followthrough as JsonObject[])[0].step, "assess_request");
+  assert.equal(((final.context as JsonObject).unfinished_delegations as JsonObject).incoming_count, 1);
+  assert.equal(JSON.parse(readFileSync(join(f.base, "reads", request.request_id + ".json"), "utf8")).kind, "receiver_cli_read");
+});
+
+for (const bytes of [Buffer.from("\ufeff{}"), Buffer.from([0xff]), Buffer.alloc(128_001, 32)]) {
+  test(`unreadable record bytes remain visible as incomplete evidence (${bytes.length})`, async t => {
+    const f = fixture(t), request = f.publish(1);
+    writeFileSync(join(f.folder, request.request_id + ".json"), bytes);
+    await assert.rejects(f.call("receiver_read"), /manager context record/);
+    const context = (await f.call("turn_context")).unfinished_delegations as JsonObject;
+    assert.equal(context.unavailable_count, 1);
+    assert.equal(context.scan_complete, false);
+    assert.deepEqual(readdirSync(f.base), ["entries"]);
+  });
+}
+
+test("malformed receipts retain the request and recovery advice", async t => {
+  const f = fixture(t), request = f.publish(1);
+  mkdirSync(join(f.base, "decisions"));
+  writeFileSync(join(f.base, "decisions", request.request_id + ".json"), "{broken");
+  const response = await f.call("receiver_read"), item = ((response.result as JsonObject).items as JsonObject[])[0];
+  assert.equal(item.inbox_state, "receipt_unavailable");
+  assert.equal(item.recorded_decision, null);
+  assert.equal((response.followthrough as JsonObject[])[0].step, "recover_evidence");
+  assert.equal(readFileSync(join(f.base, "decisions", request.request_id + ".json"), "utf8"), "{broken");
+});
+
+test("unreadable source content is incomplete evidence rather than coerced prose", async t => {
+  const f = fixture(t);
+  f.publish(1, {message: {text: "unreadable"}});
+  await assert.rejects(f.call("receiver_read"), /invalid Inbox source content/);
+  const context = (await f.call("turn_context")).unfinished_delegations as JsonObject;
+  assert.equal(context.scan_complete, false);
+  assert.equal(context.unavailable_count, 1);
+  assert.deepEqual(context.items, []);
+  assert.deepEqual(readdirSync(f.base), ["entries"]);
+});
+
+for (const status of [null, ["queued"]]) {
+test(`malformed delivery status retains its recovery distinction (${JSON.stringify(status)})`, async t => {
+  const f = fixture(t), request = f.publish(1), folder = join(f.base, "replies", request.request_id);
+  mkdirSync(folder, {recursive: true});
+  mkdirSync(join(f.base, "decisions"));
+  const identity = Object.fromEntries(["request_id", "goal_id", "agent_id", "source_id", "goal_ref"].map(key => [key, request[key as keyof typeof request]]));
+  writeFileSync(join(f.base, "decisions", request.request_id + ".json"), JSON.stringify({...identity, decision: "adopt"}));
+  writeFileSync(join(folder, "conclusion.json"), JSON.stringify({...identity, phase: "conclusion", decision: "adopt", text: "Findings"}));
+  writeFileSync(join(folder, "conclusion.delivery.json"), JSON.stringify({status}));
+  const context = (await f.call("turn_context")).unfinished_delegations as JsonObject;
+  const agenda = context.items as JsonObject[];
+  if (status === null) {
+    assert.equal((agenda[0].returns as JsonObject[])[0].status, "explicit_unverified");
+    assert.equal(agenda[0].step, "await_return_delivery");
+  } else assert.equal(agenda[0].step, "recover_evidence");
+});
+}
+
+test("a legal combined Inbox and peer-return page retains content beyond the inline wire budget", async t => {
+  const f = fixture(t), message = "x".repeat(50_000), text = "😀".repeat(20_000);
+  const operations = join(f.base, "peer-operations", targets.exact);
+  const receiver = join(f.base, "entries", "5ab0deb53efce4b1d0cf830afb67ce662e0e9aff3127e2424a4ce768212b5586");
+  for (const folder of [operations, receiver, join(f.base, "decisions"), join(f.base, "roundtrips")]) mkdirSync(folder, {recursive: true});
+  for (let n = 1; n <= 20; n++) {
+    f.publish(n, {message});
+    const row = f.row(100 + n, {agent_id: "reviewer", source_kind: "peer", source_agent_id: "worker",
+      brief: {schema_version: "collaboration_brief_v0", purpose: "Independent review", context: "Inspect the artifact",
+        constraints: [], inputs: [], acceptance: ["Return findings"], return_requirement: "Reply to the requester"}});
+    const identity = Object.fromEntries(["request_id", "goal_id", "agent_id", "source_id", "goal_ref"].map(key => [key, row[key as keyof typeof row]]));
+    writeFileSync(join(receiver, row.request_id + ".json"), JSON.stringify(row));
+    writeFileSync(join(operations, row.request_id + ".json"), JSON.stringify(row));
+    writeFileSync(join(f.base, "roundtrips", row.request_id + ".json"), JSON.stringify(identity));
+    writeFileSync(join(f.base, "decisions", row.request_id + ".json"), JSON.stringify({...identity, decision: "adopt"}));
+    const replies = join(f.base, "replies", row.request_id);
+    mkdirSync(replies, {recursive: true});
+    writeFileSync(join(replies, "conclusion.json"), JSON.stringify({...identity, phase: "conclusion",
+      decision: "adopt", text, created_at: "2026-10-19T00:00:00Z"}));
+  }
+  const observed = await f.call("receiver_read"), page = observed.result as JsonObject;
+  assert.equal(observed.completed, true);
+  assert.equal(Buffer.byteLength(JSON.stringify(observed)) > 2 * 1024 * 1024, true);
+  assert.equal((page.items as JsonObject[]).length, 20);
+  assert.equal(((page.peer_returns as JsonObject).items as JsonObject[]).length, 20);
+  for (const item of page.items as JsonObject[]) assert.equal(item.message, message);
+  for (const item of (page.peer_returns as JsonObject).items as JsonObject[]) assert.equal(item.text, text);
+});
+
+test("a large private agenda lets another runtime request finish before the scan completes", async t => {
+  const large = fixture(t), other = fixture(t);
+  for (let n = 1; n <= 256; n++) large.publish(n);
+  let finished = false;
+  const scanning = large.call("turn_context").then(context => {finished = true; return context;});
+  const independent = await other.call("turn_context");
+  assert.equal((independent.unfinished_delegations as JsonObject).incoming_count, 0);
+  assert.equal(finished, false);
+  assert.equal(((await scanning).unfinished_delegations as JsonObject).incoming_count, 256);
+});
diff --git a/tests/test_collaboration_goal_instance.py b/tests/test_collaboration_goal_instance.py
index fa1c7e8210..8280c0b00c 100644
--- a/tests/test_collaboration_goal_instance.py
+++ b/tests/test_collaboration_goal_instance.py
@@ -374,14 +374,14 @@ def test_inbox_read_rejects_recreation_before_response_commit(
     replacement = tmp_path / "replacement"
     replacement.mkdir()
     (replacement / input_path.name).write_text("instance B", encoding="utf-8")
-    render_followthrough = links.receiver_followthrough
+    read_work = links.receiver_work_observations
 
     def recreate_before_input_read(root, registry_path, items):
-        rendered = render_followthrough(root, registry_path, items)
+        rendered = read_work(root, registry_path, items)
         _recreate(registry, repo=replacement)
         return rendered
 
-    monkeypatch.setattr(links, "receiver_followthrough", recreate_before_input_read)
+    monkeypatch.setattr(links, "receiver_work_observations", recreate_before_input_read)
 
     with pytest.raises(ValueError, match="historical_mutation_forbidden"):
         read_inbox(
@@ -448,13 +448,22 @@ def test_inbox_read_does_not_commit_peer_return_before_response_succeeds(
         caller_goal_ref=goal_ref,
     )
 
+    # A real incoming input requires Host enrichment before the peer read commits.
+    input_path = tmp_path / "input.txt"
+    input_path.write_text("Current instance input", encoding="utf-8")
+    incoming_brief = {
+        **_brief("Inspect the current input before acknowledging peer delivery"),
+        "inputs": [{"ref": input_path.name, "description": "Current source input"}],
+    }
+    request(tmp_path, registry, "delivery", "reviewer", "builder", "read-input", incoming_brief)
+
     def interrupt_response(*_args):
         if interruption == "goal_recreation":
             _recreate(registry)
             return None
         raise RuntimeError("response enrichment failed")
 
-    monkeypatch.setattr(links, "receiver_followthrough", interrupt_response)
+    monkeypatch.setattr(links, "receiver_work_observations", interrupt_response)
 
     expected_error = ValueError if interruption == "goal_recreation" else RuntimeError
     expected_message = (