diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py index 19e2e8a795..fc369c5358 100644 --- a/src/agent_comms/historical_views.py +++ b/src/agent_comms/historical_views.py @@ -10,7 +10,7 @@ from collections.abc import Callable, Iterator from contextlib import contextmanager from dataclasses import dataclass, replace -from functools import partial +from functools import cache, partial from pathlib import Path from typing import TYPE_CHECKING @@ -556,20 +556,26 @@ def page(self, view: HistoryView, presentation: BusPresentation, bounds: PageBou def integrated_page( self, view: HistoryView, presentation: BusPresentation, bounds: PageBounds ) -> MessagePage: - page = self._source_page(view, presentation, bounds) - pages = deque((page,)) - while (following := bounds.after(page)) is not None: - part = self._source_page(view, presentation, following) - if part.selection.history_revision != page.selection.history_revision: + # One traversal asks once whether live messages follow its archives. + live_follows = cache(lambda: bool(view.live_page(presentation, PageBounds(limit=1)).messages)) + first = self._source_page(view, presentation, bounds, live_follows) + pages = deque((first,)) + # Each part extends the traversal's edge, so the remaining budget and continuation + # follow from the newest part; the pages are combined once, at the end. + following = bounds.after(first) + while following is not None: + part = self._source_page(view, presentation, following, live_follows) + if part.selection.history_revision != first.selection.history_revision: raise HistoricalSourceChanged("Historical sources changed; reload history") - if not bounds.fits(len(page.messages) + len(part.messages)): + if not following.fits(len(part.messages)): break bounds.traversal.append(pages, part) - page = MessagePage.combine(tuple(pages)) - return page + following = following.after(part) + return MessagePage.combine(tuple(pages)) if len(pages) > 1 else first def _source_page( - self, view: HistoryView, presentation: BusPresentation, bounds: PageBounds + self, view: HistoryView, presentation: BusPresentation, bounds: PageBounds, + live_follows: Callable[[], bool], ) -> MessagePage: historical = bounds.traversal.historical if not self.sources() and not historical: @@ -580,11 +586,10 @@ def historical_page() -> MessagePage: view, presentation, bounds if historical else replace(bounds, traversal=OlderTraversal()) ) if history.messages: - latest = view.live_page(presentation, PageBounds(limit=1)) return replace( history, selection=replace(history.selection, history_revision=history_revision), - has_newer=history.has_newer or bool(latest.messages), + has_newer=history.has_newer or live_follows(), ) selected = history if historical else page result = replace(selected, selection=replace(selected.selection, history_revision=history_revision)) diff --git a/src/agent_comms/message_page.py b/src/agent_comms/message_page.py index c1e7b933b0..8ef7b83b0b 100644 --- a/src/agent_comms/message_page.py +++ b/src/agent_comms/message_page.py @@ -40,10 +40,14 @@ def combine(cls, pages: tuple[MessagePage, ...]) -> MessagePage: source_pages=pages, ) - def evidence_for(self, message: Message) -> MessagePage | None: - page = next(page for page in (self.source_pages or (self,)) - if any(row.view_key == message.view_key for row in page.messages)) - return page if page.selection.has_evidence else None + def message_evidence(self) -> tuple[tuple[Message, MessagePage | None], ...]: + """Each message with the source page whose selection proves it, if that page has + evidence. The source pages are indexed once, not rescanned per message.""" + holders = {row.view_key: page for page in (self.source_pages or (self,)) + for row in page.messages} + return tuple( + (message, holder if (holder := holders[message.view_key]).selection.has_evidence else None) + for message in self.messages) @property def oldest_cursor(self) -> PageCursor | None: