P404: repair install lifetime and boundary ownership

src/agent_comms/cli_bridges/shared.py

@@ -617,79 +617,106 @@ class SessionBridge:
     claim: WatchClaim
     observed: SessionInformation
 
-    def pause(self, resources: ExitStack) -> tuple[PausedSessionBridge, ...]:
+    def pause(self, resources: ExitStack) -> BridgePause:
         if not self.observed.owner.alive():
             if self.observed.process.alive():
                 self.observed.process.signal(signal.SIGTERM)
-            print(f'Session bridge {self.claim.thread}: owner exited; left stopped', file=sys.stderr)
-            return ()
+            return OwnerExitedBridge(self.claim.thread)
         self.observed.owner.require_alive()
         peer = resources.enter_context(PauseSession(
             owner=self.observed.owner, process=self.observed.process).connection(self.claim))
         stream = resources.enter_context(peer.makefile('rb'))
         REPLIES.validate_json(stream.readline(4096)).require()
-        return (PausedSessionBridge(self, peer),)
+        return PausedSessionBridge(self, peer)
 
 
-class SessionBridgeResumeError(ValueError):
-    """A paused session bridge did not come back as the same process on the new runtime."""
+class BridgePause(ABC):
+    """A captured bridge either awaits resume or already has its final outcome."""
 
+    @abstractmethod
+    def resume(self) -> BridgeResume: ...
 
-@dataclass
-class SessionBridgeResumes:
-    """Every bridge an install paused. One failed resume cancels neither another resume nor
-    the install steps after it: each failure is reported, then the install exits non-zero."""
 
-    paused: list[PausedSessionBridge] = field(default_factory=list)
-    failures: list[SessionBridgeResumeError] = field(default_factory=list)
+@dataclass(frozen=True)
+class BridgeResume(DeclaredFamily, BridgePause, affix="Bridge"):
+    """One captured bridge's outcome; it owns its final report line."""
 
-    def pause(self, bridge: SessionBridge, resources: ExitStack) -> None:
-        self.paused.extend(bridge.pause(resources))
+    name: str
 
-    def resume(self) -> None:
-        for paused in self.paused:
-            try:
-                paused.resume()
-            except SessionBridgeResumeError as error:
-                self.failures.append(error)
+    def resume(self) -> BridgeResume:
+        return self
+
+    def line(self) -> str:
+        return f"Session bridge {self.name}: {self.description()}"
+
+    @abstractmethod
+    def description(self) -> str: ...
+
+
+@dataclass(frozen=True)
+class ResumedBridge(BridgeResume):
+    from_runtime: LaunchRuntime
+    to_runtime: LaunchRuntime
+
+    def description(self) -> str:
+        return f"{self.from_runtime.prefix} -> {self.to_runtime.prefix}"
 
-    def lines(self) -> tuple[str, ...]:
-        return tuple(f'Not resumed: {error}' for error in self.failures)
 
-    def require_resumed(self) -> None:
-        if self.failures:
-            raise ExceptionGroup('Session bridges not resumed', self.failures)
+@dataclass(frozen=True)
+class OwnerExitedBridge(BridgeResume):
+    def description(self) -> str:
+        return "owner exited; left stopped"
+
+
+@dataclass(frozen=True)
+class ReplacedBridge(BridgeResume):
+    observed: ProcessIdentity
+    current: ProcessIdentity
+
+    def description(self) -> str:
+        return (f"process or owner changed (observed process {self.observed.pid}"
+                f"@{self.observed.start_time}, current process {self.current.pid}"
+                f"@{self.current.start_time}); re-arm the session bridge")
 
 
 @dataclass(frozen=True)
-class PausedSessionBridge:
+class UnobservedBridge(BridgeResume):
+    reason: str = "successor runtime was not observed"
+
+    def description(self) -> str:
+        return f"{self.reason}; re-arm the session bridge"
+
+
+@dataclass(frozen=True)
+class PausedSessionBridge(BridgePause):
     bridge: SessionBridge
     peer: socket.socket
 
