Publish saved transcripts through registry session locations

tests/app/test_context_reader_session_revision.py

@@ -10,12 +10,12 @@
 def declared_reader(workspace):
     from agent_comms.fresh_private_session import create_fresh_private_session
     from agent_comms.native_session_files import NativeSessionFiles
-    from agent_comms.session_file import SavedSessionFile
+    from agent_comms.session_file import RegistrySessionLocation
 
     identity = create_fresh_private_session(workspace.comms.root, worktree=workspace.project)
     registry = workspace.comms.registry
     registry.declare(registry.new_thread(
-        "context-reader", frozenset(), str(workspace.project), session_file=SavedSessionFile(identity.session_file)))
+        "context-reader", frozenset(), str(workspace.project), session_file=RegistrySessionLocation.from_path(identity.session_file, workspace.comms.root)))
     thread = registry.require("context-reader")
     files = NativeSessionFiles.for_identity(
         identity, root=workspace.comms.root, owner=thread.incarnation.require_recorded())

tests/app/test_native_message_parts.py

@@ -1,5 +1,6 @@
 """One saved native assistant message keeps its Markdown parts together under a single timestamp."""
 
+from agent_comms.session_file import RegistrySessionLocation
 from support.journals import assistant_entry, write_journal
 
 
@@ -8,7 +9,7 @@ async def test_saved_message_in_parts_shows_as_one_message(workspace):
     entry = assistant_entry("", id="native-message")
     entry["message"]["content"] = [{"type": "text", "text": part} for part in parts]
     session = write_journal(workspace.project / "native.jsonl", entry)
-    workspace.comms.registry.declare(workspace.comms.registry.new_thread("worker", frozenset(), str(workspace.project), session_file=str(session)))
+    workspace.comms.registry.declare(workspace.comms.registry.new_thread("worker", frozenset(), str(workspace.project), session_file=RegistrySessionLocation.from_path(session, workspace.comms.root)))
     async with workspace.app(size=(110, 38)) as ui:
         await ui.app.thread_navigation.open(
             owner_mode=ui.selected_tab, project_path=workspace.project, target="worker")

tests/app/test_recorded_transcript_reader.py

@@ -7,7 +7,7 @@
 
 from agent_comms.acp_extension import TranscriptSnapshotUpdate, decode_updates, encode_updates
 from agent_comms.coordination_errors import HistoricalSourceChanged, StaleRevision
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.transcripts import RecordedTranscriptReadIdentity
 from support.roots import installed_wire
 from support.journals import assistant_entry, write_journal
@@ -21,7 +21,7 @@ def recorded(tmp_path, workspace):
     old, live = installed_wire(tmp_path / "old"), workspace.comms
     native = write_journal(tmp_path / "authored.jsonl", assistant_entry("recorded answer", id="answer-1"))
     old.registry.declare(old.registry.new_thread("reader", frozenset(), str(old.root),
-                                created_at=10.0, session_file=SavedSessionFile(str(native))))
+                                created_at=10.0, session_file=RegistrySessionLocation.from_path(str(native), old.root)))
     live.registry.declare(live.registry.new_thread("reader", frozenset(), str(live.root), created_at=20.0))
     source = live.views.attach_history(old.root)
     return live, source, native

tests/app/test_transcript_recipient.py

@@ -1,5 +1,6 @@
 """A replay belongs to its requested tab, not whichever tab is still opening."""
 
+from agent_comms.session_file import RegistrySessionLocation
 from dataclasses import replace
 
 from agent_comms.acp_extension import TranscriptSnapshotUpdate
@@ -16,7 +17,7 @@ async def test_unbound_tab_refuses_another_threads_snapshot(workspace):
         entries = (assistant_entry("Alpha-only history", id="alpha-message"),) if name == "alpha" else ()
         journal = write_journal(workspace.project / f"{name}.jsonl", *entries)
         workspace.comms.registry.declare(workspace.comms.registry.new_thread(
-            name, frozenset(), str(workspace.project), session_file=str(journal)))
+            name, frozenset(), str(workspace.project), session_file=RegistrySessionLocation.from_path(journal, workspace.comms.root)))
         read = workspace.comms.transcripts.capture_page_read(name)
         updates[name] = TranscriptSnapshotUpdate(read.read(), read.identity)
 

tests/native/test_native_journey.py

@@ -3,6 +3,7 @@
 The model is the controlled loopback provider; everything else is the real path.
 """
 
+from agent_comms.session_file import RegistrySessionLocation
 import asyncio
 import json
 from dataclasses import replace
@@ -118,7 +119,7 @@ async def test_input_queued_behind_a_running_turn_is_consumed_and_saved(workspac
         assert view.transcript.displayed_cursor is not None
 
         native_rows = [json.loads(line) for line in
-                       Path(comms.registry.require("beta").session_file).read_text().splitlines()]
+                       Path(comms.registry.require("beta").runtime.path).read_text().splitlines()]
         disposition = InputDispositions(comms.root / InputDispositions.filename).read().rows["acp:" + queued]
         assert any(row.get("message", {}).get("inputId") == disposition.native_id for row in native_rows)
         assert not disposition.unresolved

tests/support/peers/delayed_load.py

@@ -5,6 +5,7 @@
 one-answer session file, records it on the thread and publishes the transcript page.
 """
 
+from agent_comms.session_file import RegistrySessionLocation
 import asyncio
 from dataclasses import replace
 import json
@@ -45,7 +46,7 @@ async def load_session(self, session_id, cwd, **kwargs):
         saved = self.project / f"sdk-history-{session_id}.jsonl"
         saved.write_text(json.dumps({"type": "message", "id": "saved-first-open", "message": {
             "role": "assistant", "content": [{"type": "text", "text": f"SAVED_FIRST_OPEN_{session_id}"}]}}) + "\n")
