diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py index 7877412183..e3c798f7cc 100644 --- a/src/agent_comms/cli_bridges/shared.py +++ b/src/agent_comms/cli_bridges/shared.py @@ -617,63 +617,91 @@ 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: @@ -681,15 +709,14 @@ def resume(self) -> str: 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): diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index d90dc32075..f50ac42039 100644 --- a/src/agent_comms/cli_commands.py +++ b/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,7 +1628,7 @@ class InstallCliCommand(CliCommand): ] = False spill_dir: Annotated[ Path, - PathText, + AbsolutePathText, Argument( flags=("--spill-dir",), options={"help": "Parent directory for live backups"}, @@ -1635,7 +1636,7 @@ class InstallCliCommand(CliCommand): ] = 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) diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index dbca5bdbd0..c6814118a7 100644 --- a/src/agent_comms/commander/config.py +++ b/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.""" diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index 8a775a8fa0..7f88162fa6 100644 --- a/src/agent_comms/connector.py +++ b/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() diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index c3783daf06..27ecaf9c5a 100644 --- a/src/agent_comms/field_codec.py +++ b/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 diff --git a/src/agent_comms/host_services/__init__.py b/src/agent_comms/host_services/__init__.py index bf3c6bbb62..a2b01f0b00 100644 --- a/src/agent_comms/host_services/__init__.py +++ b/src/agent_comms/host_services/__init__.py @@ -13,11 +13,15 @@ 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 @@ -25,6 +29,30 @@ 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. diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index 7f9d1b1f0e..59184eda84 100644 --- a/src/agent_comms/installer/backup.py +++ b/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()): diff --git a/src/agent_comms/installer/launch.py b/src/agent_comms/installer/launch.py index e05581d751..c57b9b3b75 100644 --- a/src/agent_comms/installer/launch.py +++ b/src/agent_comms/installer/launch.py @@ -11,72 +11,166 @@ import subprocess import sys import tempfile -from dataclasses import dataclass, replace +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field from pathlib import Path -from typing import Annotated, ClassVar, TextIO +from traceback import FrameSummary, extract_tb +from typing import Annotated, ClassVar, TextIO, cast import agent_comms.installer from agent_comms.cli_commands import InstallCliCommand, InstallResult from agent_comms.comms import client_wire -from agent_comms.bus_publication import AgentOrigin -from agent_comms.field_codec import FieldCodec, Id128Text, PathText -from agent_comms.installer.backup import UpgradeRefused -from agent_comms.installer.upgrade import TargetTree, Upgrade, run, upgrade -from agent_comms.messages import InfoKind, Message +from agent_comms.field_codec import AbsolutePathText, FieldCodec, Id128Text, PathText +from agent_comms.installer.outcome import ( + CarriedInstallError, + CrashedInstall, + FailedInstall, + FinishedInstall, + InstallFailure, + InstallOutcome, + UnfinishedInstall, +) +from agent_comms.installer.reporting import InstalledWriter, RootWriter +from agent_comms.installer.runtime_resolution import CorePackage +from agent_comms.installer.upgrade import ( + CarriedPass, + IncompleteRun, + Install, + LiveRun, + Stack, + TargetTree, + Upgrade, + run, + upgrade, +) +from agent_comms.launch_runtime import LaunchRuntime +from agent_comms.mro_dispatch import MroDispatch, handles +from agent_comms.repo import Repo from agent_comms.service_sender import ServiceSender from agent_comms.storage_rules import StorageRule -@dataclass(frozen=True) -class InstallReport(ServiceSender): - """Publish the install's own typed result, never an interpretation of its logs.""" +@dataclass +class InstallReport(ServiceSender, MroDispatch): + """One run owns its writing permission and exactly one terminal result.""" sender: ClassVar[str] = "installer" + channel: ClassVar[str] = "comms" root: Path + outgoing: Stack + writer: RootWriter = field(init=False) + outcome: InstallOutcome = field(init=False) - def post(self, lines: tuple[str, ...]) -> Message: - with client_wire(self.root) as comms: - self.register(comms.registry) - return comms.bus.publisher.publish_ordinary( - Message(sender=self.sender, target="#comms", body="\n".join(lines), - type=InfoKind), origin=AgentOrigin()) - - def starting(self, upgrade: Upgrade) -> Message: - return self.post(("Install starting", *upgrade.lines())) + def __post_init__(self) -> None: + self.writer = InstalledWriter(self.outgoing) + self.outcome = UnfinishedInstall(IncompleteRun()) - def finished(self, upgrade: Upgrade) -> Message: - return self.post(("Install finished", *upgrade.lines())) - - def failed(self, error: UpgradeRefused) -> Message: - return self.post(error.lines()) + @classmethod + def for_run(cls, request: InstallRun) -> InstallReport: + return cls(request.root, Stack(request.command.checkout, runtime=LaunchRuntime.running())) + + @property + def target(self) -> str: + return f"#{self.channel}" + + @property + def source(self) -> Repo: + return Repo(checkout=self.outgoing.checkout, github=CorePackage.repository) + + def writing(self, writer: RootWriter) -> None: + self.writer = writer + writer.prepare(self) + + @handles(CarriedPass) + def carried(self, passed: CarriedPass) -> None: + if passed.root != self.root: + raise ValueError("Carried pass belongs to a different install root") + self.writing(passed.writer()) + + def record_live(self, result: LiveRun) -> LiveRun: + self.outcome = UnfinishedInstall(result) + return result + + def complete(self, result: Upgrade) -> Upgrade: + self.outcome = FinishedInstall(result) + return result + + def failure(self, error: BaseException) -> InstallFailure: + """Decode once at the exception boundary using its nominal capabilities.""" + return cast(InstallFailure, next(self.handlers_for(error))(error)) + + @handles(InstallFailure) + def reportable(self, error: InstallFailure) -> InstallFailure: + return error + + @handles(CarriedInstallError) + def carried_failure(self, error: CarriedInstallError) -> InstallFailure: + self.carried(error.passed) + return self.failure(error.error) + + @handles(BaseException) + def crashed(self, error: BaseException) -> CrashedInstall: + # The context supplies a Core frame even if the failing library is external. + # Its package directory works in both a checkout and a built wheel. + package = Path(__file__).resolve().parents[1] + frame = next( + frame + for frame in reversed(extract_tb(error.__traceback__)) + if Path(frame.filename).resolve().is_relative_to(package) + ) + module = ".".join( + Path(frame.filename).resolve().relative_to(package.parent).with_suffix("").parts + ) + location = FrameSummary( + str(self.source.module_file(module)), frame.lineno, frame.name, lookup_line=False, + ) + return CrashedInstall(error, location) + + @contextmanager + def running(self, request: InstallRun) -> Iterator[type[Install]]: + try: + with TargetTree.from_run(request): + member = Install.member(self.root, request.command.retire) + self.writing(member.initial_writer(self.root, self.outgoing)) + self.writer.post(self.root, self, ( + "Install starting", f"Target: {request.target[:12]}", + )) + yield member + # A caller cannot accidentally leave the context without its result. + self.outcome.lines() + except BaseException as error: + failure = self.failure(error) + self.outcome = FailedInstall(self.outcome.live_run(), failure) + original = failure.exception() + raise original from original.__cause__ + finally: + self.writer.post(self.root, self, self.outcome.lines()) @dataclass(frozen=True) class InstallRun: command: InstallCliCommand root: Annotated[Path, PathText] - scratch: Annotated[Path, PathText] + scratch: Annotated[Path, AbsolutePathText] target: str @classmethod def start(cls, command: InstallCliCommand, root: Path) -> InstallRun: - checkout = command.checkout.resolve() - run("git", "-C", checkout, "fetch", "-q", "origin") - target = run("git", "-C", checkout, "rev-parse", "--verify", + run("git", "-C", command.checkout, "fetch", "-q", "origin") + target = run("git", "-C", command.checkout, "rev-parse", "--verify", f"{command.to}^{{commit}}").strip() parent = command.installer.scratch(StorageRule.load(root)) parent.mkdir(mode=0o700, parents=True, exist_ok=True) - scratch = Path(tempfile.mkdtemp(dir=parent, prefix="install-")) - TargetTree(checkout, scratch, target).add() - # Paths resolve here: the install child runs in the target tree, not the caller's cwd. - return cls(replace(command, checkout=checkout, project=command.project.resolve(), - repos=command.repos.resolved()), - root, scratch, target) + scratch = AbsolutePathText.from_text(tempfile.mkdtemp(dir=parent, prefix="install-")) + request = cls(command, root, scratch, target) + TargetTree.from_run(request).add() + return request def launch(self, output: TextIO) -> int: """The install outlives its caller: a new session, its output in files beside the bus. Retiring the caller's process group must not cut an install short.""" - tree = TargetTree(self.command.checkout, self.scratch, self.target) + tree = TargetTree.from_run(self) logs = self.root.parent / f".{self.root.name}.installer-{Id128Text.mint()}" logs.mkdir(mode=0o700) result, errors = logs / "result.json", logs / "stderr.log" @@ -97,10 +191,7 @@ def launch(self, output: TextIO) -> int: return process.returncode def run(self) -> Upgrade: - command = self.command - return upgrade(command.checkout, self.root, self.target, command.installer, - command.rehearse, command.retire, command.spill_dir, self.scratch, - self.harnesses, InstallReport(self.root)) + return upgrade(self, InstallReport.for_run(self)) def harnesses(self) -> InstallResult: with client_wire(self.root) as comms: diff --git a/src/agent_comms/installer/outcome.py b/src/agent_comms/installer/outcome.py new file mode 100644 index 0000000000..a67cf3ade7 --- /dev/null +++ b/src/agent_comms/installer/outcome.py @@ -0,0 +1,86 @@ +"""An install's result; reportable exceptions participate directly.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from traceback import FrameSummary +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from agent_comms.installer.upgrade import CarriedPass, LiveRun, Upgrade + + +@dataclass +class CarriedInstallError(BaseException): + """Completion failed after publication; the root did not roll back.""" + + passed: CarriedPass + error: BaseException + + +class InstallOutcome(ABC): + @abstractmethod + def lines(self) -> tuple[str, ...]: ... + + @abstractmethod + def live_run(self) -> LiveRun: ... + + +@dataclass(frozen=True) +class ObservedInstall(InstallOutcome): + """The live observation while no complete Upgrade owns it.""" + + live: LiveRun + + def live_run(self) -> LiveRun: + return self.live + + +class UnfinishedInstall(ObservedInstall): + def lines(self) -> tuple[str, ...]: + raise RuntimeError("Install exited without producing its result") + + +@dataclass(frozen=True) +class FinishedInstall(InstallOutcome): + upgrade: Upgrade + + def lines(self) -> tuple[str, ...]: + return ("Install finished", *self.upgrade.lines()) + + def live_run(self) -> LiveRun: + return self.upgrade.live_run() + + +class InstallFailure(ABC): + """An error which supplies its terminal report and original exception.""" + + @abstractmethod + def lines(self) -> tuple[str, ...]: ... + + @abstractmethod + def exception(self) -> BaseException: ... + + +@dataclass(frozen=True) +class FailedInstall(ObservedInstall): + failure: InstallFailure + + def lines(self) -> tuple[str, ...]: + return (*self.failure.lines(), *self.live.bridge_lines()) + + +@dataclass(frozen=True) +class CrashedInstall(InstallFailure): + error: BaseException + frame: FrameSummary + + def exception(self) -> BaseException: + return self.error + + def lines(self) -> tuple[str, ...]: + return ( + f"Install failed: {type(self.error).__name__}: {self.error}", + f"{self.frame.filename}:{self.frame.lineno}", + ) diff --git a/src/agent_comms/installer/reporting.py b/src/agent_comms/installer/reporting.py new file mode 100644 index 0000000000..0c2c24e1e4 --- /dev/null +++ b/src/agent_comms/installer/reporting.py @@ -0,0 +1,92 @@ +"""An install's permission to write its root at each schema stage.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, cast + +from agent_comms.bus_publication import AgentOrigin +from agent_comms.comms import Comms, client_wire +from agent_comms.declared_family import DeclaredFamily +from agent_comms.messages import InfoKind, Message +from agent_comms.publisher import InitialWrites + +if TYPE_CHECKING: + from agent_comms.installer.launch import InstallReport + from agent_comms.installer.upgrade import Stack + + +class RootWriter(DeclaredFamily, ABC, affix="Writer"): + """The runtime permitted to write this stage of an install's root.""" + + def prepare(self, report: InstallReport) -> None: + """Existing and carried roots are already initialized.""" + + def post(self, root: Path, report: InstallReport, lines: tuple[str, ...]) -> None: + self.write(root, report, "\n".join(lines)) + + @abstractmethod + def write(self, root: Path, report: InstallReport, body: str) -> None: ... + + @staticmethod + def publish(comms: Comms, report: InstallReport, body: str) -> None: + report.register(comms.registry) + publisher = cast(InitialWrites, comms.bus.publisher) + publisher.publish_ordinary( + Message( + sender=report.sender, + target=report.target, + body=body, + type=InfoKind, + ), + origin=AgentOrigin(), + ) + + +@dataclass(frozen=True) +class InstalledWriter(RootWriter): + stack: Stack + + def write(self, root: Path, report: InstallReport, body: str) -> None: + self.stack.cli(root, "register", "--name", report.sender, "--worktree", str(root)) + self.stack.cli(root, "send", "--from", report.sender, "--to", report.target, "--body", body) + + +@dataclass(frozen=True) +class SourceWriter: + """Shared source-side I/O; only the nominal RootWriter cases grant permission.""" + + root: Path + + def write(self, root: Path, report: InstallReport, body: str) -> None: + if root != self.root: + raise ValueError("Install writer belongs to a different root") + with client_wire(self.root) as comms: + RootWriter.publish(comms, report, body) + + +class CarriedWriter(SourceWriter, RootWriter): + """Constructed only by the pass which read the carried root back.""" + + +class FreshWriter(SourceWriter, RootWriter): + """The creation policy selected this empty root before any source-side I/O.""" + + def prepare(self, report: InstallReport) -> None: + comms = Comms(self.root) + root_id = comms.messaging.initialize_private_initial_protocol() + comms.channels.create_tag(report.channel) + report.writing(PreparedFreshWriter(comms, root_id)) + + +@dataclass(frozen=True) +class PreparedFreshWriter(RootWriter): + comms: Comms + root_id: str + + def write(self, root: Path, report: InstallReport, body: str) -> None: + if root != self.comms.root: + raise ValueError("Fresh install writer belongs to a different root") + self.publish(self.comms, report, body) diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index e3c7bbc267..c8e8197f4e 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -29,30 +29,23 @@ import tempfile import time from abc import ABC, abstractmethod -from collections.abc import Callable, Mapping +from collections.abc import Iterator, Mapping from concurrent.futures import ThreadPoolExecutor -from contextlib import ExitStack +from contextlib import ExitStack, contextmanager from dataclasses import dataclass, field, fields, replace from pathlib import Path -from typing import TYPE_CHECKING, Annotated +from typing import TYPE_CHECKING, Annotated, cast from agent_comms.active_route import ActiveRoute, active_route_path, route_new_bus +from agent_comms.cli_bridges.shared import BridgePause, BridgeResume, Serve from agent_comms.cli_commands import InstallResult, ThreadsResult, WhoResult from agent_comms.cli_result import CliResult -from agent_comms.cli_bridges.shared import Serve, SessionBridgeResumes from agent_comms.declared_family import DeclaredFamily -from agent_comms.comms import Comms from agent_comms.external_transcript import ExternalTranscript from agent_comms.field_codec import FieldCodec, PathText from agent_comms.history_result import EverythingHistoryPageResult -from agent_comms.storage_rules import StorageRule from agent_comms.host_services import HostService -from agent_comms.launch_runtime import LaunchRuntime, RestartEnvironment, SpecifiedImportPath -from agent_comms.maintenance_barrier import HOLDER_NONCE, MaintenanceBarrier, MaintenanceOperator -from agent_comms.registry_store import RegistryStore -from agent_comms.repo import Checkout, ScratchCheckout -from agent_comms.store_files import _store_lock -from agent_comms.wire_log import WireLog +from agent_comms.install_choices import InstallerChoice from agent_comms.installer import publication, step_worker from agent_comms.installer.backup import ( BackupReport, @@ -65,13 +58,26 @@ ) from agent_comms.installer.carry import STEPS, Carried, Carry, CarryChain from agent_comms.installer.practice import PracticeMeasurement, practice_copy, prepare_runtime_home +from agent_comms.installer.outcome import CarriedInstallError +from agent_comms.installer.reporting import ( + CarriedWriter, + FreshWriter, + InstalledWriter, + PreparedFreshWriter, + RootWriter, +) from agent_comms.installer.runtime_resolution import RuntimeResolution -from agent_comms.install_choices import InstallerChoice +from agent_comms.launch_runtime import LaunchRuntime, RestartEnvironment, SpecifiedImportPath +from agent_comms.maintenance_barrier import HOLDER_NONCE, MaintenanceBarrier, MaintenanceOperator +from agent_comms.registry_store import RegistryStore +from agent_comms.repo import Checkout, ScratchCheckout +from agent_comms.storage_rules import StorageRule +from agent_comms.store_files import _store_lock +from agent_comms.wire_log import WireLog if TYPE_CHECKING: - from agent_comms.installer.launch import InstallReport + from agent_comms.installer.launch import InstallReport, InstallRun -HOME = Path.home() CLIENT_IMPORTS = "import toad.app, agent_comms.acp" @@ -178,7 +184,7 @@ def __init__(self, error: BaseException, started: tuple[OwnerStart, ...], backup def lines(self) -> tuple[str, ...]: return ( - f"Install rolled back: {type(self.error).__name__}: {self.error}", + f"Install failed: rolled back after {type(self.error).__name__}: {self.error}", f"Original runtime selected again: {self.runtime.prefix}", *(line for owner in self.started for line in owner.lines()), f"Backup: {self.backup}", @@ -209,6 +215,18 @@ class CarriedPass(Extendable): steps: tuple[StepRun, ...] files: RehearsalFiles read: BusReadBack + root: Annotated[Path, PathText] = field(kw_only=True) + + def writer(self) -> CarriedWriter: + return CarriedWriter(self.root) + + @contextmanager + def finishing(self) -> Iterator[None]: + """Keep the carried permission if completion fails without rolling back.""" + try: + yield + except BaseException as error: + raise CarriedInstallError(self, error) from error @dataclass(frozen=True) @@ -251,11 +269,27 @@ class RetiringRehearsal(FreshRehearsal): """The bus at the root moves aside before a fresh one is created there.""" +@dataclass(frozen=True, kw_only=True) class LiveRun(DeclaredFamily, affix="Run"): - """What a live install changed.""" + """What a live install changed, including every captured session bridge.""" + + bridges: tuple[BridgeResume, ...] = () + + def lines(self) -> tuple[str, ...]: + return (*self.details(), *self.bridge_lines()) + + def bridge_lines(self) -> tuple[str, ...]: + return tuple(bridge.line() for bridge in self.bridges) @abstractmethod - def lines(self) -> tuple[str, ...]: ... + def details(self) -> tuple[str, ...]: ... + + +class IncompleteRun(LiveRun): + """No live result was returned; captured bridge outcomes remain reportable.""" + + def details(self) -> tuple[str, ...]: + return () @dataclass(frozen=True) @@ -265,7 +299,7 @@ class CarriedRun(CarriedPass, StoppedOwners, RestartedServices, LiveRun): backup: BackupReport retired_backups: tuple[Annotated[Path, PathText], ...] - def lines(self) -> tuple[str, ...]: + def details(self) -> tuple[str, ...]: return ( *(f"Carried {step.carry}: {step.carried.changed} changed" for step in self.steps), *self.stopped_lines(), @@ -279,7 +313,7 @@ def lines(self) -> tuple[str, ...]: class UnchangedRun(LiveRun): """Published the installed runtime again and planned the next install.""" - def lines(self) -> tuple[str, ...]: + def details(self) -> tuple[str, ...]: return ("Runtime republished; nothing carried",) @@ -287,7 +321,7 @@ def lines(self) -> tuple[str, ...]: class CreatedRun(RestartedServices, LiveRun, Extendable): route: str - def lines(self) -> tuple[str, ...]: + def details(self) -> tuple[str, ...]: return (f"Created bus; route: {self.route}", *self.service_lines()) @@ -295,8 +329,8 @@ def lines(self) -> tuple[str, ...]: class RetiredRun(CreatedRun, StoppedOwners): retired: Annotated[Path, PathText] - def lines(self) -> tuple[str, ...]: - return (*super().lines(), *self.stopped_lines(), f"Retired bus: {self.retired}") + def details(self) -> tuple[str, ...]: + return (*super().details(), *self.stopped_lines(), f"Retired bus: {self.retired}") @dataclass(frozen=True) @@ -308,6 +342,9 @@ class Upgrade(DeclaredFamily, affix="Upgrade"): root: Annotated[Path, PathText] pending: tuple[str, ...] + def live_run(self) -> LiveRun: + return IncompleteRun() + def lines(self) -> tuple[str, ...]: return (f"{self.source[:12]} -> {self.target[:12]}", f"Pending carries: {', '.join(self.pending) or 'none'}") @@ -323,11 +360,12 @@ class InstalledUpgrade(RehearsedUpgrade): live: LiveRun install: InstallResult """The harnesses registered on the bus the target now reads.""" - unresumed: tuple[str, ...] - """Session bridges paused for the swap that did not come back.""" + + def live_run(self) -> LiveRun: + return self.live def lines(self) -> tuple[str, ...]: - return (*super().lines(), *self.live.lines(), *self.unresumed) + return (*super().lines(), *self.live.lines()) @dataclass(frozen=True) @@ -335,7 +373,7 @@ class Stack: """One built installation and the source checkout containing its install tools.""" checkout: Path - home: Path = HOME + home: Path = field(default_factory=Path.home) runtime: LaunchRuntime = field(default_factory=LaunchRuntime.installed) @property @@ -356,7 +394,7 @@ def require_uv(self) -> None: raise UpgradeRefused(f"No uv executable at {self.uv}; the installer runs only that uv") @classmethod - def build(cls, checkout: Path, home: Path = HOME) -> Stack: + def build(cls, checkout: Path, home: Path) -> Stack: """Resolve current mains once, then build or reuse their immutable installation.""" Checkout(checkout).prepare() source = cls(checkout, home, runtime=LaunchRuntime.running()) @@ -516,6 +554,17 @@ class TargetTree: scratch: Path revision: str + @classmethod + def from_run(cls, request: InstallRun) -> TargetTree: + return cls(request.command.checkout, request.scratch, request.target) + + def __enter__(self) -> TargetTree: + return self + + def __exit__(self, *_exception: object) -> None: + self.close() + shutil.rmtree(self.scratch) + @property def path(self) -> Path: return self.scratch / f"tree-{self.revision[:12]}" @@ -598,17 +647,16 @@ def carry_and_read(root: Path, before: dict[str, SavedFile], pending: tuple[type step's process, so a step that writes the transcript store could not run under it.""" steps = (*stopping, *carry_all(root, pending, staged, tree)) files = carried(root, before, steps, RehearsalFiles.declared(steps)) - return CarriedPass(steps, files, reader.verify(root)) + return CarriedPass(steps, files, reader.verify(root), root=root) -def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice, rehearse_only: bool, - retire: bool, spill_dir: Path, scratch: Path, - harnesses: Callable[[], InstallResult], report: InstallReport) -> Upgrade: - """``scratch`` holds the target's worktree the launcher runs this source from; this - run owns it and removes it. A live run ends by registering the ``harnesses``.""" - current = run("git", "-C", checkout, "rev-parse", "HEAD").strip() - tree = TargetTree(checkout, scratch, target) - try: +def upgrade(request: InstallRun, report: InstallReport) -> Upgrade: + """The decoded request owns the input; the report owns the whole run's lifetime.""" + with report.running(request) as member: + command = request.command + checkout, root, target = command.checkout, request.root, request.target + tree = TargetTree.from_run(request) + current = run("git", "-C", checkout, "rev-parse", "HEAD").strip() if subprocess.run(["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target]).returncode: raise UpgradeRefused(f"{target[:12]} does not follow the checkout's {current[:12]}") @@ -618,33 +666,22 @@ def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice, chain = CarryChain.at(tree.path) except ValueError as error: raise UpgradeRefused(str(error)) from None - member = Install.member(root, retire) - installed = Stack(checkout) + installed = report.outgoing installed.require_uv() if not installed.python.exists(): raise UpgradeRefused(f"No installed stack at {installed.python}") - staged = Stack.build(tree.path) + staged = Stack.build(tree.path, home=installed.home) pending = member.pending_steps(root, chain, staged, tree) install = member.of(checkout, root, current, target, pending, installed, staged, - tree, installer, spill_dir) + tree, command.installer, command.spill_dir, report) rehearsal = install.rehearsal() names = tuple(step.name() for step in pending) - if rehearse_only: - return RehearsedUpgrade(current, target, root, names, rehearsal) - install.announce(report) - bridges = SessionBridgeResumes() - live_run = install.install(rehearsal, bridges) + if command.rehearse: + return report.complete(RehearsedUpgrade(current, target, root, names, rehearsal)) + live_run = install.install(rehearsal) result = InstalledUpgrade(current, target, root, names, rehearsal, live_run, - harnesses(), bridges.lines()) - report.finished(result) - bridges.require_resumed() - return result - except UpgradeRefused as error: - report.failed(error) - raise - finally: - tree.close() - shutil.rmtree(scratch, ignore_errors=True) + request.harnesses()) + return report.complete(result) @dataclass(frozen=True) @@ -662,10 +699,11 @@ class Install[Rehearsed: Rehearsal](ABC): tree: TargetTree installer: InstallerChoice spill_dir: Path + report: InstallReport - def announce(self, report: InstallReport) -> None: - report.starting(Upgrade(self.current, self.target, self.root, - tuple(step.name() for step in self.pending))) + @classmethod + def initial_writer(cls, root: Path, installed: Stack) -> RootWriter: + return InstalledWriter(installed) @staticmethod def member(root: Path, retire: bool) -> type[Install]: @@ -685,18 +723,18 @@ def pending_steps(cls, root: Path, chain: CarryChain, staged: Stack, @classmethod def of(cls, checkout: Path, root: Path, current: str, target: str, pending: tuple[type[Carry], ...], installed: Stack, staged: Stack, tree: TargetTree, - installer: InstallerChoice, spill_dir: Path) -> Install: + installer: InstallerChoice, spill_dir: Path, report: InstallReport) -> Install: """A bus already carried, on the target, built from the installed lock needs nothing.""" unchanged = not pending and current == target and RuntimeResolution.installed().matches( staged.lock, staged.python, target) member = (ChangedInstall, UnchangedInstall)[unchanged] return member(checkout, root, current, target, pending, installed, staged, - tree, installer, spill_dir) + tree, installer, spill_dir, report) @abstractmethod def rehearsal(self) -> Rehearsed: ... - def install(self, rehearsal: Rehearsed, bridges: SessionBridgeResumes) -> LiveRun: + def install(self, rehearsal: Rehearsed) -> LiveRun: """Session bridges close their bus resources before units stop, then exec in place. Each serve retains its recorded owner's PID and start time after the hook exits. @@ -704,10 +742,17 @@ def install(self, rehearsal: Rehearsed, bridges: SessionBridgeResumes) -> LiveRu installed pointer, not a copied target version, selects its successor. """ with ExitStack() as resources: - for bridge in Serve.running(self.installed.home): - bridges.pause(bridge, resources) - resources.callback(bridges.resume) - return self.live(rehearsal) + paused: list[BridgePause] = [] + result: LiveRun = IncompleteRun() + try: + for bridge in Serve.running(self.installed.home): + paused.append(bridge.pause(resources)) + result = self.live(rehearsal) + self.report.dispatch_sync(result) + finally: + result = self.report.record_live(replace( + result, bridges=tuple(bridge.resume() for bridge in paused))) + return result @abstractmethod def live(self, rehearsal: Rehearsed) -> LiveRun: ... @@ -736,9 +781,8 @@ def live(self, rehearsal: UnchangedRehearsal) -> UnchangedRun: return UnchangedRun() -class FreshInstall(Install[FreshRehearsal]): - """Nothing was ever written at the root: no carry, rehearsal, hold or owner to restart. - The target's source creates the bus already in its format and routes to it.""" +class CreatingInstall(Install[FreshRehearsal]): + """Shared creation after the selected policy has made the root writable as fresh.""" @classmethod def pending_steps(cls, root: Path, chain: CarryChain, staged: Stack, @@ -748,29 +792,41 @@ def pending_steps(cls, root: Path, chain: CarryChain, staged: Stack, @classmethod def of(cls, checkout: Path, root: Path, current: str, target: str, pending: tuple[type[Carry], ...], installed: Stack, staged: Stack, tree: TargetTree, - installer: InstallerChoice, spill_dir: Path) -> Install: + installer: InstallerChoice, spill_dir: Path, report: InstallReport) -> Install: return cls(checkout, root, current, target, pending, installed, staged, - tree, installer, spill_dir) + tree, installer, spill_dir, report) - def announce(self, report: InstallReport) -> None: - """A fresh bus has nowhere to publish a start message yet.""" + def creation_writer(self) -> PreparedFreshWriter: + self.report.writing(FreshWriter(self.root)) + return cast(PreparedFreshWriter, self.report.writer) def rehearsal(self) -> FreshRehearsal: return FreshRehearsal() def live(self, rehearsal: FreshRehearsal) -> CreatedRun: + writer = self.creation_writer() run("git", "-C", self.checkout, "merge", "--ff-only", "-q", self.target) Checkout(self.checkout).prepare() self.staged.prepare_native() self.root.mkdir(mode=0o700, parents=True, exist_ok=True) - root_id = Comms(self.root).messaging.initialize_private_initial_protocol() self.staged.publish() - route = route_new_bus(ActiveRoute(self.root.absolute(), root_id), active_route_path()) + route = route_new_bus(ActiveRoute(self.root.absolute(), writer.root_id), active_route_path()) services = HostService.restart_running(self.staged.python) return CreatedRun(services, route) -class RetiringInstall(FreshInstall): +class FreshInstall(CreatingInstall): + """The root has no old schema: its source owns initialization and both report posts.""" + + @classmethod + def initial_writer(cls, root: Path, installed: Stack) -> RootWriter: + return FreshWriter(root) + + def creation_writer(self) -> PreparedFreshWriter: + return cast(PreparedFreshWriter, self.report.writer) + + +class RetiringInstall(CreatingInstall): """The bus at the root is runtime state the target cannot carry: its owners stop, it moves aside intact, and a fresh bus takes its place.""" @@ -857,8 +913,10 @@ def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[t runtime=installed.runtime) from error # Installed: the carries ran and the runtime is published. Starting the owners it # stopped is reported per owner and never rolls the install back. - started = staged.start(root, running) - retired_backups = backup.retain(spill_dir) - return passed.extended(CarriedRun, stopped=tuple(running), services=services, started=started, - restore_space=restore, backup=backup.report(), - retired_backups=retired_backups) + with passed.finishing(): + started = staged.start(root, running) + retired_backups = backup.retain(spill_dir) + return passed.extended( + CarriedRun, stopped=tuple(running), services=services, started=started, + restore_space=restore, backup=backup.report(), retired_backups=retired_backups, + ) diff --git a/src/agent_comms/mro_dispatch.py b/src/agent_comms/mro_dispatch.py index 8ab004be9b..b5401d174e 100644 --- a/src/agent_comms/mro_dispatch.py +++ b/src/agent_comms/mro_dispatch.py @@ -53,6 +53,12 @@ def handler_names_for(cls, value_type: type) -> Iterator[str]: if capability in classes: yield name + @classmethod + def class_handlers_for(cls, value_type: type) -> Iterator[Callable[..., object]]: + """Bind factory handlers to their declaring class, not an arbitrary instance.""" + for name in cls.handler_names_for(value_type): + yield getattr(cls, name) + def handlers_for(self, value: Any): for name in self.handler_names_for(type(value)): yield getattr(self, name) diff --git a/tests/conftest.py b/tests/conftest.py index 583c1416b1..c727815887 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,6 +3,7 @@ import asyncio import os +import pwd import subprocess import sys import tempfile @@ -71,13 +72,15 @@ def source_programs(tmp_path): return SourcePrograms.create(tmp_path) -@pytest.fixture -def bridge_home(tmp_path, monkeypatch): - """Short per-worker socket addresses; all state stays in persistent test scratch.""" - destination = tmp_path / 'home' - destination.mkdir() - monkeypatch.chdir(Path(__file__).resolve().parents[1]) - home = Path(f'.bridge-home-{uuid.uuid4().hex[:8]}') +@contextmanager +def short_home(destination: Path) -> Iterator[Path]: + """A short absolute socket address, with all data in the caller's owned scratch. + + The OS account home locates only the disposable alias. HOME may already name + another private runtime, and relative aliases break when install children chdir. + """ + destination.mkdir(parents=True, exist_ok=True) + home = Path(pwd.getpwuid(os.getuid()).pw_dir) / f'.bh-{uuid.uuid4().hex[:8]}' home.symlink_to(destination, target_is_directory=True) try: yield home @@ -85,6 +88,14 @@ def bridge_home(tmp_path, monkeypatch): home.unlink() +@pytest.fixture +def bridge_home(tmp_path, monkeypatch): + """Short per-worker socket addresses; all state stays in persistent test scratch.""" + monkeypatch.chdir(Path(__file__).resolve().parents[1]) + with short_home(tmp_path / 'home') as home: + yield home + + _MANAGED_RUNTIME_ENVIRONMENT = frozenset({ "AGENT_COMMS_MANAGED", "AGENT_COMMS_PROJECT_REQUEST", diff --git a/tests/source_programs.py b/tests/source_programs.py index aeac66f78f..34399751d2 100644 --- a/tests/source_programs.py +++ b/tests/source_programs.py @@ -10,10 +10,12 @@ from pathlib import Path from typing import Self +from agent_comms.launch_runtime import LaunchRuntime + @dataclass(frozen=True) -class SourcePrograms: - prefix: Path +class SourcePrograms(LaunchRuntime): + directory: Path entries: tuple[EntryPoint, ...] packages: tuple[Path, ...] @@ -25,7 +27,37 @@ def bind(self, directory: Path) -> None: def script(self, function: Callable[..., object]) -> Path: target = f'{function.__module__}:{function.__name__}' (entry,) = (entry for entry in self.entries if entry.value == target) - return self.prefix / 'bin' / entry.name + return self.directory / entry.name + + @classmethod + def declared(cls, interpreter: Path, directory: Path) -> Self: + project = Path(__file__).resolve().parents[1] + manifest = tomllib.loads((project / 'pyproject.toml').read_text()) + packages = tuple(project / path for path in + manifest['tool']['hatch']['build']['targets']['wheel']['packages']) + entries = tuple(EntryPoint(name=name, value=value, group='console_scripts') + for name, value in manifest['project']['scripts'].items()) + return cls(interpreter, directory, entries, packages) + + def written(self) -> Self: + self.directory.mkdir(parents=True, exist_ok=True) + sources = tuple(dict.fromkeys(str(package.parent) for package in self.packages)) + for entry in self.entries: + program = self.directory / entry.name + program.write_text( + f'#!{self.interpreter}\nimport sys\n' + f'sys.path[:0] = {sources!r}\n' + 'from importlib.metadata import EntryPoint\n' + f'sys.exit(EntryPoint(name={entry.name!r}, value={entry.value!r}, ' + f'group={entry.group!r}).load()())\n' + ) + program.chmod(0o700) + return self + + @classmethod + def current(cls, directory: Path) -> Self: + """Create only source launchers; keep the existing interpreter and dependencies.""" + return cls.declared(Path(sys.executable), directory).written() @classmethod def create(cls, directory: Path) -> Self: @@ -41,21 +73,4 @@ def create(cls, directory: Path) -> Self: site = prefix / Path(shared).relative_to(sys.prefix) site.mkdir(parents=True) (site / 'shared-python.pth').write_text(f'import site; site.addsitedir({shared!r})\n') - project = Path(__file__).resolve().parents[1] - manifest = tomllib.loads((project / 'pyproject.toml').read_text()) - packages = tuple(project / path for path in manifest['tool']['hatch']['build']['targets']['wheel']['packages']) - sources = tuple(dict.fromkeys(str(package.parent) for package in packages)) - scripts = manifest['project']['scripts'] - entries = tuple(EntryPoint(name=name, value=value, group='console_scripts') - for name, value in scripts.items()) - for entry in entries: - program = binary / entry.name - program.write_text( - f'#!{interpreter}\nimport sys\n' - f'sys.path[:0] = {sources!r}\n' - 'from importlib.metadata import EntryPoint\n' - f'sys.exit(EntryPoint(name={entry.name!r}, value={entry.value!r}, ' - f'group={entry.group!r}).load()())\n' - ) - program.chmod(0o700) - return cls(prefix, entries, packages) + return cls.declared(interpreter, binary).written() diff --git a/tests/test_bound_listener.py b/tests/test_bound_listener.py new file mode 100644 index 0000000000..ac410fbf86 --- /dev/null +++ b/tests/test_bound_listener.py @@ -0,0 +1,41 @@ +"""The connector's readiness witness comes from its real listening socket.""" +import asyncio +import socket + +import pytest +from aiohttp import web + +from agent_comms.host_services import BoundListener, ServiceNotServingError, SystemdUserService +from connector_readiness import ConnectorReadiness + + +def test_readiness_requires_the_started_sites_listening_socket(monkeypatch): + async def check(): + runner = web.AppRunner(web.Application()) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + try: + with pytest.raises(TypeError): + BoundListener(site) + with pytest.raises(ServiceNotServingError): + BoundListener.of(site) + with pytest.raises(TypeError): + SystemdUserService.ready() + await site.start() + listener = BoundListener.of(site) + assert listener.site is site + notification = ConnectorReadiness() + try: + monkeypatch.setenv(SystemdUserService.notify_variable, "@" + notification.name) + SystemdUserService.ready(listener) + assert notification.socket.recv(4096) == b"READY=1" + with socket.socket() as connected: + connected.connect(("127.0.0.1", listener.site.port)) + finally: + notification.close() + finally: + await runner.cleanup() + with pytest.raises(ServiceNotServingError): + BoundListener.of(site) + + asyncio.run(check()) diff --git a/tests/test_install_announcements.py b/tests/test_install_announcements.py index d656f34f84..6f65e8f895 100644 --- a/tests/test_install_announcements.py +++ b/tests/test_install_announcements.py @@ -1,96 +1,143 @@ -"""Typed installer results use the real publisher and retained channel history.""" +"""Run-scoped installer reporting and typed results on a real private bus.""" +from dataclasses import replace from pathlib import Path +import os import subprocess -import sys import pytest -from agent_comms.cli_commands import HistoryCliCommand, InstallResult +from agent_comms.child_process import ProcessIdentity +from agent_comms.cli_bridges.shared import ( + OwnerExitedBridge, ReplacedBridge, ResumedBridge, UnobservedBridge, +) +from agent_comms.cli_commands import HistoryCliCommand, InstallCliCommand, InstallResult from agent_comms.commander.config import NoConfigFile -from agent_comms.commander_heartbeat import CommanderHeartbeat -from agent_comms.errors import UnknownChannelError +from agent_comms.comms import Comms from agent_comms.field_codec import FieldCodec from agent_comms.installer.backup import BackupReport, RestoreSpace, UpgradeRefused from agent_comms.installer.carry import Carried -from agent_comms.installer.launch import InstallReport +from agent_comms.installer.launch import InstallReport, InstallRun +from agent_comms.installer.outcome import CrashedInstall, FailedInstall +from agent_comms.installer.reporting import InstalledWriter, PreparedFreshWriter from agent_comms.installer.upgrade import ( - BusReadBack, CarriedRun, CreatedRun, FreshRehearsal, InstalledUpgrade, - InstallRolledBack, RefusedOwner, RehearsalFiles, RetiredRun, StartedOwner, - StepRun, UnchangedRun, Upgrade, + BusReadBack, CarriedRun, CreatedRun, FreshRehearsal, IncompleteRun, InstalledUpgrade, + RefusedOwner, RehearsalFiles, RetiredRun, StartedOwner, + StepRun, UnchangedRun, Upgrade, upgrade, ) from agent_comms.launch_runtime import LaunchRuntime from agent_comms.thread_execution import ExternalThreadExecution from conftest import declare_test_bus -@pytest.mark.parametrize("live", ( - CarriedRun( +@pytest.mark.parametrize("live_factory", ( + pytest.param(lambda root: CarriedRun( steps=(StepRun("carry_order_scopes", Carried(6)),), files=RehearsalFiles(3, 1, ("todos.sqlite3",), 1024), - read=BusReadBack(2, 2, 1), stopped=("worker",), services=("connector",), + read=BusReadBack(2, 2, 1), root=root, stopped=("worker",), services=("connector",), started=(StartedOwner("worker"), RefusedOwner("busy", "already running")), restore_space=RestoreSpace(1024, 4096, 0.01), backup=BackupReport("saved-before-install", 1024), retired_backups=(), - ), - UnchangedRun(), - CreatedRun(("connector",), "created route"), - RetiredRun(services=("connector",), route="created route", stopped=("worker",), - retired=Path("retired-bus")), -), ids=("carried", "unchanged", "created", "retired")) -def test_typed_results_publish_as_installer_and_survive_history_read(private_comms, live): - comms = declare_test_bus(private_comms, channels=("comms",)) - report = InstallReport(comms.root) - found = Upgrade("a" * 40, "b" * 40, comms.root, ("carry_order_scopes",)) + ), id="carried"), + pytest.param(lambda root: UnchangedRun(), id="unchanged"), + pytest.param(lambda root: IncompleteRun(), id="incomplete"), + pytest.param(lambda root: CreatedRun(("connector",), "created route"), id="created"), + pytest.param(lambda root: RetiredRun( + services=("connector",), route="created route", stopped=("worker",), + retired=Path("retired-bus")), id="retired"), +)) +def test_live_result_family_survives_codec_and_retained_history(report_bus, live_factory): + comms = report_bus + outgoing = LaunchRuntime(Path("/outgoing/bin/python")) + target = LaunchRuntime(Path("/target/bin/python")) + owner = ProcessIdentity.capture(os.getpid()) + successor = ProcessIdentity.capture(os.getppid()) + live = replace(live_factory(comms.root), bridges=( + ResumedBridge("resumed", outgoing, target), OwnerExitedBridge("exited"), + ReplacedBridge("replaced", owner, successor), UnobservedBridge("unobserved"), + )) installed = InstalledUpgrade( - found.source, found.target, found.root, found.pending, FreshRehearsal(), - live, InstallResult(NoConfigFile(), {}), (), + "a" * 40, "b" * 40, comms.root, ("carry_order_scopes",), FreshRehearsal(), + live, InstallResult(NoConfigFile(), {}), ) - # The wire codec still carries the same result family and fields after factoring. decoded = FieldCodec.decode(Upgrade, FieldCodec.encode(installed)) assert decoded == installed - started = report.starting(found) - finished = report.finished(decoded) - messages = tuple(reversed(HistoryCliCommand(target="#comms").apply(comms).messages)) - assert len(messages) == 2 - assert tuple(message.sender for message in messages) == ("installer", "installer") - assert tuple(message.reference for message in messages) == (started.reference, finished.reference) - assert all(found.target[:12] in message.body for message in messages) - assert all(line in messages[1].body for line in live.lines()) - sender = comms.registry.require("installer") - assert sender.execution is ExternalThreadExecution + InstallReport.register(comms.registry) + comms.messaging.send(InstallReport.sender, f"#{InstallReport.channel}", + "\n".join(decoded.lines())) + messages = HistoryCliCommand(target=f"#{InstallReport.channel}").apply(comms).messages + assert len(messages) == 1 + assert all(line in messages[0].body for line in live.lines()) + assert comms.registry.require(InstallReport.sender).execution is ExternalThreadExecution + + +@pytest.fixture +def report_bus(comms, durable_home): + """Reporting starts no native owner: use the ordinary bus, not a native-SDK fixture.""" + comms.messaging.initialize_private_initial_protocol() + return declare_test_bus(comms, channels=(InstallReport.channel,)) -def test_rollback_and_refusal_are_published_from_their_error_types(private_comms): - comms = declare_test_bus(private_comms, channels=("comms",)) - error = UpgradeRefused("carry_order_scopes refused its rows") - runtime = LaunchRuntime(Path("/saved-original-runtime/bin/python")) - restored = InstallRolledBack(error, (StartedOwner("worker"),), "saved-original-bus", - runtime=runtime) - report = InstallReport(comms.root) - report.failed(error) - report.failed(restored) - messages = tuple(reversed(HistoryCliCommand(target="#comms").apply(comms).messages)) +@pytest.fixture +def refused_request(report_bus, tmp_path): + """An actual ancestry refusal, before stack building or dependency work.""" + checkout = tmp_path / "checkout" + checkout.mkdir() + git = ["git", "-C", str(checkout)] + subprocess.run([*git, "init", "-q"], check=True) + subprocess.run([*git, "-c", "user.name=Installer test", "-c", "user.email=test@example.invalid", + "commit", "-q", "--allow-empty", "-m", "old"], check=True) + target = subprocess.check_output([*git, "rev-parse", "HEAD"], text=True).strip() + subprocess.run([*git, "-c", "user.name=Installer test", "-c", "user.email=test@example.invalid", + "commit", "-q", "--allow-empty", "-m", "new"], check=True) + comms = report_bus + request = InstallRun(InstallCliCommand(checkout=checkout, project=checkout), + comms.root, tmp_path / "scratch", target) + return request, comms + + +@pytest.mark.parametrize("scratch_present,error_kind,outcome_kind", ( + (True, UpgradeRefused, UpgradeRefused), + (False, FileNotFoundError, CrashedInstall), +), ids=("early-refusal", "cleanup-crash")) +def test_install_failure_family_through_upgrade(refused_request, scratch_present, + error_kind, outcome_kind, durable_home): + request, comms = refused_request + if scratch_present: + request.scratch.mkdir() + report = InstallReport.for_run(request) + with pytest.raises(error_kind) as raised: + upgrade(request, report) + messages = tuple(reversed(HistoryCliCommand( + target=f"#{report.channel}").apply(comms).messages)) assert len(messages) == 2 - assert all(message.sender == "installer" for message in messages) - assert all(line in messages[1].body for line in restored.lines()) - assert str(error) in messages[1].body - assert str(runtime.prefix) in messages[1].body - completed = subprocess.run( - [sys.executable, "-m", "agent_comms.cli", "--root", str(comms.root), - "history", "--target", "#comms"], - capture_output=True, text=True, timeout=30, - ) - assert completed.returncode == 0, completed.stderr - assert FieldCodec.parse(completed.stdout) == HistoryCliCommand(target="#comms").apply(comms).to_wire() + assert messages[0].body.splitlines()[0] == "Install starting" + assert messages[1].body.splitlines()[0].startswith("Install failed") + assert all(message.sender == report.sender for message in messages) + assert isinstance(report.outcome, FailedInstall) + assert isinstance(report.outcome.failure, outcome_kind) + assert report.outcome.failure.exception() is raised.value + assert isinstance(report.writer, InstalledWriter) + assert report.outgoing.home == durable_home + assert not request.scratch.exists() + assert all(line in messages[1].body for line in report.outcome.lines()) + assert report.writer.stack.runtime == LaunchRuntime.running() -def test_publication_failure_raises_after_registering_the_sender(private_comms): - # No #comms channel exists: the real publisher refuses it after registering its sender. - CommanderHeartbeat.register(private_comms.registry) - report = InstallReport(private_comms.root) - with pytest.raises(UnknownChannelError): - report.failed(UpgradeRefused("stopped before carrying")) - original = private_comms.registry.require("installer") - report.register(private_comms.registry) - assert private_comms.registry.require("installer").incarnation == original.incarnation - assert private_comms.registry.require(CommanderHeartbeat.sender).execution is ExternalThreadExecution +@pytest.mark.parametrize("fresh,writer_kind", ( + (False, InstalledWriter), (True, PreparedFreshWriter), +), ids=("outgoing-schema", "fresh-source")) +def test_root_writer_family_through_upgrade(refused_request, fresh, writer_kind): + request, _existing = refused_request + root = request.root.with_name("fresh") if fresh else request.root + request = replace(request, root=root) + request.scratch.mkdir() + report = InstallReport.for_run(request) + with pytest.raises(UpgradeRefused): + upgrade(request, report) + assert isinstance(report.writer, writer_kind) + messages = tuple(reversed(HistoryCliCommand( + target=report.target).apply(Comms(root)).messages)) + assert len(messages) == 2 + assert messages[0].body.splitlines()[0] == "Install starting" + assert messages[1].body.splitlines()[0].startswith("Install failed") + assert not request.scratch.exists() diff --git a/tests/test_install_paths.py b/tests/test_install_paths.py new file mode 100644 index 0000000000..fd049aad97 --- /dev/null +++ b/tests/test_install_paths.py @@ -0,0 +1,42 @@ +"""Install arguments retain the caller's paths across the target child's cwd.""" +from dataclasses import replace +from pathlib import Path + +from agent_comms.cli import build_parser +from agent_comms.cli_commands import InstallCliCommand +from agent_comms.commander.config import GivenRepos +from agent_comms.field_codec import FieldCodec +from agent_comms.installer.launch import InstallRun + + +def test_install_paths_decode_before_the_target_child_changes_directory( + private_directory, monkeypatch, +): + home = private_directory / "home" + home.mkdir() + child = private_directory / "target" + child.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.chdir(private_directory) + arguments = build_parser().parse_args([ + "install", "--checkout", "checkout", "--project", "~/project", + "--spill-dir", "backups", "--repos", "one, ~/two", + ]) + command = InstallCliCommand.from_arguments(arguments) + assert command.checkout == private_directory / "checkout" + assert command.project == home / "project" + assert command.spill_dir == private_directory / "backups" + assert command.repos == GivenRepos((private_directory / "one", home / "two")) + + # This is the payload InstallRun.launch sends to the target's installer. + install = InstallRun(command, private_directory / "bus", child, "a" * 40) + alias = private_directory / "target-alias" + alias.symlink_to(child, target_is_directory=True) + payload = FieldCodec.encode_record(replace(install, scratch=alias)) + monkeypatch.chdir(child) + assert FieldCodec.decode(InstallRun, payload) == install + + defaults = InstallCliCommand.from_arguments(build_parser().parse_args(["install"])) + assert defaults.checkout == defaults.project == child + assert defaults.spill_dir == Path.home() / ".cache/agent-rehearsal" + assert defaults.spill_dir.is_absolute() diff --git a/tests/test_install_rehearsal.py b/tests/test_install_rehearsal.py index f38ceb9616..6193b9825a 100644 --- a/tests/test_install_rehearsal.py +++ b/tests/test_install_rehearsal.py @@ -14,15 +14,37 @@ import subprocess import sys import tarfile +from collections.abc import Iterator +from contextlib import AbstractContextManager, nullcontext +from dataclasses import dataclass from io import BytesIO from pathlib import Path +from traceback import FrameSummary, extract_stack +from typing import ClassVar, NoReturn import pytest +import agent_comms +from agent_comms.cli_bridges.monitor import InspectSession +from agent_comms.cli_bridges.shared import ResumedBridge +from agent_comms.cli_commands import HistoryCliCommand, InstallCliCommand +from agent_comms.comms import Comms from agent_comms.field_codec import FieldCodec -from agent_comms.installer.backup import bus_files -from agent_comms.installer.upgrade import CarriedRehearsal, RehearsedUpgrade, Upgrade -from conftest import declare_test_bus, publish_initial +from agent_comms.harness_install import replace_link +from agent_comms.installer.launch import InstallReport, InstallRun +from agent_comms.installer.outcome import CrashedInstall, FailedInstall +from agent_comms.installer.reporting import CarriedWriter +from agent_comms.installer.runtime_resolution import RuntimeResolution +from agent_comms.installer.upgrade import ( + CarriedRehearsal, + RehearsedUpgrade, + Stack, + Upgrade, + upgrade, +) +from agent_comms.launch_runtime import LaunchRuntime +from conftest import declare_test_bus, publish_initial, short_home +from test_install_session_bridge import session_bridge REPOSITORY = Path(__file__).resolve().parents[1] @@ -32,8 +54,14 @@ def git(*arguments: str) -> str: capture_output=True, text=True).stdout.strip() +@dataclass(frozen=True) +class InstalledHome: + environment: dict[str, str] + outgoing: LaunchRuntime + + @pytest.fixture(scope="module") -def installed_home(tmp_path_factory) -> dict[str, str]: +def installed_home(tmp_path_factory: pytest.TempPathFactory) -> InstalledHome: """A home holding what an installed machine has: the installer's uv, and as the installed runtime a build of the merge base this change lands on. Both cases share it, so the second reuses the staged runtime the first built, as a second install @@ -41,41 +69,186 @@ def installed_home(tmp_path_factory) -> dict[str, str]: home = tmp_path_factory.mktemp("installed-home") uv = home / ".local/state/agent-comms/upgrade-bus/uv" uv.parent.mkdir(parents=True) - uv.symlink_to(shutil.which("uv")) + uv_program = shutil.which("uv") + assert uv_program is not None, "The installed-runtime fixture requires uv" + uv.symlink_to(uv_program) caches = subprocess.run([uv, "cache", "dir"], check=True, capture_output=True, text=True) environment = {**os.environ, "HOME": str(home), "UV_CACHE_DIR": caches.stdout.strip(), "npm_config_cache": str(Path.home() / ".npm")} environment.pop("PYTHONPATH", None) base = home / "outgoing-source" - archive = subprocess.run(["git", "-C", REPOSITORY, "archive", git("merge-base", "HEAD", "origin/main")], - check=True, capture_output=True).stdout + archive = subprocess.run( + ["git", "-C", REPOSITORY, "archive", git("merge-base", "HEAD", "origin/main")], + check=True, capture_output=True).stdout with tarfile.open(fileobj=BytesIO(archive)) as tree: tree.extractall(base, filter="data") runtime = home / ".local/share/agent-comms/runtimes/outgoing" - subprocess.run([uv, "venv", "-q", "--python", sys.executable, runtime], check=True, env=environment) + subprocess.run([uv, "venv", "-q", "--python", sys.executable, runtime], + check=True, env=environment) subprocess.run([uv, "pip", "install", "-q", "--python", runtime / "bin/python", base], check=True, env=environment) (home / ".local/share/agent-comms/current").symlink_to(runtime, target_is_directory=True) - return environment + return InstalledHome(environment, LaunchRuntime(runtime / "bin/python")) @pytest.mark.timeout(280) @pytest.mark.parametrize("repos", ((), ("--repos",)), ids=("kept-repos", "given-repos")) -def test_install_rehearses_without_changing_the_bus(private_comms, installed_home, tmp_path, repos): +def test_install_rehearses_preserving_existing_data_and_publishing_its_result( + private_comms: Comms, installed_home: InstalledHome, tmp_path: Path, + repos: tuple[str, ...]) -> None: comms = private_comms declare_test_bus(comms, channels=("comms",), threads=( comms.registry.new_thread(name="worker", tags=frozenset(), worktree=str(tmp_path)),)) publish_initial(comms, "worker", "#all", "One message before the install") checkout = tmp_path / "checkout" subprocess.run(["git", "clone", "-q", REPOSITORY, checkout], check=True) - before = bus_files(comms.root) + before = HistoryCliCommand().apply(comms).messages + worker = comms.registry.require("worker") given = (*repos, str(checkout)) if repos else () installed = subprocess.run( [Path(sys.executable).with_name("agent-comms"), "--root", comms.root, "install", "--rehearse", "--checkout", checkout, "--to", "HEAD", "--project", checkout, *given], - cwd=checkout, env={**installed_home, "PYTHONPATH": os.environ.get("PYTHONPATH", "")}, + cwd=checkout, + env={**installed_home.environment, "PYTHONPATH": os.environ.get("PYTHONPATH", "")}, capture_output=True, text=True, timeout=270) assert installed.returncode == 0, installed.stderr result = FieldCodec.decode(Upgrade, FieldCodec.parse(installed.stdout)) assert isinstance(result, RehearsedUpgrade) and isinstance(result.rehearsal, CarriedRehearsal) - assert bus_files(comms.root) == before + after = HistoryCliCommand().apply(comms).messages + assert comms.registry.require("worker") == worker + original = {message.reference: message for message in before} + retained = {message.reference: message for message in after} + assert {reference: retained[reference] for reference in original} == original + assert len(after) == len(before) + 2 + reports = tuple(reversed(HistoryCliCommand( + target=f"#{InstallReport.channel}").apply(comms).messages)) + assert len(reports) == 2 + assert all(message.sender == InstallReport.sender for message in reports) + assert reports[0].body.splitlines()[0] == "Install starting" + assert reports[1].body.splitlines()[0] == "Install finished" + + +class PublicationFault(BaseException): + """The injected failure retains the actual source caller, not a copied line number.""" + + def __init__(self, site: FrameSummary): + super().__init__("The target was published before this failure") + self.site = site + + +class PublicationRuntimeError(PublicationFault, RuntimeError): + pass + + +class PublicationInterrupt(PublicationFault, KeyboardInterrupt): + pass + + +class PublicationFailingRun(InstallRun): + """Inject only the requested final-step failure; the entire install is real.""" + + error_kind: ClassVar[type[PublicationFault]] = PublicationRuntimeError + + def harnesses(self) -> NoReturn: + assert RuntimeResolution.at(LaunchRuntime.installed().prefix).core_commit == self.target + raise self.error_kind(extract_stack(limit=2)[0]) + + +class PublicationInterruptedRun(PublicationFailingRun): + error_kind = PublicationInterrupt + + +@pytest.fixture +def publication_home(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch, + tmp_path: Path) -> Iterator[tuple[Path, LaunchRuntime]]: + """The hosted installed-runtime job supplies the build; source workers never build one.""" + candidate = LaunchRuntime.running() + if not Path(agent_comms.__file__).resolve().is_relative_to(candidate.prefix.resolve()): + pytest.skip("Full publication proof requires the hosted built-runtime job") + installed: InstalledHome = request.getfixturevalue("installed_home") + for name, value in installed.environment.items(): + monkeypatch.setenv(name, value) + replace_link(LaunchRuntime.current_pointer(), installed.outgoing.prefix) + # An isolated home must never restart the test runner's real user services. + binary = tmp_path / "host-programs" + binary.mkdir() + systemctl = binary / "systemctl" + systemctl.write_text( + '#!/bin/sh\n' + 'if [ "$1" = "--user" ] && [ "$2" = "show" ]; then\n' + " printf '0\\n'\n" + 'else\n' + ' exit 1\n' + 'fi\n' + ) + systemctl.chmod(0o700) + monkeypatch.setenv("PATH", os.pathsep.join((str(binary), os.environ["PATH"]))) + # Only the bridge socket needs a short locator; CLI storage keeps its trusted home. + with short_home(Path(installed.environment["HOME"])) as home: + monkeypatch.setenv("HOME", str(home)) + yield home, candidate + + +@pytest.mark.timeout(280) +@pytest.mark.parametrize("run_type,completion,terminal", ( + pytest.param(InstallRun, nullcontext(), "Install finished", id="finished"), + pytest.param( + PublicationFailingRun, pytest.raises(RuntimeError), "Install failed", id="crashed"), + pytest.param( + PublicationInterruptedRun, pytest.raises(KeyboardInterrupt), "Install failed", + id="interrupted", + ), +)) +def test_live_upgrade_family_after_real_publication( + publication_home: tuple[Path, LaunchRuntime], tmp_path: Path, + run_type: type[InstallRun], completion: AbstractContextManager[object], + terminal: str) -> None: + home, candidate = publication_home + checkout = tmp_path / "checkout" + subprocess.run(["git", "clone", "-q", REPOSITORY, checkout], check=True) + target = git("rev-parse", "HEAD") + subprocess.run(["git", "-C", checkout, "checkout", "--detach", "-q", + git("rev-parse", "HEAD^")], check=True) + outgoing = Stack(checkout, home=home, runtime=LaunchRuntime.installed()) + program = (str(candidate.interpreter.with_name("agent-comms-claude-bridge")),) + with session_bridge(tmp_path, home, program, session=tmp_path.name) as copied: + comms = Comms(tmp_path / "bus") + declare_test_bus(comms, channels=(InstallReport.channel,)) + publish_initial(comms, copied.binding.bus, "#all", "Retained before the upgrade") + before = HistoryCliCommand().apply(comms).messages + request = run_type.start(InstallCliCommand(checkout=checkout, to=target, project=checkout), + comms.root) + report = InstallReport(comms.root, outgoing) + with completion as caught: + upgrade(request, report) + assert isinstance(report.writer, CarriedWriter) + if isinstance(report.outcome, FailedInstall): + failure = report.outcome.failure + assert isinstance(failure, CrashedInstall) + error = failure.error + assert isinstance(error, PublicationFault) + assert isinstance(caught, pytest.ExceptionInfo) + assert caught.value is error + assert (failure.frame.filename, failure.frame.lineno) == ( + str(report.source.module_file(upgrade.__module__)), error.site.lineno, + ) + assert RuntimeResolution.at(LaunchRuntime.installed().prefix).core_commit == target + assert LaunchRuntime.installed().prefix != outgoing.runtime.prefix + resumed = InspectSession().information(copied.claim) + assert resumed.process == copied.observed.process + assert resumed.process.pid == copied.child.pid + assert resumed.runtime.prefix == LaunchRuntime.installed().prefix.resolve() + assert copied.child.poll() is None, copied.log.read_text() + reports = tuple(reversed(HistoryCliCommand(target=report.target).apply(comms).messages)) + assert len(reports) == 2 + assert reports[0].body.splitlines()[0] == "Install starting" + assert reports[1].body.splitlines()[0].startswith(terminal) + assert all(line in reports[1].body for line in report.outcome.lines()) + bridge_line = ResumedBridge( + copied.claim.thread, copied.observed.runtime, resumed.runtime).line() + assert reports[1].body.splitlines().count(bridge_line) == 1 + retained = { + message.reference: message for message in HistoryCliCommand().apply(comms).messages + } + assert all(retained[message.reference] == message for message in before) + assert not request.scratch.exists() diff --git a/tests/test_install_session_bridge.py b/tests/test_install_session_bridge.py index 78f4c03361..ed7ffeabbc 100644 --- a/tests/test_install_session_bridge.py +++ b/tests/test_install_session_bridge.py @@ -7,15 +7,23 @@ import sys import time from contextlib import ExitStack, contextmanager +from dataclasses import dataclass from pathlib import Path import pytest from agent_comms.child_process import ProcessIdentity -from agent_comms.cli_bridges.monitor import InspectSession, PauseSession from agent_comms.cli_bridges import claude +from agent_comms.cli_bridges.monitor import InspectSession, PauseSession from agent_comms.cli_bridges.shared import ( - MONITOR_FRAME, Binding, MonitorProbe, Serve, SessionBridgeResumes, + MONITOR_FRAME, + Binding, + BridgeResume, + MonitorProbe, + ResumedBridge, + Serve, + SessionBridge, + UnobservedBridge, ) from agent_comms.comms import Comms from agent_comms.field_codec import FieldCodec @@ -25,18 +33,29 @@ from agent_comms_hooks.watch import LapsedWatch, LiveWatch, WatchClaim, WatchRecord from source_programs import SourcePrograms -SERVE_FROM_SOURCE = (sys.executable, '-c', 'from agent_comms.cli_bridges.claude import main; main()') +SERVE_FROM_SOURCE = ( + sys.executable, '-c', 'from agent_comms.cli_bridges.claude import main; main()') + + +@dataclass(frozen=True) +class CopiedSession(SessionBridge): + """The captured bridge plus its owned test process, binding and saved directory.""" + + binding: Binding + state: Path + child: subprocess.Popen[bytes] + log: Path @contextmanager -def session_bridge(tmp_path, home, program=SERVE_FROM_SOURCE): +def session_bridge(tmp_path, home, program=SERVE_FROM_SOURCE, *, session='copied-session'): comms = Comms(tmp_path / 'bus') comms.messaging.initialize_private_initial_protocol() comms.registry.declare(comms.registry.new_thread( 'bridge-copy', frozenset(), str(tmp_path), execution=ExternalThreadExecution)) - transcript = tmp_path / 'copied-session.jsonl' + transcript = tmp_path / f'{session}.jsonl' transcript.write_text('') - state = home / '.local/state/agent-comms/claude-bridge-copy' + state = home / '.local/state/agent-comms' / f'claude-bridge-{transcript.stem}' binding = Binding(str(comms.root), 'bridge-copy', transcript.stem) with Binding.open(state) as db: binding.bind(db) @@ -62,11 +81,27 @@ def session_bridge(tmp_path, home, program=SERVE_FROM_SOURCE): time.sleep(.05) else: pytest.fail(log.read_text()) - yield home, claim, observed, child, log + yield CopiedSession(claim, observed, binding, state, child, log) finally: stop(child) +def test_copied_sessions_share_a_short_home_without_rebinding(tmp_path, bridge_home): + with ExitStack() as resources: + bridges = [] + for session in ('first-session', 'second-session'): + directory = tmp_path / session + directory.mkdir() + bridges.append(resources.enter_context( + session_bridge(directory, bridge_home, session=session))) + assert len({bridge.claim.socket_path for bridge in bridges}) == len(bridges) + for bridge in bridges: + assert InspectSession().information(bridge.claim).process == bridge.observed.process + assert bridge.child.poll() is None, bridge.log.read_text() + assert {bridge.claim for bridge in Serve.running(bridge_home)} == { + bridge.claim for bridge in bridges} + + def stop(child): if child.poll() is None: child.terminate() @@ -83,25 +118,26 @@ def probe(child): def test_real_monitor_survives_bridge_retirement_and_requires_a_fresh_probe(tmp_path, bridge_home): - with session_bridge(tmp_path, bridge_home) as (home, claim, _, child, log): - env = {**os.environ, 'HOME': str(home)} + with session_bridge(tmp_path, bridge_home) as copied: + env = {**os.environ, 'HOME': str(copied.claim.home)} with ExitStack() as cleanup: monitor = subprocess.Popen( (sys.executable, '-c', 'import asyncio; from agent_comms.cli_bridges.monitor import read_monitor; ' 'from agent_comms_hooks.watch import WatchClaim; ' - 'asyncio.run(read_monitor(WatchClaim("copied-session")))'), + f'asyncio.run(read_monitor(WatchClaim({copied.claim.thread!r})))'), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) cleanup.callback(stop, monitor) first = probe(monitor) - (bridge,) = Serve.running(home) + (bridge,) = Serve.running(copied.claim.home) with ExitStack() as resources: bridge.pause(resources) assert monitor.poll() is None - assert child.wait(timeout=15) == 0, log.read_text() - with log.open('ab') as output: - successor = subprocess.Popen(child.args, env=env, stdout=output, stderr=output) + assert copied.child.wait(timeout=15) == 0, copied.log.read_text() + with copied.log.open('ab') as output: + successor = subprocess.Popen( + copied.child.args, env=env, stdout=output, stderr=output) cleanup.callback(stop, successor) second = probe(monitor) assert second.session == first.session @@ -112,41 +148,43 @@ def test_real_monitor_survives_bridge_retirement_and_requires_a_fresh_probe(tmp_ @pytest.mark.parametrize('wrong_identity', ('owner', 'process')) def test_wrong_pause_identity_cannot_stop_the_real_session_bridge( tmp_path, bridge_home, wrong_identity): - with session_bridge(tmp_path, bridge_home) as (_, claim, observed, child, log): + with session_bridge(tmp_path, bridge_home) as copied: other = ProcessIdentity.capture(os.getppid()) request = PauseSession( - owner=other if wrong_identity == 'owner' else observed.owner, - process=other if wrong_identity == 'process' else observed.process, + owner=other if wrong_identity == 'owner' else copied.observed.owner, + process=other if wrong_identity == 'process' else copied.observed.process, ) with pytest.raises(ValueError, match='different'): - request.exchange(claim) - assert InspectSession().information(claim).process == observed.process - assert child.poll() is None, log.read_text() + request.exchange(copied.claim) + assert InspectSession().information(copied.claim).process == copied.observed.process + assert copied.child.poll() is None, copied.log.read_text() def test_installer_pause_closes_the_real_bridge_and_eof_never_replays_or_relaunches( tmp_path, bridge_home): - with session_bridge(tmp_path, bridge_home) as (home, claim, observed, child, log): - directory = home / '.local/state/agent-comms' - with Binding.open(directory / 'claude-bridge-copy') as db: - (binding,) = Binding.select(db) - with Binding.open(directory / 'claude-bridge-historical-copy') as db: - binding.bind(db) + with session_bridge(tmp_path, bridge_home) as copied: + with Binding.open(copied.state.with_name(copied.state.name + '-historical')) as db: + copied.binding.bind(db) # Historical journals can name the same canonical live claim. - (bridge,) = Serve.running(home) - assert bridge.observed.process == observed.process + (bridge,) = Serve.running(copied.claim.home) + assert bridge.observed.process == copied.observed.process with ExitStack() as resources: - (paused,) = bridge.pause(resources) + paused = bridge.pause(resources) assert paused.bridge == bridge - assert child.poll() is None + assert copied.child.poll() is None # The paused process keeps the claim; no session hook sees it lapse. - assert isinstance(claim.state(), LiveWatch) - assert not claim.socket_path.exists() + assert isinstance(copied.claim.state(), LiveWatch) + assert not copied.claim.socket_path.exists() # Losing the installer connection grants no fallback runtime or retry. - assert child.wait(timeout=15) == 0, log.read_text() - assert isinstance(claim.state(), LapsedWatch) - with Binding.open(home / '.local/state/agent-comms/claude-bridge-copy') as db: - assert Binding.select(db)[0].thread == claim.thread + assert copied.child.wait(timeout=15) == 0, copied.log.read_text() + assert isinstance(copied.claim.state(), LapsedWatch) + outcome = paused.resume() + assert isinstance(outcome, UnobservedBridge) + assert outcome.name == copied.claim.thread + assert outcome.reason != UnobservedBridge(copied.claim.thread).reason + assert FieldCodec.decode(BridgeResume, FieldCodec.encode(outcome)) == outcome + with Binding.open(copied.state) as db: + assert Binding.select(db)[0].thread == copied.claim.thread def test_session_start_and_runtime_replacement_share_the_declared_launcher(tmp_path): @@ -164,7 +202,7 @@ def test_session_start_and_runtime_replacement_share_the_declared_launcher(tmp_p def bridges_serving(state): """Every process serving this test's state; a second bridge would show here.""" - key = os.fsencode(state) # under the per-test home; the successor's absolute path ends with it + key = os.fsencode(state) serving = [] for entry in Path('/proc').iterdir(): try: @@ -178,50 +216,49 @@ def bridges_serving(state): def test_install_swap_keeps_the_claim_and_resumes_the_same_process_on_the_new_runtime( tmp_path, bridge_home, monkeypatch): - old, new = (SourcePrograms.create(tmp_path / name) for name in ('old', 'new')) - with session_bridge(tmp_path, bridge_home, (str(old.script(claude.main)),)) as ( - home, claim, observed, child, log): - current = home / LaunchRuntime.current_pointer().relative_to(Path.home()) + old = SourcePrograms.current(tmp_path / 'source-programs') + new = LaunchRuntime.installed() + with session_bridge(tmp_path, bridge_home, (str(old.script(claude.main)),)) as copied: + current = copied.claim.home / LaunchRuntime.current_pointer().relative_to(Path.home()) current.parent.mkdir(parents=True) current.symlink_to(new.prefix.absolute(), target_is_directory=True) - monkeypatch.setenv('HOME', str(home)) - assert observed.runtime.prefix == old.prefix.resolve() - env = {**os.environ, 'HOME': str(home)} - packet = json.dumps(dict(session_id=claim.thread, hook_event_name='SessionStart', - transcript_path=str(observed.transcript))) - (bridge,) = Serve.running(home) - bridges = SessionBridgeResumes() + monkeypatch.setenv('HOME', str(copied.claim.home)) + assert copied.observed.runtime.prefix == old.prefix.resolve() + env = {**os.environ, 'HOME': str(copied.claim.home)} + packet = json.dumps(dict(session_id=copied.claim.thread, hook_event_name='SessionStart', + transcript_path=str(copied.observed.transcript))) + (bridge,) = Serve.running(copied.claim.home) with ExitStack() as cleanup: with ExitStack() as resources: - bridges.pause(bridge, resources) - resources.callback(bridges.resume) - assert not claim.socket_path.exists() - assert isinstance(claim.state(), LiveWatch) + paused = bridge.pause(resources) + assert not copied.claim.socket_path.exists() + assert isinstance(copied.claim.state(), LiveWatch) hook = subprocess.run( (str(old.script(started)),), input=packet, text=True, capture_output=True, - timeout=40, env={**env, 'CLAUDE_CODE_SESSION_ID': claim.thread, + timeout=40, env={**env, 'CLAUDE_CODE_SESSION_ID': copied.claim.thread, 'CLAUDE_PID': str(os.getpid())}) assert hook.returncode != 0 assert 'No second bridge launched' in hook.stdout + hook.stderr - assert bridges_serving(home / '.local/state/agent-comms/claude-bridge-copy') == [child.pid] - # A Monitor started inside the gap waits for the endpoint instead of exiting. + assert bridges_serving(copied.state) == [copied.child.pid] monitor = subprocess.Popen( (sys.executable, '-c', 'import asyncio; from agent_comms.cli_bridges.monitor import read_monitor; ' 'from agent_comms_hooks.watch import WatchClaim; ' - f'asyncio.run(read_monitor(WatchClaim({claim.thread!r})))'), + f'asyncio.run(read_monitor(WatchClaim({copied.claim.thread!r})))'), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) cleanup.callback(stop, monitor) - bridges.require_resumed() - assert bridges.lines() == () - resumed = InspectSession().information(claim) - assert resumed.process == observed.process - assert resumed.process.pid == child.pid + outcome = paused.resume() + resumed = InspectSession().information(copied.claim) + assert outcome == ResumedBridge(copied.claim.thread, copied.observed.runtime, + resumed.runtime) + assert resumed.process == copied.observed.process + assert resumed.process.pid == copied.child.pid assert resumed.runtime.prefix == new.prefix.resolve() - state = claim.state() + state = copied.claim.state() assert isinstance(state, LiveWatch) - assert Path(state.record.command[0]).resolve() == new.script(claude.main).resolve() - assert bridges_serving(home / '.local/state/agent-comms/claude-bridge-copy') == [child.pid] - assert probe(monitor).session == claim.thread - assert child.poll() is None, log.read_text() + assert Path(state.record.command[0]).resolve() == new.interpreter.with_name( + old.script(claude.main).name).resolve() + assert bridges_serving(copied.state) == [copied.child.pid] + assert probe(monitor).session == copied.claim.thread + assert copied.child.poll() is None, copied.log.read_text() diff --git a/tests/test_session_bridge_owner.py b/tests/test_session_bridge_owner.py index 2ddb7a37cb..d8ae59fd14 100644 --- a/tests/test_session_bridge_owner.py +++ b/tests/test_session_bridge_owner.py @@ -20,10 +20,12 @@ from agent_comms.thread_execution import ExternalThreadExecution from agent_comms_hooks.session import started from agent_comms_hooks.watch import LiveWatch, WatchClaim, WatchRecord +from source_programs import SourcePrograms def test_session_start_hook_exit_preserves_the_recorded_owner_for_install_pause( - tmp_path, bridge_home, source_programs): + tmp_path, bridge_home): + source_programs = SourcePrograms.current(tmp_path / 'source-programs') home = bridge_home session = 'copied-session' comms = Comms(tmp_path / 'bus') @@ -88,7 +90,7 @@ def unexpected_serve(): assert observed.process.alive() (bridge,) = Serve.running(home) with ExitStack() as resources: - (paused,) = bridge.pause(resources) + paused = bridge.pause(resources) assert paused.bridge.observed.owner == expected assert isinstance(claim.state(), LiveWatch) assert owner.poll() is None