Let declared file owners identify installer rewrites

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

src/agent_comms/installer/backup.py

@@ -16,14 +16,14 @@
 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
+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,21 +55,23 @@ def matches_extraction(self, copied: SavedFile) -> bool:
         return self.size == copied.size
 
 
-def bus_files(root: Path) -> dict[str, 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)
-    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
+    return {
+        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(root / resource.path, observed)
+        } for resource, owner in RootFile.owned_files(root, observed).items()
+    }
 
 
 @dataclass(frozen=True)
@@ -172,9 +174,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],
-                                 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], ...],
+                                 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()}')
@@ -228,9 +235,10 @@ 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(sorted(resource.path for resource in RootFile.owned_files(root, files)))
+        files = tuple(root / path for path in selected)
         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):
@@ -239,13 +247,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(
+            {PathText.encode(file): digest for file, digest in contents.items()}, indent=1))
         return backup
 
     def retain(self, spill_dir: Path) -> tuple[Path, ...]:
@@ -288,29 +298,28 @@ 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(set(owner.files(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[str, str]:
-        contents = {}
+    def capture(self, files: tuple[Path, ...]) -> dict[Path, 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)
@@ -319,7 +328,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[Path, str]) -> bool:
         """An unchanged selected file keeps its identity as well as its contents."""
         target = self.root / name
         if name not in saved:
@@ -334,14 +343,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)
@@ -353,7 +364,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()

src/agent_comms/installer/carry.py

@@ -15,11 +15,10 @@
 import inspect
 import json
 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
-from importlib import import_module
 from pathlib import Path
 from typing import Annotated, ClassVar
 
@@ -37,40 +36,26 @@ 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."""
 
 
 @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."""
@@ -132,19 +117,19 @@ 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()
-        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]
+        tree = tree.resolve()
+        directory = tree / STEPS
+        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:

src/agent_comms/installer/step_worker.py

@@ -1,28 +1,26 @@
 """Run one carry operation over a pass of steps with the staged stack's interpreter.
 
 ``python -m agent_comms.installer.step_worker OPERATION RESULT STEPS``, with the target
-tree's ``src`` and ``tools`` on the path. ``STEPS`` lists each step's module name and
+tree's ``src`` and ``tools`` on the path. ``STEPS`` lists each step's declared name and
 encoded input in pass order; one process answers the whole pass, so a pass pays the
 interpreter's start once, not once per step. Writes the encoded answers to ``RESULT``.
 """
 
 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:
     encoded_operation, result_path, encoded_steps = sys.argv[1:]
     operation = FieldCodec.decode(CarryOperation, FieldCodec.parse(encoded_operation))
     steps = []
     for name, encoded_step in FieldCodec.parse(encoded_steps):
-        import_module(f"{STEPS.name}.{name}")
-        steps.append(FieldCodec.decode(Carry.declared[name], encoded_step))
+        steps.append(FieldCodec.decode(Carry.member_named(name), encoded_step))
     Path(result_path).write_text(operation.encode_answers(
         tuple(operation.run(step) for step in steps)))