Combined carrier: Toad #1319 paired with agent-comms #3566

src/toad/acp/agent.py

@@ -15,7 +15,8 @@
 from agent_comms.goal_presentation import GoalSnapshot, IdleGoalExecution
 from agent_comms.goals import Goal, GoalDeclaration, NoGoal
 from agent_comms.goal_history import RecordedGoalHistory
-from agent_comms.ui_model.observation import ReadOwnerGoal
+from agent_comms.order_tools import OwnerGoalRead
+from agent_comms.ui_model.observation import ReadOwnerOperation
 
 from toad import paths
 from toad.acp.agent_configuration import (
@@ -238,7 +239,7 @@ async def get_goal_snapshot(self) -> GoalSnapshot:
         authority = ClientSessionRequest(self, self.session_id)
         consumer = OwnerSnapshotConsumer(self, self.session_id, turn_token=self.presentation.turns.sequence)
         snapshot = await self.controller.transcripts.read(
-            ReadOwnerGoal(coordination.wire_root, coordination.thread))
+            ReadOwnerOperation(coordination.wire_root, coordination.thread, operation=OwnerGoalRead()))
         self.controller.require_owner(coordination, authority)
         for fact in snapshot.meta.batch.updates:
             consumer.dispatch_sync(fact)

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
@@ -11,6 +11,7 @@
 from acp.exceptions import RequestError
 from agent_comms.acp_extension import (
     ClearQueueRequest,
+    CommsRequest,
     CompactRequest,
     CoordinationChangedUpdate,
     InputFailedUpdate,
@@ -23,16 +24,19 @@
 from agent_comms.coordination_errors import Refused
 from agent_comms.runtime import RuntimeConnection, socket_path
 from agent_comms.transcripts import TranscriptReadIdentity
-from agent_comms.ui_model.delivery import DeliveryView
-from agent_comms.ui_model.observation import ReadDeliveryOverview
+from agent_comms.order_tools import InputDeliveryRead
+from agent_comms.ui_model.delivery import DeliveryView, ReadDeliveryView
+from agent_comms.ui_model.observation import ReadOwnerOperation
 
-from toad.acp.client_session import ClientSessionRequest
+from toad.acp.client_session import ClientSessionRequest, QueueSessionRequest
 from toad.acp.status import StopReason
 from toad.core import events as core_events
 from toad.core.events import LogAgentFail
 from toad.plan import PlanItem
 from toad.surface_binding import DetachedSurfaceBinding, SurfaceBinding
 
+from toad.acp.queue_attachment import QueueActionRefused
+
 from .prompt import build as build_prompt
 from .session_output import SessionOutput
 from .terminal_owner import OperationalTerminalOwner
@@ -270,9 +274,9 @@
             if submission is not None:
                 self._deferred_submissions.discard(submission)
 
-    async def submit_blocks(self, content: list[dict[str, str]], command: ClearQueueRequest | None = None):
+    async def submit_blocks(self, content: list[dict[str, str]], command: CommsRequest):
         agent = self.agent
-        authority = ClientSessionRequest(agent, agent.session_id)
+        authority = QueueSessionRequest(agent, agent.session_id)
         return await self.operate(
             self._prompt_operation(
                 partial(self._prompt, content, command, authority, agent.queue_attachment.scope), authority
@@ -285,11 +289,11 @@
         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(
-                agent, content, encode_request(command) if command is not None else {}, require
+                agent, content, encode_request(command), require
             )
         except RequestError as error:
             # The agent answered: this request failed and its session stays open.
@@ -301,15 +305,15 @@
             report = agent.session.record_failure
         else:
             if authority.retired:
-                return None
+                return authority.unconfirmed("the captured ACP session retired before its reply")
             return StopReason.member_named(result.stop_reason)
         if authority.retired:
-            return None
+            return authority.unconfirmed("the captured ACP session retired during the request")
         failure = request_failure.failure
         self._prompt_failed(
             command, authority, queue_scope, failure, report, published=request_failure.notification_published
         )
-        return None
+        return authority.unconfirmed(failure.feedback)
 
     def _prompt_failed(
         self,
@@ -322,7 +326,7 @@
         published: bool = False,
     ) -> None:
         agent = self.agent
-        user_text = command.draft_text if command is not None else None
+        user_text = command.draft_text
         if user_text:
             agent.events.publish(
                 core_events.CommsUpdated(
@@ -338,19 +342,16 @@
     async def clear_queue(self) -> None:
         await self.submit_blocks([{"type": "text", "text": " "}], ClearQueueRequest())
 
-    async def send_now(self) -> bool:
-        authority = ClientSessionRequest(self.agent, self.agent.session_id)
+    async def send_now(self) -> None:
+        authority = QueueSessionRequest(self.agent, self.agent.session_id)
         queue_scope = self.agent.queue_attachment.scope
         if pending := tuple(self._deferred_submissions):
             await asyncio.gather(*(asyncio.shield(task) for task in pending))
-        return (
-            await self.operate(
-                self._prompt_operation(
-                    partial(self._prompt, [{"type": "text", "text": " "}], SendNowRequest(), authority, queue_scope),
-                    authority,
-                )
+        await self.operate(
+            self._prompt_operation(
+                partial(self._prompt, [{"type": "text", "text": " "}], SendNowRequest(), authority, queue_scope),
+                authority,
             )
-            is not None
         )
 
     async def compact_context(self, instructions=None):
@@ -427,11 +428,11 @@
         if coordination is None:
             return DeliveryView()
         authority = ClientSessionRequest(self.agent, self.agent.session_id)
-        delivery = await self.transcripts.read(
-            ReadDeliveryOverview(coordination.wire_root, coordination.thread, include_history=include_history)
-        )
+        overview = await self.transcripts.read(ReadOwnerOperation(
+            coordination.wire_root, coordination.thread, InputDeliveryRead(include_history=include_history),
+        ))
         self.require_owner(coordination, authority)
-        return delivery
+        return ReadDeliveryView(overview)
 
     async def dismiss_historical_inputs(self):
         return await self.request_owner("dismiss_historical_inputs")

src/toad/acp/client_session.py

@@ -1,6 +1,7 @@
 """One captured session authority for ACP client-side effects."""
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, NoReturn
+from .queue_attachment import QueueActionRefused
 
 if TYPE_CHECKING:
     from .agent import Agent
@@ -34,6 +35,16 @@
         if not self.current:
             raise RequestError.invalid_params({'details': 'ACP client request belongs to a retired session'})
 
+    def unconfirmed(self, reason: str) -> None:
+        """Ordinary input failure is presented by its existing failure publication."""
+
+
+class QueueSessionRequest(ClientSessionRequest):
+    """A queue mutation requires confirmation; an absent reply is never success."""
+
+    def unconfirmed(self, reason: str) -> NoReturn:
+        raise QueueActionRefused(f"Queue action not confirmed: {reason}. Inspect the queue before another action.")
+
 
 class ClientRequestOwner:
     def __init__(self, agent):

src/toad/acp/queue_attachment.py

@@ -14,6 +14,10 @@
 from .projection_attachment import ProjectionAttachment
 
 
+class QueueActionRefused(Exception):
+    """The captured queue cannot confirm or authorize this action; never replay it."""
+
+
 class QueueAttachment(ProjectionAttachment):
     def __init__(self):
         super().__init__()

src/toad/acp/transcript_reader.py

@@ -1,3 +1,3 @@
 """A view's reads, answered by Core's observation service.
 
 The UI process does not read Core's transcript stores or ask the live owner
@@ -16,6 +16,7 @@
 from functools import partial
 
 from agent_comms.acp_extension import TranscriptDraftUpdate, TranscriptSnapshotUpdate
+from agent_comms.transcripts import AtFrontier, LatestPage, LivePageSource, PageLimit, PagePosition, PageSource
 from agent_comms.thread_presentation import ThreadPresentation
 from agent_comms.ui_model.observation import (
     AwaitedRead,
@@ -68,35 +69,31 @@
         self,
         scope: TranscriptScope,
         *,
-        before=None,
-        after=None,
-        through=None,
-        historical_source: str | None = None,
+        position: PagePosition = LatestPage(),
+        limit: PageLimit = AtFrontier(),
+        page_source: PageSource = LivePageSource(),
     ):
         snapshot = await self.snapshot(
             scope,
-            before=before,
-            after=after,
-            through=through,
-            historical_source=historical_source,
+            position=position,
+            limit=limit,
+            page_source=page_source,
         )
         return snapshot.page
 
     async def snapshot(
         self,
         scope: TranscriptScope,
         *,
-        before=None,
-        after=None,
-        through=None,
-        historical_source: str | None = None,
+        position: PagePosition = LatestPage(),
+        limit: PageLimit = AtFrontier(),
+        page_source: PageSource = LivePageSource(),
     ) -> TranscriptSnapshotUpdate:
         return await self.read(
             scope.snapshot_request(
-                before=before,
-                after=after,
-                through=through,
-                historical_source=historical_source,
+                position=position,
+                limit=limit,
+                page_source=page_source,
             )
         )
 
@@ -105,6 +102,6 @@
         scope: TranscriptScope,
         previous: TranscriptPublicationState,
         *,
-        after=None,
+        position: PagePosition = LatestPage(),
     ) -> TranscriptSnapshotUpdate | TranscriptDraftUpdate:
-        return await self.read(scope.update_request(previous, after=after))
+        return await self.read(scope.update_request(previous, position=position))

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

@@ -178,9 +178,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,15 @@
 """Agent-owned optional capabilities and derived attachment presentation."""
+from __future__ import annotations
+
 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.acp.queue_attachment import QueueActionRefused
+from toad.slash_command import AgentAdvertisedCommand
 
 
 @dataclass(frozen=True)
@@ -38,6 +42,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) -> None:
+        raise QueueActionRefused("No live ACP queue authority is available")
+
     @property
     def prompt_in_flight(self):
         return 0
@@ -74,7 +93,6 @@ def auth_methods(self):
 
     @property
     def commands(self):
-        from toad.slash_command import AgentAdvertisedCommand
         return tuple(AgentAdvertisedCommand.from_acp(record)
                      for record in self.agent.controller.commands)
 
@@ -87,6 +105,28 @@ 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) -> None:
+        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")
+
     @property
     def uses_managed_turns(self):
         return self.turns.managed

src/toad/conversation_kind.py

@@ -31,6 +31,9 @@
 from agent_comms.thread_identity import ThreadIncarnation
 from textual.app import ComposeResult
 
+from toad.conversation_submission import ConversationSubmissions
+from toad.widgets.channel_prompt import ChannelPrompt
+from toad.widgets.prompt import MessagePrompt
 from toad.status_thread import NoStatusThread, StatusThread
 from toad.widgets.channel_activity import ChannelActivity
 from toad.widgets.session_details import ConversationSessionDetails
@@ -133,7 +136,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: ConversationSubmissions): ...
 
     @classmethod
     def seen_page(cls, history: WireHistory, seen):
@@ -206,18 +209,14 @@ 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):
-        from toad.widgets.channel_prompt import ChannelPrompt
-
-        return ChannelPrompt(project=project, placeholder=self.initial_placeholder)
+    def prompt(self, project: Checkout, *, submissions: ConversationSubmissions):
+        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
 
     @classmethod
     async def update_mentions(cls, view: CommsChatView, comms: Comms):
-        from toad.widgets.channel_prompt import ChannelPrompt
-
         history = view.message_history
         source = history.source_snapshot()
         conversation = view.conversation
@@ -293,10 +292,8 @@ def compose_activity(cls, view) -> ComposeResult:
         view.thread_activity = ConversationSessionDetails(view.conversation.peer)
         yield view.thread_activity
 
-    def prompt(self, project: Checkout):
-        from toad.widgets.prompt import MessagePrompt
-
-        return MessagePrompt(project=project)
+    def prompt(self, project: Checkout, *, submissions: ConversationSubmissions):
+        return MessagePrompt(project=project, submissions=submissions)
 
     async def status_text(self, comms: Comms, *, read_only: bool) -> str:
         def read():