diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index 1f38224f21..491f54e003 100644 --- a/src/agent_comms/cli_commands.py +++ b/src/agent_comms/cli_commands.py @@ -929,14 +929,7 @@ def describe(self, comms: Comms, target: CommandTarget, project: str) -> TargetA def run(self, ctx: Comms, output: TextIO) -> int: """Run this command and write its declared result.""" - return self.written(self.apply(ctx), output) - - @staticmethod - def written(result: CliResult, output: TextIO) -> int: - """Write one declared result to the caller's stream; its exit code.""" - json.dump(result.to_wire(), output, indent=2) - output.write("\n") - return result.exit_code() + return self.apply(ctx).write(output) @classmethod def published_launch(cls) -> LaunchRoot: @@ -1390,8 +1383,41 @@ def apply(self, ctx: Comms) -> CliResult: return InboxResult(tuple(ctx.bus.inbox(self.thread))) +@dataclass(frozen=True) +class WatchStream(CliResult): + """Unread and arriving messages, written as JSON lines for as long as the watch lasts.""" + + comms: Comms + watch: WatchCliCommand + + def messages(self) -> AsyncIterator[Message]: + return self.comms.bus.watch(self.watch.thread, acknowledgment=self.watch.acknowledgment, + after=self.watch.after) + + def write(self, output: TextIO) -> int: + emit = self.watch.acknowledgment.line_output(output) + async def stream() -> None: + async with aclosing(self.messages()) as messages: + async for message in messages: + emit(ArrivedMessage(message=message).model_dump_json()) + + async def watch() -> None: + thread = self.comms.registry.require(self.watch.thread) + source = thread.runtime.history_source(self.comms.root, thread.incarnation) + async with asyncio.TaskGroup() as tasks: + tasks.create_task(stream()) + tasks.create_task(source.watch_requests( + lambda row: emit(EarlierItems.from_row(row).model_dump_json()))) + + try: + asyncio.run(watch()) + except* MaintenanceHeldError as held: # the bridge reads it as its one held line + raise held.exceptions[0] from None + return self.exit_code() + + @dataclass(frozen=True, kw_only=True) -class WatchCliCommand(ThreadArgument, CliCommand, declared_name="watch"): +class WatchCliCommand(ThreadArgument, CliCommand[WatchStream], declared_name="watch"): help = "Stream unread and arriving messages as JSON lines" after: Annotated[ InboxStart, @@ -1409,29 +1435,8 @@ class WatchCliCommand(ThreadArgument, CliCommand, declared_name="watch"): Argument(flags=("--acknowledgment",)), ] = OnReadWatchAcknowledgment - def apply(self, ctx: Comms) -> AsyncIterator[Message]: - return ctx.bus.watch(self.thread, acknowledgment=self.acknowledgment, after=self.after) - - def run(self, ctx: Comms, output: TextIO) -> int: - emit = self.acknowledgment.line_output(output) - async def stream() -> None: - async with aclosing(self.apply(ctx)) as messages: - async for message in messages: - emit(ArrivedMessage(message=message).model_dump_json()) - - async def watch() -> None: - thread = ctx.registry.require(self.thread) - source = thread.runtime.history_source(ctx.root, thread.incarnation) - async with asyncio.TaskGroup() as tasks: - tasks.create_task(stream()) - tasks.create_task(source.watch_requests( - lambda row: emit(EarlierItems.from_row(row).model_dump_json()))) - - try: - asyncio.run(watch()) - except* MaintenanceHeldError as held: # the bridge reads it as its one held line - raise held.exceptions[0] from None - return 0 + def apply(self, ctx: Comms) -> WatchStream: + return WatchStream(ctx, self) @dataclass(frozen=True, kw_only=True) @@ -1813,7 +1818,7 @@ def apply(self, ctx: Comms) -> BusReadBack: def execute(self, route: CommsRoute, output: TextIO) -> int: """Before its carries, the installer reads only the existing registry.""" - return self.written(self.answer(route.root), output) + return self.answer(route.root).write(output) @dataclass(frozen=True, kw_only=True) @@ -1837,7 +1842,7 @@ def apply(self, ctx: Comms) -> ReconciledStoresResult: def execute(self, route: CommsRoute, output: TextIO) -> int: """Reconciliation reads its declared files before opening runtime consumers.""" - return self.written(self.answer(route.root), output) + return self.answer(route.root).write(output) @dataclass(frozen=True, kw_only=True) @@ -2569,4 +2574,4 @@ def apply(self, ctx: Comms) -> RouteReplacedResult: def execute(self, route: CommsRoute, output: TextIO) -> int: """Every client borrows the published route, and a route that selects no bus refuses each borrower, so the repair of that route borrows none.""" - return self.written(self.answer(route.root), output) + return self.answer(route.root).write(output) diff --git a/src/agent_comms/cli_result.py b/src/agent_comms/cli_result.py index 9412845682..7dfc30a2af 100644 --- a/src/agent_comms/cli_result.py +++ b/src/agent_comms/cli_result.py @@ -2,7 +2,9 @@ from __future__ import annotations +import json from dataclasses import dataclass +from typing import TextIO from .pi_vocabulary import ServiceTier from .tool_result import DeclaredReply @@ -22,6 +24,12 @@ def open_targets(self) -> tuple[str, ...]: def exit_code(self) -> int: return 0 + def write(self, output: TextIO) -> int: + """Write this result to the caller's stream; its exit code.""" + json.dump(self.to_wire(), output, indent=2) + output.write("\n") + return self.exit_code() + @dataclass(frozen=True) class ToolCatalogResult(CliResult): diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 0e376b5362..274a2f93fb 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -11,7 +11,7 @@ from dataclasses import dataclass, field, replace from functools import partial from pathlib import Path -from typing import TYPE_CHECKING, Annotated, ClassVar, Literal, NoReturn, TextIO, Type +from typing import TYPE_CHECKING, Annotated, ClassVar, Literal, NoReturn, Type from pydantic import JsonValue @@ -55,7 +55,7 @@ ThreadsCliCommand, ThreadStoppedResult, ) -from .cli_result import ToolCatalogResult +from .cli_result import CliResult, ToolCatalogResult from .commander.vibe_check import OverloadDegradation from .commander_hud import CommanderHud from .coordination_tables.refusals import SendRefusal @@ -1703,8 +1703,18 @@ def apply(self, ctx: Comms) -> ToolCatalogResult: return self.name.catalog(ctx, tuple(NativeRequestTools.catalog())) +@dataclass(frozen=True) +class ToolReplyResult(CliResult): + """The tool protocol's own reply, the one every adapter receives, written unwrapped.""" + + reply: dict[str, JsonValue] + + def to_wire(self) -> dict[str, JsonValue]: + return self.reply + + @dataclass(frozen=True, kw_only=True) -class InvokeCliCommand(CliCommand, declared_name="tool"): +class InvokeCliCommand(CliCommand[ToolReplyResult], declared_name="tool"): help = "Invoke one declared adapter tool, as --from THREAD when given" caller: Annotated[ ThreadCaller | NoThreadCaller, @@ -1729,14 +1739,8 @@ def argv(self, root: Path) -> tuple[str, ...]: self.declared_name, *self.caller.from_flag(), '--tool', self.tool, '--arguments', json.dumps(self.arguments)) - def apply(self, ctx: Comms) -> dict[str, JsonValue]: - return invoke_tool(ctx, self.tool, self.arguments, caller=self.caller) - - def run(self, ctx: Comms, output: TextIO) -> int: - """The tool protocol's own reply, the one every adapter receives, not a CLI result.""" - json.dump(self.apply(ctx), output, indent=2) - output.write("\n") - return 0 + def apply(self, ctx: Comms) -> ToolReplyResult: + return ToolReplyResult(invoke_tool(ctx, self.tool, self.arguments, caller=self.caller)) def context_tool_catalog(context: str) -> list[dict[str, object]]: