diff --git a/src/agent_comms/annotation_disclosure.py b/src/agent_comms/annotation_disclosure.py
index 4a8f2bd21a..3b792c79f0 100644
--- a/src/agent_comms/annotation_disclosure.py
+++ b/src/agent_comms/annotation_disclosure.py
@@ -5,7 +5,7 @@
 import re
 from abc import abstractmethod
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 from .declared_family import DeclaredFamily
 
@@ -43,15 +43,22 @@ def require_text(self, text: str) -> str:
         return text
 
 
-class WithheldAnnotations:
-    """Source text cannot be disclosed to an annotation provider."""
+class Annotations:
+    """The declared policy owns disclosure; messages and segments share its construction."""
+
+    disclosure_policy: ClassVar[type[DisclosurePolicy]]
 
     def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy:
-        return WithheldDisclosure()
+        return self.disclosure_policy()
 
 
-class PublicInstructionAnnotations:
+class WithheldAnnotations(Annotations):
+    """Source text cannot be disclosed to an annotation provider."""
+
+    disclosure_policy = WithheldDisclosure
+
+
+class PublicInstructionAnnotations(Annotations):
     """Instruction text uses the existing credential-filtering disclosure policy."""
 
-    def disclosure_for(self, span: ContextSpan) -> DisclosurePolicy:
-        return PublicInstructionDisclosure()
+    disclosure_policy = PublicInstructionDisclosure
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/comms.py b/src/agent_comms/comms.py
index d0aaf0d1b2..ca82715d4e 100644
--- a/src/agent_comms/comms.py
+++ b/src/agent_comms/comms.py
@@ -14,6 +14,7 @@
 from .context_store import ContextStore
 from .goal_management import Goals
 from .history_views import HistoryViews
+from .launch_runtime import LaunchRuntime
 from .locked_store import RegistryFile
 from .maintenance_barrier import MaintenanceBarrier
 from .message_bus import MessageBus
@@ -22,6 +23,7 @@
 from .publisher import OwnerPublisher, Publisher
 from .registration import Registration
 from .relationships import ThreadRelationships
+from .root_runtime import RootRuntime
 from .thread_configuration import ThreadConfiguration
 from .thread_management import ThreadManagement
 from .transcripts import Transcripts
@@ -32,6 +34,7 @@ def __init__(
         self, root: CommsRoute | Path, publisher: type[Publisher] = OwnerPublisher,
     ) -> None:
         self.route = resolve_comms_route(root)
+        RootRuntime.require_attachable(self.route.root, LaunchRuntime.running())
         self.route.observe_root()
         self.registry = Registration(RegistryFile.path_in(self.root))
         self.bus = MessageBus(
diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py
index b7c07ca49a..225b9bf7ac 100644
--- a/src/agent_comms/field_codec.py
+++ b/src/agent_comms/field_codec.py
@@ -1098,6 +1098,11 @@ def from_wire(cls, data):
             raise TypeError("A JSON shape factory must return its own family")
         return result
 
+    @classmethod
+    def wire_structure(cls, structural: ValueDecoder[object]) -> ValueDecoder[object]:
+        """The wire value is one owned member's value; a member owns only its own."""
+        return compile_decoder(Union[tuple(annotation for _, annotation in cls._json_owners.values())])
+
     @classmethod
     def schema(cls) -> dict[str, Any]:
         alternatives = [FieldCodec.value_schema(annotation)
diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py
index d16d2e3908..7eedc75d5b 100644
--- a/src/agent_comms/installer/backup.py
+++ b/src/agent_comms/installer/backup.py
@@ -16,15 +16,16 @@
 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.active_route import ActiveRoute, _route_directory, active_route_path, 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.registry_store import RegistryStore
 from agent_comms.store_files import AtomicFile
 
 
@@ -55,21 +56,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 +175,27 @@ 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()))
+
+    @classmethod
+    def interrupted(cls, root: Path, spill_dir: Path) -> tuple[StoreBackup, ...]:
+        if not spill_dir.exists():
+            return ()
+        route = active_route_path()
+        backups = (
+            cls(root, route, spill_dir / name.value)
+            for name in sorted(BackupName.matching(
+                root, (path.name for path in spill_dir.iterdir() if path.is_dir())
+            ), reverse=True)
+        )
+        return tuple(backup for backup in backups if not backup.original())
 
     def retire(self) -> Path:
         archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}')
@@ -226,11 +247,15 @@ def route_original(self) -> bool:
 
     @classmethod
     def take(cls, root: Path, route: Path, files: tuple[Path, ...], spill_dir: Path) -> StoreBackup:
+        registry = RegistryStore.at(root).written
+        if any(path in files for path in registry):
+            files = tuple(dict.fromkeys((*files, *registry)))
         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 +264,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 +315,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 +345,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 +360,24 @@ 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()):
+        selected = self.selected_files()
+        registry = tuple(path.relative_to(self.root) for path in RegistryStore.at(self.root).written)
+        # The registry owns this ordering: restore its snapshot before publishing
+        # the matching guard, after every unrelated selected file.
+        ordered = (
+            *(name for name in selected if name not in registry),
+            *(name for name in registry if name in selected),
+        )
+        for name in ordered:
             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 +389,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..20f4fa65a9 100644
--- a/src/agent_comms/installer/upgrade.py
+++ b/src/agent_comms/installer/upgrade.py
@@ -104,7 +104,9 @@
 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.root_runtime import RootRuntime, RootRuntimeStore
