Skip to content
3 changes: 0 additions & 3 deletions src/toad/acp/agent_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,6 @@ def __init__(self, agent) -> None:
self.connection: ClientSideConnection | None = None
self.runner = None
self.retirement = None
self.cleanup = ExitStack()
self.sessions = set()
self.initialization = None
self.command = None
Expand Down Expand Up @@ -92,7 +91,6 @@ async def start(self, agent, processes: tuple[AgentProcess, ...]=()):
connection.sessions.add(agent.session)
agent.controller.replace_terminal_session()
connection.runner = asyncio.create_task(connection.run())
connection.cleanup.enter_context(acquisition.pop_all())
return connection

async def prompt(self, agent, content, metadata: dict[str, JsonValue], require: Callable[[], None]):
Expand Down Expand Up @@ -243,7 +241,6 @@ async def _retire(self) -> None:
await self.process.stop()
self.process = None
self.sessions.clear()
self.cleanup.close()

async def detach(self, session: AgentSession):
async with self.attachment_lock:
Expand Down
27 changes: 24 additions & 3 deletions src/toad/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,10 @@
from functools import cached_property, partial
from pathlib import Path
from time import monotonic
from typing import TYPE_CHECKING, Callable, TypeVar, cast
from typing import TYPE_CHECKING, Callable, Never, TypeVar, cast

from agent_comms.acp_extension import TranscriptChangedUpdate
from rich.console import RenderableType
from textual._time import after_input
from textual import events, on, work
from textual.app import App
Expand Down Expand Up @@ -58,10 +59,11 @@
from toad.textual_colours import ThemedApp
from toad.application_actions import ApplicationCommands
from toad.widget_actions import ApplicationAction, DeclaredWidgetActions
from toad.application_lifetime import ApplicationLifetime
from toad.application_lifetime import ApplicationLifetime, Ended, Succeeded, TerminalOutcome
from toad.resource_release import AsyncResourceRelease

if TYPE_CHECKING:
from agent_comms.launch_runtime import LaunchRuntime
from toad.db import DB
from toad.render_backend import RenderTask
from toad.screens.main import MainScreen
Expand Down Expand Up @@ -127,7 +129,7 @@ def __new__(cls, minimum_width: int, side_panel_width: Scalar | None):
WIDE = (100, Scalar(50, Unit.PERCENT, Unit.WIDTH))


class ToadApp(CoreEventReceiver, DeclaredWidgetActions, WorkspaceSessionShutdown, ThemedApp, App, inherit_bindings=False):
class ToadApp(CoreEventReceiver, DeclaredWidgetActions, WorkspaceSessionShutdown, ThemedApp, App[TerminalOutcome], inherit_bindings=False):
"""The top level app."""

CSS_PATH = ["toad.tcss", "screens/comms.tcss"]
Expand Down Expand Up @@ -193,9 +195,28 @@ def __init__(
from toad.file_kind import FileKind
FileKind.prepare_terminal()
super().__init__()
self._return_value: TerminalOutcome = Ended(self.application.version)
self.session_navigation.bind_events()
self.project_dir = Path(project_dir or "./").expanduser().resolve()

@property
def return_value(self) -> TerminalOutcome:
return self._return_value

def exit(
self, result: TerminalOutcome | None = None, return_code: int = 0,
message: RenderableType | None = None,
) -> None:
"""Textual's omitted result is a normal terminal ending."""
super().exit(
Ended(self.application.version) if result is None else result,
return_code=return_code, message=message,
)

def succeed_onto(self, runtime: LaunchRuntime) -> Never:
self.call_from_thread(self.exit, Succeeded(runtime))
raise asyncio.CancelledError


async def _close_all(self) -> None:
# View release can fail. The application still owns its observation
Expand Down
27 changes: 27 additions & 0 deletions src/toad/application_lifetime.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
"""Application startup, graceful exit and optional reporting own their lifetime."""
from __future__ import annotations

import os
import sys
from abc import ABC, abstractmethod
from datetime import datetime, timezone
from functools import cached_property, partial
from importlib import import_module
Expand All @@ -15,9 +18,33 @@
if TYPE_CHECKING:
from ssl import SSLContext

from agent_comms.launch_runtime import LaunchRuntime
from toad.app import ToadApp


class TerminalOutcome(ABC):
@abstractmethod
def finish(self) -> None: ...


@dataclass(frozen=True)
class Ended(TerminalOutcome):
version: VersionMonitor

def finish(self) -> None:
self.version.status.print_notice()


@dataclass(frozen=True)
class Succeeded(TerminalOutcome):
runtime: LaunchRuntime

def finish(self) -> None:
self.runtime.require_available()
executable = str(self.runtime.interpreter.with_name("toad"))
os.execve(executable, [executable, *sys.argv[1:]], self.runtime.environment(os.environ))


class UsageReports:
"""All producers use one opt-in policy, identity and external HTTP boundary."""

Expand Down
2 changes: 1 addition & 1 deletion src/toad/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ def run_terminal(app: ToadApp) -> None:
persist_terminal_failure(app._exception)
if app._exception is not None:
raise SystemExit(app.return_code or 1)
app.application.version.status.print_notice()
app.return_value.finish()


def check_directory(path: str) -> None:
Expand Down
40 changes: 21 additions & 19 deletions src/toad/comms_root.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,16 @@
from weakref import WeakKeyDictionary, WeakSet

from agent_comms.child_process import join_retirement
from agent_comms.comms import Comms
from agent_comms.coordination_errors import StaleRevision
from agent_comms.display_references import PendingReferences, UnavailableReferences
from agent_comms.errors import RouteChangedError
from agent_comms.launch_runtime import LaunchRuntime
from agent_comms.mro_dispatch import MroDispatch, handles
from agent_comms.presentation import MessageNotification, NoRevision, ReadState
from agent_comms.revision_views import ExternalTranscriptsView, StatusView
from agent_comms.route_selection import RouteSelection, run_selected_write
from agent_comms.root_runtime import RootRuntime
from agent_comms.ui_model.observation import (
NotificationsObserved,
NotificationsUnreadable,
Expand Down Expand Up @@ -65,7 +68,6 @@
from toad.resource_release import AsyncResourceRelease, ResourceRelease

if TYPE_CHECKING:
from agent_comms.comms import Comms
from agent_comms.display_order import DisplayOrder
from agent_comms.display_references import ReferenceContext
from agent_comms.thread_identity import ThreadIncarnation
Expand Down Expand Up @@ -334,7 +336,6 @@ def __init__(self, app: ToadApp) -> None:
self.revision: ReadState = NoRevision()
self.route_task: asyncio.Task[None] | None = None
self.timer = None
self.cleanup = ExitStack()
self.opening = threading.RLock()
self.preparation = app.preparation
self.retiring: set[asyncio.Task[None]] = set()
Expand Down Expand Up @@ -369,22 +370,24 @@ def require(self, selected: RouteSelection) -> Observing:
return self.observation.require(self, selected)

def open_selected(self, selected: RouteSelection) -> Observing:
from agent_comms.comms import Comms

previous = self.observation
with ExitStack() as acquisition:
acquisition.enter_context(selected.route.hold_client_route())
service = Comms(selected.route)
if service.root.resolve() != selected.root or RouteSelection.capture() != selected:
raise RouteChangedError("Comms route changed while opening the service")
observed = Observing(
selected, service, SidebarModel(self.app.call_next), ObservationProcess(service.route),
self.app.running_loop, access=self)
self.cleanup.enter_context(acquisition.pop_all())
observed.subscriptions.callback(observed.retire, self)
self.observation = observed
observed.loop.call_soon_threadsafe(observed.attach, self, previous)
return observed
def attach() -> Observing:
previous = self.observation
with ExitStack() as acquisition:
acquisition.enter_context(selected.route.hold_client_route())
service = Comms(selected.route)
if service.root.resolve() != selected.root or RouteSelection.capture() != selected:
raise RouteChangedError("Comms route changed while opening the service")
observed = Observing(
selected, service, SidebarModel(self.app.call_next), ObservationProcess(service.route),
self.app.running_loop, access=self)
observed.subscriptions.callback(observed.retire, self)
self.observation = observed
observed.loop.call_soon_threadsafe(observed.attach, self, previous)
return observed

return RootRuntime.attachment(selected.root, LaunchRuntime.running()).proceed(
attach, self.app.succeed_onto,
)

@handles(RevisionObserved)
def revision_observed(self, result: RevisionObserved, observed: Observing) -> None:
Expand Down Expand Up @@ -586,7 +589,6 @@ async def stop_observing(self) -> None:
async def close(self) -> None:
"""Retire the observation once the views it served have closed."""
async with AsyncResourceRelease() as release:
release.callback(self.cleanup.close)
try:
with ResourceRelease() as observation:
observation.callback(lambda: self.observation.subscriptions.close())
Expand Down
1 change: 0 additions & 1 deletion tests/model/test_sessions_share_one_stdio_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,6 @@ async def test_closing_one_session_leaves_the_others_on_the_shared_process(attac
await agent.stop()
await asyncio.gather(*permissions, return_exceptions=True)
assert shared is not None and shared.process is None and not shared.sessions
assert not shared.cleanup._exit_callbacks


async def test_stopping_a_session_the_agent_refused_leaves_the_shared_process(attachment_scope):
Expand Down
4 changes: 0 additions & 4 deletions tests/textual/test_observation_shutdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,22 +2,19 @@

import asyncio
import sys
import threading
import pytest

from agent_comms.ui_model.observation import ObservationProcess, ObserverDied, ObserveNotifications, ReadNotifications
from toad.comms_root import AnsweringReads


async def test_an_observer_dying_during_shutdown_still_releases_its_resources(workspace):
cleaned = threading.Event()
with pytest.raises(BaseExceptionGroup) as failures:
async with workspace.app() as ui:
access = ui.app.coordination_access
await access.current_sidebar()
observed = access.observation.observed()
process = observed.process
access.cleanup.callback(cleaned.set)
# 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())
Expand All @@ -29,7 +26,6 @@ async def test_an_observer_dying_during_shutdown_still_releases_its_resources(wo
assert process.connection.closed
assert not process.process.is_alive()
assert not access.retiring
assert cleaned.is_set()



Expand Down
Loading