Fix local identity routing and typed page sources

src/toad/screens/main.py

@@ -233,11 +233,10 @@ def read_visible(self, screen: MainScreen) -> PaintedRead:
         access = app.coordination_access
         identity = screen.thread_identity
         project = screen.project_path
-        source_root = identity.wire_root()
         worktree = app.project_dir
 
         def acknowledge(cursor) -> None:
-            selected = access.selected(source_root)
+            selected = identity.selected_route(access)
             comms = access.require(selected).service
 
             def mark() -> None:

src/toad/session_scope.py

@@ -406,7 +406,7 @@ def require(self) -> ProjectSessionOrigin:
 
     def resolve_thread(self) -> str:
         identity = self.thread_identity
-        selected = self.app.coordination_access.selected(identity.wire_root())
+        selected = identity.selected_route(self.app.coordination_access)
         service = self.app.coordination_access.require(selected).service
         return identity.resolve_thread(service.registry, self.project_root)
 

src/toad/thread_identity.py

@@ -9,8 +9,10 @@
 from typing import TYPE_CHECKING
 
 from agent_comms.errors import UnregisteredThreadError
+from agent_comms.route_selection import RouteSelection
 from agent_comms.thread_identity import ThreadIncarnation, UnrecordedThreadIncarnation
 
+from toad.comms_root import CoordinationAccess
 from toad.format_path import format_path
 from toad.session_scope import KnownTranscript, NoTranscript, TranscriptScope
 from toad.transcript_state import UnboundTranscript, UnreadTranscript
@@ -45,6 +47,10 @@ def with_session(self, session_id: str):
     def wire_root(self) -> str | None:
         return None
 
+    def selected_route(self, access: CoordinationAccess) -> RouteSelection:
+        """An unbound identity resolves within the app's own route."""
+        return access.selection
+
     def transcript_scope(self) -> TranscriptScope:
         return NoTranscript()
 
@@ -157,6 +163,10 @@ def bind_history(self, histories) -> None:
     def wire_root(self) -> str:
         return self.root
 
+    def selected_route(self, access: CoordinationAccess) -> RouteSelection:
+        """A managed identity requires its recorded wire source."""
+        return access.selected(self.root)
+
     def transcript_scope(self) -> KnownTranscript:
         return KnownTranscript(self.root, self.thread_name())
 

src/toad/transcript_preparation.py

@@ -449,9 +449,8 @@ def __init__(self, request: PageRequest, through: TranscriptCursor | None, page:
 
 
 @dataclass(frozen=True)
-class TranscriptPageWork[Cursor = TranscriptCursor | None](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]):
+class PreparedPageWork[Cursor = TranscriptCursor | None](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]):
     scope: PreparationScope
-    loader: Callable[..., Awaitable[TranscriptPage]]
     through: Cursor
     request: PageRequest[Cursor]
     lane = WorkLane.MODEL
@@ -464,6 +463,11 @@ def work_key(self) -> WorkKey:
     def result_size(self, result: PreparedTranscriptPage) -> int:
         return result.retained_bytes
 
+
+@dataclass(frozen=True)
+class TranscriptPageWork[Cursor](PreparedPageWork[Cursor]):
+    loader: Callable[..., Awaitable[TranscriptPage]]
+
     async def execute(
         self: TranscriptPageWork[TranscriptCursor | None], runtime: PreparationRuntime, key: WorkKey,
     ) -> PreparedResult[PreparedTranscriptPage]:
@@ -495,56 +499,64 @@ class ReplaceablePageSource(PreparedPageSource):
     def replacement(self, through) -> ReplaceablePageSource: ...
 
 
-class SavedHistory(ReplaceablePageSource):
-    """Prepared pages for a snapshot; storage and work limits belong to the runtime.
+class SavedPageSource(ReplaceablePageSource):
+    """Snapshot preparation and lifetime shared by native and wire readers.
 
-    The loader retains ownership of transcript/routing semantics and off-loop
-    I/O. CPU preparation uses the application's existing renderer. Foreground
-    consumers share in-flight requests; cancelling one waiter cannot release a
-    still-running read's slot or publish it into a closed view.
+    Each member owns its actual loader and work. Cancelling a foreground waiter
+    cannot release a still-running read's slot or publish into a closed view.
     """
 
-    # The work that reads and prepares one page of this kind of source.
-    work: ClassVar[type[TranscriptPageWork]] = TranscriptPageWork
-
-    def __init__(
-        self, loader: Callable[..., Awaitable[TranscriptPage]], through: TranscriptCursor,
-        runtime: PreparationRuntime,
-    ) -> None:
+    def __init__(self, through: TranscriptCursor, runtime: PreparationRuntime) -> None:
         super().__init__()
-        self.loader, self.through, self.runtime = loader, through, runtime
+        self.through, self.runtime = through, runtime
         self.scope = PreparationScope()
 
     def prefix(self, through) -> PreparedPageSource:
-        return SavedPrefix(self, self.loader, through, self.runtime)
-
-    def replacement(self, through) -> SavedHistory:
-        return type(self)(self.loader, through, self.runtime)
+        return SavedPrefix(self, through, self.runtime)
 
-    def read_task(self, request: PageRequest) -> TranscriptPageWork:
-        return self.work(self.scope, self.loader, self.through, request)
+    @abstractmethod
+    def read_task(
+        self, request: PageRequest, scope: PreparationScope, through: TranscriptCursor,
+    ) -> ReusableWork[PreparedTranscriptPage]: ...
 
     async def _get(self, request: PageRequest) -> PreparedTranscriptPage:
-        result = await self.runtime.submit(self.read_task(request))
+        result = await self.runtime.submit(self.read_task(request, self.scope, self.through))
         return result.for_view()
 
     async def _prepare(self, request: PageRequest) -> None:
-        await self.runtime.prepare(self.read_task(request))
+        await self.runtime.prepare(self.read_task(request, self.scope, self.through))
 
 
-class SavedPrefix(SavedHistory):
-    def __init__(self, earlier: SavedHistory, loader: Callable[..., Awaitable[TranscriptPage]], through, runtime: PreparationRuntime) -> None:
-        super().__init__(loader, through, runtime)
-        self.earlier = earlier
+class SavedHistory(SavedPageSource):
+    """Native transcript pages read by the source's asynchronous loader."""
+
+    def __init__(
+        self, loader: Callable[..., Awaitable[TranscriptPage]], through: TranscriptCursor,
+        runtime: PreparationRuntime,
+    ) -> None:
+        super().__init__(through, runtime)
+        self.loader = loader
 
     def replacement(self, through) -> SavedHistory:
+        return type(self)(self.loader, through, self.runtime)
+
+    def read_task(self, request: PageRequest, scope: PreparationScope, through: TranscriptCursor) -> TranscriptPageWork:
+        return TranscriptPageWork(scope=scope, loader=self.loader, through=through, request=request)
+
+
+class SavedPrefix(SavedPageSource):
+    def __init__(self, earlier: SavedPageSource, through, runtime: PreparationRuntime) -> None:
+        super().__init__(through, runtime)
+        self.earlier = earlier
+
+    def replacement(self, through) -> ReplaceablePageSource:
         # A replacement reads the complete source; it must not delegate older
         # reads to the superseded prefix's preparation scope.
         return self.earlier.replacement(through)
 
-    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)
+    def read_task(self, request: PageRequest, scope: PreparationScope, through: TranscriptCursor):
+        # The original source supplies its native/wire work in this prefix's scope.
+        return self.earlier.read_task(request, scope, through)
 
     async def _get(self, request: PageRequest):
         return await request.end.edge((self.earlier.get, super()._get))(request)

src/toad/wire_history.py

@@ -38,11 +38,13 @@
 from toad.message_rows import MessageRow, WireMessageStyle
 from toad.transcript_preparation import (
     PageRequest,
+    PreparationRuntime,
+    PreparationScope,
     PreparedTranscriptPage,
     ReplaceablePageSource,
-    SavedHistory,
+    PreparedPageWork,
+    SavedPageSource,
     SourceObservation,
-    TranscriptPageWork,
     Unchanged,
 )
 from toad.transcript_state import (
@@ -386,8 +388,10 @@ async def apply(self, history, source) -> bool:
 
 
 @dataclass(frozen=True)
-class WirePageWork(TranscriptPageWork[WireCut]):
-    """Read one wire page off the UI thread; `loader` is the WireSource."""
+class WirePageWork(PreparedPageWork[WireCut]):
+    """Read one wire page off the UI thread from its typed source."""
+
+    loader: WireSource
 
     @property
     def retain_result(self) -> bool:
@@ -402,15 +406,17 @@ async def execute(self, runtime, key):
         return await runtime.run_thread(self.finish_result, key, prepared)
 
 
-class WirePageBuffer(SavedHistory):
+class WirePageBuffer(SavedPageSource):
     """The line history's page reader over a wire source."""
 
-    work = WirePageWork
-
-    def __init__(self, *args, revision: ConversationObservation, **kwargs):
-        super().__init__(*args, **kwargs)
+    def __init__(self, loader: WireSource, through: WireCut, runtime: PreparationRuntime, *, revision: ConversationObservation):
+        super().__init__(through, runtime)
+        self.loader = loader
         self.revision = revision
 
+    def read_task(self, request: PageRequest, scope: PreparationScope, through: WireCut) -> WirePageWork:
+        return WirePageWork(scope=scope, loader=self.loader, through=through, request=request)
+
     def replacement(self, through) -> WirePageBuffer:
         return type(self)(self.loader, through, self.runtime, revision=self.revision)