One declaration per bus read: UI, MCP and owner-socket reads derive one operation

src/agent_comms/acp_extension.py

@@ -12,7 +12,7 @@
 
 from abc import abstractmethod
 from dataclasses import dataclass, field, replace
-from typing import TYPE_CHECKING, Annotated, Callable, ClassVar
+from typing import TYPE_CHECKING, Annotated, Callable, ClassVar, Self
 
 
 from .acp_failure import (
@@ -52,8 +52,10 @@
 from .reported_turn import ReportedTurn, ReportedTurnState
 from .routing import MessageRoute, NoMessageRoute
 from .thread_identity import ClaimScoped, ThreadIncarnation
-from .tool_result import DeclaredReply
-from .transcripts import NativeCommittedPrefix, TranscriptCursor, TranscriptPage, TranscriptReadIdentity
+from .tool_result import ToolResult
+from .transcripts import (
+    NativeCommittedPrefix, TranscriptCursor, TranscriptPage, TranscriptRead, TranscriptReadIdentity,
+)
 from .turn_context import ContextManifest
 from .turn_lease import TurnState
 from .working_memory_labels import ModelLabel
@@ -138,7 +140,7 @@ def of(cls, *updates: AgentCommsUpdate) -> UpdateMetadata:
 
 
 @dataclass(frozen=True)
-class GoalSnapshotReply(DeclaredReply, GoalSnapshot):
+class GoalSnapshotReply(ToolResult, GoalSnapshot):
     """A goal request's reply: the owner's goal snapshot and its turn state."""
 
     execution: Annotated[GoalExecution, WireName("goalExecution")]
@@ -698,13 +700,13 @@ class TranscriptDraftUpdate(AgentCommsUpdate):
 
 
 @dataclass(frozen=True)
-class TranscriptSnapshotUpdate(AgentCommsUpdate):
+class TranscriptSnapshotUpdate(AgentCommsUpdate, ToolResult):
     page: TranscriptPage
     identity: TranscriptReadIdentity
     draft: DraftRow | None = None
 
     @classmethod
-    def from_read(cls, read):
+    def from_read(cls, read: TranscriptRead) -> Self:
         page, draft = read.read_snapshot()
         return cls(page, read.identity, draft.external_row())
 

src/agent_comms/bus_read.py

@@ -0,0 +1,119 @@
+"""One declaration per bus read: the operation owns its arguments and its read once.
+
+A ``BusRead`` member is the operation. Each boundary that serves it keeps only what is
+its own: the UI its root, request id and refusals (``ReadRequest``), MCP its caller,
+exposure and reply (``ToolRequest``). Neither boundary restates the read, and the
+reader's authority reaches the operation as a ``BusReader`` the boundary supplies,
+never as an argument the caller may choose.
+"""
+from __future__ import annotations
+
+import asyncio
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, NoReturn
+
+from .launch_runtime import CallingThreadUnavailable
+from .owner_connection import (
+    OwnerConnection, OwnerReading, OwnerReplyConnection, ResultRuntimeRequest, RuntimeRequestContext,
+)
+from .presentation import AllOutcomes
+from .tool_declaration import ToolDeclaration
+from .tool_result import ToolResult
+
+if TYPE_CHECKING:
+    from .comms import Comms
+    from .presentation import OutcomeViewer
+
+
+class BusReader(ABC):
+    """Whose authority a read runs under; each boundary supplies its own reader."""
+
+    @abstractmethod
+    def outcome_viewer(self, comms: Comms) -> OutcomeViewer:
+        """Whose recipient outcomes this reader may read."""
+
+    def subject(self, named: str) -> str:
+        return named or self.own_thread()
+
+    def own_thread(self) -> str:
+        """The thread a read targets when its call names none; a reader with no thread has none."""
+        return self.refuse_thread()
+
+    def refuse_thread(self) -> NoReturn:
+        raise CallingThreadUnavailable("This tool requires a thread caller.")
+
+
+class OperatorOutcomes:
+    """A reader speaking for the owner reads every recipient outcome, as the owner's UI does."""
+
+    __slots__ = ()
+
+    def outcome_viewer(self, comms: Comms) -> OutcomeViewer:
+        return AllOutcomes()
+
+
+class OperatorReader(OperatorOutcomes, BusReader):
+    """The owner's UI: it observes the whole bus as the operator."""
+
+    __slots__ = ()
+
+
+class BusRead[Result](ToolDeclaration, ABC):
+    """A read of Core's stores, declared once with its arguments; boundaries derive it."""
+
+    def apply(self, comms: Comms, caller: BusReader) -> Result:
+        """The MCP boundary runs the read under its calling thread's authority."""
+        return self.read_as(comms, caller)
+
+    @abstractmethod
+    def read_as(self, comms: Comms, reader: BusReader) -> Result:
+        """The read under the reader's authority."""
+
+
+class UiRead[Result](BusRead[Result], ABC):
+    """The UI serves this read as the operator; its envelope (root, request id, refusal)
+    stays on ``ReadRequest``."""
+
+    def read(self, comms: Comms) -> Result:
+        return self.read_as(comms, OperatorReader())
+
+
+class OwnerLiveRead[Result: ToolResult = ToolResult](BusRead[Result], ABC):
+    """A read only the selected thread's live native owner can answer, from its own queue.
+
+    MCP and the UI both reach the owner through ``OwnerReadRuntimeRequest``; neither answers
+    from its own process, and owner loss reaches the caller as the original refusal. The
+    owner runs ``owned`` against its own registration and session, never against the public
+    thread selector.
+    """
+
+    if TYPE_CHECKING:
+        # The composed ThreadReadTool declares the field; an annotation here would shadow it.
+        thread: str
+
+    def read_as(self, comms: Comms, reader: BusReader) -> Result:
+        owner = comms.registry.require(reader.subject(self.thread)).incarnation
+        request = OwnerReadRuntimeRequest(thread=owner.name, operation=self)
+        return self.decode_reply(asyncio.run(OwnerReplyConnection.answer(comms, owner, request)))
+
+    @abstractmethod
+    async def owned(self, ctx: RuntimeRequestContext) -> Result:
+        """The answer computed inside the owner, under its own context."""
+
+    @abstractmethod
+    def decode_reply(self, raw: object) -> Result: ...
+
+
+@dataclass(frozen=True, kw_only=True)
+class OwnerReadRuntimeRequest(OwnerReading[object], ResultRuntimeRequest):
+    """The one runtime envelope for every owner-live read: sent once to the held owner, and
+    answered there by the operation under the owner's context."""
+
+    operation: OwnerLiveRead
+
+    async def read(self, owner: OwnerConnection) -> object:
+        return await owner.request(self.declared_name, operation=self.operation)
+
+    async def result(self, ctx: RuntimeRequestContext) -> ToolResult:
+        return await self.operation.owned(ctx)

src/agent_comms/commander_hud.py

@@ -147,7 +147,7 @@ class CommanderOrderDetail:
 
 
 @dataclass(frozen=True)
-class CommanderUnitDetail:
+class CommanderUnitDetail(ToolResult):
     """A lazy disclosure; activity uses the existing bounded turn wire projection."""
 
     unit: Annotated[CommanderUnit, DisclosureIdentity()]

src/agent_comms/connector.py

@@ -54,6 +54,7 @@
 from .declared_family import DeclaredFamily, UnknownFamilyMember
 from .errors import RelationViolationError, ToolRefusedError, UnregisteredThreadError
 from .coordination_errors import CoordinationError, Refused
+from .work_branches import WorkBranchError
 from .field_codec import (
     Argument, FieldCodec, GitCommitText, MalformedRecord, PathText, WireName, WireOmitDefault,
     RecordMember, boundary_decoder, projected,
@@ -459,7 +460,10 @@ async def result(self, connector: Connector, caller: ConnectorCaller) -> ToolsCa
         try:
             reply = await connector.call(tool, self.params.arguments, caller)
             return ToolsCallResult((TextContent(json.dumps(reply)),), False, reply)
-        except (ToolRefusedError, RelationViolationError, CoordinationError) as error:
+        except (
+            ToolRefusedError, RelationViolationError, CoordinationError, WorkBranchError,
+            MalformedRecord, Refused, ConnectionError, TimeoutError,
+        ) as error:
             logging.exception("Connector tool %s failed", self.params.name)
             return ToolsCallResult.failed(error)
 

src/agent_comms/connector_caller.py

@@ -13,7 +13,8 @@
 from .head_review import HeadReviewOrder, PullRequestHead
 from .thread_identity import ThreadIncarnation
 from .todos import DirectiveText
-from .tool_caller import CommanderForkCaller, NoThreadCaller, OperatorOutcomes, ThreadCaller, ToolCaller
+from .bus_read import OperatorOutcomes
+from .tool_caller import CommanderForkCaller, NoThreadCaller, ThreadCaller, ToolCaller
 from .tool_request import ToolArgumentsRefusedError, ToolRequest
 
 

src/agent_comms/goal_history.py

@@ -29,7 +29,7 @@
 from .locked_store import DurableFile
 from .private_registry_guard import PrivateRegistryGuard
 from .store_files import AtomicFile
-from .tool_result import DeclaredReply
+from .tool_result import ToolResult
 from .reported_turn import ReportedTurnState
 from .typed_table import (
     Column,
@@ -170,7 +170,7 @@ def reconcile(
 
 
 @dataclass(frozen=True)
-class RecordedGoalHistory(DeclaredReply):
+class RecordedGoalHistory(ToolResult):
     """A thread's goal revisions beside the attempt ledger's failures, which only the ledger writes."""
 
     history: tuple[RecordedGoalRevision, ...]

src/agent_comms/goal_management.py

@@ -14,7 +14,7 @@
 from dataclasses import replace
 from functools import cached_property
 from pathlib import Path
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING
 
 from agent_comms.field_codec import GoalId
 from .agent_activity import AgentActivity
@@ -72,7 +72,6 @@
 from .todos import SendBackVerdict, StrikeReason, Todo, JudgedCall, TodoBatch, TodoCommand, TodoStore
 
 _LOG = logging.getLogger(__name__)
-HistoryReply = TypeVar("HistoryReply", bound=RecordedGoalHistory)
 
 
 class Goals(GoalWaitRecovery):
@@ -294,12 +293,10 @@ def goal_history(
         """Read durable revisions for one thread incarnation within ``scope``."""
         return self.registry.goal_history(name, scope)
 
-    def recorded_history(
-        self, reply: type[HistoryReply], name: str, scope: GoalScope
-    ) -> HistoryReply:
+    def recorded_history(self, name: str, scope: GoalScope) -> RecordedGoalHistory:
         """Revisions from goal history, execution failures from the attempt ledger."""
         owner = self.registry.require(name).incarnation.wire_birth
-        return reply(
+        return RecordedGoalHistory(
             tuple(entry.recorded() for entry in self.goal_history(name, scope)),
             self.open_goal_store().failures(owner, scope),
         )

src/agent_comms/historical_views.py

@@ -7,7 +7,7 @@
 import tempfile
 from abc import ABC, abstractmethod
 from collections import deque
-from collections.abc import Callable, Iterator
+from collections.abc import Callable, Iterator, Sequence
 from contextlib import contextmanager
 from dataclasses import dataclass, replace
 from functools import partial
@@ -31,7 +31,9 @@
     MessagePageRequest,
     PageBounds,
 )
+from .message_reference import MessageReference
 from .messages import CarriesEnvelope, Message
+from .notification_evidence import NotificationWindow
 from .page_traversal import HistoryCursor, OlderTraversal
 from .presentation import MessageNotification, OutcomeViewer
 from .private_path import FileObservation, FileRevision, ObservedPath, PrivateDirectoryRole
@@ -352,8 +354,9 @@ def require(cls, sources: tuple[HistorySource, ...], key: str) -> HistorySource:
         source.validate()
         return source
 
-    def notification_references(self, sources: tuple[HistorySource, ...], references,
-                                viewer: OutcomeViewer):
+    def notification_references(self, sources: tuple[HistorySource, ...],
+                                references: Sequence[MessageReference],
+                                viewer: OutcomeViewer) -> NotificationWindow:
         """Read this namespace within its archive's current catalogue read, as this viewer
         may see it."""