diff --git a/src/agent_comms/assignment_states.py b/src/agent_comms/assignment_states.py
index 8af56a6cbd..d0212f04b8 100644
--- a/src/agent_comms/assignment_states.py
+++ b/src/agent_comms/assignment_states.py
@@ -235,7 +235,7 @@ def notification(
         sender_lookup: str,
         *,
         observation: RecipientActivity,
-        activity: AssignmentActivity | None = None,
+        activity: AssignmentActivity,
         updated_at_ms: int = 0,
         triage_inflight: bool = False,
     ) -> MessageNotification:
@@ -396,7 +396,7 @@ def notification(
         sender_lookup: str,
         *,
         observation: RecipientActivity,
-        activity: AssignmentActivity | None,
+        activity: AssignmentActivity,
         updated_at_ms: int = 0,
         triage_inflight: bool = False,
     ) -> MessageNotification:
@@ -474,7 +474,7 @@ def notification(
         sender_lookup: str,
         *,
         observation: RecipientActivity,
-        activity: AssignmentActivity | None,
+        activity: AssignmentActivity,
         updated_at_ms: int = 0,
         triage_inflight: bool = False,
     ) -> MessageNotification:
diff --git a/src/agent_comms/field_codec.py b/src/agent_comms/field_codec.py
index f5fce8564e..ccf85d6216 100644
--- a/src/agent_comms/field_codec.py
+++ b/src/agent_comms/field_codec.py
@@ -1010,7 +1010,7 @@ def argument_options(cls) -> dict[str, object]:
 class DeclaredWireType(ExternalJsonShape, WireValue):
     """A member whose whole wire value is one declared JSON type, converted by its own
     from_wire and to_wire. Its shape and schema derive from that type; inherited from a
-    union, its shape would be the union holding it, and admitting would never return."""
+    union, its shape would be the union holding it, and the shape check would never return."""
 
     wire_type: ClassVar[object]
 
@@ -1949,16 +1949,16 @@ def encode(self, value: T) -> object: ...
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return ()
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         """Check outer wire shape without decoding nested values.
 
-        This may admit a value decoding rejects, but must never reject one decoding accepts.
+        This may pass a value decoding rejects, but must never reject one decoding accepts.
         """
         return True
 
-    def admits_external(self, data: object) -> bool:
+    def fits_external(self, data: object) -> bool:
         """A sibling-aware external reader may accept more than the stored wire shape."""
-        return any(self.external_decoders(data)) or self.admits_wire(data)
+        return any(self.external_decoders(data)) or self.fits_wire(data)
 
     def selects_record(self, select: WireRecordFilter) -> bool:
         return False
@@ -2110,7 +2110,7 @@ class _ExactScalarDecoder(_ScalarDecoder):
     def accepts(self, data: object) -> bool:
         return type(data) is self.target
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return self.accepts(data)
 
     def accepts_value(self, value: object) -> bool:
@@ -2158,11 +2158,11 @@ class _RealDecoder(_NumberDecoder):
     __slots__ = ()
     annotations = (float,)
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return type(data) in (int, float)
 
     def accepts(self, data: object) -> bool:
-        return self.admits_wire(data) and math.isfinite(cast(float, data))
+        return self.fits_wire(data) and math.isfinite(cast(float, data))
 
 
 @dataclass(frozen=True, slots=True)
@@ -2180,9 +2180,15 @@ def wire_structure(self) -> ValueDecoder[object]:
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return (self.wire_structure,)
 
-    def admits_wire(self, data: object) -> bool:
+    @boundary_decoder
+    def fits_wire(self, data: object) -> bool:
         # Null handlers can change the structural null contract; leave that decision to decode.
-        return data is None or self.wire_structure.admits_wire(data)
+        return data is None or self.wire_structure.fits_wire(data)
+
+    def fits_external(self, data: object) -> bool:
+        # The representation reads through its wire structure, so an external read takes
+        # whatever that structure's external reader takes.
+        return ValueDecoder.fits_external(self, data) or self.wire_structure.fits_external(data)
 
     def _rewrite_records(
         self, data: object, change: WireRecordChange, select: WireRecordFilter,
@@ -2293,7 +2299,7 @@ def accepts_value(self, value: object) -> bool:
 
 
 class _FamilyNameDecoder(_TypeDecoder, _ChoiceDecoder[Any]):
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return type(data) is str
 
     def encode(self, value: type[DeclaredFamily]) -> str:
@@ -2321,8 +2327,8 @@ def from_annotation(cls, target, args):
     def accepts_value(self, value: object) -> bool:
         return any(type(value) is type(item) and value == item for item in self.values)
 
-    def admits_wire(self, data: object) -> bool:
-        return any(plan.admits_wire(data) for plan in self.value_plans)
+    def fits_wire(self, data: object) -> bool:
+        return any(plan.fits_wire(data) for plan in self.value_plans)
 
     def encode(self, value: object) -> object:
         for item, plan in zip(self.values, self.value_plans, strict=True):
@@ -2348,8 +2354,8 @@ class UnionDecoder[T](_ContextDecoder[T]):
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return self.alternatives
 
-    def admits_wire(self, data: object) -> bool:
-        return any(alternative.admits_wire(data) for alternative in self.alternatives)
+    def fits_wire(self, data: object) -> bool:
+        return any(alternative.fits_wire(data) for alternative in self.alternatives)
 
     def _rewrite_records(
         self, data: object, change: WireRecordChange, select: WireRecordFilter,
@@ -2385,7 +2391,7 @@ def decode_external(self, data: object, record: Mapping[str, object]) -> T:
             return self.decode_bound(data, record)
         return self.decode_resolved(self.decode_candidates(
             lambda alternative: alternative.decode_external(data, record),
-            admits=lambda alternative: alternative.admits_external(data),
+            fits=lambda alternative: alternative.fits_external(data),
         ))
 
     @property
@@ -2407,18 +2413,19 @@ def encode(self, value: T) -> object:
     def decode_bound(self, data: object, context: Mapping[str, object]) -> T:
         return self.decode_candidates(
             lambda alternative: alternative.decode_bound(data, context),
-            admits=lambda alternative: alternative.admits_wire(data),
+            fits=lambda alternative: alternative.fits_wire(data),
         )
 
     def decode_candidates(
         self, decode: Callable[[ValueDecoder[T]], T], *,
-        admits: Callable[[ValueDecoder[T]], bool],
+        fits: Callable[[ValueDecoder[T]], bool],
     ) -> T:
         first_error = None
+        # Shapes only rule alternatives out; when they rule out every one, the first
+        # alternative's own decode states the refusal.
+        candidates = tuple(filter(fits, self.alternatives)) or self.alternatives[:1]
         try:
-            for alternative in self.alternatives:
-                if not admits(alternative):
-                    continue
+            for alternative in candidates:
                 try:
                     return decode(alternative)
                 except MalformedRecord as error:
@@ -2444,7 +2451,7 @@ class _ArrayDecoder(ValueDecoder[Any]):
     def wire_children(self) -> tuple[ValueDecoder[object], ...]:
         return self.items
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, list)
 
     def _rewrite_records(
@@ -2523,8 +2530,8 @@ class _VariadicTupleDecoder(_TupleDecoder):
 
 
 class _FixedTupleDecoder(_TupleDecoder):
-    def admits_wire(self, data: object) -> bool:
-        return super().admits_wire(data) and len(cast(list[object], data)) == len(self.items)
+    def fits_wire(self, data: object) -> bool:
+        return super().fits_wire(data) and len(cast(list[object], data)) == len(self.items)
 
     def plan_items(self, data: list[object]) -> Iterable[tuple[ValueDecoder[object], object]]:
         with MalformedRecord.refusing("Tuple length does not match declaration"):
@@ -2625,7 +2632,7 @@ def decode_keys(self, data, keys, selections):
 class _StringKeyDictDecoder(_DictDecoder):
     entries = staticmethod(_object_entries)
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, dict)
 
     @staticmethod
@@ -2646,7 +2653,7 @@ def decode_document(self, data: object) -> dict[str, object]:
 class _PairKeyDictDecoder(_DictDecoder):
     entries = staticmethod(_object_or_pair_entries)
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, (dict, list))
 
     @staticmethod
@@ -2664,7 +2671,7 @@ class _FamilyRecordDecoder[Member: DeclaredFamily](_ContextDecoder[Member]):
 
     target: type[Member]
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         return isinstance(data, dict) and isinstance(data.get(self.target.family_discriminator), str)
 
     @property
@@ -2713,9 +2720,9 @@ def decode_bound(self, data: object, context: Mapping[str, object]) -> Member:
 class _LeafDocumentDecoder(_FamilyRecordDecoder):
     """A directly declared leaf document omits its family selector."""
 
-    def admits_wire(self, data: object) -> bool:
-        return (super().admits_wire(data)
-                or _member_decoder(self.target).admits_wire(data))
+    def fits_wire(self, data: object) -> bool:
+        return (super().fits_wire(data)
+                or _member_decoder(self.target).fits_wire(data))
 
     def encode_document(self, value: DeclaredFamily) -> object:
         return _member_decoder(type(value)).encode(value)
@@ -2736,7 +2743,7 @@ class _RecordDecoder(_ContextDecoder[Any]):
 
     target: type
 
-    def admits_wire(self, data: object) -> bool:
+    def fits_wire(self, data: object) -> bool:
         if not isinstance(data, dict):
             return False
         plan = FieldCodec._record_plan(self.target)
diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py
index 2690c5b1d7..5fe859c7bd 100644
--- a/src/agent_comms/historical_views.py
+++ b/src/agent_comms/historical_views.py
@@ -13,7 +13,7 @@
 from typing import TYPE_CHECKING
 
 from .bus import Bus
-from .channel_targets import CommandTarget, NamedTarget, ThreadTarget
+from .channel_targets import NamedTarget, ThreadTarget
 from .channels import Channel
 from .coordination_errors import HistoricalSourceChanged
 from .errors import UnregisteredThreadError
@@ -194,11 +194,11 @@ def capture(self, snapshot: RegistryProvenance) -> ChannelDisplayScope:
 @dataclass(frozen=True, slots=True)
 class DMHistory(HistoryView):
     first: str
-    second: str | ThreadTarget
+    second: ThreadTarget
 
     def capture(self, snapshot: RegistryProvenance) -> DMDisplayScope:
         return DMDisplayScope.capture(
-            self.first, CommandTarget.named(self.second).selected_name(snapshot), snapshot)
+            self.first, self.second.selected_name(snapshot), snapshot)
 
 
 @dataclass(frozen=True, slots=True)
@@ -231,10 +231,8 @@ def live_page(self, presentation: BusPresentation, bounds: PageBounds) -> Messag
         from .read_basis import DMDisplayBasis
 
         bus = presentation.bus
-        target = CommandTarget.named(self.second)
-        target.require_dm_peer()
         with presentation.registry.store.reading_snapshot() as snapshot:
-            peer = target.selected_name(snapshot)
+            peer = self.second.selected_name(snapshot)
             missing = peer not in snapshot.threads
         if missing and bus.history.threads(peer):
             return MessagePage((), False, False, NoDisplaySelection())
diff --git a/src/agent_comms/history_views.py b/src/agent_comms/history_views.py
index 3cb4562f63..7193e3352b 100644
--- a/src/agent_comms/history_views.py
+++ b/src/agent_comms/history_views.py
@@ -119,7 +119,7 @@ def external(self):
         self.through.wire.require_source(AssignedSourceIdentity(str(self.views.root), self.thread.incarnation))
         page = DMDisplayBasis.fetch_page(
             self.views.bus, self.views.registry, self.views.root, self.viewer.name,
-            self.thread.name, worktree=self.viewer.worktree,
+            ThreadTarget(self.thread.name), worktree=self.viewer.worktree,
             bounds=PageBounds(LiveCursor(self.through.wire.sequence + 1).older()),
         )
         if page.messages:
@@ -216,7 +216,7 @@ def historical_threads(self, name: str | None = None) -> tuple[HistoricalThread,
         return self.bus.history.threads(name)
 
     def dm_display_page(
-        self, peer: str | ThreadTarget, *, worktree: str, bounds: PageBounds = NEWEST_PAGE
+        self, peer: ThreadTarget, *, worktree: str, bounds: PageBounds = NEWEST_PAGE
     ) -> MessagePage:
         viewer = self.messaging.user_identity(worktree).name
         return DMDisplayHistory(viewer, peer, worktree).page(self.presentation, bounds)
diff --git a/src/agent_comms/installer/backup.py b/src/agent_comms/installer/backup.py
index 466ef1f380..8232dc3719 100644
--- a/src/agent_comms/installer/backup.py
+++ b/src/agent_comms/installer/backup.py
@@ -26,6 +26,7 @@
 from agent_comms.field_codec import FieldCodec, MalformedRecord, PathText
 from agent_comms.installer.interpreter_operation import RouteSelection
 from agent_comms.locked_store import DeclaredFile, LockedStore, RootFile, RuntimeFile, UndeclaredFile
+from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator
 from agent_comms.private_path import FileObservation, FileRevision, MissingFile
 from agent_comms.registry_store import RegistryStore
 from agent_comms.store_files import AtomicFile
@@ -199,6 +200,21 @@ def interrupted(cls, root: Path, spill_dir: Path) -> tuple[StoreBackup, ...]:
         )
         return tuple(backup for backup in backups if not backup.original())
 
+    @classmethod
+    def complete_interrupted(cls, root: Path, spill_dir: Path) -> tuple[Path, ...]:
+        """Finish every restore a killed install left unfinished, before anything reads
+        the bus: take its abandoned maintenance hold, restore, release, then retire the
+        backup so the next install never meets it again."""
+        retired = []
+        for backup in cls.interrupted(root, spill_dir):
+            maintenance = MaintenanceBarrier(RegistryStore.at(root).path)
+            hold = maintenance.begin(MaintenanceOperator.current())
+            with hold.holding():
+                backup.restore()
+            maintenance.clear(hold)
+            retired.append(backup.retire())
+        return tuple(retired)
+
     def retire(self) -> Path:
         archived = self.path.with_name(f'{self.path.name}.restored-{time.time_ns()}')
         self.path.rename(archived)
diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py
index 6b181fea45..95a5c949ce 100644
--- a/src/agent_comms/installer/upgrade.py
+++ b/src/agent_comms/installer/upgrade.py
@@ -771,6 +771,7 @@ def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice,
     tree = TargetTree(checkout, scratch, target)
     bridges = report.bridges
     with commands, report.running(tree):
+        StoreBackup.complete_interrupted(root, spill_dir)
         installed = Stack(tree.path, commands)
         current = commands.run("git", "-C", checkout, "rev-parse", "HEAD").strip()
         ancestry = ["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target]
diff --git a/src/agent_comms/locked_store.py b/src/agent_comms/locked_store.py
index 70303e9c17..eba890b9b7 100644
--- a/src/agent_comms/locked_store.py
+++ b/src/agent_comms/locked_store.py
@@ -59,6 +59,14 @@ class RootFile:
     filename: ClassVar[str]
     root_directory: ClassVar[str] = "."
 
+    def __init_subclass__(cls, **kwargs: object) -> None:
+        """A declared file is either retained across installations or rewritten by its
+        running owner; a declaration that is neither has no install behaviour at all."""
+        super().__init_subclass__(**kwargs)
+        if "filename" in vars(cls) and not issubclass(cls, (DurableFile, RuntimeFile)):
+            raise TypeError(f"{cls.__qualname__} declares {cls.filename!r} but is neither "
+                            "a DurableFile nor a RuntimeFile")
+
     @classmethod
     def path_in(cls, root: Path | str, **parts: str) -> Path:
         return Path(root) / cls.filename.format(**parts)
@@ -336,6 +344,12 @@ class DurableFile(RootFile):
     __slots__ = ()
 
 
+class RuntimeFile(RootFile):
+    """A bus file no carry writes: running owners rewrite it at any time, maintenance included."""
+
+    __slots__ = ()
+
+
 class BusFile(DurableFile):
     """Declared data and lifecycle files that rule out a fresh bus installation."""
 
@@ -386,7 +400,7 @@ def copy_group(cls, source: Path, target: Path) -> frozenset[Path]:
         return copied
 
 
-class RetiredCommanderFile(RootFile):
+class RetiredCommanderFile(DurableFile):
     """The retired commander file that tools/cutover/carry_commander_project.py consumes.
     The practice copy holds declared files only, so an undeclared input made the practice
     carry find nothing while the live carry deleted it. Deleted with that carry."""
@@ -642,12 +656,6 @@ def publish_change(cls, path: Path) -> None:
             ) from error
 
 
-class RuntimeFile(RootFile):
-    """A bus file no carry writes: running owners rewrite it at any time, maintenance included."""
-
-    __slots__ = ()
-
-
 @dataclass(frozen=True, slots=True)
 class LockedStore(FileStore, StoreLockOwner, JsonDocument[T], declared_name=None):
     """Ordered document transactions; callbacks never mutate or reenter."""
diff --git a/src/agent_comms/read_basis.py b/src/agent_comms/read_basis.py
index cded4de71e..f631db74b9 100644
--- a/src/agent_comms/read_basis.py
+++ b/src/agent_comms/read_basis.py
@@ -37,7 +37,7 @@ class DisplayViewer(SubclassUnion, affix="Viewer"):
     """The reader owns read evidence, aliases and cursor selection."""
 
     @abstractmethod
-    def direct_history(self, direct: type[DMHistory], peer: str | ThreadTarget) -> DMHistory: ...
+    def direct_history(self, direct: type[DMHistory], peer: ThreadTarget) -> DMHistory: ...
 
     @abstractmethod
     def resolve(self, snapshot: RegistryProvenance) -> DisplayViewer: ...
@@ -69,7 +69,7 @@ class AnonymousViewer(NullMember, DisplayViewer):
     def resolve(self, snapshot: RegistryProvenance) -> AnonymousViewer:
         return self
 
-    def direct_history(self, direct: type[DMHistory], peer: str | ThreadTarget) -> DMHistory:
+    def direct_history(self, direct: type[DMHistory], peer: ThreadTarget) -> DMHistory:
         raise ValueError("DM history requires --as <participant>.")
 
     def historical_selection(self, page: MessagePage, presentation: BusPresentation, historical: HistorySource, source: Bus, document: ReadDocument, snapshot: RegistrySnapshot) -> NoDisplaySelection:
@@ -101,7 +101,7 @@ class NamedViewer(SingleFieldValue, DisplayViewer):
     def resolve(self, snapshot: RegistryProvenance) -> NamedViewer:
         return type(self)(snapshot.canonical_name(self.name))
 
-    def direct_history(self, direct: type[DMHistory], peer: str | ThreadTarget) -> DMHistory:
+    def direct_history(self, direct: type[DMHistory], peer: ThreadTarget) -> DMHistory:
         return direct(self.name, peer)
 
     def historical_selection(self, page: MessagePage, presentation: BusPresentation, historical: HistorySource, source: Bus, document: ReadDocument, snapshot: RegistrySnapshot) -> HistoricalSelection:
@@ -679,7 +679,7 @@ def fetch_page(
         registry: Registration,
         root: Path,
         viewer: str,
-        peer: str | ThreadTarget,
+        peer: ThreadTarget,
         *,
         worktree: str,
         bounds: PageBounds,
@@ -689,11 +689,9 @@ def fetch_page(
         Fetching is not paint proof. A UI may use the basis only after proving
         that the corresponding inbound tail was contiguous and visibly painted.
         """
-        target = CommandTarget.named(peer)
-        target.require_dm_peer()
         with registry.store.reading_snapshot() as snapshot:
             viewer_name = snapshot.canonical_name(viewer)
-            peer_name = target.selected_name(snapshot)
+            peer_name = peer.selected_name(snapshot)
             scope = DMDisplayScope.capture_human(viewer_name, peer_name, snapshot)
             viewer_thread = snapshot.threads[viewer_name]
             peer_thread = snapshot.participant(peer_name)
@@ -708,7 +706,7 @@ def fetch_page(
             root=str(root.resolve()),
             root_identity=(root_info.st_dev, root_info.st_ino),
             worktree=str(Path(worktree).resolve()),
-            requested_peer=target.name,
+            requested_peer=peer.name,
             viewer_identity=viewer_thread.incarnation,
             viewer_names=viewer_names,
             peer_identity=peer_thread.incarnation,
diff --git a/stack/uv.lock b/stack/uv.lock
index 1d127d4ff7..c0abdbcb4b 100644
--- a/stack/uv.lock
+++ b/stack/uv.lock
@@ -243,7 +243,7 @@ wheels = [
 [[package]]
 name = "batrachian-toad"
 version = "0.6.20"
-source = { git = "https://github.com/OpenHCSDev/toad.git?branch=main#aeef59ee093561b7a1e5a87d1f33ff952e8343ef" }
+source = { git = "https://github.com/OpenHCSDev/toad.git?branch=main#336cf3ec3074496aac2c0b783d31719ac0da3801" }
 dependencies = [
     { name = "agent-client-protocol" },
     { name = "agent-comms" },
diff --git a/tests/guards/test_debt_ratchet.py b/tests/guards/test_debt_ratchet.py
index 4be1de7b30..13a9c89234 100644
--- a/tests/guards/test_debt_ratchet.py
+++ b/tests/guards/test_debt_ratchet.py
@@ -429,6 +429,14 @@ def owner(name: str, lines: int, indent: str = "") -> str:
     )
 
 
+def module(*owners: str) -> str:
+    """The owners as one module in Black's form, so the formatting ratchet counts nothing and
+    each comparison measures only the classes."""
+    import black
+
+    return black.format_str("".join(owners), mode=black.Mode())
+
+
 @pytest.mark.parametrize("root", ["src/agent_comms", "src/toad", "src/textual"])
 def test_family_flattening_cannot_offset_another_file_or_admit_a_named_facade(
     tmp_path: Path, root: str, ratchet
@@ -512,7 +520,7 @@ def test_family_name_owner_reads_and_codec_detours(source: str, expected: int):
 
 def test_small_owners_stay_within_the_class_threshold(repo: Repository) -> None:
     base = repo.commit({"owners.py": owner("Small", 2)})
-    head = repo.commit({"owners.py": owner("Small", 500) + owner("New", 500)})
+    head = repo.commit({"owners.py": module(owner("Small", 500), owner("New", 500))})
     status, report = repo.compare(base, head)
     assert status == 0
     assert report["delta"][f"GodClassExcess:{repo.root}/owners.py::Small"] == 0
@@ -520,8 +528,8 @@ def test_small_owners_stay_within_the_class_threshold(repo: Repository) -> None:
 
 
 def test_class_growth_cannot_be_cancelled_by_another_class(repo: Repository) -> None:
-    base = repo.commit({"classes.py": owner("Growing", 501) + owner("Shrinking", 600)})
-    head = repo.commit({"classes.py": owner("Growing", 502) + owner("Shrinking", 500)})
+    base = repo.commit({"classes.py": module(owner("Growing", 501), owner("Shrinking", 600))})
+    head = repo.commit({"classes.py": module(owner("Growing", 502), owner("Shrinking", 500))})
     status, report = repo.compare(base, head)
     assert status == 1
     assert report["delta"][f"GodClassExcess:{repo.root}/classes.py::Growing"] == 1
@@ -540,13 +548,13 @@ def test_new_or_crossing_god_class_is_refused(repo: Repository, before: int) ->
 
 def test_class_move_keeps_baseline_and_new_small_owner_has_no_class_debt(repo: Repository) -> None:
     base = repo.commit({"old.py": owner("Existing", 501)})
-    moved = repo.commit({"old.py": None, "new.py": owner("Existing", 501) + owner("NewOwner", 25)})
+    moved = repo.commit({"old.py": None, "new.py": module(owner("Existing", 501), owner("NewOwner", 25))})
     status, report = repo.compare(base, moved)
     assert status == 0
     assert report["base"][f"GodClassExcess:{repo.root}/new.py::Existing"] == 1
     assert report["delta"][f"GodClassExcess:{repo.root}/new.py::Existing"] == 0
     assert report["delta"][f"GodClassExcess:{repo.root}/new.py::NewOwner"] == 0
-    grown = repo.commit({"new.py": owner("Existing", 502) + owner("NewOwner", 26)})
+    grown = repo.commit({"new.py": module(owner("Existing", 502), owner("NewOwner", 26))})
     assert repo.compare(moved, grown)[0] == 1
 
 
diff --git a/tests/source_programs.py b/tests/source_programs.py
index aeac66f78f..8950544292 100644
--- a/tests/source_programs.py
+++ b/tests/source_programs.py
@@ -10,6 +10,8 @@
 from pathlib import Path
 from typing import Self
 
+from agent_comms.launch_runtime import LaunchRuntime
+
 
 @dataclass(frozen=True)
 class SourcePrograms:
@@ -22,6 +24,11 @@ def bind(self, directory: Path) -> None:
         for package in self.packages:
             (directory / package.name).symlink_to(package, target_is_directory=True)
 
+    @property
+    def runtime(self) -> LaunchRuntime:
+        """The runtime these programs report: the interpreter this prefix declares."""
+        return LaunchRuntime(self.prefix / 'bin' / 'python')
+
     def script(self, function: Callable[..., object]) -> Path:
         target = f'{function.__module__}:{function.__name__}'
         (entry,) = (entry for entry in self.entries if entry.value == target)
diff --git a/tests/test_bus_files.py b/tests/test_bus_files.py
index 663347c0c8..65dbd40916 100644
--- a/tests/test_bus_files.py
+++ b/tests/test_bus_files.py
@@ -14,7 +14,7 @@
 from agent_comms.comms import Comms, wire
 from agent_comms.coordination_session import NativeBoundary
 from agent_comms.envelope_claim_transitions import ExistingFileClaim, NativeClaim
-from agent_comms.locked_store import DurableFile, RootFile, RuntimeFile, SQLiteFile
+from agent_comms.locked_store import DeclaredFile, DurableFile, RootFile, RuntimeFile, SQLiteFile
 from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.repository_name import RepositoryName
 from agent_comms.todos import TodoStore
@@ -81,7 +81,7 @@ def test_parameterized_sqlite_resources_follow_their_file_owner(tmp_path: Path,
         actual = {path for path in tmp_path.rglob("*") if path.is_file()}
         assert len(actual) > 1  # The real SQLite writer has an open journal/WAL.
         assert actual <= NativeInputProof.files_in(tmp_path)
-        assert str(proof.relative_to(tmp_path)) in bus_files(tmp_path)
+        assert DeclaredFile(NativeInputProof, proof.relative_to(tmp_path)) in bus_files(tmp_path)
 
 
 def test_installer_reconciles_a_native_proof_journal_onto_its_declaration(tmp_path: Path) -> None:
diff --git a/tests/test_carry_bus_history.py b/tests/test_carry_bus_history.py
index 3a387158a0..cf479e34d5 100644
--- a/tests/test_carry_bus_history.py
+++ b/tests/test_carry_bus_history.py
@@ -181,14 +181,20 @@ def test_predecessor_publisher_history_round_trips_into_bus(tmp_path):
     spill = tmp_path / "hdd-spill"
     spill.mkdir()
     StoreBackup.copy_tree(original, root)
-    # This publisher authored a pre-carry registry. Its finished-turn carry
-    # precedes the goal-request carry and every current registry reader.
-    finished = FinishedTurns(root, original, spill)
-    finished_dry_run = finished.carry(apply=False)
-    finished_result = finished.carry(apply=True)
-    assert finished_result.changed > 0
-    assert finished_result.detail == finished_dry_run.detail
-    assert finished.carry(apply=False).changed == 0
+    # This publisher authored a pre-carry registry: every step the chain orders before the
+    # goal-request carry runs first, as the installer runs them.
+    preceding = {}
+    maintenance = MaintenanceBarrier(RegistryStore.at(root).path)
+    hold = maintenance.begin(MaintenanceOperator.current())
+    with hold.holding():
+        for step in chain.steps[:chain.steps.index(OwnerRequests)]:
+            carry = step(root, original, spill)
+            dry_run = carry.carry(apply=False)
+            preceding[step] = carry.carry(apply=True)
+            assert preceding[step].detail == dry_run.detail
+            assert carry.carry(apply=False).changed == 0
+    maintenance.clear(hold)
+    assert preceding[FinishedTurns].changed > 0
     relatives = (Path(), archived.relative_to(original))
     originals = {relative: (original / relative / BusHistory.filename).read_bytes()
                  for relative in relatives}
diff --git a/tests/test_carry_dead_turns.py b/tests/test_carry_dead_turns.py
index 376fea5d95..ef53f8538e 100644
--- a/tests/test_carry_dead_turns.py
+++ b/tests/test_carry_dead_turns.py
@@ -20,6 +20,6 @@ def test_deleted_turn_files_are_named_and_the_current_one_kept(tmp_path):
     step = DeadTurns(comms.root, comms.root, tmp_path / "spill")
     assert step.carry(apply=False).rewritten == ()
     carried = step.carry(apply=True)
-    assert carried.rewritten == (str(dead.relative_to(comms.root)),)
+    assert carried.rewritten == (dead.relative_to(comms.root),)
     assert current.exists() and not dead.exists()
     assert step.carry(apply=False).changed == 0
diff --git a/tests/test_channel_display.py b/tests/test_channel_display.py
index 8ed1d25f24..dfbfe103e5 100644
--- a/tests/test_channel_display.py
+++ b/tests/test_channel_display.py
@@ -7,7 +7,7 @@
 
 import pytest
 
-from agent_comms.channel_targets import CommandTarget
+from agent_comms.channel_targets import CommandTarget, ThreadTarget
 from agent_comms.channels import AnyOfMatch, ViewPredicate, ActivityView
 from agent_comms.comms import wire
 from agent_comms.coordination_errors import DMPaintedBoundChanged, DMScopeChanged
@@ -213,7 +213,7 @@ def burst():
             page = comms.views.channel_display_page("#api", viewer=NamedViewer(comms.messaging.user_identity(str(tmp_path)).name))
             if page.newest_seq is not None:
                 comms.views.mark_page_read(page, worktree=str(tmp_path))
-            dm = comms.views.dm_display_page("alice", worktree=str(tmp_path))
+            dm = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path))
             assert bodies(dm) == ["DM original"]
             comms.views.mark_page_read(dm, worktree=str(tmp_path))
     finally:
diff --git a/tests/test_dm_scoped_ack.py b/tests/test_dm_scoped_ack.py
index 86d05e0cff..9cc5bf2a4c 100644
--- a/tests/test_dm_scoped_ack.py
+++ b/tests/test_dm_scoped_ack.py
@@ -1,7 +1,7 @@
 """No-provider bounded human DM read evidence; never use global inbox ACK."""
 
 from __future__ import annotations
-from agent_comms.channel_targets import CommandTarget
+from agent_comms.channel_targets import CommandTarget, ThreadTarget
 from agent_comms.registration import Registration
 
 import os
@@ -29,7 +29,7 @@ def test_dm_page_marks_covered_range_preserving_omitted_older_unread(tmp_path: P
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     for index in range(50):
         comms.messaging.send("peer", viewer, f"old {index}").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path), bounds=PageBounds(limit=5))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path), bounds=PageBounds(limit=5))
     assert len(page.messages) == 5 and page.has_older
     assert isinstance(page.selection, DMSelection)
     comms.views.mark_page_read(page, worktree=str(tmp_path))
