"""``hermes peer`` — bot-to-bot DMs across machines/gateways. A *peer* is another Hermes gateway running the ``api_server`` platform; its stock API is the transport (no new server surface). ``dm`` resolves the remote canonical "Bot Chat" session (creating it when missing) and runs ONE synchronous turn — the cross-machine twin of ``hermes -p chat --in ~ -c "Bot Chat"``. ``run``/``status`` /``stop`` do the same turn through the async Runs API. Peer labels/URLs live in config.yaml (``bot_peers``); the key lives in ``~/.hermes/.env`` as ``HERMES_PEER__KEY``. ``/`` targets the ``/p//`` mirror. """ from __future__ import annotations import argparse import json import re import sys import urllib.error import urllib.parse import urllib.request import uuid BOT_CHAT_TITLE = "Bot Chat" _PEER_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") _PROFILE_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$") # One synchronous agent turn can legitimately take minutes. DM_TIMEOUT_S = 600 LIST_TIMEOUT_S = 30 def _peer_key_env(name: str) -> str: return f"HERMES_PEER_{name.upper().replace('-', '_')}_KEY" def _load_peers() -> dict: from hermes_cli.config import load_config cfg = load_config() or {} peers = cfg.get("bot_peers") return peers if isinstance(peers, dict) else {} def _save_peers(peers: dict) -> None: from hermes_cli.config import load_config, save_config cfg = load_config() or {} cfg["bot_peers"] = peers save_config(cfg) def _peer_secret(name: str) -> str: """The peer's API key: profile-scoped secret store first, raw env fallback.""" env_name = _peer_key_env(name) try: from agent.secret_scope import get_secret return (get_secret(env_name, "") or "").strip() except Exception: import os return (os.environ.get(env_name) or "").strip() def _request( url: str, key: str, *, method: str = "GET", body: dict | None = None, timeout: int = LIST_TIMEOUT_S, headers: dict[str, str] | None = None) -> dict: from hermes_cli.urllib_security import open_credentialed_url data = json.dumps(body).encode("utf-8") if body is not None else None request_headers = { "Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": "hermes-peer-dm"} if headers: request_headers.update(headers) req = urllib.request.Request(url, data=data, method=method, headers=request_headers) # The peer URL is user-registered (``hermes peer add``); a redirect to a # different origin must not carry the Authorization: Bearer key with it — # a compromised/MITM'd peer could otherwise harvest it. open_credentialed_url # strips non-safelisted headers across a cross-origin redirect. with open_credentialed_url(req, timeout=timeout) as resp: payload = resp.read().decode("utf-8", "replace") try: parsed = json.loads(payload) except ValueError as exc: raise RuntimeError(f"Peer returned non-JSON response: {payload[:200]}") from exc if not isinstance(parsed, dict): raise RuntimeError("Peer returned a non-object JSON response") return parsed def _base_url(peer: dict, profile: str | None) -> str: url = str(peer.get("url") or "").rstrip("/") if profile: # Multiplex mirror: same handlers, scoped to the named profile. return f"{url}/p/{urllib.parse.quote(profile, safe='')}" return url def _find_bot_chat(base: str, key: str) -> str | None: """The remote canonical Bot Chat's session id, or None. Bot Mode always HIDES canonical chats, so the plain listing (which excludes hidden sessions) misses an existing Bot Chat and the caller would try to create a duplicate that the peer's UNIQUE(title) guard rejects (issue #91583). Newer peers support an exact-title lookup with ``include_hidden=1``; older peers ignore the unknown query params and return the ordinary visible listing, so this single request degrades to exactly the previous behavior against them. """ query = urllib.parse.urlencode({"limit": 200, "title": BOT_CHAT_TITLE, "include_hidden": 1}) listing = _request(f"{base}/api/sessions?{query}", key) for session in listing.get("data") or []: if isinstance(session, dict) and (session.get("title") or "").strip() == BOT_CHAT_TITLE: return str(session.get("id") or "") or None return None def _ensure_bot_chat(base: str, key: str) -> str: existing = _find_bot_chat(base, key) if existing: return existing try: created = _request( f"{base}/api/sessions", key, method="POST", body={"title": BOT_CHAT_TITLE, "source": "bot_peer_dm"}) except urllib.error.HTTPError as exc: detail = _http_error_detail(exc) if exc.code == 400 and "title" in detail.lower(): # Older peer (no title/include_hidden lookup support): its # canonical Bot Chat exists but is hidden, so we couldn't see it # and the create collided with the UNIQUE(title) guard. raise RuntimeError( f"Peer already has a '{BOT_CHAT_TITLE}' session but it is hidden and the " f"peer's gateway is too old to expose hidden sessions to this lookup " f"(HTTP 400: {detail}). Update the peer's hermes-agent, or unhide the " f"session there: PATCH /api/sessions/ {{\"hidden\": false}}.") from exc raise # Real api_server wraps the row: {"object": "hermes.session", "session": {...}}. session = created.get("session") if isinstance(created.get("session"), dict) else created session_id = str(session.get("id") or session.get("session_id") or "") if not session_id: raise RuntimeError("Peer did not return a session id for the new Bot Chat") return session_id def _parse_target(target: str) -> tuple[str, str | None]: """```` or ``/`` → (peer, profile|None).""" raw = (target or "").strip() peer, _, profile = raw.partition("/") peer = peer.strip() profile = profile.strip() or None if not peer: raise ValueError("Peer name required (hermes peer dm [/] ...)") if profile and not _PROFILE_RE.match(profile): raise ValueError(f"Invalid agent/profile name: {profile!r}") return peer, profile def _http_error_detail(exc: urllib.error.HTTPError) -> str: try: body = exc.read().decode("utf-8", "replace") parsed = json.loads(body) message = parsed.get("error", {}).get("message") if isinstance(parsed, dict) else None return message or body[:200] except Exception: return str(exc) def _resolve_peer_target(target: str) -> tuple[str, str | None, dict, str]: """Resolve a registered target to ``(name, profile, config, key)``.""" peer_name, profile = _parse_target(target) peer = _load_peers().get(peer_name) if not isinstance(peer, dict) or not peer.get("url"): raise LookupError(f"No peer named '{peer_name}'. Run: hermes peer list") key = _peer_secret(peer_name) if not key: raise PermissionError( f"No API key for peer '{peer_name}'. Set it: hermes peer add {peer_name} " f"--url --key (or add {_peer_key_env(peer_name)}= to ~/.hermes/.env)") return peer_name, profile, peer, key def _message_from_args(args) -> str: message = (getattr(args, "message", None) or "").strip() if not message and not sys.stdin.isatty(): message = sys.stdin.read().strip() return message def _peer_run_durability(base: str, key: str) -> bool | None: """Return durable support, or None when an older peer cannot advertise it.""" try: capabilities = _request(f"{base}/v1/capabilities", key) except Exception: return None features = capabilities.get("features") if not isinstance(features, dict): return None contract = features.get("runs_idempotency") if not isinstance(contract, dict) or not contract.get("supported"): return None return bool(contract.get("durable")) def _peer_failure(peer_name: str, exc: Exception) -> int: """Print a peer HTTP rejection or transport failure to stderr; always exit 1.""" if isinstance(exc, urllib.error.HTTPError): detail = _http_error_detail(exc) print(f"Peer '{peer_name}' rejected the request (HTTP {exc.code}): {detail}", file=sys.stderr) else: print(f"Could not reach peer '{peer_name}': {exc}", file=sys.stderr) return 1 def _emit(args, payload: dict, text_lines: list[str]) -> int: if getattr(args, "json", False): print(json.dumps(payload)) else: for line in text_lines: print(line) return 0 def _peer_add(args) -> int: name = (args.name or "").strip().lower() if not _PEER_NAME_RE.match(name): print(f"Invalid peer name: {name!r} (lowercase, digits, -, _; max 64)", file=sys.stderr) return 2 url = (args.url or "").strip() if not url.lower().startswith(("http://", "https://")): print("Peer --url must be an http(s) gateway base URL, e.g. http://spark.lan:8377", file=sys.stderr) return 2 peers = _load_peers() peers[name] = {"url": url.rstrip("/"), **({"note": args.note.strip()} if getattr(args, "note", "") else {})} _save_peers(peers) key = (getattr(args, "key", "") or "").strip() if key: from hermes_cli.config import save_env_value save_env_value(_peer_key_env(name), key) print(f"Peer '{name}' saved ({url}) — key stored as {_peer_key_env(name)} in ~/.hermes/.env") else: print( f"Peer '{name}' saved ({url}). No key given — set the peer's API_SERVER_KEY with:\n" f" hermes peer add {name} --url {url} --key \n" f" (or add {_peer_key_env(name)}= to ~/.hermes/.env)") return 0 def _peer_remove(args) -> int: name = (args.name or "").strip().lower() peers = _load_peers() if name not in peers: print(f"No peer named '{name}'.", file=sys.stderr) return 1 peers.pop(name) _save_peers(peers) print(f"Peer '{name}' removed (its {_peer_key_env(name)} entry in .env is kept; delete it manually if unused).") return 0 def _peer_list(args) -> int: peers = _load_peers() if not peers: print("No peers registered. Add one: hermes peer add --url http://host:port --key ") return 0 for name in sorted(peers): entry = peers[name] if isinstance(peers[name], dict) else {} has_key = "key set" if _peer_secret(name) else f"NO KEY ({_peer_key_env(name)} unset)" note = f" — {entry.get('note')}" if entry.get("note") else "" print(f"{name}\t{entry.get('url', '?')}\t[{has_key}]{note}") return 0 def _peer_run_ctl(args, action: str, peer_name: str, profile: str | None, base: str, key: str) -> int: """``status`` / ``stop`` on an asynchronous run.""" run_id = (getattr(args, "run_id", None) or "").strip() if not run_id: print("Run ID required.", file=sys.stderr) return 2 stop = action == "stop" try: result = _request( f"{base}/v1/runs/{urllib.parse.quote(run_id, safe='')}" + ("/stop" if stop else ""), key, method="POST" if stop else "GET", body={} if stop else None) except (urllib.error.URLError, TimeoutError, OSError, RuntimeError) as exc: return _peer_failure(peer_name, exc) if getattr(args, "json", False): print(json.dumps({"peer": peer_name, "profile": profile, **result})) return 0 print(f"{run_id}: {result.get('status', 'unknown')}") if not stop and result.get("output"): print(result["output"]) elif not stop and result.get("error"): print(result["error"], file=sys.stderr) return 0 def _turn_body(message: str, *, message_key: str, **extra) -> dict: """Request body for one turn. ``author`` is added only when a dispatcher set HERMES_TURN_AUTHOR.""" from agent.turn_author import turn_author_from_env body = {message_key: message, **extra} author = turn_author_from_env() if author is not None: body["author"] = author return body def _peer_run(args, message: str, peer_name: str, profile: str | None, base: str, key: str) -> int: idempotency_key = (getattr(args, "idempotency_key", None) or f"peer-{uuid.uuid4().hex}").strip() if (not idempotency_key or len(idempotency_key) > 255 or re.search(r"[\r\n\x00]", idempotency_key)): print("Idempotency key must be 1-255 characters without control newlines.", file=sys.stderr) return 2 try: if _peer_run_durability(base, key) is not True: print( "Warning: this peer does not advertise restart-durable " "run replay; keep the run ID and avoid blind retries " "after a gateway restart.", file=sys.stderr) session_id = _ensure_bot_chat(base, key) result = _request( f"{base}/v1/runs", key, method="POST", body=_turn_body(message, message_key="input", session_id=session_id), headers={"Idempotency-Key": idempotency_key}) except (urllib.error.URLError, TimeoutError, OSError, RuntimeError) as exc: return _peer_failure(peer_name, exc) run_id = str(result.get("run_id") or "") if not run_id: print(f"Peer '{peer_name}' did not return a run ID.", file=sys.stderr) return 1 payload = { "peer": peer_name, "profile": profile, "session_id": session_id, "run_id": run_id, "status": result.get("status") or "started", "idempotency_key": idempotency_key, "replayed": bool(result.get("replayed", False))} replay = " (replayed)" if payload["replayed"] else "" return _emit(args, payload, [ f"{run_id}: {payload['status']}{replay}", f"session_id: {session_id}", f"idempotency_key: {idempotency_key}"]) def _peer_dm(args, message: str, peer_name: str, profile: str | None, base: str, key: str) -> int: session_id = "" try: session_id = _ensure_bot_chat(base, key) result = _request( f"{base}/api/sessions/{urllib.parse.quote(session_id, safe='')}/chat", key, method="POST", body=_turn_body(message, message_key="message"), timeout=DM_TIMEOUT_S) except RuntimeError as exc: print(f"Peer '{peer_name}': {exc}", file=sys.stderr) return 1 except (urllib.error.URLError, TimeoutError, OSError) as exc: # A timeout while awaiting the response, once the Bot Chat is known, means the peer took # this turn: the message is already in its Bot Chat and the gateway runs the turn to # completion regardless of this client, so reporting it unreachable makes the sender resend # and deliver it twice. urllib raises that timeout bare; one it wraps in URLError hit while # connecting or sending, so the request never arrived and "could not reach" is the truth. if session_id and isinstance(exc, TimeoutError): print(f"Peer '{peer_name}' accepted the message but its turn is still running after " f"{DM_TIMEOUT_S}s: the message is already in its Bot Chat (session {session_id}) " "and will be answered there. The reply cannot come back on this call. Do NOT resend.", file=sys.stderr) return 1 return _peer_failure(peer_name, exc) if result.get("object") == "hermes.session.chat.queued": # The peer's Bot Chat is open in its Desktop and that turn outlasted the peer's wait: the # message is in the open chat and is answered there, so a resend would run it twice. queued_in = result.get("session_id") or session_id return _emit(args, {"peer": peer_name, "profile": profile, "session_id": queued_in, "status": result.get("status") or "queued", "delivery_id": result.get("delivery_id")}, [f"Peer '{peer_name}' has its Bot Chat open, so the message went into that chat (session " f"{queued_in}) and is answered there. The reply cannot come back on this call. Do NOT resend."]) msg = result.get("message") reply = str(msg.get("content") or "") if isinstance(msg, dict) else "" # A successful bare silence marker is a delivery decision, not a message: # the turn stays in the peer's own transcript, the sending agent never # sees NO_REPLY/[SILENT] as a real reply. Same rule as the gateway's live # Bot Chat completion, the Desktop bot_relay.deliver RPC and the one-shot # local `hermes chat -Q` transport (tools/bot_mode_dm.py) — this is the # 4th Bot Mode delivery door and was missing the same check. from gateway.response_filters import is_intentional_silence_response if is_intentional_silence_response(reply): reply = "" payload = {"peer": peer_name, "profile": profile, "session_id": result.get("session_id") or session_id, "reply": reply} return _emit(args, payload, [reply or "(no reply)"]) _REGISTRY_ACTIONS = { "add": _peer_add, "set": _peer_add, "remove": _peer_remove, "rm": _peer_remove, "list": _peer_list, "ls": _peer_list, None: _peer_list} def cmd_peer(args) -> int: action = getattr(args, "peer_action", None) if action in _REGISTRY_ACTIONS: return _REGISTRY_ACTIONS[action](args) if action not in {"dm", "run", "status", "stop"}: print("Unknown peer action. See: hermes peer --help", file=sys.stderr) return 2 try: peer_name, profile, peer, key = _resolve_peer_target(args.target) except ValueError as exc: print(str(exc), file=sys.stderr) return 2 except (LookupError, PermissionError) as exc: print(str(exc), file=sys.stderr) return 1 base = _base_url(peer, profile) if action in {"status", "stop"}: return _peer_run_ctl(args, action, peer_name, profile, base, key) message = _message_from_args(args) if not message: print("Message required (argument or stdin).", file=sys.stderr) return 2 handler = _peer_run if action == "run" else _peer_dm return handler(args, message, peer_name, profile, base, key) def build_peer_parser(subparsers) -> None: """Attach the ``peer`` subcommand to ``subparsers``.""" parser = subparsers.add_parser( "peer", help="Bot-to-bot DMs across machines (peer Hermes gateways)", description="Register other Hermes gateways as peers and message their agents. " "'hermes peer dm [/] \"...\"' delivers into the remote " "agent's canonical Bot Chat over the peer's API server and prints " "the reply — the cross-machine twin of 'hermes -p chat'. " "The peer must run the api_server platform; its API_SERVER_KEY is " "stored locally as a credential in ~/.hermes/.env.", epilog=( "Examples:\n" " hermes peer add spark --url http://spark.lan:8377 --key \n" " hermes peer list\n" ' hermes peer dm spark "Message from 🤖 dixie (@dixie): disk status?"\n' ' hermes peer dm spark/researcher "..." # named profile on a multiplexed peer\n' " hermes peer run spark --idempotency-key ticket-123 < long-task.txt\n" " hermes peer status spark run_abc123\n" " hermes peer stop spark run_abc123\n" " hermes peer remove spark\n" "\n" "Exit codes: 0 ok, 1 delivery/peer error, 2 usage error."), formatter_class=argparse.RawDescriptionHelpFormatter) peer_sub = parser.add_subparsers(dest="peer_action") add_p = peer_sub.add_parser("add", aliases=["set"], help="Register (or update) a peer gateway") add_p.add_argument("name", help="Peer name (lowercase slug, e.g. spark, homelab)") add_p.add_argument("--url", required=True, help="Peer gateway base URL, e.g. http://spark.lan:8377") add_p.add_argument("--key", default="", help="The peer's API_SERVER_KEY (stored in ~/.hermes/.env)") add_p.add_argument("--note", default="", help="Optional description") peer_sub.add_parser("list", aliases=["ls"], help="List registered peers") rm_p = peer_sub.add_parser("remove", aliases=["rm"], help="Remove a peer") rm_p.add_argument("name", help="Peer name") def _remote(name: str, help: str, *, run_id: bool): sp = peer_sub.add_parser(name, help=help) sp.add_argument("target", help=" or / (named profile on a multiplexed peer)") if run_id: sp.add_argument("run_id", help="Run ID returned by 'hermes peer run'") else: sp.add_argument("message", nargs="?", default=None, help="Message text (or stdin)") if name == "run": sp.add_argument( "--idempotency-key", default=None, help="Stable retry key (generated when omitted)") sp.add_argument("--json", action="store_true", default=False, help="Emit a JSON result") _remote("dm", "Message an agent on a peer gateway and print its reply", run_id=False) _remote("run", "Start a long peer turn asynchronously and return its run ID", run_id=False) _remote("status", "Read the status and final output of an asynchronous peer run", run_id=True) _remote("stop", "Stop one asynchronous peer run without affecting another turn", run_id=True) parser.set_defaults(func=cmd_peer)