diff --git a/src/agent_comms/session_file.py b/src/agent_comms/session_file.py index 1449840427..2567fded98 100644 --- a/src/agent_comms/session_file.py +++ b/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) diff --git a/tests/test_carry_session_locations.py b/tests/test_carry_session_locations.py index 005f05f726..a505637218 100644 --- a/tests/test_carry_session_locations.py +++ b/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):