From 63fd98ddb2317702026019934a3b9eb64591ef76 Mon Sep 17 00:00:00 2001 From: cermm <53539590+cermm@users.noreply.github.com> Date: Sun, 16 Aug 2026 19:03:45 +0200 Subject: [PATCH] feat(cli): add result metadata fd transport Refs cermm/wc-infrastructure#358 Refs cermm/wc-infrastructure#401 --- cli.py | 203 ++++- hermes_cli/_parser.py | 24 + hermes_cli/main.py | 186 ++++- hermes_cli/result_metadata.py | 373 +++++++++ tests/hermes_cli/test_result_metadata.py | 290 +++++++ tests/hermes_cli/test_result_metadata_cli.py | 797 +++++++++++++++++++ website/docs/reference/cli-commands.md | 38 + 7 files changed, 1882 insertions(+), 29 deletions(-) create mode 100644 hermes_cli/result_metadata.py create mode 100644 tests/hermes_cli/test_result_metadata.py create mode 100644 tests/hermes_cli/test_result_metadata_cli.py diff --git a/cli.py b/cli.py index 5d462805a604..dc170f023232 100644 --- a/cli.py +++ b/cli.py @@ -3697,6 +3697,7 @@ def __init__( pass_session_id: bool = False, ignore_rules: bool = False, reasoning: str = None, + result_meta_fd=None, ): """ Initialize the Hermes CLI. @@ -3877,6 +3878,7 @@ def __init__( self.max_turns = 90 else: self.max_turns = 90 + self.result_meta_fd = result_meta_fd # Parse and validate toolsets self.enabled_toolsets = toolsets @@ -4149,6 +4151,65 @@ def __init__( self._background_tasks: Dict[str, threading.Thread] = {} self._background_task_counter = 0 + def _publish_result_metadata(self, result: Any) -> None: + """Publish requested query metadata or terminate on publication failure.""" + + owner = getattr(self, "result_meta_fd", None) + if owner is None: + return + from hermes_cli.result_metadata import ( + PUBLIC_ERROR_MESSAGE, + ResultMetadataError, + build_result_metadata, + write_result_metadata_fd, + ) + + publication_failed = False + try: + metadata = build_result_metadata(result, max_iterations=self.max_turns) + write_result_metadata_fd(owner, metadata) + except ResultMetadataError: + publication_failed = True + try: + self._close_result_metadata_fd() + except ResultMetadataError: + publication_failed = True + if publication_failed: + print(PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(1) from None + + def _publish_abnormal_result_metadata(self, *, interrupted: bool) -> None: + """Publish a closed failure frame without reflecting exception details.""" + + from hermes_cli.result_metadata import MAX_API_CALLS + + api_calls = 0 + try: + summary = self.agent.get_activity_summary() + candidate = summary.get("api_call_count") + upper_bound = min(self.max_turns + 1, MAX_API_CALLS) + if type(candidate) is int and 0 <= candidate <= upper_bound: + api_calls = candidate + except Exception: + pass + self._publish_result_metadata( + { + "completed": False, + "failed": not interrupted, + "partial": False, + "interrupted": interrupted, + "api_calls": api_calls, + } + ) + + def _close_result_metadata_fd(self) -> None: + """Release the owned result-metadata descriptor exactly once.""" + + owner = getattr(self, "result_meta_fd", None) + self.result_meta_fd = None + if owner is not None: + owner.close() + def _claim_active_session(self, surface: str = "cli", *, stderr: bool = False) -> bool: """Claim a global active-session slot for this CLI process.""" if self._active_session_lease is not None: @@ -12530,6 +12591,9 @@ def run_agent(): sys.stdout.flush() time.sleep(0.15) + if getattr(self, "result_meta_fd", None) is not None: + self._publish_result_metadata(result) + # Update history with full conversation self.conversation_history = result.get("messages", self.conversation_history) if result else self.conversation_history @@ -15764,10 +15828,11 @@ def _block(reason: str) -> None: ) -def main( +def _main_impl( query: str = None, q: str = None, image: str = None, + result_meta_fd=None, toolsets: str = None, skills: str | list[str] | tuple[str, ...] = None, model: str = None, @@ -15789,6 +15854,7 @@ def main( ignore_user_config: bool = False, ignore_rules: bool = False, reasoning: str = None, + _result_meta_fd_ownership=None, ): """ Hermes Agent CLI - Interactive AI Assistant @@ -15926,7 +15992,10 @@ def main( checkpoints=checkpoints, pass_session_id=pass_session_id, ignore_rules=ignore_rules, + result_meta_fd=result_meta_fd, ) + if _result_meta_fd_ownership is not None: + _result_meta_fd_ownership.transfer_to(cli) if parsed_skills: skills_prompt, loaded_skills, missing_skills = build_preloaded_skills_prompt( @@ -16178,8 +16247,18 @@ def _signal_handler_q(signum, frame): ) except KeyboardInterrupt: _emit_interrupted_session_end(cli, reason="keyboard_interrupt") + if getattr(cli, "result_meta_fd", None) is not None: + cli._publish_abnormal_result_metadata(interrupted=True) + print(f"\nsession_id: {cli.session_id}", file=sys.stderr) + sys.exit(0) print(f"\nsession_id: {cli.session_id}", file=sys.stderr) sys.exit(130) + except Exception: + if getattr(cli, "result_meta_fd", None) is None: + raise + cli._publish_abnormal_result_metadata(interrupted=False) + print(f"\nsession_id: {cli.session_id}", file=sys.stderr) + sys.exit(0) # Sync session_id if mid-run compression created a # continuation session. The exit line below reports # session_id to stderr for automation wrappers; without @@ -16189,6 +16268,8 @@ def _signal_handler_q(signum, frame): and cli.agent.session_id != cli.session_id ): cli.session_id = cli.agent.session_id + if getattr(cli, "result_meta_fd", None) is not None: + cli._publish_result_metadata(result) response = result.get("final_response", "") if isinstance(result, dict) else str(result) # Surface backend errors that produced no visible output # (e.g. invalid model slug → provider 4xx). Mirrors the @@ -16233,7 +16314,11 @@ def _signal_handler_q(signum, frame): # permanently block the card. Non-kanban runs keep the # plain 0/1 contract automation wrappers expect. _exit_code = 0 - if isinstance(result, dict) and result.get("failed"): + if ( + result_meta_fd is None + and isinstance(result, dict) + and result.get("failed") + ): _exit_code = 1 if os.environ.get("HERMES_KANBAN_TASK") and result.get( "failure_reason" @@ -16269,7 +16354,15 @@ def _signal_handler_q(signum, frame): # Surface security advisories before the agent runs — short # banner, doesn't depend on the welcome banner being shown. cli._show_security_advisories() - cli.chat(query, images=single_query_images or None) + try: + cli.chat(query, images=single_query_images or None) + except KeyboardInterrupt: + _emit_interrupted_session_end(cli, reason="keyboard_interrupt") + if getattr(cli, "result_meta_fd", None) is not None: + cli._publish_abnormal_result_metadata(interrupted=True) + cli._print_exit_summary(clear_screen=False) + sys.exit(0) + raise cli._print_exit_summary(clear_screen=False) finally: _finalize_single_query(cli) @@ -16279,6 +16372,110 @@ def _signal_handler_q(signum, frame): cli.run() +class _ResultMetadataFDOwnershipGuard: + """Close a claimed result-metadata descriptor across every main() exit.""" + + __slots__ = ("_pending_owner", "_cli_owner") + + def __init__(self, owner) -> None: + self._pending_owner = owner + self._cli_owner = None + + def transfer_to(self, cli) -> None: + if self._pending_owner is None or self._cli_owner is not None: + raise RuntimeError("result metadata descriptor ownership already transferred") + if getattr(cli, "result_meta_fd", None) is not self._pending_owner: + raise RuntimeError("result metadata descriptor ownership transfer mismatch") + self._cli_owner = cli + self._pending_owner = None + + def close(self) -> None: + cli = self._cli_owner + self._cli_owner = None + if cli is not None: + cli._close_result_metadata_fd() + return + + owner = self._pending_owner + self._pending_owner = None + if owner is not None: + owner.close() + + +def main( + query: str = None, + q: str = None, + image: str = None, + result_meta_fd=None, + toolsets: str = None, + skills: str | list[str] | tuple[str, ...] = None, + model: str = None, + provider: str = None, + api_key: str = None, + base_url: str = None, + max_turns: int = None, + verbose: Optional[bool] = None, + quiet: bool = False, + compact: bool = False, + list_tools: bool = False, + list_toolsets: bool = False, + gateway: bool = False, + resume: str = None, + worktree: bool = False, + w: bool = False, + checkpoints: bool = False, + pass_session_id: bool = False, + ignore_user_config: bool = False, + ignore_rules: bool = False, + reasoning: str = None, +): + """Run Hermes while guarding direct-API result-metadata FD ownership.""" + + call_kwargs = locals().copy() + query = query or q + call_kwargs["query"] = query + call_kwargs["q"] = None + if result_meta_fd is None: + return _main_impl(**call_kwargs) + + from hermes_cli.result_metadata import ( + PUBLIC_ERROR_MESSAGE, + ResultMetadataError, + ResultMetadataFD, + claim_result_metadata_fd, + ) + + try: + owner = ( + result_meta_fd + if isinstance(result_meta_fd, ResultMetadataFD) + else claim_result_metadata_fd(result_meta_fd) + ) + except ResultMetadataError: + print(PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(2) from None + + ownership = _ResultMetadataFDOwnershipGuard(owner) + call_kwargs["result_meta_fd"] = owner + try: + if not query: + print("Error: --result-meta-fd requires --query.", file=sys.stderr) + raise SystemExit(2) + return _main_impl( + **call_kwargs, + _result_meta_fd_ownership=ownership, + ) + finally: + try: + ownership.close() + except ResultMetadataError: + print(PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(1) from None + + +main.__doc__ = _main_impl.__doc__ + + if __name__ == "__main__": import fire diff --git a/hermes_cli/_parser.py b/hermes_cli/_parser.py index f23906797796..56520c6306c7 100644 --- a/hermes_cli/_parser.py +++ b/hermes_cli/_parser.py @@ -12,6 +12,19 @@ import argparse +from hermes_cli.result_metadata import parse_result_metadata_fd + + +class _StoreUnique(argparse.Action): + """Store one option value and reject ambiguous repeated ownership flags.""" + + def __call__(self, parser, namespace, values, option_string=None): + if getattr(namespace, self.dest, None) is not None: + raise argparse.ArgumentError( + self, f"{option_string or self.dest} may only be specified once" + ) + setattr(namespace, self.dest, values) + # `--profile` / `-p` is consumed by ``main._apply_profile_override`` before # argparse runs (it sets ``HERMES_HOME`` and strips itself from ``sys.argv``), @@ -274,6 +287,17 @@ def build_top_level_parser(): chat_parser.add_argument( "-q", "--query", help="Single query (non-interactive mode)" ) + chat_parser.add_argument( + "--result-meta-fd", + action=_StoreUnique, + type=parse_result_metadata_fd, + metavar="FD", + default=None, + help=( + "Write JSON metadata as one frame to a pre-opened blocking anonymous " + "pipe write descriptor after --query (classic CLI, POSIX/WSL only)" + ), + ) chat_parser.add_argument( "--image", help="Optional local image path to attach to a single query" ) diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 0fb3b81feafd..e8e8d1375bcc 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -2248,9 +2248,94 @@ def _resolve_use_tui(args) -> bool: return False +def _claim_result_metadata_args(args): + """Claim chat result-metadata descriptor before startup side effects.""" + + from hermes_cli import result_metadata + + owner = None + raw_fd = getattr(args, "result_meta_fd", None) + if raw_fd is None: + return None + if isinstance(raw_fd, result_metadata.ResultMetadataFD): + return raw_fd + try: + owner = result_metadata.claim_result_metadata_fd(raw_fd) + except result_metadata.ResultMetadataError: + print(result_metadata.PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(2) from None + args.result_meta_fd = owner + return owner + + +def _early_claim_result_metadata_argv(argv: list[str]): + """Claim --result-meta-fd from raw argv before config/container startup.""" + + marker = "--result-meta-fd" + raw_values = [] + for index, token in enumerate(argv): + if token == marker: + raw_values.append(argv[index + 1] if index + 1 < len(argv) else None) + elif token.startswith(f"{marker}="): + raw_values.append(token.split("=", 1)[1]) + + if not raw_values: + return None + + from hermes_cli import result_metadata + + if len(raw_values) > 1: + print(result_metadata.PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(2) + + raw_value = raw_values[0] + if raw_value is None: + return None + + from argparse import ArgumentTypeError + + from hermes_cli._parser import parse_result_metadata_fd + + try: + return result_metadata.claim_result_metadata_fd( + parse_result_metadata_fd(raw_value) + ) + except (ValueError, ArgumentTypeError, result_metadata.ResultMetadataError): + print(result_metadata.PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(2) from None + + def cmd_chat(args): - """Run interactive chat CLI.""" + """Claim result-metadata descriptors before config or agent startup.""" + + from hermes_cli import result_metadata + + owner = _claim_result_metadata_args(args) + try: + return _cmd_chat(args) + finally: + if owner is not None: + try: + owner.close() + except result_metadata.ResultMetadataError: + print(result_metadata.PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(1) from None + + +def _cmd_chat(args): + """Run interactive chat CLI after descriptor ownership is established.""" + result_meta_fd = getattr(args, "result_meta_fd", None) + if result_meta_fd is not None and not getattr(args, "query", None): + print("Error: --result-meta-fd requires --query.", file=sys.stderr) + raise SystemExit(2) + use_tui = _resolve_use_tui(args) + if result_meta_fd is not None and use_tui: + print( + "Error: --result-meta-fd is available only in the classic CLI.", + file=sys.stderr, + ) + raise SystemExit(2) _apply_safe_mode(args) @@ -2441,6 +2526,7 @@ def cmd_chat(args): "verbose": getattr(args, "verbose", None), "quiet": getattr(args, "quiet", False), "query": args.query, + "result_meta_fd": result_meta_fd, "image": getattr(args, "image", None), "resume": getattr(args, "resume", None), "worktree": getattr(args, "worktree", False), @@ -12604,7 +12690,18 @@ def _try_termux_fast_cli_launch() -> bool: if getattr(args, "accept_hooks", False): os.environ["HERMES_ACCEPT_HOOKS"] = "1" else: - _prepare_agent_startup(args) + preclaimed_result_meta_owner = None + if getattr(args, "result_meta_fd", None) is not None: + preclaimed_result_meta_owner = _claim_result_metadata_args(args) + try: + _prepare_agent_startup(args) + except BaseException: + if preclaimed_result_meta_owner is not None: + try: + preclaimed_result_meta_owner.close() + except Exception: + pass + raise cmd_chat(args) return True @@ -12819,6 +12916,9 @@ def main(): except Exception: pass + _early_processed_argv = _coalesce_session_name_args(sys.argv[1:]) + early_result_meta_owner = _early_claim_result_metadata_argv(_early_processed_argv) + # Sweep stale ``hermes.exe.old.*`` quarantine files left by previous # ``hermes update`` runs on Windows. Silent no-op on non-Windows or when # there's nothing to clean. See ``_quarantine_running_hermes_exe``. @@ -12843,9 +12943,9 @@ def main(): except Exception: pass - if _try_termux_fast_tui_launch(): + if early_result_meta_owner is None and _try_termux_fast_tui_launch(): return - if _try_termux_fast_cli_launch(): + if early_result_meta_owner is None and _try_termux_fast_cli_launch(): return from hermes_cli._parser import build_top_level_parser @@ -14658,17 +14758,26 @@ def _export_one(session_id: str): # the managed container. This MUST run before parse_args() so that # --help, unrecognised flags, and every subcommand are forwarded # transparently instead of being intercepted by argparse on the host. + _processed_argv = _early_processed_argv + from hermes_cli.config import get_container_exec_info container_info = get_container_exec_info() if container_info: + if early_result_meta_owner is not None: + try: + early_result_meta_owner.close() + except Exception: + pass + from hermes_cli import result_metadata + + print(result_metadata.PUBLIC_ERROR_MESSAGE, file=sys.stderr) + raise SystemExit(2) from None _exec_in_container(container_info, sys.argv[1:]) # Unreachable: os.execvp never returns on success (process is replaced) # and raises OSError on failure (which propagates as a traceback). sys.exit(1) - _processed_argv = _coalesce_session_name_args(sys.argv[1:]) - # ── Defensive subparser routing (bpo-9338 workaround) ─────────── # On some Python versions (notably <3.11), argparse fails to route # subcommand tokens when the parent parser has nargs='?' optional @@ -14688,27 +14797,38 @@ def _export_one(session_id: str): t in _known_cmds for t in _processed_argv if not t.startswith("-") ) - if _has_cmd_token: - subparsers.required = True - _saved_stderr = sys.stderr - try: - sys.stderr = _io.StringIO() - args = parser.parse_args(_processed_argv) - sys.stderr = _saved_stderr - except SystemExit as exc: - sys.stderr = _saved_stderr - # Help/version flags (exit code 0) already printed output — - # re-raise immediately to avoid a second parse_args printing - # the same help text again (#10230). - if exc.code == 0: - raise - # Subcommand name was consumed as a flag value (e.g. -c model). - # Fall back to optional subparsers so argparse handles it normally. + try: + if _has_cmd_token: + subparsers.required = True + _saved_stderr = sys.stderr + try: + sys.stderr = _io.StringIO() + args = parser.parse_args(_processed_argv) + sys.stderr = _saved_stderr + except SystemExit as exc: + sys.stderr = _saved_stderr + # Help/version flags (exit code 0) already printed output — + # re-raise immediately to avoid a second parse_args printing + # the same help text again (#10230). + if exc.code == 0: + raise + # Subcommand name was consumed as a flag value (e.g. -c model). + # Fall back to optional subparsers so argparse handles it normally. + subparsers.required = False + args = parser.parse_args(_processed_argv) + else: subparsers.required = False args = parser.parse_args(_processed_argv) - else: - subparsers.required = False - args = parser.parse_args(_processed_argv) + except BaseException: + if early_result_meta_owner is not None: + try: + early_result_meta_owner.close() + except Exception: + pass + raise + + if early_result_meta_owner is not None: + args.result_meta_fd = early_result_meta_owner # Handle --version flag if args.version: @@ -14724,12 +14844,26 @@ def _export_one(session_id: str): if getattr(args, "yolo", False): os.environ["HERMES_YOLO_MODE"] = "1" + preclaimed_result_meta_owner = None + if getattr(args, "result_meta_fd", None) is not None and ( + getattr(args, "command", None) in {None, "chat"} + ): + preclaimed_result_meta_owner = _claim_result_metadata_args(args) + # Discover Python plugins and register shell hooks once, before any # command that can fire lifecycle hooks. Both are idempotent; gated # so introspection/management commands (hermes hooks list, cron # list, gateway status, mcp add, ...) don't pay discovery cost or # trigger consent prompts for hooks the user is still inspecting. - _prepare_agent_startup(args) + try: + _prepare_agent_startup(args) + except BaseException: + if preclaimed_result_meta_owner is not None: + try: + preclaimed_result_meta_owner.close() + except Exception: + pass + raise # Handle top-level --oneshot / -z: single-shot mode, stdout = final # response only, nothing else. Bypasses cli.py entirely. diff --git a/hermes_cli/result_metadata.py b/hermes_cli/result_metadata.py new file mode 100644 index 000000000000..fbb4d15ab601 --- /dev/null +++ b/hermes_cli/result_metadata.py @@ -0,0 +1,373 @@ +"""Closed-world metadata for non-interactive Hermes query results. + +This module deliberately projects the rich internal conversation result onto a +small, versioned schema. It never serializes responses, errors, prompts, +session identifiers, provider/model names, tool output, or traceback text. +""" + +from __future__ import annotations + +import json +import os +import stat +from collections.abc import Mapping +from typing import Any + +try: + import fcntl +except ImportError: # pragma: no cover - exercised on native Windows + fcntl = None # type: ignore[assignment] + +SCHEMA_VERSION = "hermes-agent-result-meta-v1" +PUBLIC_ERROR_MESSAGE = "Error: failed to publish result metadata." +MAX_METADATA_BYTES = 1024 +MAX_API_CALLS = 32 +_RESULT_KEYS = frozenset( + { + "schema_version", + "completed", + "failed", + "partial", + "interrupted", + "api_calls", + "failure_class", + } +) +_FAILURE_CLASSES = frozenset( + { + "none", + "interrupted", + "content_policy_blocked", + "provider_api_terminal", + "max_turns_or_incomplete", + "unknown_failure", + } +) +_CLAIM_TOKEN = object() + + +class ResultMetadataError(RuntimeError): + """The requested metadata cannot be projected or published safely.""" + + +class ResultMetadataFD: + """Single owner for a validated result-metadata FIFO write endpoint.""" + + __slots__ = ("_fd",) + + def __init__(self, fd: int, *, _claim_token: object | None = None) -> None: + if _claim_token is not _CLAIM_TOKEN: + raise ResultMetadataError( + "result metadata descriptor owner must be claimed" + ) + self._fd = fd + + @property + def closed(self) -> bool: + return self._fd < 0 + + def fileno(self) -> int: + if self.closed: + raise ResultMetadataError("result metadata descriptor is closed") + return self._fd + + def close(self) -> None: + if self.closed: + return + fd = self._fd + self._fd = -1 + try: + os.close(fd) + except OSError as exc: + raise ResultMetadataError( + "result metadata descriptor close failed" + ) from exc + + +def parse_result_metadata_fd(value: str) -> int: + """Parse argparse input as a canonical decimal descriptor number.""" + + if not isinstance(value, str) or not value.isascii() or not value.isdecimal(): + raise ValueError("result metadata descriptor must be a canonical integer") + if value != str(int(value)): + raise ValueError("result metadata descriptor must be a canonical integer") + fd = int(value) + if fd < 3: + raise ValueError("result metadata descriptor must be at least 3") + return fd + + +def _validate_result_metadata_fd(fd: Any) -> int: + if os.name != "posix" or fcntl is None: + raise ResultMetadataError( + "result metadata descriptor transport requires POSIX" + ) + if type(fd) is not int or fd < 3: + raise ResultMetadataError( + "result metadata descriptor must be an integer at least 3" + ) + try: + opened = os.fstat(fd) + flags = fcntl.fcntl(fd, fcntl.F_GETFL) + except (OSError, OverflowError, TypeError, ValueError) as exc: + raise ResultMetadataError( + "result metadata descriptor is invalid or closed" + ) from exc + if not stat.S_ISFIFO(opened.st_mode): + raise ResultMetadataError("result metadata descriptor must be a FIFO") + descriptor_target = None + for descriptor_root in ("/proc/self/fd", "/dev/fd"): + try: + descriptor_target = os.readlink(f"{descriptor_root}/{fd}") + break + except OSError: + continue + if descriptor_target is None: + raise ResultMetadataError( + "result metadata descriptor identity is unavailable" + ) + if not descriptor_target.startswith("pipe:"): + raise ResultMetadataError( + "result metadata descriptor must be an anonymous pipe" + ) + if flags & os.O_ACCMODE != os.O_WRONLY: + raise ResultMetadataError( + "result metadata descriptor must be an anonymous-pipe write endpoint" + ) + if flags & os.O_NONBLOCK: + raise ResultMetadataError("result metadata descriptor must be blocking") + try: + pipe_buf = os.fpathconf(fd, "PC_PIPE_BUF") + except (OSError, OverflowError, TypeError, ValueError) as exc: + raise ResultMetadataError( + "result metadata pipe atomic-write bound is unavailable" + ) from exc + if type(pipe_buf) is not int or pipe_buf < MAX_METADATA_BYTES: + raise ResultMetadataError( + "result metadata pipe atomic-write bound is too small" + ) + return fd + + +def _close_unclaimed_result_metadata_fd(fd: Any) -> None: + """Best-effort cleanup for a descriptor rejected before ownership exists.""" + + if type(fd) is not int or fd < 3: + return + try: + os.close(fd) + except (OSError, OverflowError): + pass + + +def claim_result_metadata_fd(fd: Any) -> ResultMetadataFD: + """Validate and take ownership of a pre-opened metadata descriptor.""" + + try: + validated_fd = _validate_result_metadata_fd(fd) + os.set_inheritable(validated_fd, False) + except ResultMetadataError: + _close_unclaimed_result_metadata_fd(fd) + raise + except OSError as exc: + _close_unclaimed_result_metadata_fd(fd) + raise ResultMetadataError( + "result metadata descriptor could not be isolated" + ) from exc + return ResultMetadataFD(validated_fd, _claim_token=_CLAIM_TOKEN) + + +def _api_call_count( + result: Mapping[str, Any], max_iterations: int +) -> tuple[int, bool]: + if type(max_iterations) is not int or max_iterations < 0: + max_iterations = 90 + upper_bound = min(max_iterations + 1, MAX_API_CALLS) + value = result.get("api_calls") + if "api_calls" in result and type(value) is int and 0 <= value <= upper_bound: + return value, True + return 0, False + + +def _strict_statuses(result: Mapping[str, Any]) -> tuple[dict[str, bool], bool]: + statuses: dict[str, bool] = {} + valid = True + for key in ("completed", "failed", "partial", "interrupted"): + defaultable = key != "completed" + value = result.get(key, False) + if (key not in result and not defaultable) or type(value) is not bool: + valid = False + value = False + statuses[key] = value + return statuses, valid + + +def _is_max_turn_or_incomplete( + result: Mapping[str, Any], statuses: Mapping[str, bool] +) -> bool: + if statuses["partial"] or not statuses["completed"]: + return True + exit_reason = result.get("turn_exit_reason") + return isinstance(exit_reason, str) and ( + exit_reason.startswith("max_iterations_reached(") + or exit_reason + in {"budget_exhausted", "all_retries_exhausted_no_response"} + ) + + +def _is_trusted_provider_failure_reason(value: Any) -> bool: + if not isinstance(value, str): + return False + try: + from agent.error_classifier import FailoverReason + + FailoverReason(value) + return True + except (ImportError, ValueError): + return False + + +def _failure_class_invariant_error( + failure_class: str, statuses: Mapping[str, bool] +) -> str | None: + completed = statuses["completed"] + failed = statuses["failed"] + partial = statuses["partial"] + interrupted = statuses["interrupted"] + multi_status = sum(int(value) for value in statuses.values()) > 1 + if failure_class != "unknown_failure" and multi_status: + return "non-unknown metadata violates status exclusivity invariants" + if failure_class == "none" and not ( + completed and not failed and not partial and not interrupted + ): + return "success metadata violates status invariants" + if failure_class == "interrupted" and ( + not interrupted or completed or failed or partial + ): + return "interrupted metadata violates status invariants" + if failure_class in {"content_policy_blocked", "provider_api_terminal"} and not failed: + return "terminal failure metadata violates status invariants" + if failure_class == "max_turns_or_incomplete" and not ( + partial and not completed and not failed and not interrupted + ): + return "incomplete metadata violates status invariants" + return None + + +def build_result_metadata( + result: Any, *, max_iterations: int +) -> dict[str, Any]: + """Project a trusted turn result onto the public v1 metadata schema.""" + + if not isinstance(result, Mapping): + result = {} + valid_shape = False + else: + valid_shape = True + + statuses, statuses_valid = _strict_statuses(result) + api_calls, api_calls_valid = _api_call_count(result, max_iterations) + valid = valid_shape and statuses_valid and api_calls_valid + failure_class = "unknown_failure" + + if valid and sum(int(value) for value in statuses.values()) > 1: + pass + elif valid and statuses["interrupted"]: + failure_class = "interrupted" + elif ( + valid + and statuses["failed"] + and isinstance(result.get("error"), str) + and result["error"].startswith("content_policy_blocked:") + ): + failure_class = "content_policy_blocked" + elif valid and statuses["failed"] and _is_trusted_provider_failure_reason( + result.get("failure_reason") + ): + failure_class = "provider_api_terminal" + elif valid and statuses["failed"]: + pass + elif valid and _is_max_turn_or_incomplete(result, statuses): + failure_class = "max_turns_or_incomplete" + elif valid and statuses["completed"]: + failure_class = "none" + + if _failure_class_invariant_error(failure_class, statuses) is not None: + failure_class = "unknown_failure" + + return { + "schema_version": SCHEMA_VERSION, + "completed": statuses["completed"], + "failed": statuses["failed"], + "partial": statuses["partial"], + "interrupted": statuses["interrupted"], + "api_calls": api_calls, + "failure_class": failure_class, + } + + +def serialize_result_metadata(metadata: Mapping[str, Any]) -> bytes: + """Return deterministic UTF-8 JSON with exactly the v1 public keys.""" + + if set(metadata) != _RESULT_KEYS: + raise ResultMetadataError("metadata does not match the closed public schema") + if metadata.get("schema_version") != SCHEMA_VERSION: + raise ResultMetadataError("unsupported metadata schema version") + if metadata.get("failure_class") not in _FAILURE_CLASSES: + raise ResultMetadataError("unsupported failure class") + for key in ("completed", "failed", "partial", "interrupted"): + if type(metadata.get(key)) is not bool: + raise ResultMetadataError("status fields must be strict booleans") + if ( + type(metadata.get("api_calls")) is not int + or not 0 <= metadata["api_calls"] <= MAX_API_CALLS + ): + raise ResultMetadataError("api_calls must be a bounded non-negative integer") + + statuses = { + key: metadata[key] + for key in ("completed", "failed", "partial", "interrupted") + } + invariant_error = _failure_class_invariant_error( + metadata["failure_class"], statuses + ) + if invariant_error is not None: + raise ResultMetadataError(invariant_error) + + payload = ( + json.dumps( + dict(metadata), + ensure_ascii=True, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ) + + "\n" + ).encode("utf-8") + if len(payload) > MAX_METADATA_BYTES: + raise ResultMetadataError("result metadata exceeds the fixed size bound") + return payload + + +def write_result_metadata_fd( + owner: ResultMetadataFD, metadata: Mapping[str, Any] +) -> dict[str, Any]: + """Publish one bounded atomic frame through a validated FIFO writer.""" + + if not isinstance(owner, ResultMetadataFD): + raise ResultMetadataError("result metadata descriptor owner is invalid") + payload = serialize_result_metadata(metadata) + fd = owner.fileno() + try: + written = os.write(fd, payload) + if written != len(payload): + raise ResultMetadataError( + "short write while publishing result metadata" + ) + except ResultMetadataError: + raise + except (OSError, TypeError, ValueError) as exc: + raise ResultMetadataError( + "could not write result metadata descriptor" + ) from exc + return dict(metadata) diff --git a/tests/hermes_cli/test_result_metadata.py b/tests/hermes_cli/test_result_metadata.py new file mode 100644 index 000000000000..0c7f583719c9 --- /dev/null +++ b/tests/hermes_cli/test_result_metadata.py @@ -0,0 +1,290 @@ +from __future__ import annotations + +import errno +import json +import os +import tempfile + +import pytest + + +def _success_result() -> dict[str, object]: + return { + "completed": True, + "failed": False, + "partial": False, + "interrupted": False, + "api_calls": 1, + "final_response": "secret response marker", + "error": "secret error marker", + "messages": [{"role": "user", "content": "secret prompt marker"}], + "provider": "secret provider marker", + "model": "secret model marker", + "tool_output": "secret tool-output marker", + } + + +def test_result_metadata_shape_is_closed_world_bounded_and_secret_free(): + from hermes_cli.result_metadata import ( + MAX_METADATA_BYTES, + SCHEMA_VERSION, + build_result_metadata, + serialize_result_metadata, + ) + + metadata = build_result_metadata(_success_result(), max_iterations=3) + encoded = serialize_result_metadata(metadata) + + assert metadata == { + "schema_version": SCHEMA_VERSION, + "completed": True, + "failed": False, + "partial": False, + "interrupted": False, + "api_calls": 1, + "failure_class": "none", + } + assert json.loads(encoded) == metadata + assert len(encoded) <= MAX_METADATA_BYTES + for marker in ( + b"secret response marker", + b"secret error marker", + b"secret prompt marker", + b"secret provider marker", + b"secret model marker", + b"secret tool-output marker", + ): + assert marker not in encoded + + +def test_result_metadata_fd_claim_accepts_only_blocking_fifo_writer(): + from hermes_cli.result_metadata import ( + ResultMetadataError, + claim_result_metadata_fd, + ) + + read_fd, write_fd = os.pipe() + owner = claim_result_metadata_fd(write_fd) + try: + assert owner.fileno() == write_fd + assert os.get_inheritable(write_fd) is False + finally: + owner.close() + os.close(read_fd) + + with pytest.raises(OSError): + os.fstat(write_fd) + + with pytest.raises(ResultMetadataError): + claim_result_metadata_fd(10**100) + + read_fd, write_fd = os.pipe() + try: + with pytest.raises(ResultMetadataError): + claim_result_metadata_fd(read_fd) + with pytest.raises(OSError): + os.fstat(read_fd) + finally: + os.close(write_fd) + + +def test_result_metadata_fd_claim_rejects_unsupported_platform(monkeypatch): + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + monkeypatch.setattr(result_metadata, "fcntl", None) + try: + with pytest.raises(result_metadata.ResultMetadataError): + result_metadata.claim_result_metadata_fd(write_fd) + with pytest.raises(OSError): + os.fstat(write_fd) + finally: + os.close(read_fd) + + +def test_result_metadata_fd_claim_closes_when_isolation_fails(monkeypatch): + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + + def fail_set_inheritable(_fd, _inheritable): + raise OSError(errno.EIO, "secret isolation detail") + + monkeypatch.setattr(result_metadata.os, "set_inheritable", fail_set_inheritable) + try: + with pytest.raises(result_metadata.ResultMetadataError): + result_metadata.claim_result_metadata_fd(write_fd) + with pytest.raises(OSError): + os.fstat(write_fd) + finally: + os.close(read_fd) + + +def test_result_metadata_fd_claim_rejects_named_fifo(): + from hermes_cli import result_metadata + + with tempfile.TemporaryDirectory() as tmpdir: + fifo_path = os.path.join(tmpdir, "result-meta") + os.mkfifo(fifo_path) + read_fd = os.open(fifo_path, os.O_RDONLY | os.O_NONBLOCK) + write_fd = os.open(fifo_path, os.O_WRONLY) + try: + with pytest.raises(result_metadata.ResultMetadataError): + result_metadata.claim_result_metadata_fd(write_fd) + with pytest.raises(OSError): + os.fstat(write_fd) + finally: + os.close(read_fd) + + +def test_result_metadata_fd_claim_uses_dev_fd_when_proc_fd_missing(monkeypatch): + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + real_readlink = os.readlink + + def fake_readlink(path): + if path.startswith("/proc/self/fd/"): + raise OSError("procfs unavailable") + return real_readlink(path) + + monkeypatch.setattr(result_metadata.os, "readlink", fake_readlink) + owner = result_metadata.claim_result_metadata_fd(write_fd) + try: + assert owner.fileno() == write_fd + finally: + owner.close() + os.close(read_fd) + + +def test_result_metadata_fd_owner_cannot_be_directly_constructed(): + from hermes_cli.result_metadata import ResultMetadataError, ResultMetadataFD + + with pytest.raises(ResultMetadataError): + ResultMetadataFD(3) + + +@pytest.mark.parametrize("invalid", [True, False, "3", 3.0, None, 0, 1, 2]) +def test_result_metadata_fd_claim_rejects_noncanonical_values(invalid): + from hermes_cli.result_metadata import ResultMetadataError, claim_result_metadata_fd + + with pytest.raises(ResultMetadataError): + claim_result_metadata_fd(invalid) + + +def test_result_metadata_fd_writes_one_exact_frame_and_closes(): + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + owner = result_metadata.claim_result_metadata_fd(write_fd) + metadata = result_metadata.build_result_metadata( + _success_result(), max_iterations=3 + ) + try: + result_metadata.write_result_metadata_fd(owner, metadata) + assert os.read(read_fd, result_metadata.MAX_METADATA_BYTES) == ( + result_metadata.serialize_result_metadata(metadata) + ) + finally: + owner.close() + os.close(read_fd) + + +def test_result_metadata_fd_close_fault_is_fail_closed(monkeypatch): + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + owner = result_metadata.claim_result_metadata_fd(write_fd) + real_close = os.close + + def fail_close(_fd): + raise OSError(errno.EIO, "secret close detail") + + monkeypatch.setattr(result_metadata.os, "close", fail_close) + try: + with pytest.raises(result_metadata.ResultMetadataError): + owner.close() + assert owner.closed is True + finally: + real_close(write_fd) + real_close(read_fd) + + +@pytest.mark.parametrize("fault", ["short", "epipe", "eagain"]) +def test_result_metadata_fd_write_fails_closed_without_retry(monkeypatch, fault): + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + owner = result_metadata.claim_result_metadata_fd(write_fd) + calls = 0 + + def faulty_write(_fd, payload): + nonlocal calls + calls += 1 + if fault == "short": + return len(payload) - 1 + error_number = errno.EPIPE if fault == "epipe" else errno.EAGAIN + raise OSError(error_number, fault) + + monkeypatch.setattr(result_metadata.os, "write", faulty_write) + try: + with pytest.raises(result_metadata.ResultMetadataError): + result_metadata.write_result_metadata_fd( + owner, + result_metadata.build_result_metadata( + _success_result(), max_iterations=3 + ), + ) + assert calls == 1 + finally: + owner.close() + os.close(read_fd) + + +def test_result_metadata_api_calls_match_consumer_bound(): + from hermes_cli import result_metadata + + accepted = result_metadata.build_result_metadata( + {**_success_result(), "api_calls": result_metadata.MAX_API_CALLS}, + max_iterations=90, + ) + rejected = result_metadata.build_result_metadata( + {**_success_result(), "api_calls": result_metadata.MAX_API_CALLS + 1}, + max_iterations=90, + ) + + assert accepted["api_calls"] == result_metadata.MAX_API_CALLS + assert accepted["failure_class"] == "none" + assert rejected["api_calls"] == 0 + assert rejected["failure_class"] == "unknown_failure" + with pytest.raises(result_metadata.ResultMetadataError): + result_metadata.serialize_result_metadata( + {**accepted, "api_calls": result_metadata.MAX_API_CALLS + 1} + ) + + +def test_result_metadata_contradictory_statuses_fail_to_unknown_failure(): + from hermes_cli import result_metadata + + metadata = result_metadata.build_result_metadata( + { + "completed": False, + "failed": True, + "partial": False, + "interrupted": True, + "api_calls": 1, + }, + max_iterations=90, + ) + + assert metadata["failed"] is True + assert metadata["interrupted"] is True + assert metadata["failure_class"] == "unknown_failure" + with pytest.raises(result_metadata.ResultMetadataError): + result_metadata.serialize_result_metadata( + {**metadata, "failure_class": "interrupted"} + ) + with pytest.raises(result_metadata.ResultMetadataError): + result_metadata.serialize_result_metadata( + {**metadata, "failure_class": "provider_api_terminal"} + ) diff --git a/tests/hermes_cli/test_result_metadata_cli.py b/tests/hermes_cli/test_result_metadata_cli.py new file mode 100644 index 000000000000..560b24ed8cdc --- /dev/null +++ b/tests/hermes_cli/test_result_metadata_cli.py @@ -0,0 +1,797 @@ +from __future__ import annotations + +import os +import sys +import types + +import pytest + + +def _install_fake_cli_dependencies(monkeypatch, fake_main): + fake_cli = types.ModuleType("cli") + fake_cli.main = fake_main + monkeypatch.setitem(sys.modules, "cli", fake_cli) + + +def _install_direct_api_fake_cli(monkeypatch): + import cli as cli_mod + + real_cli = cli_mod.HermesCLI + + class FakeCLI(real_cli): + def __init__(self, **kwargs): + self.result_meta_fd = kwargs.get("result_meta_fd") + self.session_id = "session" + self.system_prompt = "" + self.preloaded_skills = [] + + def show_banner(self): + pass + + def show_tools(self): + pass + + monkeypatch.setattr(cli_mod, "HermesCLI", FakeCLI) + return cli_mod + + +def test_parser_result_meta_fd_is_opt_in_and_canonical(): + from hermes_cli._parser import build_top_level_parser + + parser, _subparsers, _chat = build_top_level_parser() + + absent = parser.parse_args(["chat", "--query", "hello"]) + present = parser.parse_args( + ["chat", "--query", "hello", "--result-meta-fd", "9"] + ) + + assert absent.result_meta_fd is None + assert present.result_meta_fd == 9 + + for value in ("03", "+3", "-3", "3.0", "true", "2"): + with pytest.raises(SystemExit) as raised: + parser.parse_args( + ["chat", "--query", "hello", "--result-meta-fd", value] + ) + assert raised.value.code == 2 + + +@pytest.mark.parametrize( + "duplicate_args", + [ + ["--result-meta-fd", "7", "--result-meta-fd", "9"], + ["--result-meta-fd=7", "--result-meta-fd=9"], + ["--result-meta-fd", "7", "--result-meta-fd=9"], + ], +) +def test_parser_rejects_duplicate_result_meta_fd(duplicate_args): + from hermes_cli._parser import build_top_level_parser + + parser, _subparsers, _chat = build_top_level_parser() + + with pytest.raises(SystemExit) as raised: + parser.parse_args(["chat", "--query", "hello", *duplicate_args]) + + assert raised.value.code == 2 + + +def test_cmd_chat_claims_forwards_and_closes_result_meta_fd(monkeypatch): + import hermes_cli.main as main_mod + from hermes_cli._parser import build_top_level_parser + + read_fd, write_fd = os.pipe() + captured = {} + + def fake_main(**kwargs): + owner = kwargs["result_meta_fd"] + captured["fd"] = owner.fileno() + assert os.get_inheritable(write_fd) is False + + _install_fake_cli_dependencies(monkeypatch, fake_main) + monkeypatch.setattr(main_mod, "_has_any_provider_configured", lambda: True) + monkeypatch.setattr(main_mod, "_pin_kanban_board_env", lambda: None) + monkeypatch.setattr(main_mod, "_termux_should_prefetch_update_check", lambda: False) + monkeypatch.setattr(main_mod, "_sync_bundled_skills_for_startup", lambda: None) + + parser, _subparsers, chat = build_top_level_parser() + chat.set_defaults(func=main_mod.cmd_chat) + main_mod.cmd_chat( + parser.parse_args( + ["chat", "--query", "hello", "--result-meta-fd", str(write_fd)] + ) + ) + + assert captured["fd"] == write_fd + with pytest.raises(OSError): + os.fstat(write_fd) + os.close(read_fd) + + +def test_cmd_chat_real_bridge_publishes_metadata(monkeypatch, capsys): + import signal + + import cli as cli_mod + import hermes_cli.main as main_mod + from hermes_cli import result_metadata + from hermes_cli._parser import build_top_level_parser + + result = { + "completed": True, + "failed": False, + "partial": False, + "interrupted": False, + "api_calls": 1, + "final_response": "exact response", + "messages": [], + } + real_cli = cli_mod.HermesCLI + + class FakeAgent: + session_id = "session" + quiet_mode = False + suppress_status_output = False + stream_delta_callback = object() + tool_gen_callback = object() + + def run_conversation(self, *_args, **_kwargs): + return dict(result) + + class FakeCLI(real_cli): + def __init__(self, **kwargs): + self.result_meta_fd = kwargs.get("result_meta_fd") + self.max_turns = kwargs.get("max_turns") or 90 + self.agent = FakeAgent() + self.session_id = "session" + self.conversation_history = [] + self._active_agent_route_signature = "same" + + def _claim_active_session(self, *_args, **_kwargs): + return True + + def _release_active_session(self): + pass + + def _ensure_runtime_credentials(self): + return True + + def _resolve_turn_agent_config(self, _query): + return { + "signature": "same", + "model": None, + "runtime": None, + "request_overrides": None, + } + + def _init_agent(self, **_kwargs): + return True + + monkeypatch.setattr(cli_mod, "HermesCLI", FakeCLI) + monkeypatch.setattr(cli_mod, "_finalize_single_query", lambda _cli: None) + monkeypatch.setattr(signal, "signal", lambda *_args: None) + monkeypatch.setattr(main_mod, "_has_any_provider_configured", lambda: True) + monkeypatch.setattr(main_mod, "_pin_kanban_board_env", lambda: None) + monkeypatch.setattr(main_mod, "_termux_should_prefetch_update_check", lambda: False) + monkeypatch.setattr(main_mod, "_sync_bundled_skills_for_startup", lambda: None) + + parser, _subparsers, chat = build_top_level_parser() + chat.set_defaults(func=main_mod.cmd_chat) + read_fd, write_fd = os.pipe() + with pytest.raises(SystemExit) as raised: + main_mod.cmd_chat( + parser.parse_args( + [ + "chat", + "--query", + "hello", + "--quiet", + "--toolsets", + "safe", + "--result-meta-fd", + str(write_fd), + ] + ) + ) + captured = capsys.readouterr() + payload = os.read(read_fd, result_metadata.MAX_METADATA_BYTES) + os.close(read_fd) + + assert raised.value.code == 0 + assert captured.out == "exact response\n" + assert captured.err == "\nsession_id: session\n" + assert result_metadata.serialize_result_metadata( + result_metadata.build_result_metadata(result, max_iterations=90) + ) == payload + + +def test_invalid_result_meta_fd_fails_before_config_or_provider(monkeypatch, capsys): + import hermes_cli.main as main_mod + from hermes_cli import result_metadata + from hermes_cli._parser import build_top_level_parser + + read_fd, closed_fd = os.pipe() + os.close(closed_fd) + monkeypatch.setattr( + main_mod, + "_resolve_use_tui", + lambda _args: (_ for _ in ()).throw(AssertionError("config resolution ran")), + ) + monkeypatch.setattr( + main_mod, + "_has_any_provider_configured", + lambda: (_ for _ in ()).throw(AssertionError("provider check ran")), + ) + parser, _subparsers, chat = build_top_level_parser() + chat.set_defaults(func=main_mod.cmd_chat) + + with pytest.raises(SystemExit) as raised: + main_mod.cmd_chat( + parser.parse_args( + ["chat", "--query", "hello", "--result-meta-fd", str(closed_fd)] + ) + ) + + captured = capsys.readouterr() + assert raised.value.code == 2 + assert captured.out == "" + assert captured.err == result_metadata.PUBLIC_ERROR_MESSAGE + "\n" + os.close(read_fd) + + +@pytest.mark.parametrize("raw_fd", ["999999", "03"]) +def test_main_invalid_result_meta_fd_fails_before_startup(monkeypatch, capsys, raw_fd): + import hermes_cli.config as config_mod + import hermes_cli.main as main_mod + from hermes_cli import result_metadata + + events = [] + + def record_event(name): + def _inner(*_args, **_kwargs): + events.append(name) + raise AssertionError(f"{name} must not run before fd claim") + return _inner + + monkeypatch.setattr(main_mod, "_cleanup_quarantined_exes", record_event("cleanup_quarantined_exes")) + monkeypatch.setattr(main_mod, "_recover_from_interrupted_install", record_event("recover_from_interrupted_install")) + monkeypatch.setattr(config_mod, "get_container_exec_info", record_event("get_container_exec_info")) + monkeypatch.setattr(main_mod, "_prepare_agent_startup", record_event("prepare_agent_startup")) + monkeypatch.setattr(main_mod, "_try_termux_fast_cli_launch", lambda: False) + monkeypatch.setattr(main_mod, "_try_termux_fast_tui_launch", lambda: False) + monkeypatch.setattr( + sys, + "argv", + ["hermes", "chat", "--query", "hello", "--result-meta-fd", raw_fd], + ) + + with pytest.raises(SystemExit) as raised: + main_mod.main() + + captured = capsys.readouterr() + assert raised.value.code == 2 + assert events == [] + assert captured.out == "" + assert captured.err == result_metadata.PUBLIC_ERROR_MESSAGE + "\n" + + +def test_main_duplicate_result_meta_fd_rejects_before_claim_or_startup( + monkeypatch, capsys +): + import hermes_cli.config as config_mod + import hermes_cli.main as main_mod + from hermes_cli import result_metadata + + events = [] + + def record_event(name): + def _inner(*_args, **_kwargs): + events.append(name) + raise AssertionError(f"{name} must not run for duplicate result metadata FDs") + + return _inner + + monkeypatch.setattr( + result_metadata, + "claim_result_metadata_fd", + record_event("claim_result_metadata_fd"), + ) + monkeypatch.setattr( + main_mod, + "_cleanup_quarantined_exes", + record_event("cleanup_quarantined_exes"), + ) + monkeypatch.setattr( + main_mod, + "_recover_from_interrupted_install", + record_event("recover_from_interrupted_install"), + ) + monkeypatch.setattr( + config_mod, + "get_container_exec_info", + record_event("get_container_exec_info"), + ) + monkeypatch.setattr( + main_mod, + "_exec_in_container", + record_event("exec_in_container"), + ) + monkeypatch.setattr( + main_mod, + "_prepare_agent_startup", + record_event("prepare_agent_startup"), + ) + monkeypatch.setattr( + main_mod, + "_try_termux_fast_cli_launch", + record_event("termux_fast_cli_launch"), + ) + monkeypatch.setattr( + main_mod, + "_try_termux_fast_tui_launch", + record_event("termux_fast_tui_launch"), + ) + + read_fds = [] + write_fds = [] + try: + for _ in range(2): + read_fd, write_fd = os.pipe() + os.set_inheritable(write_fd, True) + read_fds.append(read_fd) + write_fds.append(write_fd) + monkeypatch.setattr( + sys, + "argv", + [ + "hermes", + "chat", + "--query", + "hello", + "--result-meta-fd", + str(write_fds[0]), + f"--result-meta-fd={write_fds[1]}", + ], + ) + + with pytest.raises(SystemExit) as raised: + main_mod.main() + + captured = capsys.readouterr() + assert raised.value.code == 2 + assert events == [] + assert captured.out == "" + assert captured.err == result_metadata.PUBLIC_ERROR_MESSAGE + "\n" + for write_fd in write_fds: + os.fstat(write_fd) + assert os.get_inheritable(write_fd) is True + finally: + for fd in [*write_fds, *read_fds]: + os.close(fd) + + +def test_nonquiet_query_interrupt_publishes_result_metadata(monkeypatch, capsys): + import cli as cli_mod + from hermes_cli import result_metadata + + real_cli = cli_mod.HermesCLI + + class FakeConsole: + def print(self, *_args, **_kwargs): + pass + + class FakeAgent: + def get_activity_summary(self): + return {"api_call_count": 1} + + class FakeCLI(real_cli): + def __init__(self, **kwargs): + self.result_meta_fd = kwargs.get("result_meta_fd") + self.max_turns = kwargs.get("max_turns") or 90 + self.session_id = "session" + self.console = FakeConsole() + self.agent = FakeAgent() + self.conversation_history = [] + self._active_agent_route_signature = "same" + + def _claim_active_session(self, *_args, **_kwargs): + return True + + def _release_active_session(self): + pass + + def _show_security_advisories(self): + pass + + def chat(self, *_args, **_kwargs): + raise KeyboardInterrupt() + + def _print_exit_summary(self, clear_screen=False): + print(f"\nsession_id: {self.session_id}", file=sys.stderr) + + monkeypatch.setattr(cli_mod, "HermesCLI", FakeCLI) + monkeypatch.setattr(cli_mod, "_finalize_single_query", lambda _cli: None) + read_fd, write_fd = os.pipe() + with pytest.raises(SystemExit) as raised: + cli_mod.main(query="hello", quiet=False, result_meta_fd=write_fd) + + captured = capsys.readouterr() + payload = os.read(read_fd, result_metadata.MAX_METADATA_BYTES) + os.close(read_fd) + metadata = result_metadata.build_result_metadata( + { + "completed": False, + "failed": False, + "partial": False, + "interrupted": True, + "api_calls": 1, + }, + max_iterations=90, + ) + + assert raised.value.code == 0 + assert captured.err == "\nsession_id: session\n" + assert result_metadata.serialize_result_metadata(metadata) == payload + + +@pytest.mark.parametrize("extra_args", [[], ["--tui", "--query", "hello"]]) +def test_cmd_chat_rejects_result_meta_fd_without_query_or_with_tui( + monkeypatch, extra_args +): + import hermes_cli.main as main_mod + from hermes_cli._parser import build_top_level_parser + + read_fd, write_fd = os.pipe() + monkeypatch.setattr( + main_mod, + "_has_any_provider_configured", + lambda: (_ for _ in ()).throw(AssertionError("provider check ran")), + ) + parser, _subparsers, chat = build_top_level_parser() + chat.set_defaults(func=main_mod.cmd_chat) + + with pytest.raises(SystemExit) as raised: + main_mod.cmd_chat( + parser.parse_args( + ["chat", *extra_args, "--result-meta-fd", str(write_fd)] + ) + ) + + assert raised.value.code == 2 + with pytest.raises(OSError): + os.fstat(write_fd) + os.close(read_fd) + + +def test_publish_result_metadata_fd_is_silent_and_closes_owner(capsys): + from cli import HermesCLI + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + cli = HermesCLI.__new__(HermesCLI) + cli.result_meta_fd = result_metadata.claim_result_metadata_fd(write_fd) + cli.max_turns = 7 + + cli._publish_result_metadata( + { + "completed": True, + "failed": False, + "partial": False, + "interrupted": False, + "api_calls": 2, + "final_response": "raw response must not leak", + } + ) + + captured = capsys.readouterr() + payload = os.read(read_fd, result_metadata.MAX_METADATA_BYTES) + assert captured.out == captured.err == "" + assert b'"api_calls":2' in payload + assert b"raw response must not leak" not in payload + with pytest.raises(OSError): + os.fstat(write_fd) + os.close(read_fd) + + +def test_publish_result_metadata_fd_write_failure_is_fixed_and_secret_free( + monkeypatch, capsys +): + from cli import HermesCLI + from hermes_cli import result_metadata + + read_fd, write_fd = os.pipe() + cli = HermesCLI.__new__(HermesCLI) + cli.result_meta_fd = result_metadata.claim_result_metadata_fd(write_fd) + cli.max_turns = 7 + + def fail_write(_fd, _payload): + raise OSError("secret provider detail") + + monkeypatch.setattr(result_metadata.os, "write", fail_write) + + with pytest.raises(SystemExit) as raised: + cli._publish_result_metadata( + { + "completed": True, + "failed": False, + "partial": False, + "interrupted": False, + "api_calls": 1, + "final_response": "raw response must not leak", + } + ) + + captured = capsys.readouterr() + assert raised.value.code == 1 + assert captured.out == "" + assert captured.err == result_metadata.PUBLIC_ERROR_MESSAGE + "\n" + assert "secret provider detail" not in captured.err + assert "raw response must not leak" not in captured.err + with pytest.raises(OSError): + os.fstat(write_fd) + os.close(read_fd) + + +def test_direct_api_closes_result_meta_fd_on_post_construction_error(monkeypatch): + cli_mod = _install_direct_api_fake_cli(monkeypatch) + read_fd, write_fd = os.pipe() + + with pytest.raises(ValueError, match=r"Unknown skill\(s\): __missing_skill__"): + cli_mod.main( + query="x", + quiet=True, + toolsets="safe", + skills="__missing_skill__", + result_meta_fd=write_fd, + ) + + with pytest.raises(OSError): + os.fstat(write_fd) + os.close(read_fd) + + +def test_direct_api_closes_result_meta_fd_on_pre_query_exit(monkeypatch): + cli_mod = _install_direct_api_fake_cli(monkeypatch) + read_fd, write_fd = os.pipe() + + with pytest.raises(SystemExit) as raised: + cli_mod.main(query="x", list_tools=True, result_meta_fd=write_fd) + + assert raised.value.code == 0 + with pytest.raises(OSError): + os.fstat(write_fd) + os.close(read_fd) + + +def test_direct_api_accepts_claimed_result_meta_owner(monkeypatch): + import cli as cli_mod + from hermes_cli import result_metadata + + _install_direct_api_fake_cli(monkeypatch) + read_fd, write_fd = os.pipe() + owner = result_metadata.claim_result_metadata_fd(write_fd) + + with pytest.raises(SystemExit) as raised: + cli_mod.main(query="x", list_tools=True, result_meta_fd=owner) + + assert raised.value.code == 0 + assert owner.closed is True + with pytest.raises(OSError): + os.fstat(write_fd) + os.close(read_fd) + + +@pytest.mark.parametrize("failed", [False, True]) +def test_quiet_query_preserves_legacy_exit_without_metadata_and_uses_frame_exit_contract( + monkeypatch, capsys, failed +): + import signal + + import cli as cli_mod + from hermes_cli import result_metadata + + result = { + "completed": not failed, + "failed": failed, + "partial": False, + "interrupted": False, + "api_calls": 1, + "final_response": "" if failed else "exact response", + "error": "secret failure detail" if failed else "", + "messages": [], + } + real_cli = cli_mod.HermesCLI + + class FakeAgent: + session_id = "session" + quiet_mode = False + suppress_status_output = False + stream_delta_callback = object() + tool_gen_callback = object() + + def run_conversation(self, *_args, **_kwargs): + return dict(result) + + def get_activity_summary(self): + return {"api_call_count": 1} + + class FakeCLI(real_cli): + def __init__(self, **kwargs): + self.result_meta_fd = kwargs.get("result_meta_fd") + self.max_turns = kwargs.get("max_turns") or 90 + self.agent = FakeAgent() + self.session_id = "session" + self.conversation_history = [] + self._active_agent_route_signature = "same" + + def _claim_active_session(self, *_args, **_kwargs): + return True + + def _release_active_session(self): + pass + + def _ensure_runtime_credentials(self): + return True + + def _resolve_turn_agent_config(self, _query): + return { + "signature": "same", + "model": None, + "runtime": None, + "request_overrides": None, + } + + def _init_agent(self, **_kwargs): + return True + + monkeypatch.setattr(cli_mod, "HermesCLI", FakeCLI) + monkeypatch.setattr(cli_mod, "_finalize_single_query", lambda _cli: None) + monkeypatch.setattr(signal, "signal", lambda *_args: None) + + with pytest.raises(SystemExit) as baseline_exit: + cli_mod.main(query="hello", quiet=True, toolsets="safe") + baseline = capsys.readouterr() + + read_fd, write_fd = os.pipe() + with pytest.raises(SystemExit) as metadata_exit: + cli_mod.main( + query="hello", + quiet=True, + toolsets="safe", + result_meta_fd=write_fd, + ) + with_metadata = capsys.readouterr() + payload = os.read(read_fd, result_metadata.MAX_METADATA_BYTES) + os.close(read_fd) + + assert baseline_exit.value.code == (1 if failed else 0) + assert metadata_exit.value.code == 0 + expected_stdout = "" if failed else "exact response\n" + assert baseline.out == with_metadata.out == expected_stdout + expected_stderr = ( + "Error: secret failure detail\n\nsession_id: session\n" + if failed + else "\nsession_id: session\n" + ) + assert baseline.err == with_metadata.err == expected_stderr + assert result_metadata.serialize_result_metadata( + result_metadata.build_result_metadata(result, max_iterations=90) + ) == payload + + +@pytest.mark.parametrize( + ("raised_error", "legacy_error", "failure_class", "expected_statuses"), + [ + ( + KeyboardInterrupt(), + SystemExit, + "interrupted", + (False, False, False, True), + ), + ( + RuntimeError("secret exception detail /private/path"), + RuntimeError, + "unknown_failure", + (False, True, False, False), + ), + ], +) +def test_quiet_query_abnormal_exit_publishes_closed_failure_frame( + monkeypatch, + capsys, + raised_error, + legacy_error, + failure_class, + expected_statuses, +): + import json + import signal + + import cli as cli_mod + from hermes_cli import result_metadata + + real_cli = cli_mod.HermesCLI + + class FakeAgent: + session_id = "session" + quiet_mode = False + suppress_status_output = False + stream_delta_callback = object() + tool_gen_callback = object() + + def run_conversation(self, *_args, **_kwargs): + raise raised_error + + def get_activity_summary(self): + return {"api_call_count": 3, "private": "/private/path"} + + class FakeCLI(real_cli): + def __init__(self, **kwargs): + self.result_meta_fd = kwargs.get("result_meta_fd") + self.max_turns = kwargs.get("max_turns") or 90 + self.agent = FakeAgent() + self.session_id = "session" + self.conversation_history = [] + self._active_agent_route_signature = "same" + + def _claim_active_session(self, *_args, **_kwargs): + return True + + def _release_active_session(self): + pass + + def _ensure_runtime_credentials(self): + return True + + def _resolve_turn_agent_config(self, _query): + return { + "signature": "same", + "model": None, + "runtime": None, + "request_overrides": None, + } + + def _init_agent(self, **_kwargs): + return True + + monkeypatch.setattr(cli_mod, "HermesCLI", FakeCLI) + monkeypatch.setattr(cli_mod, "_finalize_single_query", lambda _cli: None) + monkeypatch.setattr(signal, "signal", lambda *_args: None) + + with pytest.raises(legacy_error) as baseline_error: + cli_mod.main(query="hello", quiet=True, toolsets="safe") + baseline = capsys.readouterr() + if isinstance(raised_error, KeyboardInterrupt): + assert baseline_error.value.code == 130 + + read_fd, write_fd = os.pipe() + with pytest.raises(SystemExit) as metadata_exit: + cli_mod.main( + query="hello", + quiet=True, + toolsets="safe", + result_meta_fd=write_fd, + ) + captured = capsys.readouterr() + payload = os.read(read_fd, result_metadata.MAX_METADATA_BYTES) + os.close(read_fd) + + decoded = json.loads(payload) + assert metadata_exit.value.code == 0 + assert decoded["failure_class"] == failure_class + assert tuple( + decoded[key] for key in ("completed", "failed", "partial", "interrupted") + ) == expected_statuses + assert decoded["api_calls"] == 3 + assert len(payload) <= result_metadata.MAX_METADATA_BYTES + assert set(decoded) == { + "schema_version", + "completed", + "failed", + "partial", + "interrupted", + "api_calls", + "failure_class", + } + for secret in ("secret exception detail", "/private/path"): + assert secret not in payload.decode("utf-8") + assert secret not in captured.out + assert secret not in captured.err diff --git a/website/docs/reference/cli-commands.md b/website/docs/reference/cli-commands.md index 804a25aab411..659335b5b6dd 100644 --- a/website/docs/reference/cli-commands.md +++ b/website/docs/reference/cli-commands.md @@ -109,6 +109,7 @@ Common options: | `-s`, `--skills ` | Preload one or more skills for the session (can be repeated or comma-separated). | | `-v`, `--verbose` | Verbose output. | | `-Q`, `--quiet` | Programmatic mode: suppress banner/spinner/tool previews. | +| `--result-meta-fd ` | Write closed-world JSON status as one bounded frame to a pre-opened POSIX/WSL anonymous-pipe write descriptor; single `--query`, classic CLI only. | | `--image ` | Attach a local image to a single query. | | `--resume ` / `--continue [name]` | Resume a session directly from `chat`. | | `--worktree` | Create an isolated git worktree for this run. | @@ -134,6 +135,43 @@ hermes chat --ignore-user-config --ignore-rules -q "Repro without my personal se hermes chat --safe-mode -q "Is this bug mine or Hermes'?" ``` +### Structured query-result metadata + +Automation that needs a machine-readable turn outcome without inspecting the +model response can create an anonymous pipe, retain its read endpoint, pass only +the blocking write endpoint to Hermes, and select it with +`--result-meta-fd `. The option requires a single `--query`, is available +only in the classic CLI on POSIX systems (including WSL), and has no effect when +absent. + +Hermes accepts only a canonical integer descriptor numbered 3 or higher that is +an open blocking anonymous-pipe write endpoint with `PC_PIPE_BUF >= 1024`. It marks the +descriptor non-inheritable, writes exactly one JSON frame of at most 1024 bytes, +requires a full write, and closes the owned endpoint on success or failure. +Invalid descriptors fail before config, model, or agent construction. `EPIPE`, +`EAGAIN`, short writes, and close faults fail closed without retry or fallback, +emit only `Error: failed to publish result metadata.`, and exit nonzero. +Once a valid frame is published, the process exits zero even when the frame +reports a failed or interrupted turn; callers must use `failure_class` and the +status fields as the turn outcome. Without `--result-meta-fd`, legacy process +statuses remain unchanged (including exit 1 for failed turns and 130 for +`KeyboardInterrupt`). + +The versioned `hermes-agent-result-meta-v1` object contains only: +`schema_version`, `completed`, `failed`, `partial`, `interrupted`, `api_calls`, +and `failure_class`. Failure classes are `none`, `interrupted`, +`content_policy_blocked`, `provider_api_terminal`, +`max_turns_or_incomplete`, and `unknown_failure`. The projection never includes +response/error text, prompts, messages, tool output, provider/model names, +session IDs, paths, hashes, or exceptions. Unknown, malformed, or contradictory +internal results become `unknown_failure`. +`KeyboardInterrupt` and unexpected query exceptions publish `interrupted` and +`unknown_failure` frames respectively. Their API-call count comes only from the +agent's bounded activity counter; invalid or unavailable counters become zero. + +The security boundary is producer-bound against ordinary filesystem +substitution, not root or arbitrary same-principal ptrace/proc-fd compromise. + ### `hermes -z ` — scripted one-shot For programmatic callers (shell scripts, CI, cron, parent processes piping in a prompt), `hermes -z` is the purest one-shot entry point: **single prompt in, final response text out, nothing else on stdout or stderr.** No banner, no spinner, no tool previews, no `Session:` line — just the agent's final reply as plain text.