diff --git a/src/agent_comms/check_root.py b/src/agent_comms/check_root.py
index f122b9fb59..e16e28a9e1 100644
--- a/src/agent_comms/check_root.py
+++ b/src/agent_comms/check_root.py
@@ -27,7 +27,7 @@
 from .registry_document import RosterVisibility
 from .storage_rules import StorageRoots
 from .thread_text import NoThreadText
-from .session_file import SavedSessionFile
+from .session_file import RegistrySessionLocation
 from .threads import Thread
 from .transcript_routes import TranscriptRoutes
 from .turn_lease import NoTurnRecord
@@ -114,7 +114,7 @@ def _populate(self, copied: Comms) -> None:
                 task=NoThreadText(),
                 goal=NoGoal(),
                 process_identity=NoProcess(),
-                session_file=SavedSessionFile(session),
+                session_file=RegistrySessionLocation.from_path(session, copied.root),
                 earlier_projects=(),
                 project_resume_due=False,
                 storage_roots=StorageRoots(copied.root / "scratch" / slot),
diff --git a/src/agent_comms/fresh_private_session.py b/src/agent_comms/fresh_private_session.py
index c7e29b75a9..cae912ec38 100644
--- a/src/agent_comms/fresh_private_session.py
+++ b/src/agent_comms/fresh_private_session.py
@@ -12,13 +12,13 @@
 from .native_entries import SessionEntry
 from .native_failures import NativePiUnavailable
 from agent_comms.native_session_files import NativeSessionFiles
-from .native_session_files import PrivateSession, SavedSessionFile
+from .native_session_files import PrivateSession, BusTranscriptFile
 from .native_session_reopen import NativeSessionIdentity
 from .private_path import NativeSessionFileRole
 
 
 def create_fresh_private_session(
-    root: Path, *, worktree: Path, location: type[SavedSessionFile] = PrivateSession, **parts: str
+    root: Path, *, worktree: Path, location: type[BusTranscriptFile] = PrivateSession, **parts: str
 ) -> NativeSessionIdentity:
     """Exclusively create and sync the SDK header; never reuse an uncertain file."""
     worktree = Path(worktree).absolute()
diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py
index 5fe859c7bd..fc4927035d 100644
--- a/src/agent_comms/historical_views.py
+++ b/src/agent_comms/historical_views.py
@@ -259,6 +259,10 @@ class HistorySource:
     snapshot_registry_revision: FileObservation
     provenance: RegistryProvenance
 
+    def __post_init__(self) -> None:
+        # Archived transcripts stay with the original root; the snapshot copies no sessions.
+        object.__setattr__(self, "provenance", self.provenance.bound_to(Path(self.original_root)))
+
     @property
     def key(self) -> str:
         return self.root
