diff --git a/src/toad/acp/agent_process.py b/src/toad/acp/agent_process.py
index fe565ce7a..54c7b624c 100644
--- a/src/toad/acp/agent_process.py
+++ b/src/toad/acp/agent_process.py
@@ -4,6 +4,7 @@
 import asyncio
 import json
 import os
+from collections.abc import Callable
 from contextlib import ExitStack
 from contextvars import ContextVar
 from pathlib import Path
@@ -19,7 +20,9 @@
 )
 from agent_comms.launch_root import LaunchRoot
 from agent_comms.route_selection import RouteSelection
+from pydantic import JsonValue
 
+from toad.acp.agent_session import AgentSession
 from toad.acp.shell_command import ShellCommand
 from toad.core.events import LogAgentFail
 
@@ -92,7 +95,7 @@ async def start(self, agent, processes: tuple[AgentProcess, ...]=()):
             connection.cleanup.enter_context(acquisition.pop_all())
             return connection
 
-    async def prompt(self, agent, content, metadata, require):
+    async def prompt(self, agent, content, metadata: dict[str, JsonValue], require: Callable[[], None]):
         """The agent-comms-acp child writes the prompt under Core's maintenance gate."""
         require()
         return await self.rpc.prompt(agent.session_id, content, **metadata)
@@ -119,66 +122,104 @@ def recipient(self, session_id: str):
             return members[0]
         raise RequestError.invalid_params({'details': 'ACP session is retired or unknown'})
 
-    async def request_permission(self, session_id, tool_call, options, **kwargs):
+    async def request_permission(
+        self,
+        session_id: str,
+        tool_call: schema.ToolCallUpdate,
+        options: list[schema.PermissionOption],
+        **kwargs: JsonValue,
+    ):
         agent = self.recipient(session_id)
         return await agent.session.start_operation(
             agent.permissions.request_permission(session_id, options, tool_call, kwargs))
 
-    async def session_update(self, session_id, update, **kwargs) -> None:
+    async def session_update(
+        self,
+        session_id: str,
+        update: schema.UserMessageChunk
+        | schema.AgentMessageChunk
+        | schema.AgentThoughtChunk
+        | schema.ToolCallStart
+        | schema.ToolCallProgress
+        | schema.AgentPlanUpdate
+        | schema.AgentPlanContentUpdate
+        | schema.AgentPlanRemovedUpdate
+        | schema.AvailableCommandsUpdate
+        | schema.CurrentModeUpdate
+        | schema.ConfigOptionUpdate
+        | schema.SessionInfoUpdate
+        | schema.UsageUpdate,
+        **kwargs: JsonValue,
+    ) -> None:
         agent = self.recipient(session_id)
         notification = schema.SessionNotification.model_construct(
             session_id=session_id, update=update, field_meta=kwargs or None)
         raw = self._incoming.get()['params']['update']
         await agent.session.start_operation(agent.updates.receive(notification, raw))
 
-    async def read_text_file(self, session_id, path, line=None, limit: Optional[int]=None, **kwargs):
+    async def read_text_file(
+        self,
+        session_id: str,
+        path: str,
+        line=None,
+        limit: Optional[int] = None,
+        **kwargs: JsonValue,
+    ):
         agent = self.recipient(session_id)
         return await agent.session.start_operation(asyncio.to_thread(
             agent.files.read_text_file, session_id, path, line, limit))
 
-    async def write_text_file(self, session_id, path, content, **kwargs):
+    async def write_text_file(self, session_id: str, path: str, content: str, **kwargs: JsonValue):
         agent = self.recipient(session_id)
         return await agent.session.start_operation(asyncio.to_thread(
             agent.files.write_text_file, session_id, path, content))
 
-    async def create_terminal(self, session_id, command, args=None, env=None,
-                              cwd=None, output_byte_limit=None, **kwargs):
+    async def create_terminal(
+        self,
+        session_id: str,
+        command: str,
+        args=None,
+        env=None,
+        cwd=None,
+        output_byte_limit=None,
+        **kwargs: JsonValue,
+    ):
         agent = self.recipient(session_id)
         return await agent.session.start_operation(agent.controller.terminal_create(
             session_id, command, kwargs, args, cwd, env, output_byte_limit))
 
-    async def terminal_output(self, session_id, terminal_id, **kwargs):
+    async def terminal_output(self, session_id: str, terminal_id: str, **kwargs: JsonValue):
         agent = self.recipient(session_id)
         return await agent.session.start_operation(
             agent.controller.terminal_output(session_id, terminal_id, kwargs))
 
-    async def release_terminal(self, session_id, terminal_id, **kwargs):
+    async def release_terminal(self, session_id: str, terminal_id: str, **kwargs: JsonValue):
         agent = self.recipient(session_id)
         return await agent.session.start_operation(
             agent.controller.terminal_release(session_id, terminal_id, kwargs))
 
-    async def wait_for_terminal_exit(self, session_id, terminal_id, **kwargs):
+    async def wait_for_terminal_exit(self, session_id: str, terminal_id: str, **kwargs: JsonValue):
         agent = self.recipient(session_id)
         return await agent.session.start_operation(
             agent.controller.terminal_wait_for_exit(session_id, terminal_id, kwargs))
 
-    async def kill_terminal(self, session_id, terminal_id, **kwargs):
+    async def kill_terminal(self, session_id: str, terminal_id: str, **kwargs: JsonValue):
         agent = self.recipient(session_id)
         return agent.controller.terminal_kill(session_id, terminal_id, kwargs)
 
-    async def create_elicitation(self, message, mode, **kwargs):
+    async def create_elicitation(self, message: str, mode: schema.ElicitationMode, **kwargs: JsonValue):
         raise RequestError.method_not_found('elicitation/create')
 
-    async def complete_elicitation(self, elicitation_id, **kwargs):
+    async def complete_elicitation(self, elicitation_id: str, **kwargs: JsonValue):
         raise RequestError.method_not_found('elicitation/complete')
 
-    async def ext_method(self, method, params):
+    async def ext_method(self, method: str, params: dict[str, JsonValue]):
         raise RequestError.method_not_found(f'_{method}')
 
-    async def ext_notification(self, method, params) -> None:
+    async def ext_notification(self, method: str, params: dict[str, JsonValue]) -> None:
         self.agent.log(f'[ACP unsupported notification] _{method}: {params}')
 
-    def on_connect(self, conn) -> None:
+    def on_connect(self, conn: ClientSideConnection) -> None:
         pass
 
     async def retire(self) -> None:
@@ -204,7 +245,7 @@ async def _retire(self) -> None:
             self.sessions.clear()
             self.cleanup.close()
 
-    async def detach(self, session):
+    async def detach(self, session: AgentSession):
         async with self.attachment_lock:
             if session not in self.sessions:
                 return
diff --git a/src/toad/comms_root.py b/src/toad/comms_root.py
index 34911a97b..f67a9affd 100644
--- a/src/toad/comms_root.py
+++ b/src/toad/comms_root.py
@@ -11,43 +11,71 @@
 
 from __future__ import annotations
 
-from agent_comms.errors import RouteChangedError
-
 import asyncio
 import threading
-from dataclasses import dataclass, field
-from functools import partial
 from collections.abc import Callable, Mapping
 from contextlib import ExitStack
-from typing import TYPE_CHECKING, TypeVar
+from dataclasses import dataclass, field
+from functools import partial
 from pathlib import Path
+from typing import TYPE_CHECKING, TypeVar
 from weakref import WeakKeyDictionary, WeakSet
 
-from agent_comms.coordination_errors import StaleRevision
 from agent_comms.child_process import join_retirement
+from agent_comms.coordination_errors import StaleRevision
+from agent_comms.display_references import PendingReferences, UnavailableReferences
+from agent_comms.errors import RouteChangedError
 from agent_comms.mro_dispatch import MroDispatch, handles
 from agent_comms.presentation import MessageNotification, NoRevision, ReadState
 from agent_comms.revision_views import ExternalTranscriptsView, StatusView
 from agent_comms.route_selection import RouteSelection, run_selected_write
 from agent_comms.ui_model.observation import (
-    ObserveNotifications, NotificationsObserved, NotificationsUnreadable, ReadNotifications,
-    ObservationProcess, ObserverDied, ObserveSidebar, ObserveThreads, ObserveViews, RevisionObserved,
-    SidebarObserved, SidebarPreparing, SidebarUnreadable, ViewsRetired,
-    StopObserving, StopSidebar, ThreadsObserved,
+    NotificationsObserved,
+    NotificationsUnreadable,
+    ObservationProcess,
+    ObserveNotifications,
+    ObserverDied,
+    ObserveSidebar,
+    ObserveThreads,
+    ObserveViews,
+    ReadNotifications,
+    RevisionObserved,
+    SidebarObserved,
+    SidebarPreparing,
+    SidebarUnreadable,
+    StopObserving,
+    StopSidebar,
+    ThreadsObserved,
+    ViewsRetired,
 )
 from agent_comms.ui_model.sidebar import SidebarModel, SidebarObservation
 from agent_comms.ui_model.status import OpenThreadStatus
-from toad.core.events import CoreEventStream, CoordinationObserved, ExternalTranscriptChanged, OpenTabsChanged, LinkContextChanged, ObservedServiceChanged, ThreadRowsChanged
-from toad.core_event_carrier import ThreadCallbackSubscription
-from toad.resource_release import AsyncResourceRelease, ResourceRelease
+
+from toad.core.events import (
+    CoordinationObserved,
+    CoreEventStream,
+    ExternalTranscriptChanged,
+    LinkContextChanged,
+    ObservedServiceChanged,
+    OpenTabsChanged,
+    ThreadRowsChanged,
+)
+from toad.core_event_carrier import CoreEventReceiver, ThreadCallbackSubscription
 from toad.preferences import SETTINGS
-from agent_comms.display_references import PendingReferences, UnavailableReferences
+from toad.resource_release import AsyncResourceRelease, ResourceRelease
 
 if TYPE_CHECKING:
-    from toad.widgets.message_notifications import HistoryNotifications
-    from toad.widgets.sidebar_tree import SidebarModelView
     from agent_comms.comms import Comms
+    from agent_comms.display_order import DisplayOrder
+    from agent_comms.display_references import ReferenceContext
+    from agent_comms.thread_identity import ThreadIncarnation
+
     from toad.app import ToadApp
+    from toad.session_entry import SessionEntry
+    from toad.session_navigation import SessionNavigation
+    from toad.widgets.linked_rows import ReferenceRendering
+    from toad.widgets.message_notifications import HistoryNotifications
+    from toad.widgets.sidebar_tree import SidebarModelView
 
 T = TypeVar("T")
 
@@ -72,25 +100,25 @@ def observed(self):
     def is_current(self) -> bool:
         return self is self.access.observation
 
-    def observes(self, root) -> bool:
+    def observes(self, root: str | Path) -> bool:
         return False
 
-    def uses(self, service) -> bool:
+    def uses(self, service: Comms) -> bool:
         return False
 
-    def tab(self, entry, sessions):
+    def tab(self, entry: SessionEntry, sessions: SessionNavigation):
         return entry.unobserved_tab()
 
-    def tabs(self, sessions):
+    def tabs(self, sessions: SessionNavigation):
         return {entry.mode: self.tab(entry, sessions) for entry in sessions.members}
-    def require(self, access, selected):
+    def require(self, access: CoordinationAccess, selected: RouteSelection):
         return access.open_selected(selected)
-    def references_for(self, root):
+    def references_for(self, root: str) -> ReferenceContext:
         return PendingReferences(root)
-    async def current_references(self, access, root):
+    async def current_references(self, access: CoordinationAccess, root: str):
         return access.reference_context(root)
 
-    def failed(self, error, selected):
+    def failed(self, error: Exception, selected):
         failed = FailedObservation(error, selected, access=self.access)
         if failed == self:
             return self
@@ -147,16 +175,16 @@ def service(self):
     def sidebar(self):
         raise self.error
 
-    async def write_order(self, access, root, channel, order):
+    async def write_order(self, access: CoordinationAccess, root, channel, order: type[DisplayOrder]):
         raise self.error
 
-    def references_for(self, root):
+    def references_for(self, root: str) -> ReferenceContext:
         return UnavailableReferences(root, str(self.error))
 
-    def show_sidebar(self, view):
+    def show_sidebar(self, view: SidebarModelView):
         view.bind_failed_observation(self)
 
-    def activate(self, access):
+    def activate(self, access: CoordinationAccess):
         self.observe_sidebar(None)
 
     def observe_sidebar(self, wanted):
@@ -164,11 +192,11 @@ def observe_sidebar(self, wanted):
         for view in tuple(access.sidebar_views):
             self.show_sidebar(view)
 
-    def observe_notifications(self, wanted):
+    def observe_notifications(self, wanted: ObserveNotifications):
         for view in tuple(self.access.notification_views):
             view.notifications_failed(self.error)
 
-    def thread_interest(self, access):
+    def thread_interest(self, access: CoordinationAccess):
         access.app.show_error(self.error, "Thread status unavailable")
 
 
@@ -180,29 +208,29 @@ class Observing(Observation):
     process: ObservationProcess = field(compare=False)
     loop: asyncio.AbstractEventLoop
 
-    def observes(self, root) -> bool:
+    def observes(self, root: str | Path) -> bool:
         return self.service.root == Path(root).resolve()
 
-    def tab(self, entry, sessions):
+    def tab(self, entry: SessionEntry, sessions: SessionNavigation):
         return entry.tab(sessions, self)
 
-    def show_sidebar(self, view) -> None:
+    def show_sidebar(self, view: SidebarModelView) -> None:
         if self.is_current():
             view.bind_observation(self)
 
-    def uses(self, service) -> bool:
+    def uses(self, service: Comms) -> bool:
         return self.service is service
 
-    async def write_order(self, access, root, channel, order):
+    async def write_order(self, access: CoordinationAccess, root, channel, order: type[DisplayOrder]):
         if root is not None and not self.observes(root):
             raise ValueError("Comms route changed before sorting")
         return await access.preparation.run_thread(
             access.write, self.selection, order.store_channel_order, self.service.channels, channel)
 
-    def require(self, access, selected):
+    def require(self, access: CoordinationAccess, selected: RouteSelection):
         return self if self.selection == selected else access.open_selected(selected)
 
-    def activate(self, access):
+    def activate(self, access: CoordinationAccess):
         self.loop.add_reader(self.process.fileno(), self.receive, access)
         access.update_sidebar_interest()
         self.thread_interest(access)
@@ -222,14 +250,14 @@ def stop_updates(self) -> AnsweringReads:
             self.selection, self.service, self.sidebar, self.process, self.loop,
             access=self.access, subscriptions=self.subscriptions, updates=self.updates)
 
-    def retire(self, access):
+    def retire(self, access: CoordinationAccess):
         self.loop.remove_reader(self.process.fileno())
         self.process.abandon_reads(StaleRevision("The Comms service was replaced before the read was answered"))
         task = asyncio.ensure_future(asyncio.to_thread(self.process.close))
         access.retiring.add(task)
         task.add_done_callback(access.retiring.discard)
 
-    def receive(self, access):
+    def receive(self, access: CoordinationAccess):
         if not self.is_current():
             return
         try:
@@ -242,7 +270,8 @@ def receive(self, access):
 
     def views_changed(self, event=None, subscription=None):
         access, root = self.access, str(self.service.root)
