From ea0fffa58a6beba699d8d72152986e8f179654c3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 30 Oct 2026 13:44:40 -0400 Subject: [PATCH 1/2] Keep identity routes and page loaders on their owning families --- src/toad/screens/main.py | 3 +- src/toad/session_scope.py | 2 +- src/toad/thread_identity.py | 10 ++++ src/toad/transcript_preparation.py | 74 +++++++++++++++++------------- src/toad/wire_history.py | 22 +++++---- 5 files changed, 68 insertions(+), 43 deletions(-) diff --git a/src/toad/screens/main.py b/src/toad/screens/main.py index c0aa182e2..2c3cf3da0 100644 --- a/src/toad/screens/main.py +++ b/src/toad/screens/main.py @@ -233,11 +233,10 @@ def read_visible(self, screen: MainScreen) -> PaintedRead: access = app.coordination_access identity = screen.thread_identity project = screen.project_path - source_root = identity.wire_root() worktree = app.project_dir def acknowledge(cursor) -> None: - selected = access.selected(source_root) + selected = identity.selected_route(access) comms = access.require(selected).service def mark() -> None: diff --git a/src/toad/session_scope.py b/src/toad/session_scope.py index 3783818e3..e6b1133b2 100644 --- a/src/toad/session_scope.py +++ b/src/toad/session_scope.py @@ -406,7 +406,7 @@ def require(self) -> ProjectSessionOrigin: def resolve_thread(self) -> str: identity = self.thread_identity - selected = self.app.coordination_access.selected(identity.wire_root()) + selected = identity.selected_route(self.app.coordination_access) service = self.app.coordination_access.require(selected).service return identity.resolve_thread(service.registry, self.project_root) diff --git a/src/toad/thread_identity.py b/src/toad/thread_identity.py index b4b1539dd..d144e9a2f 100644 --- a/src/toad/thread_identity.py +++ b/src/toad/thread_identity.py @@ -9,8 +9,10 @@ from typing import TYPE_CHECKING from agent_comms.errors import UnregisteredThreadError +from agent_comms.route_selection import RouteSelection from agent_comms.thread_identity import ThreadIncarnation, UnrecordedThreadIncarnation +from toad.comms_root import CoordinationAccess from toad.format_path import format_path from toad.session_scope import KnownTranscript, NoTranscript, TranscriptScope from toad.transcript_state import UnboundTranscript, UnreadTranscript @@ -45,6 +47,10 @@ def with_session(self, session_id: str): def wire_root(self) -> str | None: return None + def selected_route(self, access: CoordinationAccess) -> RouteSelection: + """An unbound identity resolves within the app's own route.""" + return access.selection + def transcript_scope(self) -> TranscriptScope: return NoTranscript() @@ -157,6 +163,10 @@ def bind_history(self, histories) -> None: def wire_root(self) -> str: return self.root + def selected_route(self, access: CoordinationAccess) -> RouteSelection: + """A managed identity requires its recorded wire source.""" + return access.selected(self.root) + def transcript_scope(self) -> KnownTranscript: return KnownTranscript(self.root, self.thread_name()) diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py index 6738dda57..45d02501f 100644 --- a/src/toad/transcript_preparation.py +++ b/src/toad/transcript_preparation.py @@ -449,9 +449,8 @@ def __init__(self, request: PageRequest, through: TranscriptCursor | None, page: @dataclass(frozen=True) -class TranscriptPageWork[Cursor = TranscriptCursor | None](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]): +class PreparedPageWork[Cursor = TranscriptCursor | None](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]): scope: PreparationScope - loader: Callable[..., Awaitable[TranscriptPage]] through: Cursor request: PageRequest[Cursor] lane = WorkLane.MODEL @@ -464,6 +463,11 @@ def work_key(self) -> WorkKey: def result_size(self, result: PreparedTranscriptPage) -> int: return result.retained_bytes + +@dataclass(frozen=True) +class TranscriptPageWork[Cursor = TranscriptCursor | None](PreparedPageWork[Cursor]): + loader: Callable[..., Awaitable[TranscriptPage]] + async def execute( self: TranscriptPageWork[TranscriptCursor | None], runtime: PreparationRuntime, key: WorkKey, ) -> PreparedResult[PreparedTranscriptPage]: @@ -495,56 +499,64 @@ class ReplaceablePageSource(PreparedPageSource): def replacement(self, through) -> ReplaceablePageSource: ... -class SavedHistory(ReplaceablePageSource): - """Prepared pages for a snapshot; storage and work limits belong to the runtime. +class SavedPageSource(ReplaceablePageSource): + """Snapshot preparation and lifetime shared by native and wire readers. - The loader retains ownership of transcript/routing semantics and off-loop - I/O. CPU preparation uses the application's existing renderer. Foreground - consumers share in-flight requests; cancelling one waiter cannot release a - still-running read's slot or publish it into a closed view. + Each member owns its actual loader and work. Cancelling a foreground waiter + cannot release a still-running read's slot or publish into a closed view. """ - # The work that reads and prepares one page of this kind of source. - work: ClassVar[type[TranscriptPageWork]] = TranscriptPageWork - - def __init__( - self, loader: Callable[..., Awaitable[TranscriptPage]], through: TranscriptCursor, - runtime: PreparationRuntime, - ) -> None: + def __init__(self, through: TranscriptCursor, runtime: PreparationRuntime) -> None: super().__init__() - self.loader, self.through, self.runtime = loader, through, runtime + self.through, self.runtime = through, runtime self.scope = PreparationScope() def prefix(self, through) -> PreparedPageSource: - return SavedPrefix(self, self.loader, through, self.runtime) - - def replacement(self, through) -> SavedHistory: - return type(self)(self.loader, through, self.runtime) + return SavedPrefix(self, through, self.runtime) - def read_task(self, request: PageRequest) -> TranscriptPageWork: - return self.work(self.scope, self.loader, self.through, request) + @abstractmethod + def read_task( + self, request: PageRequest, scope: PreparationScope, through: TranscriptCursor, + ) -> ReusableWork[PreparedTranscriptPage]: ... async def _get(self, request: PageRequest) -> PreparedTranscriptPage: - result = await self.runtime.submit(self.read_task(request)) + result = await self.runtime.submit(self.read_task(request, self.scope, self.through)) return result.for_view() async def _prepare(self, request: PageRequest) -> None: - await self.runtime.prepare(self.read_task(request)) + await self.runtime.prepare(self.read_task(request, self.scope, self.through)) -class SavedPrefix(SavedHistory): - def __init__(self, earlier: SavedHistory, loader: Callable[..., Awaitable[TranscriptPage]], through, runtime: PreparationRuntime) -> None: - super().__init__(loader, through, runtime) - self.earlier = earlier +class SavedHistory(SavedPageSource): + """Native transcript pages read by the source's asynchronous loader.""" + + def __init__( + self, loader: Callable[..., Awaitable[TranscriptPage]], through: TranscriptCursor, + runtime: PreparationRuntime, + ) -> None: + super().__init__(through, runtime) + self.loader = loader def replacement(self, through) -> SavedHistory: + return type(self)(self.loader, through, self.runtime) + + def read_task(self, request: PageRequest, scope: PreparationScope, through: TranscriptCursor) -> TranscriptPageWork: + return TranscriptPageWork(scope=scope, loader=self.loader, through=through, request=request) + + +class SavedPrefix(SavedPageSource): + def __init__(self, earlier: SavedPageSource, through, runtime: PreparationRuntime) -> None: + super().__init__(through, runtime) + self.earlier = earlier + + def replacement(self, through) -> ReplaceablePageSource: # A replacement reads the complete source; it must not delegate older # reads to the superseded prefix's preparation scope. return self.earlier.replacement(through) - 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) + def read_task(self, request: PageRequest, scope: PreparationScope, through: TranscriptCursor): + # The original source supplies its native/wire work in this prefix's scope. + return self.earlier.read_task(request, scope, through) async def _get(self, request: PageRequest): return await request.end.edge((self.earlier.get, super()._get))(request) diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index 04505e2ea..8c707866c 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -40,9 +40,9 @@ PageRequest, PreparedTranscriptPage, ReplaceablePageSource, - SavedHistory, + PreparedPageWork, + SavedPageSource, SourceObservation, - TranscriptPageWork, Unchanged, ) from toad.transcript_state import ( @@ -386,8 +386,10 @@ 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(PreparedPageWork[WireCut]): + """Read one wire page off the UI thread from its typed source.""" + + loader: WireSource @property def retain_result(self) -> bool: @@ -402,15 +404,17 @@ async def execute(self, runtime, key): return await runtime.run_thread(self.finish_result, key, prepared) -class WirePageBuffer(SavedHistory): +class WirePageBuffer(SavedPageSource): """The line history's page reader over a wire source.""" - work = WirePageWork - - def __init__(self, *args, revision: ConversationObservation, **kwargs): - super().__init__(*args, **kwargs) + def __init__(self, loader: WireSource, through, runtime, *, revision: ConversationObservation): + super().__init__(through, runtime) + self.loader = loader self.revision = revision + def read_task(self, request, scope, through) -> WirePageWork: + return WirePageWork(scope=scope, loader=self.loader, through=through, request=request) + def replacement(self, through) -> WirePageBuffer: return type(self)(self.loader, through, self.runtime, revision=self.revision) From 16c91d9c76a6c66170a9252637fea477a454e3d7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 30 Oct 2026 15:49:38 -0400 Subject: [PATCH 2/2] Clear the NRA gate growth this repair introduced Annotate the wire page reader's new parameters and drop the second `| None` default on TranscriptPageWork, which only restated PreparedPageWork's. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XbFMSpSd8zQxyozgEV2Re7 --- src/toad/transcript_preparation.py | 2 +- src/toad/wire_history.py | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py index 45d02501f..1a6ecd4ed 100644 --- a/src/toad/transcript_preparation.py +++ b/src/toad/transcript_preparation.py @@ -465,7 +465,7 @@ def result_size(self, result: PreparedTranscriptPage) -> int: @dataclass(frozen=True) -class TranscriptPageWork[Cursor = TranscriptCursor | None](PreparedPageWork[Cursor]): +class TranscriptPageWork[Cursor](PreparedPageWork[Cursor]): loader: Callable[..., Awaitable[TranscriptPage]] async def execute( diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index 8c707866c..d741ae986 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -38,6 +38,8 @@ from toad.message_rows import MessageRow, WireMessageStyle from toad.transcript_preparation import ( PageRequest, + PreparationRuntime, + PreparationScope, PreparedTranscriptPage, ReplaceablePageSource, PreparedPageWork, @@ -407,12 +409,12 @@ async def execute(self, runtime, key): class WirePageBuffer(SavedPageSource): """The line history's page reader over a wire source.""" - def __init__(self, loader: WireSource, through, runtime, *, revision: ConversationObservation): + def __init__(self, loader: WireSource, through: WireCut, runtime: PreparationRuntime, *, revision: ConversationObservation): super().__init__(through, runtime) self.loader = loader self.revision = revision - def read_task(self, request, scope, through) -> WirePageWork: + def read_task(self, request: PageRequest, scope: PreparationScope, through: WireCut) -> WirePageWork: return WirePageWork(scope=scope, loader=self.loader, through=through, request=request) def replacement(self, through) -> WirePageBuffer: