From d1229ad4233d050ce4942ef200e5f9a96e7cb6e8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 17:11:27 -0400 Subject: [PATCH 1/2] Release watch route locks between client operations --- src/agent_comms/cli_commands.py | 12 +++++++--- src/agent_comms/comms.py | 37 +++++++++++++++++++++++++++-- src/agent_comms/message_bus.py | 27 --------------------- tests/test_cli_error_family.py | 2 +- tests/test_message_bus_watch_seq.py | 6 ++--- tests/test_monitor_delivery.py | 2 +- tests/test_thread_watch.py | 2 +- tests/test_tool_output.py | 2 +- 8 files changed, 51 insertions(+), 39 deletions(-) diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index 18f257fef0..b874a49ba1 100644 --- a/src/agent_comms/cli_commands.py +++ b/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): diff --git a/src/agent_comms/comms.py b/src/agent_comms/comms.py index fa99dab351..a4e4a1b545 100644 --- a/src/agent_comms/comms.py +++ b/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, diff --git a/src/agent_comms/message_bus.py b/src/agent_comms/message_bus.py index 36f3412e78..d671b1f03d 100644 --- a/src/agent_comms/message_bus.py +++ b/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) diff --git a/tests/test_cli_error_family.py b/tests/test_cli_error_family.py index 6dae7abeb8..8e02d06ee2 100644 --- a/tests/test_cli_error_family.py +++ b/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()) diff --git a/tests/test_message_bus_watch_seq.py b/tests/test_message_bus_watch_seq.py index 1a5baa892e..28ade1ee42 100644 --- a/tests/test_message_bus_watch_seq.py +++ b/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() diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index a9ab9f40bb..da971d0a23 100644 --- a/tests/test_monitor_delivery.py +++ b/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: diff --git a/tests/test_thread_watch.py b/tests/test_thread_watch.py index 524f0e8133..49d344a3f0 100644 --- a/tests/test_thread_watch.py +++ b/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 diff --git a/tests/test_tool_output.py b/tests/test_tool_output.py index c499ca11a7..2942bb08a6 100644 --- a/tests/test_tool_output.py +++ b/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] From d0ea1698fec34bcf0142974715d9a7007661a8c0 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 17:12:56 -0400 Subject: [PATCH 2/2] Wait for operation-scoped route readers without a deadline --- src/agent_comms/active_route.py | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/src/agent_comms/active_route.py b/src/agent_comms/active_route.py index b1f32191cb..5c58caa4d6 100644 --- a/src/agent_comms/active_route.py +++ b/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: