From bfc03e737b833e58b387450b14d668d38c481f35 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 07:04:16 -0400 Subject: [PATCH] Annotate ACP lifecycle parameters with existing owner types Apply checked autotyping/Pyrefly inference to process, controller, session and configuration contracts. Use SurfaceBinding for attachment and AgentProcess for reusable connections. Reject the inferred object-based prompt list; the boundary still accepts two different payload representations. --- src/toad/acp/agent.py | 50 ++++++++++-------- src/toad/acp/agent_configuration.py | 29 +++++----- src/toad/acp/agent_controller.py | 82 ++++++++++++++++------------- src/toad/acp/agent_process.py | 36 ++++++++----- src/toad/acp/agent_session.py | 72 ++++++++++++++----------- src/toad/agent.py | 15 +++--- src/toad/agent_binding.py | 31 ++++++----- 7 files changed, 178 insertions(+), 137 deletions(-) diff --git a/src/toad/acp/agent.py b/src/toad/acp/agent.py index 472264fbb..8facae879 100644 --- a/src/toad/acp/agent.py +++ b/src/toad/acp/agent.py @@ -1,16 +1,3 @@ -from toad.acp.agent_session import AgentSession -from toad.acp.session_updates import SessionNotificationOwner -from toad.acp.client_session import ClientSessionRequest -from toad.acp.client_files import FileClientRequestOwner -from toad.acp.tool_calls import SessionToolCalls -from toad.acp.agent_configuration import ModelConfigurationSetting, ThinkingConfigurationSetting -from toad.acp.context_measurement import ContextMeasurement -from toad.acp.agent_log import AgentLog -from toad.acp.agent_process import AgentProcess -from toad.acp.agent_controller import AgentController -from toad.acp.transcript_reader import CoordinationTranscriptReader -from toad.acp.permission_controller import PermissionController -from toad.agent_presentation import ACPAgentPresentation import asyncio from contextlib import suppress from dataclasses import replace @@ -20,12 +7,11 @@ import rich.repr from agent_comms.acp_extension import ( CoordinationChangedUpdate, - PromptRequest, InputStartedUpdate, + PromptRequest, ) -from agent_comms.acp_failure import ACPFailure -from agent_comms.field_codec import FieldCodec from agent_comms.context_reading import UnmeasuredContext +from agent_comms.field_codec import FieldCodec from agent_comms.goal_actions import GoalAction from agent_comms.goal_presentation import GoalSnapshot, IdleGoalExecution from agent_comms.goals import Goal, GoalDeclaration, NoGoal @@ -33,14 +19,34 @@ import toad from toad import paths -from toad.core import events as core_events +from toad.acp.agent_configuration import ( + ModelConfigurationSetting, + ThinkingConfigurationSetting, +) +from toad.acp.agent_controller import AgentController +from toad.acp.agent_log import AgentLog +from toad.acp.agent_process import AgentProcess +from toad.acp.agent_session import AgentSession from toad.acp.attachment_presentation import CursorPresentation, QueuePresentation -from toad.acp.comms_updates import CommsUpdateConsumer, NoPendingThreadRead, PendingThreadRead, ThreadRead +from toad.acp.client_files import FileClientRequestOwner +from toad.acp.client_session import ClientSessionRequest +from toad.acp.comms_updates import ( + CommsUpdateConsumer, + NoPendingThreadRead, + PendingThreadRead, + ThreadRead, +) +from toad.acp.context_measurement import ContextMeasurement +from toad.acp.permission_controller import PermissionController from toad.acp.projection_attachment import CursorAttachment from toad.acp.queue_attachment import QueueAttachment -from toad.core.events import UnsupportedResumeAgentFail, AgentReady +from toad.acp.session_updates import SessionNotificationOwner +from toad.acp.tool_calls import SessionToolCalls +from toad.acp.transcript_reader import CoordinationTranscriptReader from toad.agent import AgentBase +from toad.agent_presentation import ACPAgentPresentation from toad.agent_schema import AgentDefinition +from toad.core import events as core_events def generate_datetime_filename( @@ -131,7 +137,7 @@ def log(self, line: str) -> None: def get_info(self) -> str: return self.definition.name - async def start(self, processes=()) -> None: + async def start(self, processes: tuple[AgentProcess, ...] = ()) -> None: """Start the operational agent; attachment resources are acquired separately.""" await self.process.start(self, processes) @@ -159,7 +165,7 @@ def session_id(self): return self.controller.session.session_id @session_id.setter - def session_id(self, value): + def session_id(self, value) -> None: self.controller.bind_session(value) @property @@ -170,7 +176,7 @@ def ready(self): def current_turn(self): return self.presentation.turns.owner - async def retire_surface(self, surface): + async def retire_surface(self, surface) -> None: if self.controller.surface.owns(surface): self.detach_surface(surface) await self.stop() diff --git a/src/toad/acp/agent_configuration.py b/src/toad/acp/agent_configuration.py index 8cc891fd6..a0ab3112c 100644 --- a/src/toad/acp/agent_configuration.py +++ b/src/toad/acp/agent_configuration.py @@ -5,27 +5,30 @@ from dataclasses import dataclass from typing import ClassVar -from acp.schema import (SessionConfigSelectOption, SessionConfigSelectGroup, - SessionConfigOptionSelect) +from acp.exceptions import RequestError +from acp.schema import ( + SessionConfigOptionSelect, + SessionConfigSelectGroup, + SessionConfigSelectOption, +) from agent_comms.acp_failure import ACPFailure from agent_comms.declared_family import DeclaredFamily from agent_comms.mro_dispatch import MroDispatch, handles -from acp.exceptions import RequestError class SelectChoices(MroDispatch): """Flatten only the SDK's decoded external flat/group variants.""" - def __init__(self, option: SessionConfigOptionSelect): + def __init__(self, option: SessionConfigOptionSelect) -> None: self.values: list[SessionConfigSelectOption] = [] for value in option.options: self.dispatch_sync(value) @handles(SessionConfigSelectOption) - def choice(self, value): + def choice(self, value) -> None: self.values.append(value) @handles(SessionConfigSelectGroup) - def group(self, value): + def group(self, value) -> None: self.values.extend(value.options) @@ -39,7 +42,7 @@ class ConfigurationSetting(DeclaredFamily, affix="ConfigurationSetting"): def matches(cls, option: SessionConfigOptionSelect) -> bool: ... @classmethod - def bind(cls, option): + def bind(cls, option: SessionConfigOptionSelect): if option.current_value not in {value.value for value in SelectChoices(option).values}: raise ValueError(f"ACP configuration {option.id!r} has no advertised current value") return cls(option) @@ -115,11 +118,11 @@ def matches(cls, option): class ConfigurationAdvertisements(MroDispatch): - def __init__(self, selections): + def __init__(self, selections: dict[type[ConfigurationSetting], ConfigurationSetting]) -> None: self.selections = selections @handles(SessionConfigOptionSelect) - def selection(self, option): + def selection(self, option) -> None: for kind in ConfigurationSetting.members_with(ConfigurationSetting): if kind.matches(option): self.selections[kind] = kind.bind(option) @@ -128,11 +131,11 @@ def selection(self, option): class AgentConfiguration: """Single actual advertisement owner; detached surfaces keep this same value.""" - def __init__(self, agent): + def __init__(self, agent) -> None: self.agent = agent self.reset() - def reset(self): + def reset(self) -> None: self.selections = {kind: kind() for kind in ConfigurationSetting.members_with(ConfigurationSetting)} def setting[T: ConfigurationSetting](self, kind: type[T]) -> T: @@ -155,7 +158,7 @@ def receive_current_mode(self, value: str) -> None: self.selections[ModeConfigurationSetting] = self.mode.with_current(value) self.publish() - def receive(self, options): + def receive(self, options) -> None: if options is None: return # Official external decoder once; consumers use its guaranteed typed fields. @@ -166,6 +169,6 @@ def receive(self, options): self.selections = selections self.publish() - def publish(self): + def publish(self) -> None: from toad.core.events import ConfigurationChanged self.agent.events.publish(ConfigurationChanged()) diff --git a/src/toad/acp/agent_controller.py b/src/toad/acp/agent_controller.py index cfe1e51ff..e1606858d 100644 --- a/src/toad/acp/agent_controller.py +++ b/src/toad/acp/agent_controller.py @@ -1,30 +1,38 @@ """ACP operations remain owned by the controller after their view retires.""" from __future__ import annotations -from toad.acp.status import StopReason import asyncio from abc import ABC, abstractmethod from dataclasses import dataclass, replace from functools import partial -from toad import surface_binding -from toad.plan import PlanItem -from .terminal_owner import OperationalTerminalOwner -from .transcript_reader import CoordinationTranscriptReader -from .client_session import ClientSessionRequest -from .prompt import build as build_prompt -from toad.core import events as core_events from acp.exceptions import RequestError -from toad.core.events import LogAgentFail from agent_comms.acp_extension import ( - PromptRequest, QueuePromptRequest, ClearQueueRequest, SendNowRequest, - CompactRequest, CoordinationChangedUpdate, InputFailedUpdate, encode_request, + ClearQueueRequest, + CompactRequest, + CoordinationChangedUpdate, + InputFailedUpdate, + PromptRequest, + QueuePromptRequest, + SendNowRequest, + encode_request, ) from agent_comms.acp_failure import ACPFailure, RequestFailure from agent_comms.transcripts import TranscriptReadIdentity from agent_comms.ui_model.delivery import InputDelivery from agent_comms.ui_model.observation import ReadInputDelivery +from toad.acp.client_session import ClientSessionRequest +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 .prompt import build as build_prompt +from .terminal_owner import OperationalTerminalOwner +from .transcript_reader import CoordinationTranscriptReader + @dataclass(frozen=True) class SessionBinding: @@ -32,10 +40,10 @@ class SessionBinding: session_id: str | None @property - def bound(self): + def bound(self) -> bool: return self.session_id is not None - def accepts_notification(self, session_id): + def accepts_notification(self, session_id: str): return self.bound and self.session_id == session_id def accepts_snapshot(self, identity: TranscriptReadIdentity, @@ -57,7 +65,7 @@ async def run(self, task): class ProcessTaskRunner(TaskRunner): - def __init__(self, processes): + def __init__(self, processes) -> None: self.processes = processes async def run(self, task): @@ -66,9 +74,9 @@ async def run(self, task): class AgentController(OperationalTerminalOwner): """One operational source; the surface is an optional weak projection.""" - def __init__(self, agent, transcripts: CoordinationTranscriptReader): + def __init__(self, agent, transcripts: CoordinationTranscriptReader) -> None: super().__init__(agent) - self.surface: surface_binding.SurfaceBinding = surface_binding.DetachedSurfaceBinding() + self.surface: SurfaceBinding = DetachedSurfaceBinding() self.task_runner: TaskRunner = ThreadTaskRunner() self._deferred_submissions: set[asyncio.Task] = set() self.prompt_in_flight = 0 @@ -78,7 +86,7 @@ def __init__(self, agent, transcripts: CoordinationTranscriptReader): self.commands = [] self.plan_entries: list[PlanItem] | None = None - def bind_session(self, session_id): + def bind_session(self, session_id) -> None: if session_id != self.session.session_id: self.replace_terminal_session() self.agent.permissions.cancel() @@ -87,7 +95,7 @@ def bind_session(self, session_id): self.session = SessionBinding(session_id) self.reset_configuration() - def attach(self, binding: surface_binding.SurfaceBinding): + def attach(self, binding: SurfaceBinding) -> None: previous = self.surface.target if self.surface.owns(binding.target): binding.close() @@ -101,16 +109,15 @@ def attach(self, binding: surface_binding.SurfaceBinding): else: self.start_terminal_presentation(binding.target) - def detach(self, target): + def detach(self, target) -> None: if self.surface.owns(target): self.surface.close() - self.surface = surface_binding.DetachedSurfaceBinding() + self.surface = DetachedSurfaceBinding() self.agent.permissions.detach(target) self.terminals.detach() - async def restore(self, binding): - from toad.core.events import CommsUpdated - from toad.core.events import AvailableCommandsUpdate + async def restore(self, binding: SurfaceBinding) -> None: + from toad.core.events import AvailableCommandsUpdate, CommsUpdated if self.surface is not binding: return agent = self.agent @@ -131,7 +138,7 @@ async def restore(self, binding): - async def publish_transcript_snapshot(self, update, binding, session): + async def publish_transcript_snapshot(self, update, binding, session) -> None: if binding.target is None or not session.accepts_snapshot(update.identity, self.coordination): return # No view or matching request owns this publication. snapshot = await self.transcripts.publication(update) @@ -143,7 +150,7 @@ async def publish_transcript_snapshot(self, update, binding, session): return self.agent.events.publish(core_events.CommsUpdated(snapshot, session.session_id)) - def connection_closed(self): + def connection_closed(self) -> None: from toad.core.events import McpClientStopped agent = self.agent agent.session.closed() @@ -152,7 +159,7 @@ def connection_closed(self): agent.presentation.turns.reset() agent.events.publish(McpClientStopped()) - def reset_configuration(self): + def reset_configuration(self) -> None: self.agent.configuration.reset() self.agent.configuration.publish() @@ -162,7 +169,7 @@ def publish_plan(self, entries: list[PlanItem]) -> None: self.plan_entries = entries self.agent.events.publish(Plan(entries)) - def publish_commands(self, commands): + def publish_commands(self, commands) -> None: from toad.core.events import AvailableCommandsUpdate self.commands = commands self.agent.events.publish(AvailableCommandsUpdate()) @@ -181,7 +188,7 @@ async def submit(self, prompt: str, *, request: PromptRequest | None = None): prompt, command, authority, agent.queue_attachment.scope, agent.project_root_path), authority)) - async def _prompt_operation(self, operation, authority): + async def _prompt_operation(self, operation, authority: ClientSessionRequest): """Keep the request in flight until its operation ends.""" coordination = self.coordination self.prompt_in_flight += 1 @@ -192,7 +199,7 @@ async def _prompt_operation(self, operation, authority): if self.prompt_in_flight == 0 and authority.current and coordination is not None: self.start_operation(self.reconcile_turn(coordination, authority)) - async def reconcile_turn(self, coordination, authority): + async def reconcile_turn(self, coordination, authority) -> None: """Release snapshot ordering through a fresh read of the same owner.""" self.require_owner(coordination, authority) await self.agent.thread_read().reconcile(self.transcripts) @@ -202,7 +209,7 @@ def accepts_turn_snapshot(self, sequence): return (self.prompt_in_flight == 0 and sequence == self.agent.presentation.turns.sequence) - async def _submit(self, prompt, command, authority, queue_scope, project): + async def _submit(self, prompt: str, command, authority, queue_scope, project): submission = asyncio.current_task() if command.defer_display else None if submission is not None: self._deferred_submissions.add(submission) @@ -226,7 +233,7 @@ async def _submit(self, prompt, command, authority, queue_scope, project): if submission is not None: self._deferred_submissions.discard(submission) - async def submit_blocks(self, content, command=None): + async def submit_blocks(self, content: list[dict[str, str]], command: ClearQueueRequest | None=None): agent = self.agent authority = ClientSessionRequest(agent, agent.session_id) return await self.operate(self._prompt_operation(partial(self._prompt, @@ -263,7 +270,7 @@ def require(): published=request_failure.notification_published) return None - def _prompt_failed(self, command, authority, queue_scope, failure, report, *, published=False): + def _prompt_failed(self, command, authority, queue_scope, failure: ACPFailure, report, *, published: bool=False) -> None: agent = self.agent user_text = command.draft_text if command is not None else None if user_text: @@ -271,10 +278,10 @@ def _prompt_failed(self, command, authority, queue_scope, failure, report, *, pu if not published: report(LogAgentFail(failure.title, failure.feedback, log_path=agent.presentation.log_path)) - async def clear_queue(self): + async def clear_queue(self) -> None: await self.submit_blocks([{'type': 'text', 'text': ' '}], ClearQueueRequest()) - async def send_now(self): + async def send_now(self) -> bool: authority = ClientSessionRequest(self.agent, self.agent.session_id) queue_scope = self.agent.queue_attachment.scope if pending := tuple(self._deferred_submissions): @@ -306,7 +313,7 @@ async def cancel_prompt(self): return await self.operate(self._cancel_prompt( ClientSessionRequest(self.agent, self.agent.session_id))) - async def _cancel_prompt(self, authority): + async def _cancel_prompt(self, authority: ClientSessionRequest): authority.require() try: await self.agent.process.rpc.cancel(authority.session_id) @@ -318,6 +325,7 @@ async def request_owner(self, method: str, **params): if self.coordination is None: raise ValueError('This action requires an agent-comms thread.') from agent_comms.runtime import RuntimeProxy, socket_path + from toad.owner_preparation import OwnerRequestContext coordination = self.coordination authority = ClientSessionRequest(self.agent, self.agent.session_id) @@ -339,13 +347,13 @@ def resolve(): finally: await proxy.close() - def require_owner(self, coordination, authority): + def require_owner(self, coordination, authority: ClientSessionRequest): if authority.retired or self.coordination is None: raise ValueError('The connected owner session retired; refresh its state.') if (self.coordination.wire_root, self.coordination.thread) != (coordination.wire_root, coordination.thread): raise ValueError('The owner identity changed while preparing the request.') - async def input_delivery(self, *, include_history=False) -> InputDelivery: + async def input_delivery(self, *, include_history: bool=False) -> InputDelivery: """The owner's delivery notices, read by the observation service.""" coordination = self.coordination if coordination is None: diff --git a/src/toad/acp/agent_process.py b/src/toad/acp/agent_process.py index 8d8dca11c..fe565ce7a 100644 --- a/src/toad/acp/agent_process.py +++ b/src/toad/acp/agent_process.py @@ -1,22 +1,31 @@ """ACP connection, admitted subprocess and attached-session lifetime.""" from __future__ import annotations + import asyncio import json import os from contextlib import ExitStack from contextvars import ContextVar from pathlib import Path +from typing import Optional + from acp import Client, ClientSideConnection, schema from acp.connection import StreamDirection, StreamEvent from acp.exceptions import RequestError -from agent_comms.child_process import AttachedChild, StreamingChildStdio, join_retirement +from agent_comms.child_process import ( + AttachedChild, + StreamingChildStdio, + join_retirement, +) from agent_comms.launch_root import LaunchRoot +from agent_comms.route_selection import RouteSelection + from toad.acp.shell_command import ShellCommand from toad.core.events import LogAgentFail class AgentProcess(Client): - def __init__(self, agent): + def __init__(self, agent) -> None: self.agent = agent self.env = None self.cwd = None @@ -38,7 +47,7 @@ def rpc(self) -> ClientSideConnection: raise ConnectionError('ACP process is not connected') return self.connection - def accepts_attachment(self, agent, env, cwd, selection): + def accepts_attachment(self, agent, env: dict[str, str], cwd: str, selection: RouteSelection): response = self.initialization capabilities = response.agent_capabilities if response is not None else None lifecycle = capabilities.session_capabilities if capabilities is not None else None @@ -51,12 +60,11 @@ def accepts_attachment(self, agent, env, cwd, selection): and agent.session_id is not None and all(member.agent.session_id != agent.session_id for member in self.sessions)) - async def start(self, agent, processes=()): + async def start(self, agent, processes: tuple[AgentProcess, ...]=()): agent.session.reopen() agent.session.starting() await asyncio.to_thread( agent.presentation.log_path.parent.mkdir, parents=True, exist_ok=True) - from agent_comms.route_selection import RouteSelection env = os.environ.copy() cwd = str(agent.project_root_path.resolve()) selection = RouteSelection.capture(agent.controller.transcripts.access.launch_root.path) @@ -89,7 +97,7 @@ async def prompt(self, agent, content, metadata, require): require() return await self.rpc.prompt(agent.session_id, content, **metadata) - def observe_stream(self, event: StreamEvent): + def observe_stream(self, event: StreamEvent) -> None: """Keep the original frame with its SDK handler.""" if event.direction is StreamDirection.INCOMING: # SDK creates the handler task immediately after this observer; its @@ -116,14 +124,14 @@ async def request_permission(self, session_id, tool_call, options, **kwargs): return await agent.session.start_operation( agent.permissions.request_permission(session_id, options, tool_call, kwargs)) - async def session_update(self, session_id, update, **kwargs): + async def session_update(self, session_id, update, **kwargs) -> None: agent = self.recipient(session_id) notification = schema.SessionNotification.model_construct( session_id=session_id, update=update, field_meta=kwargs or None) raw = self._incoming.get()['params']['update'] await agent.session.start_operation(agent.updates.receive(notification, raw)) - async def read_text_file(self, session_id, path, line=None, limit=None, **kwargs): + async def read_text_file(self, session_id, path, line=None, limit: Optional[int]=None, **kwargs): agent = self.recipient(session_id) return await agent.session.start_operation(asyncio.to_thread( agent.files.read_text_file, session_id, path, line, limit)) @@ -167,13 +175,13 @@ async def complete_elicitation(self, elicitation_id, **kwargs): async def ext_method(self, method, params): raise RequestError.method_not_found(f'_{method}') - async def ext_notification(self, method, params): + async def ext_notification(self, method, params) -> None: self.agent.log(f'[ACP unsupported notification] _{method}: {params}') - def on_connect(self, conn): + def on_connect(self, conn) -> None: pass - async def retire(self): + async def retire(self) -> None: for session in tuple(self.sessions): session.close() for task in tuple(session.responses): @@ -182,7 +190,7 @@ async def retire(self): self.retirement = asyncio.create_task(self._retire()) await join_retirement(self.retirement) - async def _retire(self): + async def _retire(self) -> None: try: if self.connection is not None: await self.connection.close() @@ -216,7 +224,7 @@ async def detach(self, session): if not self.sessions and self.retirement is None and self.runner is not None: self.runner.cancel() - async def stop(self, agent): + async def stop(self, agent) -> None: agent.session.close() await agent.session.retire() if not self.sessions: @@ -226,7 +234,7 @@ async def stop(self, agent): await self.retire() self.runner = None - async def run(self): + async def run(self) -> None: try: await self.communicate() finally: diff --git a/src/toad/acp/agent_session.py b/src/toad/acp/agent_session.py index d4a99c0ac..2f3e88998 100644 --- a/src/toad/acp/agent_session.py +++ b/src/toad/acp/agent_session.py @@ -1,28 +1,40 @@ """ACP session handshake, durable metadata and reconnection.""" from __future__ import annotations -from agent_comms.errors import RouteChangedError -from agent_comms.declared_family import DeclaredFamily -from agent_comms.child_process import join_retirement import asyncio import os -from dataclasses import dataclass, replace from collections.abc import Callable +from dataclasses import dataclass, replace + +from acp import schema +from acp.exceptions import RequestError +from agent_comms.acp_extension import HistoryOnlyQueueProjection +from agent_comms.acp_failure import ACPFailure +from agent_comms.child_process import join_retirement +from agent_comms.declared_family import DeclaredFamily +from agent_comms.errors import RouteChangedError +from agent_comms.input_attempt import NotSentInput +from agent_comms.owner_protocol import IncompatibleOwner from agent_comms.route_selection import RouteSelection +from agent_comms.session_load import ( + EnsuringSessionLoadPlan, + ExistingSessionLoadPlan, + FailedSessionLoadPlan, + SessionLoadPlan, +) from agent_comms.thread_presentation import SessionDisconnected, ThreadPresentation + import toad from toad import constants -from acp.exceptions import RequestError -from toad.core import events as core_events -from acp import schema from toad.acp.client_session import ClientSessionRequest -from toad.core.events import AgentFail, AgentReady, UnsupportedResumeAgentFail, LogAgentFail +from toad.core import events as core_events +from toad.core.events import ( + AgentFail, + AgentReady, + LogAgentFail, + UnsupportedResumeAgentFail, +) from toad.db import DB, SessionMeta -from agent_comms.acp_failure import ACPFailure -from agent_comms.acp_extension import HistoryOnlyQueueProjection -from agent_comms.input_attempt import NotSentInput -from agent_comms.owner_protocol import IncompatibleOwner -from agent_comms.session_load import EnsuringSessionLoadPlan, ExistingSessionLoadPlan, FailedSessionLoadPlan, SessionLoadPlan PROTOCOL_VERSION = 1 @@ -115,7 +127,7 @@ class AgentSession: reconnect_timeout_seconds = 30 - def __init__(self, agent, pk, app_events: core_events.CoreEventStream): + def __init__(self, agent, pk, app_events: core_events.CoreEventStream) -> None: self.agent = agent # The app's stream: its thread rows show this session's state, attached to a view or not. self.app_events = app_events @@ -167,14 +179,14 @@ def start_operation(self, operation): task.add_done_callback(self.responses.discard) return task - def close(self): + def close(self) -> None: self.disposition = self.disposition.close(self.agent) - def reopen(self): + def reopen(self) -> None: self.disposition = ActiveSessionDisposition(self) self.retirement = None - def start(self): + def start(self) -> None: self.task = asyncio.create_task(self.run()) self.task.add_done_callback(self.session_finished) @@ -185,10 +197,10 @@ def begin_retirement(self): self.retirement.add_done_callback(self.retirement_finished) return self.retirement - async def retire(self): + async def retire(self) -> None: await join_retirement(self.begin_retirement()) - async def _retire(self): + async def _retire(self) -> None: pending = tuple(task for task in (self.task, *self.responses) if task is not None) for task in pending: @@ -199,14 +211,14 @@ async def _retire(self): self.task = None await self.agent.process.detach(self) - def session_finished(self, task): + def session_finished(self, task) -> None: if task.cancelled(): return error = task.exception() if error is not None and self.accepts_updates: self.startup_failed(f"{type(error).__name__}: {error}") - def retirement_finished(self, task): + def retirement_finished(self, task) -> None: if task.cancelled(): return error = task.exception() @@ -225,13 +237,13 @@ def record_failure( self.agent.events.publish(event) self.begin_retirement() - def session_failed(self, failure): + def session_failed(self, failure: ACPFailure) -> None: self.record_failure( LogAgentFail(failure.title, failure.feedback, log_path=self.agent.presentation.log_path), RefusedSessionDisposition, ) - def startup_failed(self, details): + def startup_failed(self, details: str) -> None: self.record_failure( LogAgentFail("ACP session startup failed", details=details, log_path=self.agent.presentation.log_path), @@ -251,20 +263,20 @@ def ready(self): def supports_load(self): return self.capabilities.load_session - def starting(self): + def starting(self) -> None: """Starting clears readiness before any asynchronous work.""" self.connected = False self.settled.clear() self.agent.controller.reset_configuration() - def closed(self): + def closed(self) -> None: self.connected = False - async def touch(self): + async def touch(self) -> None: if self.pk is not None: await DB().session_update_last_used(self.pk) - def coordinated_title(self, title): + def coordinated_title(self, title) -> None: """A coordination update presents the thread's title. A name the user chose wins: it is announced as the session's title @@ -283,7 +295,7 @@ def coordinated_title(self, title): if self.pk is not None: self.start_operation(self._adopt_thread_title(title)) - async def _adopt_thread_title(self, title): + async def _adopt_thread_title(self, title) -> None: """A session without a chosen name keeps its saved title following the thread.""" await DB().session_update_title(self.pk, title) @@ -378,7 +390,7 @@ async def authenticate(self, method_id: str) -> None: authority.require() - async def initialize(self, *, force=False): + async def initialize(self, *, force: bool=False) -> None: """The connection owns its original initialization response.""" authority = ClientSessionRequest(self.agent, self.agent.session_id) authority.require() @@ -539,6 +551,6 @@ async def set_name(self, name: str) -> None: await db.session_update_title(self.pk, name) - def publish_configuration(self, response): + def publish_configuration(self, response) -> None: self.load_plan = SessionLoadPlan.at_response(response.field_meta) self.agent.configuration.receive(response.config_options) diff --git a/src/toad/agent.py b/src/toad/agent.py index d42114491..43bf05a7b 100644 --- a/src/toad/agent.py +++ b/src/toad/agent.py @@ -9,6 +9,7 @@ from toad.agent_presentation import LocalAgentPresentation from toad.conversation_turn import NoTurn from toad.core.events import CoreEventStream +from toad.surface_binding import SurfaceBinding class AgentBase(AgentConnection): @@ -43,14 +44,14 @@ def matches_session(self, session_id): def borrowed_processes(self): return (self.process,) - async def reconnect_session(self, source, plan): + async def reconnect_session(self, source, plan) -> None: await self.session.reconnect(plan) - async def change_project(self, view, path): + async def change_project(self, view, path) -> None: project = await self.update_project(str(path)) view.feedback.flash(Content(f"Project changed to {project}"), style="success") - def request_compaction(self, view, instructions): + def request_compaction(self, view, instructions) -> None: view.compact_context(instructions) def begin_submission(self, owner, submission): @@ -68,11 +69,11 @@ def accepts_failed_input(self, message): def send_now(self, owner): return owner.send_bound_now(self) - def invalidate_observation(self, observation): + def invalidate_observation(self, observation) -> None: observation.schedule(self) @property - def automatic_titles(self): + def automatic_titles(self) -> bool: return self.observed_thread() is None async def project_changed(self, path) -> None: @@ -121,10 +122,10 @@ def session_disposition(self): def current_turn(self): return NoTurn() - def attach_surface(self, binding): + def attach_surface(self, binding: SurfaceBinding) -> None: binding.close() - async def retire_surface(self, surface): + async def retire_surface(self, surface) -> None: await self.stop() def observed_thread(self) -> tuple[str, str] | None: diff --git a/src/toad/agent_binding.py b/src/toad/agent_binding.py index f175b578d..1706c14bb 100644 --- a/src/toad/agent_binding.py +++ b/src/toad/agent_binding.py @@ -1,12 +1,15 @@ """The agent itself is a bound connection; a saved view may be unbound.""" from __future__ import annotations + from abc import ABC, abstractmethod from functools import cached_property + from agent_comms.thread_presentation import ThreadPresentation from textual.content import Content -from toad.agent_presentation import LocalAgentPresentation + from toad.acp.agent_configuration import AgentConfiguration from toad.acp.agent_session import UnopenedSessionDisposition +from toad.agent_presentation import LocalAgentPresentation class AgentConnection(ABC): @@ -112,10 +115,10 @@ async def project_changed(self, path) -> None: def refresh_context(self, view) -> None: """There is no native measurement to display.""" - async def set_model(self, model_id): + async def set_model(self, model_id) -> str: return "No agent is bound yet" - async def set_mode(self, mode_id): + async def set_mode(self, mode_id) -> str: return "No agent is bound yet" async def compact_context(self, instructions) -> None: @@ -127,41 +130,41 @@ async def cancel_turn(self, view) -> None: def begin_submission(self, owner, submission): return owner.unbound(submission) - def accepts_failed_input(self, message): + def accepts_failed_input(self, message) -> bool: return False - def send_now(self, owner): + def send_now(self, owner) -> None: """No remote queue is bound.""" async def get_goal_history(self, goal_id): return () - async def update_goal(self, action, text): + async def update_goal(self, action, text: str): raise ValueError("Persistent goals require an agent-comms session") - def cancel_allowed(self, turn): + def cancel_allowed(self, turn) -> bool: return False - async def cancel(self): + async def cancel(self) -> bool: return False - def matches_session(self, session_id): + def matches_session(self, session_id) -> bool: return False def borrowed_processes(self): return () - async def reconnect_session(self, source, plan): + async def reconnect_session(self, source, plan) -> None: source.presentation.sources.attach_started(source, plan) - async def change_project(self, view, path): + async def change_project(self, view, path) -> None: view.feedback.flash("Project changes require an agent-comms session", style="error") - def request_compaction(self, view, instructions): + def request_compaction(self, view, instructions) -> None: view.feedback.flash("Context compaction requires an agent-comms session", style="error") - async def observe_thread_presentation(self, presentation): + async def observe_thread_presentation(self, presentation) -> None: """There is no connected agent to inform.""" - def invalidate_observation(self, observation): + def invalidate_observation(self, observation) -> None: """No native source exists to schedule a read against."""