+from agent_comms.runtime_processes import RuntimeProcesses
 from agent_comms.maintenance_barrier import (
     MaintenanceBarrier,
     MaintenanceOperator,
@@ -136,7 +138,7 @@ class RehearsalFiles:
 
     identical: int
     rewritten: int
-    names: tuple[str, ...]
+    names: tuple[Annotated[Path, PathText], ...]
     restore_bytes: int
 
     @classmethod
@@ -145,7 +147,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 +157,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 +257,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)
         ))
 
@@ -402,12 +409,14 @@ class InstalledUpgrade(RehearsedUpgrade, Extendable):
     live: LiveRun
     install: InstallResult
     """The harnesses registered on the bus the target now reads."""
+    processes: RuntimeProcesses = field(kw_only=True, default=RuntimeProcesses(()))
+    """Applications still running an outgoing runtime; reported, never signalled."""
 
     finished: ClassVar[str] = "Install finished"
     """The headline of the post that reports this upgrade."""
 
     def lines(self) -> tuple[str, ...]:
-        return (*super().lines(), *self.live.lines())
+        return (*super().lines(), *self.live.lines(), *self.processes.lines())
 
 
 @dataclass(frozen=True)
@@ -543,12 +552,14 @@ def keep_native(self, native: Path, layout: Path) -> None:
     def prepare_native(self) -> None:
         self.probe(PrepareNativeProbe())
 
-    def publish(self) -> None:
-        """The selected interpreter owns console metadata and harness declarations."""
+    def publish(self, root: Path) -> None:
+        """The selected interpreter owns console metadata and harness declarations, and
+        ``root`` belongs to this stack's runtime from now on."""
         self.commands.run(
                           self.python, "-m", publication.__name__,
                           cwd=self.checkout, env=self.environment(),
         )
+        RootRuntime.publish(root, self.runtime)
 
     def environment(self, *paths: Path) -> dict:
         """Use the artifact's packages; carry steps also select the target tree's source."""
@@ -595,7 +606,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())
@@ -776,7 +787,7 @@ def publish(self, backup: StoreBackup, selected: RouteSelection) -> str:
         """The converted bus's route is ready before new runtime entrypoints open it."""
         incoming = ActiveRoute(self.root.absolute(), RecordedBusIdentity.capture(Bus.path_in(self.root)))
         route = backup.rebind_route(incoming, selected)
-        self.staged.publish()
+        self.staged.publish(self.root)
         return route
 
     @staticmethod
@@ -798,7 +809,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,
@@ -807,7 +818,8 @@ def finished(self, rehearsal: Rehearsed, bridges: SessionBridgeResumes,
         live_run = self.install(rehearsal, bridges, report)
         finished = InstalledUpgrade(
             rehearsed.source, rehearsed.target, rehearsed.root, rehearsed.pending,
-            rehearsed.rehearsal, live_run, harnesses(), commands=tuple(self.staged.commands.ran))
+            rehearsed.rehearsal, live_run, harnesses(), commands=tuple(self.staged.commands.ran),
+            processes=RuntimeProcesses.observe(self.staged.runtime))
         return (finished.extended(UnresumedUpgrade, unresumed=bridges.lines())
                 if bridges.failures else finished)
 
@@ -878,7 +890,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 +906,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(
@@ -933,7 +945,7 @@ def rehearsal(self, practice: PracticeBus) -> CarriedRehearsal:
                 RootFile.content_owner(source).copy_file(source, destination)
         self.installed.running_owners(copied)
         maintenance = MaintenanceBarrier(RegistryStore.at(copied).path)
-        hold = maintenance.begin(MaintenanceOperator.current())
+        hold = maintenance.retarget(maintenance.begin(MaintenanceOperator.current()), stack.runtime)
         with hold.holding():
             migration = PreparedMigration.capture(hold, self.installed)
             passed = migration.carry(pending, stack, self.tree)
@@ -946,7 +958,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(
@@ -954,6 +966,7 @@ def live(self, rehearsal: RehearsedFiles, report: InstallReport) -> CarriedRun:
             sum(path.stat().st_size for path in selected if path.exists()),
         ), self.root.parent)
         restore.require()
+        RootRuntimeStore.at(self.root).adopt(self.installed.runtime)
         running = self.installed.running_owners(self.root)
         maintenance = MaintenanceBarrier(RegistryStore.at(self.root).path)
         held = maintenance.begin(MaintenanceOperator.current())
@@ -962,6 +975,9 @@ def live(self, rehearsal: RehearsedFiles, report: InstallReport) -> CarriedRun:
         try:
             try:
                 self.installed.owners(self.root, StopCliCommand, running)
+                held = maintenance.retarget(held, self.staged.runtime)
+                holder.close()
+                holder.enter_context(held.holding())
                 planned = self.staged.cli(self.root, PlannedReconcileCliCommand)
                 unrehearsed = sorted(frozenset(planned.rewritten) - frozenset(rehearsal.files.names))
                 if unrehearsed:
@@ -1003,7 +1019,7 @@ def live(self, rehearsal: RehearsedFiles, report: InstallReport) -> CarriedRun:
                 if head != self.current:
                     self.installed.commands.run("git", "-C", self.checkout, "reset", "-q", "--keep", self.current)
                     Checkout(self.checkout).prepare()
-                self.installed.publish()
+                self.installed.publish(self.root)
                 holder.close()
                 maintenance.clear(restored_hold)
                 started = self.installed.start(self.root, running)
@@ -1042,7 +1058,7 @@ def uncopied(self) -> UnchangedRehearsal:
         return UnchangedRehearsal()
 
     def live(self, rehearsal: UnchangedRehearsal, report: InstallReport) -> UnchangedRun:
-        self.installed.publish()
+        self.installed.publish(self.root)
         report.writer = RuntimeWriter(self.root, self.installed, InstalledProgram())
         return UnchangedRun()
 
