Run install CLI probes in the existing runtime worker

src/agent_comms/cli.py

@@ -1,3 +1,3 @@
 """OpenHCS agent communications — JSON CLI.
 
 Commands print one JSON result, except watch, which streams messages as JSON
@@ -9,7 +9,7 @@
 
 from __future__ import annotations
 
-from agent_comms.cli_arguments import CliArgumentParser, CliArguments
+from agent_comms.cli_arguments import CliArgumentParser, CliArguments, CliParserExit
 
 import argparse
 import sys
@@ -22,48 +22,61 @@
 from agent_comms.launch_root import LaunchRoot
 
 
-def _fail(error: BaseException) -> int:
+def _fail(error: BaseException, output: TextIO) -> int:
     """The declared wire family owns classification, causes and JSON encoding."""
-    print(CoreError.from_exception(error).model_dump_json(), flush=True)
+    print(CoreError.from_exception(error).model_dump_json(), file=output, flush=True)
     return 1
 
 
 def build_parser(arguments: Sequence[str] = ()) -> argparse.ArgumentParser:
     parser = CliArgumentParser(prog="agent-comms", description=__doc__)
     parser.add_root_options()
-    selection = CliArgumentParser.command_parsing(arguments)
+    selection = parser.command_parsing(arguments)
     subparsers = parser.add_subparsers(required=True)
     for declaration in selection.declarations():
         declaration.add_parser(subparsers)
     return parser
 
 
-def execute(arguments: Sequence[str], output: TextIO) -> int:
-    """Own parsing, root selection and dispatch for every registered command."""
-    arguments = list(arguments)
-    args, unknown = build_parser(arguments).parse_known_args(arguments, namespace=CliArguments())
-    if unknown and (len(unknown) != 2 or unknown[0] != LaunchRoot.argument):
-        raise ValueError(f"unrecognized arguments: {' '.join(unknown)}")
-    declaration = args.command_declaration
-    LaunchRoot.startup(
-        arguments,
-        lambda: args.root.resolve(declaration.published_launch),
-    )
-    route = resolve_comms_route(args.root.resolve(LaunchRoot.current).path)
-    with LaunchRoot.from_path(route.root).selected():
-        command = declaration.from_arguments(args)
-        return command.execute(route, output)
+class CliInvocation:
+    """The CLI boundary shared by console entry and selected-runtime probes."""
+
+    def execute(self, arguments: Sequence[str], output: TextIO) -> int:
+        """Own parsing, root selection and dispatch for every registered command."""
+        arguments = list(arguments)
+        args, unknown = build_parser(arguments).parse_known_args(
+            arguments, namespace=CliArguments())
+        if unknown and (len(unknown) != 2 or unknown[0] != LaunchRoot.argument):
+            raise ValueError(f"unrecognized arguments: {' '.join(unknown)}")
+        declaration = args.command_declaration
+        LaunchRoot.startup(
+            arguments,
+            lambda: args.root.resolve(declaration.published_launch),
+        )
+        route = resolve_comms_route(args.root.resolve(LaunchRoot.current).path)
+        with LaunchRoot.from_path(route.root).selected():
+            command = declaration.from_arguments(args)
+            return command.execute(route, output)
+
+    def invoke(self, arguments: Sequence[str], output: TextIO) -> int:
+        """A call borrows its root; success, refusal and parser exit all restore it."""
+        with LaunchRoot.unstarted().selected():
+            try:
+                return self.execute(arguments, output)
+            except CLI_FAILURES as exc:
+                return _fail(exc, output)
+            except CliParserExit as exited:
+                for text in exited.text.texts():
+                    output.write(text)
+                return exited.code
 
 
 def main(argv: Sequence[str] | None = None) -> int:
     try:
         arguments = argv if isinstance(argv, Sequence) else sys.argv[1:]
-        return execute(arguments, sys.stdout)
+        return CliInvocation().invoke(arguments, sys.stdout)
     except KeyboardInterrupt:
         return 130
-    except CLI_FAILURES as exc:
-        return _fail(exc)
-    return 0
 
 
 if __name__ == "__main__":

src/agent_comms/cli_arguments.py

@@ -1,3 +1,3 @@
 """CLI parsing keeps the selected declaration, not a second spelling of it."""
 
 from __future__ import annotations
@@ -9,6 +9,8 @@
 
 from agent_comms.cli_command import CliCommand
 from agent_comms.launch_root import DefaultLaunch, LaunchRequest, LaunchRoot
+from agent_comms.field_codec import SubclassUnion
+from agent_comms.thread_text import AbsentText, GivenText, TextChoice
 
 if TYPE_CHECKING:
     from agent_comms.cli_bridges.shared import BindingThread
@@ -44,25 +46,58 @@
     binding_thread: BindingThread
 
 
+class CliParserExit(SystemExit):
+    """A parser exit carries its declared integer status, not arbitrary process state."""
+
+    code: int
+    text: CliParserText
+
+    def __init__(self, status: int, text: CliParserText) -> None:
+        super().__init__(status)
+        self.text = text
+
+
+class CliParserText(SubclassUnion, TextChoice, affix="CliParserText"):
+    """The parser's text carried to the invocation that owns its output."""
+
+
+@dataclass(frozen=True)
+class GivenCliParserText(GivenText, CliParserText):
+    text: str
+
+
+@dataclass(frozen=True)
+class NoCliParserText(AbsentText, CliParserText):
+    pass
+
+
 class CliArgumentParser(argparse.ArgumentParser):
     """Argument errors use Core's JSON failure boundary, including subcommands."""
 
