diff --git a/src/toad/acp/agent.py b/src/toad/acp/agent.py index db02cf442..cb0c79bbe 100644 --- a/src/toad/acp/agent.py +++ b/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) diff --git a/src/toad/acp/agent_controller.py b/src/toad/acp/agent_controller.py index 91057609d..8b2923135 100644 --- a/src/toad/acp/agent_controller.py +++ b/src/toad/acp/agent_controller.py @@ -11,6 +11,7 @@ from acp.exceptions import RequestError from agent_comms.acp_extension import ( ClearQueueRequest, + CommsRequest, CompactRequest, CoordinationChangedUpdate, InputFailedUpdate, @@ -22,16 +23,19 @@ from agent_comms.acp_failure import ACPFailure, RequestFailure from agent_comms.coordination_errors import Refused 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 @@ -269,9 +273,9 @@ async def _submit( 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 @@ -284,11 +288,11 @@ async def _prompt(self, content, command, authority: ClientSessionRequest, queue 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. @@ -300,15 +304,15 @@ def require(): 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, @@ -321,7 +325,7 @@ def _prompt_failed( 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( @@ -337,19 +341,16 @@ def _prompt_failed( 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): @@ -430,11 +431,11 @@ async def input_delivery(self, *, include_history: bool = False) -> DeliveryView 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") diff --git a/src/toad/acp/client_session.py b/src/toad/acp/client_session.py index dba294c23..d38ec88f0 100644 --- a/src/toad/acp/client_session.py +++ b/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 @@ def require(self): 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): diff --git a/src/toad/acp/queue_attachment.py b/src/toad/acp/queue_attachment.py index c1db617d5..81e60896b 100644 --- a/src/toad/acp/queue_attachment.py +++ b/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__() diff --git a/src/toad/acp/transcript_reader.py b/src/toad/acp/transcript_reader.py index aa5f6fac8..aa7d4d80a 100644 --- a/src/toad/acp/transcript_reader.py +++ b/src/toad/acp/transcript_reader.py @@ -17,6 +17,7 @@ from agent_comms.acp_extension import TranscriptDraftUpdate, TranscriptSnapshotUpdate from agent_comms.route_selection import RouteSelection +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, @@ -69,17 +70,15 @@ async def page( 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 @@ -87,17 +86,15 @@ 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, ) ) @@ -106,6 +103,6 @@ async def updates( 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)) diff --git a/src/toad/agent.py b/src/toad/agent.py index adca76eb7..5cb6de45f 100644 --- a/src/toad/agent.py +++ b/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) diff --git a/src/toad/agent_binding.py b/src/toad/agent_binding.py index 58a134bd0..258fdc2a2 100644 --- a/src/toad/agent_binding.py +++ b/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() diff --git a/src/toad/agent_presentation.py b/src/toad/agent_presentation.py index 55d878b82..127d77205 100644 --- a/src/toad/agent_presentation.py +++ b/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 diff --git a/src/toad/conversation_kind.py b/src/toad/conversation_kind.py index a6e386cde..24c2b4b36 100644 --- a/src/toad/conversation_kind.py +++ b/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(): diff --git a/src/toad/conversation_submission.py b/src/toad/conversation_submission.py index 6f670d5c2..18ba4ffc8 100644 --- a/src/toad/conversation_submission.py +++ b/src/toad/conversation_submission.py @@ -6,7 +6,7 @@ from abc import ABC, abstractmethod from dataclasses import dataclass from functools import partial -from typing import TYPE_CHECKING, ClassVar +from typing import ClassVar from acp.exceptions import RequestError from agent_comms.acp_extension import ( @@ -21,16 +21,14 @@ 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 from toad.core.input_events import UserInputSubmitted from toad.core_event_carrier import CoreEventMessage from toad.input_decision import InputDecision, InputRefused from toad.widgets.prompt import Prompt -if TYPE_CHECKING: - from toad.acp.agent import Agent - from toad.widgets.conversation import Conversation - - @dataclass(frozen=True) class InputSubmission(DeclaredFamily, affix="InputSubmission"): source_text: str @@ -42,17 +40,17 @@ def text(self): @classmethod @abstractmethod - def matches(cls, event: UserInputSubmitted, view: Conversation): ... + def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): ... @abstractmethod async def execute(self, owner: ConversationSubmissions): ... @classmethod - def from_event(cls, event: UserInputSubmitted, view: Conversation): + 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) - if kind.matches(event, view) + if kind.matches(event, owner) ) return kind(event.body) @@ -61,7 +59,7 @@ class NoInputSubmission(InputSubmission): priority = -1 @classmethod - def matches(cls, event: UserInputSubmitted, view: Conversation): + def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): return True async def execute(self, owner: ConversationSubmissions): @@ -72,10 +70,10 @@ class QueueNowInputSubmission(InputSubmission): priority = 100 @classmethod - def matches(cls, event: UserInputSubmitted, view: Conversation): + def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): if event.shell or event.body.strip(): return False - return event.immediate and view.agent_binding.presentation.queue_supported + return event.immediate and owner.view.agent_binding.presentation.queue_supported async def execute(self, owner: ConversationSubmissions): projection = owner.queue_projection @@ -87,7 +85,7 @@ class ShellInputSubmission(InputSubmission): priority = 90 @classmethod - def matches(cls, event: UserInputSubmitted, view: Conversation): + def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): return event.shell and bool(event.body.strip()) async def execute(self, owner): @@ -147,7 +145,7 @@ async def execute(self, owner: ConversationSubmissions): class OrdinaryInputSubmission(AgentInputSubmission): @classmethod - def matches(cls, event: UserInputSubmitted, view: Conversation): + def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): if event.shell: return False return bool(event.body.strip()) @@ -161,11 +159,12 @@ class DeferredInputSubmission(OrdinaryInputSubmission): priority = 70 @classmethod - def matches(cls, event: UserInputSubmitted, view: Conversation): - if not super().matches(event, view): + def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): + if not super().matches(event, owner): return False if event.immediate: return False + view = owner.view return view.turns.owner.busy and view.agent_binding.presentation.queue_supported @property @@ -181,10 +180,10 @@ class ImmediateInputSubmission(OrdinaryInputSubmission): priority = 80 @classmethod - def matches(cls, event: UserInputSubmitted, view: Conversation): - if not super().matches(event, view): + def matches(cls, event: UserInputSubmitted, owner: ConversationSubmissions): + if not super().matches(event, owner): return False - return event.immediate and view.agent_binding.presentation.queue_supported + return event.immediate and owner.view.agent_binding.presentation.queue_supported @property def request(self): @@ -240,17 +239,7 @@ async def send(self): reason = await self.agent.send_prompt(self.submission.text) except RequestError as error: if self.current: - from toad.widgets.conversation import INTERNAL_EROR - from toad.widgets.markdown_note import MarkdownNote - - view.turns.finish_client() - view.prompt.restore_draft(self.submission.text) - await view.post( - MarkdownNote( - INTERNAL_EROR.replace("$ERROR", str(error) or "no details were provided"), - classes="-stop-reason", - ) - ) + await view.submission_request_failed(error, self.submission.text) finally: self.owner.finish(self) if self.current and local: @@ -292,6 +281,16 @@ async def submit(self, event: UserInputSubmitted): def accepts_failure(self, message: CoreEventMessage): return self.view.agent_binding.accepts_failed_input(message) + @property + def queue_evidence(self) -> QueueEvidence: + return QueueEvidence( + self.view.agent_binding.presentation, self.delivering, + self.view.send_block_reason, self.view.contents.notification_targets) + + @property + def queue_summary(self) -> str: + return self.view.queue_source.queue_summary(self.queue_evidence) + @property def queue_projection(self): return self.view.agent_binding.presentation.queue @@ -335,20 +334,16 @@ def reset(self): self.publish_pending() def send_now(self): - return self.view.agent_binding.send_now(self) - - def send_bound_now(self, agent: Agent): - session, scope = agent.session_id, agent.queue_attachment.scope + source = self.queue_source return self.view.run_worker( - partial(self._send_now, agent, session, scope), group="send-queued-now", exclusive=True + partial(self._send_now, source, self.view.agent_binding.presentation, + source.queue_scope(self.view.agent_binding.presentation)), + group="send-queued-now", exclusive=True, ) - async def _send_now(self, agent: Agent, session, scope: QueueScope): + async def _send_now(self, source: QueueSource, presentation, scope: QueueScope) -> None: try: - if await agent.controller.send_now(): - return - except RequestError as error: - if self.view.agent_binding is agent and agent.session_id == session: - self.view.feedback.flash(f"Send now failed: {error}", style="error") - if self.view.agent_binding is agent and agent.queue_attachment.accepts_request(scope): - self.publish_pending() + await source.queue_send_now(presentation, scope) + except (RequestError, QueueActionRefused) as error: + self.view.feedback.flash(f"Send now failed: {error}", style="error") + self.publish_pending() diff --git a/src/toad/core/input_events.py b/src/toad/core/input_events.py index 5f6a606a5..aabe5f52d 100644 --- a/src/toad/core/input_events.py +++ b/src/toad/core/input_events.py @@ -5,7 +5,6 @@ from toad.acp.agent_configuration import ConfigurationSetting from toad.input_history import InputHistory -from toad.navigation_target import NavigationTarget from toad.goal_interaction import GoalInteraction from .events import CoreEvent @@ -68,11 +67,6 @@ class OpenAgentDetails(CoreEvent): identity: str -@dataclass(frozen=True) -class SelectTarget(CoreEvent): - target: NavigationTarget - - @dataclass(frozen=True) class SelectHistoricalIdentity(CoreEvent): name: str diff --git a/src/toad/delivery_failure_view.py b/src/toad/delivery_failure_view.py index 04c2a3cec..cb6a6533c 100644 --- a/src/toad/delivery_failure_view.py +++ b/src/toad/delivery_failure_view.py @@ -91,9 +91,13 @@ def send_block_reason(self) -> str: """Why this view's own send holds compose; empty when it does not.""" return self._send_state.reason + def queue_evidence_changed(self) -> None: + """An unbound delivery surface has no queue consumer to repaint.""" + def _enter_send_state(self, state: SendState) -> None: """The one place a send changes state; compose and the status line follow it.""" self._send_state = state + self.queue_evidence_changed() if self.is_attached: async def refresh() -> None: read_only = await self.composer_read_only() diff --git a/src/toad/messages.py b/src/toad/messages.py index e9f92abf7..3c5ddc543 100644 --- a/src/toad/messages.py +++ b/src/toad/messages.py @@ -7,6 +7,10 @@ from textual.message import Message +class QueueEvidenceChanged(Message): + """Repaint the local queue after its existing evidence changes.""" + + class SendPromptNow(Message): pass diff --git a/src/toad/navigation_target.py b/src/toad/navigation_target.py index 1a4eb6769..063e28bde 100644 --- a/src/toad/navigation_target.py +++ b/src/toad/navigation_target.py @@ -11,6 +11,7 @@ from agent_comms.channel_targets import ThreadTarget as CoreThreadTarget from agent_comms.declared_family import DeclaredFamily +from toad.core.events import CoreEvent from toad.constants import ALL_COMMS_TARGET from toad.navigation_context import NavigationContext, NavigationOwner from toad.conversation_kind import ( @@ -63,6 +64,11 @@ async def open(self, context: NavigationContext) -> str: pass +@dataclass(frozen=True) +class SelectTarget(CoreEvent): + target: NavigationTarget + + @dataclass(frozen=True) class NoTarget(NavigationTarget): name: str = field(default="", init=False) diff --git a/src/toad/notification_status.py b/src/toad/notification_status.py index c4e687879..78ef9da51 100644 --- a/src/toad/notification_status.py +++ b/src/toad/notification_status.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Iterable, Iterator, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import ClassVar, TYPE_CHECKING from agent_comms.message_reference import MessageReference @@ -38,10 +38,18 @@ class NotificationStatus: def title(self) -> str: return "" + @property + def queue_title(self) -> str: + return self.title + @property def details(self) -> DetailLines: return () + @property + def queue_details(self) -> DetailLines: + return self.details + @property def disclosure(self) -> SpanColumns: return SpanColumns(self.details, indent=2) @@ -54,6 +62,21 @@ def blocks(self, shown: Shown, link: LineLink) -> tuple[LineBlock, ...]: if shown.details_open else (summary,)) +@dataclass(frozen=True) +class MessageQueueEntry: + reference: MessageReference + status: NotificationStatus + + @property + def summary(self) -> str: + return f"{self.reference.seq}: {self.status.queue_title}" + + @property + def details(self) -> DetailLines: + return ((Span(f"Original {self.reference.seq}/{self.reference.message_id}", MutedRole),), + *self.status.queue_details) + + @dataclass(frozen=True) class UncheckedNotifications(NotificationStatus): # The row is drawn before Core answers, so the answer changes its text @@ -71,7 +94,12 @@ def for_references(cls, references: Iterable[MessageReference], """Yield nothing when this answer has no matching reference.""" loaded = results.keys() & {(reference.seq, reference.message_id) for reference in references} if loaded: - yield cls(tuple(item for reference in loaded for item in results[reference])) + yield cls.from_notifications(tuple(item for reference in loaded for item in results[reference])) + + @classmethod + def from_notifications(cls, notifications: tuple[MessageNotification, ...]) -> CheckedNotifications: + """Keep a matched empty result distinct from an observed recipient.""" + return cls(notifications) if notifications else EmptyCheckedNotifications() @property def working(self) -> bool: @@ -81,6 +109,24 @@ def working(self) -> bool: def title(self) -> str: return MessageNotification.summary(self.notifications) + @property + def queue_title(self) -> str: + # Keep the leading recipient's actual proof; routing is not arrival. + first = self.notifications[0] + return f"@{first.recipient} {first.label}" + + @property + def queue_details(self) -> DetailLines: + leading = CheckedNotifications.from_notifications(self.notifications[:1]) + return ( + (Span(self.title, MutedRole),), + *((Span(f"Recipient {item.recipient_identity.recipient_lookup}", MutedRole),) + for item in leading.notifications), + *leading.details, + *((Span(text, MutedRole),) + for item in leading.notifications for text in item.answer.detail_texts()), + ) + @property def details(self) -> DetailLines: lines = [] @@ -90,9 +136,19 @@ def details(self) -> DetailLines: line += (Span(" · Read by ", MutedRole), Span(reader.name, NameRole, OpenTarget(reader.name))) lines.append(line) - return (tuple(lines) if lines else ( - (Span("No notification result is recorded for this message. " - "This does not confirm the message was received.", MutedRole),),)) + return tuple(lines) + + +@dataclass(frozen=True) +class EmptyCheckedNotifications(CheckedNotifications): + """The original matched, but no recipient feedback is visible to this viewer.""" + + notifications: tuple[()] = field(default=(), init=False) + queue_title = "No recipient feedback" + + @property + def details(self) -> DetailLines: + return ((Span("No recipient feedback to display.", MutedRole),),) @dataclass(frozen=True) diff --git a/src/toad/owner_preparation.py b/src/toad/owner_preparation.py index 559d689f2..d93a1be92 100644 --- a/src/toad/owner_preparation.py +++ b/src/toad/owner_preparation.py @@ -75,5 +75,6 @@ class OwnerRequestContext: def read_thread_presentation(comms: Comms, name: str) -> ThreadPresentation: - """Read the existing core presentation on a worker, without UI dependencies.""" - return comms.views.thread_presentations((name,)).read(name) + """Read the existing core presentation on a worker, without UI dependencies; only its + published projection leaves the worker, never the acquired delivery evidence.""" + return comms.views.thread_presentations((name,)).read(name).published() diff --git a/src/toad/screens/commander.py b/src/toad/screens/commander.py index cf46144e6..d6988555f 100644 --- a/src/toad/screens/commander.py +++ b/src/toad/screens/commander.py @@ -10,7 +10,8 @@ from agent_comms.coordination_errors import DeferredRead from agent_comms.errors import RelationViolationError, RouteChangedError, ToolRefusedError from agent_comms.thread_identity import ThreadIncarnation -from agent_comms.ui_model.observation import ReadCommanderUnit, UnreadableHistory +from agent_comms.order_tools import CommanderUnitRead +from agent_comms.ui_model.observation import ReadOperation, UnreadableHistory from agent_comms.work_branches import WorkBranchError from metaclass_registry.mro_dispatch import MroDispatch, handles from rich.cells import cell_len @@ -130,10 +131,10 @@ async def read_detail(self) -> None: access = self.app.coordination_access try: observed = await access.preparation.run_thread(access.observation.observed) - request = ReadCommanderUnit( - str(observed.service.root), self.commander, card.unit, - tuple(dict.fromkeys(order.order for order in card.orders)), - ) + request = ReadOperation(str(observed.service.root), CommanderUnitRead( + commander=self.commander, unit=card.unit, + orders=tuple(dict.fromkeys(order.order for order in card.orders)), + )) detail = await observed.process.read(request) except ( DeferredRead, RelationViolationError, RouteChangedError, diff --git a/src/toad/screens/comms.py b/src/toad/screens/comms.py index 4a5ba97c4..2a21c160b 100644 --- a/src/toad/screens/comms.py +++ b/src/toad/screens/comms.py @@ -23,11 +23,10 @@ from toad.conversation_kind import ConversationKind from toad.widgets.comms_chat import CommsChatView from toad.core.input_events import SelectHistoricalIdentity -from toad.core.input_events import SelectTarget from toad.widgets.comms_sidebar import CoordinationStatus, CommsSidebar from toad.session_tracker import SidebarState from toad.navigation_context import NavigationContext -from toad.navigation_target import HistoryTarget, NavigationTarget +from toad.navigation_target import HistoryTarget, NavigationTarget, SelectTarget from toad.widgets.session_tabs import SessionsTabs from toad.widgets.side_bar import SideBar, ThreadSidebar, TabHistoryControls from toad.widgets.thread_comms import RelationshipSort, ThreadCommsSidebar diff --git a/src/toad/screens/goal_details.py b/src/toad/screens/goal_details.py index 905851b41..8a6974487 100644 --- a/src/toad/screens/goal_details.py +++ b/src/toad/screens/goal_details.py @@ -11,7 +11,7 @@ from toad.widget_actions import DismissAction from textual.widgets import Button, Static -from toad.core.input_events import SelectTarget +from toad.navigation_target import SelectTarget from toad.goal_interaction import GoalSession from toad.project_path_owner import ProjectPathOwner from toad.widgets.goal_text import GoalText diff --git a/src/toad/screens/goal_edit.py b/src/toad/screens/goal_edit.py index 89a9bb795..79f72a6b0 100644 --- a/src/toad/screens/goal_edit.py +++ b/src/toad/screens/goal_edit.py @@ -16,7 +16,7 @@ from toad.core.input_events import UserInputSubmitted from toad.core_event_carrier import CoreEventReceiver, CoreEventMessage from metaclass_registry.mro_dispatch import handles -from toad.widgets.channel_prompt import ChannelPrompt +from toad.widgets.channel_prompt import DraftPrompt class GoalEdit(CoreEventReceiver, DismissibleScreen[str | None]): @@ -43,7 +43,7 @@ def __init__( self.goal = goal self.on_save = on_save self.saving = False - self.editor = ChannelPrompt(project=project) + self.editor = DraftPrompt(project=project) self.editor.set_mention_candidates(candidates) def compose(self) -> ComposeResult: diff --git a/src/toad/screens/historical_sessions.py b/src/toad/screens/historical_sessions.py index 2395cc0a4..d43970c8e 100644 --- a/src/toad/screens/historical_sessions.py +++ b/src/toad/screens/historical_sessions.py @@ -9,9 +9,12 @@ from agent_comms.comms import Comms from agent_comms.coordination_errors import StaleRevision from metaclass_registry.mro_dispatch import handles -from agent_comms.ui_model.observation import ReadTranscript, ReadRecordedNotifications +from agent_comms.notification_reads import RecordedNotificationRead +from agent_comms.order_tools import TranscriptPageRead +from agent_comms.ui_model.observation import ReadTranscript from toad.core_event_carrier import CoreEventMessage, CoreEventReceiver -from agent_comms.historical_views import HistoricalThread +from agent_comms.historical_views import HistoricalThread, HistorySourceKey +from agent_comms.transcripts import RecordedPageSource from textual import on, work from textual.app import ComposeResult from textual.containers import Vertical @@ -28,8 +31,8 @@ class RecordedContents(Contents): """The notification subscriber reads the same archive as its history.""" - def __init__(self, source: str, **kwargs): - self.notification_request = partial(ReadRecordedNotifications, source=source) + def __init__(self, source: HistorySourceKey, **kwargs): + self.notification_operation = partial(RecordedNotificationRead, source=source) super().__init__(**kwargs) @@ -121,8 +124,8 @@ async def selected(self, event: Select.Changed) -> None: access, root = self.app.coordination_access, str(self.comms.root) async def load(**kwargs): - snapshot = await access.transcripts.read(ReadTranscript( - root, item.thread.name, historical_source=item.source.key, **kwargs)) + snapshot = await access.transcripts.read(ReadTranscript(root, TranscriptPageRead( + thread=item.thread.name, page_source=RecordedPageSource(item.source.key), **kwargs))) return snapshot.page try: diff --git a/src/toad/screens/main.py b/src/toad/screens/main.py index 398bb77fb..61565a6e9 100644 --- a/src/toad/screens/main.py +++ b/src/toad/screens/main.py @@ -14,6 +14,7 @@ from metaclass_registry.mro_dispatch import handles from agent_comms.route_selection import RouteSelection from agent_comms.repo import Checkout +from agent_comms.transcripts import TranscriptCursor from textual import containers, getters, on from textual.app import ComposeResult from textual.binding import ModeBinding @@ -36,13 +37,13 @@ from toad.core import events as core_events from toad.core import input_events, session_requests from toad.core.events import SessionChangedEvent -from toad.core.input_events import SelectTarget from toad.format_path import format_path from toad.navigation_context import NavigationContext from toad.navigation_target import ( DirectTarget, FeedTarget, NavigationTarget, + SelectTarget, ThreadTarget, ) from toad.screens.mcp_inventory import MCPInventoryScreen @@ -225,11 +226,11 @@ def focus_sidebar_return(self, screen: MainScreen) -> None: screen.screen.set_focus(target, scroll_visible=False) def read_visible(self, screen: MainScreen) -> PaintedRead: - """Only mounted content can supply a painted reader cursor.""" + return screen.conversation.painted_read(partial(self.read_work, screen)) + + def read_work(self, screen: MainScreen, through: TranscriptCursor) -> ReadWork: + """Only a painted cursor reaches route selection and the write worker.""" conversation = screen.conversation - through = conversation.visible_read_cursor - if through is None: - return NothingToRead() app = screen.app access = app.coordination_access identity = screen.thread_identity diff --git a/src/toad/screens/queue.py b/src/toad/screens/queue.py new file mode 100644 index 000000000..816ef9ece --- /dev/null +++ b/src/toad/screens/queue.py @@ -0,0 +1,71 @@ +"""Inspect the captured queue source through the actual conversation view.""" + +from collections.abc import Awaitable, Callable + +from acp.exceptions import RequestError +from agent_comms.acp_extension import QueueScope +from textual import on +from textual.app import ComposeResult +from textual.containers import HorizontalGroup, Vertical, VerticalScroll +from textual.widgets import Button, Static + +from toad.acp.queue_attachment import QueueActionRefused +from toad.agent_presentation import AgentPresentation +from toad.line_blocks import SpanColumns +from toad.screens.modal import DismissibleScreen +from toad.submission_queue import QueueEvidence, QueueSource +from toad.widgets.tool_content import ToolContent + + +class QueueInspection(DismissibleScreen[None]): + """One screen captures the source, presentation and scope it actually displays.""" + + DEFAULT_CSS = """ + QueueInspection { align: center middle; background: $background 40%; } + QueueInspection > Vertical { width: 85%; max-width: 110; height: 75%; padding: 1 2; background: $surface; } + QueueInspection VerticalScroll { height: 1fr; } + QueueInspection HorizontalGroup { height: auto; } + QueueInspection #queue-error { height: auto; color: $error; } + """ + + def __init__(self, source: QueueSource, evidence: QueueEvidence, changed: Callable[[], None]) -> None: + super().__init__() + self.source, self.evidence = source, evidence + self.scope = source.queue_scope(evidence.presentation) + self.changed = changed + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static("Queue and delivery") + with VerticalScroll(): + details = ToolContent() + details.update((SpanColumns(self.source.queue_details(self.evidence)),)) + yield details + yield Static("Individual edit/delete/reorder requires ID-targeted, revision-checked backend support.") + yield Static("", id="queue-error") + with HorizontalGroup(): + available = self.source.queue_actions_available(self.evidence.presentation) + yield Button("Send now", id="queue-send-now", disabled=not available) + yield Button("Clear queued inputs", id="queue-clear", disabled=not available) + yield Button("Close", id="queue-close") + + async def apply_action(self, action: Callable[[AgentPresentation, QueueScope], Awaitable[None]]) -> None: + try: + await action(self.evidence.presentation, self.scope) + except (QueueActionRefused, RequestError) as error: + self.query_one("#queue-error", Static).update(str(error)) + return + self.changed() + self.dismiss() + + @on(Button.Pressed, "#queue-send-now") + async def send_now(self) -> None: + await self.apply_action(self.source.queue_send_now) + + @on(Button.Pressed, "#queue-clear") + async def clear_queue(self) -> None: + await self.apply_action(self.source.queue_clear) + + @on(Button.Pressed, "#queue-close") + def close_queue(self) -> None: + self.dismiss() diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py index 340a19ffe..47c3f7a94 100644 --- a/src/toad/session_presentation.py +++ b/src/toad/session_presentation.py @@ -24,6 +24,7 @@ from toad.widgets.conversation import Conversation from toad.widgets.history_anchor import ParkedPosition from toad.widgets.prompt import MailboxPrompt +from toad.submission_queue import BusQueueSource, NativeQueueSource if TYPE_CHECKING: from toad.app import ToadApp @@ -141,7 +142,7 @@ async def release_binding( async def attach_binding(self, conversation: Conversation) -> None: ... -class OperationalSessionSources: +class OperationalSessionSources(NativeQueueSource): """Own existing operational resources while the view is detached.""" def __init__(self) -> None: @@ -181,7 +182,7 @@ def prompt(self, conversation: Conversation): async def submit(self, owner, event): from toad.conversation_submission import InputSubmission - return await InputSubmission.from_event(event, owner.view).execute(owner) + return await InputSubmission.from_event(event, owner).execute(owner) def input_decision(self, conversation: Conversation) -> InputDecision: agent = conversation.agent_binding @@ -252,7 +253,7 @@ async def close(self) -> None: release.push_async_callback(agent.stop) -class MailboxSessionSources(OperationalSessionSources): +class MailboxSessionSources(BusQueueSource, OperationalSessionSources): """CLI execution receives user messages through Core's inbox, not ACP.""" def activity_row(self, conversation: Conversation): @@ -262,7 +263,7 @@ def agent_info(self, conversation: Conversation) -> str: return conversation.thread_activity.row.presentation.title def prompt(self, conversation: Conversation): - return MailboxPrompt(project=conversation.project_path) + return MailboxPrompt(project=conversation.project_path, submissions=conversation.submissions) def thread_presented(self, conversation: Conversation) -> None: conversation.run_worker( diff --git a/src/toad/session_scope.py b/src/toad/session_scope.py index 5137fa9c5..2871661f2 100644 --- a/src/toad/session_scope.py +++ b/src/toad/session_scope.py @@ -15,6 +15,8 @@ from agent_comms.errors import RelationViolationError from agent_comms.route_selection import RouteSelection 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 from agent_comms.ui_model.observation import ( ReadTranscript, ReadTranscriptUpdate, @@ -35,7 +37,6 @@ if TYPE_CHECKING: from toad.widget_actions import DeclaredWidgetActions from agent_comms.messages import Message - from agent_comms.transcripts import TranscriptCursor from agent_comms.thread_presentation import ThreadPresentation from textual.command import Provider from textual.dom import DOMNode @@ -83,10 +84,9 @@ def observe( def snapshot_request( self, *, - before: TranscriptCursor | None = None, - after: TranscriptCursor | None = None, - through: TranscriptCursor | None = None, - historical_source: str | None = None, + position: PagePosition = LatestPage(), + limit: PageLimit = AtFrontier(), + page_source: PageSource = LivePageSource(), ) -> ReadTranscript: ... @abstractmethod @@ -94,7 +94,7 @@ def update_request( self, previous: TranscriptPublicationState, *, - after: TranscriptCursor | None = None, + position: PagePosition = LatestPage(), ) -> ReadTranscriptUpdate: ... @abstractmethod @@ -162,27 +162,26 @@ def observe( def snapshot_request( self, *, - before: TranscriptCursor | None = None, - after: TranscriptCursor | None = None, - through: TranscriptCursor | None = None, - historical_source: str | None = None, + position: PagePosition = LatestPage(), + limit: PageLimit = AtFrontier(), + page_source: PageSource = LivePageSource(), ) -> ReadTranscript: - return ReadTranscript( - self.root, - self.name, - before=before, - after=after, - through=through, - historical_source=historical_source, - ) + return ReadTranscript(self.root, TranscriptPageRead( + thread=self.name, + position=position, + limit=limit, + page_source=page_source, + )) def update_request( self, previous: TranscriptPublicationState, *, - after: TranscriptCursor | None = None, + position: PagePosition = LatestPage(), ) -> ReadTranscriptUpdate: - return ReadTranscriptUpdate(self.root, self.name, previous=previous, after=after) + return ReadTranscriptUpdate( + self.root, TranscriptPageRead(thread=self.name, position=position), previous=previous + ) def mailbox_target(self) -> ThreadTarget: return ThreadTarget(self.name) @@ -262,10 +261,9 @@ def unavailable(self) -> Never: def snapshot_request( self, *, - before: TranscriptCursor | None = None, - after: TranscriptCursor | None = None, - through: TranscriptCursor | None = None, - historical_source: str | None = None, + position: PagePosition = LatestPage(), + limit: PageLimit = AtFrontier(), + page_source: PageSource = LivePageSource(), ) -> Never: self.unavailable() @@ -273,7 +271,7 @@ def update_request( self, previous: TranscriptPublicationState, *, - after: TranscriptCursor | None = None, + position: PagePosition = LatestPage(), ) -> Never: self.unavailable() diff --git a/src/toad/submission_queue.py b/src/toad/submission_queue.py new file mode 100644 index 000000000..8c3d57bf9 --- /dev/null +++ b/src/toad/submission_queue.py @@ -0,0 +1,136 @@ +"""Source-owned queue capabilities and captured evidence, independent of screens.""" +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Callable +from dataclasses import dataclass +from functools import cached_property +from typing import ClassVar + +from agent_comms.acp_extension import NoQueueScope, QueueItem, QueueScope +from agent_comms.message_reference import MessageReference +from agent_comms.notification_evidence import NotificationWindow + +from toad.acp.queue_attachment import QueueActionRefused +from toad.agent_presentation import AgentPresentation +from toad.line_blocks import MutedRole, Span +from toad.notification_status import DetailLines, MessageQueueEntry +from toad.widgets.message_notifications import SourceNotifications + + +@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[SourceNotifications, ...]] + + @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) >= NotificationWindow.window_limit: + return tuple(entries.values()) + return tuple(entries.values()) + + +class QueueSource(ABC): + """The source selects its capability; the actual presentation owns native facts.""" + + preview_limit: ClassVar[int] = 3 + + @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: ... + + def queue_summary(self, evidence: QueueEvidence) -> str: + return "Queue unavailable" + + def queue_details(self, evidence: QueueEvidence) -> DetailLines: + return ((Span("No queue observation is available for this source.", MutedRole),),) + + +class ReadOnlyQueueSource(QueueSource): + """Delivery evidence grants no native queue mutation authority.""" + + def queue_scope(self, presentation: AgentPresentation) -> QueueScope: + return NoQueueScope() + + def queue_actions_available(self, presentation: AgentPresentation) -> bool: + return False + + async def queue_send_now(self, presentation: AgentPresentation, scope: QueueScope) -> None: + raise QueueActionRefused("Send now unavailable: this source has no native queue authority") + + async def queue_clear(self, presentation: AgentPresentation, scope: QueueScope) -> None: + raise QueueActionRefused("Clear unavailable: BUS messages cannot be withdrawn through an ACP queue") + + +class NativeQueueSource(QueueSource): + def queue_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"), + ): + if texts: + parts.append(f"{label} ({len(texts)}{detail}): " + " · ".join(texts[:self.preview_limit])) + return " | ".join(parts) + + def queue_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) + + def queue_scope(self, presentation: AgentPresentation) -> QueueScope: + return presentation.queue_scope + + def queue_actions_available(self, presentation: AgentPresentation) -> bool: + return presentation.queue_actions_available + + async def queue_send_now(self, presentation: AgentPresentation, scope: QueueScope) -> None: + presentation.require_queue_action(scope) + await presentation.agent.controller.send_now() + + async def queue_clear(self, presentation: AgentPresentation, scope: QueueScope) -> None: + presentation.require_queue_action(scope) + await presentation.agent.controller.clear_queue() + + +class BusQueueSource(ReadOnlyQueueSource): + def queue_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) + + def queue_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(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 6738dda57..0aae526a0 100644 --- a/src/toad/transcript_preparation.py +++ b/src/toad/transcript_preparation.py @@ -219,13 +219,6 @@ def __eq__(self, other): and self.start == other.start and self.stop == other.stop) -@dataclass(frozen=True) -class PageRevision[Cursor = TranscriptCursor | None](WorkRevision): - through: Cursor | None - request: PageRequest[Cursor] - size: ClassVar[int] = 512 - - @dataclass(frozen=True) class FilteredTranscriptBatch: prepared: PreparedTranscriptPage @@ -289,7 +282,7 @@ def close(self) -> None: pass -class PreparedPageSource(ABC): +class PreparedPageSource: """One lifetime-owned source of prepared pages, independent of widget paging.""" runtime: PreparationRuntime @@ -440,7 +433,7 @@ def execute(self) -> PreparedTranscriptPage: class TranscriptNoProgress(ValueError): """A page read returned no new edge while its end still claimed more.""" - def __init__(self, request: PageRequest, through: TranscriptCursor | None, page: TranscriptPage) -> None: + def __init__(self, request: PageRequest, through: TranscriptCursor, page: TranscriptPage) -> None: super().__init__( f"Transcript history made no cursor progress: {type(request.end).__name__} " f"from {request.cursor!r} through {through!r} returned before={page.before!r} " @@ -449,7 +442,7 @@ def __init__(self, request: PageRequest, through: TranscriptCursor | None, page: @dataclass(frozen=True) -class TranscriptPageWork[Cursor = TranscriptCursor | None](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]): +class TranscriptPageWork[Cursor = TranscriptCursor](ReusableWork[PreparedTranscriptPage], ScopedWork[PreparedTranscriptPage]): scope: PreparationScope loader: Callable[..., Awaitable[TranscriptPage]] through: Cursor @@ -459,13 +452,13 @@ class TranscriptPageWork[Cursor = TranscriptCursor | None](ReusableWork[Prepared @property def work_key(self) -> WorkKey: # An older page is immutable as its source grows; a newer read ends at the source end. - return WorkKey(type(self), PageRevision(self.request.end.work_cursor(self.through), self.request), self.scope) + return WorkKey(type(self), self.request.end.work_revision(self.request, self.through), self.scope) def result_size(self, result: PreparedTranscriptPage) -> int: return result.retained_bytes async def execute( - self: TranscriptPageWork[TranscriptCursor | None], runtime: PreparationRuntime, key: WorkKey, + self: TranscriptPageWork[TranscriptCursor], runtime: PreparationRuntime, key: WorkKey, ) -> PreparedResult[PreparedTranscriptPage]: request = self.request page = await request.end.load_page(self.loader, request.cursor, self.through) diff --git a/src/toad/transcript_publication.py b/src/toad/transcript_publication.py index 75f7bc1ec..e5f125bd0 100644 --- a/src/toad/transcript_publication.py +++ b/src/toad/transcript_publication.py @@ -28,9 +28,15 @@ from agent_comms.thread_presentation import ThreadPresentation from agent_comms.transcript_events import TranscriptCoverage from agent_comms.transcripts import ( + AtFrontier, + LatestPage, + PageLimit, + PagePosition, + Through, TranscriptCursor, TranscriptPage, TranscriptReadIdentity, + UnreadTranscriptCursor, ) from agent_comms.ui_model.observation import ( NotPublished, @@ -272,7 +278,7 @@ async def publish_snapshot(self, snapshot: TranscriptSnapshotUpdate) -> bool: if bound != page.after: self.owner.dirty = self.owner.commit_required = True page = await self.owner.get_transcript_page( - after=snapshot.identity.after, through=bound + position=snapshot.identity.position.later_only(), limit=Through(bound) ) # Preserve this cut's original cohort across the bounded read. if not await self.application_current(): @@ -425,7 +431,7 @@ async def publish_source_page(self, page: TranscriptPage) -> bool: prepared.history, TranscriptCoverage(page.events).native_inputs, prepared.outgoing, - identity=replace(self.identity, through=page.after), + identity=replace(self.identity, limit=Through(page.after)), ) retired = retirement_candidates(contents.children, evidence) if not plan.permits(view, retired): @@ -767,7 +773,7 @@ def __init__(self) -> None: self.generation = 0 self.dirty = False self.commit_required = False - self.displayed_cursor: TranscriptCursor | None = None + self.displayed_cursor: TranscriptCursor = UnreadTranscriptCursor() self.current_read: TranscriptPublicationState = NotPublished() self.worker: Worker[None] | None = None self.source_requests = SourcePublicationRequests(self) @@ -799,16 +805,13 @@ def readable(self) -> bool: return self.binding.thread_scope.readable async def get_transcript_page( - self, *, before=None, after=None, - through=None + self, *, position: PagePosition = LatestPage(), limit: PageLimit = AtFrontier(), ) -> TranscriptPage: frame = self.binding.capture(self, BoundFrame) scope = self.binding.thread_scope if frame is None: raise ValueError("Transcript paging requires a bound thread") - page = await frame.binding.reader.page( - scope, before=before, after=after, through=through - ) + page = await frame.binding.reader.page(scope, position=position, limit=limit) if self.binding.thread_scope != scope: raise ValueError("Thread source changed while reading transcript") return page @@ -850,10 +853,10 @@ async def publish(self, kind: type[TranscriptPublication], *args: ThreadPresenta return await publication.publish() return True - def changed(self, cursor: TranscriptCursor | None) -> None: + def changed(self, cursor: TranscriptCursor) -> None: self.invalidate_publications() self.dirty = True - if cursor is None: + if cursor.requires_commit(): self.commit_required = True self.binding.request(self) @@ -897,7 +900,7 @@ async def refresh_cursor(self) -> TranscriptCursor: lambda: ready.set_result(self.displayed_cursor) ) cursor = await ready - if cursor is None: + if cursor.requires_commit(): raise TranscriptChanged() return cursor @@ -926,7 +929,7 @@ async def close(self) -> None: await self.suspend() finally: self._unbind_view() - self.displayed_cursor = None + self.displayed_cursor = UnreadTranscriptCursor() def _close_documents(self) -> None: documents, self.documents = self.documents, () @@ -1007,7 +1010,7 @@ async def changing_source(self, origin: SessionOrigin): self.current_read = NotPublished() self._close_documents() self.dirty = self.commit_required = False - self.displayed_cursor = None + self.displayed_cursor = UnreadTranscriptCursor() worker, self.worker = self.worker, None if worker is not None: worker.cancel() @@ -1041,4 +1044,4 @@ async def detach(self, view: Conversation) -> None: self._close_documents() self.documents = tuple(documents) self._unbind_view() - self.displayed_cursor = None + self.displayed_cursor = UnreadTranscriptCursor() diff --git a/src/toad/widgets/channel_prompt.py b/src/toad/widgets/channel_prompt.py index a17b86465..b76262f18 100644 --- a/src/toad/widgets/channel_prompt.py +++ b/src/toad/widgets/channel_prompt.py @@ -118,3 +118,13 @@ def dismiss_control(self) -> None: self.mention_list.display = False else: super().dismiss_control() + + +class DraftPrompt(ChannelPrompt): + """Mention-enabled editing without a conversation or delivery authority.""" + + def compose_delivery_controls(self) -> ComposeResult: + return () + + def _update_queue_summary(self) -> None: + pass diff --git a/src/toad/widgets/committed_presentation.py b/src/toad/widgets/committed_presentation.py index 54100346f..82267b020 100644 --- a/src/toad/widgets/committed_presentation.py +++ b/src/toad/widgets/committed_presentation.py @@ -367,7 +367,7 @@ async def read_snapshot[Snapshot]( self, read: Callable[..., Awaitable[Snapshot]], history: CommittedHistory, ) -> Snapshot: try: - return await read(after=history.page_source.through) + return await read(position=history.page_source.through.later_page()) except TranscriptChanged: # A replaced source cannot extend the old reader's source range. return await read() diff --git a/src/toad/widgets/comms_chat.py b/src/toad/widgets/comms_chat.py index 1caaeca1c..0999a516c 100644 --- a/src/toad/widgets/comms_chat.py +++ b/src/toad/widgets/comms_chat.py @@ -7,6 +7,7 @@ from toad.widgets.wire_history import WireHistory from toad.message_rows import IrcMessageStyle from toad.wire_history import WireSource +from toad.submission_queue import QueueSource, ReadOnlyQueueSource from toad.transcript_publication import NoView, ViewBinding import re @@ -61,6 +62,7 @@ class HistoryLoading(ConversationBlock, Static): class CommsChatView(Conversation): """A wire-backed conversation using Toad's normal transcript primitives.""" + queue_source: QueueSource = ReadOnlyQueueSource() def __init__( self, @@ -96,7 +98,7 @@ def compose(self) -> ComposeResult: with containers.Vertical(id="prompt-stack"): yield from self.conversation.compose_activity(self) yield self.throbber_above - yield self.conversation.prompt(self.project_path).data_bind( + yield self.conversation.prompt(self.project_path, submissions=self.submissions).data_bind( project_path=Conversation.project_path, working_directory=Conversation.working_directory, agent_info=Conversation.agent_info, @@ -115,10 +117,9 @@ async def initialize_view(self) -> None: ) comms = shared await self.app.coordination_access.current_sidebar() - self.message_history.initialize_source(WireSource( - comms, self.conversation, self.project_path, self._me, - IrcMessageStyle(), - )) + source = WireSource(comms, self.conversation, self.project_path, self._me, IrcMessageStyle()) + self.queue_source = source + self.message_history.initialize_source(source) self.refresh_identity(comms.root) self.prepare_prompt() self.window.anchor() diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py index 69493d711..0813a3e13 100644 --- a/src/toad/widgets/conversation.py +++ b/src/toad/widgets/conversation.py @@ -11,8 +11,11 @@ from typing import TYPE_CHECKING, Callable from acp import schema as acp_protocol +from acp.exceptions import RequestError +from toad.widgets.markdown_note import MarkdownNote from agent_comms.channel_targets import CommandTarget from agent_comms.repo import Checkout +from agent_comms.transcripts import TranscriptCursor from agent_comms.goal_presentation import GoalExecution, IdleGoalExecution from agent_comms.goals import NoGoal from metaclass_registry.mro_dispatch import MroDispatch, handles @@ -42,6 +45,9 @@ from toad.acp.status import StopReason from toad.agent import AgentBase from toad.agent_binding import AgentConnection, NoAgentConnection +from toad.messages import QueueEvidenceChanged +from toad.submission_queue import QueueSource +from toad.screens.queue import QueueInspection from toad.agent_schema import AgentDefinition from toad.answer import Answer from toad.conversation_actions import ConversationAction, WindowAction @@ -101,7 +107,7 @@ ) from toad.widgets.menu import Menu from toad.widgets.message_divider import StickyHeader -from toad.widgets.message_notifications import HistoryNotifications, SourceNotifications +from toad.widgets.message_notifications import HistoryNotifications, NothingToRead, PaintedRead, SourceNotifications from toad.widgets.native_history import NativeHistory from toad.widgets.note import Note from toad.widgets.observed_thread_activity import ( @@ -752,7 +758,6 @@ async def session_ready(self) -> None: ): welcome = agent_data.welcome if welcome is not None: - from toad.widgets.markdown_note import MarkdownNote await self.post(MarkdownNote(welcome)) if ( @@ -844,18 +849,24 @@ def keyboard_scopes(self) -> tuple[DeclaredWidgetActions, ...]: window = getters.query_one(Window) cursor = getters.query_one(CursorContainer) - @property - def visible_read_cursor(self): - """Only a painted, unfiltered tail can acknowledge the full transcript.""" + async def submission_request_failed(self, error: RequestError, text: str) -> None: + self.turns.finish_client() + self.prompt.restore_draft(text) + await self.post(MarkdownNote( + INTERNAL_EROR.replace("$ERROR", str(error) or "no details were provided"), + classes="-stop-reason", + )) + + def painted_read(self, acknowledge: Callable[[TranscriptCursor], PaintedRead]) -> PaintedRead: + """Decide before constructing a worker or opening a read/write context.""" if not self.is_mounted or self.visible_categories != all_categories(): - return None + return NothingToRead() window = self.window - through = self.transcript.displayed_cursor - if through is None or not window.follows_tail: - return None + if not window.follows_tail: + return NothingToRead() if any(history.blocks_visible_read for history in window.histories): - return None - return through + return NothingToRead() + return self.transcript.displayed_cursor.painted_read(acknowledge, NothingToRead()) def watch_visible_categories( self, @@ -1329,7 +1340,6 @@ async def agent_turn_over(self, stop_reason: type[StopReason] | None) -> None: self._turn_count += 1 if stop_reason is not None and stop_reason.note: - from toad.widgets.markdown_note import MarkdownNote await self.post( MarkdownNote( @@ -1656,22 +1666,21 @@ async def compact_context(self, instructions: str | None) -> None: await self.agent_binding.compact_context(instructions) self.feedback.flash("Context compacted", style="success") - def open_queue_menu(self) -> None: - """Remote edits require exact input IDs and backend revision-CAS support.""" - if self.submissions.queue_projection.items: - self._queue_edit_unavailable() - - def _queue_edit_unavailable(self) -> None: - self.feedback.flash( - "Remote queue editing unavailable: requires ID-targeted, revision-checked backend support", - style="error", - ) + @property + def queue_source(self) -> QueueSource: + return self.operational_sources - def _on_queue_choice(self, action: str | None) -> None: - # A stale menu callback must not edit the draft or mutate the remote - # queue. Text/index matching cannot identify a sent input safely. - if action: - self._queue_edit_unavailable() + def queue_evidence_changed(self) -> None: + self.submissions.publish_pending() + + @on(QueueEvidenceChanged) + def queue_observation_changed(self, event: QueueEvidenceChanged) -> None: + event.stop() + self.queue_evidence_changed() + + def open_queue_menu(self) -> None: + self.app.push_screen(QueueInspection( + self.queue_source, self.submissions.queue_evidence, self.submissions.publish_pending)) @handles(PreferenceChanged) async def _settings_changed(self, event: PreferenceChanged) -> None: diff --git a/src/toad/widgets/linked_rows.py b/src/toad/widgets/linked_rows.py index c261d84a0..f63cd73e0 100644 --- a/src/toad/widgets/linked_rows.py +++ b/src/toad/widgets/linked_rows.py @@ -18,13 +18,13 @@ from toad.block_navigation import ConversationEntry from toad.core.events import LinkContextChanged -from toad.core.input_events import SelectHistoricalIdentity, SelectTarget +from toad.core.input_events import SelectHistoricalIdentity from toad.core_event_carrier import CoreEventReceiver from toad.frame_presentation import AnimationEffects from toad.line_blocks import LineLink, OpenHistoricalTarget, OpenReference, OpenTarget from toad.menus import MenuItem from toad.message_rows import MessageRow -from toad.navigation_target import NavigationTarget +from toad.navigation_target import NavigationTarget, SelectTarget from toad.project_link import ProjectLink from toad.project_path_owner import ProjectPathOwner from toad.widgets.transcript_lines import Lines diff --git a/src/toad/widgets/message_notifications.py b/src/toad/widgets/message_notifications.py index d5a2fe5d5..7a1bee21a 100644 --- a/src/toad/widgets/message_notifications.py +++ b/src/toad/widgets/message_notifications.py @@ -3,7 +3,7 @@ from __future__ import annotations from abc import ABC, abstractmethod -from collections.abc import Awaitable +from collections.abc import Awaitable, Iterable from dataclasses import dataclass, replace from typing import TYPE_CHECKING @@ -11,10 +11,14 @@ from textual.content import Content from textual.widgets import Collapsible +from agent_comms.notification_reads import LiveNotificationRead from agent_comms.ui_model.observation import ReadNotifications +from agent_comms.read_basis import DisplayIdentity, NoDisplaySelection +from agent_comms.message_reference import MessageReference +from toad.messages import QueueEvidenceChanged from toad.notification_status import ( - CheckedNotifications, NotificationStatus, UnavailableNotifications, UncheckedNotifications, + CheckedNotifications, MessageQueueEntry, NotificationStatus, UnavailableNotifications, UncheckedNotifications, ) if TYPE_CHECKING: from toad.widgets.tool_content import ToolContent @@ -110,7 +114,7 @@ class HistoryNotifications: """Visible history rows subscribe to Core; paging never decides when outcomes changed.""" NOTIFICATION_GROUP = "message-notifications" - notification_request = ReadNotifications + notification_operation = LiveNotificationRead # Core feedback for every visible bus message, native or wire. @@ -119,14 +123,20 @@ def check_notifications(self) -> None: references = tuple(dict.fromkeys( reference for fragment in fragments for reference in fragment.notification_references)) self.subscribe_notifications(references) + self.post_message(QueueEvidenceChanged()) - def notification_read(self, references) -> ReadNotifications: + def notification_read(self, references: Iterable[MessageReference]) -> ReadNotifications: from toad.project_path_owner import ProjectPathOwner root = ProjectPathOwner.containing(self).coordination_root if root is None: raise ValueError("Notification observation requires a bound thread root") - return self.notification_request(root, references) + references = tuple(references) + wanted = frozenset(references) + display = next((target.notification_identity_for(wanted) + for target in self.notification_targets() + if wanted.intersection(target.notification_references)), NoDisplaySelection()) + return ReadNotifications(root, self.notification_operation(references), display=display) def subscribe_notifications(self, references) -> None: access = self.app.coordination_access @@ -144,6 +154,7 @@ def notifications_observed(self, results) -> None: async def _show_notifications(self, results): for target in self.notification_targets(): await target.show_notifications(results) + self.post_message(QueueEvidenceChanged()) def notifications_failed(self, failure: Exception) -> None: if self.can_draw_on_screen: @@ -153,6 +164,7 @@ def notifications_failed(self, failure: Exception) -> None: async def _show_notification_error(self, status): for target in self.notification_targets(): await target.show_notification_status(status) + self.post_message(QueueEvidenceChanged()) def on_mount(self) -> None: self.call_after_refresh(self.check_notifications) @@ -170,6 +182,16 @@ def on_show(self) -> None: class SourceNotifications: notification_view: HistoryNotifications + def queue_entries(self) -> Iterable[MessageQueueEntry]: + for fragment in self.visible_fragments(): + yield from fragment.queue_entries() + + def notification_identity_for(self, references: frozenset[MessageReference]) -> DisplayIdentity | NoDisplaySelection: + """Use the visible BUS fragments' construction-time display identity.""" + return next((fragment.notification_identity + for fragment in reversed(tuple(self.visible_fragments())) + if references.intersection(fragment.notification_references)), NoDisplaySelection()) + def rows_published(self) -> None: super().rows_published() self.notification_view.call_after_refresh(self.notification_view.check_notifications) diff --git a/src/toad/widgets/presentation_window.py b/src/toad/widgets/presentation_window.py index 013505056..5320891a3 100644 --- a/src/toad/widgets/presentation_window.py +++ b/src/toad/widgets/presentation_window.py @@ -13,6 +13,7 @@ from agent_comms.message_page import PageBounds from agent_comms.transcripts import TranscriptCursor +from toad.work_preparation import WorkRevision if TYPE_CHECKING: from collections.abc import Callable @@ -173,10 +174,10 @@ def at_boundary(self, window: HistoryWindow) -> bool: return all((position - boundary) * self.step >= 0 for position in (window.scroll_y, window.scroll_target_y)) - def work_cursor[Cursor](self, through: Cursor) -> Cursor | None: - return self.edge((None, through)) @abstractmethod - async def load_page(self, loader, cursor: TranscriptCursor | None, through) -> TranscriptPage: ... + def work_revision[Cursor](self, request: "PageRequest[Cursor]", through: Cursor) -> WorkRevision: ... + @abstractmethod + async def load_page(self, loader, cursor: TranscriptCursor, through) -> TranscriptPage: ... def progressed(self, page: "TranscriptPage", cursor: TranscriptCursor) -> bool: edge = self.edge((page.before, page.after)) start, stop = self.edge(((edge, cursor), (cursor, edge))) @@ -233,12 +234,15 @@ def trim(self, view: TranscriptPageView, count: int) -> None: class OlderEnd(PageEnd): + def work_revision[Cursor](self, request: "PageRequest[Cursor]", through: Cursor) -> WorkRevision: + return PageRevision(request) + step = -1 pop = staticmethod(deque.popleft) append = staticmethod(deque.appendleft) async def load_page(self, loader, cursor, through) -> TranscriptPage: - return await through.read_page(loader, before=cursor) + return await through.read_page(loader, cursor.earlier_page()) def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence) -> MessagePage: return source.conversation.page(source, PageBounds( @@ -265,6 +269,9 @@ def leave_tail(self, window: HistoryWindow) -> None: class NewerEnd(PageEnd): + def work_revision[Cursor](self, request: "PageRequest[Cursor]", through: Cursor) -> WorkRevision: + return GrowingPageRevision(request, through) + step = 1 accepts_destination = False pop = staticmethod(deque.pop) @@ -274,7 +281,7 @@ def has_more(self, through, page): return super().has_more(through, page) or page.after != through async def load_page(self, loader, cursor, through) -> TranscriptPage: - return await through.read_page(loader, after=cursor) + return await through.read_page(loader, cursor.later_page()) def wire_page(self, source: WireSource, cursor: WireCut, through: AfterSequence) -> MessagePage: page = source.conversation.page(source, PageBounds(cursor.newer(), source.page_limit)) @@ -308,12 +315,27 @@ async def mount(self, history: LineHistory, view: TranscriptPageView, pages: deq @dataclass(frozen=True) -class PageRequest[Cursor = TranscriptCursor | None]: +class PageRequest[Cursor = TranscriptCursor]: end: PageEnd cursor: Cursor - def progressed(self: PageRequest[TranscriptCursor | None], page: "TranscriptPage") -> bool: - return self.cursor is None or self.end.progressed(page, self.cursor) + def progressed(self: PageRequest[TranscriptCursor], page: "TranscriptPage") -> bool: + return self.end.progressed(page, self.cursor) + + +@dataclass(frozen=True) +class PageRevision[Cursor = TranscriptCursor](WorkRevision): + """An older request is bounded by its own cursor, independent of new arrivals.""" + + request: PageRequest[Cursor] + size: ClassVar[int] = 512 + + +@dataclass(frozen=True) +class GrowingPageRevision[Cursor = TranscriptCursor](PageRevision[Cursor]): + """A newer request also depends on the frontier this read reached.""" + + through: Cursor class PreparationDemand(ABC): diff --git a/src/toad/widgets/prompt.py b/src/toad/widgets/prompt.py index fc2c49be3..e8e2639aa 100644 --- a/src/toad/widgets/prompt.py +++ b/src/toad/widgets/prompt.py @@ -8,7 +8,6 @@ from agent_comms.repo import Checkout from typing import TYPE_CHECKING, Literal, Self -from agent_comms.acp_extension import QueueItem from rich.text import Text as RichText from textual import containers, events, getters, on, work @@ -687,29 +686,11 @@ def on_mount(self) -> None: self.call_after_refresh(self.sync_queue) def _update_queue_summary(self) -> None: - submissions = self.operational_submissions - projection = submissions.queue_projection - queued = [row.preview for row in submissions.queued_inputs] - restored = [row.preview for row in projection.restored] - feedback = projection.feedback(self.agent_binding.presentation.queue_supported) - delivering = [QueueItem.summarize(text) for text in submissions.delivering] - self.set_class(bool(queued or restored or delivering or feedback), "-has-queue") - parts: list[str] = [] - if feedback: - parts.append(feedback) - for label, texts, detail in ( - ("Submitting", delivering, ""), - ("Queued", queued, ""), - ("Restored", restored, ", read-only"), - ): - if texts: - parts.append( - f"{label} ({len(texts)}{detail}): " - + " · ".join(texts[:3]) - ) + text = self.operational_submissions.queue_summary + self.set_class(bool(text), "-has-queue") summary = self.query_one(".queue-summary", Label) - summary.tooltip = "Submitting is an outstanding local request, not acceptance. Queued is the producer's accepted input, not consumption or completion. Restored rows are not local drafts." - summary.update(Content(" | ".join(parts))) + summary.tooltip = "Inspect this source's actual queue or visible message delivery evidence. Acceptance is not consumption; unsupported actions remain unavailable." + summary.update(Content(text)) @on(messages.SendPromptNow) async def on_send_prompt_now(self, event): @@ -929,9 +910,6 @@ def sync_turn(self) -> None: def sync_configuration(self) -> None: pass - def _update_queue_summary(self) -> None: - pass - def watch_agent_info(self, agent_info: Content) -> None: if self.is_mounted: self.update_prompt() @@ -949,9 +927,6 @@ def update_prompt(self) -> None: def label_glyph(self) -> Content: return PromptLabel.AI - def compose_delivery_controls(self) -> ComposeResult: - return () - def compose_info_popups(self) -> ComposeResult: return () diff --git a/src/toad/widgets/thread_comms.py b/src/toad/widgets/thread_comms.py index a2f2747bb..c10c16079 100644 --- a/src/toad/widgets/thread_comms.py +++ b/src/toad/widgets/thread_comms.py @@ -7,7 +7,7 @@ from toad.widgets.thread_identity import ShowsThreadIdentity from toad.core import events as core_events -from toad.navigation_target import NavigationTarget, ThreadTarget +from toad.navigation_target import NavigationTarget, SelectTarget, ThreadTarget import asyncio from dataclasses import dataclass, field @@ -22,7 +22,6 @@ from textual.geometry import Offset from textual.widgets import Checkbox, Static -from toad.core.input_events import SelectTarget from toad.widgets.comms_sidebar import CommsRow, ThreadStatusRow from agent_comms.ui_model.message_category import MessageCategory from toad.widgets.session_sort import SortControl, ThreadSortForm diff --git a/src/toad/widgets/transcript_fragments.py b/src/toad/widgets/transcript_fragments.py index 77489a592..0e5a5daab 100644 --- a/src/toad/widgets/transcript_fragments.py +++ b/src/toad/widgets/transcript_fragments.py @@ -8,6 +8,7 @@ from agent_comms.field_codec import ToolCallId from agent_comms.message_reference import MessageReference +from agent_comms.read_basis import DisplayIdentity, NoDisplaySelection from agent_comms.turn_item import NativeItemId from metaclass_registry.mro_dispatch import handles from agent_comms.sent_message import PublishedSentMessage @@ -45,7 +46,7 @@ from toad.markdown_preparation import PreparedMarkdownPart from toad.menus import MenuItem from toad.message_rows import MarkdownMessageStyle, MessageRow, WireMessageStyle -from toad.notification_status import NotificationStatus, UncheckedNotifications +from toad.notification_status import MessageQueueEntry, NotificationStatus, UncheckedNotifications from toad.widgets.agent_activity import AgentActivityPresentation from toad.widgets.transcript_lines import AcquiredRows, NoRows, RowAcquisition @@ -340,6 +341,13 @@ def shown_live_by(self, claim: CommitClaim) -> bool: def notification_references(self) -> tuple[MessageReference, ...]: return () + @property + def notification_identity(self) -> DisplayIdentity | NoDisplaySelection: + return 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.""" return False @@ -498,6 +506,8 @@ class BusMessageFragment(TranscriptFragment): shown: MessageShown = field(default_factory=MessageShown) text_part: str row: MessageRow + notification_identity: DisplayIdentity | NoDisplaySelection = field( + default=NoDisplaySelection(), repr=False, compare=False) @classmethod def capture_part(cls, event: TextTranscript, text: str, **kwargs): @@ -516,6 +526,10 @@ def has_details(self) -> bool: def notification_references(self) -> tuple[MessageReference, ...]: return self.part_position.notification_references(self.events[0]) + 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: return self.events[0].source == message.reference diff --git a/src/toad/widgets/wire_history.py b/src/toad/widgets/wire_history.py index b34faf17e..a8eb51097 100644 --- a/src/toad/widgets/wire_history.py +++ b/src/toad/widgets/wire_history.py @@ -57,7 +57,7 @@ def initialize_source(self, source: WireSource) -> None: @property def undrawn(self) -> bool: """Nothing of the bound source has been drawn yet.""" - return all(view.prepared.page.undrawn for view in self.pages) + return all(view.prepared.page.after.requires_read() for view in self.pages) @property def source_identity(self): diff --git a/src/toad/wire_history.py b/src/toad/wire_history.py index 93e9d610c..0ac7fe6b9 100644 --- a/src/toad/wire_history.py +++ b/src/toad/wire_history.py @@ -22,7 +22,7 @@ ConversationObservation, ConversationRevision, ) -from agent_comms.message_page import MessagePage +from agent_comms.message_page import MessagePage, MessagePageRequest from agent_comms.messages import ConversationRole, Membership, Message, NoticeRole from agent_comms.page_traversal import ( LiveCursor, @@ -50,6 +50,7 @@ ) from toad.widgets.history_anchor import LatestPosition from toad.widgets.message_notifications import PaintedRead, ReadWork +from toad.submission_queue import BusQueueSource from toad.widgets.transcript_fragments import ( BusMessageFragment, MessageShown, @@ -102,6 +103,19 @@ class AfterMessage(WireCut): bounded = True kept = True + def accepts(self, current: MessagePage, incoming: MessagePage, + view: type[ConversationHistoryView]) -> bool: + return view.same_display(incoming, current) + + def advanced_source(self, current: MessagePage, incoming: MessagePage) -> MessagePage: + return current + + def observed(self) -> AfterMessage: + return self + + def requires_read(self) -> bool: + return False + @dataclass(frozen=True) class AfterSequence(AfterMessage): @@ -127,6 +141,19 @@ class NothingRead(AfterSequence): cursor: LiveCursor = field(default_factory=lambda: LiveCursor(0)) + def accepts(self, current: MessagePage, incoming: MessagePage, + view: type[ConversationHistoryView]) -> bool: + return True + + def advanced_source(self, current: MessagePage, incoming: MessagePage) -> MessagePage: + return incoming + + def observed(self) -> AfterSequence: + return AfterSequence(self.cursor) + + def requires_read(self) -> bool: + return True + @dataclass(frozen=True) class WirePage: @@ -137,21 +164,16 @@ class WirePage: after: AfterMessage has_older: bool has_newer: bool - source: MessagePage | None + source: MessagePage @classmethod def unread(cls) -> WirePage: - """What is drawn before the first read: nothing, ending before any answer.""" - return cls((), None, NothingRead(), False, False, None) - - @property - def undrawn(self) -> bool: - return self.source is None + """The declared unread cursor carries an empty, unscoped placeholder.""" + return cls((), None, NothingRead(), False, False, MessagePageRequest.empty) def accepts(self, page: MessagePage, view: type[ConversationHistoryView]) -> bool: """An initial placeholder accepts its first scope; Core compares actual pages.""" - source = self.source - return source is None or view.same_display(page, source) + return self.after.accepts(self.source, page, view) def advanced(self, interval: WirePage, *, older: bool) -> WirePage: """An empty interval advances its cut and establishes the first Core page. @@ -159,9 +181,10 @@ def advanced(self, interval: WirePage, *, older: bool) -> WirePage: Existing fragments retain their original page evidence; a placeholder must adopt the actual empty Core page rather than remain unobserved. """ - source = interval.source if self.undrawn else self.source + source = self.after.advanced_source(self.source, interval.source) return ( - replace(self, before=interval.before, has_older=interval.has_older, source=source) + replace(self, before=interval.before, after=self.after.observed(), + has_older=interval.has_older, source=source) if older else replace(self, after=interval.after, has_newer=interval.has_newer, source=source) ) @@ -201,7 +224,7 @@ def of(cls, message: Message, *, reader: str, **fields) -> WireFragment: return cls._roles.dispatch(type(message.role))(message, reader=reader, **fields) @classmethod - def from_message(cls, message, *, reader, **fields): + def from_message(cls, message, *, reader, page: MessagePage, **fields): return cls((IncomingTranscript.of(message, reader=reader),), **fields) @property @@ -254,11 +277,12 @@ class MessageFragment(BusMessageFragment, WireFragment, roles=(ConversationRole, """A sent message: its header, text and the state of its notifications.""" @classmethod - def from_message(cls, message, *, reader, **fields): + def from_message(cls, message, *, reader, page: MessagePage, **fields): return cls( (IncomingTranscript.of(message, reader=reader),), text_part=message.body, row=MessageRow(message, reader), + notification_identity=page.selection.notification_identity, **fields, ) @@ -282,7 +306,7 @@ def leading(self): @dataclass(frozen=True, eq=False) -class WireSource: +class WireSource(BusQueueSource): """One conversation as one viewer reads it from Core, and how its messages are styled. A page read to the newest messages is short so that it is drawn at once; @@ -323,13 +347,13 @@ def read(self, request: PageRequest[WireCut], through: AfterSequence) -> Prepare page = request.end.wire_page(self, request.cursor, through) shown = MessageShown(style=self.style) fragments = tuple( - WireFragment.of(message, shown=shown, evidence=page.evidence_for(message), reader=self.me) + WireFragment.of(message, shown=shown, evidence=page.evidence_for(message), reader=self.me, page=page) for message in page.messages ) wire = WirePage( page.messages, BeforeMessage(page.messages[0].view_cursor) if page.messages else None, - AfterMessage(page.messages[-1].view_cursor) if page.has_newer and page.messages else through, + AfterMessage(page.messages[-1].view_cursor) if page.has_newer and page.messages else through.observed(), page.has_older and bool(page.messages), page.has_newer and bool(page.messages), page, diff --git a/tests/app/test_recorded_transcript_reader.py b/tests/app/test_recorded_transcript_reader.py index 09fbe55ed..2aa5b5ee3 100644 --- a/tests/app/test_recorded_transcript_reader.py +++ b/tests/app/test_recorded_transcript_reader.py @@ -8,9 +8,11 @@ from agent_comms.acp_extension import TranscriptSnapshotUpdate, decode_updates, encode_updates from agent_comms.coordination_errors import HistoricalSourceChanged, StaleRevision +from agent_comms.order_tools import TranscriptPageRead from agent_comms.session_file import RegistrySessionLocation -from agent_comms.transcripts import RecordedTranscriptReadIdentity +from agent_comms.transcripts import EarlierThan, LaterThan, RecordedPageSource, RecordedTranscriptReadIdentity from agent_comms.threads import Thread +from agent_comms.ui_model.observation import NotPublished from support.roots import installed_wire from support.journals import assistant_entry, write_journal from toad.acp.comms_updates import OwnerSnapshotConsumer @@ -34,13 +36,14 @@ async def test_recorded_snapshot_stays_immutable_when_original_grows_and_rejects async with workspace.app() as ui: reader = ui.app.coordination_access.transcripts - update = await reader.snapshot(KnownTranscript(str(live.root), "reader"), historical_source=source.key) + update = await reader.snapshot(KnownTranscript(str(live.root), "reader"), + page_source=RecordedPageSource(source.key)) assert isinstance(update.identity, RecordedTranscriptReadIdentity) assert decode_updates(encode_updates(update)) == (update,) with native.open("a") as output: output.write(json.dumps(assistant_entry("later recorded answer", id="answer-2", parent="answer-1")) + "\n") refreshed = await reader.publication(update) - assert refreshed.identity.historical_source == source.key + assert refreshed.identity.page_source == RecordedPageSource(source.key) assert refreshed.identity.thread.incarnation == source.provenance.threads["reader"].incarnation assert [event.text for event in refreshed.page.events] == ["recorded answer"] assert refreshed == update @@ -55,10 +58,24 @@ async def test_recorded_snapshot_stays_immutable_when_original_grows_and_rejects def test_recorded_presentation_does_not_acquire_live_owner(recorded): live, source, _native = recorded presentation = live.views.thread_presentations(("reader",)).read("reader") - identity = live.transcripts.capture_page_read("reader", historical_source=source.key).identity + identity = live.transcripts.capture_page_read("reader", page_source=RecordedPageSource(source.key)).identity # There is no live agent/controller to acquire: the recorded declaration # cannot publish a turn even when delivered to the ordinary snapshot consumer. consumer = OwnerSnapshotConsumer(None, "authored", turn_token=1) consumer.thread_presentation(replace(presentation, read_identity=identity)) assert TranscriptSnapshotUpdate.capture(live.transcripts, "reader", - historical_source=source.key).identity == identity + page_source=RecordedPageSource(source.key)).identity == identity + + +def test_known_transcript_requests_carry_the_page_operation_and_keep_root_at_the_envelope(recorded): + live, source, _native = recorded + scope = KnownTranscript(str(live.root), "reader") + cursor = live.transcripts.capture_page_read("reader", page_source=RecordedPageSource(source.key)).read().after + snapshot = scope.snapshot_request(position=EarlierThan(cursor), page_source=RecordedPageSource(source.key)) + assert snapshot.root == str(live.root) + assert snapshot.operation == TranscriptPageRead( + thread="reader", position=EarlierThan(cursor), page_source=RecordedPageSource(source.key)) + previous = NotPublished() + update = scope.update_request(previous, position=LaterThan(cursor)) + assert update.root == str(live.root) and update.previous is previous + assert update.operation == TranscriptPageRead(thread="reader", position=LaterThan(cursor)) diff --git a/tests/native/test_native_goal_owner.py b/tests/native/test_native_goal_owner.py index 8fc1016d1..8c7cf29dd 100644 --- a/tests/native/test_native_goal_owner.py +++ b/tests/native/test_native_goal_owner.py @@ -3,6 +3,8 @@ import pytest from agent_comms.goal_actions import SetGoalAction from agent_comms.goal_history import RecordedGoalHistory +from agent_comms.input_attempt import OwnerQueueInputDeliveryScope +from agent_comms.ui_model.delivery import ReadDeliveryView from support.native import native_session @@ -37,3 +39,18 @@ async def test_editing_a_goal_keeps_its_identity_and_refuses_a_stale_edit(worksp for entry in history.history) snapshot = await agent.get_goal_snapshot() assert snapshot.goal == changed and snapshot.execution.goal_id == changed.id + + +async def test_input_delivery_is_read_from_the_live_native_owner_queue(workspace, native_package): + """The owner_queue scope can only come from the live owner; the no-coordination + DeliveryView and the unobserved projection never carry it.""" + async with native_session(workspace) as native: + await native.ui.until_agent_ready() + controller = native.agent.controller + for delivery in (await controller.input_delivery(), + await controller.input_delivery(include_history=True)): + assert type(delivery) is ReadDeliveryView + overview = delivery.overview + assert overview.current_scope is OwnerQueueInputDeliveryScope + assert overview.inputs == () and overview.historical_inputs == () + assert overview.historical_count == 0 and overview.dismissed_historical_count == 0 diff --git a/tests/native/test_native_reconnect_history.py b/tests/native/test_native_reconnect_history.py index 1aa14d0dd..14641f93b 100644 --- a/tests/native/test_native_reconnect_history.py +++ b/tests/native/test_native_reconnect_history.py @@ -87,7 +87,7 @@ def reply(request, number): await ui.until(return_ready.is_set, seconds=90, why="the hidden native rounds to finish") if hidden_rounds > 1: pending = await view.transcript.get_transcript_page( - after=view.transcript.binding.history.page_source.through) + position=view.transcript.binding.history.page_source.through.later_page()) assert pending.has_newer, 'The native hidden gap must exceed one real source page' await ui.select_tab(tab) assert ui.conversation is view diff --git a/tests/support/native_steps.py b/tests/support/native_steps.py index cd5819cfa..ac501ca90 100644 --- a/tests/support/native_steps.py +++ b/tests/support/native_steps.py @@ -7,7 +7,7 @@ from contextlib import asynccontextmanager from agent_comms.channel_targets import ChannelTarget -from agent_comms.notification_progress import AnsweredProgress, ConsumedDelivery, IgnoredProgress +from agent_comms.notification_progress import AnsweredProgress, ConsumedDelivery from toad.conversation_kind import ChannelConversation, DmConversation from toad.core import input_events from toad.navigation_target import DirectTarget, channel_target @@ -82,7 +82,7 @@ def replies(): feedback = feedback_of(chat, sent) assert isinstance(feedback, CheckedNotifications) assert any(item.recipient == target and isinstance(item.state, AnsweredProgress) - and item.answer == reply.reference + and item.state.response.reference == reply.reference for item in feedback.notifications) await ui.close_tab(screen.id) await DirectTarget(target).open(navigation) @@ -130,8 +130,8 @@ async def channel_message_feedback(native, owner_tab: str) -> None: why="seen, not answering feedback after declining the channel message") feedback = feedback_of(chat, sent) assert isinstance(feedback, CheckedNotifications) - assert any(item.recipient == native.thread and isinstance(item.state, IgnoredProgress) - and item.state.seen_not_answering and item.answer is None for item in feedback.notifications) + assert any(item.recipient == native.thread and item.state.seen_not_answering + for item in feedback.notifications) @asynccontextmanager diff --git a/tests/textual/test_goal_owner_views.py b/tests/textual/test_goal_owner_views.py index 18a78e60c..6ead72c29 100644 --- a/tests/textual/test_goal_owner_views.py +++ b/tests/textual/test_goal_owner_views.py @@ -23,7 +23,7 @@ from support.agents import RecordedEvents from support.textual_steps import channel_roster from toad.core.events import CommsUpdated -from toad.core.input_events import SelectTarget +from toad.navigation_target import SelectTarget from toad.screens.goal_details import GoalDetails from toad.screens.goal_edit import GoalEdit from toad.widgets.goal_bar import GoalBar diff --git a/tests/textual/test_historical_notification_source.py b/tests/textual/test_historical_notification_source.py index 4959f5871..fcf3798e1 100644 --- a/tests/textual/test_historical_notification_source.py +++ b/tests/textual/test_historical_notification_source.py @@ -2,7 +2,7 @@ from agent_comms.repo import Checkout from agent_comms.historical_views import HistoricalThread -from agent_comms.ui_model.observation import ReadRecordedNotifications +from agent_comms.notification_reads import RecordedNotificationRead from agent_comms.threads import Thread from support.roots import installed_wire @@ -28,9 +28,7 @@ async def test_historical_notification_keeps_selected_source_and_lifetime(worksp await ui.until(lambda: content in access.notification_views, why="the historical body's notification subscription") request = access.notification_views[content] - assert isinstance(request, ReadRecordedNotifications) - assert request.source == source.key - assert request.references == (message.reference,) + assert request.operation == RecordedNotificationRead((message.reference,), source.key) await ui.until(lambda: request in access.notification_outcomes, why="the archive's actual observation reply") outcomes = access.notification_outcomes[request] diff --git a/tests/textual/test_late_live_answer.py b/tests/textual/test_late_live_answer.py index de08978ea..f9891a2f0 100644 --- a/tests/textual/test_late_live_answer.py +++ b/tests/textual/test_late_live_answer.py @@ -237,7 +237,7 @@ async def test_returning_to_busy_tab_loads_the_saved_gap(workspace, hidden_count parent=f'hidden-{index - 1}' if index else 'before')) + '\n') hidden = latest_assistant_commit(workspace.comms, "background", journal) pending_page = workspace.comms.transcripts.capture_page_read( - 'background', after=view.transcript.binding.history.page_source.through).read() + 'background', position=view.transcript.binding.history.page_source.through.later_page()).read() assert pending_page.has_newer == (hidden_count > 20) if pending_page.has_newer: assert pending_page.after.offset < hidden.page_bound.offset diff --git a/tests/textual/test_message_notification_title.py b/tests/textual/test_message_notification_title.py index 1bd02b4f3..0d917efe5 100644 --- a/tests/textual/test_message_notification_title.py +++ b/tests/textual/test_message_notification_title.py @@ -38,14 +38,12 @@ async def test_changed_details_and_errors_keep_the_title_layout_but_changed_summ await ui.until(lambda: "waiting-peer: Queued" in " ".join(ui.painted(feedback).split()), why="the changed detail") assert str(feedback.title) == MessageNotification.summary((responding, queued)) assert str(feedback.title) == MessageNotification.summary((responding, waiting)) - assert str(feedback.title) == "Answering 1 · Delivered 2/2 · Consumed 1/2 · Read 0/2" assert feedback.has_class("-working") completed = replace(responding, state=CompletedProgress(ShownToModelProgress(1))) feedback.show(CheckedNotifications((completed, queued))) await ui.step() assert str(feedback.title) == MessageNotification.summary((completed, queued)) assert str(feedback.title) != MessageNotification.summary((responding, queued)) - assert str(feedback.title) == queued.label + " · Delivered 2/2 · Consumed 1/2 · Read 0/2" assert not feedback.has_class("-working") feedback.show(UnavailableNotifications("First read error")) await ui.step() diff --git a/tests/textual/test_message_notifications.py b/tests/textual/test_message_notifications.py index 670fa0bc2..7097802ba 100644 --- a/tests/textual/test_message_notifications.py +++ b/tests/textual/test_message_notifications.py @@ -35,7 +35,7 @@ async def publish(self, ui, chat): view = chat.message_history.notification_view access = ui.app.coordination_access request = access.notification_views[view] - outcomes = self.answer(request.references) + outcomes = self.answer(request.operation.references) expected, = CheckedNotifications.for_references( chat.message_history.visible_fragments()[-1].notification_references, outcomes) access.notifications_observed(NotificationsObserved(access.revision, request, outcomes), access.observation.observed()) @@ -88,10 +88,10 @@ async def test_visible_references_are_bounded_and_typed_outcomes_paint_their_det request = ui.app.coordination_access.notification_views[history.notification_view] on_screen = {reference for fragment in history.visible_fragments() for reference in fragment.notification_references} - assert 0 < len(request.references) <= len(on_screen) < 30 - assert set(request.references) <= on_screen + assert 0 < len(request.operation.references) <= len(on_screen) < 30 + assert set(request.operation.references) <= on_screen await notifications.publish(ui, chat) - await ui.until_shows("Delivered 45/45 · Consumed 0/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) assert newest_status(chat).working await ui.click("▶ Sent", occurrence=len(on_screen) - 1) await ui.until_shows("waiting-9: Sent") @@ -110,11 +110,11 @@ async def test_the_line_follows_typed_state_and_read_errors_without_new_messages await settled(ui, chat) notifications.state = ConsumingProgress(0) await notifications.publish(ui, chat) - await ui.until_shows("Answering 1 · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) assert newest_status(chat).working notifications.state = IgnoredProgress(ShownToModelProgress(0)) await notifications.publish(ui, chat) - await ui.until_shows("Seen, not answering 1 · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) assert not newest_status(chat).working access = ui.app.coordination_access request = access.notification_views[chat.message_history.notification_view] @@ -124,7 +124,7 @@ async def test_the_line_follows_typed_state_and_read_errors_without_new_messages assert "fixture busy database" in newest_status(chat).error notifications.state = CompletedProgress(ShownToModelProgress(0)) await notifications.publish(ui, chat) - await ui.until_shows("▶ Answered · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) async def test_a_hidden_view_unsubscribes_and_discards_a_late_reply(workspace): @@ -144,7 +144,7 @@ async def test_a_hidden_view_unsubscribes_and_discards_a_late_reply(workspace): await ui.select_tab(owner) assert view not in ui.app.coordination_access.notification_views notifications.state = CompletedProgress(ShownToModelProgress(0)) - view.notifications_observed(notifications.answer(request.references)) + view.notifications_observed(notifications.answer(request.operation.references)) await ui.step() assert view not in ui.app.coordination_access.notification_views assert {fragment.message.reference: fragment.shown.status @@ -162,7 +162,8 @@ async def test_markdown_uses_the_same_typed_line_and_an_empty_recorded_result(wo await chat.message_history.toggle_style() await ui.step() await notifications.publish(ui, chat) - await ui.until_shows("▶ Answered · Delivered 45/45 · Consumed 1/45 · Read 0/45") + await ui.until_shows(newest_status(chat).title) notifications.recorded = False await notifications.publish(ui, chat) - await ui.until_shows("▶ No recorded notification result") + await ui.until_shows(newest_status(chat).title) + assert newest_status(chat).notifications == () diff --git a/tests/textual/test_observation_shutdown.py b/tests/textual/test_observation_shutdown.py index 4e4aa236c..0a2b134db 100644 --- a/tests/textual/test_observation_shutdown.py +++ b/tests/textual/test_observation_shutdown.py @@ -1,35 +1,93 @@ """Textual-facing: widget unmount; unmount ends interest, not the real observer's already-acquired reads.""" import asyncio +import os +import signal import sys +from contextlib import asynccontextmanager + import pytest -from agent_comms.ui_model.observation import ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications +from agent_comms.notification_reads import LiveNotificationRead +from agent_comms.ui_model.observation import ( + ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications, StopObserving, + StopSidebar, +) from toad.comms_root import AnsweringReads +async def stopped(pid: int) -> None: + """Wait, bounded, until the stop is reported; WUNTRACED reports it without reaping.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + 5 + while loop.time() < deadline: + found, status = os.waitpid(pid, os.WNOHANG | os.WUNTRACED) + if found: + assert os.WIFSTOPPED(status) and os.WSTOPSIG(status) == signal.SIGSTOP + return + await asyncio.sleep(0.01) + raise AssertionError(f"observer child {pid} did not stop") + + +@asynccontextmanager +async def held(process: ObservationProcess): + """Hold this test's own observer child; resume it on every path while it lives.""" + pid = process.process.pid + os.kill(pid, signal.SIGSTOP) + try: + await stopped(pid) + yield pid + finally: + if process.process.is_alive(): + os.kill(pid, signal.SIGCONT) + + +async def settled(task: asyncio.Task) -> None: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +def empty_read(workspace) -> ReadNotifications: + return ReadNotifications(str(workspace.roots.wire), LiveNotificationRead(())) + + async def test_an_observer_dying_during_shutdown_still_releases_its_resources(workspace): - with pytest.raises(BaseExceptionGroup) as failures: + died = None + try: async with workspace.app() as ui: access = ui.app.coordination_access await access.current_sidebar() - observed = access.observation.observed() - process = observed.process - # The death has not been delivered to receive(); stop_updates must - # encounter it at its actual StopObserving request during shutdown. - asyncio.get_running_loop().remove_reader(process.fileno()) - process.process.kill() - await asyncio.to_thread(process.process.join, 5) - assert process.process.exitcode is not None - assert not process.connection.closed - assert failures.value.subgroup(ObserverDied) is not None + process = access.observation.observed().process + assert not (await asyncio.wait_for(process.read(empty_read(workspace)), 30)).outcomes + reading = None + try: + async with held(process) as pid: + reading = asyncio.create_task(process.read(empty_read(workspace))) + await asyncio.sleep(0) + assert process.reads and not reading.done() + os.kill(pid, signal.SIGKILL) + with pytest.raises(ObserverDied) as death: + await asyncio.wait_for(reading, 10) + finally: + if reading is not None: + await settled(reading) + died = death.value + assert died.pid == pid + assert died.exit_status == -signal.SIGKILL + for request in (StopSidebar(), StopObserving()): + with pytest.raises(ObserverDied) as again: + process.request(request) + assert again.value is died + except* ObserverDied as failures: + assert died is not None + _same, other = failures.split(lambda error: error is died) + assert other is None assert process.connection.closed assert not process.process.is_alive() assert not access.retiring - - async def test_stopping_observation_keeps_reads_but_sends_no_late_interest(workspace): workspace.thread("sender", title="Sender") workspace.thread("reader", title="Reader") @@ -43,50 +101,48 @@ async def test_stopping_observation_keeps_reads_but_sends_no_late_interest(works # Retiring the old process must not publish its state over the new one. assert access.observation is observing process = observing.process - loop = asyncio.get_running_loop() - acquired = asyncio.Event() - request = ReadNotifications(str(workspace.roots.wire), (message.reference,)) - - async def read_notifications(): - acquired.set() - return await process.read(request) - - # Hold the readiness callback, not the service, so this acquired read - # cannot resolve until stop_observing installs the read-only reader. - loop.remove_reader(process.fileno()) - async with asyncio.TaskGroup() as reads: - reading = reads.create_task(read_notifications()) - await acquired.wait() - assert process.reads and not reading.done() - await access.stop_observing() - answering = access.observation - assert type(answering) is AnsweringReads - assert answering.process is process - assert answering.service is observing.service - assert answering.selection is observing.selection - assert answering.sidebar is observing.sidebar - assert answering.subscriptions is observing.subscriptions - assert answering.updates is observing.updates - - requests = [] - - def record_request(frame, event, result): - if (event == "call" and frame.f_code is ObservationProcess.request.__code__ - and frame.f_locals["self"] is process): - requests.append(frame.f_locals["request"]) - - previous = sys.getprofile() - sys.setprofile(record_request) - try: - answering.views_changed() - answering.thread_interest(access) - answering.observe_sidebar(access.sidebar_request()) - answering.observe_notifications(ObserveNotifications(frozenset())) + assert not (await asyncio.wait_for(process.read(empty_read(workspace)), 30)).outcomes + request = ReadNotifications(str(workspace.roots.wire), LiveNotificationRead((message.reference,))) + reading = None + try: + # The observer child itself holds the answer; the UI's one receiver keeps running. + async with held(process): + reading = asyncio.create_task(process.read(request)) + await asyncio.sleep(0) + assert process.reads and not reading.done() await access.stop_observing() - assert access.observation is answering - finally: - sys.setprofile(previous) - assert requests == [] - result = await reading - assert set(result) == {(message.seq, message.message_id)} - assert process.process.is_alive() + answering = access.observation + assert type(answering) is AnsweringReads + assert answering.process is process + assert answering.service is observing.service + assert answering.selection is observing.selection + assert answering.sidebar is observing.sidebar + assert answering.subscriptions is observing.subscriptions + assert answering.updates is observing.updates + + requests = [] + + def record_request(frame, event, result): + if (event == "call" and frame.f_code is ObservationProcess.request.__code__ + and frame.f_locals["self"] is process): + requests.append(frame.f_locals["request"]) + + previous = sys.getprofile() + sys.setprofile(record_request) + try: + answering.views_changed() + answering.thread_interest(access) + answering.observe_sidebar(access.sidebar_request()) + answering.observe_notifications(ObserveNotifications(frozenset())) + await access.stop_observing() + assert access.observation is answering + finally: + sys.setprofile(previous) + assert requests == [] + assert not reading.done() + result = await asyncio.wait_for(reading, 30) + finally: + if reading is not None: + await settled(reading) + assert set(result.outcomes) == {(message.seq, message.message_id)} + assert process.process.is_alive() diff --git a/tests/textual/test_wire_message_lines.py b/tests/textual/test_wire_message_lines.py index d1ec839c4..6851be392 100644 --- a/tests/textual/test_wire_message_lines.py +++ b/tests/textual/test_wire_message_lines.py @@ -60,7 +60,7 @@ def test_rows_fit_the_width_wrap_on_words_and_keep_drawn_names_clickable_at_ever def test_opening_notification_details_draws_more_rows_than_the_collapsed_message(): - status = CheckedNotifications(()) + status = CheckedNotifications.from_notifications(()) for style in (IrcMessageStyle(), MarkdownMessageStyle()): shown = fragment("#all", style) shown.request_show(MessageShown(style=style, status=status)) diff --git a/tests/textual/test_wire_source.py b/tests/textual/test_wire_source.py index 59635665e..385569e54 100644 --- a/tests/textual/test_wire_source.py +++ b/tests/textual/test_wire_source.py @@ -1,6 +1,7 @@ """Textual-facing: the presentation window; a wire source pages one conversation and notices when Core replaces its reading.""" import os +from pathlib import Path from agent_comms.repo import Checkout from agent_comms.channel_targets import ChannelTarget, ThreadTarget @@ -15,7 +16,10 @@ from toad.transcript_preparation import PageRequest, Unchanged from toad.widgets.presentation_window import OLDER_END, NEWER_END from toad.message_rows import IrcMessageStyle -from toad.wire_history import AfterSequence, GrowingWireObservation, MessageFragment, ReplacedWireObservation, WireSource +from toad.wire_history import ( + AfterSequence, GrowingWireObservation, MessageFragment, NothingRead, + PreparedWirePage, ReplacedWireObservation, WirePage, WireSource, +) def channel_source(comms, target, project): @@ -32,6 +36,23 @@ def observed(source): return first, newest +def test_empty_first_read_keeps_its_real_identity_after_the_unread_placeholder(tmp_path: Path) -> None: + comms = installed_wire(tmp_path / 'wire') + comms.registry.declare(Thread('peer', frozenset({'team'}), Checkout(tmp_path))) + source = channel_source(comms, '#team', tmp_path) + placeholder = PreparedWirePage.empty(WirePage.unread()) + assert placeholder.page.after.requires_read() and not placeholder.page.source.selection.has_evidence + initial = NothingRead() + prepared = source.read(PageRequest(OLDER_END, initial), initial) + assert not prepared.fragments and not prepared.page.after.requires_read() + identity = prepared.page.source.selection.notification_identity + assert identity.viewer == comms.registry.require(source.viewer.name).incarnation + assert placeholder.page.accepts(prepared.page.source, source.conversation.history_view) + placeholder.advance_empty(prepared, older=True) + assert not placeholder.page.after.requires_read() + assert placeholder.page.source.selection.notification_identity == identity + + def test_a_turn_claim_on_the_peer_keeps_the_direct_message_page_identity(tmp_path): comms = installed_wire(tmp_path / "wire") comms.registry.declare(Thread("peer", frozenset(), Checkout(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()))) @@ -59,6 +80,8 @@ def bodies(prepared): assert newest.page.after == first.through and not newest.page.has_newer and newest.page.has_older assert all(isinstance(fragment, MessageFragment) and not fragment.outgoing_sources for fragment in newest.fragments) + assert all(fragment.notification_identity == newest.page.source.selection.notification_identity + for fragment in newest.fragments) earlier = source.read(PageRequest(OLDER_END, newest.page.before), first.through) assert bodies(earlier) == [f"Message {index}" for index in range(12, 52)] assert earlier.page.has_newer and earlier.page.has_older