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)), + )