+    def print_help(self, file: object = None) -> Never:
+        raise CliParserExit(0, GivenCliParserText(self.format_help()))
+
+    def print_usage(self, file: object = None) -> Never:
+        raise CliParserExit(0, GivenCliParserText(self.format_usage()))
+
     def add_root_options(self) -> None:
         self.add_argument(
             "--root", type=LaunchRoot.from_path, default=DefaultLaunch(),
             help="Explicit Comms root directory (otherwise uses the published active route)",
         )
         self.add_argument(LaunchRoot.argument, help=argparse.SUPPRESS)
 
-    @classmethod
-    def command_parsing(cls, arguments: Sequence[str]) -> CommandParsing:
-        selection = cls(add_help=False)
+    def command_parsing(self, arguments: Sequence[str]) -> CommandParsing:
+        selection = type(self)(add_help=False)
         selection.add_root_options()
         selection.add_argument(
             "command", nargs="?", type=SelectedCommand.from_name, default=CommandParsing(),
         )
         args, _ = selection.parse_known_args(arguments, namespace=_CommandParsingArguments())
         return args.command
 
+    def exit(self, status: int = 0, message: object = None) -> Never:
+        raise CliParserExit(status, NoCliParserText())
+
     def error(self, message: str) -> Never:
         raise ValueError(message)

src/agent_comms/installer/interpreter_operation.py

