Delete the second bus install after a root is claimed

tests/acp_error_payload_installed_journey.py

@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 
-from agent_comms.bus import Bus
 
 from agent_comms.session_file import RegistrySessionLocation
 from agent_comms.pi_vocabulary import OffThinkingLevel
@@ -32,7 +31,6 @@
 from agent_comms.comms import wire
 from agent_comms.fresh_private_session import create_fresh_private_session
 from agent_comms.field_codec import FieldCodec
-from agent_comms.input_disposition import InputDispositions
 from agent_comms.threads import Thread
 
 from native_context_budget_installed_journey import ReceiptSubscriber
@@ -126,7 +124,6 @@ def log_message(self, *_):
         "retry": {"enabled": False, "maxRetries": 0, "provider": {"maxRetries": 0}},
     }))
     comms = wire(root)
-    Bus.install(Bus.path_in(comms.root))
     fresh = create_fresh_private_session(evidence / "sessions", worktree=project)
     comms.registry.declare(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()

tests/bus_record_fixture.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 
 from agent_comms.audience_manifest import FrozenRecipient, freeze_audience
-from agent_comms.bus import Bus, PrivateRecord
+from agent_comms.bus import PrivateRecord
 from agent_comms.comms import Comms, wire
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.bus_publication import Delivered, KeyedPublication
@@ -26,7 +26,6 @@
 def prepare_native_comms(comms: Comms) -> Comms:
     """Validate native-root authority before installing data or opening a coordinator."""
     comms.registry.prepare_native_root()
-    Bus.install(Bus.path_in(comms.root))
     return comms
 
 

tests/c4_coordination_installed_ui.py

@@ -1,12 +1,10 @@
 """Installed snapshot values, original notification evidences and real sidebar rows."""
 
 
-from agent_comms.bus import Bus
 from conftest import publish_initial
 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
 import argparse
 import asyncio
 from importlib.resources import files
@@ -42,7 +40,6 @@ async def main():
     os.environ.update(XDG_CONFIG_HOME=str(root / 'config'), XDG_STATE_HOME=str(root / 'state'),
                       XDG_DATA_HOME=str(root / 'data'), AGENT_COMMS_ROOT=str(root / 'wire'))
     comms = wire(root / 'wire')
-    Bus.install(Bus.path_in(comms.root))
     owner = root.name
     for name in (owner, 'stopped', 'archived'):
         source = root / (name + '.jsonl')

tests/config_option_authority_installed_pilot.py

@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 
-from agent_comms.bus import Bus
 
 import argparse
 import asyncio
@@ -108,7 +107,6 @@ async def run(stage, package, models, ui, menu_contention=False):
     project = stage / 'project'
     project.mkdir()
     service = wire(stage / 'wire')
-    Bus.install(Bus.path_in(service.root))
     environment = dict(os.environ)
     environment.update(
         AGENT_COMMS_ROOT=str(service.root),

tests/goal_owner_fixture.py

@@ -5,14 +5,12 @@
 
 import pytest
 
-from agent_comms.bus import Bus
 from agent_comms.comms import wire
 
 
 def canonical_goal_wire(root):
     """Create an empty real checked source for goal/wait fixture operations."""
     comms = wire(root)
-    Bus.install(Bus.path_in(comms.root))
     return comms
 
 

tests/native_backend_fixture.py

@@ -1,7 +1,6 @@
 """Acquired original native fixture shared by pytest and installed drivers."""
 
 
-from agent_comms.bus import Bus
 from agent_comms.native_package import VerifiedNativePackage
 from agent_comms.native_arguments import NativeArguments
 
@@ -263,7 +262,7 @@ async def serve(reader, writer):
     (root / "storage.json").write_text(json.dumps(FieldCodec.encode(
         StorageRule(scratch_base=project.parent),
     )))
-    Bus.install(Bus.path_in(wire(root).root))
+    wire(root)
     environment = {
         NativeArguments.binary_key: NativeArguments.default_binary,
         "AGENT_COMMS_ROOT": str(root),

tests/native_request_wait_installed_journey.py

@@ -1,7 +1,6 @@
 """Real installed Toad/ACP/native waits with only the localhost provider controlled."""
 
 
-from agent_comms.bus import Bus
 from agent_comms.lifecycle_capability import Unresolved
 from agent_comms.pi_vocabulary import OffThinkingLevel
 import argparse
@@ -100,7 +99,6 @@ async def provider(reader, writer):
     (config / "settings.json").write_text(json.dumps({"compaction": {"enabled": False},
         "retry": {"enabled": False, "maxRetries": 0, "provider": {"maxRetries": 0}}}))
     service = wire(stage / "wire")
-    Bus.install(Bus.path_in(service.root))
     environment = dict(os.environ, AGENT_COMMS_ROOT=str(service.root),
         AGENT_COMMS_NATIVE_CONFIG_DIR=str(config), PI_CODING_AGENT_DIR=str(config),
         AGENT_COMMS_AGENT_MODELS="wait-local/fixture", AGENT_COMMS_AGENT_BIN="pi",

tests/ordinary_claims_live_journey.py

@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 
-from agent_comms.bus import Bus
 
 import argparse
 import asyncio
@@ -84,7 +83,6 @@ async def journey(arguments):
                       PI_CODING_AGENT_DIR=str(Path.home() / '.pi/agent'))
     env = dict(os.environ)
     comms = wire(root)
-    Bus.install(Bus.path_in(comms.root))
     name = 'scratch-sleeper'
     comms.registry.declare(replace(Thread(
         name, frozenset(), str(project), model=DeclaredModel(arguments.model), thinking_level=HighThinkingLevel),

tests/test_acp.py

@@ -25,7 +25,6 @@
     decode_updates,
 )
 from agent_comms.activity import ActivityState
-from agent_comms.bus import Bus
 from agent_comms.channel_targets import CommandTarget
 from agent_comms.channels import AnyChannel
 from agent_comms.child_process import ProcessIdentity
@@ -302,7 +301,6 @@ def test_real_stdio_roundtrip(self, native_stdio_root):
         (tmp_path / "proj").mkdir()
         root = tmp_path / "wire"
         comms = wire(root)
-        Bus.install(Bus.path_in(comms.root))
         env = dict(
             __import__("os").environ,
             AGENT_COMMS_ROOT=str(root),

tests/test_carry.py

@@ -1,6 +1,5 @@
 """The carry family: one chain of idempotent steps whose dry runs say what a bus still needs."""
 
-from agent_comms.bus import Bus
 
 import os
 import sqlite3
@@ -287,7 +286,6 @@ def test_restart_selection_uses_receiving_declaration_with_connected_cli_inbox(c
 
 
 def test_owner_request_failure_joins_the_entire_set_before_propagating(comms, stack):
-    Bus.install(Bus.path_in(comms.root))
     names = ["failed", "other", "last"]
     for name in names[1:]:
         comms.registry.declare(

tests/test_carry_session_locations.py

@@ -6,7 +6,6 @@
 
 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
@@ -88,9 +87,7 @@ def test_carry_converts_the_archived_catalogue_before_reading_its_roots(tmp_path
     session = old.root / "native-sessions" / "bus.jsonl"
     old.registry.declare(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())

tests/test_channel_coding_tools.py

@@ -1,7 +1,6 @@
 """Normal coding policies use existing claims; no provider or user work is touched."""
 
 
-from agent_comms.bus import Bus
 
 from agent_comms.pi_payloads import ReportedFlag, ReportedText
 import json
@@ -43,7 +42,6 @@ def test_create_claim_competes_before_file_exists_and_releases_without_creation(
         for name, created in [("a", 21.0), ("b", 22.0)]:
             thread = Thread(name, frozenset({'team'}), str(work), created_at=created)
             c.registry.declare(replace(thread, process_identity=ProcessIdentity.capture(os.getpid())))
-        Bus.install(Bus.path_in(c.root))
         resource = WritableFileClaim(Path("new/nested/file.py"))
         def acquire(name):
             with NativeBoundary(c.bus) as boundary:
@@ -80,7 +78,6 @@ def test_stopped_holder_reclaimed_in_copied_root(tmp_path, monkeypatch, existing
         original.registry.declare(replace(
             thread, process_identity=ProcessIdentity.capture(os.getpid())
         ))
-    Bus.install(Bus.path_in(original.root))
 
     def acquire(comms, name):
         with NativeBoundary(comms.bus) as boundary:

tests/test_check_root.py

@@ -1,7 +1,6 @@
 """Selected check history is small, non-executable and scoped to the command."""
 
 
-from agent_comms.bus import Bus
 import json
 from pathlib import Path
 import sys
@@ -22,7 +21,6 @@
 def source(scratch_root, durable_home):
     root = scratch_root / "source"
     c = wire(root)
-    Bus.install(Bus.path_in(c.root))
     for name in ("A", "B", "other"):
         c.registry.declare(Thread(name, frozenset({"comms"}), str(scratch_root)),
                            StoppedThreadStatus())

tests/test_command_families.py

@@ -1,4 +1,3 @@
-from agent_comms.bus import Bus
 from agent_comms.field_codec import FieldCodec
 
 """External contracts and adding commands without editing any consumer catalog."""
@@ -151,7 +150,6 @@ def test_normal_history_cli_keeps_migrated_content_and_source_identity(
 ):
     old, live = wire(tmp_path / "old"), wire(tmp_path / "live")
     for comms in (old, live):
-        Bus.install(Bus.path_in(comms.root))
         for name in ("alice", "bob"):
             comms.registry.declare(
                 Thread(name, frozenset({"team"}), str(tmp_path))

tests/test_commander_forks.py

@@ -1,7 +1,6 @@
 """Commander forks use registered identity and the original Todo assignment."""
 
 
-from agent_comms.bus import Bus
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import replace
 import json
@@ -51,7 +50,6 @@ def fork_bus(tmp_path):
     source.git(source.checkout, 'commit', '-qm', 'Declare shared assignment owner')
     source.git(source.checkout, 'push', 'origin', 'main')
     comms = wire(tmp_path / 'saved-bus')
-    Bus.install(Bus.path_in(comms.root))
     comms.channels.create_tag('comms')
     # Two order holders never share one worktree, so each unit has its own checkout.
     units = tuple((name, tmp_path / f'{name}-checkout') for name in ('worker', 'other', 'retained'))

tests/test_deleted_thread_observation.py

@@ -1,6 +1,5 @@
 """A stale open view is a registry-owned deletion result, not observer failure."""
 
-from agent_comms.bus import Bus
 import asyncio
 import pickle
 from dataclasses import replace
@@ -22,7 +21,6 @@
 def declare(comms, *names):
     for name in names:
         comms.registry.declare(Thread(name, frozenset(), str(comms.root.parent)))
-    Bus.install(Bus.path_in(comms.root))
 
 
 def observe(reading, comms):

tests/test_envelope_bus_integration.py

@@ -6,7 +6,6 @@
 from __future__ import annotations
 
 
-from agent_comms.bus import Bus
 
 from bus_history import read_history
 import json
@@ -48,7 +47,6 @@ def _participants(tmp_path: Path) -> tuple[Comms, Path]:
             replace(owner, process_identity=ProcessIdentity.capture(os.getpid())),
             RunningThreadStatus(),
         )
-    Bus.install(Bus.path_in(comms.root))
     return comms, worktree
 
 
@@ -237,7 +235,6 @@ def test_sigkill_across_claim_acquisition_recovers_one_row_or_none(
 from pathlib import Path
 sys.path.insert(0, sys.argv[2])
 sys.path.insert(0, str(Path(sys.argv[2]).parent / 'tests'))
-from agent_comms.bus import Bus
 from agent_comms.comms import wire
 from test_envelope_bus_integration import acquire_message
 """ + kill + """

tests/test_goal_input_review.py

@@ -1,6 +1,5 @@
 """Canonical inspection unblocks standby without fabricating native delivery."""
 
-from agent_comms.bus import Bus
 from agent_comms.lifecycle_capability import Unresolved
 from agent_comms.channel_targets import CommandTarget
 import os
@@ -36,7 +35,6 @@ def test_inspected_canonical_dependencies_allow_standby_without_replay(tmp_path,
         comms.registry.declare(
             Thread(name, frozenset(), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()))
         )
-    Bus.install(Bus.path_in(comms.root))
     comms.agents.begin_turn("parent", "parent-delegation-in-flight")
     goal = comms.goals.update_goal("worker", SetGoalAction(text="Delegate and wait"))
     messages = [

tests/test_goal_owner_pause.py

@@ -1,6 +1,5 @@
 """Owner pause remains attributable and cannot be undone by a model tool."""
 
-from agent_comms.bus import Bus
 
 import pytest
 
@@ -20,7 +19,6 @@
 
 def test_owner_pause_survives_reopen_and_explains_stale_model_report(tmp_path, monkeypatch):
     comms = wire(tmp_path)
-    Bus.install(Bus.path_in(comms.root))
     comms.registry.declare(Thread(name='worker', tags=frozenset(), worktree=str(tmp_path)))
     goal = comms.goals.update_goal("worker", SetGoalAction(text="Read fifty files"))
     comms.goals.update_goal(
@@ -55,7 +53,6 @@ def test_owner_pause_survives_reopen_and_explains_stale_model_report(tmp_path, m
 
 def test_model_cannot_pause(tmp_path, monkeypatch):
     comms = wire(tmp_path)
-    Bus.install(Bus.path_in(comms.root))
     comms.registry.declare(Thread(name='worker', tags=frozenset(), worktree=str(tmp_path)))
     goal = comms.goals.update_goal("worker", SetGoalAction(text="Read fifty files"))
     monkeypatch.setenv("AGENT_COMMS_THREAD", "worker")

tests/test_held_sends.py

@@ -1,7 +1,6 @@
 """Real send entrypoints wait through a private bus hold and publish once."""
 
 
-from agent_comms.bus import Bus
 import asyncio
 import os
 import sqlite3
@@ -55,7 +54,6 @@ def test_refusal_writer_keeps_registry_permission_until_its_sql_write_finishes(p
 @pytest.fixture
 def held_bus(tmp_path):
     comms = wire(tmp_path)
-    Bus.install(Bus.path_in(comms.root))
     for name in ('sender', 'recipient'):
         comms.registry.declare(Thread(name, frozenset(), str(tmp_path)))
     assert comms.bus.inbox('recipient') == []

tests/test_historical_page_scope.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from agent_comms.bus import Bus, PublicRecord
+from agent_comms.bus import PublicRecord
 from agent_comms.channel_targets import CommandTarget
 from agent_comms.comms import wire
 from agent_comms.mentions import ThreadMention
@@ -22,7 +22,6 @@ def history(tmp_path):
     for created, name in enumerate(("bob", "carol"), start=11):
         old.registry.declare(Thread(name, frozenset({'ops'}), str(old.root), created_at=created))
     old.registry.rename("alice-old", "alice")
-    Bus.install(Bus.path_in(old.root))
     rows = []
     for seq in range(1, 241):
         message = Message(
@@ -49,7 +48,6 @@ def history(tmp_path):
         )
         rows.append(Message.from_message(published))
     live = wire(tmp_path / "live")
-    Bus.install(Bus.path_in(live.root))
     # Deliberately different membership: source scope must not use live tags.
     live.registry.declare(Thread('alice', frozenset({'ops'}), str(live.root), created_at=500))
     live.registry.declare(Thread('alice-old', frozenset({'api'}), str(live.root), created_at=501))

tests/test_historical_views.py

@@ -53,7 +53,6 @@ def setup(root, count=5):
     comms = wire(root)
     comms.registry.declare(Thread('alice', frozenset({'team'}), str(root), created_at=10.0))
     comms.registry.declare(Thread('bob', frozenset({'team'}), str(root), created_at=11.0))
-    Bus.install(Bus.path_in(comms.root))
     viewer = comms.messaging.user_identity(str(root)).name
     for i in range(count):
         comms.messaging.send("alice", "#team", f"{root.name} channel {i}").message.message_id