diff --git a/src/agent_comms/manual_compaction_bridge.py b/src/agent_comms/manual_compaction_bridge.py
index 1ce81cabfa..0f176f5b4b 100644
--- a/src/agent_comms/manual_compaction_bridge.py
+++ b/src/agent_comms/manual_compaction_bridge.py
@@ -73,7 +73,7 @@ async def compact_context(
         if await Coordination.run_worker(partial(runner.session_busy, session_id)):
             return FailedCompaction("Wait for the current response before compacting.")
         thread = await Coordination.run_worker(partial(runner.comms.registry.require, thread_name))
-        return await thread.runtime.compact(partial(
+        return await thread.writable_runtime.compact(partial(
             _compact_started, runner, session_id, thread_name, instructions,
         ))
 
diff --git a/src/agent_comms/native_entries.py b/src/agent_comms/native_entries.py
index bd8ce98f70..79bcaeb0fc 100644
--- a/src/agent_comms/native_entries.py
+++ b/src/agent_comms/native_entries.py
@@ -15,7 +15,7 @@
 from typing import TYPE_CHECKING, Annotated, ClassVar, Literal, TextIO, cast
 
 from .locked_store import DurableFile
-from .native_session_files import SavedSessionFile
+from .native_session_files import BusTranscriptFile
 
 if TYPE_CHECKING:
     from .native_pi import NativeContextProof
@@ -189,7 +189,7 @@ def unread_reply(self) -> bool:
 
 
 @dataclass
-class ImportedSessionWriter(DurableFile, SavedSessionFile):
+class ImportedSessionWriter(DurableFile, BusTranscriptFile):
     filename = "imported_sessions/{session_id}.jsonl"
     stream: TextIO
     parent: NativeEntry
diff --git a/src/agent_comms/native_session_files.py b/src/agent_comms/native_session_files.py
index 1c1f20c073..f8235a63f9 100644
--- a/src/agent_comms/native_session_files.py
+++ b/src/agent_comms/native_session_files.py
@@ -217,7 +217,7 @@ def table_schema(cls) -> type[TableSchema]:
         return NativeContextJournal
 
 
-class SavedSessionFile(PatternFile):
+class BusTranscriptFile(PatternFile):
     """A bus location holding saved native transcripts and their writer locks."""
 
     __slots__ = ()
@@ -246,14 +246,14 @@ def sessions_in(cls, root: Path) -> tuple[NativeSessionLocks, ...]:
         )
 
 
-class PrivateSession(DurableFile, SavedSessionFile):
+class PrivateSession(DurableFile, BusTranscriptFile):
     """A thread's own session, created fresh for it."""
 
     __slots__ = ()
     filename = "sessions/enrolled-{session_id}.jsonl"
 
 
-class CheckSession(DurableFile, SavedSessionFile):
+class CheckSession(DurableFile, BusTranscriptFile):
     """A check root's copy of one selected worker's history, under its record's slot."""
 
     __slots__ = ()
diff --git a/src/agent_comms/pi_durable_backend.py b/src/agent_comms/pi_durable_backend.py
index e42d786097..ca52fa8b62 100644
--- a/src/agent_comms/pi_durable_backend.py
+++ b/src/agent_comms/pi_durable_backend.py
@@ -223,10 +223,10 @@ def new_location(cls, root, worktree):
     async def open(self, thread):
         async with self.lock:
             if self.child is None:
-                path = Path(thread.runtime.path)
+                path = Path(thread.writable_runtime.path)
                 path.parent.mkdir(parents=True, exist_ok=True)
                 with ExitStack() as resources:
-                    lock = resources.enter_context(thread.runtime.writer_lock.open('a+b'))
+                    lock = resources.enter_context(thread.writable_runtime.writer_lock.open('a+b'))
                     fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
                     child = await PiChild.start_command(('node', str(Path(__file__).with_name('pi_durable_host.mjs'))),
                         cwd=thread.worktree, pass_fds=(lock.fileno(),), event_type=DurablePiEvent, resources=resources.pop_all())
diff --git a/src/agent_comms/pi_native_backend.py b/src/agent_comms/pi_native_backend.py
index 8c8ee18764..0a3a2ceae8 100644
--- a/src/agent_comms/pi_native_backend.py
+++ b/src/agent_comms/pi_native_backend.py
@@ -243,7 +243,7 @@ async def prepare(self, thread: Thread, *, open_native=None) -> StateData:
             owner=thread.incarnation.require_recorded(),
             session_directory=thread.native_session_directory(self.comms.root),
             environment=environment,
-            session_file=thread.runtime.read_key,
+            session_file=thread.writable_runtime.read_key,
             observe=partial(self.observe, thread),
         )
 
@@ -269,7 +269,7 @@ async def run_turn(
             delivery.environment,
             **images,
             context_contributions=content.contributions,
-            **thread.runtime.arguments(),
+            **thread.writable_runtime.arguments(),
             steering_queue=delivery.inbox,
             finish_event=delivery.finish_event,
             send_boundary=delivery.send_boundary,
diff --git a/src/agent_comms/registry_provenance.py b/src/agent_comms/registry_provenance.py
index c0c6333630..a0e70e4d19 100644
--- a/src/agent_comms/registry_provenance.py
+++ b/src/agent_comms/registry_provenance.py
@@ -6,8 +6,9 @@
 import re
 from abc import ABC, abstractmethod
 from collections.abc import Iterator, Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from functools import cached_property, partial
+from pathlib import Path
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Generic, TypeVar
 
@@ -171,6 +172,15 @@ def __post_init__(self) -> None:
         if any(target not in names or alias in names for alias, target in self.aliases.items()):
             raise RelationViolationError("Provenance alias has no exclusive retained owner")
 
+    def bound_to(self, root: Path) -> RegistryProvenance:
+        """Recorded transcripts resolve under the root this namespace was captured from."""
+        threads = {name: replace(thread, session_file=thread.session_file.bound_to(root))
+                   for name, thread in self.threads.items()}
+        if all(threads[name].session_file is thread.session_file
+               for name, thread in self.threads.items()):
+            return self
+        return replace(self, threads=threads)
+
     @cached_property
     def sender_bindings(self) -> Mapping[str, str]:
         """One lazy SQL value for this namespace, including its historical names."""
diff --git a/src/agent_comms/registry_store.py b/src/agent_comms/registry_store.py
index a664d2fc1b..85d78e76e1 100644
--- a/src/agent_comms/registry_store.py
+++ b/src/agent_comms/registry_store.py
@@ -32,7 +32,7 @@
 from .routing import TurnRouting
 from .store_files import SharedLock
 from .thread_identity import TurnIdentity
-from .session_file import SavedSessionFile
+from .session_file import RegistrySessionLocation
 from .threads import Thread
 from .turn_lease import ActiveTurn, RecordedGeneration, StoredTurnRecord
 
@@ -147,8 +147,9 @@ class RegistryEdit:
     def prepare_native_source(self, owner: RegistryOwner, session_file: str) -> NativeSourcePublication:
         current = owner.require_source_snapshot(self.original.snapshot)
         status = self.document.status(current.name)
+        location = RegistrySessionLocation.from_path(session_file, self.store.path.parent)
         return NativeSourcePublication(
-            thread=replace(current, session_file=SavedSessionFile(session_file)), status=status,
+            thread=replace(current, session_file=location), status=status,
             previous=current, previous_status=status,
         )
 
@@ -284,9 +285,11 @@ def _encode(self, value: RegistryDocument) -> dict:
         return FieldCodec.encode(value)
 
     def _bind_thread(self, thread: Thread) -> Thread:
-        record = thread.turn_record
-        bound = record.bound_to(self.path, thread.turn_identity)
-        return thread if bound is record else replace(thread, turn_record=bound)
+        record = thread.turn_record.bound_to(self.path, thread.turn_identity)
+        session = thread.session_file.bound_to(self.path.parent)
+        if record is thread.turn_record and session is thread.session_file:
+            return thread
+        return replace(thread, turn_record=record, session_file=session)
 
     def _bind_document(self, document: RegistryDocument) -> RegistryDocument:
         for name, thread in document.threads.items():
@@ -312,6 +315,7 @@ def mark_historical(self) -> None:
 
     def save_unlocked(self, document: RegistryDocument) -> None:
         self.read_revision_unlocked().document.purpose.require_write()
+        document = self._bind_document(document)
         raw = self._encode(document)
         self._write_unlocked(json.dumps(raw, indent=2))
         # Still under the exclusive lock: the written file is this document.
diff --git a/src/agent_comms/session_file.py b/src/agent_comms/session_file.py
index 94b29c0d35..1449840427 100644
--- a/src/agent_comms/session_file.py
+++ b/src/agent_comms/session_file.py
@@ -1,47 +1,163 @@
-"""Saved-session representation owns binding to the selected runtime."""
+"""The registry owns whether a saved transcript follows its bus or its harness."""
 
 from __future__ import annotations
 
 from abc import abstractmethod
-from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING, Annotated
 
-from .field_codec import SubclassUnion
-from .thread_text import AbsentText, GivenText, TextChoice
+from .active_route import CommsRoute
+from .errors import RelationViolationError
+from .field_codec import NullMember, PathText, RecordMember, SubclassUnion, WireTransient
 from .unstarted_runtime import UnstartedRuntime
 
 if TYPE_CHECKING:
     from .agent_backend import AgentBackend, RuntimeLocation
 
 
-class SessionFile(SubclassUnion, TextChoice, affix="SessionFile"):
-    """A thread's saved native session file, or null before its runtime first saves one."""
+class SessionFile(SubclassUnion, affix="SessionFile"):
+    """A registry transcript location, or no saved transcript yet."""
 
     @abstractmethod
-    def location(self, backend: type[AgentBackend]) -> RuntimeLocation:
-        """The runtime this file binds under the thread's backend."""
+    def bound_to(self, root: Path) -> SessionFile:
+        """The location read from the registry under ``root``."""
 
     @abstractmethod
-    def applied(self, previous: SessionFile) -> SessionFile:
-        """The file a redeclaration keeps: its own, or the previous one."""
+    def location(self, backend: type[AgentBackend]) -> RuntimeLocation: ...
+
+    @abstractmethod
+    def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation: ...
 
+    @abstractmethod
+    def applied(self, previous: SessionFile) -> SessionFile: ...
 
-@dataclass(frozen=True, slots=True)
-class SavedSessionFile(GivenText, SessionFile):
-    text: str
+
+@dataclass(frozen=True, slots=True, eq=False)
+class RegistrySessionLocation(RecordMember, SessionFile, declared_name=None):
+    """A saved transcript whose storage owner was selected at publication."""
+
+    path: Annotated[Path, PathText]
+
+    def __eq__(self, other: object) -> bool:
+        """A bus path is relative and an external one absolute, so the path is the location;
+        the root a registry bound it to says where it is read, not which one it is."""
+        return isinstance(other, RegistrySessionLocation) and self.path == other.path
+
+    def __hash__(self) -> int:
+        return hash(self.path)
+
+    @classmethod
+    def from_path(cls, path: Path | str, root: Path) -> BoundSessionLocation:
+        absolute = Path(path)
+        if not absolute.is_absolute():
+            raise ValueError(f"Saved session publication requires an absolute path: {path}")
+        absolute = absolute.resolve()
+        root = root.resolve()
+        if absolute.is_relative_to(root):
+            return BusSessionLocation(absolute.relative_to(root)).bound_to(root)
+        return ExternalSessionLocation(absolute).bound_to(root)
 
     def location(self, backend: type[AgentBackend]) -> RuntimeLocation:
-        return backend.location_type(self.text)
+        raise RelationViolationError(f"Saved session {self.path} must be bound by its registry")
+
+    def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation:
+        return self.location(backend)
 
     def applied(self, previous: SessionFile) -> SessionFile:
         return self
 
+    @abstractmethod
+    def resolve(self, root: Path) -> Path: ...
+
+    def writable(self, root: Path) -> Path:
+        return self.resolve(root)
+
+
+@dataclass(frozen=True, eq=False)
+class BoundSessionLocation(RegistrySessionLocation, declared_name=None):
+    """A wire location bound to the root whose registry decoded it."""
+
+    root: Annotated[Path, WireTransient()] = field(repr=False)
+
+    def bound_to(self, root: Path) -> BoundSessionLocation:
+        """Publishing into another root keeps the transcript this location names."""
+        if self.root == root or self.root == (root := root.resolve()):
+            return self
+        return RegistrySessionLocation.from_path(self.resolve(self.root), root)
 
-@dataclass(frozen=True, slots=True)
-class NoSessionFile(AbsentText, SessionFile):
     def location(self, backend: type[AgentBackend]) -> RuntimeLocation:
+        return self.located(backend, self.resolve(self.root))
+
+    def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation:
+        return self.located(backend, self.writable(self.root))
+
+    @staticmethod
+    def located(backend: type[AgentBackend], path: Path) -> RuntimeLocation:
+        return backend.location_type(str(path))
 
+
+@dataclass(frozen=True, slots=True, eq=False)
+class BusSessionLocation(RegistrySessionLocation):
+    """A transcript stored under the bus root; a copied root reads its own copy."""
+
+    def __post_init__(self) -> None:
+        if self.path.is_absolute() or ".." in self.path.parts:
+            raise ValueError(f"Bus session must be relative to its root: {self.path}")
+
+    def bound_to(self, root: Path) -> BoundSessionLocation:
+        return BoundBusSession(self.path, root.resolve())
+
+    def resolve(self, root: Path) -> Path:
+        root = root.resolve()
+        path = (root / self.path).resolve()
+        if not path.is_relative_to(root):
+            raise ValueError(f"Bus session escapes its root: {self.path}")
+        return path
+
+
+@dataclass(frozen=True, slots=True, eq=False)
+class ExternalSessionLocation(RegistrySessionLocation):
+    """A harness transcript outside every bus root; only the published root writes it."""
+
+    def __post_init__(self) -> None:
+        if not self.path.is_absolute():
+            raise ValueError(f"External session must be absolute: {self.path}")
+
+    def bound_to(self, root: Path) -> BoundSessionLocation:
+        return BoundExternalSession(self.path, root.resolve())
+
+    def resolve(self, root: Path) -> Path:
+        return self.path
+
+    def writable(self, root: Path) -> Path:
+        if root.resolve() != CommsRoute.current().observe_root():
+            raise RelationViolationError(
+                f"Copied bus {root} cannot write external session {self.path}"
+            )
+        return self.resolve(root)
+
+
+@dataclass(frozen=True, eq=False)
+class BoundBusSession(BoundSessionLocation, BusSessionLocation, declared_name=None):
+    pass
+
+
+@dataclass(frozen=True, eq=False)
+class BoundExternalSession(BoundSessionLocation, ExternalSessionLocation, declared_name=None):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class NoSessionFile(NullMember, SessionFile):
+    def bound_to(self, root: Path) -> NoSessionFile:
+        return self
+
+    def location(self, backend: type[AgentBackend]) -> RuntimeLocation:
         return UnstartedRuntime(backend)
 
+    def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation:
+        return self.location(backend)
+
     def applied(self, previous: SessionFile) -> SessionFile:
         return previous
diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py
index 556a3c1afa..7fed8114a3 100644
--- a/src/agent_comms/thread_management.py
+++ b/src/agent_comms/thread_management.py
@@ -43,7 +43,7 @@
 )
 from .registry_document import RegistrySnapshot
 from .registry_provenance import RegistryNames
-from .session_file import SavedSessionFile
+from .session_file import RegistrySessionLocation
 from .store_files import AtomicFile
 from .thread_identity import AgentThreadRole, ThreadRole
 from .thread_model import DeclaredModel, ThreadModel, UndeclaredModel
@@ -160,7 +160,8 @@ def prepared(self, threads: ThreadManagement, pi_bin: str, runtime: LaunchRuntim
         threads.registry.require_declaration_available(thread)
         location = thread.runtime.new_location(threads.root, project)
         with threads.registry.store.locked():
-            yield replace(thread, session_file=SavedSessionFile(location.read_key))
+            saved = RegistrySessionLocation.from_path(location.read_key, threads.root)
+            yield replace(thread, session_file=saved)
 
 
 @dataclass(frozen=True, slots=True, kw_only=True)
@@ -175,16 +176,17 @@ def _parent(self, threads: ThreadManagement) -> Thread:
         parent.runtime.require_forkable()
         return parent
 
-    def child(self, parent: Thread, runtime: RuntimeLocation, registry: Registration) -> Thread:
-        return replace(self.declaration(self.project, registry), parent=ForkedParent(parent.name),
-                       tags=self.tags.for_child(parent.tags),
-                       backend=parent.backend, session_file=SavedSessionFile(runtime.read_key))
+    def child(self, parent: Thread, runtime: RuntimeLocation, threads: ThreadManagement) -> Thread:
+        saved = RegistrySessionLocation.from_path(runtime.read_key, threads.root)
+        return replace(self.declaration(self.project, threads.registry),
+                       parent=ForkedParent(parent.name), tags=self.tags.for_child(parent.tags),
+                       backend=parent.backend, session_file=saved)
 
     @contextmanager
     def prepared(self, threads: ThreadManagement, pi_bin: str, runtime: LaunchRuntime) -> Iterator[Thread]:
         # Copy saved history before locking the registry for its declaration.
         parent = self._parent(threads)
-        threads.registry.require_declaration_available(self.child(parent, parent.runtime, threads.registry))
+        threads.registry.require_declaration_available(self.child(parent, parent.runtime, threads))
         location = asyncio.run(parent.backend.fork(parent, pi_bin, runtime))
         with threads.registry.store.locked():
             current = self._parent(threads)
@@ -193,7 +195,7 @@ def prepared(self, threads: ThreadManagement, pi_bin: str, runtime: LaunchRuntim
                     f"Parent thread {self.parent!r} changed its saved session during the fork; "
                     f"the created fork {location} was not registered."
                 )
-            yield self.child(current, location, threads.registry)
+            yield self.child(current, location, threads)
 
 
 @dataclass(frozen=True, slots=True)
@@ -246,7 +248,7 @@ def import_thread(
             name,
             tags,
             str(project),
-            session_file=SavedSessionFile(str(session_path)),
+            session_file=RegistrySessionLocation.from_path(session_path, self.root),
             title=snapshot.title,
         ))
         with self.registry.store.locked():
@@ -455,7 +457,7 @@ def attach_session(
                     model=saved_model, thinking_level=NoThinkingOverride,
                 ),
                 process_identity=process,
-                session_file=SavedSessionFile(location.read_key),
+                session_file=RegistrySessionLocation.from_path(location.read_key, self.root),
             )
             self.registry.register(attached)
             return attached
diff --git a/src/agent_comms/threads.py b/src/agent_comms/threads.py
index 97838548f7..c13ec74ffe 100644
--- a/src/agent_comms/threads.py
+++ b/src/agent_comms/threads.py
@@ -199,6 +199,11 @@ def runtime(self) -> RuntimeLocation:
         """The codec decoded both stored fields; their record owns this binding."""
         return self.session_file.location(self.backend)
 
+    @cached_property
+    def writable_runtime(self) -> RuntimeLocation:
+        """The runtime a launch writes; a copied bus refuses a harness transcript."""
+        return self.session_file.writable_location(self.backend)
+
     @property
     def display_title(self) -> str:
         return self.title.otherwise(self.name)
