diff --git a/pyproject.toml b/pyproject.toml
index 564a0629e..f709fc75e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -84,3 +84,4 @@ markers = ["refactor_guard: permanent architecture deletion guards"]
 
 [tool.mypy]
 follow_untyped_imports = true
+mypy_path = ["src", "tools"]
diff --git a/src/toad/acp/agent.py b/src/toad/acp/agent.py
index db02cf442..cb0c79bbe 100644
--- a/src/toad/acp/agent.py
+++ b/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)
diff --git a/src/toad/acp/agent_controller.py b/src/toad/acp/agent_controller.py
index 91057609d..c2302c28d 100644
--- a/src/toad/acp/agent_controller.py
+++ b/src/toad/acp/agent_controller.py
@@ -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 @@ async def input_delivery(self, *, include_history: bool = False) -> DeliveryView
         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")
diff --git a/src/toad/owner_preparation.py b/src/toad/owner_preparation.py
index 04a5c8eb1..3e2372860 100644
--- a/src/toad/owner_preparation.py
+++ b/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()
diff --git a/src/toad/screens/commander.py b/src/toad/screens/commander.py
index 5be6bf5e4..4528d3af2 100644
--- a/src/toad/screens/commander.py
+++ b/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,
diff --git a/src/toad/screens/historical_sessions.py b/src/toad/screens/historical_sessions.py
index 2395cc0a4..7125d6c46 100644
--- a/src/toad/screens/historical_sessions.py
+++ b/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:
diff --git a/src/toad/session_scope.py b/src/toad/session_scope.py
index fed3c6f28..a3c9d0375 100644
--- a/src/toad/session_scope.py
+++ b/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,14 +159,13 @@ 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,
@@ -173,7 +173,9 @@ def update_request(
         *,
         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)
diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py
index d5a2fe5d5..4f73df4fb 100644
--- a/src/toad/widgets/message_notifications.py
+++ b/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
diff --git a/tests/app/test_recorded_transcript_reader.py b/tests/app/test_recorded_transcript_reader.py
index 09fbe55ed..1263b70f9 100644
--- a/tests/app/test_recorded_transcript_reader.py
+++ b/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)}
diff --git a/tests/native/test_native_goal_owner.py b/tests/native/test_native_goal_owner.py
index 8fc1016d1..8c7cf29dd 100644
--- a/tests/native/test_native_goal_owner.py
+++ b/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
diff --git a/tests/textual/test_historical_notification_source.py b/tests/textual/test_historical_notification_source.py
index 4959f5871..307672f37 100644
--- a/tests/textual/test_historical_notification_source.py
+++ b/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]
diff --git a/tests/textual/test_message_notifications.py b/tests/textual/test_message_notifications.py
index 670fa0bc2..7e9908783 100644
--- a/tests/textual/test_message_notifications.py
+++ b/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
diff --git a/tests/textual/test_observation_shutdown.py b/tests/textual/test_observation_shutdown.py
index 4e4aa236c..0a2b134db 100644
--- a/tests/textual/test_observation_shutdown.py
+++ b/tests/textual/test_observation_shutdown.py
@@ -1,35 +1,93 @@
 """Textual-facing: widget unmount; unmount ends interest, not the real observer's already-acquired reads."""
 
 import asyncio
+import os
+import signal
 import sys
+from contextlib import asynccontextmanager
+
 import pytest
 
-from agent_comms.ui_model.observation import ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications
+from agent_comms.notification_reads import LiveNotificationRead
+from agent_comms.ui_model.observation import (
+    ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications, StopObserving,
+    StopSidebar,
+)
 from toad.comms_root import AnsweringReads
 
 
+async def stopped(pid: int) -> None:
+    """Wait, bounded, until the stop is reported; WUNTRACED reports it without reaping."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + 5
+    while loop.time() < deadline:
+        found, status = os.waitpid(pid, os.WNOHANG | os.WUNTRACED)
+        if found:
+            assert os.WIFSTOPPED(status) and os.WSTOPSIG(status) == signal.SIGSTOP
+            return
+        await asyncio.sleep(0.01)
+    raise AssertionError(f"observer child {pid} did not stop")
+
+
+@asynccontextmanager
+async def held(process: ObservationProcess):
+    """Hold this test's own observer child; resume it on every path while it lives."""
+    pid = process.process.pid
+    os.kill(pid, signal.SIGSTOP)
+    try:
+        await stopped(pid)
+        yield pid
+    finally:
+        if process.process.is_alive():
+            os.kill(pid, signal.SIGCONT)
+
+
+async def settled(task: asyncio.Task) -> None:
+    if not task.done():
+        task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+
+
+def empty_read(workspace) -> ReadNotifications:
+    return ReadNotifications(str(workspace.roots.wire), LiveNotificationRead(()))
+
+
 async def test_an_observer_dying_during_shutdown_still_releases_its_resources(workspace):
-    with pytest.raises(BaseExceptionGroup) as failures:
+    died = None
+    try:
         async with workspace.app() as ui:
             access = ui.app.coordination_access
             await access.current_sidebar()
