From eebf7ff2cb8a6ff90fad9907bde760e185f10740 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:36:20 -0400 Subject: [PATCH 01/21] Discover installer carry steps through DeclaredFamily --- src/agent_comms/installer/carry.py | 30 +++++++----------------- src/agent_comms/installer/step_worker.py | 6 ++--- src/agent_comms/installer/upgrade.py | 8 +++---- tools/cutover/carry_bus_history.py | 2 +- 4 files changed, 16 insertions(+), 30 deletions(-) diff --git a/src/agent_comms/installer/carry.py b/src/agent_comms/installer/carry.py index 0ffc905be3..e8f6321888 100644 --- a/src/agent_comms/installer/carry.py +++ b/src/agent_comms/installer/carry.py @@ -14,7 +14,7 @@ import inspect import subprocess -from abc import ABC, abstractmethod +from abc import abstractmethod from collections.abc import Callable from dataclasses import dataclass, field from graphlib import TopologicalSorter @@ -42,34 +42,20 @@ class Carried: @dataclass(frozen=True) -class Carry(ABC): +class Carry(DeclaredFamily, affix="", discovery_package="cutover"): """One migration's decoded source, destination and retirement location.""" root: Annotated[Path, PathText] original_root: Annotated[Path, PathText] spill_dir: Annotated[Path, PathText] - declared: ClassVar[dict[str, type[Carry]]] = {} precedes: ClassVar[tuple[type[Carry], ...]] = () """Steps whose decoders require this step's output, regardless of addition date.""" - """Every step imported so far, by its module name.""" - - def __init_subclass__(cls, **kwargs: object) -> None: - super().__init_subclass__(**kwargs) - if inspect.isabstract(cls): - return - if cls.name() in Carry.declared and Carry.declared[cls.name()] is not cls: - raise TypeError(f"{cls.name()} declares a second carry step") - Carry.declared[cls.name()] = cls @classmethod def module(cls) -> Path: return Path(inspect.getfile(cls)).resolve() - @classmethod - def name(cls) -> str: - return cls.module().stem - @abstractmethod def backup_files(self) -> tuple[Annotated[Path, PathText], ...]: """The existing files this step rewrites or deletes, backed up before it applies.""" @@ -124,19 +110,21 @@ class CarryChain: def at(cls, tree: Path) -> CarryChain: """Import the tree's steps (its ``tools`` is on the path) and order them by the first-parent commit that added each file.""" - directory = (tree / STEPS).resolve() + tree = tree.resolve() + directory = tree / STEPS for path in sorted(directory.glob("*.py")): import_module(f"{directory.name}.{path.stem}") - steps = [step for step in Carry.declared.values() if step.module().parent == directory] + steps = [step for step in Carry.members_with(Carry) if step.module().parent == directory] log = subprocess.check_output( ["git", "-C", str(tree), "log", "-m", "--first-parent", "--no-renames", "--diff-filter=A", "--reverse", "--name-only", "--format=", "HEAD", "--", str(STEPS)]).decode().split() - added = {Path(path).stem: position for position, path in reversed(list(enumerate(log)))} - unordered = sorted(step.name() for step in steps if step.name() not in added) + added = {Path(path): position for position, path in reversed(list(enumerate(log)))} + unordered = sorted(step.declared_name for step in steps + if step.module().relative_to(tree) not in added) if unordered: raise ValueError(f"{unordered} are not on the main line, so nothing orders them") - ordered = sorted(steps, key=lambda step: added[step.name()]) + ordered = sorted(steps, key=lambda step: added[step.module().relative_to(tree)]) dependencies: dict[type[Carry], set[type[Carry]]] = {step: set() for step in ordered} for step in ordered: for later in step.precedes: diff --git a/src/agent_comms/installer/step_worker.py b/src/agent_comms/installer/step_worker.py index 180cae19a8..464756c8f6 100644 --- a/src/agent_comms/installer/step_worker.py +++ b/src/agent_comms/installer/step_worker.py @@ -7,17 +7,15 @@ from __future__ import annotations import sys -from importlib import import_module from pathlib import Path from agent_comms.field_codec import FieldCodec -from agent_comms.installer.carry import STEPS, Carry, CarryOperation +from agent_comms.installer.carry import Carry, CarryOperation def main() -> None: name, encoded_step, encoded_operation, result_path = sys.argv[1:] - import_module(f"{STEPS.name}.{name}") - step = FieldCodec.decode(Carry.declared[name], FieldCodec.parse(encoded_step)) + step = FieldCodec.decode(Carry.member_named(name), FieldCodec.parse(encoded_step)) operation = FieldCodec.decode(CarryOperation, FieldCodec.parse(encoded_operation)) result = operation.run(step) Path(result_path).write_text(operation.encode_answer(result)) diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index e714f97ba9..f0be06f94e 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -250,7 +250,7 @@ class CarriedPass(Extendable): def after_acceptance(self, pending: tuple[Carry, ...], stack: Stack, tree: TargetTree) -> Self: return replace(self, retired_inputs=tuple( - StepRun(step.name(), stack.call_step(step, tree, RetireInputsOperation())) + StepRun(step.declared_name, stack.call_step(step, tree, RetireInputsOperation())) for step in pending )) @@ -564,7 +564,7 @@ def call_step[Answer](self, step: Carry, tree: TargetTree, """Run the step in the target tree with this stack's interpreter.""" with tempfile.TemporaryDirectory() as scratch: result = Path(scratch) / "result.json" - self.commands.run(self.python, "-m", step_worker.__name__, step.name(), + self.commands.run(self.python, "-m", step_worker.__name__, step.declared_name, json.dumps(FieldCodec.encode(step)), json.dumps(FieldCodec.encode(operation)), result, cwd=tree.path, env=self.environment(*tree.source_paths())) @@ -763,7 +763,7 @@ def rehearsed(cls, checkout: Path, root: Path, current: str, target: str, def reported(self, rehearsal: Rehearsed) -> RehearsedUpgrade: return RehearsedUpgrade( self.current, self.target, self.root, - tuple(step.name() for step in self.pending), rehearsal, + tuple(step.declared_name for step in self.pending), rehearsal, commands=tuple(self.staged.commands.ran)) def finished(self, rehearsal: Rehearsed, bridges: SessionBridgeResumes, @@ -861,7 +861,7 @@ def carry(self, pending: tuple[Carry, ...], staged: Stack, """One migration sequence for rehearsal and the backed-up live bus.""" steps = [] for step in pending: - steps.append(StepRun(step.name(), staged.call_step( + steps.append(StepRun(step.declared_name, staged.call_step( step, tree, MigrateOperation(apply=True)))) reconciled = staged.cli(self.root, ReconcileStoresCliCommand) steps.append(StepRun(ReconcileStoresCliCommand.declared_name, Carried( diff --git a/tools/cutover/carry_bus_history.py b/tools/cutover/carry_bus_history.py index e22e44881d..03adf9786a 100644 --- a/tools/cutover/carry_bus_history.py +++ b/tools/cutover/carry_bus_history.py @@ -970,7 +970,7 @@ def after_acceptance(self) -> Carried: self.root, *(Path(source.root) for source in HistoryArchive(Bus.path_in(self.root)).sources()), ) - stage = Path(tempfile.mkdtemp(prefix=f"{self.name()}-", dir=spill)) + stage = Path(tempfile.mkdtemp(prefix=f"{self.declared_name}-", dir=spill)) rewritten = tuple( name for directory in roots From 654e80b99b027119273063e362aff1bd3b31a014 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:39:16 -0400 Subject: [PATCH 02/21] Let registry discovery own carry loading without a second import walk --- src/agent_comms/installer/carry.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/agent_comms/installer/carry.py b/src/agent_comms/installer/carry.py index e8f6321888..a08f908471 100644 --- a/src/agent_comms/installer/carry.py +++ b/src/agent_comms/installer/carry.py @@ -18,7 +18,6 @@ from collections.abc import Callable from dataclasses import dataclass, field from graphlib import TopologicalSorter -from importlib import import_module from pathlib import Path from typing import Annotated, ClassVar @@ -112,8 +111,6 @@ def at(cls, tree: Path) -> CarryChain: first-parent commit that added each file.""" tree = tree.resolve() directory = tree / STEPS - for path in sorted(directory.glob("*.py")): - import_module(f"{directory.name}.{path.stem}") steps = [step for step in Carry.members_with(Carry) if step.module().parent == directory] log = subprocess.check_output( ["git", "-C", str(tree), "log", "-m", "--first-parent", "--no-renames", From ded6cdd83091a8af62f26e5df56556d5a547ddde Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:42:59 -0400 Subject: [PATCH 03/21] Compare declared file groups and carry typed rewrite paths --- src/agent_comms/cli_commands.py | 4 +-- src/agent_comms/installer/backup.py | 34 +++++++++++++------- src/agent_comms/installer/carry.py | 2 +- src/agent_comms/installer/upgrade.py | 37 ++++++++++++---------- src/agent_comms/locked_store.py | 34 ++++++++++++++++++-- tools/cutover/carry_bus_history.py | 17 +++++----- tools/cutover/carry_context_request_key.py | 2 +- tools/cutover/carry_dead_turns.py | 2 +- tools/cutover/carry_drop_connectivity.py | 2 +- tools/cutover/carry_finished_turns.py | 8 ++--- tools/cutover/carry_owner_requests.py | 14 ++++---- tools/cutover/carry_turn_owner.py | 4 +-- 12 files changed, 103 insertions(+), 57 deletions(-) diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index e30e9c60c5..171a2b4d00 100644 --- a/src/agent_comms/cli_commands.py +++ b/src/agent_comms/cli_commands.py @@ -1765,7 +1765,7 @@ def apply(self, ctx: Comms) -> CliResult: class ReconciledStoresResult(CliResult): """The bus files whose schema the reconcile changed, relative to the bus root.""" - rewritten: tuple[str, ...] + rewritten: tuple[Annotated[Path, PathText], ...] @dataclass(frozen=True) @@ -1810,7 +1810,7 @@ def changes(owner: type[ReconciledFile], path: Path) -> bool: def answer(self, root: Path) -> ReconciledStoresResult: return ReconciledStoresResult(tuple( - str(path.relative_to(root)) + path.relative_to(root) for path, owner in sorted(ReconciledFile.existing_in(root).items()) if self.changes(owner, path))) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index d16d2e3908..6699742b88 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -23,7 +23,7 @@ from agent_comms.coordination_errors import CoordinationError from agent_comms.field_codec import FieldCodec from agent_comms.installer.interpreter_operation import RouteSelection -from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile +from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile, UndeclaredFile from agent_comms.private_path import FileObservation, FileRevision, MissingFile from agent_comms.store_files import AtomicFile @@ -55,21 +55,31 @@ def matches_extraction(self, copied: SavedFile) -> bool: return self.size == copied.size -def bus_files(root: Path) -> dict[str, SavedFile]: +def bus_files(root: Path) -> dict[tuple[type[RootFile] | UndeclaredFile, Path], dict[Path, SavedFile]]: """Read saved file identity and modification metadata, never contents.""" excluded = RuntimeFile.paths_in(root) - saved = {} + observed: dict[Path, os.stat_result] = {} for path in root.rglob("*"): - if path in excluded or path.name.endswith((".lock", "-shm", "-journal")): + if path in excluded: continue - observed = path.lstat() - # Empty WAL files appear on a read-only SQLite open; they hold no committed frames. - if path.name.endswith("-wal") and observed.st_size == 0: - continue - if stat.S_ISREG(observed.st_mode): - saved[str(path.relative_to(root))] = SavedFile( - observed.st_size, observed.st_mtime_ns, observed.st_ino) - return saved + info = path.lstat() + if stat.S_ISREG(info.st_mode): + observed[path] = info + declared = RootFile.existing_in(root) + owners = {path: owner for path, owner in declared.items() if path in observed} + companions = {member for path, owner in owners.items() + for member in owner.member_paths(path) if member != path} + owners.update((path, RootFile.content_owner(path)) + for path in observed.keys() - owners.keys() - companions) + companions.update(member for path, owner in owners.items() + for member in owner.member_paths(path) if member != path) + return { + (declared.get(path, UndeclaredFile(path.relative_to(root))), path.relative_to(root)): { + member.relative_to(root): SavedFile( + observed[member].st_size, observed[member].st_mtime_ns, observed[member].st_ino, + ) for member in owner.snapshot_members(path, observed) + } for path, owner in owners.items() if path not in companions + } @dataclass(frozen=True) diff --git a/src/agent_comms/installer/carry.py b/src/agent_comms/installer/carry.py index a08f908471..252de64b98 100644 --- a/src/agent_comms/installer/carry.py +++ b/src/agent_comms/installer/carry.py @@ -35,7 +35,7 @@ class Carried: changed: int detail: dict[str, int | bool] = field(default_factory=dict) - rewritten: tuple[str, ...] = () + rewritten: tuple[Annotated[Path, PathText], ...] = () """Files under the bus root this run wrote or deleted; every other file stays byte-identical.""" diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index f0be06f94e..9ae0326b2a 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -104,7 +104,7 @@ from agent_comms.installer.reporting import RootWriter, RuntimeWriter, UnpostedWriter from agent_comms.installer.runtime_resolution import RuntimeResolution from agent_comms.launch_runtime import LaunchRuntime, RestartEnvironment, SpecifiedImportPath -from agent_comms.locked_store import BusFile, RootFile +from agent_comms.locked_store import BusFile, RootFile, UndeclaredFile from agent_comms.maintenance_barrier import ( MaintenanceBarrier, MaintenanceOperator, @@ -136,7 +136,7 @@ class RehearsalFiles: identical: int rewritten: int - names: tuple[str, ...] + names: tuple[Annotated[Path, PathText], ...] restore_bytes: int @classmethod @@ -145,7 +145,7 @@ def none(cls) -> RehearsalFiles: return cls(0, 0, (), 0) @staticmethod - def declared(steps: tuple[StepRun, ...]) -> frozenset[str]: + def declared(steps: tuple[StepRun, ...]) -> frozenset[Path]: return frozenset(name for step in steps for name in step.carried.rewritten) def require_rehearsal(self, rehearsal: RehearsalFiles) -> None: @@ -155,25 +155,30 @@ def require_rehearsal(self, rehearsal: RehearsalFiles) -> None: -def carried(root: Path, before: dict[str, SavedFile], steps: tuple[StepRun, ...], - expected: frozenset[str]) -> RehearsalFiles: +def carried( + root: Path, + before: dict[tuple[type[RootFile] | UndeclaredFile, Path], dict[Path, SavedFile]], + steps: tuple[StepRun, ...], expected: frozenset[Path], +) -> RehearsalFiles: """Only the files a step names as rewritten may differ from ``before``.""" after = bus_files(root) - physical = {name for name in before.keys() | after.keys() - if before.get(name) != after.get(name)} - # SQLite's committed WAL belongs to its database declaration, not a second carry. - changed = {name[:-4] if name.endswith('-wal') and name[:-4] in before.keys() | after.keys() - else name for name in physical} - unnamed = sorted(changed - RehearsalFiles.declared(steps)) + physical = {key for key in before.keys() | after.keys() + if before.get(key) != after.get(key)} + members = {member: primary for (_owner, primary), snapshot in (*before.items(), *after.items()) + for member in snapshot} + changed = {primary for _owner, primary in physical} + declared = {members.get(path, path) for path in RehearsalFiles.declared(steps)} + unnamed = sorted(changed - declared) if unnamed: raise UpgradeRefused(f"Carries changed files no step names: {unnamed}") - outside = sorted(changed - expected) + outside = sorted(changed - {members.get(path, path) for path in expected}) if outside: raise UpgradeRefused(f"Carries changed files outside the expected set: {outside}") return RehearsalFiles( - identical=len(before.keys() - changed), rewritten=len(changed), + identical=len({primary for _owner, primary in before} - changed), rewritten=len(changed), names=tuple(sorted(changed)), - restore_bytes=sum(before[name].size for name in physical & before.keys())) + restore_bytes=sum(file.size for key in physical & before.keys() + for file in before[key].values())) @dataclass(frozen=True) @@ -843,7 +848,7 @@ class PreparedMigration(Extendable): """The held bus as the outgoing runtime read it, before any migration.""" hold: MaintenanceState - before: dict[str, SavedFile] + before: dict[tuple[type[RootFile] | UndeclaredFile, Path], dict[Path, SavedFile]] read: BusReadBack @property @@ -914,7 +919,7 @@ def uncopied(self) -> UncopiedRehearsal: def live(self, rehearsal: RehearsedFiles, report: InstallReport) -> CarriedRun: expected = frozenset(rehearsal.files.names) | frozenset( - str(path.relative_to(self.root)) + path.relative_to(self.root) for path in self.staged.backup_files(self.pending, self.tree)) selected = tuple(self.root / name for name in sorted(expected)) restore = RestoreSpace.measure(max( diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index ad86b3f852..532459a160 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -162,6 +162,13 @@ def member_paths(cls, path: Path) -> frozenset[Path]: """The resources composed onto one declared primary file.""" return frozenset((path, store_lock_path(path))) + @classmethod + def snapshot_members( + cls, path: Path, observed: dict[Path, os.stat_result], + ) -> frozenset[Path]: + """Content changes count; lock and coordination files are not content.""" + return frozenset((path,)) + @classmethod def files_in(cls, root: Path) -> frozenset[Path]: return cls.member_paths(cls.path_in(root)) @@ -237,6 +244,13 @@ def paths_in(cls, root: Path) -> frozenset[Path]: ) +@dataclass(frozen=True) +class UndeclaredFile(RootFile): + """A physical file without a stored declaration, still identified by its path.""" + + path: Path + + class FileStore(RootFile, ABC): """Construction belongs only to owners whose initializer takes a file.""" @@ -366,7 +380,8 @@ class SQLiteFile(RootFile): """SQLite connections, snapshots, copies and companions share one file owner.""" __slots__ = () - companion_suffixes: ClassVar[tuple[str, ...]] = ("-wal", "-shm", "-journal") + wal_suffix: ClassVar[str] = "-wal" + transient_suffixes: ClassVar[tuple[str, ...]] = ("-shm", "-journal") @classmethod def writer( @@ -497,7 +512,22 @@ def digest(cls, path: Path) -> str: @classmethod def files(cls, path: Path) -> tuple[Path, ...]: - return (path, *(path.with_name(path.name + suffix) for suffix in cls.companion_suffixes)) + return (path, cls.wal_path(path), + *(path.with_name(path.name + suffix) for suffix in cls.transient_suffixes)) + + @classmethod + def wal_path(cls, path: Path) -> Path: + return path.with_name(path.name + cls.wal_suffix) + + @classmethod + def snapshot_members( + cls, path: Path, observed: dict[Path, os.stat_result], + ) -> frozenset[Path]: + wal = cls.wal_path(path) + # A read can create an empty WAL; only a nonempty WAL holds content. + return super().snapshot_members(path, observed) | frozenset( + (wal,) if wal in observed and observed[wal].st_size else () + ) @classmethod def member_paths(cls, path: Path) -> frozenset[Path]: diff --git a/tools/cutover/carry_bus_history.py b/tools/cutover/carry_bus_history.py index 03adf9786a..af69e32a4f 100644 --- a/tools/cutover/carry_bus_history.py +++ b/tools/cutover/carry_bus_history.py @@ -150,7 +150,7 @@ def publish( if path.read_bytes() == payload.encode(): return Carried(changed, rewritten=rewritten) AtomicFile(path).write_text(payload) - return Carried(changed + len(converted), rewritten=(*rewritten, HistoryArchive.filename)) + return Carried(changed + len(converted), rewritten=(*rewritten, path.relative_to(root))) def mapped_root(self, root: Path, original_root: Path) -> Path: original = Path(self.root) @@ -219,7 +219,7 @@ def converted( source.validate() return source, Carried( 1, - rewritten=tuple(str(path.relative_to(root)) for path in registry.written), + rewritten=tuple(path.relative_to(root) for path in registry.written), ) @@ -819,7 +819,7 @@ def inspect(self) -> Carried: return replace(result, changed=changed) @staticmethod - def moved(root: Path, directory: Path, spill: Path) -> tuple[str, ...]: + def moved(root: Path, directory: Path, spill: Path) -> tuple[Path, ...]: """Retire each old file after acceptance, never while rollback can need its inode.""" rewritten = [] for name in (BusHistory.filename, HistoryMetadata.filename): @@ -832,7 +832,7 @@ def moved(root: Path, directory: Path, spill: Path) -> tuple[str, ...]: Carry.require_confined(spill, target) source.chmod(0o400) shutil.move(source, target) - rewritten.append(str(source.relative_to(root))) + rewritten.append(source.relative_to(root)) return tuple(rewritten) @@ -904,7 +904,7 @@ def rebind_reads(root: Path, original: RecordedBusIdentity) -> Carried: "read_transcripts_before": len(document.transcripts), "read_transcripts_after": len(rebound.transcripts), }, - rewritten=(ledger.filename,), + rewritten=(ledger.path.relative_to(root),), ) def carry(self, *, apply: bool) -> Carried: @@ -936,7 +936,7 @@ def carry(self, *, apply: bool) -> Carried: results = [] read_results = [] bindings = {} - rewritten: set[str] = set() + rewritten: set[Path] = set() for source in inputs: result = self.transfer(source.source, source.bus, source.metadata) read = self.rebind_reads(source.bus.path.parent, source.original_identity) @@ -945,7 +945,7 @@ def carry(self, *, apply: bool) -> Carried: bindings[source.original_identity] = RecordedBusIdentity.capture(source.bus.path) directory = source.bus.path.parent.relative_to(self.root) rewritten.update( - str(directory / name) for name in (*result.rewritten, *read.rewritten) + directory / name for name in (*result.rewritten, *read.rewritten) ) catalog_result = CarryCatalogSource.publish( self.root, self.original_root, catalog, bindings @@ -1039,7 +1039,8 @@ def transfer(cls, source: Path, bus: Bus, metadata: HistoryMetadata) -> Carried: return Carried( changed, detail=dict(rows=count, continuous=count, round_tripped=count), - rewritten=(bus.path.name, *(path.name for path in bus.source_paths(bus.path.parent))), + rewritten=(bus.path.relative_to(bus.path.parent), + *(path.relative_to(bus.path.parent) for path in bus.source_paths(bus.path.parent))), ) diff --git a/tools/cutover/carry_context_request_key.py b/tools/cutover/carry_context_request_key.py index eab80c62ec..170233e4d0 100644 --- a/tools/cutover/carry_context_request_key.py +++ b/tools/cutover/carry_context_request_key.py @@ -43,4 +43,4 @@ def carry(self, *, apply: bool) -> Carried: with SQLiteFile.write_transaction(db): ContextRequest.rebuild(db) ContextStore.require_schema(db) - return Carried(1, {"context_requests": rows}, (ContextStore.filename,)) + return Carried(1, {"context_requests": rows}, (self.path.relative_to(self.root),)) diff --git a/tools/cutover/carry_dead_turns.py b/tools/cutover/carry_dead_turns.py index df146abfde..5e326e9a06 100644 --- a/tools/cutover/carry_dead_turns.py +++ b/tools/cutover/carry_dead_turns.py @@ -46,5 +46,5 @@ def carry(self, *, apply: bool) -> Carried: if apply: path.unlink() return Carried(len(dead), detail={"kept": len(current), "deleted": len(dead)}, - rewritten=tuple(str(path.relative_to(self.root)) for path in dead) + rewritten=tuple(path.relative_to(self.root) for path in dead) if apply else ()) diff --git a/tools/cutover/carry_drop_connectivity.py b/tools/cutover/carry_drop_connectivity.py index 33cdebac74..e27234ef89 100644 --- a/tools/cutover/carry_drop_connectivity.py +++ b/tools/cutover/carry_drop_connectivity.py @@ -45,4 +45,4 @@ def carry(self, *, apply: bool) -> Carried: CoordinationStore.schema_change(self.path).require() with SQLiteFile.write_transaction(db): db.execute(f'DROP TABLE "{TABLE}"') - return Carried(1, {TABLE: rows}, (CoordinationStore.filename,)) + return Carried(1, {TABLE: rows}, (self.path.relative_to(self.root),)) diff --git a/tools/cutover/carry_finished_turns.py b/tools/cutover/carry_finished_turns.py index 2c3f91c8fd..24c926a227 100644 --- a/tools/cutover/carry_finished_turns.py +++ b/tools/cutover/carry_finished_turns.py @@ -58,7 +58,7 @@ def carry(self, *, apply: bool) -> Carried: resealed = True rebound.append(source) rewritten = [ - str(root.relative_to(self.root) / name) + root.relative_to(self.root) / name for root, result in results for name in result.rewritten ] @@ -66,7 +66,7 @@ def carry(self, *, apply: bool) -> Carried: AtomicFile(HistoryArchive.path_in(self.root)).write_text( json.dumps(FieldCodec.encode(tuple(rebound))) ) - rewritten.append(HistoryArchive.filename) + rewritten.append(HistoryArchive.path_in(self.root).relative_to(self.root)) finished = sum(int(result.detail.get("finished", 0)) for _, result in results) return Carried( sum(result.changed for _, result in results), @@ -100,8 +100,8 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: return Carried(0) raw = json.dumps(document) registry._decode(FieldCodec.parse(raw)) - rewritten: tuple[str, ...] = () + rewritten: tuple[Path, ...] = () if apply: registry._write_unlocked(raw) - rewritten = tuple(str(path.relative_to(root)) for path in registry.written) + rewritten = tuple(path.relative_to(root) for path in registry.written) return Carried(carried, {"finished": finished}, rewritten) diff --git a/tools/cutover/carry_owner_requests.py b/tools/cutover/carry_owner_requests.py index 4f583b76e1..e78a2db2fa 100644 --- a/tools/cutover/carry_owner_requests.py +++ b/tools/cutover/carry_owner_requests.py @@ -153,18 +153,18 @@ def carry(self, *, apply: bool) -> Carried: mapped = source.mapped_root(self.root, self.original_root) result = self.carry_root(mapped, apply=apply) results.append((mapped, result)) - if apply and RegistryStore.at(mapped).path.name in result.rewritten: + if apply and RegistryStore.at(mapped).path.relative_to(mapped) in result.rewritten: source = source.carried_registry(mapped) resealed = True rebound.append(source) rewritten = [ - str(root.relative_to(self.root) / name) + root.relative_to(self.root) / name for root, result in results for name in result.rewritten ] if resealed: AtomicFile(HistoryArchive.path_in(self.root)).write_text(json.dumps(FieldCodec.encode(tuple(rebound)))) - rewritten.append(HistoryArchive.filename) + rewritten.append(HistoryArchive.path_in(self.root).relative_to(self.root)) detail: dict[str, int] = {} for _, result in results: for key, value in result.detail.items(): @@ -341,7 +341,7 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: for entry in GoalHistoryEntry.stream(history): for prefix in entry.request_revisions: prefix.validate(requests) - rewritten: list[str] = [] + rewritten: list[Path] = [] if apply: # Prepared in-memory files are private to this carry. Replacing a # retained file requires the installer's actual maintenance hold. @@ -358,15 +358,15 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried: with closing(SQLiteFile.writer(path)) as destination: prepared.backup(destination) path.chmod(0o600) - rewritten.append(str(path.relative_to(root))) + rewritten.append(path.relative_to(root)) if todo_changed: TodoStore.publish_change(todo_path) rewritten.extend( - str(path.relative_to(root)) for path in TodoStore.source_paths(root) + path.relative_to(root) for path in TodoStore.source_paths(root) ) if current: registry._write_unlocked(raw) - rewritten.extend(str(path.relative_to(root)) for path in registry.written) + rewritten.extend(path.relative_to(root) for path in registry.written) return Carried( new_rows + int(old_links) + int(history_changed) + len(current), detail={ diff --git a/tools/cutover/carry_turn_owner.py b/tools/cutover/carry_turn_owner.py index 568a557dee..8d8547aa94 100644 --- a/tools/cutover/carry_turn_owner.py +++ b/tools/cutover/carry_turn_owner.py @@ -36,7 +36,7 @@ def carry(self, *, apply: bool) -> Carried: registry = RegistryStore.at(self.root) if not registry.path.exists(): return Carried(0) - rewritten: list[str] = [] + rewritten: list[Path] = [] for thread in registry.read_snapshot_unlocked().threads.values(): if not thread.executing: continue @@ -56,5 +56,5 @@ def carry(self, *, apply: bool) -> Carried: FieldCodec.decode(ActiveTurn, carried) if apply: AtomicFile(path).write_text(json.dumps(carried)) - rewritten.append(str(path.relative_to(self.root))) + rewritten.append(path.relative_to(self.root)) return Carried(len(rewritten), rewritten=tuple(rewritten) if apply else ()) From 8b7166b7ca00549e74f60ec4029e87e4134c04cf Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:46:13 -0400 Subject: [PATCH 04/21] Keep backup manifests and carry fixture paths typed at their boundary --- src/agent_comms/installer/backup.py | 41 ++++++++++++++++----------- tests/test_carry.py | 10 +++---- tests/test_carry_drop_connectivity.py | 2 +- tests/test_install_announcements.py | 2 +- 4 files changed, 32 insertions(+), 23 deletions(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index 6699742b88..f397b317cc 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -16,12 +16,12 @@ from functools import cache from pathlib import Path from string import Formatter -from typing import ClassVar, Iterator, Self +from typing import Annotated, ClassVar, Iterator, Self from agent_comms.child_process import Platform from agent_comms.active_route import ActiveRoute, _route_directory, route_directory from agent_comms.coordination_errors import CoordinationError -from agent_comms.field_codec import FieldCodec +from agent_comms.field_codec import FieldCodec, PathText from agent_comms.installer.interpreter_operation import RouteSelection from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile, UndeclaredFile from agent_comms.private_path import FileObservation, FileRevision, MissingFile @@ -74,7 +74,8 @@ def bus_files(root: Path) -> dict[tuple[type[RootFile] | UndeclaredFile, Path], companions.update(member for path, owner in owners.items() for member in owner.member_paths(path) if member != path) return { - (declared.get(path, UndeclaredFile(path.relative_to(root))), path.relative_to(root)): { + (declared[path] if path in declared else UndeclaredFile(path.relative_to(root)), + path.relative_to(root)): { member.relative_to(root): SavedFile( observed[member].st_size, observed[member].st_mtime_ns, observed[member].st_ino, ) for member in owner.snapshot_members(path, observed) @@ -182,10 +183,14 @@ class StoreBackup(ResourceBackup): def spill_directory(self, configured: Path) -> Path: return self.path - def files(self) -> dict[str, str]: - return FieldCodec.decode(dict[str, str], + def files(self) -> dict[Annotated[Path, PathText], str]: + return FieldCodec.decode(dict[Annotated[Path, PathText], str], FieldCodec.parse((self.path / 'files.json').read_text())) + def selected_files(self) -> tuple[Annotated[Path, PathText], ...]: + return FieldCodec.decode(tuple[Annotated[Path, PathText], ...], + FieldCodec.parse((self.path / 'selected.json').read_text())) + def retire(self) -> Path: archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}') self.path.rename(archived) @@ -238,9 +243,9 @@ def route_original(self) -> bool: def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path) -> StoreBackup: path = spill_dir / BackupName.now(root).value backup = cls(root, route, path) - selected = tuple(str(file.relative_to(root)) for file in files) + selected = tuple(file.relative_to(root) for file in files) if path.exists(): - if tuple(json.loads((path / 'selected.json').read_text())) != selected: + if backup.selected_files() != selected: raise UpgradeRefused('Existing backup belongs to different resources') backup.files() # An interrupted snapshot is never treated as complete. with route_directory(route): @@ -249,13 +254,15 @@ def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path) backup.route_revision = FileRevision.capture(route) return backup path.mkdir(mode=0o700, parents=True) - AtomicFile(path / 'selected.json').write_text(json.dumps(selected)) + AtomicFile(path / 'selected.json').write_text(json.dumps(FieldCodec.encode( + selected, tuple[Annotated[Path, PathText], ...]))) contents = backup.capture(files) with route_directory(route): if route.exists(): shutil.copy2(route, path / 'active-route.json') backup.route_revision = FileRevision.capture(route) - AtomicFile(path / 'files.json').write_text(json.dumps(FieldCodec.encode(contents), indent=1)) + AtomicFile(path / 'files.json').write_text(json.dumps(FieldCodec.encode( + contents, dict[Annotated[Path, PathText], str]), indent=1)) return backup def retain(self, spill_dir: Path) -> tuple[Path, ...]: @@ -315,12 +322,12 @@ def transient(directory: str, names: list[str]) -> set[str]: source, target, symlinks=True, ignore=transient, copy_function=cls.copy_file ) - def capture(self, files: tuple[Path, ...]) -> dict[str, str]: - contents = {} + def capture(self, files: tuple[Path, ...]) -> dict[Annotated[Path, PathText], str]: + contents: dict[Path, str] = {} for source in files: if not source.exists(): continue - name = str(source.relative_to(self.root)) + name = source.relative_to(self.root) target = self.data_root / name target.parent.mkdir(parents=True, exist_ok=True) self.copy_file(source, target) @@ -329,7 +336,7 @@ def capture(self, files: tuple[Path, ...]) -> dict[str, str]: Platform.current().sync_directory(target.parent) return contents - def file_original(self, name: str, saved: dict[str, str]) -> bool: + def file_original(self, name: Path, saved: dict[Annotated[Path, PathText], str]) -> bool: """An unchanged selected file keeps its identity as well as its contents.""" target = self.root / name if name not in saved: @@ -344,14 +351,16 @@ def stores_original(self) -> bool: return False saved = self.files() return all(self.file_original(name, saved) - for name in json.loads((self.path / 'selected.json').read_text())) + for name in self.selected_files()) def restore(self) -> None: saved = self.files() - for name in json.loads((self.path / 'selected.json').read_text()): + for name in self.selected_files(): if self.file_original(name, saved): continue target = self.root / name + observed = self.data_root / name if name in saved else target + owner = RootFile.content_owner(observed) if observed.exists() else RootFile if name in saved: descriptor, temporary = tempfile.mkstemp(prefix='.restore-', dir=target.parent) os.close(descriptor) @@ -363,7 +372,7 @@ def restore(self) -> None: stage.unlink(missing_ok=True) else: target.unlink(missing_ok=True) - for companion in SQLiteFile.files(target)[1:]: + for companion in set(owner.files(target)) - {target}: companion.unlink(missing_ok=True) Platform.current().sync_directory(self.root) self.restore_route() diff --git a/tests/test_carry.py b/tests/test_carry.py index 08fb091cf1..5148d8813c 100644 --- a/tests/test_carry.py +++ b/tests/test_carry.py @@ -178,11 +178,11 @@ def noisy_step(tmp_path_factory): " (self.root / 'saved').write_text('migrated')\n" " if (self.root / 'fail').exists():\n" " raise RuntimeError('failed after changing the saved file')\n" - " return Carried(1, {'applied': apply}, ('saved',))\n" + " return Carried(1, {'applied': apply}, (Path('saved'),))\n" ) # The parent needs only the selected step's identity; its implementation is loaded # by the worker from the target tree. - class SelectedStep(Carry): + class Noisy(Carry): @classmethod def module(cls): return module @@ -193,7 +193,7 @@ def backup_files(self): def carry(self, *, apply): raise AssertionError("The selected interpreter executes the carry, never its caller") - return tree, SelectedStep + return tree, Noisy def test_step_answers_are_separate_from_diagnostics_and_keep_their_types( @@ -205,10 +205,10 @@ def test_step_answers_are_separate_from_diagnostics_and_keep_their_types( request = step(root, root, tmp_path / "spill") assert stack.call_step(request, tree, BackupFilesOperation()) == (root / "saved",) assert stack.call_step(request, tree, MigrateOperation(False)) == ( - Carried(1, {"applied": False}, ("saved",))) + Carried(1, {"applied": False}, (Path("saved"),))) assert not (root / "saved").exists() assert stack.call_step(request, tree, MigrateOperation(True)) == ( - Carried(1, {"applied": True}, ("saved",))) + Carried(1, {"applied": True}, (Path("saved"),))) assert (root / "saved").read_text() == "migrated" diff --git a/tests/test_carry_drop_connectivity.py b/tests/test_carry_drop_connectivity.py index 992ad09b8e..4e00638c50 100644 --- a/tests/test_carry_drop_connectivity.py +++ b/tests/test_carry_drop_connectivity.py @@ -21,7 +21,7 @@ def test_carry_drops_the_table_and_its_triggers_once(tmp_path): with maintenance_held(tmp_path): assert step.carry(apply=False).changed == 1 - assert step.carry(apply=True).rewritten == (CoordinationStore.filename,) + assert step.carry(apply=True).rewritten == (step.path.relative_to(step.root),) assert step.carry(apply=False).changed == 0 with sqlite3.connect(path) as db: assert not db.execute( diff --git a/tests/test_install_announcements.py b/tests/test_install_announcements.py index c07df78cae..224ff4831a 100644 --- a/tests/test_install_announcements.py +++ b/tests/test_install_announcements.py @@ -83,7 +83,7 @@ def test_failed_rehearsal_does_not_publish_to_the_source_bus(private_comms, tmp_ @pytest.mark.parametrize("live", ( CarriedRun( steps=(StepRun("carry_order_scopes", Carried(6)),), - files=RehearsalFiles(3, 1, ("todos.sqlite3",), 1024), + files=RehearsalFiles(3, 1, (Path("todos.sqlite3"),), 1024), read=BusReadBack(2), stopped=("worker",), services=(RESTARTED,), started=(StartedOwner("worker"), RefusedOwner("busy", "already running")), restore_space=RestoreSpace(1024, 4096, 0.01), From 54b2e14c7af0d8a02e6f41910667ca428571c8fe Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:47:16 -0400 Subject: [PATCH 05/21] Decode backup JSON paths once and derive copy exclusions from owners --- src/agent_comms/installer/backup.py | 32 ++++++++++++++--------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index f397b317cc..2ad8a7e15e 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -23,7 +23,7 @@ from agent_comms.coordination_errors import CoordinationError from agent_comms.field_codec import FieldCodec, PathText from agent_comms.installer.interpreter_operation import RouteSelection -from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile, UndeclaredFile +from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, UndeclaredFile from agent_comms.private_path import FileObservation, FileRevision, MissingFile from agent_comms.store_files import AtomicFile @@ -183,9 +183,10 @@ class StoreBackup(ResourceBackup): def spill_directory(self, configured: Path) -> Path: return self.path - def files(self) -> dict[Annotated[Path, PathText], str]: - return FieldCodec.decode(dict[Annotated[Path, PathText], str], - FieldCodec.parse((self.path / 'files.json').read_text())) + def files(self) -> dict[Path, str]: + stored = FieldCodec.decode(dict[str, str], + FieldCodec.parse((self.path / 'files.json').read_text())) + return {PathText.decode(path): digest for path, digest in stored.items()} def selected_files(self) -> tuple[Annotated[Path, PathText], ...]: return FieldCodec.decode(tuple[Annotated[Path, PathText], ...], @@ -261,8 +262,8 @@ def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path) if route.exists(): shutil.copy2(route, path / 'active-route.json') backup.route_revision = FileRevision.capture(route) - AtomicFile(path / 'files.json').write_text(json.dumps(FieldCodec.encode( - contents, dict[Annotated[Path, PathText], str]), indent=1)) + AtomicFile(path / 'files.json').write_text(json.dumps( + {PathText.encode(file): digest for file, digest in contents.items()}, indent=1)) return backup def retain(self, spill_dir: Path) -> tuple[Path, ...]: @@ -305,24 +306,23 @@ def copy_tree( """Copy committed files; their owners declare transient names before any stat.""" def transient(directory: str, names: list[str]) -> set[str]: - omitted: set[str] = set() - for name in sorted(names): - path = Path(directory) / name - if name in omitted or path in excluded or LockedStore.is_staging(path): - omitted.add(name) + omitted: set[Path] = set() + for path in sorted(Path(directory) / name for name in names): + if path in omitted or path in excluded or LockedStore.is_staging(path): + omitted.add(path) continue if stat.S_ISSOCK(path.lstat().st_mode): - omitted.add(name) + omitted.add(path) elif path.is_file() and not path.is_symlink(): owner = RootFile.content_owner(path) - omitted.update(companion.name for companion in owner.files(path)[1:]) - return omitted + omitted.update(owner.member_paths(path) - {path}) + return {path.name for path in omitted} shutil.copytree( source, target, symlinks=True, ignore=transient, copy_function=cls.copy_file ) - def capture(self, files: tuple[Path, ...]) -> dict[Annotated[Path, PathText], str]: + def capture(self, files: tuple[Path, ...]) -> dict[Path, str]: contents: dict[Path, str] = {} for source in files: if not source.exists(): @@ -336,7 +336,7 @@ def capture(self, files: tuple[Path, ...]) -> dict[Annotated[Path, PathText], st Platform.current().sync_directory(target.parent) return contents - def file_original(self, name: Path, saved: dict[Annotated[Path, PathText], str]) -> bool: + def file_original(self, name: Path, saved: dict[Path, str]) -> bool: """An unchanged selected file keeps its identity as well as its contents.""" target = self.root / name if name not in saved: From 1a6fc31c19050c04ee571e08e66d1696831c3216 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:50:40 -0400 Subject: [PATCH 06/21] Give each snapshot key one typed file identity --- src/agent_comms/installer/backup.py | 10 +++++----- src/agent_comms/installer/upgrade.py | 12 ++++++------ src/agent_comms/locked_store.py | 8 ++++++++ tests/test_install_rehearsal.py | 5 +++-- 4 files changed, 22 insertions(+), 13 deletions(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index 2ad8a7e15e..cb70ae0a5b 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -23,7 +23,7 @@ from agent_comms.coordination_errors import CoordinationError from agent_comms.field_codec import FieldCodec, PathText from agent_comms.installer.interpreter_operation import RouteSelection -from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, UndeclaredFile +from agent_comms.locked_store import DeclaredFile, LockedStore, RootFile, RuntimeFile, UndeclaredFile from agent_comms.private_path import FileObservation, FileRevision, MissingFile from agent_comms.store_files import AtomicFile @@ -55,8 +55,8 @@ def matches_extraction(self, copied: SavedFile) -> bool: return self.size == copied.size -def bus_files(root: Path) -> dict[tuple[type[RootFile] | UndeclaredFile, Path], dict[Path, SavedFile]]: - """Read saved file identity and modification metadata, never contents.""" +def bus_files(root: Path) -> dict[DeclaredFile | UndeclaredFile, dict[Path, SavedFile]]: + """Record owner identities and their content members' physical metadata.""" excluded = RuntimeFile.paths_in(root) observed: dict[Path, os.stat_result] = {} for path in root.rglob("*"): @@ -74,8 +74,8 @@ def bus_files(root: Path) -> dict[tuple[type[RootFile] | UndeclaredFile, Path], companions.update(member for path, owner in owners.items() for member in owner.member_paths(path) if member != path) return { - (declared[path] if path in declared else UndeclaredFile(path.relative_to(root)), - path.relative_to(root)): { + (DeclaredFile(declared[path], path.relative_to(root)) if path in declared + else UndeclaredFile(path.relative_to(root))): { member.relative_to(root): SavedFile( observed[member].st_size, observed[member].st_mtime_ns, observed[member].st_ino, ) for member in owner.snapshot_members(path, observed) diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index 9ae0326b2a..13ba3e2e89 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -104,7 +104,7 @@ from agent_comms.installer.reporting import RootWriter, RuntimeWriter, UnpostedWriter from agent_comms.installer.runtime_resolution import RuntimeResolution from agent_comms.launch_runtime import LaunchRuntime, RestartEnvironment, SpecifiedImportPath -from agent_comms.locked_store import BusFile, RootFile, UndeclaredFile +from agent_comms.locked_store import BusFile, DeclaredFile, RootFile, UndeclaredFile from agent_comms.maintenance_barrier import ( MaintenanceBarrier, MaintenanceOperator, @@ -157,16 +157,16 @@ def require_rehearsal(self, rehearsal: RehearsalFiles) -> None: def carried( root: Path, - before: dict[tuple[type[RootFile] | UndeclaredFile, Path], dict[Path, SavedFile]], + before: dict[DeclaredFile | UndeclaredFile, dict[Path, SavedFile]], steps: tuple[StepRun, ...], expected: frozenset[Path], ) -> RehearsalFiles: """Only the files a step names as rewritten may differ from ``before``.""" after = bus_files(root) physical = {key for key in before.keys() | after.keys() if before.get(key) != after.get(key)} - members = {member: primary for (_owner, primary), snapshot in (*before.items(), *after.items()) + members = {member: resource.path for resource, snapshot in (*before.items(), *after.items()) for member in snapshot} - changed = {primary for _owner, primary in physical} + changed = {resource.path for resource in physical} declared = {members.get(path, path) for path in RehearsalFiles.declared(steps)} unnamed = sorted(changed - declared) if unnamed: @@ -175,7 +175,7 @@ def carried( if outside: raise UpgradeRefused(f"Carries changed files outside the expected set: {outside}") return RehearsalFiles( - identical=len({primary for _owner, primary in before} - changed), rewritten=len(changed), + identical=len({resource.path for resource in before} - changed), rewritten=len(changed), names=tuple(sorted(changed)), restore_bytes=sum(file.size for key in physical & before.keys() for file in before[key].values())) @@ -848,7 +848,7 @@ class PreparedMigration(Extendable): """The held bus as the outgoing runtime read it, before any migration.""" hold: MaintenanceState - before: dict[tuple[type[RootFile] | UndeclaredFile, Path], dict[Path, SavedFile]] + before: dict[DeclaredFile | UndeclaredFile, dict[Path, SavedFile]] read: BusReadBack @property diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index 532459a160..9439b0eae3 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -244,6 +244,14 @@ def paths_in(cls, root: Path) -> frozenset[Path]: ) +@dataclass(frozen=True) +class DeclaredFile(RootFile): + """A declared store and its primary path form one physical resource identity.""" + + owner: type[RootFile] + path: Path + + @dataclass(frozen=True) class UndeclaredFile(RootFile): """A physical file without a stored declaration, still identified by its path.""" diff --git a/tests/test_install_rehearsal.py b/tests/test_install_rehearsal.py index c43f74ca07..d7952994b7 100644 --- a/tests/test_install_rehearsal.py +++ b/tests/test_install_rehearsal.py @@ -147,7 +147,8 @@ def test_failed_live_migration_restores_data_checkout_and_runtime( (comms.root / "fail-live").touch() current = subprocess.check_output(["git", "-C", checkout, "rev-parse", "HEAD"], text=True).strip() step = checkout / "tools/cutover/carry_fault.py" - step.write_text('''from agent_comms.installer.carry import Carry, Carried + step.write_text('''from pathlib import Path +from agent_comms.installer.carry import Carry, Carried class Fault(Carry): def backup_files(self): return (self.root / "saved",) @@ -156,7 +157,7 @@ def carry(self, *, apply): (self.root / "saved").write_text("changed by the worker") if (self.root / "fail-live").exists(): raise RuntimeError("intentional failure after writing the resource") - return Carried(1, rewritten=("saved",)) + return Carried(1, rewritten=(Path("saved"),)) ''') subprocess.run(["git", "-C", checkout, "add", step], check=True) subprocess.run(["git", "-C", checkout, "-c", "user.name=Installer test", From 27e8722f7992347f78edf828eb96201d9759ab4a Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:54:53 -0400 Subject: [PATCH 07/21] Discover unfinished store backups through their declared names and original state --- src/agent_comms/installer/backup.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index d16d2e3908..8b3aa99dec 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -19,7 +19,7 @@ from typing import ClassVar, Iterator, Self from agent_comms.child_process import Platform -from agent_comms.active_route import ActiveRoute, _route_directory, route_directory +from agent_comms.active_route import ActiveRoute, _route_directory, active_route_path, route_directory from agent_comms.coordination_errors import CoordinationError from agent_comms.field_codec import FieldCodec from agent_comms.installer.interpreter_operation import RouteSelection @@ -176,6 +176,19 @@ def files(self) -> dict[str, str]: return FieldCodec.decode(dict[str, str], FieldCodec.parse((self.path / 'files.json').read_text())) + @classmethod + def interrupted(cls, root: Path, spill_dir: Path) -> tuple[StoreBackup, ...]: + if not spill_dir.exists(): + return () + route = active_route_path() + backups = ( + cls(root, route, spill_dir / name.value) + for name in sorted(BackupName.matching( + root, (path.name for path in spill_dir.iterdir() if path.is_dir()) + ), reverse=True) + ) + return tuple(backup for backup in backups if not backup.original()) + def retire(self) -> Path: archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}') self.path.rename(archived) From 7b243695e1382d35961b522c8f9baffa8bb6465e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:58:02 -0400 Subject: [PATCH 08/21] Let RootFile own primary selection for both inventories and backups --- src/agent_comms/installer/backup.py | 18 +++++----------- src/agent_comms/locked_store.py | 33 ++++++++++++++++++++++++++--- 2 files changed, 35 insertions(+), 16 deletions(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index cb70ae0a5b..c7c354c1b1 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -65,21 +65,12 @@ def bus_files(root: Path) -> dict[DeclaredFile | UndeclaredFile, dict[Path, Save info = path.lstat() if stat.S_ISREG(info.st_mode): observed[path] = info - declared = RootFile.existing_in(root) - owners = {path: owner for path, owner in declared.items() if path in observed} - companions = {member for path, owner in owners.items() - for member in owner.member_paths(path) if member != path} - owners.update((path, RootFile.content_owner(path)) - for path in observed.keys() - owners.keys() - companions) - companions.update(member for path, owner in owners.items() - for member in owner.member_paths(path) if member != path) return { - (DeclaredFile(declared[path], path.relative_to(root)) if path in declared - else UndeclaredFile(path.relative_to(root))): { + resource: { member.relative_to(root): SavedFile( observed[member].st_size, observed[member].st_mtime_ns, observed[member].st_ino, - ) for member in owner.snapshot_members(path, observed) - } for path, owner in owners.items() if path not in companions + ) for member in owner.snapshot_members(root / resource.path, observed) + } for resource, owner in RootFile.owned_files(root, observed).items() } @@ -244,7 +235,8 @@ def route_original(self) -> bool: def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path) -> StoreBackup: path = spill_dir / BackupName.now(root).value backup = cls(root, route, path) - selected = tuple(file.relative_to(root) for file in files) + selected = tuple(sorted(resource.path for resource in RootFile.owned_files(root, files))) + files = tuple(root / path for path in selected) if path.exists(): if backup.selected_files() != selected: raise UpgradeRefused('Existing backup belongs to different resources') diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index 9439b0eae3..1995b5dd3b 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -10,7 +10,7 @@ import stat from abc import ABC, abstractmethod from collections import defaultdict -from collections.abc import Callable, Iterator +from collections.abc import Callable, Collection, Iterator, Mapping from contextlib import closing, contextmanager from dataclasses import dataclass from functools import cache @@ -164,7 +164,7 @@ def member_paths(cls, path: Path) -> frozenset[Path]: @classmethod def snapshot_members( - cls, path: Path, observed: dict[Path, os.stat_result], + cls, path: Path, observed: Mapping[Path, os.stat_result], ) -> frozenset[Path]: """Content changes count; lock and coordination files are not content.""" return frozenset((path,)) @@ -236,6 +236,33 @@ def existing_in(cls, root: Path) -> dict[Path, type[Self]]: return {path: owner for owner, directory in cls.declarations_in(root) for path in owner.matching_in(directory)} + @classmethod + def owned_files( + cls, root: Path, paths: Collection[Path], + ) -> dict[DeclaredFile | UndeclaredFile, type[RootFile]]: + """Resolve primaries once; companions belong to their primary's declaration. + + Missing selected files still identify an absent resource to restore. An + undeclared existing file acquires content behavior without inventing a + store declaration for it. + """ + declarations = cls.existing_in(root) + selected = frozenset(paths) + owners: dict[Path, type[RootFile]] = { + path: owner for path, owner in declarations.items() if path in selected + } + companions = {member for path, owner in owners.items() + for member in owner.member_paths(path) if member != path} + owners.update((path, cls.content_owner(path) if path.exists() else RootFile) + for path in selected - owners.keys() - companions) + companions.update(member for path, owner in owners.items() + for member in owner.member_paths(path) if member != path) + return { + (DeclaredFile(declarations[path], path.relative_to(root)) if path in declarations + else UndeclaredFile(path.relative_to(root))): owner + for path, owner in owners.items() if path not in companions + } + @classmethod def paths_in(cls, root: Path) -> frozenset[Path]: return frozenset( @@ -529,7 +556,7 @@ def wal_path(cls, path: Path) -> Path: @classmethod def snapshot_members( - cls, path: Path, observed: dict[Path, os.stat_result], + cls, path: Path, observed: Mapping[Path, os.stat_result], ) -> frozenset[Path]: wal = cls.wal_path(path) # A read can create an empty WAL; only a nonempty WAL holds content. From d7a7cdc6e7961bab086d5a1699bea538c10caaf8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 04:58:59 -0400 Subject: [PATCH 09/21] Restore the registry snapshot and its declared guard after unrelated files --- src/agent_comms/installer/backup.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index 8b3aa99dec..4379455b4b 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -25,6 +25,7 @@ from agent_comms.installer.interpreter_operation import RouteSelection from agent_comms.locked_store import LockedStore, RootFile, RuntimeFile, SQLiteFile from agent_comms.private_path import FileObservation, FileRevision, MissingFile +from agent_comms.registry_store import RegistryStore from agent_comms.store_files import AtomicFile @@ -351,7 +352,15 @@ def stores_original(self) -> bool: def restore(self) -> None: saved = self.files() - for name in json.loads((self.path / 'selected.json').read_text()): + selected = json.loads((self.path / 'selected.json').read_text()) + registry = tuple(str(path.relative_to(self.root)) for path in RegistryStore.at(self.root).written) + # The registry owns this ordering: restore its snapshot before publishing + # the matching guard, after every unrelated selected file. + ordered = ( + *(name for name in selected if name not in registry), + *(name for name in registry if name in selected), + ) + for name in ordered: if self.file_original(name, saved): continue target = self.root / name From 4cad001cc755ea92bb947da5b18fdd790f5d92b9 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:00:19 -0400 Subject: [PATCH 10/21] Give each bus root one recorded runtime owner Ported from #2527 onto current main. RootRuntime records the runtime prefix that owns a root; Comms refuses a foreign runtime unless the maintenance hold targets it. MaintenanceTarget is a declared family on the hold (no target, or TargetRuntime); a phase that ends maintenance drops the target, and every rewrite of a hold goes through MaintenanceState.advanced. retarget takes only the registry lock, like begin and clear. RuntimeProcesses reports old-runtime applications without signalling them. The installer wiring follows on this branch after the batch carrying #2698 lands, because it rewrites the same upgrade.py sites. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/comms.py | 3 + src/agent_comms/launch_runtime.py | 41 +++++---- src/agent_comms/maintenance_barrier.py | 78 ++++++++++++++--- src/agent_comms/root_runtime.py | 65 ++++++++++++++ src/agent_comms/runtime_processes.py | 112 +++++++++++++++++++++++++ src/agent_comms/store_order.py | 3 +- tests/test_root_runtime.py | 67 +++++++++++++++ tests/test_runtime_processes.py | 70 ++++++++++++++++ 8 files changed, 410 insertions(+), 29 deletions(-) create mode 100644 src/agent_comms/root_runtime.py create mode 100644 src/agent_comms/runtime_processes.py create mode 100644 tests/test_root_runtime.py create mode 100644 tests/test_runtime_processes.py diff --git a/src/agent_comms/comms.py b/src/agent_comms/comms.py index d0aaf0d1b2..ca82715d4e 100644 --- a/src/agent_comms/comms.py +++ b/src/agent_comms/comms.py @@ -14,6 +14,7 @@ from .context_store import ContextStore from .goal_management import Goals from .history_views import HistoryViews +from .launch_runtime import LaunchRuntime from .locked_store import RegistryFile from .maintenance_barrier import MaintenanceBarrier from .message_bus import MessageBus @@ -22,6 +23,7 @@ from .publisher import OwnerPublisher, Publisher from .registration import Registration from .relationships import ThreadRelationships +from .root_runtime import RootRuntime from .thread_configuration import ThreadConfiguration from .thread_management import ThreadManagement from .transcripts import Transcripts @@ -32,6 +34,7 @@ def __init__( self, root: CommsRoute | Path, publisher: type[Publisher] = OwnerPublisher, ) -> None: self.route = resolve_comms_route(root) + RootRuntime.require_attachable(self.route.root, LaunchRuntime.running()) self.route.observe_root() self.registry = Registration(RegistryFile.path_in(self.root)) self.bus = MessageBus( diff --git a/src/agent_comms/launch_runtime.py b/src/agent_comms/launch_runtime.py index dc39d0c93e..5280e6448e 100644 --- a/src/agent_comms/launch_runtime.py +++ b/src/agent_comms/launch_runtime.py @@ -26,7 +26,15 @@ WireOmitDefault, ) from agent_comms.launch_root import LaunchRoot -from agent_comms.thread_text import AbsentText, GivenText, NoParent, NoThreadText, TextChoice, ThreadParent, ThreadText +from agent_comms.thread_text import ( + AbsentText, + GivenText, + NoParent, + NoThreadText, + TextChoice, + ThreadParent, + ThreadText, +) if TYPE_CHECKING: from agent_comms.threads import Thread @@ -62,10 +70,15 @@ class LaunchRuntime: interpreter: Annotated[Path, PathText] + @classmethod + def from_interpreter(cls, interpreter: str | Path) -> Self: + """Keep the invoked interpreter: resolving its symlink loses the runtime.""" + return cls(Path(interpreter)) + @classmethod def running(cls) -> LaunchRuntime: """Keep ordinary launches on the calling process's installation.""" - return cls(Path(sys.executable)) + return cls.from_interpreter(sys.executable) def child_environment(self, environment: Mapping[str, str]) -> dict[str, str]: """A child of this process's own interpreter imports the source this process @@ -239,9 +252,7 @@ class RestartEnvironment(EnvironmentRecord): stay in /proc and are read only from the exact selected process at execution. """ - home: Annotated[ - EnvironmentValue, WireOmitDefault(), WireName("HOME") - ] = UnsetVariable() + home: Annotated[EnvironmentValue, WireOmitDefault(), WireName("HOME")] = UnsetVariable() path: Annotated[ EnvironmentValue, WireOmitDefault(), @@ -260,9 +271,9 @@ class RestartEnvironment(EnvironmentRecord): WireName("VIRTUAL_ENV"), RuntimeScope(), ] = UnsetVariable() - config: Annotated[ - EnvironmentValue, WireOmitDefault(), WireName("XDG_CONFIG_HOME") - ] = UnsetVariable() + config: Annotated[EnvironmentValue, WireOmitDefault(), WireName("XDG_CONFIG_HOME")] = ( + UnsetVariable() + ) default_agent_directory: ClassVar[Path] = Path("~/.pi/agent") native_config: Annotated[ Path, @@ -270,9 +281,9 @@ class RestartEnvironment(EnvironmentRecord): WireName("AGENT_COMMS_NATIVE_CONFIG_DIR"), NativeScope(), ] = default_agent_directory - agent_directory: Annotated[ - Path, PathText, WireName("PI_CODING_AGENT_DIR"), NativeScope() - ] = default_agent_directory + agent_directory: Annotated[Path, PathText, WireName("PI_CODING_AGENT_DIR"), NativeScope()] = ( + default_agent_directory + ) def __post_init__(self) -> None: object.__setattr__(self, "native_config", self.absolute(self.native_config)) @@ -384,12 +395,8 @@ class ThreadLaunch(EnvironmentRecord): WireOmitDefault(), WireName("AGENT_COMMS_TAGS"), ] = frozenset() - parent: Annotated[ - ThreadParent, WireOmitDefault(), WireName("PI_PARENT_ID") - ] = NoParent() - task: Annotated[ - ThreadText, WireOmitDefault(), WireName("PI_TASK") - ] = NoThreadText() + parent: Annotated[ThreadParent, WireOmitDefault(), WireName("PI_PARENT_ID")] = NoParent() + task: Annotated[ThreadText, WireOmitDefault(), WireName("PI_TASK")] = NoThreadText() managed: Annotated[ bool, FlagText, diff --git a/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py index cf7ed02035..1e27421a74 100644 --- a/src/agent_comms/maintenance_barrier.py +++ b/src/agent_comms/maintenance_barrier.py @@ -14,14 +14,23 @@ from abc import ABC, abstractmethod from collections.abc import AsyncIterator, Callable, Iterator from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager -from dataclasses import dataclass, replace +from dataclasses import dataclass, field, replace from pathlib import Path -from typing import Literal, Self +from typing import Annotated, Literal, Self from .child_process import Platform, ProcessIdentity from .declared_family import DeclaredFamily from .errors import RelationViolationError, ToolRefusedError -from .field_codec import FieldCodec, Id128Text, MalformedRecord +from .field_codec import ( + FieldCodec, + Id128Text, + MalformedRecord, + NullMember, + RecordMember, + SubclassUnion, + WireOmitDefault, +) +from .launch_runtime import LaunchRuntime from .lifecycle import LifecycleState from .locked_store import DurableFile, LockedStore, RegistryFile from .send_outcome import GrantedSend, HeldSend @@ -79,9 +88,32 @@ def require_open(): pass +class MaintenanceTarget(SubclassUnion): + """A hold without a target grants no additional runtime attachment.""" + + @abstractmethod + def allows_runtime(self, runtime: LaunchRuntime) -> bool: ... + + +@dataclass(frozen=True) +class NoMaintenanceTarget(NullMember, MaintenanceTarget): + def allows_runtime(self, runtime: LaunchRuntime) -> bool: + return False + + +@dataclass(frozen=True) +class TargetRuntime(LaunchRuntime, RecordMember, MaintenanceTarget): + def allows_runtime(self, runtime: LaunchRuntime) -> bool: + return self.prefix == runtime.prefix + + class MaintenancePhase(DeclaredFamily, LifecycleState, affix="Phase"): new_sends = running_sends = HeldSends + @classmethod + def target_after(cls, target: MaintenanceTarget) -> MaintenanceTarget: + return target + @classmethod def holder_dead(cls, state: MaintenanceState) -> bool: return not state.operator.alive() @@ -147,6 +179,10 @@ class ReadyPhase(MaintenancePhase): def require_held_by(cls, state: MaintenanceState, holder: str, change: str) -> None: raise MaintenanceNotHeldError.unheld(change) + @classmethod + def target_after(cls, target: MaintenanceTarget) -> MaintenanceTarget: + return NoMaintenanceTarget() + @classmethod def holder_dead(cls, state: MaintenanceState) -> bool: return False @@ -187,6 +223,9 @@ class MaintenanceStatus: def turn_pending(self) -> bool: return self.phase.new_sends.pending() + def allows_runtime(self, runtime: LaunchRuntime) -> bool: + return False + def observed(self, barrier: MaintenanceBarrier) -> MaintenanceStatus: return self @@ -234,6 +273,12 @@ class MaintenanceState(MaintenanceMarker, MaintenanceStatus): operator: MaintenanceOperator nonce: str phase: type[MaintenancePhase] + target: Annotated[MaintenanceTarget, WireOmitDefault()] = field( + default_factory=NoMaintenanceTarget + ) + + def allows_runtime(self, runtime: LaunchRuntime) -> bool: + return self.target.allows_runtime(runtime) def __post_init__(self): super().__post_init__() @@ -277,9 +322,15 @@ def holding(self) -> Iterator[Self]: def successor(self, phase: type[MaintenancePhase], nonce: str) -> MaintenanceState: if not self.phase().may_become(phase()): raise ValueError('Maintenance transition requires a valid expected successor') - return MaintenanceState(version=self.version, root=self.root, - generation=self.generation + 1, operator=self.operator, - nonce=nonce, phase=phase) + return self.in_phase(phase, nonce) + + def in_phase(self, phase: type[MaintenancePhase], nonce: str) -> MaintenanceState: + """Ordinary and dead-holder transitions share the phase's target lifetime.""" + return self.advanced(nonce, phase=phase, target=phase.target_after(self.target)) + + def advanced(self, nonce: str, **changes: object) -> MaintenanceState: + """Every rewrite of a hold takes the next generation and a fresh nonce.""" + return replace(self, generation=self.generation + 1, nonce=nonce, **changes) def require_open_to(self, holder: str) -> None: if holder != self.nonce: @@ -372,10 +423,7 @@ def _release_dead_unlocked(self, expected: MaintenanceState) -> MaintenanceStatu current = self._current_unlocked() if current != expected: return current - self._write_unlocked(replace( - expected, generation=expected.generation + 1, nonce=Id128Text.mint(), - phase=DeadHolderReleasedPhase, - )) + self._write_unlocked(expected.in_phase(DeadHolderReleasedPhase, Id128Text.mint())) raise MaintenanceHeldError( f"Maintenance released dead holder pid={expected.operator.pid} " f"start_time={expected.operator.start_time}; ingress reopened" @@ -420,6 +468,14 @@ def begin(self, operator: MaintenanceOperator) -> MaintenanceState: self._write_unlocked(hold) return hold + def retarget(self, expected: MaintenanceState, runtime: LaunchRuntime) -> MaintenanceState: + """Transfer a drained hold after its outgoing readers have stopped.""" + with _store_lock(self.registry_path): + self.require_draining_unlocked(expected) + hold = expected.advanced(Id128Text.mint(), target=TargetRuntime(runtime.interpreter)) + self._write_unlocked(hold) + return hold + def _advance_unlocked(self, expected: MaintenanceState, phase: type[MaintenancePhase]) -> MaintenanceState: if self.current_unlocked() != expected: @@ -443,7 +499,7 @@ def require_draining_unlocked(self, expected: MaintenanceState) -> None: expected.require_draining() if self.current_unlocked() != expected: raise RelationViolationError( - 'Cutover maintenance phase changed before owner retirement' + 'Maintenance generation/operator changed' ) @contextmanager diff --git a/src/agent_comms/root_runtime.py b/src/agent_comms/root_runtime.py new file mode 100644 index 0000000000..7c25157e44 --- /dev/null +++ b/src/agent_comms/root_runtime.py @@ -0,0 +1,65 @@ +"""A bus root belongs to exactly one published runtime.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Annotated + +from .errors import RelationViolationError +from .field_codec import PathText +from .launch_runtime import LaunchRuntime +from .locked_store import DurableFile, LockedStore, RegistryFile +from .maintenance_barrier import MaintenanceBarrier + + +class ForeignRuntimeError(RelationViolationError): + """The caller is neither this root's owner nor its maintenance target.""" + + +@dataclass(frozen=True) +class RootRuntime: + prefix: Annotated[Path, PathText] + + @classmethod + def require_attachable(cls, root: Path, caller: LaunchRuntime) -> None: + store = RootRuntimeStore.at(root) + owner = store.owner(caller) + if owner.prefix == caller.prefix: + return + if MaintenanceBarrier(RegistryFile.path_in(root)).read().allows_runtime(caller): + return + raise ForeignRuntimeError( + f"Root {root} belongs to runtime {owner.prefix}; " + f"foreign runtime {caller.prefix} cannot attach" + ) + + @classmethod + def publish(cls, root: Path, runtime: LaunchRuntime) -> None: + RootRuntimeStore.at(root).replace(cls(runtime.prefix)) + + +@dataclass(frozen=True) +class RootRuntimeStore(LockedStore[RootRuntime], DurableFile): + filename = ".root-runtime.json" + record_type = RootRuntime + + def __post_init__(self) -> None: + # This is the first file on a new root, before publisher preparation. + self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + + def empty(self) -> RootRuntime: + raise FileNotFoundError(self.path) + + def owner(self, creator: LaunchRuntime) -> RootRuntime: + try: + return self.read() + except FileNotFoundError: + return self.adopt(creator) + + def adopt(self, runtime: LaunchRuntime) -> RootRuntime: + """A root without a runtime record belongs to ``runtime``; a recorded owner stays.""" + with self.locked(): + if not self.path.exists(): + self._publish_unlocked(RootRuntime(runtime.prefix)) + return self._read_unlocked() diff --git a/src/agent_comms/runtime_processes.py b/src/agent_comms/runtime_processes.py new file mode 100644 index 0000000000..85f282959a --- /dev/null +++ b/src/agent_comms/runtime_processes.py @@ -0,0 +1,112 @@ +"""Report old-runtime applications without acquiring authority to stop them.""" + +from __future__ import annotations + +import os +from abc import ABC, abstractmethod +from dataclasses import dataclass +from pathlib import Path + +from agent_comms.child_process import IdentityMismatchError, Platform, ProcessIdentity +from agent_comms.declared_family import DeclaredFamily +from agent_comms.launch_runtime import LaunchRuntime +from agent_comms.runtime_children import ApplicationChildInitiator + + +class RuntimeProcessOutcome(DeclaredFamily, ABC, affix="RuntimeProcess"): + @abstractmethod + def line(self) -> str: ... + + +@dataclass(frozen=True) +class UnreadableRuntimeProcess(RuntimeProcessOutcome): + pid: int + reason: str + + def line(self) -> str: + return f"Unknown runtime PID {self.pid}: {self.reason}" + + +@dataclass(frozen=True) +class RuntimeApplication(ApplicationChildInitiator): + """The observed parent application's exact identity and display name.""" + + name: str + + def describe(self) -> str: + return f"{self.name} (application PID {self.process.pid})" + + +@dataclass(frozen=True, eq=False) +class RuntimeProcess(ProcessIdentity, LaunchRuntime, declared_name=None): + application: RuntimeApplication + + +@dataclass(frozen=True) +class UnretiredRuntimeProcess(RuntimeProcessOutcome): + process: RuntimeProcess + + def line(self) -> str: + return ( + f"Unretired old-runtime PID {self.process.pid}: {self.process.prefix}; " + f"owner {self.process.application.describe()}; not signalled" + ) + + +@dataclass(frozen=True) +class RuntimeProcesses: + """A read-only snapshot, not a process handle or permission to signal an app. + + Installer-launched children retain their own ChildProcess stop/reap authority. + Discovering another application's interpreter grants no such authority. + """ + + outcomes: tuple[RuntimeProcessOutcome, ...] + + @classmethod + def observe(cls, target: LaunchRuntime, proc: Path = Path("/proc")) -> RuntimeProcesses: + directory = target.prefix.parent + outcomes: list[RuntimeProcessOutcome] = [] + platform = Platform.current() + for entry in proc.iterdir(): + if not entry.name.isdecimal(): + continue + pid = int(entry.name) + try: + identity = ProcessIdentity.capture(pid) + invoked = (entry / "cmdline").read_bytes().partition(b"\0")[0] + identity.require_alive() + except ProcessLookupError, FileNotFoundError: + continue + except (OSError, IdentityMismatchError) as error: + outcomes.append(UnreadableRuntimeProcess(pid, str(error))) + continue + # Empty command lines declare no userspace invocation (kernel/exited + # tasks). Other namespaces do not name an agent-comms installation. + interpreter = Path(os.fsdecode(invoked)) + if not invoked or not interpreter.is_relative_to(directory): + continue + runtime = LaunchRuntime.from_interpreter(interpreter) + if runtime.prefix == target.prefix: + continue + try: + parent = platform.parent_identity(pid) + application = RuntimeApplication( + parent, (proc / str(parent.pid) / "comm").read_text().strip() + ) + parent.require_alive() + identity.require_alive() + except (OSError, IdentityMismatchError) as error: + outcomes.append(UnreadableRuntimeProcess(pid, str(error))) + continue + outcomes.append( + UnretiredRuntimeProcess( + RuntimeProcess( + pid=identity.pid, + start_time=identity.start_time, + interpreter=runtime.interpreter, + application=application, + ) + ) + ) + return cls(tuple(outcomes)) diff --git a/src/agent_comms/store_order.py b/src/agent_comms/store_order.py index 19c3bce3e2..5c0f8f9b82 100644 --- a/src/agent_comms/store_order.py +++ b/src/agent_comms/store_order.py @@ -5,6 +5,7 @@ from .relationships import RelationshipStore from .runtime_info import RuntimeInfoStore from .runtime_children import RuntimeChildrenStore +from .root_runtime import RootRuntimeStore from .shared_ledger import SharedLedger from .collaboration_ledger import CollaborationLedger from .connector_store import ConnectorStore @@ -14,7 +15,7 @@ STORE_LOCK_ORDER = ( RegistryStore, ChannelCatalog, ReadLedger, RelationshipStore, RuntimeInfoStore, SharedLedger, CollaborationLedger, ConnectorStore, - RuntimeChildrenStore, ServicePassStore, + RuntimeChildrenStore, RootRuntimeStore, ServicePassStore, ) StoreLockOwner.declare_lock_order(STORE_LOCK_ORDER) diff --git a/tests/test_root_runtime.py b/tests/test_root_runtime.py new file mode 100644 index 0000000000..374260ad22 --- /dev/null +++ b/tests/test_root_runtime.py @@ -0,0 +1,67 @@ +"""Root attachment and publication use one persisted runtime owner.""" + +import pytest + +from agent_comms.comms import Comms +from agent_comms.errors import RelationViolationError +from agent_comms.launch_runtime import LaunchRuntime +from agent_comms.locked_store import RegistryFile +from agent_comms.maintenance_barrier import ( + MaintenanceBarrier, + MaintenanceOperator, + NoMaintenanceTarget, +) +from agent_comms.root_runtime import ForeignRuntimeError, RootRuntime, RootRuntimeStore +from agent_comms.bus import Bus + + +def test_root_runtime_attachment_family(tmp_path): + owner, target, foreign = ( + LaunchRuntime(tmp_path / name / "bin/python") + for name in ("owner", "target", "foreign") + ) + root = tmp_path / "bus" + RootRuntime.require_attachable(root, owner) + store = RootRuntimeStore.at(root) + assert store.read() == RootRuntime(owner.prefix) + RootRuntime.require_attachable(root, owner) + with pytest.raises(ForeignRuntimeError) as refused: + RootRuntime.require_attachable(root, foreign) + assert str(owner.prefix) in str(refused.value) + assert str(foreign.prefix) in str(refused.value) + + barrier = MaintenanceBarrier(RegistryFile.path_in(root)) + original = barrier.begin(MaintenanceOperator.current()) + assert not original.allows_runtime(target) + with pytest.raises(ForeignRuntimeError): + RootRuntime.require_attachable(root, target) + held = barrier.retarget(original, target) + assert held.generation == original.generation + 1 + with pytest.raises(RelationViolationError): + barrier.retarget(original, foreign) + assert not NoMaintenanceTarget().allows_runtime(foreign) + RootRuntime.require_attachable(root, target) + RootRuntime.require_attachable(root, owner) + with pytest.raises(ForeignRuntimeError): + RootRuntime.require_attachable(root, foreign) + cleared = barrier.clear(held) + assert cleared.target == NoMaintenanceTarget() + with pytest.raises(ForeignRuntimeError): + RootRuntime.require_attachable(root, target) + RootRuntime.publish(root, target) + RootRuntime.require_attachable(root, target) + with pytest.raises(ForeignRuntimeError): + RootRuntime.require_attachable(root, owner) + + +def test_comms_creates_a_private_root_and_refuses_foreign_stores(tmp_path): + created = Comms(tmp_path / "created") + assert RootRuntimeStore.at(created.root).read().prefix == LaunchRuntime.running().prefix + + root = tmp_path / "bus" + other = LaunchRuntime(tmp_path / "other/bin/python") + RootRuntime.publish(root, other) + with pytest.raises(ForeignRuntimeError): + Comms(root) + assert not RegistryFile.path_in(root).exists() + assert not Bus.path_in(root).exists() diff --git a/tests/test_runtime_processes.py b/tests/test_runtime_processes.py new file mode 100644 index 0000000000..36e20e0c07 --- /dev/null +++ b/tests/test_runtime_processes.py @@ -0,0 +1,70 @@ +"""A fake runtimes tree with real applications; observations never signal them.""" + +import os +import subprocess +import sys +from contextlib import ExitStack +from pathlib import Path + +from agent_comms.child_process import ProcessIdentity +from agent_comms.field_codec import FieldCodec +from agent_comms.installer.upgrade import LiveRun, UnchangedRun +from agent_comms.launch_runtime import LaunchRuntime +from agent_comms.runtime_processes import ( + RuntimeProcessOutcome, + RuntimeProcesses, + UnreadableRuntimeProcess, + UnretiredRuntimeProcess, +) + + +def stop(child): + if child.poll() is None: + child.terminate() + child.wait(timeout=10) + + +def test_runtime_process_family(tmp_path): + runtimes = tmp_path / "runtimes" + old, target = (LaunchRuntime(runtimes / name / "bin/python") for name in ("old", "target")) + for runtime in (old, target): + runtime.interpreter.parent.mkdir(parents=True) + runtime.interpreter.symlink_to(sys.executable) + with ExitStack() as cleanup: + children = [] + for runtime in (old, target): + child = subprocess.Popen([str(runtime.interpreter), "-c", "import time; time.sleep(120)"]) + cleanup.callback(stop, child) + children.append(child) + snapshot = RuntimeProcesses.observe(target) + (outcome,) = snapshot.outcomes + assert isinstance(outcome, UnretiredRuntimeProcess) + assert outcome.process == ProcessIdentity.capture(children[0].pid) + assert outcome.process.prefix == old.prefix + assert outcome.process.application.process == ProcessIdentity.capture(os.getpid()) + assert outcome.process.application.name == Path(f"/proc/{os.getpid()}/comm").read_text().strip() + # Resolving the interpreter would collapse both runtimes onto the shared Python. + assert old.interpreter.resolve() == target.interpreter.resolve() + assert str(children[0].pid) in outcome.line() + assert outcome.process.application.describe() in outcome.line() + # Neither discovery nor rendering gains the right to terminate an application. + assert all(child.poll() is None for child in children) + run = UnchangedRun(processes=snapshot.outcomes) + decoded = FieldCodec.decode(LiveRun, FieldCodec.encode(run)) + assert decoded == run + assert outcome.line() in decoded.lines() + assert FieldCodec.decode(RuntimeProcessOutcome, FieldCodec.encode(outcome)) == outcome + stop(children[0]) # Only its actual launcher closes this child. + assert not RuntimeProcesses.observe(target).outcomes + assert children[1].poll() is None + + # A named process whose command line cannot be read is not silently discarded. + proc = tmp_path / "proc" + entry = proc / str(os.getpid()) + (entry / "cmdline").mkdir(parents=True) + (unknown,) = RuntimeProcesses.observe(target, proc).outcomes + assert isinstance(unknown, UnreadableRuntimeProcess) + assert unknown.pid == os.getpid() + assert str(unknown.pid) in unknown.line() + assert unknown.reason + assert FieldCodec.decode(RuntimeProcessOutcome, FieldCodec.encode(unknown)) == unknown From e056098e94e38064fe27311ff604f2285c47f43d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:01:00 -0400 Subject: [PATCH 11/21] Keep abandoned maintenance held and cover killed registry restoration --- src/agent_comms/maintenance_barrier.py | 7 ++- tests/test_interrupted_rollback.py | 84 ++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 2 deletions(-) create mode 100644 tests/test_interrupted_rollback.py diff --git a/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py index cf7ed02035..2ca10c4de6 100644 --- a/src/agent_comms/maintenance_barrier.py +++ b/src/agent_comms/maintenance_barrier.py @@ -412,8 +412,11 @@ def _write_unlocked(self, hold: MaintenanceState) -> None: def begin(self, operator: MaintenanceOperator) -> MaintenanceState: """Hold new turns before fleet drain; an uncertain write stays closed.""" with _store_lock(self.registry_path): - current = self.current_unlocked() - current.require_open() + current = self._current_unlocked() + # A successor takes an abandoned hold without briefly reopening + # ingress or making the owner repeat its install command. + if not current.phase.holder_dead(current): + current.require_open() hold = MaintenanceState(version=2, root=str(self.registry_path.parent.resolve()), generation=current.generation + 1, operator=operator, nonce=Id128Text.mint(), phase=DrainingPhase) diff --git a/tests/test_interrupted_rollback.py b/tests/test_interrupted_rollback.py new file mode 100644 index 0000000000..565a4e587a --- /dev/null +++ b/tests/test_interrupted_rollback.py @@ -0,0 +1,84 @@ +"""A killed restore must be completed by the next install, before it reads the bus.""" + +import os +import signal +import subprocess +import sys +from dataclasses import replace +from pathlib import Path + +import pytest + +from agent_comms.comms import Comms +from agent_comms.install_choices import RehearsalMode, UnnamedInstaller +from agent_comms.installer.backup import StoreBackup, UpgradeRefused +from agent_comms.installer.carry import Carried, Carry +from agent_comms.installer.launch import InstallReport +from agent_comms.installer.upgrade import upgrade +from agent_comms.registry_store import RegistryStore +from bus_record_fixture import prepare_native_comms + + +class RegistryTitleCarry(Carry): + def carry(self, *, apply): + registry = RegistryStore.at(self.root) + if apply: + with registry.editing() as edit: + thread = edit.document.require("worker") + edit.document.threads[thread.name] = replace(thread, title="Carried title") + return Carried(1, rewritten=tuple(str(path.relative_to(self.root)) for path in registry.written)) + + +KILLED_RESTORE = """ +import os, signal, sys +from pathlib import Path +from agent_comms.installer.backup import StoreBackup +from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator +from agent_comms.registry_store import RegistryStore + +root, route, path = map(Path, sys.argv[1:4]) +registry = RegistryStore.at(root) +publication = registry.written[int(sys.argv[4])] +original_replace = os.replace +def killed_publication(source, target, **directories): + original_replace(source, target, **directories) + if Path(target).name == publication.name: + os.kill(os.getpid(), signal.SIGKILL) +os.replace = killed_publication +hold = MaintenanceBarrier(registry.path).begin(MaintenanceOperator.current()) +with hold.holding(): + StoreBackup(root, route, path).restore() +raise AssertionError("The selected registry publication was never restored") +""" + + +@pytest.mark.skipif(os.name != "posix", reason="SIGKILL and private registry require POSIX") +@pytest.mark.parametrize("publication", (0, 1), ids=("after-registry", "after-guard")) +def test_next_install_completes_a_killed_registry_restore(tmp_path, monkeypatch, publication): + # HOME selects only this test's route. Neither process can replace the developer's route. + monkeypatch.setenv("HOME", str(tmp_path / "home")) + comms = prepare_native_comms(Comms(tmp_path / "wire")) + comms.registry.declare(comms.registry.new_thread("worker", frozenset(), str(tmp_path))) + registry = RegistryStore.at(comms.root) + route = Path.home() / ".local/state/agent-comms/active-route.json" + route.parent.mkdir(mode=0o700, parents=True) + spill = tmp_path / "backups" + # Put the guard first, as the interrupted historical backup did. Restore ordering + # must come from the registry declaration, not from this manifest's ordering. + backup = StoreBackup.take(comms.root, route, tuple(reversed(registry.written)), spill) + expected = {path: StoreBackup.digest(path) for path in registry.written} + RegistryTitleCarry(comms.root, comms.root, spill).carry(apply=True) + killed = subprocess.run( + [sys.executable, "-c", KILLED_RESTORE, str(comms.root), str(route), str(backup.path), str(publication)], + capture_output=True, text=True, timeout=30, + ) + assert killed.returncode == -signal.SIGKILL, killed.stdout + killed.stderr + # Enter the real installer. Its next Git command deliberately fails: rollback + # recovery must finish even before the checkout can be inspected or a stack built. + with pytest.raises(UpgradeRefused, match="git failed"): + upgrade( + tmp_path / "missing-checkout", comms.root, "HEAD", UnnamedInstaller(), RehearsalMode(), + False, spill, tmp_path / "install-scratch", lambda: None, InstallReport(), + ) + assert {path: StoreBackup.digest(path) for path in registry.written} == expected + assert comms.registry.require("worker").title != "Carried title" From ab68d3c583add66fd4d97ad97869b12b90cc778f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:01:42 -0400 Subject: [PATCH 12/21] Share annotation policy construction without changing native disclosure precedence --- src/agent_comms/annotation_disclosure.py | 21 ++++++++++++++------- src/agent_comms/turn_context.py | 22 ++++++++++------------ 2 files changed, 24 insertions(+), 19 deletions(-) diff --git a/src/agent_comms/annotation_disclosure.py b/src/agent_comms/annotation_disclosure.py index 4a8f2bd21a..3b792c79f0 100644 --- a/src/agent_comms/annotation_disclosure.py +++ b/src/agent_comms/annotation_disclosure.py @@ -5,7 +5,7 @@ import re from abc import abstractmethod from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, ClassVar from .declared_family import DeclaredFamily @@ -43,15 +43,22 @@ def require_text(self, text: str) -> str: return text -class WithheldAnnotations: - """Source text cannot be disclosed to an annotation provider.""" +class Annotations: + """The declared policy owns disclosure; messages and segments share its construction.""" + + disclosure_policy: ClassVar[type[DisclosurePolicy]] def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy: - return WithheldDisclosure() + return self.disclosure_policy() -class PublicInstructionAnnotations: +class WithheldAnnotations(Annotations): + """Source text cannot be disclosed to an annotation provider.""" + + disclosure_policy = WithheldDisclosure + + +class PublicInstructionAnnotations(Annotations): """Instruction text uses the existing credential-filtering disclosure policy.""" - def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy: - return PublicInstructionDisclosure() + disclosure_policy = PublicInstructionDisclosure diff --git a/src/agent_comms/turn_context.py b/src/agent_comms/turn_context.py index d8afc93af9..ef93d6cb25 100644 --- a/src/agent_comms/turn_context.py +++ b/src/agent_comms/turn_context.py @@ -29,6 +29,7 @@ from .message_reference import MessageReference from .native_session_reopen import NativeSessionIdentity from .peer_relevance import AlwaysListed, GoalShown, PeerRelevance +from .pi_payloads import PiMessage from .thread_identity import ThreadIncarnation, TurnId, TurnIdentity from .tool_result import DeclaredReply @@ -916,13 +917,9 @@ def provider_value(self): return list(self.messages) def public_text(self) -> str: - from .pi_payloads import PiMessage - return "\n".join(PiMessage.from_wire(message).text for message in self.messages) def source_ranges(self) -> tuple[ContributionCoordinates, ...]: - from .pi_payloads import PiMessage - offset = 0 ranges = [] parts = self.recorded_parts(self.measured_manifest()) @@ -933,20 +930,21 @@ def source_ranges(self) -> tuple[ContributionCoordinates, ...]: offset += len(text.encode("utf-8")) + 1 return tuple(ranges) - def disclosure_for(self, span: ContextSpan): - from .pi_payloads import PiMessage + def render_into(self, prompt_parts, provider): + provider.setdefault("messages", []).extend(self.messages) + +class NativeAnnotations(NativeMessages): + """A transcript span keeps the disclosure policy of its original native message.""" + + def disclosure_for(self, span: ContextSpan): for source, message in zip(self.source_ranges(), self.messages, strict=True): if source.contains(span.coordinates): return PiMessage.from_wire(message).disclosure_for(span) raise ValueError("Annotation names no original native message") - def render_into(self, prompt_parts, provider): - provider.setdefault("messages", []).extend(self.messages) - - @dataclass(frozen=True, kw_only=True) -class TranscriptSegment(NativeMessages, MeasuredNativeSegment): +class TranscriptSegment(NativeAnnotations, MeasuredNativeSegment): keyframe_interval: ClassVar[int] = 32 @classmethod @@ -975,7 +973,7 @@ class CompactionSummarySegment(PublicInstructionAnnotations, NativeMessages, Mea @dataclass(frozen=True, kw_only=True) -class InjectionMessageSegment(NativeMessages, MeasuredNativeSegment): +class InjectionMessageSegment(NativeAnnotations, MeasuredNativeSegment): pass From 22a7f6fdea156c52eb5c398b7f366d1a2642d95e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:02:17 -0400 Subject: [PATCH 13/21] Wire the root runtime owner into the installer Stack.publish(root) publishes the root's owner, on success and on restore. A live install adopts the root for the installed runtime before it stops owners; once they stop, the hold targets the staged runtime and its holder is re-entered with the new nonce before the first staged CLI call. The rehearsal holds its copy for the staged stack the same way. InstalledUpgrade reports the RuntimeProcesses still on an outgoing runtime after the bridges resume; RuntimeProcesses owns its lines. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/installer/upgrade.py | 29 +++++++++++++++++++--------- src/agent_comms/runtime_processes.py | 3 +++ tests/test_runtime_processes.py | 8 +++----- 3 files changed, 26 insertions(+), 14 deletions(-) diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index c2774c8291..c174191fff 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -105,6 +105,8 @@ from agent_comms.installer.runtime_resolution import RuntimeResolution from agent_comms.launch_runtime import LaunchRuntime, RestartEnvironment, SpecifiedImportPath from agent_comms.locked_store import BusFile, RootFile +from agent_comms.root_runtime import RootRuntime, RootRuntimeStore +from agent_comms.runtime_processes import RuntimeProcesses from agent_comms.maintenance_barrier import ( MaintenanceBarrier, MaintenanceOperator, @@ -402,12 +404,14 @@ class InstalledUpgrade(RehearsedUpgrade, Extendable): live: LiveRun install: InstallResult """The harnesses registered on the bus the target now reads.""" + processes: RuntimeProcesses = field(kw_only=True, default=RuntimeProcesses(())) + """Applications still running an outgoing runtime; reported, never signalled.""" finished: ClassVar[str] = "Install finished" """The headline of the post that reports this upgrade.""" def lines(self) -> tuple[str, ...]: - return (*super().lines(), *self.live.lines()) + return (*super().lines(), *self.live.lines(), *self.processes.lines()) @dataclass(frozen=True) @@ -543,12 +547,14 @@ def keep_native(self, native: Path, layout: Path) -> None: def prepare_native(self) -> None: self.probe(PrepareNativeProbe()) - def publish(self) -> None: - """The selected interpreter owns console metadata and harness declarations.""" + def publish(self, root: Path) -> None: + """The selected interpreter owns console metadata and harness declarations, and + ``root`` belongs to this stack's runtime from now on.""" self.commands.run( self.python, "-m", publication.__name__, cwd=self.checkout, env=self.environment(), ) + RootRuntime.publish(root, self.runtime) def environment(self, *paths: Path) -> dict: """Use the artifact's packages; carry steps also select the target tree's source.""" @@ -776,7 +782,7 @@ def publish(self, backup: StoreBackup, selected: RouteSelection) -> str: """The converted bus's route is ready before new runtime entrypoints open it.""" incoming = ActiveRoute(self.root.absolute(), RecordedBusIdentity.capture(Bus.path_in(self.root))) route = backup.rebind_route(incoming, selected) - self.staged.publish() + self.staged.publish(self.root) return route @staticmethod @@ -807,7 +813,8 @@ def finished(self, rehearsal: Rehearsed, bridges: SessionBridgeResumes, live_run = self.install(rehearsal, bridges, report) finished = InstalledUpgrade( rehearsed.source, rehearsed.target, rehearsed.root, rehearsed.pending, - rehearsed.rehearsal, live_run, harnesses(), commands=tuple(self.staged.commands.ran)) + rehearsed.rehearsal, live_run, harnesses(), commands=tuple(self.staged.commands.ran), + processes=RuntimeProcesses.observe(self.staged.runtime)) return (finished.extended(UnresumedUpgrade, unresumed=bridges.lines()) if bridges.failures else finished) @@ -933,7 +940,7 @@ def rehearsal(self, practice: PracticeBus) -> CarriedRehearsal: RootFile.content_owner(source).copy_file(source, destination) self.installed.running_owners(copied) maintenance = MaintenanceBarrier(RegistryStore.at(copied).path) - hold = maintenance.begin(MaintenanceOperator.current()) + hold = maintenance.retarget(maintenance.begin(MaintenanceOperator.current()), stack.runtime) with hold.holding(): migration = PreparedMigration.capture(hold, self.installed) passed = migration.carry(pending, stack, self.tree) @@ -954,6 +961,7 @@ def live(self, rehearsal: RehearsedFiles, report: InstallReport) -> CarriedRun: sum(path.stat().st_size for path in selected if path.exists()), ), self.root.parent) restore.require() + RootRuntimeStore.at(self.root).adopt(self.installed.runtime) running = self.installed.running_owners(self.root) maintenance = MaintenanceBarrier(RegistryStore.at(self.root).path) held = maintenance.begin(MaintenanceOperator.current()) @@ -962,6 +970,9 @@ def live(self, rehearsal: RehearsedFiles, report: InstallReport) -> CarriedRun: try: try: self.installed.owners(self.root, StopCliCommand, running) + held = maintenance.retarget(held, self.staged.runtime) + holder.close() + holder.enter_context(held.holding()) planned = self.staged.cli(self.root, PlannedReconcileCliCommand) unrehearsed = sorted(frozenset(planned.rewritten) - frozenset(rehearsal.files.names)) if unrehearsed: @@ -1003,7 +1014,7 @@ def live(self, rehearsal: RehearsedFiles, report: InstallReport) -> CarriedRun: if head != self.current: self.installed.commands.run("git", "-C", self.checkout, "reset", "-q", "--keep", self.current) Checkout(self.checkout).prepare() - self.installed.publish() + self.installed.publish(self.root) holder.close() maintenance.clear(restored_hold) started = self.installed.start(self.root, running) @@ -1042,7 +1053,7 @@ def uncopied(self) -> UnchangedRehearsal: return UnchangedRehearsal() def live(self, rehearsal: UnchangedRehearsal, report: InstallReport) -> UnchangedRun: - self.installed.publish() + self.installed.publish(self.root) report.writer = RuntimeWriter(self.root, self.installed, InstalledProgram()) return UnchangedRun() @@ -1082,7 +1093,7 @@ def live(self, rehearsal: FreshRehearsal, report: InstallReport) -> CreatedRun: services = HostService.restart_running(self.staged.python) except BaseException: backup.restore_route() - self.installed.publish() + self.installed.publish(self.root) raise backup.retain(self.spill_dir) return CreatedRun(services, route) diff --git a/src/agent_comms/runtime_processes.py b/src/agent_comms/runtime_processes.py index 85f282959a..c030333768 100644 --- a/src/agent_comms/runtime_processes.py +++ b/src/agent_comms/runtime_processes.py @@ -63,6 +63,9 @@ class RuntimeProcesses: outcomes: tuple[RuntimeProcessOutcome, ...] + def lines(self) -> tuple[str, ...]: + return tuple(outcome.line() for outcome in self.outcomes) + @classmethod def observe(cls, target: LaunchRuntime, proc: Path = Path("/proc")) -> RuntimeProcesses: directory = target.prefix.parent diff --git a/tests/test_runtime_processes.py b/tests/test_runtime_processes.py index 36e20e0c07..c3d067faad 100644 --- a/tests/test_runtime_processes.py +++ b/tests/test_runtime_processes.py @@ -8,7 +8,6 @@ from agent_comms.child_process import ProcessIdentity from agent_comms.field_codec import FieldCodec -from agent_comms.installer.upgrade import LiveRun, UnchangedRun from agent_comms.launch_runtime import LaunchRuntime from agent_comms.runtime_processes import ( RuntimeProcessOutcome, @@ -49,10 +48,9 @@ def test_runtime_process_family(tmp_path): assert outcome.process.application.describe() in outcome.line() # Neither discovery nor rendering gains the right to terminate an application. assert all(child.poll() is None for child in children) - run = UnchangedRun(processes=snapshot.outcomes) - decoded = FieldCodec.decode(LiveRun, FieldCodec.encode(run)) - assert decoded == run - assert outcome.line() in decoded.lines() + decoded = FieldCodec.decode(RuntimeProcesses, FieldCodec.encode(snapshot)) + assert decoded == snapshot + assert decoded.lines() == (outcome.line(),) assert FieldCodec.decode(RuntimeProcessOutcome, FieldCodec.encode(outcome)) == outcome stop(children[0]) # Only its actual launcher closes this child. assert not RuntimeProcesses.observe(target).outcomes From 0c3e7d7eff72b6ae9359be8cfb982624f3d72630 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:02:38 -0400 Subject: [PATCH 14/21] Preserve durable companion intent files while grouping SQLite content --- src/agent_comms/installer/backup.py | 2 +- src/agent_comms/locked_store.py | 23 +++++++++++++++++++---- 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index c7c354c1b1..9aa3b8d217 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -307,7 +307,7 @@ def transient(directory: str, names: list[str]) -> set[str]: omitted.add(path) elif path.is_file() and not path.is_symlink(): owner = RootFile.content_owner(path) - omitted.update(owner.member_paths(path) - {path}) + omitted.update(set(owner.files(path)) - {path}) return {path.name for path in omitted} shutil.copytree( diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index 1995b5dd3b..9ea6d88ad3 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -162,6 +162,10 @@ def member_paths(cls, path: Path) -> frozenset[Path]: """The resources composed onto one declared primary file.""" return frozenset((path, store_lock_path(path))) + @classmethod + def transient_paths(cls, path: Path) -> frozenset[Path]: + return frozenset((store_lock_path(path),)) + @classmethod def snapshot_members( cls, path: Path, observed: Mapping[Path, os.stat_result], @@ -246,17 +250,22 @@ def owned_files( undeclared existing file acquires content behavior without inventing a store declaration for it. """ - declarations = cls.existing_in(root) + primaries = cls.existing_in(root) + declarations = {member: owner for path, owner in primaries.items() + for member in owner.member_paths(path)} + declarations.update(primaries) selected = frozenset(paths) owners: dict[Path, type[RootFile]] = { - path: owner for path, owner in declarations.items() if path in selected + path: owner for path, owner in primaries.items() if path in selected } companions = {member for path, owner in owners.items() - for member in owner.member_paths(path) if member != path} + for member in (*owner.files(path), *owner.transient_paths(path)) + if member != path} owners.update((path, cls.content_owner(path) if path.exists() else RootFile) for path in selected - owners.keys() - companions) companions.update(member for path, owner in owners.items() - for member in owner.member_paths(path) if member != path) + for member in (*owner.files(path), *owner.transient_paths(path)) + if member != path) return { (DeclaredFile(declarations[path], path.relative_to(root)) if path in declarations else UndeclaredFile(path.relative_to(root))): owner @@ -554,6 +563,12 @@ def files(cls, path: Path) -> tuple[Path, ...]: def wal_path(cls, path: Path) -> Path: return path.with_name(path.name + cls.wal_suffix) + @classmethod + def transient_paths(cls, path: Path) -> frozenset[Path]: + return super().transient_paths(path) | frozenset(( + *(set(cls.files(path)) - {path, cls.wal_path(path)}), StoreOwner.snapshot_lock(path), + )) + @classmethod def snapshot_members( cls, path: Path, observed: Mapping[Path, os.stat_result], From 49f77daab8f9a332d82774328e54a962097c32b1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:03:45 -0400 Subject: [PATCH 15/21] Capture the declared registry pair whenever either member is selected --- src/agent_comms/installer/backup.py | 3 +++ tests/test_interrupted_rollback.py | 3 ++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index 4379455b4b..34db54e0ae 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -240,6 +240,9 @@ def route_original(self) -> bool: @classmethod def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path) -> StoreBackup: + registry = RegistryStore.at(root).written + if any(path in files for path in registry): + files = tuple(dict.fromkeys((*files, *registry))) path = spill_dir / BackupName.now(root).value backup = cls(root, route, path) selected = tuple(str(file.relative_to(root)) for file in files) diff --git a/tests/test_interrupted_rollback.py b/tests/test_interrupted_rollback.py index 565a4e587a..adb3d8fd86 100644 --- a/tests/test_interrupted_rollback.py +++ b/tests/test_interrupted_rollback.py @@ -10,6 +10,7 @@ import pytest from agent_comms.comms import Comms +from agent_comms.active_route import active_route_path from agent_comms.install_choices import RehearsalMode, UnnamedInstaller from agent_comms.installer.backup import StoreBackup, UpgradeRefused from agent_comms.installer.carry import Carried, Carry @@ -60,7 +61,7 @@ def test_next_install_completes_a_killed_registry_restore(tmp_path, monkeypatch, comms = prepare_native_comms(Comms(tmp_path / "wire")) comms.registry.declare(comms.registry.new_thread("worker", frozenset(), str(tmp_path))) registry = RegistryStore.at(comms.root) - route = Path.home() / ".local/state/agent-comms/active-route.json" + route = active_route_path() route.parent.mkdir(mode=0o700, parents=True) spill = tmp_path / "backups" # Put the guard first, as the interrupted historical backup did. Restore ordering From 8bd448ee640bc6207ae6246705708449deb7a93c Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:03:45 -0400 Subject: [PATCH 16/21] Carry newly landed thread-name rewrite paths through the same typed boundary --- src/agent_comms/locked_store.py | 2 +- tools/cutover/thread_name_owners.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py index 9ea6d88ad3..d4cedd2607 100644 --- a/src/agent_comms/locked_store.py +++ b/src/agent_comms/locked_store.py @@ -282,7 +282,7 @@ def paths_in(cls, root: Path) -> frozenset[Path]: @dataclass(frozen=True) class DeclaredFile(RootFile): - """A declared store and its primary path form one physical resource identity.""" + """A declared store and one retained file path identify a resource.""" owner: type[RootFile] path: Path diff --git a/tools/cutover/thread_name_owners.py b/tools/cutover/thread_name_owners.py index 4eace6ef03..aa85b8f97c 100644 --- a/tools/cutover/thread_name_owners.py +++ b/tools/cutover/thread_name_owners.py @@ -41,7 +41,7 @@ def carry(self, *, apply: bool) -> Carried: raise RelationViolationError("Inspect the unfinished old rename before carrying its bus.") changed = 0 detail: dict[str, int] = {} - rewritten: list[str] = [] + rewritten: list[Path] = [] for path in paths: with ExitStack() as lifetime: sidecar = path.name == PromptBinding.filename @@ -117,5 +117,5 @@ def carry(self, *, apply: bool) -> Carried: else: with closing(SQLiteFile.writer(path)) as destination: db.backup(destination) - rewritten.append(str(path.relative_to(self.root))) + rewritten.append(path.relative_to(self.root)) return Carried(changed, detail, tuple(rewritten)) From 98619073b5b3c21e84c0351a4f300fc786dad7a7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:04:28 -0400 Subject: [PATCH 17/21] Track the shared wire-member owner in duplicate evidence --- src/agent_comms/plan_phases.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 6a355cdc22..9219817a84 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -406,7 +406,7 @@ class DuplicatesPhase(MultipleInheritancePhase): ("__init__",), ("durable_turn.py", "execution_store.py", "exporting.py", "publisher.py"), ), - (("wire_member",), ("pi_payloads.py",)), + (("wire_member",), ("pi_payloads.py", "field_codec.py")), ( ( "_assert_response_schema", From 433a7f77c7394b0fcfbd18d9faa662d4109e97ac Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:04:36 -0400 Subject: [PATCH 18/21] Check the schema procedure at its TableSchema owner --- src/agent_comms/plan_phases.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 9219817a84..65f42a6971 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -412,11 +412,13 @@ class DuplicatesPhase(MultipleInheritancePhase): "_assert_response_schema", "assert_cohort_schema", "assert_native_runtime_schema", + "require_schema", ), ( "cohort_schema.py", "coordinated_runtime_schema.py", "coordination_response.py", + "typed_table.py", ), ), ( From 35d2080c4125eb208db5746c5875c21d698fb1b2 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:04:54 -0400 Subject: [PATCH 19/21] Keep duplicate disclosure evidence on the shared policy owner --- src/agent_comms/plan_phases.py | 2 +- src/agent_comms/turn_context.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 65f42a6971..f5dc43c1fd 100644 --- a/src/agent_comms/plan_phases.py +++ b/src/agent_comms/plan_phases.py @@ -431,7 +431,7 @@ class DuplicatesPhase(MultipleInheritancePhase): ), ( ("annotation_disclosure", "disclosure_for"), - ("pi_payloads.py", "turn_context.py"), + ("pi_payloads.py", "turn_context.py", "annotation_disclosure.py"), ), ) ) diff --git a/src/agent_comms/turn_context.py b/src/agent_comms/turn_context.py index ef93d6cb25..4a19469de3 100644 --- a/src/agent_comms/turn_context.py +++ b/src/agent_comms/turn_context.py @@ -943,6 +943,7 @@ def disclosure_for(self, span: ContextSpan): return PiMessage.from_wire(message).disclosure_for(span) raise ValueError("Annotation names no original native message") + @dataclass(frozen=True, kw_only=True) class TranscriptSegment(NativeAnnotations, MeasuredNativeSegment): keyframe_interval: ClassVar[int] = 32 From d3f611a48d0533cae80c9c8bb22f5cc21b418ce7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:08:35 -0400 Subject: [PATCH 20/21] Preserve the existing native message import boundary --- src/agent_comms/turn_context.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/turn_context.py b/src/agent_comms/turn_context.py index 4a19469de3..21fee46bf8 100644 --- a/src/agent_comms/turn_context.py +++ b/src/agent_comms/turn_context.py @@ -29,7 +29,6 @@ from .message_reference import MessageReference from .native_session_reopen import NativeSessionIdentity from .peer_relevance import AlwaysListed, GoalShown, PeerRelevance -from .pi_payloads import PiMessage from .thread_identity import ThreadIncarnation, TurnId, TurnIdentity from .tool_result import DeclaredReply @@ -917,9 +916,13 @@ def provider_value(self): return list(self.messages) def public_text(self) -> str: + from .pi_payloads import PiMessage + return "\n".join(PiMessage.from_wire(message).text for message in self.messages) def source_ranges(self) -> tuple[ContributionCoordinates, ...]: + from .pi_payloads import PiMessage + offset = 0 ranges = [] parts = self.recorded_parts(self.measured_manifest()) @@ -938,6 +941,8 @@ class NativeAnnotations(NativeMessages): """A transcript span keeps the disclosure policy of its original native message.""" def disclosure_for(self, span: ContextSpan): + from .pi_payloads import PiMessage + for source, message in zip(self.source_ranges(), self.messages, strict=True): if source.contains(span.coordinates): return PiMessage.from_wire(message).disclosure_for(span) From 470089cacb01584ff22d42f34eb9f649da81d283 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 05:09:12 -0400 Subject: [PATCH 21/21] Derive a JSON shape family wire structure from its owned member values Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/field_codec.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py index b7c07ca49a..225b9bf7ac 100644 --- a/src/agent_comms/field_codec.py +++ b/src/agent_comms/field_codec.py @@ -1098,6 +1098,11 @@ def from_wire(cls, data): raise TypeError("A JSON shape factory must return its own family") return result + @classmethod + def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]: + """The wire value is one owned member's value; a member owns only its own.""" + return compile_decoder(Union[tuple(annotation for _, annotation in cls._json_owners.values())]) + @classmethod def schema(cls) -> dict[str, Any]: alternatives = [FieldCodec.value_schema(annotation)