Keep queue actions on captured sources and original recipients

src/toad/comms_root.py

@@ -110,6 +110,9 @@ class Observation(ABC):
     def observed(self):
         return self
 
+    def queue_service(self) -> Comms:
+        raise RouteChangedError("The queue's original Comms source is unavailable")
+
     @abstractmethod
     async def read(self, request: AwaitedRead) -> object: ...
 
@@ -237,6 +240,11 @@ class Observing(Observation):
     sidebar: SidebarModel = field()
     process: ObservationProcess = field(compare=False)
 
+    def queue_service(self) -> Comms:
+        if not self.is_current():
+            raise RouteChangedError("The queue's original Comms source retired")
+        return self.service
+
     async def read(self, request: AwaitedRead) -> object:
         return await self.process.read(request)
 

src/toad/conversation_submission.py

@@ -286,7 +286,9 @@ def queue_evidence(self) -> QueueEvidence:
         return QueueEvidence(
             self.view.agent_binding.presentation, self.delivering,
             self.view.send_block_reason, self.view.contents.notification_targets,
-            lambda: self.view.app.coordination_access.observation.observed().service)
+            self.view.queue_source.capture_queue_service(
+                self.view.transcript.binding.thread_scope.queue_service,
+                self.view.app.coordination_access.observation.queue_service))
 
     @property
     def queue_summary(self) -> str:

src/toad/notification_status.py

@@ -6,6 +6,7 @@
 from dataclasses import dataclass, field
 from typing import ClassVar, TYPE_CHECKING
 
+from agent_comms.frozen_recipient import FrozenRecipient
 from agent_comms.message_reference import MessageReference
 from agent_comms.presentation import MessageNotification
 from toad.line_blocks import LineBlock, LineLink, MutedRole, NameRole, NoticeRole, OpenTarget, Span, SpanColumns, WorkingRole
@@ -55,8 +56,8 @@ def queue_details(self) -> DetailLines:
         return self.details
 
     @property
-    def queued_recipients(self) -> tuple[str, ...]:
-        """Recipients whose model has not yet taken this message; an unchecked status knows none."""
+    def queued_recipients(self) -> tuple[FrozenRecipient, ...]:
+        """Original recipients whose model has not yet taken this message."""
         return ()
 
     @property
@@ -114,8 +115,8 @@ def failed(self, failure: NotificationStatus) -> NotificationStatus:
         return self
 
     @property
-    def queued_recipients(self) -> tuple[str, ...]:
-        return tuple(item.recipient for item in self.notifications if not item.state.consumed)
+    def queued_recipients(self) -> tuple[FrozenRecipient, ...]:
+        return tuple(item.recipient_identity for item in self.notifications if not item.state.consumed)
 
     @property
     def working(self) -> bool:

src/toad/screens/queue.py

@@ -5,6 +5,7 @@
 
 from acp.exceptions import RequestError
 from agent_comms.acp_extension import QueueScope
+from agent_comms.errors import RelationViolationError, RouteChangedError, UnregisteredThreadError
 from textual import on
 from textual.app import ComposeResult
 from textual.containers import HorizontalGroup, Vertical, VerticalScroll
@@ -57,7 +58,8 @@ def compose(self) -> ComposeResult:
     async def apply_action(self, action: Callable[[QueueEvidence, QueueScope], Awaitable[None]]) -> None:
         try:
             await action(self.evidence, self.scope)
-        except (QueueActionRefused, RequestError) as error:
+        except (QueueActionRefused, RequestError, RelationViolationError,
+                RouteChangedError, UnregisteredThreadError) as error:
             self.query_one("#queue-error", Static).update(str(error))
             return
         self.submissions.publish_pending()

src/toad/session_presentation.py

@@ -2,13 +2,16 @@
 
 import asyncio
 from abc import ABC, abstractmethod
+from collections.abc import Callable
+from functools import partial
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING
 from weakref import ref
 
 from agent_comms.acp_extension import CoordinationChangedUpdate
+from agent_comms.comms import Comms
 from agent_comms.session_load import EnsuringSessionLoadPlan
 from agent_comms.ui_model.message_category import MessageCategory
 from textual.document._editing import EditingMode
@@ -258,6 +261,12 @@ async def close(self) -> None:
 class MailboxSessionSources(BusQueueSource, OperationalSessionSources):
     """CLI execution receives user messages through Core's inbox, not ACP."""
 
+    def capture_queue_service(
+        self, transcript: Callable[[Callable[[], Comms]], Comms],
+        service: Callable[[], Comms],
+    ) -> Callable[[], Comms]:
+        return partial(transcript, super().capture_queue_service(transcript, service))
+
     def activity_row(self, conversation: Conversation):
         return conversation.thread_activity.row
 

src/toad/session_scope.py

@@ -10,9 +10,10 @@
 from typing import TYPE_CHECKING, ClassVar, Never
 
 from agent_comms.channel_targets import ThreadTarget
+from agent_comms.comms import Comms
 from agent_comms.repo import Checkout
 from agent_comms.coordination_errors import CoordinationReadUnavailable, StaleRevision
-from agent_comms.errors import RelationViolationError
+from agent_comms.errors import RelationViolationError, RouteChangedError
 from agent_comms.thread_identity import ThreadIncarnation
 from agent_comms.order_tools import TranscriptPageRead
 from agent_comms.transcripts import AtFrontier, LatestPage, LivePageSource, PageLimit, PagePosition, PageSource
@@ -96,6 +97,9 @@ def update_request(
         position: PagePosition = LatestPage(),
     ) -> ReadTranscriptUpdate: ...
 
