P904: share source-owned queue UI across native, channel and mailbox

src/toad/acp/agent_controller.py

@@ -1,3 +1,3 @@
 """ACP operations remain owned by the controller after their view retires."""
 
 from __future__ import annotations
@@ -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 @@
         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 @@
         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)

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)
 

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()
 

src/toad/agent_presentation.py

@@ -1,11 +1,19 @@
 """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 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
+from toad.submission_queue import QueueActionRefused
+
+
+if TYPE_CHECKING:
+    from toad.acp.agent import Agent
 
 
 @dataclass(frozen=True)
@@ -38,6 +46,21 @@ def queue_supported(self) -> bool:
     def queue(self):
         """The actual source owns its input queue projection."""
 
+    @property
+    def queue_scope(self) -> QueueScope:
+        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: QueueScope) -> Agent:
+        raise QueueActionRefused("No live ACP queue authority is available")
+
     @property
     def prompt_in_flight(self):
         return 0
@@ -87,6 +110,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) -> QueueScope:
+        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: QueueScope) -> Agent:
+        if not self.queue_supported or not self.agent.queue_attachment.accepts_request(scope):
+            raise QueueActionRefused("Queue attachment changed or is unavailable; inspect the current queue again")
+        return self.agent
+
     @property
     def uses_managed_turns(self):
         return self.turns.managed

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():

src/toad/conversation_submission.py

@@ -1,3 +1,3 @@
 """One submission transaction owns input intent, request identity and draft recovery."""
 
 from __future__ import annotations
@@ -21,12 +21,16 @@
 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
+from agent_comms.notification_evidence import NotificationWindow
 
 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
 
 
@@ -292,6 +296,33 @@
     def accepts_failure(self, message: CoreEventMessage):
         return self.view.agent_binding.accepts_failed_input(message)
 
+    @property
+    def queue_source(self) -> QueueSource:
+        return self.view.queue_source
+
+    @property
+    def queue_summary(self) -> str:
+        return self.queue_source.queue_summary(self)
+
+    @property
+    def message_queue_entries(self) -> tuple[MessageQueueEntry, ...]:
+        """Borrow the existing visible fragment proofs; never retain another inbox."""
+        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):
+                    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 +366,19 @@
         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: 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 await agent.controller.send_now():
-                return
-        except RequestError as error:
-            if self.view.agent_binding is agent and agent.session_id == session:
+            await source.queue_send_now(self, scope)
+        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")
-        if self.view.agent_binding is agent and agent.queue_attachment.accepts_request(scope):
-            self.publish_pending()
+        self.publish_pending()

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()

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
@@ -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)
@@ -71,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:
@@ -81,6 +94,24 @@ def working(self) -> bool:
     def title(self) -> str:
         return MessageNotification.summary(self.notifications)
 
+    @property
+    def queue_title(self) -> str:
+        # 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.from_notifications(self.notifications[:1])
+        return (
+            (Span(self.title, 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 = []
@@ -91,8 +122,15 @@ 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)
+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)

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:

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)