@@ -1082,7 +1098,7 @@ def live(self, rehearsal: FreshRehearsal, report: InstallReport) -> CreatedRun:
             services = HostService.restart_running(self.staged.python)
         except BaseException:
             backup.restore_route()
-            self.installed.publish()
+            self.installed.publish(self.root)
             raise
         backup.retain(self.spill_dir)
         return CreatedRun(services, route)
diff --git a/src/agent_comms/launch_runtime.py b/src/agent_comms/launch_runtime.py
index dc39d0c93e..5280e6448e 100644
--- a/src/agent_comms/launch_runtime.py
+++ b/src/agent_comms/launch_runtime.py
@@ -26,7 +26,15 @@
     WireOmitDefault,
 )
 from agent_comms.launch_root import LaunchRoot
-from agent_comms.thread_text import AbsentText, GivenText, NoParent, NoThreadText, TextChoice, ThreadParent, ThreadText
+from agent_comms.thread_text import (
+    AbsentText,
+    GivenText,
+    NoParent,
+    NoThreadText,
+    TextChoice,
+    ThreadParent,
+    ThreadText,
+)
 
 if TYPE_CHECKING:
     from agent_comms.threads import Thread
@@ -62,10 +70,15 @@ class LaunchRuntime:
 
     interpreter: Annotated[Path, PathText]
 
+    @classmethod
+    def from_interpreter(cls, interpreter: str | Path) -> Self:
+        """Keep the invoked interpreter: resolving its symlink loses the runtime."""
+        return cls(Path(interpreter))
+
     @classmethod
     def running(cls) -> LaunchRuntime:
         """Keep ordinary launches on the calling process's installation."""
-        return cls(Path(sys.executable))
+        return cls.from_interpreter(sys.executable)
 
     def child_environment(self, environment: Mapping[str, str]) -> dict[str, str]:
         """A child of this process's own interpreter imports the source this process
@@ -239,9 +252,7 @@ class RestartEnvironment(EnvironmentRecord):
     stay in /proc and are read only from the exact selected process at execution.
     """
 
-    home: Annotated[
-        EnvironmentValue, WireOmitDefault(), WireName("HOME")
-    ] = UnsetVariable()
+    home: Annotated[EnvironmentValue, WireOmitDefault(), WireName("HOME")] = UnsetVariable()
     path: Annotated[
         EnvironmentValue,
         WireOmitDefault(),
@@ -260,9 +271,9 @@ class RestartEnvironment(EnvironmentRecord):
         WireName("VIRTUAL_ENV"),
         RuntimeScope(),
     ] = UnsetVariable()
-    config: Annotated[
-        EnvironmentValue, WireOmitDefault(), WireName("XDG_CONFIG_HOME")
-    ] = UnsetVariable()
+    config: Annotated[EnvironmentValue, WireOmitDefault(), WireName("XDG_CONFIG_HOME")] = (
+        UnsetVariable()
+    )
     default_agent_directory: ClassVar[Path] = Path("~/.pi/agent")
     native_config: Annotated[
         Path,
@@ -270,9 +281,9 @@ class RestartEnvironment(EnvironmentRecord):
         WireName("AGENT_COMMS_NATIVE_CONFIG_DIR"),
         NativeScope(),
     ] = default_agent_directory
-    agent_directory: Annotated[
-        Path, PathText, WireName("PI_CODING_AGENT_DIR"), NativeScope()
-    ] = default_agent_directory
+    agent_directory: Annotated[Path, PathText, WireName("PI_CODING_AGENT_DIR"), NativeScope()] = (
+        default_agent_directory
+    )
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "native_config", self.absolute(self.native_config))
@@ -384,12 +395,8 @@ class ThreadLaunch(EnvironmentRecord):
         WireOmitDefault(),
         WireName("AGENT_COMMS_TAGS"),
     ] = frozenset()
