Input delivery reads through Core's one owner-live operation (agent-comms#3538)

pyproject.toml

@@ -84,3 +84,4 @@ markers = ["refactor_guard: permanent architecture deletion guards"]
 
 [tool.mypy]
 follow_untyped_imports = true
+mypy_path = ["src", "tools"]

src/toad/acp/agent.py

@@ -15,7 +15,8 @@
 from agent_comms.goal_presentation import GoalSnapshot, IdleGoalExecution
 from agent_comms.goals import Goal, GoalDeclaration, NoGoal
 from agent_comms.goal_history import RecordedGoalHistory
-from agent_comms.ui_model.observation import ReadOwnerGoal
+from agent_comms.order_tools import OwnerGoalRead
+from agent_comms.ui_model.observation import ReadOwnerOperation
 
 from toad import paths
 from toad.acp.agent_configuration import (
@@ -238,7 +239,7 @@ async def get_goal_snapshot(self) -> GoalSnapshot:
         authority = ClientSessionRequest(self, self.session_id)
         consumer = OwnerSnapshotConsumer(self, self.session_id, turn_token=self.presentation.turns.sequence)
         snapshot = await self.controller.transcripts.read(
-            ReadOwnerGoal(coordination.wire_root, coordination.thread))
+            ReadOwnerOperation(coordination.wire_root, coordination.thread, operation=OwnerGoalRead()))
         self.controller.require_owner(coordination, authority)
         for fact in snapshot.meta.batch.updates:
             consumer.dispatch_sync(fact)

src/toad/acp/agent_controller.py

@@ -1,3 +1,3 @@
 """ACP operations remain owned by the controller after their view retires."""
 
 from __future__ import annotations
@@ -22,8 +22,9 @@
 from agent_comms.acp_failure import ACPFailure, RequestFailure
 from agent_comms.coordination_errors import Refused
 from agent_comms.transcripts import TranscriptReadIdentity
-from agent_comms.ui_model.delivery import DeliveryView
-from agent_comms.ui_model.observation import ReadDeliveryOverview
+from agent_comms.order_tools import InputDeliveryRead
+from agent_comms.ui_model.delivery import DeliveryView, ReadDeliveryView
+from agent_comms.ui_model.observation import ReadOwnerOperation
 
 from toad.acp.client_session import ClientSessionRequest
 from toad.acp.status import StopReason
@@ -430,11 +431,11 @@
         if coordination is None:
             return DeliveryView()
         authority = ClientSessionRequest(self.agent, self.agent.session_id)
-        delivery = await self.transcripts.read(
-            ReadDeliveryOverview(coordination.wire_root, coordination.thread, include_history=include_history)
-        )
+        overview = await self.transcripts.read(ReadOwnerOperation(
+            coordination.wire_root, coordination.thread, InputDeliveryRead(include_history=include_history),
+        ))
         self.require_owner(coordination, authority)
-        return delivery
+        return ReadDeliveryView(overview)
 
     async def dismiss_historical_inputs(self):
         return await self.request_owner("dismiss_historical_inputs")

src/toad/owner_preparation.py

@@ -68,5 +68,6 @@ class OwnerRequestContext:
 
 
 def read_thread_presentation(comms: Comms, name: str) -> ThreadPresentation:
-    """Read the existing core presentation on a worker, without UI dependencies."""
-    return comms.views.thread_presentations((name,)).read(name)
+    """Read the existing core presentation on a worker, without UI dependencies; only its
+    published projection leaves the worker, never the acquired delivery evidence."""
+    return comms.views.thread_presentations((name,)).read(name).published()

src/toad/screens/commander.py

@@ -11,7 +11,8 @@
 from agent_comms.coordination_errors import DeferredRead
 from agent_comms.errors import RelationViolationError, RouteChangedError, ToolRefusedError
 from agent_comms.thread_identity import ThreadIncarnation
-from agent_comms.ui_model.observation import ReadCommanderUnit, UnreadableHistory
+from agent_comms.order_tools import CommanderUnitRead
+from agent_comms.ui_model.observation import ReadOperation, UnreadableHistory
 from agent_comms.work_branches import WorkBranchError
 from metaclass_registry.mro_dispatch import MroDispatch, handles
 from textual import on, work
@@ -118,10 +119,10 @@ async def read_detail(self) -> None:
         access = self.app.coordination_access
         try:
             observed = await access.preparation.run_thread(access.observation.observed)
