diff --git a/src/toad/conversation_kind.py b/src/toad/conversation_kind.py index a6e386cde..36e298542 100644 --- a/src/toad/conversation_kind.py +++ b/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 diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py index 6738dda57..5cb9c1d1a 100644 --- a/src/toad/transcript_preparation.py +++ b/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 diff --git a/src/toad/widgets/comms_chat.py b/src/toad/widgets/comms_chat.py index 5942c65a9..556bdb9db 100644 --- a/src/toad/widgets/comms_chat.py +++ b/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, diff --git a/src/toad/widgets/presentation_window.py b/src/toad/widgets/presentation_window.py index 013505056..df3544d89 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, 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 diff --git a/src/toad/widgets/thread_comms.py b/src/toad/widgets/thread_comms.py index f1404be99..d11dbfa97 100644 --- a/src/toad/widgets/thread_comms.py +++ b/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.""" diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index 04505e2ea..9ee45816b 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -9,20 +9,22 @@ import asyncio from abc import ABC +from collections.abc import Callable +from contextlib import ExitStack from dataclasses import dataclass, field, replace -from functools import partial, singledispatch +from functools import cached_property, partial, singledispatch from pathlib import Path +from threading import RLock from typing import TYPE_CHECKING, ClassVar from agent_comms.comms import Comms from agent_comms.repo import Checkout -from agent_comms.display_references import CurrentDisplayReferences from agent_comms.historical_views import ( ConversationHistoryView, ConversationObservation, ConversationRevision, ) -from agent_comms.message_page import MessagePage +from agent_comms.message_page import MessagePage, PageBounds from agent_comms.messages import ConversationRole, Membership, Message, NoticeRole from agent_comms.page_traversal import ( LiveCursor, @@ -31,6 +33,7 @@ PageCursor, ) from agent_comms.transcript_events import IncomingTranscript +from agent_comms.transcripts import TranscriptCursor from toad.comms_root import CoordinationAccess from toad.conversation_kind import ConversationKind @@ -41,6 +44,7 @@ PreparedTranscriptPage, ReplaceablePageSource, SavedHistory, + SavedPrefix, SourceObservation, TranscriptPageWork, Unchanged, @@ -49,6 +53,7 @@ LatestReplacementViewportRequest, ReplacementViewportRequest, ) +from toad.work_preparation import PreparationRuntime, PreparationScope from toad.widgets.history_anchor import LatestPosition from toad.widgets.message_notifications import PaintedRead, ReadWork from toad.widgets.transcript_fragments import ( @@ -303,10 +308,6 @@ class WireSource: def viewer(self) -> NamedViewer: return self.comms.views.display_viewer(self.project) - def references(self) -> CurrentDisplayReferences: - """Resolve stable destinations through this read's registry cut.""" - return CurrentDisplayReferences.from_registry(str(self.comms.root), self.comms.registry.snapshot()) - def observe(self, seen: ConversationObservation) -> SourceObservation: """The observed state owns how changed or unchanged history is drawn.""" revision = self.conversation.history_view.revision( @@ -320,9 +321,8 @@ def observe(self, seen: ConversationObservation) -> SourceObservation: observation = ReplacedWireObservation if revision.display_changed_since(seen) else GrowingWireObservation return observation(revision, high) - def read(self, request: PageRequest[WireCut], through: AfterSequence) -> PreparedTranscriptPage: - """Read the page a request names, never past `through`, as fragments to draw.""" - page = request.end.wire_page(self, request.cursor, through) + def prepare(self, page: MessagePage, through: AfterSequence) -> PreparedTranscriptPage: + """Draw the original page and its evidence without reacquiring its source.""" shown = MessageShown(style=self.style) fragments = tuple( WireFragment.of(message, shown=shown, evidence=evidence, reader=self.me) @@ -342,6 +342,59 @@ def restyled(self, style: WireMessageStyle) -> WireSource: return replace(self, style=style) +class WireReading(ExitStack): + """One lazy Core cut, borrowed by independently revocable source scopes. + + Scope borrowing/release belongs to the preparation loop. Workers serialize + acquisition, page reads and final close through the same lock. + """ + + def __init__(self, loader: WireSource) -> None: + super().__init__() + self.loader = loader + self._use = RLock() + self._borrowers = 0 + + @cached_property + def pages(self) -> Callable[[PageBounds], MessagePage]: + return self.enter_context(self.loader.conversation.reading(self.loader)) + + def borrow(self) -> WireScope: + self._borrowers += 1 + return WireScope(self) + + def release(self, runtime: PreparationRuntime) -> None: + self._borrowers -= 1 + if self._borrowers == 0: + runtime.retire_thread(self.close) + + def read(self, scope: WireScope, request: PageRequest[WireCut], + through: AfterSequence) -> PreparedTranscriptPage: + with self._use: + if scope.closed: + raise asyncio.CancelledError + page = request.end.wire_page(self.loader, request.cursor, through, self.pages) + return self.loader.prepare(page, through) + + def close(self) -> None: + with self._use: + super().close() + + +class WireScope(PreparationScope): + """Revoking one source does not revoke another source's borrowed cut.""" + + def __init__(self, reading: WireReading) -> None: + super().__init__() + self.reading = reading + + def _fork(self) -> WireScope: + return self.reading.borrow() + + def release(self, runtime: PreparationRuntime) -> None: + self.reading.release(runtime) + + @dataclass(frozen=True) class RestyledViewportRequest(ReplacementViewportRequest): loader: WireSource @@ -375,6 +428,7 @@ async def apply(self, history: WireHistory, source: ReplaceablePageSource) -> bo class GrowingWireObservation(WireObservation): async def apply(self, history, source) -> bool: + source.renew_reading(self.through) if not history.undrawn: published = await history.extend(self.through) else: @@ -386,27 +440,41 @@ 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(TranscriptPageWork[WireCut, WireSource]): + """Read through the source scope's original cut, not a fresh per-page cut.""" + + scope: WireScope @property def retain_result(self) -> bool: return self.request.cursor.kept async def execute(self, runtime, key): - import asyncio - - prepared = await runtime.run_thread(self.loader.read, self.request, self.through) + prepared = await runtime.run_thread(self.scope.reading.read, self.scope, self.request, self.through) if self.scope.closed: raise asyncio.CancelledError return await runtime.run_thread(self.finish_result, key, prepared) -class WirePageBuffer(SavedHistory): +class WirePageBuffer(SavedHistory[WireSource]): """The line history's page reader over a wire source.""" work = WirePageWork + def new_scope(self) -> WireScope: + return WireReading(self.loader).borrow() + + def prefix(self, through: TranscriptCursor) -> SavedPrefix[WireSource]: + return SavedPrefix(self, self.loader, through, self.runtime) + + def renew_reading(self, through: AfterSequence) -> None: + # Read/notification-only revisions keep their cut. Growth needs a new + # main cut; prefixes retain the old cut under their own scopes. + if through != self.through: + previous = self.scope + self.scope = self.runtime.open_scope(self.new_scope) + self.runtime.discard_scope(previous) + def __init__(self, *args, revision: ConversationObservation, **kwargs): super().__init__(*args, **kwargs) self.revision = revision diff --git a/src/toad/work_preparation.py b/src/toad/work_preparation.py index a6ca9be7f..5b2cef1ed 100644 --- a/src/toad/work_preparation.py +++ b/src/toad/work_preparation.py @@ -5,7 +5,7 @@ import asyncio from abc import ABC, abstractmethod from collections import OrderedDict -from collections.abc import Callable +from collections.abc import Callable, Coroutine from copy import deepcopy from dataclasses import dataclass from enum import Enum @@ -31,6 +31,22 @@ class PreparationScope: def __init__(self) -> None: self.closed = False + def fork(self) -> PreparationScope: + if self.closed: + raise ValueError("Cannot borrow a revoked preparation scope") + return self._fork() + + def _fork(self) -> PreparationScope: + return PreparationScope() + + def revoke(self, runtime: PreparationRuntime) -> None: + if not self.closed: + self.closed = True + self.release(runtime) + + def release(self, runtime: PreparationRuntime) -> None: + """Scopes with resources retire them through the preparation owner.""" + class WorkRevision(ABC): """A key's declared, bounded metadata charge; never a source graph.""" @@ -262,6 +278,7 @@ def __init__(self, renderer: Renderer, *, max_entries: int = 256, self._admitted = dict.fromkeys(WorkLane, 0) self._threads = asyncio.Semaphore(4) self._thread_tasks: set[asyncio.Task] = set() + self._scopes: set[PreparationScope] = set() self._changed = asyncio.Event() self._closed = False self._shutdown: asyncio.Task[None] | None = None @@ -277,23 +294,48 @@ async def run_thread[*Args, Value]( if self._closed: self._threads.release() raise asyncio.CancelledError - context = self.renderer.execution_context() - task = asyncio.create_task(asyncio.to_thread(function, *args), name="preparation-thread", - context=context) - self._thread_tasks.add(task) - - def finished(completed): - self._thread_tasks.discard(completed) - if not completed.cancelled(): - completed.exception() - self._threads.release() - - task.add_done_callback(finished, context=context) + task = self._track_thread(asyncio.to_thread(function, *args)) + task.add_done_callback(lambda completed: self._threads.release(), + context=self.renderer.execution_context()) await self.renderer.wait_for_work(task) return task.result() + def _track_thread[Value](self, work: Coroutine[object, object, Value]) -> asyncio.Task[Value]: + task = asyncio.create_task(work, name="preparation-thread", + context=self.renderer.execution_context()) + self._thread_tasks.add(task) + task.add_done_callback(self._thread_finished, context=self.renderer.execution_context()) + return task + + def _thread_finished(self, completed: asyncio.Task) -> None: + self._thread_tasks.discard(completed) + if not completed.cancelled(): + completed.exception() + + def retire_thread(self, release: Callable[[], None]) -> None: + """Retirement still takes a thread slot and is joined at close, even after new work stops.""" + self._track_thread(self._retire_thread(release)) + + async def _retire_thread(self, release: Callable[[], None]) -> None: + try: + async with self._threads: + await super().run_thread(release) + except Exception as error: + asyncio.get_running_loop().call_exception_handler({ + "message": "Preparation retirement failed", "exception": error, + }) + raise + + def open_scope[Scope: PreparationScope](self, create: Callable[[], Scope]) -> Scope: + if self._closed: + raise asyncio.CancelledError + scope = create() + self._scopes.add(scope) + return scope + def discard_scope(self, scope: PreparationScope) -> None: - scope.closed = True + scope.revoke(self) + self._scopes.discard(scope) for key in tuple(self._ready): if key.scope is scope: _, size = self._ready.pop(key) @@ -380,6 +422,8 @@ async def aclose(self) -> None: self._shutdown.result() async def _close(self) -> None: + for scope in tuple(self._scopes): + self.discard_scope(scope) if self._pending: await asyncio.gather(*tuple(self._pending.values()), return_exceptions=True) if self._thread_tasks: diff --git a/tests/app/test_channel_scope_expansion.py b/tests/app/test_channel_scope_expansion.py index b5adb5890..b155aa86d 100644 --- a/tests/app/test_channel_scope_expansion.py +++ b/tests/app/test_channel_scope_expansion.py @@ -1,5 +1,7 @@ """A mounted any-mode channel reveals older messages entering its scope.""" +from agent_comms.repo import Checkout + async def test_mounted_any_mode_channel_reveals_older_messages_entering_its_scope(workspace): for name, tags in (("alice", ("team",)), ("bob", ()), ("carol", ()), ("dave", ())): @@ -8,7 +10,10 @@ async def test_mounted_any_mode_channel_reveals_older_messages_entering_its_scop comms.messaging.send("carol", "dave", "older newly visible DM") comms.messaging.send("alice", "#team", "current channel message") comms.channels.set_channel_any_mode("#team", True) - assert [m.body for m in comms.views.channel_display_page("#team", worktree=str(workspace.project)).messages] == [ + view = comms.views.channel_display_view( + "#team", viewer=comms.views.display_viewer(Checkout(workspace.project)), + ) + assert [m.body for m in view.page(comms.views.presentation).messages] == [ "current channel message"] async with workspace.app(size=(100, 30)) as ui: diff --git a/tests/textual/test_channel_any_mode.py b/tests/textual/test_channel_any_mode.py index 5057ca9f1..29a94c845 100644 --- a/tests/textual/test_channel_any_mode.py +++ b/tests/textual/test_channel_any_mode.py @@ -9,7 +9,9 @@ def bodies(comms): - return [message.body for message in comms.views.channel_display_page("#team").messages] + return [message.body for message in comms.views.channel_display_view( + '#team', + ).page(comms.views.presentation).messages] async def open_menu(ui, sidebar, channel): diff --git a/tests/textual/test_channel_read_ahead.py b/tests/textual/test_channel_read_ahead.py index 0e29e002f..d27d255ee 100644 --- a/tests/textual/test_channel_read_ahead.py +++ b/tests/textual/test_channel_read_ahead.py @@ -1,6 +1,4 @@ -"""Textual-facing: the channel history reads earlier pages from Core before the reader's scrolling reaches its edge.""" - -from unittest.mock import patch +"""Textual-facing: scrolling consumes prepared pages on the original channel reading.""" from toad.widgets.comms_chat import CommsChatView @@ -11,30 +9,24 @@ async def test_a_channel_reads_the_next_older_page_before_scrolling_reaches_its_ workspace.comms.messaging.send("sender", "#scroll", f"Message {index}: " + "wrapped body " * 6) async with workspace.app(size=(145, 50)) as ui: app, _pilot = ui.textual - views = app.coordination_access.service.views - earlier_reads = [] - original = views.channel_display_page - - def counted(target, **bounds): - if bounds.get("before") is not None: - earlier_reads.append(bounds["limit"]) - return original(target, **bounds) - - with patch.object(views, "channel_display_page", side_effect=counted): - await ui.open_channel("#scroll", actor="sender") - chat = app.screen.query_one(CommsChatView) - history = chat.message_history - await ui.until(lambda: history.messages, why="channel messages") - source = history.source - assert source.comms.views is views - # The newest messages come as a short page drawn at once; every earlier read is a whole page. - await ui.until(lambda: len(history.messages) > source.tail_limit and len(earlier_reads) >= 3 - and history.state.accepts_source_work, why="earlier messages and a page read ahead") - assert earlier_reads[0] == source.tail_limit and set(earlier_reads[1:]) == {source.page_limit}, earlier_reads - oldest, reads = history.messages[0].seq, len(earlier_reads) - chat.window.focus() - for _ in range(6): - await ui.press("pageup") - await ui.until(lambda: history.messages[0].seq <= oldest - 24, why="earlier messages drawn while paging up") - # Six pages of scrolling did not cost six Core reads. - assert len(earlier_reads) - reads <= 3 and set(earlier_reads[1:]) == {source.page_limit}, earlier_reads + await ui.open_channel("#scroll", actor="sender") + chat = app.screen.query_one(CommsChatView) + history = chat.message_history + await ui.until(lambda: history.fragments, why="channel messages") + source = history.page_source.loader + assert source.comms.views is app.coordination_access.service.views + # Earlier pages are prepared before the reader starts scrolling. + await ui.until(lambda: len(history.fragments) > source.tail_limit + and history.state.accepts_source_work, why="earlier messages prepared") + scope = history.page_source.scope + reading = scope.reading + oldest = history.fragments[0].message.seq + hits = app.preparation.hits + chat.window.focus() + for _ in range(6): + await ui.press("pageup") + await ui.until(lambda: history.fragments[0].message.seq <= oldest - 24, + why="earlier messages drawn while paging up") + assert history.page_source.scope is scope and scope.reading is reading + assert not scope.closed + assert app.preparation.hits > hits diff --git a/tests/textual/test_channel_scope_rows.py b/tests/textual/test_channel_scope_rows.py index b8e10e58e..f7673dc85 100644 --- a/tests/textual/test_channel_scope_rows.py +++ b/tests/textual/test_channel_scope_rows.py @@ -15,7 +15,10 @@ async def test_any_mode_channel_rows_include_older_messages_after_a_refresh(work comms.messaging.send("carol", "dave", "older newly visible DM") comms.messaging.send("alice", "#team", "current channel message") comms.channels.set_channel_any_mode("#team", True) - assert [m.body for m in comms.views.channel_display_page("#team", viewer=comms.views.display_viewer(Checkout(project))).messages] == [ + assert [m.body for m in comms.views.channel_display_view( + '#team', + viewer=comms.views.display_viewer(Checkout(project)), + ).page(comms.views.presentation).messages] == [ "current channel message"] async with workspace.app(size=(100, 30)) as ui: @@ -30,9 +33,15 @@ async def test_any_mode_channel_rows_include_older_messages_after_a_refresh(work await ui.app.coordination_access.stop_observing() history = chat.message_history source, snapshot = history.page_source, history.source_snapshot() - marked = comms.views.channel_display_page("#team", viewer=comms.views.display_viewer(Checkout(project))) + marked = comms.views.channel_display_view( + '#team', + viewer=comms.views.display_viewer(Checkout(project)), + ).page(comms.views.presentation) comms.channels.update_tags("carol", add=frozenset({"team"})) - expanded = comms.views.channel_display_page("#team", viewer=comms.views.display_viewer(Checkout(project))) + expanded = comms.views.channel_display_view( + '#team', + viewer=comms.views.display_viewer(Checkout(project)), + ).page(comms.views.presentation) assert [m.body for m in expanded.messages][:2] == ["older newly visible DM", "current channel message"] await history._mark_page(marked, (), snapshot) await ui.until(lambda: history.page_source is not source, diff --git a/tests/textual/test_channel_updates_and_returns.py b/tests/textual/test_channel_updates_and_returns.py index be4771fdc..60971cec8 100644 --- a/tests/textual/test_channel_updates_and_returns.py +++ b/tests/textual/test_channel_updates_and_returns.py @@ -1,5 +1,7 @@ """A real shared observer updates a shown channel and leaves its parked view alone.""" +import asyncio + async def test_live_update_paints_parked_channel_reads_nothing_and_return_shows_the_update(workspace): from toad.widgets.comms_chat import CommsChatView, session_thread_name @@ -32,3 +34,64 @@ async def test_live_update_paints_parked_channel_reads_nothing_and_return_shows_ await ui.until_shows("PARKED UPDATE") await ui.until(lambda: bool(access.notification_interest.reads), why="the returned channel to resubscribe") assert app.selected_session.query_one(CommsChatView) is chat + + +async def test_first_wire_page_paints_before_pending_sidebar_and_keeps_source_fences(workspace): + from agent_comms.launch_root import LaunchRoot + from support.roots import installed_wire + from toad.preferences import SETTINGS + from toad.widgets.channel_prompt import ChannelPrompt + from toad.widgets.comms_chat import CommsChatView + + sender = workspace.thread("sender", "first-page") + workspace.thread("reader", "first-page") + workspace.comms.messaging.send("sender", "#first-page", "FIRST_PAGE_WITHOUT_SIDEBAR") + other = installed_wire(workspace.roots.attempt / "other-route") + async with workspace.app(size=(113, 34)) as ui: + app = ui.app + access = app.coordination_access + await access.current_sidebar() + observed = access.observation.observed() + process = observed.process + # Join the actual receiver, not a replacement service or fake answer. + # Its child keeps running and the real sidebar reply remains in its pipe. + await process._reader.aclose() + async with asyncio.timeout(20), asyncio.TaskGroup() as work: + try: + app.saved_settings.toggle(SETTINGS.sidebar.show_archived) + sidebar = work.create_task(access.current_sidebar()) + references = work.create_task( + observed.current_references(access, str(workspace.comms.root)), + ) + await ui.until(lambda: bool(access.sidebar_waiters), + why="the genuine changed-interest sidebar answer to remain pending") + opening = work.create_task(ui.open_channel("#first-page", actor="reader")) + await ui.until_shows("FIRST_PAGE_WITHOUT_SIDEBAR") + chat = app.selected_session.query_one(CommsChatView) + history = chat.message_history + source = history.page_source + assert not history.undrawn + assert not sidebar.done() and not references.done() + assert not chat.query_one(ChannelPrompt)._candidates + finally: + # Reattach the same observer's one reader; deliver its original reply. + process.start_receiving(observed.receive, observed.receive_failed) + mode = await opening + answer = await sidebar + catalog = await references + assert catalog.thread(sender.incarnation).name == sender.name + assert answer.require_snapshot().references.thread(sender.incarnation).name == sender.name + await ui.until( + lambda: any(candidate.name == sender.name + for candidate in chat.query_one(ChannelPrompt)._candidates), + why="the genuine channel mention label to finish after its sidebar answer", + ) + assert chat.agent_info.plain == chat.conversation.label(catalog) + snapshot = history.source_snapshot() + source = history.page_source + with LaunchRoot.from_path(other.root).selected(): + assert not await history.source_is_current(snapshot) + assert await history.source_is_current(snapshot) + await app.session_navigation.close(mode) + assert source.closed + assert not await source.route_is_current() diff --git a/tests/textual/test_conversation_kind.py b/tests/textual/test_conversation_kind.py index cdc0d63d6..b7e4e211f 100644 --- a/tests/textual/test_conversation_kind.py +++ b/tests/textual/test_conversation_kind.py @@ -10,10 +10,12 @@ from agent_comms.comms import Comms from agent_comms.active_route import resolve_comms_route from support.roots import installed_wire +from agent_comms.errors import UnregisteredThreadError from support.agents import AppAgents from agent_comms.field_codec import FieldCodec from agent_comms.installer.backup import StoreBackup from agent_comms.message_page import PageBounds +from agent_comms.thread_creation import RecordedCreationTime from agent_comms.threads import Thread from toad.constants import ALL_COMMS_TARGET @@ -54,11 +56,12 @@ def test_conversation_family_resolves_copied_pages_and_current_labels( assert resolved.viewer == viewer assert conversation.label(references) == label assert HistoryTarget.for_conversation(conversation, references) == route - page = conversation.page( - WireSource(comms, conversation, comms.registry.require(viewer).worktree, viewer, IrcMessageStyle(), - access=AppAgents.unrun(tmp_path, LaunchRoot.from_path(comms.root)).app.coordination_access), - PageBounds(limit=8), + source = WireSource( + comms, conversation, comms.registry.require(viewer).worktree, viewer, IrcMessageStyle(), + access=AppAgents.unrun(tmp_path, LaunchRoot.from_path(comms.root)).app.coordination_access, ) + with conversation.reading(source) as pages: + page = pages(PageBounds(limit=8)) assert body in [message.body for message in page.messages] assert FieldCodec.decode(type[ConversationKind], FieldCodec.encode(type(conversation))) is type(conversation) with pytest.raises(FrozenInstanceError): @@ -110,11 +113,14 @@ def test_kind_and_owner_are_part_of_identity_but_channel_address_is_shared(copie ) -def test_dm_rename_retains_key_and_history_scope_but_resolves_current_name(copied_conversations): +async def test_dm_rename_retains_key_and_history_scope_but_resolves_current_name(copied_conversations): comms, viewer = copied_conversations direct = DirectTarget("peer").prepare(comms, viewer).conversation key = CommsViewKey(str(comms.root), "owner", direct) scope = direct.input_scope + assert await direct.status_text(comms, read_only=False) == ( + comms.configuration.answer("peer").status_label + ) comms.registry.rename("peer", "renamed") reopened = DirectTarget("renamed").prepare(comms, viewer).conversation references = comms.views.viewer_snapshot(Checkout(comms.root.parent)).references @@ -124,6 +130,9 @@ def test_dm_rename_retains_key_and_history_scope_but_resolves_current_name(copie assert direct.label(references) == "@renamed" assert direct.send_target().selected_name(comms.registry.snapshot()) == "renamed" assert HistoryTarget.for_conversation(direct, references) == DirectTarget("renamed") + assert await direct.status_text(comms, read_only=False) == ( + comms.configuration.answer("renamed").status_label + ) def test_recreated_name_cannot_resolve_the_retired_dm(copied_conversations): @@ -141,6 +150,27 @@ def test_recreated_name_cannot_resolve_the_retired_dm(copied_conversations): direct.destination(references) +@pytest.mark.parametrize("recreate", (False, True)) +async def test_dm_status_refuses_deleted_or_recreated_peer(copied_conversations, recreate): + comms, viewer = copied_conversations + original = comms.registry.require("peer") + direct = DirectTarget("peer").prepare(comms, viewer).conversation + comms.registry.unregister("peer") + comms.threads.delete("peer") + if recreate: + comms.registry.declare(Thread( + "peer", frozenset(), Checkout(comms.root.parent), + created_at=RecordedCreationTime(original.created_at + 1), + )) + replacement = DirectTarget("peer").prepare(comms, viewer).conversation + assert replacement != direct + assert await replacement.status_text(comms, read_only=False) == ( + comms.configuration.answer("peer").status_label + ) + with pytest.raises(UnregisteredThreadError): + await direct.status_text(comms, read_only=False) + + def test_history_picker_and_channel_status_use_the_same_declared_value(copied_conversations): from toad.screens.historical_sessions import HistoricalSessions diff --git a/tests/textual/test_thread_relationship_sidebar.py b/tests/textual/test_thread_relationship_sidebar.py index 4191d2d3e..dcac44d28 100644 --- a/tests/textual/test_thread_relationship_sidebar.py +++ b/tests/textual/test_thread_relationship_sidebar.py @@ -76,6 +76,48 @@ async def test_relationship_groups_follow_core_edits_tools_sorting_and_renames(w assert peer_group.entries["owner"].target == "owner" assert peer_group.entries["owner"].detail == "Review the sidebar" + # Use Core's actual entries and models in this already-open panel. + collaboration = tree.groups[CollaboratingRelationshipGroup] + person_entry = collaboration.model.entries["peer"] + person, = person_entry.person.views() + person_row = collaboration.new_row("peer") + channel_group = tree.groups[OutboundRelationshipGroup] + channel_entry = channel_group.model.entries["#team"] + channel_row = channel_group.new_row("#team") + await collaboration.member_container.mount(person_row) + await channel_group.member_container.mount(channel_row) + try: + person_row.present(person_entry, {}) + await ui.until( + lambda: person_row.thread_row is not None, + why="an uncached person to finish real model preparation", + ) + assert person_row.thread_row is not None + assert person_row.thread_row.incarnation == person.thread.incarnation + assert person_row.thread_row.name == "peer" + assert "Review the sidebar" in str(person_row.tooltip) + + cached = tree.model_state.shown.threads.rows["peer"] + person_row.retire_thread_row() + person_row.present(person_entry, {"peer": cached}) + assert person_row.thread_row is cached + + assert tuple(channel_entry.person.views()) == () + channel_row.present(channel_entry, tree.model_state.shown.threads.rows) + await ui.until( + lambda: "#team" in ui.painted(channel_row), + why="the nominally absent person to keep its channel label", + ) + assert channel_row.thread_row is None + assert channel_row.entry == channel_entry + assert str(channel_row.tooltip) == ( + f"{channel_entry.target}\n{channel_entry.detail}" + if channel_entry.detail else channel_entry.target + ) + finally: + await person_row.remove() + await channel_row.remove() + invoke_tool(comms, "comms_collaboration", {"action": "remove", "peer": "owner"}, caller=ThreadCaller("peer")) await ui.until(lambda: not tree.groups[CollaboratingRelationshipGroup].model.entries diff --git a/tests/textual/test_wire_source.py b/tests/textual/test_wire_source.py index 8d96d8159..890e908ab 100644 --- a/tests/textual/test_wire_source.py +++ b/tests/textual/test_wire_source.py @@ -1,6 +1,8 @@ """Textual-facing: the presentation window; a wire source pages one conversation and notices when Core replaces its reading.""" import os +from contextlib import contextmanager +from functools import partial from agent_comms.repo import Checkout from agent_comms.launch_root import LaunchRoot @@ -17,7 +19,7 @@ from toad.transcript_preparation import PageRequest, Unchanged from toad.widgets.presentation_window import OLDER_END, NEWER_END from toad.message_rows import IrcMessageStyle -from toad.wire_history import AfterSequence, GrowingWireObservation, MessageFragment, ReplacedWireObservation, WireSource +from toad.wire_history import AfterSequence, GrowingWireObservation, MessageFragment, ReplacedWireObservation, WireReading, WireSource def channel_source(comms, target, project): @@ -27,12 +29,15 @@ def channel_source(comms, target, project): access=AppAgents.unrun(project.path, LaunchRoot.from_path(comms.root)).app.coordination_access) +@contextmanager def observed(source): - """What a view that just drew the newest page remembers of it.""" + """Retain one original cut across the newest and subsequent bounded pages.""" first = source.observe(ConversationObservation()) assert isinstance(first, ReplacedWireObservation) - newest = source.read(PageRequest(OLDER_END, first.through), first.through) - return first, newest + with WireReading(source) as reading: + read = partial(reading.read, reading.borrow()) + newest = read(PageRequest(OLDER_END, first.through), first.through) + yield first, newest, read def test_a_turn_claim_on_the_peer_keeps_the_direct_message_page_identity(tmp_path): @@ -40,9 +45,15 @@ def test_a_turn_claim_on_the_peer_keeps_the_direct_message_page_identity(tmp_pat comms.registry.declare(Thread("peer", frozenset(), Checkout(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()))) viewer = comms.messaging.user_identity(Checkout(tmp_path)).name comms.messaging.send("peer", viewer, "shown across turn claim") - before = comms.views.dm_display_page(ThreadTarget("peer"), worktree=Checkout(tmp_path)) + before = comms.views.dm_display_view( + ThreadTarget('peer'), + worktree=Checkout(tmp_path), + ).page(comms.views.presentation) comms.registry.lease_local_turn("peer", TurnId.mint()) - after = comms.views.dm_display_page(ThreadTarget("peer"), worktree=Checkout(tmp_path)) + after = comms.views.dm_display_view( + ThreadTarget('peer'), + worktree=Checkout(tmp_path), + ).page(comms.views.presentation) assert DmConversation.history_view.same_display(after, before) @@ -53,33 +64,34 @@ def test_pages_lie_between_cuts_and_a_read_never_passes_its_bound(tmp_path): comms.messaging.send("sender", "#one", f"Message {index}") comms.messaging.send("sender", "#two", f"Elsewhere {index}") source = channel_source(comms, "#one", tmp_path) - first, newest = observed(source) - def bodies(prepared): return [fragment.message.body for fragment in prepared.fragments] - assert bodies(newest) == [f"Message {index}" for index in range(52, 60)] - assert newest.page.after == first.through and not newest.page.has_newer and newest.page.has_older - assert all(isinstance(fragment, MessageFragment) and not fragment.outgoing_sources - for fragment in newest.fragments) - earlier = source.read(PageRequest(OLDER_END, newest.page.before), first.through) - assert bodies(earlier) == [f"Message {index}" for index in range(12, 52)] - assert earlier.page.has_newer and earlier.page.has_older - oldest = source.read(PageRequest(OLDER_END, earlier.page.before), first.through) - assert bodies(oldest) == [f"Message {index}" for index in range(12)] and not oldest.page.has_older - assert bodies(source.read(PageRequest(NEWER_END, oldest.page.after), first.through))[:3] == [ - "Message 12", "Message 13", "Message 14"] + with observed(source) as (first, newest, read): + assert bodies(newest) == [f"Message {index}" for index in range(52, 60)] + assert newest.page.after == first.through and not newest.page.has_newer and newest.page.has_older + assert all(isinstance(fragment, MessageFragment) and not fragment.outgoing_sources + for fragment in newest.fragments) + earlier = read(PageRequest(OLDER_END, newest.page.before), first.through) + assert bodies(earlier) == [f"Message {index}" for index in range(12, 52)] + assert earlier.page.has_newer and earlier.page.has_older + oldest = read(PageRequest(OLDER_END, earlier.page.before), first.through) + assert bodies(oldest) == [f"Message {index}" for index in range(12)] and not oldest.page.has_older + assert bodies(read(PageRequest(NEWER_END, oldest.page.after), first.through))[:3] == [ + "Message 12", "Message 13", "Message 14"] comms.messaging.send("sender", "#two", "Elsewhere again") grown = source.observe(first.revision) assert grown.through.cursor.sequence > first.through.cursor.sequence and isinstance(grown, GrowingWireObservation) - nothing = source.read(PageRequest(NEWER_END, first.through), grown.through) - assert not nothing.fragments and nothing.page.after == grown.through + with observed(source) as (_grown, _newest, read): + nothing = read(PageRequest(NEWER_END, first.through), grown.through) + assert not nothing.fragments and nothing.page.after == grown.through comms.messaging.send("sender", "#one", "arrived later") comms.messaging.send("sender", "#one", "arrived after the bound") later = AfterSequence(LiveCursor(grown.through.cursor.sequence + 1)) - bounded = source.read(PageRequest(NEWER_END, grown.through), later) - assert bodies(bounded) == ["arrived later"] and bounded.page.after == later and not bounded.page.has_newer + with observed(source) as (_later, _newest, read): + bounded = read(PageRequest(NEWER_END, grown.through), later) + assert bodies(bounded) == ["arrived later"] and bounded.page.after == later and not bounded.page.has_newer current = source.observe(grown.revision) assert isinstance(source.observe(current.revision), Unchanged) @@ -92,14 +104,14 @@ def test_any_mode_expansion_replaces_the_reading_without_new_bus_rows(tmp_path): comms.messaging.send("carol", "dave", "older DM") comms.messaging.send("alice", "#team", "channel message") source = channel_source(comms, "#team", tmp_path) - first, newest = observed(source) - assert [fragment.message.body for fragment in newest.fragments] == ["channel message"] + with observed(source) as (first, newest, _read): + assert [fragment.message.body for fragment in newest.fragments] == ["channel message"] comms.channels.set_channel_any_mode("#team", True) assert comms.bus.bus.revision() == first.through.cursor.sequence expanded = source.observe(first.revision) assert isinstance(expanded, ReplacedWireObservation) and expanded.through == first.through - again = source.read(PageRequest(OLDER_END, expanded.through), expanded.through) - assert [fragment.message.body for fragment in again.fragments] == ["older DM", "channel message"] + with observed(source) as (_expanded, again, _read): + assert [fragment.message.body for fragment in again.fragments] == ["older DM", "channel message"] def test_attaching_or_detaching_history_replaces_the_reading_without_new_live_messages(tmp_path): @@ -109,15 +121,16 @@ def test_attaching_or_detaching_history_replaces_the_reading_without_new_live_me bus.registry.declare(Thread("alice", frozenset({"team"}), Checkout(tmp_path))) bus.messaging.send("alice", "#team", bus.root.name) source = channel_source(comms, "#team", tmp_path) - first, _newest = observed(source) + with observed(source) as (first, newest, _read): + assert [message.body for message in newest.page.events] == ["live"] comms.views.attach_history(old.root) attached = source.observe(first.revision) assert isinstance(attached, ReplacedWireObservation) and attached.through == first.through - page = source.read(PageRequest(OLDER_END, attached.through), attached.through) - assert [message.body for message in page.page.events] == ["old", "live"] + with observed(source) as (_attached, page, _read): + assert [message.body for message in page.page.events] == ["old", "live"] comms.bus.history.path.unlink() detached = source.observe(attached.revision) assert isinstance(detached, ReplacedWireObservation) - remaining = source.read(PageRequest(OLDER_END, detached.through), detached.through) - assert [message.body for message in remaining.page.events] == ["live"] - assert not remaining.page.has_older + with observed(source) as (_detached, remaining, _read): + assert [message.body for message in remaining.page.events] == ["live"] + assert not remaining.page.has_older