-    parent: Annotated[
-        ThreadParent, WireOmitDefault(), WireName("PI_PARENT_ID")
-    ] = NoParent()
-    task: Annotated[
-        ThreadText, WireOmitDefault(), WireName("PI_TASK")
-    ] = NoThreadText()
+    parent: Annotated[ThreadParent, WireOmitDefault(), WireName("PI_PARENT_ID")] = NoParent()
+    task: Annotated[ThreadText, WireOmitDefault(), WireName("PI_TASK")] = NoThreadText()
     managed: Annotated[
         bool,
         FlagText,
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/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py
index cf7ed02035..2f9528f1d9 100644
--- a/src/agent_comms/maintenance_barrier.py
+++ b/src/agent_comms/maintenance_barrier.py
@@ -14,14 +14,23 @@
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterator, Callable, Iterator
 from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, field, replace
 from pathlib import Path
-from typing import Literal, Self
+from typing import Annotated, Literal, Self
 
 from .child_process import Platform, ProcessIdentity
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError, ToolRefusedError
-from .field_codec import FieldCodec, Id128Text, MalformedRecord
+from .field_codec import (
+    FieldCodec,
+    Id128Text,
+    MalformedRecord,
+    NullMember,
+    RecordMember,
+    SubclassUnion,
+    WireOmitDefault,
+)
+from .launch_runtime import LaunchRuntime
 from .lifecycle import LifecycleState
 from .locked_store import DurableFile, LockedStore, RegistryFile
 from .send_outcome import GrantedSend, HeldSend
@@ -79,9 +88,32 @@ def require_open():
         pass
 
 
+class MaintenanceTarget(SubclassUnion):
+    """A hold without a target grants no additional runtime attachment."""
+
+    @abstractmethod
+    def allows_runtime(self, runtime: LaunchRuntime) -> bool: ...
+
+
+@dataclass(frozen=True)
+class NoMaintenanceTarget(NullMember, MaintenanceTarget):
+    def allows_runtime(self, runtime: LaunchRuntime) -> bool:
+        return False
+
+
+@dataclass(frozen=True)
+class TargetRuntime(LaunchRuntime, RecordMember, MaintenanceTarget):
+    def allows_runtime(self, runtime: LaunchRuntime) -> bool:
+        return self.prefix == runtime.prefix
+
+
 class MaintenancePhase(DeclaredFamily, LifecycleState, affix="Phase"):
     new_sends = running_sends = HeldSends
 
+    @classmethod
+    def target_after(cls, target: MaintenanceTarget) -> MaintenanceTarget:
+        return target
+
     @classmethod
     def holder_dead(cls, state: MaintenanceState) -> bool:
         return not state.operator.alive()
@@ -147,6 +179,10 @@ class ReadyPhase(MaintenancePhase):
     def require_held_by(cls, state: MaintenanceState, holder: str, change: str) -> None:
         raise MaintenanceNotHeldError.unheld(change)
 
+    @classmethod
+    def target_after(cls, target: MaintenanceTarget) -> MaintenanceTarget:
+        return NoMaintenanceTarget()
+
     @classmethod
     def holder_dead(cls, state: MaintenanceState) -> bool:
         return False
@@ -187,6 +223,9 @@ class MaintenanceStatus:
     def turn_pending(self) -> bool:
         return self.phase.new_sends.pending()
 
+    def allows_runtime(self, runtime: LaunchRuntime) -> bool:
+        return False
+
     def observed(self, barrier: MaintenanceBarrier) -> MaintenanceStatus:
         return self
 
@@ -234,6 +273,12 @@ class MaintenanceState(MaintenanceMarker, MaintenanceStatus):
     operator: MaintenanceOperator
     nonce: str
     phase: type[MaintenancePhase]
+    target: Annotated[MaintenanceTarget, WireOmitDefault()] = field(
+        default_factory=NoMaintenanceTarget
+    )
+
+    def allows_runtime(self, runtime: LaunchRuntime) -> bool:
+        return self.target.allows_runtime(runtime)
 
     def __post_init__(self):
         super().__post_init__()
@@ -277,9 +322,15 @@ def holding(self) -> Iterator[Self]:
     def successor(self, phase: type[MaintenancePhase], nonce: str) -> MaintenanceState:
         if not self.phase().may_become(phase()):
             raise ValueError('Maintenance transition requires a valid expected successor')
-        return MaintenanceState(version=self.version, root=self.root,
-                                generation=self.generation + 1, operator=self.operator,
-                                nonce=nonce, phase=phase)
+        return self.in_phase(phase, nonce)
+
+    def in_phase(self, phase: type[MaintenancePhase], nonce: str) -> MaintenanceState:
+        """Ordinary and dead-holder transitions share the phase's target lifetime."""
+        return self.advanced(nonce, phase=phase, target=phase.target_after(self.target))
+
+    def advanced(self, nonce: str, **changes: object) -> MaintenanceState:
+        """Every rewrite of a hold takes the next generation and a fresh nonce."""
+        return replace(self, generation=self.generation + 1, nonce=nonce, **changes)
 
     def require_open_to(self, holder: str) -> None:
         if holder != self.nonce:
@@ -372,10 +423,7 @@ def _release_dead_unlocked(self, expected: MaintenanceState) -> MaintenanceStatu
         current = self._current_unlocked()
         if current != expected:
             return current
-        self._write_unlocked(replace(
-            expected, generation=expected.generation + 1, nonce=Id128Text.mint(),
-            phase=DeadHolderReleasedPhase,
-        ))
+        self._write_unlocked(expected.in_phase(DeadHolderReleasedPhase, Id128Text.mint()))
         raise MaintenanceHeldError(
             f"Maintenance released dead holder pid={expected.operator.pid} "
             f"start_time={expected.operator.start_time}; ingress reopened"
@@ -412,14 +460,25 @@ def _write_unlocked(self, hold: MaintenanceState) -> None:
     def begin(self, operator: MaintenanceOperator) -> MaintenanceState:
         """Hold new turns before fleet drain; an uncertain write stays closed."""
         with _store_lock(self.registry_path):
-            current = self.current_unlocked()
-            current.require_open()
+            current = self._current_unlocked()
+            # A successor takes an abandoned hold without briefly reopening
+            # ingress or making the owner repeat its install command.
+            if not current.phase.holder_dead(current):
+                current.require_open()
             hold = MaintenanceState(version=2, root=str(self.registry_path.parent.resolve()),
                                     generation=current.generation + 1, operator=operator,
                                     nonce=Id128Text.mint(), phase=DrainingPhase)
             self._write_unlocked(hold)
             return hold
 
+    def retarget(self, expected: MaintenanceState, runtime: LaunchRuntime) -> MaintenanceState:
+        """Transfer a drained hold after its outgoing readers have stopped."""
+        with _store_lock(self.registry_path):
+            self.require_draining_unlocked(expected)
+            hold = expected.advanced(Id128Text.mint(), target=TargetRuntime(runtime.interpreter))
+            self._write_unlocked(hold)
+            return hold
+
     def _advance_unlocked(self, expected: MaintenanceState,
                           phase: type[MaintenancePhase]) -> MaintenanceState:
         if self.current_unlocked() != expected:
@@ -443,7 +502,7 @@ def require_draining_unlocked(self, expected: MaintenanceState) -> None:
         expected.require_draining()
         if self.current_unlocked() != expected:
             raise RelationViolationError(
-                'Cutover maintenance phase changed before owner retirement'
+                'Maintenance generation/operator changed'
             )
 
     @contextmanager
diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py
index 14f46482f7..6b750dd0ef 100644
--- a/src/agent_comms/plan_phases.py
+++ b/src/agent_comms/plan_phases.py
@@ -406,17 +406,19 @@ class DuplicatesPhase(MultipleInheritancePhase):
                     ("__init__",),
                     ("durable_turn.py", "execution_store.py", "exporting.py", "publisher.py"),
                 ),
-                (("wire_member",), ("pi_payloads.py",)),
+                (("wire_member",), ("pi_payloads.py", "field_codec.py")),
                 (
                     (
                         "_assert_response_schema",
                         "assert_cohort_schema",
                         "assert_native_runtime_schema",
+                        "require_schema",
                     ),
                     (
                         "cohort_schema.py",
                         "coordinated_runtime_schema.py",
                         "coordination_response.py",
+                        "typed_table.py",
                     ),
                 ),
                 (
@@ -429,7 +431,7 @@ class DuplicatesPhase(MultipleInheritancePhase):
                 ),
                 (
                     ("annotation_disclosure", "disclosure_for"),
-                    ("pi_payloads.py", "turn_context.py"),
+                    ("pi_payloads.py", "turn_context.py", "annotation_disclosure.py"),
                 ),
             )
         )