@@ -50,7 +50,7 @@ def test_peer_delete_same_name_rebind_rejects_stale_page_without_read_ack(tmp_pa
     comms.messaging.send("other", "#all", "OTHER_CHANNEL_4").message.message_id
     comms.messaging.send("peer", viewer, "PEER_PAINTED_5").message.message_id
     comms.messaging.send("other", viewer, "OTHER_UNPAINTED_6").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     assert [message.body for message in page.messages] == ["PEER_PAINTED_2", "PEER_PAINTED_5"]
     assert page.newest_seq == 5 and isinstance(page.selection, DMSelection)
     comms.registry.unregister("peer")
@@ -74,7 +74,7 @@ def test_foreign_root_wrong_peer_and_unbounded_or_bool_through_rejected(tmp_path
             comms.registry.declare(_thread(comms.root, name))
         viewer = comms.messaging.user_identity(str(comms.root)).name
         comms.messaging.send("peer", viewer, "painted").message.message_id
-    page = first.views.dm_display_page("peer", worktree=str(first.root))
+    page = first.views.dm_display_page(ThreadTarget("peer"), worktree=str(first.root))
     assert isinstance(page.selection, DMSelection) and page.newest_seq is not None
     for comms, target, through, error in (
         (second, "peer", page.newest_seq, DMScopeChanged),
@@ -96,7 +96,7 @@ def test_painted_bound_uses_existing_transcript_refresh_error(tmp_path: Path):
     comms.registry.declare(_thread(tmp_path, "peer"))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", viewer, "painted").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     with pytest.raises(TranscriptChanged):
         page.selection.basis.acknowledge(
             comms.bus, comms.registry, comms.root, "peer",
@@ -110,7 +110,7 @@ def test_root_replaced_after_fetch_rejects_mark_without_reading_new_root(tmp_pat
     comms.registry.declare(_thread(tmp_path, "peer"))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", viewer, "painted in original root").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     previous = tmp_path.with_name(tmp_path.name + "-previous")
     tmp_path.rename(previous)
     shutil.copytree(previous, tmp_path, symlinks=True)
@@ -124,7 +124,7 @@ def test_independent_read_is_idempotent_and_never_retargets_page(tmp_path: Path)
     comms.registry.declare(_thread(tmp_path, "peer"))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", viewer, "painted").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     comms.views.mark_user_view_read("peer", worktree=str(tmp_path))
     comms.messaging.send("peer", viewer, "unpainted after explicit mark").message.message_id
     comms.views.mark_page_read(page, worktree=str(tmp_path))
@@ -134,7 +134,7 @@ def test_independent_read_is_idempotent_and_never_retargets_page(tmp_path: Path)
 def test_empty_dm_basis_is_no_op_without_creating_read_ledger(tmp_path: Path):
     comms = installed_comms(tmp_path)
     comms.registry.declare(_thread(tmp_path, "peer"))
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     assert page.selection.basis.displayed.tail == EmptyTail()
     page.selection.basis.acknowledge(
         comms.bus, comms.registry, comms.root, "peer", worktree=str(tmp_path), through=0,
diff --git a/tests/test_dm_scoped_ack_successor.py b/tests/test_dm_scoped_ack_successor.py
index dd298368d1..f0a42c494c 100644
--- a/tests/test_dm_scoped_ack_successor.py
+++ b/tests/test_dm_scoped_ack_successor.py
@@ -8,7 +8,7 @@
 
 import pytest
 
-from agent_comms.channel_targets import CommandTarget
+from agent_comms.channel_targets import CommandTarget, ThreadTarget
 from agent_comms.comms import wire
 from agent_comms.child_process import ProcessIdentity
 from conftest import installed_comms
@@ -41,7 +41,7 @@ def test_alias_rename_invalidates_old_basis_but_fresh_alias_page_can_mark(tmp_pa
     comms.registry.declare(_peer(tmp_path, "peer"))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", viewer, "before rename").message.message_id
-    old = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    old = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     comms.threads.rename_managed_thread("peer", "renamed", owner=ProcessIdentity.capture(os.getpid()))
     markers = tmp_path / "read_ledger.json"
     marker_before = markers.read_bytes() if markers.exists() else None
@@ -52,7 +52,7 @@ def test_alias_rename_invalidates_old_basis_but_fresh_alias_page_can_mark(tmp_pa
 
     # The old spelling is now an alias for the SAME incarnation. A fresh
     # page, not stale name text, is authoritative after the rename.
-    fresh = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    fresh = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     assert isinstance(fresh.selection, DMSelection) and fresh.selection.basis.peer == "renamed"
     _mark(comms, "peer", tmp_path, fresh)
     assert comms.bus.pending_count(viewer, CommandTarget.named("renamed")) == 0
@@ -65,7 +65,7 @@ def test_delete_and_same_name_reregister_invalidates_old_peer_basis(tmp_path: Pa
     comms.registry.declare(_peer(tmp_path, "peer"))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", viewer, "old peer painted").message.message_id
-    old = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    old = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     assert isinstance(old.selection, DMSelection)
     comms.registry.unregister("peer")
     comms.registry.remove("peer")
@@ -79,7 +79,7 @@ def test_delete_and_same_name_reregister_invalidates_old_peer_basis(tmp_path: Pa
         _mark(comms, "peer", tmp_path, old)
     assert (markers.read_bytes() if markers.exists() else None) == marker_before
     assert comms.bus.pending_count(viewer, CommandTarget.named("peer")) == 1
-    fresh = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    fresh = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     assert isinstance(fresh.selection, DMSelection)
     assert fresh.selection.basis.peer_identity != old.selection.basis.peer_identity
 
@@ -89,7 +89,7 @@ def test_viewer_rebind_and_foreign_worktree_reject_old_basis(tmp_path: Path):
     comms.registry.declare(_peer(tmp_path, "peer"))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", viewer, "old viewer painted").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     assert isinstance(page.selection, DMSelection)
     with pytest.raises(DMScopeChanged):
         comms.views.mark_page_read(page, worktree=str(tmp_path / "foreign"))
@@ -109,7 +109,7 @@ def test_old_viewer_alive_but_new_human_selected_rejects_stale_basis(tmp_path: P
     comms.registry.declare(_peer(tmp_path, "peer"))
     old_viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", old_viewer, "old viewer painted").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     assert isinstance(page.selection, DMSelection) and page.selection.basis.viewer == old_viewer
     comms.registry.declare(comms.registry.new_thread('new_user', frozenset(), str(tmp_path), role=UserThreadRole))
     # Normal registry rename retains the old human declaration but moves its
@@ -134,7 +134,7 @@ def test_newest_page_mark_leaves_older_message_unread(tmp_path: Path):
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     older = comms.messaging.send("peer", viewer, "older").message
     newest = comms.messaging.send("peer", viewer, "newest").message
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path), bounds=PageBounds(limit=1))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path), bounds=PageBounds(limit=1))
     assert page.messages == (newest,)
     _mark(comms, "peer", tmp_path, page)
     assert comms.bus.inbox(viewer, CommandTarget.named("peer")) == [older]
@@ -146,7 +146,7 @@ def test_lifecycle_and_marker_lock_order_does_not_deadlock(tmp_path: Path):
         comms.registry.declare(_peer(tmp_path, name))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("peer", viewer, "painted").message.message_id
-    page = comms.views.dm_display_page("peer", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("peer"), worktree=str(tmp_path))
     with ThreadPoolExecutor(max_workers=2) as pool:
         marked = pool.submit(_mark, comms, "peer", tmp_path, page)
         lifecycle = pool.submit(
diff --git a/tests/test_field_codec.py b/tests/test_field_codec.py
index 410e262d66..5451efb836 100644
--- a/tests/test_field_codec.py
+++ b/tests/test_field_codec.py
@@ -9,6 +9,7 @@
 
 from agent_comms.declared_family import DeclaredFamily, UnknownFamilyMember
 from agent_comms.field_codec import (
+    DeclaredWireType,
     FieldCodec,
     InlineRecord,
     JsonShapeFamily,
@@ -16,6 +17,7 @@
     MalformedRecord,
     NullMember,
     PathText,
+    RecordMember,
     RecordedTimestamp,
     SubclassUnion,
     TextRepresentation,
@@ -278,7 +280,8 @@ class Second(SubclassUnion, affix=""):
 
     def declare_number(union):
         @dataclass(frozen=True, slots=True)
-        class Number(union):
+        class Number(DeclaredWireType, union):
+            wire_type = int
             value: int
 
             def to_wire(self):
@@ -289,18 +292,14 @@ def from_wire(cls, data):
                 calls.append(cls)
                 return cls(FieldCodec.decode(int, data))
 
-            @classmethod
-            def schema(cls):
-                return FieldCodec.value_schema(int)
-
         return Number
 
     first = declare_number(First)
     second = declare_number(Second)
 
     @dataclass(frozen=True, slots=True)
-    class Text(OwnedWireText, First):
-        pass
+    class Text(DeclaredWireType, OwnedWireText, First):
+        wire_type = str
 
     @dataclass(frozen=True, slots=True)
     class Empty(NullMember, First):
@@ -323,20 +322,9 @@ class Empty(NullMember, First):
         FieldCodec.decode(Second, "text")
 
     @dataclass(frozen=True, slots=True)
-    class Later(First):
+    class Later(RecordMember, First):
         entries: tuple[int, ...]
 
-        def to_wire(self):
-            return FieldCodec.encode_record(self)
-
-        @classmethod
-        def from_wire(cls, data):
-            return FieldCodec.decode_record(cls, data)
-
-        @classmethod
-        def schema(cls):
-            return FieldCodec.record_schema(cls)
-
     encoded = {"entries": [1, 2]}
     assert FieldCodec.encode(Later((1, 2))) == encoded
     assert type(FieldCodec.decode(First, encoded)) is Later
diff --git a/tests/test_historical_views.py b/tests/test_historical_views.py
index e14ab93945..da3a0c155d 100644
--- a/tests/test_historical_views.py
+++ b/tests/test_historical_views.py
@@ -8,7 +8,7 @@
 import pytest
 
 from agent_comms.bus import Bus, PublicRecord
-from agent_comms.channel_targets import CommandTarget
+from agent_comms.channel_targets import CommandTarget, ThreadTarget
 from agent_comms.comms import wire
 from agent_comms.coordination_errors import HistoricalSourceChanged
 from agent_comms.historical_views import HistoricalMessage, HistoryArchive
@@ -87,7 +87,7 @@ def migrated(tmp_path):
     "method,args,reader",
     [
         ("channel_display_page", ("#team",), lambda live: dict(viewer=live.views.display_viewer(str(live.root)))),
-        ("dm_display_page", ("alice",), lambda live: dict(worktree=str(live.root))),
+        ("dm_display_page", (ThreadTarget("alice"),), lambda live: dict(worktree=str(live.root))),
         ("channel_display_page", ("#any",), lambda live: dict(viewer=live.views.display_viewer(str(live.root)))),
     ],
 )
@@ -159,7 +159,7 @@ def test_select_painted_page_keeps_its_source_with_attached_history(migrated, co
             "#team", viewer=live.views.display_viewer(str(live.root)), bounds=bounds,
         )
     else:
-        page = live.views.dm_display_page("alice", worktree=str(live.root), bounds=bounds)
+        page = live.views.dm_display_page(ThreadTarget("alice"), worktree=str(live.root), bounds=bounds)
     message = page.messages[-1]
     selected = page.select((message.seq,))
     assert selected.messages == (message,)
@@ -309,13 +309,19 @@ def test_normal_comms_api_contains_history_and_execution_stays_live(migrated):
         live.bus.bus).messages) == 4
 
 
+def root_bytes(root: Path) -> dict[Path, bytes]:
+    """The bytes of every member the backup owner records for this root."""
+    return {path: (root / path).read_bytes()
+            for members in bus_files(root).values() for path in members}
+
+
 def test_attach_only_reads_original_roots_and_rejects_changed_snapshot(tmp_path):
     old = setup(tmp_path / "old")
     (old.root / "unknown-input.json").write_text('{"verdict":"UNKNOWN"}')
-    original = {name: (old.root / name).read_bytes() for name in bus_files(old.root)}
+    original = root_bytes(old.root)
     live = setup(tmp_path / "live", 1)
     source = live.views.attach_history(old.root)
-    assert {name: (old.root / name).read_bytes() for name in bus_files(old.root)} == original
+    assert root_bytes(old.root) == original
     assert not (Path(source.root) / "unknown-input.json").exists()
     page = live.views.channel_display_page(
         "#team", viewer=NamedViewer(live.messaging.user_identity(str(live.root)).name),
@@ -348,7 +354,7 @@ def test_missing_historical_peer_is_browsable_without_live_registration(migrated
     _, _, live, source = migrated
     live.registry.unregister("alice")
     live.registry.remove("alice")
-    page = live.views.dm_display_page("alice", worktree=str(live.root))
+    page = live.views.dm_display_page(ThreadTarget("alice"), worktree=str(live.root))
     assert page.messages
     assert all(isinstance(message, HistoricalMessage) for message in page.messages)
     assert "alice" not in live.registry
diff --git a/tests/test_install_rehearsal.py b/tests/test_install_rehearsal.py
index 62687e963b..7a9698bb5a 100644
--- a/tests/test_install_rehearsal.py
+++ b/tests/test_install_rehearsal.py
@@ -22,9 +22,8 @@
 
 import pytest
 
-from agent_comms.cli_commands import HistoryCliCommand
-from agent_comms.comms import wire
 from agent_comms.field_codec import FieldCodec
+from agent_comms.history_result import ChannelHistoryPageResult
 from agent_comms.installer.backup import bus_files
 from agent_comms.installer.commands import InstallCommands
 from agent_comms.installer.practice import TestHome
@@ -37,9 +36,12 @@
     Stack,
     Upgrade,
 )
+from agent_comms.launch_runtime import LaunchRuntime
 from agent_comms.maintenance_barrier import MaintenanceBarrier
+from agent_comms.messages import Message
 from agent_comms.registry_store import RegistryStore
 from agent_comms.repo import Checkout
+from agent_comms.root_runtime import RootRuntime
 from agent_comms.storage_rules import StorageRule
 from agent_comms.typed_table import ReconciledFile
 from conftest import declare_test_bus, publish_initial
@@ -67,6 +69,16 @@ def installed_home(tmp_path_factory) -> dict[str, str]:
     return environment
 
 
+@pytest.fixture
+def installed_comms(private_comms, installed_home):
+    """An installed machine's bus belongs to its installed runtime, not to the test
+    process that declared it."""
+    with pytest.MonkeyPatch.context() as installed:
+        installed.setenv("HOME", installed_home["HOME"])
+        RootRuntime.publish(private_comms.root, LaunchRuntime.installed())
+    return private_comms
+
+
 @pytest.fixture(scope="module")
 def install_checkout(tmp_path_factory) -> Path:
     """Unchanged targets reuse one built revision; fault cases restore this checkout."""
@@ -79,7 +91,7 @@ def install_checkout(tmp_path_factory) -> Path:
     return checkout
 
 
-def drop_one_index(root: Path) -> str:
+def drop_one_index(root: Path) -> Path:
     """One declared store's schema drifts from both runtimes' declarations."""
     for path in sorted(ReconciledFile.existing_in(root)):
         with closing(sqlite3.connect(path)) as db:
@@ -88,7 +100,7 @@ def drop_one_index(root: Path) -> str:
             if index:
                 db.execute(f'DROP INDEX "{index[0]}"')
                 db.commit()
-                return str(path.relative_to(root))
+                return path.relative_to(root)
     raise AssertionError("No declared store has an index to drop")
 
 
@@ -96,13 +108,13 @@ def drop_one_index(root: Path) -> str:
 @pytest.mark.parametrize("repos, drifted", (
     ((), False), (("--repos",), True),
 ), ids=("kept-repos-current-stores", "given-repos-drifted-store"))
-def test_install_rehearses_without_changing_the_bus(private_comms, installed_home, install_checkout, tmp_path,
+def test_install_rehearses_without_changing_the_bus(installed_comms, installed_home, install_checkout, tmp_path,
                                                     repos, drifted):
     """Retained carry steps inspect a practice copy, even when no step is pending.
 
     Only the deliberately drifted store is rewritten; neither case changes the bus.
     """
-    comms = private_comms
+    comms = installed_comms
     declare_test_bus(comms, channels=("comms",), threads=(
         comms.registry.new_thread(name="worker", tags=frozenset(), worktree=str(tmp_path)),))
     publish_initial(comms, "worker", "#all", "One message before the install")
@@ -125,10 +137,10 @@ def test_install_rehearses_without_changing_the_bus(private_comms, installed_hom
 
 
 @pytest.mark.timeout(280)
-def test_rehearsal_refuses_changed_native_inputs(private_comms, installed_home, install_checkout):
+def test_rehearsal_refuses_changed_native_inputs(installed_comms, installed_home, install_checkout):
     """The target's own installer refuses a stale pin before rehearsing the bus."""
     checkout = install_checkout
-    declare_test_bus(private_comms, channels=("comms",))
+    declare_test_bus(installed_comms, channels=("comms",))
     current = subprocess.check_output(
         ["git", "-C", checkout, "rev-parse", "HEAD"], text=True).strip()
     patch = checkout / "stack/stale-pin-test.py"
@@ -141,7 +153,7 @@ def test_rehearsal_refuses_changed_native_inputs(private_comms, installed_home,
         ["git", "-C", checkout, "rev-parse", "HEAD"], text=True).strip()
     subprocess.run(["git", "-C", checkout, "checkout", "--detach", "-q", current], check=True)
     installed = subprocess.run(
-        [Path(sys.executable).with_name("agent-comms"), "--root", private_comms.root, "install",
+        [Path(sys.executable).with_name("agent-comms"), "--root", installed_comms.root, "install",
          "--rehearse", "--checkout", checkout, "--to", target, "--project", checkout],
         cwd=checkout, env=installed_home, capture_output=True, text=True, timeout=270)
     assert installed.returncode != 0
@@ -150,8 +162,8 @@ def test_rehearsal_refuses_changed_native_inputs(private_comms, installed_home,
 
 @pytest.mark.timeout(280)
 def test_failed_live_migration_restores_data_checkout_and_runtime(
-        private_comms, installed_home, install_checkout, tmp_path):
-    comms, checkout = private_comms, install_checkout
+        installed_comms, installed_home, install_checkout, tmp_path):
+    comms, checkout = installed_comms, install_checkout
     declare_test_bus(comms, channels=("comms",), threads=(
         comms.registry.new_thread(name="worker", tags=frozenset(), worktree=str(tmp_path)),))
     publish_initial(comms, "worker", "#all", "Retained before migration")
@@ -164,8 +176,8 @@ def test_failed_live_migration_restores_data_checkout_and_runtime(
     step = checkout / "tools/cutover/carry_fault.py"
     step.write_text('''from pathlib import Path
 from agent_comms.installer.carry import Carry, Carried
-from agent_comms.locked_store import RootFile
-class Saved(RootFile):
+from agent_comms.locked_store import DurableFile
+class Saved(DurableFile):
     filename = "saved"
 class Fault(Carry):
     opened_files = (Saved,)
@@ -204,9 +216,9 @@ def carry(self, *, apply):
 
 @pytest.mark.timeout(280)
 def test_failed_publication_restores_absent_route_without_store_changes(
-        private_comms, installed_home, install_checkout, tmp_path):
+        installed_comms, installed_home, install_checkout, tmp_path):
     checkout = install_checkout
-    declare_test_bus(private_comms, channels=("comms",))
+    declare_test_bus(installed_comms, channels=("comms",))
     current = subprocess.check_output(
         ["git", "-C", checkout, "rev-parse", "HEAD"], text=True).strip()
     manifest = checkout / "pyproject.toml"
@@ -236,7 +248,7 @@ def test_failed_publication_restores_absent_route_without_store_changes(
     assert not route.exists()
 
     completed = subprocess.run(
-        [pointer / "bin/agent-comms", "--root", private_comms.root, "install",
+        [pointer / "bin/agent-comms", "--root", installed_comms.root, "install",
          "--checkout", checkout, "--to", target, "--project", checkout,
          "--spill-dir", tmp_path / "backups"], cwd=checkout,
         env={**installed_home, "HOME": str(home)}, capture_output=True, text=True, timeout=270,
@@ -249,17 +261,17 @@ def test_failed_publication_restores_absent_route_without_store_changes(
     assert blocker.read_text() == "An owner file that publication must preserve"
     assert subprocess.check_output(
         ["git", "-C", checkout, "rev-parse", "HEAD"], text=True).strip() == current
-    MaintenanceBarrier(RegistryStore.at(private_comms.root).path).read().phase.new_sends.require_open()
+    MaintenanceBarrier(RegistryStore.at(installed_comms.root).path).read().phase.new_sends.require_open()
 
 
 @pytest.mark.timeout(280)
 @pytest.mark.parametrize("fresh, live_kind", ((False, CarriedRun), (True, CreatedRun)),
                          ids=("retained-bus", "fresh-bus"))
 def test_installed_entry_reports_completion_after_target_source_retirement(
-        private_comms, installed_home, install_checkout, tmp_path, fresh, live_kind):
-    root = tmp_path / "new-bus" if fresh else private_comms.root
+        installed_comms, installed_home, install_checkout, tmp_path, fresh, live_kind):
+    root = tmp_path / "new-bus" if fresh else installed_comms.root
     if not fresh:
-        declare_test_bus(private_comms, channels=("comms",))
+        declare_test_bus(installed_comms, channels=("comms",))
     scratch = tmp_path / "install-scratch"
     root.mkdir(mode=0o700, parents=True, exist_ok=True)
     StorageRule.path_in(root).write_text(json.dumps(FieldCodec.encode(
@@ -277,7 +289,11 @@ def test_installed_entry_reports_completion_after_target_source_retirement(
     assert completed.returncode == 0, completed.stderr
     result = FieldCodec.decode(Upgrade, FieldCodec.parse(completed.stdout))
     assert isinstance(result, InstalledUpgrade) and type(result.live) is live_kind
-    history = tuple(reversed(HistoryCliCommand(target="#comms").apply(wire(root)).messages))
+    # The bus now belongs to the installed runtime, so its own CLI reads it.
+    read = subprocess.run([outgoing, "--root", root, "history", "--target", "#comms"],
+                          env=installed_home, capture_output=True, text=True, check=True, timeout=60)
+    page = FieldCodec.parse(read.stdout)[ChannelHistoryPageResult.messages_key]
+    history = tuple(reversed(FieldCodec.decode(tuple[Message, ...], page)))
     assert len(history) == (1 if fresh else 2)
     assert history[-1].sender == "installer"
     assert history[-1].body.startswith("Install finished")
diff --git a/tests/test_install_session_bridge.py b/tests/test_install_session_bridge.py
index 2bc854f706..d9263cefeb 100644
--- a/tests/test_install_session_bridge.py
+++ b/tests/test_install_session_bridge.py
@@ -23,6 +23,7 @@
 from agent_comms.comms import wire
 from agent_comms.field_codec import FieldCodec
 from agent_comms.launch_runtime import LaunchRuntime
+from agent_comms.root_runtime import RootRuntime
 from agent_comms.thread_execution import ExternalThreadExecution
 from agent_comms_hooks.session import started
 from agent_comms_hooks.watch import LapsedWatch, LiveWatch, WatchClaim, WatchRecord
@@ -32,9 +33,12 @@
 
 
 @contextmanager
-def session_bridge(tmp_path, home, program=SERVE_FROM_SOURCE, *, state_directory=None):
+def session_bridge(tmp_path, home, program=SERVE_FROM_SOURCE, *, state_directory=None,
+                   runtime=LaunchRuntime.running()):
+    """``runtime`` serves ``program`` and so owns the bus it attaches to."""
     comms = wire(tmp_path / 'bus')
     Bus.install(Bus.path_in(comms.root))
+    RootRuntime.publish(comms.root, runtime)
     comms.registry.declare(comms.registry.new_thread(
         'bridge-copy', frozenset(), str(tmp_path), execution=ExternalThreadExecution))
     transcript = tmp_path / 'copied-session.jsonl'
@@ -140,7 +144,7 @@ def test_installer_pause_closes_the_real_bridge_and_eof_never_replays_or_relaunc
         (bridge,) = Serve.running(home)
         assert bridge.observed.process == observed.process
         with ExitStack() as resources:
-            (paused,) = bridge.pause(resources)
+            paused = bridge.pause(resources)
             assert paused.bridge == bridge
             assert child.poll() is None
             # The paused process keeps the claim; no session hook sees it lapse.
@@ -165,7 +169,7 @@ def test_installer_discovers_the_live_claim_independent_of_state_directory(
         assert bridge.claim == claim
         assert bridge.observed.process == observed.process
         with ExitStack() as resources:
-            (paused,) = bridge.pause(resources)
+            paused = bridge.pause(resources)
             assert paused.bridge == bridge
             assert isinstance(claim.state(), LiveWatch)
         assert child.wait(timeout=15) == 0, log.read_text()
@@ -202,11 +206,13 @@ def bridges_serving(state):
 def test_install_swap_keeps_the_claim_and_resumes_the_same_process_on_the_new_runtime(
         tmp_path, bridge_home, monkeypatch):
     old, new = (SourcePrograms.create(tmp_path / name) for name in ('old', 'new'))
-    with session_bridge(tmp_path, bridge_home, (str(old.script(claude.main)),)) as (
+    with session_bridge(tmp_path, bridge_home, (str(old.script(claude.main)),), runtime=old.runtime) as (
             home, claim, observed, child, log):
         current = home / LaunchRuntime.current_pointer().relative_to(Path.home())
         current.parent.mkdir(parents=True)
         current.symlink_to(new.prefix.absolute(), target_is_directory=True)
+        # Publication hands the bus to the installed runtime, as the installer's does.
+        RootRuntime.publish(tmp_path / 'bus', new.runtime)
         monkeypatch.setenv('HOME', str(home))
         assert observed.runtime.prefix == old.prefix.resolve()
         env = {**os.environ, 'HOME': str(home)}
diff --git a/tests/test_interrupted_rollback.py b/tests/test_interrupted_rollback.py
index a1400a788f..2ac3e2f9d2 100644
--- a/tests/test_interrupted_rollback.py
+++ b/tests/test_interrupted_rollback.py
@@ -19,6 +19,7 @@
 from agent_comms.installer.upgrade import upgrade
 from agent_comms.locked_store import RegistryFile
 from agent_comms.registry_store import RegistryStore
+from agent_comms.thread_text import GivenThreadText
 from bus_record_fixture import prepare_native_comms
 
 
@@ -29,8 +30,9 @@ def carry(self, *, apply):
         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))
+                edit.document.threads[thread.name] = replace(thread, title=GivenThreadText("Carried title"))
+                edit.commit()
+        return Carried(1, rewritten=tuple(path.relative_to(self.root) for path in registry.written))
 
 
 KILLED_RESTORE = """
@@ -58,9 +60,9 @@ def killed_publication(source, target, **directories):
 
 @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"))
+def test_next_install_completes_a_killed_registry_restore(tmp_path, installed_runtime, publication):
+    # The temporary installation's HOME selects only this test's route and runtime.
+    # Neither process can replace the developer's route.
     comms = prepare_native_comms(wire(tmp_path / "wire"))
     comms.registry.declare(comms.registry.new_thread("worker", frozenset(), str(tmp_path)))
     registry = RegistryStore.at(comms.root)
@@ -85,4 +87,4 @@ def test_next_install_completes_a_killed_registry_restore(tmp_path, monkeypatch,
             False, spill, tmp_path / "install-scratch", lambda: None, InstallReport(),
         )
     assert {path: RegistryStore.digest(path) for path in registry.written} == expected
-    assert comms.registry.require("worker").title != "Carried title"
+    assert comms.registry.require("worker").title != GivenThreadText("Carried title")
diff --git a/tests/test_native_refusal_recovery.py b/tests/test_native_refusal_recovery.py
index 84fac30c40..04faeb05d6 100644
--- a/tests/test_native_refusal_recovery.py
+++ b/tests/test_native_refusal_recovery.py
@@ -1,50 +1,10 @@
 """Native refusal recovery: Core reports one outcome, never blocks on a retry."""
 from agent_comms.native_arguments import SelectedModel
 
-import json
-
 import pytest
 
 import agent_comms.native_pi as native_pi
 import agent_comms.pi_events as pi_events
-from test_backend import _NATIVE_PROMPT_START, _replay_stub
-
-OVERLOADED = "Our servers are currently overloaded. Please try again later."
-
-
-
-def _stub_with(tmp_path, records):
-    lines = "\n".join(json.dumps(record) for record in records)
-    return _replay_stub(
-        tmp_path,
-        lines,
-        startup="def emit(value):\n    print(json.dumps(value), flush=True)\n"
-        + _NATIVE_PROMPT_START
-        + "prompt_request = prompt\n",
-    )
-
-
-def _refused_attempt(attempt, delay_ms):
-    """The recovery announces its retry decision before the refused message_end."""
-    return [
-        {"type": "message_start", "message": {"role": "assistant", "content": []}},
-        {"type": "native_retry_start", "attempt": attempt, "maxAttempts": 3,
-         "delayMs": delay_ms, "errorMessage": OVERLOADED},
-        {"type": "message_end", "message": {"role": "assistant", "stopReason": "error",
-         "errorMessage": OVERLOADED, "content": []}},
-    ]
-
-
-def _settled_records():
-    return [
-        {"type": "agent_settled"},
-        {"type": "response", "command": "get_session_stats", "success": True,
-         "data": {"contextUsage": {}}},
-    ]
-
-
-
-
 
 
 @pytest.mark.asyncio
diff --git a/tests/test_ordinary_nk_delivery.py b/tests/test_ordinary_nk_delivery.py
index 344b26c007..8e46981449 100644
--- a/tests/test_ordinary_nk_delivery.py
+++ b/tests/test_ordinary_nk_delivery.py
@@ -7,6 +7,7 @@
 import os
 
 
+from agent_comms.channel_targets import ThreadTarget
 from agent_comms.child_process import ProcessIdentity
 from conftest import installed_comms
 from test_coordinated_runtime import tmp_path as private_root_fixture
@@ -30,4 +31,4 @@ def test_fresh_send_uses_canonical_publication_and_human_delivery_has_no_wake(tm
     assert isinstance(human.message.role, NoticeRole)
     assert human.audience.recipients == ()
     assert human.decisions == ()
-    assert comms.views.dm_display_page("sender", worktree=str(tmp_path)).messages[-1] == reply
+    assert comms.views.dm_display_page(ThreadTarget("sender"), worktree=str(tmp_path)).messages[-1] == reply
diff --git a/tests/test_outbound_transcript_source.py b/tests/test_outbound_transcript_source.py
index eeefd98c3d..9509a7982b 100644
--- a/tests/test_outbound_transcript_source.py
+++ b/tests/test_outbound_transcript_source.py
@@ -19,7 +19,6 @@
     UnrecordedProgress,
 )
 from agent_comms.pi_payloads import PublishedMessageToolDetails
-from agent_comms.thread_identity import RecordedThreadIncarnation
 from agent_comms.tool_caller import NoThreadCaller
 from agent_comms.tools import invoke_tool
 from agent_comms.transcripts import TranscriptRead
@@ -98,7 +97,7 @@ def test_frozen_unhandled_delivery_projects_recovery_without_claiming(tmp_path):
     beta = next(notice for notice in notices if notice.recipient=='beta')
     assert beta.state == ContextualProgress(UnrecordedProgress())
     assert beta.state.detail == ''
-    assert beta.recipient_identity.recipient_lookup==RecordedThreadIncarnation("fixture", comms.registry.require('beta').created_at).lookup
+    assert beta.recipient_identity.recipient_lookup==comms.registry.require('beta').incarnation.require_recorded().lookup
     with Coordination(str(root/'coordination.sqlite3')) as store:
         assert WakeAssignment.select(store.session._connection)==before
     assert not any(row.wire_seq==message.seq for row in before)
@@ -148,7 +147,7 @@ def test_publication_intent_joins_original_sender_and_target_only(tmp_path):  #
         comms.root, comms.registry.require("beta"), comms.bus.bus
     )
     original = source.rows(sequences=(reply.seq,))
-    lookup = RecordedThreadIncarnation("fixture", source.recipient.created_at).lookup
+    lookup = source.recipient.require_recorded().lookup
     intent = TurnRouting((initial.message.reference,), MessageRoute("beta", ("sender", "alpha")))
     # A later send can fail; this first committed subset stays a valid relation.
     intent.require_publications(original, lookup)
diff --git a/tests/test_read_ledger.py b/tests/test_read_ledger.py
index 15d8ec974f..048c70443c 100644
--- a/tests/test_read_ledger.py
+++ b/tests/test_read_ledger.py
@@ -7,7 +7,7 @@
 import pytest
 
 from agent_comms.bus import Bus
-from agent_comms.channel_targets import CommandTarget
+from agent_comms.channel_targets import CommandTarget, ThreadTarget
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.comms import wire
 from agent_comms.coordination_errors import DMPaintedBoundChanged, StaleRevision
@@ -189,10 +189,10 @@ def test_turn_lease_and_unrelated_registry_changes_do_not_invalidate_dm(tmp_path
     comms.registry.register(replace(peer, process_identity=ProcessIdentity.capture(os.getpid())))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("alice", viewer, "painted").message.message_id
-    page = comms.views.dm_display_page("alice", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path))
     assert page.selection.basis.peer_identity == peer.incarnation
     comms.registry.lease_local_turn("alice", "ordinary-turn")
-    fresh = comms.views.dm_display_page("alice", worktree=str(tmp_path))
+    fresh = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path))
     assert fresh.selection.basis.peer_identity == page.selection.basis.peer_identity
     comms.registry.register(comms.registry.new_thread('unrelated', frozenset(), str(tmp_path)))
     comms.views.mark_page_read(page, worktree=str(tmp_path))
@@ -211,9 +211,10 @@ def test_dm_ack_survives_process_exit_and_rebind_does_not_inherit_reads(tmp_path
             sys.executable,
             "-c",
             "import os, sys\n"
+            "from agent_comms.channel_targets import ThreadTarget\n"
             "from agent_comms.comms import wire\n"
             "comms = wire(sys.argv[1])\n"
-            "page = comms.views.dm_display_page('alice', worktree=sys.argv[1])\n"
+            "page = comms.views.dm_display_page(ThreadTarget('alice'), worktree=sys.argv[1])\n"
             "comms.views.mark_page_read(page, worktree=sys.argv[1])\n"
             "os._exit(9)\n",
             str(tmp_path),
@@ -353,7 +354,7 @@ def test_recorded_bus_identity_is_one_member(tmp_path):
     assert not ledger.read().bus_identity.matches(ObservedPath.at(comms.bus.bus.path))
     assert comms.views.viewer_snapshot(str(tmp_path)).unread["alice"].count == 1
 
-    page = comms.views.dm_display_page("alice", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path))
     comms.views.mark_page_read(page, worktree=str(tmp_path))
     recorded = RecordedBus(FileIdentity.from_stat(comms.bus.bus.path.stat()))
     assert ledger.read().bus_identity == recorded == page.selection.basis.bus_identity
@@ -374,7 +375,7 @@ def test_recorded_bus_identity_is_one_member(tmp_path):
     assert reopened.views.viewer_snapshot(str(tmp_path)).unread["alice"].count == 1
     with pytest.raises(DMScopeChanged):
         reopened.views.mark_page_read(page, worktree=str(tmp_path))
-    fresh = reopened.views.dm_display_page("alice", worktree=str(tmp_path))
+    fresh = reopened.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path))
     reopened.views.mark_page_read(fresh, worktree=str(tmp_path))
     assert reopened.views.viewer_snapshot(str(tmp_path)).unread["alice"].count == 0
 
@@ -447,7 +448,7 @@ def current_cursor(facts):
             comms.channels.set_channel_any_mode("#team", rng.choice([True, False]))
         elif operation == "display":
             if rng.choice([True, False]):
-                page = comms.views.dm_display_page("alice", worktree=str(tmp_path), bounds=PageBounds(limit=3))
+                page = comms.views.dm_display_page(ThreadTarget("alice"), worktree=str(tmp_path), bounds=PageBounds(limit=3))
                 basis = page.selection.basis.displayed
                 kind, target = "dm", "alice"
                 proof = page.selection.basis
diff --git a/tests/test_store_backup.py b/tests/test_store_backup.py
index 26005f4dbd..05b7ca9fae 100644
--- a/tests/test_store_backup.py
+++ b/tests/test_store_backup.py
@@ -130,6 +130,7 @@ def test_store_backup_restores_wal_rows_and_removes_new_files(tmp_path):
     root.mkdir()
     store = ExternalTranscript.path_in(root)
     route = tmp_path / 'route.json'
+    route.touch(mode=0o600)  # The route's owner writes it private.
     route.write_text('original route')
     created = root / 'created.json'
     with sqlite3.connect(store) as writer:
@@ -223,6 +224,7 @@ def test_route_only_backup_preserves_a_concurrent_change(tmp_path, had_route):
     root.mkdir()
     route = tmp_path / 'route.json'
     if had_route:
+        route.touch(mode=0o600)  # The route's owner writes it private.
         route.write_text('original route')
     spill_dir = tmp_path / 'spill'
     backup = StoreBackup.take(root, route, (), spill_dir)
diff --git a/tests/test_thread_identity.py b/tests/test_thread_identity.py
index 1aa92e9cc5..ec25b4610a 100644
--- a/tests/test_thread_identity.py
+++ b/tests/test_thread_identity.py
@@ -7,6 +7,7 @@
 
 import pytest
 
+from agent_comms.channel_targets import ThreadTarget
 from agent_comms.session_file import SavedSessionFile
 from agent_comms.child_process import NoProcess, ProcessIdentity
 from agent_comms.comms import wire
@@ -248,7 +249,7 @@ def test_saved_read_ledger_survives_reopen_and_new_ack(tmp_path):
     comms.registry.declare(comms.registry.new_thread('owner', frozenset(), str(tmp_path)))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     message = comms.messaging.send("owner", viewer, "already painted").message
-    page = comms.views.dm_display_page("owner", worktree=str(tmp_path))
+    page = comms.views.dm_display_page(ThreadTarget("owner"), worktree=str(tmp_path))
     comms.views.mark_page_read(page, worktree=str(tmp_path))
     path = comms.bus.reads.path
     document = FieldCodec.decode(ReadDocument, json.loads(path.read_text()))
@@ -262,7 +263,7 @@ def test_saved_read_ledger_survives_reopen_and_new_ack(tmp_path):
     conversation = reopened.bus.reads.conversation(message, snapshot)
     assert message.seq in reopened.bus.reads.cursor(viewer, snapshot).for_conversation(conversation)
     next_message = reopened.messaging.send("owner", viewer, "new paint").message
-    page = reopened.views.dm_display_page("owner", worktree=str(tmp_path))
+    page = reopened.views.dm_display_page(ThreadTarget("owner"), worktree=str(tmp_path))
     reopened.views.mark_page_read(page, worktree=str(tmp_path))
     seen = reopened.bus.reads.cursor(viewer, reopened.registry.snapshot()).for_conversation(conversation)
     assert all(sequence in seen for sequence in (message.seq, next_message.seq))
diff --git a/tests/test_user_thread_lookup.py b/tests/test_user_thread_lookup.py
index dadc74e996..91b56dd0d7 100644
--- a/tests/test_user_thread_lookup.py
+++ b/tests/test_user_thread_lookup.py
@@ -1,6 +1,6 @@
 """The stored registry revision owns human identity and UI history uses it."""
 
-from agent_comms.channel_targets import CommandTarget
+from agent_comms.channel_targets import CommandTarget, ThreadTarget
 
 
 from dataclasses import replace
@@ -41,7 +41,7 @@ def test_channel_and_dm_history_keep_the_human_as_sender(tmp_path):
     assert channel.sender_role is direct.sender_role is UserThreadRole
     channel_page = comms.views.channel_display_page(
         '#comms', viewer=comms.views.display_viewer(str(tmp_path)))
-    direct_page = comms.views.dm_display_page('peer', worktree=str(tmp_path))
+    direct_page = comms.views.dm_display_page(ThreadTarget('peer'), worktree=str(tmp_path))
     assert channel.message_id in {message.message_id for message in channel_page.messages}
     assert direct.message_id in {message.message_id for message in direct_page.messages}
     assert all(message.sender == human.name for message in (*channel_page.messages, *direct_page.messages))
diff --git a/tools/cutover/thread_name_owners.py b/tools/cutover/thread_name_owners.py
index 0b3612b26e..49aecbae1b 100644
--- a/tools/cutover/thread_name_owners.py
+++ b/tools/cutover/thread_name_owners.py
@@ -17,7 +17,7 @@
 from agent_comms.coordination_tables.participants import OwnerGenerations, Participants
 from agent_comms.errors import RelationViolationError
 from agent_comms.installer.carry import Carried, Carry
-from agent_comms.locked_store import RootFile, SQLiteFile
+from agent_comms.locked_store import DurableFile, SQLiteFile
 from agent_comms.native_prompt_binding import PromptBinding
 from agent_comms.native_runtime_input import CurrentNativeCursor, NativeRuntimeInput
 from agent_comms.private_sidecar import (
@@ -27,7 +27,7 @@
 from agent_comms.typed_table import HeldBusFile, SQLiteCatalogObject, SQLiteSchemaObject
 
 
-class UnfinishedOwnerRename(RootFile):
+class UnfinishedOwnerRename(DurableFile):
     filename = ".private-owner-rename.pending"
 
     @classmethod