diff --git a/src/agent_comms/transcripts.py b/src/agent_comms/transcripts.py
index 487fce13f3..c940ac6a4e 100644
--- a/src/agent_comms/transcripts.py
+++ b/src/agent_comms/transcripts.py
@@ -255,6 +255,16 @@ class TranscriptReadIdentity(DeclaredFamily, affix="TranscriptReadIdentity"):
     after: TranscriptCursor | None
     through: TranscriptCursor | None
 
+    def __post_init__(self) -> None:
+        # A decoded witness reads its transcript under the root that published its thread.
+        bound = self.thread.session_file.bound_to(self.transcript_root)
+        if bound is not self.thread.session_file:
+            object.__setattr__(self, "thread", replace(self.thread, session_file=bound))
+
+    @property
+    def transcript_root(self) -> Path:
+        return Path(self.root)
+
     @projected(view="content", name="thread")
     def content_thread(self) -> ThreadIncarnation:
         return self.thread.incarnation
@@ -352,6 +362,10 @@ def publish_turn(self, publish: Callable[[str, Thread], None]) -> None:
 class RecordedTranscriptReadIdentity(TranscriptReadIdentity):
     source: Annotated[HistorySource, ContentExclude()] = field(kw_only=True)
 
+    @property
+    def transcript_root(self) -> Path:
+        return Path(self.source.original_root)
+
     @property
     def historical_source(self) -> str:
         return self.source.key
diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py
index 0d0764391e..82eb125304 100644
--- a/src/agent_comms/work_dispatch.py
+++ b/src/agent_comms/work_dispatch.py
@@ -29,7 +29,7 @@
 from .repository_name import RepositoryName
 from .storage_rules import CheckoutStorageEffect, StorageProjection
 from .thread_status import StoppedThreadStatus
-from .session_file import SavedSessionFile
+from .session_file import RegistrySessionLocation
 
 if TYPE_CHECKING:
     from .comms import Comms
@@ -85,7 +85,10 @@ def start(self, comms: Comms) -> Thread:
         )
         location = candidate.runtime.new_location(comms.root, checkout)
         child = comms.registry.declare(
-            replace(candidate, session_file=SavedSessionFile(location.read_key)),
+            replace(
+                candidate,
+                session_file=RegistrySessionLocation.from_path(location.read_key, comms.root),
+            ),
             StoppedThreadStatus(),
         )
         # A failed publication leaves a stopped unit and its checkout, not an
diff --git a/tests/acp_error_payload_installed_journey.py b/tests/acp_error_payload_installed_journey.py
index c64352e38d..e7a2b7817f 100644
--- a/tests/acp_error_payload_installed_journey.py
+++ b/tests/acp_error_payload_installed_journey.py
@@ -9,7 +9,7 @@
 
 from agent_comms.bus import Bus
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.pi_vocabulary import OffThinkingLevel
 import argparse
 import asyncio
@@ -128,7 +128,7 @@ def log_message(self, *_):
     comms = wire(root)
     Bus.install(Bus.path_in(comms.root))
     fresh = create_fresh_private_session(evidence / "sessions", worktree=project)