diff --git a/src/agent_comms/root_runtime.py b/src/agent_comms/root_runtime.py
new file mode 100644
index 0000000000..7c25157e44
--- /dev/null
+++ b/src/agent_comms/root_runtime.py
@@ -0,0 +1,65 @@
+"""A bus root belongs to exactly one published runtime."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Annotated
+
+from .errors import RelationViolationError
+from .field_codec import PathText
+from .launch_runtime import LaunchRuntime
+from .locked_store import DurableFile, LockedStore, RegistryFile
+from .maintenance_barrier import MaintenanceBarrier
+
+
+class ForeignRuntimeError(RelationViolationError):
+    """The caller is neither this root's owner nor its maintenance target."""
+
+
+@dataclass(frozen=True)
+class RootRuntime:
+    prefix: Annotated[Path, PathText]
+
+    @classmethod
+    def require_attachable(cls, root: Path, caller: LaunchRuntime) -> None:
+        store = RootRuntimeStore.at(root)
+        owner = store.owner(caller)
+        if owner.prefix == caller.prefix:
+            return
+        if MaintenanceBarrier(RegistryFile.path_in(root)).read().allows_runtime(caller):
+            return
+        raise ForeignRuntimeError(
+            f"Root {root} belongs to runtime {owner.prefix}; "
+            f"foreign runtime {caller.prefix} cannot attach"
+        )
+
+    @classmethod
+    def publish(cls, root: Path, runtime: LaunchRuntime) -> None:
+        RootRuntimeStore.at(root).replace(cls(runtime.prefix))
+
+
+@dataclass(frozen=True)
+class RootRuntimeStore(LockedStore[RootRuntime], DurableFile):
+    filename = ".root-runtime.json"
+    record_type = RootRuntime
+
+    def __post_init__(self) -> None:
+        # This is the first file on a new root, before publisher preparation.
+        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+
+    def empty(self) -> RootRuntime:
+        raise FileNotFoundError(self.path)
+
+    def owner(self, creator: LaunchRuntime) -> RootRuntime:
+        try:
+            return self.read()
+        except FileNotFoundError:
+            return self.adopt(creator)
+
+    def adopt(self, runtime: LaunchRuntime) -> RootRuntime:
+        """A root without a runtime record belongs to ``runtime``; a recorded owner stays."""
+        with self.locked():
+            if not self.path.exists():
+                self._publish_unlocked(RootRuntime(runtime.prefix))
+            return self._read_unlocked()
diff --git a/src/agent_comms/runtime_processes.py b/src/agent_comms/runtime_processes.py
new file mode 100644
index 0000000000..c030333768
--- /dev/null
+++ b/src/agent_comms/runtime_processes.py
@@ -0,0 +1,115 @@
+"""Report old-runtime applications without acquiring authority to stop them."""
+
+from __future__ import annotations
+
+import os
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from pathlib import Path
+
+from agent_comms.child_process import IdentityMismatchError, Platform, ProcessIdentity
+from agent_comms.declared_family import DeclaredFamily
+from agent_comms.launch_runtime import LaunchRuntime
+from agent_comms.runtime_children import ApplicationChildInitiator
+
+
+class RuntimeProcessOutcome(DeclaredFamily, ABC, affix="RuntimeProcess"):
+    @abstractmethod
+    def line(self) -> str: ...
+
+
+@dataclass(frozen=True)
+class UnreadableRuntimeProcess(RuntimeProcessOutcome):
+    pid: int
+    reason: str
+
+    def line(self) -> str:
+        return f"Unknown runtime PID {self.pid}: {self.reason}"
+
+
+@dataclass(frozen=True)
+class RuntimeApplication(ApplicationChildInitiator):
+    """The observed parent application's exact identity and display name."""
+
+    name: str
+
+    def describe(self) -> str:
+        return f"{self.name} (application PID {self.process.pid})"
+
+
+@dataclass(frozen=True, eq=False)
+class RuntimeProcess(ProcessIdentity, LaunchRuntime, declared_name=None):
+    application: RuntimeApplication
+
+
+@dataclass(frozen=True)
+class UnretiredRuntimeProcess(RuntimeProcessOutcome):
+    process: RuntimeProcess
+
+    def line(self) -> str:
+        return (
+            f"Unretired old-runtime PID {self.process.pid}: {self.process.prefix}; "
+            f"owner {self.process.application.describe()}; not signalled"
+        )
+
+
+@dataclass(frozen=True)
+class RuntimeProcesses:
+    """A read-only snapshot, not a process handle or permission to signal an app.
+
+    Installer-launched children retain their own ChildProcess stop/reap authority.
+    Discovering another application's interpreter grants no such authority.
+    """
+
+    outcomes: tuple[RuntimeProcessOutcome, ...]
+
+    def lines(self) -> tuple[str, ...]:
+        return tuple(outcome.line() for outcome in self.outcomes)
+
+    @classmethod
+    def observe(cls, target: LaunchRuntime, proc: Path = Path("/proc")) -> RuntimeProcesses:
+        directory = target.prefix.parent
+        outcomes: list[RuntimeProcessOutcome] = []
+        platform = Platform.current()
+        for entry in proc.iterdir():
+            if not entry.name.isdecimal():
+                continue
+            pid = int(entry.name)
+            try:
+                identity = ProcessIdentity.capture(pid)
+                invoked = (entry / "cmdline").read_bytes().partition(b"\0")[0]
+                identity.require_alive()
+            except ProcessLookupError, FileNotFoundError:
+                continue
+            except (OSError, IdentityMismatchError) as error:
+                outcomes.append(UnreadableRuntimeProcess(pid, str(error)))
+                continue
+            # Empty command lines declare no userspace invocation (kernel/exited
+            # tasks). Other namespaces do not name an agent-comms installation.
+            interpreter = Path(os.fsdecode(invoked))
+            if not invoked or not interpreter.is_relative_to(directory):
+                continue
+            runtime = LaunchRuntime.from_interpreter(interpreter)
+            if runtime.prefix == target.prefix:
+                continue
+            try:
+                parent = platform.parent_identity(pid)
+                application = RuntimeApplication(
+                    parent, (proc / str(parent.pid) / "comm").read_text().strip()
+                )
+                parent.require_alive()
+                identity.require_alive()
+            except (OSError, IdentityMismatchError) as error:
+                outcomes.append(UnreadableRuntimeProcess(pid, str(error)))
+                continue
+            outcomes.append(
+                UnretiredRuntimeProcess(
+                    RuntimeProcess(
+                        pid=identity.pid,
+                        start_time=identity.start_time,
+                        interpreter=runtime.interpreter,
+                        application=application,
+                    )
+                )
+            )
+        return cls(tuple(outcomes))
diff --git a/src/agent_comms/store_order.py b/src/agent_comms/store_order.py
index 19c3bce3e2..5c0f8f9b82 100644
--- a/src/agent_comms/store_order.py
+++ b/src/agent_comms/store_order.py
@@ -5,6 +5,7 @@
 from .relationships import RelationshipStore
 from .runtime_info import RuntimeInfoStore
 from .runtime_children import RuntimeChildrenStore
