P954: pair nominal relationship presence and strict DM consumers with Core

src/toad/conversation_kind.py

@@ -6,6 +6,8 @@
 
 import asyncio
 from abc import abstractmethod
+from collections.abc import Callable
+from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from functools import partial
 from pathlib import Path
@@ -78,7 +80,7 @@ def view_identity(cls, key: CommsViewKey):
         return key
 
     @abstractmethod
-    def page(self, source: WireSource, bounds: PageBounds) -> MessagePage: ...
+    def reading(self, source: WireSource) -> AbstractContextManager[Callable[[PageBounds], MessagePage]]: ...
 
     @classmethod
     @abstractmethod
@@ -182,12 +184,10 @@ def view_identity(cls, key: CommsViewKey):
 
         return ChannelViewAddress(key.root, key.conversation)
 
-    def page(self, source: WireSource, bounds: PageBounds) -> MessagePage:
-        return source.comms.views.channel_display_page(
-            self.target.name,
-            viewer=source.viewer,
-            bounds=bounds,
-        )
+    def reading(self, source: WireSource) -> AbstractContextManager[Callable[[PageBounds], MessagePage]]:
+        return source.comms.views.channel_display_view(
+            self.target.name, viewer=source.viewer,
+        ).reading(source.comms.views.presentation)
 
     @classmethod
     def resolve(cls, comms: Comms, me: str, name: str) -> ResolvedConversation:
@@ -268,12 +268,10 @@ def threads_needed(self, snapshot: RegistrySnapshot, me: str):
     def unread(self, observation: SidebarObservation, references: CurrentDisplayReferences):
         return observation.bus_unread(self.destination(references).name)
 
-    def page(self, source: WireSource, bounds: PageBounds) -> MessagePage:
-        return source.comms.views.dm_display_page(
-            self.retained_target(),
-            worktree=source.project,
-            bounds=bounds,
-        )
+    def reading(self, source: WireSource) -> AbstractContextManager[Callable[[PageBounds], MessagePage]]:
+        return source.comms.views.dm_display_view(
+            self.retained_target(), worktree=source.project,
+        ).reading(source.comms.views.presentation)
 
     @classmethod
     def resolve(cls, comms: Comms, me: str, name: str) -> ResolvedConversation:
@@ -300,8 +298,8 @@ def prompt(self, project: Checkout):
 
     async def status_text(self, comms: Comms, *, read_only: bool) -> str:
         def read():
-            references = CurrentDisplayReferences.from_registry(str(comms.root), comms.registry.snapshot())
-            return comms.configuration.answer(self.destination(references).name)
+            snapshot = comms.registry.snapshot()
+            return comms.configuration.answer(self.peer.require_current(snapshot).name)
 
         return (await asyncio.to_thread(read)).status_label
 

src/toad/transcript_preparation.py

@@ -449,9 +449,9 @@ def __init__(self, request: PageRequest, through: TranscriptCursor | None, page:
 
 
 @dataclass(frozen=True)
-class TranscriptPageWork[Cursor = TranscriptCursor | None](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]):
+class TranscriptPageWork[Cursor = TranscriptCursor | None, Loader = Callable[..., Awaitable[TranscriptPage]]](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]):
     scope: PreparationScope
-    loader: Callable[..., Awaitable[TranscriptPage]]
+    loader: Loader
     through: Cursor
     request: PageRequest[Cursor]
     lane = WorkLane.MODEL
@@ -465,7 +465,8 @@ def result_size(self, result: PreparedTranscriptPage) -> int:
         return result.retained_bytes
 
     async def execute(
-        self: TranscriptPageWork[TranscriptCursor | None], runtime: PreparationRuntime, key: WorkKey,
+        self: TranscriptPageWork[TranscriptCursor | None, Callable[..., Awaitable[TranscriptPage]]],
+        runtime: PreparationRuntime, key: WorkKey,
     ) -> PreparedResult[PreparedTranscriptPage]:
         request = self.request
         page = await request.end.load_page(self.loader, request.cursor, self.through)
@@ -495,7 +496,7 @@ class ReplaceablePageSource(PreparedPageSource):
     def replacement(self, through) -> ReplaceablePageSource: ...
 
 
-class SavedHistory(ReplaceablePageSource):
+class SavedHistory[Loader = Callable[..., Awaitable[TranscriptPage]]](ReplaceablePageSource):
     """Prepared pages for a snapshot; storage and work limits belong to the runtime.
 
     The loader retains ownership of transcript/routing semantics and off-loop
@@ -505,23 +506,26 @@ class SavedHistory(ReplaceablePageSource):
     """
 
     # The work that reads and prepares one page of this kind of source.
