Every CLI result writes itself; watch and tool honour Answer: CliResult

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)

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):

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]]: