P68: keep reconnect failure in its session and finish cleanup

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)

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)
 

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:

src/toad/owner_preparation.py

@@ -4,24 +4,24 @@
 
 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):
         return None
 
     @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)
 
 

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