From 1c003bd3f67273354874a9a18440f5401f152fba Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 13:51:39 -0400 Subject: [PATCH 1/2] Create the bus when a root is adopted; an absent bus is a deleted bus #2784 (8ed65d3c6) made an absent bus.sqlite3 read as the declared empty schema, to fix a fresh root raising CANTOPEN. A missing file then meant two things, "never written" and "deleted" (IDEN-1), and test_outbound_transcript_source.py:245 (a deleted bus must not present as empty) went red on main. The owner of "this directory became a root" is RootRuntimeStore.adopt: it publishes the root's first record. It now creates the bus there, empty, through Bus.install. Bus.observing no longer falls back to an empty in-memory schema: an absent file fails to open. The fresh install's second Bus.install after wire(root) restated that creation and is deleted. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/bus.py | 8 +++----- src/agent_comms/installer/upgrade.py | 1 - src/agent_comms/root_runtime.py | 7 ++++++- 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 23d2d2f3b6..d377458e75 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -469,11 +469,9 @@ def table_schema(cls) -> type[BusSchema]: def observing( self, contention: StoreLockContention = UNBOUNDED_CONTENTION, ) -> Iterator[sqlite3.Connection]: - """Borrow one read-only snapshot, including for an archived bus; a bus before its - first write reads as its declared empty tables.""" - with self.committed_or_empty( - self.table_schema(), lambda: SQLiteFile.observing(self.path), contention - ) as db: + """Borrow one read-only snapshot, including for an archived bus. Root adoption + creates the bus, so an absent file is a deleted bus and fails to open.""" + with self._attached(lambda: SQLiteFile.observing(self.path), contention) as db: yield db def backup(self, destination: Path) -> None: diff --git a/src/agent_comms/installer/upgrade.py b/src/agent_comms/installer/upgrade.py index aa960a040e..6e35bb090d 100644 --- a/src/agent_comms/installer/upgrade.py +++ b/src/agent_comms/installer/upgrade.py @@ -1154,7 +1154,6 @@ def live(self, rehearsal: FreshRehearsal, report: InstallReport) -> CreatedRun: Checkout(self.checkout).prepare() self.root.mkdir(mode=0o700, parents=True, exist_ok=True) wire(self.root).registry.prepare_native_root() - Bus.install(Bus.path_in(self.root)) report.writer = RuntimeWriter(self.root, self.staged) backup, selected_route = self.backed_up(()) try: diff --git a/src/agent_comms/root_runtime.py b/src/agent_comms/root_runtime.py index 60309399f7..4e7f23177b 100644 --- a/src/agent_comms/root_runtime.py +++ b/src/agent_comms/root_runtime.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import Annotated +from .bus import Bus from .errors import RelationViolationError from .field_codec import PathText from .launch_runtime import LaunchRuntime @@ -59,8 +60,12 @@ def owner(self, creator: LaunchRuntime) -> RootRuntime: return self.adopt(creator) def adopt(self, runtime: LaunchRuntime) -> RootRuntime: - """A root without a runtime record belongs to ``runtime``; a recorded owner stays.""" + """A root without a runtime record belongs to ``runtime``; a recorded owner stays. + + Adoption is where a directory becomes a root, so its bus is created here, empty: + from then on an absent bus file means a deleted bus, never one not yet written.""" with self.locked(): if not self.path.exists(): + Bus.install(Bus.path_in(self.path.parent)) self._publish_unlocked(RootRuntime(runtime.prefix)) return self._read_unlocked() From ecd695e15ad9615739621b4ac81366f6c223f031 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 13:55:57 -0400 Subject: [PATCH 2/2] Pin the adopted root's empty bus; fix main's red at test_start.py:35 test_root_runtime asserts a root adopted by wire() reads an empty bus. test_start.py:35 was red on main: it compared the registry's typed BoundExternalSession to a str. It now compares against the declared location bound to the root, built from the same declaration it registers. Its Bus.install after wire() restated the creation adoption now owns and is deleted. Co-Authored-By: Claude Opus 5.5 --- tests/test_root_runtime.py | 1 + tests/test_start.py | 7 +++---- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_root_runtime.py b/tests/test_root_runtime.py index b34b9c17a9..c2ae09294b 100644 --- a/tests/test_root_runtime.py +++ b/tests/test_root_runtime.py @@ -58,6 +58,7 @@ def test_root_runtime_attachment_family(tmp_path): def test_comms_creates_a_private_root_and_refuses_foreign_stores(tmp_path): created = wire(tmp_path / "created") assert RootRuntimeStore.at(created.root).read().prefix == LaunchRuntime.running().prefix + assert Bus(Bus.path_in(created.root)).revision() == 0 root = tmp_path / "bus" other = LaunchRuntime(tmp_path / "other/bin/python") diff --git a/tests/test_start.py b/tests/test_start.py index 8926fd7840..a24970ab6d 100644 --- a/tests/test_start.py +++ b/tests/test_start.py @@ -1,5 +1,4 @@ -from agent_comms.bus import Bus import asyncio import sys @@ -21,8 +20,8 @@ async def test_real_stopped_owner_starts_through_shared_agent_tool(tmp_path, mon source = tmp_path / "saved.jsonl" source.touch() comms = wire(tmp_path / "wire") - Bus.install(Bus.path_in(comms.root)) - comms.registry.declare(Thread('worker', frozenset(), str(tmp_path), session_file=RegistrySessionLocation.from_path(str(source), comms.root))) + declared = RegistrySessionLocation.from_path(str(source), comms.root) + comms.registry.declare(Thread('worker', frozenset(), str(tmp_path), session_file=declared)) comms.registry.unregister("worker") try: result = await asyncio.to_thread(invoke_tool, comms, "comms_start", {"name": "worker"}, caller=NoThreadCaller()) @@ -32,6 +31,6 @@ async def test_real_stopped_owner_starts_through_shared_agent_tool(tmp_path, mon await asyncio.sleep(0.05) again = await asyncio.to_thread(invoke_tool, comms, "comms_start", {"name": "worker"}, caller=NoThreadCaller()) assert not again["launched"] and again["pid"] == result["pid"] - assert comms.registry.require("worker").session_file == str(source) + assert comms.registry.require("worker").session_file == declared.bound_to(comms.root) finally: await asyncio.to_thread(comms.owners.stop, "worker")