-            observed = access.observation.observed()
-            process = observed.process
-            # The death has not been delivered to receive(); stop_updates must
-            # encounter it at its actual StopObserving request during shutdown.
-            asyncio.get_running_loop().remove_reader(process.fileno())
-            process.process.kill()
-            await asyncio.to_thread(process.process.join, 5)
-            assert process.process.exitcode is not None
-            assert not process.connection.closed
-    assert failures.value.subgroup(ObserverDied) is not None
+            process = access.observation.observed().process
+            assert not (await asyncio.wait_for(process.read(empty_read(workspace)), 30)).outcomes
+            reading = None
+            try:
+                async with held(process) as pid:
+                    reading = asyncio.create_task(process.read(empty_read(workspace)))
+                    await asyncio.sleep(0)
+                    assert process.reads and not reading.done()
+                    os.kill(pid, signal.SIGKILL)
+                with pytest.raises(ObserverDied) as death:
+                    await asyncio.wait_for(reading, 10)
+            finally:
+                if reading is not None:
+                    await settled(reading)
+            died = death.value
+            assert died.pid == pid
+            assert died.exit_status == -signal.SIGKILL
+            for request in (StopSidebar(), StopObserving()):
+                with pytest.raises(ObserverDied) as again:
+                    process.request(request)
+                assert again.value is died
+    except* ObserverDied as failures:
+        assert died is not None
+        _same, other = failures.split(lambda error: error is died)
+        assert other is None
     assert process.connection.closed
     assert not process.process.is_alive()
     assert not access.retiring
 
 
-
-
 async def test_stopping_observation_keeps_reads_but_sends_no_late_interest(workspace):
     workspace.thread("sender", title="Sender")
     workspace.thread("reader", title="Reader")
@@ -43,50 +101,48 @@ async def test_stopping_observation_keeps_reads_but_sends_no_late_interest(works
         # Retiring the old process must not publish its state over the new one.
         assert access.observation is observing
         process = observing.process
-        loop = asyncio.get_running_loop()
-        acquired = asyncio.Event()
-        request = ReadNotifications(str(workspace.roots.wire), (message.reference,))
-
-        async def read_notifications():
-            acquired.set()
-            return await process.read(request)
-
-        # Hold the readiness callback, not the service, so this acquired read
-        # cannot resolve until stop_observing installs the read-only reader.
-        loop.remove_reader(process.fileno())
-        async with asyncio.TaskGroup() as reads:
-            reading = reads.create_task(read_notifications())
-            await acquired.wait()
-            assert process.reads and not reading.done()
-            await access.stop_observing()
-            answering = access.observation
-            assert type(answering) is AnsweringReads
-            assert answering.process is process
-            assert answering.service is observing.service
-            assert answering.selection is observing.selection
-            assert answering.sidebar is observing.sidebar
-            assert answering.subscriptions is observing.subscriptions
-            assert answering.updates is observing.updates
-
-            requests = []
-
-            def record_request(frame, event, result):
-                if (event == "call" and frame.f_code is ObservationProcess.request.__code__
-                        and frame.f_locals["self"] is process):
-                    requests.append(frame.f_locals["request"])
-
-            previous = sys.getprofile()
-            sys.setprofile(record_request)
-            try:
-                answering.views_changed()
-                answering.thread_interest(access)
-                answering.observe_sidebar(access.sidebar_request())
-                answering.observe_notifications(ObserveNotifications(frozenset()))
+        assert not (await asyncio.wait_for(process.read(empty_read(workspace)), 30)).outcomes
+        request = ReadNotifications(str(workspace.roots.wire), LiveNotificationRead((message.reference,)))
+        reading = None
+        try:
+            # The observer child itself holds the answer; the UI's one receiver keeps running.
+            async with held(process):
+                reading = asyncio.create_task(process.read(request))
+                await asyncio.sleep(0)
+                assert process.reads and not reading.done()
                 await access.stop_observing()
-                assert access.observation is answering
-            finally:
-                sys.setprofile(previous)
-            assert requests == []
-            result = await reading
-            assert set(result) == {(message.seq, message.message_id)}
-            assert process.process.is_alive()
+                answering = access.observation
+                assert type(answering) is AnsweringReads
+                assert answering.process is process
+                assert answering.service is observing.service
+                assert answering.selection is observing.selection
+                assert answering.sidebar is observing.sidebar
+                assert answering.subscriptions is observing.subscriptions
+                assert answering.updates is observing.updates
+
+                requests = []
+
+                def record_request(frame, event, result):
+                    if (event == "call" and frame.f_code is ObservationProcess.request.__code__
+                            and frame.f_locals["self"] is process):
+                        requests.append(frame.f_locals["request"])
+
+                previous = sys.getprofile()
+                sys.setprofile(record_request)
+                try:
+                    answering.views_changed()
+                    answering.thread_interest(access)
+                    answering.observe_sidebar(access.sidebar_request())
+                    answering.observe_notifications(ObserveNotifications(frozenset()))
+                    await access.stop_observing()
+                    assert access.observation is answering
+                finally:
+                    sys.setprofile(previous)
+                assert requests == []
+                assert not reading.done()
+            result = await asyncio.wait_for(reading, 30)
+        finally:
+            if reading is not None:
+                await settled(reading)
+        assert set(result.outcomes) == {(message.seq, message.message_id)}
+        assert process.process.is_alive()