From b7088a3383996261bf33e5497648f86e86e70e66 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 31 Oct 2026 19:07:25 -0400 Subject: [PATCH 01/24] Use one submission lifetime for native, mailbox and channel input --- src/toad/acp/agent.py | 69 +++- src/toad/agent.py | 3 - src/toad/agent_binding.py | 9 +- src/toad/conversation_submission.py | 299 ++++++++++++------ src/toad/delivery_failure_view.py | 133 +------- src/toad/input_decision.py | 4 + src/toad/queue_capability.py | 51 +++ src/toad/session_presentation.py | 23 +- src/toad/session_scope.py | 54 ++-- src/toad/submission_queue.py | 123 +++---- src/toad/transcript_preparation.py | 4 - src/toad/transcript_publication.py | 2 +- src/toad/widgets/comms_chat.py | 18 +- src/toad/widgets/conversation.py | 2 +- src/toad/wire_history.py | 25 +- tests/textual/test_input_failure_draft.py | 25 -- .../test_painted_view_marks_itself_visible.py | 29 -- tests/textual/test_stale_route_writes.py | 7 +- 18 files changed, 467 insertions(+), 413 deletions(-) create mode 100644 src/toad/queue_capability.py delete mode 100644 tests/textual/test_painted_view_marks_itself_visible.py diff --git a/src/toad/acp/agent.py b/src/toad/acp/agent.py index bab886c5d..bd1038c67 100644 --- a/src/toad/acp/agent.py +++ b/src/toad/acp/agent.py @@ -1,9 +1,10 @@ import asyncio -from dataclasses import replace +from dataclasses import dataclass, replace from datetime import datetime from pathlib import Path import rich.repr +from acp.exceptions import RequestError from agent_comms.acp_extension import ( CoordinationChangedUpdate, InputStartedUpdate, @@ -47,6 +48,9 @@ from toad.agent_presentation import ACPAgentPresentation from toad.agent_schema import AgentDefinition from toad.core import events as core_events +from toad.conversation_submission import ( + PromptInputSubmission, ConversationSubmissions, NoSend, SendPending, SendState, SubmissionExecution, +) from toad.platform import host @@ -157,6 +161,9 @@ async def stop(self) -> None: await self.process.stop(self) + def begin_submission(self, owner: ConversationSubmissions, submission: PromptInputSubmission): + return owner.start(NativeSubmissionExecution(owner, submission, self)) + async def send_prompt(self, prompt: str, *, request: PromptRequest | None = None): """Submit through AgentBase's contract; the controller runs the request.""" return await self.controller.submit(prompt, request=request) @@ -306,3 +313,63 @@ async def cancel(self) -> bool: async def set_session_name(self, name: str) -> None: await self.session.set_name(name) + + +@dataclass(frozen=True) +class NativeInputPresented(SendState): + started: InputStartedUpdate + + def response_finished(self, execution) -> None: + execution.enter(NoSend()) + + +class NativeSubmissionExecution(SubmissionExecution): + """ACP owns its captured request and native acknowledgement, not the shared transaction.""" + + def __init__(self, owner: ConversationSubmissions, submission: PromptInputSubmission, agent: Agent): + super().__init__(owner, submission) + self.agent = agent + self.authority = ClientSessionRequest(agent, agent.session_id) + self.scope = agent.queue_attachment.scope + self.request = submission.request + + @property + def current(self): + view = self.owner.view + return (view.agent_binding is self.agent and not self.authority.retired + and self.agent.queue_attachment.accepts_request(self.scope)) + + async def feedback(self): + await self.submission.feedback(self) + + def accepted(self, accepted) -> bool: + return self.request.input_id in accepted or not self.agent.presentation.uses_managed_turns + + def native_input_presented(self, started: InputStartedUpdate) -> None: + if self.current and started.started_request(self.request.input_id): + self.enter(NativeInputPresented(started)) + + async def deliver(self): + view = self.owner.view + local = not self.agent.presentation.uses_managed_turns + reason = None + if local: + view.turns.start_client() + self.enter(SendPending()) + try: + if self.agent.presentation.queue_supported: + reason = await self.agent.send_prompt(self.submission.text, request=self.request) + else: + reason = await self.agent.send_prompt(self.submission.text) + except RequestError as error: + self.enter(NoSend()) + if self.current: + await view.submission_request_failed(error, self.submission.text) + else: + self.enter(NoSend()) + if self.current and local: + view.call_later(self.complete, reason) + + async def complete(self, reason): + if self.current: + await self.owner.view.agent_turn_over(reason) diff --git a/src/toad/agent.py b/src/toad/agent.py index 5cb6de45f..aaa8fb09b 100644 --- a/src/toad/agent.py +++ b/src/toad/agent.py @@ -66,9 +66,6 @@ async def change_project(self, view: Conversation, path: Path) -> None: def request_compaction(self, view: Conversation, instructions) -> None: view.compact_context(instructions) - def begin_submission(self, owner, submission): - return owner.start(submission, self) - def accepts_failed_input(self, message): if message.publisher is not self or message.event.session_id != self.session_id: return False diff --git a/src/toad/agent_binding.py b/src/toad/agent_binding.py index 6497fabf1..275b18b48 100644 --- a/src/toad/agent_binding.py +++ b/src/toad/agent_binding.py @@ -25,7 +25,7 @@ from agent_comms.session_load import ExistingSessionLoadPlan from toad.conversation_submission import ( - AgentInputSubmission, + PromptInputSubmission, ConversationSubmissions, SubmissionStart, ) @@ -91,8 +91,8 @@ def connected(self, view: ConversationSessionBinding) -> None: ... @abstractmethod def accepts_message(self, message: CoreEventMessage) -> bool: ... - @abstractmethod - def begin_submission(self, owner: ConversationSubmissions, submission: AgentInputSubmission) -> SubmissionStart: ... + def begin_submission(self, owner: ConversationSubmissions, submission: PromptInputSubmission) -> SubmissionStart: + return owner.unbound(submission, self.readiness_refusal) @abstractmethod def accepts_failed_input(self, message: CoreEventMessage) -> bool: ... @@ -186,9 +186,6 @@ async def compact_context(self, instructions) -> None: async def cancel_turn(self, view: Conversation) -> None: """No native turn exists to cancel.""" - def begin_submission(self, owner: ConversationSubmissions, submission: AgentInputSubmission): - return owner.unbound(submission, self.readiness_refusal) - def accepts_failed_input(self, message: CoreEventMessage) -> bool: return False diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py index 3e442566a..09cfb6519 100644 --- a/src/toad/conversation_submission.py +++ b/src/toad/conversation_submission.py @@ -4,26 +4,27 @@ import asyncio from abc import ABC, abstractmethod +from collections.abc import Awaitable, Callable from dataclasses import dataclass from functools import partial -from typing import ClassVar +from pathlib import Path from acp.exceptions import RequestError from agent_comms.acp_extension import ( InputStartedUpdate, - PendingQueueProjection, QueuePromptRequest, QueueScope, SteerPromptRequest, ) from agent_comms.nominal.declared_family import DeclaredFamily +from agent_comms.delivery_failure import delivery_failure +from agent_comms.errors import CommittedMessageError, HumanSendError, HumanSendRefusedError, RelationViolationError, RouteChangedError +from agent_comms.messages import Message from toad import messages -from toad.acp.client_session import ClientSessionRequest from toad.command_catalog import CommandCatalog -from toad.acp.agent import Agent from toad.acp.queue_attachment import QueueActionRefused -from toad.submission_queue import QueueEvidence, QueueSource +from toad.queue_capability import QueueActions, QueueEvidence from toad.core.input_events import UserInputSubmitted from toad.core_event_carrier import CoreEventMessage from toad.input_decision import InputDecision, InputRefused @@ -32,7 +33,6 @@ @dataclass(frozen=True) class InputSubmission(DeclaredFamily, affix="InputSubmission"): source_text: str - priority: ClassVar[int] = 0 @property def text(self): @@ -49,41 +49,35 @@ async def execute(self, owner: ConversationSubmissions): ... def from_event(cls, event: UserInputSubmitted, owner: ConversationSubmissions): kind = next( kind - for kind in sorted(cls.members_with(InputSubmission), key=lambda kind: kind.priority, reverse=True) + for kind in sorted(cls.members_with(InputSubmission), key=lambda kind: len(kind.__mro__), reverse=True) if kind.matches(event, owner) ) return kind(event.body) class NoInputSubmission(InputSubmission): - priority = -1 - @classmethod def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): - return True + return not event.body.strip() async def execute(self, owner: ConversationSubmissions): pass -class QueueNowInputSubmission(InputSubmission): - priority = 100 - +class QueueNowInputSubmission(NoInputSubmission): @classmethod def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): if event.shell or event.body.strip(): return False - return event.immediate and owner.view.agent_binding.presentation.queue_supported + return event.immediate and owner.view.queue_source.queue_actions_available(owner.view.agent_binding.presentation) async def execute(self, owner: ConversationSubmissions): projection = owner.queue_projection - if projection.status == "available" and projection.items: + if projection.items: owner.send_now() class ShellInputSubmission(InputSubmission): - priority = 90 - @classmethod def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): return event.shell and bool(event.body.strip()) @@ -109,7 +103,7 @@ async def prepare(self): ... class UnboundSubmissionStart(SubmissionStart): def __init__( - self, owner: ConversationSubmissions, submission: AgentInputSubmission, refusal: InputRefused + self, owner: ConversationSubmissions, submission: PromptInputSubmission, refusal: InputRefused ): self.owner, self.submission, self.refusal = owner, submission, refusal @@ -120,7 +114,7 @@ async def prepare(self): ) -class AgentInputSubmission(InputSubmission): +class PromptInputSubmission(InputSubmission): @property @abstractmethod def request(self): ... @@ -143,7 +137,7 @@ async def execute(self, owner: ConversationSubmissions): return await owner.begin(self).prepare() -class OrdinaryInputSubmission(AgentInputSubmission): +class OrdinaryInputSubmission(PromptInputSubmission): @classmethod def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): if event.shell: @@ -156,8 +150,6 @@ def request(self): class DeferredInputSubmission(OrdinaryInputSubmission): - priority = 70 - @classmethod def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): if not super().matches(event, owner): @@ -165,7 +157,7 @@ def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): if event.immediate: return False view = owner.view - return view.turns.owner.busy and view.agent_binding.presentation.queue_supported + return view.queue_source.defers_input(view) @property def request(self): @@ -177,77 +169,198 @@ async def feedback(self, execution: SubmissionExecution): class ImmediateInputSubmission(OrdinaryInputSubmission): - priority = 80 - @classmethod def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): if not super().matches(event, owner): return False - return event.immediate and owner.view.agent_binding.presentation.queue_supported + return event.immediate and owner.view.queue_source.queue_actions_available(owner.view.agent_binding.presentation) @property def request(self): return SteerPromptRequest(self.text) +class SendState: + """One transaction's local outcome; widgets only draw this answer.""" + + reason = "" + + def settle(self, execution) -> None: + execution.owner.retire(execution) + + def drawn_in(self, view) -> bool: + return False + + def response_finished(self, execution) -> None: + """A retained outcome already names the transaction's answer.""" + + def local_intent(self, execution, accepted) -> tuple[str, ...]: + return () + + +class NoSend(SendState): + pass + + +class RetainedSend(SendState): + def settle(self, execution) -> None: + """A pending, undrawn or UNKNOWN result keeps its original transaction.""" + + +class CheckingSend(RetainedSend): + reason = "Checking source; do not retry" + + def response_finished(self, execution) -> None: + execution.enter(NoSend()) + + def local_intent(self, execution, accepted) -> tuple[str, ...]: + return () if execution.accepted(accepted) else (execution.submission.text,) + + +class SendPending(CheckingSend): + reason = "Send pending; do not retry" + + def response_finished(self, execution) -> None: + execution.delivery_blocked( + "Request ended without a known outcome; inspect the source, do not retry", + execution.submission.source_text, + ) + + +@dataclass(frozen=True) +class SentUndrawn(RetainedSend): + root: Path + sent: Message + + @property + def reason(self) -> str: + return f"Sent {self.root}/{self.sent.seq}/{self.sent.message_id}; waiting to be drawn, do not retry" + + def drawn_in(self, view) -> bool: + return view.queue_source.draws_message(view, self.root, self.sent) + + +@dataclass(frozen=True) +class SendBlocked(RetainedSend): + description: str + + @property + def reason(self) -> str: + return f"Send blocked: {self.description}" + + class SubmissionExecution(SubmissionStart): - """Immutable source binding and the real outstanding local request.""" + """Shared history, worker and outcome lifetime over an immutable source binding.""" + + def __init__(self, owner: ConversationSubmissions, submission: PromptInputSubmission): + self.owner, self.submission = owner, submission + self.state: SendState = CheckingSend() - def __init__(self, owner: ConversationSubmissions, submission: AgentInputSubmission, agent: Agent): - self.owner, self.submission, self.agent = owner, submission, agent - self.authority = ClientSessionRequest(agent, agent.session_id) - self.scope = agent.queue_attachment.scope - self.request = submission.request - self.started_update: InputStartedUpdate | None = None + @property + @abstractmethod + def current(self) -> bool: ... + + @property + def retired(self): + return not self.current + + def enter(self, state: SendState) -> None: + self.state = state + self.owner.view.refresh_submission_state() + + def accepted(self, accepted) -> bool: + return False + + def native_input_presented(self, started: InputStartedUpdate) -> None: + """A non-native source has no native start to observe.""" + + async def feedback(self): + """Only a source's own acceptance can confirm its input.""" async def prepare(self): - owner, submission = self.owner, self.submission - view = owner.view - owner.publish_pending() + view, submission = self.owner.view, self.submission + self.owner.publish_pending() view.run_worker(partial(view.input_histories.prompt.record, submission.source_text), group="history") try: - await submission.feedback(self) + await self.feedback() if self.current: return view.run_worker(partial(self.send), group="conversation-submission") - owner.finish(self) + self.delivery_refused("Message source changed; reopen this view", submission.source_text) except BaseException: - owner.finish(self) + self.state.response_finished(self) + self.owner.finish(self) raise + self.owner.finish(self) - @property - def current(self): - view = self.owner.view - if view.agent_binding is not self.agent or self.authority.retired: - return False - return self.agent.queue_attachment.accepts_request(self.scope) - - @property - def retired(self): - return not self.current + @abstractmethod + async def deliver(self): ... async def send(self): - view = self.owner.view - managed = self.agent.presentation.uses_managed_turns - local = not managed - reason = None - if local: - view.turns.start_client() try: - if self.agent.presentation.queue_supported: - reason = await self.agent.send_prompt(self.submission.text, request=self.request) - else: - reason = await self.agent.send_prompt(self.submission.text) - except RequestError as error: if self.current: - await view.submission_request_failed(error, self.submission.text) + await self.deliver() + else: + self.delivery_refused("Message source changed; reopen this view", self.submission.source_text) finally: + self.state.response_finished(self) self.owner.finish(self) - if self.current and local: - view.call_later(self.complete, reason) - async def complete(self, reason): + def delivery_blocked(self, description: str, body: str) -> None: + self.enter(SendBlocked(description)) + self.owner.view.prompt.restore_draft(body) + self.owner.view.feedback.flash(self.state.reason, style="error") + + def delivery_refused(self, description: str, body: str) -> None: + self.enter(NoSend()) + self.owner.view.delivery_refused(description, body) + + +class MessageSubmissionExecution(SubmissionExecution): + """An original Core publication, including a worker whose outcome is UNKNOWN.""" + + def __init__(self, owner: ConversationSubmissions, submission: PromptInputSubmission, + root: Path, conversation, current: Callable[[], bool], + present: Callable[[Message], Awaitable[None]]): + super().__init__(owner, submission) + self.root, self.conversation = root, conversation + self.source_current, self.present = current, present + + @property + def current(self) -> bool: + return self.source_current() + + async def deliver(self): + view, body = self.owner.view, self.submission.source_text + try: + comms = await view.app.coordination_access.transcripts.service(self.root) + read_only = await view.app.preparation.run_thread(self.conversation.read_only, comms) + if read_only: + raise HumanSendRefusedError(RelationViolationError("Read-only saved view")) + route_current = await view.app.preparation.run_thread( + lambda: view.app.coordination_access.current(self.root)) + if not route_current or not self.current: + raise HumanSendRefusedError(RelationViolationError("Message route changed; reopen this view")) + self.enter(SendPending()) + task = asyncio.create_task(asyncio.to_thread( + comms.messaging.send_user_message, self.conversation.send_target(), body, + worktree=view.project_path)) + try: + sent = await asyncio.shield(task) + except asyncio.CancelledError: + self.delivery_blocked("Send cancelled while worker may still write; outcome UNKNOWN, inspect the bus, do not retry", body) + task.add_done_callback(lambda done: done.exception() if not done.cancelled() else None) + return + except CommittedMessageError as error: + sent = error.record + self.enter(SentUndrawn(comms.root, sent)) + except asyncio.CancelledError: + self.delivery_refused("Cancelled before sending", body) + raise + except (HumanSendError, RouteChangedError) as error: + delivery_failure(error).present(self, body) + return if self.current: - await self.owner.view.agent_turn_over(reason) + await self.present(sent) class ConversationSubmissions: @@ -257,26 +370,37 @@ def __init__(self, view): self.view = view self.active: list[SubmissionExecution] = [] - def begin(self, submission: AgentInputSubmission): - return self.view.agent_binding.begin_submission(self, submission) + def begin(self, submission: PromptInputSubmission): + return self.view.queue_source.begin_submission(self, submission) - def unbound(self, submission: AgentInputSubmission, refusal: InputRefused): + def unbound(self, submission: PromptInputSubmission, refusal: InputRefused): return UnboundSubmissionStart(self, submission, refusal) - def start(self, submission: AgentInputSubmission, agent: Agent): - execution = SubmissionExecution(self, submission, agent) + def start(self, execution: SubmissionExecution): self.active.append(execution) + self.view.refresh_submission_state() return execution + def start_message(self, submission: PromptInputSubmission, root: Path, conversation, + current: Callable[[], bool], present: Callable[[Message], Awaitable[None]]): + return self.start(MessageSubmissionExecution(self, submission, root, conversation, current, present)) + + @property + def send_block_reason(self) -> str: + return next((item.state.reason for item in self.active if item.state.reason), "") + @property def input_decision(self) -> InputDecision: - return self.view.operational_sources.input_decision(self.view) + return self.view.queue_source.input_decision(self.view) async def submit(self, event: UserInputSubmitted): text = event.body.strip() if not event.shell and text.startswith("/") and await CommandCatalog.execute(text, self.view): return - return await self.view.operational_sources.submit(self, event) + if self.send_block_reason: + self.view.delivery_refused(self.send_block_reason, event.body) + return + return await InputSubmission.from_event(event, self).execute(self) def accepts_failure(self, message: CoreEventMessage): return self.view.agent_binding.accepts_failed_input(message) @@ -285,7 +409,7 @@ def accepts_failure(self, message: CoreEventMessage): def queue_evidence(self) -> QueueEvidence: return QueueEvidence( self.view.agent_binding.presentation, self.delivering, - self.view.send_block_reason, self.view.contents.notification_targets) + self.send_block_reason, self.view.contents.notification_targets) @property def queue_summary(self) -> str: @@ -304,35 +428,32 @@ def queued_inputs(self): def delivering(self): queue = self.queue_projection accepted = {row.input_id for row in (*queue.items, *queue.restored)} - return tuple( - item.submission.text - for item in self.active - if item.current - and item.agent.presentation.uses_managed_turns - and item.request.input_id not in accepted - and item.started_update is None - ) + return tuple(text for item in self.active if item.current + for text in item.state.local_intent(item, accepted)) def native_input_presented(self, started: InputStartedUpdate): """Retain this request's actual response while its RPC is outstanding.""" for execution in self.active: - if execution.current and started.started_request(execution.request.input_id): - execution.started_update = started + execution.native_input_presented(started) def publish_pending(self): if (prompt := self.view.query_one_optional(Prompt)) is not None: prompt.sync_queue() - def finish(self, execution: SubmissionExecution): + def retire(self, execution: SubmissionExecution) -> None: if execution in self.active: self.active.remove(execution) - if execution.current: - self.publish_pending() - def reset(self): - self.active.clear() + def finish(self, execution: SubmissionExecution): + execution.state.settle(execution) self.publish_pending() + def settle_drawn(self) -> None: + for execution in tuple(self.active): + if execution.state.drawn_in(self.view): + execution.enter(NoSend()) + self.finish(execution) + def send_now(self): source = self.view.queue_source presentation = self.view.agent_binding.presentation @@ -342,7 +463,7 @@ def send_now(self): group="send-queued-now", exclusive=True, ) - async def _send_now(self, source: QueueSource, presentation, scope: QueueScope) -> None: + async def _send_now(self, source: QueueActions, presentation, scope: QueueScope) -> None: try: await source.queue_send_now(presentation, scope) except (RequestError, QueueActionRefused) as error: diff --git a/src/toad/delivery_failure_view.py b/src/toad/delivery_failure_view.py index 7ff10a298..5db4a9e98 100644 --- a/src/toad/delivery_failure_view.py +++ b/src/toad/delivery_failure_view.py @@ -1,18 +1,7 @@ -"""The concrete compose controls used by shared delivery-failure declarations.""" +"""Compose and error presentation derived from the submission owner.""" - -import asyncio from abc import abstractmethod -from dataclasses import dataclass -from pathlib import Path -from agent_comms.errors import ( - CommittedMessageError, - HumanSendError, - HumanSendRefusedError, - RelationViolationError, - RouteChangedError, -) from agent_comms.messages import Message from rich.text import Text as RichText from textual import getters @@ -21,60 +10,20 @@ from textual.widget import Widget from toad.app_node import ToadNode +from toad.conversation_submission import ConversationSubmissions from toad.widgets.flash import Flash from toad.widgets.prompt import Prompt -class SendState: - """Where this view's own human-message send stands; compose follows it.""" - - reason = "" - - def drawn_in(self, view) -> bool: - """Whether this send is now drawn in `view`, so compose can open again.""" - return False - - -class NoSend(SendState): - pass - - -class SendPending(SendState): - reason = "Send pending; do not retry" - - -@dataclass(frozen=True) -class SentUndrawn(SendState): - root: Path - sent: Message - - @property - def reason(self) -> str: - return f"Sent {self.root}/{self.sent.seq}/{self.sent.message_id}; waiting to be drawn, do not retry" - - def drawn_in(self, view) -> bool: - return view.draws_user_message(self.sent) - - -@dataclass(frozen=True) -class SendBlocked(SendState): - description: str - - @property - def reason(self) -> str: - return f"Send blocked: {self.description}" - - class DeliveryFailureView(ToadNode, Widget): - """The composed input, status and delivery behavior shared by conversations.""" + """Draw submission state; this widget owns no transport or send lifecycle.""" + submissions: ConversationSubmissions prompt = getters.query_one(Prompt) feedback = getters.query_one(Flash) status: var[str | Content | RichText] = var("") thread_status: var[str | Content | RichText] = var("") - """The thread's own status from its owner; a send or read-only block shows over it.""" composer_block: var[str] = var("") - """Why compose is held: this view's send, or a read-only saved view; empty when open.""" @abstractmethod def draws_user_message(self, message: Message) -> bool: ... @@ -82,26 +31,18 @@ def draws_user_message(self, message: Message) -> bool: ... async def composer_read_only(self) -> bool: return False - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self._send_state: SendState = NoSend() - @property def send_block_reason(self) -> str: - """Why this view's own send holds compose; empty when it does not.""" - return self._send_state.reason + return self.submissions.send_block_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 + def refresh_submission_state(self) -> None: self.queue_evidence_changed() if self.is_attached: async def refresh() -> None: - read_only = await self.composer_read_only() - await self.refresh_composer(read_only) + await self.refresh_composer(await self.composer_read_only()) self.run_worker(refresh(), group="composer-access", exclusive=True) @@ -109,7 +50,6 @@ def compute_status(self) -> str | Content | RichText: return self.composer_block or self.thread_status async def refresh_composer(self, read_only: bool) -> None: - """Draw compose from the caller's Core answer and the current send state.""" if not self.is_attached: return reason = self.send_block_reason or ("Read-only saved view" if read_only else "") @@ -120,65 +60,6 @@ async def refresh_composer(self, read_only: bool) -> None: if not area.disabled and self.screen.focused is None: self.screen.set_focus(area) - async def submit_comms_message(self, root, conversation, body, current, present): - """CLI transcripts and wire pages use the same Core-send checks.""" - from agent_comms.delivery_failure import delivery_failure - - if not body.strip(): - return - if self.send_block_reason: - self.feedback.flash(self.send_block_reason, style="error") - return - try: - comms = await self.app.coordination_access.transcripts.service(root) - read_only = await self.app.preparation.run_thread(conversation.read_only, comms) - await self.refresh_composer(read_only) - if read_only: - self.prompt.restore_draft(body) - self.feedback.flash(self.status, style="error") - return - self._enter_send_state(SendPending()) - route_current = await self.app.preparation.run_thread( - lambda: self.app.coordination_access.current(root) - ) - if not route_current or not current(): - raise HumanSendRefusedError(RelationViolationError("Message route changed; reopen this view")) - sent = await self.commit_user_message(comms, conversation.send_target(), body) - except asyncio.CancelledError: - self.delivery_refused("Cancelled before sending", body) - raise - except (HumanSendError, RouteChangedError) as error: - delivery_failure(error).present(self, body) - return - if sent is None: - return - await present(sent) - - async def commit_user_message(self, comms, target, body): - task = asyncio.create_task(asyncio.to_thread( - comms.messaging.send_user_message, target, body, worktree=self.project_path)) - try: - sent = await asyncio.shield(task) - except asyncio.CancelledError: - self.delivery_blocked("Send cancelled while worker may still write; outcome UNKNOWN, inspect the bus, do not retry", body) - task.add_done_callback(lambda done: done.exception() if not done.cancelled() else None) - return None - except CommittedMessageError as error: - sent = error.record - self._enter_send_state(SentUndrawn(comms.root, sent)) - return sent - - def settle_user_message(self) -> None: - """Enable compose only when the sent message is drawn.""" - if self._send_state.drawn_in(self): - self._enter_send_state(NoSend()) - - def delivery_blocked(self, description: str, body: str) -> None: - self.prompt.restore_draft(body) - self._enter_send_state(SendBlocked(description)) - self.feedback.flash(self.send_block_reason, style="error") - def delivery_refused(self, description: str, body: str) -> None: self.prompt.restore_draft(body) - self._enter_send_state(NoSend()) self.feedback.flash(f"Send failed: {description}", style="error") diff --git a/src/toad/input_decision.py b/src/toad/input_decision.py index 9656ef8ed..a6bcf483d 100644 --- a/src/toad/input_decision.py +++ b/src/toad/input_decision.py @@ -75,3 +75,7 @@ class ModelSelectionRequired(InputRefused): class OwnerNotAttached(InputRefused): message = "No live owner binding for this thread; its input stays read-only. Your draft is kept." + + +class InputSourceUnavailable(InputRefused): + message = "Conversation source is not ready. Your draft is kept." diff --git a/src/toad/queue_capability.py b/src/toad/queue_capability.py new file mode 100644 index 000000000..ddd4c1bb9 --- /dev/null +++ b/src/toad/queue_capability.py @@ -0,0 +1,51 @@ +"""Captured queue evidence and mutation capability, below submission execution.""" +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Callable +from dataclasses import dataclass +from functools import cached_property + +from agent_comms.acp_extension import QueueScope +from agent_comms.message_reference import MessageReference +from agent_comms.notification_assignment import NotificationSelection + +from toad.agent_presentation import AgentPresentation +from toad.block_navigation import ConversationEntry +from toad.notification_status import MessageQueueEntry + + +@dataclass(frozen=True) +class QueueEvidence: + """One UI read; only a BUS source asks for its visible message entries.""" + + presentation: AgentPresentation + delivering: tuple[str, ...] + transport: str + read_targets: Callable[[], tuple[ConversationEntry, ...]] + + @cached_property + def entries(self) -> tuple[MessageQueueEntry, ...]: + entries: dict[MessageReference, MessageQueueEntry] = {} + for target in self.read_targets(): + for entry in target.queue_entries(): + entries[entry.reference] = entry + if len(entries) >= NotificationSelection.window_limit: + return tuple(entries.values()) + return tuple(entries.values()) + + +class QueueActions(ABC): + """An exact captured source, not a queue label, grants mutation authority.""" + + @abstractmethod + def queue_scope(self, presentation: AgentPresentation) -> QueueScope: ... + + @abstractmethod + def queue_actions_available(self, presentation: AgentPresentation) -> bool: ... + + @abstractmethod + async def queue_send_now(self, presentation: AgentPresentation, scope: QueueScope) -> None: ... + + @abstractmethod + async def queue_clear(self, presentation: AgentPresentation, scope: QueueScope) -> None: ... diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py index 47c3f7a94..2c826042d 100644 --- a/src/toad/session_presentation.py +++ b/src/toad/session_presentation.py @@ -9,6 +9,7 @@ from weakref import ref from agent_comms.acp_extension import CoordinationChangedUpdate +from agent_comms.messages import Message from agent_comms.session_load import EnsuringSessionLoadPlan from agent_comms.ui_model.message_category import MessageCategory from textual.document._editing import EditingMode @@ -18,6 +19,7 @@ from toad.agent_binding import NoAgentConnection from toad.agent_surface import AttachedSurfaceBinding from toad.conversation_kind import DmConversation +from toad.conversation_submission import PromptInputSubmission, ConversationSubmissions, SubmissionStart from toad.input_history import InputHistories from toad.input_decision import InputDecision from toad.resource_release import AsyncResourceRelease @@ -179,11 +181,6 @@ def prompt(self, conversation: Conversation): return Prompt(project=conversation.project_path, turns=conversation.turns, submissions=conversation.submissions) - async def submit(self, owner, event): - from toad.conversation_submission import InputSubmission - - return await InputSubmission.from_event(event, owner).execute(owner) - def input_decision(self, conversation: Conversation) -> InputDecision: agent = conversation.agent_binding return agent.readiness_decision.then( @@ -256,6 +253,9 @@ async def close(self) -> None: class MailboxSessionSources(BusQueueSource, OperationalSessionSources): """CLI execution receives user messages through Core's inbox, not ACP.""" + def draws_message(self, view: Conversation, root: Path, sent: Message) -> bool: + return view.transcript.binding.thread_scope.draws_message(view, root, sent) + def activity_row(self, conversation: Conversation): return conversation.thread_activity.row @@ -279,21 +279,16 @@ async def present(self, conversation: Conversation) -> None: self.external_transcript_changed(conversation) self.thread_presented(conversation) - async def submit(self, owner, event): + def begin_submission(self, owner: ConversationSubmissions, + submission: PromptInputSubmission) -> SubmissionStart: view = owner.view async def presented(sent): - view.run_worker( - view.input_histories.prompt.record(event.body), group="history" - ) view.window.jump_to_latest() self.external_transcript_changed(view) - await view.transcript.binding.thread_scope.submit_mailbox( - view, - event, - DmConversation, - presented, + return view.transcript.binding.thread_scope.bind_submission( + owner, submission, DmConversation, presented, ) diff --git a/src/toad/session_scope.py b/src/toad/session_scope.py index 02a40a697..dd7c5e293 100644 --- a/src/toad/session_scope.py +++ b/src/toad/session_scope.py @@ -13,6 +13,7 @@ from agent_comms.repo import Checkout from agent_comms.coordination_errors import CoordinationReadUnavailable, StaleRevision from agent_comms.errors import RelationViolationError +from agent_comms.messages import Message 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 @@ -24,9 +25,10 @@ from toad.app_node import ToadNode from toad.constants import WELCOME +from toad.conversation_submission import PromptInputSubmission, ConversationSubmissions, SubmissionStart from toad.core.events import ThreadRowsChanged from toad.core_event_carrier import ThreadCallbackSubscription -from toad.input_decision import InputAccepted, InputDecision +from toad.input_decision import InputAccepted, InputDecision, InputSourceUnavailable from toad.navigation_context import NavigationOwner from toad.owner_preparation import ExistingOwnerPlan from toad.project_path_owner import CheckoutRoot, ProjectPathOwner @@ -35,7 +37,6 @@ if TYPE_CHECKING: from toad.widget_actions import DeclaredWidgetActions - from agent_comms.messages import Message from agent_comms.thread_presentation import ThreadPresentation from textual.command import Provider from toad.widgets.side_bar import SideBar @@ -43,7 +44,6 @@ from toad.agent_schema import AgentDefinition from toad.comms_root import CoordinationAccess from toad.conversation_kind import DmConversation - from toad.core.input_events import UserInputSubmitted from toad.navigation_target import NavigationTarget from toad.session_presentation import OperationalSessionPresentation from toad.thread_identity import SessionIdentity @@ -96,17 +96,18 @@ def update_request( position: PagePosition = LatestPage(), ) -> ReadTranscriptUpdate: ... + def draws_message(self, view: Conversation, root: Path, sent: Message) -> bool: + return False + @abstractmethod def mailbox_target(self) -> ThreadTarget: ... @abstractmethod - async def submit_mailbox( - self, - view: Conversation, - event: UserInputSubmitted, + def bind_submission( + self, owner: ConversationSubmissions, submission: PromptInputSubmission, conversation_type: type[DmConversation], - presented: Callable[[Message | None], Awaitable[None]], - ) -> None: ... + presented: Callable[[Message], Awaitable[None]], + ) -> SubmissionStart: ... @abstractmethod async def show_mailbox_status( @@ -182,22 +183,21 @@ def update_request( self.root, TranscriptPageRead(thread=self.name, position=position), previous=previous ) + def draws_message(self, view: Conversation, root: Path, sent: Message) -> bool: + return Path(self.root) == root and self.shown_by(view) and view.draws_user_message(sent) + def mailbox_target(self) -> ThreadTarget: return ThreadTarget(self.name) - async def submit_mailbox( - self, - view: Conversation, - event: UserInputSubmitted, + def bind_submission( + self, owner: ConversationSubmissions, submission: PromptInputSubmission, conversation_type: type[DmConversation], - presented: Callable[[Message | None], Awaitable[None]], - ) -> None: - await view.submit_comms_message( - self.root, - self.mailbox_conversation(view, conversation_type), - event.body, - lambda: self.shown_by(view), - presented, + presented: Callable[[Message], Awaitable[None]], + ) -> SubmissionStart: + view = owner.view + return owner.start_message( + submission, Path(self.root), self.mailbox_conversation(view, conversation_type), + lambda: self.shown_by(view), presented, ) async def show_mailbox_status( @@ -277,14 +277,12 @@ def update_request( def mailbox_target(self) -> Never: self.unavailable() - async def submit_mailbox( - self, - view: Conversation, - event: UserInputSubmitted, + def bind_submission( + self, owner: ConversationSubmissions, submission: PromptInputSubmission, conversation_type: type[DmConversation], - presented: Callable[[Message | None], Awaitable[None]], - ) -> None: - view.delivery_refused("CLI transcript is not ready", event.body) + presented: Callable[[Message], Awaitable[None]], + ) -> SubmissionStart: + return owner.unbound(submission, InputSourceUnavailable()) async def show_mailbox_status( self, view: Conversation, conversation_type: type[DmConversation] diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py index 60902c09c..660dda49f 100644 --- a/src/toad/submission_queue.py +++ b/src/toad/submission_queue.py @@ -1,64 +1,60 @@ -"""Source-owned queue capabilities and captured evidence, independent of screens.""" +"""Source-bound submission capabilities and one queue presentation workflow.""" from __future__ import annotations -from abc import ABC, abstractmethod -from collections.abc import Callable -from dataclasses import dataclass -from functools import cached_property +from pathlib import Path from typing import ClassVar from agent_comms.acp_extension import NoQueueScope, QueueItem, QueueScope -from agent_comms.message_reference import MessageReference -from agent_comms.notification_assignment import NotificationSelection +from agent_comms.messages import Message from toad.acp.queue_attachment import QueueActionRefused from toad.agent_presentation import AgentPresentation -from toad.block_navigation import ConversationEntry +from toad.conversation_submission import PromptInputSubmission, ConversationSubmissions, SubmissionStart +from toad.input_decision import InputAccepted, InputDecision, InputSourceUnavailable from toad.line_blocks import MutedRole, Span -from toad.notification_status import DetailLines, MessageQueueEntry +from toad.notification_status import DetailLines +from toad.queue_capability import QueueActions, QueueEvidence -@dataclass(frozen=True) -class QueueEvidence: - """One UI read; only a BUS source asks for its visible message entries.""" - - presentation: AgentPresentation - delivering: tuple[str, ...] - transport: str - read_targets: Callable[[], tuple[ConversationEntry, ...]] - - @cached_property - def entries(self) -> tuple[MessageQueueEntry, ...]: - entries: dict[MessageReference, MessageQueueEntry] = {} - for target in self.read_targets(): - for entry in target.queue_entries(): - entries[entry.reference] = entry - if len(entries) >= NotificationSelection.window_limit: - return tuple(entries.values()) - return tuple(entries.values()) - - -class QueueSource(ABC): - """The source selects its capability; the actual presentation owns native facts.""" +class QueueSource(QueueActions): + """The captured source owns admission, observation and mutation capability.""" preview_limit: ClassVar[int] = 3 - @abstractmethod - def queue_scope(self, presentation: AgentPresentation) -> QueueScope: ... + def begin_submission(self, owner: ConversationSubmissions, + submission: PromptInputSubmission) -> SubmissionStart: + return owner.unbound(submission, InputSourceUnavailable()) - @abstractmethod - def queue_actions_available(self, presentation: AgentPresentation) -> bool: ... + def input_decision(self, view) -> InputDecision: + return InputSourceUnavailable() - @abstractmethod - async def queue_send_now(self, presentation: AgentPresentation, scope: QueueScope) -> None: ... + def defers_input(self, view) -> bool: + return False - @abstractmethod - async def queue_clear(self, presentation: AgentPresentation, scope: QueueScope) -> None: ... + def draws_message(self, view, root: Path, sent: Message) -> bool: + """Only the original message source can confirm its committed row.""" + return False def queue_summary(self, evidence: QueueEvidence) -> str: + local = evidence.delivering + submitting = (f"Submitting ({len(local)}): " + " · ".join( + QueueItem.summarize(text) for text in local[:self.preview_limit])) if local else "" + return " | ".join(part for part in ( + evidence.transport, submitting, self.observed_summary(evidence)) if part) + + def observed_summary(self, evidence: QueueEvidence) -> str: return "Queue unavailable" def queue_details(self, evidence: QueueEvidence) -> DetailLines: + lines = [(Span("Submitting is local intent; acceptance is not consumption.", MutedRole),)] + lines.extend((Span(f"Submitting: {QueueItem.summarize(text)}", MutedRole),) + for text in evidence.delivering) + if evidence.transport: + lines.append((Span(evidence.transport, MutedRole),)) + lines.extend(self.observed_details(evidence)) + return tuple(lines) + + def observed_details(self, evidence: QueueEvidence) -> DetailLines: return ((Span("No queue observation is available for this source.", MutedRole),),) @@ -79,29 +75,35 @@ async def queue_clear(self, presentation: AgentPresentation, scope: QueueScope) class NativeQueueSource(QueueSource): - def queue_summary(self, evidence: QueueEvidence) -> str: + def begin_submission(self, owner: ConversationSubmissions, + submission: PromptInputSubmission) -> SubmissionStart: + return owner.view.agent_binding.begin_submission(owner, submission) + + def defers_input(self, view) -> bool: + return view.turns.owner.busy and view.agent_binding.presentation.queue_supported + + def observed_summary(self, evidence: QueueEvidence) -> str: presentation = evidence.presentation projection = presentation.queue feedback = projection.feedback(presentation.queue_supported) parts = [feedback] if feedback else [] - for label, texts, detail in ( - ("Submitting", tuple(QueueItem.summarize(text) for text in evidence.delivering), ""), - ("Queued", tuple(row.preview for row in projection.items), ""), - ("Restored", tuple(row.preview for row in projection.restored), ", read-only"), + for label, rows, detail in ( + ("Queued", projection.items, ""), + ("Restored", projection.restored, ", read-only"), ): - if texts: - parts.append(f"{label} ({len(texts)}{detail}): " + " · ".join(texts[:self.preview_limit])) + if rows: + parts.append(f"{label} ({len(rows)}{detail}): " + " · ".join( + row.preview for row in rows[:self.preview_limit])) return " | ".join(parts) - def queue_details(self, evidence: QueueEvidence) -> DetailLines: + def observed_details(self, evidence: QueueEvidence) -> DetailLines: projection = evidence.presentation.queue - 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 evidence.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(evidence.presentation.queue_identity, MutedRole),)) - return tuple(lines) + return ( + *((Span(f"Queued {row.input_id}: {row.preview}", MutedRole),) for row in projection.items), + *((Span(f"Restored {row.input_id} (read-only): {row.preview}", MutedRole),) + for row in projection.restored), + (Span(evidence.presentation.queue_identity, MutedRole),), + ) def queue_scope(self, presentation: AgentPresentation) -> QueueScope: return presentation.queue_scope @@ -119,17 +121,18 @@ async def queue_clear(self, presentation: AgentPresentation, scope: QueueScope) class BusQueueSource(ReadOnlyQueueSource): - def queue_summary(self, evidence: QueueEvidence) -> str: + def input_decision(self, view) -> InputDecision: + return InputAccepted() + + def observed_summary(self, evidence: QueueEvidence) -> str: entries = evidence.entries previews = " · ".join(entry.summary for entry in entries[:self.preview_limit]) coverage = f"Delivery ({len(entries)} visible messages)" - return " | ".join(part for part in ( - evidence.transport, coverage + (f": {previews}" if previews else ": unobserved")) if part) + return coverage + (f": {previews}" if previews else ": unobserved") - def queue_details(self, evidence: QueueEvidence) -> DetailLines: + def observed_details(self, evidence: QueueEvidence) -> DetailLines: return ( - (Span("Visible original message references only; this is not an ACP queue or a complete inbox count.", MutedRole),), - (Span(evidence.transport, MutedRole),), + (Span("Visible original message references only; not a complete inbox count.", 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 evidence.entries[:self.preview_limit] for line in entry.details), (Span("Clear/send-now unavailable: this BUS source has no native queue authority.", MutedRole),), diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py index 0aae526a0..80effea37 100644 --- a/src/toad/transcript_preparation.py +++ b/src/toad/transcript_preparation.py @@ -42,7 +42,6 @@ from agent_comms.message_page import MessagePage from toad.comms_root import CoordinationAccess, Observation - from toad.core.input_events import UserInputSubmitted from toad.transcript_source_preparation import TranscriptSourcePreparation from toad.widgets.comms_chat import CommsChatView from toad.widgets.conversation import Contents @@ -584,9 +583,6 @@ def status_thread(self, access: CoordinationAccess): def command_context(self, view: CommsChatView): return None - async def submit_input(self, view: CommsChatView, event: UserInputSubmitted) -> None: - view.delivery_refused("Conversation source is not ready", event.body) - def schedule_refresh(self, view: CommsChatView): return None diff --git a/src/toad/transcript_publication.py b/src/toad/transcript_publication.py index aef0c8613..6346298cc 100644 --- a/src/toad/transcript_publication.py +++ b/src/toad/transcript_publication.py @@ -109,7 +109,7 @@ def painted(self, cursor: TranscriptCursor) -> None: self.owner.displayed_cursor = cursor view.app.history_displayed() view.query_ancestor(SessionView).history_painted() - view.settle_user_message() + view.submissions.settle_drawn() class TranscriptPublication(BoundFrame, ABC): diff --git a/src/toad/widgets/comms_chat.py b/src/toad/widgets/comms_chat.py index c8998f076..ad0c7125d 100644 --- a/src/toad/widgets/comms_chat.py +++ b/src/toad/widgets/comms_chat.py @@ -2,7 +2,6 @@ from __future__ import annotations from toad.widgets.presentation_window import OLDER_END, NEWER_END -from toad.core import input_events from functools import partial from toad.widgets.wire_history import WireHistory from toad.message_rows import IrcMessageStyle @@ -41,7 +40,6 @@ from toad.widgets.flash import Flash from toad.widgets.prompt import Prompt from toad.session_scope import NoTranscript, SessionScope -from toad.command_catalog import CommandCatalog def session_thread_name(project_path) -> str: @@ -153,7 +151,7 @@ async def on_transcript_source_work_finished(self, event: TranscriptSourceWorkFi access = self.app.coordination_access if message.publisher is not history: return - self.settle_user_message() + self.submissions.settle_drawn() if (not history.state.accepts_source_work or not source.uses_observation(access.observation.observed())): return @@ -256,7 +254,7 @@ async def _refresh_source(self) -> None: await self.refresh_composer(read_only) self._refreshed_from = announced # A sent message may now be drawn: composing resumes before mentions are read. - self.settle_user_message() + self.submissions.settle_drawn() except (CoordinationReadUnavailable, StaleRevision): raise # Composer mentions and DM status read the observer's snapshot @@ -312,19 +310,11 @@ async def composer_read_only(self) -> bool: async def command_target_context(self): return self.message_history.page_source.command_context(self) - async def submit_input(self, event: input_events.UserInputSubmitted) -> None: - if event.body.strip().startswith("/") and await CommandCatalog.execute(event.body.strip(), self): - return - if not event.body.strip(): - return - await self.message_history.page_source.submit_input(self, event) - - async def _show_sent(self, sent: WireMessage, body: str) -> None: + async def _show_sent(self, sent: WireMessage) -> None: """Follow the conversation's end and read it now; composing resumes once `sent` is drawn.""" - self.run_worker(partial(self.input_histories.prompt.record, body), group="history") self.window.jump_to_latest() await self._refresh() - self.settle_user_message() + self.submissions.settle_drawn() def draws_user_message(self, sent) -> bool: return self.message_history.draws(sent) diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py index 91e2c5093..39c3c7a3b 100644 --- a/src/toad/widgets/conversation.py +++ b/src/toad/widgets/conversation.py @@ -835,7 +835,7 @@ def watch_agent_binding(self, binding: AgentConnection) -> None: attachments = binding.presentation.attachments self.native_history_status = attachments.cursor self._private_cursor_sequence = attachments.cursor_sequence - self.submissions.reset() + self.submissions.publish_pending() # The observed presentation is Core's answer for this thread, not the # agent's; it stays until the next observation replaces it. self.arm_thread_read() diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index e8070a13f..a913ff798 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -34,6 +34,7 @@ from toad.comms_root import CoordinationAccess from toad.conversation_kind import ConversationKind +from toad.conversation_submission import PromptInputSubmission, ConversationSubmissions, SubmissionStart from toad.line_blocks import Body, Gap, LineBlock, MutedRole, Summary from toad.message_rows import MessageRow, WireMessageStyle from toad.transcript_preparation import ( @@ -62,7 +63,6 @@ from agent_comms.read_basis import NamedViewer from toad.comms_root import Observation - from toad.core.input_events import UserInputSubmitted from toad.widgets.comms_chat import CommsChatView from toad.widgets.transcript_lines import Lines from toad.widgets.wire_history import WireHistory @@ -323,6 +323,20 @@ class WireSource(BusQueueSource): tail_limit: int = 8 page_limit: int = 40 + def draws_message(self, view, root: Path, sent: Message) -> bool: + return self.comms.root == root and view.draws_user_message(sent) + + def begin_submission(self, owner: ConversationSubmissions, + submission: PromptInputSubmission) -> SubmissionStart: + view = owner.view + source = view.message_history.page_source + return owner.start_message( + submission, self.comms.root, self.conversation, + lambda: (view.is_attached and source is view.message_history.page_source + and self.conversation == view.conversation), + view._show_sent, + ) + @property def viewer(self) -> NamedViewer: return self.comms.views.display_viewer(self.project) @@ -465,15 +479,6 @@ def command_context(self, view): view.query_ancestor(SessionScope).id, ) - async def submit_input(self, view: CommsChatView, event: UserInputSubmitted) -> None: - conversation = self.loader.conversation - current = lambda: ( - view.is_attached and self is view.message_history.page_source and conversation == view.conversation - ) - await view.submit_comms_message( - self.loader.comms.root, conversation, event.body, current, partial(view._show_sent, body=event.body) - ) - def schedule_refresh(self, view: CommsChatView): history = view.message_history return history.schedule_source_work(view._refresh_source) diff --git a/tests/textual/test_input_failure_draft.py b/tests/textual/test_input_failure_draft.py index f7e07b41c..aaa4efe25 100644 --- a/tests/textual/test_input_failure_draft.py +++ b/tests/textual/test_input_failure_draft.py @@ -107,28 +107,3 @@ async def test_a_successful_wire_send_keeps_a_different_unsent_draft(workspace): why="the committed message drawn and composing released") assert chat.prompt.text == "A different unsent draft" assert editor.document is document and editor.history is history - - -async def test_a_sent_message_scrolled_past_by_a_long_reply_still_releases_composing(workspace): - """Drawn means in the history's drawn rows, not inside the viewport: a long reply - landing in the same read pushes the sent message off screen, and composing must open.""" - from toad.delivery_failure_view import DeliveryFailureView - from toad.widgets.comms_chat import CommsChatView - - comms = workspace.comms - workspace.thread("peer") - commit = DeliveryFailureView.commit_user_message - - async def commit_then_reply(view, *args, **kwargs): - sent = await commit(view, *args, **kwargs) - comms.messaging.send("peer", workspace.viewer, "\n\n".join(f"reply line {n}" for n in range(80))) - return sent - - async with workspace.app(size=(110, 24)) as ui: - await ui.open_direct("peer", actor=workspace.viewer) - chat = ui.app.selected_session.query_one(CommsChatView) - with patch.object(DeliveryFailureView, "commit_user_message", commit_then_reply): - await chat.submit_input(UserInputSubmitted("SCROLLED_PAST_SEND")) - await ui.until(lambda: ui.shows("reply line 79") and not chat.send_block_reason, - why="the long reply drawn and composing released") - assert not ui.shows("SCROLLED_PAST_SEND") diff --git a/tests/textual/test_painted_view_marks_itself_visible.py b/tests/textual/test_painted_view_marks_itself_visible.py deleted file mode 100644 index ae15999e0..000000000 --- a/tests/textual/test_painted_view_marks_itself_visible.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Textual-facing: a painted transcript marks the session that holds it visible, even when the last tab closed before the paint landed, and a hidden tab's paint leaves it unread.""" - -from types import SimpleNamespace - -import pytest - -from toad.app import HeapFrozen, ToadApp -from toad.screens.session_view import SessionView -from toad.transcript_publication import BoundFrame - - -@pytest.mark.parametrize("on_screen, marked", [(True, ["painted"]), (False, [])]) -def test_a_paint_after_the_last_tab_closed_marks_only_its_own_session(on_screen, marked): - seen = [] - session = SimpleNamespace(is_current=on_screen, mark_visible=lambda: seen.append("painted")) - session.history_painted = lambda: SessionView.history_painted(session) - app = SimpleNamespace(heap=HeapFrozen(), selected_session=None) - app.history_displayed = lambda: ToadApp.history_displayed(app) - view = SimpleNamespace( - app=app, - query_ancestor=lambda kind: session if kind is SessionView else None, - settle_user_message=lambda: None, - ) - owner = SimpleNamespace(displayed_cursor=None) - - BoundFrame(SimpleNamespace(view=view), owner, 0).record("cursor") - - assert seen == marked - assert owner.displayed_cursor == "cursor" diff --git a/tests/textual/test_stale_route_writes.py b/tests/textual/test_stale_route_writes.py index 94657b292..2d1fc5404 100644 --- a/tests/textual/test_stale_route_writes.py +++ b/tests/textual/test_stale_route_writes.py @@ -5,6 +5,7 @@ from support.roots import installed_wire from toad.core.input_events import UserInputSubmitted +from toad.conversation_submission import OrdinaryInputSubmission from toad.widgets.comms_chat import CommsChatView @@ -17,8 +18,10 @@ async def test_a_view_that_changed_during_send_preflight_keeps_the_draft_editabl view = ui.app.selected_session.query_one(CommsChatView) messaging = view.message_history.page_source.loader.comms.messaging before = messaging.bus.bus.revision() - await view.submit_comms_message(workspace.roots.wire, view.conversation, "STALE_DRAFT", - lambda: False, view._show_sent) + await view.submissions.start_message( + OrdinaryInputSubmission("STALE_DRAFT"), workspace.roots.wire, + view.conversation, lambda: False, view._show_sent, + ).prepare() await ui.until(lambda: not view.prompt.prompt_text_area.disabled, why="the refused draft editable again") assert not view.send_block_reason From 0c75ad5beeae6b8508545e36a289d716dc6b7de1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 31 Oct 2026 19:35:08 -0400 Subject: [PATCH 02/24] Declare the shared queue-entry capability instead of assuming a widget --- src/toad/block_navigation.py | 9 ++------- src/toad/notification_status.py | 7 +++++++ src/toad/queue_capability.py | 5 ++--- src/toad/widgets/message_notifications.py | 4 ++-- 4 files changed, 13 insertions(+), 12 deletions(-) diff --git a/src/toad/block_navigation.py b/src/toad/block_navigation.py index 027cdb3b9..0a844b2f5 100644 --- a/src/toad/block_navigation.py +++ b/src/toad/block_navigation.py @@ -3,7 +3,6 @@ from __future__ import annotations from abc import abstractmethod -from collections.abc import Iterable from functools import cached_property from typing import TYPE_CHECKING, Callable, ClassVar, cast @@ -16,7 +15,7 @@ from toad.block_content import BlockContent from toad.frame_presentation import AnimationEffects -from toad.notification_status import MessageQueueEntry +from toad.notification_status import QueueEntries from toad.widgets.presentation_window import NewerEnd, OlderEnd, PageEnd if TYPE_CHECKING: @@ -377,7 +376,7 @@ def clear(self): self._child = None -class ConversationEntry(AnimationEffects[Mix], BlockContent): +class ConversationEntry(AnimationEffects[Mix], BlockContent, QueueEntries): """Conversation content, with an owned atomic cursor by default. The rows drawn inside an entry paint its fades (`LinkedRows.arrival`). @@ -404,10 +403,6 @@ def notification_identity_for(self, references: frozenset[MessageReference]) -> """Live entries have no captured BUS display cut; source rows override with theirs.""" return NoDisplaySelection() - def queue_entries(self) -> Iterable[MessageQueueEntry]: - """Live entries supply no captured BUS delivery rows; source rows override with theirs.""" - return () - @cached_property def block_cursor(self) -> BlockCursor: return AtomicBlockCursor(self) diff --git a/src/toad/notification_status.py b/src/toad/notification_status.py index 78ef9da51..05b37b071 100644 --- a/src/toad/notification_status.py +++ b/src/toad/notification_status.py @@ -77,6 +77,13 @@ def details(self) -> DetailLines: *self.status.queue_details) +class QueueEntries: + """A drawn target supplies its original message queue evidence, or none.""" + + def queue_entries(self) -> Iterable[MessageQueueEntry]: + return () + + @dataclass(frozen=True) class UncheckedNotifications(NotificationStatus): # The row is drawn before Core answers, so the answer changes its text diff --git a/src/toad/queue_capability.py b/src/toad/queue_capability.py index ddd4c1bb9..6767f4296 100644 --- a/src/toad/queue_capability.py +++ b/src/toad/queue_capability.py @@ -11,8 +11,7 @@ from agent_comms.notification_assignment import NotificationSelection from toad.agent_presentation import AgentPresentation -from toad.block_navigation import ConversationEntry -from toad.notification_status import MessageQueueEntry +from toad.notification_status import MessageQueueEntry, QueueEntries @dataclass(frozen=True) @@ -22,7 +21,7 @@ class QueueEvidence: presentation: AgentPresentation delivering: tuple[str, ...] transport: str - read_targets: Callable[[], tuple[ConversationEntry, ...]] + read_targets: Callable[[], tuple[QueueEntries, ...]] @cached_property def entries(self) -> tuple[MessageQueueEntry, ...]: diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py index 4eee1f4cc..678d8d4c6 100644 --- a/src/toad/widgets/message_notifications.py +++ b/src/toad/widgets/message_notifications.py @@ -20,7 +20,7 @@ from toad.messages import QueueEvidenceChanged from toad.notification_status import ( - CheckedNotifications, MessageQueueEntry, NotificationStatus, UnavailableNotifications, UncheckedNotifications, + CheckedNotifications, MessageQueueEntry, QueueEntries, NotificationStatus, UnavailableNotifications, UncheckedNotifications, ) if TYPE_CHECKING: from toad.widgets.tool_content import ToolContent @@ -191,7 +191,7 @@ def on_show(self) -> None: self.call_after_refresh(self.check_notifications) -class SourceNotifications: +class SourceNotifications(QueueEntries): notification_view: HistoryNotifications def queue_entries(self) -> Iterable[MessageQueueEntry]: From e3237ef49626ede2acf1a44820522366eac1dd8f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 31 Oct 2026 20:48:43 -0400 Subject: [PATCH 03/24] Settle submissions through worker completion before view retirement --- src/toad/conversation_submission.py | 61 ++++++++++------- src/toad/delivery_failure_view.py | 2 +- src/toad/session_presentation.py | 6 +- src/toad/session_scope.py | 6 +- src/toad/submission_queue.py | 2 +- src/toad/transcript_preparation.py | 5 ++ src/toad/transcript_publication.py | 2 +- src/toad/widgets/comms_chat.py | 14 ++-- src/toad/widgets/conversation.py | 14 ++-- src/toad/widgets/transcript_fragments.py | 6 +- src/toad/widgets/transcript_history.py | 8 +-- src/toad/wire_history.py | 4 +- tests/native/test_native_journey.py | 6 +- tests/textual/test_input_failure_draft.py | 81 ++++++++++++++++++++++- 14 files changed, 163 insertions(+), 54 deletions(-) diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py index 09cfb6519..12be997d7 100644 --- a/src/toad/conversation_submission.py +++ b/src/toad/conversation_submission.py @@ -188,7 +188,7 @@ class SendState: def settle(self, execution) -> None: execution.owner.retire(execution) - def drawn_in(self, view) -> bool: + def acknowledged_by(self, view) -> bool: return False def response_finished(self, execution) -> None: @@ -211,7 +211,9 @@ class CheckingSend(RetainedSend): reason = "Checking source; do not retry" def response_finished(self, execution) -> None: - execution.enter(NoSend()) + execution.delivery_refused( + "Request ended before handing input to its source", execution.submission.source_text, + ) def local_intent(self, execution, accepted) -> tuple[str, ...]: return () if execution.accepted(accepted) else (execution.submission.text,) @@ -228,16 +230,16 @@ def response_finished(self, execution) -> None: @dataclass(frozen=True) -class SentUndrawn(RetainedSend): +class SentAwaitingSource(RetainedSend): root: Path sent: Message @property def reason(self) -> str: - return f"Sent {self.root}/{self.sent.seq}/{self.sent.message_id}; waiting to be drawn, do not retry" + return f"Sent {self.root}/{self.sent.seq}/{self.sent.message_id}; waiting for source acknowledgement, do not retry" - def drawn_in(self, view) -> bool: - return view.queue_source.draws_message(view, self.root, self.sent) + def acknowledged_by(self, view) -> bool: + return view.queue_source.contains_message(view, self.root, self.sent) @dataclass(frozen=True) @@ -252,6 +254,8 @@ def reason(self) -> str: class SubmissionExecution(SubmissionStart): """Shared history, worker and outcome lifetime over an immutable source binding.""" + worker_group = __qualname__ + def __init__(self, owner: ConversationSubmissions, submission: PromptInputSubmission): self.owner, self.submission = owner, submission self.state: SendState = CheckingSend() @@ -282,28 +286,26 @@ async def prepare(self): self.owner.publish_pending() view.run_worker(partial(view.input_histories.prompt.record, submission.source_text), group="history") try: - await self.feedback() - if self.current: - return view.run_worker(partial(self.send), group="conversation-submission") - self.delivery_refused("Message source changed; reopen this view", submission.source_text) + worker = view.run_worker(partial(self.send), group=self.worker_group) + worker._add_done_callback(lambda _worker: self.finish()) + return worker except BaseException: - self.state.response_finished(self) - self.owner.finish(self) + self.finish() raise - self.owner.finish(self) @abstractmethod async def deliver(self): ... async def send(self): - try: - if self.current: - await self.deliver() - else: - self.delivery_refused("Message source changed; reopen this view", self.submission.source_text) - finally: - self.state.response_finished(self) - self.owner.finish(self) + await self.feedback() + if self.current: + await self.deliver() + else: + self.delivery_refused("Message source changed; reopen this view", self.submission.source_text) + + def finish(self) -> None: + self.state.response_finished(self) + self.owner.finish(self) def delivery_blocked(self, description: str, body: str) -> None: self.enter(SendBlocked(description)) @@ -352,7 +354,7 @@ async def deliver(self): return except CommittedMessageError as error: sent = error.record - self.enter(SentUndrawn(comms.root, sent)) + self.enter(SentAwaitingSource(comms.root, sent)) except asyncio.CancelledError: self.delivery_refused("Cancelled before sending", body) raise @@ -370,6 +372,17 @@ def __init__(self, view): self.view = view self.active: list[SubmissionExecution] = [] + @property + def retains_surface(self) -> bool: + """Pending, unacknowledged and UNKNOWN intent still owns this view's resources.""" + return bool(self.active) + + async def close(self) -> None: + workers = self.view.workers.cancel_group(self.view, SubmissionExecution.worker_group) + # Worker reports errors itself; retirement waits for every completion, + # including cancellation, before the view releases its draft resources. + await asyncio.gather(*(worker.wait() for worker in workers), return_exceptions=True) + def begin(self, submission: PromptInputSubmission): return self.view.queue_source.begin_submission(self, submission) @@ -448,9 +461,9 @@ def finish(self, execution: SubmissionExecution): execution.state.settle(execution) self.publish_pending() - def settle_drawn(self) -> None: + def settle_acknowledged(self) -> None: for execution in tuple(self.active): - if execution.state.drawn_in(self.view): + if execution.state.acknowledged_by(self.view): execution.enter(NoSend()) self.finish(execution) diff --git a/src/toad/delivery_failure_view.py b/src/toad/delivery_failure_view.py index 5db4a9e98..604281921 100644 --- a/src/toad/delivery_failure_view.py +++ b/src/toad/delivery_failure_view.py @@ -26,7 +26,7 @@ class DeliveryFailureView(ToadNode, Widget): composer_block: var[str] = var("") @abstractmethod - def draws_user_message(self, message: Message) -> bool: ... + def contains_user_message(self, message: Message) -> bool: ... async def composer_read_only(self) -> bool: return False diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py index 2c826042d..091a35570 100644 --- a/src/toad/session_presentation.py +++ b/src/toad/session_presentation.py @@ -253,8 +253,8 @@ async def close(self) -> None: class MailboxSessionSources(BusQueueSource, OperationalSessionSources): """CLI execution receives user messages through Core's inbox, not ACP.""" - def draws_message(self, view: Conversation, root: Path, sent: Message) -> bool: - return view.transcript.binding.thread_scope.draws_message(view, root, sent) + def contains_message(self, view: Conversation, root: Path, sent: Message) -> bool: + return view.transcript.binding.thread_scope.contains_message(view, root, sent) def activity_row(self, conversation: Conversation): return conversation.thread_activity.row @@ -411,6 +411,8 @@ async def evict(self, screen: MainScreen) -> None: editor = conversation.prompt.prompt_text_area if not await editor.editing_mode.flush(editor): return # Retain the actual view/engine while its modal input is pending. + if conversation.submissions.retains_surface: + return # Eviction cannot discard an unresolved transaction or its draft recovery. self.state = RetainedSessionViewState.capture(conversation) await self.remove_native_tree(screen) diff --git a/src/toad/session_scope.py b/src/toad/session_scope.py index dd7c5e293..0b226223c 100644 --- a/src/toad/session_scope.py +++ b/src/toad/session_scope.py @@ -96,7 +96,7 @@ def update_request( position: PagePosition = LatestPage(), ) -> ReadTranscriptUpdate: ... - def draws_message(self, view: Conversation, root: Path, sent: Message) -> bool: + def contains_message(self, view: Conversation, root: Path, sent: Message) -> bool: return False @abstractmethod @@ -183,8 +183,8 @@ def update_request( self.root, TranscriptPageRead(thread=self.name, position=position), previous=previous ) - def draws_message(self, view: Conversation, root: Path, sent: Message) -> bool: - return Path(self.root) == root and self.shown_by(view) and view.draws_user_message(sent) + def contains_message(self, view: Conversation, root: Path, sent: Message) -> bool: + return Path(self.root) == root and self.shown_by(view) and view.contains_user_message(sent) def mailbox_target(self) -> ThreadTarget: return ThreadTarget(self.name) diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py index 660dda49f..806a1a6a1 100644 --- a/src/toad/submission_queue.py +++ b/src/toad/submission_queue.py @@ -31,7 +31,7 @@ def input_decision(self, view) -> InputDecision: def defers_input(self, view) -> bool: return False - def draws_message(self, view, root: Path, sent: Message) -> bool: + def contains_message(self, view, root: Path, sent: Message) -> bool: """Only the original message source can confirm its committed row.""" return False diff --git a/src/toad/transcript_preparation.py b/src/toad/transcript_preparation.py index 80effea37..071de7291 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.messages import Message from agent_comms.sent_message import OutgoingMessageReference from agent_comms.transcripts import ( TranscriptCursor, @@ -101,6 +102,10 @@ def configure(self, *, batch_size: int) -> None: if self.fragment_range is None: self.set_fragment_range(slice(max(0, len(self.fragments) - batch_size), len(self.fragments))) + def contains_message(self, message: Message) -> bool: + """Loaded source evidence does not disappear when its drawing range moves.""" + return any(fragment.contains_message(message) for fragment in reversed(self.fragments)) + @property def fragments_in_range(self): return self.fragments[self.start:self.stop] diff --git a/src/toad/transcript_publication.py b/src/toad/transcript_publication.py index 6346298cc..7f3fd0378 100644 --- a/src/toad/transcript_publication.py +++ b/src/toad/transcript_publication.py @@ -109,7 +109,7 @@ def painted(self, cursor: TranscriptCursor) -> None: self.owner.displayed_cursor = cursor view.app.history_displayed() view.query_ancestor(SessionView).history_painted() - view.submissions.settle_drawn() + view.submissions.settle_acknowledged() class TranscriptPublication(BoundFrame, ABC): diff --git a/src/toad/widgets/comms_chat.py b/src/toad/widgets/comms_chat.py index ad0c7125d..937bc8992 100644 --- a/src/toad/widgets/comms_chat.py +++ b/src/toad/widgets/comms_chat.py @@ -151,7 +151,7 @@ async def on_transcript_source_work_finished(self, event: TranscriptSourceWorkFi access = self.app.coordination_access if message.publisher is not history: return - self.submissions.settle_drawn() + self.submissions.settle_acknowledged() if (not history.state.accepts_source_work or not source.uses_observation(access.observation.observed())): return @@ -253,8 +253,8 @@ async def _refresh_source(self) -> None: return await self.refresh_composer(read_only) self._refreshed_from = announced - # A sent message may now be drawn: composing resumes before mentions are read. - self.submissions.settle_drawn() + # The source may now acknowledge the commit, before viewport paint or mention reads. + self.submissions.settle_acknowledged() except (CoordinationReadUnavailable, StaleRevision): raise # Composer mentions and DM status read the observer's snapshot @@ -311,10 +311,10 @@ async def command_target_context(self): return self.message_history.page_source.command_context(self) async def _show_sent(self, sent: WireMessage) -> None: - """Follow the conversation's end and read it now; composing resumes once `sent` is drawn.""" + """Read the conversation's end; composing resumes when its source acknowledges `sent`.""" self.window.jump_to_latest() await self._refresh() - self.submissions.settle_drawn() + self.submissions.settle_acknowledged() - def draws_user_message(self, sent) -> bool: - return self.message_history.draws(sent) + def contains_user_message(self, sent) -> bool: + return self.message_history.contains_message(sent) diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py index 39c3c7a3b..9da18822d 100644 --- a/src/toad/widgets/conversation.py +++ b/src/toad/widgets/conversation.py @@ -686,7 +686,8 @@ def _initialize_session( ) self.cli_draft = CliDraft() self.tool_expansions: dict[str, bool] = {} - # Both explicit eviction and Textual unmount consume this same lifetime. + # Explicit eviction and message-loop exit consume this same lifetime + # before the child editor and feedback surfaces are released. # The stack removes callbacks before calling them and continues on error. self._source_resources = AsyncResourceRelease() self._source_resources.push_async_callback( @@ -697,12 +698,17 @@ def _initialize_session( self._source_resources.callback(self.output.retire) self._source_resources.push_async_callback(self.transcript.detach, self) self._source_resources.callback(self.goal_controls.close) + self._source_resources.push_async_callback(self.submissions.close) + + async def _message_loop_exit(self) -> None: + async with AsyncResourceRelease() as release: + release.push_async_callback(super()._message_loop_exit) + await self._release_source_resources() async def on_unmount(self) -> None: async with AsyncResourceRelease() as release: release.callback(self._publish_session_end) release.push_async_callback(self.agent_binding.retire_surface, self) - release.push_async_callback(self._release_source_resources) def _publish_session_end(self) -> None: if self._agent_data is not None and self.session_start_time is not None: @@ -1132,8 +1138,8 @@ async def interrupt_shell(self) -> None: def cursor_block(self) -> ConversationEntry | None: return self.navigation.current - def draws_user_message(self, sent) -> bool: - return self.transcript.binding.history.draws(sent) + def contains_user_message(self, sent) -> bool: + return self.transcript.binding.history.contains_message(sent) async def _apply_session_name(self, name: str) -> None: await self.agent_binding.set_session_name(name) diff --git a/src/toad/widgets/transcript_fragments.py b/src/toad/widgets/transcript_fragments.py index 7ce9011b2..715a5d352 100644 --- a/src/toad/widgets/transcript_fragments.py +++ b/src/toad/widgets/transcript_fragments.py @@ -352,8 +352,8 @@ def notification_identity(self) -> DisplayIdentity | NoDisplaySelection: def queue_entries(self) -> tuple[MessageQueueEntry, ...]: return () - def draws(self, message: BusMessage) -> bool: - """Whether this fragment draws that bus message; only bus-message fragments do.""" + def contains_message(self, message: BusMessage) -> bool: + """Only bus-message fragments contain the original message reference.""" return False @classmethod @@ -541,7 +541,7 @@ def queue_entries(self) -> tuple[MessageQueueEntry, ...]: return tuple(MessageQueueEntry(reference, self.shown.status) for reference in self.notification_references) - def draws(self, message: BusMessage) -> bool: + def contains_message(self, message: BusMessage) -> bool: return self.events[0].source == message.reference def shown_live_by(self, claim: CommitClaim) -> bool: diff --git a/src/toad/widgets/transcript_history.py b/src/toad/widgets/transcript_history.py index 1dbb4f3b8..f0e1535dc 100644 --- a/src/toad/widgets/transcript_history.py +++ b/src/toad/widgets/transcript_history.py @@ -611,7 +611,7 @@ async def select_range( async with window.preserve_history(root=self): return self.prepared.replace_range(self, selected, current) -class DrawsBusMessages: +class ContainsBusMessages: """The fragments the loaded pages draw, and whether they include a bus message, scrolled into view or not.""" pages: deque[TranscriptPageView] @@ -620,11 +620,11 @@ class DrawsBusMessages: def fragments_in_range(self) -> tuple[TranscriptFragment, ...]: return tuple(fragment for page in self.pages for fragment in page.prepared.fragments_in_range) - def draws(self, message: BusMessage) -> bool: - return any(fragment.draws(message) for fragment in reversed(self.fragments_in_range)) + def contains_message(self, message: BusMessage) -> bool: + return any(page.prepared.contains_message(message) for page in reversed(self.pages)) -class LineHistory(DrawsBusMessages, SourceNotifications, OpensLinks, TranscriptSourcePreparation, ConversationEntry, VerticalGroup): +class LineHistory(ContainsBusMessages, SourceNotifications, OpensLinks, TranscriptSourcePreparation, ConversationEntry, VerticalGroup): """History drawn as lines, a bounded range of pages at a time. It owns paging in both directions, lookahead, trimming, the jump to the newest page and the reader's position. diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index a913ff798..d72033832 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -323,8 +323,8 @@ class WireSource(BusQueueSource): tail_limit: int = 8 page_limit: int = 40 - def draws_message(self, view, root: Path, sent: Message) -> bool: - return self.comms.root == root and view.draws_user_message(sent) + def contains_message(self, view, root: Path, sent: Message) -> bool: + return self.comms.root == root and view.contains_user_message(sent) def begin_submission(self, owner: ConversationSubmissions, submission: PromptInputSubmission) -> SubmissionStart: diff --git a/tests/native/test_native_journey.py b/tests/native/test_native_journey.py index 6a4e8f1d9..5df02ea8e 100644 --- a/tests/native/test_native_journey.py +++ b/tests/native/test_native_journey.py @@ -159,11 +159,15 @@ async def test_input_queued_behind_a_running_turn_is_consumed_and_saved(workspac async def test_input_queued_while_the_tab_is_hidden_is_shown_on_return(workspace, native_package): async with native_session(workspace) as native: ui, agent = native.ui, native.agent - owner_tab = ui.selected_tab + owner_tab, owner = ui.selected_tab, ui.app.selected_session await send_first_input_and_hold_its_reply(native) await ui.write_draft("unsent local draft") + original = ui.conversation process, queue = agent.process.process, agent.queue_attachment await ui.new_session() + assert original.submissions.retains_surface + await owner.presentation.evict(owner) + assert owner.presentation.widget is original, "Eviction discarded a running submission owner" assert agent.process.process is process and process.returncode is None await asyncio.wait_for(agent.send_prompt( "QUEUED_NATIVE_INPUT", request=QueuePromptRequest("QUEUED_NATIVE_INPUT", True)), 10) diff --git a/tests/textual/test_input_failure_draft.py b/tests/textual/test_input_failure_draft.py index aaa4efe25..f3d7f3016 100644 --- a/tests/textual/test_input_failure_draft.py +++ b/tests/textual/test_input_failure_draft.py @@ -1,15 +1,21 @@ """Textual-facing: the conversation's submissions owner puts a locally failed prompt back in the composer, and remote failure facts never overwrite what the user is typing.""" +import asyncio +from contextlib import closing +import sqlite3 from unittest.mock import patch +import pytest +from textual.worker import WorkerCancelled + from acp.exceptions import RequestError from agent_comms.acp_extension import InputFailedUpdate, QueuePromptRequest, RequestFailedUpdate, encode_updates from agent_comms.acp_failure import BackendDeliveryFailure from support.agents import deliver -from toad.conversation_submission import ImmediateInputSubmission +from toad.conversation_submission import ImmediateInputSubmission, SendBlocked, SendPending from toad.core.input_events import UserInputSubmitted @@ -107,3 +113,76 @@ async def test_a_successful_wire_send_keeps_a_different_unsent_draft(workspace): why="the committed message drawn and composing released") assert chat.prompt.text == "A different unsent draft" assert editor.document is document and editor.history is history + (sent,) = comms.bus.bus.latest_addressed( + frozenset((workspace.viewer,)), frozenset(("peer",))) + assert sent.body == "The submitted message" + reply = "\n\n".join(f"LONG_REPLY_{line}" for line in range(200)) + comms.messaging.send("peer", workspace.viewer, reply) + chat.window.jump_to_latest() + await ui.until(lambda: "LONG_REPLY_199" in ui.painted(chat.window), + why="the real long reply at the latest viewport") + assert "The submitted message" not in ui.painted(chat.window) + assert chat.contains_user_message(sent), "the loaded source still contains the original commit" + assert not chat.send_block_reason + assert chat.prompt.text == "A different unsent draft" + assert editor.document is document and editor.history is history + + +async def test_cancelling_before_worker_start_restores_the_same_editor_without_sending(workspace): + workspace.thread("peer") + async with workspace.app(size=(110, 32)) as ui: + await ui.open_direct("peer", actor=workspace.viewer) + chat = ui.conversation + await ui.write_draft("Different draft") + editor = chat.prompt.prompt_text_area + document, history = editor.document, editor.history + bus = workspace.comms.bus.bus + before = bus.revision() + worker = await chat.submissions.submit(UserInputSubmitted("CANCELLED_BEFORE_START")) + worker.cancel() # No yield between admission and cancellation: the callable never ran. + with pytest.raises(WorkerCancelled): + await asyncio.wait_for(worker.wait(), 10) + await chat.submissions.close() + assert chat.prompt.text == "Different draft\n\nCANCELLED_BEFORE_START" + assert editor.document is document and editor.history is history + assert not chat.submissions.retains_surface + assert not chat.send_block_reason + assert bus.revision() == before + + +async def test_cancelled_bus_write_keeps_unknown_and_never_retries_the_publication(workspace): + workspace.thread("peer") + comms = workspace.comms + async with workspace.app(size=(110, 32)) as ui: + await ui.open_direct("peer", actor=workspace.viewer) + chat = ui.conversation + body = "CANCELLED_DURING_REAL_BUS_WRITE" + await ui.write_draft("") + bus = comms.bus.bus + before = bus.revision() + # Hold SQLite's real writer, not a patched messaging method. Readers still run. + with closing(sqlite3.connect(bus.path)) as blocker: + blocker.execute("BEGIN IMMEDIATE") + try: + worker = await chat.submissions.submit(UserInputSubmitted(body)) + (execution,) = chat.submissions.active + await ui.until(lambda: isinstance(execution.state, SendPending), + why="the original write handed to its worker") + worker.cancel() + await asyncio.wait_for(worker.wait(), 10) + assert isinstance(execution.state, SendBlocked) + assert chat.submissions.retains_surface + assert chat.prompt.text == body + finally: + blocker.rollback() + # The shielded original may finish; observing it is not permission to replay. + await ui.until(lambda: bus.revision() > before, why="the original publication after writer relief") + (sent,) = bus.latest_addressed(frozenset((workspace.viewer,)), frozenset(("peer",))) + assert sent.body == body + assert execution.root == comms.root + assert isinstance(execution.state, SendBlocked) + await chat.submissions.submit(UserInputSubmitted(body)) + await chat.submissions.close() + assert bus.revision() == before + 1 + assert chat.prompt.text == body + assert chat.submissions.retains_surface From a1339a999f009112267a8a4bba46f8df1c7aded4 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 31 Oct 2026 21:29:05 -0400 Subject: [PATCH 04/24] Exercise submission failures through the foreign ACP peer --- tests/support/agents.py | 2 +- tests/support/peers/scripted.py | 6 + tests/textual/test_input_failure_draft.py | 145 ++++++++++++++-------- 3 files changed, 99 insertions(+), 54 deletions(-) diff --git a/tests/support/agents.py b/tests/support/agents.py index 6512cce11..85101ef93 100644 --- a/tests/support/agents.py +++ b/tests/support/agents.py @@ -84,7 +84,7 @@ async def deliver(agent, update: dict, *, session_id: str | None = None): def script(*steps: dict) -> str: - """A prompt for ``peer("scripted")``: steps of {"update": ...}, {"say": ...} or {"wait": name}.""" + """A prompt for ``peer("scripted")``: updates, replies, waits or an SDK error's error object.""" import json return json.dumps(list(steps)) diff --git a/tests/support/peers/scripted.py b/tests/support/peers/scripted.py index 8ef2338d6..0a7c6ec41 100644 --- a/tests/support/peers/scripted.py +++ b/tests/support/peers/scripted.py @@ -3,6 +3,7 @@ - ``{"update": {...}}`` sends that ``session/update`` as written, valid or not. - ``{"say": "text"}`` sends an agent message chunk. - ``{"wait": "name"}`` waits until the test releases ``name`` (``Workspace.release``). +- ``{"error": RequestError.to_error_obj()}`` raises that SDK error through the real request connection. - ``{"ask": "title", "options": ["Allow", "Reject"]}`` asks the user for permission and records the answer as ``{"answered":