diff --git a/src/agent_comms/cli.py b/src/agent_comms/cli.py index 9086dae635..8a3f20802f 100644 --- a/src/agent_comms/cli.py +++ b/src/agent_comms/cli.py @@ -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__": diff --git a/src/agent_comms/cli_arguments.py b/src/agent_comms/cli_arguments.py index 04584982a0..4fa63717af 100644 --- a/src/agent_comms/cli_arguments.py +++ b/src/agent_comms/cli_arguments.py @@ -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,9 +46,40 @@ class CliArguments(argparse.Namespace): 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(), @@ -54,9 +87,8 @@ def add_root_options(self) -> None: ) 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(), @@ -64,5 +96,8 @@ def command_parsing(cls, arguments: Sequence[str]) -> 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) diff --git a/src/agent_comms/installer/interpreter_operation.py b/src/agent_comms/installer/interpreter_operation.py index 08a9195719..8bf78c3190 100644 --- a/src/agent_comms/installer/interpreter_operation.py +++ b/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()) diff --git a/src/agent_comms/installer/probes/commands.py b/src/agent_comms/installer/probes/commands.py index bf2c757590..05e8bae90a 100644 --- a/src/agent_comms/installer/probes/commands.py +++ b/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()) diff --git a/src/agent_comms/installer/probes/owner_actions.py b/src/agent_comms/installer/probes/owner_actions.py deleted file mode 100644 index f93075ecf1..0000000000 --- a/src/agent_comms/installer/probes/owner_actions.py +++ /dev/null @@ -1,33 +0,0 @@ -"""The selected runtime reads its own CLI result, then emits an installer report. - -OwnerActionsProbe supplies this module's source to an isolated interpreter. Do -not import the installing Core here: both CLI execution and batch decoding must -use the selected runtime, whose result shape may differ from the installing one. -""" - -from __future__ import annotations - -import sys -from io import StringIO - -from agent_comms.cli import _fail, execute -from agent_comms.cli_command import TargetBatchResult -from agent_comms.cli_failures import CLI_FAILURES -from agent_comms.field_codec import FieldCodec -from agent_comms.installer.owner_starts import OwnerStart - - -def main(arguments: list[str]) -> int: - output = StringIO() - execute(arguments, output) - batch = FieldCodec.decode(TargetBatchResult, FieldCodec.parse(output.getvalue())) - owners = tuple(outcome.owner_start() for outcome in batch.outcomes) - print(FieldCodec.compact(FieldCodec.encode(owners, tuple[OwnerStart, ...]))) - return 0 - - -if __name__ == "__main__": - try: - raise SystemExit(main(sys.argv[1:])) - except CLI_FAILURES as error: - raise SystemExit(_fail(error)) diff --git a/src/agent_comms/installer/probes/owners.py b/src/agent_comms/installer/probes/owners.py index 14da5e6ed1..72d6d68dff 100644 --- a/src/agent_comms/installer/probes/owners.py +++ b/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) diff --git a/tests/test_cli_error_family.py b/tests/test_cli_error_family.py index 0e07e02d72..3dfaea4629 100644 --- a/tests/test_cli_error_family.py +++ b/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 diff --git a/tests/test_command_families.py b/tests/test_command_families.py index 99ce697d4b..e241ada8bb 100644 --- a/tests/test_command_families.py +++ b/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"]) diff --git a/tests/test_install_worker_holds.py b/tests/test_install_worker_holds.py index 25c7df4de4..c29fba781f 100644 --- a/tests/test_install_worker_holds.py +++ b/tests/test_install_worker_holds.py @@ -1,13 +1,20 @@ -import io import os +import sys +from collections.abc import Sequence from dataclasses import dataclass from functools import partial +from io import StringIO from pathlib import Path +from typing import TextIO import pytest +from agent_comms.cli import CliInvocation +from agent_comms.cli_arguments import CliArgumentParser +from agent_comms.cli_verbs.read_back import ReadBackCliCommand from agent_comms.cli_verbs.start import StartCliCommand from agent_comms.cli_verbs.stop import StopCliCommand +from agent_comms.comms import Comms from agent_comms.field_codec import ClaimId, FieldCodec from agent_comms.installer.commands import InstallCommands from agent_comms.installer.interpreter_operation import ( @@ -18,11 +25,13 @@ ) from agent_comms.installer.owner_starts import OwnerStart, RefusedOwner, StartedOwner from agent_comms.installer.probes.commands import CliCommandProbe, CliOutput, CliRefusedError -from agent_comms.installer.probes.owners import OwnerActionsProbe from agent_comms.installer.probes.packages import DistributionVersionsProbe, InstalledPackagesProbe from agent_comms.installer.refusals import InterpreterWorkerDied, UpgradeRefused from agent_comms.installer.upgrade import Stack +from agent_comms.launch_root import LaunchRoot from agent_comms.launch_runtime import LaunchRuntime +from agent_comms.repo import Checkout +from agent_comms.threads import Thread from agent_comms.maintenance_barrier import HOLDER_NONCE, MaintenanceClaimHolder @@ -38,15 +47,72 @@ def test_owner_actions_decode_in_the_selected_worker(comms, command, source_repo outcomes[0].require_success(UpgradeRefused) -@pytest.mark.parametrize("probe_type", [CliCommandProbe, OwnerActionsProbe]) -def test_cli_reply_preserves_declared_failure_from_each_entry(comms, probe_type): +def test_cli_probes_share_worker_and_keep_roots_and_refusals_separate( + comms: Comms, tmp_path: Path, +) -> None: + comms.registry.declare(Thread("only-first-root", frozenset(), Checkout(tmp_path))) runtime = LaunchRuntime.running() with InstallCommands() as commands: stack = Stack(Path(__file__).resolve().parents[1], commands, runtime=runtime) - probe = probe_type(("--root", str(comms.root), "not-a-command")) - with pytest.raises(UpgradeRefused) as refused: - stack.cli_reply(probe, partial(CliOutput.decoded, result=tuple[OwnerStart, ...])) - assert isinstance(refused.value.__cause__, CliRefusedError) + stack.probe(DistributionVersionsProbe()) + (worker,) = commands.workers.values() + pid = worker.child.pid + assert stack.cli(comms.root, ReadBackCliCommand).threads == 1 + assert stack.cli(tmp_path / "other-root", ReadBackCliCommand).threads == 0 + help_output = stack.probe(CliCommandProbe(("--help",))) + assert help_output.returncode == 0 + assert help_output.stdout + for probe_type in CliCommandProbe.members_with(CliCommandProbe): + with pytest.raises(UpgradeRefused) as refused: + stack.cli_reply(probe_type(("--root", str(comms.root), "not-a-command")), + partial(CliOutput.decoded, result=tuple[OwnerStart, ...])) + assert isinstance(refused.value.__cause__, CliRefusedError) + assert stack.cli(comms.root, ReadBackCliCommand).threads == 1 + stack.probe(DistributionVersionsProbe()) + assert tuple(commands.workers.values()) == (worker,) + assert worker.child.pid == pid + + +@pytest.mark.parametrize("arguments", [("read-back",), ("not-a-command",), ("--help",)]) +def test_cli_invocation_restores_the_callers_root( + comms: Comms, tmp_path: Path, arguments: tuple[str, ...], +) -> None: + previous = LaunchRoot.from_path(tmp_path / "caller-root") + with previous.selected(): + CliCommandProbe(("--root", str(comms.root), *arguments)).run() + assert LaunchRoot.current() == previous + + +@pytest.mark.parametrize("arguments", [("--help",), ("read-back", "--help")]) +def test_parser_help_uses_only_the_invocations_output( + arguments: tuple[str, ...], capsys: pytest.CaptureFixture[str], +) -> None: + output = StringIO() + assert CliInvocation().invoke(arguments, output) == 0 + assert output.getvalue() + assert capsys.readouterr() == ("", "") + + +def test_cli_invocation_preserves_parser_exit_status_without_console_output( + capsys: pytest.CaptureFixture[str], +) -> None: + class ExitingInvocation(CliInvocation): + def execute(self, arguments: Sequence[str], output: TextIO) -> int: + CliArgumentParser().exit(2) + + output = StringIO() + assert ExitingInvocation().invoke((), output) == 2 + assert output.getvalue() == "" + assert capsys.readouterr() == ("", "") + + +def test_cli_invocation_does_not_turn_interrupt_into_an_answer() -> None: + class InterruptedInvocation(CliInvocation): + def execute(self, arguments: Sequence[str], output: TextIO) -> int: + raise KeyboardInterrupt + + with pytest.raises(KeyboardInterrupt): + InterruptedInvocation().invoke((), StringIO()) def test_owner_reports_own_their_refusal(): @@ -73,6 +139,7 @@ def test_worker_answers_each_request_as_that_request_s_holder(monkeypatch, tmp_p @dataclass(frozen=True) class HolderSeenProbe(RuntimeProbe[str]): def run(self) -> str: + assert sys.stdin.read() == "", "Operations must not read the request transport" return os.environ.get(HOLDER_NONCE, "") try: @@ -81,9 +148,10 @@ def run(self) -> str: FieldCodec.compact(FieldCodec.encode(ProbeRequest(HolderSeenProbe(), holder))) + "\n" for holder in holders ) - # serve() takes over the descriptor of its stdout, so the answers need a real file. - answers = tmp_path / "answers" - monkeypatch.setattr("sys.stdin", io.StringIO(lines)) + # serve() owns both transport descriptors; operations inherit EOF, not requests. + requests, answers = tmp_path / "requests", tmp_path / "answers" + requests.write_text(lines) + monkeypatch.setattr("sys.stdin", requests.open()) monkeypatch.setattr("sys.stdout", answers.open("w")) monkeypatch.setenv(HOLDER_NONCE, str(ClaimId.mint())) serve() diff --git a/tests/test_page_cursor.py b/tests/test_page_cursor.py index c12e7aaa9f..53ab53d973 100644 --- a/tests/test_page_cursor.py +++ b/tests/test_page_cursor.py @@ -4,7 +4,6 @@ import pytest from agent_comms.cli import build_parser -from agent_comms.cli_command import CliCommand from agent_comms.field_codec import FieldCodec from agent_comms.page_traversal import ( HistoryCursor, @@ -27,7 +26,7 @@ def test_cursor_owns_traversal_and_cli_json(cursor, wire, source): assert FieldCodec.encode(cursor) == wire assert FieldCodec.decode(PageCursor, wire) == cursor args = build_parser().parse_args(['history', '--before', json.dumps(wire)]) - command = CliCommand.member_named(args.command).from_arguments(args) + command = args.command_declaration.from_arguments(args) assert command.before == cursor