From 7fa5b861c7b3542dc7f8f77a7a06bfb32f21d168 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 30 Oct 2026 02:37:38 -0400 Subject: [PATCH] Index a page's evidence once and read live history once per traversal MessagePage.evidence_for rescanned every source page for each message (O(page^2) per history read); message_evidence indexes the source pages by view key once and pairs every message with its evidence page. HistoricalSources.integrated_page re-combined every page into a new MessagePage on each loop, and each historical part opened a full live page read only to learn whether live messages follow. The remaining budget and continuation now follow from the newest part, the pages are combined once, and one traversal asks the live bus once. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/historical_views.py | 29 +++++++++++++++++------------ src/agent_comms/message_page.py | 12 ++++++++---- 2 files changed, 25 insertions(+), 16 deletions(-) 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: