Create the bus when a root is adopted; an absent bus is a deleted bus

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:

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:

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

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

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