From b2844a2383cd747866c530b4a79dc65dd719a150 Mon Sep 17 00:00:00 2001 From: Citation fixture Date: Fri, 23 Oct 2026 09:43:29 -0400 Subject: [PATCH 1/3] Finish install 24's half restore of install 22's backup, and restore deleted parents Install 24 (before #2779) took install 22's completed backup for an interrupted restore and copied it over the live root. It replaced .root-runtime.json, .registry-owner-guard, bus.sqlite3 and coordination.sqlite3 with their pre-22 contents, then died at mkstemp(dir=target.parent): install 22's carry had deleted imported_sessions/selected-tool-ledger. The live root is now owned by f3f069fb8 while current selects 701746802, so no client attaches (install 25's ForeignRuntimeError). - StoreBackup.restore recreates a saved file's missing parent directories, so a rollback after a carry that removed a directory restores it instead of dying. - tools/cutover/finish_interrupted_restore.py finishes that restore once with the installer's own StoreBackup.restore under the maintenance hold, then selects the runtime the restored root names through Stack.publish. The backup is never retired. Delete it after its one live run. - carry_selected_tool_slots comes back (deleted in a00663d28): the restored root holds the ledger files again, so the next install carries them again. SavedSessionFile is BusTranscriptFile since 502f3d851. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/installer/backup.py | 10 ++ tests/test_store_backup.py | 17 ++ tools/cutover/carry_selected_tool_slots.py | 183 ++++++++++++++++++++ tools/cutover/finish_interrupted_restore.py | 93 ++++++++++ 4 files changed, 303 insertions(+) create mode 100644 tools/cutover/carry_selected_tool_slots.py create mode 100644 tools/cutover/finish_interrupted_restore.py diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py index eb1f4c0e2e..df6f345ee0 100644 --- a/src/agent_comms/installer/backup.py +++ b/src/agent_comms/installer/backup.py @@ -403,6 +403,15 @@ def stores_original(self) -> bool: owners = self.resource_owners(selected, saved) return all(self.file_original(name, saved, owners[name]) for name in selected) + def restore_parents(self, target: Path) -> None: + """A carry may delete the directory that held a saved file (install 24 met + imported_sessions/selected-tool-ledger gone); restoring the file restores it.""" + missing = [parent for parent in target.relative_to(self.root).parents + if not (self.root / parent).exists()] + for parent in reversed(missing): + (self.root / parent).mkdir(mode=0o700) + Platform.current().sync_directory((self.root / parent).parent) + def restore(self) -> None: saved = self.files() selected = self.selected_files() @@ -421,6 +430,7 @@ def restore(self) -> None: target = self.root / name owner = owners[name] if name in saved: + self.restore_parents(target) descriptor, temporary = tempfile.mkstemp(prefix='.restore-', dir=target.parent) os.close(descriptor) stage = Path(temporary) diff --git a/tests/test_store_backup.py b/tests/test_store_backup.py index 45fcb9355e..a7b33f8640 100644 --- a/tests/test_store_backup.py +++ b/tests/test_store_backup.py @@ -161,6 +161,23 @@ def test_store_backup_restores_wal_rows_and_removes_new_files(tmp_path): assert archived.exists() +def test_store_backup_restores_a_file_whose_directory_a_carry_deleted(tmp_path): + """Install 24 died restoring a ledger file whose directory install 22's carry had + removed; restoring the file restores its directory.""" + root = tmp_path / 'wire' + saved = root / 'imported_sessions' / 'retained.jsonl' + saved.parent.mkdir(parents=True) + saved.write_text('{"retained": true}\n') + route = tmp_path / 'route.json' + backup = StoreBackup.take(root, route, (saved,), tmp_path / 'spill') + saved.unlink() + saved.parent.rmdir() + backup.restore() + assert saved.read_text() == '{"retained": true}\n' + assert backup.stores_original() + assert not backup.restoring.exists() + + @pytest.mark.parametrize('root_name', ('ROOT', 'wire%[x]+.{}')) def test_backup_name_recognizes_only_its_own_complete_names(tmp_path, root_name): root = tmp_path / root_name diff --git a/tools/cutover/carry_selected_tool_slots.py b/tools/cutover/carry_selected_tool_slots.py new file mode 100644 index 0000000000..6183093ce8 --- /dev/null +++ b/tools/cutover/carry_selected_tool_slots.py @@ -0,0 +1,183 @@ +"""Move the selected-tool files into their one coordination row, then retire them. + +The filename and its contents are the original slot identity. No input table or +transcript can reconstruct them. Delete this carry after every root has crossed. +Brought back once: install 24 restored install 22's pre-carry backup over the live +root, so the live root holds the files again. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from contextlib import closing +from dataclasses import dataclass +from pathlib import Path +from typing import ClassVar + +from agent_comms.child_process import Platform +from agent_comms.coordination_database import CoordinationStore +from agent_comms.errors import RelationViolationError +from agent_comms.installer.carry import Carried, Carry +from agent_comms.locked_store import SQLiteFile +from agent_comms.native_runtime_input import ( + AcknowledgedSlot, + ConsumedSlot, + SelectedSlotState, + SelectedToolSlot, +) +from agent_comms.native_session_files import BusTranscriptFile +from agent_comms.native_tool_call import SelectedToolDenied +from agent_comms.typed_table import HeldBusFile + + +@dataclass(frozen=True) +class LedgerSlot(ABC): + """One legacy slot; its file kind owns validation and the declared state.""" + + consumed: Path + + @property + @abstractmethod + def state(self) -> type[SelectedSlotState]: ... + + @classmethod + def at(cls, consumed: Path, acknowledgements: frozenset[Path]) -> LedgerSlot: + acknowledgement = consumed.with_name(consumed.name + ".done") + if acknowledgement in acknowledgements: + return AcknowledgedLedgerSlot(consumed, acknowledgement) + return ConsumedLedgerSlot(consumed) + + @property + def files(self) -> tuple[Path, ...]: + return (self.consumed,) + + def contents(self) -> bytes: + return self.consumed.read_bytes() + + def retire_acknowledgement(self) -> None: + for acknowledgement in self.files[1:]: + acknowledgement.unlink() + + def read(self) -> SelectedToolSlot: + original = self.contents() + return SelectedToolSlot( + self.consumed.name, + original.decode("ascii").removesuffix("\n"), + self.state, + ) + + +class ConsumedLedgerSlot(LedgerSlot): + state: ClassVar[type[SelectedSlotState]] = ConsumedSlot + + +@dataclass(frozen=True) +class AcknowledgedLedgerSlot(LedgerSlot): + acknowledgement: Path + state: ClassVar[type[SelectedSlotState]] = AcknowledgedSlot + + @property + def files(self) -> tuple[Path, ...]: + return (self.consumed, self.acknowledgement) + + def contents(self) -> bytes: + original = super().contents() + if self.acknowledgement.read_bytes() != original: + raise RelationViolationError( + f"Selected tool acknowledgement differs from {self.consumed}: " + f"{self.acknowledgement}" + ) + return original + + +class SelectedToolSlots(Carry): + opened_files = (CoordinationStore,) + + def directories(self) -> tuple[Path, ...]: + directories = { + path + for owner, root in BusTranscriptFile.declarations_in(self.root) + for path in root.glob(str(owner.pattern().parent / "selected-tool-ledger")) + } + for directory in directories: + self.require_confined(self.root, directory) + return tuple(sorted(directories)) + + def slots(self) -> tuple[LedgerSlot, ...]: + slots: list[LedgerSlot] = [] + for directory in self.directories(): + files = frozenset(directory.iterdir()) + for path in files: + self.require_confined(self.root, path) + if not path.is_file(): + raise RelationViolationError( + f"Selected tool ledger entry is not a file: {path}" + ) + acknowledgements = frozenset(path for path in files if path.match("*.done")) + consumed = files - acknowledgements + for path in acknowledgements: + if path.with_suffix("") not in consumed: + raise RelationViolationError( + f"Selected tool acknowledgement has no consumed file: {path}" + ) + slots.extend(LedgerSlot.at(path, acknowledgements) for path in sorted(consumed)) + return tuple(slots) + + def input_files(self) -> tuple[Path, ...]: + return (*super().input_files(), *(path for slot in self.slots() for path in slot.files)) + + def carry(self, *, apply: bool) -> Carried: + slots = self.slots() + directories = self.directories() + if not directories: + return Carried(0) + rows = tuple(slot.read() for slot in slots) + files = tuple(path for slot in slots for path in slot.files) + acknowledged = len(files) - len(slots) + detail: dict[str, int | bool] = { + "consumed_files": len(slots), + "acknowledgement_files": acknowledged, + "consumed_rows": len(slots) - acknowledged, + "acknowledged_rows": acknowledged, + "slot_rows": len(slots), + "deleted_files": len(files) if apply else 0, + } + changed = len(files) + len(directories) - len({slot.consumed.parent for slot in slots}) + if not apply: + return Carried(changed, detail) + path = CoordinationStore.path_in(self.root) + self.require_confined(self.root, path) + HeldBusFile(CoordinationStore, path).require() + if rows: + with closing(SQLiteFile.writer(path)) as db, SQLiteFile.write_transaction(db): + SelectedToolSlot.create(db, if_not_exists=True) + for slot, row in zip(slots, rows, strict=True): + if row.insert_if_absent(db).rowcount == 0: + current = SelectedToolSlot.require(db, input_id=row.input_id) + try: + current.require_call(row.call_id) + except SelectedToolDenied as error: + raise RelationViolationError( + f"Selected tool slot differs from committed row: {slot.consumed}" + ) from error + if current.state is not row.state: + raise RelationViolationError( + f"Selected tool state differs from committed row: {slot.consumed}" + ) + # Every row is committed before the first unlink. Persist removal of + # acknowledgements before removing consumed files: a killed carry can + # leave a consumed file, but can never manufacture an orphan .done. + for slot in slots: + slot.retire_acknowledgement() + for directory in directories: + Platform.current().sync_directory(directory) + for slot in slots: + slot.consumed.unlink() + for directory in directories: + directory.rmdir() + Platform.current().sync_directory(directory.parent) + return Carried( + changed, + detail, + (path.relative_to(self.root), *(file.relative_to(self.root) for file in files)), + ) diff --git a/tools/cutover/finish_interrupted_restore.py b/tools/cutover/finish_interrupted_restore.py new file mode 100644 index 0000000000..bc793208f8 --- /dev/null +++ b/tools/cutover/finish_interrupted_restore.py @@ -0,0 +1,93 @@ +"""Finish the restore install 24 began and did not complete on the live root, once. + +Install 24's ``complete_interrupted`` (before #2779) took install 22's backup for an +interrupted restore and copied it over the live root. It replaced ``.root-runtime.json``, +``.registry-owner-guard``, ``bus.sqlite3`` and ``coordination.sqlite3`` with their +pre-install-22 contents, then died on a selected-tool ledger directory install 22's carry +had deleted. The root is half pre-22 and half post-22, owned by runtime f3f069fb8 while +``current`` selects 701746802, so no client attaches. + +This finishes that restore with the installer's own ``StoreBackup.restore`` under the +maintenance hold, then selects the runtime the restored root names through the installer's +own ``Stack.publish``: the rollback path's "original runtime selected again". The backup +directory is the only complete copy and is never retired or renamed here. Run it once on +the live root, then delete it. +""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass +from pathlib import Path + +from agent_comms.active_route import active_route_path, route_directory +from agent_comms.host_services import HostService +from agent_comms.installer.backup import StoreBackup, UpgradeRefused +from agent_comms.installer.upgrade import InstallCommands, Stack +from agent_comms.launch_runtime import LaunchRuntime +from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator +from agent_comms.private_path import FileRevision +from agent_comms.registry_store import RegistryStore +from agent_comms.root_runtime import RootRuntimeStore + + +@dataclass(frozen=True) +class InterruptedRestore: + root: Path + backup: Path + runtime: Path + checkout: Path + home: Path + + @property + def store_backup(self) -> StoreBackup: + return StoreBackup(self.root, active_route_path(), self.backup) + + @property + def stack(self) -> Stack: + return Stack(self.checkout, InstallCommands(), self.home, + LaunchRuntime(self.runtime / "bin/python")) + + def require_half_restored(self) -> None: + """Only the root this backup's restore already began on: its runtime record is + the backup's byte for byte.""" + name = RootRuntimeStore.path_in(self.root).relative_to(self.root) + if (self.root / name).read_bytes() != (self.backup / "stores" / name).read_bytes(): + raise UpgradeRefused(f"{self.root} does not hold {self.backup}'s runtime record") + if not self.stack.python.exists(): + raise UpgradeRefused(f"No runtime at {self.stack.python}") + + def finish(self) -> None: + self.require_half_restored() + backup = self.store_backup + with route_directory(backup.route): + backup.route_revision = FileRevision.capture(backup.route) + maintenance = MaintenanceBarrier(RegistryStore.at(self.root).path) + hold = maintenance.begin(MaintenanceOperator.current()) + with hold.holding(): + backup.restore() + self.stack.publish(self.root) + maintenance.clear(hold) + + def run(self) -> None: + """Nothing may write the root while it is restored; services start again on the + runtime that owns it.""" + stopped = HostService.stop_running() + self.finish() + self.stack.start_services(stopped) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--root", type=Path, required=True) + parser.add_argument("--backup", type=Path, required=True) + parser.add_argument("--runtime", type=Path, required=True) + parser.add_argument("--checkout", type=Path, required=True) + arguments = parser.parse_args() + InterruptedRestore(arguments.root, arguments.backup, arguments.runtime, + arguments.checkout, Path.home()).run() + print(f"Restored {arguments.root} from {arguments.backup}; runtime {arguments.runtime}") + + +if __name__ == "__main__": + main() From 3c1d623d76dd55035a90237c19492c40ef8ff93c Mon Sep 17 00:00:00 2001 From: Citation fixture Date: Fri, 23 Oct 2026 10:04:48 -0400 Subject: [PATCH 2/3] Run the restore finish under an installed runtime's interpreter Backup digests hash SQLite iterdump text; the runtime's SQLite 3.50.4 matches files.json, a newer SQLite does not, and restore then refuses on bus.sqlite3. Co-Authored-By: Claude Opus 5.5 --- tools/cutover/finish_interrupted_restore.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tools/cutover/finish_interrupted_restore.py b/tools/cutover/finish_interrupted_restore.py index bc793208f8..acd24dfb49 100644 --- a/tools/cutover/finish_interrupted_restore.py +++ b/tools/cutover/finish_interrupted_restore.py @@ -12,6 +12,10 @@ own ``Stack.publish``: the rollback path's "original runtime selected again". The backup directory is the only complete copy and is never retired or renamed here. Run it once on the live root, then delete it. + +Run it with an installed runtime's interpreter: the backup's digests are SQLite dumps +taken with the runtimes' SQLite 3.50.4, and a dev environment's newer SQLite dumps the +same bus differently, so the restore would refuse its own result. """ from __future__ import annotations From cd2c51c0d07a1e89d9f30cd62e70663c8ac34994 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 12:17:11 -0400 Subject: [PATCH 3/3] Delete the one-shot restore tool after its live run finish_interrupted_restore.py ran once on the live root on 2026-10-23 12:06:22-12:12:41 with exit 0; install 26 from 66e63fe8b then exited 0. The selected-tool carry stays: the live root still holds 118,271 legacy ledger files and selected_tool_slot has 0 rows. Co-Authored-By: Claude Opus 5.5 --- tools/cutover/finish_interrupted_restore.py | 97 --------------------- 1 file changed, 97 deletions(-) delete mode 100644 tools/cutover/finish_interrupted_restore.py diff --git a/tools/cutover/finish_interrupted_restore.py b/tools/cutover/finish_interrupted_restore.py deleted file mode 100644 index acd24dfb49..0000000000 --- a/tools/cutover/finish_interrupted_restore.py +++ /dev/null @@ -1,97 +0,0 @@ -"""Finish the restore install 24 began and did not complete on the live root, once. - -Install 24's ``complete_interrupted`` (before #2779) took install 22's backup for an -interrupted restore and copied it over the live root. It replaced ``.root-runtime.json``, -``.registry-owner-guard``, ``bus.sqlite3`` and ``coordination.sqlite3`` with their -pre-install-22 contents, then died on a selected-tool ledger directory install 22's carry -had deleted. The root is half pre-22 and half post-22, owned by runtime f3f069fb8 while -``current`` selects 701746802, so no client attaches. - -This finishes that restore with the installer's own ``StoreBackup.restore`` under the -maintenance hold, then selects the runtime the restored root names through the installer's -own ``Stack.publish``: the rollback path's "original runtime selected again". The backup -directory is the only complete copy and is never retired or renamed here. Run it once on -the live root, then delete it. - -Run it with an installed runtime's interpreter: the backup's digests are SQLite dumps -taken with the runtimes' SQLite 3.50.4, and a dev environment's newer SQLite dumps the -same bus differently, so the restore would refuse its own result. -""" - -from __future__ import annotations - -import argparse -from dataclasses import dataclass -from pathlib import Path - -from agent_comms.active_route import active_route_path, route_directory -from agent_comms.host_services import HostService -from agent_comms.installer.backup import StoreBackup, UpgradeRefused -from agent_comms.installer.upgrade import InstallCommands, Stack -from agent_comms.launch_runtime import LaunchRuntime -from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator -from agent_comms.private_path import FileRevision -from agent_comms.registry_store import RegistryStore -from agent_comms.root_runtime import RootRuntimeStore - - -@dataclass(frozen=True) -class InterruptedRestore: - root: Path - backup: Path - runtime: Path - checkout: Path - home: Path - - @property - def store_backup(self) -> StoreBackup: - return StoreBackup(self.root, active_route_path(), self.backup) - - @property - def stack(self) -> Stack: - return Stack(self.checkout, InstallCommands(), self.home, - LaunchRuntime(self.runtime / "bin/python")) - - def require_half_restored(self) -> None: - """Only the root this backup's restore already began on: its runtime record is - the backup's byte for byte.""" - name = RootRuntimeStore.path_in(self.root).relative_to(self.root) - if (self.root / name).read_bytes() != (self.backup / "stores" / name).read_bytes(): - raise UpgradeRefused(f"{self.root} does not hold {self.backup}'s runtime record") - if not self.stack.python.exists(): - raise UpgradeRefused(f"No runtime at {self.stack.python}") - - def finish(self) -> None: - self.require_half_restored() - backup = self.store_backup - with route_directory(backup.route): - backup.route_revision = FileRevision.capture(backup.route) - maintenance = MaintenanceBarrier(RegistryStore.at(self.root).path) - hold = maintenance.begin(MaintenanceOperator.current()) - with hold.holding(): - backup.restore() - self.stack.publish(self.root) - maintenance.clear(hold) - - def run(self) -> None: - """Nothing may write the root while it is restored; services start again on the - runtime that owns it.""" - stopped = HostService.stop_running() - self.finish() - self.stack.start_services(stopped) - - -def main() -> None: - parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) - parser.add_argument("--root", type=Path, required=True) - parser.add_argument("--backup", type=Path, required=True) - parser.add_argument("--runtime", type=Path, required=True) - parser.add_argument("--checkout", type=Path, required=True) - arguments = parser.parse_args() - InterruptedRestore(arguments.root, arguments.backup, arguments.runtime, - arguments.checkout, Path.home()).run() - print(f"Restored {arguments.root} from {arguments.backup}; runtime {arguments.runtime}") - - -if __name__ == "__main__": - main()