+    def queue_service(self, service: Callable[[], Comms]) -> Comms:
+        raise RouteChangedError("The queue has no original transcript source")
+
     @abstractmethod
     def mailbox_target(self) -> ThreadTarget: ...
 
@@ -120,6 +124,12 @@ class KnownTranscript(TranscriptScope):
     name: str
     readable = True
 
+    def queue_service(self, service: Callable[[], Comms]) -> Comms:
+        comms = service()
+        if comms.root != Path(self.root):
+            raise RouteChangedError("The queue's original transcript root changed")
+        return comms
+
     def owner_decision(self, presentation: ThreadPresentation) -> InputDecision:
         return ExistingOwnerPlan.from_binding(presentation.binding).decision
 

src/toad/submission_queue.py

@@ -7,11 +7,9 @@
 from dataclasses import dataclass
 from functools import cached_property
 from typing import ClassVar
-
-from acp.exceptions import RequestError
 from agent_comms.acp_extension import NoQueueScope, QueueItem, QueueScope
 from agent_comms.comms import Comms
-from agent_comms.errors import RelationViolationError
+from agent_comms.frozen_recipient import FrozenRecipient
 from agent_comms.message_reference import MessageReference
 from agent_comms.notification_assignment import NotificationSelection
 from agent_comms.runtime_requests import ClearQueueRuntimeRequest, QueueActionRuntimeRequest, SendNowRuntimeRequest
@@ -22,7 +20,6 @@
 from toad.line_blocks import MutedRole, Span
 from toad.notification_status import DetailLines, MessageQueueEntry
 
-
 @dataclass(frozen=True)
 class QueueEvidence:
     """One UI read; only a BUS source asks for its visible message entries."""
@@ -44,16 +41,23 @@ def entries(self) -> tuple[MessageQueueEntry, ...]:
         return tuple(entries.values())
 
     @cached_property
-    def queued_recipients(self) -> tuple[str, ...]:
-        """Each recipient still holding one of the visible messages, once."""
-        return tuple(dict.fromkeys(name for entry in self.entries for name in entry.status.queued_recipients))
+    def queued_recipients(self) -> tuple[FrozenRecipient, ...]:
+        """Each original recipient still holding a visible message, once."""
+        return tuple(dict.fromkeys(recipient for entry in self.entries
+                                   for recipient in entry.status.queued_recipients))
 
 
 class QueueSource(ABC):
     """The source selects its capability; the actual presentation owns native facts."""
 
     preview_limit: ClassVar[int] = 3
 
+    def capture_queue_service(
+        self, transcript: Callable[[Callable[[], Comms]], Comms],
+        service: Callable[[], Comms],
+    ) -> Callable[[], Comms]:
+        return service
+
     @abstractmethod
     def queue_scope(self, evidence: QueueEvidence) -> QueueScope: ...
 
@@ -140,23 +144,23 @@ def queue_actions_available(self, evidence: QueueEvidence) -> bool:
         return bool(evidence.queued_recipients)
 
     async def queue_send_now(self, evidence: QueueEvidence, scope: QueueScope) -> None:
-        await self.at_owners(evidence, SendNowRuntimeRequest, "Send now")
+        await self.at_owners(evidence, SendNowRuntimeRequest)
 
     async def queue_clear(self, evidence: QueueEvidence, scope: QueueScope) -> None:
-        await self.at_owners(evidence, ClearQueueRuntimeRequest, "Clear")
+        await self.at_owners(evidence, ClearQueueRuntimeRequest)
+
+    @staticmethod
+    def at_original_owner(comms: Comms, recipient: FrozenRecipient,
+                          action: type[QueueActionRuntimeRequest]) -> None:
+        original = comms.registry.snapshot().born(recipient.lookup)
+        action.at_owner(comms.root, original)
 
     @staticmethod
-    async def at_owners(evidence: QueueEvidence, action: type[QueueActionRuntimeRequest], title: str) -> None:
-        """Ask every recipient's owner; name each one that refused."""
-        comms = evidence.service()
-        refused = []
-        for name in evidence.queued_recipients:
-            try:
-                await asyncio.to_thread(action.at_owner, comms.root, comms.registry.require(name))
-            except (RequestError, OSError, RelationViolationError) as error:
-                refused.append(f"{name}: {error}")
-        if refused:
-            raise QueueActionRefused(f"{title} refused by " + "; ".join(refused))
+    async def at_owners(evidence: QueueEvidence, action: type[QueueActionRuntimeRequest]) -> None:
+        """Act on the original recipients; stop and propagate a failed action."""
+        for recipient in evidence.queued_recipients:
+            comms = evidence.service()
+            await asyncio.to_thread(BusQueueSource.at_original_owner, comms, recipient, action)
 
     def queue_summary(self, evidence: QueueEvidence) -> str:
         entries = evidence.entries
@@ -172,5 +176,6 @@ def queue_details(self, evidence: QueueEvidence) -> DetailLines:
             (Span(evidence.transport, MutedRole),),
             (Span(f"Showing up to {self.preview_limit} visible messages and their leading observed recipient.", MutedRole),),
             *(line for entry in evidence.entries[:self.preview_limit] for line in entry.details),
-            (Span("Queued at: " + (", ".join(recipients) if recipients else "no recipient"), MutedRole),),
+            (Span("Queued at: " + (", ".join(recipient.canonical_thread for recipient in recipients)
+                                  if recipients else "no recipient"), MutedRole),),
         )