Let commit destinations acquire and publish their saved viewport

src/toad/transcript_publication.py

@@ -221,22 +221,18 @@ async def publish_applied(self) -> bool: ...
 class CanonicalSourcePublication(MroDispatch, TranscriptPublication):
     """An observed source change refreshes source pages, never appends a notice."""
 
+    def __init__(self, binding: BoundView, owner: TranscriptPresentation, generation: int) -> None:
+        super().__init__(binding, owner, generation)
+        self.plan = commit_plan(binding.window)
+
     async def read_snapshot(self):
-        return await self.read_past_history(
+        return await self.plan.read_snapshot(
             partial(
                 self.binding.reader.updates, self.thread_scope, self.owner.current_read
-            )
+            ),
+            self.binding.history,
         )
 
-    async def read_past_history(self, read, **whole):
-        """What follows the mounted history's end while its source only grew, else the whole latest page."""
-        history = self.binding.history
-        try:
-            return await read(after=history.page_source.through)
-        except TranscriptChanged:
-            pass  # Another source file replaced the history's.
-        return await read(**whole)
-
     async def publish_applied(self) -> bool:
         if not self.owner.readable:
             return True
@@ -370,14 +366,10 @@ def cancel(self) -> Worker[None] | None:
 
 
 class CommitPublication(CanonicalSourcePublication):
-    def __init__(self, binding: BoundView, owner: TranscriptPresentation, generation: int) -> None:
-        super().__init__(binding, owner, generation)
-        self.plan = commit_plan(binding.window)
-
     async def read_snapshot(self):
