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..9aa3b8d217 100644
--- a/src/agent_comms/installer/backup.py
+++ b/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()
diff --git a/src/agent_comms/installer/carry.py b/src/agent_comms/installer/carry.py
index 77df6f3a6e..9cf4ea396b 100644
--- a/src/agent_comms/installer/carry.py
+++ b/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:
diff --git a/src/agent_comms/installer/step_worker.py b/src/agent_comms/installer/step_worker.py
index 57906b4f61..4608fdb5a1 100644
--- a/src/agent_comms/installer/step_worker.py
+++ b/src/agent_comms/installer/step_worker.py
@@ -1,7 +1,7 @@
 """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``.
 """
@@ -9,11 +9,10 @@
 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:
@@ -21,8 +20,7 @@ def main() -> None:
     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)))
 
diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py
index c2774c8291..031be0254d 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, DeclaredFile, 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[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 = {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: resource.path for resource, snapshot in (*before.items(), *after.items())
+               for member in snapshot}
+    changed = {resource.path for resource 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({resource.path for resource 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)
@@ -250,7 +255,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(), retired) for step, retired in
+            StepRun(step.declared_name, retired) for step, retired in
             zip(pending, stack.call_steps(pending, tree, RetireInputsOperation()), strict=True)
         ))
 
@@ -595,7 +600,7 @@ def call_steps[Answer](self, steps: tuple[Carry, ...], tree: TargetTree,
             result = Path(scratch) / "result.json"
             self.commands.run(self.python, "-m", step_worker.__name__,
                               json.dumps(FieldCodec.encode(operation)), result,
-                              json.dumps([[step.name(), FieldCodec.encode(step)]
+                              json.dumps([[step.declared_name, FieldCodec.encode(step)]
                                           for step in steps]),
                               cwd=tree.path, env=self.environment(*tree.source_paths()))
             return operation.decode_answers(result.read_text())
@@ -798,7 +803,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,
@@ -878,7 +883,7 @@ class PreparedMigration(Extendable):
     """The held bus as the outgoing runtime read it, before any migration."""
 
     hold: MaintenanceState
-    before: dict[str, SavedFile]
+    before: dict[DeclaredFile | UndeclaredFile, dict[Path, SavedFile]]
     read: BusReadBack
 
     @property
@@ -894,7 +899,7 @@ def capture(cls, hold: MaintenanceState, outgoing: Stack) -> PreparedMigration:
     def carry(self, pending: tuple[Carry, ...], staged: Stack,
               tree: TargetTree) -> CarriedPass:
         """One migration sequence for rehearsal and the backed-up live bus."""
-        steps = [StepRun(step.name(), carried) for step, carried in zip(
+        steps = [StepRun(step.declared_name, carried) for step, carried in zip(
             pending, staged.call_steps(pending, tree, MigrateOperation(apply=True)), strict=True)]
         reconciled = staged.cli(self.root, ReconcileStoresCliCommand)
         steps.append(StepRun(ReconcileStoresCliCommand.declared_name, Carried(
@@ -946,7 +951,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..d4cedd2607 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
@@ -162,6 +162,17 @@ 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],
+    ) -> 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))
@@ -229,6 +240,38 @@ 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.
+        """
+        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 primaries.items() if path in selected
+        }
+        companions = {member for path, owner in owners.items()
+                      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.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
+            for path, owner in owners.items() if path not in companions
+        }
+
     @classmethod
     def paths_in(cls, root: Path) -> frozenset[Path]:
         return frozenset(
@@ -237,6 +280,21 @@ def paths_in(cls, root: Path) -> frozenset[Path]:
         )
 
 
+@dataclass(frozen=True)
+class DeclaredFile(RootFile):
+    """A declared store and one retained file path identify a resource."""
+
+    owner: type[RootFile]
+    path: 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 +424,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 +556,28 @@ 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 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],
+    ) -> 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/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),
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",
diff --git a/tools/cutover/carry_bus_history.py b/tools/cutover/carry_bus_history.py
index 0f3b03bb73..5eaf261f0b 100644
--- a/tools/cutover/carry_bus_history.py
+++ b/tools/cutover/carry_bus_history.py
@@ -152,7 +152,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)
@@ -221,7 +221,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),
         )
 
 
@@ -829,7 +829,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):
@@ -842,7 +842,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)
 
 
@@ -914,7 +914,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:
@@ -946,7 +946,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)
@@ -955,7 +955,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
@@ -980,7 +980,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
@@ -1049,7 +1049,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 ())
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))