-        threads, channels = set(), set()
+        threads: set[ThreadIncarnation] = set()
+        channels: set[str] = set()
         for member in access.app.session_navigation.members:
             threads.update(identity for owner, identity in member.original_threads(access.app.session_navigation) if owner == root)
             channels.update(target for owner, target in member.original_channels() if owner == root)
@@ -251,23 +280,23 @@ def views_changed(self, event=None, subscription=None):
             access.views_interest = request
             self.process.request(request)
 
-    def references_for(self, root):
+    def references_for(self, root: str) -> ReferenceContext:
         return self.sidebar.observation.references_for(root)
 
-    async def current_references(self, access, root):
+    async def current_references(self, access: CoordinationAccess, root: str):
         try:
             await access.current_sidebar()
         except RouteChangedError:
             return await access.observation.current_references(access, root)
         return access.reference_context(root)
 
-    def thread_interest(self, access):
+    def thread_interest(self, access: CoordinationAccess):
         self.process.request(ObserveThreads(access.events.observed_threads(str(self.service.root))))
 
     def observe_sidebar(self, wanted):
         self.process.request(wanted if wanted is not None else StopSidebar())
 
-    def observe_notifications(self, wanted):
+    def observe_notifications(self, wanted: ObserveNotifications):
         self.process.request(wanted)
 
 class AnsweringReads(Observing):
@@ -279,13 +308,13 @@ def stop_updates(self) -> AnsweringReads:
     def views_changed(self, event=None, subscription=None):
         pass
 
-    def thread_interest(self, access):
+    def thread_interest(self, access: CoordinationAccess):
         pass
 
     def observe_sidebar(self, wanted):
         pass
 
-    def observe_notifications(self, wanted):
+    def observe_notifications(self, wanted: ObserveNotifications):
         pass
 
 
@@ -474,7 +503,12 @@ def sidebar_request(self) -> ObserveSidebar:
             SETTINGS.sidebar.show_archived.bind(state).value,
         )
 
-    def observe_thread_changes(self, view, scope: tuple[str, str], resume) -> ResourceRelease:
+    def observe_thread_changes(
+        self,
+        view: CoreEventReceiver,
+        scope: tuple[str, str],
+        resume: Callable[[ThreadRowsChanged], None],
+    ) -> ResourceRelease:
         """Hold the shared model while this native read waits for its own thread."""
         release = ResourceRelease()
 
@@ -498,14 +532,14 @@ def show_sidebar(self, view: SidebarModelView, shown: bool) -> None:
             self.sidebar_views.discard(view)
         self.update_sidebar_interest()
 
-    def show_references(self, view, shown):
+    def show_references(self, view: ReferenceRendering, shown: bool):
         if shown:
             self.reference_views.add(view)
         else:
             self.reference_views.discard(view)
         self.update_sidebar_interest()
 
-    def reference_context(self, root):
+    def reference_context(self, root: str) -> ReferenceContext:
         return self.observation.references_for(root)
 
     async def current_sidebar(self) -> SidebarObservation:
diff --git a/src/toad/core/context_inspection.py b/src/toad/core/context_inspection.py
index eb61bf769..787b83a16 100644
--- a/src/toad/core/context_inspection.py
+++ b/src/toad/core/context_inspection.py
@@ -8,13 +8,17 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Awaitable, Callable
 
+from acp.exceptions import RequestError
 from agent_comms.coordinator import Coordination
 from agent_comms.declared_family import DeclaredFamily
 from agent_comms.field_codec import FieldCodec
 from agent_comms.importing import ImportedSessionMetadata
 from agent_comms.mro_dispatch import handles
+from agent_comms.native_session_files import NativeSessionFiles
+from agent_comms.native_session_reopen import NativeSessionIdentity
 from agent_comms.native_turn_context import NativeContextData
 from agent_comms.pi_payloads import PiMessage
+from agent_comms.presentation import ReadState
 from agent_comms.registry_document import RegistryEntry
 from agent_comms.session_revision import SessionObservation, SessionRevision
 from agent_comms.thread_status import ActiveThreadPresence, ThreadStatus
@@ -31,13 +35,17 @@
 from agent_comms.working_memory_annotations import WorkingMemoryAnnotations
 from agent_comms.working_memory_labels import JevClassifier, ModelLabel
 from agent_comms.working_memory_questions import SpanAnswer
+from pydantic import JsonValue
 
 from toad.core.projection import MroProjection
 from toad.thread_identity import ThreadIdentity
 
 if TYPE_CHECKING:
     from agent_comms.comms import Comms
-    from agent_comms.context_store import ContextManifestPage
+    from agent_comms.context_store import (
+        CapturedContext,
+        ContextManifestPage,
+    )
 
 
 @dataclass(frozen=True)
@@ -87,7 +95,7 @@ def read():
             yield match
 
     @staticmethod
-    def search_roots(current_roots):
+    def search_roots(current_roots: Callable[[], tuple[ContextNode, ...]]):
         return current_roots()
 
     def search_description(self) -> str:
@@ -108,7 +116,7 @@ async def export(self, destination: Path) -> None:
         await Coordination.run_worker(partial(self._write_export, destination, detail))
 
     @staticmethod
-    def _write_export(destination, detail) -> None:
+    def _write_export(destination: Path, detail: str) -> None:
         with destination.open("x", encoding="utf-8") as output:
             output.write(detail)
 
@@ -138,7 +146,7 @@ def search_roots(self, current_roots):
     def search_description(self):
         return self.label
 
-    async def find(self, query, *, limit: int=100):
+    async def find(self, query: str, *, limit: int=100):
         original = await self.read_source(self.source)
         if query in original.text.casefold() or query in original.description.casefold():
             yield self
@@ -196,7 +204,7 @@ def detail(self) -> str:
                 "Historical request observation, not an additional active context section. "
                 "Segment counts are estimates, not provider measurements.")
 
-    async def find(self, query, *, limit: int=100):
+    async def find(self, query: str, *, limit: int=100):
         count = 0
         for child in self.children():
             async for match in child.find(query, limit=limit - count):
@@ -240,7 +248,7 @@ async def read(self) -> str:
         original = await self.source_text()
         return f"{original.description}\n\n{original.text}"
 
-    async def find(self, query, *, limit: int=100):
+    async def find(self, query: str, *, limit: int=100):
         # One complete selected value includes its annotation children. Search
         # that value once; opening a child still uses that exact child's path.
         original = await self.source_text()
@@ -296,7 +304,7 @@ def search_roots(self, current_roots):
     def search_description(self):
         return self.label
 
-    async def find(self, query, *, limit: int=100):
+    async def find(self, query: str, *, limit: int=100):
         if query in (await self.read()).casefold():
             yield self
 
@@ -324,7 +332,7 @@ def label(self):
     def children(self):
         return self.answers
 
-    def reader_path(self, key):
+    def reader_path(self, key: str):
         # Label refinement can move an original answer between sections. Its
         # request/span key stays owned by the answer, not the grouping label.
         for answer in self.answers:
@@ -343,7 +351,7 @@ def search_roots(self, current_roots):
     def search_description(self):
         return self.label
 
-    async def find(self, query, *, limit: int=100):
+    async def find(self, query: str, *, limit: int=100):
         # One search borrows each authenticated selected value once. These
         # buffers leave with the operation; no corpus or text index is kept.
         acquired = {}
@@ -412,16 +420,16 @@ def detail(self):
         return f"{super().detail()}\n\n{usage}" if usage else super().detail()
 
 
-class SegmentNodes(MroProjection):
-    def __init__(self, key: str, read_source) -> None:
+class SegmentNodes(MroProjection[NativeSegmentNode]):
+    def __init__(self, key: str, read_source: Callable[[Provenance], Awaitable[ContextSourceText]]) -> None:
         self.key, self.read_source = key, read_source
 
     @handles(ContextSegment)
-    def segment(self, segment):
+    def segment(self, segment: ContextSegment) -> NativeSegmentNode:
         return NativeSegmentNode(self.key, segment, self.read_source)
 
     @handles(NativeMessages)
-    def messages(self, segment):
+    def messages(self, segment) -> NativeSegmentNode:
         return NativeMessagesNode(self.key, segment, self.read_source)
 
 
@@ -493,10 +501,10 @@ def status(self) -> str:
 class AcquiredCoreInspection(CoreInspection):
     context: TurnContext
 
-    def roots(self, inspection):
+    def roots(self, inspection: ContextInspection):
         return inspection.contributors(self.context)
 
-    def groups(self, inspection):
+    def groups(self, inspection: ContextInspection):
         return (("Current Core instructions · before next input", self.roots(inspection), True),)
 
     @property
@@ -520,7 +528,13 @@ def owner(self):
         return self.entry.thread
 
     @staticmethod
-    def read(comms, owner, *, cursor=None, previous=()):
+    def read(
+        comms: Comms,
+        owner: str,
+        *,
+        cursor=None,
+        previous: tuple[CapturedContext, ...] = (),
+    ):
         entry = comms.registry.entry(owner).entry
         thread = entry.thread
         page = comms.contexts.page(owner, comms.registry,
@@ -529,11 +543,15 @@ def read(comms, owner, *, cursor=None, previous=()):
             comms.root / "coordination.sqlite3",
             tuple(resource.value for resource in page.resources), JevClassifier.version())
         imported = ImportedSessionMetadata.sources_for_owner(comms.registry, thread)
-        inspection_type = ThreadInspectionSource().dispatch_sync(entry.status)
-        return inspection_type(entry, page, SessionRevision.observe(thread.session_file), comms,
+        inspection_type = ThreadInspectionSource().project_value(entry.status)
+        files = NativeSessionFiles.for_identity(
+            NativeSessionIdentity.locate(thread.runtime.path),
+            root=comms.root, owner=thread.incarnation.require_recorded(),
+        )
+        return inspection_type(entry, page, SessionRevision.observe(files), comms,
                                annotations, imported)
 
-    async def acquire(self, revision):
+    async def acquire(self, revision: ReadState):
         return HoldingInspection(await self.acquire_core(), revision)
 
     def same_native_source(self, other: ContextInspection):
@@ -558,7 +576,7 @@ def imported(self):
             f"imported/{source.sha256}/{source.offset}/{source.instruction}", source,
             self.imported_source) for source in self.imported_sources)
 
-    async def imported_source(self, source):
+    async def imported_source(self, source: Provenance):
         return await Coordination.run_worker(partial(
             ImportedSessionMetadata.public_source_text,
             self.service.registry, self.owner, source, self.service))
