Keep saved transcripts with their bus or external harness owner

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

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

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

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

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

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__ = ()

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

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,

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."""

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.

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