-    work: ClassVar[type[TranscriptPageWork]] = TranscriptPageWork
+    work: ClassVar[type[TranscriptPageWork[object, object]]] = TranscriptPageWork
 
     def __init__(
-        self, loader: Callable[..., Awaitable[TranscriptPage]], through: TranscriptCursor,
+        self, loader: Loader, through: TranscriptCursor,
         runtime: PreparationRuntime,
     ) -> None:
         super().__init__()
         self.loader, self.through, self.runtime = loader, through, runtime
-        self.scope = PreparationScope()
+        self.scope = runtime.open_scope(self.new_scope)
+
+    def new_scope(self) -> PreparationScope:
+        return PreparationScope()
 
     def prefix(self, through) -> PreparedPageSource:
-        return SavedPrefix(self, self.loader, through, self.runtime)
+        return ReusedPrefix(self, self.loader, through, self.runtime)
 
-    def replacement(self, through) -> SavedHistory:
+    def replacement(self, through) -> SavedHistory[Loader]:
         return type(self)(self.loader, through, self.runtime)
 
-    def read_task(self, request: PageRequest) -> TranscriptPageWork:
+    def read_task(self, request: PageRequest) -> TranscriptPageWork[object, object]:
         return self.work(self.scope, self.loader, self.through, request)
 
     async def _get(self, request: PageRequest) -> PreparedTranscriptPage:
@@ -532,12 +536,18 @@ async def _prepare(self, request: PageRequest) -> None:
         await self.runtime.prepare(self.read_task(request))
 
 
-class SavedPrefix(SavedHistory):
-    def __init__(self, earlier: SavedHistory, loader: Callable[..., Awaitable[TranscriptPage]], through, runtime: PreparationRuntime) -> None:
-        super().__init__(loader, through, runtime)
+class SavedPrefix[Loader = Callable[..., Awaitable[TranscriptPage]]](SavedHistory[Loader]):
+    def __init__(self, earlier: SavedHistory[Loader], loader: Loader, through, runtime: PreparationRuntime) -> None:
         self.earlier = earlier
+        super().__init__(loader, through, runtime)
+
+    def new_scope(self) -> PreparationScope:
+        return self.earlier.scope.fork()
 
-    def replacement(self, through) -> SavedHistory:
+    def prefix(self, through: TranscriptCursor) -> PreparedPageSource:
+        return type(self)(self, self.loader, through, self.runtime)
+
+    def replacement(self, through) -> SavedHistory[Loader]:
         # A replacement reads the complete source; it must not delegate older
         # reads to the superseded prefix's preparation scope.
         return self.earlier.replacement(through)
@@ -546,6 +556,10 @@ 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)
 
+
+class ReusedPrefix[Loader = Callable[..., Awaitable[TranscriptPage]]](SavedPrefix[Loader]):
+    """An immutable native prefix can reuse its original prepared older pages."""
+
     async def _get(self, request: PageRequest):
         return await request.end.edge((self.earlier.get, super()._get))(request)
 
@@ -575,7 +589,7 @@ class ClosedHistory(ReplaceablePageSource):
     def __init__(self, through: TranscriptCursor, runtime: PreparationRuntime) -> None:
         super().__init__()
         self.through, self.runtime = through, runtime
-        self.scope = PreparationScope()
+        self.scope = runtime.open_scope(PreparationScope)
 
     def uses_observation(self, observation: Observation) -> bool:
         return False

src/toad/widgets/comms_chat.py

@@ -114,7 +114,6 @@ async def initialize_view(self) -> None:
                 f"Conversation root {self._bound_root} differs from observed root {shared.root}"
             )
         comms = shared
-        await self.app.coordination_access.current_sidebar()
         self.message_history.initialize_source(WireSource(
             comms, self.conversation, self.project_path, self._me,
             IrcMessageStyle(), access=self.app.coordination_access,

src/toad/widgets/presentation_window.py

@@ -2,7 +2,7 @@
 
 from abc import ABC, abstractmethod
 from collections import deque
-from collections.abc import Iterable, Sequence
+from collections.abc import Callable, Iterable, Sequence
 from dataclasses import dataclass, fields, replace
 from functools import partial
 from itertools import zip_longest
@@ -15,8 +15,6 @@
 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
@@ -183,7 +181,8 @@ def progressed(self, page: "TranscriptPage", cursor: TranscriptCursor) -> bool:
         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: WireSource, cursor: WireCut, through: AfterSequence) -> MessagePage: ...
+    def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence,
+                  pages: Callable[[PageBounds], MessagePage]) -> MessagePage: ...
 
     def entry(self, length: int):
         return self.edge((length - 1, 0))
@@ -240,10 +239,10 @@ class OlderEnd(PageEnd):
     async def load_page(self, loader, cursor, through) -> TranscriptPage:
         return await through.read_page(loader, before=cursor)
 
