From d6e7cd6806b9870e0f7312f598654e6f6e5de635 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 13:29:52 -0400 Subject: [PATCH 1/4] P233: keep the Claude bridge for its session and replace Monitor readers --- pyproject.toml | 2 + src/agent_comms/child_process.py | 65 +++- src/agent_comms/cli_bridges/claude.py | 27 +- .../cli_bridges/claude_protocol.py | 4 +- src/agent_comms/cli_bridges/monitor.py | 277 ++++++++++++++++++ src/agent_comms/cli_bridges/shared.py | 166 +++++++---- src/agent_comms/cli_protocol.py | 22 +- src/agent_comms/skills/comms/SKILL.md | 2 +- src/agent_comms_hooks/session.py | 95 ++++++ src/agent_comms_hooks/watch.py | 65 +++- tests/test_monitor_delivery.py | 22 -- tests/test_session_bridge.py | 180 ++++++++++++ tests/test_watch_guard.py | 16 +- 13 files changed, 833 insertions(+), 110 deletions(-) create mode 100644 src/agent_comms/cli_bridges/monitor.py create mode 100644 src/agent_comms_hooks/session.py create mode 100644 tests/test_session_bridge.py diff --git a/pyproject.toml b/pyproject.toml index 19f22ffcc6..29eeb94ca7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,6 +60,8 @@ agent-comms-commander = "agent_comms.commander.cli:main" agent-comms-watch-guard = "agent_comms_hooks.watch:main" agent-comms-commander-skill = "agent_comms_hooks.skill:main" agent-comms-owner-directive = "agent_comms_hooks.directive:main" +agent-comms-claude-session-started = "agent_comms_hooks.session:started" +agent-comms-claude-session-ended = "agent_comms_hooks.session:ended" agent-comms-connector = "agent_comms.connector:main" opencode-comms = "agent_comms.cli_bridges.opencode_launch:main" diff --git a/src/agent_comms/child_process.py b/src/agent_comms/child_process.py index e52bec8117..9ab938ebe4 100644 --- a/src/agent_comms/child_process.py +++ b/src/agent_comms/child_process.py @@ -263,6 +263,20 @@ def owns_turn(self, turn: ActiveTurn) -> bool: def signal(self, signum: int) -> None: Platform.current().send(self, signum) + def require_ancestor_of(self, pid: int) -> None: + """A supplied process is authority only for its actual descendant.""" + platform = Platform.current() + platform.require(self) + current = platform.identity(pid) + while current.pid > 1: + if current == self: + return + current = platform.parent_identity(current.pid) + raise IdentityMismatchError('The declared process is not an ancestor of this hook') + + async def wait_exited(self) -> None: + await Platform.current().wait_exited(self) + def require_release_by(self, process: ProcessIdentity) -> None: if self != process: raise RelationViolationError("Only the registered owner may release itself.") @@ -428,6 +442,12 @@ def force_group(self, identity: ProcessIdentity) -> None: @abstractmethod def identity(self, pid: int) -> ProcessIdentity: ... + def parent_identity(self, pid: int) -> ProcessIdentity: + raise NotImplementedError('This platform does not expose exact parent identities') + + async def wait_exited(self, identity: ProcessIdentity) -> None: + raise NotImplementedError('This platform does not expose process exit observation') + def matches(self, identity: ProcessIdentity) -> bool: try: return self.identity(identity.pid) == identity @@ -494,9 +514,16 @@ def send(self, identity: ProcessIdentity, signum: int) -> None: os.kill(identity.pid, signum) +@dataclass(frozen=True) +class LinuxProcessRecord: + identity: ProcessIdentity + group: int + parent: int + + class LinuxPlatform(NamespaceContainment, PidfdHandles, PosixPlatform): @staticmethod - def _stat(pid: int) -> tuple[ProcessIdentity, int]: + def _stat(pid: int) -> LinuxProcessRecord: try: # comm can contain whitespace and ')'; fields after its final ')' # start at field 3. Kernel starttime is field 22, pgrp is field 5. @@ -506,19 +533,45 @@ def _stat(pid: int) -> tuple[ProcessIdentity, int]: if fields[0] in {"Z", "X"}: raise ProcessLookupError(pid) group = int(fields[2]) - return ProcessIdentity.in_group(pid, int(fields[19]), group), group + return LinuxProcessRecord( + ProcessIdentity.in_group(pid, int(fields[19]), group), group, int(fields[1])) def identity(self, pid: int) -> ProcessIdentity: - return self._stat(pid)[0] + return self._stat(pid).identity + + def parent_identity(self, pid: int) -> ProcessIdentity: + return self.identity(self._stat(pid).parent) + + async def wait_exited(self, identity: ProcessIdentity) -> None: + try: + descriptor = self.open_pidfd(identity.pid) + except ProcessLookupError: + return + try: + self.require(identity) + loop = asyncio.get_running_loop() + exited = loop.create_future() + def observe() -> None: + if not exited.done(): + exited.set_result(None) + loop.add_reader(descriptor, observe) + try: + await exited + finally: + loop.remove_reader(descriptor) + except ProcessLookupError: + return + finally: + os.close(descriptor) def members_of_group(self, pgid: int) -> tuple[ProcessIdentity, ...]: members = [] for entry in Path("/proc").iterdir(): if entry.name.isdecimal(): with suppress(ProcessLookupError): - identity, group = self._stat(int(entry.name)) - if group == pgid: - members.append(identity) + record = self._stat(int(entry.name)) + if record.group == pgid: + members.append(record.identity) return tuple(members) def send(self, identity: ProcessIdentity, signum: int) -> None: diff --git a/src/agent_comms/cli_bridges/claude.py b/src/agent_comms/cli_bridges/claude.py index 4e8cd46d8e..909b109dd7 100755 --- a/src/agent_comms/cli_bridges/claude.py +++ b/src/agent_comms/cli_bridges/claude.py @@ -12,9 +12,10 @@ from agent_comms.coordination_database import CoordinationStore from agent_comms.private_path import FileRevision from agent_comms.wire_watch import WatchSource, WireWatch +from agent_comms_hooks.session import SessionHook from .claude_protocol import CLAUDE_RECORD, ClaudeRecord, ClaudeSession, ClaudeMonitorFrame from .history import ClaudeHistory, JsonLinesFile, OfflineHistory, OnlineHistory -from .shared import Binding, LineMonitorBridge, Options, run_bridge +from .shared import Binding, Enrol, LineMonitorBridge, Options, run_bridge from agent_comms.runtime_updates import RuntimeUpdate, SessionFacts @@ -104,9 +105,13 @@ def history_start(self, stream: BinaryIO, boundary: int) -> int: async def transcript(self) -> AsyncIterator[RuntimeUpdate]: session = self.start_state() - with self.source.open('rb') as stream: - original = FileRevision.from_file(stream) - async with aclosing(WireWatch.observations(WatchSource.modified_file(self.source))) as changes: + source = self.options.operation.transcript_source() + async with aclosing(WireWatch.observations(WatchSource.modified_file(source))) as changes: + await anext(changes) + while not source.exists(): + await anext(changes) + with source.open('rb') as stream: + original = FileRevision.from_file(stream) for update in self.decoded(stream, self.history_start(stream, original.size), original.size, session): yield update # Only the current status is published from the recent window. @@ -114,7 +119,7 @@ async def transcript(self) -> AsyncIterator[RuntimeUpdate]: yield update async for _ in changes: opened = FileRevision.from_file(stream) - current = FileRevision.capture(self.source) + current = FileRevision.capture(source) if (current is None or current.identity != original.identity or min(opened.size, current.size) < stream.tell()): raise ValueError('Claude transcript was replaced or truncated') @@ -122,5 +127,17 @@ async def transcript(self) -> AsyncIterator[RuntimeUpdate]: yield update +class Connect(Enrol): + @classmethod + def serves(cls, bridge: type) -> bool: + return issubclass(bridge, ClaudeBridge) + + async def run(self, bridge: LineMonitorBridge) -> None: + await super().run(bridge) + SessionHook.current( + bridge.binding.thread, str(RunningClaude(bridge.binding.thread).source) + ).start() + + def main() -> None: run_bridge(ClaudeBridge, Options.parse(__doc__, Path.home() / '.local/state/claude-comms-inbox', ClaudeBridge)) diff --git a/src/agent_comms/cli_bridges/claude_protocol.py b/src/agent_comms/cli_bridges/claude_protocol.py index 700621760a..295cf90074 100644 --- a/src/agent_comms/cli_bridges/claude_protocol.py +++ b/src/agent_comms/cli_bridges/claude_protocol.py @@ -43,8 +43,8 @@ from .shared import ( MONITOR_FRAME, ForeignMonitorFrame, - IncomingFrame, MonitorFrame, + MonitorIncomingFrame, MonitorObservation, MonitorPart, MonitorProbe, @@ -52,7 +52,7 @@ ) -class ClaudeMonitorFrame(IncomingFrame): +class ClaudeMonitorFrame(MonitorIncomingFrame): """Claude Monitor caps lines at 500 and joined batches at 3000 UTF-16 units.""" EVENT_CHARS = 500 BATCH_CHARS = 3000 diff --git a/src/agent_comms/cli_bridges/monitor.py b/src/agent_comms/cli_bridges/monitor.py new file mode 100644 index 0000000000..a31591f63a --- /dev/null +++ b/src/agent_comms/cli_bridges/monitor.py @@ -0,0 +1,277 @@ +"""Session-owned endpoint and replaceable Monitor readers.""" + +from __future__ import annotations + +import asyncio +import secrets +import socket +import sys +from abc import abstractmethod +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +from pydantic import TypeAdapter, ValidationError + +from agent_comms.child_process import ProcessIdentity +from agent_comms.cli_protocol import Outcome, TaggedWire +from agent_comms.messages import Message +from agent_comms_hooks.watch import WatchClaim + +if TYPE_CHECKING: + from .shared import LineMonitorBridge, MonitorProbe + + +@dataclass +class MonitorPeer: + reader: asyncio.StreamReader + writer: asyncio.StreamWriter + + async def write(self, line: str) -> None: + self.writer.write((line + "\n").encode()) + await self.writer.drain() + + async def close(self) -> None: + self.writer.close() + await self.writer.wait_closed() + + +class ReaderState: + async def prepare_delivery(self, bridge: LineMonitorBridge) -> AttachedReader: + async with bridge.reader_changed: + await bridge.reader_changed.wait_for(lambda: bridge.reader is not self) + return await bridge.reader.prepare_delivery(bridge) + + async def deliver(self, bridge: LineMonitorBridge, message: Message) -> type[Outcome]: + reader = await self.prepare_delivery(bridge) + return await reader.deliver(bridge, message) + + async def confirm_probe(self, bridge: LineMonitorBridge, probe: MonitorProbe) -> None: + """Only an unproven attachment consumes a probe.""" + + async def part_observed(self, message_id: str, number: int) -> None: + """Only an attached reader has a flow-control window.""" + + async def attach(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: + return SessionRefused.model_validate( + dict(error="This session already has a Monitor reader") + ) + + +class NoReader(ReaderState): + async def attach(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: + bridge = service.bridge + attachment = bridge.new_attachment(peer, secrets.token_hex(24)) + async with bridge.reader_changed: + bridge.reader = UnreadyReader(attachment) + try: + await peer.write(SessionReady().model_dump_json()) + await peer.write(attachment.probe.model_dump_json()) + await peer.reader.read() + return PeerGone() + finally: + async with bridge.reader_changed: + bridge.reader = NoReader() + bridge.reader_changed.notify_all() + async with attachment.window_changed: + attachment.window_changed.notify_all() + print( + "Monitor detached; delivery paused, bridge remains connected.", + file=sys.stderr, + flush=True, + ) + + +@dataclass +class MonitorAttachment: + peer: MonitorPeer + probe: MonitorProbe + pending_parts: dict[tuple[str, int], int] = field(default_factory=dict) + window_changed: asyncio.Condition = field(default_factory=asyncio.Condition) + + async def emit(self, message_id: str, number: int, line: str, window_bytes: int) -> None: + size = len((line + "\n").encode("utf-16-le")) + async with self.window_changed: + await self.window_changed.wait_for( + lambda: self.peer.writer.is_closing() + or not self.pending_parts + or sum(self.pending_parts.values()) + size <= window_bytes + ) + self.pending_parts[message_id, number] = size + await self.peer.write(line) + + async def observed(self, message_id: str, number: int) -> None: + async with self.window_changed: + self.pending_parts.pop((message_id, number), None) + self.window_changed.notify_all() + + +@dataclass +class UnreadyReader(ReaderState): + attachment: MonitorAttachment + + async def confirm_probe(self, bridge: LineMonitorBridge, probe: MonitorProbe) -> None: + if probe == self.attachment.probe: + async with bridge.reader_changed: + bridge.reader = AttachedReader(self.attachment) + bridge.reader_changed.notify_all() + print("Fresh Monitor probe observed; delivery enabled.", file=sys.stderr, flush=True) + + +@dataclass +class AttachedReader(UnreadyReader): + async def prepare_delivery(self, bridge: LineMonitorBridge) -> AttachedReader: + return self + + async def confirm_probe(self, bridge: LineMonitorBridge, probe: MonitorProbe) -> None: + """Overlapping observations cannot prove a different attachment.""" + + async def part_observed(self, message_id: str, number: int) -> None: + await self.attachment.observed(message_id, number) + + async def deliver(self, bridge: LineMonitorBridge, message: Message) -> type[Outcome]: + return await bridge.deliver_to(self.attachment, message) + + +class SessionReply(TaggedWire): + __registry_config__ = TaggedWire.family_registry() + + def require(self) -> None: + """Successful replies carry no refusal.""" + + async def publish(self, peer: MonitorPeer) -> None: + await peer.write(self.model_dump_json()) + + def end_session(self, service: MonitorService) -> None: + """A refusal cannot stop the service.""" + + +class SessionReady(SessionReply): + wire_tag = "ready" + + def end_session(self, service: MonitorService) -> None: + asyncio.get_running_loop().call_soon(service.stop) + + +class SessionRefused(SessionReply): + wire_tag = "refused" + error: str + + def require(self) -> None: + raise ValueError(self.error) + + +class PeerGone(SessionReply): + wire_tag = "gone" + + async def publish(self, peer: MonitorPeer) -> None: + """The stream already ended; there is no reply to send.""" + + +REPLIES = TypeAdapter(SessionReply.declared_union()) + + +class SessionRequest(TaggedWire): + __registry_config__ = TaggedWire.family_registry() + + def exchange(self, claim: WatchClaim) -> None: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as peer: + peer.settimeout(5) + peer.connect(str(claim.socket_path)) + peer.sendall((self.model_dump_json() + "\n").encode()) + with peer.makefile("rb") as stream: + REPLIES.validate_json(stream.readline(4096)).require() + + @abstractmethod + async def handle(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: ... + + +class AttachMonitor(SessionRequest): + wire_tag = "attach" + + async def handle(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: + return await service.bridge.reader.attach(service, peer) + + +class VerifySession(SessionRequest): + wire_tag = "verify" + owner: ProcessIdentity + + async def handle(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: + if self.owner != service.owner: + return SessionRefused.model_validate( + dict(error="This bridge belongs to a different Claude process incarnation") + ) + return SessionReady() + + +class EndSession(VerifySession): + wire_tag = "end" + + async def handle(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: + reply = await super().handle(service, peer) + reply.end_session(service) + return reply + + +REQUESTS = TypeAdapter(SessionRequest.declared_union()) + + +@dataclass +class MonitorService: + bridge: LineMonitorBridge + owner: ProcessIdentity + stop: Callable[[], None] + + async def accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + peer = MonitorPeer(reader, writer) + try: + try: + reply: SessionReply + try: + request = REQUESTS.validate_json(await reader.readline()) + except ValidationError as error: + reply = SessionRefused.model_validate(dict(error=str(error))) + else: + reply = await request.handle(self, peer) + await reply.publish(peer) + finally: + await peer.close() + except (BrokenPipeError, ConnectionResetError) as error: + await self.bridge.report_failure(f"Monitor connection ended: {error}") + + async def run(self) -> None: + path = WatchClaim(self.bridge.binding.thread).socket_path + path.unlink(missing_ok=True) + try: + async with asyncio.TaskGroup() as clients: + server = await asyncio.start_unix_server( + lambda reader, writer: clients.create_task(self.accept(reader, writer)), + str(path), + ) + async with server: + print( + f"Session bridge ready for Claude PID {self.owner.pid}.", + file=sys.stderr, + flush=True, + ) + try: + await self.owner.wait_exited() + finally: + self.stop() + finally: + path.unlink(missing_ok=True) + + +async def read_monitor(claim: WatchClaim) -> None: + with claim.reader_held(): + reader, writer = await asyncio.open_unix_connection(str(claim.socket_path)) + peer = MonitorPeer(reader, writer) + try: + await peer.write(AttachMonitor().model_dump_json()) + REPLIES.validate_json(await reader.readline()).require() + async for line in reader: + sys.stdout.write(line.decode()) + sys.stdout.flush() + finally: + await peer.close() diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py index 1494a24749..6996ac7706 100644 --- a/src/agent_comms/cli_bridges/shared.py +++ b/src/agent_comms/cli_bridges/shared.py @@ -6,7 +6,6 @@ import asyncio import json import os -import secrets import signal import sys import traceback @@ -49,6 +48,7 @@ from agent_comms.send_result import SendResult from agent_comms.typed_table import Column, TypedTable from agent_comms_hooks.watch import WatchClaim, WatchRecord +from .monitor import MonitorAttachment, MonitorPeer, MonitorService, NoReader, ReaderState, read_monitor class DeclaredOptions: @@ -231,6 +231,7 @@ def message(self) -> messages.Message: return messages.Message.from_wire(json.loads(self.payload)) async def finish(self, bridge: CliBridge, outcome: type[Outcome]) -> None: + outcome = self.outcome.followed_by(outcome) with bridge.db: replace(self, outcome=outcome).upsert(bridge.db) await outcome.acknowledge(bridge, self.message) @@ -249,6 +250,9 @@ def binding(self, arguments: argparse.Namespace) -> Binding | None: @abstractmethod async def execute(self, bridge: OperationBridge) -> None: ... + def transcript_source(self) -> Path: + raise ValueError('This command does not own a live transcript subscription') + @classmethod def serves(cls, bridge: type[CliBridge]) -> bool: """Whether this command runs on that bridge; every bridge sends.""" @@ -417,21 +421,7 @@ def start(self, bridge: OperationBridge, tasks: asyncio.TaskGroup) -> None: tasks.create_task(bridge.watch()) async def execute(self, bridge: OperationBridge) -> None: - # The claim records how this watch was started, so the session's hook can - # name the re-arm command once the lease lapses. - record = WatchRecord(bridge.binding.bus, bridge.binding.root, - (Path(sys.argv[0]).name, *sys.argv[1:])) - with WatchClaim(bridge.binding.thread).held(record): - from agent_comms.registration import Registration - - registry = await asyncio.to_thread( - Registration, Path(bridge.binding.root) / 'registry.json') - # Retry only endpoint acquisition, never the running bridge's delivery work. - with ExitStack() as endpoint: - await bridge.comms.bus.when_open(lambda: endpoint.enter_context( - registry.external_process(bridge.binding.bus, ProcessIdentity.capture(os.getpid())))) - async with asyncio.TaskGroup() as tasks: - self.start(bridge, tasks) + await bridge.registered_watch(self) class Watch(Inbox): @@ -439,11 +429,47 @@ class Watch(Inbox): def serves(cls, bridge: type[CliBridge]) -> bool: return issubclass(bridge, PublishesTranscript) + async def execute(self, bridge: OperationBridge) -> None: + await bridge.watch_operation(self) + def start(self, bridge: TranscriptBridge, tasks: asyncio.TaskGroup) -> None: super().start(bridge, tasks) tasks.create_task(bridge.publish_transcript()) +@dataclass(frozen=True, kw_only=True) +class Serve(Watch): + owner: ProcessIdentity = option('--owner', type=FieldCodec.parse) + transcript: Path = option('--transcript', type=Path) + + @classmethod + def serves(cls, bridge: type[CliBridge]) -> bool: + return issubclass(bridge, LineMonitorBridge) + + def transcript_source(self) -> Path: + return self.transcript + + async def execute(self, bridge: LineMonitorBridge) -> None: + if self.transcript.stem != bridge.binding.thread: + raise ValueError('The live transcript belongs to another Claude session') + self.owner.require_ancestor_of(os.getpid()) + await bridge.registered_watch(self) + + def start(self, bridge: LineMonitorBridge, tasks: asyncio.TaskGroup) -> None: + super().start(bridge, tasks) + tasks.create_task(MonitorService(bridge, self.owner, bridge.stop).run()) + + +class Enrol(Operation): + @classmethod + def serves(cls, bridge: type[CliBridge]) -> bool: + return issubclass(bridge, LineMonitorBridge) + + async def execute(self, bridge: LineMonitorBridge) -> None: + WatchClaim(bridge.binding.thread).enrol(bridge.watch_record()) + print('Session binding enrolled; SessionStart owns the bridge. Monitor runs watch.', flush=True) + + @dataclass(frozen=True) class IncomingFrame(ABC): message: messages.Message @@ -470,6 +496,24 @@ def __init__(self, options: BridgeOptions, binding: Binding, db: CoordinationSto def comms(self) -> Comms: return Comms(Path(self.binding.root)) + def watch_record(self) -> WatchRecord: + return WatchRecord(self.binding.bus, self.binding.root, + (Path(sys.argv[0]).name, *sys.argv[1:])) + + async def registered_watch(self, operation: Inbox) -> None: + from agent_comms.registration import Registration + registry = await asyncio.to_thread( + Registration, Path(self.binding.root) / 'registry.json') + with ExitStack() as endpoint: + await self.comms.bus.when_open(lambda: endpoint.enter_context( + registry.external_process(self.binding.bus, ProcessIdentity.capture(os.getpid())))) + with WatchClaim(self.binding.thread).held(self.watch_record()): + async with asyncio.TaskGroup() as tasks: + operation.start(self, tasks) + + async def watch_operation(self, operation: Watch) -> None: + await self.registered_watch(operation) + @abstractmethod async def show_sent(self, result: SendResult) -> None: """Display a completed send without publishing it again.""" @@ -524,7 +568,8 @@ async def watch_child(self) -> None: async def run(self) -> None: current = asyncio.current_task() assert current is not None - asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, current.cancel) + self.stop = current.cancel + asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, self.stop) try: await self.options.operation.execute(self) except* BlockingIOError: @@ -589,7 +634,7 @@ class MonitorProbe(MonitorFrame): token: str async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> None: - bridge.confirm_probe(self) + await bridge.confirm_probe(self) class MonitorPart(MonitorFrame): @@ -638,71 +683,76 @@ class TranscriptConfirmsDelivery: """Delivery is confirmed only by observing it in the bridge's own transcript subscription.""" +class MonitorIncomingFrame(IncomingFrame): + @property + @abstractmethod + def parts(self) -> tuple[str, ...]: ... + + class LineMonitorBridge(TranscriptConfirmsDelivery, TranscriptBridge[Options]): - """Monitor delivery requires the same live transcript subscription.""" + """Session-owned subscription and delivery journal, with replaceable readers.""" - incoming_frame: ClassVar[type[IncomingFrame]] + incoming_frame: ClassVar[type[MonitorIncomingFrame]] window_bytes: ClassVar[int] - PROBE_NOTICE_SECONDS: ClassVar[float] = 15.0 def __init__(self, options: Options, binding: Binding, db: CoordinationStore) -> None: super().__init__(options, binding, db) - self.probe = MonitorProbe(session=binding.thread, token=secrets.token_hex(24)) - self.monitor_ready = asyncio.Event() - self.pending_parts: dict[tuple[str, int], int] = {} - self.window_changed = asyncio.Condition() - self.write_line = ExplicitWatchAcknowledgment.line_output(sys.stdout) + self.reader: ReaderState = NoReader() + self.reader_changed = asyncio.Condition() - async def prepare_delivery(self) -> None: - # The probe is on stdout before the wait: no later write exists to flush it. - print(self.probe.model_dump_json(), flush=True) - print('Waiting for this session to observe the monitor probe; inbox remains unread.', - file=sys.stderr, flush=True) - # The harness owns notification scheduling. Missing proof never opens - # the inbox; the watch's existing cancellation owns this wait's lifetime. + def watch_record(self) -> WatchRecord: + command = (str(Path(sys.argv[0]).absolute()), Watch.__name__.lower(), + '--state', str(self.options.state.absolute())) + return WatchRecord(self.binding.bus, self.binding.root, command) + + async def watch_operation(self, operation: Watch) -> None: + await read_monitor(WatchClaim(self.binding.thread)) + + async def accept(self, message: messages.Message) -> None: + # Wait before recording an attempt. Detachment never creates + # an attempt for a message that has not reached an attached reader. + await self.prepare_delivery() try: - await asyncio.wait_for(self.monitor_ready.wait(), self.PROBE_NOTICE_SECONDS) - except TimeoutError: - # A line-lagging filter (grep aliased to ugrep) holds the last line until another - # arrives; this notice says the wait out loud and releases a held probe. - print(f'Monitor probe not observed after {self.PROBE_NOTICE_SECONDS:g}s; ' - 'inbox remains unread until this session observes it.', flush=True) - await self.monitor_ready.wait() - - def confirm_probe(self, probe: MonitorProbe) -> None: - if probe == self.probe: - self.monitor_ready.set() + await super().accept(message) + except (BrokenPipeError, ConnectionResetError) as error: + await self.report_failure( + f'Monitor failed during {message.message_id}: {error}; ' + 'the recorded attempt stays uncertain and will not be replayed.') + + async def prepare_delivery(self) -> None: + await self.reader.prepare_delivery(self) + + async def confirm_probe(self, probe: MonitorProbe) -> None: + await self.reader.confirm_probe(self, probe) async def part_observed(self, message_id: str, number: int) -> None: - async with self.window_changed: - self.pending_parts.pop((message_id, number), None) - self.window_changed.notify_all() + await self.reader.part_observed(message_id, number) @staticmethod def encode(text: str) -> str: # One reversible JSON string per line; no truncation or lost line breaks. return json.dumps(text, ensure_ascii=False) + def new_attachment(self, peer: MonitorPeer, token: str) -> MonitorAttachment: + return MonitorAttachment(peer, MonitorProbe(session=self.binding.thread, token=token)) + async def deliver(self, message: messages.Message) -> type[Outcome]: + return await self.reader.deliver(self, message) + + async def deliver_to(self, attachment: MonitorAttachment, message: messages.Message) -> type[Outcome]: parts = self.incoming_frame(message).parts with self.db: - DeliveryRow.update(self.db, where='id=?', parameters=(message.message_id,), monitor=ReceivingMonitor(len(parts))) + DeliveryRow.update(self.db, where='id=?', parameters=(message.message_id,), + monitor=ReceivingMonitor(len(parts))) for number, part in enumerate(parts, 1): - line = self.encode(part) - size = len((line + '\n').encode('utf-16-le')) - async with self.window_changed: - await self.window_changed.wait_for( - lambda: not self.pending_parts - or sum(self.pending_parts.values()) + size <= self.window_bytes) - self.pending_parts[message.message_id, number] = size - self.write_line(line) + await attachment.emit(message.message_id, number, self.encode(part), self.window_bytes) return EmittedOutcome async def show_sent(self, result: SendResult) -> None: - self.write_line(self.encode(self.sent_text(result))) + print(self.encode(self.sent_text(result)), flush=True) async def report_failure(self, text: str) -> None: - self.write_line(self.encode(text)) + print(text, file=sys.stderr, flush=True) def run_bridge(bridge_type: type[CliBridge[BridgeOptions]], options: BridgeOptions) -> None: diff --git a/src/agent_comms/cli_protocol.py b/src/agent_comms/cli_protocol.py index 93aa3650e3..ff7c949703 100644 --- a/src/agent_comms/cli_protocol.py +++ b/src/agent_comms/cli_protocol.py @@ -5,10 +5,12 @@ import builtins from abc import ABC, abstractmethod from graphlib import TopologicalSorter -from typing import TYPE_CHECKING, Annotated, AsyncIterator, ClassVar, NoReturn, Union +from typing import TYPE_CHECKING, Annotated, AsyncIterator, ClassVar, NoReturn, Self, Union from metaclass_registry import AutoRegisterMeta, RegistryConfig from pydantic import BaseModel, ConfigDict, Discriminator, InstanceOf, PlainSerializer, PlainValidator, Tag, TypeAdapter, computed_field, field_validator, model_validator +from pydantic._internal._model_construction import ModelMetaclass +from typing_extensions import TypeForm from .coordination_errors import CoordinationReadUnavailable from .declared_family import DeclaredFamily @@ -52,7 +54,7 @@ def epoch(cls, value: str) -> float: return TimestampText.from_text(value).at.timestamp() -class RegisteredWireMeta(AutoRegisterMeta, type(WireModel)): +class RegisteredWireMeta(AutoRegisterMeta, ModelMetaclass): """Compose declaration registration with external-field validation.""" @@ -65,6 +67,10 @@ class TaggedWire(WireModel, ABC, metaclass=RegisteredWireMeta): wire_tag: ClassVar[str] header: ClassVar[builtins.type[WireHeader]] = WireHeader + @staticmethod + def family_registry() -> RegistryConfig: + return RegistryConfig(registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + @computed_field @property def type(self) -> str: @@ -75,7 +81,7 @@ def tag_of(cls, value: object) -> str: return cls.header.model_validate(value).type @classmethod - def declared_union(cls) -> object: + def declared_union(cls) -> TypeForm[Self]: def tag(value: object) -> str: if isinstance(value, cls): return value.wire_tag @@ -112,8 +118,18 @@ class Outcome(DeclaredFamily, affix='Outcome'): async def acknowledge(cls, bridge: CliBridge, message: Message) -> None: """Pending transport outcomes cannot advance Core's read state.""" + @classmethod + def followed_by(cls, observation: type[Outcome]) -> type[Outcome]: + return observation + class FilteredOutcome(Outcome): + @classmethod + def followed_by(cls, observation: type[Outcome]) -> type[Outcome]: + # Transcript confirmation may race the final socket drain. An emission + # report cannot turn an already confirmed attempt back into pending. + return cls + @classmethod async def acknowledge(cls, bridge: CliBridge, message: Message) -> None: await asyncio.to_thread(bridge.comms.messaging.acknowledge, bridge.binding.bus, (message.reference,)) diff --git a/src/agent_comms/skills/comms/SKILL.md b/src/agent_comms/skills/comms/SKILL.md index 234c2a4f9e..3e5cb8080d 100644 --- a/src/agent_comms/skills/comms/SKILL.md +++ b/src/agent_comms/skills/comms/SKILL.md @@ -9,7 +9,7 @@ description: Connect this coding session to the agent-comms bus and send or rece ## Receiving -- **Claude Code:** run `agent-comms-claude-bridge watch` under Monitor so each message arrives as an event. The first run binds the state dir to this session: `agent-comms-claude-bridge watch --root ROOT --bus-thread NAME --session ID`, where NAME is this session's bus thread (install registers the commander name) and ID is the newest `~/.claude/projects//*.jsonl` file name without `.jsonl`. Later runs need no flags. Run the watch with the longest Monitor timeout (`timeout_ms=1800000`), and on every expiry notice re-arm it at once with the same command; a unit with open orders never lets its watch lapse. +- **Claude Code:** the session owns its bridge; Monitor is only its event reader. On first binding, use Bash (not Monitor) to run `agent-comms-claude-bridge connect --root ROOT --bus-thread NAME --session "$CLAUDE_CODE_SESSION_ID"`, where NAME is this session's bus thread (install registers the commander name). This records the explicit binding and starts the detached bridge for this Claude process. Subsequent SessionStart hooks start or verify that same session bridge; SessionEnd or Claude process exit stops it. Then run `agent-comms-claude-bridge watch` under Monitor so each message arrives as an event. On reader expiry, re-arm the same command. The bridge stays connected while no reader is attached, but new delivery pauses until a fresh reader's probe appears in this session's transcript. An emitted or uncertain attempt is never replayed. `enrol` with the same binding can preconfigure a known session ID before startup without starting a process. Use `--state DIR` consistently for a nondefault binding directory. - **Codex:** bind once with `agent-comms-codex-bridge watch --root ROOT --bus-thread codex --session CODEX_THREAD_ID` (stop it after it starts), then `systemctl --user daemon-reload && systemctl --user enable --now codex-comms-inbox`. Messages are pushed into that Codex conversation. - **OpenCode:** start it with `opencode-comms` instead of `opencode`. The installed TUI plugin attaches the session as thread `opencode-`. diff --git a/src/agent_comms_hooks/session.py b/src/agent_comms_hooks/session.py new file mode 100644 index 0000000000..c3a597e537 --- /dev/null +++ b/src/agent_comms_hooks/session.py @@ -0,0 +1,95 @@ +"""Claude session hooks start/stop the configured bridge, never a Monitor reader.""" +from __future__ import annotations + +import json +import os +import time +from dataclasses import dataclass + +from agent_comms.child_process import ParentedProcess, ProcessIdentity +from agent_comms.cli_bridges.monitor import EndSession, VerifySession +from agent_comms.cli_bridges.shared import Serve +from agent_comms.field_codec import FieldCodec +from agent_comms_hooks import ClaudeHook, ClaudeHookProgram +from agent_comms_hooks.watch import LiveWatch, RecordedWatch, WatchClaim + + +@dataclass(frozen=True) +class SessionHook(ClaudeHook): + transcript_path: str + + @classmethod + def current(cls, session: str, transcript: str) -> SessionHook: + return cls(session_id=session, hook_event_name='SessionStart', transcript_path=transcript) + + def owner(self) -> ProcessIdentity: + if os.environ.get('CLAUDE_CODE_SESSION_ID') != self.session_id: + raise ValueError('Claude hook session and CLAUDE_CODE_SESSION_ID disagree') + owner = ProcessIdentity.capture(int(os.environ['CLAUDE_PID'])) + owner.require_ancestor_of(os.getpid()) + return owner + + def start(self) -> None: + claim = WatchClaim(self.session_id) + if not isinstance(claim.state(), RecordedWatch): + return + owner = self.owner() + request = VerifySession(owner=owner) + with claim.launching(): + state = claim.state() + if not isinstance(state, RecordedWatch): + return + try: + request.exchange(claim) + return + except (FileNotFoundError, ConnectionRefusedError) as error: + if isinstance(state, LiveWatch): + raise RuntimeError( + 'An older watch still owns this session. Stop that Monitor once, then connect.' + ) from error + command = (state.record.command[0], Serve.__name__.lower(), *state.record.command[2:], + '--owner', json.dumps(FieldCodec.encode_record(owner)), + '--transcript', self.transcript_path) + log = claim.path.with_suffix('.log') + with log.open('ab') as output: + child = ParentedProcess.launch(command, env=dict(os.environ), output=output) + deadline = time.monotonic() + 30 + while child.alive and time.monotonic() < deadline: + try: + request.exchange(claim) + return + except (FileNotFoundError, ConnectionRefusedError): + time.sleep(0.05) + raise RuntimeError( + f'Session bridge startup not observed (PID {child.pid}); inspect {log}. No launch retry.' + ) + + def end(self) -> None: + claim = WatchClaim(self.session_id) + if not isinstance(claim.state(), LiveWatch): + return + EndSession(owner=self.owner()).exchange(claim) + + +class SessionStarted(ClaudeHookProgram): + events = ('SessionStart',) + + @classmethod + def command(cls) -> str: + return cls.console_script(f'{started.__module__}:{started.__name__}') + + +class SessionEnded(ClaudeHookProgram): + events = ('SessionEnd',) + + @classmethod + def command(cls) -> str: + return cls.console_script(f'{ended.__module__}:{ended.__name__}') + + +def started() -> None: + SessionHook.read().start() + + +def ended() -> None: + SessionHook.read().end() diff --git a/src/agent_comms_hooks/watch.py b/src/agent_comms_hooks/watch.py index e8760b9cdf..2d9605722f 100644 --- a/src/agent_comms_hooks/watch.py +++ b/src/agent_comms_hooks/watch.py @@ -31,6 +31,9 @@ class WatchRecord: root: str command: Sequence[str] + def __post_init__(self) -> None: + object.__setattr__(self, 'command', tuple(self.command)) + @dataclass(frozen=True) class WatchClaim: @@ -41,6 +44,48 @@ def path(self) -> Path: key = hashlib.sha256(self.thread.encode()).hexdigest() return Path.home() / ".local/state/agent-comms/cli-watch" / f"{key}.lock" + @property + def socket_path(self) -> Path: + # sockaddr_un is short. Keep the locator derived from the session claim. + return self.path.with_name(self.path.stem[:32] + '.sock') + + @property + def reader_path(self) -> Path: + return self.path.with_suffix('.reader.lock') + + @contextmanager + def launching(self) -> Iterator[None]: + self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + with self.path.with_suffix('.launch.lock').open('a') as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + yield + + @contextmanager + def reader_held(self) -> Iterator[None]: + with self.reader_path.open('a') as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + yield + + def has_reader(self) -> bool: + if not self.reader_path.exists(): + return False + with self.reader_path.open() as lock: + try: + fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB) + except BlockingIOError: + return True + return False + + def enrol(self, record: WatchRecord) -> None: + with self.launching(): + current = self.state() + if isinstance(current, LiveWatch): + if current.record != record: + raise ValueError('This session is already bound to another bridge command') + return + with self.held(record): + pass + @contextmanager def held(self, record: WatchRecord) -> Iterator[None]: """One watch per conversation, independent of --state and bus aliases.""" @@ -59,7 +104,7 @@ def state(self) -> WatchState: try: fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB) except BlockingIOError: - return LiveWatch() + return LiveWatch(self, WatchRecord(**json.load(lock))) text = lock.read() if not text: return UnclaimedWatch() @@ -75,15 +120,23 @@ class UnclaimedWatch(WatchState): pass -class LiveWatch(WatchState): - pass - - @dataclass(frozen=True) -class LapsedWatch(WatchState): +class RecordedWatch(WatchState): claim: WatchClaim record: WatchRecord + +class LiveWatch(RecordedWatch): + def report(self, hook: ClaudeHook) -> None: + if not self.claim.has_reader(): + hook.warn( + f'agent-comms: the session bridge for {self.record.bus} is connected, but no ' + f'Monitor reader is running. Delivery is paused. Attach Monitor now: ' + f'{shlex.join(self.record.command)}') + + +class LapsedWatch(RecordedWatch): + @property def warning(self) -> str: bus = self.record.bus diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index 7b6f54d21f..b6f470792e 100644 --- a/tests/test_monitor_delivery.py +++ b/tests/test_monitor_delivery.py @@ -168,25 +168,3 @@ async def test_explicit_watch_does_not_mark_streamed_messages_read(wired): assert {m.reference for m in wired.bus.inbox('fixer')}=={first.reference,second.reference} assert wired.messaging.acknowledge('fixer',(first.reference,))==1 assert [m.reference for m in wired.bus.inbox('fixer')]==[second.reference] - - -@pytest.mark.asyncio -async def test_watch_prints_its_monitor_probe_before_waiting(tmp_path, capsys): - import asyncio - with Binding.open(tmp_path/'state') as db: - binding=Binding(str(tmp_path),'viewer','session') - Binding.bind(db, binding) - bridge=ClaudeBridge(Options(Watch(),None,tmp_path/'state'),binding,db) - bridge.PROBE_NOTICE_SECONDS=0.05 - waiting=asyncio.create_task(bridge.prepare_delivery()) - await asyncio.sleep(0) - # The probe is flushed before the wait, with no later loop pass to flush it. - assert capsys.readouterr().out==bridge.probe.model_dump_json()+'\n' - assert not waiting.done() - async with asyncio.timeout(3): - while 'not observed' not in (notice:=capsys.readouterr().out): - await asyncio.sleep(0.01) - assert notice.startswith('Monitor probe not observed after 0.05s') and not waiting.done() - bridge.confirm_probe(bridge.probe) - async with asyncio.timeout(3): - await waiting diff --git a/tests/test_session_bridge.py b/tests/test_session_bridge.py new file mode 100644 index 0000000000..c9b22f0be2 --- /dev/null +++ b/tests/test_session_bridge.py @@ -0,0 +1,180 @@ +"""Socket/admission owners already exercised by the real Claude probe.""" +import asyncio +from contextlib import asynccontextmanager +import json +import os + +import pytest +import pytest_asyncio + +from agent_comms.child_process import ProcessIdentity +from agent_comms.cli_bridges.claude import ClaudeBridge +from agent_comms.cli_bridges.claude_protocol import ClaudeMonitorFrame +from agent_comms.cli_bridges.monitor import AttachMonitor, EndSession, MonitorService, NoReader, REPLIES, VerifySession +from agent_comms.cli_bridges.shared import Binding, DeliveryRow, MONITOR_FRAME, ObservedMonitor, Options, Serve +from agent_comms.cli_protocol import DeliveredOutcome, EmittedOutcome, UnconfirmedOutcome +from agent_comms.messages import InfoKind, Message +from agent_comms_hooks.watch import WatchClaim, WatchRecord + + +@pytest_asyncio.fixture +async def endpoint(tmp_path, monkeypatch): + monkeypatch.setattr(WatchClaim, 'path', property(lambda self: tmp_path / 'session.lock')) + monkeypatch.setattr(WatchClaim, 'socket_path', property(lambda self: tmp_path.parent / 'session.sock')) + claim = WatchClaim('session') + owner = ProcessIdentity.capture(os.getpid()) + acknowledgments = [] + class Core: + def __init__(self): + self.messaging = self + def acknowledge(self, thread, references): + acknowledgments.extend(references) + with Binding.open(tmp_path / 'state') as db, claim.held(WatchRecord('viewer', str(tmp_path), ('bridge', 'watch'))): + binding = Binding(str(tmp_path), 'viewer', 'session') + bridge = ClaudeBridge(Options(Serve(owner, tmp_path / 'session.jsonl'), binding, tmp_path / 'state'), binding, db) + bridge.comms = Core() + async def run(): + await MonitorService(bridge, owner, asyncio.current_task().cancel).run() + task = asyncio.create_task(run()) + async with asyncio.timeout(3): + while not claim.socket_path.exists(): + await asyncio.sleep(0) + try: + yield bridge, claim, owner, acknowledgments, task + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@asynccontextmanager +async def reader(claim): + stream, writer = await asyncio.open_unix_connection(str(claim.socket_path)) + writer.write((AttachMonitor().model_dump_json() + '\n').encode()) + await writer.drain() + REPLIES.validate_json(await stream.readline()).require() + probe = MONITOR_FRAME.validate_json(await stream.readline()) + try: + yield stream, writer, probe + finally: + writer.close() + await writer.wait_closed() + + +def message(body, seq=1): + return Message(sender='peer', target='viewer', type=InfoKind, body=body, timestamp=float(seq), seq=seq) + + +async def detached(bridge): + async with asyncio.timeout(3): + async with bridge.reader_changed: + await bridge.reader_changed.wait_for(lambda: isinstance(bridge.reader, NoReader)) + + +async def no_line(stream): + with pytest.raises(TimeoutError): + async with asyncio.timeout(0.03): + await stream.readline() + + +@pytest.mark.asyncio +async def test_admission_waits_for_current_probe_then_delivers_once(endpoint): + bridge, claim, owner, acknowledgments, task = endpoint + first = message('first') + delivery = asyncio.create_task(bridge.accept(first)) + await asyncio.sleep(0) + assert DeliveryRow.select(bridge.db) == [] + async with reader(claim) as (stream, writer, old_probe): + assert not delivery.done() + assert DeliveryRow.select(bridge.db) == [] + await old_probe.confirm(bridge, ObservedMonitor('probe-1', 1.)) + part = ClaudeMonitorFrame.decode_line((await stream.readline()).decode().strip()) + await delivery + assert DeliveryRow.select(bridge.db)[0].outcome is EmittedOutcome + await part.confirm(bridge, ObservedMonitor('part-1', 2.)) + assert acknowledgments == [first.reference] + await detached(bridge) + assert not task.done() + second = message('second', 2) + delivery = asyncio.create_task(bridge.accept(second)) + await asyncio.sleep(0) + assert DeliveryRow.select(bridge.db, id=second.message_id) == [] + async with reader(claim) as (stream, writer, fresh_probe): + assert fresh_probe != old_probe + await old_probe.confirm(bridge, ObservedMonitor('old-history', 3.)) + await asyncio.sleep(0) + assert not delivery.done() + assert DeliveryRow.select(bridge.db, id=second.message_id) == [] + await fresh_probe.confirm(bridge, ObservedMonitor('probe-2', 4.)) + part = ClaudeMonitorFrame.decode_line((await stream.readline()).decode().strip()) + await delivery + await part.confirm(bridge, ObservedMonitor('part-2', 5.)) + await bridge.accept(first) + await no_line(stream) + assert acknowledgments == [first.reference, second.reference, first.reference] + assert all(row.outcome is DeliveredOutcome for row in DeliveryRow.select(bridge.db)) + + +@pytest.mark.asyncio +async def test_detached_partial_attempt_is_retained_without_replay_or_old_window(endpoint): + bridge, claim, owner, acknowledgments, task = endpoint + uncertain = message('x' * 10000) + async with reader(claim) as (stream, writer, probe): + await probe.confirm(bridge, ObservedMonitor('probe-1', 1.)) + delivery = asyncio.create_task(bridge.accept(uncertain)) + await stream.readline() + await detached(bridge) + async with asyncio.timeout(3): + await delivery + assert DeliveryRow.select(bridge.db)[0].outcome is UnconfirmedOutcome + assert not acknowledgments and not task.done() + async with reader(claim) as (stream, writer, probe): + await probe.confirm(bridge, ObservedMonitor('probe-2', 2.)) + await bridge.accept(uncertain) + await no_line(stream) + fresh = message('new message after the uncertain one', 2) + await bridge.accept(fresh) + part = ClaudeMonitorFrame.decode_line((await stream.readline()).decode().strip()) + assert part.id == fresh.message_id + await part.confirm(bridge, ObservedMonitor('fresh-part', 3.)) + assert acknowledgments == [fresh.reference] + assert DeliveryRow.select(bridge.db, id=uncertain.message_id)[0].outcome is UnconfirmedOutcome + + +@pytest.mark.asyncio +async def test_session_control_fences_process_incarnation(endpoint): + bridge, claim, owner, acknowledgments, task = endpoint + stale = ProcessIdentity(owner.pid, owner.start_time + 1) + with pytest.raises(ValueError, match='different Claude process incarnation'): + await asyncio.to_thread(EndSession(owner=stale).exchange, claim) + assert not task.done() + await asyncio.to_thread(VerifySession(owner=owner).exchange, claim) + await asyncio.to_thread(EndSession(owner=owner).exchange, claim) + async with asyncio.timeout(3): + await asyncio.gather(task, return_exceptions=True) + assert not claim.socket_path.exists() + + +@pytest.mark.asyncio +async def test_confirmed_outcome_cannot_be_downgraded_by_socket_completion(endpoint): + bridge, claim, owner, acknowledgments, task = endpoint + original = message('observed before the final drain returned') + row = DeliveryRow(original.message_id, json.dumps(original.to_wire()), DeliveredOutcome) + row.insert(bridge.db) + bridge.db.commit() + await row.finish(bridge, EmittedOutcome) + assert DeliveryRow.select(bridge.db)[0].outcome is DeliveredOutcome + + +def test_enrol_is_idempotent_and_guard_reports_only_missing_reader(tmp_path, monkeypatch, capsys): + monkeypatch.setattr(WatchClaim, 'path', property(lambda self: tmp_path / 'session.lock')) + claim = WatchClaim('session') + record = WatchRecord('viewer', str(tmp_path), ('bridge', 'watch')) + claim.enrol(record) + assert claim.state().record == record + with claim.held(record): + claim.enrol(record) + claim.state().report('PostToolUse') + assert 'Delivery is paused' in capsys.readouterr().out + with claim.reader_held(): + claim.state().report('PostToolUse') + assert capsys.readouterr().out == '' diff --git a/tests/test_watch_guard.py b/tests/test_watch_guard.py index fa31cf7b69..6ba34bfded 100644 --- a/tests/test_watch_guard.py +++ b/tests/test_watch_guard.py @@ -21,7 +21,7 @@ def hook(event='PostToolUse', session='session'): input=stdin, capture_output=True, text=True, check=True).stdout assert hook() == '' - with WatchClaim('session').held(record): + with WatchClaim('session').held(record), WatchClaim('session').reader_held(): assert hook() == '' lapsed = json.loads(hook('UserPromptSubmit'))['hookSpecificOutput'] assert lapsed['hookEventName'] == 'UserPromptSubmit' @@ -36,12 +36,14 @@ def hook(event='PostToolUse', session='session'): ClaudeCodeHarness(tmp_path).configure() ClaudeCodeHarness(tmp_path).configure() hooks = json.loads((tmp_path / '.claude/settings.json').read_text())['hooks'] - assert {event: [hook['command'] for entry in entries for hook in entry['hooks']] - for event, entries in hooks.items()} == { - 'PreToolUse': ['agent-comms-commander send-guard'], - 'UserPromptSubmit': ['agent-comms-watch-guard', 'agent-comms-commander order peer-sync'], - 'PostToolUse': ['agent-comms-watch-guard', 'agent-comms-commander order peer-sync'], - 'Stop': ['agent-comms-commander order peer-sync']} + programs = agent_comms_hooks.ClaudeHookProgram.members() + expected = {event: {program.command() for program in programs if event in program.events} + for program in programs for event in program.events} + actual = {event: [hook['command'] for entry in entries for hook in entry['hooks']] + for event, entries in hooks.items()} + assert {event: set(commands) for event, commands in actual.items()} == expected + assert all(len(commands) == len(set(commands)) for commands in actual.values()) assert hooks['PreToolUse'][0]['matcher'] == 'Bash' assert 'matcher' not in hooks['PostToolUse'][0] + assert 'matcher' not in hooks['SessionStart'][0] From dc78647606265d770d0c1a1bd10453518fb4a1cb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 21:52:17 -0400 Subject: [PATCH 2/4] P233: use current operation execution and hook warnings throughout --- src/agent_comms/cli_bridges/claude.py | 4 ++-- tests/test_session_bridge.py | 12 +++++++----- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/src/agent_comms/cli_bridges/claude.py b/src/agent_comms/cli_bridges/claude.py index 909b109dd7..2fd83056bf 100755 --- a/src/agent_comms/cli_bridges/claude.py +++ b/src/agent_comms/cli_bridges/claude.py @@ -132,8 +132,8 @@ class Connect(Enrol): def serves(cls, bridge: type) -> bool: return issubclass(bridge, ClaudeBridge) - async def run(self, bridge: LineMonitorBridge) -> None: - await super().run(bridge) + async def execute(self, bridge: LineMonitorBridge) -> None: + await super().execute(bridge) SessionHook.current( bridge.binding.thread, str(RunningClaude(bridge.binding.thread).source) ).start() diff --git a/tests/test_session_bridge.py b/tests/test_session_bridge.py index c9b22f0be2..4db1be0bca 100644 --- a/tests/test_session_bridge.py +++ b/tests/test_session_bridge.py @@ -1,4 +1,4 @@ -"""Socket/admission owners already exercised by the real Claude probe.""" +"""Socket and delivery owners exercised by the real Claude probe.""" import asyncio from contextlib import asynccontextmanager import json @@ -14,6 +14,7 @@ from agent_comms.cli_bridges.shared import Binding, DeliveryRow, MONITOR_FRAME, ObservedMonitor, Options, Serve from agent_comms.cli_protocol import DeliveredOutcome, EmittedOutcome, UnconfirmedOutcome from agent_comms.messages import InfoKind, Message +from agent_comms_hooks import ClaudeHook from agent_comms_hooks.watch import WatchClaim, WatchRecord @@ -31,7 +32,7 @@ def acknowledge(self, thread, references): acknowledgments.extend(references) with Binding.open(tmp_path / 'state') as db, claim.held(WatchRecord('viewer', str(tmp_path), ('bridge', 'watch'))): binding = Binding(str(tmp_path), 'viewer', 'session') - bridge = ClaudeBridge(Options(Serve(owner, tmp_path / 'session.jsonl'), binding, tmp_path / 'state'), binding, db) + bridge = ClaudeBridge(Options(Serve(owner=owner, transcript=tmp_path / 'session.jsonl'), binding, tmp_path / 'state'), binding, db) bridge.comms = Core() async def run(): await MonitorService(bridge, owner, asyncio.current_task().cancel).run() @@ -77,7 +78,7 @@ async def no_line(stream): @pytest.mark.asyncio -async def test_admission_waits_for_current_probe_then_delivers_once(endpoint): +async def test_delivery_waits_for_current_probe_then_delivers_once(endpoint): bridge, claim, owner, acknowledgments, task = endpoint first = message('first') delivery = asyncio.create_task(bridge.accept(first)) @@ -173,8 +174,9 @@ def test_enrol_is_idempotent_and_guard_reports_only_missing_reader(tmp_path, mon assert claim.state().record == record with claim.held(record): claim.enrol(record) - claim.state().report('PostToolUse') + hook = ClaudeHook('session', 'PostToolUse') + claim.state().report(hook) assert 'Delivery is paused' in capsys.readouterr().out with claim.reader_held(): - claim.state().report('PostToolUse') + claim.state().report(hook) assert capsys.readouterr().out == '' From 16a6d2a1cc9bc751873742329e5e80e43dcba672 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 21:54:22 -0400 Subject: [PATCH 3/4] P233: decode transcript paths and fixture records through their existing codecs --- src/agent_comms/cli_bridges/shared.py | 6 ++--- src/agent_comms_hooks/session.py | 39 ++++++++++++++++----------- tests/test_monitor_delivery.py | 20 ++++++++------ 3 files changed, 38 insertions(+), 27 deletions(-) diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py index 6996ac7706..f20bb3524d 100644 --- a/src/agent_comms/cli_bridges/shared.py +++ b/src/agent_comms/cli_bridges/shared.py @@ -15,7 +15,7 @@ from dataclasses import dataclass, field, replace from functools import cached_property, partial from pathlib import Path -from typing import ClassVar, Generic, Literal, Self, TypeVar +from typing import Annotated, ClassVar, Generic, Literal, Self, TypeVar from acp.schema import SessionConfigSelectOption from metaclass_registry import RegistryConfig from pydantic import Field, TypeAdapter, model_validator @@ -39,7 +39,7 @@ from agent_comms.coordination_errors import CoordinationError from agent_comms.declared_family import DeclaredFamily from agent_comms import field_codec -from agent_comms.field_codec import FieldCodec, option +from agent_comms.field_codec import FieldCodec, PathText, option from agent_comms.launch_runtime import LaunchRuntime from agent_comms.message_reference import MessageReference from agent_comms import messages @@ -440,7 +440,7 @@ def start(self, bridge: TranscriptBridge, tasks: asyncio.TaskGroup) -> None: @dataclass(frozen=True, kw_only=True) class Serve(Watch): owner: ProcessIdentity = option('--owner', type=FieldCodec.parse) - transcript: Path = option('--transcript', type=Path) + transcript: Annotated[Path, PathText] = option('--transcript') @classmethod def serves(cls, bridge: type[CliBridge]) -> bool: diff --git a/src/agent_comms_hooks/session.py b/src/agent_comms_hooks/session.py index c3a597e537..3bfd8a43e6 100644 --- a/src/agent_comms_hooks/session.py +++ b/src/agent_comms_hooks/session.py @@ -1,4 +1,5 @@ """Claude session hooks start/stop the configured bridge, never a Monitor reader.""" + from __future__ import annotations import json @@ -20,12 +21,12 @@ class SessionHook(ClaudeHook): @classmethod def current(cls, session: str, transcript: str) -> SessionHook: - return cls(session_id=session, hook_event_name='SessionStart', transcript_path=transcript) + return cls(session_id=session, hook_event_name="SessionStart", transcript_path=transcript) def owner(self) -> ProcessIdentity: - if os.environ.get('CLAUDE_CODE_SESSION_ID') != self.session_id: - raise ValueError('Claude hook session and CLAUDE_CODE_SESSION_ID disagree') - owner = ProcessIdentity.capture(int(os.environ['CLAUDE_PID'])) + if os.environ.get("CLAUDE_CODE_SESSION_ID") != self.session_id: + raise ValueError("Claude hook session and CLAUDE_CODE_SESSION_ID disagree") + owner = ProcessIdentity.capture(int(os.environ["CLAUDE_PID"])) owner.require_ancestor_of(os.getpid()) return owner @@ -45,23 +46,29 @@ def start(self) -> None: except (FileNotFoundError, ConnectionRefusedError) as error: if isinstance(state, LiveWatch): raise RuntimeError( - 'An older watch still owns this session. Stop that Monitor once, then connect.' + "An older watch still owns this session. Stop that Monitor once, then connect." ) from error - command = (state.record.command[0], Serve.__name__.lower(), *state.record.command[2:], - '--owner', json.dumps(FieldCodec.encode_record(owner)), - '--transcript', self.transcript_path) - log = claim.path.with_suffix('.log') - with log.open('ab') as output: + command = ( + state.record.command[0], + Serve.__name__.lower(), + *state.record.command[2:], + "--owner", + json.dumps(FieldCodec.encode_record(owner)), + "--transcript", + self.transcript_path, + ) + log = claim.path.with_suffix(".log") + with log.open("ab") as output: child = ParentedProcess.launch(command, env=dict(os.environ), output=output) deadline = time.monotonic() + 30 while child.alive and time.monotonic() < deadline: try: request.exchange(claim) return - except (FileNotFoundError, ConnectionRefusedError): + except FileNotFoundError, ConnectionRefusedError: time.sleep(0.05) raise RuntimeError( - f'Session bridge startup not observed (PID {child.pid}); inspect {log}. No launch retry.' + f"Session bridge startup not observed (PID {child.pid}); inspect {log}. No launch retry." ) def end(self) -> None: @@ -72,19 +79,19 @@ def end(self) -> None: class SessionStarted(ClaudeHookProgram): - events = ('SessionStart',) + events = ("SessionStart",) @classmethod def command(cls) -> str: - return cls.console_script(f'{started.__module__}:{started.__name__}') + return cls.console_script(f"{started.__module__}:{started.__name__}") class SessionEnded(ClaudeHookProgram): - events = ('SessionEnd',) + events = ("SessionEnd",) @classmethod def command(cls) -> str: - return cls.console_script(f'{ended.__module__}:{ended.__name__}') + return cls.console_script(f"{ended.__module__}:{ended.__name__}") def started() -> None: diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index b6f470792e..b0f4ea7835 100644 --- a/tests/test_monitor_delivery.py +++ b/tests/test_monitor_delivery.py @@ -4,9 +4,9 @@ import pytest -from agent_comms.cli_bridges.claude import ClaudeBridge +from agent_comms.cli_bridges.claude import ClaudeBridge, ClaudeJsonLines from agent_comms.cli_bridges.claude_protocol import ( - CLAUDE_RECORD, ClaudeSession, ClaudeMonitorFrame, ForeignTaskNotification, + ClaudeSession, ClaudeMonitorFrame, ForeignTaskNotification, ForeignAttachment, ) from agent_comms.cli_bridges.shared import ( @@ -18,8 +18,12 @@ from agent_comms.read_basis import ExplicitWatchAcknowledgment +def decoded_record(data): + return ClaudeJsonLines().decode_line(json.dumps(data).encode(), 0) + + def notification(line, record_id='notification-record'): - return CLAUDE_RECORD.validate_python(dict( + return decoded_record(dict( type='user', uuid=record_id, timestamp='2026-10-16T16:40:05.123Z', sessionId='session', promptId='prompt', origin=dict(kind='task-notification'), message=dict(content=''+escape(line)+ @@ -27,7 +31,7 @@ def notification(line, record_id='notification-record'): def attachment_notification(line, record_id='notification-record'): - return CLAUDE_RECORD.validate_python(dict( + return decoded_record(dict( type='attachment', uuid=record_id, timestamp='2026-10-16T16:40:05.123Z', attachment=dict(type='queued_command', commandMode='task-notification', origin=dict(kind='task-notification', producer='session-task'), @@ -95,14 +99,14 @@ def test_frame_cap_and_exact_reconstruction(body, sender): def test_foreign_task_notifications_are_named_and_malformed_matching_xml_fails(): - record=CLAUDE_RECORD.validate_python(dict(type='user',uuid='record',timestamp='2026-10-16T16:40:05.123Z', + record=decoded_record(dict(type='user',uuid='record',timestamp='2026-10-16T16:40:05.123Z', sessionId='session',promptId='prompt',origin=dict(kind='task-notification'), message=dict(content='Another task producer'))) assert isinstance(record.message.content[0].text,ForeignTaskNotification) assert record.updates(ClaudeSession())==() from xml.etree.ElementTree import ParseError with pytest.raises(ParseError): - CLAUDE_RECORD.validate_python(dict(type='user',uuid='record',timestamp='2026-10-16T16:40:05.123Z', + decoded_record(dict(type='user',uuid='record',timestamp='2026-10-16T16:40:05.123Z', sessionId='session',promptId='prompt',origin=dict(kind='task-notification'), message=dict(content=''))) assert isinstance(ClaudeMonitorFrame.decode_line('ordinary unrelated monitor event'),ForeignMonitorFrame) @@ -110,12 +114,12 @@ def test_foreign_task_notifications_are_named_and_malformed_matching_xml_fails() def test_non_task_attachments_and_pending_queue_records_do_not_confirm(): - record=CLAUDE_RECORD.validate_python(dict(type='attachment', uuid='foreign-record', + record=decoded_record(dict(type='attachment', uuid='foreign-record', timestamp='2026-10-16T16:40:05.123Z', attachment=dict(type='queued_command', origin=dict(kind='peer'), prompt='Not a Monitor notification'))) assert isinstance(record.attachment, ForeignAttachment) assert record.updates(ClaudeSession())==() - queued=CLAUDE_RECORD.validate_python(dict(type='queue-operation', operation='enqueue', + queued=decoded_record(dict(type='queue-operation', operation='enqueue', sessionId='session', content='not yet received')) assert queued.updates(ClaudeSession())==() From 2dd8bac38d112b4ea1b11c06647fd5a6c2973cfe Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 21:59:41 -0400 Subject: [PATCH 4/4] P233: decode saved commands before freezing and observe child liveness --- src/agent_comms_hooks/session.py | 17 +++++++++++------ src/agent_comms_hooks/watch.py | 29 ++++++++++++++++------------- 2 files changed, 27 insertions(+), 19 deletions(-) diff --git a/src/agent_comms_hooks/session.py b/src/agent_comms_hooks/session.py index 3bfd8a43e6..6577ab0599 100644 --- a/src/agent_comms_hooks/session.py +++ b/src/agent_comms_hooks/session.py @@ -26,7 +26,10 @@ def current(cls, session: str, transcript: str) -> SessionHook: def owner(self) -> ProcessIdentity: if os.environ.get("CLAUDE_CODE_SESSION_ID") != self.session_id: raise ValueError("Claude hook session and CLAUDE_CODE_SESSION_ID disagree") - owner = ProcessIdentity.capture(int(os.environ["CLAUDE_PID"])) + pid = os.environ.get("CLAUDE_PID", "") + if not pid: + raise ValueError("Claude hook has no exported CLAUDE_PID") + owner = ProcessIdentity.capture(int(pid)) owner.require_ancestor_of(os.getpid()) return owner @@ -35,7 +38,7 @@ def start(self) -> None: if not isinstance(claim.state(), RecordedWatch): return owner = self.owner() - request = VerifySession(owner=owner) + request = VerifySession.model_validate(dict(owner=owner)) with claim.launching(): state = claim.state() if not isinstance(state, RecordedWatch): @@ -46,7 +49,8 @@ def start(self) -> None: except (FileNotFoundError, ConnectionRefusedError) as error: if isinstance(state, LiveWatch): raise RuntimeError( - "An older watch still owns this session. Stop that Monitor once, then connect." + "An older watch still owns this session. " + "Stop that Monitor once, then connect." ) from error command = ( state.record.command[0], @@ -61,21 +65,22 @@ def start(self) -> None: with log.open("ab") as output: child = ParentedProcess.launch(command, env=dict(os.environ), output=output) deadline = time.monotonic() + 30 - while child.alive and time.monotonic() < deadline: + while child.alive() and time.monotonic() < deadline: try: request.exchange(claim) return except FileNotFoundError, ConnectionRefusedError: time.sleep(0.05) raise RuntimeError( - f"Session bridge startup not observed (PID {child.pid}); inspect {log}. No launch retry." + f"Session bridge startup not observed (PID {child.pid}); inspect {log}. " + "No launch retry." ) def end(self) -> None: claim = WatchClaim(self.session_id) if not isinstance(claim.state(), LiveWatch): return - EndSession(owner=self.owner()).exchange(claim) + EndSession.model_validate(dict(owner=self.owner())).exchange(claim) class SessionStarted(ClaudeHookProgram): diff --git a/src/agent_comms_hooks/watch.py b/src/agent_comms_hooks/watch.py index 2d9605722f..3f7ba989dd 100644 --- a/src/agent_comms_hooks/watch.py +++ b/src/agent_comms_hooks/watch.py @@ -29,10 +29,12 @@ class WatchRecord: bus: str root: str - command: Sequence[str] + command: tuple[str, ...] - def __post_init__(self) -> None: - object.__setattr__(self, 'command', tuple(self.command)) + @classmethod + def decode(cls, bus: str, root: str, command: Sequence[str]) -> WatchRecord: + """Normalize the saved JSON array before constructing the immutable record.""" + return cls(bus, root, tuple(command)) @dataclass(frozen=True) @@ -47,22 +49,22 @@ def path(self) -> Path: @property def socket_path(self) -> Path: # sockaddr_un is short. Keep the locator derived from the session claim. - return self.path.with_name(self.path.stem[:32] + '.sock') + return self.path.with_name(self.path.stem[:32] + ".sock") @property def reader_path(self) -> Path: - return self.path.with_suffix('.reader.lock') + return self.path.with_suffix(".reader.lock") @contextmanager def launching(self) -> Iterator[None]: self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) - with self.path.with_suffix('.launch.lock').open('a') as lock: + with self.path.with_suffix(".launch.lock").open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX) yield @contextmanager def reader_held(self) -> Iterator[None]: - with self.reader_path.open('a') as lock: + with self.reader_path.open("a") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) yield @@ -81,7 +83,7 @@ def enrol(self, record: WatchRecord) -> None: current = self.state() if isinstance(current, LiveWatch): if current.record != record: - raise ValueError('This session is already bound to another bridge command') + raise ValueError("This session is already bound to another bridge command") return with self.held(record): pass @@ -104,11 +106,11 @@ def state(self) -> WatchState: try: fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB) except BlockingIOError: - return LiveWatch(self, WatchRecord(**json.load(lock))) + return LiveWatch(self, WatchRecord.decode(**json.load(lock))) text = lock.read() if not text: return UnclaimedWatch() - return LapsedWatch(self, WatchRecord(**json.loads(text))) + return LapsedWatch(self, WatchRecord.decode(**json.loads(text))) class WatchState: @@ -130,9 +132,10 @@ class LiveWatch(RecordedWatch): def report(self, hook: ClaudeHook) -> None: if not self.claim.has_reader(): hook.warn( - f'agent-comms: the session bridge for {self.record.bus} is connected, but no ' - f'Monitor reader is running. Delivery is paused. Attach Monitor now: ' - f'{shlex.join(self.record.command)}') + f"agent-comms: the session bridge for {self.record.bus} is connected, but no " + f"Monitor reader is running. Delivery is paused. Attach Monitor now: " + f"{shlex.join(self.record.command)}" + ) class LapsedWatch(RecordedWatch):