+from .root_runtime import RootRuntimeStore
 from .shared_ledger import SharedLedger
 from .collaboration_ledger import CollaborationLedger
 from .connector_store import ConnectorStore
@@ -14,7 +15,7 @@
 STORE_LOCK_ORDER = (
     RegistryStore, ChannelCatalog, ReadLedger,
     RelationshipStore, RuntimeInfoStore, SharedLedger, CollaborationLedger, ConnectorStore,
-    RuntimeChildrenStore, ServicePassStore,
+    RuntimeChildrenStore, RootRuntimeStore, ServicePassStore,
 )
 
 StoreLockOwner.declare_lock_order(STORE_LOCK_ORDER)
diff --git a/src/agent_comms/turn_context.py b/src/agent_comms/turn_context.py
index aa73437dee..e37539e8e8 100644
--- a/src/agent_comms/turn_context.py
+++ b/src/agent_comms/turn_context.py
@@ -933,6 +933,13 @@ def source_ranges(self) -> tuple[ContributionCoordinates, ...]:
             offset += len(text.encode("utf-8")) + 1
         return tuple(ranges)
 
+    def render_into(self, prompt_parts, provider):
+        provider.setdefault("messages", []).extend(self.messages)
+
+
+class NativeAnnotations(NativeMessages):
+    """A transcript span keeps the disclosure policy of its original native message."""
+
     def disclosure_for(self, span: ContextSpan):
         from .pi_payloads import PiMessage
 
@@ -941,12 +948,9 @@ def disclosure_for(self, span: ContextSpan):
                 return PiMessage.from_wire(message).disclosure_for(span)
         raise ValueError("Annotation names no original native message")
 
-    def render_into(self, prompt_parts, provider):
-        provider.setdefault("messages", []).extend(self.messages)
-
 
 @dataclass(frozen=True, kw_only=True)
-class TranscriptSegment(NativeMessages, MeasuredNativeSegment):
+class TranscriptSegment(NativeAnnotations, MeasuredNativeSegment):
     keyframe_interval: ClassVar[int] = 32
 
     @classmethod