-        comms.registry.declare(replace(comms.registry.require(session_id), session_file=str(saved)))
+        comms.registry.declare(replace(comms.registry.require(session_id), session_file=RegistrySessionLocation.from_path(saved, comms.root)))
         read = comms.transcripts.capture_page_read(session_id)
         page = read.read()
         await self.connection.session_update(session_id=session_id, update=AgentMessageChunk(

tests/terminal/test_resize.py

@@ -2,7 +2,7 @@
 
 import sys
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 
 from support.journals import assistant_entry, write_journal
 from support.terminal import terminal_running, toad_command
@@ -18,7 +18,7 @@ async def test_every_terminal_resize_repaints_transcript_sidebar_and_composer(
         "AGENT_COMMS_AGENT_BIN": "pi", "AGENT_COMMS_AGENT_MODELS": "test/model",
         "AGENT_COMMS_AGENT_ARGS": "--model test/model"}
     workspace.comms.registry.declare(workspace.comms.registry.new_thread(
-        "resize", frozenset(), str(workspace.project), session_file=SavedSessionFile(str(journal))))
+        "resize", frozenset(), str(workspace.project), session_file=RegistrySessionLocation.from_path(str(journal), workspace.comms.root)))
     command = toad_command("acp", f"{sys.executable} -m agent_comms.acp", "--session", "resize")
     async with terminal_running(workspace, terminal_emulator, command, environment=environment) as terminal:
         await terminal.until(lambda: terminal.shows("A paragraph") or terminal.shows("REPLAY_MARKER_39"),

tests/textual/test_app_close_order.py

@@ -9,7 +9,7 @@
 from agent_comms.acp_extension import CoordinationChangedUpdate, encode_updates
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.context_reading import UnmeasuredContext
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.fresh_private_session import create_fresh_private_session
 
 from support.agents import deliver
@@ -36,7 +36,7 @@ async def test_closing_the_app_after_the_observation_process_exits_raises_nothin
         stream.write(json.dumps({"type": "message", "id": "ask",
             "timestamp": "2026-10-22T00:00:00Z", "message": {
         "role": "user", "content": [{"type": "text", "text": "ASK_ROW"}]}}) + "\n")
-    thread = comms.registry.declare(replace(workspace.thread("source"), session_file=SavedSessionFile(str(saved))))
+    thread = comms.registry.declare(replace(workspace.thread("source"), session_file=RegistrySessionLocation.from_path(str(saved), workspace.comms.root)))
     workspace.thread("peer")
     comms.messaging.send("peer", "source", "SAVED_NOTICE")
     async with workspace.app(size=(120, 35)) as ui:

tests/textual/test_archive_from_thread_menu.py

@@ -2,6 +2,7 @@
 choosing Archive archives the thread on the wire while keeping its messages, goal history, session file
 and saved session."""
 
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.cli_commands import ArchiveCliCommand, CliCommand
 from agent_comms.goal_actions import SetGoalAction
 from agent_comms.thread_status import ArchivedThreadStatus, StoppedThreadStatus
@@ -25,7 +26,7 @@ class RetainAction(ArchiveCliCommand):
     session_file = project / "peer.jsonl"
     session_file.write_text('{"type":"session","id":"retained"}\n')
     comms.registry.register(comms.registry.new_thread("owner", frozenset(), str(project)))
-    comms.registry.register(comms.registry.new_thread("peer", frozenset(), str(project), session_file=str(session_file)),
+    comms.registry.register(comms.registry.new_thread("peer", frozenset(), str(project), session_file=RegistrySessionLocation.from_path(session_file, workspace.comms.root)),
                             StoppedThreadStatus())
     comms.messaging.send("peer", "owner", "Retain this archived conversation")
     comms.goals.update_goal("peer", SetGoalAction(text="Retain the goal revisions"))

tests/textual/test_arrival_fade.py

@@ -10,7 +10,7 @@
 from agent_comms.context_reading import UnmeasuredContext
 from agent_comms.turn_lease import ActiveTurn, TurnState
 from agent_comms.turn_phase import PublishingPhase
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.fresh_private_session import create_fresh_private_session
 from ui_theme.effects import NEW_TEXT
 
@@ -44,7 +44,7 @@ async def test_a_commit_retiring_a_live_block_still_fades_another_senders_row(wo
         "timestamp": "2026-10-22T00:00:00Z", "message": {
         "role": "user", "content": [{"type": "text", "text": "ASK_ROW"}]}}) + "\n"
     saved.write_text(prompt)
-    thread = comms.registry.declare(replace(workspace.thread("source"), session_file=SavedSessionFile(str(saved))))
+    thread = comms.registry.declare(replace(workspace.thread("source"), session_file=RegistrySessionLocation.from_path(str(saved), workspace.comms.root)))
     workspace.thread("peer")
     comms.messaging.send("peer", "source", "EARLIER_NOTICE")
     async with workspace.app(size=(120, 35)) as ui:

tests/textual/test_checkpoint_first_source.py

@@ -3,6 +3,7 @@
 Textual owns mounting and removing the history and response widgets that the checkpoint replaces.
 """
 
+from agent_comms.session_file import RegistrySessionLocation
 import asyncio
 from dataclasses import replace
 import os
@@ -42,7 +43,7 @@ async def test_first_native_file_replaces_the_wire_history_and_the_covered_live_
             journals.user_entry("FIRST_SAVED_NATIVE_USER", id="native-user-original"),
             journals.assistant_entry("FIRST_SAVED_NATIVE_RESPONSE", id="native-assistant-original",
                                      parent="native-user-original"))
-        installed = comms.registry.declare(replace(thread, session_file=str(source)))
+        installed = comms.registry.declare(replace(thread, session_file=RegistrySessionLocation.from_path(source, workspace.comms.root)))
         assert installed.incarnation == thread.incarnation
         page = await view.transcript.get_transcript_page()
         assert page.after.session_file != old_page.after.session_file

tests/textual/test_late_live_answer.py

@@ -9,7 +9,7 @@
 from agent_comms.acp_extension import (
     TranscriptSnapshotUpdate, TurnChangedUpdate, encode_updates,
 )
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.turn_lease import ActiveTurn, TurnState
 from agent_comms.turn_phase import PublishingPhase
 
