diff --git a/src/toad/transcript_publication.py b/src/toad/transcript_publication.py index 6e3e7e206..75f7bc1ec 100644 --- a/src/toad/transcript_publication.py +++ b/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, diff --git a/src/toad/widgets/committed_presentation.py b/src/toad/widgets/committed_presentation.py index 218bd5f2f..54100346f 100644 --- a/src/toad/widgets/committed_presentation.py +++ b/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) diff --git a/src/toad/widgets/transcript_history.py b/src/toad/widgets/transcript_history.py index ba7afc168..afbec9ccf 100644 --- a/src/toad/widgets/transcript_history.py +++ b/src/toad/widgets/transcript_history.py @@ -1240,26 +1240,6 @@ def retain_committed(self, through: TranscriptCursor) -> None: 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."""