-    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 wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence,
+                  pages: Callable[[PageBounds], MessagePage]) -> MessagePage:
+        return pages(PageBounds(cursor.older(),
+                                source.tail_limit if cursor == through else source.page_limit))
 
     def advance_empty(self, edge: PreparedTranscriptPage, staged: PreparedTranscriptPage) -> None:
         edge.page = replace(edge.page, before=staged.page.before, has_older=staged.page.has_older)
@@ -276,8 +275,9 @@ def has_more(self, through, page):
     async def load_page(self, loader, cursor, through) -> TranscriptPage:
         return await through.read_page(loader, after=cursor)
 
-    def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence) -> MessagePage:
-        page = source.conversation.page(source, PageBounds(cursor.newer(), source.page_limit))
+    def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence,
+                  pages: Callable[[PageBounds], MessagePage]) -> MessagePage:
+        page = pages(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
 

src/toad/widgets/thread_comms.py

@@ -10,6 +10,7 @@
 from toad.navigation_target import NavigationTarget, ThreadTarget
 
 import asyncio
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 
 from agent_comms.display_order import ThreadSort
@@ -43,10 +44,9 @@
 
 
 def entry_target(entry: RelationshipEntry) -> NavigationTarget:
-    person = entry.person
-    if person is None:
-        return NavigationTarget.named(entry.target)
-    return ThreadTarget(person.thread.name)
+    for person in entry.person.views():
+        return ThreadTarget(person.thread.name)
+    return NavigationTarget.named(entry.target)
 
 
 @dataclass
@@ -110,7 +110,9 @@ class RelationshipRow(ThreadStatusRow, actions=(CopyIdentityAction,)):
     available = True
     entry: RelationshipEntry | None = None
 
-    def present(self, entry, model) -> None:
+    def present(
+        self, entry: RelationshipEntry, rows: Mapping[str, ThreadRowModel]
+    ) -> None:
         previous, self.entry = self.entry, entry
         self.available, self.target = entry.available, entry_target(entry)
         if not entry.available:
@@ -119,36 +121,39 @@ def present(self, entry, model) -> None:
                 Content(f"? {entry.target}\n  Unavailable · Ctrl+C copies name"),
                 layout=False,
             )
-        elif entry.person is not None:
-            if (
-                model is not None
-                and model.incarnation == entry.person.thread.incarnation
-            ):
+            self._show_entry_detail(entry)
+            return
+        for person in entry.person.views():
+            model = rows.get(person.thread.name)
+            if model is not None and model.incarnation == person.thread.incarnation:
                 self.workers.cancel_group(self, "relationship-row")
                 self._show_person(entry, model)
             elif previous != entry:
                 self.run_worker(
-                    self._present_person(entry),
+                    self._present_person(entry, person),
                     group="relationship-row",
                     exclusive=True,
                 )
-        elif previous != entry:
+            return
+        if previous != entry:
             self.set_label(entry.target)
-        if entry.person is None:
-            self.tooltip = (
-                Content(f"{entry.target}\n{entry.detail}")
-                if entry.detail
-                else entry.target
-            )
+        self._show_entry_detail(entry)
+
+    def _show_entry_detail(self, entry):
+        self.tooltip = (
+            Content(f"{entry.target}\n{entry.detail}")
+            if entry.detail
+            else entry.target
+        )
 
     def _show_person(self, entry, model):
         self.show(model)
         if entry.detail:
-            self.tooltip = Content(f"{entry.target}\n{entry.detail}")
+            self._show_entry_detail(entry)
 
-    async def _present_person(self, entry):
+    async def _present_person(self, entry, person):
         model = await self.app.preparation.run_thread(
-            ThreadRowModel.of, entry.person, UnobservedUnread()
+            ThreadRowModel.of, person, UnobservedUnread()
         )
         if self.entry == entry:
             self._show_person(entry, model)
@@ -206,7 +211,8 @@ def keys_for_threads(self, names: frozenset[str]) -> frozenset[str]:
         return frozenset(
             entry.target
             for entry in self.model.entries.values()
-            if entry.person is not None and entry.person.thread.name in names
+            for person in entry.person.views()
+            if person.thread.name in names
         )
 
     def new_row(self, key: str):
@@ -216,10 +222,7 @@ def new_row(self, key: str):
     def show_member(self, key: str, row: RelationshipRow) -> None:
         entry = self.model.entries[key]
         threads = self.query_ancestor(ThreadCommsSidebar).model_state.shown.threads.rows
-        row.present(
-            entry,
-            threads.get(entry.person.thread.name) if entry.person is not None else None,
-        )
+        row.present(entry, threads)
 
     def sync_members(self, changed: frozenset[str]) -> None:
         """Reconcile member rows from this group's entries and the tree's derived thread rows."""