@@ -97,7 +97,7 @@ async def held_read(*arguments, **options):
             await asyncio.wait_for(held.wait(), 5)
             # The session file and the end of the turn arrive while the live answer is still unapplied.
             saved = write_journal(project / "first-native.jsonl", assistant_entry(TOKEN, id="first-assistant"))
-            comms.registry.declare(replace(thread, session_file=SavedSessionFile(str(saved))))
+            comms.registry.declare(replace(thread, session_file=RegistrySessionLocation.from_path(str(saved), workspace.comms.root)))
             pending.append(asyncio.create_task(deliver(agent, {
                 "sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": ""},
                 "_meta": encode_updates(TurnChangedUpdate(TurnState(finished_turn_id="first-turn")))})))
@@ -153,7 +153,7 @@ async def test_an_advancing_commit_retires_the_outgoing_body_and_its_tool_report
     workspace.thread("peer")
     journal = write_journal(workspace.project / "source.jsonl",
                             assistant_entry("SAVED_NATIVE_PREFIX", id="prefix"))
-    workspace.comms.registry.declare(replace(source, session_file=SavedSessionFile(str(journal))))
+    workspace.comms.registry.declare(replace(source, session_file=RegistrySessionLocation.from_path(str(journal), workspace.comms.root)))
     async with workspace.app(size=(120, 35), resume="source") as ui:
         view = ui.conversation
         await ui.until_shows("SAVED_NATIVE_PREFIX")

tests/textual/test_main_menu_transfer.py

@@ -1,5 +1,6 @@
 """Textual-facing: the main menu button opens Toad's menu and its dialogs drive Core's wire export and stopped-thread import."""
 
+from agent_comms.session_file import RegistrySessionLocation
 import json
 from pathlib import Path
 
@@ -94,4 +95,4 @@ async def test_main_menu_imports_a_session_as_a_stopped_thread(workspace):
         await ui.until(lambda: comms.registry.snapshot().taken("imported"), seconds=10, why="the imported thread")
         thread = comms.registry.require("imported")
         assert comms.registry.status(thread.name).stopped
-        assert "Imported history" in Path(thread.session_file).read_text()
+        assert "Imported history" in Path(thread.runtime.path).read_text()

tests/textual/test_message_filter_restyle.py

@@ -5,7 +5,7 @@
 
 from toad.widgets.agent_response import AgentResponse
 from toad.widgets.agent_thought import AgentThought
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.ui_model.message_category import ThinkingCategory, UserCategory, all_categories
 from agent_comms.ui_model.message_category import InboundCategory
 from toad.message_rows import MarkdownMessageStyle
