refactor(inbox): consolidate native read transport in the IO boundary

loopx/capabilities/manager_context/__init__.py

@@ -238,7 +238,7 @@ def turn_start_hook(
     def produce():
         nonlocal snapshot
         try:
-            from ...control_plane.collaboration.inbox_read import read_inbox_transaction
+            from ...control_plane.collaboration.inbox import read_inbox_transaction
             from ...control_plane.collaboration.inbox import _pending_result
 
             with collaboration_goal_scope(

loopx/control_plane/collaboration/inbox.py

@@ -147,13 +147,48 @@ def plan_request_publication(request, *, entry, operation=None):
 
 
 
+def read_inbox_transaction(root: Path, request: dict, *, scope=None) -> dict:
+    """Adapt the existing Inbox IO boundary to its native transaction owner."""
+    from ..effect_runtime import (
+        EffectRuntimeIOError,
+        EffectRuntimeRejected,
+        effect_runtime_result,
+    )
+
+    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:
+        # Preserve the IO classification and native retryable/permanent code.
+        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
+
+
 def pending(
     runtime_root: Path, goal_id: str, agent_id: str, *, cursor: str | None = None,
     operation_cursor: str | None = None, request_id: str | None = None,
     scope: CollaborationGoalScope | None = None,
 ) -> dict:
-    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,
@@ -184,8 +219,6 @@ def _pending_result(root, snapshot, scope, operation_cursor):
 
 
 def _pending_receipt_states(runtime_root, rows):
-    from .inbox_read import read_inbox_transaction
-
     return read_inbox_transaction(runtime_root, {
         "operation": "receipt_states", "rows": rows,
     })["items"]
@@ -313,8 +346,6 @@ def record_read(root: Path, items: list[dict], *, scope: CollaborationGoalScope
     """Record supplied source without certifying comprehension or adoption."""
     if not items:
         return
-    from .inbox_read import read_inbox_transaction
-
     result = read_inbox_transaction(root, {
         "operation": "record_read", "goal_id": items[0]["goal_id"],
         "agent_id": items[0]["agent_id"], "items": items, "deliveries": [],

loopx/control_plane/collaboration/peers.py

@@ -241,7 +241,7 @@ def request(
 
 
 def returns(root, goal_id, agent_id, *, mark_read=False, scope=None):
-    from .inbox_read import read_inbox_transaction
+    from .inbox import read_inbox_transaction
 
     return read_inbox_transaction(root, {
         "operation": "returns", "goal_id": goal_id, "agent_id": agent_id, "mark_read": mark_read,
@@ -486,7 +486,7 @@ def read_inbox(
     caller_goal_ref=None,
 ):
     from .inbox import _pending_result
-    from .inbox_read import read_inbox_transaction
+    from .inbox import read_inbox_transaction
     from .turn_context_observation import delegation_read_command
     from .links import receiver_work_observations
 

loopx/control_plane/collaboration/turn_context_observation.py

@@ -3,7 +3,7 @@
 from pathlib import Path
 
 from .goal_instance_scope import collaboration_goal_scope
-from .inbox_read import read_inbox_transaction
+from .inbox import read_inbox_transaction
 
 
 def delegation_read_command(root, registry, goal_id, agent_id):

tests/control_plane/test_inbox_read_recovery.py

@@ -108,7 +108,7 @@ def test_operation_source_appearing_after_committed_read_is_observed_on_next_rea
     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 import inbox
     from loopx.control_plane.collaboration.peers import read_inbox
 
     registry = tmp_path / "registry.json"
@@ -120,7 +120,7 @@ def test_operation_source_appearing_after_committed_read_is_observed_on_next_rea
     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
+    transport = inbox.read_inbox_transaction
 
     def source_appears(root, query, **kwargs):
         response = transport(root, query, **kwargs)
@@ -130,11 +130,11 @@ def source_appears(root, query, **kwargs):
             path.write_text("{broken", encoding="utf-8")
         return response
 
-    monkeypatch.setattr(inbox_read, "read_inbox_transaction", source_appears)
+    monkeypatch.setattr(inbox, "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)
+    monkeypatch.setattr(inbox, "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