diff --git a/src/toad/acp/agent_session.py b/src/toad/acp/agent_session.py index 7128cf6af..1f0c3c3c6 100644 --- a/src/toad/acp/agent_session.py +++ b/src/toad/acp/agent_session.py @@ -1,10 +1,14 @@ -"""ACP session handshake, durable metadata and reconnection custody.""" +"""ACP session handshake, durable metadata and reconnection.""" from __future__ import annotations from agent_comms.declared_family import DeclaredFamily from agent_comms.child_process import join_retirement import asyncio import os +from pathlib import Path from dataclasses import dataclass, replace +from agent_comms.acp_ingress import AcpIngress +from agent_comms.errors import RelationViolationError +from agent_comms.route_selection import RouteChanged, RouteSelection import toad from toad import constants, jsonrpc from toad.core import events as core_events @@ -15,7 +19,7 @@ from toad.db import DB, SessionMeta from agent_comms.acp_failure import ACPFailure from agent_comms.input_attempt import NotSentInput -from agent_comms.session_load import EnsuringSessionLoadAdmission, ExistingSessionLoadAdmission, FailedSessionLoadAdmission, SessionLoadAdmission +from agent_comms.session_load import EnsuringSessionLoadPlan, ExistingSessionLoadPlan, FailedSessionLoadPlan, SessionLoadPlan PROTOCOL_VERSION = 1 @@ -70,8 +74,15 @@ class RefusedSessionDisposition(FailedSessionDisposition): held_by_agent = False +class LoginNotReady(ValueError): + """A terminal login finished, but the session it restarted did not become ready.""" + + class AgentSession: """Own readiness and durable session effects; binding identity stays on the controller.""" + + reconnect_timeout_seconds = 30 + def __init__(self, agent, pk): self.agent = agent self.task = None @@ -82,7 +93,7 @@ def __init__(self, agent, pk): self.chosen_name = None self.connected = False self.reconnecting = False - self.load_admission: SessionLoadAdmission = EnsuringSessionLoadAdmission() + self.load_plan: SessionLoadPlan = EnsuringSessionLoadPlan() self.settled = asyncio.Event() @property @@ -118,11 +129,15 @@ def start(self): self.task = asyncio.create_task(self.run()) self.task.add_done_callback(self.session_finished) - async def retire(self): + def begin_retirement(self): self.close() if self.retirement is None: self.retirement = asyncio.create_task(self._retire()) - await join_retirement(self.retirement) + self.retirement.add_done_callback(self.retirement_finished) + return self.retirement + + async def retire(self): + await join_retirement(self.begin_retirement()) async def _retire(self): pending = tuple(task for task in (self.task, *self.responses) @@ -141,8 +156,6 @@ def session_finished(self, task): error = task.exception() if error is not None and self.accepts_updates: self.startup_failed(f"{type(error).__name__}: {error}") - self.retirement = asyncio.create_task(self._retire()) - self.retirement.add_done_callback(self.retirement_finished) def retirement_finished(self, task): if task.cancelled(): @@ -161,6 +174,7 @@ def record_failure( self.failed() self.disposition = state(event) self.agent.events.publish(event) + self.begin_retirement() def session_failed(self, failure): self.record_failure( @@ -189,7 +203,7 @@ def supports_load(self): return self.capabilities.load_session def starting(self): - """Start admission clears readiness before any asynchronous work.""" + """Starting clears readiness before any asynchronous work.""" self.connected = False self.settled.clear() self.agent.controller.reset_configuration() @@ -269,20 +283,14 @@ async def observe_owner(self, presentation) -> None: if self.agent.coordination is None: return binding = presentation.binding - if not self.load_admission.superseded_by(binding): + if not self.load_plan.superseded_by(binding): return - await self.reconnect(ExistingSessionLoadAdmission(binding)) + await self.reconnect(ExistingSessionLoadPlan(binding)) - async def reconnect(self, admission: SessionLoadAdmission | None = None) -> None: + async def reconnect(self, plan: SessionLoadPlan | None = None) -> None: """Reattach the existing view after login or an explicit owner start.""" if self.reconnecting: return - if self.agent.controller.session.bound and not self.supports_load: - raise ValueError("This agent cannot resume its session.") - from pathlib import Path - from agent_comms.acp_ingress import AcpIngress - from agent_comms.route_selection import RouteSelection - requested_cwd = str(self.agent.project_root_path.resolve()) self.reconnecting = True try: @@ -291,21 +299,35 @@ async def reconnect(self, admission: SessionLoadAdmission | None = None) -> None RouteSelection.for_child, os.environ.copy(), requested_cwd) await AcpIngress(selection, Path(requested_cwd), self.agent.process.attached_root(self.agent)).preflight() - except Exception as error: - raise ValueError( - f"Reconnect not attempted: maintenance admission denied: {error}" - ) from error + except (RouteChanged, RelationViolationError) as error: + self.startup_failed(f"Reconnect blocked before restart: {error}") + return await self.agent.stop() - self.load_admission = admission or EnsuringSessionLoadAdmission() + self.load_plan = plan or EnsuringSessionLoadPlan() self.settled.clear() await self.agent.start((self.agent.process,)) - await asyncio.wait_for(self.settled.wait(), timeout=30) - if not self.connected: - raise ValueError("The agent could not reconnect.") + try: + await asyncio.wait_for( + self.settled.wait(), timeout=self.reconnect_timeout_seconds) + except TimeoutError: + self.startup_failed( + f"Reconnect timed out after {self.reconnect_timeout_seconds} seconds " + "waiting for the session handshake to settle." + ) + except (OSError, jsonrpc.APIError) as error: + self.startup_failed(f"Reconnect failed: {type(error).__name__}: {error}") finally: self.reconnecting = False + async def reconnect_after_login(self) -> None: + """Only a session that is ready after its restart counts as a finished login.""" + await self.reconnect() + if not self.ready: + failure = self.disposition.current_failure() + raise LoginNotReady("Session did not become ready after login" if failure is None + else f"{failure.message}: {failure.details}") + async def authenticate(self, method_id: str) -> None: authority = ClientSessionRequest(self.agent, self.agent.session_id) authority.require() @@ -413,17 +435,17 @@ async def load(self) -> None: if agent_data := session.meta_json.agent_data: self.agent.definition = agent_data - admission, self.load_admission = self.load_admission, EnsuringSessionLoadAdmission() + plan, self.load_plan = self.load_plan, EnsuringSessionLoadPlan() with self.agent.request(): session_load_response = api.session_load(cwd, [], request_session_id, - admission.metadata()) + plan.metadata()) try: response = await session_load_response.wait() except jsonrpc.APIError as error: authority.require() if self.agent.queue_attachment.is_current_request(queue_token): - if failed_command := FailedSessionLoadAdmission.from_failure(error.data): - self.load_admission = failed_command + if failed_command := FailedSessionLoadPlan.from_failure(error.data): + self.load_plan = failed_command raise authority.require() if ( @@ -496,7 +518,7 @@ async def set_name(self, name: str) -> None: def publish_configuration(self, response): - self.load_admission = SessionLoadAdmission.at_response(response.field_meta) + self.load_plan = SessionLoadPlan.at_response(response.field_meta) if (modes := response.modes) is not None: self.agent.controller.publish_modes(modes) self.agent.configuration.receive(response.config_options) diff --git a/src/toad/app.py b/src/toad/app.py index b384b98df..05e414e6c 100644 --- a/src/toad/app.py +++ b/src/toad/app.py @@ -56,6 +56,7 @@ from toad.application_actions import ApplicationAction, ApplicationCommands from toad.widget_actions import DeclaredWidgetActions from toad.application_lifetime import ApplicationLifetime +from toad.resource_release import AsyncResourceRelease if TYPE_CHECKING: from toad.db import DB @@ -319,19 +320,18 @@ def __init__( async def _close_all(self) -> None: - await self.thread_navigation.close() - await self.thread_actions.close() - await self.coordination_access.stop_observing() - await super()._close_all() - # Native clients and the preparation owner have joined actual I/O. - # A failed retirement keeps its route resource instead of certifying exit. - await self.coordination_access.close() - - async def on_unmount(self) -> None: - self.colours.stop() - self.terminal_attention.close() - await self.navigation_reader.aclose() - await self.render_processes.aclose() + # View release can fail. The application still owns its observation + # pipe and process pools; an Unmount event may never follow that error. + async with AsyncResourceRelease() as release: + release.push_async_callback(self.render_processes.aclose) + release.push_async_callback(self.navigation_reader.aclose) + release.callback(self.terminal_attention.close) + release.callback(self.colours.stop) + release.push_async_callback(self.coordination_access.close) + release.push_async_callback(super()._close_all) + release.push_async_callback(self.coordination_access.stop_observing) + release.push_async_callback(self.thread_actions.close) + release.push_async_callback(self.thread_navigation.close) if self._background_render_tasks: await asyncio.gather(*tuple(self._background_render_tasks), return_exceptions=True) diff --git a/src/toad/navigation_preparation.py b/src/toad/navigation_preparation.py index d8a0ca5d0..948a398b2 100644 --- a/src/toad/navigation_preparation.py +++ b/src/toad/navigation_preparation.py @@ -14,7 +14,8 @@ from agent_comms.registry_document import RegistrySnapshot from agent_comms.declared_family import DeclaredFamily from agent_comms.thread_execution import ConversationPreparation -from agent_comms.session_load import ExistingSessionLoadAdmission +from agent_comms.session_load import ExistingSessionLoadPlan +from toad.owner_preparation import ExistingOwnerPlan from toad.session_tracker import CommsViewKey from toad.conversation_kind import ConversationKind @@ -99,10 +100,10 @@ def bind(self, screen) -> None: @dataclass(frozen=True) class NativeThreadNavigation(ThreadNavigation): - admission: ExistingSessionLoadAdmission + plan: ExistingSessionLoadPlan def bind(self, screen) -> None: - screen.presentation.sources.load_admission = self.admission + screen.presentation.sources.load_plan = self.plan @dataclass(frozen=True) @@ -157,13 +158,11 @@ def external(self) -> ThreadNavigation: return ExternalThreadNavigation(self.root, self.thread, self.project) def native(self) -> ThreadNavigation: - from toad.owner_preparation import ExistingOwnerAdmission - binding = self.thread.execution.owner_binding(self.snapshot, self.thread) - admission = ExistingOwnerAdmission.from_binding(binding) - if admission is None: + plan = ExistingOwnerPlan.from_binding(binding) + if plan is None: return ThreadNavigation(self.root, self.thread, self.project) - return NativeThreadNavigation(self.root, self.thread, self.project, admission) + return NativeThreadNavigation(self.root, self.thread, self.project, plan) class ThreadOpening: diff --git a/src/toad/owner_preparation.py b/src/toad/owner_preparation.py index 7136f055a..12b44a68f 100644 --- a/src/toad/owner_preparation.py +++ b/src/toad/owner_preparation.py @@ -4,13 +4,13 @@ from agent_comms.comms import Comms from agent_comms.thread_presentation import ThreadPresentation, ThreadOwnerBinding, LiveThreadOwnerBinding -from agent_comms.session_load import ExistingSessionLoadAdmission +from agent_comms.session_load import ExistingSessionLoadPlan from agent_comms.mro_dispatch import handles from toad.core.projection import MroProjection -class ExistingOwnerAdmission(MroProjection): - """Project Core's binding into read-only attachment admission, without starting an owner.""" +class ExistingOwnerPlan(MroProjection): + """Project Core's binding into a read-only attachment plan without starting an owner.""" @handles(ThreadOwnerBinding) def unavailable(self, binding): @@ -18,10 +18,10 @@ def unavailable(self, binding): @handles(LiveThreadOwnerBinding) def live(self, binding): - return ExistingSessionLoadAdmission(binding) + return ExistingSessionLoadPlan(binding) @classmethod - def from_binding(cls, binding: ThreadOwnerBinding) -> ExistingSessionLoadAdmission | None: + def from_binding(cls, binding: ThreadOwnerBinding) -> ExistingSessionLoadPlan | None: return cls().dispatch_sync(binding) diff --git a/src/toad/screens/main.py b/src/toad/screens/main.py index cd41ea4c7..1e9abc891 100644 --- a/src/toad/screens/main.py +++ b/src/toad/screens/main.py @@ -112,7 +112,16 @@ class ScreenContent(DeclaredFamily, affix="Content"): """Content state owns its label, preparation admission and completion.""" @abstractmethod - def render(self, screen: "MainScreen") -> Static: ... + def line(self) -> str: + """The opening line this content state shows; empty hides it.""" + + def render(self, screen: "MainScreen") -> Static: + """The one decision of the opening line: a session failure outranks the state's line.""" + agent = screen.presentation.operational_agent + if agent is not None and (failure := agent.session.disposition.current_failure()) is not None: + return self.label(screen, f"Thread disconnected: {failure.message}: {failure.details}", visible=True) + line = self.line() + return self.label(screen, line, visible=bool(line)) @abstractmethod async def wait(self, screen: "MainScreen") -> None: ... @@ -128,7 +137,8 @@ def close(self, screen: "MainScreen") -> None: """Terminal content has no unfinished waiters to release.""" async def attached(self, screen: "MainScreen") -> None: - """Admitted content already presents the live session.""" + """Refresh connection feedback from the session that just became ready.""" + self.render(screen) def label(self, screen: "MainScreen", text: str, *, visible: bool) -> Static: label = screen.query_one_optional("#session-opening", Static) @@ -143,10 +153,10 @@ class OpeningContent(ScreenContent): def __init__(self) -> None: self.completion = asyncio.get_running_loop().create_future() - def render(self, screen: "MainScreen") -> Static: + def line(self) -> str: from toad.widgets.conversation import ThreadLoading - return self.label(screen, ThreadLoading.LABEL, visible=True) + return ThreadLoading.LABEL async def wait(self, screen: "MainScreen") -> None: await asyncio.shield(self.completion) @@ -172,8 +182,8 @@ class ReadyContent(ScreenContent): async def prepare(self, screen: "MainScreen") -> None: await screen._prepare_content() - def render(self, screen: "MainScreen") -> Static: - return self.label(screen, "", visible=False) + def line(self) -> str: + return "" async def wait(self, screen: "MainScreen") -> None: """The content has already been admitted.""" @@ -183,8 +193,8 @@ class FailedContent(ScreenContent): def __init__(self, error: BaseException) -> None: self.error = error - def render(self, screen: "MainScreen") -> Static: - return self.label(screen, f"Thread unavailable: {self.error}", visible=True) + def line(self) -> str: + return f"Thread unavailable: {self.error}" async def wait(self, screen: "MainScreen") -> None: raise self.error @@ -452,10 +462,10 @@ async def on_session_attached(self, message: CoreEventMessage) -> None: @handles(core_events.AgentFail) def on_session_failure(self, message: CoreEventMessage) -> None: agent = self.presentation.operational_agent - if agent is not None and (failure := agent.session.disposition.current_failure()) is not None: - # Prompt/tool failures do not fail the session. Its disposition is - # the original handshake/connection authority, not this view. - self.opening_failed(RuntimeError(f"{failure.message}: {failure.details}")) + if agent is not None and agent.session.disposition.current_failure() is not None: + # A disconnected session is not a failed history-opening operation. + # Keep its reason on the session; every content view reads it there. + self.content.render(self) _last_dm_target: str | None = None diff --git a/src/toad/session_entry.py b/src/toad/session_entry.py index 6dbc7a351..6ccc63a61 100644 --- a/src/toad/session_entry.py +++ b/src/toad/session_entry.py @@ -13,6 +13,8 @@ from agent_comms.declared_family import DeclaredFamily from agent_comms.comms import wire +from agent_comms.route_selection import RouteSelection +from toad.owner_preparation import ExistingOwnerPlan from agent_comms.thread_identity import ThreadIncarnation from agent_comms.ui_model.unread import UnobservedUnread, UnreadPresentation @@ -109,8 +111,6 @@ def source(self, sessions: SessionNavigation) -> MainScreen | None: return sessions.app.workspace_sessions.views.get(self.mode) async def reconnect(self, sessions, selected, snapshot, targets): - from agent_comms.route_selection import RouteSelection - if sessions.get(self.mode) is not self or RouteSelection.capture() != selected: return source = self.source(sessions) @@ -123,16 +123,14 @@ async def reconnect(self, sessions, selected, snapshot, targets): if original is None: return thread = snapshot.require(snapshot.canonical_name(original.name)) - from toad.owner_preparation import ExistingOwnerAdmission - binding = thread.execution.owner_binding(snapshot, thread) - admission = ExistingOwnerAdmission.from_binding(binding) - if admission is None: + plan = ExistingOwnerPlan.from_binding(binding) + if plan is None: return if agent := source.presentation.operational_agent: - await agent.session.reconnect(admission) + await agent.session.reconnect(plan) else: - source.presentation.sources.attach_started(source, admission) + source.presentation.sources.attach_started(source, plan) def accepts_launch(self, sessions: SessionNavigation, agent_identity: str, session_id: str) -> bool: source = self.source(sessions) diff --git a/src/toad/session_presentation.py b/src/toad/session_presentation.py index 0e0e2cabb..f7afee8e4 100644 --- a/src/toad/session_presentation.py +++ b/src/toad/session_presentation.py @@ -1,4 +1,4 @@ -"""Session-owned mounted presentation, with one bounded workspace admission owner.""" +"""Session-owned mounted presentation, with one bounded workspace owner.""" import asyncio from abc import ABC, abstractmethod @@ -13,7 +13,8 @@ from textual.widgets.text_area import TextAreaState from agent_comms.acp_extension import CoordinationChangedUpdate -from toad.owner_preparation import ExistingOwnerAdmission +from agent_comms.session_load import EnsuringSessionLoadPlan +from toad.owner_preparation import ExistingOwnerPlan from toad.input_history import InputHistories from toad.agent_surface import AttachedSurfaceBinding @@ -101,11 +102,9 @@ class OperationalSessionSources: """Own existing operational resources while the view is detached.""" def __init__(self) -> None: - from agent_comms.session_load import EnsuringSessionLoadAdmission - self.agent = None self.shell = None - self.load_admission = EnsuringSessionLoadAdmission() + self.load_plan = EnsuringSessionLoadPlan() def status_row(self, conversation: Conversation): return self.activity_row(conversation).with_delivery(conversation.input_delivery) @@ -146,13 +145,13 @@ def input_ready(self, conversation: Conversation) -> bool: if conversation.transcript.readable: presentation = conversation.thread_presentation return (presentation is not None - and ExistingOwnerAdmission.from_binding(presentation.binding) is not None) + and ExistingOwnerPlan.from_binding(presentation.binding) is not None) return True def unavailable_reason(self, conversation: Conversation) -> str: presentation = conversation.thread_presentation if (conversation.transcript.readable and presentation is not None - and ExistingOwnerAdmission.from_binding(presentation.binding) is None): + and ExistingOwnerPlan.from_binding(presentation.binding) is None): return "Thread is stopped. Choose Start thread from its menu; your draft is kept." agent = conversation.agent if agent is not None and agent.configuration.model.requires_selection: @@ -160,7 +159,7 @@ def unavailable_reason(self, conversation: Conversation) -> str: return "Agent is not ready. Please wait while the agent connects…" def start_agent(self, conversation: Conversation) -> None: - """The operational source owns connection creation and load admission.""" + """The operational source owns connection creation and the load plan.""" from toad.acp.agent import Agent from toad.core import events as core_events @@ -169,16 +168,16 @@ def start_agent(self, conversation: Conversation) -> None: assert conversation._agent_data is not None agent = Agent(conversation.project_path, conversation._agent_data, conversation._agent_session_id, conversation._session_pk) - agent.session.load_admission = self.load_admission + agent.session.load_plan = self.load_plan conversation.agent = agent conversation._native_agent_started_here = True conversation.bind_agent(agent) agent.controller.start_operation(agent.start(conversation.app.session_navigation.processes)) conversation.publish_core(core_events.SessionSubtitleChanged(conversation.agent_title)) - def attach_started(self, screen: "MainScreen", admission) -> None: + def attach_started(self, screen: "MainScreen", plan) -> None: """An explicit Start action may attach the same retained view to its new owner.""" - self.load_admission = admission + self.load_plan = plan screen._agent = screen._agent or screen.app.agent_data conversation = screen.presentation.widget if conversation is not None: diff --git a/src/toad/widgets/conversation.py b/src/toad/widgets/conversation.py index c6acfae9f..7c1309ef7 100644 --- a/src/toad/widgets/conversation.py +++ b/src/toad/widgets/conversation.py @@ -1209,7 +1209,7 @@ async def action_provider_login(self) -> None: ) is None: await self.prune_window(0, 0) self.output.boundary() - await agent.session.reconnect() + await agent.session.reconnect_after_login() else: await agent.session.authenticate(method_id) self.flash(