Finish install 24's half restore of install 22's backup

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)

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

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