From 3772cf0f796cba739a1c6103eeeefcf77ed9e797 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 06:33:59 -0400 Subject: [PATCH 1/9] Input delivery reads through Core's one owner-live operation (agent-comms#3538) ReadDeliveryOverview restated Core's input-delivery read with its own include_history; Core deleted it. Toad now sends ReadOwnerOperation carrying InputDeliveryRead(include_history=...) to the held owner and wraps the owner's DeliveryOverview in ReadDeliveryView. No-coordination DeliveryView, require_owner and dismiss_historical_inputs are unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- src/toad/acp/agent_controller.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/toad/acp/agent_controller.py b/src/toad/acp/agent_controller.py index 91057609d..c2302c28d 100644 --- a/src/toad/acp/agent_controller.py +++ b/src/toad/acp/agent_controller.py @@ -22,8 +22,9 @@ from agent_comms.acp_failure import ACPFailure, RequestFailure from agent_comms.coordination_errors import Refused from agent_comms.transcripts import TranscriptReadIdentity -from agent_comms.ui_model.delivery import DeliveryView -from agent_comms.ui_model.observation import ReadDeliveryOverview +from agent_comms.order_tools import InputDeliveryRead +from agent_comms.ui_model.delivery import DeliveryView, ReadDeliveryView +from agent_comms.ui_model.observation import ReadOwnerOperation from toad.acp.client_session import ClientSessionRequest from toad.acp.status import StopReason @@ -430,11 +431,11 @@ async def input_delivery(self, *, include_history: bool = False) -> DeliveryView if coordination is None: return DeliveryView() authority = ClientSessionRequest(self.agent, self.agent.session_id) - delivery = await self.transcripts.read( - ReadDeliveryOverview(coordination.wire_root, coordination.thread, include_history=include_history) - ) + overview = await self.transcripts.read(ReadOwnerOperation( + coordination.wire_root, coordination.thread, InputDeliveryRead(include_history=include_history), + )) self.require_owner(coordination, authority) - return delivery + return ReadDeliveryView(overview) async def dismiss_historical_inputs(self): return await self.request_owner("dismiss_historical_inputs") From 0be58d8a3e560b80f55400af10707f4568253fe9 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 07:11:08 -0400 Subject: [PATCH 2/9] Native Toad reads input delivery from the live owner queue (agent-comms#3538) tests/native/test_native_goal_owner.py: with the real native session, after until_agent_ready(), controller.input_delivery() and controller.input_delivery(include_history=True) are each a ReadDeliveryView whose overview has current_scope OwnerQueueInputDeliveryScope (only the live owner yields it; the no-coordination DeliveryView and the unobserved projection never do), and on the fresh private wire empty inputs and historical_inputs with zero counts. No mocked read, operation, owner or reply. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- tests/native/test_native_goal_owner.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/native/test_native_goal_owner.py b/tests/native/test_native_goal_owner.py index 8fc1016d1..8c7cf29dd 100644 --- a/tests/native/test_native_goal_owner.py +++ b/tests/native/test_native_goal_owner.py @@ -3,6 +3,8 @@ import pytest from agent_comms.goal_actions import SetGoalAction from agent_comms.goal_history import RecordedGoalHistory +from agent_comms.input_attempt import OwnerQueueInputDeliveryScope +from agent_comms.ui_model.delivery import ReadDeliveryView from support.native import native_session @@ -37,3 +39,18 @@ async def test_editing_a_goal_keeps_its_identity_and_refuses_a_stale_edit(worksp for entry in history.history) snapshot = await agent.get_goal_snapshot() assert snapshot.goal == changed and snapshot.execution.goal_id == changed.id + + +async def test_input_delivery_is_read_from_the_live_native_owner_queue(workspace, native_package): + """The owner_queue scope can only come from the live owner; the no-coordination + DeliveryView and the unobserved projection never carry it.""" + async with native_session(workspace) as native: + await native.ui.until_agent_ready() + controller = native.agent.controller + for delivery in (await controller.input_delivery(), + await controller.input_delivery(include_history=True)): + assert type(delivery) is ReadDeliveryView + overview = delivery.overview + assert overview.current_scope is OwnerQueueInputDeliveryScope + assert overview.inputs == () and overview.historical_inputs == () + assert overview.historical_count == 0 and overview.dismissed_historical_count == 0 From a9a330ba6d6ac9ae93ba11d151ad7cf25f816d74 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 08:26:30 -0400 Subject: [PATCH 3/9] Notification views build a root-free operation; one envelope binds the root (agent-comms#3538) Core moved the notification read off the UI request: LiveNotificationRead and RecordedNotificationRead own references (and the archive), and ReadNotifications(root, operation) is the one UI envelope. - HistoryNotifications.notification_request (a ReadNotifications class used as a root+references constructor) becomes notification_operation, a root-free operation constructor (LiveNotificationRead by default). notification_read binds the ProjectPathOwner root: ReadNotifications(root, operation(refs)). - RecordedContents sets partial(RecordedNotificationRead, source=source). ReadRecordedNotifications is gone. - comms_root keeps keying views and outcomes by the full UI request. - tests: the historical source test requires the generic envelope with RecordedNotificationRead((reference,), source.key); the observation shutdown test builds its request with LiveNotificationRead. Assertions unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- src/toad/screens/historical_sessions.py | 5 +++-- src/toad/widgets/message_notifications.py | 5 +++-- tests/textual/test_historical_notification_source.py | 8 ++++---- tests/textual/test_observation_shutdown.py | 3 ++- 4 files changed, 12 insertions(+), 9 deletions(-) diff --git a/src/toad/screens/historical_sessions.py b/src/toad/screens/historical_sessions.py index 2395cc0a4..367524055 100644 --- a/src/toad/screens/historical_sessions.py +++ b/src/toad/screens/historical_sessions.py @@ -9,7 +9,8 @@ from agent_comms.comms import Comms from agent_comms.coordination_errors import StaleRevision from metaclass_registry.mro_dispatch import handles -from agent_comms.ui_model.observation import ReadTranscript, ReadRecordedNotifications +from agent_comms.notification_reads import RecordedNotificationRead +from agent_comms.ui_model.observation import ReadTranscript from toad.core_event_carrier import CoreEventMessage, CoreEventReceiver from agent_comms.historical_views import HistoricalThread from textual import on, work @@ -29,7 +30,7 @@ class RecordedContents(Contents): """The notification subscriber reads the same archive as its history.""" def __init__(self, source: str, **kwargs): - self.notification_request = partial(ReadRecordedNotifications, source=source) + self.notification_operation = partial(RecordedNotificationRead, source=source) super().__init__(**kwargs) diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py index d5a2fe5d5..4f73df4fb 100644 --- a/src/toad/widgets/message_notifications.py +++ b/src/toad/widgets/message_notifications.py @@ -11,6 +11,7 @@ from textual.content import Content from textual.widgets import Collapsible +from agent_comms.notification_reads import LiveNotificationRead from agent_comms.ui_model.observation import ReadNotifications from toad.notification_status import ( @@ -110,7 +111,7 @@ class HistoryNotifications: """Visible history rows subscribe to Core; paging never decides when outcomes changed.""" NOTIFICATION_GROUP = "message-notifications" - notification_request = ReadNotifications + notification_operation = LiveNotificationRead # Core feedback for every visible bus message, native or wire. @@ -126,7 +127,7 @@ def notification_read(self, references) -> ReadNotifications: root = ProjectPathOwner.containing(self).coordination_root if root is None: raise ValueError("Notification observation requires a bound thread root") - return self.notification_request(root, references) + return ReadNotifications(root, self.notification_operation(references)) def subscribe_notifications(self, references) -> None: access = self.app.coordination_access diff --git a/tests/textual/test_historical_notification_source.py b/tests/textual/test_historical_notification_source.py index 4959f5871..307672f37 100644 --- a/tests/textual/test_historical_notification_source.py +++ b/tests/textual/test_historical_notification_source.py @@ -2,7 +2,8 @@ from agent_comms.repo import Checkout from agent_comms.historical_views import HistoricalThread -from agent_comms.ui_model.observation import ReadRecordedNotifications +from agent_comms.notification_reads import RecordedNotificationRead +from agent_comms.ui_model.observation import ReadNotifications from agent_comms.threads import Thread from support.roots import installed_wire @@ -28,9 +29,8 @@ async def test_historical_notification_keeps_selected_source_and_lifetime(worksp await ui.until(lambda: content in access.notification_views, why="the historical body's notification subscription") request = access.notification_views[content] - assert isinstance(request, ReadRecordedNotifications) - assert request.source == source.key - assert request.references == (message.reference,) + assert type(request) is ReadNotifications + assert request.operation == RecordedNotificationRead((message.reference,), source.key) await ui.until(lambda: request in access.notification_outcomes, why="the archive's actual observation reply") outcomes = access.notification_outcomes[request] diff --git a/tests/textual/test_observation_shutdown.py b/tests/textual/test_observation_shutdown.py index 4e4aa236c..90bf272ca 100644 --- a/tests/textual/test_observation_shutdown.py +++ b/tests/textual/test_observation_shutdown.py @@ -4,6 +4,7 @@ import sys import pytest +from agent_comms.notification_reads import LiveNotificationRead from agent_comms.ui_model.observation import ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications from toad.comms_root import AnsweringReads @@ -45,7 +46,7 @@ async def test_stopping_observation_keeps_reads_but_sends_no_late_interest(works process = observing.process loop = asyncio.get_running_loop() acquired = asyncio.Event() - request = ReadNotifications(str(workspace.roots.wire), (message.reference,)) + request = ReadNotifications(str(workspace.roots.wire), LiveNotificationRead((message.reference,))) async def read_notifications(): acquired.set() From 54db5d11ad7e320ddd1d74cb03fc0873078bc823 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 08:54:05 -0400 Subject: [PATCH 4/9] Shutdown tests hold the real observer child; paint fixtures read the operation's references (agent-comms#3538) Both shutdown tests called process.fileno() to unhook a readiness callback that ObservationProcess no longer has. The one receiver started at activation keeps running after stop_observing, so the answer has to be held where it is made: in this test's own observer child. - test_stopping_observation_keeps_reads_but_sends_no_late_interest proves the child is ready with an empty-window read and SIGSTOPs it (stop confirmed by a bounded WNOHANG|WUNTRACED wait). It then registers the real read, stops observing with every AnsweringReads identity check and the no-late-interest profile, and SIGCONTs the child. The same receiver settles the read: set(result.outcomes) is the message's seq/id. - test_an_observer_dying_during_shutdown_still_releases_its_resources holds the child, registers a pending empty-window read and SIGKILLs the child. - The read fails with ObserverDied carrying that pid and -SIGKILL. - StopSidebar and StopObserving each raise that same object. - Teardown may report it only as that same cause, and other grouped errors propagate. - The connection is closed, the child is gone and nothing is retiring. - Every path resumes a still-stopped live child and settles any unconsumed read. - The four paint-fixture reads of request.references now read request.operation.references; ReadNotifications has no references of its own. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- tests/textual/test_message_notifications.py | 8 +- tests/textual/test_observation_shutdown.py | 173 +++++++++++++------- 2 files changed, 118 insertions(+), 63 deletions(-) diff --git a/tests/textual/test_message_notifications.py b/tests/textual/test_message_notifications.py index 670fa0bc2..7e9908783 100644 --- a/tests/textual/test_message_notifications.py +++ b/tests/textual/test_message_notifications.py @@ -35,7 +35,7 @@ async def publish(self, ui, chat): view = chat.message_history.notification_view access = ui.app.coordination_access request = access.notification_views[view] - outcomes = self.answer(request.references) + outcomes = self.answer(request.operation.references) expected, = CheckedNotifications.for_references( chat.message_history.visible_fragments()[-1].notification_references, outcomes) access.notifications_observed(NotificationsObserved(access.revision, request, outcomes), access.observation.observed()) @@ -88,8 +88,8 @@ async def test_visible_references_are_bounded_and_typed_outcomes_paint_their_det request = ui.app.coordination_access.notification_views[history.notification_view] on_screen = {reference for fragment in history.visible_fragments() for reference in fragment.notification_references} - assert 0 < len(request.references) <= len(on_screen) < 30 - assert set(request.references) <= on_screen + assert 0 < len(request.operation.references) <= len(on_screen) < 30 + assert set(request.operation.references) <= on_screen await notifications.publish(ui, chat) await ui.until_shows("Delivered 45/45 · Consumed 0/45 · Read 0/45") assert newest_status(chat).working @@ -144,7 +144,7 @@ async def test_a_hidden_view_unsubscribes_and_discards_a_late_reply(workspace): await ui.select_tab(owner) assert view not in ui.app.coordination_access.notification_views notifications.state = CompletedProgress(ShownToModelProgress(0)) - view.notifications_observed(notifications.answer(request.references)) + view.notifications_observed(notifications.answer(request.operation.references)) await ui.step() assert view not in ui.app.coordination_access.notification_views assert {fragment.message.reference: fragment.shown.status diff --git a/tests/textual/test_observation_shutdown.py b/tests/textual/test_observation_shutdown.py index 90bf272ca..0a2b134db 100644 --- a/tests/textual/test_observation_shutdown.py +++ b/tests/textual/test_observation_shutdown.py @@ -1,36 +1,93 @@ """Textual-facing: widget unmount; unmount ends interest, not the real observer's already-acquired reads.""" import asyncio +import os +import signal import sys +from contextlib import asynccontextmanager + import pytest from agent_comms.notification_reads import LiveNotificationRead -from agent_comms.ui_model.observation import ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications +from agent_comms.ui_model.observation import ( + ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications, StopObserving, + StopSidebar, +) from toad.comms_root import AnsweringReads +async def stopped(pid: int) -> None: + """Wait, bounded, until the stop is reported; WUNTRACED reports it without reaping.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + 5 + while loop.time() < deadline: + found, status = os.waitpid(pid, os.WNOHANG | os.WUNTRACED) + if found: + assert os.WIFSTOPPED(status) and os.WSTOPSIG(status) == signal.SIGSTOP + return + await asyncio.sleep(0.01) + raise AssertionError(f"observer child {pid} did not stop") + + +@asynccontextmanager +async def held(process: ObservationProcess): + """Hold this test's own observer child; resume it on every path while it lives.""" + pid = process.process.pid + os.kill(pid, signal.SIGSTOP) + try: + await stopped(pid) + yield pid + finally: + if process.process.is_alive(): + os.kill(pid, signal.SIGCONT) + + +async def settled(task: asyncio.Task) -> None: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +def empty_read(workspace) -> ReadNotifications: + return ReadNotifications(str(workspace.roots.wire), LiveNotificationRead(())) + + async def test_an_observer_dying_during_shutdown_still_releases_its_resources(workspace): - with pytest.raises(BaseExceptionGroup) as failures: + died = None + try: async with workspace.app() as ui: access = ui.app.coordination_access await access.current_sidebar() - observed = access.observation.observed() - process = observed.process - # The death has not been delivered to receive(); stop_updates must - # encounter it at its actual StopObserving request during shutdown. - asyncio.get_running_loop().remove_reader(process.fileno()) - process.process.kill() - await asyncio.to_thread(process.process.join, 5) - assert process.process.exitcode is not None - assert not process.connection.closed - assert failures.value.subgroup(ObserverDied) is not None + process = access.observation.observed().process + assert not (await asyncio.wait_for(process.read(empty_read(workspace)), 30)).outcomes + reading = None + try: + async with held(process) as pid: + reading = asyncio.create_task(process.read(empty_read(workspace))) + await asyncio.sleep(0) + assert process.reads and not reading.done() + os.kill(pid, signal.SIGKILL) + with pytest.raises(ObserverDied) as death: + await asyncio.wait_for(reading, 10) + finally: + if reading is not None: + await settled(reading) + died = death.value + assert died.pid == pid + assert died.exit_status == -signal.SIGKILL + for request in (StopSidebar(), StopObserving()): + with pytest.raises(ObserverDied) as again: + process.request(request) + assert again.value is died + except* ObserverDied as failures: + assert died is not None + _same, other = failures.split(lambda error: error is died) + assert other is None assert process.connection.closed assert not process.process.is_alive() assert not access.retiring - - async def test_stopping_observation_keeps_reads_but_sends_no_late_interest(workspace): workspace.thread("sender", title="Sender") workspace.thread("reader", title="Reader") @@ -44,50 +101,48 @@ async def test_stopping_observation_keeps_reads_but_sends_no_late_interest(works # Retiring the old process must not publish its state over the new one. assert access.observation is observing process = observing.process - loop = asyncio.get_running_loop() - acquired = asyncio.Event() + assert not (await asyncio.wait_for(process.read(empty_read(workspace)), 30)).outcomes request = ReadNotifications(str(workspace.roots.wire), LiveNotificationRead((message.reference,))) - - async def read_notifications(): - acquired.set() - return await process.read(request) - - # Hold the readiness callback, not the service, so this acquired read - # cannot resolve until stop_observing installs the read-only reader. - loop.remove_reader(process.fileno()) - async with asyncio.TaskGroup() as reads: - reading = reads.create_task(read_notifications()) - await acquired.wait() - assert process.reads and not reading.done() - await access.stop_observing() - answering = access.observation - assert type(answering) is AnsweringReads - assert answering.process is process - assert answering.service is observing.service - assert answering.selection is observing.selection - assert answering.sidebar is observing.sidebar - assert answering.subscriptions is observing.subscriptions - assert answering.updates is observing.updates - - requests = [] - - def record_request(frame, event, result): - if (event == "call" and frame.f_code is ObservationProcess.request.__code__ - and frame.f_locals["self"] is process): - requests.append(frame.f_locals["request"]) - - previous = sys.getprofile() - sys.setprofile(record_request) - try: - answering.views_changed() - answering.thread_interest(access) - answering.observe_sidebar(access.sidebar_request()) - answering.observe_notifications(ObserveNotifications(frozenset())) + reading = None + try: + # The observer child itself holds the answer; the UI's one receiver keeps running. + async with held(process): + reading = asyncio.create_task(process.read(request)) + await asyncio.sleep(0) + assert process.reads and not reading.done() await access.stop_observing() - assert access.observation is answering - finally: - sys.setprofile(previous) - assert requests == [] - result = await reading - assert set(result) == {(message.seq, message.message_id)} - assert process.process.is_alive() + answering = access.observation + assert type(answering) is AnsweringReads + assert answering.process is process + assert answering.service is observing.service + assert answering.selection is observing.selection + assert answering.sidebar is observing.sidebar + assert answering.subscriptions is observing.subscriptions + assert answering.updates is observing.updates + + requests = [] + + def record_request(frame, event, result): + if (event == "call" and frame.f_code is ObservationProcess.request.__code__ + and frame.f_locals["self"] is process): + requests.append(frame.f_locals["request"]) + + previous = sys.getprofile() + sys.setprofile(record_request) + try: + answering.views_changed() + answering.thread_interest(access) + answering.observe_sidebar(access.sidebar_request()) + answering.observe_notifications(ObserveNotifications(frozenset())) + await access.stop_observing() + assert access.observation is answering + finally: + sys.setprofile(previous) + assert requests == [] + assert not reading.done() + result = await asyncio.wait_for(reading, 30) + finally: + if reading is not None: + await settled(reading) + assert set(result.outcomes) == {(message.seq, message.message_id)} + assert process.process.is_alive() From cdd6e15d612860a0978a73fbb14b1aedaf38dd93 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 10:19:02 -0400 Subject: [PATCH 5/9] Owner presentation reads leave their worker as the published projection (agent-comms#3538) read_thread_presentation returned the acquired presentation to Conversation and ObservedSourcePublication, carrying the notification window and backend records into the UI. It now returns .published(), the source-free ThreadPresentation; settlement still reads its read_identity. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- src/toad/owner_preparation.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/toad/owner_preparation.py b/src/toad/owner_preparation.py index 04a5c8eb1..3e2372860 100644 --- a/src/toad/owner_preparation.py +++ b/src/toad/owner_preparation.py @@ -68,5 +68,6 @@ class OwnerRequestContext: def read_thread_presentation(comms: Comms, name: str) -> ThreadPresentation: - """Read the existing core presentation on a worker, without UI dependencies.""" - return comms.views.thread_presentations((name,)).read(name) + """Read the existing core presentation on a worker, without UI dependencies; only its + published projection leaves the worker, never the acquired delivery evidence.""" + return comms.views.thread_presentations((name,)).read(name).published() From 9b6c482e2d56ea59cb07b0437e10125f17da1236 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 12:05:58 -0400 Subject: [PATCH 6/9] Commander detail reads through the shared comms_commander_unit operation (#1302) The commander page wraps CommanderUnitRead(commander, unit, orders) in the UI's ReadOperation envelope instead of the deleted ReadCommanderUnit. Request id, root, card/collapsed fencing, worker cancellation and the unavailable-detail display are unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- src/toad/screens/commander.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/toad/screens/commander.py b/src/toad/screens/commander.py index 5be6bf5e4..4528d3af2 100644 --- a/src/toad/screens/commander.py +++ b/src/toad/screens/commander.py @@ -11,7 +11,8 @@ from agent_comms.coordination_errors import DeferredRead from agent_comms.errors import RelationViolationError, RouteChangedError, ToolRefusedError from agent_comms.thread_identity import ThreadIncarnation -from agent_comms.ui_model.observation import ReadCommanderUnit, UnreadableHistory +from agent_comms.order_tools import CommanderUnitRead +from agent_comms.ui_model.observation import ReadOperation, UnreadableHistory from agent_comms.work_branches import WorkBranchError from metaclass_registry.mro_dispatch import MroDispatch, handles from textual import on, work @@ -118,10 +119,10 @@ async def read_detail(self) -> None: access = self.app.coordination_access try: observed = await access.preparation.run_thread(access.observation.observed) - request = ReadCommanderUnit( - str(observed.service.root), self.commander, card.unit, - tuple(dict.fromkeys(order.order for order in card.orders)), - ) + request = ReadOperation(str(observed.service.root), CommanderUnitRead( + commander=self.commander, unit=card.unit, + orders=tuple(dict.fromkeys(order.order for order in card.orders)), + )) detail = await observed.process.read(request) except ( DeferredRead, RelationViolationError, RouteChangedError, From 6f41bc6f9f88e14329b6731c03e942c64e85cbab Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 12:09:56 -0400 Subject: [PATCH 7/9] The goal snapshot reads through the shared comms_owner_goal operation (#1302) acp/agent.get_goal_snapshot sends ReadOwnerOperation(wire_root, thread, operation=OwnerGoalRead()) through its existing transcript reader instead of the deleted ReadOwnerGoal. The local NoGoal/Idle answer without coordination, the ClientSessionRequest authority, OwnerSnapshotConsumer's captured turn token, the post-read require_owner and dispatch of every meta.batch update are unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- src/toad/acp/agent.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/toad/acp/agent.py b/src/toad/acp/agent.py index db02cf442..cb0c79bbe 100644 --- a/src/toad/acp/agent.py +++ b/src/toad/acp/agent.py @@ -15,7 +15,8 @@ from agent_comms.goal_presentation import GoalSnapshot, IdleGoalExecution from agent_comms.goals import Goal, GoalDeclaration, NoGoal from agent_comms.goal_history import RecordedGoalHistory -from agent_comms.ui_model.observation import ReadOwnerGoal +from agent_comms.order_tools import OwnerGoalRead +from agent_comms.ui_model.observation import ReadOwnerOperation from toad import paths from toad.acp.agent_configuration import ( @@ -238,7 +239,7 @@ async def get_goal_snapshot(self) -> GoalSnapshot: authority = ClientSessionRequest(self, self.session_id) consumer = OwnerSnapshotConsumer(self, self.session_id, turn_token=self.presentation.turns.sequence) snapshot = await self.controller.transcripts.read( - ReadOwnerGoal(coordination.wire_root, coordination.thread)) + ReadOwnerOperation(coordination.wire_root, coordination.thread, operation=OwnerGoalRead())) self.controller.require_owner(coordination, authority) for fact in snapshot.meta.batch.updates: consumer.dispatch_sync(fact) From 9572b2f2c45923d6f065e846287617eceef47c51 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 12:18:24 -0400 Subject: [PATCH 8/9] Transcript reads carry the shared comms_transcript_page operation (#1302) KnownTranscript.snapshot_request and update_request, and the historical sessions screen, build their UI envelope around TranscriptPageRead with every original cursor and the attached source key; root and publication state stay on the envelope. acp/transcript_reader's RefreshTranscript keeps its original identity. A recorded-reader test asserts the exact page arguments on the operation and root/previous on the envelope. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- src/toad/screens/historical_sessions.py | 5 +++-- src/toad/session_scope.py | 12 +++++++----- tests/app/test_recorded_transcript_reader.py | 20 +++++++++++++++++++- 3 files changed, 29 insertions(+), 8 deletions(-) diff --git a/src/toad/screens/historical_sessions.py b/src/toad/screens/historical_sessions.py index 367524055..7125d6c46 100644 --- a/src/toad/screens/historical_sessions.py +++ b/src/toad/screens/historical_sessions.py @@ -10,6 +10,7 @@ from agent_comms.coordination_errors import StaleRevision from metaclass_registry.mro_dispatch import handles from agent_comms.notification_reads import RecordedNotificationRead +from agent_comms.order_tools import TranscriptPageRead from agent_comms.ui_model.observation import ReadTranscript from toad.core_event_carrier import CoreEventMessage, CoreEventReceiver from agent_comms.historical_views import HistoricalThread @@ -122,8 +123,8 @@ async def selected(self, event: Select.Changed) -> None: access, root = self.app.coordination_access, str(self.comms.root) async def load(**kwargs): - snapshot = await access.transcripts.read(ReadTranscript( - root, item.thread.name, historical_source=item.source.key, **kwargs)) + snapshot = await access.transcripts.read(ReadTranscript(root, TranscriptPageRead( + thread=item.thread.name, historical_source=item.source.key, **kwargs))) return snapshot.page try: diff --git a/src/toad/session_scope.py b/src/toad/session_scope.py index fed3c6f28..a3c9d0375 100644 --- a/src/toad/session_scope.py +++ b/src/toad/session_scope.py @@ -15,6 +15,7 @@ from agent_comms.errors import RelationViolationError from agent_comms.route_selection import RouteSelection from agent_comms.thread_identity import ThreadIncarnation +from agent_comms.order_tools import TranscriptPageRead from agent_comms.ui_model.observation import ( ReadTranscript, ReadTranscriptUpdate, @@ -158,14 +159,13 @@ def snapshot_request( through: TranscriptCursor | None = None, historical_source: str | None = None, ) -> ReadTranscript: - return ReadTranscript( - self.root, - self.name, + return ReadTranscript(self.root, TranscriptPageRead( + thread=self.name, before=before, after=after, through=through, historical_source=historical_source, - ) + )) def update_request( self, @@ -173,7 +173,9 @@ def update_request( *, after: TranscriptCursor | None = None, ) -> ReadTranscriptUpdate: - return ReadTranscriptUpdate(self.root, self.name, previous=previous, after=after) + return ReadTranscriptUpdate( + self.root, TranscriptPageRead(thread=self.name, after=after), previous=previous + ) def mailbox_target(self) -> ThreadTarget: return ThreadTarget(self.name) diff --git a/tests/app/test_recorded_transcript_reader.py b/tests/app/test_recorded_transcript_reader.py index 09fbe55ed..1263b70f9 100644 --- a/tests/app/test_recorded_transcript_reader.py +++ b/tests/app/test_recorded_transcript_reader.py @@ -2,15 +2,17 @@ from agent_comms.repo import Checkout import json -from dataclasses import replace +from dataclasses import fields, replace import pytest from agent_comms.acp_extension import TranscriptSnapshotUpdate, decode_updates, encode_updates from agent_comms.coordination_errors import HistoricalSourceChanged, StaleRevision +from agent_comms.order_tools import TranscriptPageRead from agent_comms.session_file import RegistrySessionLocation from agent_comms.transcripts import RecordedTranscriptReadIdentity from agent_comms.threads import Thread +from agent_comms.ui_model.observation import NotPublished from support.roots import installed_wire from support.journals import assistant_entry, write_journal from toad.acp.comms_updates import OwnerSnapshotConsumer @@ -62,3 +64,19 @@ def test_recorded_presentation_does_not_acquire_live_owner(recorded): consumer.thread_presentation(replace(presentation, read_identity=identity)) assert TranscriptSnapshotUpdate.capture(live.transcripts, "reader", historical_source=source.key).identity == identity + + +def test_known_transcript_requests_carry_the_page_operation_and_keep_root_at_the_envelope(recorded): + live, source, _native = recorded + scope = KnownTranscript(str(live.root), "reader") + cursor = live.transcripts.capture_page_read("reader", historical_source=source.key).read().after + snapshot = scope.snapshot_request(before=cursor, historical_source=source.key) + assert snapshot.root == str(live.root) + assert snapshot.operation == TranscriptPageRead( + thread="reader", before=cursor, historical_source=source.key) + previous = NotPublished() + update = scope.update_request(previous, after=cursor) + assert update.root == str(live.root) and update.previous is previous + assert update.operation == TranscriptPageRead(thread="reader", after=cursor) + # The shared operation never carries the UI's root, request or publication witness. + assert not {"root", "request_id", "previous", "identity"} & {f.name for f in fields(TranscriptPageRead)} From ef21cbf251516e2ad486e38f3531fc50f4245786 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 21:31:38 -0400 Subject: [PATCH 9/9] mypy reads Toad's declared source roots, src and tools Config only, authorized by root (bus 36182) for the P883 matched cut. The checker guessed the root from src/dot; tools/acceptance.py:34-36 already declares tools as a script root for its sibling imports (performance.*), so mypy_path names both roots instead of guessing one. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RnH6dS2Kmg1WxNnAc78YHA --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 564a0629e..f709fc75e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -84,3 +84,4 @@ markers = ["refactor_guard: permanent architecture deletion guards"] [tool.mypy] follow_untyped_imports = true +mypy_path = ["src", "tools"]