-    comms.registry.declare(comms.registry.new_thread('error-native', frozenset(), str(project), session_file=SavedSessionFile(str(fresh.path)), model=DeclaredModel('error-local/error-fixture'), thinking_level=OffThinkingLevel))
+    comms.registry.declare(comms.registry.new_thread('error-native', frozenset(), str(project), session_file=RegistrySessionLocation.from_path(str(fresh.path), comms.root), model=DeclaredModel('error-local/error-fixture'), thinking_level=OffThinkingLevel))
     environment = {key: value for key, value in os.environ.items()
                    if not key.startswith(("PI_", "AGENT_COMMS_")) and key != "PYTHONPATH"}
     environment.update(
diff --git a/tests/c4_coordination_installed_ui.py b/tests/c4_coordination_installed_ui.py
index 000dd8f7d8..ffdf4a4749 100644
--- a/tests/c4_coordination_installed_ui.py
+++ b/tests/c4_coordination_installed_ui.py
@@ -3,7 +3,7 @@
 
 from agent_comms.bus import Bus
 from conftest import publish_initial
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.channel_targets import CommandTarget
 from agent_comms.read_basis import NamedViewer
 from agent_comms.presentation import ReadThrough
@@ -47,7 +47,7 @@ async def main():
     for name in (owner, 'stopped', 'archived'):
         source = root / (name + '.jsonl')
         source.touch()
-        comms.registry.declare(comms.registry.new_thread(name, frozenset({'team'}), str(root), session_file=SavedSessionFile(str(source))))
+        comms.registry.declare(comms.registry.new_thread(name, frozenset({'team'}), str(root), session_file=RegistrySessionLocation.from_path(str(source), comms.root)))
     comms.owners.stop('stopped')
     comms.owners.stop('archived')
     comms.threads.archive('archived')
diff --git a/tests/reply_relevance_installed_journey.py b/tests/reply_relevance_installed_journey.py
index 89b3e64591..7076cd3bd6 100644
--- a/tests/reply_relevance_installed_journey.py
+++ b/tests/reply_relevance_installed_journey.py
@@ -5,7 +5,7 @@
 
 from bus_record_fixture import prepare_native_comms
 from agent_comms.lifecycle_capability import Complete
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.native_entries import NativeEvidenceScope
 import asyncio
 import hashlib
@@ -86,7 +86,7 @@ async def run(stage, package):
         environment.pop(key, None)
     os.environ.clear()
     os.environ.update(environment)
-    child = service.registry.new_thread('reply505', source.tags, source.worktree, parent=ForkedParent(source.name), task=source.task, session_file=SavedSessionFile(fork.session_file), model=source.model, thinking_level=source.thinking_level, execution=source.execution)
+    child = service.registry.new_thread('reply505', source.tags, source.worktree, parent=ForkedParent(source.name), task=source.task, session_file=RegistrySessionLocation.from_path(fork.session_file, service.root), model=source.model, thinking_level=source.thinking_level, execution=source.execution)
     service.registry.declare(child)
     assert 'openhcs' in child.tags
     packets = []
diff --git a/tests/shared_bus_restart_native.py b/tests/shared_bus_restart_native.py
index 640bd3104e..301c3627f1 100644
--- a/tests/shared_bus_restart_native.py
+++ b/tests/shared_bus_restart_native.py
@@ -10,7 +10,7 @@
 from bus_record_fixture import prepare_native_comms
 from agent_comms.lifecycle_capability import Complete
 from agent_comms.coordination_session import NativeBoundary
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.native_entries import NativeEvidenceScope
 from conftest import publish_initial
 from agent_comms.pi_vocabulary import OffThinkingLevel
@@ -150,7 +150,7 @@ async def configured_pure_channel(arguments):
         assert hashlib.sha256(original.read_bytes()).hexdigest() == before
         source_hashes[str(original)] = before
         name = f'purechannel-owner-{index}'
-        child = service.registry.new_thread(name, source.tags, source.worktree, parent=ForkedParent(source.name), task=source.task, session_file=SavedSessionFile(fork.session_file), model=source.model, thinking_level=source.thinking_level, execution=source.execution)
+        child = service.registry.new_thread(name, source.tags, source.worktree, parent=ForkedParent(source.name), task=source.task, session_file=RegistrySessionLocation.from_path(fork.session_file, service.root), model=source.model, thinking_level=source.thinking_level, execution=source.execution)
         service.registry.declare(child)
         names.append(name)
         settings.append({'source_owner':source.name, 'private_owner':name,
@@ -832,7 +832,7 @@ async def serve(reader, writer):
                 with closing(sqlite3.connect(str(saved)+'.input-proof')) as copied:
                     original.backup(copied)
             Path(str(saved)+'.input-proof').chmod(0o600)
-            service.registry.register(replace(owner, session_file=SavedSessionFile(str(saved))))
+            service.registry.register(replace(owner, session_file=RegistrySessionLocation.from_path(str(saved), service.root)))
     for index in range(arguments.history):
         publish_initial(service, 'fixture-sender','history-recipient',
             f'Retained unrelated canonical source {index}: ' + 'history '*250)
diff --git a/tests/test_agent_backend.py b/tests/test_agent_backend.py
index bdc0671b49..bf2d5b7226 100644
--- a/tests/test_agent_backend.py
+++ b/tests/test_agent_backend.py
@@ -5,6 +5,7 @@
 import json
 import runpy
 import sys
+from dataclasses import replace
 from pathlib import Path
 from types import MappingProxyType, SimpleNamespace
 
@@ -19,7 +20,7 @@
 from agent_comms.threads import Thread
 from agent_comms.unstarted_runtime import UnstartedRuntime
 from agent_comms.thread_model import DeclaredModel
-from agent_comms.session_file import NoSessionFile, SavedSessionFile
+from agent_comms.session_file import NoSessionFile, RegistrySessionLocation
 
 # -- registry ---------------------------------------------------------------
 
@@ -52,12 +53,14 @@ def test_pi_native_serves_threads_with_every_thread_capability():
 
 
 @pytest.mark.parametrize("session_file, location", [
-    (SavedSessionFile("/home/ts/wt/test/session.jsonl"), PiSessionFile("/home/ts/wt/test/session.jsonl")),
+    (RegistrySessionLocation.from_path("/home/ts/wt/test/session.jsonl", Path("/home/ts/wt")),
+     PiSessionFile("/home/ts/wt/test/session.jsonl")),
     (NoSessionFile(), UnstartedRuntime(PiNativeBackend)),
 ])
 def test_runtime_location_round_trip_uses_declared_backend(session_file, location, *, thread_registration):
     thread = thread_registration.new_thread('location', frozenset(), '/home/ts/wt/test', session_file=session_file)
     restored = FieldCodec.decode(Thread, FieldCodec.encode(thread))
+    restored = replace(restored, session_file=restored.session_file.bound_to(Path("/home/ts/wt")))
     assert restored.runtime == location
     assert restored.runtime is restored.runtime
     assert restored.backend is PiNativeBackend
diff --git a/tests/test_atomic_turn_source_owner.py b/tests/test_atomic_turn_source_owner.py
index 46e9c0b483..6578c7886a 100644
--- a/tests/test_atomic_turn_source_owner.py
+++ b/tests/test_atomic_turn_source_owner.py
@@ -12,7 +12,7 @@
 from agent_comms.comms import wire
 from conftest import installed_comms
 from agent_comms.coordination_errors import StaleFence
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.goal_actions import SetGoalAction
 from agent_comms.native_input_owner import RegistryOwner
 from agent_comms.owned_turn import OwnedTurn
@@ -51,7 +51,7 @@ async def test_begin_callback_and_native_attachment_share_original_owner(tmp_pat
             saved = str(native_file)
             await turn.progress.before_native_metadata(AgentInfo(session_file=saved))
             assert turn.registry_owner is not initial
-            assert turn.thread.session_file == SavedSessionFile(saved)
+            assert turn.thread.session_file == RegistrySessionLocation.from_path(saved, comms.root)
             assert turn.progress.thread is turn.thread
             assert turn.progress.turn_lease == initial.turn_lease
             assert comms.registry.require("owner").turn_lease == initial.turn_lease
diff --git a/tests/test_carry_session_locations.py b/tests/test_carry_session_locations.py
new file mode 100644
index 0000000000..005f05f726
--- /dev/null
+++ b/tests/test_carry_session_locations.py
@@ -0,0 +1,93 @@
+"""A saved transcript is the bus's own or an external harness file, and a copy reads its own."""
+
+import json
+import shutil
+from pathlib import Path
+
+import pytest
+
+from agent_comms.bus import Bus
+from agent_comms.comms import wire
+from agent_comms.errors import RelationViolationError
+from agent_comms.historical_views import HistoryArchive
+from agent_comms.installer.carry import Carried
+from agent_comms.pi_native_backend import PiNativeBackend
+from agent_comms.registry_store import RegistryStore
+from agent_comms.session_file import BusSessionLocation, ExternalSessionLocation, RegistrySessionLocation
+from conftest import maintenance_held
+from cutover.carry_session_locations import SessionLocations
+
+
+def old_registry(tmp_path):
+    comms = wire(tmp_path / "wire")
+    bus = comms.root / "native-sessions" / "bus.jsonl"
+    external = tmp_path / "harness" / "external.jsonl"
+    for name, path in (("bus", bus), ("harness", external)):
+        comms.registry.declare(comms.registry.new_thread(
+            name, frozenset(), str(tmp_path),
+            session_file=RegistrySessionLocation.from_path(path, comms.root)))
+    raw = json.loads(comms.registry.store.path.read_text())
+    raw["threads"]["bus"]["session_file"] = str(bus)
+    raw["threads"]["harness"]["session_file"] = str(external)
+    comms.registry.store.path.write_text(json.dumps(raw, indent=2))
+    return comms, bus, external
+
+
+def test_carry_splits_bus_and_external_transcripts_once(tmp_path, monkeypatch):
+    # The published route is this machine's; the copy is never it.
+    monkeypatch.setenv("HOME", str(tmp_path / "home"))
+    comms, bus, external = old_registry(tmp_path)
+    step = SessionLocations(comms.root, comms.root, tmp_path / "spill")
+    with maintenance_held(comms.root):
+        counts = {"BusSessionLocation": 1, "ExternalSessionLocation": 1}
+        assert step.carry(apply=False) == Carried(2, counts)
+        assert step.carry(apply=True).detail == counts
+        assert step.carry(apply=False) == Carried(0)
+    stored = json.loads(comms.registry.store.path.read_text())["threads"]
+    assert stored["bus"]["session_file"] == {"path": "native-sessions/bus.jsonl"}
+    assert stored["harness"]["session_file"] == {"path": str(external)}
+
+    copy = tmp_path / "copy"
+    shutil.copytree(comms.root, copy, symlinks=True)
+    copied = RegistryStore.at(copy).read_snapshot_unlocked()
+    assert copied.require("bus").runtime.path == str(copy.resolve() / "native-sessions" / "bus.jsonl")
+    assert copied.require("harness").runtime.path == str(external)
+    with pytest.raises(RelationViolationError, match="cannot write external session"):
+        copied.require("harness").writable_runtime
+    assert copied.require("bus").writable_runtime.path.startswith(str(copy.resolve()))
+
+
+def test_unbound_location_refuses_to_guess_its_root():
+    for location in (BusSessionLocation(Path("native-sessions/a.jsonl")),
+                     ExternalSessionLocation(Path("/harness/a.jsonl"))):
+        with pytest.raises(RelationViolationError, match="must be bound by its registry"):
+            location.location(PiNativeBackend)
+
+
+def test_publishing_into_another_root_keeps_the_named_transcript(tmp_path):
+    first = RegistrySessionLocation.from_path(tmp_path / "a" / "s.jsonl", tmp_path / "a")
+    moved = first.bound_to(tmp_path / "b")
+    assert isinstance(moved, ExternalSessionLocation)
+    assert moved.location(PiNativeBackend).path == str((tmp_path / "a" / "s.jsonl").resolve())
+
+
+def test_carry_converts_the_archived_catalogue_before_reading_its_roots(tmp_path):
+    old = wire(tmp_path / "old")
+    session = old.root / "native-sessions" / "bus.jsonl"
+    old.registry.declare(old.registry.new_thread(
+        "bus", frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(session, old.root)))
+    Bus.install(Bus.path_in(old.root))
+    live = wire(tmp_path / "live")
+    Bus.install(Bus.path_in(live.root))
+    live.views.attach_history(old.root)
+    catalog = HistoryArchive.path_in(live.root)
+    recorded = json.loads(catalog.read_text())
+    recorded[0]["provenance"]["threads"]["bus"]["session_file"] = str(session)
+    catalog.write_text(json.dumps(recorded))
+    step = SessionLocations(live.root, live.root, tmp_path / "spill")
+    with maintenance_held(live.root):
+        assert step.carry(apply=False) == Carried(1, {"BusSessionLocation": 1})
+        assert step.carry(apply=True).changed == 1
+        assert step.carry(apply=False) == Carried(0)
+    (source,) = live.bus.history.sources()
+    assert source.provenance.require("bus").runtime.path == str(session.resolve())
diff --git a/tests/test_channels.py b/tests/test_channels.py
index 64a6761c12..e4e176fadc 100644
--- a/tests/test_channels.py
+++ b/tests/test_channels.py
@@ -1,4 +1,4 @@
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.pi_vocabulary import HighThinkingLevel
 import os
 from dataclasses import replace
@@ -138,7 +138,7 @@ def test_exclusive_tag_deletion_uses_original_storage_and_preserves_survivors(tm
         path.write_text('{"original_saved_history":"' + name + '"}\n')
         saved[name] = path.read_bytes()
         process = ProcessIdentity.capture(os.getpid()) if name == 'active' else NoProcess()
-        comms.registry.declare(comms.registry.new_thread(name, frozenset(tags), str(tmp_path), session_file=SavedSessionFile(str(path)), process_identity=process), status)
+        comms.registry.declare(comms.registry.new_thread(name, frozenset(tags), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), comms.root), process_identity=process), status)
     comms.channels.set_channel_pinned('remove', True)
     comms.channels.set_thread_pinned('keep', 'multi', True)
     original_active = comms.registry.require('active')
@@ -383,7 +383,7 @@ def test_all_implicit_tag_introduction_paths_honor_name_reservations(tmp_path):
     parent_session = tmp_path / "parent.jsonl"
     parent_session.write_text("{}\n")
     comms.registry.declare(
-        comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(parent_session)))
+        comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(parent_session), comms.root))
     )
     with pytest.raises(ValueError):
         comms.threads.create(
diff --git a/tests/test_check_root.py b/tests/test_check_root.py
index cb24a1bee7..ea71ec1024 100644
--- a/tests/test_check_root.py
+++ b/tests/test_check_root.py
@@ -55,7 +55,7 @@ def test_external_projection_preserves_event_clocks_or_refuses_unknown(source, s
         return
     external.finish(turn_id="saved", item_id="item", events=events)
     with CheckRoot(source, (owner.name,), root).open() as copied:
-        session = Path(copied.registry.require(owner.name).session_file.require("Missing check session"))
+        session = Path(copied.registry.require(owner.name).runtime.path)
         rows = [NativeEntry.read(line) for line in session.read_bytes().splitlines()]
         assert [entry.message.text for entry in rows[1:]] == [event.text for event in events[:2]]
         assert [entry.timestamp.at.timestamp() for entry in rows[1:]] == [
diff --git a/tests/test_cli.py b/tests/test_cli.py
index bff651a5cf..82b8d96c8c 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -9,7 +9,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.channel_targets import CommandTarget, PerTargetChannels
 from agent_comms.cli import main
 from agent_comms.threads import ChosenServiceTier, NextTurnConfiguration
@@ -599,7 +599,7 @@ def test_mixed_read_marks_human_views_without_advancing_executor_delivery(self,
         source.write_text(json.dumps({'type': 'message', 'timestamp': '2026-09-20T00:00:01Z', 'message': {
             'role': 'assistant', 'content': [{'type': 'text', 'text': 'Saved reply'}],
         }}) + '\n')
-        comms.registry.declare(comms.registry.new_thread('alpha', frozenset({'team'}), str(tmp_path), session_file=SavedSessionFile(str(source))))
+        comms.registry.declare(comms.registry.new_thread('alpha', frozenset({'team'}), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(source), comms.root)))
         comms.channels.update_tags('beta', add=frozenset({'team'}))
         viewer = comms.messaging.user_identity(str(tmp_path)).name
         comms.messaging.send('alpha', 'beta', 'Executor message').message.message_id
diff --git a/tests/test_declarations.py b/tests/test_declarations.py
index 3fea359e48..321ede91b5 100644
--- a/tests/test_declarations.py
+++ b/tests/test_declarations.py
@@ -12,7 +12,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.agent_events import AgentInfo, ContextUsageObservation
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.comms import wire
@@ -219,7 +219,7 @@ def test_owner_claim_survives_session_metadata_and_rotates_on_restart(
         owner = registry.new_thread(name='a', tags=frozenset(), worktree='/wt', process_identity=ProcessIdentity.capture(os.getpid()))
         registry.register(owner)
         before = registry.snapshot().claim_generation("a")
-        registry.register(replace(owner, session_file=SavedSessionFile(str(tmp_path / "session.jsonl"))))
+        registry.register(replace(owner, session_file=RegistrySessionLocation.from_path(str(tmp_path / "session.jsonl"), tmp_path)))
         assert registry.snapshot().claim_generation("a") == before
         assert Registration(registry.store.path).snapshot().claim_generation("a") == before
 
diff --git a/tests/test_durable_host_records.py b/tests/test_durable_host_records.py
index 9038352d27..07021aea84 100644
--- a/tests/test_durable_host_records.py
+++ b/tests/test_durable_host_records.py
@@ -3,7 +3,7 @@
 import json
 
 import pytest
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.pi_payloads import ReportedText, RequestId
 from agent_comms.field_codec import FieldCodec, MalformedRecord
 from agent_comms.native_entries import NativeEntry, StoredEntry, TranscriptProjection
@@ -121,7 +121,7 @@ async def journey():
         backend = PiDurableBackend()
         location = DurableDatabase(str(tmp_path / 'records.sqlite3'))
         thread = thread_registration.new_thread('records', frozenset(), str(tmp_path), backend=PiDurableBackend,
-                        session_file=SavedSessionFile(location.path))
+                        session_file=RegistrySessionLocation.from_path(location.path, tmp_path))
         try:
             outcome = await backend.compact(thread, 'retain the empty plan')
             assert isinstance(outcome, FailedCompaction)
diff --git a/tests/test_durable_runtime_location.py b/tests/test_durable_runtime_location.py
index 13eebcdcc1..baf06e2bfc 100644
--- a/tests/test_durable_runtime_location.py
+++ b/tests/test_durable_runtime_location.py
@@ -1,11 +1,12 @@
 """P117 location behavior over a real, paused durable store; no provider inputs."""
 import asyncio
 import subprocess
+from dataclasses import replace
 from pathlib import Path
 
 import pytest
 
-from agent_comms.session_file import NoSessionFile, SavedSessionFile
+from agent_comms.session_file import NoSessionFile, RegistrySessionLocation
 from agent_comms.agent_backend import NewThreadBackend
 from agent_comms.field_codec import FieldCodec
 from agent_comms.turn_digests import LastTurns
@@ -19,8 +20,9 @@
 @pytest.mark.parametrize('saved', [False, True])
 def test_durable_thread_roundtrip_and_worker_selection(tmp_path, saved, *, thread_registration):
     location = DurableDatabase(str(tmp_path / 'session.sqlite3'))
-    thread = thread_registration.new_thread('durable', frozenset(), str(tmp_path), backend=PiDurableBackend, session_file=SavedSessionFile(location.read_key) if saved else NoSessionFile())
+    thread = thread_registration.new_thread('durable', frozenset(), str(tmp_path), backend=PiDurableBackend, session_file=RegistrySessionLocation.from_path(location.read_key, tmp_path) if saved else NoSessionFile())
     restored = FieldCodec.decode(Thread, FieldCodec.encode(thread))
+    restored = replace(restored, session_file=restored.session_file.bound_to(tmp_path))
     assert restored.runtime == (location if saved else UnstartedRuntime(PiDurableBackend))
     assert restored.runtime.backend is PiDurableBackend
     assert NewThreadBackend.selected() is PiNativeBackend
@@ -71,7 +73,7 @@ def test_real_durable_history_pages_while_writer_owns_database(tmp_path, thread_
     async def journey():
         backend = PiDurableBackend()
         thread = thread_registration.new_thread('history', frozenset(), str(tmp_path), backend=PiDurableBackend,
-                        session_file=SavedSessionFile(location.path))
+                        session_file=RegistrySessionLocation.from_path(location.path, tmp_path))
         host = await backend.open(thread)
         try:
             interrupted = await host.request(Inspect())
diff --git a/tests/test_external_transcript.py b/tests/test_external_transcript.py
index da0161a25e..b241d00f14 100644
--- a/tests/test_external_transcript.py
+++ b/tests/test_external_transcript.py
@@ -3,7 +3,7 @@
 import pytest
 
 from agent_comms.bus import Bus
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.comms import wire
 from agent_comms.field_codec import FieldCodec
 from agent_comms.transcript_events import AssistantTranscript, NoticeTranscript
@@ -135,7 +135,7 @@ async def test_earlier_page_requests_reach_initial_and_committed_watch_observati
 
 def test_a_native_session_thread_takes_no_cli_items(tmp_path):
     comms = wire(tmp_path / "wire")
-    comms.registry.declare(comms.registry.new_thread('pi', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(tmp_path / 'session.jsonl'))))
+    comms.registry.declare(comms.registry.new_thread('pi', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(tmp_path / 'session.jsonl'), comms.root)))
     with pytest.raises(ValueError):
         comms.transcripts.external("pi")
 
diff --git a/tests/test_historical_views.py b/tests/test_historical_views.py
index da3a0c155d..e5e3848c24 100644
--- a/tests/test_historical_views.py
+++ b/tests/test_historical_views.py
@@ -28,7 +28,7 @@
 )
 from agent_comms.read_ledger import ReadLedger
 from agent_comms.routing import MessageRoute, TurnRouting
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 
 
 @pytest.mark.parametrize('exists', [False, True])
@@ -199,7 +199,7 @@ def test_duplicate_and_newer_incarnations_sessions(tmp_path):
     session.write_text("".join(json.dumps(r) + "\n" for r in records))
     with old.registry.store.editing() as edit:
         edit.document.threads["bob"] = replace(edit.document.threads["bob"], created_at=10.0)
-        edit.document.threads["alice"] = replace(edit.document.threads["alice"], session_file=SavedSessionFile(str(session)))
+        edit.document.threads["alice"] = replace(edit.document.threads["alice"], session_file=RegistrySessionLocation.from_path(str(session), tmp_path))
         edit.commit()
     routing = TurnRouting(reply=MessageRoute("alice", ("#team",)))
     old.transcripts.routes.record(str(session), ("two",), routing)
diff --git a/tests/test_native_owner_launcher_resolution.py b/tests/test_native_owner_launcher_resolution.py
index b7527770e7..1c1a0cdd38 100644
--- a/tests/test_native_owner_launcher_resolution.py
+++ b/tests/test_native_owner_launcher_resolution.py
@@ -7,7 +7,7 @@
 from pathlib import Path
 
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.launch_root import LaunchRoot
 from agent_comms.launch_runtime import LaunchRuntime
 from agent_comms.native_package import VerifiedNativePackage
@@ -41,7 +41,7 @@ async def test_fork_runs_selected_python_before_copying_saved_history(tmp_path,
     parent = create_fresh_private_session(tmp_path / "sessions", worktree=tmp_path)
     original = parent.path.read_bytes()
     copied = await PiNativeBackend.fork(
-        thread_registration.new_thread('parent', tags=frozenset(), worktree=str(tmp_path), session_file=SavedSessionFile(parent.session_file)),
+        thread_registration.new_thread('parent', tags=frozenset(), worktree=str(tmp_path), session_file=RegistrySessionLocation.from_path(parent.session_file, comms.root)),
         str(package.cli), LaunchRuntime(interpreter),
     )
     assert marker.read_text() == str(selected)
diff --git a/tests/test_native_unknown_recovery.py b/tests/test_native_unknown_recovery.py
index b762de51a0..b2498bc3bb 100644
--- a/tests/test_native_unknown_recovery.py
+++ b/tests/test_native_unknown_recovery.py
@@ -24,7 +24,7 @@
 from agent_comms.launch_root import LaunchRoot
 from agent_comms.runtime import socket_path
 from agent_comms.thread_model import DeclaredModel
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from functools import partial
 from native_backend_fixture import native_backend_fixture
 from agent_comms.child_process import ProcessIdentity
@@ -155,7 +155,7 @@ def test_undispatched_owner_loss_refuses_a_live_native_child(released_starting,
     directory = released_starting / "native-sessions" / owner.incarnation.lookup
     session = directory / "undispatched.jsonl"
     comms.registry.register(
-        replace(owner, session_file=SavedSessionFile(str(session))),
+        replace(owner, session_file=RegistrySessionLocation.from_path(str(session), comms.root)),
         status=comms.registry.status(owner.name),
     )
     selected = session if argument == "--session" else directory
diff --git a/tests/test_operations.py b/tests/test_operations.py
index 617380df98..0fd54251c7 100644
--- a/tests/test_operations.py
+++ b/tests/test_operations.py
@@ -8,7 +8,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.native_entries import CompactionEntry, ModelChangeEntry
 from agent_comms.channel_targets import ChannelTarget, CommandTarget, ThreadTarget
 from agent_comms.channels import AnyChannel
@@ -176,7 +176,7 @@ def test_thread_transcript_normalizes_persisted_pi_events(self, wired, tmp_path)
         ]
         session_file.write_text("\n".join(json.dumps(record) for record in records) + "\n")
         wired.registry.declare(
-            wired.registry.new_thread(name='transcript-thread', tags=frozenset(), worktree=str(tmp_path), session_file=SavedSessionFile(str(session_file)))
+            wired.registry.new_thread(name='transcript-thread', tags=frozenset(), worktree=str(tmp_path), session_file=RegistrySessionLocation.from_path(str(session_file), wired.root))
         )
 
         events = wired.transcripts.thread_transcript_page("transcript-thread").events
@@ -202,7 +202,7 @@ def test_thread_transcript_includes_saved_compaction_summary(self, wired, tmp_pa
             + "\n"
         )
         wired.registry.declare(
-            wired.registry.new_thread(name='compacted-thread', tags=frozenset(), worktree=str(tmp_path), session_file=SavedSessionFile(str(session_file)))
+            wired.registry.new_thread(name='compacted-thread', tags=frozenset(), worktree=str(tmp_path), session_file=RegistrySessionLocation.from_path(str(session_file), wired.root))
         )
 
         events = wired.transcripts.thread_transcript_page("compacted-thread").events
@@ -303,7 +303,7 @@ def test_attach_session_preserves_declaration_and_updates_runtime(self, wired, t
         assert attached.parent == ForkedParent("PR111")
         assert attached.task == GivenThreadText("fix auth")
         assert attached.process_identity == ProcessIdentity.capture(os.getpid())
-        assert attached.session_file == SavedSessionFile(str(session_file.resolve()))
+        assert attached.session_file == RegistrySessionLocation.from_path(str(session_file.resolve()), wired.root)
         assert wired.registry.status("fixer").declared_name == "running"
 
 
@@ -512,7 +512,7 @@ def test_recent_transcript_is_bounded_and_ordered(self, wired, tmp_path):
                     + "\n"
                 )
         wired.registry.declare(
-            wired.registry.new_thread(name='large', tags=frozenset(), worktree=str(tmp_path), session_file=SavedSessionFile(str(path)))
+            wired.registry.new_thread(name='large', tags=frozenset(), worktree=str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), wired.root))
         )
         page = wired.transcripts.thread_transcript_page("large")
         assert page.has_older
@@ -581,10 +581,10 @@ def test_explicit_tags_replace_previous(self, wired):
 
     def test_missing_session_file_inherits_previous(self, wired, tmp_path):
         wired.registry.declare(
-            wired.registry.new_thread(name='PR111', tags=frozenset({'base'}), worktree='/tmp/wt1', session_file=SavedSessionFile(str(tmp_path / 's.json')))
+            wired.registry.new_thread(name='PR111', tags=frozenset({'base'}), worktree='/tmp/wt1', session_file=RegistrySessionLocation.from_path(str(tmp_path / 's.json'), wired.root))
         )
         wired.registry.declare(wired.registry.new_thread(name='PR111', tags=frozenset({'base'}), worktree='/tmp/wt1'))
-        assert wired.registry.require("PR111").session_file == SavedSessionFile(str(tmp_path / "s.json"))
+        assert wired.registry.require("PR111").session_file == RegistrySessionLocation.from_path(str(tmp_path / "s.json"), wired.root)
 
     def test_fresh_declaration_is_unaffected(self, wired):
         wired.registry.declare(wired.registry.new_thread(name='fresh', tags=frozenset(), worktree='/wt'))
diff --git a/tests/test_registration_inheritance.py b/tests/test_registration_inheritance.py
index 87b2a9eaf7..693e903bf2 100644
--- a/tests/test_registration_inheritance.py
+++ b/tests/test_registration_inheritance.py
@@ -3,7 +3,7 @@
 from typing import Annotated
 from dataclasses import dataclass, replace
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms import field_codec
 from agent_comms.comms import wire
 from agent_comms.pi_vocabulary import HighThinkingLevel
@@ -45,7 +45,7 @@ def test_registry_redeclaration_inherits_metadata_but_respects_explicit_values(t
         "owner",
         frozenset({"team"}),
         '/old',
-        session_file=SavedSessionFile(str(tmp_path / "session.jsonl")),
+        session_file=RegistrySessionLocation.from_path(str(tmp_path / "session.jsonl"), comms.root),
         model=DeclaredModel('provider/model'),
         thinking_level=HighThinkingLevel,
         auto_title_pending=True,
@@ -113,7 +113,7 @@ def test_project_change_retains_executing_source_but_refuses_another_input(tmp_p
         observed = comms.registry.attach_native_session(original, saved)
         current = comms.registry.require('owner')
         assert current.worktree == str(new) and current.earlier_projects == (old,)
-        assert current.session_file == observed.thread.session_file == SavedSessionFile(saved)
+        assert current.session_file == observed.thread.session_file == RegistrySessionLocation.from_path(saved, comms.root)
         assert observed.thread.worktree == str(old)
         assert observed.turn_lease == original.turn_lease == current.turn_lease
         # Later native observations use the same source and preserve old -> new.
diff --git a/tests/test_registration_process_identity.py b/tests/test_registration_process_identity.py
index d69dc96098..8a85e8198d 100644
--- a/tests/test_registration_process_identity.py
+++ b/tests/test_registration_process_identity.py
@@ -16,7 +16,7 @@
 )
 from agent_comms.thread_status import NoBridgeThreadStatus
 from agent_comms.thread_text import GivenThreadText
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 
 
 @pytest.mark.parametrize('phases', [(), (PausedPhase,), (PausedPhase, InstallingPhase)])
@@ -111,7 +111,7 @@ def test_claim_attach_and_reload_preserve_the_captured_process(tmp_path):
     saved = restored.threads[owner.name]
     assert saved.process_identity == NoProcess() and not saved.executing
     assert saved.incarnation == owner.incarnation
-    assert saved.session_file == SavedSessionFile(str(session))
+    assert saved.session_file == RegistrySessionLocation.from_path(str(session), comms.root)
 
 
 def test_same_pid_different_birth_advances_both_authority_domains(tmp_path):
diff --git a/tests/test_registration_transitions.py b/tests/test_registration_transitions.py
index 215cccac5b..582cc610f0 100644
--- a/tests/test_registration_transitions.py
+++ b/tests/test_registration_transitions.py
@@ -5,7 +5,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.errors import RelationViolationError
 from agent_comms.registration import Registration
@@ -30,7 +30,7 @@ def test_registration_transition_preserves_or_revokes_exact_lease(tmp_path, tran
     if transition == 'metadata':
         updated = replace(leased, title=GivenThreadText('new title'), tags=frozenset({'changed'}))
     elif transition == 'session':
-        updated = replace(leased, session_file=SavedSessionFile(str(tmp_path / 'new.jsonl')))
+        updated = replace(leased, session_file=RegistrySessionLocation.from_path(str(tmp_path / 'new.jsonl'), tmp_path))
     elif transition == 'restart':
         kwargs['new_owner'] = True
         advances_owner = True
diff --git a/tests/test_restart.py b/tests/test_restart.py
index ec7d6dc678..8989fcfd7a 100644
--- a/tests/test_restart.py
+++ b/tests/test_restart.py
@@ -11,7 +11,7 @@
 from agent_comms.native_arguments import GivenBinary
 from agent_comms.restart_scope import OwnersNamed
 from agent_comms.thread_model import DeclaredModel
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from native_backend_fixture import native_backend_fixture
 
 
@@ -23,7 +23,7 @@ async def test_real_idle_owner_is_replaced_without_losing_session(
         comms = wire(native.root)
         with LaunchRoot.from_path(comms.root).selected():
             comms.registry.declare(comms.registry.new_thread(
-                'worker', frozenset(), str(native.project), session_file=SavedSessionFile(str(native.session))))
+                'worker', frozenset(), str(native.project), session_file=RegistrySessionLocation.from_path(str(native.session), comms.root)))
             comms.agents.observe_runtime_info('worker', ContextUsageObservation(
                 context_used=PiCount.from_wire(23)))
             comms.agents.observe_runtime_info('worker', AgentInfo(
@@ -48,7 +48,7 @@ async def ready(pid):
                 assert results[0].pid != process.pid
                 assert not owner.process_alive
                 await ready(results[0].pid)
-                assert comms.registry.require('worker').session_file == SavedSessionFile(str(native.session))
+                assert comms.registry.require('worker').session_file == RegistrySessionLocation.from_path(str(native.session), comms.root)
                 assert not comms.registry.require('worker').executing
                 assert comms.agents.runtime_info.path.read_bytes() == metadata
                 assert comms.ledger.path.read_bytes() == collaboration
diff --git a/tests/test_retained_context_operations.py b/tests/test_retained_context_operations.py
index 8801d27404..19a5fae5e3 100644
--- a/tests/test_retained_context_operations.py
+++ b/tests/test_retained_context_operations.py
@@ -11,7 +11,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.cli import main
 from agent_comms.cli_commands import ConstraintPinnedResult
 from agent_comms.field_codec import FieldCodec
@@ -34,7 +34,7 @@ def test_original_retained_source_narrow_export(comms, tmp_path, capsys):
     saved = tmp_path / "original.jsonl"
     saved_source(saved)
     goal = Goal("Original task goal, not an exported human constraint", "original-goal", published_seq=0, revision=7)
-    owner = comms.registry.declare(comms.registry.new_thread('reader', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(saved)), goal=goal))
+    owner = comms.registry.declare(comms.registry.new_thread('reader', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(saved), comms.root), goal=goal))
     wording = "Original human constraint λ; preserve UNKNOWN and its source."
     original = comms.messaging.send_user_message(CommandTarget.named(owner.name), wording, worktree=owner.worktree)
     comms.messaging.pin_user_constraint(owner.name, original.reference, worktree=owner.worktree)
diff --git a/tests/test_roster_restoration.py b/tests/test_roster_restoration.py
index 32a8e40944..984a4fabea 100644
--- a/tests/test_roster_restoration.py
+++ b/tests/test_roster_restoration.py
@@ -12,7 +12,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.coordination_tables.executions import NoPointedExecution
 from agent_comms.thread_identity import RecordedThreadIncarnation
 from agent_comms.child_process import NoProcess, ProcessIdentity
@@ -28,7 +28,7 @@ def test_restore_keeps_live_owner_and_bus_while_importing_saved_stopped_identity
     old = installed_comms(tmp_path / "old")
     current = installed_comms(tmp_path / "current")
     live = old.registry.new_thread('live', frozenset({'comms'}), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()))
-    missing = old.registry.new_thread('missing', frozenset({'nra', 'openhcs'}), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()), session_file=SavedSessionFile(str(tmp_path / 'saved.jsonl')), title=GivenThreadText('Preserved title'), task=GivenThreadText('Original task'))
+    missing = old.registry.new_thread('missing', frozenset({'nra', 'openhcs'}), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()), session_file=RegistrySessionLocation.from_path(str(tmp_path / 'saved.jsonl'), old.root), title=GivenThreadText('Preserved title'), task=GivenThreadText('Original task'))
     old.registry.declare(live)
     old.registry.declare(missing)
     missing = old.agents.begin_turn("missing", "old-turn").thread
diff --git a/tests/test_runtime.py b/tests/test_runtime.py
index 9c1e9ac3d0..1638bc4934 100644
--- a/tests/test_runtime.py
+++ b/tests/test_runtime.py
@@ -23,7 +23,7 @@
 )
 from agent_comms.private_path import PrivateSocketRole
 from agent_comms.thread_identity import RecordedThreadIncarnation
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.native_arguments import GivenBinary
 from delivery_owner_fixture import canonical_agent
 
@@ -235,7 +235,7 @@ async def receive(reader, writer):
         )
         # Session metadata may be filled in after initial owner attachment.
         comms.registry.register(
-            replace(comms.registry.require("worker"), session_file=SavedSessionFile(str(tmp_path / "session.jsonl")))
+            replace(comms.registry.require("worker"), session_file=RegistrySessionLocation.from_path(str(tmp_path / "session.jsonl"), comms.root))
         )
         assert await proxy.request("cancel") == {"owner": "old"}
         comms.threads.rename_managed_thread("worker", "renamed", owner=old)
diff --git a/tests/test_runtime_subscription_handshake.py b/tests/test_runtime_subscription_handshake.py
index 380018b0a1..eddaa93065 100644
--- a/tests/test_runtime_subscription_handshake.py
+++ b/tests/test_runtime_subscription_handshake.py
@@ -8,7 +8,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.acp import CommsAgent
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.comms import wire
@@ -30,7 +30,7 @@ async def test_subscription_busy_read_retains_original_traceback(tmp_path, caplo
         "timestamp": "2026-10-10T00:00:00.000Z", "cwd": str(tmp_path),
     }) + "\n")
     session_file.chmod(0o600)
-    thread = comms.registry.new_thread('owner', frozenset(), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()), session_file=SavedSessionFile(str(session_file)))
+    thread = comms.registry.new_thread('owner', frozenset(), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()), session_file=RegistrySessionLocation.from_path(str(session_file), comms.root))
     comms.registry.declare(thread)
     await owner.sessions.bind_owned(thread, thread.name)
     database = comms.root / "coordination.sqlite3"
diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py
index 9d67036eb5..e36f71cd2a 100644
--- a/tests/test_standing_rules.py
+++ b/tests/test_standing_rules.py
@@ -3,7 +3,7 @@
 from dataclasses import replace
 
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.todos import (DirectiveText)
 
 
@@ -24,10 +24,10 @@ def working(comms, name, created):
     path.write_text("".join(json.dumps(row) + "\n" for row in rows))
     path.chmod(0o600)
     if name in comms.registry.all_threads():
-        comms.registry.register(replace(comms.registry.require(name), session_file=SavedSessionFile(str(path))))
+        comms.registry.register(replace(comms.registry.require(name), session_file=RegistrySessionLocation.from_path(str(path), comms.root)))
     else:
         comms.registry.declare(comms.registry.new_thread(
-            name, frozenset(), str(comms.root), session_file=SavedSessionFile(str(path)), created_at=created))
+            name, frozenset(), str(comms.root), session_file=RegistrySessionLocation.from_path(str(path), tmp_path), created_at=created))
 
 
 def test_plain_messages_need_no_order_labels():
diff --git a/tests/test_start.py b/tests/test_start.py
index 3b929bf10a..55b229cbaf 100644
--- a/tests/test_start.py
+++ b/tests/test_start.py
@@ -5,7 +5,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.tools import invoke_tool
 from agent_comms.comms import wire
 from agent_comms.tool_caller import NoThreadCaller
@@ -21,7 +21,7 @@ async def test_real_stopped_owner_starts_through_shared_agent_tool(tmp_path, mon
     source.touch()
     comms = wire(tmp_path / "wire")
     Bus.install(Bus.path_in(comms.root))
-    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(source))))
+    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(source), comms.root)))
     comms.registry.unregister("worker")
     try:
         result = await asyncio.to_thread(invoke_tool, comms, "comms_start", {"name": "worker"}, caller=NoThreadCaller())
diff --git a/tests/test_task_decisions.py b/tests/test_task_decisions.py
index b2ca1528a4..9a175b4a65 100644
--- a/tests/test_task_decisions.py
+++ b/tests/test_task_decisions.py
@@ -5,7 +5,7 @@
 
 from agent_comms.bus import Bus
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.bus_publication import AgentOrigin
 from agent_comms.messages import NoticeRole
 import os
@@ -391,7 +391,7 @@ def test_cross_audience_corrections_capture_only_eligible_owned_lineage(comms, m
         owner = admit(comms, name)
         saved = comms.root / f"{name}-source.jsonl"
         saved_source(saved)
-        comms.registry.register(replace(owner, session_file=SavedSessionFile(str(saved))))
+        comms.registry.register(replace(owner, session_file=RegistrySessionLocation.from_path(str(saved), comms.root)))
     owner = comms.registry.require("alpha")
     monkeypatch.setenv("AGENT_COMMS_THREAD", "alpha")
     request = {"chosen": "PRIVATE_ORIGINAL_CHOICE", "rejected": ["PRIVATE_ALTERNATIVE"], "to": "beta"}
diff --git a/tests/test_thread_identity.py b/tests/test_thread_identity.py
index ec25b4610a..541cc1c74f 100644
--- a/tests/test_thread_identity.py
+++ b/tests/test_thread_identity.py
@@ -8,7 +8,7 @@
 import pytest
 
 from agent_comms.channel_targets import ThreadTarget
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.child_process import NoProcess, ProcessIdentity
 from agent_comms.comms import wire
 from conftest import installed_comms
@@ -316,12 +316,12 @@ def test_external_unregister_revokes_owner_while_bridge_remains_connected(tmp_pa
 
 def test_same_launch_ignores_metadata_and_follows_process_session_and_model(tmp_path, *, thread_registration):
     process = ProcessIdentity.capture(os.getpid())
-    thread = thread_registration.new_thread('owner', frozenset(), str(tmp_path), process_identity=process, session_file=SavedSessionFile(str(tmp_path / 'a.jsonl')), model=DeclaredModel('model-a'), created_at=1.0)
+    thread = thread_registration.new_thread('owner', frozenset(), str(tmp_path), process_identity=process, session_file=RegistrySessionLocation.from_path(str(tmp_path / 'a.jsonl'), tmp_path), model=DeclaredModel('model-a'), created_at=1.0)
 
     assert thread.same_launch(replace(thread, title=GivenThreadText("Renamed"), tags=frozenset({"team"}), turn_generation=3))
     for changed in (
         replace(thread, process_identity=NoProcess()),
-        replace(thread, session_file=SavedSessionFile(str(tmp_path / "b.jsonl"))),
+        replace(thread, session_file=RegistrySessionLocation.from_path(str(tmp_path / "b.jsonl"), tmp_path)),
         replace(thread, model=DeclaredModel("model-b")),
         replace(thread, worktree=str(tmp_path / "elsewhere")),
         replace(thread, created_at=2.0),
diff --git a/tests/test_thread_status.py b/tests/test_thread_status.py
index 5e7e1e188a..f13edd13cd 100644
--- a/tests/test_thread_status.py
+++ b/tests/test_thread_status.py
@@ -8,7 +8,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.activity import Activity, ActivityState
 from agent_comms.comms import wire
 from conftest import installed_comms
@@ -62,7 +62,7 @@ def test_saved_presence_roundtrip_and_wire_views_preserve_data(tmp_path, status)
     comms = installed_comms(tmp_path)
     child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
     try:
-        thread = comms.registry.new_thread('owner', frozenset({'team'}), str(tmp_path), process_identity=ProcessIdentity.capture(child.pid), title=GivenThreadText('Retained title'), session_file=SavedSessionFile('/retained/session.jsonl'), created_at=12.5)
+        thread = comms.registry.new_thread('owner', frozenset({'team'}), str(tmp_path), process_identity=ProcessIdentity.capture(child.pid), title=GivenThreadText('Retained title'), session_file=RegistrySessionLocation.from_path('/retained/session.jsonl', comms.root), created_at=12.5)
         comms.registry.register(thread, status)
         before = comms.registry.snapshot()
         saved = comms.registry.store.path.read_bytes()
diff --git a/tests/test_thread_unread.py b/tests/test_thread_unread.py
index 54442e3375..b2f4f74170 100644
--- a/tests/test_thread_unread.py
+++ b/tests/test_thread_unread.py
@@ -10,7 +10,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.comms import wire
 from conftest import installed_comms
 from agent_comms.private_path import FileRevision
@@ -49,7 +49,7 @@ def setup_thread(tmp_path):
     session(source)
     comms = installed_comms(tmp_path / 'wire')
     comms.registry.declare(comms.registry.new_thread('worker', frozenset({'ci'}),
-        str(tmp_path), session_file=SavedSessionFile(str(source))))
+        str(tmp_path), session_file=RegistrySessionLocation.from_path(str(source), comms.root)))
     return comms, source
 
 
diff --git a/tests/test_transcript_input_display.py b/tests/test_transcript_input_display.py
index aa6be40033..68b83b66a8 100644
--- a/tests/test_transcript_input_display.py
+++ b/tests/test_transcript_input_display.py
@@ -6,7 +6,7 @@
 import pytest
 
 from conftest import transcript_records
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.acp import CommsAgent
 from agent_comms.acp_extension import TranscriptSnapshotUpdate, decode_updates
 from agent_comms.comms import wire
@@ -97,7 +97,7 @@ async def test_acp_saved_transcript_replay_hides_only_owned_internal_input(tmp_p
             ]
         )
     )
-    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(session))))
+    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(session), comms.root)))
     comms.transcripts.routes.record_input_display("a" * 32, None)
     comms.transcripts.routes.record_input_display("b" * 32, "test2")
     agent = CommsAgent(wire(comms.root))
diff --git a/tests/test_transcript_pages.py b/tests/test_transcript_pages.py
index cf936064dc..bf2a23b02a 100644
--- a/tests/test_transcript_pages.py
+++ b/tests/test_transcript_pages.py
@@ -1,5 +1,5 @@
 from conftest import transcript_records
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.coordination_errors import TranscriptChanged
 import json
 from dataclasses import replace
@@ -31,7 +31,7 @@ def test_pages_reach_beginning_and_return_to_tail_without_duplicates(tmp_path):
     path = tmp_path / "session.jsonl"
     transcript(path, 105)
     comms = installed_comms(tmp_path / "wire")
-    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(path))))
+    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), comms.root)))
     page = comms.transcripts.thread_transcript_page("worker", max_messages=10)
     through = page.after
     seen = [event.text for event in page.events]
@@ -65,12 +65,12 @@ def test_oversized_message_and_file_identity(tmp_path):
         json.dumps(transcript_records.message({"role": "assistant", "content": [{"type": "text", "text": text}]})) + "\n"
     )
     comms = installed_comms(tmp_path / "wire")
-    thread = comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(path)))
+    thread = comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), comms.root))
     comms.registry.declare(thread)
     page = comms.transcripts.thread_transcript_page("worker", max_bytes=100)
     assert page.events[0].text == text
     assert not page.has_older
-    comms.registry.declare(replace(thread, session_file=SavedSessionFile(str(tmp_path / "other.jsonl"))))
+    comms.registry.declare(replace(thread, session_file=RegistrySessionLocation.from_path(str(tmp_path / "other.jsonl"), comms.root)))
     with pytest.raises(TranscriptChanged):
         comms.transcripts.thread_transcript_page("worker", before=page.before)
     with pytest.raises(ValueError):
@@ -91,7 +91,7 @@ def test_new_fork_projects_parent_history_and_instruction_until_own_session_exis
     transcript(parent_path, 3)
     comms = installed_comms(tmp_path / "wire")
     comms.registry.declare(
-        comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(parent_path)))
+        comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(parent_path), comms.root))
     )
     comms.registry.declare(
         comms.registry.new_thread('child', frozenset(), str(tmp_path), parent=ForkedParent('parent'), task=GivenThreadText('Inspect the renderer'))
@@ -125,7 +125,7 @@ def test_inherited_scroll_window_survives_child_session_persistence(tmp_path):
     transcript(parent_path, 35)
     comms = installed_comms(tmp_path / "wire")
     comms.registry.declare(
-        comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(parent_path)))
+        comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(parent_path), comms.root))
     )
     comms.registry.declare(
         comms.registry.new_thread('child', frozenset(), str(tmp_path), parent=ForkedParent('parent'), task=GivenThreadText('Child task'))
@@ -173,7 +173,7 @@ def test_never_started_native_child_keeps_no_native_cursor_and_no_read_mark(tmp_
     transcript(path, 3)
     comms = installed_comms(tmp_path / "wire")
     comms.registry.declare(comms.registry.new_thread(
-        "parent", frozenset(), str(tmp_path), session_file=SavedSessionFile(str(path)),
+        "parent", frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), comms.root),
         execution=NativeThreadExecution,
     ))
     RegisterCliCommand(name="child", worktree=str(tmp_path), parent=ForkedParent("parent"),
diff --git a/tests/test_transcript_partial_tail.py b/tests/test_transcript_partial_tail.py
index 7c7a073c12..2ea1b1f39b 100644
--- a/tests/test_transcript_partial_tail.py
+++ b/tests/test_transcript_partial_tail.py
@@ -3,7 +3,7 @@
 from agent_comms.private_runtime_schema import PrivateCoordinatorTable
 import json
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.comms import wire
 from agent_comms.coordinator import Coordination
 
@@ -22,7 +22,7 @@ def test_newer_read_moves_past_a_line_still_being_appended(tmp_path):
     session = tmp_path / "saved.jsonl"
     session.write_text(line({"type": "session", "version": 3, "id": "saved-session",
                              "timestamp": "2026-10-20T00:00:00.000Z"}) + line(answer("first", "First answer")))
-    comms.registry.declare(comms.registry.new_thread("saved", frozenset(), str(tmp_path), session_file=SavedSessionFile(str(session))))
+    comms.registry.declare(comms.registry.new_thread("saved", frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(session), comms.root)))
     with Coordination(str(comms.root / "coordination.sqlite3")) as store:
         PrivateCoordinatorTable.install(store)
     complete = session.stat().st_size
diff --git a/tests/test_transcript_read_identity.py b/tests/test_transcript_read_identity.py
index c4ff0b8e7d..5abc38b32b 100644
--- a/tests/test_transcript_read_identity.py
+++ b/tests/test_transcript_read_identity.py
@@ -6,7 +6,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from conftest import installed_comms
 from agent_comms.coordination_errors import StaleRevision
 from agent_comms.thread_text import ForkedParent, GivenThreadText
@@ -22,7 +22,7 @@ def test_saved_input_annotation_revokes_page_without_journal_change(tmp_path, jo
             "role": "user", "inputId": native_id, "content": "private wrapper",
         },
     }) + "\n")
-    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(path))))
+    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), comms.root)))
     comms.transcripts.routes.record_input_display("b" * 32, None)
     with closing(sqlite3.connect(comms.transcripts.routes.path)) as connection:
         assert connection.execute("PRAGMA journal_mode=" + journal_mode).fetchone()[0] == journal_mode.lower()
