From 2fc67db407727b5572167215afe80a3b27dd76ca Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 20 Oct 2026 00:13:16 +0800 Subject: [PATCH] refactor(inbox): consolidate native read transport in the IO boundary Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/manager_context/__init__.py | 2 +- loopx/control_plane/collaboration/inbox.py | 43 ++++++++++++++++--- .../control_plane/collaboration/inbox_read.py | 42 ------------------ loopx/control_plane/collaboration/peers.py | 4 +- .../collaboration/turn_context_observation.py | 2 +- .../control_plane/test_inbox_read_recovery.py | 8 ++-- 6 files changed, 45 insertions(+), 56 deletions(-) delete mode 100644 loopx/control_plane/collaboration/inbox_read.py diff --git a/loopx/capabilities/manager_context/__init__.py b/loopx/capabilities/manager_context/__init__.py index 218d437dd3..1305d76597 100644 --- a/loopx/capabilities/manager_context/__init__.py +++ b/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( diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py index 3aa08c7368..b4b11f5932 100644 --- a/loopx/control_plane/collaboration/inbox.py +++ b/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": [], diff --git a/loopx/control_plane/collaboration/inbox_read.py b/loopx/control_plane/collaboration/inbox_read.py deleted file mode 100644 index 0a712b58fd..0000000000 --- a/loopx/control_plane/collaboration/inbox_read.py +++ /dev/null @@ -1,42 +0,0 @@ -"""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/peers.py b/loopx/control_plane/collaboration/peers.py index 14e3ce2742..d79725156a 100644 --- a/loopx/control_plane/collaboration/peers.py +++ b/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 diff --git a/loopx/control_plane/collaboration/turn_context_observation.py b/loopx/control_plane/collaboration/turn_context_observation.py index 2aec38e178..acc6d1ffa3 100644 --- a/loopx/control_plane/collaboration/turn_context_observation.py +++ b/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): diff --git a/tests/control_plane/test_inbox_read_recovery.py b/tests/control_plane/test_inbox_read_recovery.py index ee78226ac4..68de34865a 100644 --- a/tests/control_plane/test_inbox_read_recovery.py +++ b/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