@@ -975,7 +979,7 @@ class CompactionSummarySegment(PublicInstructionAnnotations, NativeMessages, Mea
 
 
 @dataclass(frozen=True, kw_only=True)
-class InjectionMessageSegment(NativeMessages, MeasuredNativeSegment):
+class InjectionMessageSegment(NativeAnnotations, MeasuredNativeSegment):
     pass
 
 
diff --git a/tests/test_carry.py b/tests/test_carry.py
index efdfcee76b..e1e57b5c5c 100644
--- a/tests/test_carry.py
+++ b/tests/test_carry.py
@@ -179,11 +179,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
@@ -194,7 +194,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(
@@ -206,10 +206,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/tests/test_interrupted_rollback.py b/tests/test_interrupted_rollback.py
new file mode 100644
index 0000000000..adb3d8fd86
--- /dev/null
+++ b/tests/test_interrupted_rollback.py
@@ -0,0 +1,85 @@
+"""A killed restore must be completed by the next install, before it reads the bus."""
+
+import os
+import signal
+import subprocess
+import sys
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from agent_comms.comms import Comms
+from agent_comms.active_route import active_route_path
+from agent_comms.install_choices import RehearsalMode, UnnamedInstaller
+from agent_comms.installer.backup import StoreBackup, UpgradeRefused
+from agent_comms.installer.carry import Carried, Carry
+from agent_comms.installer.launch import InstallReport
+from agent_comms.installer.upgrade import upgrade
+from agent_comms.registry_store import RegistryStore
+from bus_record_fixture import prepare_native_comms
+
+
+class RegistryTitleCarry(Carry):
+    def carry(self, *, apply):
+        registry = RegistryStore.at(self.root)
+        if apply:
+            with registry.editing() as edit:
+                thread = edit.document.require("worker")
+                edit.document.threads[thread.name] = replace(thread, title="Carried title")
+        return Carried(1, rewritten=tuple(str(path.relative_to(self.root)) for path in registry.written))
+
+
+KILLED_RESTORE = """
+import os, signal, sys
+from pathlib import Path
+from agent_comms.installer.backup import StoreBackup
+from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator
+from agent_comms.registry_store import RegistryStore
+
+root, route, path = map(Path, sys.argv[1:4])
+registry = RegistryStore.at(root)
+publication = registry.written[int(sys.argv[4])]
+original_replace = os.replace
+def killed_publication(source, target, **directories):
+    original_replace(source, target, **directories)
+    if Path(target).name == publication.name:
+        os.kill(os.getpid(), signal.SIGKILL)
+os.replace = killed_publication
+hold = MaintenanceBarrier(registry.path).begin(MaintenanceOperator.current())
+with hold.holding():
+    StoreBackup(root, route, path).restore()
+raise AssertionError("The selected registry publication was never restored")
+"""
+
+
+@pytest.mark.skipif(os.name != "posix", reason="SIGKILL and private registry require POSIX")
+@pytest.mark.parametrize("publication", (0, 1), ids=("after-registry", "after-guard"))
+def test_next_install_completes_a_killed_registry_restore(tmp_path, monkeypatch, publication):
+    # HOME selects only this test's route. Neither process can replace the developer's route.
+    monkeypatch.setenv("HOME", str(tmp_path / "home"))
+    comms = prepare_native_comms(Comms(tmp_path / "wire"))
+    comms.registry.declare(comms.registry.new_thread("worker", frozenset(), str(tmp_path)))
+    registry = RegistryStore.at(comms.root)
+    route = active_route_path()
+    route.parent.mkdir(mode=0o700, parents=True)
+    spill = tmp_path / "backups"
+    # Put the guard first, as the interrupted historical backup did. Restore ordering
+    # must come from the registry declaration, not from this manifest's ordering.
+    backup = StoreBackup.take(comms.root, route, tuple(reversed(registry.written)), spill)
+    expected = {path: StoreBackup.digest(path) for path in registry.written}
+    RegistryTitleCarry(comms.root, comms.root, spill).carry(apply=True)
+    killed = subprocess.run(
+        [sys.executable, "-c", KILLED_RESTORE, str(comms.root), str(route), str(backup.path), str(publication)],
+        capture_output=True, text=True, timeout=30,
+    )
+    assert killed.returncode == -signal.SIGKILL, killed.stdout + killed.stderr
+    # Enter the real installer. Its next Git command deliberately fails: rollback
+    # recovery must finish even before the checkout can be inspected or a stack built.
+    with pytest.raises(UpgradeRefused, match="git failed"):
+        upgrade(
+            tmp_path / "missing-checkout", comms.root, "HEAD", UnnamedInstaller(), RehearsalMode(),
+            False, spill, tmp_path / "install-scratch", lambda: None, InstallReport(),
+        )
+    assert {path: StoreBackup.digest(path) for path in registry.written} == expected
+    assert comms.registry.require("worker").title != "Carried title"
diff --git a/tests/test_root_runtime.py b/tests/test_root_runtime.py
new file mode 100644
index 0000000000..374260ad22
--- /dev/null
+++ b/tests/test_root_runtime.py
@@ -0,0 +1,67 @@
+"""Root attachment and publication use one persisted runtime owner."""
+
+import pytest
+
+from agent_comms.comms import Comms
+from agent_comms.errors import RelationViolationError
+from agent_comms.launch_runtime import LaunchRuntime
+from agent_comms.locked_store import RegistryFile
+from agent_comms.maintenance_barrier import (
+    MaintenanceBarrier,
+    MaintenanceOperator,
+    NoMaintenanceTarget,
+)
+from agent_comms.root_runtime import ForeignRuntimeError, RootRuntime, RootRuntimeStore
+from agent_comms.bus import Bus
+
+
+def test_root_runtime_attachment_family(tmp_path):
+    owner, target, foreign = (
+        LaunchRuntime(tmp_path / name / "bin/python")
+        for name in ("owner", "target", "foreign")
+    )
+    root = tmp_path / "bus"
+    RootRuntime.require_attachable(root, owner)
+    store = RootRuntimeStore.at(root)
+    assert store.read() == RootRuntime(owner.prefix)
+    RootRuntime.require_attachable(root, owner)
+    with pytest.raises(ForeignRuntimeError) as refused:
+        RootRuntime.require_attachable(root, foreign)
+    assert str(owner.prefix) in str(refused.value)
+    assert str(foreign.prefix) in str(refused.value)
+
+    barrier = MaintenanceBarrier(RegistryFile.path_in(root))
+    original = barrier.begin(MaintenanceOperator.current())
+    assert not original.allows_runtime(target)
+    with pytest.raises(ForeignRuntimeError):
+        RootRuntime.require_attachable(root, target)
+    held = barrier.retarget(original, target)
+    assert held.generation == original.generation + 1
+    with pytest.raises(RelationViolationError):
+        barrier.retarget(original, foreign)
+    assert not NoMaintenanceTarget().allows_runtime(foreign)
+    RootRuntime.require_attachable(root, target)
+    RootRuntime.require_attachable(root, owner)
+    with pytest.raises(ForeignRuntimeError):
+        RootRuntime.require_attachable(root, foreign)
+    cleared = barrier.clear(held)
+    assert cleared.target == NoMaintenanceTarget()
+    with pytest.raises(ForeignRuntimeError):
+        RootRuntime.require_attachable(root, target)
+    RootRuntime.publish(root, target)
+    RootRuntime.require_attachable(root, target)
+    with pytest.raises(ForeignRuntimeError):
+        RootRuntime.require_attachable(root, owner)
+
+
+def test_comms_creates_a_private_root_and_refuses_foreign_stores(tmp_path):
+    created = Comms(tmp_path / "created")
+    assert RootRuntimeStore.at(created.root).read().prefix == LaunchRuntime.running().prefix
+
+    root = tmp_path / "bus"
+    other = LaunchRuntime(tmp_path / "other/bin/python")
+    RootRuntime.publish(root, other)
+    with pytest.raises(ForeignRuntimeError):
+        Comms(root)
+    assert not RegistryFile.path_in(root).exists()
+    assert not Bus.path_in(root).exists()
diff --git a/tests/test_runtime_processes.py b/tests/test_runtime_processes.py
new file mode 100644
index 0000000000..c3d067faad
--- /dev/null
+++ b/tests/test_runtime_processes.py
@@ -0,0 +1,68 @@
+"""A fake runtimes tree with real applications; observations never signal them."""
+
+import os
+import subprocess
+import sys
+from contextlib import ExitStack
+from pathlib import Path
+
+from agent_comms.child_process import ProcessIdentity
+from agent_comms.field_codec import FieldCodec
+from agent_comms.launch_runtime import LaunchRuntime
+from agent_comms.runtime_processes import (
+    RuntimeProcessOutcome,
+    RuntimeProcesses,
+    UnreadableRuntimeProcess,
+    UnretiredRuntimeProcess,
+)
+
+
+def stop(child):
+    if child.poll() is None:
+        child.terminate()
+    child.wait(timeout=10)
+
+
+def test_runtime_process_family(tmp_path):
+    runtimes = tmp_path / "runtimes"
+    old, target = (LaunchRuntime(runtimes / name / "bin/python") for name in ("old", "target"))
+    for runtime in (old, target):
+        runtime.interpreter.parent.mkdir(parents=True)
+        runtime.interpreter.symlink_to(sys.executable)
+    with ExitStack() as cleanup:
+        children = []
+        for runtime in (old, target):
+            child = subprocess.Popen([str(runtime.interpreter), "-c", "import time; time.sleep(120)"])
+            cleanup.callback(stop, child)
+            children.append(child)
+        snapshot = RuntimeProcesses.observe(target)
+        (outcome,) = snapshot.outcomes
+        assert isinstance(outcome, UnretiredRuntimeProcess)
+        assert outcome.process == ProcessIdentity.capture(children[0].pid)
+        assert outcome.process.prefix == old.prefix
+        assert outcome.process.application.process == ProcessIdentity.capture(os.getpid())
+        assert outcome.process.application.name == Path(f"/proc/{os.getpid()}/comm").read_text().strip()
+        # Resolving the interpreter would collapse both runtimes onto the shared Python.
+        assert old.interpreter.resolve() == target.interpreter.resolve()
+        assert str(children[0].pid) in outcome.line()
+        assert outcome.process.application.describe() in outcome.line()
+        # Neither discovery nor rendering gains the right to terminate an application.
+        assert all(child.poll() is None for child in children)
+        decoded = FieldCodec.decode(RuntimeProcesses, FieldCodec.encode(snapshot))
+        assert decoded == snapshot
+        assert decoded.lines() == (outcome.line(),)
+        assert FieldCodec.decode(RuntimeProcessOutcome, FieldCodec.encode(outcome)) == outcome
+        stop(children[0])  # Only its actual launcher closes this child.
+        assert not RuntimeProcesses.observe(target).outcomes
+        assert children[1].poll() is None
+
+    # A named process whose command line cannot be read is not silently discarded.
+    proc = tmp_path / "proc"
+    entry = proc / str(os.getpid())
+    (entry / "cmdline").mkdir(parents=True)
+    (unknown,) = RuntimeProcesses.observe(target, proc).outcomes
+    assert isinstance(unknown, UnreadableRuntimeProcess)
+    assert unknown.pid == os.getpid()
+    assert str(unknown.pid) in unknown.line()
+    assert unknown.reason
+    assert FieldCodec.decode(RuntimeProcessOutcome, FieldCodec.encode(unknown)) == unknown
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 32eb055c62..5f2b3c0215 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),
@@ -101,8 +101,8 @@ def carry_root(self, root: Path, *, apply: bool) -> Carried:
             raw = json.dumps(document)
             # Later carries still own other predecessor fields, including goal
             # orders. A current registry decode belongs after those carries.
-            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))