diff --git a/docs/cli-bridges/claude-code.md b/docs/cli-bridges/claude-code.md index c161976594..3c7ffb57ff 100644 --- a/docs/cli-bridges/claude-code.md +++ b/docs/cli-bridges/claude-code.md @@ -2,7 +2,37 @@ Claude Code's Monitor injects stdout lines into its conversation. The adapter is `agent_comms.cli_bridges.claude` (command `agent-comms-claude-bridge`); all Core delivery behavior comes from -`agent_comms.cli_bridges.shared`, shared with Codex. No Claude settings hook is needed. +`agent_comms.cli_bridges.shared`, shared with Codex. No persistent Claude settings change is needed; +managed sessions receive their lifecycle hooks through per-process settings. + +## Start a session that can wake from its first bus message + +Monitor needs a tool call first, so it cannot wake a never-prompted session. +Instead, explicitly launch a fresh Claude session with its stdin owned by the bridge: + +```sh +agent-comms-claude-bridge run \ + --root /path/to/private-bus --bus-thread claude-code \ + --session NEW_UUID --state /path/to/private-bridge-state +``` + +Declare the external bus thread first. Run launches the real `claude` executable +in that thread's worktree, with stream-JSON input/output and the supplied session +UUID. It sends no bootstrap prompt. The first incoming conversation message is +its first user input; subsequent bus messages use the same process and session. +Claude's real SessionStart hook starts the same session-owned daemon used by +Monitor. Run only attaches a managed-input frontend to that daemon; it owns no +registry entry, inbox watch or transcript reader. Registration retains the +`IncomingFrame` family's turn-start behavior with the daemon's live endpoint. Delivery is acknowledged only when Claude replays that +input's UUID, not when the pipe flushes. The session's owner process exit stops the daemon and its bus watch, leaving +uncertain attempts in the existing journal without replay. + +`--claude PATH` selects the executable; repeat `--claude-arg=ARG` to pass Claude +options. The bridge does not override the configured model, tools or permission +policy. Raw Claude stream-JSON output goes to stdout; progress goes to stderr. +The existing transcript reader publishes Claude's actual saved history to Core. +This is opt-in and starts a new managed session; it does not take over an +already-running interactive Claude session or replace its Monitor transport. ## Start the monitor @@ -42,8 +72,9 @@ and cancellation; an internal startup timer cannot safely guess its latency. - `DeliveryJournal` records message IDs and transport results in `~/.local/state/claude-comms-inbox/delivery.sqlite3`. Core still owns the inbox. - `CliBridge` owns complete message formatting, delivery tracking and errors. -- `Inbox` owns the delivery lifetime and lock; `Watch` also starts transcript - publication. `Send` owns a required target, body and kind. Argument decoding +- The SessionStart-owned `Serve` daemon owns registration, inbox delivery and + transcript publication. `Watch` attaches Monitor; `Run` attaches native stdin + and holds the same reader lock. `Send` owns a required target, body and kind. Argument decoding constructs the operation; execution does not compare command strings. - `ClaudeMonitorFrame` owns Monitor's 500-character event cap. It splits a body into numbered parts; each begins `Comms in · sender · id · k/n`. The encoded @@ -56,10 +87,10 @@ and cancellation; an internal startup timer cannot safely guess its latency. publication. Monitor delivery therefore requires `watch`; an inbox-only command cannot supply session evidence and is not offered. Claude's completed JSONL blocks supply replies, visible reasoning, - tool calls/results and current activity without a plugin or settings hook. + tool calls/results and current activity without a transcript plugin. Hidden thinking supplies activity only. Unknown records fail visibly. - Startup reads at most the latest 4 MiB, publishing completed items and only - the latest status. The bridge then reads appended complete lines every 0.2 s; + the latest status. Filesystem notifications admit appended complete lines; a partial final line waits for its newline. Replacement or truncation fails. No offset is saved. Core makes repeated completed items idempotent. diff --git a/pyproject.toml b/pyproject.toml index 90fe6d07c9..2c68aab9f9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,10 @@ agent-comms-codex-bridge = "agent_comms.cli_bridges.codex:main" agent-comms-opencode-bridge = "agent_comms.cli_bridges.opencode:main" agent-comms-commander = "agent_comms.commander.cli:main" agent-comms-watch-guard = "agent_comms_hooks.watch:main" +agent-comms-claude-precompact = "agent_comms_hooks.compaction:starting" +agent-comms-claude-compacted = "agent_comms_hooks.compaction:finished" +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 f12253f903..0bed42ffba 100644 --- a/src/agent_comms/child_process.py +++ b/src/agent_comms/child_process.py @@ -34,6 +34,8 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar +from pydantic import ConfigDict, TypeAdapter + from .declared_family import DeclaredFamily from .errors import RelationViolationError from .field_codec import FieldCodec, NullMember, RecordMember, SubclassUnion @@ -263,6 +265,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 +444,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 +516,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 +535,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: @@ -1339,6 +1394,15 @@ class StreamingChildStdio(ChildStdio): limit: int = 65536 stderr: int | io.IOBase = subprocess.PIPE + _output: ClassVar[TypeAdapter[asyncio.StreamReader]] = TypeAdapter( + asyncio.StreamReader, config=ConfigDict(arbitrary_types_allowed=True) + ) + + @classmethod + def output(cls, process: asyncio.subprocess.Process) -> asyncio.StreamReader: + """Decode the external subprocess result against this declared PIPE contract.""" + return cls._output.validate_python(process.stdout) + @property def options(self) -> dict: return { diff --git a/src/agent_comms/cli_bridges/claude.py b/src/agent_comms/cli_bridges/claude.py index 4e8cd46d8e..340ea9b09e 100755 --- a/src/agent_comms/cli_bridges/claude.py +++ b/src/agent_comms/cli_bridges/claude.py @@ -1,21 +1,41 @@ """Claude Code sessions: their stored transcripts, and the live bridge that follows one.""" + from __future__ import annotations +import argparse +import asyncio import json -from collections.abc import Iterator +import sys +from collections.abc import AsyncIterator, Iterator from contextlib import aclosing +from dataclasses import dataclass from functools import cached_property from pathlib import Path -from typing import AsyncIterator, BinaryIO, Self +from typing import BinaryIO, Self -from agent_comms.cli_protocol import WireModel +from agent_comms.child_process import AttachedChild, StreamingChildStdio +from agent_comms.cli_protocol import WireModel, raise_child_exit from agent_comms.coordination_database import CoordinationStore from agent_comms.private_path import FileRevision +from agent_comms.runtime_updates import RuntimeUpdate, SessionFacts from agent_comms.wire_watch import WatchSource, WireWatch -from .claude_protocol import CLAUDE_RECORD, ClaudeRecord, ClaudeSession, ClaudeMonitorFrame +from agent_comms_hooks.session import SessionEnded, SessionHook, SessionStarted +from agent_comms_hooks.watch import WatchClaim + +from .claude_protocol import ( + CLAUDE_OUTPUT, + CLAUDE_RECORD, + ClaudeHookCommand, + ClaudeHookMatcher, + ClaudeHookSettings, + ClaudeInputFrame, + ClaudeMonitorFrame, + ClaudeRecord, + ClaudeSession, +) from .history import ClaudeHistory, JsonLinesFile, OfflineHistory, OnlineHistory -from .shared import Binding, LineMonitorBridge, Options, run_bridge -from agent_comms.runtime_updates import RuntimeUpdate, SessionFacts +from .monitor import REPLIES, AttachInput +from .shared import Binding, CliBridge, Enrol, LineMonitorBridge, Operation, Options, run_bridge class SubagentMeta(WireModel): @@ -29,18 +49,23 @@ class ClaudeJsonLines(JsonLinesFile): state = ClaudeSession def decode_line(self, raw: bytes, position: int): - return self.records.validate_python({**json.loads(raw), 'position': position}) + return self.records.validate_python({**json.loads(raw), "position": position}) - def record_updates(self, record: ClaudeRecord, state: ClaudeSession, offset: int, - raw: bytes) -> Iterator[RuntimeUpdate]: + def record_updates( + self, record: ClaudeRecord, state: ClaudeSession, offset: int, raw: bytes + ) -> Iterator[RuntimeUpdate]: yield from state.located(record.cwd) for update in record.updates(state): yield from update.during(state.phase) def facts(self, stream: BinaryIO, boundary: int) -> Iterator[RuntimeUpdate]: """The file is named by its session; a subagent's task description is beside it, not in it.""" - meta = self.source.with_suffix('.meta.json') - title = SubagentMeta.model_validate_json(meta.read_bytes()).description if meta.is_file() else '' + meta = self.source.with_suffix(".meta.json") + title = ( + SubagentMeta.model_validate_json(meta.read_bytes()).description + if meta.is_file() + else "" + ) yield SessionFacts(source_id=self.source.stem, title=title) @@ -48,18 +73,22 @@ class ClaudeTranscript(ClaudeJsonLines, ClaudeHistory, OfflineHistory): """Claude Code writes one transcript file per session (`.jsonl`) and per subagent (`/subagents/agent-.jsonl`), under ~/.claude/projects.""" - suffixes = frozenset({'.jsonl'}) + suffixes = frozenset({".jsonl"}) def facts(self, stream: BinaryIO, boundary: int) -> Iterator[RuntimeUpdate]: if self.session_id not in (None, self.source.stem): - raise ValueError('The transcript belongs to a different Claude session.') + raise ValueError("The transcript belongs to a different Claude session.") return super().facts(stream, boundary) @staticmethod def subagents(session: Path) -> tuple[Path, ...]: """The subagent transcripts a stored session started, oldest first.""" - return tuple(sorted(session.with_suffix('').glob('subagents/agent-*.jsonl'), - key=lambda path: path.stat().st_mtime)) + return tuple( + sorted( + session.with_suffix("").glob("subagents/agent-*.jsonl"), + key=lambda path: path.stat().st_mtime, + ) + ) class RunningClaude(ClaudeJsonLines, ClaudeHistory, OnlineHistory): @@ -71,9 +100,11 @@ def __init__(self, thread: str) -> None: @cached_property def source(self) -> Path: - matches = tuple((Path.home() / '.claude/projects').glob(f'*/{self.thread}.jsonl')) + matches = tuple((Path.home() / ".claude/projects").glob(f"*/{self.thread}.jsonl")) if len(matches) != 1: - raise ValueError(f'Expected one Claude transcript for {self.thread}; found {len(matches)}') + raise ValueError( + f"Expected one Claude transcript for {self.thread}; found {len(matches)}" + ) return matches[0] @classmethod @@ -85,6 +116,7 @@ class ClaudeBridge(RunningClaude, LineMonitorBridge, declared_name=None): """Follow this Claude session's file without submitting an input.""" incoming_frame = ClaudeMonitorFrame + input_frame = ClaudeInputFrame window_bytes = ClaudeMonitorFrame.BATCH_CHARS * 2 RECENT_BYTES = 4 * 1024 * 1024 @@ -104,23 +136,132 @@ 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: - for update in self.decoded(stream, self.history_start(stream, original.size), original.size, session): + 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. for update in session.phase.window_ended(session): 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') + raise ValueError("Claude transcript was replaced or truncated") for update in self.decoded(stream, stream.tell(), opened.size, session): 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() + + +@dataclass(frozen=True) +class Run(Operation[ClaudeBridge]): + """Own real Claude stdin, not a second registration, inbox or history reader.""" + + claude: str + arguments: tuple[str, ...] + + @classmethod + def serves(cls, bridge: type[CliBridge]) -> bool: + return issubclass(bridge, ClaudeBridge) + + @classmethod + def add_arguments(cls, parser: argparse.ArgumentParser) -> None: + parser.add_argument("--claude", default="claude") + parser.add_argument("--claude-arg", dest="arguments", action="append", default=[]) + + async def forward(self, reader: asyncio.StreamReader, child: AttachedChild) -> None: + async for line in reader: + await child.write(line) + raise ConnectionResetError("Session bridge disconnected; no input retried") + + async def observe(self, output: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + async for line in output: + event = CLAUDE_OUTPUT.validate_json(line) + sys.stdout.buffer.write(line) + sys.stdout.buffer.flush() + await event.observed(writer) + + async def attached( + self, child: AttachedChild, output: asyncio.StreamReader, claim: WatchClaim + ) -> None: + async with aclosing( + WireWatch.observations(WatchSource.modified_file(claim.socket_path)) + ) as changes: + async for _ in changes: + if claim.socket_path.exists(): + break + reader, writer = await asyncio.open_unix_connection(str(claim.socket_path)) + try: + request = AttachInput.model_validate(dict(owner=child.identity)) + writer.write((request.model_dump_json() + "\n").encode()) + await writer.drain() + REPLIES.validate_json(await reader.readline()).require() + print( + "Managed Claude input attached to SessionStart bridge.", file=sys.stderr, flush=True + ) + async with asyncio.TaskGroup() as tasks: + tasks.create_task(self.forward(reader, child)) + tasks.create_task(self.observe(output, writer)) + finally: + writer.close() + await writer.wait_closed() + + async def exited(self, child: AttachedChild) -> None: + await raise_child_exit(child, "Claude session exited; no input retried") + + async def run(self, bridge: ClaudeBridge) -> None: + claim = WatchClaim(bridge.binding.thread) + claim.enrol(bridge.watch_record()) + settings = ClaudeHookSettings( + hooks={ + event: ( + ClaudeHookMatcher(hooks=(ClaudeHookCommand(command=program.source_command()),)), + ) + for program in (SessionStarted, SessionEnded) + for event in program.events + } + ) + stdio = StreamingChildStdio(limit=8 * 1024 * 1024, stderr=sys.stderr.fileno()) + child = await AttachedChild.start( + ClaudeInputFrame.command( + self.claude, + bridge.binding.thread, + (*self.arguments, "--settings", settings.model_dump_json()), + ), + cwd=bridge.comms.registry.require(bridge.binding.bus).worktree, + stdio=stdio, + ) + try: + with claim.reader_held(): + output = stdio.output(child.process) + async with asyncio.TaskGroup() as tasks: + tasks.create_task(self.attached(child, output, claim)) + tasks.create_task(self.exited(child)) + finally: + await child.stop() + + def main() -> None: - run_bridge(ClaudeBridge, Options.parse(__doc__, Path.home() / '.local/state/claude-comms-inbox', ClaudeBridge)) + 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 6b40971ae3..23c1a0129b 100644 --- a/src/agent_comms/cli_bridges/claude_protocol.py +++ b/src/agent_comms/cli_bridges/claude_protocol.py @@ -1,15 +1,19 @@ """Claude's completed JSONL blocks own transcript items and activity.""" + from __future__ import annotations +import asyncio import json from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import ClassVar, Literal +from uuid import NAMESPACE_URL, UUID, uuid5 from xml.etree import ElementTree from metaclass_registry import RegistryConfig from pydantic import ( + ConfigDict, Field, JsonValue, TypeAdapter, @@ -18,7 +22,8 @@ ) from agent_comms.activity import ActivityState -from agent_comms.cli_protocol import TaggedWire, WireHeader, WireModel +from agent_comms.agent_backend import InputId +from agent_comms.cli_protocol import OpenWire, TaggedWire, WireHeader, WireModel from agent_comms.runtime_updates import ( ActiveItem, FinishedItem, @@ -41,11 +46,13 @@ UserTranscript, ) +from .monitor import InputAcknowledgment from .shared import ( MONITOR_FRAME, ForeignMonitorFrame, IncomingFrame, MonitorFrame, + MonitorIncomingFrame, MonitorObservation, MonitorPart, MonitorProbe, @@ -53,20 +60,117 @@ ) -class ClaudeMonitorFrame(IncomingFrame): +class ClaudeHookCommand(WireModel): + model_config = ConfigDict(frozen=True) + type: Literal["command"] = "command" + command: str + + +class ClaudeHookMatcher(WireModel): + model_config = ConfigDict(frozen=True) + hooks: tuple[ClaudeHookCommand, ...] + + +class ClaudeHookSettings(WireModel): + model_config = ConfigDict(frozen=True) + hooks: dict[str, tuple[ClaudeHookMatcher, ...]] + + +class ClaudeInputMessage(WireModel): + model_config = ConfigDict(frozen=True) + role: Literal["user"] = "user" + content: str + + +class ClaudeInputSubmission(WireModel): + model_config = ConfigDict(frozen=True) + type: Literal["user"] = "user" + uuid: UUID + message: ClaudeInputMessage + + +class ClaudeInputFrame(IncomingFrame): + @property + def identity(self) -> UUID: + return uuid5(NAMESPACE_URL, self.message.message_id) + + @property + def input_id(self) -> InputId: + return InputId(str(self.identity)) + + @property + def payload(self) -> tuple[bytes, ...]: + submitted = ClaudeInputSubmission( + uuid=self.identity, message=ClaudeInputMessage(content=self.text) + ) + return ((submitted.model_dump_json() + "\n").encode(),) + + @staticmethod + def command(executable: str, session: str, arguments: tuple[str, ...]) -> tuple[str, ...]: + return ( + executable, + *arguments, + "-p", + "--input-format", + "stream-json", + "--output-format", + "stream-json", + "--verbose", + "--replay-user-messages", + "--session-id", + session, + ) + + +class ClaudeOutput(OpenWire): + """The SDK stream's external tag is decoded once; variants own its effect.""" + + __registry_config__ = TaggedWire.family_registry() + model_config = ConfigDict(frozen=True) + + async def observed(self, writer: asyncio.StreamWriter) -> None: + """Other SDK output is displayed, not input acceptance.""" + + +class ClaudeUserOutput(ClaudeOutput): + wire_tag = "user" + uuid: UUID + + async def observed(self, writer: asyncio.StreamWriter) -> None: + writer.write((InputAcknowledgment(identity=self.uuid).model_dump_json() + "\n").encode()) + await writer.drain() + + +class ForeignClaudeOutput(ClaudeOutput): + """Other SDK output is displayed without confirming input delivery.""" + + wire_tag = "foreign" + + +CLAUDE_OUTPUT = TypeAdapter(ClaudeOutput.declared_union()) + + +class ClaudeMonitorFrame(MonitorIncomingFrame): """Claude Monitor caps lines at 500 and joined batches at 3000 UTF-16 units.""" + + @property + def payload(self) -> tuple[bytes, ...]: + return tuple((json.dumps(part, ensure_ascii=False) + "\n").encode() for part in self.parts) + EVENT_CHARS = 500 BATCH_CHARS = 3000 - HEADER = 'Comms in' + HEADER = "Comms in" JSON_TEXT = TypeAdapter(str) @classmethod def event_size(cls, text: str) -> int: - return len(json.dumps(text, ensure_ascii=False).encode('utf-16-le')) // 2 + return len(json.dumps(text, ensure_ascii=False).encode("utf-16-le")) // 2 def header(self, number: int, total: int) -> str: - return (f'{self.HEADER} · {self.message.sender} · ' - f'{self.message.message_id} · {number}/{total}\n') + return ( + f"{self.HEADER} · {self.message.sender} · " + f"{self.message.message_id} · {number}/{total}\n" + ) @property def parts(self) -> tuple[str, ...]: @@ -80,16 +184,15 @@ def parts(self) -> tuple[str, ...]: for character in self.message.body: size = self.event_size(character) - 2 # JSON quotes belong to the header budget. if size > budget: - raise ValueError('Claude Monitor header leaves no room for a body character') + raise ValueError("Claude Monitor header leaves no room for a body character") if used + size > budget: - bodies.append(''.join(current)) + bodies.append("".join(current)) current, used = [], 0 current.append(character) used += size - bodies.append(''.join(current)) + bodies.append("".join(current)) total = len(bodies) - return tuple(self.header(number, total) + body - for number, body in enumerate(bodies, 1)) + return tuple(self.header(number, total) + body for number, body in enumerate(bodies, 1)) @classmethod def decode_line(cls, line: str) -> MonitorFrame: @@ -97,22 +200,23 @@ def decode_line(cls, line: str) -> MonitorFrame: return MONITOR_FRAME.validate_json(line) # Classify the external header before decoding its JSON string. Old or # unrelated monitor text, including truncated text, is a named foreign frame. - if not line.startswith('"' + cls.HEADER + ' · '): + if not line.startswith('"' + cls.HEADER + " · "): return ForeignMonitorFrame(text=line) - raw_header, separator, _ = line.partition(r'\n') + raw_header, separator, _ = line.partition(r"\n") if not separator or not line.endswith('"'): return ForeignMonitorFrame(text=line) - columns = raw_header[1:].rsplit(' · ', 2) + columns = raw_header[1:].rsplit(" · ", 2) if len(columns) != 3: return ForeignMonitorFrame(text=line) - number, slash, total = columns[-1].partition('/') + number, slash, total = columns[-1].partition("/") if not slash or not number.isdecimal() or not total.isdecimal(): return ForeignMonitorFrame(text=line) decoded = cls.JSON_TEXT.validate_json(line) - header, _, body = decoded.partition('\n') - sender, message_id, _ = header.removeprefix(cls.HEADER + ' · ').rsplit(' · ', 2) - return MonitorPart(id=message_id, sender=sender, number=int(number), - total=int(total), body=body) + header, _, body = decoded.partition("\n") + sender, message_id, _ = header.removeprefix(cls.HEADER + " · ").rsplit(" · ", 2) + return MonitorPart( + id=message_id, sender=sender, number=int(number), total=int(total), body=body + ) @classmethod def decode_event(cls, event: str) -> tuple[MonitorFrame, ...]: @@ -159,31 +263,35 @@ def window_ended(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return () def missing_tool(self, tool_id: str) -> tuple[RuntimeUpdate, ...]: - raise ValueError(f'Claude result has no tool start: {tool_id}') + raise ValueError(f"Claude result has no tool start: {tool_id}") def unturned_finish(self) -> tuple[RuntimeUpdate, ...]: - raise ValueError('Claude completed an item before identifying its turn') + raise ValueError("Claude completed an item before identifying its turn") class ClaudeTurn(ABC): """The transcript's identified turn owns item identities.""" @abstractmethod - def finished(self, item: str, events: tuple[TranscriptEvent, ...], - phase: ClaudeReaderPhase) -> tuple[RuntimeUpdate, ...]: ... + def finished( + self, item: str, events: tuple[TranscriptEvent, ...], phase: ClaudeReaderPhase + ) -> tuple[RuntimeUpdate, ...]: ... @abstractmethod - def active(self, tool_id: str, phase: ClaudeReaderPhase, - state: ActivityState, detail: str) -> tuple[RuntimeUpdate, ...]: ... + def active( + self, tool_id: str, phase: ClaudeReaderPhase, state: ActivityState, detail: str + ) -> tuple[RuntimeUpdate, ...]: ... class UnknownTurn(ClaudeTurn): - def finished(self, item: str, events: tuple[TranscriptEvent, ...], - phase: ClaudeReaderPhase) -> tuple[RuntimeUpdate, ...]: + def finished( + self, item: str, events: tuple[TranscriptEvent, ...], phase: ClaudeReaderPhase + ) -> tuple[RuntimeUpdate, ...]: return phase.unturned_finish() - def active(self, tool_id: str, phase: ClaudeReaderPhase, - state: ActivityState, detail: str) -> tuple[RuntimeUpdate, ...]: + def active( + self, tool_id: str, phase: ClaudeReaderPhase, state: ActivityState, detail: str + ) -> tuple[RuntimeUpdate, ...]: return phase.unturned_finish() @@ -191,12 +299,14 @@ def active(self, tool_id: str, phase: ClaudeReaderPhase, class KnownTurn(ClaudeTurn): id: str - def finished(self, item: str, events: tuple[TranscriptEvent, ...], - phase: ClaudeReaderPhase) -> tuple[RuntimeUpdate, ...]: + def finished( + self, item: str, events: tuple[TranscriptEvent, ...], phase: ClaudeReaderPhase + ) -> tuple[RuntimeUpdate, ...]: return (FinishedItem(self.id, item, events),) - def active(self, tool_id: str, phase: ClaudeReaderPhase, - state: ActivityState, detail: str) -> tuple[RuntimeUpdate, ...]: + def active( + self, tool_id: str, phase: ClaudeReaderPhase, state: ActivityState, detail: str + ) -> tuple[RuntimeUpdate, ...]: return (ActiveItem(self.id, tool_id, state, detail),) @@ -205,7 +315,7 @@ class ClaudeSession: turn: ClaudeTurn = field(default_factory=UnknownTurn) tools: dict[str, ToolCall] = field(default_factory=dict) phase: ClaudeReaderPhase = field(default_factory=CatchUp) - project: str = '' + project: str = "" def located(self, cwd: str) -> tuple[RuntimeUpdate, ...]: if not cwd or cwd == self.project: @@ -216,13 +326,17 @@ def located(self, cwd: str) -> tuple[RuntimeUpdate, ...]: def started(self, call: ToolCall) -> None: self.tools[call.id] = call - def tool_result(self, tool_id: str, text: str, ok: bool, position: int) -> tuple[RuntimeUpdate, ...]: + def tool_result( + self, tool_id: str, text: str, ok: bool, position: int + ) -> tuple[RuntimeUpdate, ...]: try: call = self.tools.pop(tool_id) except KeyError: return self.phase.missing_tool(tool_id) - return (*self.finished(tool_id, call.finished(text, ok, position)), - *self.turn.active(tool_id, self.phase, ActivityState.WORKING, '')) + return ( + *self.finished(tool_id, call.finished(text, ok, position)), + *self.turn.active(tool_id, self.phase, ActivityState.WORKING, ""), + ) def finished(self, item: str, events: tuple[TranscriptEvent, ...]) -> tuple[RuntimeUpdate, ...]: return self.turn.finished(item, events, self.phase) @@ -233,37 +347,38 @@ class TimedRecord(WireModel): uuid: str timestamp: float - @field_validator('timestamp', mode='before') + @field_validator("timestamp", mode="before") @classmethod def epoch(cls, value: str) -> float: return datetime.fromisoformat(value).timestamp() class ClaudeHeader(WireHeader): - @model_validator(mode='before') + @model_validator(mode="before") @classmethod def local_command(cls, value: dict) -> dict: # Local slash commands (/login, /cost) are written as user records # without a prompt; they never reach the model. Decode that shape here. # Harness notes (reminders, skill text) are meta records with no sender; # a compaction summary is a user record flagged as one. - if value.get('type') == 'user': - if value.get('origin', {}).get('kind') == TaskNotification.wire_tag: - return {'type': TaskUserRecord.wire_tag} - if 'promptId' not in value: - return {'type': LocalCommandOutput.wire_tag} - if value.get('isCompactSummary'): - return {'type': CompactSummary.wire_tag} - if value.get('isMeta') and 'origin' not in value: - return {'type': HarnessNote.wire_tag} + if value.get("type") == "user": + if value.get("origin", {}).get("kind") == TaskNotification.wire_tag: + return {"type": TaskUserRecord.wire_tag} + if "promptId" not in value: + return {"type": LocalCommandOutput.wire_tag} + if value.get("isCompactSummary"): + return {"type": CompactSummary.wire_tag} + if value.get("isMeta") and "origin" not in value: + return {"type": HarnessNote.wire_tag} return value class ClaudeRecord(TaggedWire, ABC): __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) header = ClaudeHeader - cwd: str = '' + cwd: str = "" def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return () @@ -274,12 +389,13 @@ def begins_history(self) -> bool: def request(self) -> str: """The prompt this record carries, read without the session around it.""" - return '' + return "" class AssistantBlock(TaggedWire, ABC): __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) @abstractmethod def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: ... @@ -294,37 +410,49 @@ def result_text(self) -> str: class ImageContent(WireModel): def result_text(self) -> str: - return '[image]' + return "[image]" class AssistantText(AssistantBlock, TextContent): - wire_tag = 'text' + wire_tag = "text" def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return session.finished(record.uuid, (AssistantTranscript(self.text, timestamp=record.timestamp, position=record.position),)) + return session.finished( + record.uuid, + (AssistantTranscript(self.text, timestamp=record.timestamp, position=record.position),), + ) class RedactedThinking(AssistantBlock): - wire_tag = 'redacted_thinking' + wire_tag = "redacted_thinking" def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return session.turn.active(record.uuid, session.phase, ActivityState.THINKING, '') + return session.turn.active(record.uuid, session.phase, ActivityState.THINKING, "") class ThinkingBlock(RedactedThinking): - wire_tag = 'thinking' + wire_tag = "thinking" thinking: str def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: activity = super().updates(record, session) if not self.thinking: return activity - return (*activity, *session.finished( - record.uuid, (ThinkingTranscript(self.thinking, timestamp=record.timestamp, position=record.position),))) + return ( + *activity, + *session.finished( + record.uuid, + ( + ThinkingTranscript( + self.thinking, timestamp=record.timestamp, position=record.position + ), + ), + ), + ) class ToolUse(AssistantBlock): - wire_tag = 'tool_use' + wire_tag = "tool_use" id: str name: str input: dict[str, JsonValue] @@ -334,7 +462,6 @@ def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeU session.started(call) return session.turn.active(self.id, session.phase, ActivityState.WORKING, call.title) - def call(self) -> ToolCall: return CliToolArguments.decode(ToolCall(self.id, self.name, self.input), self.input) @@ -344,54 +471,60 @@ class AssistantMessage(WireModel): class AssistantRecord(ClaudeRecord, TimedRecord): - wire_tag = 'assistant' + wire_tag = "assistant" message: AssistantMessage def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return tuple(update for block in self.message.content for update in block.updates(self, session)) + return tuple( + update for block in self.message.content for update in block.updates(self, session) + ) class ResultPart(TaggedWire, ABC): __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) @abstractmethod def result_text(self) -> str: ... class ResultText(TextContent, ResultPart): - wire_tag = 'text' + wire_tag = "text" class ResultImage(ImageContent, ResultPart): - wire_tag = 'image' + wire_tag = "image" class ResultToolReference(ResultPart): - wire_tag = 'tool_reference' + wire_tag = "tool_reference" tool_name: str def result_text(self) -> str: - return f'Tool available: {self.tool_name}' + return f"Tool available: {self.tool_name}" class KindHeader(WireHeader): - type: str = Field(alias='kind') + type: str = Field(alias="kind") class PromptOrigin(TaggedWire, ABC): """Who wrote a user record. Only a person's prompt is shown live as input; another agent's message is conversation history the live view hides.""" + __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) header = KindHeader - def prompt(self, record: TimedRecord, text: str, - session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: + def prompt( + self, record: TimedRecord, text: str, session: ClaudeSession + ) -> tuple[RuntimeUpdate, ...]: return () def request(self, text: str) -> str: - return '' + return "" class Prompting(PromptOrigin, ABC): @@ -404,96 +537,118 @@ def request(self, text: str) -> str: class AgentPrompt(Prompting): """Input another agent wrote: the commanding session, or a peer session.""" - def prompt(self, record: TimedRecord, text: str, - session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return (HiddenItem((UserTranscript(text, timestamp=record.timestamp, position=record.position),)),) + def prompt( + self, record: TimedRecord, text: str, session: ClaudeSession + ) -> tuple[RuntimeUpdate, ...]: + return ( + HiddenItem( + (UserTranscript(text, timestamp=record.timestamp, position=record.position),) + ), + ) class HumanPrompt(Prompting): - wire_tag = 'human' + wire_tag = "human" - def prompt(self, record: TimedRecord, text: str, - session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return (TurnStarted(), *session.finished( - record.uuid, (UserTranscript(text, timestamp=record.timestamp, position=record.position),))) + def prompt( + self, record: TimedRecord, text: str, session: ClaudeSession + ) -> tuple[RuntimeUpdate, ...]: + return ( + TurnStarted(), + *session.finished( + record.uuid, + (UserTranscript(text, timestamp=record.timestamp, position=record.position),), + ), + ) class TaskNotification(PromptOrigin): """Its specialized user record owns task-envelope decoding.""" - wire_tag = 'task-notification' - kind: Literal['task-notification'] + + wire_tag = "task-notification" + kind: Literal["task-notification"] class PeerMessage(AgentPrompt): - wire_tag = 'peer' + wire_tag = "peer" class CoordinatorMessage(AgentPrompt): """The session that started this subagent, writing to it while it works.""" - wire_tag = 'coordinator' + + wire_tag = "coordinator" class AutoContinuation(PromptOrigin): - wire_tag = 'auto-continuation' + wire_tag = "auto-continuation" class ToolSupplied(Prompting): """Text Claude records without a sender: a subagent's task, or a local command's echo. It starts a turn; the live view does not show it as the person's prompt.""" - def prompt(self, record: TimedRecord, text: str, - session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return (TurnStarted(), HiddenItem((UserTranscript(text, timestamp=record.timestamp, position=record.position),))) + def prompt( + self, record: TimedRecord, text: str, session: ClaudeSession + ) -> tuple[RuntimeUpdate, ...]: + return ( + TurnStarted(), + HiddenItem( + (UserTranscript(text, timestamp=record.timestamp, position=record.position),) + ), + ) class UserBlock(TaggedWire, ABC): __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return (TurnStarted(),) def prompt_text(self) -> str: - return '' + return "" class UserText(UserBlock, TextContent): - wire_tag = 'text' + wire_tag = "text" def prompt_text(self) -> str: return self.text def updates(self, record: UserRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - if self.text.startswith('[Request interrupted by user'): + if self.text.startswith("[Request interrupted by user"): return (TurnEnded(),) return record.origin.prompt(record, self.text, session) class UserImage(UserBlock, ImageContent): - wire_tag = 'image' + wire_tag = "image" class ContentBlocks(WireModel): """Claude's string shorthand is decoded once for both content families.""" - @field_validator('content', mode='before', check_fields=False) + @field_validator("content", mode="before", check_fields=False) @classmethod def blocks(cls, value: str | list) -> list: - return [{'type': 'text', 'text': value}] if isinstance(value, str) else value + return [{"type": "text", "text": value}] if isinstance(value, str) else value class TaskEnvelope(WireModel, ABC): @classmethod def from_text(cls, text: str) -> TaskEnvelope: - if not text.startswith(''): + if not text.startswith(""): return ForeignTaskNotification(text=text) # Claude Code appends sibling blocks (system-reminder) after the envelope in the same record. - envelope, closing, _ = text.partition('') + envelope, closing, _ = text.partition("") root = ElementTree.fromstring(envelope + closing) # Malformed envelope fails at ingress. - event = root.find('event') + event = root.find("event") if event is None: return ForeignTaskNotification(text=text) - return MonitorTaskNotification(frames=ClaudeMonitorFrame.decode_event(''.join(event.itertext()))) + return MonitorTaskNotification( + frames=ClaudeMonitorFrame.decode_event("".join(event.itertext())) + ) @abstractmethod def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: ... @@ -510,16 +665,18 @@ class MonitorTaskNotification(TaskEnvelope): frames: tuple[MonitorFrame, ...] def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: - return tuple(MonitorObservation(frame, - ObservedMonitor(record.uuid, record.timestamp)) - for frame in self.frames) + return tuple( + MonitorObservation(frame, ObservedMonitor(record.uuid, record.timestamp)) + for frame in self.frames + ) class TaskContent(WireModel): """Both external task-record shapes share envelope decoding and observations.""" + text: TaskEnvelope - @field_validator('text', mode='before') + @field_validator("text", mode="before") @classmethod def envelope(cls, value: str) -> TaskEnvelope: return TaskEnvelope.from_text(value) @@ -529,7 +686,7 @@ def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: class TaskText(TaskContent): - type: Literal['text'] = 'text' + type: Literal["text"] = "text" class TaskMessage(ContentBlocks): @@ -540,13 +697,18 @@ def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: class ToolResult(UserBlock, ContentBlocks): - wire_tag = 'tool_result' + wire_tag = "tool_result" tool_use_id: str content: list[ResultPart.declared_union()] is_error: bool = False def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return session.tool_result(self.tool_use_id, '\n'.join(part.result_text() for part in self.content), not self.is_error, record.position) + return session.tool_result( + self.tool_use_id, + "\n".join(part.result_text() for part in self.content), + not self.is_error, + record.position, + ) class UserMessage(ContentBlocks): @@ -554,18 +716,20 @@ class UserMessage(ContentBlocks): class UserRecord(ClaudeRecord, TimedRecord): - wire_tag = 'user' + wire_tag = "user" promptId: str message: UserMessage origin: PromptOrigin.declared_union() = Field(default_factory=ToolSupplied) def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: session.turn = KnownTurn(self.promptId) - return tuple(update for block in self.message.content for update in block.updates(self, session)) + return tuple( + update for block in self.message.content for update in block.updates(self, session) + ) @property def text(self) -> str: - return '\n'.join(text for block in self.message.content if (text := block.prompt_text())) + return "\n".join(text for block in self.message.content if (text := block.prompt_text())) def request(self) -> str: return self.origin.request(self.text) @@ -573,18 +737,19 @@ def request(self) -> str: class CompactSummary(UserRecord): """Claude's summary of the conversation before its latest compaction.""" - wire_tag = 'compact-summary' + + wire_tag = "compact-summary" def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: session.turn = KnownTurn(self.promptId) return (TurnStarted(), Summarized(self.text)) def request(self) -> str: - return '' + return "" class TaskUserRecord(ClaudeRecord, TimedRecord): - wire_tag = 'task-user' + wire_tag = "task-user" origin: TaskNotification message: TaskMessage @@ -593,12 +758,13 @@ def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: class SubtypeHeader(WireHeader): - type: str = Field(alias='subtype') + type: str = Field(alias="subtype") class SystemEvent(TaggedWire, ABC): __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) header = SubtypeHeader def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: @@ -609,60 +775,67 @@ def begins_history(self) -> bool: class TurnDuration(SystemEvent): - wire_tag = 'turn_duration' + wire_tag = "turn_duration" def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return (TurnEnded(),) class CompactBoundary(SystemEvent, TimedRecord): - wire_tag = 'compact_boundary' + wire_tag = "compact_boundary" def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return session.finished(self.uuid, (NoticeTranscript('Context compacted', timestamp=self.timestamp, position=self.position),)) + return session.finished( + self.uuid, + ( + NoticeTranscript( + "Context compacted", timestamp=self.timestamp, position=self.position + ), + ), + ) def begins_history(self) -> bool: return True class StopHookSummary(SystemEvent): - wire_tag = 'stop_hook_summary' + wire_tag = "stop_hook_summary" class Informational(SystemEvent): - wire_tag = 'informational' + wire_tag = "informational" class ScheduledTaskFire(SystemEvent): - wire_tag = 'scheduled_task_fire' + wire_tag = "scheduled_task_fire" class LocalCommand(SystemEvent): - wire_tag = 'local_command' + wire_tag = "local_command" class AwaySummary(SystemEvent): - wire_tag = 'away_summary' + wire_tag = "away_summary" class AgentsKilled(SystemEvent): - wire_tag = 'agents_killed' + wire_tag = "agents_killed" class ModelRefusalNoFallback(SystemEvent): - wire_tag = 'model_refusal_no_fallback' + wire_tag = "model_refusal_no_fallback" class SystemRecord(ClaudeRecord): - wire_tag = 'system' + wire_tag = "system" subtype: SystemEvent.declared_union() - @model_validator(mode='before') + @model_validator(mode="before") @classmethod def decode_subtype(cls, value: dict) -> dict: # Claude places both discriminators on one envelope. Decode that shape # once; its declared subtype owns the behavior and any retained fields. - return {'subtype': value} + return {"subtype": value} def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return self.subtype.updates(session) @@ -676,18 +849,21 @@ class MetadataRecord(ClaudeRecord, ABC): class AttachmentHeader(WireHeader): - @model_validator(mode='before') + @model_validator(mode="before") @classmethod def task_notification(cls, value: dict) -> dict: - if (value.get('type') == QueuedTaskNotification.wire_tag - and value.get('origin', {}).get('kind') == TaskNotification.wire_tag): + if ( + value.get("type") == QueuedTaskNotification.wire_tag + and value.get("origin", {}).get("kind") == TaskNotification.wire_tag + ): return value - return {'type': ForeignAttachment.wire_tag} + return {"type": ForeignAttachment.wire_tag} class AttachmentPayload(TaggedWire, ABC): __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) header = AttachmentHeader @abstractmethod @@ -695,20 +871,20 @@ def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: ... class QueuedTaskNotification(TaskContent, AttachmentPayload): - wire_tag = 'queued_command' - text: TaskEnvelope = Field(alias='prompt') + wire_tag = "queued_command" + text: TaskEnvelope = Field(alias="prompt") origin: TaskNotification class ForeignAttachment(AttachmentPayload): - wire_tag = 'foreign' + wire_tag = "foreign" def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: return () class Attachment(ClaudeRecord, TimedRecord): - wire_tag = 'attachment' + wire_tag = "attachment" attachment: AttachmentPayload.declared_union() def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: @@ -716,82 +892,93 @@ def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: class QueueOperation(MetadataRecord): - wire_tag = 'queue-operation' + wire_tag = "queue-operation" class Mode(MetadataRecord): - wire_tag = 'mode' + wire_tag = "mode" class PermissionMode(MetadataRecord): - wire_tag = 'permission-mode' + wire_tag = "permission-mode" class AtisLatch(MetadataRecord): - wire_tag = 'atis-latch' + wire_tag = "atis-latch" class LastPrompt(MetadataRecord): - wire_tag = 'last-prompt' + wire_tag = "last-prompt" class AiTitle(MetadataRecord): - wire_tag = 'ai-title' + wire_tag = "ai-title" class PrLink(MetadataRecord): - wire_tag = 'pr-link' + wire_tag = "pr-link" class FileHistorySnapshot(MetadataRecord): - wire_tag = 'file-history-snapshot' + wire_tag = "file-history-snapshot" class FileHistoryDelta(MetadataRecord): - wire_tag = 'file-history-delta' + wire_tag = "file-history-delta" class CostState(MetadataRecord): - wire_tag = 'cost-state' + wire_tag = "cost-state" class ApiRequest(MetadataRecord): """The parameters of one model request the harness sent.""" - wire_tag = 'api-request' + + wire_tag = "api-request" class ApiRequestShape(MetadataRecord): """The system prompt and tools a request was sent with, stored once by hash.""" - wire_tag = 'api-request-shape' + + wire_tag = "api-request-shape" class ApiRequestBlob(MetadataRecord): """One message a request carried, stored once by hash.""" - wire_tag = 'api-request-blob' + + wire_tag = "api-request-blob" class LocalCommandOutput(MetadataRecord): """A terminal-only slash command and its output.""" - wire_tag = 'local-command' + + wire_tag = "local-command" class HarnessNote(MetadataRecord): """Text the harness adds for the model without a sender: reminders, skill bodies, image notes.""" - wire_tag = 'harness-note' + + wire_tag = "harness-note" class ForkContextReference(MetadataRecord): """A forked subagent starts from its parent's context, stored in the parent's transcript.""" - wire_tag = 'fork-context-ref' + + wire_tag = "fork-context-ref" parentSessionId: str def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return (HistoryNotice(f'Context inherited from session {self.parentSessionId} is not imported.'),) + return ( + HistoryNotice( + f"Context inherited from session {self.parentSessionId} is not imported." + ), + ) class CustomTitle(MetadataRecord): """The name a person gave the session.""" - wire_tag = 'custom-title' + + wire_tag = "custom-title" customTitle: str def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: @@ -800,7 +987,8 @@ def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: class AgentName(MetadataRecord): """The name the session runs under as an agent.""" - wire_tag = 'agent-name' + + wire_tag = "agent-name" agentName: str def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: @@ -809,11 +997,14 @@ def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: class ContinuedIn(MetadataRecord): """The conversation went on in another session.""" - wire_tag = 'continued-in' + + wire_tag = "continued-in" continuedInSessionId: str def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return (HistoryNotice(f'The conversation continued in session {self.continuedInSessionId}.'),) + return ( + HistoryNotice(f"The conversation continued in session {self.continuedInSessionId}."), + ) CLAUDE_RECORD = TypeAdapter(ClaudeRecord.declared_union()) diff --git a/src/agent_comms/cli_bridges/monitor.py b/src/agent_comms/cli_bridges/monitor.py new file mode 100644 index 0000000000..421d826877 --- /dev/null +++ b/src/agent_comms/cli_bridges/monitor.py @@ -0,0 +1,435 @@ +"""Session-owned endpoint and replaceable, explicitly modeled input readers.""" + +from __future__ import annotations + +import asyncio +import secrets +import socket +import sys +from abc import ABC, abstractmethod +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING +from uuid import UUID + +from pydantic import ConfigDict, TypeAdapter, ValidationError + +from agent_comms.agent_backend import InputId +from agent_comms.child_process import ProcessIdentity +from agent_comms.cli_protocol import ( + DeliveredOutcome, + Outcome, + TaggedWire, + UnconfirmedOutcome, + WireModel, +) +from agent_comms.messages import Message +from agent_comms_hooks.watch import WatchClaim + +if TYPE_CHECKING: + from .shared import IncomingFrame, LineMonitorBridge, MonitorProbe + + +class PeerOutcome(ABC): + @abstractmethod + def write(self, transport: asyncio.Transport, data: bytes) -> bool: ... + + +class ConnectedPeer(PeerOutcome): + def write(self, transport: asyncio.Transport, data: bytes) -> bool: + transport.write(data) + return True + + +class GonePeer(PeerOutcome): + def write(self, transport: asyncio.Transport, data: bytes) -> bool: + return False + + +class MonitorPeer(asyncio.Protocol): + """Own transport loss and closing once; callers consume outcomes, not errors.""" + + def __init__(self, service: MonitorService): + self.service = service + self.reader = asyncio.StreamReader() + self.outcome: PeerOutcome = ConnectedPeer() + self.writable = asyncio.Event() + self.writable.set() + self.closed = asyncio.Event() + + def connection_made(self, transport: asyncio.BaseTransport) -> None: + if not isinstance(transport, asyncio.Transport): + raise TypeError("Monitor requires a byte-stream transport") + self.transport = transport + self.service.clients.create_task(self.service.accept(self)) + + def data_received(self, data: bytes) -> None: + self.reader.feed_data(data) + + def connection_lost(self, exc: object) -> None: + self.outcome = GonePeer() + self.reader.feed_eof() + self.writable.set() + self.closed.set() + + def pause_writing(self) -> None: + self.writable.clear() + + def resume_writing(self) -> None: + self.writable.set() + + async def write(self, line: str) -> bool: + await self.writable.wait() + return self.outcome.write(self.transport, (line + "\n").encode()) + + async def close(self) -> None: + self.transport.close() + await self.closed.wait() + + +class ReaderState: + """One bridge reader state, including absence and unproven attachment.""" + + async def prepare_delivery(self, bridge: LineMonitorBridge) -> ReadyReader: + async with bridge.reader_changed: + await bridge.reader_changed.wait_for(lambda: bridge.reader is not self) + return await bridge.reader.prepare_delivery(bridge) + + def frame(self, bridge: LineMonitorBridge, message: Message) -> IncomingFrame: + raise ConnectionError("No proven input reader is attached") + + async def send( + self, bridge: LineMonitorBridge, identity: InputId, payloads: tuple[bytes, ...] + ) -> type[Outcome]: + reader = await self.prepare_delivery(bridge) + return await reader.send(bridge, identity, payloads) + + async def start(self, bridge: LineMonitorBridge) -> None: + """Already proven input transports need no Monitor challenge.""" + + async def receive(self) -> None: + raise RuntimeError("An absent reader cannot receive input acknowledgments") + + async def detached(self) -> None: + """A missing reader has no pending input or window.""" + + async def confirm_probe(self, bridge: LineMonitorBridge, probe: MonitorProbe) -> None: + """A missing or already-proven reader cannot accept a new probe.""" + + async def part_observed(self, message_id: str, number: int) -> None: + """An unattached reader has no flow-control window to release.""" + + async def attach( + self, service: MonitorService, peer: MonitorPeer, request: AttachReader + ) -> SessionReply: + return SessionRefused.model_validate(dict(error="This session already has an input reader")) + + +class NoReader(ReaderState): + async def attach( + self, service: MonitorService, peer: MonitorPeer, request: AttachReader + ) -> SessionReply: + async with service.bridge.reader_changed: + if service.bridge.reader is not self: + return SessionRefused.model_validate(dict(error="Another reader attached first")) + reader = request.open_reader(service, peer) + service.bridge.reader = reader + try: + if await peer.write(SessionReady().model_dump_json()): + await reader.start(service.bridge) + async with service.bridge.reader_changed: + service.bridge.reader_changed.notify_all() + await reader.receive() + return PeerGone() + finally: + async with service.bridge.reader_changed: + service.bridge.reader = NoReader() + service.bridge.reader_changed.notify_all() + await reader.detached() + print("Input detached; bridge remains connected.", file=sys.stderr, flush=True) + + +class ReadyReader(ReaderState): + async def prepare_delivery(self, bridge: LineMonitorBridge) -> ReadyReader: + return self + + +@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) -> bool: + size = len((line + "\n").encode("utf-16-le")) + async with self.window_changed: + await self.window_changed.wait_for( + lambda: ( + self.peer.closed.is_set() + or not self.pending_parts + or sum(self.pending_parts.values()) + size <= window_bytes + ) + ) + self.pending_parts[message_id, number] = size + return 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 start(self, bridge: LineMonitorBridge) -> None: + await self.attachment.peer.write(self.attachment.probe.model_dump_json()) + + async def receive(self) -> None: + await self.attachment.peer.reader.read() + + async def detached(self) -> None: + async with self.attachment.window_changed: + self.attachment.window_changed.notify_all() + + 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(ReadyReader, UnreadyReader): + def frame(self, bridge: LineMonitorBridge, message: Message) -> IncomingFrame: + return bridge.incoming_frame(message) + + async def confirm_probe(self, bridge: LineMonitorBridge, probe: MonitorProbe) -> None: + """A proven attachment ignores overlapping transcript probe observations.""" + + async def part_observed(self, message_id: str, number: int) -> None: + await self.attachment.observed(message_id, number) + + async def send( + self, bridge: LineMonitorBridge, identity: InputId, payloads: tuple[bytes, ...] + ) -> type[Outcome]: + return await bridge.deliver_to(self.attachment, identity, payloads) + + +class InputAcknowledgment(WireModel): + model_config = ConfigDict(frozen=True) + identity: UUID + + @property + def input_id(self) -> InputId: + return InputId(str(self.identity)) + + +@dataclass +class InputReader(ReadyReader): + peer: MonitorPeer + pending: dict[InputId, asyncio.Future[type[Outcome]]] = field(default_factory=dict) + + def frame(self, bridge: LineMonitorBridge, message: Message) -> IncomingFrame: + return bridge.input_frame(message) + + async def send( + self, bridge: LineMonitorBridge, identity: InputId, payloads: tuple[bytes, ...] + ) -> type[Outcome]: + answer: asyncio.Future[type[Outcome]] = asyncio.get_running_loop().create_future() + self.pending[identity] = answer + try: + for payload in payloads: + if not await self.peer.write(payload.decode().removesuffix("\n")): + return UnconfirmedOutcome + return await answer + finally: + del self.pending[identity] + + async def receive(self) -> None: + async for line in self.peer.reader: + answer = InputAcknowledgment.model_validate_json(line) + if answer.input_id in self.pending: + pending = self.pending[answer.input_id] + if not pending.done(): + pending.set_result(DeliveredOutcome) + + async def detached(self) -> None: + for answer in self.pending.values(): + if not answer.done(): + answer.set_result(UnconfirmedOutcome) + + +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: + """Refused requests cannot stop the service.""" + + async def attach( + self, service: MonitorService, peer: MonitorPeer, request: AttachReader + ) -> SessionReply: + return self + + +class SessionReady(SessionReply): + wire_tag = "ready" + + def end_session(self, service: MonitorService) -> None: + asyncio.get_running_loop().call_soon(service.stop) + + async def attach( + self, service: MonitorService, peer: MonitorPeer, request: AttachReader + ) -> SessionReply: + return await service.bridge.reader.attach(service, peer, request) + + +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: + """A disconnected peer receives no reply.""" + + +REPLIES = TypeAdapter(SessionReply.declared_union()) + + +class SessionRequest(TaggedWire): + __registry_config__ = TaggedWire.family_registry() + + def exchange(self, claim: WatchClaim) -> None: + """Control requests never inject input or retry an emitted event.""" + 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 AttachReader(SessionRequest): + @abstractmethod + def open_reader(self, service: MonitorService, peer: MonitorPeer) -> ReaderState: ... + + +class AttachMonitor(AttachReader): + wire_tag = "attach" + + def open_reader(self, service: MonitorService, peer: MonitorPeer) -> ReaderState: + return UnreadyReader(service.bridge.new_attachment(peer, secrets.token_hex(24))) + + async def handle(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: + return await service.bridge.reader.attach(service, peer, self) + + +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 AttachInput(VerifySession, AttachReader): + wire_tag = "input" + + def open_reader(self, service: MonitorService, peer: MonitorPeer) -> ReaderState: + return InputReader(peer) + + async def handle(self, service: MonitorService, peer: MonitorPeer) -> SessionReply: + verified = await super().handle(service, peer) + return await verified.attach(service, peer, self) + + +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] + clients: asyncio.TaskGroup = field(default_factory=asyncio.TaskGroup) + + async def accept(self, peer: MonitorPeer) -> None: + try: + line = await peer.reader.readline() + reply: SessionReply + try: + request = REQUESTS.validate_json(line) + 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() + + async def run(self) -> None: + path = WatchClaim(self.bridge.binding.thread).socket_path + path.unlink(missing_ok=True) + try: + async with self.clients: + server = await asyncio.get_running_loop().create_unix_server( + lambda: MonitorPeer(self), 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)) + try: + writer.write((AttachMonitor().model_dump_json() + "\n").encode()) + await writer.drain() + REPLIES.validate_json(await reader.readline()).require() + async for line in reader: + sys.stdout.write(line.decode()) + sys.stdout.flush() + finally: + writer.close() + await writer.wait_closed() diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py index e48dbc3430..be0862e6c7 100644 --- a/src/agent_comms/cli_bridges/shared.py +++ b/src/agent_comms/cli_bridges/shared.py @@ -1,24 +1,26 @@ """Shared Core inbox delivery and process lifetime for external CLI adapters.""" + from __future__ import annotations import argparse import asyncio import json import os -import secrets import signal import sys import traceback from abc import ABC, abstractmethod +from collections.abc import Awaitable, Callable from contextlib import ExitStack, closing, contextmanager 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 metaclass_registry import RegistryConfig from pydantic import Field, TypeAdapter, model_validator +from agent_comms.agent_backend import InputId from agent_comms.agent_runtime import DeliversInput, ExternalRuntime, PublishesTranscript from agent_comms.child_process import AttachedChild, ProcessIdentity, StreamingChildStdio from agent_comms.cli_protocol import ( @@ -35,41 +37,55 @@ from agent_comms.coordination_database import CoordinationStore from agent_comms.coordination_errors import CoordinationError from agent_comms.declared_family import DeclaredFamily -from agent_comms.field_codec import FieldCodec +from agent_comms.field_codec import FieldCodec, PathText from agent_comms.launch_runtime import LaunchRuntime from agent_comms.message_reference import MessageReference from agent_comms.messages import InfoKind, Message, MessageKind from agent_comms.read_basis import ExplicitWatchAcknowledgment +from agent_comms.registration import RegisteredInput, Registration from agent_comms.runtime_updates import RuntimeUpdate, ThreadOutput 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, +) + @dataclass(frozen=True) class Binding(TypedTable): root: str - bus: str = field(metadata={'wire_name': 'bus_thread'}) - thread: str = field(metadata={'wire_name': 'session'}) + bus: str = field(metadata={"wire_name": "bus_thread"}) + thread: str = field(metadata={"wire_name": "session"}) @classmethod def add_arguments(cls, parser: argparse.ArgumentParser) -> None: - group = parser.add_argument_group('Core/CLI binding (all or none)') + group = parser.add_argument_group("Core/CLI binding (all or none)") for _, key in FieldCodec.record_fields(cls): - group.add_argument('--' + key.replace('_', '-')) + group.add_argument("--" + key.replace("_", "-")) @classmethod def from_arguments(cls, arguments: argparse.Namespace) -> Binding | None: values = {key: vars(arguments)[key] for _, key in FieldCodec.record_fields(cls) if vars(arguments)[key] is not None} - return FieldCodec.decode(cls, {cls.family_discriminator: cls.declared_name, **values}) if values else None + return ( + FieldCodec.decode(cls, {cls.family_discriminator: cls.declared_name, **values}) + if values + else None + ) @classmethod @contextmanager def open(cls, state: Path): state.mkdir(parents=True, exist_ok=True, mode=0o700) - with closing(CoordinationStore.writer(state / 'delivery.sqlite3')) as db: - db.execute('PRAGMA synchronous=FULL') + with closing(CoordinationStore.writer(state / "delivery.sqlite3")) as db: + db.execute("PRAGMA synchronous=FULL") for row in (cls, DeliveryRow, ObservedPart): row.create(db, if_not_exists=True) db.commit() @@ -79,16 +95,16 @@ def open(cls, state: Path): def bind(cls, db: CoordinationStore, requested: Binding | None) -> Binding: with db: if requested is not None: - db.execute('BEGIN IMMEDIATE') + db.execute("BEGIN IMMEDIATE") rows = cls.select(db) if not rows and requested is not None: requested.insert(db) return requested if len(rows) != 1: - raise ValueError('Supply one complete Core/CLI binding for this state directory') + raise ValueError("Supply one complete Core/CLI binding for this state directory") current = rows[0] if requested is not None and requested != current: - raise ValueError('This state directory belongs to another Core/CLI conversation') + raise ValueError("This state directory belongs to another Core/CLI conversation") return current @@ -102,7 +118,7 @@ class Options: def parse(cls, description: str | None, state: Path, bridge_type: type[CliBridge]) -> Self: common = argparse.ArgumentParser(add_help=False) Binding.add_arguments(common) - common.add_argument('--state', type=Path, default=state) + common.add_argument("--state", type=Path, default=state) bridge_type.add_arguments(common) parser = argparse.ArgumentParser(description=description) commands = parser.add_subparsers(required=True) @@ -113,15 +129,25 @@ def parse(cls, description: str | None, state: Path, bridge_type: type[CliBridge arguments = parser.parse_args() arguments.operation = arguments.operation.from_arguments(arguments) arguments.binding = arguments.operation.binding(arguments) - return cls(**{declared.name: vars(arguments)[key] - for declared, key in FieldCodec.record_fields(cls)}) + return cls( + **{ + declared.name: vars(arguments)[key] + for declared, key in FieldCodec.record_fields(cls) + } + ) + @dataclass(frozen=True) class MonitorEvidence(DeclaredFamily): """Unsent and terminal observations cannot accept another delivery.""" - async def confirm(self, bridge: LineMonitorBridge, delivery: DeliveryRow, - notification: MonitorPart, record: ObservedMonitor) -> None: + async def confirm( + self, + bridge: LineMonitorBridge, + delivery: DeliveryRow, + notification: MonitorPart, + record: ObservedMonitor, + ) -> None: pass @@ -134,24 +160,43 @@ class UnrecordedMonitor(MonitorEvidence): class ReceivingMonitor(MonitorEvidence): total: int - async def confirm(self, bridge: LineMonitorBridge, delivery: DeliveryRow, - notification: MonitorPart, record: ObservedMonitor) -> None: + async def confirm( + self, + bridge: LineMonitorBridge, + delivery: DeliveryRow, + notification: MonitorPart, + record: ObservedMonitor, + ) -> None: if notification.total != self.total: return - part = ObservedPart(delivery.id, notification.number, notification.sender, - notification.body, record.record_id, record.timestamp) + part = ObservedPart( + delivery.id, + notification.number, + notification.sender, + notification.body, + record.record_id, + record.timestamp, + ) with bridge.db: part.insert_if_absent(bridge.db) await bridge.part_observed(delivery.id, notification.number) if ObservedPart.count(bridge.db, id=delivery.id) != self.total: return - parts = ObservedPart.select(bridge.db, id=delivery.id, order_by=('number',)) + parts = ObservedPart.select(bridge.db, id=delivery.id, order_by=("number",)) message = delivery.message - if (''.join(part.body for part in parts) != message.body - or any(part.sender != message.sender for part in parts)): + if "".join(part.body for part in parts) != message.body or any( + part.sender != message.sender for part in parts + ): with bridge.db: - replace(delivery, monitor=MismatchedMonitor('Observed parts do not equal the original message')).upsert(bridge.db) - print(f'Monitor parts mismatch; message remains pending: {delivery.id}', file=sys.stderr, flush=True) + replace( + delivery, + monitor=MismatchedMonitor("Observed parts do not equal the original message"), + ).upsert(bridge.db) + print( + f"Monitor parts mismatch; message remains pending: {delivery.id}", + file=sys.stderr, + flush=True, + ) return await replace(delivery, monitor=record).finish(bridge, DeliveredOutcome) @@ -169,8 +214,8 @@ class ObservedMonitor(MonitorEvidence): @dataclass(frozen=True) class ObservedPart(TypedTable): - id: str = field(metadata={'sql': Column(primary_key=True)}) - number: int = field(metadata={'sql': Column(primary_key=True)}) + id: str = field(metadata={"sql": Column(primary_key=True)}) + number: int = field(metadata={"sql": Column(primary_key=True)}) sender: str body: str record_id: str @@ -184,23 +229,27 @@ class DeliveryRow(TypedTable): @classmethod def schema_objects(cls, *, if_not_exists: bool = False): objects = super().schema_objects(if_not_exists=if_not_exists) - existence = 'IF NOT EXISTS ' if if_not_exists else '' - objects[f'{cls.declared_name}_sequence'] = ( + existence = "IF NOT EXISTS " if if_not_exists else "" + objects[f"{cls.declared_name}_sequence"] = ( f'CREATE INDEX {existence}"{cls.declared_name}_sequence" ' - f'ON "{cls.declared_name}" ({cls.sequence_sql})') + f'ON "{cls.declared_name}" ({cls.sequence_sql})' + ) return objects @classmethod def resume_after(cls, db: CoordinationStore) -> MessageReference | None: """The existing journal owns attempted progress, including uncertain emissions.""" - latest = cls.read(db.execute( - f'SELECT {cls._column_list(tuple(cls.column_projection()))} FROM "{cls.declared_name}" ' - f'ORDER BY {cls.sequence_sql} DESC LIMIT 1')) + latest = cls.read( + db.execute( + f'SELECT {cls._column_list(tuple(cls.column_projection()))} FROM "{cls.declared_name}" ' + f"ORDER BY {cls.sequence_sql} DESC LIMIT 1" + ) + ) return latest[0].message.reference if latest else None - id: str = field(metadata={'sql': Column(primary_key=True)}) + id: str = field(metadata={"sql": Column(primary_key=True)}) payload: str - outcome: type[Outcome] = field(metadata={'sql': Column(index=True)}) + outcome: type[Outcome] = field(metadata={"sql": Column(index=True)}) monitor: MonitorEvidence = field(default_factory=UnrecordedMonitor) @cached_property @@ -208,12 +257,14 @@ def message(self) -> Message: return 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) -BridgeOptions = TypeVar('BridgeOptions', bound=Options) -OperationBridge = TypeVar('OperationBridge', bound='CliBridge') + +BridgeOptions = TypeVar("BridgeOptions", bound=Options) +OperationBridge = TypeVar("OperationBridge", bound="CliBridge") @dataclass(frozen=True) @@ -229,11 +280,16 @@ def add_arguments(cls, parser: argparse.ArgumentParser) -> None: @classmethod def from_arguments(cls, arguments: argparse.Namespace) -> Operation: - return FieldCodec.decode(cls, {key: vars(arguments)[key] for _, key in FieldCodec.record_fields(cls)}) + return FieldCodec.decode( + cls, {key: vars(arguments)[key] for _, key in FieldCodec.record_fields(cls)} + ) @abstractmethod async def run(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.""" @@ -253,19 +309,22 @@ def offered(cls, bridge: type[CliBridge]) -> tuple[type[Operation], ...]: @dataclass(frozen=True) class Send(Operation): - target: str = field(metadata={'wire_name': 'to'}) + target: str = field(metadata={"wire_name": "to"}) body: str - kind: type[MessageKind] = field(metadata={'wire_name': 'type'}) + kind: type[MessageKind] = field(metadata={"wire_name": "type"}) @classmethod def add_arguments(cls, parser: argparse.ArgumentParser) -> None: - parser.add_argument('--to', required=True) - parser.add_argument('--body', required=True) - parser.add_argument('--type', default=FieldCodec.encode(InfoKind), choices=MessageKind.sendable_names()) + parser.add_argument("--to", required=True) + parser.add_argument("--body", required=True) + parser.add_argument( + "--type", default=FieldCodec.encode(InfoKind), choices=MessageKind.sendable_names() + ) async def run(self, bridge: CliBridge) -> None: - result = await asyncio.to_thread(bridge.comms.messaging.send, bridge.binding.bus, - self.target, self.body, self.kind) + result = await asyncio.to_thread( + bridge.comms.messaging.send, bridge.binding.bus, self.target, self.body, self.kind + ) await bridge.show_sent(result) @@ -279,21 +338,7 @@ def start(self, bridge: OperationBridge, tasks: asyncio.TaskGroup) -> None: tasks.create_task(bridge.watch()) async def run(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): @@ -301,20 +346,80 @@ class Watch(Inbox): def serves(cls, bridge: type[CliBridge]) -> bool: return issubclass(bridge, PublishesTranscript) + async def run(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) +class Serve(Watch): + owner: ProcessIdentity + transcript: Annotated[Path, PathText] + + @classmethod + def serves(cls, bridge: type[CliBridge]) -> bool: + return issubclass(bridge, LineMonitorBridge) + + @classmethod + def add_arguments(cls, parser: argparse.ArgumentParser) -> None: + parser.add_argument("--owner", type=json.loads, required=True) + parser.add_argument("--transcript", required=True) + + def transcript_source(self) -> Path: + return self.transcript + + async def run(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 run(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: Message + async def turn_start( + self, send: Callable[[InputId, tuple[bytes, ...]], Awaitable[type[Outcome]]] + ) -> type[Outcome]: + return await send(self.input_id, self.payload) + + @property + def input_id(self) -> InputId: + return InputId(self.message.message_id) + + @property + def payload(self) -> tuple[bytes, ...]: + return ((self.text + "\n").encode(),) + @property def text(self) -> str: origin = self.message.sender_role.origin_label - return (f'Comms in · {self.message.sender} ({origin})\n' - f'{self.message.body}\nMessage ID: {self.message.message_id}') + return ( + f"Comms in · {self.message.sender} ({origin})\n" + f"{self.message.body}\nMessage ID: {self.message.message_id}" + ) + class CliBridge(ExternalRuntime, DeliversInput, Generic[BridgeOptions]): """Core delivery and lifetime shared by CLI-specific transports.""" @@ -332,6 +437,27 @@ 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: + registry = await asyncio.to_thread(Registration, Path(self.binding.root) / "registry.json") + with ExitStack() as endpoint: + await self.comms.bus.when_open(lambda: self.register_endpoint(registry, endpoint)) + with WatchClaim(self.binding.thread).held(self.watch_record()): + async with asyncio.TaskGroup() as tasks: + operation.start(self, tasks) + + def register_endpoint(self, registry: Registration, endpoint: ExitStack) -> None: + endpoint.enter_context( + registry.external_process(self.binding.bus, ProcessIdentity.capture(os.getpid())) + ) + + 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.""" @@ -342,8 +468,14 @@ async def report_failure(self, text: str) -> None: def sent_text(self, result: SendResult) -> str: message = result.message - return '\n'.join((f'Comms out · {message.sender} → {message.target}', - message.body, f'Message ID: {message.message_id}', *result.notices)) + return "\n".join( + ( + f"Comms out · {message.sender} → {message.target}", + message.body, + f"Message ID: {message.message_id}", + *result.notices, + ) + ) async def prepare_delivery(self) -> None: """API transports already bind the intended harness on each request.""" @@ -351,54 +483,76 @@ async def prepare_delivery(self) -> None: async def accept(self, message: Message) -> None: conversation = message.type.conversation_message outcome = UnconfirmedOutcome if conversation else FilteredOutcome - delivery = DeliveryRow(message.message_id, json.dumps(message.to_wire(), separators=(',', ':')) + '\n', outcome) + delivery = DeliveryRow( + message.message_id, json.dumps(message.to_wire(), separators=(",", ":")) + "\n", outcome + ) with self.db: first = delivery.insert_if_absent(self.db).rowcount if not first: - delivery, = DeliveryRow.select(self.db, id=message.message_id) + (delivery,) = DeliveryRow.select(self.db, id=message.message_id) await delivery.outcome.acknowledge(self, message) return result = await self.deliver(message) if conversation else outcome - delivery, = DeliveryRow.select(self.db, id=message.message_id) + (delivery,) = DeliveryRow.select(self.db, id=message.message_id) await delivery.finish(self, result) - print(f'{result.declared_name}: {message.message_id} from {message.sender}', file=sys.stderr, flush=True) + print( + f"{result.declared_name}: {message.message_id} from {message.sender}", + file=sys.stderr, + flush=True, + ) async def watch(self) -> None: await self.prepare_delivery() - print(f'Watching {self.binding.bus} → {self.binding.thread}; unconfirmed attempts retained without replay', file=sys.stderr, flush=True) + print( + f"Watching {self.binding.bus} → {self.binding.thread}; unconfirmed attempts retained without replay", + file=sys.stderr, + flush=True, + ) await self.comms.bus.when_open(self.watch_child) async def watch_child(self) -> None: """One child stream; the bus owns waiting and its next attempt after a hold.""" - child = await AttachedChild.start(( - str(LaunchRuntime.running().interpreter.with_name('agent-comms')), - '--root', self.binding.root, 'watch', '--thread', self.binding.bus, - '--after', json.dumps(FieldCodec.encode(DeliveryRow.resume_after(self.db))), - '--acknowledgment', FieldCodec.encode(ExplicitWatchAcknowledgment)), - stdio=StreamingChildStdio(input_enabled=False)) + child = await AttachedChild.start( + ( + str(LaunchRuntime.running().interpreter.with_name("agent-comms")), + "--root", + self.binding.root, + "watch", + "--thread", + self.binding.bus, + "--after", + json.dumps(FieldCodec.encode(DeliveryRow.resume_after(self.db))), + "--acknowledgment", + FieldCodec.encode(ExplicitWatchAcknowledgment), + ), + stdio=StreamingChildStdio(input_enabled=False), + ) try: async for item in WatchLine.stream(child): await item.handle(self, child) - await raise_child_exit(child, '') + await raise_child_exit(child, "") finally: await child.stop() 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.run(self) except* BlockingIOError: raise except* (OSError, ValueError, CoordinationError) as group: - failure = 'Comms bridge stopped:\n' + ''.join( + failure = "Comms bridge stopped:\n" + "".join( traceback.TracebackException.from_exception( - group, max_group_width=sys.maxsize, max_group_depth=sys.maxsize, + group, + max_group_width=sys.maxsize, + max_group_depth=sys.maxsize, ).format() ) print(failure, file=sys.stderr, flush=True) - await self.report_failure(failure + '\nNo message was retried.') + await self.report_failure(failure + "\nNo message was retried.") raise @@ -408,7 +562,8 @@ class TranscriptBridge(PublishesTranscript, CliBridge[BridgeOptions]): async def publish_transcript(self) -> None: comms = self.comms output = ThreadOutput( - await asyncio.to_thread(comms.transcripts.external, self.binding.bus), comms.agents) + await asyncio.to_thread(comms.transcripts.external, self.binding.bus), comms.agents + ) # This subscription can assemble only items whose start it receives. # A prefix from the previous subscription cannot stand for that stream. await comms.bus.when_open(output.transcript.end_turn) @@ -435,37 +590,41 @@ async def read() -> None: pending.append(queue.get_nowait()) await comms.bus.when_open(partial(output.publish_batch, pending)) + class MonitorFrame(TaggedWire, ABC): """Our line protocol inside the harness's task-notification envelope.""" + __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( - registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) + registry_dict={}, key_attribute="wire_tag", skip_if_no_key=True + ) + @abstractmethod async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> None: ... class MonitorProbe(MonitorFrame): - wire_tag = 'probe' - protocol: Literal['agent-comms-monitor'] = 'agent-comms-monitor' - wire_prefix: ClassVar[str] = '{"protocol":' + json.dumps(protocol) + ',' + wire_tag = "probe" + protocol: Literal["agent-comms-monitor"] = "agent-comms-monitor" + wire_prefix: ClassVar[str] = '{"protocol":' + json.dumps(protocol) + "," session: str token: str async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> None: - bridge.confirm_probe(self) + await bridge.confirm_probe(self) class MonitorPart(MonitorFrame): - wire_tag = 'part' + wire_tag = "part" id: str sender: str number: int = Field(ge=1) total: int = Field(ge=1) body: str - @model_validator(mode='after') + @model_validator(mode="after") def valid_number(self): if self.number > self.total: - raise ValueError('Monitor part number exceeds total') + raise ValueError("Monitor part number exceeds total") return self async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> None: @@ -474,7 +633,7 @@ async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> N class ForeignMonitorFrame(MonitorFrame): - wire_tag = 'foreign' + wire_tag = "foreign" text: str async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> None: @@ -500,71 +659,103 @@ 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]] + input_frame: ClassVar[type[IncomingFrame]] window_bytes: ClassVar[int] - PROBE_NOTICE_SECONDS: ClassVar[float] = 15.0 + input: RegisteredInput 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() + + 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) + + def register_endpoint(self, registry: Registration, endpoint: ExitStack) -> None: + self.input = endpoint.enter_context( + registry.external_input( + self.binding.bus, ProcessIdentity.capture(os.getpid()), self.frame, self.send_input + ) + ) + + def frame(self, message: Message) -> IncomingFrame: + return self.reader.frame(self, message) + + async def send_input(self, identity: InputId, payloads: tuple[bytes, ...]) -> type[Outcome]: + return await self.reader.send(self, identity, payloads) + + async def watch_operation(self, operation: Watch) -> None: + await read_monitor(WatchClaim(self.binding.thread)) + + async def accept(self, message: 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() + await super().accept(message) 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. - 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 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: Message) -> type[Outcome]: - parts = self.incoming_frame(message).parts + return await self.input.turn_start(message) + + async def deliver_to( + self, attachment: MonitorAttachment, identity: InputId, payloads: tuple[bytes, ...] + ) -> type[Outcome]: with self.db: - 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) + DeliveryRow.update( + self.db, + where="id=?", + parameters=(identity.value,), + monitor=ReceivingMonitor(len(payloads)), + ) + for number, payload in enumerate(payloads, 1): + if not await attachment.emit( + identity.value, number, payload.decode().removesuffix("\n"), self.window_bytes + ): + print( + f"Monitor detached during {identity.value}; attempt retained without replay.", + file=sys.stderr, + flush=True, + ) + return UnconfirmedOutcome 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: @@ -576,4 +767,4 @@ def run_bridge(bridge_type: type[CliBridge[BridgeOptions]], options: BridgeOptio except* asyncio.CancelledError: pass except* BlockingIOError: - sys.exit('Another watcher owns this conversation') + sys.exit("Another watcher owns this conversation") diff --git a/src/agent_comms/cli_protocol.py b/src/agent_comms/cli_protocol.py index 9b62b5b419..467d3ee889 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 +from pydantic._internal._model_construction import ModelMetaclass +from typing_extensions import TypeForm from .coordination_errors import CoordinationReadUnavailable from .declared_family import DeclaredFamily @@ -32,7 +34,7 @@ class WireModel(BaseModel): model_config = ConfigDict(strict=True) -class RegisteredWireMeta(AutoRegisterMeta, type(WireModel)): +class RegisteredWireMeta(AutoRegisterMeta, ModelMetaclass): """Compose declaration registration with external-field validation.""" @@ -45,6 +47,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: @@ -55,7 +61,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 @@ -92,8 +98,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/registration.py b/src/agent_comms/registration.py index 91dc1ad65a..080b63a1f3 100644 --- a/src/agent_comms/registration.py +++ b/src/agent_comms/registration.py @@ -2,7 +2,14 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence +from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .agent_backend import InputId + from .cli_bridges.shared import IncomingFrame + from .cli_protocol import Outcome + from .messages import Message from contextlib import contextmanager from dataclasses import dataclass, replace from pathlib import Path @@ -14,16 +21,19 @@ from .maintenance_barrier import MaintenanceBarrier from .native_input_owner import RegistryOwner from .registration_change import RegistrationChange -from .registry_document import RegistrySnapshot, UserThreadLookup +from .registry_document import RegistryEntry, RegistrySnapshot, UserThreadLookup from .registry_store import RegistryEdit, RegistryEntryRevision, RegistryStore from .routing import TurnRouting +from .storage_rules import StorageRule from .store_files import SharedLock, _store_lock -from .thread_identity import GenerationCounter, TurnId +from .thread_identity import GenerationCounter, OwnerIdentity, TurnId from .thread_status import ( - ConnectedThreadStatus, NoBridgeThreadStatus, RunningThreadStatus, ThreadStatus, + ConnectedThreadStatus, + NoBridgeThreadStatus, + RunningThreadStatus, + ThreadStatus, ) from .threads import Thread -from .storage_rules import StorageRule from .turn_lease import TurnCompletion, TurnLeaseFence _RUNNING_STATUS = RunningThreadStatus() @@ -63,6 +73,17 @@ class SeveralProjectThreads(ProjectThreadSelection): threads: tuple[Thread, ...] +@dataclass(frozen=True) +class RegisteredInput: + """An external process's input capability, retained for its registration lifetime.""" + + frame: Callable[[Message], IncomingFrame] + send: Callable[[InputId, tuple[bytes, ...]], Awaitable[type[Outcome]]] + + async def turn_start(self, message: Message) -> type[Outcome]: + return await self.frame(message).turn_start(self.send) + + class Registration: """Own registration transactions; never borrow a Comms or facade's state.""" @@ -91,18 +112,55 @@ def external_process(self, name: str, process: ProcessIdentity): if thread.process_alive and thread.process_identity != process: raise RelationViolationError("Another live process owns this CLI endpoint.") bound = replace(thread, process_identity=process) - self._commit_registration(edit, edit.document.prepare_registration( - bound, ConnectedThreadStatus(), new_owner=True)) + self._commit_registration( + edit, + edit.document.prepare_registration(bound, ConnectedThreadStatus(), new_owner=True), + ) owner = edit.document.entry(bound.name).owner_identity try: yield owner.incarnation finally: with self.store.editing() as edit: - entry = edit.document.entry_if_present(owner.incarnation.name) - if entry is not None and entry.owned_by(owner, process): - self._commit_registration(edit, edit.document.prepare_registration( - replace(entry.thread, process_identity=NoProcess()), NoBridgeThreadStatus(), - new_owner=False)) + + def release(entry: RegistryEntry) -> None: + self._commit_registration( + edit, + edit.document.prepare_registration( + replace(entry.thread, process_identity=NoProcess()), + NoBridgeThreadStatus(), + new_owner=False, + ), + ) + + edit.document.external_endpoint(owner, process).when_owned(release) + + @contextmanager + def external_input( + self, + name: str, + process: ProcessIdentity, + frame: Callable[[Message], IncomingFrame], + send: Callable[[InputId, tuple[bytes, ...]], Awaitable[type[Outcome]]], + ) -> Iterator[RegisteredInput]: + """Keep the declared input behavior with the endpoint that can actually invoke it.""" + with self.external_process(name, process): + yield RegisteredInput(frame, send) + + def observe_external_status( + self, + owner: OwnerIdentity, + process: ProcessIdentity, + status: ThreadStatus, + ) -> None: + """An observation changes only the bridge lease it actually witnessed.""" + with self.store.editing() as edit: + + def observe(entry: RegistryEntry) -> None: + entry.thread.execution.require_external() + edit.document.statuses[entry.thread.name] = status.with_bridge() + edit.commit() + + edit.document.external_endpoint(owner, process).when_owned(observe) def attach_native_session(self, original: RegistryOwner, session_file: str) -> RegistryOwner: """Publish an observed native source under its original executable lease. @@ -111,14 +169,20 @@ def attach_native_session(self, original: RegistryOwner, session_file: str) -> R phase is retained from the locked document, not compared to an old view. Native session identity and context proof remain with their producers. """ - with _store_lock(self.store.path.parent / "wire", mode=SharedLock()), self.store.editing() as edit: + with ( + _store_lock(self.store.path.parent / "wire", mode=SharedLock()), + self.store.editing() as edit, + ): change = edit.prepare_native_source(original, session_file) change = self._commit_registration(edit, change) # The document owns the next turn's selection. Publishing an # observed source must not rewrite this authorized turn's settings. - observed = replace(change.thread, - worktree=original.thread.worktree, model=original.thread.model, - thinking_level=original.thread.thinking_level) + observed = replace( + change.thread, + worktree=original.thread.worktree, + model=original.thread.model, + thinking_level=original.thread.thinking_level, + ) return replace(original, thread=observed) def declare(self, thread: Thread, status: ThreadStatus = _RUNNING_STATUS) -> Thread: @@ -151,7 +215,9 @@ def _prepare_declaration(document, channels, thread: Thread, status: ThreadStatu return change def require_declaration_available( - self, thread: Thread, status: ThreadStatus = _RUNNING_STATUS, + self, + thread: Thread, + status: ThreadStatus = _RUNNING_STATUS, ) -> None: """Refuse a known-invalid declaration before its producer performs side effects.""" catalog = ChannelCatalog.at(self.store.path.parent) @@ -167,7 +233,9 @@ def _declare_in(self, edit: RegistryEdit, thread: Thread, status: ThreadStatus) channels.remember_tags(change.thread.tags) return change.thread - def _commit_registration(self, edit: RegistryEdit, change: RegistrationChange) -> RegistrationChange: + def _commit_registration( + self, edit: RegistryEdit, change: RegistrationChange + ) -> RegistrationChange: change.thread.role.require_maintenance_open( MaintenanceBarrier(self.store.path), change.claims_owner ) @@ -183,9 +251,17 @@ def _commit_registration(self, edit: RegistryEdit, change: RegistrationChange) - history.commit(intent) return change - def new_thread(self, name, tags, worktree, *args, kind: type[Thread] = Thread, **values) -> Thread: - return kind(name, tags, worktree, *args, - storage_roots=self.storage_rule.initial_roots(name), **values) + def new_thread( + self, name, tags, worktree, *args, kind: type[Thread] = Thread, **values + ) -> Thread: + return kind( + name, + tags, + worktree, + *args, + storage_roots=self.storage_rule.initial_roots(name), + **values, + ) def decide_storage(self, name, effects): current = self.require(name) @@ -210,9 +286,10 @@ def fence_idle_owners( """Commit all original idle owner fences together, or change none.""" with self.store.editing() as edit: fenced = tuple( - (thread, edit.document.fence_idle_owner( - thread, expected_claimed_generation=generation - )) + ( + thread, + edit.document.fence_idle_owner(thread, expected_claimed_generation=generation), + ) for thread, generation in selected ) edit.commit() @@ -379,9 +456,9 @@ def session_thread( for thread in located: if thread.name == selected: return FoundProjectThread(thread) - return ProjectThreadSelection.from_threads(tuple( - member for thread in located for member in thread.role.acp_thread(thread) - )) + return ProjectThreadSelection.from_threads( + tuple(member for thread in located for member in thread.role.acp_thread(thread)) + ) def user_thread(self) -> UserThreadLookup[Thread]: """The human answer from this stored revision, without copying its roster.""" diff --git a/src/agent_comms/registry_document.py b/src/agent_comms/registry_document.py index 62e212a515..d31c498889 100644 --- a/src/agent_comms/registry_document.py +++ b/src/agent_comms/registry_document.py @@ -2,13 +2,14 @@ from __future__ import annotations +import logging import math import time from abc import ABC, abstractmethod from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence from dataclasses import Field, dataclass, field, replace from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Literal, cast +from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast from .child_process import NoProcess, ProcessIdentity from .errors import RelationViolationError, UnregisteredThreadError @@ -92,6 +93,40 @@ def channel_audience(self) -> Iterator[Thread]: return self.visible_threads(DEFAULT_ROSTER) +class ExternalEndpoint(ABC): + @abstractmethod + def when_owned(self, update: Callable[[RegistryEntry], None]) -> None: ... + + +@dataclass(frozen=True) +class RejectedEndpoint(ExternalEndpoint): + name: str + reason: ClassVar[str] + + def when_owned(self, update: Callable[[RegistryEntry], None]) -> None: + logging.getLogger(__name__).warning('Bridge observation for %s: %s', self.name, self.reason) + + +class AbsentEndpoint(RejectedEndpoint): + reason = 'entry absent' + + +class ForeignEndpoint(RejectedEndpoint): + reason = 'foreign owner or process incarnation' + + +class DeadEndpoint(RejectedEndpoint): + reason = 'owner process dead' + + +@dataclass(frozen=True) +class OwnedEndpoint(ExternalEndpoint): + entry: RegistryEntry + + def when_owned(self, update: Callable[[RegistryEntry], None]) -> None: + update(self.entry) + + class UserThreadLookup[Value](ABC): """The registry's human selection, including a registry with no human yet.""" @@ -223,8 +258,8 @@ def copy(cls, value: int) -> int: @ThreadSlice.required @dataclass(slots=True) class RegistryDocument(RegistryPresence): - version: Literal[1] = field( - default=1, metadata={"thread_slice": WholeDocumentValue, "wire_required": True}) + version: Literal[2] = field( + default=2, metadata={"thread_slice": WholeDocumentValue, "wire_required": True}) threads: dict[str, Thread] = field(default_factory=dict, metadata={"thread_slice": OwnRow}) statuses: dict[str, ThreadStatus] = field( default_factory=dict, metadata={"thread_slice": OwnRow}) @@ -288,6 +323,17 @@ def entry(self, name: str) -> RegistryEntry: def entry_if_present(self, name: str) -> RegistryEntry | None: return self.entry(name) if self.canonical_name(name) in self.threads else None + def external_endpoint(self, owner: OwnerIdentity, process: ProcessIdentity) -> ExternalEndpoint: + name = owner.incarnation.name + if self.canonical_name(name) not in self.threads: + return AbsentEndpoint(name) + entry = self.entry(name) + if not entry.owned_by(owner, process): + return ForeignEndpoint(name) + if not process.alive(): + return DeadEndpoint(name) + return OwnedEndpoint(entry) + @classmethod def entry_from_wire(cls, raw: dict, name: str) -> RegistryEntry: """Decode one thread's entry without decoding the other declarations. 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/thread_execution.py b/src/agent_comms/thread_execution.py index 7606a790a8..1605c8b7eb 100644 --- a/src/agent_comms/thread_execution.py +++ b/src/agent_comms/thread_execution.py @@ -8,7 +8,7 @@ from .declared_family import DeclaredFamily from .child_process import ProcessState from .errors import RelationViolationError -from .thread_status import ConnectedThreadStatus, NoBridgeThreadStatus, ThreadStatus +from .thread_status import NoBridgeThreadStatus, ThreadStatus if TYPE_CHECKING: from .threads import Thread @@ -165,7 +165,7 @@ def require_external(cls) -> None: def registered_status(cls, thread: Thread, requested: ThreadStatus) -> ThreadStatus: """Core launches no process here: the inbox is connected while its bridge lives.""" if requested.active and thread.process_alive: - return ConnectedThreadStatus() + return requested.with_bridge() return NoBridgeThreadStatus() @classmethod diff --git a/src/agent_comms/thread_status.py b/src/agent_comms/thread_status.py index 3e79341069..c1dcd399d9 100644 --- a/src/agent_comms/thread_status.py +++ b/src/agent_comms/thread_status.py @@ -10,7 +10,10 @@ from .declared_family import DeclaredFamily from .errors import RelationViolationError from .field_codec import FieldCodec, TextRepresentation -from .thread_presentation import ThreadOwnerBinding, ThreadPresentation, InactiveThreadPresentationStatus +from .thread_presentation import ( + ThreadOwnerBinding, ThreadPresentation, InactiveThreadPresentationStatus, + WorkingThreadPresentationStatus, +) if TYPE_CHECKING: from .activity import ObservedActivity @@ -78,6 +81,10 @@ def native_presentation( ) -> ThreadPresentation: return ordinary + def with_bridge(self) -> ThreadStatus: + """An attached bridge supplies connected presence, not a native lifecycle.""" + return ConnectedThreadStatus() + def external_presentation( self, process: ProcessState, ordinary: ThreadPresentation, ) -> ThreadPresentation: @@ -167,6 +174,19 @@ def after_heartbeat(self, name: str) -> ThreadStatus: class ConnectedThreadStatus(CliInboxPresence, RunningThreadStatus): """A CLI inbox whose bridge process is watching it.""" + def with_bridge(self) -> ThreadStatus: + return self + + +class CompactingThreadStatus(ConnectedThreadStatus): + """The bridge is still watching while its CLI compacts context.""" + + receiving = False + + def presentation(self, title: str, activity: ObservedActivity) -> ThreadPresentation: + return ThreadPresentation(title, "⌛", "Compacting context", + status=WorkingThreadPresentationStatus()) + class NoBridgeThreadStatus(CliInboxPresence, StoppedThreadStatus): """A CLI inbox with no bridge attached; its messages wait unread. diff --git a/src/agent_comms/wire_watch.py b/src/agent_comms/wire_watch.py index 7428d7ab84..8455fe4492 100644 --- a/src/agent_comms/wire_watch.py +++ b/src/agent_comms/wire_watch.py @@ -32,7 +32,7 @@ _IN_Q_OVERFLOW = 0x00004000 _IN_IGNORED = 0x00008000 _MASK = ( - _IN_CLOSE_WRITE | _IN_MOVED_FROM | _IN_MOVED_TO | _IN_DELETE + _IN_CREATE | _IN_CLOSE_WRITE | _IN_MOVED_FROM | _IN_MOVED_TO | _IN_DELETE | _IN_DELETE_SELF | _IN_MOVE_SELF ) _HEADER = struct.Struct("=iIII") @@ -303,8 +303,7 @@ def _open_file_watcher[WatchT: NativeFileWatch]( watch = add( fd, os.fsencode(directory), _MASK - | (_IN_MODIFY if watched.modified else 0) - | (_IN_CREATE if watched.reopen else 0), + | (_IN_MODIFY if watched.modified else 0), ) if watch < 0: error = OSError(ctypes.get_errno(), f"inotify_add_watch failed for {directory}") diff --git a/src/agent_comms_hooks/__init__.py b/src/agent_comms_hooks/__init__.py index e5fc0fb914..b3abcc83f0 100644 --- a/src/agent_comms_hooks/__init__.py +++ b/src/agent_comms_hooks/__init__.py @@ -12,15 +12,24 @@ import json import pkgutil +import shlex import sys from abc import ABC, abstractmethod -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass, fields from importlib import import_module from importlib.metadata import entry_points from typing import ClassVar, Self, get_type_hints +class HookValue(ABC): + """A declared scalar boundary with its own decoding behavior.""" + + @classmethod + @abstractmethod + def decode(cls, value: object) -> HookValue: ... + + @dataclass(frozen=True) class HookRecord: """A JSON object Claude Code writes to a hook, decoded once by its declared fields.""" @@ -40,6 +49,8 @@ def field(declared: type, value: object) -> object: """A nested record is decoded by its own fields; a JSON scalar is kept as written.""" if isinstance(value, Mapping) and issubclass(declared, HookRecord): return declared.decode(value) + if issubclass(declared, HookValue): + return declared.decode(value) return value @@ -113,6 +124,22 @@ def entry(cls) -> dict[str, object]: return {"hooks": [{"type": "command", "command": cls.command()}]} +class PythonHookCommand: + """One callable declaration supplies installed and source-only hook commands.""" + + target: ClassVar[Callable[[], None]] + + @classmethod + def command(cls) -> str: + return ClaudeHookProgram.console_script(f"{cls.target.__module__}:{cls.target.__name__}") + + @classmethod + def source_command(cls) -> str: + target = cls.target + program = f"from {target.__module__} import {target.__name__}; {target.__name__}()" + return shlex.join((sys.executable, "-c", program)) + + class ClaudeToolHookProgram(ClaudeHookProgram): """A program that reads one tool's input, so Claude Code runs it for that tool only.""" diff --git a/src/agent_comms_hooks/compaction.py b/src/agent_comms_hooks/compaction.py new file mode 100644 index 0000000000..d22a7367f7 --- /dev/null +++ b/src/agent_comms_hooks/compaction.py @@ -0,0 +1,83 @@ +"""Claude's real compaction hooks observe the connected watch, never create one.""" + +from __future__ import annotations + +from dataclasses import dataclass +from functools import partial +from pathlib import Path +from typing import ClassVar + +from agent_comms.registration import Registration +from agent_comms.thread_status import CompactingThreadStatus, ConnectedThreadStatus, ThreadStatus +from agent_comms_hooks import ClaudeHook, ClaudeHookProgram, HookValue +from agent_comms_hooks.watch import WatchClaim, WatchRecord + + +def observe(record: WatchRecord, *, status: ThreadStatus) -> None: + registry = Registration(Path(record.root) / "registry.json") + original = registry.entry(record.bus).entry + registry.observe_external_status( + original.owner_identity, original.thread.require_process(), status + ) + + +class BeforeCompaction(ClaudeHookProgram): + events = ("PreCompact",) + + @classmethod + def command(cls) -> str: + return cls.console_script(f"{starting.__module__}:{starting.__name__}") + + +class AfterCompaction(ClaudeHookProgram): + events = ("SessionStart",) + + @classmethod + def command(cls) -> str: + return cls.console_script(f"{finished.__module__}:{finished.__name__}") + + @classmethod + def entry(cls) -> dict[str, object]: + return {"matcher": CompactedSource.name, **super().entry()} + + +class SessionSource(HookValue): + name: ClassVar[str] + + @classmethod + def decode(cls, value: object) -> SessionSource: + for member in cls.__subclasses__(): + if member.name == value: + return member() + return OtherSource() + + def observe(self, claim: WatchClaim) -> None: + """Startup, resume and future non-compaction sources do not end compaction.""" + + +class OtherSource(SessionSource): + name: ClassVar[str] = "" + + +class CompactedSource(SessionSource): + name: ClassVar[str] = "compact" + + def observe(self, claim: WatchClaim) -> None: + claim.state().observe_compaction(partial(observe, status=ConnectedThreadStatus())) + + +@dataclass(frozen=True) +class SessionStartHook(ClaudeHook): + source: SessionSource + + +def starting() -> None: + hook = ClaudeHook.read() + WatchClaim(hook.session_id).state().observe_compaction( + partial(observe, status=CompactingThreadStatus()) + ) + + +def finished() -> None: + hook = SessionStartHook.read() + hook.source.observe(WatchClaim(hook.session_id)) diff --git a/src/agent_comms_hooks/session.py b/src/agent_comms_hooks/session.py new file mode 100644 index 0000000000..e732087e65 --- /dev/null +++ b/src/agent_comms_hooks/session.py @@ -0,0 +1,107 @@ +"""Claude session hooks start/stop the configured bridge, never a Monitor reader.""" + +from __future__ import annotations + +import asyncio +import json +import os +from contextlib import aclosing +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.wire_watch import WatchSource, WireWatch +from agent_comms_hooks import ClaudeHook, ClaudeHookProgram, PythonHookCommand +from agent_comms_hooks.watch import 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 + + async def ready(self, request: VerifySession, claim: WatchClaim) -> None: + async with ( + asyncio.timeout(30), + aclosing( + WireWatch.observations(WatchSource.modified_file(claim.socket_path)) + ) as changes, + ): + async for _ in changes: + try: + await asyncio.to_thread(request.exchange, claim) + except FileNotFoundError, ConnectionRefusedError: + continue + return + + def start(self) -> None: + claim = WatchClaim(self.session_id) + with claim.launching(): + claim.state().start_session(self) + + def start_recorded(self, state: RecordedWatch) -> None: + owner = self.owner() + request = VerifySession.model_validate(dict(owner=owner)) + try: + request.exchange(state.claim) + return + except (FileNotFoundError, ConnectionRefusedError) as error: + state.require_launch(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 = state.claim.path.with_suffix(".log") + with log.open("ab") as output: + child = ParentedProcess.launch(command, env=dict(os.environ), output=output) + try: + asyncio.run(self.ready(request, state.claim)) + except TimeoutError as error: + raise RuntimeError( + f"Session bridge startup not observed (PID {child.pid}); inspect {log}. " + "No launch retry." + ) from error + + def end(self) -> None: + WatchClaim(self.session_id).state().end_session(self) + + def end_recorded(self, state: RecordedWatch) -> None: + try: + EndSession.model_validate(dict(owner=self.owner())).exchange(state.claim) + except FileNotFoundError, ConnectionRefusedError: + return + + +def started() -> None: + SessionHook.read().start() + + +def ended() -> None: + SessionHook.read().end() + + +class SessionStarted(PythonHookCommand, ClaudeHookProgram): + events = ("SessionStart",) + target = staticmethod(started) + + +class SessionEnded(PythonHookCommand, ClaudeHookProgram): + events = ("SessionEnd",) + target = staticmethod(ended) diff --git a/src/agent_comms_hooks/watch.py b/src/agent_comms_hooks/watch.py index ef65dd78a5..48ad5290fd 100644 --- a/src/agent_comms_hooks/watch.py +++ b/src/agent_comms_hooks/watch.py @@ -11,10 +11,14 @@ import hashlib import json import shlex -from collections.abc import Iterator, Sequence +from collections.abc import Callable, Iterator, Sequence from contextlib import contextmanager from dataclasses import asdict, dataclass from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from agent_comms_hooks.session import SessionHook from agent_comms_hooks import ClaudeHook, ClaudeHookProgram @@ -25,6 +29,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: @@ -35,6 +42,42 @@ 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(): + self.state().enrol(self, record) + @contextmanager def held(self, record: WatchRecord) -> Iterator[None]: """One watch per conversation, independent of --state and bus aliases.""" @@ -53,7 +96,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() @@ -61,6 +104,19 @@ def state(self) -> WatchState: class WatchState: + def enrol(self, claim: WatchClaim, record: WatchRecord) -> None: + with claim.held(record): + pass + + def start_session(self, hook: SessionHook) -> None: + """A session without a recorded bridge never acquires one from a hook.""" + + def end_session(self, hook: SessionHook) -> None: + """An absent or lapsed bridge has nothing to stop.""" + + def observe_compaction(self, observe: Callable[[WatchRecord], None]) -> None: + """An absent watch never acquires bridge presence from a hook.""" + def report(self, event: str) -> None: """Silent unless the conversation is meant to be watched and is not.""" @@ -69,15 +125,51 @@ class UnclaimedWatch(WatchState): pass -class LiveWatch(WatchState): - pass - - @dataclass(frozen=True) -class LapsedWatch(WatchState): +class RecordedWatch(WatchState): claim: WatchClaim record: WatchRecord + def start_session(self, hook: SessionHook) -> None: + hook.start_recorded(self) + + def require_launch(self, error: OSError) -> None: + """A lapsed record may launch its bridge once under the launch lock.""" + + def report_warning(self, event: str, warning: str) -> None: + context = {"hookEventName": event, "additionalContext": warning} + print(json.dumps({"hookSpecificOutput": context})) + + +class LiveWatch(RecordedWatch): + def enrol(self, claim: WatchClaim, record: WatchRecord) -> None: + if self.record != record: + raise ValueError("This session is already bound to another bridge command") + + def require_launch(self, error: OSError) -> None: + raise RuntimeError( + "An older watch still owns this session. Stop that Monitor once, then connect." + ) from error + + def end_session(self, hook: SessionHook) -> None: + hook.end_recorded(self) + + def report(self, event: str) -> None: + if not self.claim.has_reader(): + self.report_warning( + event, + 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)}", + ) + + def observe_compaction(self, observe: Callable[[WatchRecord], None]) -> None: + # Keep prompt/tool hooks stdlib-only; compaction supplies its own observer. + if self.claim.state() == self: + observe(self.record) + + +class LapsedWatch(RecordedWatch): @property def warning(self) -> str: bus = self.record.bus @@ -89,8 +181,7 @@ def warning(self) -> str: ) def report(self, event: str) -> None: - context = {"hookEventName": event, "additionalContext": self.warning} - print(json.dumps({"hookSpecificOutput": context})) + self.report_warning(event, self.warning) class WatchGuard(ClaudeHookProgram): 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_thread_status.py b/tests/test_thread_status.py index 1e8bd583a6..680cde7cd7 100644 --- a/tests/test_thread_status.py +++ b/tests/test_thread_status.py @@ -17,6 +17,9 @@ from agent_comms.registration import Registration from agent_comms.thread_status import ( ArchivedThreadStatus, + CompactingThreadStatus, + ConnectedThreadStatus, + NoBridgeThreadStatus, DeletingThreadStatus, IdleThreadStatus, RunningThreadStatus, @@ -26,6 +29,47 @@ from agent_comms.threads import Thread +def test_connected_compaction_preserves_endpoint_and_presentation(tmp_path, monkeypatch): + from agent_comms.thread_execution import ExternalThreadExecution + from agent_comms_hooks.watch import WatchClaim, WatchRecord + + monkeypatch.setenv('HOME', str(tmp_path)) + comms = wire(tmp_path / 'bus') + registry = comms.registry + registry.declare(registry.new_thread( + 'watched', frozenset(), str(tmp_path), execution=ExternalThreadExecution)) + process = ProcessIdentity.capture(os.getpid()) + claim = WatchClaim('session') + record = WatchRecord('watched', str(comms.root), ('watch',)) + with registry.external_process('watched', process): + with claim.held(record): + original = registry.entry('watched').entry + watch = claim.state() + watch.observe_compaction(CompactingThreadStatus()) + compacting = registry.snapshot() + assert compacting.statuses['watched'] == CompactingThreadStatus() + assert compacting.threads['watched'].process_identity == process + assert compacting.owner_identity('watched') == original.owner_identity + registry.heartbeat('watched') + registry.register(replace(registry.require('watched'), title='Current title'), registry.status('watched')) + shown = comms.views.thread_presentations(('watched',)).require('watched') + assert shown.summary == 'Compacting context' and shown.status.working + assert shown.title == 'Current title' + assert registry.status('watched') == CompactingThreadStatus() + watch.observe_compaction(ConnectedThreadStatus()) + assert registry.status('watched') == ConnectedThreadStatus() + assert registry.entry('watched').entry.owner_identity == original.owner_identity + watch.observe_compaction(CompactingThreadStatus()) + assert registry.status('watched') == NoBridgeThreadStatus() + watch.observe_compaction(CompactingThreadStatus()) + assert registry.status('watched') == NoBridgeThreadStatus() + with registry.external_process('watched', process): + # An earlier bridge's delayed observation cannot alter the new lease, + # even when the process is the same and only the generation changed. + registry.observe_external_status(original.owner_identity, process, CompactingThreadStatus()) + assert registry.status('watched') == ConnectedThreadStatus() + + def test_captured_owner_presentation_survives_exit_until_fresh_acquisition(tmp_path): comms = wire(tmp_path) child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) diff --git a/tests/test_watch_guard.py b/tests/test_watch_guard.py index d95680877f..8e82ac9d4d 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' @@ -40,7 +40,11 @@ def hook(event='PostToolUse', session='session'): for event, entries in hooks.items()} == { 'PreToolUse': ['agent-comms-commander send-guard'], 'UserPromptSubmit': ['agent-comms-watch-guard'], - 'PostToolUse': ['agent-comms-watch-guard']} + 'PostToolUse': ['agent-comms-watch-guard'], + 'PreCompact': ['agent-comms-claude-precompact'], + 'SessionStart': ['agent-comms-claude-session-started', 'agent-comms-claude-compacted'], + 'SessionEnd': ['agent-comms-claude-session-ended']} assert hooks['PreToolUse'][0]['matcher'] == 'Bash' assert 'matcher' not in hooks['PostToolUse'][0] + assert any(entry.get('matcher') == 'compact' for entry in hooks['SessionStart']) diff --git a/tools/cutover/carry_compacting_status.py b/tools/cutover/carry_compacting_status.py new file mode 100644 index 0000000000..1242a56ef3 --- /dev/null +++ b/tools/cutover/carry_compacting_status.py @@ -0,0 +1,40 @@ +"""Carry registry version 1 before publishing connected compaction presence.""" + +import json +from dataclasses import fields +from pathlib import Path + +from agent_comms.registry_document import RegistryDocument +from agent_comms.registry_store import RegistryStore + +from .carry import Carried, Carry + + +class CompactingStatus(Carry): + @classmethod + def backup_files(cls, root: Path) -> tuple[Path, ...]: + store = RegistryStore(root / "registry.json") + return store.path, store.guard_path + + @classmethod + def carry(cls, root: Path, *, apply: bool) -> Carried: + store = RegistryStore(root / "registry.json") + if not store.path.exists(): + return Carried(0) + version = fields(RegistryDocument)[0] + with store.locked(): + raw = json.loads(store.path.read_text()) + previous = raw[version.name] + if previous == version.default: + return Carried(0) + if previous != 1: + raise ValueError(f"Unexpected previous registry version: {previous}") + raw[version.name] = version.default + document = RegistryDocument.from_wire(raw) + if apply: + store.save_unlocked(document) + return Carried(1) + + +if __name__ == "__main__": + CompactingStatus.main()