Keep route readers scoped to individual operations

src/agent_comms/active_route.py

@@ -18,7 +18,6 @@
 from typing import TYPE_CHECKING, Annotated, ClassVar
 
 from .child_process import Platform
-from .coordination_errors import CoordinationError
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
 from .field_codec import FieldCodec, MalformedRecord, PathText
@@ -263,20 +262,9 @@ def guard_default_route_write(
         os.close(directory)
 
 
-class RouteReadersHold(CoordinationError):
-    """In-flight client calls exceeded the route publication's contention budget."""
-
-
 def acquire_exclusive_route_lock(directory: int) -> None:
-    """Wait up to ten seconds for in-flight calls, then name the remaining holders."""
-    try:
-        StoreLockContention(10).acquire(directory, Platform.current(), mode=ExclusiveLock())
-    except BlockingIOError as error:
-        detail = '; '.join(holder.describe() for holder in Platform.current().lock_holders(directory))
-        raise RouteReadersHold(
-            f'default-route readers hold the route: '
-            f'{detail or "unknown (lock changed during inspection)"}; close them and rerun'
-        ) from error
+    """Wait for in-flight operations to finish before publishing the route."""
+    UNBOUNDED_CONTENTION.acquire(directory, Platform.current(), mode=ExclusiveLock())
 
 
 def publish_active_route(route: ActiveRoute, path: Path | None = None) -> None:

src/agent_comms/cli_commands.py

@@ -1397,7 +1397,7 @@ class WatchStream(CliResult):
     watch: WatchCliCommand
 
     def messages(self) -> AsyncIterator[Message]:
-        return self.comms.bus.watch(self.watch.thread, acknowledgment=self.watch.acknowledgment,
+        return self.comms.watch(self.watch.thread, acknowledgment=self.watch.acknowledgment,
                                     after=self.watch.after)
 
     def write(self, output: TextIO) -> int:
@@ -1408,8 +1408,9 @@ async def stream() -> None:
                     emit(ArrivedMessage(message=message).model_dump_json())
 
         async def watch() -> None:
-            thread = self.comms.registry.require(self.watch.thread)
-            source = thread.runtime.history_source(self.comms.root, thread.incarnation)
+            with Comms.client(self.comms.route) as comms:
+                thread = comms.registry.require(self.watch.thread)
+                source = thread.runtime.history_source(comms.root, thread.incarnation)
             async with asyncio.TaskGroup() as tasks:
                 tasks.create_task(stream())
                 tasks.create_task(source.watch_requests(
@@ -1444,6 +1445,11 @@ class WatchCliCommand(ThreadArgument, CliCommand[WatchStream], declared_name="wa
     def apply(self, ctx: Comms) -> WatchStream:
         return WatchStream(ctx, self)
 
+    def execute(self, route: CommsRoute, output: TextIO) -> int:
+        with Comms.client(route) as comms:
+            stream = self.apply(comms)
+        return stream.write(output)
+
 
 @dataclass(frozen=True, kw_only=True)
 class AckCliCommand(ThreadArgument, CliCommand):

src/agent_comms/comms.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterator
-from contextlib import AbstractContextManager, contextmanager
+from collections.abc import AsyncIterator, Iterator
+from contextlib import AbstractContextManager, aclosing, contextmanager
+from functools import partial
 from pathlib import Path
 from typing import ClassVar
 
@@ -13,21 +14,27 @@
 from .channel_management import ChannelManagement
 from .collaboration_ledger import CollaborationLedger
 from .context_store import ContextStore
+from .coordinator import Coordination
+from .errors import UnregisteredThreadError
 from .goal_management import Goals
 from .history_views import HistoryViews
 from .launch_runtime import LaunchRuntime
 from .locked_store import RegistryFile
 from .maintenance_barrier import MaintenanceBarrier
 from .message_bus import MessageBus
+from .message_reference import AfterMessageStart, FirstUnreadStart, InboxStart
+from .messages import Message
 from .messaging import Messaging
 from .owner_lifecycle import OwnerLifecycle
 from .publisher import OwnerPublisher, Publisher
 from .registration import Registration
 from .relationships import ThreadRelationships
+from .read_basis import WatchAcknowledgment
 from .root_runtime import RootRuntime
 from .thread_configuration import ThreadConfiguration
 from .thread_management import ThreadManagement
 from .transcripts import Transcripts
+from .watched import FileSource
 
 
 class Comms:
@@ -85,6 +92,32 @@ def client(cls, route: CommsRoute) -> Iterator[Comms]:
         with route.hold_client_route():
             yield cls(route)
 
+    async def watch(
+        self, name: str, *, acknowledgment: type[WatchAcknowledgment],
+        after: InboxStart = FirstUnreadStart(),
+    ) -> AsyncIterator[Message]:
+        """Wait unlocked; each inbox read and acknowledgment owns one client call."""
+        with Comms.client(self.route) as comms:
+            reader = comms.registry.require(name).incarnation
+        maintenance = MaintenanceBarrier(self.registry.store.path)
+        async with aclosing(self.bus.bus.watch(FileSource(maintenance.state_path))) as changes:
+            async for _ in changes:
+                while True:
+                    with Comms.client(self.route) as comms:
+                        snapshot = comms.registry.snapshot()
+                        batch = await Coordination.run_worker(partial(
+                            comms.bus.inbox_at, name, snapshot, after=after,
+                        ))
+                        if not reader.current(comms.registry.snapshot()):
+                            raise UnregisteredThreadError("Watched thread was removed or replaced")
+                    for message in batch.page.messages:
+                        yield message
+                        with Comms.client(self.route) as comms:
+                            acknowledgment.taken(comms.bus.reads, batch.displayed.through(message.seq))
+                        after = AfterMessageStart(message.reference)
+                    if not batch.remaining:
+                        break
+
 
 def wire(
     root: Path | str | None = None,

src/agent_comms/message_bus.py

@@ -7,7 +7,6 @@
 import inspect
 from abc import ABC, abstractmethod
 from collections.abc import (
-    AsyncIterator,
     Awaitable,
     Callable,
     Iterable,
@@ -24,12 +23,10 @@
 from agent_comms.channel_targets import NamedTarget
 from agent_comms.channels import AnyChannel
 from agent_comms.coordinator import Coordination
-from agent_comms.errors import UnregisteredThreadError
 from agent_comms.historical_views import HistoryArchive
 from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceHeldError
 from agent_comms.message_page import InboxPage, MessagePage, MessagePageRequest, PageBounds
 from agent_comms.message_reference import (
-    AfterMessageStart,
     FirstUnreadStart,
     NewestUnreadStart,
     InboxStart,
@@ -45,7 +42,6 @@
     DisplayViewer,
     ListedDeliveries,
     NamedViewer,
-    WatchAcknowledgment,
 )
 from agent_comms.routing import DeliveryScope
 from agent_comms.store_files import UNBOUNDED_CONTENTION, StoreLockContention
@@ -151,29 +147,6 @@ def inbox_at(
         ).displayed(self.reads, document, delivery.actor, snapshot, snapshot, self.bus)
         return InboxPage(result.page, result.displayed, total - len(result.page.messages))
 
-    async def watch(
-        self, name: str, *, acknowledgment: type[WatchAcknowledgment],
-        after: InboxStart = FirstUnreadStart(),
-    ) -> AsyncIterator[Message]:
-        from .watched import FileSource
-
-        reader = self._registry.require(name).incarnation
-        maintenance = MaintenanceBarrier(self._registry.store.path)
-        async with aclosing(self.bus.watch(FileSource(maintenance.state_path))) as changes:
-            async for _ in changes:
-                maintenance.read().require_open()
-                snapshot = self._registry.snapshot()
-                while True:
-                    batch = await asyncio.to_thread(self.inbox_at, name, snapshot, after=after)
-                    if not reader.current(self._registry.snapshot()):
-                        raise UnregisteredThreadError("Watched thread was removed or replaced")
-                    for message in batch.page.messages:
-                        yield message
-                        acknowledgment.taken(self.reads, batch.displayed.through(message.seq))
-                        after = AfterMessageStart(message.reference)
-                    if not batch.remaining:
-                        break
-
     async def when_open[Result](self, operation: Callable[[], Result | Awaitable[Result]]) -> Result:
         maintenance = MaintenanceBarrier(self._registry.store.path)
 

tests/test_cli_error_family.py

@@ -180,7 +180,7 @@ async def test_watched_record_crosses_the_watch_line_as_its_envelope(root):
     for name in ("a", "b"):
         comms.registry.declare(Thread(name=name, tags=frozenset(), worktree='/wt'))
     sent = comms.messaging.send("a", "b", "crosses").message
-    async with aclosing(comms.bus.watch("b", acknowledgment=ExplicitWatchAcknowledgment)) as watched:
+    async with aclosing(comms.watch("b", acknowledgment=ExplicitWatchAcknowledgment)) as watched:
         async with asyncio.timeout(5):
             record = await anext(watched)
     line = WATCH_LINE.validate_json(ArrivedMessage(message=record).model_dump_json())

tests/test_message_bus_watch_seq.py

@@ -6,7 +6,7 @@
 import pytest
 
 from agent_comms.locked_store import RegistryFile
-from agent_comms.message_bus import MessageBus
+from agent_comms.comms import wire
 from agent_comms.publisher import Publisher
 from agent_comms.read_basis import ExplicitWatchAcknowledgment
 from agent_comms.registration import Registration
@@ -23,7 +23,7 @@ async def test_real_message_bus_watch_recipient_suffix_and_rollback(bus):
             name, frozenset(), str(root))).incarnation.require_recorded()
         for name in ("sender", "target", "idle")
     }
-    messages = MessageBus(root, registry, Publisher)
+    messages = wire(root, Publisher)
     primers = {
         name: append_reply(bus, people["sender"], people[name], f"prime {name}")
         for name in ("target", "idle")
@@ -65,7 +65,7 @@ def rolled_back(seq, db):
             assert bus.revision() == before
             done, _ = await asyncio.wait((target, idle), timeout=0.2)
             assert not done, "unaddressed or rolled-back append yielded a message"
-            print("real MessageBus.watch: exact new recipient suffix; unaddressed and rollback silence")
+            print("real Comms.watch: exact new recipient suffix; unaddressed and rollback silence")
         finally:
             for task in tasks:
                 task.cancel()

tests/test_monitor_delivery.py

@@ -206,7 +206,7 @@ async def test_explicit_watch_does_not_mark_streamed_messages_read(wired):
     from contextlib import aclosing
     import asyncio
     first=wired.messaging.send('PR111','fixer','first').message
-    async with aclosing(wired.bus.watch('fixer', acknowledgment=ExplicitWatchAcknowledgment)) as stream:
+    async with aclosing(wired.watch('fixer', acknowledgment=ExplicitWatchAcknowledgment)) as stream:
         assert (await anext(stream)).reference==first.reference
         pending=asyncio.create_task(anext(stream))
         try:

tests/test_thread_watch.py

@@ -8,7 +8,7 @@
 async def test_watch_emits_startup_unread_then_arrival_and_leaves_untaken_unread(wired):
     first = wired.messaging.send("PR111", "fixer", "already unread").message
 
-    async with aclosing(wired.bus.watch("fixer", acknowledgment=OnReadWatchAcknowledgment)) as watcher:
+    async with aclosing(wired.watch("fixer", acknowledgment=OnReadWatchAcknowledgment)) as watcher:
         initial = await asyncio.wait_for(anext(watcher), timeout=3)
         assert initial.message_id == first.message_id
 

tests/test_tool_output.py

@@ -284,7 +284,7 @@ def test_inbox_page_bound_and_no_ack_keep_the_next_page_unread(inbox_comms):
 async def test_watch_drains_multiple_pages_without_another_file_event(inbox_comms):
     comms = inbox_comms
     expected = [comms.messaging.send("a", "b", str(index)).message for index in range(201)]
-    watcher = comms.bus.watch("b", acknowledgment=ExplicitWatchAcknowledgment)
+    watcher = comms.watch("b", acknowledgment=ExplicitWatchAcknowledgment)
     try:
         async with asyncio.timeout(5):
             received = [await anext(watcher) for _ in expected]