@@ -501,11 +501,14 @@ def answer(self) -> ProbeReply:
 def serve() -> None:
     """Answer declared operations from stdin, one encoded answer per request line, until it
     closes. An operation that raises ends the worker: its traceback goes to stderr, and the
-    caller reads the exit in place of an answer. The answers own stdout's descriptor; every
-    other write, an operation's exit handlers included, lands on stderr."""
-    with os.fdopen(os.dup(sys.stdout.fileno()), "w") as answers:
+    caller reads the exit in place of an answer. The transport owns its descriptors:
+    operations read EOF on stdin, and their stdout lands on stderr."""
+    with (os.fdopen(os.dup(sys.stdin.fileno())) as requests,
+          os.fdopen(os.dup(sys.stdout.fileno()), "w") as answers,
+          open(os.devnull) as empty_input):
+        os.dup2(empty_input.fileno(), sys.stdin.fileno())
         os.dup2(sys.stderr.fileno(), sys.stdout.fileno())
-        for line in sys.stdin:
+        for line in requests:
             print(
                 FieldCodec.compact(
                     FieldCodec.encode(FieldCodec.decode(ProbeRequest, FieldCodec.parse(line)).answer())

src/agent_comms/installer/probes/commands.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
-import subprocess
-import sys
+from contextlib import redirect_stderr
 from dataclasses import dataclass
+from io import StringIO
 
 from pydantic import TypeAdapter, ValidationError
 
+from agent_comms.cli import CliInvocation
 from agent_comms.cli_protocol import CoreError
 from agent_comms.cli_result import CliResult
 from agent_comms.field_codec import FieldCodec, MalformedRecord
@@ -31,7 +32,9 @@ class CliUndecodableError(Exception):
 class CliOutput:
     """What a command line printed. The worker encodes it on the runtime that ran the
     command (an older Core than the installer's, during an upgrade), so it stays plain
-    text; the installer decodes it into the command's answer here, once."""
+    text; the installer decodes it into the command's answer here, once. stderr contains
+    Python writes during the call; native writes and pre-bound logging go to worker diagnostics.
+    """
 
     returncode: int
     stdout: str
@@ -68,28 +71,19 @@ def decoded[Answer](self, result: type[Answer]) -> Answer:
 
 
 @dataclass(frozen=True)
-class CliCommandProbe(RuntimeProbe[CliOutput]):
-    """One command-line call, answered by the selected runtime's own command line in its
-    own process of this interpreter."""
+class CliCommandProbe(CliInvocation, RuntimeProbe[CliOutput]):
+    """One CLI invocation in the selected runtime's existing interpreter worker.
+
+    The worker key owns cwd and environment; CliInvocation scopes each call's root.
+    """
 
     arguments: tuple[str, ...]
 
     def install_phase(self) -> CliProbePhase:
         return CliProbePhase(probe=type(self), arguments=self.arguments)
 
-    def entry(self) -> tuple[str, ...]:
-        return "-m", "agent_comms.cli"
-
-    def source(self) -> str:
-        """Ordinary CLI invocations supply no stdin; scripted probes override it."""
-        return ""
-
     def run(self) -> CliOutput:
-        completed = subprocess.run(
-            (sys.executable, *self.entry(), *self.arguments),
-            capture_output=True,
-            input=self.source(),
-            text=True,
-            check=False,
-        )
-        return CliOutput(completed.returncode, completed.stdout, completed.stderr)
+        output, errors = StringIO(), StringIO()
+        with redirect_stderr(errors):
+            returncode = self.invoke(self.arguments, output)
+        return CliOutput(returncode, output.getvalue(), errors.getvalue())

src/agent_comms/installer/probes/owners.py

@@ -1,28 +1,30 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from importlib.resources import files
+from collections.abc import Sequence
+from io import StringIO
 from pathlib import Path
-from typing import Annotated
+from typing import Annotated, TextIO
 
+from agent_comms.cli_command import TargetBatchResult
 from agent_comms.comms import wire
-from agent_comms.field_codec import PathText
+from agent_comms.field_codec import FieldCodec, PathText
 from agent_comms.installer.interpreter_operation import RuntimeProbe
 from agent_comms.installer.probes.commands import CliCommandProbe
+from agent_comms.installer.owner_starts import OwnerStart
 from agent_comms.owner_restart import OwnerRestartRequest
 
 
 class OwnerActionsProbe(CliCommandProbe):
-    """Execute and decode the batch in one isolated, selected-runtime process.
-
-    The entry source ships with the selected runtime's own installer.
-    """
-
-    def entry(self) -> tuple[str, ...]:
-        return ("-",)
-
-    def source(self) -> str:
-        return files(__package__).joinpath("owner_actions.py").read_text()
+    """The selected runtime decodes its own batch before emitting owner reports."""
+
+    def execute(self, arguments: Sequence[str], output: TextIO) -> int:
+        command_output = StringIO()
+        super().execute(arguments, command_output)
+        batch = FieldCodec.decode(TargetBatchResult, FieldCodec.parse(command_output.getvalue()))
+        owners = tuple(outcome.owner_start() for outcome in batch.outcomes)
+        print(FieldCodec.compact(FieldCodec.encode(owners, tuple[OwnerStart, ...])), file=output)
+        return 0
 
 
 @dataclass(frozen=True)

tests/test_cli_error_family.py

@@ -27,7 +27,7 @@
     (ValueError('first\nsecond'), FailedCoreError),
 ])
 def test_exception_declaration_encodes_one_line(error, expected, capsys):
-    assert _fail(error) == 1
+    assert _fail(error, sys.stdout) == 1
     output = capsys.readouterr().out
     assert len(output.splitlines()) == 1
     decoded = WATCH_LINE.validate_json(output)
@@ -42,7 +42,7 @@ def test_error_keeps_cause_without_using_cause_text_as_policy(capsys):
     except ValueError as cause:
         error = CoordinationReadUnavailable('not ready')
         error.__cause__ = cause
-    _fail(error)
+    _fail(error, sys.stdout)
     decoded = WATCH_LINE.validate_json(capsys.readouterr().out)
     assert isinstance(decoded, UnavailableCoreError)
     assert 'ValueError: bad\ninput' in decoded.error

tests/test_command_families.py

@@ -1,4 +1,4 @@
-from agent_comms.repo import Checkout
+from agent_comms.repo import Checkout, Repo
 from agent_comms.field_codec import FieldCodec, GoalId
 
 """External contracts and adding commands without editing any consumer catalog."""
@@ -46,7 +46,8 @@ def parse(args):
     send_arguments = ["user-send", "--to", "a", "--body", "hello"]
     with pytest.raises(ValueError):
         parse(send_arguments)
-    assert parse([*send_arguments, "--worktree", str(tmp_path)]).worktree == str(tmp_path)
+    Repo.git(tmp_path, "init", "--quiet")
+    assert parse([*send_arguments, "--worktree", str(tmp_path)]).worktree == Checkout.checked(tmp_path)
     invoke = parse(["tool", "--tool", "comms_threads"])
     assert invoke.arguments == {}
     send = parse(["send", "--from", "a", "--to", "b", "--body", "hello"])