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() 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")