@@ -43,7 +43,7 @@ def test_native_append_preserves_captured_cut_and_fork_source_change_revokes_it(
     parent.write_text(json.dumps({"type": "message", "id": "parent-1", "timestamp": "2026-10-19T00:00:00Z", "message": {
         "role": "assistant", "content": [{"type": "text", "text": "parent answer"}],
     }}) + "\n")
-    comms.registry.declare(comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(parent))))
+    comms.registry.declare(comms.registry.new_thread('parent', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(parent), comms.root)))
     comms.registry.declare(comms.registry.new_thread('child', frozenset(), str(tmp_path), parent=ForkedParent('parent'), task=GivenThreadText('continue')))
     captured = comms.transcripts.capture_page_read("child")
     assert any(event.text == "parent answer" for event in captured.read().events)
@@ -78,7 +78,7 @@ def test_native_rewrite_inside_captured_cut_revokes_the_read(tmp_path):
     replacement = original.replace("first answer", "other answer")
     assert len(replacement) == len(original)
     path.write_text(original)
-    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(path))))
+    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), comms.root)))
     captured = comms.transcripts.capture_page_read("worker")
 
     path.write_text(replacement)
@@ -97,7 +97,7 @@ def test_thread_activity_and_annotations_do_not_replay_saved_content(tmp_path):
     path.write_text(json.dumps({"type": "message", "id": "entry", "timestamp": "2026-10-19T00:00:00Z", "message": {
         "role": "assistant", "content": [{"type": "text", "text": "Original retained answer"}],
     }}) + "\n")
-    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(path))))
+    comms.registry.declare(comms.registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(path), comms.root)))
     captured = comms.transcripts.capture_page_read("worker")
     original = captured.read()
     comms.registry.heartbeat("worker")
diff --git a/tests/test_transcript_read_lifetime.py b/tests/test_transcript_read_lifetime.py
index 00b9ecfb08..7dcd4b90a0 100644
--- a/tests/test_transcript_read_lifetime.py
+++ b/tests/test_transcript_read_lifetime.py
@@ -10,7 +10,7 @@
 
 import pytest
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.view_unread import transcript_read_state
 from agent_comms.comms import wire
 from native_unread_fixture import native_reader
@@ -25,7 +25,7 @@ def test_collected_reader_cancels_and_joins_its_worker_then_closes_sqlite(tmp_pa
     }}) + "\n")
     state = native_reader(tmp_path / 'read_ledger.json')
     registry = wire(tmp_path).registry
-    registry.declare(registry.new_thread('worker', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(source))))
+    registry.declare(registry.new_thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(source), tmp_path)))
     assert state.counts('viewer', registry=registry.snapshot(), document=state.reads.read(), visibility=DEFAULT_ROSTER).counts == {"worker": IndexingUnread()}
     for future in state.watch(frozenset(), lambda _: None):
         future.result(timeout=10)
diff --git a/tests/test_turn_digests.py b/tests/test_turn_digests.py
index 1132111152..0c8dee963a 100644
--- a/tests/test_turn_digests.py
+++ b/tests/test_turn_digests.py
@@ -19,7 +19,7 @@
 from agent_comms.field_codec import FieldCodec
 from agent_comms.runtime_updates import FinishedItem
 from agent_comms.thread_execution import ExternalThreadExecution
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.transcript_events import AssistantTranscript, TurnDigest
 from agent_comms.turn_digests import LastTurns, TurnBound, TurnsOlderThan, TurnsBefore, TurnsSince
 from agent_comms.bus import Bus
@@ -107,7 +107,7 @@ def test_native_turns_keep_original_calls_results_and_input_identity(tmp_path):
     ]
     source.write_text(''.join(json.dumps(row) + '\n' for row in rows))
     source.chmod(0o600)
-    comms.registry.declare(comms.registry.new_thread('native', frozenset(), str(tmp_path), session_file=SavedSessionFile(str(source))))
+    comms.registry.declare(comms.registry.new_thread('native', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(source), comms.root)))
     answer = endpoint(comms, 'native', LastTurns(1))
     turn, = answer['turns']
     assert turn['turn_id'] == 'u2'
diff --git a/tests/test_unit_state.py b/tests/test_unit_state.py
index 12171a2d4a..781526a8c4 100644
--- a/tests/test_unit_state.py
+++ b/tests/test_unit_state.py
@@ -15,7 +15,7 @@
 from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.thread_status import StoppedThreadStatus
 from agent_comms.thread_text import ForkedParent
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.todos import DeliverTodo, OperationDelivery, ProgressRead
 from agent_comms.ui_model.unit_state import (
     AwaitingCommander,
@@ -96,7 +96,7 @@ def recorded_turn(comms, calls):
     path.write_text("".join(json.dumps(row) + "\n" for row in rows))
     path.chmod(0o600)
     thread = comms.registry.require("worker")
-    comms.registry.register(replace(thread, session_file=SavedSessionFile(str(path))))
+    comms.registry.register(replace(thread, session_file=RegistrySessionLocation.from_path(str(path), comms.root)))
 
 
 def unread_checkout(comms, tmp_path):
diff --git a/tests/test_user_view_mark_read.py b/tests/test_user_view_mark_read.py
index bc6658f2e1..9e666413cc 100644
--- a/tests/test_user_view_mark_read.py
+++ b/tests/test_user_view_mark_read.py
@@ -1,6 +1,6 @@
 """Explicit human Mark read never acknowledges a persistent agent's inbox."""
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.channel_targets import CommandTarget
 from agent_comms.read_basis import NamedViewer
 import json
@@ -27,7 +27,7 @@ def test_mark_read_clears_thread_dm_and_channel_views_without_agent_delivery(tmp
     source = tmp_path / "sender.jsonl"
     source.touch()
     answer(source, "Saved reply")
-    comms.registry.declare(comms.registry.new_thread('sender', frozenset({'ci'}), str(tmp_path), session_file=SavedSessionFile(str(source))))
+    comms.registry.declare(comms.registry.new_thread('sender', frozenset({'ci'}), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(source), comms.root)))
     comms.registry.declare(comms.registry.new_thread('receiver', frozenset({'ci'}), str(tmp_path)))
     viewer = comms.messaging.user_identity(str(tmp_path)).name
     comms.messaging.send("sender", "receiver", "Agent inbox remains unread").message.message_id
diff --git a/tools/cutover/carry_session_locations.py b/tools/cutover/carry_session_locations.py
new file mode 100644
index 0000000000..fb68c2a686
--- /dev/null
+++ b/tools/cutover/carry_session_locations.py
@@ -0,0 +1,112 @@
+"""Store each saved transcript as a bus location or an external one, once.
+
+The registry stored every saved transcript as one absolute path string, so a
+copied root read and wrote the transcripts of the bus it was copied from. A path
+under the bus that published it becomes a location relative to the root that
+reads it; any other path is an external harness transcript, which only the
+published root may write. Delete this step after every retained root has run it.
+"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from agent_comms.field_codec import FieldCodec, compile_decoder
+from agent_comms.historical_views import HistoryArchive, HistorySource
+from agent_comms.installer.carry import Carried
+from agent_comms.locked_store import RegistryFile
+from agent_comms.registry_document import RegistryDocument
+from agent_comms.registry_store import RegistryStore
+from agent_comms.session_file import RegistrySessionLocation, SessionFile
+from agent_comms.store_files import AtomicFile, _store_lock
+from agent_comms.threads import ThreadProvenance
+from agent_comms.typed_table import HeldBusFile
+
+from .archived_roots import ArchivedCarry, CarryCatalogSource
+from .carry_claim_fields import ClaimFields
+from .carry_parked_threads import ParkedThreads
+
+
+@dataclass
+class Relocation:
+    """Classify each old path string against the root that published its thread.
+
+    FieldCodec walks records before their children, so an archived source names
+    its original root before the threads it holds are rewritten.
+    """
+
+    published: Path
+    carried: Counter[str] = field(default_factory=Counter)
+
+    @staticmethod
+    def selects(record: type) -> bool:
+        return record is HistorySource or issubclass(record, ThreadProvenance)
+
+    def change(self, record: type, raw: dict[str, object]) -> dict[str, object]:
+        if record is HistorySource:
+            self.published = Path(
+                str(raw.get(FieldCodec.record_key(HistorySource, "original_root")))
+            )
+            return raw
+        key = FieldCodec.record_key(ThreadProvenance, "session_file")
+        saved = raw.get(key)
+        if not isinstance(saved, str):
+            return raw
+        wire = FieldCodec.encode(
+            RegistrySessionLocation.from_path(saved, self.published), SessionFile
+        )
+        self.carried[type(FieldCodec.decode(SessionFile, wire)).__name__] += 1
+        return {**raw, key: wire}
+
+
+class SessionLocations(ArchivedCarry):
+    opened_files = (RegistryFile,)
+    # Both decode registries with this source, and so do the steps they precede.
+    precedes = (ClaimFields, ParkedThreads)
+
+    def carry(self, *, apply: bool) -> Carried:
+        if apply:
+            HeldBusFile(RegistryFile, RegistryFile.path_in(self.root)).require()
+        rewritten: list[Path] = []
+        catalog = HistoryArchive.path_in(self.root)
+        archived = Relocation(self.original_root)
+        with _store_lock(catalog):
+            recorded = json.loads(catalog.read_bytes()) if catalog.exists() else []
+            current = compile_decoder(tuple[HistorySource, ...]).rewrite_document(
+                recorded, archived.change, Relocation.selects
+            )
+            if archived.carried and not apply:
+                # The catalogue names the archived roots and decodes only once carried.
+                return Carried(archived.carried.total(), dict(archived.carried))
+            if archived.carried:
+                AtomicFile(catalog).write_text(json.dumps(current))
+                rewritten.append(Path(HistoryArchive.filename))
+            sources = CarryCatalogSource.read(self.root, self.original_root)
+            registries = Relocation(self.original_root)
+            for root in self.roots():
+                registry = RegistryStore.at(root)
+                if not registry.path.exists():
+                    continue
+                with registry.locked():
+                    registry.private_guard_unlocked()
+                    before = registries.carried.total()
+                    raw = json.loads(registry.path.read_bytes())
+                    carried = compile_decoder(RegistryDocument).rewrite_document(
+                        raw, registries.change, Relocation.selects
+                    )
+                    if apply and registries.carried.total() != before:
+                        registry._write_unlocked(json.dumps(carried, indent=2))
+                        rewritten.extend(path.relative_to(self.root) for path in registry.written)
+            if apply and rewritten:
+                rebound = tuple(
+                    source.carried_registry(source.mapped_root(self.root, self.original_root))
+                    for source in sources
+                )
+                if rebound:
+                    AtomicFile(catalog).write_text(json.dumps(FieldCodec.encode(rebound)))
+                    rewritten.append(Path(HistoryArchive.filename))
+        total = archived.carried + registries.carried
+        return Carried(total.total(), dict(total), tuple(dict.fromkeys(rewritten)))