diff --git a/src/toad/comms_root.py b/src/toad/comms_root.py index 1ef8747e6..c3096de65 100644 --- a/src/toad/comms_root.py +++ b/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) diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py index 7e13dc2bc..0c2c346b9 100644 --- a/src/toad/conversation_submission.py +++ b/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: diff --git a/src/toad/notification_status.py b/src/toad/notification_status.py index 4b9c1961c..aab15fd0a 100644 --- a/src/toad/notification_status.py +++ b/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: diff --git a/src/toad/screens/queue.py b/src/toad/screens/queue.py index 9770c0021..d47c02b3b 100644 --- a/src/toad/screens/queue.py +++ b/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() diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py index 8becf6a2c..d39c9caae 100644 --- a/src/toad/session_presentation.py +++ b/src/toad/session_presentation.py @@ -2,6 +2,8 @@ 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 @@ -9,6 +11,7 @@ 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 diff --git a/src/toad/session_scope.py b/src/toad/session_scope.py index 02a40a697..373270d00 100644 --- a/src/toad/session_scope.py +++ b/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 diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py index 9f276c62e..b56ad5c7a 100644 --- a/src/toad/submission_queue.py +++ b/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,9 +41,10 @@ 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): @@ -54,6 +52,12 @@ class QueueSource(ABC): 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),), )