From 44abe7ce378c208bca6cdb194b5a22670b54e888 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 22:49:16 -0400 Subject: [PATCH 1/8] Share source-owned queue inspection across native and bus composers --- src/toad/agent.py | 3 - src/toad/agent_binding.py | 3 - src/toad/agent_presentation.py | 40 ++++- src/toad/conversation_kind.py | 10 +- src/toad/conversation_submission.py | 50 ++++-- src/toad/delivery_failure_view.py | 4 + src/toad/notification_status.py | 25 +++ src/toad/screens/goal_edit.py | 4 +- src/toad/session_presentation.py | 7 +- src/toad/submission_queue.py | 205 ++++++++++++++++++++++ src/toad/transcript_preparation.py | 6 + src/toad/widgets/channel_prompt.py | 10 ++ src/toad/widgets/comms_chat.py | 6 +- src/toad/widgets/conversation.py | 29 ++- src/toad/widgets/message_notifications.py | 4 + src/toad/widgets/prompt.py | 33 +--- src/toad/widgets/transcript_fragments.py | 8 + src/toad/wire_history.py | 7 +- tests/textual/test_common_queue.py | 73 ++++++++ 19 files changed, 452 insertions(+), 75 deletions(-) create mode 100644 src/toad/submission_queue.py create mode 100644 tests/textual/test_common_queue.py diff --git a/src/toad/agent.py b/src/toad/agent.py index adca76eb7..5cb6de45f 100644 --- a/src/toad/agent.py +++ b/src/toad/agent.py @@ -78,9 +78,6 @@ def accepts_failed_input(self, message): else True ) - def send_now(self, owner): - return owner.send_bound_now(self) - def invalidate_observation(self, observation: SessionObservation) -> None: observation.schedule(self) diff --git a/src/toad/agent_binding.py b/src/toad/agent_binding.py index 1b8989060..e2bde404e 100644 --- a/src/toad/agent_binding.py +++ b/src/toad/agent_binding.py @@ -167,9 +167,6 @@ def begin_submission(self, owner: ConversationSubmissions, submission: AgentInpu def accepts_failed_input(self, message: CoreEventMessage) -> bool: return False - def send_now(self, owner: ConversationSubmissions) -> None: - """No remote queue is bound.""" - async def get_goal_history(self, goal_id: str) -> RecordedGoalHistory: return RecordedGoalHistory.empty() diff --git a/src/toad/agent_presentation.py b/src/toad/agent_presentation.py index 55d878b82..4fa243948 100644 --- a/src/toad/agent_presentation.py +++ b/src/toad/agent_presentation.py @@ -3,7 +3,7 @@ from dataclasses import dataclass from pathlib import Path from agent_comms.declared_family import DeclaredFamily -from agent_comms.acp_extension import PendingQueueProjection, QueueProjection +from agent_comms.acp_extension import NoQueueScope, PendingQueueProjection, QueueProjection from toad.private_native_cursor import CursorStatus, HiddenCursorStatus from toad.conversation_turn import TurnBinding, LocalTurnBinding, ManagedTurnBinding @@ -38,6 +38,21 @@ def queue_supported(self) -> bool: def queue(self): """The actual source owns its input queue projection.""" + @property + def queue_scope(self): + return NoQueueScope() + + @property + def queue_identity(self) -> str: + return "No live ACP queue attachment" + + @property + def queue_actions_available(self) -> bool: + return False + + def require_queue_action(self, scope): + raise ValueError("No live ACP queue authority is available") + @property def prompt_in_flight(self): return 0 @@ -87,6 +102,29 @@ def queue_supported(self) -> bool: """Advertisement belongs to the original coordination declaration.""" return self.agent.coordination is not None + @property + def queue_scope(self): + return self.agent.queue_attachment.scope + + @property + def queue_identity(self) -> str: + attachment = self.agent.queue_attachment + return attachment.state.project( + attachment.current, + available=lambda current: f"Scope {current.scope!r} · revision {current.revision}", + pending="Checking queue attachment", + unavailable="Queue attachment unavailable", + ) + + @property + def queue_actions_available(self) -> bool: + return self.queue_supported and self.agent.queue_attachment.accepts_request(self.queue_scope) + + def require_queue_action(self, scope): + if not self.queue_supported or not self.agent.queue_attachment.accepts_request(scope): + raise ValueError("Queue attachment changed or is unavailable; inspect the current queue again") + return self.agent + @property def uses_managed_turns(self): return self.turns.managed diff --git a/src/toad/conversation_kind.py b/src/toad/conversation_kind.py index a6e386cde..34fce26d5 100644 --- a/src/toad/conversation_kind.py +++ b/src/toad/conversation_kind.py @@ -133,7 +133,7 @@ def active_members(self, model: SidebarModel): def compose_activity(cls, view: CommsChatView) -> ComposeResult: ... @abstractmethod - def prompt(self, project: Checkout): ... + def prompt(self, project: Checkout, *, submissions): ... @classmethod def seen_page(cls, history: WireHistory, seen): @@ -206,10 +206,10 @@ async def status_text(self, comms: Comms, *, read_only: bool) -> str: def initial_placeholder(self) -> str: return f"Message {self.target.name}" - def prompt(self, project: Checkout): + def prompt(self, project: Checkout, *, submissions): from toad.widgets.channel_prompt import ChannelPrompt - return ChannelPrompt(project=project, placeholder=self.initial_placeholder) + return ChannelPrompt(project=project, placeholder=self.initial_placeholder, submissions=submissions) def read_only(self, comms: Comms) -> bool: return comms.bus.require_channel(self.target.name).read_only @@ -293,10 +293,10 @@ def compose_activity(cls, view) -> ComposeResult: view.thread_activity = ConversationSessionDetails(view.conversation.peer) yield view.thread_activity - def prompt(self, project: Checkout): + def prompt(self, project: Checkout, *, submissions): from toad.widgets.prompt import MessagePrompt - return MessagePrompt(project=project) + return MessagePrompt(project=project, submissions=submissions) async def status_text(self, comms: Comms, *, read_only: bool) -> str: def read(): diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py index 8fa854b58..9d2dd56f1 100644 --- a/src/toad/conversation_submission.py +++ b/src/toad/conversation_submission.py @@ -24,6 +24,7 @@ from toad.core.input_events import UserInputSubmitted from toad.core_event_carrier import CoreEventMessage from toad.widgets.prompt import Prompt +from agent_comms.notification_evidence import NotificationWindow if TYPE_CHECKING: from toad.acp.agent import Agent @@ -292,6 +293,33 @@ async def submit(self, event: UserInputSubmitted): def accepts_failure(self, message: CoreEventMessage): return self.view.agent_binding.accepts_failed_input(message) + @property + def queue_source(self): + return self.view.queue_source + + @property + def queue_summary(self) -> str: + return self.queue_source.queue_summary(self) + + @property + def message_queue_entries(self): + """Borrow the existing visible fragment proofs; never retain another inbox.""" + entries = {} + for target in self.view.contents.notification_targets(): + for fragment in target.visible_fragments(): + for entry in fragment.queue_entries(target.notification_view): + entries[entry.request] = entry + if len(entries) >= NotificationWindow.window_limit: + return tuple(entries.values()) + return tuple(entries.values()) + + def inspect_queue(self): + from toad.submission_queue import QueueInspection + + inspection = QueueInspection(self) + self.view.app.push_screen(inspection) + return inspection + @property def queue_projection(self): return self.view.agent_binding.presentation.queue @@ -335,20 +363,18 @@ def reset(self): self.publish_pending() def send_now(self): - return self.view.agent_binding.send_now(self) - - def send_bound_now(self, agent: Agent): - session, scope = agent.session_id, agent.queue_attachment.scope + source = self.queue_source return self.view.run_worker( - partial(self._send_now, agent, session, scope), group="send-queued-now", exclusive=True + partial(self._send_now, source, self.view.agent_binding, source.queue_scope(self)), + group="send-queued-now", exclusive=True, ) - async def _send_now(self, agent: Agent, session, scope: QueueScope): + async def _send_now(self, source, agent, scope: QueueScope): try: - if await agent.controller.send_now(): - return - except RequestError as error: - if self.view.agent_binding is agent and agent.session_id == session: + if self.queue_source is not source or self.view.agent_binding is not agent: + raise ValueError("The source changed before send now") + await source.queue_send_now(self, scope) + except (RequestError, ValueError) as error: + if self.queue_source is source and self.view.agent_binding is agent: self.view.feedback.flash(f"Send now failed: {error}", style="error") - if self.view.agent_binding is agent and agent.queue_attachment.accepts_request(scope): - self.publish_pending() + self.publish_pending() diff --git a/src/toad/delivery_failure_view.py b/src/toad/delivery_failure_view.py index 04c2a3cec..cb6a6533c 100644 --- a/src/toad/delivery_failure_view.py +++ b/src/toad/delivery_failure_view.py @@ -91,9 +91,13 @@ def send_block_reason(self) -> str: """Why this view's own send holds compose; empty when it does not.""" return self._send_state.reason + def queue_evidence_changed(self) -> None: + """An unbound delivery surface has no queue consumer to repaint.""" + def _enter_send_state(self, state: SendState) -> None: """The one place a send changes state; compose and the status line follow it.""" self._send_state = state + self.queue_evidence_changed() if self.is_attached: async def refresh() -> None: read_only = await self.composer_read_only() diff --git a/src/toad/notification_status.py b/src/toad/notification_status.py index c4e687879..17fccce8d 100644 --- a/src/toad/notification_status.py +++ b/src/toad/notification_status.py @@ -38,10 +38,18 @@ class NotificationStatus: def title(self) -> str: return "" + @property + def queue_title(self) -> str: + return self.title + @property def details(self) -> DetailLines: return () + @property + def queue_details(self) -> DetailLines: + return self.details + @property def disclosure(self) -> SpanColumns: return SpanColumns(self.details, indent=2) @@ -81,6 +89,23 @@ def working(self) -> bool: def title(self) -> str: return MessageNotification.summary(self.notifications) + @property + def queue_title(self) -> str: + # Keep each recipient's actual proof label; do not turn routing into arrival. + return " · ".join(f"@{item.recipient} {item.label}" for item in self.notifications[:1]) or "Unobserved" + + @property + def queue_details(self) -> DetailLines: + leading = CheckedNotifications(self.notifications[:1]) + return ( + (Span(f"{len(self.notifications)} observed recipients; leading result shown", MutedRole),), + *((Span(f"Recipient {item.recipient_identity.recipient_lookup}", MutedRole),) + for item in leading.notifications), + *leading.details, + *((Span(f"Answer {item.answer.seq}/{item.answer.message_id}", MutedRole),) + for item in leading.notifications if item.answer is not None), + ) + @property def details(self) -> DetailLines: lines = [] diff --git a/src/toad/screens/goal_edit.py b/src/toad/screens/goal_edit.py index 89a9bb795..79f72a6b0 100644 --- a/src/toad/screens/goal_edit.py +++ b/src/toad/screens/goal_edit.py @@ -16,7 +16,7 @@ from toad.core.input_events import UserInputSubmitted from toad.core_event_carrier import CoreEventReceiver, CoreEventMessage from metaclass_registry.mro_dispatch import handles -from toad.widgets.channel_prompt import ChannelPrompt +from toad.widgets.channel_prompt import DraftPrompt class GoalEdit(CoreEventReceiver, DismissibleScreen[str | None]): @@ -43,7 +43,7 @@ def __init__( self.goal = goal self.on_save = on_save self.saving = False - self.editor = ChannelPrompt(project=project) + self.editor = DraftPrompt(project=project) self.editor.set_mention_candidates(candidates) def compose(self) -> ComposeResult: diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py index e13b2259c..edd469e04 100644 --- a/src/toad/session_presentation.py +++ b/src/toad/session_presentation.py @@ -23,6 +23,7 @@ from toad.widgets.conversation import Conversation from toad.widgets.history_anchor import ParkedPosition from toad.widgets.prompt import MailboxPrompt +from toad.submission_queue import BusQueueSource, NativeQueueSource if TYPE_CHECKING: from toad.app import ToadApp @@ -104,7 +105,7 @@ async def release_binding( async def attach_binding(self, conversation: Conversation) -> None: ... -class OperationalSessionSources: +class OperationalSessionSources(NativeQueueSource): """Own existing operational resources while the view is detached.""" def __init__(self) -> None: @@ -235,7 +236,7 @@ async def close(self) -> None: release.push_async_callback(agent.stop) -class MailboxSessionSources(OperationalSessionSources): +class MailboxSessionSources(BusQueueSource, OperationalSessionSources): """CLI execution receives user messages through Core's inbox, not ACP.""" def activity_row(self, conversation: Conversation): @@ -245,7 +246,7 @@ def agent_info(self, conversation: Conversation) -> str: return conversation.thread_activity.row.presentation.title def prompt(self, conversation: Conversation): - return MailboxPrompt(project=conversation.project_path) + return MailboxPrompt(project=conversation.project_path, submissions=conversation.submissions) def thread_presented(self, conversation: Conversation) -> None: conversation.run_worker( diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py new file mode 100644 index 000000000..4757fd408 --- /dev/null +++ b/src/toad/submission_queue.py @@ -0,0 +1,205 @@ +"""Source-owned queue capabilities; the prompt is their one presentation consumer. + +These are read-only views of existing attachments, transport and notification +proofs. They never reserve input, keep membership, or turn BUS references into +ACP request identities. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from acp.exceptions import RequestError +from agent_comms.acp_extension import NoQueueScope, QueueItem +from agent_comms.message_reference import MessageReference +from agent_comms.ui_model.observation import ReadNotifications +from textual import on +from textual.containers import HorizontalGroup, Vertical, VerticalScroll +from textual.message import Message +from textual.widgets import Button, Static + +from toad.line_blocks import MutedRole, Span, SpanColumns +from toad.notification_status import DetailLines, NotificationStatus +from toad.screens.modal import DismissibleScreen +from toad.widgets.tool_content import ToolContent + +if TYPE_CHECKING: + from agent_comms.acp_extension import QueueScope + from toad.conversation_submission import ConversationSubmissions + + +class QueueEvidenceChanged(Message): + """Local repaint after existing notification/viewport evidence changes.""" + + +@dataclass(frozen=True) +class MessageQueueEntry: + request: ReadNotifications + status: NotificationStatus + + @property + def reference(self) -> MessageReference: + return self.request.references[0] + + @property + def summary(self) -> str: + return f"{self.reference.seq}: {self.status.queue_title}" + + @property + def details(self) -> DetailLines: + return ( + (Span(f"{self.request.root} · {self.reference.seq}/{self.reference.message_id}", MutedRole),), + *self.status.queue_details, + ) + + +class QueueSource: + """An absent source grants neither queue membership nor mutation authority.""" + + preview_limit = 3 + + def queue_summary(self, owner: ConversationSubmissions) -> str: + return "Queue unavailable" + + def queue_details(self, owner: ConversationSubmissions) -> DetailLines: + return ((Span("No queue observation is available for this source.", MutedRole),),) + + +class ReadOnlyQueueSource(QueueSource): + """A source may expose delivery evidence without granting native actions.""" + + def queue_scope(self, owner: ConversationSubmissions) -> QueueScope: + return NoQueueScope() + + def queue_actions_available(self, owner: ConversationSubmissions) -> bool: + return False + + async def queue_send_now(self, owner: ConversationSubmissions, scope: QueueScope) -> None: + raise ValueError("Send now unavailable: this source has no native queue authority") + + async def queue_clear(self, owner: ConversationSubmissions, scope: QueueScope) -> None: + raise ValueError("Clear unavailable: BUS messages cannot be withdrawn through an ACP queue") + + +class NativeQueueSource(QueueSource): + """Native queue presentation and actions borrow the actual ACP attachment.""" + + def queue_summary(self, owner: ConversationSubmissions) -> str: + projection = owner.queue_projection + feedback = projection.feedback(owner.view.agent_binding.presentation.queue_supported) + parts = [feedback] if feedback else [] + for label, texts, detail in ( + ("Submitting", tuple(QueueItem.summarize(text) for text in owner.delivering), ""), + ("Queued", tuple(row.preview for row in projection.items), ""), + ("Restored", tuple(row.preview for row in projection.restored), ", read-only"), + ): + if texts: + parts.append(f"{label} ({len(texts)}{detail}): " + " · ".join(texts[:self.preview_limit])) + return " | ".join(parts) + + def queue_details(self, owner: ConversationSubmissions) -> DetailLines: + agent = owner.view.agent_binding + projection = owner.queue_projection + lines = [(Span("Submitting is local intent; Queued is producer acceptance, not consumption.", MutedRole),)] + lines.extend((Span(f"Submitting: {QueueItem.summarize(text)}", MutedRole),) + for text in owner.delivering) + lines.extend((Span(f"Queued {row.input_id}: {row.preview}", MutedRole),) + for row in projection.items) + lines.extend((Span(f"Restored {row.input_id} (read-only): {row.preview}", MutedRole),) + for row in projection.restored) + lines.append((Span(agent.presentation.queue_identity, MutedRole),)) + return tuple(lines) + + def queue_scope(self, owner: ConversationSubmissions) -> QueueScope: + return owner.view.agent_binding.presentation.queue_scope + + def queue_actions_available(self, owner: ConversationSubmissions) -> bool: + return owner.view.agent_binding.presentation.queue_actions_available + + def require_queue_action(self, owner: ConversationSubmissions, scope: QueueScope): + return owner.view.agent_binding.presentation.require_queue_action(scope) + + async def queue_send_now(self, owner: ConversationSubmissions, scope: QueueScope) -> None: + agent = self.require_queue_action(owner, scope) + if not await agent.controller.send_now(): + raise ValueError("The native owner did not accept send now") + + async def queue_clear(self, owner: ConversationSubmissions, scope: QueueScope) -> None: + await self.require_queue_action(owner, scope).controller.clear_queue() + + +class BusQueueSource(ReadOnlyQueueSource): + """BUS views borrow original-reference/per-recipient evidence, not ACP rows.""" + + def queue_summary(self, owner: ConversationSubmissions) -> str: + entries = owner.message_queue_entries + transport = owner.view.send_block_reason + evidence = " · ".join(entry.summary for entry in entries[:self.preview_limit]) + coverage = f"Delivery ({len(entries)} visible messages)" + return " | ".join(part for part in (transport, coverage + (f": {evidence}" if evidence else ": unobserved")) if part) + + def queue_details(self, owner: ConversationSubmissions) -> DetailLines: + return ( + (Span("Visible original message references only; this is not an ACP queue or a complete inbox count.", MutedRole),), + (Span(owner.view.send_block_reason, MutedRole),), + (Span(f"Showing up to {self.preview_limit} visible messages and their leading observed recipient; not complete recipient coverage.", MutedRole),), + *(line for entry in owner.message_queue_entries[:self.preview_limit] for line in entry.details), + (Span("Clear/send-now unavailable: this BUS source has no native queue authority.", MutedRole),), + ) + + +class QueueInspection(DismissibleScreen[None]): + """One inspect/action surface for native, channel and bridge composers.""" + + DEFAULT_CSS = """ + QueueInspection { align: center middle; background: $background 40%; } + QueueInspection > Vertical { width: 85%; max-width: 110; height: 75%; padding: 1 2; background: $surface; } + QueueInspection VerticalScroll { height: 1fr; } + QueueInspection HorizontalGroup { height: auto; } + QueueInspection #queue-error { height: auto; color: $error; } + """ + + def __init__(self, owner: ConversationSubmissions): + super().__init__() + self.submissions = owner + self.source = owner.queue_source + self.agent = owner.view.agent_binding + self.scope = self.source.queue_scope(owner) + + def compose(self): + with Vertical(): + yield Static("Queue and delivery") + with VerticalScroll(): + details = ToolContent() + details.update((SpanColumns(self.source.queue_details(self.submissions)),)) + yield details + yield Static("Individual edit/delete/reorder requires ID-targeted, revision-checked backend support.") + yield Static("", id="queue-error") + with HorizontalGroup(): + available = self.source.queue_actions_available(self.submissions) + yield Button("Send now", id="queue-send-now", disabled=not available) + yield Button("Clear queued inputs", id="queue-clear", disabled=not available) + yield Button("Close", id="queue-close") + + async def apply_action(self, action): + try: + if self.submissions.queue_source is not self.source or self.submissions.view.agent_binding is not self.agent: + raise ValueError("The source changed; inspect its current queue again") + await action(self.submissions, self.scope) + except (ValueError, RequestError) as error: + self.query_one("#queue-error", Static).update(str(error)) + return + self.submissions.publish_pending() + self.dismiss() + + @on(Button.Pressed, "#queue-send-now") + async def send_now(self): + await self.apply_action(self.source.queue_send_now) + + @on(Button.Pressed, "#queue-clear") + async def clear_queue(self): + await self.apply_action(self.source.queue_clear) + + @on(Button.Pressed, "#queue-close") + def close_queue(self): + self.dismiss() diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py index 6738dda57..b643021e7 100644 --- a/src/toad/transcript_preparation.py +++ b/src/toad/transcript_preparation.py @@ -290,6 +290,12 @@ def close(self) -> None: class PreparedPageSource(ABC): + @property + def queue_source(self): + from toad.submission_queue import ReadOnlyQueueSource + + return ReadOnlyQueueSource() + """One lifetime-owned source of prepared pages, independent of widget paging.""" runtime: PreparationRuntime diff --git a/src/toad/widgets/channel_prompt.py b/src/toad/widgets/channel_prompt.py index a17b86465..b76262f18 100644 --- a/src/toad/widgets/channel_prompt.py +++ b/src/toad/widgets/channel_prompt.py @@ -118,3 +118,13 @@ def dismiss_control(self) -> None: self.mention_list.display = False else: super().dismiss_control() + + +class DraftPrompt(ChannelPrompt): + """Mention-enabled editing without a conversation or delivery authority.""" + + def compose_delivery_controls(self) -> ComposeResult: + return () + + def _update_queue_summary(self) -> None: + pass diff --git a/src/toad/widgets/comms_chat.py b/src/toad/widgets/comms_chat.py index 1caaeca1c..76c454fab 100644 --- a/src/toad/widgets/comms_chat.py +++ b/src/toad/widgets/comms_chat.py @@ -81,6 +81,10 @@ def __init__( self.message_contents = Contents(id="contents") self.message_history = WireHistory.empty(view=self, notification_view=self.message_contents) + @property + def queue_source(self): + return self.message_history.page_source.queue_source + def transcript_binding(self) -> ViewBinding: """A wire conversation has no native transcript to publish.""" return NoView() @@ -96,7 +100,7 @@ def compose(self) -> ComposeResult: with containers.Vertical(id="prompt-stack"): yield from self.conversation.compose_activity(self) yield self.throbber_above - yield self.conversation.prompt(self.project_path).data_bind( + yield self.conversation.prompt(self.project_path, submissions=self.submissions).data_bind( project_path=Conversation.project_path, working_directory=Conversation.working_directory, agent_info=Conversation.agent_info, diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py index 47c98594d..6d8f3e61e 100644 --- a/src/toad/widgets/conversation.py +++ b/src/toad/widgets/conversation.py @@ -42,6 +42,7 @@ from toad.acp.status import StopReason from toad.agent import AgentBase from toad.agent_binding import AgentConnection, NoAgentConnection +from toad.submission_queue import QueueEvidenceChanged from toad.agent_schema import AgentDefinition from toad.answer import Answer from toad.conversation_actions import ConversationAction, WindowAction @@ -1622,22 +1623,20 @@ async def compact_context(self, instructions: str | None) -> None: await self.agent_binding.compact_context(instructions) self.feedback.flash("Context compacted", style="success") - def open_queue_menu(self) -> None: - """Remote edits require exact input IDs and backend revision-CAS support.""" - if self.submissions.queue_projection.items: - self._queue_edit_unavailable() - - def _queue_edit_unavailable(self) -> None: - self.feedback.flash( - "Remote queue editing unavailable: requires ID-targeted, revision-checked backend support", - style="error", - ) + @property + def queue_source(self): + return self.operational_sources + + def queue_evidence_changed(self) -> None: + self.submissions.publish_pending() - def _on_queue_choice(self, action: str | None) -> None: - # A stale menu callback must not edit the draft or mutate the remote - # queue. Text/index matching cannot identify a sent input safely. - if action: - self._queue_edit_unavailable() + @on(QueueEvidenceChanged) + def queue_observation_changed(self, event: QueueEvidenceChanged) -> None: + event.stop() + self.queue_evidence_changed() + + def open_queue_menu(self) -> None: + self.submissions.inspect_queue() @handles(PreferenceChanged) async def _settings_changed(self, event: PreferenceChanged) -> None: diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py index d5a2fe5d5..a4538b942 100644 --- a/src/toad/widgets/message_notifications.py +++ b/src/toad/widgets/message_notifications.py @@ -12,6 +12,7 @@ from textual.widgets import Collapsible from agent_comms.ui_model.observation import ReadNotifications +from toad.submission_queue import QueueEvidenceChanged from toad.notification_status import ( CheckedNotifications, NotificationStatus, UnavailableNotifications, UncheckedNotifications, @@ -119,6 +120,7 @@ def check_notifications(self) -> None: references = tuple(dict.fromkeys( reference for fragment in fragments for reference in fragment.notification_references)) self.subscribe_notifications(references) + self.post_message(QueueEvidenceChanged()) def notification_read(self, references) -> ReadNotifications: from toad.project_path_owner import ProjectPathOwner @@ -144,6 +146,7 @@ def notifications_observed(self, results) -> None: async def _show_notifications(self, results): for target in self.notification_targets(): await target.show_notifications(results) + self.post_message(QueueEvidenceChanged()) def notifications_failed(self, failure: Exception) -> None: if self.can_draw_on_screen: @@ -153,6 +156,7 @@ def notifications_failed(self, failure: Exception) -> None: async def _show_notification_error(self, status): for target in self.notification_targets(): await target.show_notification_status(status) + self.post_message(QueueEvidenceChanged()) def on_mount(self) -> None: self.call_after_refresh(self.check_notifications) diff --git a/src/toad/widgets/prompt.py b/src/toad/widgets/prompt.py index 8a9085b77..4d9423e19 100644 --- a/src/toad/widgets/prompt.py +++ b/src/toad/widgets/prompt.py @@ -7,7 +7,6 @@ from agent_comms.repo import Checkout from typing import TYPE_CHECKING, Literal, Self -from agent_comms.acp_extension import QueueItem from rich.text import Text as RichText from textual import containers, events, getters, on, work @@ -666,29 +665,11 @@ def on_mount(self) -> None: self.call_after_refresh(self.sync_queue) def _update_queue_summary(self) -> None: - submissions = self.operational_submissions - projection = submissions.queue_projection - queued = [row.preview for row in submissions.queued_inputs] - restored = [row.preview for row in projection.restored] - feedback = projection.feedback(self.agent_binding.presentation.queue_supported) - delivering = [QueueItem.summarize(text) for text in submissions.delivering] - self.set_class(bool(queued or restored or delivering or feedback), "-has-queue") - parts: list[str] = [] - if feedback: - parts.append(feedback) - for label, texts, detail in ( - ("Submitting", delivering, ""), - ("Queued", queued, ""), - ("Restored", restored, ", read-only"), - ): - if texts: - parts.append( - f"{label} ({len(texts)}{detail}): " - + " · ".join(texts[:3]) - ) + text = self.operational_submissions.queue_summary + self.set_class(bool(text), "-has-queue") summary = self.query_one(".queue-summary", Label) - summary.tooltip = "Submitting is an outstanding local request, not acceptance. Queued is the producer's accepted input, not consumption or completion. Restored rows are not local drafts." - summary.update(Content(" | ".join(parts))) + summary.tooltip = "Inspect this source's actual queue or visible message delivery evidence. Acceptance is not consumption; unsupported actions remain unavailable." + summary.update(Content(text)) @on(messages.SendPromptNow) async def on_send_prompt_now(self, event): @@ -914,9 +895,6 @@ def sync_turn(self) -> None: def sync_configuration(self) -> None: pass - def _update_queue_summary(self) -> None: - pass - def watch_agent_info(self, agent_info: Content) -> None: if self.is_mounted: self.update_prompt() @@ -934,9 +912,6 @@ def update_prompt(self) -> None: def label_glyph(self) -> Content: return PromptLabel.AI - def compose_delivery_controls(self) -> ComposeResult: - return () - def compose_info_popups(self) -> ComposeResult: return () diff --git a/src/toad/widgets/transcript_fragments.py b/src/toad/widgets/transcript_fragments.py index 77489a592..464929ac0 100644 --- a/src/toad/widgets/transcript_fragments.py +++ b/src/toad/widgets/transcript_fragments.py @@ -46,6 +46,7 @@ from toad.menus import MenuItem from toad.message_rows import MarkdownMessageStyle, MessageRow, WireMessageStyle from toad.notification_status import NotificationStatus, UncheckedNotifications +from toad.submission_queue import MessageQueueEntry from toad.widgets.agent_activity import AgentActivityPresentation from toad.widgets.transcript_lines import AcquiredRows, NoRows, RowAcquisition @@ -340,6 +341,9 @@ def shown_live_by(self, claim: CommitClaim) -> bool: def notification_references(self) -> tuple[MessageReference, ...]: return () + def queue_entries(self, notification_view): + return () + def draws(self, message: BusMessage) -> bool: """Whether this fragment draws that bus message; only bus-message fragments do.""" return False @@ -516,6 +520,10 @@ def has_details(self) -> bool: def notification_references(self) -> tuple[MessageReference, ...]: return self.part_position.notification_references(self.events[0]) + def queue_entries(self, notification_view): + return tuple(MessageQueueEntry(notification_view.notification_read((reference,)), self.shown.status) + for reference in self.notification_references) + def draws(self, message: BusMessage) -> bool: return self.events[0].source == message.reference diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index 93e9d610c..e6f7afacd 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -50,6 +50,7 @@ ) from toad.widgets.history_anchor import LatestPosition from toad.widgets.message_notifications import PaintedRead, ReadWork +from toad.submission_queue import BusQueueSource from toad.widgets.transcript_fragments import ( BusMessageFragment, MessageShown, @@ -282,7 +283,7 @@ def leading(self): @dataclass(frozen=True, eq=False) -class WireSource: +class WireSource(BusQueueSource): """One conversation as one viewer reads it from Core, and how its messages are styled. A page read to the newest messages is short so that it is drawn at once; @@ -409,6 +410,10 @@ def __init__(self, *args, revision: ConversationObservation, **kwargs): super().__init__(*args, **kwargs) self.revision = revision + @property + def queue_source(self): + return self.loader + def replacement(self, through) -> WirePageBuffer: return type(self)(self.loader, through, self.runtime, revision=self.revision) diff --git a/tests/textual/test_common_queue.py b/tests/textual/test_common_queue.py new file mode 100644 index 000000000..089285510 --- /dev/null +++ b/tests/textual/test_common_queue.py @@ -0,0 +1,73 @@ +"""Common queue chrome uses real source references and never grants BUS ACP actions. + +These controls use the private wire, real observer, prompt and submission owner. +No product monkeypatches and no provider input are needed. +""" +import pytest +from agent_comms.acp_extension import NoQueueScope +from toad.session_presentation import MailboxSessionSources +from toad.widgets.comms_chat import CommsChatView +from toad.widgets.prompt import QueueSummary +from textual.widgets import Button + + +async def open_channel(ui, viewer): + return await ui.open_channel("#team", actor=viewer) + + +async def open_direct(ui, viewer): + return await ui.open_direct("peer", actor=viewer) + + +@pytest.mark.parametrize("open_surface", [open_channel, open_direct]) +async def test_bus_queue_borrows_original_proofs_and_preserves_draft(workspace, open_surface): + workspace.thread("peer", "team") + workspace.thread("recipient", "team") + originals = ( + workspace.comms.messaging.send("peer", "#team", "CHANNEL ORIGINAL").message, + workspace.comms.messaging.send("peer", workspace.viewer, "DIRECT ORIGINAL").message, + ) + async with workspace.app(size=(115, 38)) as ui: + await open_surface(ui, workspace.viewer) + view = ui.app.selected_session.query_one(CommsChatView) + owner = view.submissions + await ui.until(lambda: bool(owner.message_queue_entries), why="visible original-reference evidence") + await ui.write_draft("@peer kept draft") + entries = owner.message_queue_entries + assert all(entry.reference in {message.reference for message in originals} for entry in entries) + assert all(entry.request.root == str(workspace.comms.root) for entry in entries) + assert view.query_one(QueueSummary) in ui.app.screen._compositor.visible_widgets + assert not owner.queue_source.queue_actions_available(owner) + with pytest.raises(ValueError): + await owner.queue_source.queue_clear(owner, NoQueueScope()) + with pytest.raises(ValueError): + await owner.queue_source.queue_send_now(owner, NoQueueScope()) + + previous_screen = ui.app.screen + screen = owner.inspect_queue() + await ui.until(lambda: ui.app.screen is screen, why="shared queue inspection") + assert screen.query_one("#queue-send-now", Button).disabled + assert screen.query_one("#queue-clear", Button).disabled + await ui.textual[1].click("#queue-close") + await ui.until(lambda: ui.app.screen is previous_screen, why="queue inspection closed") + assert ui.draft() == "@peer kept draft" + + workspace.comms.messaging.send("peer", "#team", "LATER CHANNEL ORIGINAL") + workspace.comms.messaging.send("peer", workspace.viewer, "LATER DIRECT ORIGINAL") + previous = {entry.reference for entry in entries} + await ui.until(lambda: bool({entry.reference for entry in owner.message_queue_entries} - previous), + why="source subscription to refresh the existing queue consumer") + + +async def test_mailbox_factory_supplies_same_owner_and_readonly_capability_wins_mro(workspace): + async with workspace.app(size=(110, 35)) as ui: + view = ui.conversation + source = MailboxSessionSources() + prompt = source.prompt(view) + assert prompt.operational_submissions is view.submissions + assert not source.queue_actions_available(view.submissions) + with pytest.raises(ValueError): + await source.queue_send_now(view.submissions, source.queue_scope(view.submissions)) + with pytest.raises(ValueError): + await source.queue_clear(view.submissions, source.queue_scope(view.submissions)) + await source.close() From 9494d869150412c137bc3c666bc812d96524299d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 28 Oct 2026 23:36:02 -0400 Subject: [PATCH 2/8] Declare queue capabilities and remove duplicated title facts --- src/toad/agent_presentation.py | 17 +++++++++---- src/toad/conversation_submission.py | 10 +++++--- src/toad/submission_queue.py | 25 +++++++++++++++---- src/toad/widgets/conversation.py | 4 +-- src/toad/wire_history.py | 2 +- .../test_message_notification_title.py | 2 -- tests/textual/test_message_notifications.py | 10 ++++---- 7 files changed, 46 insertions(+), 24 deletions(-) diff --git a/src/toad/agent_presentation.py b/src/toad/agent_presentation.py index 4fa243948..e017da152 100644 --- a/src/toad/agent_presentation.py +++ b/src/toad/agent_presentation.py @@ -1,13 +1,20 @@ """Agent-owned optional capabilities and derived attachment presentation.""" +from __future__ import annotations + +from typing import TYPE_CHECKING from abc import abstractmethod from dataclasses import dataclass from pathlib import Path from agent_comms.declared_family import DeclaredFamily -from agent_comms.acp_extension import NoQueueScope, PendingQueueProjection, QueueProjection +from agent_comms.acp_extension import NoQueueScope, PendingQueueProjection, QueueProjection, QueueScope from toad.private_native_cursor import CursorStatus, HiddenCursorStatus from toad.conversation_turn import TurnBinding, LocalTurnBinding, ManagedTurnBinding +if TYPE_CHECKING: + from toad.acp.agent import Agent + + @dataclass(frozen=True) class AgentAttachmentView: cursor: type[CursorStatus] @@ -39,7 +46,7 @@ def queue(self): """The actual source owns its input queue projection.""" @property - def queue_scope(self): + def queue_scope(self) -> QueueScope: return NoQueueScope() @property @@ -50,7 +57,7 @@ def queue_identity(self) -> str: def queue_actions_available(self) -> bool: return False - def require_queue_action(self, scope): + def require_queue_action(self, scope: QueueScope) -> Agent: raise ValueError("No live ACP queue authority is available") @property @@ -103,7 +110,7 @@ def queue_supported(self) -> bool: return self.agent.coordination is not None @property - def queue_scope(self): + def queue_scope(self) -> QueueScope: return self.agent.queue_attachment.scope @property @@ -120,7 +127,7 @@ def queue_identity(self) -> str: def queue_actions_available(self) -> bool: return self.queue_supported and self.agent.queue_attachment.accepts_request(self.queue_scope) - def require_queue_action(self, scope): + def require_queue_action(self, scope: QueueScope) -> Agent: if not self.queue_supported or not self.agent.queue_attachment.accepts_request(scope): raise ValueError("Queue attachment changed or is unavailable; inspect the current queue again") return self.agent diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py index 9d2dd56f1..48322d7ad 100644 --- a/src/toad/conversation_submission.py +++ b/src/toad/conversation_submission.py @@ -28,6 +28,8 @@ if TYPE_CHECKING: from toad.acp.agent import Agent + from toad.submission_queue import QueueSource, MessageQueueEntry + from agent_comms.ui_model.observation import ReadNotifications from toad.widgets.conversation import Conversation @@ -294,7 +296,7 @@ def accepts_failure(self, message: CoreEventMessage): return self.view.agent_binding.accepts_failed_input(message) @property - def queue_source(self): + def queue_source(self) -> QueueSource: return self.view.queue_source @property @@ -302,9 +304,9 @@ def queue_summary(self) -> str: return self.queue_source.queue_summary(self) @property - def message_queue_entries(self): + def message_queue_entries(self) -> tuple[MessageQueueEntry, ...]: """Borrow the existing visible fragment proofs; never retain another inbox.""" - entries = {} + entries: dict[ReadNotifications, MessageQueueEntry] = {} for target in self.view.contents.notification_targets(): for fragment in target.visible_fragments(): for entry in fragment.queue_entries(target.notification_view): @@ -369,7 +371,7 @@ def send_now(self): group="send-queued-now", exclusive=True, ) - async def _send_now(self, source, agent, scope: QueueScope): + async def _send_now(self, source: QueueSource, agent, scope: QueueScope) -> None: try: if self.queue_source is not source or self.view.agent_binding is not agent: raise ValueError("The source changed before send now") diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py index 4757fd408..a280f5ea4 100644 --- a/src/toad/submission_queue.py +++ b/src/toad/submission_queue.py @@ -6,8 +6,10 @@ """ from __future__ import annotations +from abc import ABC, abstractmethod +from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, ClassVar from acp.exceptions import RequestError from agent_comms.acp_extension import NoQueueScope, QueueItem @@ -25,6 +27,7 @@ if TYPE_CHECKING: from agent_comms.acp_extension import QueueScope + from toad.acp.agent import Agent from toad.conversation_submission import ConversationSubmissions @@ -53,10 +56,22 @@ def details(self) -> DetailLines: ) -class QueueSource: +class QueueSource(ABC): """An absent source grants neither queue membership nor mutation authority.""" - preview_limit = 3 + preview_limit: ClassVar[int] = 3 + + @abstractmethod + def queue_scope(self, owner: ConversationSubmissions) -> QueueScope: ... + + @abstractmethod + def queue_actions_available(self, owner: ConversationSubmissions) -> bool: ... + + @abstractmethod + async def queue_send_now(self, owner: ConversationSubmissions, scope: QueueScope) -> None: ... + + @abstractmethod + async def queue_clear(self, owner: ConversationSubmissions, scope: QueueScope) -> None: ... def queue_summary(self, owner: ConversationSubmissions) -> str: return "Queue unavailable" @@ -116,7 +131,7 @@ def queue_scope(self, owner: ConversationSubmissions) -> QueueScope: def queue_actions_available(self, owner: ConversationSubmissions) -> bool: return owner.view.agent_binding.presentation.queue_actions_available - def require_queue_action(self, owner: ConversationSubmissions, scope: QueueScope): + def require_queue_action(self, owner: ConversationSubmissions, scope: QueueScope) -> Agent: return owner.view.agent_binding.presentation.require_queue_action(scope) async def queue_send_now(self, owner: ConversationSubmissions, scope: QueueScope) -> None: @@ -181,7 +196,7 @@ def compose(self): yield Button("Clear queued inputs", id="queue-clear", disabled=not available) yield Button("Close", id="queue-close") - async def apply_action(self, action): + async def apply_action(self, action: Callable[[ConversationSubmissions, QueueScope], Awaitable[None]]) -> None: try: if self.submissions.queue_source is not self.source or self.submissions.view.agent_binding is not self.agent: raise ValueError("The source changed; inspect its current queue again") diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py index 6d8f3e61e..e12c50085 100644 --- a/src/toad/widgets/conversation.py +++ b/src/toad/widgets/conversation.py @@ -42,7 +42,7 @@ from toad.acp.status import StopReason from toad.agent import AgentBase from toad.agent_binding import AgentConnection, NoAgentConnection -from toad.submission_queue import QueueEvidenceChanged +from toad.submission_queue import QueueEvidenceChanged, QueueSource from toad.agent_schema import AgentDefinition from toad.answer import Answer from toad.conversation_actions import ConversationAction, WindowAction @@ -1624,7 +1624,7 @@ async def compact_context(self, instructions: str | None) -> None: self.feedback.flash("Context compacted", style="success") @property - def queue_source(self): + def queue_source(self) -> QueueSource: return self.operational_sources def queue_evidence_changed(self) -> None: diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index e6f7afacd..a7bc45277 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -411,7 +411,7 @@ def __init__(self, *args, revision: ConversationObservation, **kwargs): self.revision = revision @property - def queue_source(self): + def queue_source(self) -> WireSource: return self.loader def replacement(self, through) -> WirePageBuffer: diff --git a/tests/textual/test_message_notification_title.py b/tests/textual/test_message_notification_title.py index 1bd02b4f3..0d917efe5 100644 --- a/tests/textual/test_message_notification_title.py +++ b/tests/textual/test_message_notification_title.py @@ -38,14 +38,12 @@ async def test_changed_details_and_errors_keep_the_title_layout_but_changed_summ await ui.until(lambda: "waiting-peer: Queued" in " ".join(ui.painted(feedback).split()), why="the changed detail") assert str(feedback.title) == MessageNotification.summary((responding, queued)) assert str(feedback.title) == MessageNotification.summary((responding, waiting)) - assert str(feedback.title) == "Answering 1 · Delivered 2/2 · Consumed 1/2 · Read 0/2" assert feedback.has_class("-working") completed = replace(responding, state=CompletedProgress(ShownToModelProgress(1))) feedback.show(CheckedNotifications((completed, queued))) await ui.step() assert str(feedback.title) == MessageNotification.summary((completed, queued)) assert str(feedback.title) != MessageNotification.summary((responding, queued)) - assert str(feedback.title) == queued.label + " · Delivered 2/2 · Consumed 1/2 · Read 0/2" assert not feedback.has_class("-working") feedback.show(UnavailableNotifications("First read error")) await ui.step() diff --git a/tests/textual/test_message_notifications.py b/tests/textual/test_message_notifications.py index 670fa0bc2..b7d7f5e1a 100644 --- a/tests/textual/test_message_notifications.py +++ b/tests/textual/test_message_notifications.py @@ -91,7 +91,7 @@ async def test_visible_references_are_bounded_and_typed_outcomes_paint_their_det assert 0 < len(request.references) <= len(on_screen) < 30 assert set(request.references) <= on_screen await notifications.publish(ui, chat) - await ui.until_shows("Delivered 45/45 · Consumed 0/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) assert newest_status(chat).working await ui.click("▶ Sent", occurrence=len(on_screen) - 1) await ui.until_shows("waiting-9: Sent") @@ -110,11 +110,11 @@ async def test_the_line_follows_typed_state_and_read_errors_without_new_messages await settled(ui, chat) notifications.state = ConsumingProgress(0) await notifications.publish(ui, chat) - await ui.until_shows("Answering 1 · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) assert newest_status(chat).working notifications.state = IgnoredProgress(ShownToModelProgress(0)) await notifications.publish(ui, chat) - await ui.until_shows("Seen, not answering 1 · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) assert not newest_status(chat).working access = ui.app.coordination_access request = access.notification_views[chat.message_history.notification_view] @@ -124,7 +124,7 @@ async def test_the_line_follows_typed_state_and_read_errors_without_new_messages assert "fixture busy database" in newest_status(chat).error notifications.state = CompletedProgress(ShownToModelProgress(0)) await notifications.publish(ui, chat) - await ui.until_shows("▶ Answered · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) async def test_a_hidden_view_unsubscribes_and_discards_a_late_reply(workspace): @@ -162,7 +162,7 @@ async def test_markdown_uses_the_same_typed_line_and_an_empty_recorded_result(wo await chat.message_history.toggle_style() await ui.step() await notifications.publish(ui, chat) - await ui.until_shows("▶ Answered · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) notifications.recorded = False await notifications.publish(ui, chat) await ui.until_shows("▶ No recorded notification result") From 0a757e54b666d3c11f64fe034933f19856b3d37b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 00:05:51 -0400 Subject: [PATCH 3/8] Read queue references from the root-free notification operation --- src/toad/submission_queue.py | 2 +- tests/textual/test_historical_notification_source.py | 6 ++---- tests/textual/test_message_notifications.py | 8 ++++---- 3 files changed, 7 insertions(+), 9 deletions(-) diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py index a280f5ea4..c4a601330 100644 --- a/src/toad/submission_queue.py +++ b/src/toad/submission_queue.py @@ -42,7 +42,7 @@ class MessageQueueEntry: @property def reference(self) -> MessageReference: - return self.request.references[0] + return self.request.operation.references[0] @property def summary(self) -> str: diff --git a/tests/textual/test_historical_notification_source.py b/tests/textual/test_historical_notification_source.py index 4959f5871..fcf3798e1 100644 --- a/tests/textual/test_historical_notification_source.py +++ b/tests/textual/test_historical_notification_source.py @@ -2,7 +2,7 @@ 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.threads import Thread from support.roots import installed_wire @@ -28,9 +28,7 @@ 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 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_message_notifications.py b/tests/textual/test_message_notifications.py index b7d7f5e1a..5a03cdfa4 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(newest_status(chat).title) 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 From 5f72bcfeb2a334d2485f9b0378dfc539a6b1fae5 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 00:41:16 -0400 Subject: [PATCH 4/8] Retain actual Core display scope through notification observation --- src/toad/notification_status.py | 5 ++- src/toad/screens/historical_sessions.py | 5 +-- src/toad/transcript_preparation.py | 6 ++++ src/toad/widgets/message_notifications.py | 17 ++++++++-- src/toad/wire_history.py | 36 ++++++++++++++++----- tests/textual/test_common_queue.py | 8 +++++ tests/textual/test_message_notifications.py | 3 +- tests/textual/test_wire_source.py | 22 ++++++++++++- 8 files changed, 85 insertions(+), 17 deletions(-) diff --git a/src/toad/notification_status.py b/src/toad/notification_status.py index 17fccce8d..4ea8184fa 100644 --- a/src/toad/notification_status.py +++ b/src/toad/notification_status.py @@ -98,7 +98,7 @@ def queue_title(self) -> str: def queue_details(self) -> DetailLines: leading = CheckedNotifications(self.notifications[:1]) return ( - (Span(f"{len(self.notifications)} observed recipients; leading result shown", MutedRole),), + (Span(self.title, MutedRole),), *((Span(f"Recipient {item.recipient_identity.recipient_lookup}", MutedRole),) for item in leading.notifications), *leading.details, @@ -116,8 +116,7 @@ def details(self) -> DetailLines: Span(reader.name, NameRole, OpenTarget(reader.name))) lines.append(line) return (tuple(lines) if lines else ( - (Span("No notification result is recorded for this message. " - "This does not confirm the message was received.", MutedRole),),)) + (Span("No recipient feedback to display.", MutedRole),),)) @dataclass(frozen=True) diff --git a/src/toad/screens/historical_sessions.py b/src/toad/screens/historical_sessions.py index 2395cc0a4..6a7d41755 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.ui_model.observation import ReadTranscript +from agent_comms.notification_reads import RecordedNotificationRead 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/transcript_preparation.py b/src/toad/transcript_preparation.py index b643021e7..1ab9f2869 100644 --- a/src/toad/transcript_preparation.py +++ b/src/toad/transcript_preparation.py @@ -11,6 +11,7 @@ from weakref import ReferenceType, ref from agent_comms.historical_views import ConversationObservation +from agent_comms.read_basis import DisplayIdentity, NoDisplaySelection from agent_comms.sent_message import OutgoingMessageReference from agent_comms.transcripts import ( TranscriptCursor, @@ -78,6 +79,11 @@ def measured(cls, page, fragments: tuple[TranscriptFragment, ...], **view): prepared.measure_sources() return prepared + @property + def notification_identity(self) -> DisplayIdentity | NoDisplaySelection: + """Native transcript pages have no captured BUS display selection.""" + return NoDisplaySelection() + @property def source_payload_bytes(self) -> int: return sum(256 + 4 * event.text_size for event in self.page.events) diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py index a4538b942..4d5ce9f40 100644 --- a/src/toad/widgets/message_notifications.py +++ b/src/toad/widgets/message_notifications.py @@ -12,6 +12,8 @@ from textual.widgets import Collapsible from agent_comms.ui_model.observation import ReadNotifications +from agent_comms.notification_reads import LiveNotificationRead +from agent_comms.read_basis import NoDisplaySelection from toad.submission_queue import QueueEvidenceChanged from toad.notification_status import ( @@ -111,7 +113,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. @@ -128,7 +130,12 @@ 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) + references = tuple(references) + wanted = frozenset(references) + display = next((target.notification_identity_for(wanted) + for target in self.notification_targets() + if wanted.intersection(target.notification_references)), NoDisplaySelection()) + return ReadNotifications(root, self.notification_operation(references), display=display) def subscribe_notifications(self, references) -> None: access = self.app.coordination_access @@ -174,6 +181,12 @@ def on_show(self) -> None: class SourceNotifications: notification_view: HistoryNotifications + def notification_identity_for(self, references): + """Use retained prepared Core pages, even after the pending paint was retired.""" + return next((page.prepared.notification_identity for page in reversed(self.pages) + if any(references.intersection(fragment.notification_references) + for fragment in page.prepared.fragments_in_range)), NoDisplaySelection()) + def rows_published(self) -> None: super().rows_published() self.notification_view.call_after_refresh(self.notification_view.check_notifications) diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index a7bc45277..7e00e0383 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -23,6 +23,7 @@ ConversationRevision, ) from agent_comms.message_page import MessagePage +from agent_comms.read_basis import NoDisplaySelection from agent_comms.messages import ConversationRole, Membership, Message, NoticeRole from agent_comms.page_traversal import ( LiveCursor, @@ -103,6 +104,13 @@ class AfterMessage(WireCut): bounded = True kept = True + @property + def undrawn(self) -> bool: + return False + + def observed(self) -> AfterMessage: + return self + @dataclass(frozen=True) class AfterSequence(AfterMessage): @@ -128,6 +136,13 @@ class NothingRead(AfterSequence): cursor: LiveCursor = field(default_factory=lambda: LiveCursor(0)) + @property + def undrawn(self) -> bool: + return True + + def observed(self) -> AfterSequence: + return AfterSequence(self.cursor) + @dataclass(frozen=True) class WirePage: @@ -138,21 +153,21 @@ class WirePage: after: AfterMessage has_older: bool has_newer: bool - source: MessagePage | None + source: MessagePage @classmethod def unread(cls) -> WirePage: - """What is drawn before the first read: nothing, ending before any answer.""" - return cls((), None, NothingRead(), False, False, None) + """The declared unread cursor carries an empty, unscoped placeholder.""" + return cls((), None, NothingRead(), False, False, + MessagePage((), False, False, NoDisplaySelection())) @property def undrawn(self) -> bool: - return self.source is None + return self.after.undrawn def accepts(self, page: MessagePage, view: type[ConversationHistoryView]) -> bool: """An initial placeholder accepts its first scope; Core compares actual pages.""" - source = self.source - return source is None or view.same_display(page, source) + return self.undrawn or view.same_display(page, self.source) def advanced(self, interval: WirePage, *, older: bool) -> WirePage: """An empty interval advances its cut and establishes the first Core page. @@ -162,7 +177,8 @@ def advanced(self, interval: WirePage, *, older: bool) -> WirePage: """ source = interval.source if self.undrawn else self.source return ( - replace(self, before=interval.before, has_older=interval.has_older, source=source) + replace(self, before=interval.before, after=self.after.observed(), + has_older=interval.has_older, source=source) if older else replace(self, after=interval.after, has_newer=interval.has_newer, source=source) ) @@ -171,6 +187,10 @@ def advanced(self, interval: WirePage, *, older: bool) -> WirePage: class PreparedWirePage(PreparedTranscriptPage): """Wire intervals preserve Core's original display page when advanced empty.""" + @property + def notification_identity(self): + return self.page.source.selection.notification_identity + def advance_empty(self, prepared: PreparedTranscriptPage, older: bool) -> None: self.page = self.page.advanced(prepared.page, older=older) @@ -330,7 +350,7 @@ def read(self, request: PageRequest[WireCut], through: AfterSequence) -> Prepare wire = WirePage( page.messages, BeforeMessage(page.messages[0].view_cursor) if page.messages else None, - AfterMessage(page.messages[-1].view_cursor) if page.has_newer and page.messages else through, + AfterMessage(page.messages[-1].view_cursor) if page.has_newer and page.messages else through.observed(), page.has_older and bool(page.messages), page.has_newer and bool(page.messages), page, diff --git a/tests/textual/test_common_queue.py b/tests/textual/test_common_queue.py index 089285510..514c35667 100644 --- a/tests/textual/test_common_queue.py +++ b/tests/textual/test_common_queue.py @@ -36,6 +36,11 @@ async def test_bus_queue_borrows_original_proofs_and_preserves_draft(workspace, entries = owner.message_queue_entries assert all(entry.reference in {message.reference for message in originals} for entry in entries) assert all(entry.request.root == str(workspace.comms.root) for entry in entries) + captured = {page.prepared.notification_identity for page in view.message_history.pages} + human = workspace.comms.registry.require(workspace.viewer).incarnation + assert all(entry.request.display in captured for entry in entries) + assert all(entry.request.display.viewer == human for entry in entries) + assert not view.message_history.undrawn assert view.query_one(QueueSummary) in ui.app.screen._compositor.visible_widgets assert not owner.queue_source.queue_actions_available(owner) with pytest.raises(ValueError): @@ -57,6 +62,9 @@ async def test_bus_queue_borrows_original_proofs_and_preserves_draft(workspace, previous = {entry.reference for entry in entries} await ui.until(lambda: bool({entry.reference for entry in owner.message_queue_entries} - previous), why="source subscription to refresh the existing queue consumer") + captured = {page.prepared.notification_identity for page in view.message_history.pages} + assert all(entry.request.display in captured for entry in owner.message_queue_entries) + assert ui.draft() == "@peer kept draft" async def test_mailbox_factory_supplies_same_owner_and_readonly_capability_wins_mro(workspace): diff --git a/tests/textual/test_message_notifications.py b/tests/textual/test_message_notifications.py index 5a03cdfa4..7097802ba 100644 --- a/tests/textual/test_message_notifications.py +++ b/tests/textual/test_message_notifications.py @@ -165,4 +165,5 @@ async def test_markdown_uses_the_same_typed_line_and_an_empty_recorded_result(wo await ui.until_shows(newest_status(chat).title) notifications.recorded = False await notifications.publish(ui, chat) - await ui.until_shows("▶ No recorded notification result") + await ui.until_shows(newest_status(chat).title) + assert newest_status(chat).notifications == () diff --git a/tests/textual/test_wire_source.py b/tests/textual/test_wire_source.py index 59635665e..c4e08a650 100644 --- a/tests/textual/test_wire_source.py +++ b/tests/textual/test_wire_source.py @@ -15,7 +15,10 @@ from toad.transcript_preparation import PageRequest, Unchanged from toad.widgets.presentation_window import OLDER_END, NEWER_END from toad.message_rows import IrcMessageStyle -from toad.wire_history import AfterSequence, GrowingWireObservation, MessageFragment, ReplacedWireObservation, WireSource +from toad.wire_history import ( + AfterSequence, GrowingWireObservation, MessageFragment, NothingRead, + PreparedWirePage, ReplacedWireObservation, WirePage, WireSource, +) def channel_source(comms, target, project): @@ -32,6 +35,23 @@ def observed(source): return first, newest +def test_empty_first_read_keeps_its_real_identity_after_the_unread_placeholder(tmp_path): + comms = installed_wire(tmp_path / 'wire') + comms.registry.declare(Thread('peer', frozenset({'team'}), Checkout(tmp_path))) + source = channel_source(comms, '#team', tmp_path) + placeholder = PreparedWirePage.empty(WirePage.unread()) + assert placeholder.page.undrawn and not placeholder.page.source.selection.has_evidence + initial = NothingRead() + prepared = source.read(PageRequest(OLDER_END, initial), initial) + assert not prepared.fragments and not prepared.page.undrawn + identity = prepared.page.source.selection.notification_identity + assert prepared.notification_identity == identity + assert identity.viewer == comms.registry.require(source.viewer.name).incarnation + assert placeholder.page.accepts(prepared.page.source, source.conversation.history_view) + placeholder.advance_empty(prepared, older=True) + assert not placeholder.page.undrawn and placeholder.notification_identity == identity + + def test_a_turn_claim_on_the_peer_keeps_the_direct_message_page_identity(tmp_path): comms = installed_wire(tmp_path / "wire") comms.registry.declare(Thread("peer", frozenset(), Checkout(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()))) From 4debca4df7854894ceef628dfcd457bd6b4aa7fb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 00:52:41 -0400 Subject: [PATCH 5/8] Close captured viewer and matched empty-feedback controls --- src/toad/widgets/message_notifications.py | 9 +++-- src/toad/wire_history.py | 4 +- tests/textual/test_common_queue.py | 46 ++++++++++++++++++++--- tests/textual/test_wire_source.py | 3 +- 4 files changed, 49 insertions(+), 13 deletions(-) diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py index 4d5ce9f40..0e71542f8 100644 --- a/src/toad/widgets/message_notifications.py +++ b/src/toad/widgets/message_notifications.py @@ -3,7 +3,7 @@ from __future__ import annotations from abc import ABC, abstractmethod -from collections.abc import Awaitable +from collections.abc import Awaitable, Iterable from dataclasses import dataclass, replace from typing import TYPE_CHECKING @@ -13,7 +13,8 @@ from agent_comms.ui_model.observation import ReadNotifications from agent_comms.notification_reads import LiveNotificationRead -from agent_comms.read_basis import NoDisplaySelection +from agent_comms.read_basis import DisplayIdentity, NoDisplaySelection +from agent_comms.message_reference import MessageReference from toad.submission_queue import QueueEvidenceChanged from toad.notification_status import ( @@ -124,7 +125,7 @@ def check_notifications(self) -> None: self.subscribe_notifications(references) self.post_message(QueueEvidenceChanged()) - def notification_read(self, references) -> ReadNotifications: + def notification_read(self, references: Iterable[MessageReference]) -> ReadNotifications: from toad.project_path_owner import ProjectPathOwner root = ProjectPathOwner.containing(self).coordination_root @@ -181,7 +182,7 @@ def on_show(self) -> None: class SourceNotifications: notification_view: HistoryNotifications - def notification_identity_for(self, references): + def notification_identity_for(self, references: frozenset[MessageReference]) -> DisplayIdentity | NoDisplaySelection: """Use retained prepared Core pages, even after the pending paint was retired.""" return next((page.prepared.notification_identity for page in reversed(self.pages) if any(references.intersection(fragment.notification_references) diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index 7e00e0383..af7e6911d 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -23,7 +23,7 @@ ConversationRevision, ) from agent_comms.message_page import MessagePage -from agent_comms.read_basis import NoDisplaySelection +from agent_comms.read_basis import DisplayIdentity, NoDisplaySelection from agent_comms.messages import ConversationRole, Membership, Message, NoticeRole from agent_comms.page_traversal import ( LiveCursor, @@ -188,7 +188,7 @@ class PreparedWirePage(PreparedTranscriptPage): """Wire intervals preserve Core's original display page when advanced empty.""" @property - def notification_identity(self): + def notification_identity(self) -> DisplayIdentity | NoDisplaySelection: return self.page.source.selection.notification_identity def advance_empty(self, prepared: PreparedTranscriptPage, older: bool) -> None: diff --git a/tests/textual/test_common_queue.py b/tests/textual/test_common_queue.py index 514c35667..bbd7e47c4 100644 --- a/tests/textual/test_common_queue.py +++ b/tests/textual/test_common_queue.py @@ -3,24 +3,36 @@ These controls use the private wire, real observer, prompt and submission owner. No product monkeypatches and no provider input are needed. """ +from collections.abc import Awaitable, Callable + import pytest +from agent_comms.bus import Bus +from agent_comms.channel_targets import ThreadTarget +from agent_comms.presentation import DisplayOutcomes, MessageNotification +from agent_comms.recorded_bus import RecordedBusIdentity +from support.ui import Ui +from support.workspace import Workspace from agent_comms.acp_extension import NoQueueScope +from toad.notification_status import CheckedNotifications from toad.session_presentation import MailboxSessionSources from toad.widgets.comms_chat import CommsChatView from toad.widgets.prompt import QueueSummary from textual.widgets import Button -async def open_channel(ui, viewer): - return await ui.open_channel("#team", actor=viewer) +async def open_channel(ui: Ui, viewer: str) -> str: + # The tab acts as an agent; the Core page is still read by the actual human. + return await ui.open_channel("#team", actor="peer") -async def open_direct(ui, viewer): +async def open_direct(ui: Ui, viewer: str) -> str: return await ui.open_direct("peer", actor=viewer) @pytest.mark.parametrize("open_surface", [open_channel, open_direct]) -async def test_bus_queue_borrows_original_proofs_and_preserves_draft(workspace, open_surface): +async def test_bus_queue_borrows_original_proofs_and_preserves_draft( + workspace: Workspace, open_surface: Callable[[Ui, str], Awaitable[str]], +) -> None: workspace.thread("peer", "team") workspace.thread("recipient", "team") originals = ( @@ -39,7 +51,9 @@ async def test_bus_queue_borrows_original_proofs_and_preserves_draft(workspace, captured = {page.prepared.notification_identity for page in view.message_history.pages} human = workspace.comms.registry.require(workspace.viewer).incarnation assert all(entry.request.display in captured for entry in entries) - assert all(entry.request.display.viewer == human for entry in entries) + namespace = RecordedBusIdentity.capture(Bus.path_in(workspace.comms.root)) + assert all(entry.request.display.outcome_viewer() == DisplayOutcomes(human, namespace) + for entry in entries) assert not view.message_history.undrawn assert view.query_one(QueueSummary) in ui.app.screen._compositor.visible_widgets assert not owner.queue_source.queue_actions_available(owner) @@ -67,7 +81,7 @@ async def test_bus_queue_borrows_original_proofs_and_preserves_draft(workspace, assert ui.draft() == "@peer kept draft" -async def test_mailbox_factory_supplies_same_owner_and_readonly_capability_wins_mro(workspace): +async def test_mailbox_factory_supplies_same_owner_and_readonly_capability_wins_mro(workspace: Workspace) -> None: async with workspace.app(size=(110, 35)) as ui: view = ui.conversation source = MailboxSessionSources() @@ -79,3 +93,23 @@ async def test_mailbox_factory_supplies_same_owner_and_readonly_capability_wins_ with pytest.raises(ValueError): await source.queue_clear(view.submissions, source.queue_scope(view.submissions)) await source.close() + + +def test_matched_empty_feedback_is_not_a_missing_original(workspace: Workspace) -> None: + workspace.thread('peer', 'team') + comms = workspace.comms + human = comms.registry.require(workspace.viewer) + first = comms.messaging.send('peer', human.name, 'matched original').message + missing = comms.messaging.send('peer', human.name, 'not in this result window').message + page = comms.views.dm_display_page(ThreadTarget('peer'), worktree=human.worktree) + comms.views.mark_page_read(page, worktree=human.worktree) + document = comms.bus.reads.read() + window = comms.views.message_notifications_for_references( + page.selection.notification_identity.outcome_viewer(), (first.reference,), + ) + checked, = CheckedNotifications.for_references((first.reference,), window.outcomes) + assert checked.notifications == () + assert not checked.unavailable and not checked.working + assert checked.title == MessageNotification.summary(checked.notifications) + assert tuple(CheckedNotifications.for_references((missing.reference,), window.outcomes)) == () + assert comms.bus.reads.read() == document diff --git a/tests/textual/test_wire_source.py b/tests/textual/test_wire_source.py index c4e08a650..f29c6b420 100644 --- a/tests/textual/test_wire_source.py +++ b/tests/textual/test_wire_source.py @@ -1,6 +1,7 @@ """Textual-facing: the presentation window; a wire source pages one conversation and notices when Core replaces its reading.""" import os +from pathlib import Path from agent_comms.repo import Checkout from agent_comms.channel_targets import ChannelTarget, ThreadTarget @@ -35,7 +36,7 @@ def observed(source): return first, newest -def test_empty_first_read_keeps_its_real_identity_after_the_unread_placeholder(tmp_path): +def test_empty_first_read_keeps_its_real_identity_after_the_unread_placeholder(tmp_path: Path) -> None: comms = installed_wire(tmp_path / 'wire') comms.registry.declare(Thread('peer', frozenset({'team'}), Checkout(tmp_path))) source = channel_source(comms, '#team', tmp_path) From 27999b1ba0216f290478f0e637b5e81e636fd2a6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 03:17:21 -0400 Subject: [PATCH 6/8] Keep queue refusals typed and source identity lifetime-owned --- src/toad/acp/agent_controller.py | 8 +- src/toad/agent_presentation.py | 5 +- src/toad/conversation_submission.py | 8 +- src/toad/notification_status.py | 24 ++++- src/toad/submission_queue.py | 21 ++-- src/toad/transcript_preparation.py | 13 ++- tests/support/peers/refuses_queue.py | 15 +++ tests/textual/test_common_queue.py | 13 ++- tests/textual/test_queue_action_boundary.py | 114 ++++++++++++++++++++ tests/textual/test_wire_message_lines.py | 2 +- 10 files changed, 190 insertions(+), 33 deletions(-) create mode 100644 tests/support/peers/refuses_queue.py create mode 100644 tests/textual/test_queue_action_boundary.py diff --git a/src/toad/acp/agent_controller.py b/src/toad/acp/agent_controller.py index 91057609d..a96c8fa13 100644 --- a/src/toad/acp/agent_controller.py +++ b/src/toad/acp/agent_controller.py @@ -32,6 +32,8 @@ from toad.plan import PlanItem from toad.surface_binding import DetachedSurfaceBinding, SurfaceBinding +from toad.submission_queue import QueueActionRefused + from .prompt import build as build_prompt from .session_output import SessionOutput from .terminal_owner import OperationalTerminalOwner @@ -284,7 +286,7 @@ async def _prompt(self, content, command, authority: ClientSessionRequest, queue def require(): authority.require() if not agent.queue_attachment.accepts_request(queue_scope): - raise ValueError("The queued input owner changed before submission; inspect the current queue.") + raise QueueActionRefused("The queued input owner changed before submission; inspect the current queue.") try: result = await agent.process.prompt( @@ -334,8 +336,8 @@ def _prompt_failed( if not published: report(LogAgentFail(failure.title, failure.feedback, log_path=agent.presentation.log_path)) - async def clear_queue(self) -> None: - await self.submit_blocks([{"type": "text", "text": " "}], ClearQueueRequest()) + async def clear_queue(self) -> bool: + return await self.submit_blocks([{"type": "text", "text": " "}], ClearQueueRequest()) is not None async def send_now(self) -> bool: authority = ClientSessionRequest(self.agent, self.agent.session_id) diff --git a/src/toad/agent_presentation.py b/src/toad/agent_presentation.py index e017da152..200445a20 100644 --- a/src/toad/agent_presentation.py +++ b/src/toad/agent_presentation.py @@ -9,6 +9,7 @@ from agent_comms.acp_extension import NoQueueScope, PendingQueueProjection, QueueProjection, QueueScope from toad.private_native_cursor import CursorStatus, HiddenCursorStatus from toad.conversation_turn import TurnBinding, LocalTurnBinding, ManagedTurnBinding +from toad.submission_queue import QueueActionRefused if TYPE_CHECKING: @@ -58,7 +59,7 @@ def queue_actions_available(self) -> bool: return False def require_queue_action(self, scope: QueueScope) -> Agent: - raise ValueError("No live ACP queue authority is available") + raise QueueActionRefused("No live ACP queue authority is available") @property def prompt_in_flight(self): @@ -129,7 +130,7 @@ def queue_actions_available(self) -> bool: def require_queue_action(self, scope: QueueScope) -> Agent: if not self.queue_supported or not self.agent.queue_attachment.accepts_request(scope): - raise ValueError("Queue attachment changed or is unavailable; inspect the current queue again") + raise QueueActionRefused("Queue attachment changed or is unavailable; inspect the current queue again") return self.agent @property diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py index 48322d7ad..7b199b684 100644 --- a/src/toad/conversation_submission.py +++ b/src/toad/conversation_submission.py @@ -21,6 +21,7 @@ from toad import messages from toad.acp.client_session import ClientSessionRequest from toad.command_catalog import CommandCatalog +from toad.submission_queue import QueueActionRefused from toad.core.input_events import UserInputSubmitted from toad.core_event_carrier import CoreEventMessage from toad.widgets.prompt import Prompt @@ -372,11 +373,12 @@ def send_now(self): ) async def _send_now(self, source: QueueSource, agent, scope: QueueScope) -> None: + if self.queue_source is not source or self.view.agent_binding is not agent: + self.view.feedback.flash("The source changed before send now", style="error") + return try: - if self.queue_source is not source or self.view.agent_binding is not agent: - raise ValueError("The source changed before send now") await source.queue_send_now(self, scope) - except (RequestError, ValueError) as error: + except (RequestError, QueueActionRefused) as error: if self.queue_source is source and self.view.agent_binding is agent: self.view.feedback.flash(f"Send now failed: {error}", style="error") self.publish_pending() diff --git a/src/toad/notification_status.py b/src/toad/notification_status.py index 4ea8184fa..4004d3a81 100644 --- a/src/toad/notification_status.py +++ b/src/toad/notification_status.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Iterable, Iterator, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import ClassVar, TYPE_CHECKING from agent_comms.message_reference import MessageReference @@ -79,7 +79,12 @@ def for_references(cls, references: Iterable[MessageReference], """Yield nothing when this answer has no matching reference.""" loaded = results.keys() & {(reference.seq, reference.message_id) for reference in references} if loaded: - yield cls(tuple(item for reference in loaded for item in results[reference])) + yield cls.from_notifications(tuple(item for reference in loaded for item in results[reference])) + + @classmethod + def from_notifications(cls, notifications: tuple[MessageNotification, ...]) -> CheckedNotifications: + """Keep a matched empty result distinct from an observed recipient.""" + return cls(notifications) if notifications else EmptyCheckedNotifications() @property def working(self) -> bool: @@ -91,12 +96,13 @@ def title(self) -> str: @property def queue_title(self) -> str: - # Keep each recipient's actual proof label; do not turn routing into arrival. - return " · ".join(f"@{item.recipient} {item.label}" for item in self.notifications[:1]) or "Unobserved" + # Keep the leading recipient's actual proof; routing is not arrival. + first = self.notifications[0] + return f"@{first.recipient} {first.label}" @property def queue_details(self) -> DetailLines: - leading = CheckedNotifications(self.notifications[:1]) + leading = CheckedNotifications.from_notifications(self.notifications[:1]) return ( (Span(self.title, MutedRole),), *((Span(f"Recipient {item.recipient_identity.recipient_lookup}", MutedRole),) @@ -119,6 +125,14 @@ def details(self) -> DetailLines: (Span("No recipient feedback to display.", MutedRole),),)) +@dataclass(frozen=True) +class EmptyCheckedNotifications(CheckedNotifications): + """The original matched, but no recipient feedback is visible to this viewer.""" + + notifications: tuple[()] = field(default=(), init=False) + queue_title = "No recipient feedback" + + @dataclass(frozen=True) class UnavailableNotifications(NotificationStatus): error: str diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py index c4a601330..9fda6d5d5 100644 --- a/src/toad/submission_queue.py +++ b/src/toad/submission_queue.py @@ -31,6 +31,10 @@ from toad.conversation_submission import ConversationSubmissions +class QueueActionRefused(Exception): + """A deliberate queue refusal, not an unexpected programming failure.""" + + class QueueEvidenceChanged(Message): """Local repaint after existing notification/viewport evidence changes.""" @@ -90,10 +94,10 @@ def queue_actions_available(self, owner: ConversationSubmissions) -> bool: return False async def queue_send_now(self, owner: ConversationSubmissions, scope: QueueScope) -> None: - raise ValueError("Send now unavailable: this source has no native queue authority") + raise QueueActionRefused("Send now unavailable: this source has no native queue authority") async def queue_clear(self, owner: ConversationSubmissions, scope: QueueScope) -> None: - raise ValueError("Clear unavailable: BUS messages cannot be withdrawn through an ACP queue") + raise QueueActionRefused("Clear unavailable: BUS messages cannot be withdrawn through an ACP queue") class NativeQueueSource(QueueSource): @@ -137,10 +141,12 @@ def require_queue_action(self, owner: ConversationSubmissions, scope: QueueScope async def queue_send_now(self, owner: ConversationSubmissions, scope: QueueScope) -> None: agent = self.require_queue_action(owner, scope) if not await agent.controller.send_now(): - raise ValueError("The native owner did not accept send now") + raise QueueActionRefused("The native owner did not accept send now") async def queue_clear(self, owner: ConversationSubmissions, scope: QueueScope) -> None: - await self.require_queue_action(owner, scope).controller.clear_queue() + agent = self.require_queue_action(owner, scope) + if not await agent.controller.clear_queue(): + raise QueueActionRefused("The native owner did not accept clear queue") class BusQueueSource(ReadOnlyQueueSource): @@ -197,11 +203,12 @@ def compose(self): yield Button("Close", id="queue-close") async def apply_action(self, action: Callable[[ConversationSubmissions, QueueScope], Awaitable[None]]) -> None: + if self.submissions.queue_source is not self.source or self.submissions.view.agent_binding is not self.agent: + self.query_one("#queue-error", Static).update("The source changed; inspect its current queue again") + return try: - if self.submissions.queue_source is not self.source or self.submissions.view.agent_binding is not self.agent: - raise ValueError("The source changed; inspect its current queue again") await action(self.submissions, self.scope) - except (ValueError, RequestError) as error: + except (QueueActionRefused, RequestError) as error: self.query_one("#queue-error", Static).update(str(error)) return self.submissions.publish_pending() diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py index 1ab9f2869..9525348f2 100644 --- a/src/toad/transcript_preparation.py +++ b/src/toad/transcript_preparation.py @@ -21,6 +21,7 @@ from agent_comms.ui_model.message_category import MessageCategory from toad.render_backend import ReusableRenderTask +from toad.submission_queue import QueueSource, ReadOnlyQueueSource from toad.transcript_state import LatestViewportRequest from toad.widgets.history_anchor import LatestPosition from toad.widgets.message_notifications import NothingToRead, PaintedRead @@ -295,15 +296,13 @@ def close(self) -> None: pass -class PreparedPageSource(ABC): - @property - def queue_source(self): - from toad.submission_queue import ReadOnlyQueueSource - - return ReadOnlyQueueSource() - +class PreparedPageSource(ReadOnlyQueueSource): """One lifetime-owned source of prepared pages, independent of widget paging.""" + @property + def queue_source(self) -> QueueSource: + return self + runtime: PreparationRuntime scope: PreparationScope through: TranscriptCursor diff --git a/tests/support/peers/refuses_queue.py b/tests/support/peers/refuses_queue.py new file mode 100644 index 000000000..dcba5a39e --- /dev/null +++ b/tests/support/peers/refuses_queue.py @@ -0,0 +1,15 @@ +"""An external ACP peer refuses a queue request without losing the session.""" + +from acp.exceptions import RequestError + +from base import Peer + + +class RefusesQueuePeer(Peer): + async def prompt(self, session_id, prompt, **kwargs): + self.record(prompt=prompt[0].text) + raise RequestError.invalid_params({"details": "Queue action refused by peer"}) + + +if __name__ == "__main__": + RefusesQueuePeer.main() diff --git a/tests/textual/test_common_queue.py b/tests/textual/test_common_queue.py index bbd7e47c4..84ccfd223 100644 --- a/tests/textual/test_common_queue.py +++ b/tests/textual/test_common_queue.py @@ -13,7 +13,8 @@ from support.ui import Ui from support.workspace import Workspace from agent_comms.acp_extension import NoQueueScope -from toad.notification_status import CheckedNotifications +from toad.notification_status import CheckedNotifications, EmptyCheckedNotifications +from toad.submission_queue import QueueActionRefused from toad.session_presentation import MailboxSessionSources from toad.widgets.comms_chat import CommsChatView from toad.widgets.prompt import QueueSummary @@ -57,9 +58,9 @@ async def test_bus_queue_borrows_original_proofs_and_preserves_draft( assert not view.message_history.undrawn assert view.query_one(QueueSummary) in ui.app.screen._compositor.visible_widgets assert not owner.queue_source.queue_actions_available(owner) - with pytest.raises(ValueError): + with pytest.raises(QueueActionRefused): await owner.queue_source.queue_clear(owner, NoQueueScope()) - with pytest.raises(ValueError): + with pytest.raises(QueueActionRefused): await owner.queue_source.queue_send_now(owner, NoQueueScope()) previous_screen = ui.app.screen @@ -88,9 +89,9 @@ async def test_mailbox_factory_supplies_same_owner_and_readonly_capability_wins_ prompt = source.prompt(view) assert prompt.operational_submissions is view.submissions assert not source.queue_actions_available(view.submissions) - with pytest.raises(ValueError): + with pytest.raises(QueueActionRefused): await source.queue_send_now(view.submissions, source.queue_scope(view.submissions)) - with pytest.raises(ValueError): + with pytest.raises(QueueActionRefused): await source.queue_clear(view.submissions, source.queue_scope(view.submissions)) await source.close() @@ -108,7 +109,9 @@ def test_matched_empty_feedback_is_not_a_missing_original(workspace: Workspace) page.selection.notification_identity.outcome_viewer(), (first.reference,), ) checked, = CheckedNotifications.for_references((first.reference,), window.outcomes) + assert isinstance(checked, EmptyCheckedNotifications) assert checked.notifications == () + assert checked.queue_title == "No recipient feedback" assert not checked.unavailable and not checked.working assert checked.title == MessageNotification.summary(checked.notifications) assert tuple(CheckedNotifications.for_references((missing.reference,), window.outcomes)) == () diff --git a/tests/textual/test_queue_action_boundary.py b/tests/textual/test_queue_action_boundary.py new file mode 100644 index 000000000..c10894f48 --- /dev/null +++ b/tests/textual/test_queue_action_boundary.py @@ -0,0 +1,114 @@ +"""Queue actions keep stale sources and deliberate refusals apart from bugs. + +Use the actual app, source retirement, agent binding and ACP transport. The +unexpected-error callback exercises the inspection's public action boundary; +it neither patches a product owner nor submits input to a model. +""" + +import pytest +from agent_comms.acp_extension import ClearQueueRequest, QueueScope, encode_request +from textual.widgets import Static + +from support.agents import peer +from support.workspace import Workspace +from toad.acp.client_session import ClientSessionRequest +from toad.conversation_submission import ConversationSubmissions +from toad.session_scope import AgentAttached +from toad.widgets.comms_chat import CommsChatView + + +async def unexpected_value_error(owner: ConversationSubmissions, scope: QueueScope) -> None: + raise ValueError("Unexpected action defect") + + +async def test_stale_source_stops_before_the_action_and_closed_source_identity_is_stable( + workspace: Workspace, +) -> None: + workspace.thread("peer", "team") + async with workspace.app(size=(115, 38)) as ui: + await ui.open_channel("#team", actor=workspace.viewer) + view = ui.app.selected_session.query_one(CommsChatView) + owner = view.submissions + await ui.write_draft("kept across source retirement") + inspection = owner.inspect_queue() + await ui.until(lambda: ui.app.screen is inspection, why="queue inspection mounted") + await view.message_history.retire_source() + view.message_history.close() + replacement = view.message_history.page_source + assert owner.queue_source is replacement + assert owner.queue_source is owner.queue_source + assert owner.queue_source is not inspection.source + # If stale checks reach this action, ValueError must fail the control. + await inspection.apply_action(unexpected_value_error) + assert ui.app.screen is inspection + assert inspection.query_one("#queue-error", Static).content + inspection.dismiss() + await ui.until(lambda: ui.app.screen is not inspection, why="inspection closed") + assert ui.draft() == "kept across source retirement" + + +async def test_stale_agent_stops_before_inspection_or_send_now_action(workspace: Workspace) -> None: + async with workspace.app(size=(115, 38)) as ui: + owner = ui.conversation.submissions + await ui.write_draft("kept across agent replacement") + source, previous = owner.queue_source, owner.view.agent_binding + scope = source.queue_scope(owner) + inspection = owner.inspect_queue() + await ui.until(lambda: ui.app.screen is inspection, why="queue inspection mounted") + replacement = ui.detached_agent("replacement") + ui.attach_agent(replacement) + assert owner.queue_source is source + assert owner.view.agent_binding is not previous + await inspection.apply_action(unexpected_value_error) + assert ui.app.screen is inspection + assert inspection.query_one("#queue-error", Static).content + await owner._send_now(source, previous, scope) + assert owner.view.feedback.display + assert owner.view.feedback.content + inspection.dismiss() + await ui.until(lambda: ui.app.screen is not inspection, why="inspection closed") + assert ui.draft() == "kept across agent replacement" + await replacement.stop() + + +async def test_unsupported_action_stays_visible_and_unrelated_value_error_escapes(workspace: Workspace) -> None: + async with workspace.app(size=(115, 38)) as ui: + owner = ui.conversation.submissions + await ui.write_draft("kept after refusal") + inspection = owner.inspect_queue() + await ui.until(lambda: ui.app.screen is inspection, why="queue inspection mounted") + await inspection.apply_action(inspection.source.queue_clear) + assert ui.app.screen is inspection + assert inspection.query_one("#queue-error", Static).content + with pytest.raises(ValueError, match="Unexpected action defect"): + await inspection.apply_action(unexpected_value_error) + assert ui.app.screen is inspection + inspection.dismiss() + await ui.until(lambda: ui.app.screen is not inspection, why="inspection closed") + assert ui.draft() == "kept after refusal" + + +async def test_real_acp_refusal_keeps_inspection_open_without_retry(workspace: Workspace) -> None: + async with workspace.app(size=(115, 38), binding=AgentAttached(peer("refuses_queue"))) as ui: + await ui.until(ui.agent_ready, why="external ACP peer ready") + owner = ui.conversation.submissions + agent = owner.view.agent_binding + authority = ClientSessionRequest(agent, agent.session_id) + await ui.write_draft("kept after ACP refusal") + inspection = owner.inspect_queue() + await ui.until(lambda: ui.app.screen is inspection, why="queue inspection mounted") + + async def refused_at_acp(submissions: ConversationSubmissions, scope: QueueScope) -> None: + # The official SDK returns RequestError from an actual child process. + await agent.process.prompt( + agent, [{"type": "text", "text": " "}], encode_request(ClearQueueRequest()), authority.require, + ) + + await inspection.apply_action(refused_at_acp) + assert ui.app.screen is inspection + assert inspection.query_one("#queue-error", Static).content + assert authority.current + assert workspace.prompts_received() == [" "] + inspection.dismiss() + await ui.until(lambda: ui.app.screen is not inspection, why="inspection closed") + assert ui.draft() == "kept after ACP refusal" diff --git a/tests/textual/test_wire_message_lines.py b/tests/textual/test_wire_message_lines.py index d1ec839c4..6851be392 100644 --- a/tests/textual/test_wire_message_lines.py +++ b/tests/textual/test_wire_message_lines.py @@ -60,7 +60,7 @@ def test_rows_fit_the_width_wrap_on_words_and_keep_drawn_names_clickable_at_ever def test_opening_notification_details_draws_more_rows_than_the_collapsed_message(): - status = CheckedNotifications(()) + status = CheckedNotifications.from_notifications(()) for style in (IrcMessageStyle(), MarkdownMessageStyle()): shown = fragment("#all", style) shown.request_show(MessageShown(style=style, status=status)) From 55b055b934f8d3c5fbfa92b86e1e509332c350d7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 03:19:13 -0400 Subject: [PATCH 7/8] Do not describe an unconfirmed queue action as definitely refused --- src/toad/submission_queue.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py index 9fda6d5d5..9f62ef8d5 100644 --- a/src/toad/submission_queue.py +++ b/src/toad/submission_queue.py @@ -141,12 +141,12 @@ def require_queue_action(self, owner: ConversationSubmissions, scope: QueueScope async def queue_send_now(self, owner: ConversationSubmissions, scope: QueueScope) -> None: agent = self.require_queue_action(owner, scope) if not await agent.controller.send_now(): - raise QueueActionRefused("The native owner did not accept send now") + raise QueueActionRefused("Send now was not confirmed; inspect the current queue before another action") async def queue_clear(self, owner: ConversationSubmissions, scope: QueueScope) -> None: agent = self.require_queue_action(owner, scope) if not await agent.controller.clear_queue(): - raise QueueActionRefused("The native owner did not accept clear queue") + raise QueueActionRefused("Clear queue was not confirmed; inspect the current queue before another action") class BusQueueSource(ReadOnlyQueueSource): From d92ac585eb8a27fd3772de2a9606a12b3344c683 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 29 Oct 2026 04:06:36 -0400 Subject: [PATCH 8/8] Read queue viewer projection from its shared display owner --- tests/textual/test_common_queue.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/textual/test_common_queue.py b/tests/textual/test_common_queue.py index 84ccfd223..b74461330 100644 --- a/tests/textual/test_common_queue.py +++ b/tests/textual/test_common_queue.py @@ -8,7 +8,8 @@ import pytest from agent_comms.bus import Bus from agent_comms.channel_targets import ThreadTarget -from agent_comms.presentation import DisplayOutcomes, MessageNotification +from agent_comms.read_basis import DisplayOutcomes +from agent_comms.presentation import MessageNotification from agent_comms.recorded_bus import RecordedBusIdentity from support.ui import Ui from support.workspace import Workspace