From 4cfd311bab6f5b799a1986b90ed2221241f3ba96 Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 08:37:44 -0400 Subject: [PATCH 1/4] Declare bus-owned and harness-owned registry session locations --- src/agent_comms/session_file.py | 76 ++++++++++++++++++++++++++------- 1 file changed, 61 insertions(+), 15 deletions(-) diff --git a/src/agent_comms/session_file.py b/src/agent_comms/session_file.py index 94b29c0d35..a496d27c47 100644 --- a/src/agent_comms/session_file.py +++ b/src/agent_comms/session_file.py @@ -1,46 +1,92 @@ -"""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 pathlib import Path +from typing import Annotated, TYPE_CHECKING -from .field_codec import SubclassUnion -from .thread_text import AbsentText, GivenText, TextChoice +from .active_route import CommsRoute +from .field_codec import NullMember, PathText, RecordMember, SubclassUnion 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 location(self, backend: type[AgentBackend]) -> RuntimeLocation: ... @abstractmethod - def applied(self, previous: SessionFile) -> SessionFile: - """The file a redeclaration keeps: its own, or the previous one.""" + def applied(self, previous: SessionFile) -> SessionFile: ... @dataclass(frozen=True, slots=True) -class SavedSessionFile(GivenText, SessionFile): - text: str +class RegistrySessionLocation(RecordMember, SessionFile, declared_name=None): + """A saved transcript whose storage owner was selected at publication.""" + + path: Annotated[Path, PathText] + + @classmethod + def from_path(cls, path: Path | str, root: Path) -> RegistrySessionLocation: + 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)) + return ExternalSessionLocation(absolute) def location(self, backend: type[AgentBackend]) -> RuntimeLocation: - return backend.location_type(self.text) + return backend.location_type(self) 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, slots=True) -class NoSessionFile(AbsentText, SessionFile): - def location(self, backend: type[AgentBackend]) -> RuntimeLocation: +class BusSessionLocation(RegistrySessionLocation): + def __post_init__(self): + if self.path.is_absolute() or ".." in self.path.parts: + raise ValueError(f"Bus session must be relative to its root: {self.path}") + + 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) +class ExternalSessionLocation(RegistrySessionLocation): + def __post_init__(self): + if not self.path.is_absolute(): + raise ValueError(f"External session must be absolute: {self.path}") + + def resolve(self, root: Path) -> Path: + return self.path + + def writable(self, root: Path) -> Path: + if root.resolve() != CommsRoute.current().observe_root(): + raise ValueError(f"Copied bus {root} cannot write external session {self.path}") + return self.resolve(root) + + +@dataclass(frozen=True, slots=True) +class NoSessionFile(NullMember, SessionFile): + def location(self, backend: type[AgentBackend]) -> RuntimeLocation: return UnstartedRuntime(backend) def applied(self, previous: SessionFile) -> SessionFile: From 502f3d851548bfde40aacc7456adb83a909f6ec7 Mon Sep 17 00:00:00 2001 From: Ratchet test Date: Fri, 23 Oct 2026 09:05:41 -0400 Subject: [PATCH 2/4] Bind saved transcript locations to the registry root that reads them The registry stored every saved transcript as one absolute string, so a copied root read and wrote the live root's transcripts. RegistryStore now binds each decoded thread's location to its own root, as it already binds turn records; a bus location resolves under that root and an external harness transcript refuses writes from any root but the published one. Writers publish through RegistrySessionLocation.from_path, and launches and compaction take the writable runtime. The duplicate SavedSessionFile pattern declaration is now BusTranscriptFile. A one-shot carry turns the old strings into bus and external locations before any step decodes the registry. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/check_root.py | 4 +- src/agent_comms/fresh_private_session.py | 4 +- src/agent_comms/manual_compaction_bridge.py | 2 +- src/agent_comms/native_entries.py | 4 +- src/agent_comms/native_session_files.py | 6 +- src/agent_comms/pi_durable_backend.py | 4 +- src/agent_comms/pi_native_backend.py | 4 +- src/agent_comms/registry_store.py | 14 ++-- src/agent_comms/session_file.py | 72 ++++++++++++++++-- src/agent_comms/thread_management.py | 18 ++--- src/agent_comms/threads.py | 5 ++ src/agent_comms/work_dispatch.py | 4 +- tests/acp_error_payload_installed_journey.py | 4 +- tests/c4_coordination_installed_ui.py | 4 +- tests/reply_relevance_installed_journey.py | 4 +- tests/shared_bus_restart_native.py | 6 +- tests/test_agent_backend.py | 7 +- tests/test_atomic_turn_source_owner.py | 4 +- tests/test_carry_session_locations.py | 67 +++++++++++++++++ tests/test_channels.py | 6 +- tests/test_cli.py | 4 +- tests/test_declarations.py | 4 +- tests/test_durable_host_records.py | 4 +- tests/test_durable_runtime_location.py | 6 +- tests/test_external_transcript.py | 4 +- tests/test_historical_views.py | 4 +- .../test_native_owner_launcher_resolution.py | 4 +- tests/test_native_unknown_recovery.py | 4 +- tests/test_operations.py | 14 ++-- tests/test_registration_inheritance.py | 6 +- tests/test_registration_process_identity.py | 4 +- tests/test_registration_transitions.py | 4 +- tests/test_restart.py | 6 +- tests/test_retained_context_operations.py | 4 +- tests/test_roster_restoration.py | 4 +- tests/test_runtime.py | 4 +- tests/test_runtime_subscription_handshake.py | 4 +- tests/test_standing_rules.py | 6 +- tests/test_start.py | 4 +- tests/test_task_decisions.py | 4 +- tests/test_thread_identity.py | 6 +- tests/test_thread_status.py | 4 +- tests/test_thread_unread.py | 4 +- tests/test_transcript_input_display.py | 4 +- tests/test_transcript_pages.py | 14 ++-- tests/test_transcript_partial_tail.py | 4 +- tests/test_transcript_read_identity.py | 10 +-- tests/test_transcript_read_lifetime.py | 4 +- tests/test_turn_digests.py | 4 +- tests/test_unit_state.py | 4 +- tests/test_user_view_mark_read.py | 4 +- tools/cutover/carry_session_locations.py | 73 +++++++++++++++++++ 52 files changed, 343 insertions(+), 133 deletions(-) create mode 100644 tests/test_carry_session_locations.py create mode 100644 tools/cutover/carry_session_locations.py 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/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_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 a496d27c47..738b444476 100644 --- a/src/agent_comms/session_file.py +++ b/src/agent_comms/session_file.py @@ -3,12 +3,13 @@ from __future__ import annotations from abc import abstractmethod -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Annotated, TYPE_CHECKING from .active_route import CommsRoute -from .field_codec import NullMember, PathText, RecordMember, SubclassUnion +from .errors import RelationViolationError +from .field_codec import NullMember, PathText, RecordMember, SubclassUnion, WireTransient from .unstarted_runtime import UnstartedRuntime if TYPE_CHECKING: @@ -18,9 +19,16 @@ class SessionFile(SubclassUnion, affix="SessionFile"): """A registry transcript location, or no saved transcript yet.""" + @abstractmethod + def bound_to(self, root: Path) -> SessionFile: + """The location read from the registry under ``root``.""" + @abstractmethod def location(self, backend: type[AgentBackend]) -> RuntimeLocation: ... + @abstractmethod + def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation: ... + @abstractmethod def applied(self, previous: SessionFile) -> SessionFile: ... @@ -32,18 +40,21 @@ class RegistrySessionLocation(RecordMember, SessionFile, declared_name=None): path: Annotated[Path, PathText] @classmethod - def from_path(cls, path: Path | str, root: Path) -> RegistrySessionLocation: + 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)) - return ExternalSessionLocation(absolute) + 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) + 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 @@ -55,12 +66,37 @@ def writable(self, root: Path) -> Path: return self.resolve(root) +@dataclass(frozen=True) +class BoundSessionLocation: + """A wire location bound to the root whose registry decoded it.""" + + root: Annotated[Path, WireTransient()] = field(repr=False, compare=False) + + def bound_to(self, root: Path) -> BoundSessionLocation: + """Publishing into another root keeps the transcript this location names.""" + root = root.resolve() + if self.root == root: + return self + return RegistrySessionLocation.from_path(self.resolve(self.root), root) + + def location(self, backend: type[AgentBackend]) -> RuntimeLocation: + return backend.location_type(str(self.resolve(self.root))) + + def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation: + return backend.location_type(str(self.writable(self.root))) + + @dataclass(frozen=True, slots=True) class BusSessionLocation(RegistrySessionLocation): + """A transcript stored under the bus root; a copied root reads its own copy.""" + def __post_init__(self): 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) -> BoundBusSession: + return BoundBusSession(self.path, root.resolve()) + def resolve(self, root: Path) -> Path: root = root.resolve() path = (root / self.path).resolve() @@ -71,23 +107,45 @@ def resolve(self, root: Path) -> Path: @dataclass(frozen=True, slots=True) class ExternalSessionLocation(RegistrySessionLocation): + """A harness transcript outside every bus root; only the published root writes it.""" + def __post_init__(self): if not self.path.is_absolute(): raise ValueError(f"External session must be absolute: {self.path}") + def bound_to(self, root: Path) -> BoundExternalSession: + 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 ValueError(f"Copied bus {root} cannot write external session {self.path}") + raise RelationViolationError( + f"Copied bus {root} cannot write external session {self.path}") return self.resolve(root) +@dataclass(frozen=True) +class BoundBusSession(BoundSessionLocation, BusSessionLocation, declared_name=None): + pass + + +@dataclass(frozen=True) +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..830ccd60ff 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,7 @@ 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)) + yield replace(thread, session_file=RegistrySessionLocation.from_path(location.read_key, threads.root)) @dataclass(frozen=True, slots=True, kw_only=True) @@ -175,16 +175,16 @@ 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), + def child(self, parent: Thread, runtime: RuntimeLocation, threads: ThreadManagement) -> Thread: + return replace(self.declaration(self.project, threads.registry), parent=ForkedParent(parent.name), tags=self.tags.for_child(parent.tags), - backend=parent.backend, session_file=SavedSessionFile(runtime.read_key)) + backend=parent.backend, session_file=RegistrySessionLocation.from_path(runtime.read_key, threads.root)) @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 +193,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 +246,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 +455,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/work_dispatch.py b/src/agent_comms/work_dispatch.py index 0d0764391e..46bea4693d 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,7 @@ 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..40e8e3b631 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, tmp_path) 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..4efe690fd4 --- /dev/null +++ b/tests/test_carry_session_locations.py @@ -0,0 +1,67 @@ +"""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.comms import wire +from agent_comms.errors import RelationViolationError +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): + 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()) diff --git a/tests/test_channels.py b/tests/test_channels.py index 64a6761c12..cbbf0c9f14 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), tmp_path), 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), tmp_path)) ) with pytest.raises(ValueError): comms.threads.create( diff --git a/tests/test_cli.py b/tests/test_cli.py index bff651a5cf..a22639605e 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), tmp_path))) 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..49137f516d 100644 --- a/tests/test_durable_runtime_location.py +++ b/tests/test_durable_runtime_location.py @@ -5,7 +5,7 @@ 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,7 +19,7 @@ @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)) assert restored.runtime == (location if saved else UnstartedRuntime(PiDurableBackend)) assert restored.runtime.backend is PiDurableBackend @@ -71,7 +71,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..46c8db7d41 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'), tmp_path))) with pytest.raises(ValueError): comms.transcripts.external("pi") diff --git a/tests/test_historical_views.py b/tests/test_historical_views.py index e14ab93945..cb050f04f3 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..9ba747429c 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, tmp_path)), 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..15fe12b78e 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), tmp_path)), 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..dc6c0e63de 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), tmp_path)) ) 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), tmp_path)) ) 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()), tmp_path) 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), tmp_path)) ) 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'), tmp_path)) ) 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"), tmp_path) 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..c4f168c77e 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"), tmp_path), 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, tmp_path) 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..812a2850c0 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), tmp_path) 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..d47a0a7cf7 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), tmp_path))) 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), tmp_path) 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..f9589b4282 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), tmp_path), 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..dc1ef391f0 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'), tmp_path), 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..0f43a8dc63 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"), tmp_path)) ) 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..b0d189b1d8 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), tmp_path)) 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..f1f9f86926 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), tmp_path))) 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..a20bb92c32 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), tmp_path))) 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..55f6bcc7e0 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), tmp_path))) 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 1aa92e9cc5..2b735b9f71 100644 --- a/tests/test_thread_identity.py +++ b/tests/test_thread_identity.py @@ -7,7 +7,7 @@ import pytest -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 @@ -315,12 +315,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..4299524283 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', tmp_path), 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..0ae1280762 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), tmp_path))) return comms, source diff --git a/tests/test_transcript_input_display.py b/tests/test_transcript_input_display.py index aa6be40033..64f55f8ed6 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), tmp_path))) 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..3970ae118f 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), tmp_path))) 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), tmp_path)) 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"), tmp_path))) 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), tmp_path)) ) 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), tmp_path)) ) 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), tmp_path), 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..f7c94f0c5c 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), tmp_path))) 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..ac4b9626e6 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), tmp_path))) 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), tmp_path))) 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), tmp_path))) 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), tmp_path))) 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..dd609898c2 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), tmp_path))) 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..77bc4b423c 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), tmp_path))) 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..8a7eccaa40 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), tmp_path))) 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..27d1f1a813 --- /dev/null +++ b/tools/cutover/carry_session_locations.py @@ -0,0 +1,73 @@ +"""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 pathlib import Path + +from agent_comms.field_codec import FieldCodec +from agent_comms.historical_views import HistoryArchive +from agent_comms.installer.carry import Carried +from agent_comms.locked_store import RegistryFile +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.typed_table import HeldBusFile + +from .archived_roots import ArchivedCarry, CarryCatalogSource +from .carry_claim_fields import ClaimFields +from .carry_parked_threads import ParkedThreads + + +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] = [] + carried: Counter[str] = Counter() + with _store_lock(HistoryArchive.path_in(self.root)): + sources = CarryCatalogSource.read(self.root, 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() + raw = json.loads(registry.path.read_bytes()) + changed = self.carry_threads(raw["threads"].values()) + carried.update(changed) + if changed and apply: + registry._write_unlocked(json.dumps(raw, 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(HistoryArchive.path_in(self.root)).write_text( + json.dumps(FieldCodec.encode(rebound))) + rewritten.append(Path(HistoryArchive.filename)) + return Carried(carried.total(), dict(carried), tuple(rewritten)) + + def carry_threads(self, threads) -> Counter[str]: + """Rewrite each old path string in place; the decoded member names the result.""" + changed: Counter[str] = Counter() + for thread in threads: + saved = thread.get("session_file") + if not isinstance(saved, str): + continue + location = RegistrySessionLocation.from_path(saved, self.original_root) + thread["session_file"] = FieldCodec.encode(location, SessionFile) + changed[type(FieldCodec.decode(SessionFile, thread["session_file"])).__name__] += 1 + return changed From f2ba027ee992d76a912def0e6a7a5df37965dc01 Mon Sep 17 00:00:00 2001 From: Citation fixture Date: Fri, 23 Oct 2026 09:21:08 -0400 Subject: [PATCH 3/4] Bind archived and witnessed transcripts to the root that published them A location's identity is its path: a bus path is relative and an external one absolute, so the bound root says where it is read, not which one it is. Archived history sources bind their provenance to the original root, and a decoded transcript read witness binds its thread to the root that published it. The carry rewrites the history catalogue against each source's original root before the registries. Tests build locations against the root of the registry that reads them instead of the test directory. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/historical_views.py | 4 +++ src/agent_comms/registry_provenance.py | 12 ++++++++- src/agent_comms/session_file.py | 25 ++++++++++++------- src/agent_comms/transcripts.py | 14 +++++++++++ tests/test_atomic_turn_source_owner.py | 2 +- tests/test_carry_session_locations.py | 24 ++++++++++++++++++ tests/test_channels.py | 4 +-- tests/test_check_root.py | 2 +- tests/test_cli.py | 2 +- tests/test_durable_runtime_location.py | 2 ++ tests/test_external_transcript.py | 2 +- .../test_native_owner_launcher_resolution.py | 2 +- tests/test_native_unknown_recovery.py | 2 +- tests/test_operations.py | 12 ++++----- tests/test_registration_inheritance.py | 4 +-- tests/test_registration_process_identity.py | 2 +- tests/test_restart.py | 4 +-- tests/test_retained_context_operations.py | 2 +- tests/test_roster_restoration.py | 2 +- tests/test_runtime.py | 2 +- tests/test_runtime_subscription_handshake.py | 2 +- tests/test_standing_rules.py | 2 +- tests/test_start.py | 2 +- tests/test_task_decisions.py | 2 +- tests/test_thread_status.py | 2 +- tests/test_thread_unread.py | 2 +- tests/test_transcript_input_display.py | 2 +- tests/test_transcript_pages.py | 12 ++++----- tests/test_transcript_partial_tail.py | 2 +- tests/test_transcript_read_identity.py | 8 +++--- tests/test_turn_digests.py | 2 +- tests/test_unit_state.py | 2 +- tests/test_user_view_mark_read.py | 2 +- tools/cutover/carry_session_locations.py | 25 +++++++++++++------ 34 files changed, 132 insertions(+), 60 deletions(-) diff --git a/src/agent_comms/historical_views.py b/src/agent_comms/historical_views.py index 2690c5b1d7..a7f911a4ea 100644 --- a/src/agent_comms/historical_views.py +++ b/src/agent_comms/historical_views.py @@ -261,6 +261,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/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/session_file.py b/src/agent_comms/session_file.py index 738b444476..3d585377ae 100644 --- a/src/agent_comms/session_file.py +++ b/src/agent_comms/session_file.py @@ -33,12 +33,20 @@ def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation: ... def applied(self, previous: SessionFile) -> SessionFile: ... -@dataclass(frozen=True, slots=True) +@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) @@ -66,16 +74,15 @@ def writable(self, root: Path) -> Path: return self.resolve(root) -@dataclass(frozen=True) +@dataclass(frozen=True, eq=False) class BoundSessionLocation: """A wire location bound to the root whose registry decoded it.""" - root: Annotated[Path, WireTransient()] = field(repr=False, compare=False) + root: Annotated[Path, WireTransient()] = field(repr=False) def bound_to(self, root: Path) -> BoundSessionLocation: """Publishing into another root keeps the transcript this location names.""" - root = root.resolve() - if self.root == root: + if self.root == root or self.root == (root := root.resolve()): return self return RegistrySessionLocation.from_path(self.resolve(self.root), root) @@ -86,7 +93,7 @@ def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation: return backend.location_type(str(self.writable(self.root))) -@dataclass(frozen=True, slots=True) +@dataclass(frozen=True, slots=True, eq=False) class BusSessionLocation(RegistrySessionLocation): """A transcript stored under the bus root; a copied root reads its own copy.""" @@ -105,7 +112,7 @@ def resolve(self, root: Path) -> Path: return path -@dataclass(frozen=True, slots=True) +@dataclass(frozen=True, slots=True, eq=False) class ExternalSessionLocation(RegistrySessionLocation): """A harness transcript outside every bus root; only the published root writes it.""" @@ -126,12 +133,12 @@ def writable(self, root: Path) -> Path: return self.resolve(root) -@dataclass(frozen=True) +@dataclass(frozen=True, eq=False) class BoundBusSession(BoundSessionLocation, BusSessionLocation, declared_name=None): pass -@dataclass(frozen=True) +@dataclass(frozen=True, eq=False) class BoundExternalSession(BoundSessionLocation, ExternalSessionLocation, declared_name=None): pass 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/tests/test_atomic_turn_source_owner.py b/tests/test_atomic_turn_source_owner.py index 40e8e3b631..6578c7886a 100644 --- a/tests/test_atomic_turn_source_owner.py +++ b/tests/test_atomic_turn_source_owner.py @@ -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 == RegistrySessionLocation.from_path(saved, tmp_path) + 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 index 4efe690fd4..65a86f67ce 100644 --- a/tests/test_carry_session_locations.py +++ b/tests/test_carry_session_locations.py @@ -6,8 +6,10 @@ 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 @@ -65,3 +67,25 @@ 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()) + + +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 cbbf0c9f14..e4e176fadc 100644 --- a/tests/test_channels.py +++ b/tests/test_channels.py @@ -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=RegistrySessionLocation.from_path(str(path), tmp_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=RegistrySessionLocation.from_path(str(parent_session), tmp_path)) + 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 a22639605e..82b8d96c8c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -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=RegistrySessionLocation.from_path(str(source), tmp_path))) + 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_durable_runtime_location.py b/tests/test_durable_runtime_location.py index 49137f516d..baf06e2bfc 100644 --- a/tests/test_durable_runtime_location.py +++ b/tests/test_durable_runtime_location.py @@ -1,6 +1,7 @@ """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 @@ -21,6 +22,7 @@ def test_durable_thread_roundtrip_and_worker_selection(tmp_path, saved, *, threa location = DurableDatabase(str(tmp_path / 'session.sqlite3')) 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 diff --git a/tests/test_external_transcript.py b/tests/test_external_transcript.py index 46c8db7d41..b241d00f14 100644 --- a/tests/test_external_transcript.py +++ b/tests/test_external_transcript.py @@ -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=RegistrySessionLocation.from_path(str(tmp_path / 'session.jsonl'), tmp_path))) + 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_native_owner_launcher_resolution.py b/tests/test_native_owner_launcher_resolution.py index 9ba747429c..1c1a0cdd38 100644 --- a/tests/test_native_owner_launcher_resolution.py +++ b/tests/test_native_owner_launcher_resolution.py @@ -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=RegistrySessionLocation.from_path(parent.session_file, tmp_path)), + 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 15fe12b78e..b2498bc3bb 100644 --- a/tests/test_native_unknown_recovery.py +++ b/tests/test_native_unknown_recovery.py @@ -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=RegistrySessionLocation.from_path(str(session), tmp_path)), + 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 dc6c0e63de..0fd54251c7 100644 --- a/tests/test_operations.py +++ b/tests/test_operations.py @@ -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=RegistrySessionLocation.from_path(str(session_file), tmp_path)) + 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=RegistrySessionLocation.from_path(str(session_file), tmp_path)) + 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 == RegistrySessionLocation.from_path(str(session_file.resolve()), tmp_path) + 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=RegistrySessionLocation.from_path(str(path), tmp_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=RegistrySessionLocation.from_path(str(tmp_path / 's.json'), tmp_path)) + 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 == RegistrySessionLocation.from_path(str(tmp_path / "s.json"), tmp_path) + 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 c4f168c77e..693e903bf2 100644 --- a/tests/test_registration_inheritance.py +++ b/tests/test_registration_inheritance.py @@ -45,7 +45,7 @@ def test_registry_redeclaration_inherits_metadata_but_respects_explicit_values(t "owner", frozenset({"team"}), '/old', - session_file=RegistrySessionLocation.from_path(str(tmp_path / "session.jsonl"), tmp_path), + 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 == RegistrySessionLocation.from_path(saved, tmp_path) + 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 812a2850c0..8a85e8198d 100644 --- a/tests/test_registration_process_identity.py +++ b/tests/test_registration_process_identity.py @@ -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 == RegistrySessionLocation.from_path(str(session), tmp_path) + 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_restart.py b/tests/test_restart.py index d47a0a7cf7..8989fcfd7a 100644 --- a/tests/test_restart.py +++ b/tests/test_restart.py @@ -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=RegistrySessionLocation.from_path(str(native.session), tmp_path))) + '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 == RegistrySessionLocation.from_path(str(native.session), tmp_path) + 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 f9589b4282..19a5fae5e3 100644 --- a/tests/test_retained_context_operations.py +++ b/tests/test_retained_context_operations.py @@ -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=RegistrySessionLocation.from_path(str(saved), tmp_path), 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 dc1ef391f0..984a4fabea 100644 --- a/tests/test_roster_restoration.py +++ b/tests/test_roster_restoration.py @@ -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=RegistrySessionLocation.from_path(str(tmp_path / 'saved.jsonl'), tmp_path), 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 0f43a8dc63..1638bc4934 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -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=RegistrySessionLocation.from_path(str(tmp_path / "session.jsonl"), tmp_path)) + 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 b0d189b1d8..eddaa93065 100644 --- a/tests/test_runtime_subscription_handshake.py +++ b/tests/test_runtime_subscription_handshake.py @@ -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=RegistrySessionLocation.from_path(str(session_file), tmp_path)) + 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 f1f9f86926..e36f71cd2a 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -24,7 +24,7 @@ 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=RegistrySessionLocation.from_path(str(path), tmp_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=RegistrySessionLocation.from_path(str(path), tmp_path), created_at=created)) diff --git a/tests/test_start.py b/tests/test_start.py index a20bb92c32..55b229cbaf 100644 --- a/tests/test_start.py +++ b/tests/test_start.py @@ -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=RegistrySessionLocation.from_path(str(source), tmp_path))) + 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 55f6bcc7e0..9a175b4a65 100644 --- a/tests/test_task_decisions.py +++ b/tests/test_task_decisions.py @@ -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=RegistrySessionLocation.from_path(str(saved), tmp_path))) + 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_status.py b/tests/test_thread_status.py index 4299524283..f13edd13cd 100644 --- a/tests/test_thread_status.py +++ b/tests/test_thread_status.py @@ -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=RegistrySessionLocation.from_path('/retained/session.jsonl', tmp_path), 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 0ae1280762..b2f4f74170 100644 --- a/tests/test_thread_unread.py +++ b/tests/test_thread_unread.py @@ -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=RegistrySessionLocation.from_path(str(source), tmp_path))) + 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 64f55f8ed6..68b83b66a8 100644 --- a/tests/test_transcript_input_display.py +++ b/tests/test_transcript_input_display.py @@ -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=RegistrySessionLocation.from_path(str(session), tmp_path))) + 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 3970ae118f..bf2a23b02a 100644 --- a/tests/test_transcript_pages.py +++ b/tests/test_transcript_pages.py @@ -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=RegistrySessionLocation.from_path(str(path), tmp_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=RegistrySessionLocation.from_path(str(path), tmp_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=RegistrySessionLocation.from_path(str(tmp_path / "other.jsonl"), tmp_path))) + 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=RegistrySessionLocation.from_path(str(parent_path), tmp_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=RegistrySessionLocation.from_path(str(parent_path), tmp_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=RegistrySessionLocation.from_path(str(path), tmp_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 f7c94f0c5c..2ea1b1f39b 100644 --- a/tests/test_transcript_partial_tail.py +++ b/tests/test_transcript_partial_tail.py @@ -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=RegistrySessionLocation.from_path(str(session), tmp_path))) + 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 ac4b9626e6..5abc38b32b 100644 --- a/tests/test_transcript_read_identity.py +++ b/tests/test_transcript_read_identity.py @@ -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=RegistrySessionLocation.from_path(str(path), tmp_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=RegistrySessionLocation.from_path(str(parent), tmp_path))) + 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=RegistrySessionLocation.from_path(str(path), tmp_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=RegistrySessionLocation.from_path(str(path), tmp_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_turn_digests.py b/tests/test_turn_digests.py index dd609898c2..0c8dee963a 100644 --- a/tests/test_turn_digests.py +++ b/tests/test_turn_digests.py @@ -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=RegistrySessionLocation.from_path(str(source), tmp_path))) + 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 77bc4b423c..781526a8c4 100644 --- a/tests/test_unit_state.py +++ b/tests/test_unit_state.py @@ -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=RegistrySessionLocation.from_path(str(path), tmp_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 8a7eccaa40..9e666413cc 100644 --- a/tests/test_user_view_mark_read.py +++ b/tests/test_user_view_mark_read.py @@ -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=RegistrySessionLocation.from_path(str(source), tmp_path))) + 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 index 27d1f1a813..72687bbac3 100644 --- a/tools/cutover/carry_session_locations.py +++ b/tools/cutover/carry_session_locations.py @@ -37,7 +37,18 @@ def carry(self, *, apply: bool) -> Carried: HeldBusFile(RegistryFile, RegistryFile.path_in(self.root)).require() rewritten: list[Path] = [] carried: Counter[str] = Counter() - with _store_lock(HistoryArchive.path_in(self.root)): + catalog = HistoryArchive.path_in(self.root) + with _store_lock(catalog): + recorded = json.loads(catalog.read_bytes()) if catalog.exists() else [] + for source in recorded: + carried.update(self.carry_threads( + source["provenance"]["threads"].values(), Path(source["original_root"]))) + if carried and not apply: + # The catalogue names the archived roots and decodes only once carried. + return Carried(carried.total(), dict(carried)) + if carried: + AtomicFile(catalog).write_text(json.dumps(recorded)) + rewritten.append(Path(HistoryArchive.filename)) sources = CarryCatalogSource.read(self.root, self.original_root) for root in self.roots(): registry = RegistryStore.at(root) @@ -46,7 +57,7 @@ def carry(self, *, apply: bool) -> Carried: with registry.locked(): registry.private_guard_unlocked() raw = json.loads(registry.path.read_bytes()) - changed = self.carry_threads(raw["threads"].values()) + changed = self.carry_threads(raw["threads"].values(), self.original_root) carried.update(changed) if changed and apply: registry._write_unlocked(json.dumps(raw, indent=2)) @@ -55,19 +66,19 @@ def carry(self, *, apply: bool) -> Carried: rebound = tuple(source.carried_registry(source.mapped_root(self.root, self.original_root)) for source in sources) if rebound: - AtomicFile(HistoryArchive.path_in(self.root)).write_text( - json.dumps(FieldCodec.encode(rebound))) + AtomicFile(catalog).write_text(json.dumps(FieldCodec.encode(rebound))) rewritten.append(Path(HistoryArchive.filename)) - return Carried(carried.total(), dict(carried), tuple(rewritten)) + return Carried(carried.total(), dict(carried), tuple(dict.fromkeys(rewritten))) - def carry_threads(self, threads) -> Counter[str]: + @staticmethod + def carry_threads(threads, published: Path) -> Counter[str]: """Rewrite each old path string in place; the decoded member names the result.""" changed: Counter[str] = Counter() for thread in threads: saved = thread.get("session_file") if not isinstance(saved, str): continue - location = RegistrySessionLocation.from_path(saved, self.original_root) + location = RegistrySessionLocation.from_path(saved, published) thread["session_file"] = FieldCodec.encode(location, SessionFile) changed[type(FieldCodec.decode(SessionFile, thread["session_file"])).__name__] += 1 return changed From b93d3fe825ba80a4191b494a9f6bdbb8132b6d3b Mon Sep 17 00:00:00 2001 From: Citation fixture Date: Fri, 23 Oct 2026 09:33:52 -0400 Subject: [PATCH 4/4] Carry session locations through the declared registry records The carry walked raw JSON by hand: string subscripts on the catalogue, provenance and thread rows, restating the shape RegistryDocument and HistorySource already declare. FieldCodec now walks both documents and hands each ThreadProvenance record to one Relocation change; an archived source names its original root before its threads, so each path is classified against the root that published it. A bound location is a member of the registry location family, so it inherits the resolve and writable it calls, and every writer's result is a SessionFile. The carry test isolates HOME so it never reads this machine's published route. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/session_file.py | 23 ++++--- src/agent_comms/thread_management.py | 10 +-- src/agent_comms/work_dispatch.py | 5 +- tests/test_carry_session_locations.py | 4 +- tools/cutover/carry_session_locations.py | 88 ++++++++++++++++-------- 5 files changed, 85 insertions(+), 45 deletions(-) diff --git a/src/agent_comms/session_file.py b/src/agent_comms/session_file.py index 3d585377ae..1449840427 100644 --- a/src/agent_comms/session_file.py +++ b/src/agent_comms/session_file.py @@ -5,7 +5,7 @@ from abc import abstractmethod from dataclasses import dataclass, field from pathlib import Path -from typing import Annotated, TYPE_CHECKING +from typing import TYPE_CHECKING, Annotated from .active_route import CommsRoute from .errors import RelationViolationError @@ -75,7 +75,7 @@ def writable(self, root: Path) -> Path: @dataclass(frozen=True, eq=False) -class BoundSessionLocation: +class BoundSessionLocation(RegistrySessionLocation, declared_name=None): """A wire location bound to the root whose registry decoded it.""" root: Annotated[Path, WireTransient()] = field(repr=False) @@ -87,21 +87,25 @@ def bound_to(self, root: Path) -> BoundSessionLocation: return RegistrySessionLocation.from_path(self.resolve(self.root), root) def location(self, backend: type[AgentBackend]) -> RuntimeLocation: - return backend.location_type(str(self.resolve(self.root))) + return self.located(backend, self.resolve(self.root)) def writable_location(self, backend: type[AgentBackend]) -> RuntimeLocation: - return backend.location_type(str(self.writable(self.root))) + 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): + 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) -> BoundBusSession: + def bound_to(self, root: Path) -> BoundSessionLocation: return BoundBusSession(self.path, root.resolve()) def resolve(self, root: Path) -> Path: @@ -116,11 +120,11 @@ def resolve(self, root: Path) -> Path: class ExternalSessionLocation(RegistrySessionLocation): """A harness transcript outside every bus root; only the published root writes it.""" - def __post_init__(self): + 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) -> BoundExternalSession: + def bound_to(self, root: Path) -> BoundSessionLocation: return BoundExternalSession(self.path, root.resolve()) def resolve(self, root: Path) -> Path: @@ -129,7 +133,8 @@ def resolve(self, root: Path) -> 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}") + f"Copied bus {root} cannot write external session {self.path}" + ) return self.resolve(root) diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py index 830ccd60ff..7fed8114a3 100644 --- a/src/agent_comms/thread_management.py +++ b/src/agent_comms/thread_management.py @@ -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=RegistrySessionLocation.from_path(location.read_key, threads.root)) + saved = RegistrySessionLocation.from_path(location.read_key, threads.root) + yield replace(thread, session_file=saved) @dataclass(frozen=True, slots=True, kw_only=True) @@ -176,9 +177,10 @@ def _parent(self, threads: ThreadManagement) -> Thread: return parent def child(self, parent: Thread, runtime: RuntimeLocation, threads: ThreadManagement) -> Thread: - return replace(self.declaration(self.project, threads.registry), parent=ForkedParent(parent.name), - tags=self.tags.for_child(parent.tags), - backend=parent.backend, session_file=RegistrySessionLocation.from_path(runtime.read_key, threads.root)) + 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]: diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py index 46bea4693d..82eb125304 100644 --- a/src/agent_comms/work_dispatch.py +++ b/src/agent_comms/work_dispatch.py @@ -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=RegistrySessionLocation.from_path(location.read_key, comms.root)), + 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/test_carry_session_locations.py b/tests/test_carry_session_locations.py index 65a86f67ce..005f05f726 100644 --- a/tests/test_carry_session_locations.py +++ b/tests/test_carry_session_locations.py @@ -33,7 +33,9 @@ def old_registry(tmp_path): return comms, bus, external -def test_carry_splits_bus_and_external_transcripts_once(tmp_path): +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): diff --git a/tools/cutover/carry_session_locations.py b/tools/cutover/carry_session_locations.py index 72687bbac3..fb68c2a686 100644 --- a/tools/cutover/carry_session_locations.py +++ b/tools/cutover/carry_session_locations.py @@ -11,15 +11,18 @@ import json from collections import Counter +from dataclasses import dataclass, field from pathlib import Path -from agent_comms.field_codec import FieldCodec -from agent_comms.historical_views import HistoryArchive +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 @@ -27,6 +30,38 @@ 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. @@ -36,49 +71,42 @@ def carry(self, *, apply: bool) -> Carried: if apply: HeldBusFile(RegistryFile, RegistryFile.path_in(self.root)).require() rewritten: list[Path] = [] - carried: Counter[str] = Counter() 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 [] - for source in recorded: - carried.update(self.carry_threads( - source["provenance"]["threads"].values(), Path(source["original_root"]))) - if carried and not apply: + 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(carried.total(), dict(carried)) - if carried: - AtomicFile(catalog).write_text(json.dumps(recorded)) + 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()) - changed = self.carry_threads(raw["threads"].values(), self.original_root) - carried.update(changed) - if changed and apply: - registry._write_unlocked(json.dumps(raw, indent=2)) + 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) + 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)) - return Carried(carried.total(), dict(carried), tuple(dict.fromkeys(rewritten))) - - @staticmethod - def carry_threads(threads, published: Path) -> Counter[str]: - """Rewrite each old path string in place; the decoded member names the result.""" - changed: Counter[str] = Counter() - for thread in threads: - saved = thread.get("session_file") - if not isinstance(saved, str): - continue - location = RegistrySessionLocation.from_path(saved, published) - thread["session_file"] = FieldCodec.encode(location, SessionFile) - changed[type(FieldCodec.decode(SessionFile, thread["session_file"])).__name__] += 1 - return changed + total = archived.carried + registries.carried + return Carried(total.total(), dict(total), tuple(dict.fromkeys(rewritten)))