diff --git a/tools/bot_mode_dm.py b/tools/bot_mode_dm.py index 2c44316c38..4b0f91c5ed 100644 --- a/tools/bot_mode_dm.py +++ b/tools/bot_mode_dm.py @@ -2,25 +2,28 @@ Lets a Bot Mode agent message a teammate (another profile on this install, an agent on a registered peer gateway, or one on another Desktop-connected machine) -as a real tool call instead of a hand-assembled ``hermes -p …`` shellout: the -message is a parameter, the target is validated against the live roster, the -attribution prefix is applied server-side, and the reply arrives later through -the background-process completion notification (fire-and-forget). +as a real tool call instead of a hand-assembled ``hermes -p …`` shellout, whose +invocation was fragile (quoting traps, temp-file choreography, dead-profile +races, raw user text forwarded verbatim): the message is a parameter, the target +is validated against the live roster, the attribution prefix is applied +server-side, and the reply arrives later through the background-process +completion notification (fire-and-forget — never blocks the sender's turn). -Containment contract (all three MUST hold): +Containment contract (all three MUST hold — reviewers check each): - The schema is injected ONLY into a bot's canonical "Bot Chat" session on a Bot-Mode-managed install — the same gate as ``tools/bot_mode_probe.py``. It is - not in the registry or any toolset; CLI, ordinary gateway, group-room, cron and - subagent sessions never see it. + not in the registry or any toolset; CLI, ordinary gateway, group-room + (titled "Group: …"), cron and subagent sessions never see it. - Dispatch re-checks the title gate (defense in depth): a forged call returns a structured error instead of delivering. - Everything is additive; the legacy ``hermes -p`` / ``hermes peer dm`` transports keep working for older prompts. -Transports: local → ``hermes -p chat --in ~ -c "Bot Chat" ---create-if-missing -Q --query-file ``; peer → ``hermes peer dm -[/] < ``. Both run via ``terminal_tool(background=True, -notify_on_complete=True)`` so the reply wakes the sender on its next turn. +Transports (unchanged and proven): local → ``hermes -p chat --in ~ -c +"Bot Chat" --create-if-missing -Q --query-file `` (one turn, reply on +stdout); peer → ``hermes peer dm [/] < ``. Both run via +``terminal_tool(background=True, notify_on_complete=True)`` so the reply wakes +the sender on its NEXT turn — the wake shape every Bot Mode agent already knows. """ from __future__ import annotations @@ -39,16 +42,10 @@ import time from pathlib import Path from typing import Any, Optional -from tools.bot_mode_probe import ( - BOT_CHAT_TITLE, - _handle, - _hermes_root, - _peers, - _profile_name as _self_profile_name, - _roster, - is_bot_mode_managed, -) -from tools.bot_relay import _HANDLE_RE as _LOCAL_TARGET_RE +# Top-level imports stay stdlib-only: this module is also executed directly as the +# background delivery runner (``python bot_mode_dm.py --run-delivery …``), where +# ``tools.*`` is resolved from whichever install is on sys.path. Hermes-side +# helpers are imported lazily inside the functions that need them. logger = logging.getLogger(__name__) @@ -63,13 +60,16 @@ MESSAGE_MAX_CHARS = 16000 _DM_DIR_NAME = "hermes-dm" _DM_STALE_SECONDS = 24 * 60 * 60 +# '/' — peer names are lowercase (``hermes peer`` normalizes them). +_PEER_TARGET_RE = re.compile(r"^([a-z0-9][a-z0-9_-]{0,63})/([a-zA-Z0-9][a-zA-Z0-9_-]{0,63})$") +# Same shape as ``tools.bot_relay._HANDLE_RE`` (kept local: see import note above). +_LOCAL_TARGET_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$") + + def _default_home() -> str: return os.getenv("HERMES_HOME") or os.path.expanduser("~/.hermes") -_PEER_TARGET_RE = re.compile(r"^([a-z0-9][a-z0-9_-]{0,63})/([a-zA-Z0-9][a-zA-Z0-9_-]{0,63})$") - - def message_agent_tool_schema() -> dict: """OpenAI-format schema for ``message_agent`` (injected, not registered).""" return { @@ -128,20 +128,22 @@ def ensure_message_agent_tool(agent: Any) -> bool: """Inject the ``message_agent`` schema into a Bot Chat agent's tool list. Called once per turn. Idempotent and deterministic for the life of a session - (the gate is stable from the first turn), so the tool list is byte-identical - across turns — prompt-cache safe. Never raises. + (the gate — canonical Bot Chat title on a Bot-Mode-managed install — is + stable from the first turn), so the tool list is byte-identical across turns + — prompt-cache safe. Every non-Bot-Chat session fails the gate on every turn + and never sees the schema. Never raises. """ try: if not getattr(agent, "_bot_mode_protocol", True): return False tools = getattr(agent, "tools", None) - if tools: - for tool in tools: - if ( - isinstance(tool, dict) - and tool.get("function", {}).get("name") == MESSAGE_AGENT_TOOL_NAME - ): - return True + if tools and any( + isinstance(t, dict) and t.get("function", {}).get("name") == MESSAGE_AGENT_TOOL_NAME + for t in tools + ): + return True + from tools.bot_mode_probe import BOT_CHAT_TITLE, is_bot_mode_managed + if _session_title(agent) != BOT_CHAT_TITLE: return False # Managed-install check, NOT section non-emptiness: a SOUL.md carrying the @@ -165,6 +167,8 @@ def ensure_message_agent_tool(agent: Any) -> bool: def _local_roster(root: Path) -> list[str]: """Profile names on this install: default + every named profile.""" + from tools.bot_mode_probe import _roster + return [name for name, _dir in _roster(root)] @@ -209,6 +213,16 @@ def message_agent_tool( # Defense-in-depth gate: only a canonical Bot Chat may deliver. home = _agent_home(agent) try: + from tools.bot_mode_probe import ( + BOT_CHAT_TITLE, + _handle, + _hermes_root, + _peers, + _profile_name as _self_profile_name, + is_bot_mode_managed, + ) + from tools.bot_relay import BOT_CHAT_TURN_ARGS + if _session_title(agent) != BOT_CHAT_TITLE: return _err( "message_agent is only available in a Bot Mode 'Bot Chat' session. " @@ -275,10 +289,11 @@ def message_agent_tool( resolved = _resolve_local_name(raw_target, roster) if _LOCAL_TARGET_RE.match(raw_target) else None if resolved is None or resolved == me: # Unknown locally, or same-name target on ANOTHER connection (this - # gateway's 'default' messaging the cloud 'default'): try the Desktop - # relay before reporting a resolution failure / self-message. + # gateway's 'default' messaging the cloud 'default'): every gateway + # connected to the user's Desktop is reachable via the relay roster, so + # try that before reporting a resolution failure / self-message. relayed = _try_relay_delivery( - root, raw_target, body, me, sender_handle, task_id=task_id, agent=agent + root, raw_target, prefix + body, me, task_id=task_id, agent=agent ) if relayed is not None: return relayed @@ -293,7 +308,7 @@ def message_agent_tool( ) return _start_delivery( - ["hermes", "-p", resolved, "chat", "--in", "~", "-c", "Bot Chat", "--create-if-missing", "-Q"], + ["hermes", "-p", resolved, *BOT_CHAT_TURN_ARGS], prefix + body, f"@{_handle(resolved)}", stdin_file=False, @@ -305,21 +320,23 @@ def message_agent_tool( def _try_relay_delivery( root: Path, raw_target: str, - body: str, + content: str, me: str, - sender_handle: str, *, task_id: Optional[str], agent: Any, ) -> Optional[str]: """Cross-connection delivery via the Desktop relay; None when the target - doesn't resolve against the relay roster. + doesn't resolve against the relay roster. ``content`` is the attributed + message (prefix already applied). - The envelope is queued on disk for the Desktop to drain; a background waiter - is spawned immediately so the relayed reply wakes the sender through the - standard completion-notification path — identical UX to a local DM. + The envelope is queued on disk; the Desktop drains it over RPC and delivers + on the target connection's own socket. A background waiter is spawned + immediately so the relayed reply wakes the sender through the standard + completion-notification path — identical UX to a local DM. """ try: + from tools.bot_mode_probe import _handle from tools.bot_relay import ( EnvelopeRefusedError, enqueue_envelope, @@ -348,9 +365,9 @@ def _try_relay_delivery( envelope = enqueue_envelope( root, target=match, - message=f"Message from 🤖 {sender_handle} (@{sender_handle}): {body}", + message=content, sender_profile=me, - sender_handle=sender_handle, + sender_handle=_handle(me), ) except EnvelopeRefusedError as exc: # Fail fast: target definitively offline — nothing was queued. @@ -394,7 +411,6 @@ def cleanup_bot_dm_cache( DM (the in-band sweep in ``_write_dm_file`` only runs when a DM is written). """ cutoff = (time.time() if now is None else now) - max_age_hours * 3600 - removed = 0 # Legacy temp-root locations too, so upgrades clean files from versions # predating the dedicated directory. temp_root = Path(tempfile.gettempdir()) @@ -406,18 +422,9 @@ def cleanup_bot_dm_cache( locations.append((_dm_dir(), "*.txt")) except OSError: pass - for directory, pattern in locations: - try: - for candidate in directory.glob(pattern): - try: - if candidate.is_file() and candidate.stat().st_mtime < cutoff: - candidate.unlink() - removed += 1 - except OSError: - pass - except OSError: - pass - return removed + from tools.bot_relay import unlink_files_older_than + + return sum(unlink_files_older_than(d, pattern, cutoff) for d, pattern in locations) def _write_dm_file(content: str) -> str: @@ -466,11 +473,54 @@ def _delivery_lock(argv: list[str], *, stdin_file: bool): or argv[1] != "-p" ): return contextlib.nullcontext() + from tools.bot_mode_probe import _hermes_root from tools.bot_relay import acquire_turn_lock return acquire_turn_lock(_hermes_root(Path(_default_home())), argv[2]) +def _run_local_turn(argv: list[str], dm_file: str) -> int: + """One Bot Chat turn via ``--query-file`` (plus the policy-gated retry); + re-emits the transport's streams and returns its exit code.""" + + def _turn(): + return subprocess.run( + [*argv, "--query-file", dm_file], + check=False, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + ) + + proc = _turn() + if proc.returncode != 0: + from tools.bot_failure_reasons import RETRY_NONE, classify_agent_error, retry_action + + detail = (proc.stderr or proc.stdout or "").strip()[-500:] + if retry_action(classify_agent_error(detail)) != RETRY_NONE: + proc = _turn() + if proc.returncode != 0 and "already has a live owner" in (proc.stderr or ""): + # The target's Bot Chat is held live by another surface (Desktop). + # The turn never ran — tell the sender plainly instead of leaking + # a raw lease error + exit code. + who = argv[argv.index("-p") + 1] if "-p" in argv[:-1] else "the teammate" + print(json.dumps({ + "error": f"Delivery failed: @{who}'s Bot Chat is open on another " + "surface right now, so your message was NOT delivered. Try again later.", + "reason": "target_busy", + })) + return 1 + # Re-emit the transport's streams: stdout is the reply text the + # completion notification carries back to the sending agent. + if proc.stdout: + sys.stdout.write(proc.stdout) + sys.stdout.flush() + if proc.stderr: + sys.stderr.write(proc.stderr) + sys.stderr.flush() + return proc.returncode + + def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int: """Run one DM transport and remove its plaintext file after consumption. @@ -487,52 +537,12 @@ def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int: """ try: with _delivery_lock(argv, stdin_file=stdin_file): - if stdin_file: - # Keep the file open until the transport exits; cleanup occurs - # after subprocess.run returns, not merely after stdin reaches EOF. - with open(dm_file, "r", encoding="utf-8") as stream: - return subprocess.run(argv, stdin=stream, check=False).returncode - - def _turn(): - return subprocess.run( - [*argv, "--query-file", dm_file], - check=False, - stdin=subprocess.DEVNULL, - capture_output=True, - text=True, - ) - - proc = _turn() - if proc.returncode != 0: - from tools.bot_failure_reasons import ( - RETRY_NONE, - classify_agent_error, - retry_action, - ) - - detail = (proc.stderr or proc.stdout or "").strip()[-500:] - if retry_action(classify_agent_error(detail)) != RETRY_NONE: - proc = _turn() - if proc.returncode != 0 and "already has a live owner" in (proc.stderr or ""): - # The target's Bot Chat is held live by another surface (Desktop). - # The turn never ran — tell the sender plainly instead of leaking - # a raw lease error + exit code. - who = argv[argv.index("-p") + 1] if "-p" in argv[:-1] else "the teammate" - print(json.dumps({ - "error": f"Delivery failed: @{who}'s Bot Chat is open on another " - "surface right now, so your message was NOT delivered. Try again later.", - "reason": "target_busy", - })) - return 1 - # Re-emit the transport's streams: stdout is the reply text the - # completion notification carries back to the sending agent. - if proc.stdout: - sys.stdout.write(proc.stdout) - sys.stdout.flush() - if proc.stderr: - sys.stderr.write(proc.stderr) - sys.stderr.flush() - return proc.returncode + if not stdin_file: + return _run_local_turn(argv, dm_file) + # Keep the file open until the transport exits; cleanup occurs + # after subprocess.run returns, not merely after stdin reaches EOF. + with open(dm_file, "r", encoding="utf-8") as stream: + return subprocess.run(argv, stdin=stream, check=False).returncode finally: _unlink_dm_file(dm_file) diff --git a/tools/bot_mode_probe.py b/tools/bot_mode_probe.py index 0d84ef454c..ca79f9966b 100644 --- a/tools/bot_mode_probe.py +++ b/tools/bot_mode_probe.py @@ -3,11 +3,17 @@ When any profile on this install carries ``ui_meta['hermes-bots']`` in its profile.yaml (Bot-Mode-managed), a bot's canonical "Bot Chat" session — and ONLY that session (the caller, agent/system_prompt.py, enforces the title gate against -``BOT_CHAT_TITLE``) — gets a "Messaging other agents" section. Silent (``""``) -when no profile is managed, when the profile's SOUL.md already carries the -heading (legacy plugin-appended text must never double up), or on any error — -a prompt build must never crash. Cached per (process, home) so compression- -triggered rebuilds produce identical bytes. Toggle: ``agent.bot_mode_protocol``. +``BOT_CHAT_TITLE``) — gets a "Messaging other agents" section so the bot can +receive teammate DMs, reply with attribution, and hand off @mentions. Regular +sessions never carry it; the desktop's composer middleware owns the @mention +send path there. + +The protocol is injected by the core at prompt-build time instead of the old +plugin-side SOUL.md append. Silent (``""``) when no profile is managed, when the +profile's SOUL.md already carries the heading (legacy plugin-appended text must +never double up), or on any error — a prompt build must never crash. Cached per +(process, home) so compression-triggered rebuilds produce identical bytes. +Toggle: ``agent.bot_mode_protocol`` in config.yaml (default True). Also hosts the path/roster helpers shared by ``bot_mode_dm`` and ``bot_relay``. """ @@ -123,18 +129,18 @@ def _soul_has_protocol(profile_dir: Path) -> bool: def _profile_role(profile_dir: Path) -> str: """Teammate role line: Bot Mode title — profile description (either may be - absent). Single-line, ≤160 chars, "" when neither exists. Never raises.""" + absent). The title is the name the user gave the bot in Bot Mode; the + description is the profile's stated purpose — either tells a teammate WHO + to message for a job. Single-line, ≤160 chars, "" when neither. Never raises.""" try: data = _read_yaml_dict(profile_dir / "profile.yaml") if not data: return "" bots = _bots_meta(data) or {} - parts = [ - p - for p in (str(bots.get("title") or "").strip(), str(data.get("description") or "").strip()) - if p - ] - return " ".join(" — ".join(parts).split())[:160] + title = str(bots.get("title") or "").strip() + description = str(data.get("description") or "").strip() + line = " — ".join(p for p in (title, description) if p) + return " ".join(line.split())[:160] except Exception: return "" @@ -284,8 +290,11 @@ _EPOCH_RE_TEXT = r"Capability epoch: ([0-9a-f]{12})" def capability_fingerprint(home: str | os.PathLike | None = None) -> str: """12-hex digest of the capability surface for ``home``'s profile. - Deliberately NOT cached — the point is detecting on-disk drift against the - epoch embedded in a stored prompt. Never raises ("unavailable" on failure). + Sources: disabled skills + enabled toolsets + MCP config (config.yaml), + SOUL.md bytes, installed skill names, the Bot-Mode roster (+ roles), peers + and the relay roster. Deliberately NOT cached — the point is detecting + on-disk drift against the epoch embedded in a stored prompt. Never raises + ("unavailable" on failure). """ import hashlib import json diff --git a/tools/bot_relay.py b/tools/bot_relay.py index e4d4272d84..1306debc02 100644 --- a/tools/bot_relay.py +++ b/tools/bot_relay.py @@ -1,9 +1,11 @@ """Bot Mode cross-connection relay — connections ARE the peer set. Gateway-side half of the relay that lets agents on ANY Desktop-connected -gateway message agents on ANY other, with ``message_agent`` as the one send -path. Plain file plumbing under ``/bot_relay/`` — no network; the Desktop -owns every socket and does all cross-connection I/O: +gateway (local, remote URL, SSH, Hermes Cloud, docker) message agents on ANY +other, with ``message_agent`` as the one send path — connections ARE the peer +set. Plain file plumbing under ``/bot_relay/`` — no network; the gateway +never holds another connection's credentials, the Desktop owns every socket and +does all cross-connection I/O: - ``roster.json`` — union roster of agents on OTHER connections, pushed by the Desktop (``bot_relay.roster.sync``); folded into the Bot Chat protocol section @@ -80,8 +82,15 @@ class EnvelopeRefusedError(RuntimeError): super().__init__(message) self.reason = reason + +# Profile names, handles and connection ids share one shape (also the local +# ``message_agent`` target grammar in ``tools/bot_mode_dm.py``). _HANDLE_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$") +# One turn in a profile's canonical Bot Chat: ``hermes -p *BOT_CHAT_TURN_ARGS``. +# ``-c "Bot Chat"`` must match ``bot_mode_probe.BOT_CHAT_TITLE``. +BOT_CHAT_TURN_ARGS = ("chat", "--in", "~", "-c", "Bot Chat", "--create-if-missing", "-Q") + def relay_root(root: Path | str) -> Path: return Path(root) / RELAY_DIR_NAME @@ -116,7 +125,9 @@ def _atomic_write_json(target: Path, payload: Any, *, prefix: str, sort_keys: bo def _normalize_roster_row(row: Any) -> Optional[dict]: """Validated, minimal roster row or None. - Rows come from the Desktop over RPC — treat as untrusted input. + Rows come from the Desktop over RPC — treat as untrusted input. A row names + an agent on another connection: profile, taggable handle, owning connection + id/label, and optional title/description for the protocol section. """ if not isinstance(row, dict): return None @@ -210,13 +221,12 @@ def resolve_remote_target(raw_target: str, roster: list[dict]) -> Any: def remote_target_forms(roster: list[dict]) -> list[str]: - """Human/agent-facing target strings, ambiguity-aware.""" - by_handle: dict[str, int] = {} - for row in roster: - by_handle[row["handle"].lower()] = by_handle.get(row["handle"].lower(), 0) + 1 + """Human/agent-facing target strings: bare handle when unique across + connections, else ``handle@connection`` (mirrors ``resolve_remote_target``).""" + handles = [row["handle"].lower() for row in roster] return [ - f"{row['handle']}@{row['connection_id']}" if by_handle[row["handle"].lower()] > 1 else row["handle"] - for row in roster + f"{row['handle']}@{row['connection_id']}" if handles.count(h) > 1 else row["handle"] + for row, h in zip(roster, handles) ] @@ -309,6 +319,34 @@ def enqueue_envelope( return envelope +def _expire_if_stale(root: Path | str, path: Path, ttl: float, now: float) -> bool: + """True when the outbox envelope at ``path`` is older than ``ttl``; writes the + 'queued_expired' error reply so the sender's waiter resolves (best effort — + an invalid id still counts as expired). Unreadable envelopes are left for + the claim attempt to deal with.""" + try: + env = json.loads(path.read_text(encoding="utf-8")) + created = float(env.get("created_at") or path.stat().st_mtime) + except (OSError, ValueError): + return False + if now - created <= ttl: + return False + handle = str(env.get("target_handle") or "?") + conn = str(env.get("target_connection") or "?") + with contextlib.suppress(OSError, ValueError): + write_reply( + root, + str(env.get("id") or ""), + error=( + f"queued message to @{handle} on {conn} expired after " + f"{ttl}s waiting for the Desktop to drain it — it was " + "NOT delivered. Resend once the Desktop reconnects." + ), + reason="queued_expired", + ) + return True + + def claim_pending_envelopes(root: Path | str) -> list[dict]: """Drain the outbox (rename → claimed/, so a second drain can't double- deliver). Sweeps stale claimed/reply artifacts opportunistically. @@ -322,37 +360,11 @@ def claim_pending_envelopes(root: Path | str) -> list[dict]: ttl = _envelope_ttl_seconds() now = time.time() out: list[dict] = [] - outbox = base / OUTBOX_DIR - for path in sorted(outbox.glob("*.json")): - if ttl > 0: - expired = False - try: - env = json.loads(path.read_text(encoding="utf-8")) - created = float(env.get("created_at") or path.stat().st_mtime) - if now - created > ttl: - expired = True - handle = str(env.get("target_handle") or "?") - conn = str(env.get("target_connection") or "?") - write_reply( - root, - str(env.get("id") or ""), - error=( - f"queued message to @{handle} on {conn} expired after " - f"{ttl}s waiting for the Desktop to drain it — it was " - "NOT delivered. Resend once the Desktop reconnects." - ), - reason="queued_expired", - ) - except (OSError, ValueError): - # Unreadable envelope / invalid id: still removed below if it - # already counted as expired, else the claim attempt handles it. - pass - if expired: - try: - path.unlink() - except OSError: - pass - continue + for path in sorted((base / OUTBOX_DIR).glob("*.json")): + if ttl > 0 and _expire_if_stale(root, path, ttl, now): + with contextlib.suppress(OSError): + path.unlink() + continue claimed = base / CLAIMED_DIR / path.name try: os.replace(path, claimed) # atomic claim @@ -368,7 +380,8 @@ def write_reply( """Persist the relayed reply (or delivery error) for the waiter. ``reason`` is an optional typed failure code (``tools.bot_failure_reasons``); - when omitted and ``error`` is non-empty it is classified from the text. + when omitted and ``error`` is non-empty it is classified from the text. The + waiter only surfaces the human ``error`` (plus the code as a tag). """ base = _ensure_dirs(root) safe = str(envelope_id or "").strip() @@ -386,21 +399,29 @@ def write_reply( return path +def unlink_files_older_than(directory: Path, pattern: str, cutoff: float) -> int: + """Remove regular files under ``directory`` matching ``pattern`` with mtime + before ``cutoff``; returns the count. Never raises (missing dir → 0).""" + removed = 0 + try: + for path in directory.glob(pattern): + try: + if path.is_file() and path.stat().st_mtime < cutoff: + path.unlink() + removed += 1 + except OSError: + continue + except OSError: + pass + return removed + + def _sweep_stale(base: Path, *, now: float | None = None) -> int: cutoff = (time.time() if now is None else now) - STALE_AFTER_SECONDS - removed = 0 - for sub in (CLAIMED_DIR, REPLIES_DIR, OUTBOX_DIR): - try: - for path in (base / sub).glob("*.json"): - try: - if path.stat().st_mtime < cutoff: - path.unlink() - removed += 1 - except OSError: - continue - except OSError: - continue - return removed + return sum( + unlink_files_older_than(base / sub, "*.json", cutoff) + for sub in (CLAIMED_DIR, REPLIES_DIR, OUTBOX_DIR) + ) def cleanup_bot_relay_artifacts(max_age_hours: float | None = None) -> int: @@ -488,8 +509,7 @@ def _hermes_cli() -> str: def local_delivery_command(profile: str, query_file: str) -> list[str]: """argv that delivers a DM into ``profile``'s Bot Chat on THIS gateway.""" - return [_hermes_cli(), "-p", profile, "chat", "--in", "~", "-c", "Bot Chat", - "--create-if-missing", "-Q", "--query-file", query_file] + return [_hermes_cli(), "-p", profile, *BOT_CHAT_TURN_ARGS, "--query-file", query_file] # ── per-profile turn lock ──────────────────────────────────────────────────── @@ -550,7 +570,8 @@ def acquire_turn_lock( (``bot_mode.turn_wait_seconds`` unless ``timeout_seconds`` is given). No ordering among waiters, but every waiter is bounded — no deadlock. Raises :class:`TurnBusyError` when the budget is exhausted. Without ``fcntl`` - (Windows) the lock is a no-op. + (Windows) the lock is a no-op — those installs never had this race path + in production. """ try: import fcntl diff --git a/tools/discord_tool.py b/tools/discord_tool.py index e20cd06d47..38ce662372 100644 --- a/tools/discord_tool.py +++ b/tools/discord_tool.py @@ -1,6 +1,8 @@ """Discord server introspection and management tool (REST API + bot token). -Only in the hermes-discord toolset. The model-visible schema is filtered by two +Talks to the Discord REST API directly with the bot token — no dependency on +the gateway adapter's client. Only in the hermes-discord toolset, so it costs +nothing on other platforms. The model-visible schema is filtered by two gates: (1) privileged intents from GET /applications/@me — actions needing an intent the bot lacks (search_members / member_info → GUILD_MEMBERS) are hidden, and fetch_messages/list_pins are annotated when MESSAGE_CONTENT is missing; @@ -135,7 +137,8 @@ _CAPABILITY_DISK_TTL_SECONDS = 24 * 3600 _capability_bg_started: set = set() _capability_bg_lock = threading.Lock() -# Permissive default: all actions exposed, call-time 403s mapped to guidance. +# Permissive default (``detected`` False = detection failed/pending): all actions +# exposed, call-time 403s mapped to guidance by ``_enrich_403``. _PERMISSIVE_CAPS = {"has_members_intent": True, "has_message_content": True, "detected": False} @@ -284,21 +287,50 @@ def _ok(message: str) -> str: def _member_summary(m: Dict[str, Any], *, full: bool) -> Dict[str, Any]: + """Member row; ``full`` adds the avatar/join fields member_info exposes + (key order is part of the result text, so the two shapes stay explicit).""" user = m.get("user", {}) - out = { + base = { "user_id": user.get("id"), "username": user.get("username"), "display_name": user.get("global_name"), "nickname": m.get("nick"), } - if full: - out["avatar"] = user.get("avatar") - out["bot"] = user.get("bot", False) - out["roles"] = m.get("roles", []) - if full: - out["joined_at"] = m.get("joined_at") - out["premium_since"] = m.get("premium_since") - return out + tail = {"bot": user.get("bot", False), "roles": m.get("roles", [])} + if not full: + return {**base, **tail} + return { + **base, + "avatar": user.get("avatar"), + **tail, + "joined_at": m.get("joined_at"), + "premium_since": m.get("premium_since"), + } + + +def _message_summary(msg: Dict[str, Any]) -> Dict[str, Any]: + author = msg.get("author", {}) + return { + "id": msg["id"], + "content": msg.get("content", ""), + "author": { + "id": author.get("id"), + "username": author.get("username"), + "display_name": author.get("global_name"), + "bot": author.get("bot", False), + }, + "timestamp": msg.get("timestamp"), + "edited_timestamp": msg.get("edited_timestamp"), + "attachments": [ + {"filename": a.get("filename"), "url": a.get("url"), "size": a.get("size")} + for a in msg.get("attachments", []) + ], + "reactions": [ + {"emoji": r.get("emoji", {}).get("name"), "count": r.get("count", 0)} + for r in msg.get("reactions", []) + ] if msg.get("reactions") else [], + "pinned": msg.get("pinned", False), + } def _int_or(value: Any, default: int) -> int: @@ -308,6 +340,11 @@ def _int_or(value: Any, default: int) -> int: return default +def _limit_param(limit: Any, default: int) -> str: + """Discord caps list endpoints at 100 per page.""" + return str(min(_int_or(limit, default), 100)) + + def _list_guilds(token: str, **_kwargs: Any) -> str: """List all guilds the bot is a member of.""" guilds = _discord_request("GET", "/users/@me/guilds", token) @@ -425,8 +462,8 @@ def _member_info(token: str, guild_id: str, user_id: str, **_kwargs: Any) -> str def _search_members(token: str, guild_id: str, query: str, limit: int = 20, **_kwargs: Any) -> str: - """Search for guild members by name.""" - params = {"query": query, "limit": str(min(_int_or(limit, 20), 100))} + """Search for guild members by name prefix (requires the GUILD_MEMBERS intent).""" + params = {"query": query, "limit": _limit_param(limit, 20)} members = _discord_request("GET", f"/guilds/{guild_id}/members/search", token, params=params) return _listing("members", [_member_summary(m, full=False) for m in members]) @@ -436,38 +473,15 @@ def _fetch_messages( before: Optional[str] = None, after: Optional[str] = None, **_kwargs: Any, ) -> str: - """Fetch recent messages from a channel.""" - params: Dict[str, str] = {"limit": str(min(_int_or(limit, 50), 100))} + """Recent messages from a channel or thread; ``before``/``after`` are + message snowflakes for reverse/forward pagination.""" + params: Dict[str, str] = {"limit": _limit_param(limit, 50)} if before: params["before"] = before if after: params["after"] = after messages = _discord_request("GET", f"/channels/{channel_id}/messages", token, params=params) - result = [] - for msg in messages: - author = msg.get("author", {}) - result.append({ - "id": msg["id"], - "content": msg.get("content", ""), - "author": { - "id": author.get("id"), - "username": author.get("username"), - "display_name": author.get("global_name"), - "bot": author.get("bot", False), - }, - "timestamp": msg.get("timestamp"), - "edited_timestamp": msg.get("edited_timestamp"), - "attachments": [ - {"filename": a.get("filename"), "url": a.get("url"), "size": a.get("size")} - for a in msg.get("attachments", []) - ], - "reactions": [ - {"emoji": r.get("emoji", {}).get("name"), "count": r.get("count", 0)} - for r in msg.get("reactions", []) - ] if msg.get("reactions") else [], - "pinned": msg.get("pinned", False), - }) - return _listing("messages", result) + return _listing("messages", [_message_summary(msg) for msg in messages]) def _list_pins(token: str, channel_id: str, **_kwargs: Any) -> str: @@ -546,11 +560,11 @@ _ACTIONS = { "remove_role": _remove_role, } +# Two tools share one action table: ``discord`` (core, the participation trio +# every bot user wants) and ``discord_admin`` (everything else). _CORE_ACTION_NAMES = frozenset({"fetch_messages", "search_members", "create_thread"}) -_ADMIN_ACTION_NAMES = frozenset(_ACTIONS.keys()) - _CORE_ACTION_NAMES - _CORE_ACTIONS = {k: v for k, v in _ACTIONS.items() if k in _CORE_ACTION_NAMES} -_ADMIN_ACTIONS = {k: v for k, v in _ACTIONS.items() if k in _ADMIN_ACTION_NAMES} +_ADMIN_ACTIONS = {k: v for k, v in _ACTIONS.items() if k not in _CORE_ACTION_NAMES} # Single source of truth: action → (required-param signature, description). # Drives the schema description AND runtime required-param validation. @@ -595,18 +609,14 @@ def _load_allowed_actions_config() -> Optional[List[str]]: raw = (cfg.get("discord") or {}).get("server_actions") if raw is None or raw == "": return None - if isinstance(raw, str): - names = [n.strip() for n in raw.split(",") if n.strip()] - elif isinstance(raw, (list, tuple)): - names = [str(n).strip() for n in raw if str(n).strip()] - else: + raw = raw.split(",") + elif not isinstance(raw, (list, tuple)): logger.warning( "discord.server_actions: unexpected type %s; ignoring.", type(raw).__name__, ) return None - - valid = [n for n in names if n in _ACTIONS] + names = [str(n).strip() for n in raw if str(n).strip()] invalid = [n for n in names if n not in _ACTIONS] if invalid: logger.warning( @@ -614,7 +624,7 @@ def _load_allowed_actions_config() -> Optional[List[str]]: "Known: %s", ", ".join(invalid), ", ".join(_ACTIONS.keys()), ) - return valid + return [n for n in names if n in _ACTIONS] def _available_actions( @@ -733,39 +743,31 @@ def get_dynamic_schema_admin() -> Optional[Dict[str, Any]]: # ── 403 error enrichment ───────────────────────────────────────────────────── +_NO_MANAGE_MESSAGES = "Bot lacks MANAGE_MESSAGES permission in this channel" +_VIEW_HISTORY = "Bot cannot view this channel (missing VIEW_CHANNEL or READ_MESSAGE_HISTORY)." +_ROLE_HIERARCHY = ( + "Either the bot lacks MANAGE_ROLES, or the target role sits higher " + "than the bot's highest role." +) + +# Per-action guidance for a call-time 403 (per-guild permissions are never pre-checked). _ACTION_403_HINT = { "pin_message": ( - "Bot lacks MANAGE_MESSAGES permission in this channel. " + f"{_NO_MANAGE_MESSAGES}. " "Ask the server admin to grant the bot a role that has MANAGE_MESSAGES, " "or a per-channel overwrite." ), - "unpin_message": ( - "Bot lacks MANAGE_MESSAGES permission in this channel." - ), - "delete_message": ( - "Bot lacks MANAGE_MESSAGES permission in this channel, or cannot view the channel/message." - ), - "create_thread": ( - "Bot lacks CREATE_PUBLIC_THREADS in this channel, or cannot view it." - ), + "unpin_message": f"{_NO_MANAGE_MESSAGES}.", + "delete_message": f"{_NO_MANAGE_MESSAGES}, or cannot view the channel/message.", + "create_thread": "Bot lacks CREATE_PUBLIC_THREADS in this channel, or cannot view it.", "add_role": ( - "Either the bot lacks MANAGE_ROLES, or the target role sits higher " - "than the bot's highest role. Roles can only be assigned below the " + f"{_ROLE_HIERARCHY} Roles can only be assigned below the " "bot's own position in the role hierarchy." ), - "remove_role": ( - "Either the bot lacks MANAGE_ROLES, or the target role sits higher " - "than the bot's highest role." - ), - "fetch_messages": ( - "Bot cannot view this channel (missing VIEW_CHANNEL or READ_MESSAGE_HISTORY)." - ), - "list_pins": ( - "Bot cannot view this channel (missing VIEW_CHANNEL or READ_MESSAGE_HISTORY)." - ), - "channel_info": ( - "Bot cannot view this channel (missing VIEW_CHANNEL)." - ), + "remove_role": _ROLE_HIERARCHY, + "fetch_messages": _VIEW_HISTORY, + "list_pins": _VIEW_HISTORY, + "channel_info": "Bot cannot view this channel (missing VIEW_CHANNEL).", "search_members": ( "Likely missing the Server Members privileged intent — enable it in the " "Discord Developer Portal under your bot's settings." @@ -859,23 +861,18 @@ def discord_admin_handler(action: str, **kwargs) -> str: def _make_handler(handler_fn): - """Create a registry-compatible handler lambda for a discord handler.""" + """Registry-compatible handler: fills every schema param from ``_HANDLER_DEFAULTS``.""" return lambda args, **kw: handler_fn( **{k: args.get(k, v) for k, v in _HANDLER_DEFAULTS.items()}, ) -_STATIC_CORE_SCHEMA = _build_schema( - list(_CORE_ACTIONS.keys()), caps={"detected": False}, tool_name="discord", -) -_STATIC_ADMIN_SCHEMA = _build_schema( - list(_ADMIN_ACTIONS.keys()), caps={"detected": False}, tool_name="discord_admin", -) - +# Static (un-detected) schemas at import; the intent/config-filtered ones come +# from get_dynamic_schema_core/admin via model_tools' dynamic schema overrides. registry.register( name="discord", toolset="discord", - schema=_STATIC_CORE_SCHEMA, + schema=_build_schema(list(_CORE_ACTIONS), caps={"detected": False}, tool_name="discord"), handler=_make_handler(discord_core), check_fn=check_discord_tool_requirements, requires_env=["DISCORD_BOT_TOKEN"], @@ -884,7 +881,7 @@ registry.register( registry.register( name="discord_admin", toolset="discord_admin", - schema=_STATIC_ADMIN_SCHEMA, + schema=_build_schema(list(_ADMIN_ACTIONS), caps={"detected": False}, tool_name="discord_admin"), handler=_make_handler(discord_admin_handler), check_fn=check_discord_tool_requirements, requires_env=["DISCORD_BOT_TOKEN"], diff --git a/tools/drive_preview_tool.py b/tools/drive_preview_tool.py index ff11397a59..9f4323162e 100644 --- a/tools/drive_preview_tool.py +++ b/tools/drive_preview_tool.py @@ -1,14 +1,19 @@ #!/usr/bin/env python3 """Interact with the in-app browser / preview pane in the Hermes desktop GUI. -Clicking, typing, scrolling, history — so the agent drives the page the user sees. +``open_preview`` shows a page, ``read_preview`` reads it; this is the third leg — +clicking, typing, scrolling, history — so the agent drives the page the user sees. Elements are addressed by legible refs from ``action="elements"`` (``btn-sign-in``, ``inp-email``). A ref lasts while the page is open — including across a re-render that rebuilds the element — and only a navigation retires it (the renderer says so rather than acting on whatever now occupies the spot). Because refs hold, the renderer answers -with a *delta* (appeared/went/changed/rebound) instead of re-sending the inventory. -Round-trips through the gateway's blocking-prompt bridge (``preview.act.request`` -> -``preview.act.respond``); this module is schema + a thin dispatcher over the callback. +with a *delta* (appeared/went/changed/rebound) instead of re-sending the inventory — +cheap only because the refs stay legible on their own several turns later. +Round-trips through the gateway's blocking-prompt bridge like ``read_preview``: +tui_gateway emits ``preview.act.request``, the renderer injects the interaction engine +into the pane's webview and answers ``preview.act.respond``. This module is schema + a +thin dispatcher over the platform-injected callback. Lives in the ``desktop_ui`` toolset, +which the GUI gateway enables only for desktop-sourced sessions. """ from typing import Callable, Optional diff --git a/tools/feishu_doc_tool.py b/tools/feishu_doc_tool.py index 6d2aad8fc6..78cdcc9b49 100644 --- a/tools/feishu_doc_tool.py +++ b/tools/feishu_doc_tool.py @@ -4,32 +4,15 @@ Provides ``feishu_doc_read`` for reading document content as plain text. Uses the same lazy-import + BaseRequest pattern as feishu_comment.py. """ -import json -import logging -import threading - +from tools.feishu_lark import ( # noqa: F401 (set_client/get_client are imported by feishu_comment) + build_request, + _check_feishu, + get_client, + raw_body, + set_client, +) from tools.registry import registry, tool_error, tool_result -logger = logging.getLogger(__name__) - -# Thread-local storage for the lark client injected by feishu_comment handler. -_local = threading.local() - - -def set_client(client): - """Store a lark client for the current thread (called by feishu_comment).""" - _local.client = client - - -def get_client(): - """Return the lark client for the current thread, or None.""" - return getattr(_local, "client", None) - - -# --------------------------------------------------------------------------- -# feishu_doc_read -# --------------------------------------------------------------------------- - _RAW_CONTENT_URI = "/open-apis/docx/v1/documents/:document_id/raw_content" FEISHU_DOC_READ_SCHEMA = { @@ -51,21 +34,6 @@ FEISHU_DOC_READ_SCHEMA = { } -def _check_feishu(): - # Use ``importlib.util.find_spec`` — it checks whether ``lark_oapi`` - # is importable without actually executing its ``__init__``. - # Executing the real import here costs ~5 seconds (the SDK eagerly - # loads websockets, dispatcher, every api/v2 model) and this probe - # fires at every ``hermes`` startup during tool-availability - # evaluation. Correctness is preserved because the actual tool - # handler still does the real import when invoked. - import importlib.util - try: - return importlib.util.find_spec("lark_oapi") is not None - except (ImportError, ValueError): - return False - - def _handle_feishu_doc_read(args: dict, **kwargs) -> str: doc_token = args.get("doc_token", "").strip() if not doc_token: @@ -76,21 +44,10 @@ def _handle_feishu_doc_read(args: dict, **kwargs) -> str: return tool_error("Feishu client not available (not in a Feishu comment context)") try: - from lark_oapi import AccessTokenType - from lark_oapi.core.enum import HttpMethod - from lark_oapi.core.model.base_request import BaseRequest + request = build_request("GET", _RAW_CONTENT_URI, paths={"document_id": doc_token}) except ImportError: return tool_error("lark_oapi not installed") - request = ( - BaseRequest.builder() - .http_method(HttpMethod.GET) - .uri(_RAW_CONTENT_URI) - .token_types({AccessTokenType.TENANT}) - .paths({"document_id": doc_token}) - .build() - ) - # Tool handlers run synchronously in a worker thread (no running event # loop), so call the blocking lark client directly. response = client.request(request) @@ -100,31 +57,22 @@ def _handle_feishu_doc_read(args: dict, **kwargs) -> str: msg = getattr(response, "msg", "unknown error") return tool_error(f"Failed to read document: code={code} msg={msg}") - raw = getattr(response, "raw", None) - if raw and hasattr(raw, "content"): + body = raw_body(response) + if body is not None: try: - body = json.loads(raw.content) - content = body.get("data", {}).get("content", "") - return tool_result(success=True, content=content) - except (json.JSONDecodeError, AttributeError): + return tool_result(success=True, content=body.get("data", {}).get("content", "")) + except AttributeError: pass - # Fallback: try response.data + # Fallback: try the typed response.data data = getattr(response, "data", None) if data: - if isinstance(data, dict): - content = data.get("content", "") - else: - content = getattr(data, "content", str(data)) + content = data.get("content", "") if isinstance(data, dict) else getattr(data, "content", str(data)) return tool_result(success=True, content=content) return tool_error("No content returned from document API") -# --------------------------------------------------------------------------- -# Registration -# --------------------------------------------------------------------------- - registry.register( name="feishu_doc_read", toolset="feishu_doc", diff --git a/tools/feishu_drive_tool.py b/tools/feishu_drive_tool.py index ff4cf14eb5..bc95c6b6d4 100644 --- a/tools/feishu_drive_tool.py +++ b/tools/feishu_drive_tool.py @@ -1,105 +1,76 @@ """Feishu Drive Tools -- document comment operations via Feishu/Lark API. List / reply-to / add document comments through the generic BaseRequest path (lazy SDK -import). The lark client is injected per-thread by the feishu_comment event handler. +import), sharing client/request plumbing with feishu_doc_tool via ``tools.feishu_lark``. +The lark client is injected per-thread by the feishu_comment event handler. """ -import json import logging -import threading +from tools.feishu_lark import ( # noqa: F401 (set_client/get_client are imported by feishu_comment) + build_request, + _check_feishu, + get_client, + response_data, + set_client, +) from tools.registry import registry, tool_error, tool_result logger = logging.getLogger(__name__) -# Thread-local storage for the lark client injected by feishu_comment handler. -_local = threading.local() - - -def set_client(client): - """Store a lark client for the current thread (called by feishu_comment).""" - _local.client = client - - -def get_client(): - """Return the lark client for the current thread, or None.""" - return getattr(_local, "client", None) - - -def _check_feishu(): - # find_spec avoids executing lark_oapi's ~5s __init__ at every hermes startup; - # the handlers do the real import when invoked. - import importlib.util - try: - return importlib.util.find_spec("lark_oapi") is not None - except (ImportError, ValueError): - return False - def _do_request(client, method, uri, paths=None, queries=None, body=None): """Build and execute a BaseRequest, return (code, msg, data_dict).""" - from lark_oapi import AccessTokenType - from lark_oapi.core.enum import HttpMethod - from lark_oapi.core.model.base_request import BaseRequest - - builder = ( - BaseRequest.builder() - .http_method(HttpMethod.GET if method == "GET" else HttpMethod.POST) - .uri(uri) - .token_types({AccessTokenType.TENANT}) - ) - if paths: - builder = builder.paths(paths) - if queries: - builder = builder.queries(queries) - if body is not None: - builder = builder.body(body) - # Tool handlers run synchronously in a worker thread (no running event # loop), so call the blocking lark client directly. - response = client.request(builder.build()) + response = client.request(build_request(method, uri, paths, queries, body)) + return getattr(response, "code", None), getattr(response, "msg", ""), response_data(response) - code = getattr(response, "code", None) - msg = getattr(response, "msg", "") - # Prefer the raw JSON body's "data"; fall back to the typed response.data. - data = {} - raw = getattr(response, "raw", None) - if raw and hasattr(raw, "content"): - try: - data = json.loads(raw.content).get("data", {}) - except (json.JSONDecodeError, AttributeError): - pass - if not data: - resp_data = getattr(response, "data", None) - if isinstance(resp_data, dict): - data = resp_data - elif resp_data and hasattr(resp_data, "__dict__"): - data = vars(resp_data) +def _prepare(args: dict, keys: tuple, missing_msg: str): + """Client check first, then required fields (stripped). Returns (client, values, error|None).""" + client = get_client() + values = tuple(args.get(k, "").strip() for k in keys) + if client is None: + return None, values, tool_error("Feishu client not available") + if not all(values): + return client, values, tool_error(missing_msg) + return client, values, None - return code, msg, data + +def _file_type(args: dict) -> str: + return args.get("file_type", "docx") or "docx" def _paged_queries(args: dict) -> list: - """Common query params for comment/reply listing endpoints.""" + """Query params shared by the comment/reply listing endpoints.""" return [ - ("file_type", args.get("file_type", "docx") or "docx"), + ("file_type", _file_type(args)), ("user_id_type", "open_id"), ("page_size", str(args.get("page_size", 100))), ] +def _with_page_token(queries: list, args: dict) -> list: + """Append page_token last (after any is_whole) so the query order stays as before.""" + page_token = args.get("page_token", "") + if page_token: + queries.append(("page_token", page_token)) + return queries + + _FILE_TOKEN_PROP = {"type": "string", "description": "The document file token."} _FILE_TYPE_PROP = {"type": "string", "description": "File type (default: docx).", "default": "docx"} _PAGE_TOKEN_PROP = {"type": "string", "description": "Pagination token for next page."} +_COMMENTS_URI = "/open-apis/drive/v1/files/:file_token/comments" +_REPLIES_URI = "/open-apis/drive/v1/files/:file_token/comments/:comment_id/replies" +_ADD_COMMENT_URI = "/open-apis/drive/v1/files/:file_token/new_comments" # --------------------------------------------------------------------------- # feishu_drive_list_comments # --------------------------------------------------------------------------- -_LIST_COMMENTS_URI = "/open-apis/drive/v1/files/:file_token/comments" - FEISHU_DRIVE_LIST_COMMENTS_SCHEMA = { "name": "feishu_drive_list_comments", "description": ( @@ -129,29 +100,20 @@ FEISHU_DRIVE_LIST_COMMENTS_SCHEMA = { def _handle_list_comments(args: dict, **kwargs) -> str: - client = get_client() - if client is None: - return tool_error("Feishu client not available") - - file_token = args.get("file_token", "").strip() - if not file_token: - return tool_error("file_token is required") + client, (file_token,), err = _prepare(args, ("file_token",), "file_token is required") + if err: + return err queries = _paged_queries(args) if args.get("is_whole", False): queries.append(("is_whole", "true")) - page_token = args.get("page_token", "") - if page_token: - queries.append(("page_token", page_token)) + _with_page_token(queries, args) code, msg, data = _do_request( - client, "GET", _LIST_COMMENTS_URI, - paths={"file_token": file_token}, - queries=queries, + client, "GET", _COMMENTS_URI, paths={"file_token": file_token}, queries=queries, ) if code != 0: return tool_error(f"List comments failed: code={code} msg={msg}") - return tool_result(data) @@ -159,8 +121,6 @@ def _handle_list_comments(args: dict, **kwargs) -> str: # feishu_drive_list_comment_replies # --------------------------------------------------------------------------- -_LIST_REPLIES_URI = "/open-apis/drive/v1/files/:file_token/comments/:comment_id/replies" - FEISHU_DRIVE_LIST_REPLIES_SCHEMA = { "name": "feishu_drive_list_comment_replies", "description": "List all replies in a comment thread on a Feishu document.", @@ -186,28 +146,19 @@ FEISHU_DRIVE_LIST_REPLIES_SCHEMA = { def _handle_list_replies(args: dict, **kwargs) -> str: - client = get_client() - if client is None: - return tool_error("Feishu client not available") - - file_token = args.get("file_token", "").strip() - comment_id = args.get("comment_id", "").strip() - if not file_token or not comment_id: - return tool_error("file_token and comment_id are required") - - queries = _paged_queries(args) - page_token = args.get("page_token", "") - if page_token: - queries.append(("page_token", page_token)) + client, (file_token, comment_id), err = _prepare( + args, ("file_token", "comment_id"), "file_token and comment_id are required" + ) + if err: + return err code, msg, data = _do_request( - client, "GET", _LIST_REPLIES_URI, + client, "GET", _REPLIES_URI, paths={"file_token": file_token, "comment_id": comment_id}, - queries=queries, + queries=_with_page_token(_paged_queries(args), args), ) if code != 0: return tool_error(f"List replies failed: code={code} msg={msg}") - return tool_result(data) @@ -215,8 +166,6 @@ def _handle_list_replies(args: dict, **kwargs) -> str: # feishu_drive_reply_comment # --------------------------------------------------------------------------- -_REPLY_COMMENT_URI = "/open-apis/drive/v1/files/:file_token/comments/:comment_id/replies" - FEISHU_DRIVE_REPLY_SCHEMA = { "name": "feishu_drive_reply_comment", "description": ( @@ -244,28 +193,21 @@ FEISHU_DRIVE_REPLY_SCHEMA = { def _handle_reply_comment(args: dict, **kwargs) -> str: - client = get_client() - if client is None: - return tool_error("Feishu client not available") - - file_token = args.get("file_token", "").strip() - comment_id = args.get("comment_id", "").strip() - content = args.get("content", "").strip() - if not file_token or not comment_id or not content: - return tool_error("file_token, comment_id, and content are required") - - file_type = args.get("file_type", "docx") or "docx" - body = {"content": {"elements": [{"type": "text_run", "text_run": {"text": content}}]}} + client, (file_token, comment_id, content), err = _prepare( + args, ("file_token", "comment_id", "content"), "file_token, comment_id, and content are required" + ) + if err: + return err + # Replies use the rich "content.elements[text_run]" body shape; file_type is a query param. code, msg, data = _do_request( - client, "POST", _REPLY_COMMENT_URI, + client, "POST", _REPLIES_URI, paths={"file_token": file_token, "comment_id": comment_id}, - queries=[("file_type", file_type)], - body=body, + queries=[("file_type", _file_type(args))], + body={"content": {"elements": [{"type": "text_run", "text_run": {"text": content}}]}}, ) if code != 0: return tool_error(f"Reply comment failed: code={code} msg={msg}") - return tool_result(success=True, data=data) @@ -273,8 +215,6 @@ def _handle_reply_comment(args: dict, **kwargs) -> str: # feishu_drive_add_comment # --------------------------------------------------------------------------- -_ADD_COMMENT_URI = "/open-apis/drive/v1/files/:file_token/new_comments" - FEISHU_DRIVE_ADD_COMMENT_SCHEMA = { "name": "feishu_drive_add_comment", "description": ( @@ -298,26 +238,20 @@ FEISHU_DRIVE_ADD_COMMENT_SCHEMA = { def _handle_add_comment(args: dict, **kwargs) -> str: - client = get_client() - if client is None: - return tool_error("Feishu client not available") - - file_token = args.get("file_token", "").strip() - content = args.get("content", "").strip() - if not file_token or not content: - return tool_error("file_token and content are required") - - file_type = args.get("file_type", "docx") or "docx" - body = {"file_type": file_type, "reply_elements": [{"type": "text", "text": content}]} + client, (file_token, content), err = _prepare( + args, ("file_token", "content"), "file_token and content are required" + ) + if err: + return err + # new_comments takes the flat "reply_elements[text]" shape with file_type in the body. code, msg, data = _do_request( client, "POST", _ADD_COMMENT_URI, paths={"file_token": file_token}, - body=body, + body={"file_type": _file_type(args), "reply_elements": [{"type": "text", "text": content}]}, ) if code != 0: return tool_error(f"Add comment failed: code={code} msg={msg}") - return tool_result(success=True, data=data) diff --git a/tools/feishu_lark.py b/tools/feishu_lark.py new file mode 100644 index 0000000000..fd8c0da0f1 --- /dev/null +++ b/tools/feishu_lark.py @@ -0,0 +1,81 @@ +"""Shared lark_oapi plumbing for the feishu_doc / feishu_drive tool modules. + +Holds the per-thread client handle, the cheap availability probe, and the generic +``BaseRequest`` builder + response unpacking (same pattern as feishu_comment.py). +No ``registry.register`` here on purpose: tool discovery must not import this module. +""" + +import json +import threading + +# The lark client is injected per-thread by the feishu_comment event handler right +# before it runs the agent, so concurrent comment events never see each other's client. +_local = threading.local() + + +def set_client(client): + """Store a lark client for the current thread (called by feishu_comment).""" + _local.client = client + + +def get_client(): + """Return the lark client for the current thread, or None.""" + return getattr(_local, "client", None) + + +def _check_feishu(): + # find_spec checks importability without executing lark_oapi's __init__, which + # eagerly loads websockets/dispatcher/every api model (~5s). This probe fires at + # every ``hermes`` startup; the handlers still do the real import when invoked. + import importlib.util + try: + return importlib.util.find_spec("lark_oapi") is not None + except (ImportError, ValueError): + return False + + +def build_request(method, uri, paths=None, queries=None, body=None): + """Build a tenant-token BaseRequest. Raises ImportError if lark_oapi is missing.""" + from lark_oapi import AccessTokenType + from lark_oapi.core.enum import HttpMethod + from lark_oapi.core.model.base_request import BaseRequest + + builder = ( + BaseRequest.builder() + .http_method(HttpMethod.GET if method == "GET" else HttpMethod.POST) + .uri(uri) + .token_types({AccessTokenType.TENANT}) + ) + if paths: + builder = builder.paths(paths) + if queries: + builder = builder.queries(queries) + if body is not None: + builder = builder.body(body) + return builder.build() + + +def raw_body(response): + """Parsed JSON object of the raw HTTP body, or None when absent/unparseable/not a dict.""" + raw = getattr(response, "raw", None) + if raw and hasattr(raw, "content"): + try: + body = json.loads(raw.content) + except (json.JSONDecodeError, AttributeError): + return None + if isinstance(body, dict): + return body + return None + + +def response_data(response) -> dict: + """``data`` of a lark response: prefer the raw JSON body, fall back to typed .data.""" + body = raw_body(response) + data = body.get("data", {}) if body is not None else {} + if not data: + resp_data = getattr(response, "data", None) + if isinstance(resp_data, dict): + data = resp_data + elif resp_data and hasattr(resp_data, "__dict__"): + data = vars(resp_data) + return data diff --git a/tools/homeassistant_tool.py b/tools/homeassistant_tool.py index 4de74292ab..aa90fd04f5 100644 --- a/tools/homeassistant_tool.py +++ b/tools/homeassistant_tool.py @@ -405,38 +405,17 @@ HA_CALL_SERVICE_SCHEMA = { # Registration # --------------------------------------------------------------------------- -registry.register( - name="ha_list_entities", - toolset="homeassistant", - schema=HA_LIST_ENTITIES_SCHEMA, - handler=_handle_list_entities, - check_fn=_check_ha_available, - emoji="🏠", -) - -registry.register( - name="ha_get_state", - toolset="homeassistant", - schema=HA_GET_STATE_SCHEMA, - handler=_handle_get_state, - check_fn=_check_ha_available, - emoji="🏠", -) - -registry.register( - name="ha_list_services", - toolset="homeassistant", - schema=HA_LIST_SERVICES_SCHEMA, - handler=_handle_list_services, - check_fn=_check_ha_available, - emoji="🏠", -) - -registry.register( - name="ha_call_service", - toolset="homeassistant", - schema=HA_CALL_SERVICE_SCHEMA, - handler=_handle_call_service, - check_fn=_check_ha_available, - emoji="🏠", -) +for _schema, _handler in ( + (HA_LIST_ENTITIES_SCHEMA, _handle_list_entities), + (HA_GET_STATE_SCHEMA, _handle_get_state), + (HA_LIST_SERVICES_SCHEMA, _handle_list_services), + (HA_CALL_SERVICE_SCHEMA, _handle_call_service), +): + registry.register( + name=_schema["name"], + toolset="homeassistant", + schema=_schema, + handler=_handler, + check_fn=_check_ha_available, + emoji="🏠", + ) diff --git a/tools/microsoft_graph_auth.py b/tools/microsoft_graph_auth.py index 46e3aa3875..f58eac5f16 100644 --- a/tools/microsoft_graph_auth.py +++ b/tools/microsoft_graph_auth.py @@ -1,4 +1,4 @@ -"""Microsoft Graph app-only authentication helpers.""" +"""Microsoft Graph app-only (client-credentials) authentication helpers.""" from __future__ import annotations @@ -15,17 +15,35 @@ DEFAULT_GRAPH_SCOPE = "https://graph.microsoft.com/.default" DEFAULT_GRAPH_AUTHORITY_URL = "https://login.microsoftonline.com" DEFAULT_TOKEN_SKEW_SECONDS = 120 +_REQUIRED_ENV = ("MSGRAPH_TENANT_ID", "MSGRAPH_CLIENT_ID", "MSGRAPH_CLIENT_SECRET") + class MicrosoftGraphAuthError(RuntimeError): """Base class for Microsoft Graph auth failures.""" class MicrosoftGraphConfigError(MicrosoftGraphAuthError): - """Raised when Graph credentials are missing or invalid.""" + """Graph credentials are missing or invalid.""" class MicrosoftGraphTokenError(MicrosoftGraphAuthError): - """Raised when token acquisition fails.""" + """Token acquisition failed.""" + + +def format_graph_error(error: Any) -> str | None: + """Render Graph's ``{"error": {"code", "message"}}`` (or bare-string ``error``) body. + + Shared by the token endpoint and the REST client so both surface + ``code: message`` the same way. ``None`` means the shape was unusable. + """ + if isinstance(error, str): + return error + if not isinstance(error, dict): + return None + code, message = error.get("code"), error.get("message") + if code and message: + return f"{code}: {message}" + return str(message) if message else None @dataclass(frozen=True) @@ -40,48 +58,24 @@ class GraphCredentials: @property def token_url(self) -> str: - base = self.authority_url.rstrip("/") tenant = self.tenant_id.strip().strip("/") - return f"{base}/{tenant}/oauth2/v2.0/token" + return f"{self.authority_url.rstrip('/')}/{tenant}/oauth2/v2.0/token" @classmethod def from_env( - cls, - environ: dict[str, str] | None = None, - *, - required: bool = True, + cls, environ: dict[str, str] | None = None, *, required: bool = True ) -> "GraphCredentials | None": env = environ if environ is not None else os.environ - tenant_id = (env.get("MSGRAPH_TENANT_ID") or "").strip() - client_id = (env.get("MSGRAPH_CLIENT_ID") or "").strip() - client_secret = (env.get("MSGRAPH_CLIENT_SECRET") or "").strip() - scope = (env.get("MSGRAPH_SCOPE") or DEFAULT_GRAPH_SCOPE).strip() - authority_url = ( - env.get("MSGRAPH_AUTHORITY_URL") or DEFAULT_GRAPH_AUTHORITY_URL - ).strip() - - missing = [ - name - for name, value in ( - ("MSGRAPH_TENANT_ID", tenant_id), - ("MSGRAPH_CLIENT_ID", client_id), - ("MSGRAPH_CLIENT_SECRET", client_secret), - ) - if not value - ] + values = [(env.get(name) or "").strip() for name in _REQUIRED_ENV] + missing = [name for name, value in zip(_REQUIRED_ENV, values) if not value] if missing: if not required: return None - raise MicrosoftGraphConfigError( - f"Missing Microsoft Graph configuration: {', '.join(missing)}" - ) - + raise MicrosoftGraphConfigError(f"Missing Microsoft Graph configuration: {', '.join(missing)}") return cls( - tenant_id=tenant_id, - client_id=client_id, - client_secret=client_secret, - scope=scope, - authority_url=authority_url, + *values, + scope=(env.get("MSGRAPH_SCOPE") or DEFAULT_GRAPH_SCOPE).strip(), + authority_url=(env.get("MSGRAPH_AUTHORITY_URL") or DEFAULT_GRAPH_AUTHORITY_URL).strip(), ) @@ -105,12 +99,8 @@ class MicrosoftGraphTokenProvider: """Acquire and cache Microsoft Graph app-only access tokens.""" def __init__( - self, - credentials: GraphCredentials, - *, - timeout: float = 20.0, - skew_seconds: int = DEFAULT_TOKEN_SKEW_SECONDS, - transport: httpx.AsyncBaseTransport | None = None, + self, credentials: GraphCredentials, *, timeout: float = 20.0, + skew_seconds: int = DEFAULT_TOKEN_SKEW_SECONDS, transport: httpx.AsyncBaseTransport | None = None, ) -> None: self.credentials = credentials self.timeout = timeout @@ -120,13 +110,8 @@ class MicrosoftGraphTokenProvider: self._lock = asyncio.Lock() @classmethod - def from_env( - cls, - environ: dict[str, str] | None = None, - **kwargs: Any, - ) -> "MicrosoftGraphTokenProvider": - credentials = GraphCredentials.from_env(environ) - return cls(credentials, **kwargs) + def from_env(cls, environ: dict[str, str] | None = None, **kwargs: Any) -> "MicrosoftGraphTokenProvider": + return cls(GraphCredentials.from_env(environ), **kwargs) def clear_cache(self) -> None: self._cached_token = None @@ -146,20 +131,20 @@ class MicrosoftGraphTokenProvider: "refresh_skew_seconds": self.skew_seconds, } - async def get_access_token(self, *, force_refresh: bool = False) -> str: + def _fresh_cached(self) -> CachedAccessToken | None: + """The cached token unless it expires within ``skew_seconds``.""" cached = self._cached_token - if not force_refresh and cached and not cached.is_expired( - skew_seconds=self.skew_seconds - ): + if cached and not cached.is_expired(skew_seconds=self.skew_seconds): + return cached + return None + + async def get_access_token(self, *, force_refresh: bool = False) -> str: + # Double-checked under the lock so concurrent callers share one fetch. + if not force_refresh and (cached := self._fresh_cached()): return cached.access_token - async with self._lock: - cached = self._cached_token - if not force_refresh and cached and not cached.is_expired( - skew_seconds=self.skew_seconds - ): + if not force_refresh and (cached := self._fresh_cached()): return cached.access_token - token = await self._fetch_access_token() self._cached_token = token return token.access_token @@ -171,43 +156,27 @@ class MicrosoftGraphTokenProvider: "client_secret": self.credentials.client_secret, "scope": self.credentials.scope, } - headers = {"Content-Type": "application/x-www-form-urlencoded"} - - async with httpx.AsyncClient( - timeout=httpx.Timeout(self.timeout), - transport=self._transport, - ) as client: + async with httpx.AsyncClient(timeout=httpx.Timeout(self.timeout), transport=self._transport) as client: response = await client.post( - self.credentials.token_url, - data=data, - headers=headers, + self.credentials.token_url, data=data, + headers={"Content-Type": "application/x-www-form-urlencoded"}, ) if response.status_code >= 400: - detail = _extract_error_detail(response) raise MicrosoftGraphTokenError( "Microsoft Graph token request failed with HTTP " - f"{response.status_code}: {detail}" + f"{response.status_code}: {_extract_error_detail(response)}" ) - try: payload = response.json() except ValueError as exc: - raise MicrosoftGraphTokenError( - "Microsoft Graph token response was not valid JSON." - ) from exc + raise MicrosoftGraphTokenError("Microsoft Graph token response was not valid JSON.") from exc access_token = str(payload.get("access_token") or "").strip() - token_type = str(payload.get("token_type") or "Bearer").strip() or "Bearer" - expires_in = payload.get("expires_in") - if not access_token: - raise MicrosoftGraphTokenError( - "Microsoft Graph token response did not include access_token." - ) - + raise MicrosoftGraphTokenError("Microsoft Graph token response did not include access_token.") try: - expires_in_seconds = int(expires_in) + expires_in_seconds = int(payload.get("expires_in")) except (TypeError, ValueError) as exc: raise MicrosoftGraphTokenError( "Microsoft Graph token response did not include a valid expires_in." @@ -215,31 +184,29 @@ class MicrosoftGraphTokenProvider: return CachedAccessToken( access_token=access_token, - token_type=token_type, + token_type=str(payload.get("token_type") or "Bearer").strip() or "Bearer", expires_at=time.time() + max(0, expires_in_seconds), ) def _extract_error_detail(response: httpx.Response) -> str: + """Best human-readable detail from a token-endpoint error body. + + The OAuth endpoint prefers ``error_description``; fall back to the + Graph-style ``error`` object/string, then a bare ``code``, then raw text. + """ try: payload = response.json() except ValueError: - text = response.text.strip() - return text or "unknown error" + return response.text.strip() or "unknown error" if isinstance(payload, dict): if isinstance(payload.get("error_description"), str): return payload["error_description"] error = payload.get("error") - if isinstance(error, dict): - message = error.get("message") - code = error.get("code") - if message and code: - return f"{code}: {message}" - if message: - return str(message) - if code: - return str(code) - if isinstance(error, str): - return error + detail = format_graph_error(error) + if detail is not None: + return detail + if isinstance(error, dict) and error.get("code"): + return str(error["code"]) return str(payload) diff --git a/tools/microsoft_graph_client.py b/tools/microsoft_graph_client.py index 60831801bf..ab8842dcf4 100644 --- a/tools/microsoft_graph_client.py +++ b/tools/microsoft_graph_client.py @@ -10,11 +10,18 @@ from typing import Any, AsyncIterator, Awaitable, Callable import httpx from agent.retry_utils import parse_retry_after_seconds -from tools.microsoft_graph_auth import GraphCredentials, MicrosoftGraphTokenProvider +from tools.microsoft_graph_auth import ( + GraphCredentials, + MicrosoftGraphTokenProvider, + format_graph_error, +) DEFAULT_GRAPH_BASE_URL = "https://graph.microsoft.com/v1.0" +Headers = dict[str, str] | None +Params = dict[str, Any] | None + class MicrosoftGraphClientError(RuntimeError): """Base class for Graph client failures.""" @@ -24,35 +31,28 @@ class MicrosoftGraphAPIError(MicrosoftGraphClientError): """Raised when a Graph API request fails.""" def __init__( - self, - status_code: int, - method: str, - url: str, - message: str, - *, - retry_after_seconds: float | None = None, - payload: Any = None, + self, status_code: int, method: str, url: str, message: str, *, + retry_after_seconds: float | None = None, payload: Any = None, ) -> None: self.status_code = status_code self.method = method self.url = url self.retry_after_seconds = retry_after_seconds self.payload = payload - super().__init__( - f"Microsoft Graph API error {status_code} for {method} {url}: {message}" - ) + super().__init__(f"Microsoft Graph API error {status_code} for {method} {url}: {message}") class MicrosoftGraphClient: - """Minimal async Microsoft Graph client with retries and pagination.""" + """Minimal async Microsoft Graph client with retries and pagination. + + Retry policy (shared by JSON requests and streaming downloads): transport + errors back off exponentially; 401 clears the token cache and refetches; + 429/5xx honor ``Retry-After``. Each attempt uses a fresh ``AsyncClient``. + """ def __init__( - self, - token_provider: MicrosoftGraphTokenProvider, - *, - base_url: str = DEFAULT_GRAPH_BASE_URL, - timeout: float = 60.0, - max_retries: int = 3, + self, token_provider: MicrosoftGraphTokenProvider, *, + base_url: str = DEFAULT_GRAPH_BASE_URL, timeout: float = 60.0, max_retries: int = 3, transport: httpx.AsyncBaseTransport | None = None, sleep: Callable[[float], Awaitable[None]] | None = None, user_agent: str = "Hermes-Agent/graph-client", @@ -67,69 +67,34 @@ class MicrosoftGraphClient: @classmethod def from_env(cls, **kwargs: Any) -> "MicrosoftGraphClient": - credentials = GraphCredentials.from_env() - provider = MicrosoftGraphTokenProvider(credentials) - return cls(provider, **kwargs) + return cls(MicrosoftGraphTokenProvider(GraphCredentials.from_env()), **kwargs) - async def get_json( - self, - path: str, - *, - params: dict[str, Any] | None = None, - headers: dict[str, str] | None = None, - ) -> Any: - response = await self._request("GET", path, params=params, headers=headers) - return self._decode_json(response) + async def get_json(self, path: str, *, params: Params = None, headers: Headers = None) -> Any: + return self._decode_json(await self._request("GET", path, params=params, headers=headers)) - async def post_json( - self, - path: str, - *, - json_body: Any | None = None, - headers: dict[str, str] | None = None, - ) -> Any: - response = await self._request("POST", path, json_body=json_body, headers=headers) - return self._decode_json(response) + async def post_json(self, path: str, *, json_body: Any | None = None, headers: Headers = None) -> Any: + return self._decode_json(await self._request("POST", path, json_body=json_body, headers=headers)) - async def patch_json( - self, - path: str, - *, - json_body: Any | None = None, - headers: dict[str, str] | None = None, - ) -> Any: + async def patch_json(self, path: str, *, json_body: Any | None = None, headers: Headers = None) -> Any: response = await self._request("PATCH", path, json_body=json_body, headers=headers) if response.status_code == 204 or not response.content: return {} return self._decode_json(response) - async def delete( - self, - path: str, - *, - headers: dict[str, str] | None = None, - ) -> dict[str, Any]: + async def delete(self, path: str, *, headers: Headers = None) -> dict[str, Any]: response = await self._request("DELETE", path, headers=headers) if response.status_code == 204 or not response.content: return {"deleted": True, "status_code": response.status_code} return self._decode_json(response) async def iterate_pages( - self, - path: str, - *, - params: dict[str, Any] | None = None, - headers: dict[str, str] | None = None, + self, path: str, *, params: Params = None, headers: Headers = None ) -> AsyncIterator[dict[str, Any]]: + # Query params go on the first request only; @odata.nextLink already embeds them. next_url: str | None = self._resolve_url(path) next_params = dict(params or {}) while next_url: - response = await self._request( - "GET", - next_url, - params=next_params or None, - headers=headers, - ) + response = await self._request("GET", next_url, params=next_params or None, headers=headers) payload = self._decode_json(response) if not isinstance(payload, dict): raise MicrosoftGraphClientError( @@ -140,11 +105,7 @@ class MicrosoftGraphClient: next_params = {} async def collect_paginated( - self, - path: str, - *, - params: dict[str, Any] | None = None, - headers: dict[str, str] | None = None, + self, path: str, *, params: Params = None, headers: Headers = None ) -> list[Any]: items: list[Any] = [] async for page in self.iterate_pages(path, params=params, headers=headers): @@ -154,180 +115,101 @@ class MicrosoftGraphClient: return items async def download_to_file( - self, - path: str, - destination: str | Path, - *, - headers: dict[str, str] | None = None, - chunk_size: int = 65536, + self, path: str, destination: str | Path, *, headers: Headers = None, chunk_size: int = 65536 ) -> dict[str, Any]: - """Download a Graph resource to disk, streaming the response body. - - The body is written chunk-by-chunk via ``response.aiter_bytes`` with - the ``httpx.AsyncClient`` kept open for the duration of the iteration, - so recordings and other large artifacts do not need to fit in memory. - """ + """Download a Graph resource to disk, streaming the body chunk-by-chunk + (recordings and other large artifacts never need to fit in memory). + Written to a ``.part`` file and renamed into place only on success.""" url = self._resolve_url(path) target = Path(destination) target.parent.mkdir(parents=True, exist_ok=True) tmp_target = target.with_suffix(target.suffix + ".part") - attempt = 0 - last_error: Exception | None = None - - while attempt <= self.max_retries: - token = await self.token_provider.get_access_token( - force_refresh=attempt > 0 and self._should_refresh_token(last_error) - ) - request_headers = { - "Authorization": f"Bearer {token}", - "Accept": "*/*", - "User-Agent": self.user_agent, - } - if headers: - request_headers.update(headers) - + async def perform(client: httpx.AsyncClient, request_headers: dict[str, str]): try: - async with httpx.AsyncClient( - timeout=httpx.Timeout(self.timeout), - transport=self._transport, - ) as client: - async with client.stream( - "GET", - url, - headers=request_headers, - ) as response: - if response.status_code >= 400: - # Materialize error body so we can surface a meaningful - # message; error bodies are small. - await response.aread() - api_error = self._build_api_error("GET", url, response) - last_error = api_error - - if ( - response.status_code == 401 - and attempt < self.max_retries - ): - self.token_provider.clear_cache() - await self._sleep( - self._retry_delay(response, attempt) - ) - attempt += 1 - continue - - if ( - self._should_retry(response) - and attempt < self.max_retries - ): - await self._sleep( - self._retry_delay(response, attempt) - ) - attempt += 1 - continue - - raise api_error - - content_type = response.headers.get("content-type") - with tmp_target.open("wb") as handle: - async for chunk in response.aiter_bytes( - chunk_size=chunk_size - ): - if chunk: - handle.write(chunk) - except httpx.HTTPError as exc: - last_error = exc + async with client.stream("GET", url, headers=request_headers) as response: + if response.status_code >= 400: + # Materialize the (small) error body so the message is meaningful. + await response.aread() + return response, None + with tmp_target.open("wb") as handle: + async for chunk in response.aiter_bytes(chunk_size=chunk_size): + if chunk: + handle.write(chunk) + return response, response.headers.get("content-type") + except httpx.HTTPError: tmp_target.unlink(missing_ok=True) - if attempt >= self.max_retries: - raise MicrosoftGraphClientError( - f"Microsoft Graph download failed for GET {url}: {exc}" - ) from exc - await self._sleep(self._retry_delay(None, attempt)) - attempt += 1 - continue + raise - os.replace(tmp_target, target) - return { - "path": str(target), - "size_bytes": target.stat().st_size, - "content_type": content_type, - } - - tmp_target.unlink(missing_ok=True) - raise MicrosoftGraphClientError( - f"Microsoft Graph download exhausted retries for GET {url}." - ) + content_type = await self._with_retries("GET", url, "*/*", None, headers, perform, "download") + os.replace(tmp_target, target) + return {"path": str(target), "size_bytes": target.stat().st_size, "content_type": content_type} async def _request( - self, - method: str, - path_or_url: str, - *, - params: dict[str, Any] | None = None, - json_body: Any | None = None, - headers: dict[str, str] | None = None, + self, method: str, path_or_url: str, *, + params: Params = None, json_body: Any | None = None, headers: Headers = None, ) -> httpx.Response: url = self._resolve_url(path_or_url) + + async def perform(client: httpx.AsyncClient, request_headers: dict[str, str]): + response = await client.request(method, url, params=params, json=json_body, headers=request_headers) + return response, response + + return await self._with_retries(method, url, "application/json", json_body, headers, perform, "request") + + async def _with_retries( + self, method: str, url: str, accept: str, json_body: Any | None, headers: Headers, + perform: Callable[[httpx.AsyncClient, dict[str, str]], Awaitable[tuple[httpx.Response, Any]]], + kind: str, + ) -> Any: + """Run ``perform`` (returning ``(response, result)``) under the retry policy. + + ``kind`` ("request"/"download") only labels the transport-failure messages. + A ``MicrosoftGraphAPIError`` for the failing status is raised once retries + are exhausted or the status is not retryable; only a 401 forces a token refresh. + """ attempt = 0 last_error: Exception | None = None while attempt <= self.max_retries: token = await self.token_provider.get_access_token( - force_refresh=attempt > 0 and self._should_refresh_token(last_error) + force_refresh=attempt > 0 + and isinstance(last_error, MicrosoftGraphAPIError) + and last_error.status_code == 401 ) - request_headers = { - "Authorization": f"Bearer {token}", - "Accept": "application/json", - "User-Agent": self.user_agent, - } + request_headers = {"Authorization": f"Bearer {token}", "Accept": accept, "User-Agent": self.user_agent} if json_body is not None: request_headers["Content-Type"] = "application/json" if headers: request_headers.update(headers) try: - async with httpx.AsyncClient( - timeout=httpx.Timeout(self.timeout), - transport=self._transport, - ) as client: - response = await client.request( - method, - url, - params=params, - json=json_body, - headers=request_headers, - ) + async with httpx.AsyncClient(timeout=httpx.Timeout(self.timeout), transport=self._transport) as client: + response, result = await perform(client, request_headers) except httpx.HTTPError as exc: last_error = exc if attempt >= self.max_retries: raise MicrosoftGraphClientError( - f"Microsoft Graph request failed for {method} {url}: {exc}" + f"Microsoft Graph {kind} failed for {method} {url}: {exc}" ) from exc await self._sleep(self._retry_delay(None, attempt)) attempt += 1 continue if response.status_code < 400: - return response + return result - api_error = self._build_api_error(method, url, response) - last_error = api_error - - if response.status_code == 401 and attempt < self.max_retries: - self.token_provider.clear_cache() + api_error = last_error = self._build_api_error(method, url, response) + status = response.status_code + if attempt < self.max_retries and (status in (401, 429) or 500 <= status < 600): + if status == 401: + self.token_provider.clear_cache() await self._sleep(self._retry_delay(response, attempt)) attempt += 1 continue - - if self._should_retry(response) and attempt < self.max_retries: - await self._sleep(self._retry_delay(response, attempt)) - attempt += 1 - continue - raise api_error - raise MicrosoftGraphClientError( - f"Microsoft Graph request exhausted retries for {method} {url}." - ) + raise MicrosoftGraphClientError(f"Microsoft Graph {kind} exhausted retries for {method} {url}.") def _resolve_url(self, path_or_url: str) -> str: if path_or_url.startswith(("http://", "https://")): @@ -345,16 +227,6 @@ class MicrosoftGraphClient: f"{response.request.method} {response.request.url}" ) from exc - @staticmethod - def _should_retry(response: httpx.Response | None) -> bool: - if response is None: - return True - return response.status_code == 429 or 500 <= response.status_code < 600 - - @staticmethod - def _should_refresh_token(error: Exception | None) -> bool: - return isinstance(error, MicrosoftGraphAPIError) and error.status_code == 401 - @staticmethod def _retry_delay(response: httpx.Response | None, attempt: int) -> float: if response is not None: @@ -364,37 +236,17 @@ class MicrosoftGraphClient: return min(8.0, 0.5 * (2 ** attempt)) @staticmethod - def _build_api_error( - method: str, - url: str, - response: httpx.Response, - ) -> MicrosoftGraphAPIError: - payload: Any = None + def _build_api_error(method: str, url: str, response: httpx.Response) -> MicrosoftGraphAPIError: message = response.text.strip() or "unknown error" try: - payload = response.json() + payload: Any = response.json() except ValueError: payload = None - if isinstance(payload, dict): - error = payload.get("error") - if isinstance(error, dict): - code = error.get("code") - inner_message = error.get("message") - if code and inner_message: - message = f"{code}: {inner_message}" - elif inner_message: - message = str(inner_message) - elif isinstance(error, str): - message = error - - retry_after: float | None = parse_retry_after_seconds(response.headers) - + detail = format_graph_error(payload.get("error")) + if detail is not None: + message = detail return MicrosoftGraphAPIError( - response.status_code, - method, - url, - message, - retry_after_seconds=retry_after, - payload=payload, + response.status_code, method, url, message, + retry_after_seconds=parse_retry_after_seconds(response.headers), payload=payload, ) diff --git a/tools/react_to_message_tool.py b/tools/react_to_message_tool.py index 96ee192c3a..fa37347fda 100644 --- a/tools/react_to_message_tool.py +++ b/tools/react_to_message_tool.py @@ -1,16 +1,12 @@ #!/usr/bin/env python3 """Let the agent react to a message with an emoji in the Hermes desktop app. -The conversational counterpart to the user's tapback: the same reaction store, -the same one-per-author semantics, just written with ``author="agent"``. - -Lives in the ``desktop_ui`` toolset (like the other GUI affordances) so it costs -nothing on every other surface — the platform adapters already expose reactions -through ``send_message(action="react")``, and this is the desktop's equivalent. - -Defaults to the message that triggered this turn (the photon precedent: the -model shouldn't have to thread row ids through tool calls), and emits -``message.reaction`` so the renderer paints it without waiting for a resume. +The conversational counterpart to the user's tapback: same reaction store, +same one-per-author semantics, written with ``author="agent"``. Lives in the +``desktop_ui`` toolset so it costs nothing on other surfaces (platform adapters +expose reactions via ``send_message(action="react")``). Defaults to the message +that triggered this turn so the model needn't thread row ids through tool calls, +and emits ``message.reaction`` so the renderer paints it without a resume. """ import json @@ -18,7 +14,6 @@ import json from gateway.session_context import get_session_env from tools import desktop_ui from tools.registry import registry, tool_error -from utils import env_var_enabled def _open_session_db(): @@ -31,25 +26,13 @@ def _open_session_db(): return None -def _react_to_message_with_db( - emoji: str, - message_row_id=None, - messages_back=None, - *, - db, - session_key: str, -) -> str: - """Attach (or with an empty ``emoji`` retract) the agent's reaction.""" - if not session_key: - return tool_error("No active session — reactions need a persisted conversation.") - +def _react(emoji: str, message_row_id, messages_back, *, db, session_key: str) -> str: row_id = message_row_id target_role = "user" if row_id is None: - # Default target: the latest user message. `messages_back` steps to - # earlier user turns (1 = the one before, etc.) for retroactive - # reactions — quoting text would be ambiguous, ids aren't visible to - # the model, but "two messages ago" is how a person thinks about it. + # Default target: the latest user message. `messages_back` steps to earlier + # user turns (1 = the one before) — ids aren't visible to the model and + # quoting text is ambiguous, but "two messages ago" is how a person thinks. back = max(0, int(messages_back or 0)) row_id = db.latest_message_row_id(session_key, role="user", offset=back) if row_id is None: @@ -57,23 +40,18 @@ def _react_to_message_with_db( f"No user message found {back} back." if back else "No user message to react to yet." ) else: - row = db.get_message_role(session_key, int(row_id)) - target_role = row or "user" + target_role = db.get_message_role(session_key, int(row_id)) or "user" try: - reactions = db.set_message_reaction( - session_key, int(row_id), emoji or None, author="agent" - ) + reactions = db.set_message_reaction(session_key, int(row_id), emoji or None, author="agent") except Exception as exc: return tool_error(f"Failed to set the reaction: {exc}") - if reactions is None: return tool_error(f"Message {row_id} is not part of this conversation.") - # Paint it live. A missing bridge (non-desktop surface) is not an error — - # the reaction is persisted either way and shows on the next load. - # `role` lets the renderer match a live message that doesn't know its - # durable row id yet (it only learns rowId on resume). + # Paint it live. A missing bridge (non-desktop surface) is not an error — the + # reaction is persisted and shows on next load. `role` lets the renderer match + # a live message that doesn't know its durable row id yet (learned on resume). try: desktop_ui.emit( "message.reaction", @@ -93,22 +71,14 @@ def react_to_message_tool(emoji: str, message_row_id=None, messages_back=None) - session_key = get_session_env("HERMES_SESSION_KEY", "") or get_session_env( "HERMES_SESSION_ID", "" ) - if not session_key: return tool_error("No active session — reactions need a persisted conversation.") db = _open_session_db() if db is None: return tool_error("Session storage is unavailable.") - try: - return _react_to_message_with_db( - emoji, - message_row_id, - messages_back, - db=db, - session_key=session_key, - ) + return _react(emoji, message_row_id, messages_back, db=db, session_key=session_key) finally: try: from hermes_state import release_or_close @@ -118,11 +88,7 @@ def react_to_message_tool(emoji: str, message_row_id=None, messages_back=None) - def check_react_requirements() -> bool: - """Opt-in feature flag — surface eligibility is the toolset's job. - - ``desktop_ui`` already restricts this to GUI sessions. What's left is the - user's own toggle (Settings → Appearance). - """ + """Opt-in flag (Settings → Appearance); ``desktop_ui`` already restricts to GUI sessions.""" return desktop_ui.user_enabled("message_reactions", default=False) diff --git a/tools/send_message_senders.py b/tools/send_message_senders.py index 2027ca8cf4..556c859a0e 100644 --- a/tools/send_message_senders.py +++ b/tools/send_message_senders.py @@ -89,7 +89,16 @@ def _display_chat_id(platform_name: str, chat_id: str) -> str: return chat_id +_TELEGRAM_TRANSIENT_MARKERS = ( + "bad gateway", "502", "too many requests", "429", + "service unavailable", "503", "gateway timeout", "504", +) + + def _telegram_retry_delay(exc: Exception, attempt: int) -> float | None: + """Seconds to wait before retrying, or None when the error is final. + Honours Telegram's ``retry_after``; timeouts are never retried (the send + may have gone through); 5xx/429 back off exponentially.""" retry_after = getattr(exc, "retry_after", None) if retry_after is not None: try: @@ -100,21 +109,13 @@ def _telegram_retry_delay(exc: Exception, attempt: int) -> float | None: text = str(exc).lower() if "timed out" in text or "timeout" in text: return None - if ( - "bad gateway" in text - or "502" in text - or "too many requests" in text - or "429" in text - or "service unavailable" in text - or "503" in text - or "gateway timeout" in text - or "504" in text - ): + if any(marker in text for marker in _TELEGRAM_TRANSIENT_MARKERS): return float(2 ** attempt) return None async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs): + """``bot.send_message`` with bounded retries on transient failures.""" for attempt in range(attempts): try: return await bot.send_message(**kwargs) @@ -178,6 +179,7 @@ def _telegram_thread_kwargs(thread_id): def _strip_mdv2_safe(text): + """Strip MarkdownV2 escapes for the plain-text fallback; identity if unavailable.""" try: from plugins.platforms.telegram.adapter import _strip_mdv2 return _strip_mdv2(text) @@ -199,15 +201,102 @@ async def _telegram_send_media(bot, chat_id, f, ext, is_voice, force_document, * return await bot.send_document(chat_id=chat_id, document=f, **kwargs) +async def _telegram_send_text_chunk(bot, chat_id, chunk, parse_mode, has_html, text_kwargs): + """Send one formatted text chunk with the adapter-matching fallbacks: + thread-not-found -> retry without ``message_thread_id`` (dropped from + ``text_kwargs`` for later chunks too); parse failure -> plain text.""" + try: + return await _send_telegram_message_with_retry( + bot, chat_id=chat_id, text=chunk, parse_mode=parse_mode, **text_kwargs + ) + except Exception as md_error: + if _is_telegram_thread_not_found(md_error) and text_kwargs.get("message_thread_id") is not None: + logger.warning( + "Thread %s not found in _send_telegram, retrying without message_thread_id", + text_kwargs.get("message_thread_id"), + ) + text_kwargs.pop("message_thread_id", None) + return await _send_telegram_message_with_retry( + bot, chat_id=chat_id, text=chunk, parse_mode=parse_mode, **text_kwargs + ) + err_text = str(md_error).lower() + if "parse" in err_text or "markdown" in err_text or "html" in err_text: + logger.warning( + "Parse mode %s failed in _send_telegram, falling back to plain text: %s", + parse_mode, + _sanitize_error_text(md_error), + ) + plain = chunk if has_html else _strip_mdv2_safe(chunk) + return await _send_telegram_message_with_retry( + bot, chat_id=chat_id, text=plain, parse_mode=None, **text_kwargs + ) + raise + + +async def _telegram_send_one_media( + bot, chat_id, media_path, is_voice, *, caption, parse_mode, has_html, thread_kwargs, force_document +): + """Upload one file with the adapter-matching fallbacks (thread-not-found -> + no ``message_thread_id``; caption parse failure -> plain caption). Each + retry re-seeks the file because the first attempt consumed it.""" + ext = os.path.splitext(media_path)[1].lower() + voice_note = ext in _VOICE_EXTS and is_voice + with open(media_path, "rb") as f: + media_kwargs = dict(thread_kwargs) + # ``caption`` is only set for a single captionable file, so this never + # double-captions a multi-file send or a voice note. + if caption is not None and not voice_note: + media_kwargs["caption"] = caption + media_kwargs["parse_mode"] = parse_mode + if voice_note or ext in _TELEGRAM_SEND_AUDIO_EXTS: + try: + from plugins.platforms.telegram.adapter import _probe_voice_duration_seconds + duration = await asyncio.to_thread(_probe_voice_duration_seconds, media_path) + if duration is not None: + media_kwargs["duration"] = duration + except Exception: + pass + + async def _send(**kw): + return await _telegram_send_media(bot, chat_id, f, ext, is_voice, force_document, **kw) + + try: + return await _send(**media_kwargs) + except Exception as media_err: + if _is_telegram_thread_not_found(media_err) and media_kwargs.get("message_thread_id"): + logger.warning( + "Thread %s not found for media send, retrying without message_thread_id", + media_kwargs["message_thread_id"], + ) + f.seek(0) + media_kwargs.pop("message_thread_id", None) + return await _send(**media_kwargs) + err_text = str(media_err).lower() + if media_kwargs.get("parse_mode") and ("parse" in err_text or "caption" in err_text): + logger.warning( + "Caption parse failed for media send, retrying plain: %s", + _sanitize_error_text(media_err), + ) + f.seek(0) + media_kwargs.pop("parse_mode", None) + if not has_html and media_kwargs.get("caption"): + media_kwargs["caption"] = _strip_mdv2_safe(media_kwargs["caption"]) + return await _send(**media_kwargs) + raise + + async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False, force_document=False): """Send via Telegram Bot API (one-shot, no polling needed). - Markdown is converted to MarkdownV2 like the gateway adapter; a message that - already contains HTML tags is sent with ``parse_mode='HTML'`` instead. + Markdown is converted to MarkdownV2 via the gateway adapter's + ``format_message`` so bold/links/headers render; a message that already + contains HTML tags skips that and is sent with ``parse_mode='HTML'``. + Parse failures fall back to plain text so the message still delivers. """ try: from telegram.constants import ParseMode + # Auto-detect HTML tags: if present, skip MarkdownV2 and send as HTML. _has_html = bool(re.search(r'<[a-zA-Z/][^>]*>', message)) if _has_html: formatted = message @@ -252,42 +341,10 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No # Chunk *after* formatting, in UTF-16 units: MarkdownV2/HTML # escaping inflates text, so a raw-<4096 message can exceed the # limit once formatted and be rejected as "Message is too long". - text_chunks = BasePlatformAdapter.truncate_message(formatted, 4096, len_fn=utf16_len) - for chunk in text_chunks: - try: - last_msg = await _send_telegram_message_with_retry( - bot, - chat_id=int_chat_id, text=chunk, - parse_mode=send_parse_mode, **text_kwargs - ) - except Exception as md_error: - # Thread not found: retry without message_thread_id so the - # message still delivers (matches the gateway adapter). - if _is_telegram_thread_not_found(md_error) and text_kwargs.get("message_thread_id") is not None: - logger.warning( - "Thread %s not found in _send_telegram, retrying without message_thread_id", - text_kwargs.get("message_thread_id"), - ) - text_kwargs.pop("message_thread_id", None) - last_msg = await _send_telegram_message_with_retry( - bot, - chat_id=int_chat_id, text=chunk, - parse_mode=send_parse_mode, **text_kwargs - ) - elif "parse" in str(md_error).lower() or "markdown" in str(md_error).lower() or "html" in str(md_error).lower(): - logger.warning( - "Parse mode %s failed in _send_telegram, falling back to plain text: %s", - send_parse_mode, - _sanitize_error_text(md_error), - ) - plain = chunk if _has_html else _strip_mdv2_safe(chunk) - last_msg = await _send_telegram_message_with_retry( - bot, - chat_id=int_chat_id, text=plain, - parse_mode=None, **text_kwargs - ) - else: - raise + for chunk in BasePlatformAdapter.truncate_message(formatted, 4096, len_fn=utf16_len): + last_msg = await _telegram_send_text_chunk( + bot, int_chat_id, chunk, send_parse_mode, _has_html, text_kwargs + ) for media_path, is_voice in media_files: if not os.path.exists(media_path): @@ -310,57 +367,12 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No ) continue - ext = os.path.splitext(media_path)[1].lower() try: - with open(media_path, "rb") as f: - media_kwargs = dict(thread_kwargs) - # _tg_caption is only set for a single captionable file, so - # this never double-captions a multi-file send or a voice note. - if _tg_caption is not None and not (ext in _VOICE_EXTS and is_voice): - media_kwargs["caption"] = _tg_caption - media_kwargs["parse_mode"] = send_parse_mode - if (ext in _VOICE_EXTS and is_voice) or ext in _TELEGRAM_SEND_AUDIO_EXTS: - try: - from plugins.platforms.telegram.adapter import _probe_voice_duration_seconds - duration = await asyncio.to_thread(_probe_voice_duration_seconds, media_path) - if duration is not None: - media_kwargs["duration"] = duration - except Exception: - pass - - async def _send(**kw): - return await _telegram_send_media( - bot, int_chat_id, f, ext, is_voice, force_document, **kw - ) - - try: - last_msg = await _send(**media_kwargs) - except Exception as media_err: - if _is_telegram_thread_not_found(media_err) and media_kwargs.get("message_thread_id"): - logger.warning( - "Thread %s not found for media send, retrying without message_thread_id", - media_kwargs["message_thread_id"], - ) - f.seek(0) # the first attempt consumed the file - media_kwargs.pop("message_thread_id", None) - last_msg = await _send(**media_kwargs) - elif media_kwargs.get("parse_mode") and ( - "parse" in str(media_err).lower() - or "caption" in str(media_err).lower() - ): - # Caption failed to parse as MarkdownV2/HTML: retry - # with a plain-text caption so media still delivers. - logger.warning( - "Caption parse failed for media send, retrying plain: %s", - _sanitize_error_text(media_err), - ) - f.seek(0) - media_kwargs.pop("parse_mode", None) - if not _has_html and media_kwargs.get("caption"): - media_kwargs["caption"] = _strip_mdv2_safe(media_kwargs["caption"]) - last_msg = await _send(**media_kwargs) - else: - raise + last_msg = await _telegram_send_one_media( + bot, int_chat_id, media_path, is_voice, + caption=_tg_caption, parse_mode=send_parse_mode, has_html=_has_html, + thread_kwargs=thread_kwargs, force_document=force_document, + ) except Exception as e: warning = _sanitize_error_text(f"Failed to send media {media_path}: {e}") logger.error(warning) diff --git a/tools/send_message_targets.py b/tools/send_message_targets.py index 5e094c067c..ec2fc5210d 100644 --- a/tools/send_message_targets.py +++ b/tools/send_message_targets.py @@ -57,26 +57,34 @@ _EMAIL_TARGET_RE = re.compile(r"^\s*[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2 _HOME_CHANNEL_ENV_OVERRIDES = {"email": "EMAIL_HOME_ADDRESS"} -# Per-platform explicit-target parsers: target_ref -> (chat_id, thread_id) or -# None to fall through to the generic rules below. Order inside each parser -# matters (e.g. Slack thread form before bare conversation id). +_UNRESOLVED = object() # sentinel: stop parsing, target is NOT explicit (skip generic rules) + + +# Per-platform explicit-target parsers: target_ref -> (chat_id, thread_id), +# None to fall through to the generic rules in _parse_target_ref, or +# _UNRESOLVED. Order inside each parser matters (e.g. Slack thread form +# before bare conversation id). +def _parse_regex_groups(regex, *, thread_group=True): + """Explicit when ``regex`` fully matches: chat_id = group 1, thread = group 2 (or None).""" + def parse(ref): + match = regex.fullmatch(ref) + if not match: + return None + return match.group(1), (match.group(2) if thread_group else None) + return parse + + def _parse_telegram(ref): - match = _TELEGRAM_TOPIC_TARGET_RE.fullmatch(ref) - if match: - return match.group(1), match.group(2) + # "[:]" or an @username (usernames must not be force-int'd). + parsed = _parse_regex_groups(_TELEGRAM_TOPIC_TARGET_RE)(ref) + if parsed: + return parsed from plugins.platforms.telegram.telegram_ids import parse_telegram_username_target username = parse_telegram_username_target(ref) return (username, None) if username else None -def _parse_topic(regex): - def parse(ref): - match = regex.fullmatch(ref) - return (match.group(1), match.group(2)) if match else None - return parse - - def _parse_slack(ref): match = _SLACK_THREAD_TARGET_RE.fullmatch(ref) if match: @@ -92,13 +100,13 @@ def _parse_slack(ref): def _parse_matrix(ref): - # ":$" addresses a thread; bare "!room" / "@user" handled below. + # ":$" addresses a thread (rfind: room ids contain ':'). + # Bare "!room" / "@user" are explicit too, but via the generic rule so the + # numeric check keeps precedence exactly as before. trimmed = ref.strip() split_idx = trimmed.rfind(":$") if split_idx > 0: return trimmed[:split_idx], trimmed[split_idx + 1 :] - if trimmed.startswith(("!", "@")): - return None # deferred: numeric check must run first (generic rule) return None @@ -109,20 +117,15 @@ def _parse_regex_stripped(regex): return parse -def _parse_group1(regex): - def parse(ref): - match = regex.fullmatch(ref) - return (match.group(1), None) if match else None - return parse - - def _parse_yuanbao(ref): + # "group:" / "direct:"; a bare number is a group code, never a + # generic numeric chat id (yuanbao never falls through to the generic rules). match = _YUANBAO_TARGET_RE.fullmatch(ref) if match: return match.group(1), None if ref.strip().isdigit(): return f"group:{ref.strip()}", None - return _UNRESOLVED # yuanbao never falls through to the generic rules + return _UNRESOLVED def _parse_nonempty(ref): @@ -133,6 +136,7 @@ def _parse_nonempty(ref): def _parse_signal(ref): + # "group:" is a native group target; an empty id is not explicit. stripped = ref.strip() if stripped.startswith("group:"): group_id = stripped[len("group:"):].strip() @@ -140,15 +144,13 @@ def _parse_signal(ref): return None -_UNRESOLVED = object() # sentinel: stop parsing, target is not explicit - _PLATFORM_PARSERS = { "telegram": _parse_telegram, - "feishu": _parse_topic(_FEISHU_TARGET_RE), - "discord": _parse_topic(_NUMERIC_TOPIC_RE), + "feishu": _parse_regex_groups(_FEISHU_TARGET_RE), + "discord": _parse_regex_groups(_NUMERIC_TOPIC_RE), # "[:]" snowflakes "slack": _parse_slack, "matrix": _parse_matrix, - "weixin": _parse_group1(_WEIXIN_TARGET_RE), + "weixin": _parse_regex_groups(_WEIXIN_TARGET_RE, thread_group=False), "yuanbao": _parse_yuanbao, "ntfy": _parse_nonempty, "email": _parse_regex_stripped(_EMAIL_TARGET_RE), @@ -165,10 +167,11 @@ _PLATFORM_PARSERS = { def _parse_target_ref(platform_name: str, target_ref: str): """Parse a tool target into (chat_id, thread_id, explicit). - Platform parser first, then the shared rules: E.164 phone numbers (with - the '+' the signal/sms/whatsapp adapters expect), bare numeric ids, - Matrix ``!room``/``@user`` and XMPP JIDs. Anything else is not explicit - and goes to channel-directory resolution. + Platform parser first, then the shared rules in this order: E.164 phone + numbers (keeping the '+' the signal/sms/whatsapp adapters expect), bare + numeric ids, Matrix ``!room``/``@user``, XMPP JIDs (user@server or + room@conference.server). Anything else is not explicit and goes to + channel-directory resolution. """ parser = _PLATFORM_PARSERS.get(platform_name) if parser is not None: diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 102fd3beec..2d08eca521 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -107,8 +107,9 @@ def _handle_list(): def _handle_react(args, remove=False): """Attach (or with ``remove=True`` retract) an emoji reaction via a live - gateway adapter exposing ``add_reaction`` / ``remove_reaction``. No - standalone fallback: reacting needs the adapter's live message-id state. + gateway adapter exposing ``add_reaction(chat_id, emoji, message_id)`` / + ``remove_reaction(chat_id, message_id)`` (e.g. photon/iMessage tapbacks). + No standalone fallback: reacting needs the adapter's live message-id state. """ target = args.get("target", "") emoji = (args.get("emoji") or "").strip() @@ -199,82 +200,40 @@ def _handle_send(args): return tool_error("Interrupted") try: - from gateway.config import load_gateway_config, Platform + from gateway.config import load_gateway_config config = load_gateway_config() except Exception as e: return json.dumps(_error(f"Failed to load gateway config: {e}")) - from gateway.platform_registry import platform_registry - - entry = platform_registry.get(platform_name) - is_builtin = platform_name in {member.value for member in Platform} - if not is_builtin and entry is None: - return tool_error( - f"Unknown or unregistered plugin platform: {platform_name}" - ) - try: - platform = Platform(platform_name) - except (ValueError, KeyError): - return tool_error(f"Unknown platform: {platform_name}") - - pconfig = config.platforms.get(platform) - if not pconfig or not pconfig.enabled: - # Weixin can be configured purely via .env; synthesize a pconfig so - # send_message and cron delivery work without a gateway.yaml entry. - pconfig = _weixin_env_pconfig() if platform_name == "weixin" else None - if pconfig is None: - return tool_error(f"Platform '{platform_name}' is not configured. Set up credentials in ~/.hermes/config.yaml or environment variables.") + platform, pconfig, entry, err = _resolve_platform_config(platform_name, config) + if err: + return tool_error(err) from gateway.platforms.base import BasePlatformAdapter # Capture [[as_document]] before extract_media strips it: image files then - # go through send_document so the original bytes survive recompression. + # go through send_document so the original bytes survive (Telegram's + # sendPhoto recompresses to 1280px). force_document_attachments = "[[as_document]]" in message media_files, cleaned_message = BasePlatformAdapter.extract_media(message) media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files) mirror_text = cleaned_message.strip() or _describe_media_for_mirror(media_files) - used_home_channel = False - if not chat_id: - home = config.get_home_channel(platform) - if not home and platform_name == "weixin": - wx_home = os.getenv("WEIXIN_HOME_CHANNEL", "").strip() - if wx_home: - from gateway.config import HomeChannel - home = HomeChannel(platform=platform, chat_id=wx_home, name="Weixin Home") - if home: - chat_id = home.chat_id - used_home_channel = True - else: - home_env = _HOME_CHANNEL_ENV_OVERRIDES.get( - platform_name, f"{platform_name.upper()}_HOME_CHANNEL" - ) - return tool_error( - f"No home channel set for {platform_name} to determine where to send the message. " - f"Either specify a channel directly with '{platform_name}:CHANNEL_NAME', " - f"or set a home channel via: hermes config set {home_env} " - ) + used_home_channel = not chat_id + if used_home_channel: + chat_id, err = _home_chat_id(config, platform, platform_name) + if err: + return tool_error(err) duplicate_skip = _maybe_skip_cron_duplicate_send(platform_name, chat_id, thread_id) if duplicate_skip: return json.dumps(duplicate_skip) - # Slack: ``user:U...`` / ``user_name:@handle`` targets (and bare U... ids - # from session metadata / home-channel config) must be opened as DM - # conversations first — chat.postMessage needs a conversation ID. if platform_name == "slack" and chat_id: - _slack_dm_target = chat_id - if _slack_dm_target.startswith("U") and _SLACK_USER_ID_RE.fullmatch(_slack_dm_target): - _slack_dm_target = f"user:{_slack_dm_target}" - if _slack_dm_target.startswith(("user:", "user_name:")): - from model_tools import _run_async - _resolved, _resolve_err = _run_async( - _resolve_slack_user_target(pconfig.token, _slack_dm_target) - ) - if _resolve_err: - return json.dumps(_resolve_err) - chat_id = _resolved + chat_id, resolve_err = _slack_dm_chat_id(pconfig, chat_id) + if resolve_err: + return json.dumps(resolve_err) try: from model_tools import _run_async @@ -283,33 +242,18 @@ def _handle_send(args): "media_files": media_files, "force_document": force_document_attachments, } - # Only custom plugin handlers receive the complete typed request. + # Only custom plugin handlers receive the complete typed request; the + # built-in call contract stays exact. if entry is not None and entry.send_message_handler is not None: send_kwargs["args"] = args result = _run_async( _send_to_platform(platform, pconfig, chat_id, cleaned_message, **send_kwargs) ) - if used_home_channel and isinstance(result, dict) and result.get("success"): - result["note"] = f"Sent to {platform_name} home channel (chat_id: {chat_id})" - - # Mirror the sent message into the target's gateway session - if isinstance(result, dict) and result.get("success") and mirror_text: - try: - from gateway.mirror import mirror_to_session - from gateway.session_context import get_session_env - source_label = get_session_env("HERMES_SESSION_PLATFORM", "cli") - user_id = get_session_env("HERMES_SESSION_USER_ID", "") or None - if mirror_to_session( - platform_name, - chat_id, - mirror_text, - source_label=source_label, - thread_id=thread_id, - user_id=user_id, - ): - result["mirrored"] = True - except Exception: - pass + if isinstance(result, dict) and result.get("success"): + if used_home_channel: + result["note"] = f"Sent to {platform_name} home channel (chat_id: {chat_id})" + if mirror_text and _mirror_sent_message(platform_name, chat_id, mirror_text, thread_id): + result["mirrored"] = True if isinstance(result, dict) and "error" in result: result["error"] = _sanitize_error_text(result["error"]) @@ -318,6 +262,88 @@ def _handle_send(args): return json.dumps(_error(f"Send failed: {e}")) +def _resolve_platform_config(platform_name, config): + """Return ``(platform, pconfig, registry_entry, error)`` for a send. + + Plugin platforms must be registered; disabled/missing platforms error, + except Weixin, which may be configured purely via .env (synthesized + pconfig so send_message and cron delivery work without a gateway.yaml entry). + """ + from gateway.config import Platform + from gateway.platform_registry import platform_registry + + entry = platform_registry.get(platform_name) + is_builtin = platform_name in {member.value for member in Platform} + if not is_builtin and entry is None: + return None, None, None, f"Unknown or unregistered plugin platform: {platform_name}" + try: + platform = Platform(platform_name) + except (ValueError, KeyError): + return None, None, None, f"Unknown platform: {platform_name}" + + pconfig = config.platforms.get(platform) + if not pconfig or not pconfig.enabled: + pconfig = _weixin_env_pconfig() if platform_name == "weixin" else None + if pconfig is None: + return None, None, None, ( + f"Platform '{platform_name}' is not configured. Set up credentials in " + "~/.hermes/config.yaml or environment variables." + ) + return platform, pconfig, entry, None + + +def _home_chat_id(config, platform, platform_name): + """Return ``(home chat_id, None)`` or ``(None, actionable error)``. + Weixin additionally honours the WEIXIN_HOME_CHANNEL env var.""" + home = config.get_home_channel(platform) + if not home and platform_name == "weixin": + wx_home = os.getenv("WEIXIN_HOME_CHANNEL", "").strip() + if wx_home: + from gateway.config import HomeChannel + home = HomeChannel(platform=platform, chat_id=wx_home, name="Weixin Home") + if home: + return home.chat_id, None + home_env = _HOME_CHANNEL_ENV_OVERRIDES.get( + platform_name, f"{platform_name.upper()}_HOME_CHANNEL" + ) + return None, ( + f"No home channel set for {platform_name} to determine where to send the message. " + f"Either specify a channel directly with '{platform_name}:CHANNEL_NAME', " + f"or set a home channel via: hermes config set {home_env} " + ) + + +def _slack_dm_chat_id(pconfig, chat_id): + """Open Slack user targets as DM conversations: ``user:U...`` / + ``user_name:@handle`` from the parser, or a bare U... id from session + metadata / home-channel config. chat.postMessage needs a conversation ID. + Returns ``(chat_id, None)`` or ``(None, error_dict)``.""" + dm_target = chat_id + if dm_target.startswith("U") and _SLACK_USER_ID_RE.fullmatch(dm_target): + dm_target = f"user:{dm_target}" + if not dm_target.startswith(("user:", "user_name:")): + return chat_id, None + from model_tools import _run_async + return _run_async(_resolve_slack_user_target(pconfig.token, dm_target)) + + +def _mirror_sent_message(platform_name, chat_id, mirror_text, thread_id): + """Best-effort mirror of the sent message into the target's gateway session.""" + try: + from gateway.mirror import mirror_to_session + from gateway.session_context import get_session_env + return bool(mirror_to_session( + platform_name, + chat_id, + mirror_text, + source_label=get_session_env("HERMES_SESSION_PLATFORM", "cli"), + thread_id=thread_id, + user_id=get_session_env("HERMES_SESSION_USER_ID", "") or None, + )) + except Exception: + return False + + def _weixin_env_pconfig(): """Synthesize a Weixin PlatformConfig from .env secrets, or None.""" wx_token = get_secret("WEIXIN_TOKEN", "").strip() @@ -408,7 +434,9 @@ async def _send_live_adapter_media( metadata=None, force_document=False, ): - """Deliver text and every media descriptor through adapter media APIs.""" + """Deliver text and every media descriptor through adapter media APIs. + Adapters that only inherit the BasePlatformAdapter stub for a media kind + are treated as unsupported rather than silently no-op'd.""" caption, separate_text = _media_caption_split( message, media_files, max_caption_len=_DEFAULT_CAPTION_LIMIT ) @@ -495,9 +523,11 @@ async def _send_via_adapter( media_files=None, force_document=False, ): - """Send via the live in-process gateway adapter, else the plugin's - ``standalone_sender_fn`` (out-of-process callers such as cron), else a - descriptive error naming both options. + """Send via the live in-process gateway adapter (``_gateway_runner_ref``), + else the plugin's ``standalone_sender_fn`` (gateway not in this process, + e.g. cron — the runner weakref is None), else a descriptive error naming + both options. Media descriptors go through the adapter's native media + APIs under the same cross-loop rules as text. """ platform_name = platform.value if hasattr(platform, "value") else str(platform) runner = _live_runner() @@ -527,7 +557,7 @@ async def _send_via_adapter( async def _dispatch(make_coro, log_message): if not cross_loop: - return await make_coro() + return await make_coro() # same loop / no gateway loop (CLI, tests) if not gateway_loop.is_running(): return {"error": "Gateway loop is not running; cannot dispatch adapter send"} from agent.async_utils import safe_schedule_threadsafe @@ -622,8 +652,9 @@ async def _send_chunks(chunks, send_one): def _platform_max_length(platform): """Max message length for chunking: the adapter constant for Signal (its - raw JSON-RPC path never sees the adapter's own chunking), the registry's - ``max_message_length`` for plugins (Slack, Feishu, ...), else None.""" + raw JSON-RPC path never sees SignalAdapter's own chunking, so signal-cli + would reject long sends whole), the registry's ``max_message_length`` for + plugins (Slack, Feishu, ...), else None (no chunking).""" from gateway.config import Platform if platform == Platform.SIGNAL: @@ -643,7 +674,11 @@ def _platform_max_length(platform): # Plugin platforms whose media (and, for Discord, all) sends go straight to the -# registry ``standalone_sender_fn`` — bypassing the live adapter on purpose. +# registry ``standalone_sender_fn`` — bypassing the live adapter on purpose: +# Discord's plugin ``_standalone_send`` handles forum channels, threads and +# multipart uploads (historically the only Discord path); Slack uploads via +# files_upload_v2; WhatsApp posts each file to the Baileys bridge /send-media so +# images/videos/audio arrive as native bubbles, not documents. # platform -> (error label, run discover_plugins first, caption-capable, # media_files sentinel for non-final chunks, forward force_document) _PLUGIN_STANDALONE_MEDIA = { @@ -683,9 +718,35 @@ async def _send_plugin_standalone( ) -# Text-only senders for built-in platforms (generic, non-media path). -# Signature: (pconfig, chat_id, chunk, thread_id) -> awaitable result dict. +# Native-media chunked routes for built-in platforms; media rides on the final +# chunk, non-final chunks get the platform's empty-media sentinel. +# platform -> (media required, empty-media sentinel, +# sender(platform, pconfig, chat_id, chunk, media, thread_id, force_document)) +# Matrix: ALL sends go through the native adapter so text is encrypted in E2EE +# rooms too (the raw-HTTP standalone path is not encryption-aware). +# Signal: attachments ride the JSON-RPC ``attachments`` param. +# Yuanbao: media needs the running gateway adapter's WebSocket. +# Slack (text; media was intercepted above): prefer the live adapter — it is +# multi-workspace aware and honors adapter gates like ignored_channels, while +# the standalone Web-API path may only have a token list — else the plugin's +# standalone sender, keeping MEDIA delivery on the cron fallback. +# WeCom: native media only through the live gateway adapter. # Names resolve at call time so tests can monkeypatch e.g. ``_send_signal``. +_CHUNKED_ROUTES = { + "matrix": (False, [], lambda p, pc, cid, chunk, media, tid, fd: _send_matrix_via_adapter( + pc, cid, chunk, media_files=media, thread_id=tid)), + "signal": (True, [], lambda p, pc, cid, chunk, media, tid, fd: _send_signal( + pc.extra, cid, chunk, media_files=media)), + "yuanbao": (True, None, lambda p, pc, cid, chunk, media, tid, fd: _send_yuanbao( + cid, chunk, media_files=media)), + "slack": (False, [], lambda p, pc, cid, chunk, media, tid, fd: _send_via_adapter( + p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd)), + "wecom": (True, None, lambda p, pc, cid, chunk, media, tid, fd: _send_via_adapter( + p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd)), +} + +# Text-only senders for built-in platforms (generic path; media is dropped +# with a warning). Signature: (pconfig, chat_id, chunk, thread_id) -> result. _TEXT_SENDERS = { "whatsapp": lambda pc, cid, chunk, tid: _registry_standalone_send("whatsapp", pc, cid, chunk, tid), "signal": lambda pc, cid, chunk, tid: _send_signal(pc.extra, cid, chunk), @@ -706,9 +767,11 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, """Route a message to the appropriate platform sender. Long messages are chunked with the adapters' smart splitter (code-block - aware, part indicators). Branch order matters: Weixin first (avoids - unrelated optional imports), then native-media platforms, then the - generic text path that drops media with a warning. + aware, part indicators). Branch order matters: Weixin first (its native + helper must not be blocked by unrelated optional imports such as + lark-oapi's heavy Feishu path), Telegram (chunks itself), plugin + standalone media routes, native chunked routes, then the generic text + path that drops media with a warning. """ from gateway.config import Platform @@ -718,8 +781,9 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, if platform == Platform.WEIXIN: return await _send_weixin(pconfig, chat_id, message, media_files=media_files) - # Telegram chunks internally on the *formatted* text (MarkdownV2/HTML - # escaping inflates length), so it gets the whole message; media follows. + # Telegram chunks internally on the *formatted* text in UTF-16 units + # (MarkdownV2/HTML escaping inflates length), so it gets the whole + # message; media attaches after all text chunks. if platform == Platform.TELEGRAM: disable_link_previews = bool(getattr(pconfig, "extra", {}) and pconfig.extra.get("disable_link_previews")) return await _send_telegram( @@ -737,55 +801,34 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, max_len = _platform_max_length(platform) chunks = BasePlatformAdapter.truncate_message(message, max_len) if max_len else [message] - # Media rides on the final chunk in every chunked branch below. - if platform == Platform.DISCORD or ( - media_files and platform_name in ("feishu", "slack", "whatsapp") - ): + if platform_name == "discord" or (media_files and platform_name in _PLUGIN_STANDALONE_MEDIA): return await _send_plugin_standalone( platform_name, pconfig, chat_id, message, chunks, media_files, thread_id=thread_id, max_len=max_len, force_document=force_document, ) - # Matrix: every send goes through the native adapter so text is encrypted - # in E2EE rooms too (the raw-HTTP standalone path is not encryption-aware). - if platform == Platform.MATRIX: - return await _send_chunks(chunks, lambda chunk, is_last: _send_matrix_via_adapter( - pconfig, chat_id, chunk, media_files=media_files if is_last else [], thread_id=thread_id, - )) - - if platform == Platform.SIGNAL and media_files: - return await _send_chunks(chunks, lambda chunk, is_last: _send_signal( - pconfig.extra, chat_id, chunk, media_files=media_files if is_last else [], - )) - - if platform == Platform.YUANBAO and media_files: - return await _send_chunks(chunks, lambda chunk, is_last: _send_yuanbao( - chat_id, chunk, media_files=media_files if is_last else None, - )) - - # Slack text: prefer the live adapter (multi-workspace aware, honors - # adapter-side gates) and fall back to the plugin's standalone sender. - # WeCom media: native delivery through the live gateway adapter. - if platform == Platform.SLACK or (platform == Platform.WECOM and media_files): - empty_media = [] if platform == Platform.SLACK else None - return await _send_chunks(chunks, lambda chunk, is_last: _send_via_adapter( - platform, pconfig, chat_id, chunk, thread_id=thread_id, - media_files=media_files if is_last else empty_media, force_document=force_document, + route = _CHUNKED_ROUTES.get(platform_name) + if route is not None and (media_files or not route[0]): + _, empty_media, sender = route + return await _send_chunks(chunks, lambda chunk, is_last: sender( + platform, pconfig, chat_id, chunk, + media_files if is_last else empty_media, thread_id, force_document, )) # --- Generic path: text only. Buzz is a plugin platform with verified - # native media delivery through _send_via_adapter, so it is exempt. - if media_files and not message.strip() and platform.value != "buzz": + # native media delivery through _send_via_adapter (media-only sends + # included), so it is exempt from the media error/warning. + if media_files and platform_name != "buzz" and not message.strip(): return { "error": ( f"send_message MEDIA delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}; " - f"target {platform.value} had only media attachments" + f"target {platform_name} had only media attachments" ) } warning = None - if media_files and platform.value != "buzz": + if media_files and platform_name != "buzz": warning = ( - f"MEDIA attachments were omitted for {platform.value}; " + f"MEDIA attachments were omitted for {platform_name}; " f"native send_message media delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}" ) @@ -798,7 +841,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, entry = platform_registry.get(platform_name) handler = entry.send_message_handler if entry is not None else None if handler is not None: - # Custom handler receives the full request once (not per chunk). + # Custom handler receives the full typed request once (not per chunk). try: import inspect diff --git a/tools/thread_context.py b/tools/thread_context.py index 8d9a272290..2a53589212 100644 --- a/tools/thread_context.py +++ b/tools/thread_context.py @@ -1,34 +1,29 @@ #!/usr/bin/env python3 """Propagate agent-turn context into worker threads that dispatch Hermes tools. -A bare ``threading.Thread`` / ``ThreadPoolExecutor`` worker starts with an -empty ``contextvars.Context`` and no thread-local approval/sudo callbacks. -Tool dispatch inside such a thread therefore silently loses: +A bare ``threading.Thread`` / ``ThreadPoolExecutor`` worker starts with an empty +``contextvars.Context`` and no thread-local approval/sudo callbacks, so tool +dispatch inside it silently loses: - * the approval *session/platform* ContextVars (``tools.approval`` / - ``gateway.session_context``) — so gateway sessions fall into + * the approval session/platform ContextVars (``tools.approval`` / + ``gateway.session_context``) — gateway sessions then fall into ``check_dangerous_command``'s non-interactive auto-approve branch and - dangerous commands run without prompting (#33057, #30882); + dangerous commands run without prompting; * the thread-local CLI approval/sudo callbacks (``tools.terminal_tool``) — - so ``prompt_dangerous_approval`` cannot reach the user - (GHSA-qg5c-hvr5-hjgr, #15216). + ``prompt_dangerous_approval`` then cannot reach the user (GHSA-qg5c-hvr5-hjgr). -This helper factors out that capture/install/clear lifecycle so the several -places that fan tool dispatch onto worker threads (``agent.tool_executor`` and -the ``execute_code`` RPC threads) share one audited implementation instead of -divergent copies. +One audited capture/install/clear lifecycle for every place that fans tool +dispatch onto worker threads (``agent.tool_executor``, ``execute_code`` RPC threads). -Usage — call :func:`propagate_context_to_thread` **on the parent thread** -(it snapshots the parent's ContextVars and callbacks at call time) and use the -returned callable as the worker's target:: +Call :func:`propagate_context_to_thread` **on the parent thread** (it snapshots +the parent's ContextVars and callbacks at call time) and use the result as the +worker target:: t = threading.Thread(target=propagate_context_to_thread(loop_fn), args=(...)) - # or executor.submit(propagate_context_to_thread(worker_fn), *args) -Approval/sudo callbacks are installed for the worker's lifetime and **always -cleared on exit**, so a recycled thread never holds a stale reference to a -disposed CLI instance. +Callbacks are installed for the worker's lifetime and **always cleared on exit**, +so a recycled thread never holds a stale reference to a disposed CLI instance. """ from __future__ import annotations @@ -43,9 +38,8 @@ logger = logging.getLogger(__name__) def _callback_api(): """Resolve the terminal_tool callback getters/setters. - Imported lazily: ``tools.terminal_tool`` imports ``tools.approval`` at - module load, so a top-level import here would risk an import cycle for - callers that live in ``tools.approval``. + Imported lazily: ``tools.terminal_tool`` imports ``tools.approval`` at module + load, so a top-level import would risk a cycle for callers in ``tools.approval``. """ from tools.terminal_tool import ( _get_approval_callback, @@ -62,18 +56,13 @@ def _callback_api(): def propagate_context_to_thread(target: Callable) -> Callable: - """Wrap *target* for execution on a worker thread with the *current* - thread's ContextVars and approval/sudo callbacks propagated. + """Wrap *target* so it runs on a worker thread with the *current* thread's + ContextVars and approval/sudo callbacks. The wrapper forwards args/kwargs. - Call this on the parent thread; pass the returned callable as the - thread/executor target. The returned callable forwards its positional - and keyword arguments to *target* and returns its result. - - Fail-closed: if callback installation raises, the callbacks are left - unset (``None``). That is the safe outcome — ``prompt_dangerous_approval`` - denies dangerous commands when no callback is registered in an interactive - context, and the gateway approval queue blocks when its notify callback is - absent. + Fail-closed: if callback installation raises, the callbacks stay unset + (``None``) — ``prompt_dangerous_approval`` denies dangerous commands with no + callback in an interactive context, and the gateway approval queue blocks + when its notify callback is absent. """ ctx = contextvars.copy_context() parent_approval_cb = parent_sudo_cb = None @@ -88,32 +77,31 @@ def propagate_context_to_thread(target: Callable) -> Callable: def _runner(*args, **kwargs): def _inner(): - if setters is not None: - set_approval, set_sudo = setters - try: - if parent_approval_cb is not None: - set_approval(parent_approval_cb) - if parent_sudo_cb is not None: - set_sudo(parent_sudo_cb) - except Exception: - logger.debug( - "Failed to install propagated approval/sudo callbacks; " - "dangerous-command approval will fail closed", - exc_info=True, - ) + if setters is None: + return target(*args, **kwargs) + set_approval, set_sudo = setters + try: + if parent_approval_cb is not None: + set_approval(parent_approval_cb) + if parent_sudo_cb is not None: + set_sudo(parent_sudo_cb) + except Exception: + logger.debug( + "Failed to install propagated approval/sudo callbacks; " + "dangerous-command approval will fail closed", + exc_info=True, + ) try: return target(*args, **kwargs) finally: - if setters is not None: - set_approval, set_sudo = setters - try: - set_approval(None) - set_sudo(None) - except Exception: - logger.debug( - "Failed to clear propagated approval/sudo callbacks", - exc_info=True, - ) + try: + set_approval(None) + set_sudo(None) + except Exception: + logger.debug( + "Failed to clear propagated approval/sudo callbacks", + exc_info=True, + ) return ctx.run(_inner) diff --git a/tools/yuanbao_tools.py b/tools/yuanbao_tools.py index 76e3bc51cb..155fa7d430 100644 --- a/tools/yuanbao_tools.py +++ b/tools/yuanbao_tools.py @@ -43,6 +43,42 @@ def _nick(m: dict) -> str: return m.get("nickname", m.get("nick_name", "")) +async def _resolve_dm_recipient(adapter, group_code: str, name: str) -> Tuple[str, str, Optional[dict]]: + """Resolve ``name`` to (user_id, nickname) via the group member list. + + Returns ``("", "", error_dict)`` on failure; >1 partial match yields candidates + for disambiguation instead of guessing. + """ + if not group_code: + return "", "", _err("group_code is required when user_id is not provided") + if not name: + return "", "", _err("name is required when user_id is not provided") + try: + raw = await adapter.get_group_member_list(group_code) + if raw is None: + return "", "", _err("get_group_member_list returned None") + filt = name.strip().lower() + matched = [ + m for m in raw.get("members", []) + if filt in (m.get("nickname") or m.get("nick_name") or "").lower() + ] + if not matched: + return "", "", _err(f'No member matching "{name}" found in group {group_code}.') + if len(matched) > 1: + return "", "", { + "success": False, + "error": f'Multiple members match "{name}". Please specify which one.', + "candidates": [ + {"user_id": m.get("user_id", ""), "nickname": _nick(m)} for m in matched + ], + } + m = matched[0] + return m.get("user_id", ""), m.get("nickname", m.get("nick_name", name)), None + except Exception as exc: + logger.exception("[yuanbao_tools] send_dm member lookup error") + return "", "", _err(str(exc)) + + async def get_group_info(group_code: str) -> dict: """查询群基本信息(群名、群主、成员数)。""" if not group_code: @@ -164,6 +200,7 @@ async def send_sticker( """向 chat_id(缺省取当前会话 HERMES_SESSION_CHAT_ID)发送一张内置贴纸(TIMFaceElem)。 ``sticker``: 名称(如 "六六六")或 sticker_id(如 "278");为空时随机发送。 + ``chat_id``: ``direct:{account_id}`` / ``group:{group_code}`` / 裸 account_id。 """ from gateway.session_context import get_session_env from gateway.platforms.yuanbao_sticker import ( @@ -242,34 +279,9 @@ async def send_dm( resolved_nickname = name.strip() if not resolved_user_id: - if not group_code: - return _err("group_code is required when user_id is not provided") - if not name: - return _err("name is required when user_id is not provided") - try: - raw = await adapter.get_group_member_list(group_code) - if raw is None: - return _err("get_group_member_list returned None") - filt = name.strip().lower() - matched = [ - m for m in raw.get("members", []) - if filt in (m.get("nickname") or m.get("nick_name") or "").lower() - ] - if not matched: - return _err(f'No member matching "{name}" found in group {group_code}.') - if len(matched) > 1: - return { - "success": False, - "error": f'Multiple members match "{name}". Please specify which one.', - "candidates": [ - {"user_id": m.get("user_id", ""), "nickname": _nick(m)} for m in matched - ], - } - resolved_user_id = matched[0].get("user_id", "") - resolved_nickname = matched[0].get("nickname", matched[0].get("nick_name", name)) - except Exception as exc: - logger.exception("[yuanbao_tools] send_dm member lookup error") - return _err(str(exc)) + resolved_user_id, resolved_nickname, err = await _resolve_dm_recipient(adapter, group_code, name) + if err is not None: + return err if not resolved_user_id: return _err("Could not resolve user_id") @@ -385,241 +397,227 @@ async def _handle_yb_send_sticker(args, **kw): )) -_TOOLSET = "hermes-yuanbao" - -registry.register( - name="yb_query_group_info", - toolset=_TOOLSET, - schema={ - "name": "yb_query_group_info", - "description": ( - "Query basic info about a group (called '派/Pai' in the app), " - "including group name, owner, and member count." - ), - "parameters": { - "type": "object", - "properties": { - "group_code": { - "type": "string", - "description": "The unique group identifier (group_code).", - }, - }, - "required": ["group_code"], - }, - }, - handler=_handle_yb_query_group_info, - check_fn=_check_yuanbao, - is_async=True, - emoji="👥", -) - -registry.register( - name="yb_query_group_members", - toolset=_TOOLSET, - schema={ - "name": "yb_query_group_members", - "description": ( - "Query members of a group (called '派/Pai' in the app). " - "Use this tool when you need to @mention someone, find a user by name, " - "list bots (including Yuanbao AI), or list all members. " - "IMPORTANT: You MUST call this tool before @mentioning any user, " - "because you need the exact nickname to construct the @mention format." - ), - "parameters": { - "type": "object", - "properties": { - "group_code": { - "type": "string", - "description": "The unique group identifier (group_code).", - }, - "action": { - "type": "string", - "enum": ["find", "list_bots", "list_all"], - "description": ( - "find — search a user by name (use when you need to @mention or look up someone); " - "list_bots — list bots and Yuanbao AI assistants; " - "list_all — list all members." - ), - }, - "name": { - "type": "string", - "description": ( - "User name to search (partial match, case-insensitive). " - "Required for 'find'. Use the name the user mentioned in the conversation." - ), - }, - "mention": { - "type": "boolean", - "description": ( - "Set to true when you need to @mention/at someone in your reply. " - "The response will include the exact @mention format to use." - ), - }, - }, - "required": ["group_code", "action"], - }, - }, - handler=_handle_yb_query_group_members, - check_fn=_check_yuanbao, - is_async=True, - emoji="📋", -) - -registry.register( - name="yb_send_dm", - toolset=_TOOLSET, - schema={ - "name": "yb_send_dm", - "description": ( - "Send a private/direct message (DM) to a user in a group, with optional media files. " - "This tool automatically looks up the user by name in the group member list " - "and sends the message. Use this when someone asks to privately message / 私信 / DM a user. " - "Supports text, images, and file attachments. " - "You can also provide user_id directly if already known." - ), - "parameters": { - "type": "object", - "properties": { - "group_code": { - "type": "string", - "description": ( - "The group where the target user belongs. " - "Extract from chat_id: 'group:328306697' → '328306697'. " - "Required when user_id is not provided." - ), - }, - "name": { - "type": "string", - "description": ( - "Target user's display name (partial match, case-insensitive). " - "Required when user_id is not provided." - ), - }, - "message": { - "type": "string", - "description": "The message text to send as a DM. Can be empty if only sending media.", - }, - "user_id": { - "type": "string", - "description": ( - "Target user's account ID. If provided, skips the member lookup. " - "Usually obtained from a previous yb_query_group_members call." - ), - }, - "media_files": { - "type": "array", - "description": ( - "Optional list of media files to send along with the DM. " - "Images (.jpg/.png/.gif/.webp/.bmp) are sent as image messages; " - "other files are sent as document attachments." - ), - "items": { - "type": "object", - "properties": { - "path": { - "type": "string", - "description": "Absolute local file path of the media to send.", - }, - "is_voice": { - "type": "boolean", - "description": "Whether this file is a voice message (default false).", - }, - }, - "required": ["path"], +# (schema, handler, emoji); the tool name is schema["name"]. +_TOOLS = ( + ( + { + "name": "yb_query_group_info", + "description": ( + "Query basic info about a group (called '派/Pai' in the app), " + "including group name, owner, and member count." + ), + "parameters": { + "type": "object", + "properties": { + "group_code": { + "type": "string", + "description": "The unique group identifier (group_code).", }, }, + "required": ["group_code"], }, - "required": [], }, - }, - handler=_handle_yb_send_dm, - check_fn=_check_yuanbao, - is_async=True, - emoji="✉️", + _handle_yb_query_group_info, + "👥", + ), + ( + { + "name": "yb_query_group_members", + "description": ( + "Query members of a group (called '派/Pai' in the app). " + "Use this tool when you need to @mention someone, find a user by name, " + "list bots (including Yuanbao AI), or list all members. " + "IMPORTANT: You MUST call this tool before @mentioning any user, " + "because you need the exact nickname to construct the @mention format." + ), + "parameters": { + "type": "object", + "properties": { + "group_code": { + "type": "string", + "description": "The unique group identifier (group_code).", + }, + "action": { + "type": "string", + "enum": ["find", "list_bots", "list_all"], + "description": ( + "find — search a user by name (use when you need to @mention or look up someone); " + "list_bots — list bots and Yuanbao AI assistants; " + "list_all — list all members." + ), + }, + "name": { + "type": "string", + "description": ( + "User name to search (partial match, case-insensitive). " + "Required for 'find'. Use the name the user mentioned in the conversation." + ), + }, + "mention": { + "type": "boolean", + "description": ( + "Set to true when you need to @mention/at someone in your reply. " + "The response will include the exact @mention format to use." + ), + }, + }, + "required": ["group_code", "action"], + }, + }, + _handle_yb_query_group_members, + "📋", + ), + ( + { + "name": "yb_send_dm", + "description": ( + "Send a private/direct message (DM) to a user in a group, with optional media files. " + "This tool automatically looks up the user by name in the group member list " + "and sends the message. Use this when someone asks to privately message / 私信 / DM a user. " + "Supports text, images, and file attachments. " + "You can also provide user_id directly if already known." + ), + "parameters": { + "type": "object", + "properties": { + "group_code": { + "type": "string", + "description": ( + "The group where the target user belongs. " + "Extract from chat_id: 'group:328306697' → '328306697'. " + "Required when user_id is not provided." + ), + }, + "name": { + "type": "string", + "description": ( + "Target user's display name (partial match, case-insensitive). " + "Required when user_id is not provided." + ), + }, + "message": { + "type": "string", + "description": "The message text to send as a DM. Can be empty if only sending media.", + }, + "user_id": { + "type": "string", + "description": ( + "Target user's account ID. If provided, skips the member lookup. " + "Usually obtained from a previous yb_query_group_members call." + ), + }, + "media_files": { + "type": "array", + "description": ( + "Optional list of media files to send along with the DM. " + "Images (.jpg/.png/.gif/.webp/.bmp) are sent as image messages; " + "other files are sent as document attachments." + ), + "items": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "Absolute local file path of the media to send.", + }, + "is_voice": { + "type": "boolean", + "description": "Whether this file is a voice message (default false).", + }, + }, + "required": ["path"], + }, + }, + }, + "required": [], + }, + }, + _handle_yb_send_dm, + "✉️", + ), + ( + { + "name": "yb_search_sticker", + "description": ( + "Search the built-in Yuanbao sticker (TIM face / 表情包) catalogue by keyword. " + "Returns the top matching candidates with sticker_id, name, and description. " + "Use this BEFORE yb_send_sticker to discover the right sticker_id. " + "Sticker = 贴纸 = TIM face — NOT a message reaction. " + "Prefer sending a sticker over bare Unicode emoji when reacting/expressing emotion." + ), + "parameters": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": ( + "Search keyword (Chinese or English, e.g. '666', '比心', 'cool', '吃瓜'). " + "Empty string returns the first N stickers." + ), + }, + "limit": { + "type": "integer", + "description": "Max number of candidates to return (default 10, max 50).", + }, + }, + "required": [], + }, + }, + _handle_yb_search_sticker, + "🔍", + ), + ( + { + "name": "yb_send_sticker", + "description": ( + "Send a built-in sticker (TIMFaceElem / 贴纸表情) to the current Yuanbao chat. " + "Call yb_search_sticker first if you don't know the sticker_id/name. " + "Sticker = 贴纸 = TIM face — NOT a message reaction. " + "CRITICAL: Whenever the user asks you to send a sticker / 贴纸 / 表情包, you MUST " + "use this tool. DO NOT draw a PNG via execute_code / Pillow / matplotlib and " + "then call send_image_file — that produces a fake 'sticker' image instead of a " + "real TIM face and is the WRONG path. If no suitable sticker_id is known, call " + "yb_search_sticker first. When the recent thread shows users sending stickers, " + "prefer matching that tone by replying with a sticker instead of (or in " + "addition to) text." + ), + "parameters": { + "type": "object", + "properties": { + "sticker": { + "type": "string", + "description": ( + "Sticker name (e.g. '六六六', '比心', 'ok') or numeric sticker_id " + "(e.g. '278'). Empty string sends a random built-in sticker." + ), + }, + "chat_id": { + "type": "string", + "description": ( + "Target chat. Defaults to the current session. " + "Format: 'direct:{account_id}', 'group:{group_code}', or bare account_id." + ), + }, + "reply_to": { + "type": "string", + "description": "Optional ref_msg_id to quote-reply (group chat only).", + }, + }, + "required": [], + }, + }, + _handle_yb_send_sticker, + "🎨", + ), ) - -registry.register( - name="yb_search_sticker", - toolset=_TOOLSET, - schema={ - "name": "yb_search_sticker", - "description": ( - "Search the built-in Yuanbao sticker (TIM face / 表情包) catalogue by keyword. " - "Returns the top matching candidates with sticker_id, name, and description. " - "Use this BEFORE yb_send_sticker to discover the right sticker_id. " - "Sticker = 贴纸 = TIM face — NOT a message reaction. " - "Prefer sending a sticker over bare Unicode emoji when reacting/expressing emotion." - ), - "parameters": { - "type": "object", - "properties": { - "query": { - "type": "string", - "description": ( - "Search keyword (Chinese or English, e.g. '666', '比心', 'cool', '吃瓜'). " - "Empty string returns the first N stickers." - ), - }, - "limit": { - "type": "integer", - "description": "Max number of candidates to return (default 10, max 50).", - }, - }, - "required": [], - }, - }, - handler=_handle_yb_search_sticker, - check_fn=_check_yuanbao, - is_async=True, - emoji="🔍", -) - - -registry.register( - name="yb_send_sticker", - toolset=_TOOLSET, - schema={ - "name": "yb_send_sticker", - "description": ( - "Send a built-in sticker (TIMFaceElem / 贴纸表情) to the current Yuanbao chat. " - "Call yb_search_sticker first if you don't know the sticker_id/name. " - "Sticker = 贴纸 = TIM face — NOT a message reaction. " - "CRITICAL: Whenever the user asks you to send a sticker / 贴纸 / 表情包, you MUST " - "use this tool. DO NOT draw a PNG via execute_code / Pillow / matplotlib and " - "then call send_image_file — that produces a fake 'sticker' image instead of a " - "real TIM face and is the WRONG path. If no suitable sticker_id is known, call " - "yb_search_sticker first. When the recent thread shows users sending stickers, " - "prefer matching that tone by replying with a sticker instead of (or in " - "addition to) text." - ), - "parameters": { - "type": "object", - "properties": { - "sticker": { - "type": "string", - "description": ( - "Sticker name (e.g. '六六六', '比心', 'ok') or numeric sticker_id " - "(e.g. '278'). Empty string sends a random built-in sticker." - ), - }, - "chat_id": { - "type": "string", - "description": ( - "Target chat. Defaults to the current session. " - "Format: 'direct:{account_id}', 'group:{group_code}', or bare account_id." - ), - }, - "reply_to": { - "type": "string", - "description": "Optional ref_msg_id to quote-reply (group chat only).", - }, - }, - "required": [], - }, - }, - handler=_handle_yb_send_sticker, - check_fn=_check_yuanbao, - is_async=True, - emoji="🎨", -) +for _schema, _handler, _emoji in _TOOLS: + registry.register( + name=_schema["name"], + toolset="hermes-yuanbao", + schema=_schema, + handler=_handler, + check_fn=_check_yuanbao, + is_async=True, + emoji=_emoji, + )