-            request = ReadCommanderUnit(
-                str(observed.service.root), self.commander, card.unit,
-                tuple(dict.fromkeys(order.order for order in card.orders)),
-            )
+            request = ReadOperation(str(observed.service.root), CommanderUnitRead(
+                commander=self.commander, unit=card.unit,
+                orders=tuple(dict.fromkeys(order.order for order in card.orders)),
+            ))
             detail = await observed.process.read(request)
         except (
             DeferredRead, RelationViolationError, RouteChangedError,

src/toad/screens/historical_sessions.py

@@ -9,7 +9,9 @@
 from agent_comms.comms import Comms
 from agent_comms.coordination_errors import StaleRevision
 from metaclass_registry.mro_dispatch import handles
-from agent_comms.ui_model.observation import ReadTranscript, ReadRecordedNotifications
+from agent_comms.notification_reads import RecordedNotificationRead
+from agent_comms.order_tools import TranscriptPageRead
+from agent_comms.ui_model.observation import ReadTranscript
 from toad.core_event_carrier import CoreEventMessage, CoreEventReceiver
 from agent_comms.historical_views import HistoricalThread
 from textual import on, work
@@ -29,7 +31,7 @@ class RecordedContents(Contents):
     """The notification subscriber reads the same archive as its history."""
 
     def __init__(self, source: str, **kwargs):
-        self.notification_request = partial(ReadRecordedNotifications, source=source)
+        self.notification_operation = partial(RecordedNotificationRead, source=source)
         super().__init__(**kwargs)
 
 
@@ -121,8 +123,8 @@ async def selected(self, event: Select.Changed) -> None:
         access, root = self.app.coordination_access, str(self.comms.root)
 
         async def load(**kwargs):
-            snapshot = await access.transcripts.read(ReadTranscript(
-                root, item.thread.name, historical_source=item.source.key, **kwargs))
+            snapshot = await access.transcripts.read(ReadTranscript(root, TranscriptPageRead(
+                thread=item.thread.name, historical_source=item.source.key, **kwargs)))
             return snapshot.page
 
         try:

src/toad/session_scope.py

@@ -15,6 +15,7 @@
 from agent_comms.errors import RelationViolationError
 from agent_comms.route_selection import RouteSelection
 from agent_comms.thread_identity import ThreadIncarnation
+from agent_comms.order_tools import TranscriptPageRead
 from agent_comms.ui_model.observation import (
     ReadTranscript,
     ReadTranscriptUpdate,
@@ -158,22 +159,23 @@ def snapshot_request(
         through: TranscriptCursor | None = None,
         historical_source: str | None = None,
     ) -> ReadTranscript:
-        return ReadTranscript(
-            self.root,
-            self.name,
+        return ReadTranscript(self.root, TranscriptPageRead(
+            thread=self.name,
             before=before,
             after=after,
             through=through,
             historical_source=historical_source,
-        )
+        ))
 
     def update_request(
         self,
         previous: TranscriptPublicationState,
         *,
         after: TranscriptCursor | None = None,
     ) -> ReadTranscriptUpdate:
-        return ReadTranscriptUpdate(self.root, self.name, previous=previous, after=after)
+        return ReadTranscriptUpdate(
+            self.root, TranscriptPageRead(thread=self.name, after=after), previous=previous
+        )
 
     def mailbox_target(self) -> ThreadTarget:
         return ThreadTarget(self.name)

src/toad/widgets/message_notifications.py

@@ -11,6 +11,7 @@
 from textual.content import Content
 from textual.widgets import Collapsible
 
+from agent_comms.notification_reads import LiveNotificationRead
 from agent_comms.ui_model.observation import ReadNotifications
 
 from toad.notification_status import (
@@ -110,7 +111,7 @@ class HistoryNotifications:
     """Visible history rows subscribe to Core; paging never decides when outcomes changed."""
 
     NOTIFICATION_GROUP = "message-notifications"
-    notification_request = ReadNotifications
+    notification_operation = LiveNotificationRead
 
     # Core feedback for every visible bus message, native or wire.
 
@@ -126,7 +127,7 @@ def notification_read(self, references) -> ReadNotifications:
         root = ProjectPathOwner.containing(self).coordination_root
         if root is None:
             raise ValueError("Notification observation requires a bound thread root")
-        return self.notification_request(root, references)
+        return ReadNotifications(root, self.notification_operation(references))
 
     def subscribe_notifications(self, references) -> None:
         access = self.app.coordination_access

tests/app/test_recorded_transcript_reader.py

@@ -2,15 +2,17 @@
 
 from agent_comms.repo import Checkout
 import json
-from dataclasses import replace
+from dataclasses import fields, replace
 
 import pytest
 
 from agent_comms.acp_extension import TranscriptSnapshotUpdate, decode_updates, encode_updates
 from agent_comms.coordination_errors import HistoricalSourceChanged, StaleRevision
+from agent_comms.order_tools import TranscriptPageRead
 from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.transcripts import RecordedTranscriptReadIdentity
 from agent_comms.threads import Thread
+from agent_comms.ui_model.observation import NotPublished
 from support.roots import installed_wire
 from support.journals import assistant_entry, write_journal
 from toad.acp.comms_updates import OwnerSnapshotConsumer
@@ -62,3 +64,19 @@ def test_recorded_presentation_does_not_acquire_live_owner(recorded):
     consumer.thread_presentation(replace(presentation, read_identity=identity))
     assert TranscriptSnapshotUpdate.capture(live.transcripts, "reader",
                                             historical_source=source.key).identity == identity
+
+
+def test_known_transcript_requests_carry_the_page_operation_and_keep_root_at_the_envelope(recorded):
+    live, source, _native = recorded
+    scope = KnownTranscript(str(live.root), "reader")
+    cursor = live.transcripts.capture_page_read("reader", historical_source=source.key).read().after
+    snapshot = scope.snapshot_request(before=cursor, historical_source=source.key)
+    assert snapshot.root == str(live.root)
+    assert snapshot.operation == TranscriptPageRead(
+        thread="reader", before=cursor, historical_source=source.key)
+    previous = NotPublished()
+    update = scope.update_request(previous, after=cursor)
+    assert update.root == str(live.root) and update.previous is previous
+    assert update.operation == TranscriptPageRead(thread="reader", after=cursor)
+    # The shared operation never carries the UI's root, request or publication witness.
+    assert not {"root", "request_id", "previous", "identity"} & {f.name for f in fields(TranscriptPageRead)}

tests/native/test_native_goal_owner.py

@@ -3,6 +3,8 @@
 import pytest
 from agent_comms.goal_actions import SetGoalAction
 from agent_comms.goal_history import RecordedGoalHistory
+from agent_comms.input_attempt import OwnerQueueInputDeliveryScope
+from agent_comms.ui_model.delivery import ReadDeliveryView
 from support.native import native_session
 
 
@@ -37,3 +39,18 @@ async def test_editing_a_goal_keeps_its_identity_and_refuses_a_stale_edit(worksp
                        for entry in history.history)
         snapshot = await agent.get_goal_snapshot()
         assert snapshot.goal == changed and snapshot.execution.goal_id == changed.id
+
+
+async def test_input_delivery_is_read_from_the_live_native_owner_queue(workspace, native_package):
+    """The owner_queue scope can only come from the live owner; the no-coordination
+    DeliveryView and the unobserved projection never carry it."""
+    async with native_session(workspace) as native:
+        await native.ui.until_agent_ready()
+        controller = native.agent.controller
+        for delivery in (await controller.input_delivery(),
+                         await controller.input_delivery(include_history=True)):
+            assert type(delivery) is ReadDeliveryView
+            overview = delivery.overview
+            assert overview.current_scope is OwnerQueueInputDeliveryScope
+            assert overview.inputs == () and overview.historical_inputs == ()
+            assert overview.historical_count == 0 and overview.dismissed_historical_count == 0

tests/textual/test_historical_notification_source.py

@@ -2,7 +2,8 @@
 
 from agent_comms.repo import Checkout
 from agent_comms.historical_views import HistoricalThread
-from agent_comms.ui_model.observation import ReadRecordedNotifications
+from agent_comms.notification_reads import RecordedNotificationRead
+from agent_comms.ui_model.observation import ReadNotifications
 from agent_comms.threads import Thread
 
 from support.roots import installed_wire
@@ -28,9 +29,8 @@ async def test_historical_notification_keeps_selected_source_and_lifetime(worksp
         await ui.until(lambda: content in access.notification_views,
                        why="the historical body's notification subscription")
         request = access.notification_views[content]
-        assert isinstance(request, ReadRecordedNotifications)
-        assert request.source == source.key
-        assert request.references == (message.reference,)
+        assert type(request) is ReadNotifications
+        assert request.operation == RecordedNotificationRead((message.reference,), source.key)
         await ui.until(lambda: request in access.notification_outcomes,
                        why="the archive's actual observation reply")
         outcomes = access.notification_outcomes[request]

tests/textual/test_message_notifications.py

@@ -35,7 +35,7 @@ async def publish(self, ui, chat):
         view = chat.message_history.notification_view
         access = ui.app.coordination_access
         request = access.notification_views[view]
-        outcomes = self.answer(request.references)
+        outcomes = self.answer(request.operation.references)
         expected, = CheckedNotifications.for_references(
             chat.message_history.visible_fragments()[-1].notification_references, outcomes)
         access.notifications_observed(NotificationsObserved(access.revision, request, outcomes), access.observation.observed())
@@ -88,8 +88,8 @@ async def test_visible_references_are_bounded_and_typed_outcomes_paint_their_det
         request = ui.app.coordination_access.notification_views[history.notification_view]
         on_screen = {reference for fragment in history.visible_fragments()
                      for reference in fragment.notification_references}
-        assert 0 < len(request.references) <= len(on_screen) < 30
-        assert set(request.references) <= on_screen
+        assert 0 < len(request.operation.references) <= len(on_screen) < 30
+        assert set(request.operation.references) <= on_screen
         await notifications.publish(ui, chat)
         await ui.until_shows("Delivered 45/45 · Consumed 0/45 · Read 0/45")
         assert newest_status(chat).working
@@ -144,7 +144,7 @@ async def test_a_hidden_view_unsubscribes_and_discards_a_late_reply(workspace):
         await ui.select_tab(owner)
         assert view not in ui.app.coordination_access.notification_views
         notifications.state = CompletedProgress(ShownToModelProgress(0))
-        view.notifications_observed(notifications.answer(request.references))
+        view.notifications_observed(notifications.answer(request.operation.references))
         await ui.step()
         assert view not in ui.app.coordination_access.notification_views
         assert {fragment.message.reference: fragment.shown.status