-        # Mounted history grows by what follows its end; without one the window needs a whole page.
-        return await self.read_past_history(
-            partial(self.binding.reader.snapshot, self.thread_scope)
+        return await self.plan.read_snapshot(
+            partial(self.binding.reader.snapshot, self.thread_scope),
+            self.binding.history,
         )
 
     def current(self) -> bool:
@@ -424,8 +416,8 @@ async def publish_source_page(self, page: TranscriptPage) -> bool:
         is_current = partial(self.source_current, page.after)
         if not is_current():
             return True
-        prepared = await plan.prepare(view, history, page, self.captured, is_current)
-        if prepared is None or not is_current():
+        prepared = await plan.prepare(view, history, page, self.captured)
+        if not is_current():
             return True
         evidence = NativeCommitEvidence(
             self.captured,

src/toad/widgets/committed_presentation.py

@@ -3,13 +3,14 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Iterable
+from collections.abc import Awaitable, Callable, Iterable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from functools import partial
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Self
 
 from agent_comms.acp_extension import InputStartedUpdate
+from agent_comms.coordination_errors import TranscriptChanged
 from agent_comms.sent_message import OutgoingMessageReference, PublishedSentMessage
 from agent_comms.transcript_events import TranscriptCoverage
 from agent_comms.transcripts import (
@@ -256,13 +257,14 @@ async def draw(self, cursor: TranscriptCursor, is_current: Callable[[], bool],
 
 
 @dataclass(frozen=True)
-class PreparedSourceCommit(PreparedCommit, ABC):
-    """Every saved-row mutation shares the live-block handoff operation."""
+class PreparedPageCommit(PreparedCommit):
+    """Publish the acquired destination page through the live-block handoff."""
 
+    history: TranscriptHistory
     page: PreparedTranscriptPage
 
     @classmethod
-    async def capture(cls, view: Conversation, history: CommittedHistory, page: TranscriptPage, sequences: frozenset[int]):
+    async def capture(cls, view: Conversation, history: TranscriptHistory, page: TranscriptPage, sequences: frozenset[int]) -> Self:
         from toad.transcript_preparation import TranscriptPageRenderTask
         from toad.work_preparation import RenderPreparation
 
@@ -279,28 +281,21 @@ async def draw(self, cursor, is_current: Callable[[], bool], live_blocks) -> boo
             if not history.source_ready or not is_current():
                 return False
             operation = history.reserve_source_work(partial(CommitTranscript, live_blocks=live_blocks))
-        return await operation.execute(history, partial(self.publish, cursor, is_current))
-
-    @abstractmethod
-    async def publish(self, cursor: TranscriptCursor, is_current: Callable[[], bool]) -> bool: ...
-
-
-class PreparedAdvanceCommit(PreparedSourceCommit):
-    async def publish(self, cursor, is_current: Callable[[], bool]):
-        return await self.history.advance_committed(cursor, is_current)
-
-
-@dataclass(frozen=True)
-class PreparedPageCommit(PreparedSourceCommit):
-    history: TranscriptHistory
-
-    async def publish(self, cursor, is_current: Callable[[], bool]):
-        return await self.history.show_newest(PreparedReplacementViewportRequest(
-            LatestPosition.hold(self.history.window), self.page), publication_current=is_current)
+        return await operation.execute(history, partial(
+            history.show_newest,
+            PreparedReplacementViewportRequest(LatestPosition.hold(history.window), self.page),
+            publication_current=is_current,
+        ))
 
 
 class CommitPlan(ABC):
-    """A viewport-owned policy; the coordinator does not branch on block types."""
+    """The captured viewport intent owns source acquisition and publication."""
+
+    @abstractmethod
+    async def read_snapshot[Snapshot](
+        self, read: Callable[..., Awaitable[Snapshot]], history: CommittedHistory,
+    ) -> Snapshot:
+        """Acquire the saved destination this plan will publish."""
 
     @abstractmethod
     def current(self, window: HistoryWindow) -> bool:
@@ -312,18 +307,22 @@ def permits(self, view: Conversation, candidates: Iterable[Widget]) -> bool:
     def ready(self, history: CommittedHistory) -> bool:
         return history.source_ready
 
-    async def prepare(self, view: Conversation, history: CommittedHistory,
-                      page: TranscriptPage, captured: frozenset[Widget],
-                      is_current: Callable[[], bool]) -> PreparedCommit | None:
-        """Grow a history that accepts the page; otherwise the page replaces it."""
+    async def prepare(self, view: Conversation, history: TranscriptHistory,
+                      page: TranscriptPage, captured: frozenset[Widget]) -> PreparedCommit:
+        """Share exact wire coverage; each destination owns its render action."""
+        sequences = TranscriptCoverage(page.events).sequences
         if history.accepts_commit(page.after):
-            return await self.extend(view, history, page, captured, is_current)
-        return await PreparedPageCommit.capture(view, history, page, TranscriptCoverage(page.events).sequences)
+            required = required_sequences(captured)
+            sequences |= history.covered_sequences(required) | frozenset(
+                sequence for sequence in required if page.after.covers_incoming(sequence)
+            )
+            return await self.extend(view, history, page, sequences)
+        return await PreparedPageCommit.capture(view, history, page, sequences)
 
     @abstractmethod
-    async def extend(self, view: Conversation, history: CommittedHistory, page: TranscriptPage,
-                     captured: frozenset[Widget], is_current: Callable[[], bool]) -> PreparedCommit | None:
-        """Commit the page's records onto the history that already ends where it starts."""
+    async def extend(self, view: Conversation, history: TranscriptHistory, page: TranscriptPage,
+                     sequences: frozenset[int]) -> PreparedCommit:
+        """Prepare publication within the history's existing source."""
 
     def commit(self, prepared: PreparedCommit, cursor: TranscriptCursor) -> None:
         """Apply source metadata only after the final retirement checks."""
@@ -340,12 +339,18 @@ class FollowTailCommit(CommitPlan):
     def __init__(self, window: HistoryWindow) -> None:
         self.revision = window.scroll_revision
 
+    async def read_snapshot[Snapshot](
+        self, read: Callable[..., Awaitable[Snapshot]], history: CommittedHistory,
+    ) -> Snapshot:
+        # No after cursor: Core reads newest-first at its captured saved cut.
+        return await read()
+
     def current(self, window: HistoryWindow) -> bool:
         return window.follows_tail and window.scroll_revision == self.revision
 
-    async def extend(self, view, history, page, captured, is_current: Callable[[], bool]):
-        return await PreparedAdvanceCommit.capture(view, history, page, TranscriptCoverage(page.events).sequences
-                                                   | history.covered_sequences(required_sequences(captured)))
+    async def extend(self, view: Conversation, history: TranscriptHistory, page: TranscriptPage,
+                     sequences: frozenset[int]) -> PreparedPageCommit:
+        return await PreparedPageCommit.capture(view, history, page, sequences)
 
     def finish(self, view, cursor) -> None:
         view.transcript.binding.painted(view.transcript, cursor)
@@ -358,6 +363,15 @@ def restore_reader(self, window) -> None:
             window.anchor()
 
 class RetainViewportCommit(CommitPlan):
+    async def read_snapshot[Snapshot](
+        self, read: Callable[..., Awaitable[Snapshot]], history: CommittedHistory,
+    ) -> Snapshot:
+        try:
+            return await read(after=history.page_source.through)
+        except TranscriptChanged:
+            # A replaced source cannot extend the old reader's source range.
+            return await read()
+
     def current(self, window: HistoryWindow) -> bool:
         # Extending the source leaves the loaded rows and reader position intact.
         # Retirement checks the current viewport again when applying the change.
@@ -378,12 +392,9 @@ def permits(self, view, candidates) -> bool:
         return not any(widget in visible and visible[widget][0].overlaps(viewport)
                        for widget in candidates)
 
-    async def extend(self, view, history, page, captured, is_current: Callable[[], bool]):
-        required = required_sequences(captured)
-        known = TranscriptCoverage(page.events).sequences | history.covered_sequences(required)
-        covered = frozenset(sequence for sequence in required
-                            if page.after.covers_incoming(sequence))
-        return PreparedCommit(history, known | covered)
+    async def extend(self, view: Conversation, history: TranscriptHistory, page: TranscriptPage,
+                     sequences: frozenset[int]) -> PreparedCommit:
+        return PreparedCommit(history, sequences)
 
     def commit(self, prepared, cursor) -> None:
         prepared.history.retain_committed(cursor)

src/toad/widgets/transcript_history.py

@@ -1,3 +1,3 @@
 """Bounded, cursor-paged saved history; all transcript interpretation is model-owned."""
 
 from __future__ import annotations
@@ -1240,26 +1240,6 @@
             raise ValueError("Commit replaces or rewinds the retained source")
         self.extend_source(through)
 
-    async def advance_committed(
-        self, through: TranscriptCursor, is_current: Callable[[], bool],
-    ) -> bool:
-        """Extend a complete followed tail; read a bounded destination across a gap.
-
-        The prepared commit owns the shared live-block handoff operation;
-        this source changes only its followed saved range.
-        """
-        if not is_current() or not self.state.can_draw:
-            return False
-        if not self.accepts_commit(through):
-            raise ValueError("Commit replaces or rewinds the saved source")
-        if not NEWER_END.visible_more(self):
-            return await self.extend(through, is_current)
-        self.retain_committed(through)
-        request = LatestViewportRequest(LatestPosition.hold(self.window))
-        return await self.jump_latest(request, is_current)
-
-
-
 class ConversationHistory[ViewT: "Conversation"](LineHistory):
     """A conversation owns its source binding and initial history state."""