-    def resume(self) -> str:
+    def resume(self) -> BridgeResume:
         runtime = LaunchRuntime.installed()
         observed = self.bridge.observed
         name = self.bridge.claim.thread
         if not observed.owner.alive():
-            print(f'Session bridge {name}: owner exited; left stopped', file=sys.stderr)
-            return f'{name}: owner exited; left stopped'
-        self.peer.sendall((json.dumps(FieldCodec.encode(runtime)) + '\n').encode())
+            return OwnerExitedBridge(name)
+        try:
+            self.peer.sendall((json.dumps(FieldCodec.encode(runtime)) + '\n').encode())
+        except OSError as error:
+            return UnobservedBridge(name, f"{type(error).__name__}: {error}")
         deadline = time.monotonic() + 30
         while observed.process.alive() and time.monotonic() < deadline:
             try:
                 current = InspectSession().information(self.bridge.claim)
             except (FileNotFoundError, ConnectionRefusedError):
                 time.sleep(.05)
                 continue
+            except (OSError, ValueError) as error:
+                return UnobservedBridge(name, f"{type(error).__name__}: {error}")
             if current.process != observed.process or current.owner != observed.owner:
-                raise SessionBridgeResumeError(f'Session bridge {name}: process or owner changed')
+                return ReplacedBridge(name, observed.process, current.process)
             if current.runtime.prefix != runtime.prefix:
-                raise SessionBridgeResumeError(
-                    f'Session bridge {name}: wrong runtime {current.runtime.prefix}')
-            print(f'Session bridge {name}: {observed.runtime.prefix} -> {current.runtime.prefix}',
-                  file=sys.stderr)
-            return name
-        raise SessionBridgeResumeError(f'Session bridge {name}: successor runtime was not observed')
+                return UnobservedBridge(name)
+            return ResumedBridge(name, observed.runtime, current.runtime)
+        return UnobservedBridge(name)
 
 
 class Enrol(Operation):

src/agent_comms/cli_commands.py

@@ -76,6 +76,7 @@
     UnsetScopeFlag,
 )
 from .field_codec import (
+    AbsolutePathText,
     FieldCodec,
     FieldPlan,
     PathText,
@@ -1603,7 +1604,7 @@ class InstallCliCommand(CliCommand):
             "harness to use the bus, with the commander skill and tools")
     checkout: Annotated[
         Path,
-        PathText,
+        AbsolutePathText,
         Argument(flags=("--checkout",)),
     ] = dataclasses.field(default_factory=Path.cwd)
     to: Annotated[str, Argument(flags=("--to",))] = "origin/main"
@@ -1616,7 +1617,7 @@ class InstallCliCommand(CliCommand):
     ] = UnnamedInstaller()
     rehearse: Annotated[
         bool,
-        Argument(flags=("--rehearse",), options={"help": "Carry and read a copy of the bus; change nothing"}),
+        Argument(flags=("--rehearse",), options={"help": "Carry and read a copy; publish its report without installing"}),
     ] = False
     retire: Annotated[
         bool,
@@ -1627,15 +1628,15 @@ class InstallCliCommand(CliCommand):
     ] = False
     spill_dir: Annotated[
         Path,
-        PathText,
+        AbsolutePathText,
         Argument(
             flags=("--spill-dir",),
             options={"help": "Parent directory for live backups"},
         ),
     ] = dataclasses.field(default_factory=lambda: Path.home() / ".cache/agent-rehearsal")
     project: Annotated[
         Path,
-        PathText,
+        AbsolutePathText,
         Argument(flags=("--project",), options={"help": "The worktree harness threads register in"}),
     ] = dataclasses.field(default_factory=Path.cwd)
     repos: Annotated[
@@ -1660,7 +1661,7 @@ def execute(self, route: CommsRoute, output: TextIO) -> int:
 
         return InstallRun.start(self, route.root).launch(output)
 
-    def apply(self, ctx: Comms) -> CliResult:
+    def apply(self, ctx: Comms) -> InstallResult:
         """The install child's last step, on the bus it carried or created."""
         home = Path.home()
         record = self.commander.configured(CommanderRecord.read(ctx.root), self.repos, ctx.root, home)

src/agent_comms/commander/config.py

@@ -19,6 +19,7 @@
 from ..comms import Comms
 from ..errors import RelationViolationError
 from ..field_codec import (
+    AbsolutePathText,
     BoundedText,
     FieldCodec,
     NullMember,
@@ -245,11 +246,6 @@ def founding(self, commander: str, root: Path, home: Path) -> RecordedCommander:
     def without_commander(self) -> None:
         """Refuse repositories given to a bus that records no commander to hold them."""
 
-    @abstractmethod
-    def resolved(self) -> Self:
-        """This choice with its paths absolute, for an install child that runs elsewhere."""
-
-
 @dataclass(frozen=True, slots=True)
 class GivenRepos(RepoChoice):
     """Checkouts spelled as comma-separated paths on the command line."""
@@ -259,7 +255,8 @@ class GivenRepos(RepoChoice):
     @classmethod
     def from_wire(cls, data: object) -> Self:
         text = FieldCodec.decode(str, data)
-        return cls(tuple(Path(part.strip()) for part in text.split(",") if part.strip()))
+        return cls(tuple(AbsolutePathText.from_text(part.strip())
+                         for part in text.split(",") if part.strip()))
 
     def to_wire(self) -> str:
         return ",".join(map(str, self.checkouts))
@@ -274,10 +271,6 @@ def without_commander(self) -> None:
         raise UpgradeRefused("--repos names a commander's repositories; this bus records no "
                              "commander, so name one with --commander")
 
-    def resolved(self) -> Self:
-        return type(self)(tuple(checkout.resolve() for checkout in self.checkouts))
-
-
 @dataclass(frozen=True, slots=True)
 class KeptRepos(NullMember, RepoChoice):
     def applied(self, recorded: tuple[Repo, ...]) -> tuple[Repo, ...]:
@@ -290,10 +283,6 @@ def founding(self, commander: str, root: Path, home: Path) -> RecordedCommander:
     def without_commander(self) -> None:
         pass
 
-    def resolved(self) -> Self:
-        return self
-
-
 class CommanderConfigFile(SubclassUnion, affix="ConfigFile"):
     """What an install left for commander tools: the config it wrote, or none."""
 

src/agent_comms/connector.py

@@ -52,7 +52,7 @@
     boundary_decoder,
 )
 from .goal_actions import DirectiveGoalAction
-from .host_services import HostService, SystemdUserService
+from .host_services import BoundListener, HostService, SystemdUserService
 from .connector_store import (
     CODE_SECONDS, ApprovalQueueFull, ApprovalRefused, AuthorizationQuery,
     AuthorizationRequest, ClaudeCallback,
@@ -894,8 +894,9 @@ async def serve(self, app: web.Application) -> None:
         for signum in (signal.SIGINT, signal.SIGTERM):
             loop.add_signal_handler(signum, stopped.set)
         try:
-            await web.TCPSite(runner, LOOPBACK, self.port).start()
-            SystemdUserService.ready()
+            site = web.TCPSite(runner, LOOPBACK, self.port)
+            await site.start()
+            SystemdUserService.ready(BoundListener.of(site))
             await stopped.wait()
         finally:
             await runner.cleanup()

src/agent_comms/field_codec.py

@@ -739,7 +739,7 @@ class ExternalNull(ExternalField, MroDispatch):
 
     @classmethod
     def normalize_external(cls, value: object, record: Mapping[str, object]) -> object:
-        handler = getattr(cls, next(cls.handler_names_for(type(value))))
+        handler = next(cls.class_handlers_for(type(value)))
         return handler(value, record)
 
     @handles(type(None))
@@ -788,7 +788,7 @@ class FieldRepresentation(MroDispatch, ABC):
 
     @classmethod
     def decode_field(cls, data: object, structural: ValueDecoder[object]) -> object:
-        handler = getattr(cls, next(cls.handler_names_for(type(data))))
+        handler = next(cls.class_handlers_for(type(data)))
         with MalformedRecord.refusing(f"Invalid {cls.__name__}"):
             return cast(object, handler(data, structural))
 
@@ -1110,6 +1110,14 @@ def from_text(cls, value: str) -> Path:
         return Path(value)
 
 
+class AbsolutePathText(PathText):
+    """An install path bound to the caller's directory before its child changes cwd."""
+
+    @classmethod
+    def from_text(cls, value: str) -> Path:
+        return Path(value).expanduser().resolve()
+
+
 class BoundedText(TextRepresentation[str], ValidatedScalar):
     """Nonempty identity text of bounded length: an owner, incarnation or message ID."""
 
@@ -1259,8 +1267,8 @@ def normalize_external(
 
     @classmethod
     def external_normalizers(cls, target: type) -> Iterator[ExternalNormalizer]:
-        for name in cls.handler_names_for(target):
-            yield getattr(cls, name)(target)
+        for handler in cls.class_handlers_for(target):
+            yield cast(ExternalNormalizer, handler(target))
 
     @handles(ExternalField)
     @classmethod

src/agent_comms/host_services/__init__.py

@@ -13,18 +13,46 @@
 import subprocess
 import time
 from abc import abstractmethod
+from asyncio import Server
 from dataclasses import dataclass
 from pathlib import Path
-from typing import ClassVar, Self
+from typing import ClassVar, Self, cast
+
+from aiohttp.web import TCPSite
 
 from ..declared_family import DeclaredFamily
+from ..field_codec import boundary_decoder
 from ..launch_runtime import LaunchRuntime
 
 
 class ServiceNotServingError(RuntimeError):
     """A restarted service did not serve its installed runtime's contract."""
 
 
+@dataclass(frozen=True)
+class BoundListener:
+    """A listening socket observed on the site that just started."""
+
+    site: TCPSite
+
+    def __new__(cls, _site: TCPSite) -> Self:
+        raise TypeError("Use of(site) to observe a listening socket")
+
+    @classmethod
+    @boundary_decoder
+    def of(cls, site: TCPSite) -> Self:
+        server = site._server
+        if server is None:
+            raise ServiceNotServingError("The connector site has not started")
+        # TCPSite.start assigns loop.create_server's Server to BaseSite's wider annotation.
+        for bound in cast(Server, server).sockets:
+            if bound.getsockopt(socket.SOL_SOCKET, socket.SO_ACCEPTCONN):
+                listener = object.__new__(cls)
+                cls.__init__(listener, site)
+                return listener
+        raise ServiceNotServingError("The connector site has no listening socket")
+
+
 class HostService(DeclaredFamily, affix="Service", discovery_package=__name__):
     """One running instance of a declared service on this host."""
 
@@ -85,8 +113,8 @@ class SystemdUserService(HostService, declared_name=None):
     notify_variable: ClassVar[str] = "NOTIFY_SOCKET"  # systemd's spelling of the supervisor socket
 
     @classmethod
-    def ready(cls) -> None:
-        """Tell the supervisor this process serves.
+    def ready(cls, listener: BoundListener) -> None:
+        """Tell the supervisor this process has a bound listener.
 
         Every launch has one: systemd for the installed unit, ``tests/connector_readiness.py``
         for a test. A launch without NOTIFY_SOCKET has nobody to tell and fails here, loudly.

src/agent_comms/installer/backup.py

@@ -26,6 +26,7 @@
 from agent_comms.coordination_errors import CoordinationError
 from agent_comms.external_transcript import ExternalTranscript
 from agent_comms.field_codec import FieldCodec
+from agent_comms.installer.outcome import InstallFailure
 from agent_comms.locked_store import LockedStore, RuntimeFile
 from agent_comms.store_files import _atomic_write_text
 
@@ -41,7 +42,7 @@
 """
 
 
-class UpgradeRefused(CoordinationError):
+class UpgradeRefused(InstallFailure, CoordinationError):
     """The update stopped without permission to discard its original data."""
     prefix: ClassVar[str] = "install: refused:"
 
@@ -51,6 +52,9 @@ def __str__(self) -> str:
     def lines(self) -> tuple[str, ...]:
         return (f"Install failed: {self}",)
 
+    def exception(self) -> BaseException:
+        return self
+
     @classmethod
     def public_message(cls, output: str) -> str:
         for line in reversed(output.splitlines()):