@@ -602,7 +620,7 @@ async def find(self, roots: tuple[ContextNode, ...], query: str, *, limit: int=1
                     return tuple(matches)
         return tuple(matches)
 
-    async def _request(self, action: str, **parameters):
+    async def _request(self, action: str, **parameters: JsonValue | tuple[int, ...]):
         from agent_comms.runtime import RuntimeConnection, socket_path
 
         owner = await Coordination.run_worker(partial(self.service.registry.require, self.owner.name))
@@ -627,7 +645,7 @@ async def acquire_core(self):
             raise ValueError("Core context belongs to another thread incarnation")
         return replace(self, core=AcquiredCoreInspection(context))
 
-    async def core_source(self, context: TurnContext, position, source):
+    async def core_source(self, context: TurnContext, position: int, source: Provenance):
         def request():
             return dict(owner=FieldCodec.encode(context.thread), segment=position,
                 manifest=FieldCodec.encode(context.segments[position].manifest(0)),
@@ -636,13 +654,13 @@ def request():
         payload = await self._request("context_core_source", **parameters)
         return await Coordination.run_worker(partial(FieldCodec.decode, ContextSourceText, payload))
 
-    async def current_source(self, context: NativeContextData, position, source):
+    async def current_source(self, context: NativeContextData, position: int, source: Provenance):
         payload = await self._request("context_source",
             observation=FieldCodec.encode(context.observation()), segment=position,
             source=FieldCodec.encode(source))
         return FieldCodec.decode(ContextSourceText, payload)
 
-    async def recorded_source(self, manifest, position, source):
+    async def recorded_source(self, manifest: ContextManifest, position: int, source: Provenance):
         payload = await self._request("context_reference",
             record=FieldCodec.encode(self.page.reference(manifest)),
             turn=FieldCodec.encode(manifest.turn), request_id=manifest.request_id.require_id(),
@@ -667,106 +685,115 @@ async def correct_annotation(self, source: RecordedSegmentNode, label: ModelLabe
 
     def active(self, context: NativeContextData):
         return tuple(SegmentNodes(f"native/{context.identity.session_id}/{i}/{segment.declared_name}/{segment.sha256}",
-                                 partial(self.current_source, context, i)).dispatch_sync(segment)
+                                 partial(self.current_source, context, i)).project_value(segment)
                      for i, segment in enumerate(context.segments))
 
     def contributors(self, context: TurnContext):
         return tuple(SegmentNodes(f"core/{context.thread.name}/{context.thread.created_at}/{i}/{segment.declared_name}",
-                     partial(self.core_source, context, i)).dispatch_sync(segment)
+                     partial(self.core_source, context, i)).project_value(segment)
                      for i, segment in enumerate(context.segments))
 
 
 class SavedContextInspection(ContextInspection):
     """A stopped thread borrows retained records, never an owner socket."""
 
-    async def acquire(self, revision):
+    async def acquire(self, revision: ReadState):
         return SavedInspection(self, revision)
 
     def saved_manifest(self, manifest):
         return self.page.reference(manifest).read(
             self.service.bus.log, self.owner.name, self.service.registry)
 
-    async def recorded_source(self, manifest, position, source):
+    async def recorded_source(self, manifest: ContextManifest, position: int, source: Provenance):
         def read():
             return self.saved_manifest(manifest).public_source_text(self.service, position, source)
         return await Coordination.run_worker(read)
 
-    async def recorded_segment(self, manifest, position, contributors: tuple[int, ...]=()):
+    async def recorded_segment(self, manifest: ContextManifest, position: int, contributors: tuple[int, ...]=()):
         def read():
             return self.saved_manifest(manifest).selected_segment(position, contributors)
         segment = await Coordination.run_worker(read)
 
-        async def unavailable(original) -> str:
+        async def unavailable(original: SegmentManifest) -> str:
             return "Public SDK text was not retained in this saved record; no live preview is available."
 
         text = await segment.public_text(unavailable)
         return ContextSourceText(segment.public_description(), text)
 
 
-class ThreadInspectionSource(MroProjection):
+class ThreadInspectionSource(MroProjection[type[ContextInspection]]):
     """The original registry state selects the inspection's source capability."""
 
     @handles(ThreadStatus)
-    def saved(self, status):
+    def saved(self, status: ThreadStatus) -> type[ContextInspection]:
         return SavedContextInspection
 
     @handles(ActiveThreadPresence)
-    def live(self, status):
+    def live(self, status: ActiveThreadPresence) -> type[ContextInspection]:
         return ContextInspection
 
 
 class InspectionState(DeclaredFamily, affix="Inspection"):
     """One inspection lifetime; Textual owns workers and Tree resources separately."""
 
-    def observed_at(self, revision) -> bool:
+    def observed_at(self, revision: ReadState) -> bool:
         return True
 
-    def observe(self, revision) -> InspectionState:
+    def observe(self, revision: ReadState) -> InspectionState:
         return self
 
     @property
     def manifest_sources(self):
         return ()
 
-    async def acquire(self, service, revision, identity: ThreadIdentity, *, cursor=None) -> InspectionState:
+    async def acquire(
+        self,
+        service: Comms,
+        revision: ReadState,
+        identity: ThreadIdentity,
+        *,
+        cursor=None,
+    ) -> InspectionState:
         if str(service.root.resolve()) != identity.root:
             raise ValueError("Selected context belongs to another wire root")
         inspection = await Coordination.run_worker(partial(ContextInspection.read, service,
             identity.thread_name(), cursor=cursor, previous=self.manifest_sources))
         return await inspection.acquire(revision)
 
-    def receive_inspection(self, acquired) -> InspectionState:
+    def receive_inspection(self, acquired: InspectionState) -> InspectionState:
         return acquired
 
-    def inspection_differs(self, inspection) -> bool:
+    def inspection_differs(self, inspection: ContextInspection) -> bool:
         return True
 
-    def needs_native(self, inspection, force: bool) -> bool:
+    def needs_native(self, inspection: ContextInspection, force: bool) -> bool:
         return True
 
-    def same_source(self, inspection) -> bool:
+    def same_source(self, inspection: ContextInspection) -> bool:
         return False
 
-    def present(self, consumer) -> None:
+    def present[Result](self, consumer: Callable[[HoldingInspection], Result]) -> None:
         """A detached or unacquired owner has no inspection to present."""
 
-    def present_acquired(self, previous, consumer) -> None:
+    def present_acquired[Result](
+        self, previous: InspectionState, consumer: Callable[[HoldingInspection], Result]
+    ) -> None:
         """Publish acquired answers through this inspection's resource state."""
         self.present(consumer)
 
-    def with_native(self, inspection, native) -> InspectionState:
+    def with_native(self, inspection: ContextInspection, native: NativeContextData) -> InspectionState:
         return self
 
-    def native_failed(self, inspection, error) -> InspectionState:
+    def native_failed(self, inspection: ContextInspection, error: RequestError) -> InspectionState:
         return self
 
-    def prepare_native(self, consumer) -> None:
+    def prepare_native[Result](self, consumer: Callable[[HoldingInspection], Result]) -> None:
         """An owner without an inspection cannot accept native context work."""
 
-    def presentation_current(self, captured) -> bool:
+    def presentation_current(self, captured: HoldingInspection) -> bool:
         return False
 
-    def contains_native(self, matches) -> bool:
+    def contains_native(self, matches: Callable[[NativeContextData], bool]) -> bool:
         """Ask the acquired native resource; unread states own no such resource."""
         return False
 
@@ -776,30 +803,37 @@ def status(self) -> str:
 
 
 class DetachedInspection(InspectionState):
-    async def acquire(self, service, revision, identity: ThreadIdentity, *, cursor=None) -> InspectionState:
+    async def acquire(
+        self,
+        service: Comms,
+        revision: ReadState,
+        identity: ThreadIdentity,
+        *,
+        cursor=None,
+    ) -> InspectionState:
         return self
 
 
 class ReadingOwnerInspection(InspectionState):
-    def observed_at(self, revision) -> bool:
+    def observed_at(self, revision: ReadState) -> bool:
         return False
 
-    def observe(self, revision) -> InspectionState:
+    def observe(self, revision: ReadState) -> InspectionState:
         return ObservedOwnerInspection(revision)
 
 
 @dataclass(frozen=True)
 class ObservedOwnerInspection(ReadingOwnerInspection):
-    revision: int
+    revision: ReadState
 
-    def observed_at(self, revision) -> bool:
+    def observed_at(self, revision: ReadState) -> bool:
         return self.revision == revision
 
 
 @dataclass(frozen=True)
 class HoldingInspection(InspectionState):
     inspection: ContextInspection
-    revision: int
+    revision: ReadState
 
     @property
     def manifest_sources(self):
@@ -809,25 +843,27 @@ def manifest_sources(self):
     def thread_identity(self) -> ThreadIdentity:
         return ThreadIdentity(self.inspection.owner.incarnation, str(self.inspection.service.root))
 
-    def observed_at(self, revision) -> bool:
+    def observed_at(self, revision: ReadState) -> bool:
         return self.revision == revision
 
-    def observe(self, revision) -> InspectionState:
+    def observe(self, revision: ReadState) -> InspectionState:
         return replace(self, revision=revision)
 
-    def same_source(self, inspection) -> bool:
+    def same_source(self, inspection: ContextInspection) -> bool:
         return inspection.same_native_source(self.inspection)
 
-    def inspection_differs(self, inspection) -> bool:
+    def inspection_differs(self, inspection: ContextInspection) -> bool:
         return not self.same_source(inspection) or inspection.changed_since(self.inspection)
 
-    def needs_native(self, inspection, force: bool) -> bool:
+    def needs_native(self, inspection: ContextInspection, force: bool) -> bool:
         return force or self.inspection_differs(inspection)
 
-    def present(self, consumer) -> None:
+    def present[Result](self, consumer: Callable[[HoldingInspection], Result]) -> None:
         consumer(self)
 
-    def present_acquired(self, previous, consumer) -> None:
+    def present_acquired[Result](
+        self, previous: InspectionState, consumer: Callable[[HoldingInspection], Result]
+    ) -> None:
         # A recorded-only acquisition is not the replacement for an already
         # painted native context. Keep that original tree while native reads
         # join; success or refusal publishes the actual replacement below.
@@ -840,27 +876,27 @@ def groups(self):
                 ("Imported instructions · historical, not current", self.inspection.imported(), False),
                 ("Recorded requests · this page's source evidence", self.inspection.recorded(), False))
 
-    def prepare_native(self, consumer) -> None:
+    def prepare_native[Result](self, consumer: Callable[[HoldingInspection], Result]) -> None:
         consumer(self)
 
-    def with_native(self, inspection, native) -> InspectionState:
+    def with_native(self, inspection: ContextInspection, native: NativeContextData) -> InspectionState:
         return NativeInspection(self.inspection, self.revision, native) if self.same_source(inspection) else self
 
-    def native_failed(self, inspection, error) -> InspectionState:
+    def native_failed(self, inspection: ContextInspection, error: RequestError) -> InspectionState:
         return UnavailableNativeInspection(self.inspection, self.revision, str(error)) if self.same_source(inspection) else self
 
     def current_roots(self):
         return self.inspection.core.roots(self.inspection)
 
-    async def find(self, query, selected):
+    async def find(self, query: str, selected):
         roots = selected.search_roots if selected is not None else ContextNode.search_roots
         sources = await Coordination.run_worker(partial(roots, self.current_roots))
         return await self.inspection.find(sources, query)
 
-    def native_matches(self, current) -> bool:
+    def native_matches(self, current: InspectionState) -> bool:
         return not current.contains_native(bool)
 
-    def presentation_current(self, captured) -> bool:
+    def presentation_current(self, captured: HoldingInspection) -> bool:
         """Prepared content survives observations, not source/resource changes."""
         return (self.thread_identity == captured.thread_identity
                 and not self.inspection_differs(captured.inspection)
@@ -874,10 +910,12 @@ def status(self) -> str:
 class SavedInspection(HoldingInspection):
     """Only saved context is available; there is no native read to schedule."""
 
-    def prepare_native(self, consumer) -> None:
+    def prepare_native[Result](self, consumer: Callable[[HoldingInspection], Result]) -> None:
         pass
 
-    def present_acquired(self, previous, consumer) -> None:
+    def present_acquired[Result](
+        self, previous: InspectionState, consumer: Callable[[HoldingInspection], Result]
+    ) -> None:
         self.present(consumer)
 
     @property
@@ -890,7 +928,9 @@ def status(self) -> str:
 class NativeInspection(HoldingInspection):
     native: NativeContextData
 
-    def present_acquired(self, previous, consumer) -> None:
+    def present_acquired[Result](
+        self, previous: InspectionState, consumer: Callable[[HoldingInspection], Result]
+    ) -> None:
         # This acquisition retains its original native source, so independently
         # changed recorded requests and annotations can publish immediately.
         self.present(consumer)
@@ -900,10 +940,10 @@ def receive_inspection(self, acquired) -> InspectionState:
             return NativeInspection(acquired.inspection, acquired.revision, self.native)
         return acquired
 
-    def needs_native(self, inspection, force: bool) -> bool:
+    def needs_native(self, inspection: ContextInspection, force: bool) -> bool:
         return force or not self.same_source(inspection)
 
-    def with_native(self, inspection, native) -> InspectionState:
+    def with_native(self, inspection: ContextInspection, native: NativeContextData) -> InspectionState:
         if not self.same_source(inspection) or native == self.native:
             return self
         return NativeInspection(self.inspection, self.revision, native)
@@ -917,10 +957,10 @@ def groups(self):
     def current_roots(self):
         return (*super().current_roots(), *self.inspection.active(self.native))
 
-    def contains_native(self, matches) -> bool:
+    def contains_native(self, matches: Callable[[NativeContextData], bool]) -> bool:
         return matches(self.native)
 
-    def native_matches(self, current):
+    def native_matches(self, current: InspectionState):
         return current.contains_native(lambda native: native is self.native)
 
     @property
diff --git a/src/toad/core/projection.py b/src/toad/core/projection.py
index f977083c2..9d608ed9e 100644
--- a/src/toad/core/projection.py
+++ b/src/toad/core/projection.py
@@ -2,8 +2,8 @@
 from agent_comms.mro_dispatch import MroDispatch
 
 
-class MroProjection(MroDispatch):
+class MroProjection[Result](MroDispatch):
     """A leaf returns projection data rather than replacing its original value."""
 
-    def consume_handlers_sync(self, value, handlers, *args, **kwargs):
-        return next(iter(handlers))(value, *args, **kwargs)
+    def project_value(self, value, *args, **kwargs) -> Result:
+        return next(self.handlers_for(value))(value, *args, **kwargs)
diff --git a/src/toad/navigation_preparation.py b/src/toad/navigation_preparation.py
index 2d5346410..0fdba6d5a 100644
--- a/src/toad/navigation_preparation.py
+++ b/src/toad/navigation_preparation.py
@@ -177,9 +177,11 @@ def external(self) -> ThreadNavigation:
     def native(self) -> ThreadNavigation:
         binding = self.thread.execution.owner_binding(self.snapshot, self.thread)
         plan = ExistingOwnerPlan.from_binding(binding)
-        if plan is None:
+        if not plan.available:
             return ThreadNavigation(self.root, self.thread, self.project)
-        return NativeThreadNavigation(self.root, self.thread, self.project, plan)
+        return NativeThreadNavigation(
+            self.root, self.thread, self.project, plan.require_plan()
+        )
 
 
 class ThreadOpening:
diff --git a/src/toad/owner_preparation.py b/src/toad/owner_preparation.py
index 68ef6aa44..4b563d448 100644
--- a/src/toad/owner_preparation.py
+++ b/src/toad/owner_preparation.py
@@ -1,6 +1,8 @@
 """Captured read-side service binding for native owner RPC preparation."""
 
+from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from typing import ClassVar
 
 from agent_comms.comms import Comms
 from agent_comms.mro_dispatch import handles
@@ -14,22 +16,43 @@
 from toad.core.projection import MroProjection
 
 
-class ExistingOwnerPlan(MroProjection):
+class OwnerPlan(ABC):
+    """A captured binding either supplies a read-only plan or is unavailable."""
+
+    available: ClassVar[bool] = False
+
+    @abstractmethod
+    def require_plan(self) -> ExistingSessionLoadPlan: ...
+
+
+class UnavailableOwnerPlan(OwnerPlan):
+    def require_plan(self) -> ExistingSessionLoadPlan:
+        raise ValueError("Read-only attachment requires a live owner binding")
+
+
+@dataclass(frozen=True)
+class AvailableOwnerPlan(OwnerPlan):
+    binding: LiveThreadOwnerBinding
+    available: ClassVar[bool] = True
+
+    def require_plan(self) -> ExistingSessionLoadPlan:
+        return ExistingSessionLoadPlan(self.binding)
+
+
+class ExistingOwnerPlan(MroProjection[OwnerPlan]):
     """Project Core's binding into a read-only attachment plan without starting an owner."""
 
     @handles(ThreadOwnerBinding)
-    def unavailable(self, binding):
-        return None
+    def unavailable(self, binding: ThreadOwnerBinding) -> OwnerPlan:
+        return UnavailableOwnerPlan()
 
     @handles(LiveThreadOwnerBinding)
-    def live(self, binding):
-        return ExistingSessionLoadPlan(binding)
+    def live(self, binding: LiveThreadOwnerBinding) -> OwnerPlan:
+        return AvailableOwnerPlan(binding)
 
     @classmethod
-    def from_binding(
-        cls, binding: ThreadOwnerBinding
-    ) -> ExistingSessionLoadPlan | None:
-        return cls().dispatch_sync(binding)
+    def from_binding(cls, binding: ThreadOwnerBinding) -> OwnerPlan:
+        return cls().project_value(binding)
 
 
 @dataclass(frozen=True, slots=True)
diff --git a/src/toad/session_entry.py b/src/toad/session_entry.py
index 07579523f..68c30680f 100644
--- a/src/toad/session_entry.py
+++ b/src/toad/session_entry.py
@@ -172,9 +172,11 @@ async def reconnect(self, sessions, selected, snapshot, targets):
         thread = snapshot.require(snapshot.canonical_name(original.name))
         binding = thread.execution.owner_binding(snapshot, thread)
         plan = ExistingOwnerPlan.from_binding(binding)
-        if plan is None:
+        if not plan.available:
             return
-        await source.presentation.operational_binding.reconnect_session(source, plan)
+        await source.presentation.operational_binding.reconnect_session(
+            source, plan.require_plan()
+        )
 
     def unobserved_tab(self) -> OpenTab:
         return NamedTab(self.mode, self.details.name.display or "New Session")
@@ -217,7 +219,7 @@ def forget(self, sessions: SessionNavigation) -> None:
         sessions.app.session_tracker.close_session(self.mode)
 
 
-class HistorySessionEntry(SessionEntry, MroProjection):
+class HistorySessionEntry(SessionEntry, MroProjection[OpenTab]):
     def __init__(
         self,
         mode: str,
@@ -268,7 +270,7 @@ def unobserved_tab(self) -> OpenTab:
         return NamedTab(self.mode, self.key.conversation.initial_label)
 
     def tab(self, sessions: SessionNavigation, observed: Observing) -> OpenTab:
-        return self.dispatch_sync(observed.references_for(self.key.root), observed)
+        return self.project_value(observed.references_for(self.key.root), observed)
 
     @handles(ReferenceContext)
     def preparing_tab(self, references, observed) -> OpenTab:
diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py
index 72209b1dd..85d718dd4 100644
--- a/src/toad/session_presentation.py
+++ b/src/toad/session_presentation.py
@@ -151,7 +151,7 @@ def input_ready(self, conversation: Conversation) -> bool:
             return False
         if conversation.transcript.readable:
             presentation = conversation.thread_presentation
-            return ExistingOwnerPlan.from_binding(presentation.binding) is not None
+            return ExistingOwnerPlan.from_binding(presentation.binding).available
         return True
 
     def unavailable_reason(self, conversation: Conversation) -> str:
@@ -159,7 +159,7 @@ def unavailable_reason(self, conversation: Conversation) -> str:
         agent = conversation.agent_binding
         if (
             conversation.transcript.readable
-            and ExistingOwnerPlan.from_binding(presentation.binding) is None
+            and not ExistingOwnerPlan.from_binding(presentation.binding).available
         ):
             line = self.activity_row(conversation).presentation.summary
         elif agent.configuration.model.requires_selection:
diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py
index 9c5164f02..6738dda57 100644
--- a/src/toad/transcript_preparation.py
+++ b/src/toad/transcript_preparation.py
@@ -7,7 +7,7 @@
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field, replace
 from itertools import islice
-from typing import TYPE_CHECKING, Awaitable, Callable, ClassVar
+from typing import TYPE_CHECKING, ClassVar
 from weakref import ReferenceType, ref
 
 from agent_comms.historical_views import ConversationObservation
@@ -39,8 +39,15 @@
 )
 
 if TYPE_CHECKING:
+    from agent_comms.message_page import MessagePage
+
+    from toad.comms_root import CoordinationAccess, Observation
+    from toad.core.input_events import UserInputSubmitted
+    from toad.transcript_source_preparation import TranscriptSourcePreparation
+    from toad.widgets.comms_chat import CommsChatView
     from toad.widgets.conversation import Contents
-    from toad.widgets.transcript_history import LineHistory
+    from toad.widgets.transcript_history import LineHistory, TranscriptPageView
+    from toad.widgets.wire_history import WireHistory
 
 
 @dataclass
@@ -85,7 +92,7 @@ def measure_sources(self) -> None:
     def empty(cls, page):
         return cls.measured(page, ())
 
-    def project(self, fragments):
+    def project(self, fragments: tuple[TranscriptFragment, ...]):
         if len(fragments) == len(self.fragments) and all(
                 selected is original for selected, original in zip(fragments, self.fragments)):
             return self.for_view()
@@ -154,7 +161,9 @@ def restore_range(self, fragment_range: PageFragmentRange) -> None:
 
     # A page draws its selected fragments as lines: drawing moves a range,
     # it never mounts or removes native members.
-    async def extend(self, view, prepared: PreparedTranscriptPage, current) -> bool:
+    async def extend(
+        self, view: TranscriptPageView, prepared: PreparedTranscriptPage, current: Callable[[], bool],
+    ) -> bool:
         """Accept the staged rows; the range owner alone combines its membership."""
         if not current():
             return False
@@ -163,7 +172,7 @@ async def extend(self, view, prepared: PreparedTranscriptPage, current) -> bool:
         selected = slice(min(self.start, prepared.start), max(self.stop, prepared.stop))
         return self.replace_range(view, selected, current)
 
-    def replace_range(self, view, selected, current) -> bool:
+    def replace_range(self, view: TranscriptPageView, selected: slice, current: Callable[[], bool]) -> bool:
         if not current():
             return False
         if self.fragment_range != (selected.start, selected.stop):
@@ -326,7 +335,9 @@ async def get(self, request: PageRequest) -> PreparedTranscriptPage:
             raise asyncio.CancelledError
         return prepared
 
-    async def accept_page(self, history, edge, prepared) -> bool:
+    async def accept_page(
+        self, history: TranscriptSourcePreparation, edge: TranscriptPageView, prepared: PreparedTranscriptPage,
+    ) -> bool:
         """Immutable saved reads already carry their validated source cut."""
         return True
 
@@ -409,7 +420,7 @@ async def project(self, page: PreparedTranscriptPage, runtime: PreparationRuntim
 class RangeCategoryProjection(CategoryProjection):
     boundary: PageFragmentRange
 
-    def fragments_for(self, page):
+    def fragments_for(self, page: PreparedTranscriptPage):
         if self.boundary.page() is page:
             return self.boundary.members
         return super().fragments_for(page)
@@ -531,25 +542,25 @@ def replacement(self, through) -> SavedHistory:
         # reads to the superseded prefix's preparation scope.
         return self.earlier.replacement(through)
 
-    def read_task(self, request):
+    def read_task(self, request: PageRequest):
         # Keep the upstream task's declared native/wire behavior, bounded here.
         return replace(self.earlier.read_task(request), scope=self.scope, through=self.through)
 
-    async def _get(self, request):
+    async def _get(self, request: PageRequest):
         return await request.end.edge((self.earlier.get, super()._get))(request)
 
-    async def _prepare(self, request) -> None:
+    async def _prepare(self, request: PageRequest) -> None:
         await request.end.edge((self.earlier.prepare, super()._prepare))(request)
 
 
 class SourceObservation:
-    async def apply(self, history, source) -> bool:
+    async def apply(self, history: WireHistory, source: ReplaceablePageSource) -> bool:
         await history.jump_latest(LatestViewportRequest(LatestPosition.hold(history.window)))
         return not history.undrawn
 
 
 class Unchanged(SourceObservation):
-    async def apply(self, history, source) -> bool:
+    async def apply(self, history: WireHistory, source: ReplaceablePageSource) -> bool:
         if not history.undrawn:
             return (history.pages[-1].prepared.page.after != source.through
                     and await history.extend(source.through))
@@ -566,63 +577,63 @@ def __init__(self, through: TranscriptCursor, runtime: PreparationRuntime) -> No
         self.through, self.runtime = through, runtime
         self.scope = PreparationScope()
 
-    def uses_observation(self, observation) -> bool:
+    def uses_observation(self, observation: Observation) -> bool:
         return False
 
     async def route_is_current(self) -> bool:
         return False
 
-    def status_thread(self, access):
+    def status_thread(self, access: CoordinationAccess):
         from toad.status_thread import NoStatusThread
 
         return NoStatusThread()
 
-    def command_context(self, view):
+    def command_context(self, view: CommsChatView):
         return None
 
-    async def submit_input(self, view, event) -> None:
+    async def submit_input(self, view: CommsChatView, event: UserInputSubmitted) -> None:
         view.delivery_refused("Conversation source is not ready", event.body)
 
-    def schedule_refresh(self, view):
+    def schedule_refresh(self, view: CommsChatView):
         return None
 
-    def painted_read(self, history) -> PaintedRead:
+    def painted_read(self, history: WireHistory) -> PaintedRead:
         return NothingToRead()
 
-    async def change_style(self, history) -> None:
+    async def change_style(self, history: WireHistory) -> None:
         return
 
     async def read_revision(self):
         raise ValueError("This transcript has no saved-source reader")
 
-    async def observe_changes(self, revision):
+    async def observe_changes(self, revision: ConversationObservation):
         return Unchanged()
 
-    def seen_page(self, history, seen):
+    def seen_page(self, history: WireHistory, seen: tuple[TranscriptFragment, ...]):
         return None
 
-    async def mark_page(self, page) -> None:
+    async def mark_page(self, page: MessagePage) -> None:
         raise ValueError("This transcript has no saved-source reader")
 
     async def read_only(self) -> bool:
         return True
 
-    async def update_mentions(self, view) -> None:
+    async def update_mentions(self, view: CommsChatView) -> None:
         return
 
-    async def status_text(self, *, read_only) -> str:
+    async def status_text(self, *, read_only: bool) -> str:
         return "Conversation source is not ready"
 
-    def has_more(self, end, page) -> bool:
+    def has_more(self, end: PageEnd, page) -> bool:
         return False
 
     def has_older(self, page) -> bool:
         return False
 
-    def edge_reached(self, request, boundary) -> bool:
+    def edge_reached(self, request: PageRequest, boundary) -> bool:
         return True
 
-    async def latest(self, prepared):
+    async def latest(self, prepared: PreparedTranscriptPage):
         return prepared.for_view()
 
     def prefix(self, through):
@@ -631,10 +642,10 @@ def prefix(self, through):
     def replacement(self, through) -> ClosedHistory:
         return ClosedHistory(through, self.runtime)
 
-    async def _get(self, request):
+    async def _get(self, request: PageRequest):
         raise ValueError("This transcript has no saved-source reader")
 
-    async def _prepare(self, request) -> None:
+    async def _prepare(self, request: PageRequest) -> None:
         # No saved read can be speculated for a retained-only range.
         return
 
diff --git a/src/toad/transcript_publication.py b/src/toad/transcript_publication.py
index aeef69dbe..23a014f8a 100644
--- a/src/toad/transcript_publication.py
+++ b/src/toad/transcript_publication.py
@@ -5,59 +5,76 @@
 import asyncio
 from abc import ABC, abstractmethod
 from collections import deque
+from collections.abc import Callable, Sequence
 from contextlib import AsyncExitStack, asynccontextmanager
 from dataclasses import dataclass
 from functools import partial
-from typing import TYPE_CHECKING, TypeGuard
+from typing import TYPE_CHECKING, Self, TypeGuard
 
-from agent_comms.transcripts import TranscriptCursor, TranscriptPage, TranscriptReadIdentity
-from agent_comms.thread_presentation import ThreadPresentation
-from agent_comms.transcript_events import TranscriptCoverage
-from agent_comms.acp_extension import InputStartedUpdate, TranscriptDraftUpdate, TranscriptSnapshotUpdate
+from agent_comms.acp_extension import (
+    InputStartedUpdate,
+    TranscriptDraftUpdate,
+    TranscriptSnapshotUpdate,
+)
+from agent_comms.child_process import join_retirement
+from agent_comms.coordination_errors import (
+    CoordinationReadUnavailable,
+    StaleRevision,
+    TranscriptChanged,
+)
 from agent_comms.external_transcript import DraftRow
 from agent_comms.mro_dispatch import MroDispatch, handles
+from agent_comms.sent_message import PublishedSentMessage
+from agent_comms.thread_presentation import ThreadPresentation
+from agent_comms.transcript_events import TranscriptCoverage
+from agent_comms.transcripts import (
+    TranscriptCursor,
+    TranscriptPage,
+    TranscriptReadIdentity,
+)
 from agent_comms.ui_model.observation import (
     NotPublished,
     Published,
     TranscriptPublicationState,
 )
-from agent_comms.coordination_errors import (
-    CoordinationReadUnavailable,
-    StaleRevision,
-    TranscriptChanged,
-)
-from textual.worker import Worker, WorkerCancelled
 from textual.await_complete import AwaitComplete
 from textual.widget import Widget
-from toad.widgets.history_anchor import (
-    HeldReaderDestination,
-    TailPosition,
-    HistoryWindow,
+from textual.worker import Worker, WorkerCancelled
+
+from toad.core.events import CoreEventStream
+from toad.core.source_events import TranscriptCoverage as CoverageEvent
+from toad.resource_release import AsyncResourceRelease, ResourceRelease
+from toad.session_scope import (
+    NoTranscript,
+    SessionOrigin,
+    SessionScope,
+    TranscriptScope,
 )
+from toad.transcript_preparation import PreparedPageSource
 from toad.widgets.committed_presentation import (
     CommitBarrier,
-    CommitParticipant,
     CommitEvidence,
+    CommitParticipant,
     commit_plan,
     protected_blocks,
     retirement_candidates,
 )
-from agent_comms.child_process import join_retirement
-
-from toad.core.events import CoreEventStream
-from toad.resource_release import AsyncResourceRelease, ResourceRelease
-from toad.transcript_preparation import PreparedPageSource
+from toad.widgets.history_anchor import (
+    HeldReaderDestination,
+    HistoryWindow,
+    TailPosition,
+)
 from toad.widgets.session_details import SessionDetails
 from toad.widgets.transcript_history import TranscriptHistory
-from toad.session_scope import (
-    NoTranscript,
-    SessionOrigin,
-    SessionScope,
-    TranscriptScope,
-)
 
 if TYPE_CHECKING:
-    from toad.widgets.conversation import Conversation, ConversationSessionBinding
+    from toad.live_output import LiveOutput
+    from toad.widgets.conversation import (
+        Contents,
+        Conversation,
+        ConversationSessionBinding,
+    )
+    from toad.widgets.transcript_history import CommittedHistory
 
 
 @dataclass(frozen=True)
@@ -67,7 +84,7 @@ class BoundFrame:
     generation: int
 
     @classmethod
-    def capture(cls, binding: BoundView, owner, *args, **extra):
+    def capture(cls, binding: BoundView, owner: TranscriptPresentation, *args, **extra) -> Self:
         return cls(binding, owner, owner.generation, *args, **extra)
 
     def current(self):
@@ -77,17 +94,17 @@ def current(self):
             and self.binding.current()
         )
 
-    def painted(self, cursor) -> None:
+    def painted(self, cursor: TranscriptCursor) -> None:
         if self.current():
             self.binding.view.retire_loading_splash()
             self.record(cursor)
 
-    def observed(self, cursor) -> None:
+    def observed(self, cursor: TranscriptCursor) -> None:
         """A changed source cursor is not proof that its page has painted."""
         if self.current():
             self.record(cursor)
 
-    def record(self, cursor) -> None:
+    def record(self, cursor: TranscriptCursor) -> None:
         self.owner.displayed_cursor = cursor
         view = self.binding.view
         view.app.history_displayed()
@@ -349,7 +366,7 @@ def cancel(self) -> Worker[None] | None:
 
 
 class CommitPublication(CanonicalSourcePublication):
-    def __init__(self, binding, owner, generation) -> None:
+    def __init__(self, binding: BoundView, owner: TranscriptPresentation, generation: int) -> None:
         super().__init__(binding, owner, generation)
         self.plan = commit_plan(binding.window)
 
@@ -463,7 +480,7 @@ def thread_scope(self) -> TranscriptScope:
 
     reports_coverage = False
 
-    def covers_outgoing(self, message) -> bool:
+    def covers_outgoing(self, message: PublishedSentMessage) -> bool:
         return False
 
     def covers_input(self, started: InputStartedUpdate) -> bool:
@@ -482,10 +499,12 @@ def history(self) -> TranscriptHistory:
     def observe_thread(self, owner: TranscriptPresentation) -> None:
         pass
 
-    def capture(self, owner: TranscriptPresentation, kind, *args):
+    def capture[FrameT: BoundFrame](
+        self, owner: TranscriptPresentation, kind: type[FrameT], *args: ThreadPresentation
+    ):
         return None
 
-    def painted(self, owner, cursor) -> None:
+    def painted(self, owner: TranscriptPresentation, cursor: TranscriptCursor) -> None:
         pass
 
     def observed(self, owner: TranscriptPresentation, cursor: TranscriptCursor) -> None:
@@ -497,16 +516,16 @@ def request(self, owner: TranscriptPresentation) -> Worker[bool] | None:
     def cancel_workers(self):
         return ()
 
-    def covered(self, owner, coverage, history):
+    def covered(self, owner: TranscriptPresentation, coverage: CoverageEvent, history: CommittedHistory):
         return AwaitComplete.nothing()
 
     def reset_reader(self) -> None:
         pass
 
-    async def restore_documents(self, owner) -> None:
+    async def restore_documents(self, owner: TranscriptPresentation) -> None:
         pass
 
-    async def restore(self, owner) -> None:
+    async def restore(self, owner: TranscriptPresentation) -> None:
         pass
 
 
@@ -520,7 +539,7 @@ class Moment(BoundFrame):
         source: SessionScope
 
         @classmethod
-        def capture(cls, binding: BoundView, owner, **extra):
+        def capture(cls, binding: BoundView, owner: TranscriptPresentation, **extra) -> Self:
             return super().capture(
                 binding,
                 owner,
@@ -590,7 +609,7 @@ def current(self):
     def history(self) -> TranscriptHistory:
         return self._history
 
-    def observe_thread(self, owner) -> None:
+    def observe_thread(self, owner: TranscriptPresentation) -> None:
         view = self.view
         owner._thread_observation.enter_context(
             self.thread_scope.observe(
@@ -602,7 +621,7 @@ def observe_thread(self, owner) -> None:
     def reports_coverage(self):
         return self.history.state.reports_coverage
 
-    def covers_outgoing(self, message):
+    def covers_outgoing(self, message: PublishedSentMessage):
         return self.history.covers_outgoing(message)
 
     def covers_input(self, started: InputStartedUpdate):
@@ -631,7 +650,7 @@ async def replace() -> None:
 
         return AwaitComplete(replace())
 
-    async def restore_documents(self, owner) -> None:
+    async def restore_documents(self, owner: TranscriptPresentation) -> None:
         if not self.view.is_attached or not owner.documents:
             return
         moment = self.Moment.capture(self, owner)
@@ -642,7 +661,7 @@ async def restore_documents(self, owner) -> None:
         if moment.current():
             owner.documents = ()
 
-    async def restore(self, owner) -> None:
+    async def restore(self, owner: TranscriptPresentation) -> None:
         view = self.view
         view.resume_retained_history()
         if not owner.readable:
@@ -661,24 +680,26 @@ async def restore(self, owner) -> None:
         publication = self.capture(owner, CanonicalSourcePublication)
         view.call_after_refresh(partial(moment.refresh, publication))
 
-    async def reveal_retained(self, owner, history) -> None:
+    async def reveal_retained(self, owner: TranscriptPresentation, history: TranscriptHistory) -> None:
         view = self.view
         if not view.is_attached:
             return
         self.painted(owner, history.page_source.through)
         view.query_one(SessionDetails)._refresh_summary()
 
-    def capture(self, owner, kind, *args):
+    def capture[FrameT: BoundFrame](
+        self, owner: TranscriptPresentation, kind: type[FrameT], *args: ThreadPresentation
+    ):
         view = self.view
         if not view.is_attached or view.is_leaving:
             return None
         return kind.capture(self, owner, *args)
 
-    def painted(self, owner, cursor) -> None:
+    def painted(self, owner: TranscriptPresentation, cursor: TranscriptCursor) -> None:
         frame = BoundFrame.capture(self, owner)
         self.view.call_after_refresh(partial(frame.painted, cursor))
 
-    def observed(self, owner, cursor) -> None:
+    def observed(self, owner: TranscriptPresentation, cursor: TranscriptCursor) -> None:
         frame = BoundFrame.capture(self, owner)
         self.view.call_after_refresh(partial(frame.observed, cursor))
 
@@ -705,7 +726,7 @@ def cancel_workers(self):
             for worker in view.workers.cancel_group(view, group)
         )
 
-    def covered(self, owner, coverage, history):
+    def covered(self, owner: TranscriptPresentation, coverage: CoverageEvent, history: CommittedHistory):
         view = self.view
         contents = self.contents
         if history.is_attached and history.parent is contents:
@@ -788,7 +809,8 @@ def readable(self) -> bool:
         return self.binding.thread_scope.readable
 
     async def get_transcript_page(
-        self, *, before=None, after=None, through=None
+        self, *, before=None, after=None,
+        through=None
     ) -> TranscriptPage:
         frame = self.binding.capture(self, BoundFrame)
         scope = self.binding.thread_scope
@@ -816,7 +838,7 @@ def received(self, page: TranscriptPage) -> None:
     def reports_coverage(self) -> bool:
         return self.binding.reports_coverage
 
-    def covers_outgoing(self, message) -> bool:
+    def covers_outgoing(self, message: PublishedSentMessage) -> bool:
         return self.binding.covers_outgoing(message)
 
     def covers_input(self, started: InputStartedUpdate) -> bool:
@@ -832,7 +854,7 @@ def invalidate_publications(self) -> None:
         """Revoke in-flight view work without forgetting its published source."""
         self.generation += 1
 
-    async def publish(self, kind: type[TranscriptPublication], *args) -> bool:
+    async def publish(self, kind: type[TranscriptPublication], *args: ThreadPresentation) -> bool:
         publication = self.binding.capture(self, kind, *args)
         if publication is not None and publication.current():
             return await publication.publish()
@@ -852,7 +874,9 @@ def require_commit(self) -> None:
         self.commit_required = self.dirty = True
         self.binding.request(self)
 
-    async def mark_read(self, through, acknowledge, selected) -> None:
+    async def mark_read(
+        self, through: TranscriptCursor, acknowledge: Callable[[TranscriptCursor], None], selected: Callable[[], bool]
+    ) -> None:
         """A replaced source gets one fresh painted cursor, never a fatal read worker."""
         try:
             await asyncio.to_thread(acknowledge, through)
@@ -943,7 +967,9 @@ async def suspend(self) -> None:
             except WorkerCancelled:
                 pass
 
-    def retire_presentations(self, contents, output, candidates: list[Widget]) -> AwaitComplete:
+    def retire_presentations(
+        self, contents: Contents, output: LiveOutput, candidates: Sequence[Widget]
+    ) -> AwaitComplete:
         """Withdraw covered native paint now; join teardown outside source locks.
 
         CommitEvidence has already selected these exact resources. Their native
@@ -966,7 +992,7 @@ def retire_presentations(self, contents, output, candidates: list[Widget]) -> Aw
         )
         return AwaitComplete(join_retirement(retirement))
 
-    async def _retire_presentations(self, contents, output, candidates) -> None:
+    async def _retire_presentations(self, contents: Contents, output: LiveOutput, candidates: Sequence[Widget]) -> None:
         try:
             await output.retire_presentations(candidates)
         finally:
diff --git a/src/toad/transcript_state.py b/src/toad/transcript_state.py
index 7842a59cd..4ec3706dd 100644
--- a/src/toad/transcript_state.py
+++ b/src/toad/transcript_state.py
@@ -5,7 +5,7 @@
 from dataclasses import dataclass
 from functools import partial
 from time import monotonic
-from typing import TYPE_CHECKING, ClassVar, cast
+from typing import TYPE_CHECKING, ClassVar, Literal, cast
 
 from agent_comms.declared_family import DeclaredFamily
 from agent_comms.lifecycle import LifecycleState
@@ -17,16 +17,22 @@
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
 
-    from agent_comms.transcripts import TranscriptCursor
+    from agent_comms.transcripts import TranscriptCursor, TranscriptPage
+    from textual.worker import Worker
 
-    from toad.transcript_preparation import PreparedTranscriptPage
+    from toad.core.events import ThreadRowsChanged
+    from toad.transcript_preparation import PreparedPageSource, PreparedTranscriptPage
     from toad.transcript_source_preparation import (
         HistorySourceSnapshot,
         TranscriptSourcePreparation,
     )
     from toad.widgets.committed_presentation import CommitParticipant
-    from toad.widgets.history_anchor import LatestPosition
-    from toad.widgets.transcript_history import ConversationHistory
+    from toad.widgets.history_anchor import HistoryWindow, LatestPosition
+    from toad.widgets.transcript_history import (
+        ConversationHistory,
+        LineHistory,
+        ProjectedTranscriptHistory,
+    )
     from toad.wire_history import WireCut
 
 
@@ -35,56 +41,60 @@ class TranscriptState(DeclaredFamily, LifecycleState, affix="Transcript"):
     reports_coverage: ClassVar[bool] = False
     accepts_source_work: ClassVar[bool] = False
 
-    def source_ready(self, owner) -> bool:
+    def source_ready(self, owner: TranscriptSourcePreparation) -> bool:
         """The source accepts work and the filter is not scanning."""
         return self.accepts_source_work and owner.filter.idle
 
-    def blocks_visible_read(self, owner) -> bool:
+    def blocks_visible_read(self, owner: TranscriptSourcePreparation) -> bool:
         """An unpublished source must settle before its cursor is read."""
         return True
 
-    async def execute(self, owner, work):
+    async def execute[Result](
+        self, owner: TranscriptSourcePreparation, work: Callable[[], Awaitable[Result]]
+    ) -> Result | Literal[False]:
         raise RuntimeError("No transcript source operation is running")
 
-    def reserve(self, owner, operation=None) -> "WorkingTranscript":
+    def reserve(
+        self, owner: TranscriptSourcePreparation, operation=None
+    ) -> WorkingTranscript:
         raise RuntimeError("The transcript source cannot start another operation")
 
-    def request_latest(self, owner, request: "LatestViewportRequest") -> None:
+    def request_latest(self, owner: TranscriptSourcePreparation, request: LatestViewportRequest) -> None:
         """An inactive source starts no destination request."""
 
-    def check_edges(self, owner) -> None:
+    def check_edges(self, owner: TranscriptSourcePreparation) -> None:
         """A source that takes no work draws nothing beyond its edges."""
 
-    def delivered(self, owner) -> None:
+    def delivered(self, owner: TranscriptSourcePreparation) -> None:
         """Only a running source operation measures time to publication."""
 
-    def arriving(self, prepared, followed_end: bool):
+    def arriving(self, prepared: PreparedTranscriptPage, followed_end: bool):
         """The staged range begins at the edge this append advances."""
-        return prepared.fragments[prepared.start:prepared.stop] if followed_end else ()
+        return prepared.fragments[prepared.start : prepared.stop] if followed_end else ()
 
-    def mounted(self, owner) -> None:
+    def mounted(self, owner: LineHistory) -> None:
         """An already-read source needs no initial destination request."""
 
-    def rows_published(self, owner) -> None:
+    def rows_published(self, owner: TranscriptSourcePreparation) -> None:
         """Only a commit's advance has live blocks that its saved rows replace."""
 
-    def retirement_source(self) -> "TranscriptState":
+    def retirement_source(self) -> TranscriptState:
         return self
 
-    def retire_preparation(self, owner: "TranscriptSourcePreparation") -> None:
+    def retire_preparation(self, owner: TranscriptSourcePreparation) -> None:
         """Final source retirement releases its bounded prepared-page scope."""
         owner.close()
 
-    def resume(self) -> "TranscriptState":
+    def resume(self) -> TranscriptState:
         raise RuntimeError("Only a parked transcript can resume publication")
 
-    def resume_if_parked(self, owner) -> None:
+    def resume_if_parked(self, owner: TranscriptSourcePreparation) -> None:
         """Showing a resource does not start work on an inactive source."""
 
-    def validate_snapshot(self, owner, page) -> None:
+    def validate_snapshot(self, owner: TranscriptSourcePreparation, page: TranscriptPage) -> None:
         """A captured source page may validate its existing parked resource."""
 
-    def observed(self, widget: Widget) -> "TranscriptState":
+    def observed(self, widget: Widget) -> TranscriptState:
         # Textual sets these before dispatching Prune/Unmount, including when
         # an ancestor is removed. Decode at this framework boundary so an
         # async page cannot publish during that interval. They are framework
@@ -95,14 +105,14 @@ def observed(self, widget: Widget) -> "TranscriptState":
             return DetachedTranscript(self)
         return self
 
-    def for_projection(self, projection, owner) -> "TranscriptState":
+    def for_projection(self, projection: ProjectedTranscriptHistory, owner) -> TranscriptState:
         """A view may publish only while its source still owns that projection."""
         if owner is None or not owner.state.can_draw or not owner.filter.owns_projection(projection):
             return RetiredProjectionTranscript(self)
         return self
 
     @abstractmethod
-    def publish(self) -> "TranscriptState": ...
+    def publish(self) -> TranscriptState: ...
 
 
 class ProvisionalTranscript(TranscriptState):
@@ -119,13 +129,13 @@ class LiveTranscript(TranscriptState):
     reports_coverage = True
     accepts_source_work = True
 
-    def blocks_visible_read(self, owner) -> bool:
+    def blocks_visible_read(self, owner: TranscriptSourcePreparation) -> bool:
         return NEWER_END.visible_more(owner) or not self.source_ready(owner)
 
-    def reserve(self, owner, operation=None) -> "WorkingTranscript":
+    def reserve(self, owner, operation=None) -> WorkingTranscript:
         return (operation or WorkingTranscript)(self, owner)
 
-    def request_latest(self, owner, request: "LatestViewportRequest") -> None:
+    def request_latest(self, owner: TranscriptSourcePreparation, request: LatestViewportRequest) -> None:
         owner.schedule_source_work(partial(owner.jump_latest, request))
 
     def check_edges(self, owner) -> None:
@@ -134,8 +144,7 @@ def check_edges(self, owner) -> None:
 
     @classmethod
     def successors(cls):
-        return (LiveTranscript, ParkedSourceTranscript, DetachedTranscript,
-                ClosingTranscript)
+        return (LiveTranscript, ParkedSourceTranscript, DetachedTranscript, ClosingTranscript)
 
     def publish(self) -> TranscriptState:
         return self
@@ -153,7 +162,7 @@ class UnreadTranscript(ProvisionalTranscript, LiveTranscript):
 
     reports_coverage = False
 
-    def mounted(self, owner) -> None:
+    def mounted(self, owner: LineHistory) -> None:
         owner.call_after_refresh(owner.load_initial_page)
 
     def rows_published(self, owner) -> None:
@@ -166,7 +175,7 @@ def rows_published(self, owner) -> None:
 class SuspendedTranscript(TranscriptState):
     source: TranscriptState
 
-    def blocks_visible_read(self, owner) -> bool:
+    def blocks_visible_read(self, owner: TranscriptSourcePreparation) -> bool:
         """A parked or closing source need not paint now."""
         return False
 
@@ -184,50 +193,49 @@ def successors(cls): ...
 
 class LatestViewportState(ABC):
     @abstractmethod
-    def resume(self, owner) -> "LatestViewportState": ...
+    def resume(self, owner: TranscriptSourcePreparation) -> LatestViewportState: ...
 
-    def prepare(self, owner) -> None:
+    def prepare(self, owner: TranscriptSourcePreparation) -> None:
         """No pending destination needs another preparation frame."""
 
 
 class NoLatestRequest(LatestViewportState):
-    def resume(self, owner) -> LatestViewportState:
+    def resume(self, owner: TranscriptSourcePreparation) -> LatestViewportState:
         return self
 
 
 @dataclass(frozen=True)
 class LatestViewportRequest(LatestViewportState):
-    destination: "LatestPosition"
+    destination: LatestPosition
 
     @contextmanager
     def reading(self, owner):
         with ResourceRelease() as release:
             yield owner.page_source, release
 
-    async def page(self, owner, source):
+    async def page(self, owner: TranscriptSourcePreparation, source: PreparedPageSource):
         return await source.latest(owner.pages[-1].prepared)
 
-    def current(self, window) -> bool:
+    def current(self, window: HistoryWindow) -> bool:
         return self.destination.current(window)
 
-    def restore(self, window) -> None:
+    def restore(self, window: HistoryWindow) -> None:
         self.destination.restore(window)
 
-    def apply(self, owner) -> None:
+    def apply(self, owner: TranscriptSourcePreparation) -> None:
         if self.current(owner.window):
             owner.state.request_latest(owner, self)
 
-    def defer(self, owner) -> LatestViewportState:
+    def defer(self, owner: TranscriptSourcePreparation) -> LatestViewportState:
         """Keep only the reader's still-current destination for another draw."""
         return self if self.current(owner.window) else NoLatestRequest()
 
-    def prepare(self, owner) -> None:
+    def prepare(self, owner: TranscriptSourcePreparation) -> None:
         owner.request_preparation()
 
-    def resume(self, owner) -> LatestViewportState:
+    def resume(self, owner: TranscriptSourcePreparation) -> LatestViewportState:
         """Return the request state after this draw's source-work admission."""
-        if (not self.current(owner.window)
-                or owner.schedule_source_work(partial(owner.jump_latest, self))):
+        if not self.current(owner.window) or owner.schedule_source_work(partial(owner.jump_latest, self)):
             return NoLatestRequest()
         return self
 
@@ -247,9 +255,10 @@ def reading(self, owner: ConversationHistory):
             release.callback(source.close)
             yield source, release
 
-    async def page(self, owner, source):
+    async def page(self, owner: TranscriptSourcePreparation, source: PreparedPageSource):
         return await source.get(PageRequest(OLDER_END, source.through))
 
+
 @dataclass(frozen=True)
 class LatestReplacementViewportRequest(ReplacementViewportRequest):
     through: TranscriptCursor | WireCut
@@ -257,6 +266,7 @@ class LatestReplacementViewportRequest(ReplacementViewportRequest):
     def source(self, owner: ConversationHistory):
         return owner.page_source.replacement(self.through)
 
+
 @dataclass(frozen=True)
 class PreparedReplacementViewportRequest(ReplacementViewportRequest):
     """An existing prepared page needs no storage read, including closed sources."""
@@ -266,30 +276,28 @@ class PreparedReplacementViewportRequest(ReplacementViewportRequest):
     def source(self, owner: ConversationHistory):
         return owner.page_source.replacement(self.initial.page.after)
 
-    async def page(self, owner, source):
+    async def page(self, owner: TranscriptSourcePreparation, source: PreparedPageSource):
         return self.initial.for_view()
 
 
 @dataclass(frozen=True, eq=False)
-class WaitingSourceWork:
+class WaitingSourceWork[Result]:
     """The pager's observation keeps one paused operation alive until release."""
 
-    owner: "TranscriptSourcePreparation"
-    snapshot: "HistorySourceSnapshot"
-    work: "Callable[[], Awaitable[object]]"
-    state: "WorkingTranscript"
+    owner: TranscriptSourcePreparation
+    snapshot: HistorySourceSnapshot
+    work: Callable[[], Awaitable[Result]]
+    state: WorkingTranscript
 
     @contextmanager
     def observe(self):
         from toad.widgets.conversation import Conversation
 
         scope = self.owner.query_ancestor(Conversation).transcript.binding.thread_scope
-        with self.owner.app.coordination_access.observe_thread_changes(
-            self.owner, scope, self.resume
-        ):
+        with self.owner.app.coordination_access.observe_thread_changes(self.owner, scope, self.resume):
             yield self
 
-    def resume(self, _event) -> None:
+    def resume(self, _event: ThreadRowsChanged) -> None:
         self.owner._source_observation.close()
         if self.snapshot.current(self.owner) and self.owner._source_state is self.state:
             self.state.schedule(self.owner, self.work)
@@ -311,10 +319,10 @@ def __init__(self, source: TranscriptState, owner):
         # The screen asked for rows beyond an edge while this work held the source.
         self.edges_wait = False
 
-    def rows_published(self, owner) -> None:
+    def rows_published(self, owner: TranscriptSourcePreparation) -> None:
         self.source.rows_published(owner)
 
-    def blocks_visible_read(self, owner) -> bool:
+    def blocks_visible_read(self, owner: TranscriptSourcePreparation) -> bool:
         """A running source operation must finish before its cursor is read."""
         return True
 
@@ -338,10 +346,10 @@ def delivered(self, owner) -> None:
             owner.window.lookahead.delivered(now - self.delivery_started)
             self.delivery_window, self.delivery_started = window, now
 
-    def request_latest(self, owner, request: LatestViewportRequest) -> None:
+    def request_latest(self, owner: TranscriptSourcePreparation, request: LatestViewportRequest) -> None:
         owner.supersede_source_work(self, request)
 
-    def check_edges(self, owner) -> None:
+    def check_edges(self, owner: TranscriptSourcePreparation) -> None:
         # One work at a time holds the source. The check is made again when
         # this work ends, even if it changed nothing itself.
         self.edges_wait = True
@@ -357,16 +365,17 @@ def can_draw(self) -> bool:
 
     @classmethod
     def successors(cls):
-        return (LiveTranscript, ProvisionalTranscript, RetiredSourceTranscript,
-                ClosingTranscript, DetachedTranscript)
+        return (LiveTranscript, ProvisionalTranscript, RetiredSourceTranscript, ClosingTranscript, DetachedTranscript)
 
-    def defer_source_work(self, owner, work) -> None:
+    def defer_source_work[Result](
+        self, owner: TranscriptSourcePreparation, work: Callable[[], Awaitable[Result]]
+    ) -> None:
         """The existing observation scope owns the paused operation's lifetime."""
         waiting = WaitingSourceWork(owner, owner.source_snapshot(), work, self)
         owner._source_observation.close()
         owner._source_observation.enter_context(waiting.observe())
 
-    async def execute(self, owner, work):
+    async def execute[Result](self, owner, work: Callable[[], Awaitable[Result]]) -> Result | Literal[False]:
         from agent_comms.coordination_errors import (
             CoordinationReadUnavailable,
             StaleRevision,
@@ -397,7 +406,7 @@ async def execute(self, owner, work):
                 owner.source_failed(error)
                 return False
 
-    def schedule(self, owner, work):
+    def schedule[Result](self, owner, work: Callable[[], Awaitable[Result]]) -> Worker[Result | Literal[False]]:
         return owner.run_worker(partial(self.execute, owner, work))
 
 
@@ -449,7 +458,7 @@ def publish(self) -> TranscriptState:
     def retirement_source(self) -> TranscriptState:
         return self.source.retirement_source()
 
-    def retire_preparation(self, owner: "TranscriptSourcePreparation") -> None:
+    def retire_preparation(self, owner: TranscriptSourcePreparation) -> None:
         """Parking retains data-only work under the shared runtime's budget.
 
         Publication and speculative workers are revoked by the pager lifetime;
@@ -460,10 +469,10 @@ def retire_preparation(self, owner: "TranscriptSourcePreparation") -> None:
     def resume(self) -> TranscriptState:
         return self.source
 
-    def resume_if_parked(self, owner) -> None:
+    def resume_if_parked(self, owner: TranscriptSourcePreparation) -> None:
         owner.resume_source()
 
-    def validate_snapshot(self, owner, page) -> None:
+    def validate_snapshot(self, owner: TranscriptSourcePreparation, page: TranscriptPage) -> None:
         # The publication already holds a certified current source page. Its
         # native/assigned cursor, not rendered content, validates this resource.
         if owner.page_source.accepts_commit(page.after):
@@ -485,17 +494,20 @@ def __init__(self, source: TranscriptState, owner, live_blocks: tuple[Widget, ..
         self.claims = tuple(cast("CommitParticipant", block).commit_claim for block in live_blocks)
         self.retirement = AwaitComplete.nothing()
 
-    async def execute(self, owner, work):
+    async def execute[Result](self, owner, work: Callable[[], Awaitable[Result]]) -> Result | Literal[False]:
         try:
             return await super().execute(owner, work)
         finally:
             await self.retirement
 
-    def arriving(self, prepared, followed_end: bool):
+    def arriving(self, prepared: PreparedTranscriptPage, followed_end: bool):
         # Rows its live blocks already showed have not arrived again; every other row has.
         # The live blocks are the commit participants `retirement_candidates` chose.
-        return tuple(fragment for fragment in super().arriving(prepared, followed_end)
-                     if not any(fragment.shown_live_by(claim) for claim in self.claims))
+        return tuple(
+            fragment
+            for fragment in super().arriving(prepared, followed_end)
+            if not any(fragment.shown_live_by(claim) for claim in self.claims)
+        )
 
     def rows_published(self, owner) -> None:
         if self.live_blocks and not NEWER_END.visible_more(owner):
diff --git a/src/toad/widgets/presentation_window.py b/src/toad/widgets/presentation_window.py
index 1dd2798ac..22336a346 100644
--- a/src/toad/widgets/presentation_window.py
+++ b/src/toad/widgets/presentation_window.py
@@ -2,7 +2,7 @@
 
 from abc import ABC, abstractmethod
 from collections import deque
-from collections.abc import Iterable
+from collections.abc import Iterable, Sequence
 from dataclasses import dataclass, fields, replace
 from functools import partial
 from itertools import zip_longest
@@ -15,9 +15,22 @@
 from agent_comms.transcripts import TranscriptCursor
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from agent_comms.message_page import MessagePage
     from agent_comms.transcripts import TranscriptPage
+    from agent_comms.ui_model.message_category import MessageCategory
     from textual.widget import Widget
 
+    from toad.transcript_preparation import (
+        PreparedTranscriptPage,
+        ProjectedTranscriptSource,
+    )
+    from toad.transcript_source_preparation import TranscriptSourcePreparation
+    from toad.widgets.history_anchor import HistoryWindow
+    from toad.widgets.transcript_history import LineHistory, TranscriptPageView
+    from toad.wire_history import AfterSequence, WireCut, WireSource
+
 
 def protected_presentations(items: Iterable["Widget"], endpoints: Iterable["Widget"]) -> set["Widget"]:
     """Resolve native selection/focus to its owning presentation, without geometry."""
@@ -122,13 +135,13 @@ class PageEnd(ABC):
     step: ClassVar[int]
     accepts_destination = True
 
-    def boundary(self, page) -> int:
+    def boundary(self, page: PreparedTranscriptPage) -> int:
         return self.entry(len(page.fragments) + 1)
 
-    def indices(self, page, boundary: int) -> range:
+    def indices(self, page: PreparedTranscriptPage, boundary: int) -> range:
         return range(boundary + min(self.step, 0), self.edge((-1, len(page.fragments))), self.step)
 
-    def can_extend(self, page) -> bool:
+    def can_extend(self, page: PreparedTranscriptPage) -> bool:
         return self.edge((page.start, len(page.fragments) - page.stop)) > 0
 
     def has_more(self, through, page) -> bool:
@@ -137,13 +150,13 @@ def has_more(self, through, page) -> bool:
     def adjacent(self, page) -> PageRequest:
         return PageRequest(self, self.edge((page.before, page.after)))
     @abstractmethod
-    def advance_empty(self, edge, staged) -> None: ...
+    def advance_empty(self, edge: PreparedTranscriptPage, staged: PreparedTranscriptPage) -> None: ...
     @abstractmethod
-    def edge(self, pages): ...
-    def edge_boundary(self, edge, local: bool, staged) -> int:
+    def edge[Item](self, pages: Sequence[Item]) -> Item: ...
+    def edge_boundary(self, edge: TranscriptPageView, local: bool, staged: PreparedTranscriptPage) -> int:
         return self.edge((edge.prepared.start, edge.prepared.stop)) if local else self.boundary(staged)
     @abstractmethod
-    async def mount(self, history, view, pages) -> None: ...
+    async def mount(self, history: LineHistory, view: TranscriptPageView, pages: deque[TranscriptPageView]) -> None: ...
     @staticmethod
     @abstractmethod
     def append[Item](pages: deque[Item], view: Item, /) -> None: ...
@@ -151,11 +164,11 @@ def append[Item](pages: deque[Item], view: Item, /) -> None: ...
     @abstractmethod
     def pop[Item](pages: deque[Item], /) -> Item: ...
 
-    def eviction_fragments(self, page):
+    def eviction_fragments(self, page: PreparedTranscriptPage):
         return page.fragments_in_range[:: -self.step]
     def rows_toward(self, runway: Runway) -> int:
         return self.edge((runway.above, runway.below))
-    def at_boundary(self, window) -> bool:
+    def at_boundary(self, window: HistoryWindow) -> bool:
         boundary = self.edge((0, window.max_scroll_y))
         return all((position - boundary) * self.step >= 0
                    for position in (window.scroll_y, window.scroll_target_y))
@@ -163,31 +176,36 @@ def at_boundary(self, window) -> bool:
     def work_cursor[Cursor](self, through: Cursor) -> Cursor | None:
         return self.edge((None, through))
     @abstractmethod
-    async def load_page(self, loader, cursor: TranscriptCursor | None, through): ...
+    async def load_page(self, loader, cursor: TranscriptCursor | None, through) -> TranscriptPage: ...
     def progressed(self, page: "TranscriptPage", cursor: TranscriptCursor) -> bool:
         edge = self.edge((page.before, page.after))
         start, stop = self.edge(((edge, cursor), (cursor, edge)))
         return cursor.accepts_source(edge) and not (
             start.contains(stop) and self.edge((page.has_older, page.has_newer)))
     @abstractmethod
-    def wire_page(self, source, cursor, through): ...
+    def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence) -> MessagePage: ...
 
     def entry(self, length: int):
         return self.edge((length - 1, 0))
 
-    async def project(self, source, initial, prepared):
+    async def project(
+        self, source: ProjectedTranscriptSource, initial: PreparedTranscriptPage, prepared: PreparedTranscriptPage,
+    ):
         return replace(prepared, page=replace(prepared.page, has_newer=True))
 
-    def visible_more(self, history):
+    def visible_more(self, history: TranscriptSourcePreparation):
         return history.has_more(self)
 
-    def fresh(self, history) -> bool:
+    def fresh(self, history: LineHistory) -> bool:
         return False
 
     def accepts_velocity(self, velocity: float) -> bool:
         return velocity * self.step > 0
 
-    async def walk(self, history, current, target, selected, start, rows_needed, limit: int):
+    async def walk(
+        self, history: LineHistory, current: Callable[[], bool], target,
+        selected: frozenset[type[MessageCategory]], start: tuple[int, int], rows_needed: int, limit: int,
+    ):
         for index in range(start[0], self.edge((-1, len(history.pages))), self.step):
             if not current() or rows_needed <= 0 or limit <= 0:
                 break
@@ -199,16 +217,16 @@ async def walk(self, history, current, target, selected, start, rows_needed, lim
             limit -= page.stop - page.start
         return rows_needed, limit
 
-    def eviction_order(self, history):
+    def eviction_order(self, history: TranscriptSourcePreparation) -> tuple[PageEnd, PageEnd]:
         return OLDER_END, NEWER_END
 
-    def can_evict(self, overlay_visible) -> bool:
+    def can_evict(self, overlay_visible: bool) -> bool:
         return True
 
-    def leave_tail(self, window) -> None:
+    def leave_tail(self, window: HistoryWindow) -> None:
         pass
 
-    def trim(self, view, count: int) -> None:
+    def trim(self, view: TranscriptPageView, count: int) -> None:
         page = view.prepared
         page.set_fragment_range(slice(page.start + self.edge((count, 0)), page.stop - self.edge((0, count))))
         view.range_changed()
@@ -219,30 +237,30 @@ class OlderEnd(PageEnd):
     pop = staticmethod(deque.popleft)
     append = staticmethod(deque.appendleft)
 
-    async def load_page(self, loader, cursor, through):
+    async def load_page(self, loader, cursor, through) -> TranscriptPage:
         return await through.read_page(loader, before=cursor)
 
-    def wire_page(self, source, cursor, through):
+    def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence) -> MessagePage:
         return source.conversation.page(source, PageBounds(
             cursor.older(),
             source.tail_limit if cursor == through else source.page_limit))
 
-    def advance_empty(self, edge, staged) -> None:
+    def advance_empty(self, edge: PreparedTranscriptPage, staged: PreparedTranscriptPage) -> None:
         edge.page = replace(edge.page, before=staged.page.before, has_older=staged.page.has_older)
 
-    def edge(self, pages):
+    def edge[Item](self, pages: Sequence[Item]) -> Item:
         return pages[0]
 
-    async def mount(self, history, view, pages) -> None:
+    async def mount(self, history: LineHistory, view: TranscriptPageView, pages: deque[TranscriptPageView]) -> None:
         await history.mount(view, before=self.edge(pages) if pages else history.newer)
 
-    def eviction_order(self, history):
+    def eviction_order(self, history: TranscriptSourcePreparation) -> tuple[PageEnd, PageEnd]:
         return super().eviction_order(history) if history.follows_source_tail else (NEWER_END, self)
 
-    def can_evict(self, overlay_visible) -> bool:
+    def can_evict(self, overlay_visible: bool) -> bool:
         return not overlay_visible
 
-    def leave_tail(self, window) -> None:
+    def leave_tail(self, window: HistoryWindow) -> None:
         window.release_anchor()
 
 
@@ -255,31 +273,33 @@ class NewerEnd(PageEnd):
     def has_more(self, through, page):
         return super().has_more(through, page) or page.after != through
 
-    async def load_page(self, loader, cursor, through):
+    async def load_page(self, loader, cursor, through) -> TranscriptPage:
         return await through.read_page(loader, after=cursor)
 
-    def wire_page(self, source, cursor, through):
+    def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence) -> MessagePage:
         page = source.conversation.page(source, PageBounds(cursor.newer(), source.page_limit))
         kept = tuple(message for message in page.messages if message.view_key[0] or message.seq <= through.cursor.sequence)
         return replace(page, messages=kept, has_newer=False) if len(kept) < len(page.messages) else page
 
-    async def project(self, source, initial, prepared):
+    async def project(
+        self, source: ProjectedTranscriptSource, initial: PreparedTranscriptPage, prepared: PreparedTranscriptPage,
+    ):
         if not prepared.fragments and not prepared.page.has_newer:
             boundary = await source.boundary()
             return replace(boundary, page=replace(boundary.page, before=initial.page.before,
                                                  has_older=initial.page.has_older))
         return await super().project(source, initial, prepared)
 
-    def advance_empty(self, edge, staged) -> None:
+    def advance_empty(self, edge: PreparedTranscriptPage, staged: PreparedTranscriptPage) -> None:
         edge.page = replace(edge.page, after=staged.page.after, has_newer=staged.page.has_newer)
 
-    def fresh(self, history):
+    def fresh(self, history: LineHistory):
         return history.drawn_tail_followed
 
-    def edge(self, pages):
+    def edge[Item](self, pages: Sequence[Item]) -> Item:
         return pages[-1]
 
-    async def mount(self, history, view, pages) -> None:
+    async def mount(self, history: LineHistory, view: TranscriptPageView, pages: deque[TranscriptPageView]) -> None:
         await history.mount(view, before=history.newer)
 
 
@@ -305,18 +325,20 @@ def rows(self, horizon: float) -> float: ...
     def moved(self, travel: float, elapsed: float) -> "PreparationDemand":
         return MovingPreparation(travel / elapsed)
 
-    def accepts_result(self, end) -> bool:
+    def accepts_result(self, end: PageEnd) -> bool:
         """A running read may finish after input settles; it starts no next read."""
         return True
 
-    def read_rows(self, preparation: "DirectionalPreparation", history, end, runway: Runway) -> int:
+    def read_rows(
+        self, preparation: "DirectionalPreparation", history: LineHistory, end: PageEnd, runway: Runway,
+    ) -> int:
         """Stationary input never requests another source page."""
         return 0
 
     def source_read(self) -> None:
         """Only a bounded destination counts the pages its request read."""
 
-    def prepare_ahead(self, preparation, history, page, end) -> None:
+    def prepare_ahead(self, preparation: DirectionalPreparation, history: LineHistory, page, end: PageEnd) -> None:
         """Only measured movement speculates beyond its current destination."""
 
     def destination_painted(self) -> "PreparationDemand":
@@ -343,16 +365,16 @@ def moved(self, travel: float, elapsed: float) -> PreparationDemand:
     def rows(self, horizon: float) -> float:
         return self.velocity * horizon
 
-    def accepts_result(self, end) -> bool:
+    def accepts_result(self, end: PageEnd) -> bool:
         return end.accepts_velocity(self.velocity)
 
-    def read_rows(self, preparation, history, end, runway: Runway) -> int:
+    def read_rows(self, preparation: DirectionalPreparation, history: LineHistory, end: PageEnd, runway: Runway) -> int:
         if not self.accepts_result(end):
             return 0
         ahead = preparation.ahead_rows(preparation.window.outer_size.height)
         return max(0, ceil(ahead) - end.rows_toward(runway))
 
-    def prepare_ahead(self, preparation, history, page, end) -> None:
+    def prepare_ahead(self, preparation: DirectionalPreparation, history, page, end: PageEnd) -> None:
         if self.accepts_result(end):
             # The existing runtime owns the single decoded/prepared result.
             # This native worker only waits for that original bounded work;
@@ -375,11 +397,11 @@ def moved(self, travel: float, elapsed: float) -> PreparationDemand:
         # An explicit reversal immediately replaces the burst.
         return super().moved(travel, elapsed) if travel < 0 else self
 
-    def accepts_result(self, end) -> bool:
+    def accepts_result(self, end: PageEnd) -> bool:
         # A newer destination supersedes older work through its source snapshot.
         return end.accepts_destination
 
-    def read_rows(self, preparation, history, end, runway: Runway) -> int:
+    def read_rows(self, preparation: DirectionalPreparation, history: LineHistory, end: PageEnd, runway: Runway) -> int:
         if not self.accepts_result(end):
             return 0
         edge = end.edge(history.pages).prepared
@@ -403,7 +425,7 @@ class DirectionalPreparation:
     The presentation budget bounds how much of that prediction can be retained.
     """
 
-    def __init__(self, window) -> None:
+    def __init__(self, window: HistoryWindow) -> None:
         self.window = window
         self.position = 0.0
         self.sampled_at = monotonic()
@@ -497,13 +519,13 @@ def destination_batch(self, budget: PresentationBudget, viewport_rows: int) -> i
         # destination items may exceed the nominal working-set item budget.
         return max(budget.item_limit(0), viewport_rows)
 
-    def read_rows(self, history, end: PageEnd) -> int:
+    def read_rows(self, history: LineHistory, end: PageEnd) -> int:
         if not end.visible_more(history):
             return 0
         runway = self.runway(history)
         return 0 if runway is None else self.demand.read_rows(self, history, end, runway)
 
-    def runway(self, history) -> Runway | None:
+    def runway(self, history: LineHistory) -> Runway | None:
         """Published rows above and below this viewport; None while the history is off screen."""
         window = self.window
         visible = window.screen._compositor.visible_widgets
@@ -513,7 +535,7 @@ def runway(self, history) -> Runway | None:
         region, frame = geometry[0], viewport[0]
         return Runway(frame.y - region.y, region.bottom - frame.bottom)
 
-    def prepare_source(self, history) -> None:
+    def prepare_source(self, history: LineHistory) -> None:
         """Request the next page from what the screen shows now, not from window state."""
         window = self.window
         if (not window.screen.is_active or not history.selected_categories
diff --git a/src/toad/widgets/session_details.py b/src/toad/widgets/session_details.py
index 752eabd7e..27f8635ac 100644
--- a/src/toad/widgets/session_details.py
+++ b/src/toad/widgets/session_details.py
@@ -30,7 +30,7 @@
     from toad.widgets.conversation import Conversation
 
 
-class SessionDetails(TonePaint, Collapsible, ObservedThreadActivity, MroProjection):
+class SessionDetails(TonePaint, Collapsible, ObservedThreadActivity, MroProjection[type[LineRole]]):
     """The observed-thread capability and disclosure share one status view.
 
     Core owns status for an observed thread, native or CLI. A standalone ACP
@@ -90,7 +90,7 @@ def disconnected_role(self, status) -> type[LineRole]:
 
     @property
     def title_role(self) -> type[LineRole]:
-        return self.dispatch_sync(
+        return self.project_value(
             self.history.status.present_status(self.status_source().status)
         )
 
diff --git a/src/toad/widgets/sidebar_tree.py b/src/toad/widgets/sidebar_tree.py
index ef0312e45..027cd2133 100644
--- a/src/toad/widgets/sidebar_tree.py
+++ b/src/toad/widgets/sidebar_tree.py
@@ -5,7 +5,7 @@
 import asyncio
 from abc import ABC, abstractmethod
 from collections import Counter
-from collections.abc import Iterable
+from collections.abc import Iterable, Mapping
 from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import wraps
@@ -14,9 +14,15 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from textual.events import Click
+    from textual.layout import WidgetPlacement
+
     from toad.app import ToadApp
     from toad.comms_root import FailedObservation, Observing
-    from toad.widgets.comms_sidebar import CommsRow
+    from toad.navigation_target import NavigationTarget
+    from toad.widgets.channel_activity import ChannelActivity
+    from toad.widgets.comms_sidebar import CommsRow, CommsSidebar
+    from toad.widgets.thread_comms import RelationshipRows
 
 from agent_comms.display_order import ThreadSort
 from agent_comms.mro_dispatch import handles
@@ -85,7 +91,7 @@ def primary_activation(self):
     def action_toggle(self) -> None:
         self.query_ancestor(SidebarGroup).toggle_members()
 
-    def on_click(self, event) -> None:
+    def on_click(self, event: Click) -> None:
         if event.button == 1:
             event.stop()
             self.action_toggle()
@@ -95,7 +101,7 @@ class SidebarReaderLayout(VerticalLayout):
     """Native flow, preserving the viewport's reader before placements change."""
 
     @height_dependency(FLOW_HEIGHT)
-    def arrange(self, parent, children, size: Size, greedy: bool=True):
+    def arrange(self, parent: Widget, children: list[Widget], size: Size, greedy: bool=True):
         placements = super().arrange(parent, children, size, greedy)
         parent.query_ancestor(TargetTree).preserve_reader(placements)
         return placements
@@ -172,7 +178,7 @@ def create_row(self, key: KeyT):
     def update_row(self, key: KeyT, row) -> None:
         pass
 
-    def retire_row(self, row) -> None:
+    def retire_row(self, row: Widget) -> None:
         row.remove()
 
     def rows_changed(self) -> None:
@@ -181,7 +187,7 @@ def rows_changed(self) -> None:
     def place_row(self, row, index: int) -> None:
         """Publish list-position facts only for a new or moved row."""
 
-    def reconcile_keys(self, keys, *, changed: frozenset[KeyT]):
+    def reconcile_keys(self, keys: Iterable[KeyT], *, changed: frozenset[KeyT]):
         rows = self.rows
         retained = set(rows.values())
         previous = {
@@ -250,8 +256,8 @@ def __init__(
         row,
         *,
         expanded: bool,
-        controls=(),
-        disclosure_type=SidebarDisclosure,
+        controls: Iterable[Widget]=(),
+        disclosure_type: type[SidebarDisclosure]=SidebarDisclosure,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -309,7 +315,7 @@ def navigation_rows(self):
         """Native-order destinations, excluding explanatory placeholder widgets."""
         return tuple(row for row in self.visible_rows if row.is_navigation_row())
 
-    def rows_for_target(self, target):
+    def rows_for_target(self, target: str):
         if not self.accepts_members():
             return ()
         header = (
@@ -350,7 +356,7 @@ def create_row(self, key: str):
         spare = self.reconciliation_tree.row_reconciliation.take(key, self)
         return spare if spare is not None else self.new_row(key)
 
-    def update_row(self, key: str, row) -> None:
+    def update_row(self, key: str, row: RowT) -> None:
         self.show_member(key, row)
 
     def place_row(self, row: RowT, index: int) -> None:
@@ -416,39 +422,39 @@ async def _mount_pending_rows(self) -> None:
 class SidebarChange:
     """The changed model rows choose which native rows need work."""
 
-    def apply_comms(self, sidebar, rows) -> None:
+    def apply_comms(self, sidebar: CommsSidebar, rows: Mapping[str, ChannelRowModel]) -> None:
         raise NotImplementedError
 
-    def apply_welcome(self, channel_list, thread_list) -> None:
+    def apply_welcome(self, channel_list: SidebarGroup, thread_list: SidebarGroup) -> None:
         raise NotImplementedError
 
-    def apply_relationships(self, groups) -> None:
+    def apply_relationships(self, groups: Iterable[RelationshipRows]) -> None:
         raise NotImplementedError
 
-    def apply_activity(self, view) -> None:
+    def apply_activity(self, view: ChannelActivity) -> None:
         raise NotImplementedError
 
-    def apply_navigation(self, sidebar) -> None:
+    def apply_navigation(self, sidebar: CommsSidebar) -> None:
         sidebar.navigation.apply()
 
 
 class AllSidebarRows(SidebarChange):
-    def apply_comms(self, sidebar, rows) -> None:
+    def apply_comms(self, sidebar: CommsSidebar, rows: Mapping[str, ChannelRowModel]) -> None:
         sidebar.open_views = sidebar.find_open_views()
         with sidebar.reconcile_members():
             for name, group in sidebar.groups.items():
                 group.present(rows[name])
                 group.sync_members(frozenset(group.member_keys))
 
-    def apply_welcome(self, channel_list, thread_list) -> None:
+    def apply_welcome(self, channel_list: SidebarGroup, thread_list: SidebarGroup) -> None:
         channel_list.sync_members(frozenset(channel_list.member_keys))
         thread_list.sync_members(frozenset(thread_list.member_keys))
 
-    def apply_relationships(self, groups) -> None:
+    def apply_relationships(self, groups: Iterable[RelationshipRows]) -> None:
         for group in groups:
             group.sync_members(frozenset(group.member_keys))
 
-    def apply_activity(self, view) -> None:
+    def apply_activity(self, view: ChannelActivity) -> None:
         view.call_next(view.sync_members, frozenset(view.member_keys))
 
 
@@ -456,7 +462,7 @@ def apply_activity(self, view) -> None:
 class ChangedChannels(SidebarChange):
     names: frozenset[str]
 
-    def apply_comms(self, sidebar, rows) -> None:
+    def apply_comms(self, sidebar: CommsSidebar, rows: Mapping[str, ChannelRowModel]) -> None:
         with sidebar.reconcile_members():
             for name in self.names:
                 group = sidebar.groups.get(name)
@@ -464,15 +470,15 @@ def apply_comms(self, sidebar, rows) -> None:
                     group.present(rows[name])
                     group.sync_members(frozenset(group.member_keys))
 
-    def apply_welcome(self, channel_list, thread_list) -> None:
+    def apply_welcome(self, channel_list: SidebarGroup, thread_list: SidebarGroup) -> None:
         channel_list.sync_members(self.names)
         # A changed channel can change the welcome thread list's order.
         thread_list.sync_members(frozenset())
 
-    def apply_relationships(self, groups) -> None:
+    def apply_relationships(self, groups: Iterable[RelationshipRows]) -> None:
         pass
 
-    def apply_activity(self, view) -> None:
+    def apply_activity(self, view: ChannelActivity) -> None:
         references = view.model_state.shown.observation.require_snapshot().references
         if view.view.conversation.destination(references).name in self.names:
             view.call_next(view.sync_members, frozenset(view.member_keys))
@@ -482,25 +488,25 @@ def apply_activity(self, view) -> None:
 class ChangedThreads(SidebarChange):
     names: frozenset[str]
 
-    def apply_comms(self, sidebar, rows) -> None:
+    def apply_comms(self, sidebar: CommsSidebar, rows: Mapping[str, ChannelRowModel]) -> None:
         for group in sidebar.groups.values():
             group.repaint_members(self.names)
 
-    def apply_welcome(self, channel_list, thread_list) -> None:
+    def apply_welcome(self, channel_list: SidebarGroup, thread_list: SidebarGroup) -> None:
         thread_list.sync_members(self.names)
 
-    def apply_relationships(self, groups) -> None:
+    def apply_relationships(self, groups: Iterable[RelationshipRows]) -> None:
         for group in groups:
             changed = group.keys_for_threads(self.names)
             if changed:
                 group.sync_members(changed=changed)
 
-    def apply_activity(self, view) -> None:
+    def apply_activity(self, view: ChannelActivity) -> None:
         changed = view.changed_keys(self.names)
         if changed:
             view.call_next(view.sync_members, changed)
 
-    def apply_navigation(self, sidebar) -> None:
+    def apply_navigation(self, sidebar: CommsSidebar) -> None:
         pass
 
 
@@ -545,7 +551,7 @@ def present(self, view: SidebarModelView):
 class Detached(SidebarModelState):
     """No opened service or model belongs to this view."""
 
-    def shows_child(self, view, child: Widget) -> bool:
+    def shows_child(self, view: SidebarModelView, child: Widget) -> bool:
         return False
 
 
@@ -553,10 +559,10 @@ def shows_child(self, view, child: Widget) -> bool:
 class FailedSidebarState(SidebarModelState):
     observed: FailedObservation
 
-    def shows_child(self, view, child: Widget) -> bool:
+    def shows_child(self, view: SidebarModelView, child: Widget) -> bool:
         return child is view.unavailable
 
-    def present(self, view):
+    def present(self, view: SidebarModelView):
         view.unavailable.update(f"Comms unavailable\n{self.observed.error}")
         view.app.coordination_access.show_sidebar(view, view.accepts_observation())
         return self
@@ -566,7 +572,7 @@ def present(self, view):
 class BoundSidebar(SidebarModelState):
     observed: Observing
 
-    def shows_child(self, view, child: Widget) -> bool:
+    def shows_child(self, view: SidebarModelView, child: Widget) -> bool:
         return (child is view.unavailable) == self.observed.sidebar.observation.shows_status()
 
     def root(self):
@@ -578,11 +584,11 @@ def serves(self, path) -> bool:
     def observes(self, observed: Observing) -> bool:
         return self.observed is observed
 
-    def unsubscribe(self, view) -> None:
+    def unsubscribe(self, view: SidebarModelView) -> None:
         self.observed.sidebar.channels.unsubscribe(view.channels_changed)
         self.observed.sidebar.threads.unsubscribe(view.threads_changed)
 
-    def present(self, view):
+    def present(self, view: SidebarModelView):
         view.unavailable.update(self.observed.sidebar.observation.status_text())
         shown = view.accepts_observation()
         view.app.coordination_access.show_sidebar(view, shown)
@@ -598,7 +604,7 @@ class Current(BoundSidebar):
     def shows_model(self) -> bool:
         return True
 
-    def apply(self, view, changes) -> None:
+    def apply(self, view: SidebarModelView, changes) -> None:
         if changes:
             view.sync(**changes)
 
@@ -623,7 +629,7 @@ class SidebarModelView(CoreEventReceiver, Widget):
     while this one is hidden are applied in one pass when it shows again.
     """
 
-    def __init__(self, *args, **kwargs) -> None:
+    def __init__(self, *args: Widget, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self.unavailable = Static("", markup=False)
 
@@ -743,7 +749,7 @@ class TargetTree(Vertical):
     def matches(self, key: str) -> bool:
         return self.search in key.casefold()
 
-    def preserve_reader(self, placements) -> None:
+    def preserve_reader(self, placements: list[WidgetPlacement]) -> None:
         """A sidebar viewport may retain its reader across a native relayout."""
 
     _member_reconciliation = None
@@ -795,7 +801,7 @@ def accepts_selection(self) -> bool:
     def row_for(self, selection):
         raise NotImplementedError
 
-    def row_selected(self, row) -> bool:
+    def row_selected(self, row: CommsRow) -> bool:
         return self.selection_for(row) in self.selection_state.selected_targets
 
     @contextmanager
@@ -810,7 +816,7 @@ def selection_change(self):
                 if (row := self.row_for(identity)) is not None:
                     row.refresh_paint()
 
-    def remember_row(self, row) -> None:
+    def remember_row(self, row: CommsRow) -> None:
         """Navigation establishes a range anchor; it is not bulk selection."""
         rows = self.rows
         if self.accepts_selection() and row in rows:
@@ -819,7 +825,7 @@ def remember_row(self, row) -> None:
                 state.selected = self.selection_for(row)
                 state.selected_targets = ()
 
-    def pointer_select(self, row, *, control: bool=False, shift: bool=False, menu: bool=False) -> None:
+    def pointer_select(self, row: CommsRow, *, control: bool=False, shift: bool=False, menu: bool=False) -> None:
         if not self.accepts_selection():
             return
         rows = self.rows
@@ -864,12 +870,12 @@ def apply_selection(self) -> None:
     def navigation_root(self):
         raise NotImplementedError
 
-    def row_current(self, row) -> bool:
+    def row_current(self, row: CommsRow) -> bool:
         return self.app.workspace_sessions.visible_target(self.navigation_root).matches(
             row.target
         )
 
-    def rows_for_target(self, target):
+    def rows_for_target(self, target: NavigationTarget):
         return tuple(
             row for group in self.navigation_groups for row in target.rows_in(group)
         )
diff --git a/src/toad/widgets/transcript_fragments.py b/src/toad/widgets/transcript_fragments.py
index 97984ecc2..7722aefd7 100644
--- a/src/toad/widgets/transcript_fragments.py
+++ b/src/toad/widgets/transcript_fragments.py
@@ -156,7 +156,7 @@ def select(first: int, stop: int, token_first: int, token_stop: int):
             yield select(pending_first, pending_stop, pending_token_first, pending_token_stop)
 
 
-class FragmentLineConsumer(MroProjection):
+class FragmentLineConsumer(MroProjection[None]):
     """Turn a fragment's events into line blocks, one handler per event family."""
 
     def __init__(self, *, show_divider: bool) -> None:
@@ -316,7 +316,7 @@ def blocks(self, shown: Shown) -> tuple[LineBlock, ...]:
         if self.starts_agent_activity and not self.continuation:
             consumer.blocks.extend(AgentActivityPresentation.group)
         for event in self.events:
-            consumer.dispatch_sync(event)
+            consumer.project_value(event)
         return tuple(consumer.blocks)
 
     @cached_property
@@ -495,7 +495,7 @@ def blocks(self, shown):
                            for event in self.events))
 
 
-class TranscriptFragmentConsumer(MroProjection):
+class TranscriptFragmentConsumer(MroProjection[None]):
     def __init__(self) -> None:
         self.fragments: list[TranscriptFragment | None] = []
         self.tools: dict[str, tuple[int, list[ToolTranscript], bool]] = {}
@@ -554,7 +554,7 @@ def transcript_fragments(
 
     consumer = TranscriptFragmentConsumer()
     for event in events:
-        consumer.dispatch_sync(event)
+        consumer.project_value(event)
     reports = list(consumer.reported_messages.values())
     tools = {}
     for index, pending, starts_activity in consumer.tools.values():
@@ -568,7 +568,7 @@ def transcript_fragments(
     canonical = {}
     for index, report in consumer.reported_messages.items():
         start = len(builder.fragments)
-        builder.dispatch_sync(resolved[id(report)])
+        builder.project_value(resolved[id(report)])
         canonical[index] = tuple(builder.fragments[start:])
     fragments = []
     for index, source in enumerate(consumer.fragments):
diff --git a/tests/app/test_context_reader_session_revision.py b/tests/app/test_context_reader_session_revision.py
new file mode 100644
index 000000000..676202c40
--- /dev/null
+++ b/tests/app/test_context_reader_session_revision.py
@@ -0,0 +1,27 @@
+"""Context inspection observes the saved session and its owner proof on a real wire."""
+
+from agent_comms.session_revision import SessionRevision
+
+from toad.core.context_inspection import ContextInspection
+
+
+def test_context_reader_observes_the_session_and_its_owner_proof(workspace):
+    from agent_comms.fresh_private_session import create_fresh_private_session
+    from agent_comms.native_session_files import NativeSessionFiles
+    from agent_comms.session_file import SavedSessionFile
+
+    identity = create_fresh_private_session(workspace.comms.root, worktree=workspace.project)
+    registry = workspace.comms.registry
+    registry.declare(registry.new_thread(
+        "context-reader", frozenset(), str(workspace.project), session_file=SavedSessionFile(identity.session_file)))
+    thread = registry.require("context-reader")
+    files = NativeSessionFiles.for_identity(
+        identity, root=workspace.comms.root, owner=thread.incarnation.require_recorded())
+
+    before = ContextInspection.read(workspace.comms, thread.name)
+    assert before.source == SessionRevision.observe(files)
+    files.input_proof.parent.mkdir(parents=True, exist_ok=True)
+    files.input_proof.touch()
+    after = ContextInspection.read(workspace.comms, thread.name)
+    assert after.source == SessionRevision.observe(files)
+    assert after.source != before.source
diff --git a/tests/model/test_recorded_transcript_reader.py b/tests/app/test_recorded_transcript_reader.py
similarity index 95%
rename from tests/model/test_recorded_transcript_reader.py
rename to tests/app/test_recorded_transcript_reader.py
index 84205a0e8..5d8b5dc98 100644
--- a/tests/model/test_recorded_transcript_reader.py
+++ b/tests/app/test_recorded_transcript_reader.py
@@ -12,6 +12,7 @@
 from support.roots import installed_wire
 from support.journals import assistant_entry, write_journal
 from toad.acp.comms_updates import OwnerSnapshotConsumer
+from toad.session_scope import KnownTranscript
 
 
 @pytest.fixture
@@ -31,7 +32,7 @@ async def test_recorded_snapshot_stays_immutable_when_original_grows_and_rejects
 
     async with workspace.app() as ui:
         reader = ui.app.coordination_access.transcripts
-        update = await reader.snapshot(str(live.root), "reader", historical_source=source.key)
+        update = await reader.snapshot(KnownTranscript(str(live.root), "reader"), historical_source=source.key)
         assert isinstance(update.identity, RecordedTranscriptReadIdentity)
         assert decode_updates(encode_updates(update)) == (update,)
         with native.open("a") as output:
diff --git a/tests/model/test_projection_contract.py b/tests/model/test_projection_contract.py
new file mode 100644
index 000000000..8814de1f9
--- /dev/null
+++ b/tests/model/test_projection_contract.py
@@ -0,0 +1,95 @@
+"""A nominal projection returns one leaf's result without replacing its input."""
+
+from dataclasses import dataclass
+
+import pytest
+from agent_comms.mro_dispatch import handles
+
+from toad.core.projection import MroProjection
+
+
+@dataclass(frozen=True)
+class Value:
+    text: str
+
+
+class SpecificValue(Value):
+    pass
+
+
+@dataclass(frozen=True)
+class Rendering:
+    text: str
+
+
+@pytest.mark.parametrize("override", [False, True])
+def test_projection_returns_the_specific_result_and_forwards_context(override: bool) -> None:
+    selected: list[str] = []
+
+    class TextProjection(MroProjection[Rendering]):
+        @handles(Value)
+        def base(self, value: Value, prefix: str, *, suffix: str) -> Rendering:
+            selected.append("base")
+            return Rendering("base")
+
+        @handles(SpecificValue)
+        def specific(self, value: SpecificValue, prefix: str, *, suffix: str) -> Rendering:
+            selected.append("specific")
+            return Rendering(f"{prefix}{value.text}{suffix}")
+
+    class OverriddenProjection(TextProjection):
+        @handles(SpecificValue)
+        def specific(self, value: SpecificValue, prefix: str, *, suffix: str) -> Rendering:
+            selected.append("override")
+            return Rendering(f"{prefix}override:{value.text}{suffix}")
+
+    projection = OverriddenProjection() if override else TextProjection()
+    value = SpecificValue("original")
+    result = projection.project_value(value, "[", suffix="]")
+
+    assert result == Rendering("[override:original]" if override else "[original]")
+    assert selected == ["override" if override else "specific"]
+    assert value == SpecificValue("original")
+
+
+def test_projection_preserves_a_none_result_from_a_side_effect_handler() -> None:
+    selected: list[str] = []
+
+    class RecordingProjection(MroProjection[None]):
+        @handles(Value)
+        def record(self, value: Value) -> None:
+            selected.append(value.text)
+
+    assert RecordingProjection().project_value(Value("recorded")) is None
+    assert selected == ["recorded"]
+
+
+def test_projection_without_a_declared_handler_raises_stop_iteration() -> None:
+    with pytest.raises(StopIteration):
+        MroProjection[Rendering]().project_value(Value("unhandled"))
+
+
+def test_unavailable_owner_projection_does_not_supply_a_load_plan():
+    from agent_comms.thread_presentation import UnavailableThreadOwnerBinding
+    from toad.owner_preparation import ExistingOwnerPlan
+
+    plan = ExistingOwnerPlan.from_binding(UnavailableThreadOwnerBinding())
+    assert not plan.available
+    with pytest.raises(ValueError, match="live owner binding"):
+        plan.require_plan()
+
+
+def test_live_owner_projection_retains_the_exact_load_binding():
+    from agent_comms.child_process import ProcessIdentity
+    from agent_comms.session_load import ExistingSessionLoadPlan
+    from agent_comms.thread_identity import OwnerIdentity, RecordedThreadIncarnation
+    from agent_comms.thread_presentation import LiveThreadOwnerBinding
+    from toad.owner_preparation import ExistingOwnerPlan
+
+    binding = LiveThreadOwnerBinding(
+        OwnerIdentity(RecordedThreadIncarnation("projection-owner", 123.0), 2),
+        ProcessIdentity(1, 1),
+    )
+    plan = ExistingOwnerPlan.from_binding(binding)
+    assert plan.available
+    assert plan.require_plan() == ExistingSessionLoadPlan(binding)
diff --git a/uv.lock b/uv.lock
index 44fddac66..66d22583f 100644
--- a/uv.lock
+++ b/uv.lock
@@ -21,7 +21,7 @@ wheels = [
 [[package]]
 name = "agent-comms"
 version = "0.1.0"
-source = { git = "https://github.com/OpenHCSDev/agent-comms.git?branch=main#2e51f3b67fc7ee4acf2c8ee6e34f42d6528d3a70" }
+source = { git = "https://github.com/OpenHCSDev/agent-comms.git?branch=main#7017468028ea2428ce71fe30261aa8d827aa222a" }
 dependencies = [
     { name = "agent-client-protocol" },
     { name = "aiohttp" },
@@ -798,12 +798,8 @@ wheels = [
 
 [[package]]
 name = "metaclass-registry"
-version = "0.4.0"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/ce/a7/480cac5c73f0b5b50359ce2b5248d816b331b8d104e5ffb0741de669c020/metaclass_registry-0.4.0.tar.gz", hash = "sha256:b66fce0cbeb6460d7835103d4206dd11be4b607247db75608a843e2e57b60fd3", size = 45713, upload-time = "2026-10-12T12:11:39.549Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/40/e9/cc748c2779cea13bae6f89f7db2e0f5ba5da47c5a9380c7c0eb73adf7ce7/metaclass_registry-0.4.0-py3-none-any.whl", hash = "sha256:9b68043883edafd5da04d171f3aeeb14e5fa27094ffcb154d959832091a91dcd", size = 33389, upload-time = "2026-10-12T12:11:38.121Z" },
-]
+version = "0.4.1"
+source = { git = "https://github.com/OpenHCSDev/metaclass-registry?branch=main#fb7c871a78c3087aa65c3a7199a1068cff858f33" }
 
 [[package]]
 name = "msgpack"