@@ -44,7 +44,7 @@ async def test_filtering_saved_history_paints_an_older_projection_and_returns_to
                             parent=f"a{index - 1}" if index else "user")
             for index in range(300)))
     workspace.comms.registry.declare(replace(
-        workspace.thread("source"), session_file=SavedSessionFile(str(journal))))
+        workspace.thread("source"), session_file=RegistrySessionLocation.from_path(str(journal), workspace.comms.root)))
     async with workspace.app(size=(120, 35), resume="source") as ui:
         view = ui.conversation
         await ui.until(lambda: "SAVED_ASSISTANT_299" in ui.painted(view.window),
@@ -69,7 +69,7 @@ async def test_filtering_saved_history_paints_an_older_projection_and_returns_to
                                 parent=f"new-a{index - 1}" if index else "new-user")
                 for index in range(300)))
         workspace.comms.registry.declare(replace(
-            workspace.comms.registry.require("source"), session_file=SavedSessionFile(str(replacement))))
+            workspace.comms.registry.require("source"), session_file=RegistrySessionLocation.from_path(str(replacement), workspace.comms.root)))
         page = await ui.app.preparation.run_thread(workspace.comms.transcripts.thread_transcript_page, "source")
         prepared = await ui.app.preparation.submit(RenderPreparation(TranscriptPageRenderTask(page)))
         assert await history.show_newest(PreparedReplacementViewportRequest(

tests/textual/test_page_read_against_changed_source.py

@@ -3,7 +3,7 @@
 import asyncio
 from dataclasses import replace
 
-from agent_comms.session_file import SavedSessionFile
+from agent_comms.session_file import RegistrySessionLocation
 from support import journals
 import pytest
 from support.textual_steps import saved_history
@@ -16,7 +16,7 @@ async def test_older_page_is_published_only_when_the_source_did_not_change_durin
     journal = journals.write_journal(workspace.project / "saved.jsonl", *(
         journals.assistant_entry(f"Saved row {i}.", id=f"a{i}", parent=f"a{i - 1}" if i else None) for i in range(4)))
     transcripts = workspace.comms.transcripts
-    workspace.comms.registry.declare(replace(workspace.thread("saved"), session_file=SavedSessionFile(str(journal))))
+    workspace.comms.registry.declare(replace(workspace.thread("saved"), session_file=RegistrySessionLocation.from_path(str(journal), workspace.comms.root)))
     page = transcripts.thread_transcript_page("saved", max_messages=1)
     assert page.has_older
     async with workspace.app(size=(110, 35)) as ui:
@@ -76,7 +76,7 @@ async def test_requested_older_page_is_read_by_the_history_worker(workspace):
     journal = journals.write_journal(workspace.project / "saved.jsonl", *(
         journals.assistant_entry(f"Saved row {i}.", id=f"a{i}", parent=f"a{i - 1}" if i else None) for i in range(4)))
     transcripts = workspace.comms.transcripts
-    workspace.comms.registry.declare(replace(workspace.thread("saved"), session_file=SavedSessionFile(str(journal))))
+    workspace.comms.registry.declare(replace(workspace.thread("saved"), session_file=RegistrySessionLocation.from_path(str(journal), workspace.comms.root)))
     page = transcripts.thread_transcript_page("saved", max_messages=1)
 
     async def loader(**bounds):

tests/textual/test_saved_history_coverage.py

@@ -5,6 +5,7 @@
 """
 
 
+from agent_comms.session_file import RegistrySessionLocation
 from support import journals
 from toad.widgets.agent_response import AgentResponse
 from toad.widgets.session_details import SessionDetails
@@ -17,7 +18,7 @@ async def saved_history(workspace, ui):
                                  "Canonical retained source owns coverage independently of painted resources.",
                                  id=f"a{number}", parent=f"a{number - 1}" if number else None)
         for number in range(8)))
-    workspace.comms.registry.declare(workspace.comms.registry.new_thread("saved", frozenset(), str(workspace.project), session_file=str(journal)))
+    workspace.comms.registry.declare(workspace.comms.registry.new_thread("saved", frozenset(), str(workspace.project), session_file=RegistrySessionLocation.from_path(journal, workspace.comms.root)))
     page = workspace.comms.transcripts.thread_transcript_page("saved")
     conversation = ui.conversation
     ui.attach_agent(ui.detached_agent())

tests/textual/test_sent_message_drawn_once.py

@@ -1,6 +1,7 @@
 """Textual-facing: a sent message whose start notice reaches the conversation after the saved history
 already holds it is drawn once, by the saved history, with no second live block under it."""
 
+from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.acp_extension import InputStartedUpdate, encode_updates
 
 from support import agents, journals
@@ -15,7 +16,7 @@ async def test_a_start_notice_after_the_saved_history_draws_the_message_once(wor
         journals.assistant_entry("An earlier answer.", id="earlier"),
         journals.user_entry(TOKEN, id="sent", parent="earlier", input_id=INPUT))
     workspace.comms.registry.declare(
-        workspace.comms.registry.new_thread("source", frozenset(), str(workspace.project), session_file=str(journal)))
+        workspace.comms.registry.new_thread("source", frozenset(), str(workspace.project), session_file=RegistrySessionLocation.from_path(journal, workspace.comms.root)))
     async with workspace.app(size=(110, 35)) as ui:
         view = ui.conversation
         agent = ui.detached_agent("source")