Let the root that published a harness transcript write it

src/agent_comms/session_file.py

@@ -7,7 +7,6 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated
 
-from .active_route import CommsRoute
 from .errors import RelationViolationError
 from .field_codec import NullMember, PathText, RecordMember, SubclassUnion, WireTransient
 from .unstarted_runtime import UnstartedRuntime
@@ -52,11 +51,15 @@ 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()
+        return cls.publish(absolute.resolve(), root, root)
+
+    @staticmethod
+    def publish(absolute: Path, root: Path, publisher: Path) -> BoundSessionLocation:
+        """A path under the root is the bus's own; any other belongs to its publishing root."""
         if absolute.is_relative_to(root):
             return BusSessionLocation(absolute.relative_to(root)).bound_to(root)
-        return ExternalSessionLocation(absolute).bound_to(root)
+        return ExternalSessionLocation(absolute, publisher).bound_to(root)
 
     def location(self, backend: type[AgentBackend]) -> RuntimeLocation:
         raise RelationViolationError(f"Saved session {self.path} must be bound by its registry")
@@ -70,6 +73,10 @@ def applied(self, previous: SessionFile) -> SessionFile:
     @abstractmethod
     def resolve(self, root: Path) -> Path: ...
 
+    @abstractmethod
+    def publisher(self, root: Path) -> Path:
+        """The root whose registry published this transcript, read under ``root``."""
+
     def writable(self, root: Path) -> Path:
         return self.resolve(root)
 
@@ -84,7 +91,7 @@ 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)
+        return self.publish(self.resolve(self.root), root, self.publisher(self.root))
 
     def location(self, backend: type[AgentBackend]) -> RuntimeLocation:
         return self.located(backend, self.resolve(self.root))
@@ -115,25 +122,39 @@ def resolve(self, root: Path) -> Path:
             raise ValueError(f"Bus session escapes its root: {self.path}")
         return path
 
+    def publisher(self, root: Path) -> Path:
+        return root.resolve()
 
-@dataclass(frozen=True, slots=True, eq=False)
+
+@dataclass(frozen=True, eq=False)
 class ExternalSessionLocation(RegistrySessionLocation):
-    """A harness transcript outside every bus root; only the published root writes it."""
+    """A harness transcript outside every bus root; only the root that published it writes it.
+
+    A copied registry keeps the published root, so the copy reads the transcript and refuses
+    to write it."""
+
+    published: Annotated[Path, PathText]
 
     def __post_init__(self) -> None:
-        if not self.path.is_absolute():
-            raise ValueError(f"External session must be absolute: {self.path}")
+        if not self.path.is_absolute() or not self.published.is_absolute():
+            raise ValueError(
+                f"External session must be absolute: {self.path} from {self.published}"
+            )
 
     def bound_to(self, root: Path) -> BoundSessionLocation:
-        return BoundExternalSession(self.path, root.resolve())
+        return BoundExternalSession(self.path, self.published, root.resolve())
 
     def resolve(self, root: Path) -> Path:
         return self.path
 
+    def publisher(self, root: Path) -> Path:
+        return self.published
+
     def writable(self, root: Path) -> Path:
-        if root.resolve() != CommsRoute.current().observe_root():
+        if root.resolve() != self.published.resolve():
             raise RelationViolationError(
-                f"Copied bus {root} cannot write external session {self.path}"
+                f"Copied bus {root} cannot write external session {self.path} "
+                f"published by {self.published}"
             )
         return self.resolve(root)
 

tests/test_carry_session_locations.py

@@ -33,9 +33,7 @@ def old_registry(tmp_path):
     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"))
+def test_carry_splits_bus_and_external_transcripts_once(tmp_path):
     comms, bus, external = old_registry(tmp_path)
     step = SessionLocations(comms.root, comms.root, tmp_path / "spill")
     with maintenance_held(comms.root):
@@ -45,7 +43,8 @@ def test_carry_splits_bus_and_external_transcripts_once(tmp_path, monkeypatch):
         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)}
+    published = {"path": str(external), "published": str(comms.root.resolve())}
+    assert stored["harness"]["session_file"] == published
 
     copy = tmp_path / "copy"
     shutil.copytree(comms.root, copy, symlinks=True)
@@ -59,7 +58,7 @@ def test_carry_splits_bus_and_external_transcripts_once(tmp_path, monkeypatch):
 
 def test_unbound_location_refuses_to_guess_its_root():
     for location in (BusSessionLocation(Path("native-sessions/a.jsonl")),
-                     ExternalSessionLocation(Path("/harness/a.jsonl"))):
+                     ExternalSessionLocation(Path("/harness/a.jsonl"), Path("/bus"))):
         with pytest.raises(RelationViolationError, match="must be bound by its registry"):
             location.location(PiNativeBackend)
 
@@ -69,6 +68,18 @@ def test_publishing_into_another_root_keeps_the_named_transcript(tmp_path):
     moved = first.bound_to(tmp_path / "b")
     assert isinstance(moved, ExternalSessionLocation)
     assert moved.location(PiNativeBackend).path == str((tmp_path / "a" / "s.jsonl").resolve())
+    assert moved.published == (tmp_path / "a").resolve()
+    with pytest.raises(RelationViolationError, match="published by"):
+        moved.writable_location(PiNativeBackend)
+
+
+def test_a_private_root_writes_the_harness_transcript_it_published(tmp_path):
+    comms = wire(tmp_path / "private")
+    harness = tmp_path / "harness" / "external.jsonl"
+    comms.registry.declare(comms.registry.new_thread(
+        "owner", frozenset(), str(tmp_path),
+        session_file=RegistrySessionLocation.from_path(harness, comms.root)))
+    assert comms.registry.require("owner").writable_runtime.path == str(harness.resolve())
 
 
 def test_carry_converts_the_archived_catalogue_before_reading_its_roots(tmp_path):