diff --git a/gateway/run_inbound.py b/gateway/run_inbound.py index 8b014b1d57..02043bd792 100644 --- a/gateway/run_inbound.py +++ b/gateway/run_inbound.py @@ -38,6 +38,88 @@ logger = logging.getLogger("gateway.run") class GatewayInboundMixin: """Inbound message pipeline (_handle_message, text/media preparation, durable-turn markers, plugin injection) for GatewayRunner.""" + def _hm_pre_gateway_dispatch_hook( + self, event: "MessageEvent", source: SessionSource + ) -> Optional["MessageEvent"]: + """Run the ``pre_gateway_dispatch`` plugin hook; None = drop, else the (maybe rewritten) event. + + Plugins may return ``{"action": "skip", "reason": ...}`` → drop; ``{"action": "rewrite", + "text": ...}`` → replace ``event.text``; ``{"action": "allow"}`` / None → normal dispatch. + Runs BEFORE auth so plugins can handle unauthorized senders without the pairing flow. + """ + try: + from hermes_cli.lifecycle import invoke_hook as _invoke_hook + _hook_results = _invoke_hook( + "pre_gateway_dispatch", + event=event, + gateway=self, + # getattr: bare-runner tests build GatewayRunner via object.__new__ without + # __init__; the hook must not fail dispatch over a missing attribute. + session_store=getattr(self, "session_store", None), + ) + except Exception as _hook_exc: + logger.warning("pre_gateway_dispatch invocation failed: %s", _hook_exc) + _hook_results = [] + + for _result in _hook_results: + if not isinstance(_result, dict): + continue + _action = _result.get("action") + if _action == "skip": + logger.info( + "pre_gateway_dispatch skip: reason=%s platform=%s chat=%s", + _result.get("reason"), + source.platform.value if source.platform else "unknown", + source.chat_id or "unknown", + ) + return None + if _action == "rewrite": + _new_text = _result.get("text") + if isinstance(_new_text, str): + event = dataclasses.replace(event, text=_new_text) + break + if _action == "allow": + break + return event + + async def _hm_offer_pairing_code(self, source: SessionSource) -> None: + """DM an unauthorized sender a pairing code (rate-limited; groups never reach here).""" + platform_name = source.platform.value if source.platform else "unknown" + pairing_store = self._pairing_store_for(source) + if pairing_store is None: + logger.error("Cannot offer pairing code on %s: no pairing store", platform_name) + return + # Rate-limit ALL pairing responses (code or rejection) so a burst of DMs doesn't spam. + if pairing_store._is_rate_limited(platform_name, source.user_id): + return + code = pairing_store.generate_code(platform_name, source.user_id, source.user_name or "") + adapter = self._adapter_for_source(source) + if code: + if adapter: + store_profile = getattr(pairing_store, "profile", None) + profile_arg = ( + f"-p {store_profile} " + if isinstance(store_profile, str) and store_profile and store_profile != "default" + else "" + ) + await adapter.send( + source.chat_id, + f"Hi~ I don't recognize you yet!\n\n" + f"Here's your pairing code: `{code}`\n\n" + f"Ask the bot owner to run:\n" + f"`hermes {profile_arg}pairing approve " + f"{platform_name} {code}`" + ) + else: + if adapter: + await adapter.send( + source.chat_id, + "Too many pairing requests right now~ " + "Please try again later!" + ) + # Record rate limit so subsequent messages are silently ignored + pairing_store._record_rate_limit(platform_name, source.user_id) + async def _hm_admit_event( self, event: "MessageEvent" ) -> Optional[Tuple["MessageEvent", SessionSource, bool]]: @@ -51,19 +133,17 @@ class GatewayInboundMixin: source = event.source # 🔴 Cross-session leak guard. This per-message task was created via create_task(), which - # copies the spawning context: if a concurrent message had already bound its session via - # set_session_vars(), we inherited ITS HERMES_SESSION_* ContextVars, and until _set_session_env - # binds ours any subprocess would read the foreign identity (the _UNSET-strip guard can't - # help — the vars are set-to-foreign). Reset to _UNSET so that window strips safe instead. + # copies the spawning context: a concurrent message may already have bound ITS + # HERMES_SESSION_* ContextVars, and until _set_session_env binds ours any subprocess would + # read the foreign identity. Reset to _UNSET so that window strips safe instead. try: from gateway.session_context import reset_session_vars reset_session_vars() except Exception: logger.debug("reset_session_vars failed at handler entry", exc_info=True) - # Most adapters resolve profile routes in build_source(), before they hand us the event. A - # few internal/voice paths construct SessionSource directly, so resolve those here as the - # shared fail-closed ingress gate before authorization, hooks, or session side effects. + # Most adapters resolve profile routes in build_source(); internal/voice paths construct + # SessionSource directly, so resolve those here as the shared fail-closed ingress gate. if ( getattr(getattr(self, "config", None), "multiplex_profiles", False) and not getattr(source, "profile", None) @@ -76,9 +156,8 @@ class GatewayInboundMixin: except ProfileRouteRejected: source.profile_route_rejected = True - # SessionSource owns a strict boolean marker. Require the literal value - # so duck-typed test/internal sources with dynamic attributes are not - # mistaken for an explicit matched-route rejection. + # SessionSource owns a strict boolean marker; require the literal value so duck-typed + # test/internal sources with dynamic attributes are not mistaken for a rejection. if getattr(source, "profile_route_rejected", False) is True: logger.warning( "Dropping inbound message because its explicit profile route " @@ -86,8 +165,7 @@ class GatewayInboundMixin: ) return None - # Internal events (e.g. background-process completion notifications) - # are system-generated and must skip user authorization. + # Internal events (e.g. background-process notifications) skip user authorization. is_internal = bool(getattr(event, "internal", False)) # Ignored-channel guard runs FIRST — before startup-restore queueing, plugin hooks, auth, @@ -114,58 +192,20 @@ class GatewayInboundMixin: self._queue_startup_restore_event(event) return None - # scale-to-zero: stamp the gateway-scoped last-inbound clock (read by is_idle) for real - # user-originated inbound only. Internal/system events are NOT traffic — counting them - # would keep a genuinely idle gateway awake. if not is_internal: + # scale-to-zero: only real user-originated inbound stamps the last-inbound clock; + # counting internal/system events would keep a genuinely idle gateway awake. self._scale_to_zero_note_real_inbound() - - # pre_gateway_dispatch plugin hook (user-originated only). Plugins may return - # {"action": "skip", "reason": ...} -> drop; {"action": "rewrite", "text": ...} -> replace - # event.text; {"action": "allow"} / None -> normal dispatch. - # Runs BEFORE auth so plugins can handle unauthorized senders without the pairing flow. - if not is_internal: - try: - from hermes_cli.lifecycle import invoke_hook as _invoke_hook - _hook_results = _invoke_hook( - "pre_gateway_dispatch", - event=event, - gateway=self, - # getattr: bare-runner tests build GatewayRunner via object.__new__ without - # __init__; the hook must not fail dispatch over a missing attribute. - session_store=getattr(self, "session_store", None), - ) - except Exception as _hook_exc: - logger.warning("pre_gateway_dispatch invocation failed: %s", _hook_exc) - _hook_results = [] - - for _result in _hook_results: - if not isinstance(_result, dict): - continue - _action = _result.get("action") - if _action == "skip": - logger.info( - "pre_gateway_dispatch skip: reason=%s platform=%s chat=%s", - _result.get("reason"), - source.platform.value if source.platform else "unknown", - source.chat_id or "unknown", - ) - return None - if _action == "rewrite": - _new_text = _result.get("text") - if isinstance(_new_text, str): - event = dataclasses.replace(event, text=_new_text) - source = event.source - break - if _action == "allow": - break + event = self._hm_pre_gateway_dispatch_hook(event, source) + if event is None: + return None + source = event.source if is_internal: pass elif source.user_id is None: - # Messages with no user identity (Telegram service messages, channel forwards, anonymous - # admin posts, sender_chat) can't be paired but may be authorized via a chat-scoped - # allowlist (e.g. TELEGRAM_GROUP_ALLOWED_CHATS), so defer to _is_user_authorized. + # No user identity (Telegram service messages, channel forwards, anonymous admin posts, + # sender_chat): can't be paired but may be authorized via a chat-scoped allowlist. if not self._is_user_authorized_for_source(source): logger.debug("Ignoring message with no user_id from %s", source.platform.value) return None @@ -174,197 +214,145 @@ class GatewayInboundMixin: # In DMs: offer pairing code. In groups: silently ignore. if ( source.chat_type == "dm" - and self._get_unauthorized_dm_behavior( - source.platform, - profile=source.profile, - ) - == "pair" + and self._get_unauthorized_dm_behavior(source.platform, profile=source.profile) == "pair" ): - platform_name = source.platform.value if source.platform else "unknown" - pairing_store = self._pairing_store_for(source) - if pairing_store is None: - logger.error( - "Cannot offer pairing code on %s: no pairing store", - platform_name, - ) - return None - # Rate-limit ALL pairing responses (code or rejection) so a burst of DMs doesn't - # spam the user with repeated messages. - if pairing_store._is_rate_limited(platform_name, source.user_id): - return None - code = pairing_store.generate_code( - platform_name, source.user_id, source.user_name or "" - ) - if code: - adapter = self._adapter_for_source(source) - if adapter: - store_profile = getattr(pairing_store, "profile", None) - profile_arg = ( - f"-p {store_profile} " - if isinstance(store_profile, str) - and store_profile - and store_profile != "default" - else "" - ) - await adapter.send( - source.chat_id, - f"Hi~ I don't recognize you yet!\n\n" - f"Here's your pairing code: `{code}`\n\n" - f"Ask the bot owner to run:\n" - f"`hermes {profile_arg}pairing approve " - f"{platform_name} {code}`" - ) - else: - adapter = self._adapter_for_source(source) - if adapter: - await adapter.send( - source.chat_id, - "Too many pairing requests right now~ " - "Please try again later!" - ) - # Record rate limit so subsequent messages are silently ignored - pairing_store._record_rate_limit(platform_name, source.user_id) + await self._hm_offer_pairing_code(source) return None return event, source, is_internal + def _hm_estop_turn_allowed(self, event: "MessageEvent", source: SessionSource) -> bool: + """Whether a turn may bypass the global emergency stop. + + Pause blocks NEW agent turns, never running work or control traffic: recognized slash + commands (/status, /approve, ... and /pause off as the in-band resume path) and replies + owned by in-flight work — pending update prompt, steering a running session, pending + slash-confirm, dangerous-command approval — all pass through. + """ + try: + _estop_cmd = event.get_command() + except Exception: + _estop_cmd = None + if _estop_cmd: + try: + from hermes_cli.commands import resolve_command as _resolve_estop_cmd + if _resolve_estop_cmd(_estop_cmd) is not None: + return True + except Exception: + pass + try: + _estop_key = self._session_key_for_source(source) + _estop_state = self._peek_session_state(_estop_key) + if _estop_state is not None and _estop_state.persistent.update_prompt_pending: + return True + if self._is_session_running(_estop_key): + # Steering / interrupting in-flight work (also covers pending clarify + tool + # approvals held by the running agent). + return True + from tools import slash_confirm as _estop_confirm_mod + if _estop_confirm_mod.get_pending(_estop_key): + return True + from tools.approval import has_blocking_approval as _estop_has_approval + if _estop_has_approval(_estop_key): + return True + except Exception: + pass + return False + def _hm_estop_gate( self, event: "MessageEvent", source: SessionSource, is_internal: bool ) -> Optional[str]: - """Return the global emergency-stop notice when this turn must be blocked, else None.""" - # Global emergency stop (`hermes pause`): new turns get a brief paused notice instead of an - # agent run. Placed after auth so unauthorized senders can't probe pause state. Pause blocks - # NEW agent turns, never running work or control traffic, so these pass through: internal - # events from IN-FLIGHT work; recognized slash commands (/status, /approve, ... and /pause off - # as the in-band resume path); replies owned by in-flight work — pending update prompt, - # clarify, slash-confirm, dangerous-command approval, or steering an already-running session. - if not is_internal: - try: - from agent.estop import paused_reply as _estop_paused_reply - _paused_notice = _estop_paused_reply() - except ImportError: - _paused_notice = None - if _paused_notice is not None: - _estop_allow = False - _estop_cmd = None - try: - _estop_cmd = event.get_command() - except Exception: - _estop_cmd = None - if _estop_cmd: - try: - from hermes_cli.commands import ( - resolve_command as _resolve_estop_cmd, - ) - _estop_allow = _resolve_estop_cmd(_estop_cmd) is not None - except Exception: - _estop_allow = False - if not _estop_allow: - try: - _estop_key = self._session_key_for_source(source) - _estop_state = self._peek_session_state(_estop_key) - if ( - _estop_state is not None - and _estop_state.persistent.update_prompt_pending - ): - _estop_allow = True - if not _estop_allow and self._is_session_running(_estop_key): - # Steering / interrupting in-flight work (also covers pending clarify + - # tool approvals held by the running agent). - _estop_allow = True - if not _estop_allow: - from tools import slash_confirm as _estop_confirm_mod - if _estop_confirm_mod.get_pending(_estop_key): - _estop_allow = True - if not _estop_allow: - from tools.approval import ( - has_blocking_approval as _estop_has_approval, - ) - if _estop_has_approval(_estop_key): - _estop_allow = True - except Exception: - pass - if not _estop_allow: - logger.info( - "Gateway turn paused by global emergency stop (platform=%s chat=%s)", - getattr(getattr(source, "platform", None), "value", "unknown"), - getattr(source, "chat_id", None) or "unknown", - ) - return _paused_notice + """Return the global emergency-stop (`hermes pause`) notice when this turn must be blocked. + + Placed after auth so unauthorized senders can't probe pause state. Internal events from + in-flight work always pass. + """ + if is_internal: + return None + try: + from agent.estop import paused_reply as _estop_paused_reply + _paused_notice = _estop_paused_reply() + except ImportError: + return None + if _paused_notice is None or self._hm_estop_turn_allowed(event, source): + return None + logger.info( + "Gateway turn paused by global emergency stop (platform=%s chat=%s)", + getattr(getattr(source, "platform", None), "value", "unknown"), + getattr(source, "chat_id", None) or "unknown", + ) + return _paused_notice + + @staticmethod + def _hm_write_update_response(response_text: str) -> Optional[str]: + """Atomically hand *response_text* to the detached update process; returns the OSError str.""" + from gateway.run import _hermes_home + response_path = _hermes_home / ".update_response" + prompt_path = _hermes_home / ".update_prompt.json" + try: + tmp = response_path.with_suffix(".tmp") + tmp.write_text(response_text, encoding="utf-8") + tmp.replace(response_path) + prompt_path.unlink(missing_ok=True) + except OSError as e: + return str(e) return None def _hm_update_prompt_reply( self, event: "MessageEvent", _quick_key: str, allow_gateway_control: bool ) -> Optional[str]: - """Consume a reply to a pending ``/update`` prompt; None when nothing was consumed.""" - from gateway.run import _hermes_home - # Route replies to a pending /update prompt back to the detached update process via - # .update_response. Recognized slash commands must bypass this or /new, /help etc. get - # silently consumed as update answers. + """Consume a reply to a pending ``/update`` prompt; None when nothing was consumed. + + Routes the reply to the detached update process via ``.update_response``. Recognized slash + commands must bypass this or /new, /help etc. get silently consumed as update answers. + """ _up_state = self._peek_session_state(_quick_key) - if ( + if not ( allow_gateway_control and _up_state is not None and _up_state.persistent.update_prompt_pending ): - raw = (event.text or "").strip() - # Accept /approve and /deny as shorthand for yes/no - cmd = event.get_command() - if cmd in {"approve", "yes"}: - response_text = "y" - elif cmd in {"deny", "no"}: - response_text = "n" + return None + raw = (event.text or "").strip() + # Accept /approve and /deny as shorthand for yes/no + cmd = event.get_command() + _recognized_cmd = None + if cmd in {"approve", "yes"}: + response_text = "y" + elif cmd in {"deny", "no"}: + response_text = "n" + else: + if cmd: + try: + from hermes_cli.commands import resolve_command as _resolve_update_cmd + _cmd_def = _resolve_update_cmd(cmd) + _recognized_cmd = _cmd_def.name if _cmd_def else None + except Exception: + _recognized_cmd = None + response_text = "" if _recognized_cmd else raw + if response_text: + err = self._hm_write_update_response(response_text) + if err is not None: + logger.warning("Failed to write update response: %s", err) + return f"✗ Failed to send response to update process: {err}" + _up_state.persistent.update_prompt_pending = False + label = response_text if len(response_text) <= 20 else response_text[:20] + "…" + return f"✓ Sent `{label}` to the update process." + # Recognized slash command during a pending update prompt: write a blank response so the + # detached update's ``_gateway_prompt`` returns the prompt's default (typically a safe + # "n" / skip) and exits instead of blocking on stdin until the watcher timeout. + if _recognized_cmd: + err = self._hm_write_update_response("") + if err is None: + logger.info( + "Recognized /%s during pending update prompt for %s; " + "cancelled prompt with default and dispatching command", + _recognized_cmd, + _quick_key, + ) else: - _recognized_cmd = None - if cmd: - try: - from hermes_cli.commands import resolve_command as _resolve_update_cmd - except Exception: - _resolve_update_cmd = None - if _resolve_update_cmd is not None: - try: - _cmd_def = _resolve_update_cmd(cmd) - _recognized_cmd = _cmd_def.name if _cmd_def else None - except Exception: - _recognized_cmd = None - response_text = "" if _recognized_cmd else raw - if response_text: - response_path = _hermes_home / ".update_response" - prompt_path = _hermes_home / ".update_prompt.json" - try: - tmp = response_path.with_suffix(".tmp") - tmp.write_text(response_text, encoding="utf-8") - tmp.replace(response_path) - prompt_path.unlink(missing_ok=True) - except OSError as e: - logger.warning("Failed to write update response: %s", e) - return f"✗ Failed to send response to update process: {e}" - _up_state.persistent.update_prompt_pending = False - label = response_text if len(response_text) <= 20 else response_text[:20] + "…" - return f"✓ Sent `{label}` to the update process." - # Recognized slash command during a pending update prompt: write a blank response so the - # detached update's ``_gateway_prompt`` returns the prompt's default (typically a safe - # "n" / skip) and exits instead of blocking on stdin until the watcher timeout. - if _recognized_cmd: - response_path = _hermes_home / ".update_response" - prompt_path = _hermes_home / ".update_prompt.json" - try: - tmp = response_path.with_suffix(".tmp") - tmp.write_text("", encoding="utf-8") - tmp.replace(response_path) - prompt_path.unlink(missing_ok=True) - logger.info( - "Recognized /%s during pending update prompt for %s; " - "cancelled prompt with default and dispatching command", - _recognized_cmd, - _quick_key, - ) - except OSError as e: - logger.warning( - "Failed to write cancel response for pending update prompt: %s", - e, - ) - _up_state.persistent.update_prompt_pending = False + logger.warning("Failed to write cancel response for pending update prompt: %s", err) + _up_state.persistent.update_prompt_pending = False return None async def _hm_clarify_reply( @@ -446,14 +434,28 @@ class GatewayInboundMixin: ) return None + # Reply → choice for a pending slash-confirm prompt; the command spelling wins over the + # bang/slash-stripped free-text spelling. + _SLASH_CONFIRM_CMD_CHOICES = { + "approve": "once", "yes": "once", "ok": "once", "confirm": "once", + "always": "always", "remember": "always", + "cancel": "cancel", "no": "cancel", "deny": "cancel", "nevermind": "cancel", + } + _SLASH_CONFIRM_TEXT_CHOICES = { + "approve": "once", "approve once": "once", "once": "once", + "always": "always", "always approve": "always", + "cancel": "cancel", "nevermind": "cancel", "no": "cancel", + } + async def _hm_slash_confirm_reply( self, event: "MessageEvent", _quick_key: str, allow_gateway_control: bool ) -> Optional[str]: - """Resolve a reply to a pending slash-confirm prompt; None when the message falls through.""" - # Replies to a pending slash-confirm prompt (/reload-mcp etc.): /approve, /always, /cancel and - # short aliases. Anything else falls through — a stale pending confirm does NOT block other - # commands. A pending dangerous-command approval takes precedence: /approve there unblocks - # the waiting tool thread; slash-confirm only catches it when no tool approval is live. + """Resolve a reply to a pending slash-confirm prompt; None when the message falls through. + + Accepts /approve, /always, /cancel and short aliases. Anything else falls through — a stale + pending confirm does NOT block other commands. A pending dangerous-command approval takes + precedence: /approve there unblocks the waiting tool thread. + """ from tools import slash_confirm as _slash_confirm_mod _pending_confirm = _slash_confirm_mod.get_pending(_quick_key) _tool_approval_live = False @@ -462,324 +464,238 @@ class GatewayInboundMixin: _tool_approval_live = has_blocking_approval(_quick_key) except Exception: _tool_approval_live = False - if allow_gateway_control and _pending_confirm and not _tool_approval_live: - _raw_reply = (event.text or "").strip() - # Accept bang-prefixed replies (`!always`, `!cancel`) verbatim: Slack/Matrix show the - # `!` prefix (typed `/` is blocked in Slack threads) and adapters only rewrite - # `!` — confirm keywords aren't commands, so the `!` survives to here. - _norm_reply = _raw_reply.lstrip("!/").lower() - _cmd_reply = event.get_command() - _confirm_choice = None - if _cmd_reply in {"approve", "yes", "ok", "confirm"}: - _confirm_choice = "once" - elif _cmd_reply in {"always", "remember"}: - _confirm_choice = "always" - elif _cmd_reply in {"cancel", "no", "deny", "nevermind"}: - _confirm_choice = "cancel" - elif _norm_reply in {"approve", "approve once", "once"}: - _confirm_choice = "once" - elif _norm_reply in {"always", "always approve"}: - _confirm_choice = "always" - elif _norm_reply in {"cancel", "nevermind", "no"}: - _confirm_choice = "cancel" - if _confirm_choice is not None: - _resolved = await _slash_confirm_mod.resolve( - _quick_key, _pending_confirm.get("confirm_id"), _confirm_choice, - ) - return _resolved or "" - # Stale pending + unrelated command: the user moved on, so drop the pending state rather - # than let the confirm block normal usage indefinitely. - _slash_confirm_mod.clear_if_stale(_quick_key) + if not (allow_gateway_control and _pending_confirm and not _tool_approval_live): + return None + # Accept bang-prefixed replies (`!always`, `!cancel`) verbatim: Slack/Matrix show the `!` + # prefix (typed `/` is blocked in Slack threads) and adapters only rewrite + # `!` — confirm keywords aren't commands, so the `!` survives to here. + _norm_reply = (event.text or "").strip().lstrip("!/").lower() + _confirm_choice = ( + self._SLASH_CONFIRM_CMD_CHOICES.get(event.get_command()) + or self._SLASH_CONFIRM_TEXT_CHOICES.get(_norm_reply) + ) + if _confirm_choice is not None: + _resolved = await _slash_confirm_mod.resolve( + _quick_key, _pending_confirm.get("confirm_id"), _confirm_choice, + ) + return _resolved or "" + # Stale pending + unrelated command: the user moved on, so drop the pending state rather + # than let the confirm block normal usage indefinitely. + _slash_confirm_mod.clear_if_stale(_quick_key) return None - def _hm_evict_stale_running_agent(self, _quick_key: str) -> None: - """Evict a leaked/reaped ``_running_agents`` slot before the busy-session fast-path.""" + def _hm_evict_idle_stale_agent(self, _quick_key: str) -> None: + """Evict a leaked lock from a hung/crashed handler. + + With inactivity-based timeout active tasks can run for hours, so evict only when the agent + has been *idle* past the threshold (or has no activity tracker and its wall-clock age is + extreme). The pending sentinel is never evicted: it has no get_activity_summary(), so the + idle check would read inf and race the async setup path. + """ from gateway.run import _AGENT_PENDING_SENTINEL, _float_env - # Staleness eviction: detect leaked locks from hung/crashed handlers. With inactivity-based - # timeout active tasks can run for hours, so evict only when the agent has been *idle* past - # the threshold (or has no activity tracker and its wall-clock age is extreme). _raw_stale_timeout = _float_env("HERMES_AGENT_TIMEOUT", 1800) _quick_state = self._peek_session_state(_quick_key) _stale_ts = _quick_state.turn.started_ts if _quick_state else 0 - if _quick_state is not None and _quick_state.turn.agent is not None and _stale_ts: - _stale_age = time.time() - _stale_ts - _stale_agent = _quick_state.turn.agent - # Never evict the pending sentinel — it was just placed during async setup before the - # real agent exists. Sentinels have no get_activity_summary(), so the idle check would - # read inf >= timeout and evict them immediately, racing the setup path. - _stale_idle = float("inf") # assume idle if we can't check - _stale_detail = "" - _activity_summary_valid = False - if _stale_agent and hasattr(_stale_agent, "get_activity_summary"): - try: - _sa = _stale_agent.get_activity_summary() - from gateway.session_stall import ( - resolve_session_idle_seconds_from_activity, - ) - - _resolved_idle = resolve_session_idle_seconds_from_activity( - _sa if isinstance(_sa, dict) else None, - now=time.time(), - ) - if _resolved_idle is not None: - _stale_idle = _resolved_idle - _activity_summary_valid = True - _stale_detail = ( - f" | last_activity={_sa.get('last_activity_desc', 'unknown') if isinstance(_sa, dict) else 'unknown'} " - f"({_stale_idle:.0f}s ago) " - f"| iteration={_sa.get('api_call_count', 0) if isinstance(_sa, dict) else 0}/{_sa.get('max_iterations', 0) if isinstance(_sa, dict) else 0}" - ) - except Exception: - pass - # A valid activity clock is authoritative: total age alone never - # makes an actively progressing turn stale. The emergency wall TTL - # is only a fallback when the agent cannot report usable activity. - _wall_ttl = max(_raw_stale_timeout * 10, 7200) if _raw_stale_timeout > 0 else float("inf") - _should_evict = ( - _stale_agent is not _AGENT_PENDING_SENTINEL - and ( - ( - _activity_summary_valid - and _raw_stale_timeout > 0 - and _stale_idle >= _raw_stale_timeout - ) - or ( - not _activity_summary_valid - and _stale_age > _wall_ttl - ) - ) - ) - if _should_evict: - logger.warning( - "Evicting stale _running_agents entry for %s " - "(age: %.0fs, idle: %.0fs, timeout: %.0fs)%s", - _quick_key, _stale_age, _stale_idle, - _raw_stale_timeout, _stale_detail, - ) - self._invalidate_session_run_generation( - _quick_key, - reason="stale_running_agent_eviction", - ) - self._release_running_agent_state(_quick_key) - - # Durable-reaped guard. A session whose routing row was ended in state.db (``ws_orphan_reap`` - # / ``agent_close``) while the gateway lived keeps its in-memory turn slot, so the fast-path - # would queue every next message into the dead runtime. Evict the stale slot so the cold - # path re-attaches via ``get_or_create_session`` → ``reopen`` or creates a fresh session. - if self._is_session_running(_quick_key): + if _quick_state is None or _quick_state.turn.agent is None or not _stale_ts: + return + _stale_age = time.time() - _stale_ts + _stale_agent = _quick_state.turn.agent + _stale_idle = float("inf") # assume idle if we can't check + _stale_detail = "" + _activity_summary_valid = False + if _stale_agent and hasattr(_stale_agent, "get_activity_summary"): try: - _reap_store = getattr(self, "session_store", None) - # Use the public, lock-held accessors: peek_session_id resolves key -> session_id - # under the store lock, and returns a non-str on stubbed stores in bare test runners - # — both the isinstance() gate and the ``is True`` gate below keep this guard inert - # unless a real SessionStore answers. - _reap_peek = getattr(_reap_store, "peek_session_id", None) - _is_ended = getattr(_reap_store, "_is_session_ended_in_db", None) - _reap_sid = _reap_peek(_quick_key) if callable(_reap_peek) else None - if ( - isinstance(_reap_sid, str) - and _reap_sid - and callable(_is_ended) - and _is_ended(_reap_sid) is True - ): - logger.warning( - "Evicting stale _running_agents entry for %s — " - "durable session %s is ended (reaped) in state.db; " - "healing routing on next message (#99106)", - _quick_key, - _reap_sid, - ) - self._invalidate_session_run_generation( - _quick_key, - reason="reaped_session_eviction", - ) - self._release_running_agent_state(_quick_key) - except Exception: - logger.debug("reaped-session staleness check failed", exc_info=True) + _sa = _stale_agent.get_activity_summary() + from gateway.session_stall import resolve_session_idle_seconds_from_activity - async def _hm_handle_running_session_message( - self, event: "MessageEvent", source: SessionSource, _quick_key: str - ) -> Optional[str]: - """Fast-path for a message that arrives while this session's agent is running.""" - from gateway.run import ( - _AGENT_PENDING_SENTINEL, - _build_media_placeholder, - merge_pending_message_event, + _resolved_idle = resolve_session_idle_seconds_from_activity( + _sa if isinstance(_sa, dict) else None, now=time.time(), + ) + if _resolved_idle is not None: + _stale_idle = _resolved_idle + _activity_summary_valid = True + _sa_d = _sa if isinstance(_sa, dict) else {} + _stale_detail = ( + f" | last_activity={_sa_d.get('last_activity_desc', 'unknown')} " + f"({_stale_idle:.0f}s ago) " + f"| iteration={_sa_d.get('api_call_count', 0)}/{_sa_d.get('max_iterations', 0)}" + ) + except Exception: + pass + # A valid activity clock is authoritative: total age alone never makes an actively + # progressing turn stale. The emergency wall TTL is only a fallback when the agent cannot + # report usable activity. + _wall_ttl = max(_raw_stale_timeout * 10, 7200) if _raw_stale_timeout > 0 else float("inf") + _should_evict = _stale_agent is not _AGENT_PENDING_SENTINEL and ( + (_activity_summary_valid and _raw_stale_timeout > 0 and _stale_idle >= _raw_stale_timeout) + or (not _activity_summary_valid and _stale_age > _wall_ttl) ) - # PRIORITY handling when an agent is already running for this session. Default behavior is - # to interrupt immediately so user text/stop messages are handled with minimal latency. - # Exception: Telegram photo bursts arrive as near-simultaneous updates — do NOT interrupt - # for photo-only follow-ups; adapter-level batching absorbs them. - # Resolve the command once; each command's mid-run behavior is declared on its - # CommandDef (busy_policy / busy_handler in hermes_cli/commands.py) and dispatched via - # _dispatch_busy_slash_command below — no per-command if-chain here. + if _should_evict: + logger.warning( + "Evicting stale _running_agents entry for %s " + "(age: %.0fs, idle: %.0fs, timeout: %.0fs)%s", + _quick_key, _stale_age, _stale_idle, + _raw_stale_timeout, _stale_detail, + ) + self._invalidate_session_run_generation(_quick_key, reason="stale_running_agent_eviction") + self._release_running_agent_state(_quick_key) + + def _hm_evict_reaped_agent(self, _quick_key: str) -> None: + """Evict the in-memory turn slot of a session whose durable row was ended while the gateway lived. + + (``ws_orphan_reap`` / ``agent_close``): otherwise the fast-path queues every next message + into the dead runtime. The cold path then re-attaches via ``get_or_create_session`` → + ``reopen`` or creates a fresh session. + """ + try: + _reap_store = getattr(self, "session_store", None) + # Public, lock-held accessors: peek_session_id resolves key -> session_id under the + # store lock and returns a non-str on stubbed stores in bare test runners — the + # isinstance() and ``is True`` gates keep this inert unless a real SessionStore answers. + _reap_peek = getattr(_reap_store, "peek_session_id", None) + _is_ended = getattr(_reap_store, "_is_session_ended_in_db", None) + _reap_sid = _reap_peek(_quick_key) if callable(_reap_peek) else None + if ( + isinstance(_reap_sid, str) + and _reap_sid + and callable(_is_ended) + and _is_ended(_reap_sid) is True + ): + logger.warning( + "Evicting stale _running_agents entry for %s — " + "durable session %s is ended (reaped) in state.db; " + "healing routing on next message (#99106)", + _quick_key, + _reap_sid, + ) + self._invalidate_session_run_generation(_quick_key, reason="reaped_session_eviction") + self._release_running_agent_state(_quick_key) + except Exception: + logger.debug("reaped-session staleness check failed", exc_info=True) + + def _hm_evict_stale_running_agent(self, _quick_key: str) -> None: + """Evict a leaked/reaped ``_running_agents`` slot before the busy-session fast-path.""" + self._hm_evict_idle_stale_agent(_quick_key) + if self._is_session_running(_quick_key): + self._hm_evict_reaped_agent(_quick_key) + + def _hm_merge_pending_for_source( + self, source: SessionSource, _quick_key: str, event: "MessageEvent", *, merge_text: bool = False + ) -> None: + """Merge *event* into the source adapter's pending slot (no-op without an adapter).""" + from gateway.run import merge_pending_message_event + adapter = self._adapter_for_source(source) + if adapter: + merge_pending_message_event(adapter._pending_messages, _quick_key, event, merge_text=merge_text) + + async def _hm_busy_slash_or_photo( + self, event: "MessageEvent", source: SessionSource, _quick_key: str + ) -> Tuple[bool, Optional[str]]: + """Slash-command and photo-burst handling on the running-agent fast-path. + + Returns ``(handled, result)``. Each command's mid-run behavior is declared on its CommandDef + (busy_policy / busy_handler in hermes_cli/commands.py) — no per-command if-chain here. + """ from hermes_cli.commands import resolve_command as _resolve_cmd_inner _evt_cmd = event.get_command() _cmd_def_inner = _resolve_cmd_inner(_evt_cmd) if _evt_cmd else None - # /status and /context are intentionally pre-gate so users - # always see session state. + # /status and /context are intentionally pre-gate so users always see session state. if _cmd_def_inner and _cmd_def_inner.name == "status": - return await self._handle_status_command(event) + return True, await self._handle_status_command(event) if _cmd_def_inner and _cmd_def_inner.name == "context": - return await self._handle_context_command(event) + return True, await self._handle_context_command(event) - # Slash command access control on the running-agent fast-path. Mirrors the cold-path - # gate below so non-admins can't bypass gating just because an agent is busy. /status - # above is intentionally pre-gate; /help and /whoami are the always-allowed floor. + # Slash access control mirrors the cold-path gate so non-admins can't bypass gating just + # because an agent is busy. /help and /whoami are the always-allowed floor. if _evt_cmd and _cmd_def_inner is not None: _denied = self._check_slash_access(source, _cmd_def_inner.name) if _denied is not None: - return _denied + return True, _denied - # Any recognized slash command: dispatch according to its declared busy_policy (dispatch - # / interrupt_then_dispatch / reject). Unrecognized commands and plain text fall through - # to the interrupt/queue logic below. + # Any recognized slash command dispatches per its declared busy_policy (dispatch / + # interrupt_then_dispatch / reject). Unrecognized commands and plain text fall through. if _cmd_def_inner: - return await self._dispatch_busy_slash_command( - event, _cmd_def_inner, _quick_key, source, - ) + return True, await self._dispatch_busy_slash_command(event, _cmd_def_inner, _quick_key, source) + # Telegram photo bursts arrive as near-simultaneous updates — never interrupt for a + # photo-only follow-up; adapter-level batching absorbs them. if event.message_type == MessageType.PHOTO: logger.debug("PRIORITY photo follow-up for session %s — queueing without interrupt", _quick_key) - adapter = self._adapter_for_source(source) - if adapter: - merge_pending_message_event(adapter._pending_messages, _quick_key, event) - return None + self._hm_merge_pending_for_source(source, _quick_key, event) + return True, None + return False, None - effective_busy_input_mode = self._effective_busy_input_mode(source) - _telegram_followup_grace = float( - os.getenv("HERMES_TELEGRAM_FOLLOWUP_GRACE_SECONDS", "3.0") - ) + def _hm_busy_telegram_grace_queue( + self, event: "MessageEvent", source: SessionSource, _quick_key: str, effective_busy_input_mode: str + ) -> bool: + """Queue a Telegram text follow-up that lands within the post-start grace window.""" + _telegram_followup_grace = float(os.getenv("HERMES_TELEGRAM_FOLLOWUP_GRACE_SECONDS", "3.0")) _grace_state = self._peek_session_state(_quick_key) _started_at = _grace_state.turn.started_ts if _grace_state else 0 - if ( + if not ( source.platform == Platform.TELEGRAM and event.message_type == MessageType.TEXT and _telegram_followup_grace > 0 and _started_at and (time.time() - _started_at) <= _telegram_followup_grace ): - logger.debug( - "Telegram follow-up arrived %.2fs after run start for %s — queueing without interrupt", - time.time() - _started_at, - _quick_key, - ) - adapter = self._adapter_for_source(source) - if adapter: - if effective_busy_input_mode == "queue": - self._enqueue_fifo(_quick_key, event, adapter) - else: - merge_pending_message_event( - adapter._pending_messages, - _quick_key, - event, - merge_text=True, - ) - return None + return False + logger.debug( + "Telegram follow-up arrived %.2fs after run start for %s — queueing without interrupt", + time.time() - _started_at, + _quick_key, + ) + adapter = self._adapter_for_source(source) + if adapter: + if effective_busy_input_mode == "queue": + self._enqueue_fifo(_quick_key, event, adapter) + else: + self._hm_merge_pending_for_source(source, _quick_key, event, merge_text=True) + return True - _ra_state = self._peek_session_state(_quick_key) - running_agent = _ra_state.turn.agent if _ra_state else None - if running_agent is _AGENT_PENDING_SENTINEL: - # Agent is being set up but not ready yet. - if event.get_command() == "stop": - # Force-clean the sentinel so the session is unlocked. - self._release_running_agent_state(_quick_key) - logger.info("HARD STOP (pending) for session %s — sentinel cleared", _quick_key) - return EphemeralReply("⚡ Force-stopped. The agent was still starting — session unlocked.") - # Queue the message so it will be picked up after the - # agent starts. - adapter = self._adapter_for_source(source) - if adapter: - merge_pending_message_event( - adapter._pending_messages, - _quick_key, - event, - merge_text=True, - ) - return None - if self._draining: - queue_during_drain = self._queue_during_drain_enabled( - effective_busy_input_mode - ) - if queue_during_drain: - self._queue_or_replace_pending_event(_quick_key, event) - return ( - f"⏳ Gateway {self._status_action_gerund()} — queued for the next turn after it comes back." - if queue_during_drain - else f"⏳ Gateway is {self._status_action_gerund()} and is not accepting another turn right now." - ) - if effective_busy_input_mode == "queue": - logger.debug("PRIORITY queue follow-up for session %s", _quick_key) - self._queue_or_replace_pending_event(_quick_key, event) - return None - if effective_busy_input_mode == "steer": - # Steer mode: inject text into the running agent mid-run via - # agent.steer(). Falls back to queue semantics if the payload - # is empty, the agent lacks steer(), or steer() rejects. - steer_text = (event.text or "").strip() - steered = False - if ( - event.message_type == MessageType.TEXT - and not event.media_urls - and not event.media_types - and steer_text - and hasattr(running_agent, "steer") - ): - try: - steered = bool(running_agent.steer(steer_text)) - except Exception as exc: - logger.warning("PRIORITY steer failed for session %s: %s", _quick_key, exc) - steered = False - if steered: - logger.debug("PRIORITY steer for session %s", _quick_key) - return None - logger.debug("PRIORITY steer-fallback-to-queue for session %s", _quick_key) - self._queue_or_replace_pending_event(_quick_key, event) - return None - # Subagent protection (PRIORITY path). Same rationale as - # ``_handle_active_session_busy_message``: an interrupt cascades through - # ``_active_children`` and aborts in-flight delegate_task work, so demote to queue - # semantics while subagents run. /stop reached its handler above — still an escape hatch. - if self._agent_has_active_subagents(running_agent): - logger.info( - "PRIORITY interrupt demoted to queue for session %s " - "because the running agent has active subagents (#30170)", - _quick_key, - ) - self._queue_or_replace_pending_event(_quick_key, event) - return None - # Compression protection (PRIORITY path), as in ``_handle_active_session_busy_message``: - # an interrupt would start a new turn on the pre-rotation parent while compression - # rotates the id away, forking orphaned siblings. Demote to queue until rotation lands. - if await self._session_has_compression_in_flight(_quick_key): - logger.info( - "PRIORITY interrupt demoted to queue for session %s " - "because context compression is in flight (#56391)", - _quick_key, - ) - self._queue_or_replace_pending_event(_quick_key, event) - return None + def _hm_busy_steer(self, event: "MessageEvent", running_agent: Any, _quick_key: str) -> None: + """Steer mode: inject text mid-run via ``agent.steer()``, else fall back to queue semantics.""" + steer_text = (event.text or "").strip() + steered = False + if ( + event.message_type == MessageType.TEXT + and not event.media_urls + and not event.media_types + and steer_text + and hasattr(running_agent, "steer") + ): + try: + steered = bool(running_agent.steer(steer_text)) + except Exception as exc: + logger.warning("PRIORITY steer failed for session %s: %s", _quick_key, exc) + steered = False + if steered: + logger.debug("PRIORITY steer for session %s", _quick_key) + return + logger.debug("PRIORITY steer-fallback-to-queue for session %s", _quick_key) + self._queue_or_replace_pending_event(_quick_key, event) + + async def _hm_busy_interrupt( + self, event: "MessageEvent", source: SessionSource, running_agent: Any, _quick_key: str + ) -> None: + """Interrupt path: redirect text-only corrections when supported, else ``agent.interrupt()``.""" + from gateway.run import _build_media_placeholder # Text-only corrections redirect the live turn (preserving displayed context) when the # runtime supports it; media/voice and older runtimes use the interrupt path below. if ( event.message_type == MessageType.TEXT and not event.media_urls and not event.media_types - and getattr(running_agent, "_supports_active_turn_redirect", False) - is True + and getattr(running_agent, "_supports_active_turn_redirect", False) is True and hasattr(running_agent, "redirect") ): try: if running_agent.redirect((event.text or "").strip()): logger.debug("PRIORITY redirect for session %s", _quick_key) - return None + return except Exception as exc: - logger.warning( - "PRIORITY redirect failed for session %s: %s", - _quick_key, - exc, - ) + logger.warning("PRIORITY redirect failed for session %s: %s", _quick_key, exc) logger.debug("PRIORITY interrupt for session %s", _quick_key) _interrupt_text = event.text _media_urls = getattr(event, "media_urls", None) or [] @@ -796,8 +712,156 @@ class GatewayInboundMixin: running_agent.interrupt(_interrupt_text) # The interrupt message is delivered via adapter._pending_messages (read by _run_agent); # don't also buffer it on self — that copy was never consumed and grew unbounded. + + async def _hm_handle_running_session_message( + self, event: "MessageEvent", source: SessionSource, _quick_key: str + ) -> Optional[str]: + """Fast-path for a message that arrives while this session's agent is running. + + Default is to interrupt immediately so user text/stop messages are handled with minimal + latency; busy_input_mode queue/steer, subagent and compression protection demote to queue. + """ + from gateway.run import _AGENT_PENDING_SENTINEL + _handled, _result = await self._hm_busy_slash_or_photo(event, source, _quick_key) + if _handled: + return _result + + effective_busy_input_mode = self._effective_busy_input_mode(source) + if self._hm_busy_telegram_grace_queue(event, source, _quick_key, effective_busy_input_mode): + return None + + _ra_state = self._peek_session_state(_quick_key) + running_agent = _ra_state.turn.agent if _ra_state else None + if running_agent is _AGENT_PENDING_SENTINEL: + # Agent is being set up but not ready yet. + if event.get_command() == "stop": + # Force-clean the sentinel so the session is unlocked. + self._release_running_agent_state(_quick_key) + logger.info("HARD STOP (pending) for session %s — sentinel cleared", _quick_key) + return EphemeralReply("⚡ Force-stopped. The agent was still starting — session unlocked.") + # Queue the message so it will be picked up after the agent starts. + self._hm_merge_pending_for_source(source, _quick_key, event, merge_text=True) + return None + if self._draining: + queue_during_drain = self._queue_during_drain_enabled(effective_busy_input_mode) + if queue_during_drain: + self._queue_or_replace_pending_event(_quick_key, event) + return ( + f"⏳ Gateway {self._status_action_gerund()} — queued for the next turn after it comes back." + if queue_during_drain + else f"⏳ Gateway is {self._status_action_gerund()} and is not accepting another turn right now." + ) + if effective_busy_input_mode == "queue": + logger.debug("PRIORITY queue follow-up for session %s", _quick_key) + self._queue_or_replace_pending_event(_quick_key, event) + return None + if effective_busy_input_mode == "steer": + self._hm_busy_steer(event, running_agent, _quick_key) + return None + # Subagent protection: an interrupt cascades through ``_active_children`` and aborts + # in-flight delegate_task work, so demote to queue while subagents run. /stop reached its + # handler above — still an escape hatch. + if self._agent_has_active_subagents(running_agent): + logger.info( + "PRIORITY interrupt demoted to queue for session %s " + "because the running agent has active subagents (#30170)", + _quick_key, + ) + self._queue_or_replace_pending_event(_quick_key, event) + return None + # Compression protection: an interrupt would start a new turn on the pre-rotation parent + # while compression rotates the id away, forking orphaned siblings. + if await self._session_has_compression_in_flight(_quick_key): + logger.info( + "PRIORITY interrupt demoted to queue for session %s " + "because context compression is in flight (#56391)", + _quick_key, + ) + self._queue_or_replace_pending_event(_quick_key, event) + return None + await self._hm_busy_interrupt(event, source, running_agent, _quick_key) return None + def _hm_quick_commands(self) -> dict: + """User-defined ``quick_commands`` mapping from config (empty dict when unset/malformed).""" + if isinstance(self.config, dict): + quick_commands = self.config.get("quick_commands", {}) or {} + else: + quick_commands = getattr(self.config, "quick_commands", {}) or {} + return quick_commands if isinstance(quick_commands, dict) else {} + + @staticmethod + def _hm_expand_alias_quick_command(event: "MessageEvent", qcmd: dict) -> Optional[str]: + """Rewrite ``event.text`` to an alias quick command's target; returns the new command name.""" + target = (qcmd.get("target") or "").strip() + if not target: + return None + target = target if target.startswith("/") else f"/{target}" + target_command = target.lstrip("/") + user_args = event.get_command_args().strip() + event.text = f"{target} {user_args}".strip() + return target_command.split()[0] if target_command else target_command + + async def _hm_command_hooks( + self, event: "MessageEvent", source: SessionSource, _quick_key: str, command: str, canonical: str + ) -> Tuple[bool, Optional[str], Optional[str]]: + """Fire ``pre_command`` (observer) and ``command:`` (interceptor) hooks. + + Returns ``(handled, result, new_command)``; ``new_command`` is set when a handler rewrote + the command. The running-agent intercept path deliberately does NOT fire these — a slow or + hostile plugin must not interfere with the operator's escape hatches for a live agent. + """ + try: + from hermes_cli.plugins import fire_pre_command_hook + fire_pre_command_hook( + surface="gateway", + command=str(canonical), + alias_used=str(command), + args_raw=event.get_command_args().strip(), + session_key=_quick_key, + platform=source.platform.value if source.platform else "", + ) + except Exception as _pre_cmd_err: + logger.debug("pre_command hook dispatch failed (non-fatal): %s", _pre_cmd_err) + + # Handlers may return ``{"decision": "deny" | "handled" | "rewrite", ...}`` to intercept + # dispatch; handlers returning nothing behave as plain observers. + raw_args = event.get_command_args().strip() + hook_ctx = { + "platform": source.platform.value if source.platform else "", + "user_id": source.user_id, + "command": canonical, + "raw_command": command, + "args": raw_args, + "raw_args": raw_args, + } + try: + hook_results = await self.hooks.emit_collect(f"command:{canonical}", hook_ctx) + except Exception as _hook_err: + logger.debug("command:%s hook dispatch failed (non-fatal): %s", canonical, _hook_err) + hook_results = [] + + for hook_result in hook_results: + if not isinstance(hook_result, dict): + continue + decision = str(hook_result.get("decision", "")).strip().lower() + if not decision or decision == "allow": + continue + message = hook_result.get("message") + message = message if isinstance(message, str) and message else None + if decision == "deny": + return True, message or f"Command `/{command}` was blocked by a hook.", None + if decision == "handled": + return True, message, None + if decision == "rewrite": + new_command = str(hook_result.get("command_name", "")).strip().lstrip("/") + if not new_command: + continue + new_args = str(hook_result.get("raw_args", "")).strip() + event.text = f"/{new_command} {new_args}".strip() + return False, None, event.get_command() + return False, None, None + async def _hm_resolve_command( self, event: "MessageEvent", source: SessionSource, _quick_key: str ) -> Tuple[bool, Optional[str], Optional[str], Optional[str]]: @@ -806,123 +870,229 @@ class GatewayInboundMixin: Returns ``(handled, result, command, canonical)``; when ``handled`` the caller returns ``result`` as-is (it may legitimately be None). """ - # Check for commands command = event.get_command() - from hermes_cli.commands import ( - is_gateway_known_command, - resolve_command as _resolve_cmd, - ) + from hermes_cli.commands import is_gateway_known_command, resolve_command as _resolve_cmd - # Resolve aliases to canonical name so dispatch and hook names - # don't depend on the exact alias the user typed. + # Resolve aliases to canonical name so dispatch and hook names don't depend on the alias. _cmd_def = _resolve_cmd(command) if command else None canonical = _cmd_def.name if _cmd_def else command # Expand alias quick commands before built-in dispatch so targets like /model openai/gpt-5.5 - # --provider openrouter reach the /model handler. Preserve built-in precedence; aliases only + # --provider openrouter reach the /model handler. Built-ins keep precedence: aliases only # need early handling when the typed command is not already known. if command and _cmd_def is None: - if isinstance(self.config, dict): - quick_commands = self.config.get("quick_commands", {}) or {} - else: - quick_commands = getattr(self.config, "quick_commands", {}) or {} - if isinstance(quick_commands, dict) and command in quick_commands: - qcmd = quick_commands[command] - if qcmd.get("type") == "alias": - target = (qcmd.get("target") or "").strip() - if target: - target = target if target.startswith("/") else f"/{target}" - target_command = target.lstrip("/") - user_args = event.get_command_args().strip() - event.text = f"{target} {user_args}".strip() - command = target_command.split()[0] if target_command else target_command - _cmd_def = _resolve_cmd(command) if command else None - canonical = _cmd_def.name if _cmd_def else command - - # Per-platform slash command access control. Only kicks in when the operator has set - # ``allow_admin_from`` for the source's scope (DM vs group). When unset → backward-compat: - # every allowed user can run every command. When set → non-admins get only - # ``user_allowed_commands`` plus the /help, /whoami floor. Plain chat is never gated. - if command and canonical and is_gateway_known_command(canonical): - _denied = self._check_slash_access(source, canonical) - if _denied is not None: - return True, _denied, command, canonical - - # pre_command observer hook (returns ignored) fires for every recognized slash command - # BEFORE core handling, mirroring cli.py. The running-agent intercept path above (/stop, - # /approve, busy_policy) deliberately does NOT fire it — a slow or hostile plugin must not - # interfere with the operator's escape hatches for a live agent. - if command and is_gateway_known_command(canonical): - try: - from hermes_cli.plugins import fire_pre_command_hook - fire_pre_command_hook( - surface="gateway", - command=str(canonical), - alias_used=str(command), - args_raw=event.get_command_args().strip(), - session_key=_quick_key, - platform=source.platform.value if source.platform else "", - ) - except Exception as _pre_cmd_err: - logger.debug( - "pre_command hook dispatch failed (non-fatal): %s", - _pre_cmd_err, - ) - - # Fire ``command:`` for any recognized slash command (built-in or plugin). - # Handlers may return ``{"decision": "deny" | "handled" | "rewrite", ...}`` to intercept - # dispatch; handlers returning nothing behave as plain observers. - if command and is_gateway_known_command(canonical): - raw_args = event.get_command_args().strip() - hook_ctx = { - "platform": source.platform.value if source.platform else "", - "user_id": source.user_id, - "command": canonical, - "raw_command": command, - "args": raw_args, - "raw_args": raw_args, - } - try: - hook_results = await self.hooks.emit_collect( - f"command:{canonical}", hook_ctx - ) - except Exception as _hook_err: - logger.debug( - "command:%s hook dispatch failed (non-fatal): %s", - canonical, _hook_err, - ) - hook_results = [] - - for hook_result in hook_results: - if not isinstance(hook_result, dict): - continue - decision = str(hook_result.get("decision", "")).strip().lower() - if not decision or decision == "allow": - continue - if decision == "deny": - message = hook_result.get("message") - if isinstance(message, str) and message: - return True, message, command, canonical - return True, f"Command `/{command}` was blocked by a hook.", command, canonical - if decision == "handled": - message = hook_result.get("message") - return True, message if isinstance(message, str) and message else None, command, canonical - if decision == "rewrite": - new_command = str( - hook_result.get("command_name", "") - ).strip().lstrip("/") - if not new_command: - continue - new_args = str(hook_result.get("raw_args", "")).strip() - event.text = f"/{new_command} {new_args}".strip() - command = event.get_command() + quick_commands = self._hm_quick_commands() + qcmd = quick_commands[command] if command in quick_commands else None + if qcmd is not None and qcmd.get("type") == "alias": + new_command = self._hm_expand_alias_quick_command(event, qcmd) + if new_command is not None: + command = new_command _cmd_def = _resolve_cmd(command) if command else None canonical = _cmd_def.name if _cmd_def else command - break + if not (command and canonical and is_gateway_known_command(canonical)): + return False, None, command, canonical + + # Per-platform slash access control: only active when the operator set ``allow_admin_from`` + # for the source's scope; then non-admins get ``user_allowed_commands`` plus the + # /help, /whoami floor. Plain chat is never gated. + _denied = self._check_slash_access(source, canonical) + if _denied is not None: + return True, _denied, command, canonical + + _handled, _result, new_command = await self._hm_command_hooks( + event, source, _quick_key, command, canonical + ) + if _handled: + return True, _result, command, canonical + if new_command is not None: + command = new_command + _cmd_def = _resolve_cmd(command) if command else None + canonical = _cmd_def.name if _cmd_def else command return False, None, command, canonical + async def _hm_cmd_new(self, event, source, _quick_key): + if await asyncio.to_thread(self._is_telegram_topic_root_lobby, source): + return True, self._telegram_topic_root_new_message() + async def _do_reset(): + return await self._handle_reset_command(event) + return True, await self._maybe_confirm_destructive_slash( + event=event, + command="new", + title="/new", + detail="This starts a fresh session and discards the current conversation history.", + execute=_do_reset, + ) + + async def _hm_cmd_start(self, event, source, _quick_key): + logger.info("Ignoring /start platform ping for session %s", _quick_key) + return True, "" + + async def _hm_cmd_egress(self, event, source, _quick_key): + from hermes_cli.proxy_cli import format_status_text + + return True, format_status_text() + + async def _hm_rewrite_turn_to_prompt(self, event, source, name: str, ack: str, build) -> Tuple[bool, Optional[str]]: + """Ack, then rewrite the turn to ``build()`` and fall through to normal agent processing. + + Mirrors the /blueprint fall-through so role alternation is preserved; no engine, works on + any backend. A failing builder replies with a retry hint instead. + """ + await self._send_command_ack(source, ack, name) + try: + event.text = build() + except Exception: + return True, f"Could not start /{name} — please try again." + return False, None + + async def _hm_cmd_learn(self, event, source, _quick_key): + from agent.learn_prompt import build_learn_prompt + + _learn_req = event.get_command_args().strip() + _ack = ( + "Learning a skill from what you described…" + if _learn_req + else "Learning a skill from this conversation…" + ) + return await self._hm_rewrite_turn_to_prompt( + event, source, "learn", _ack, lambda: build_learn_prompt(_learn_req) + ) + + async def _hm_cmd_plan(self, event, source, _quick_key): + from agent.plan_prompt import build_plan_prompt + + _plan_task = event.get_command_args().strip() + _ack = ( + f"Planning: {_plan_task[:80]}{'…' if len(_plan_task) > 80 else ''}" + if _plan_task + else "Planning from this conversation's context…" + ) + return await self._hm_rewrite_turn_to_prompt( + event, source, "plan", _ack, lambda: build_plan_prompt(_plan_task) + ) + + async def _hm_cmd_init(self, event, source, _quick_key): + # /init builds the prompt first: the ack wording depends on whether AGENTS.md exists. + from hermes_cli.init_command import build_init_prompt_for_cwd + + _init_notes = event.get_command_args().strip() + try: + _init_prompt = build_init_prompt_for_cwd(extra=_init_notes) + except Exception: + return True, "Could not start /init — please try again." + _ack = ( + "Updating AGENTS.md from a project scan…" + if "UPDATE the existing AGENTS.md" in _init_prompt + else "Generating AGENTS.md from a project scan…" + ) + await self._send_command_ack(source, _ack, "init") + event.text = _init_prompt + return False, None + + async def _hm_cmd_blueprint(self, event, source, _quick_key): + _blueprint_result = await self._handle_blueprint_command(event) + _blueprint_seed = getattr(_blueprint_result, "agent_seed", None) + if not _blueprint_seed: + return True, getattr(_blueprint_result, "text", "") or None + # Blueprint matched — rewrite the turn to the seed and fall through so the agent collects + # each slot value conversationally, then calls the cronjob tool (the /steer pattern). + _ack = getattr(_blueprint_result, "text", "") or "" + if _ack: + await self._send_command_ack(source, _ack, "blueprint") + try: + event.text = _blueprint_seed + except Exception: + return True, getattr(_blueprint_result, "text", "") or None + return False, None + + async def _hm_cmd_undo(self, event, source, _quick_key): + async def _do_undo(): + return await self._handle_undo_command(event) + _undo_n = 1 + _undo_raw = event.get_command_args().strip() + if _undo_raw: + try: + _undo_n = max(1, int(_undo_raw.split()[0])) + except (ValueError, IndexError): + _undo_n = 1 + _undo_detail = ( + "This removes the last user/assistant exchange from history." + if _undo_n == 1 + else f"This removes the last {_undo_n} user turns from history." + ) + return True, await self._maybe_confirm_destructive_slash( + event=event, command="undo", title="/undo", detail=_undo_detail, execute=_do_undo, + ) + + async def _hm_cmd_queue(self, event, source, _quick_key): + queue_payload = event.get_command_args().strip() + if not queue_payload: + return True, "Usage: /queue " + with suppress(Exception): + event.text = queue_payload + return False, None + + async def _hm_cmd_steer(self, event, source, _quick_key): + # No active agent — /steer has nothing to inject into. Strip the prefix so the rewritten + # text is sent to the agent as a regular user turn; an empty payload surfaces the usage hint. + steer_payload = event.get_command_args().strip() + if not steer_payload: + return True, "Usage: /steer (no agent is running; sending as a normal message)" + with suppress(Exception): + event.text = steer_payload + return False, None + + async def _hm_cmd_moa(self, event, source, _quick_key): + # /moa is one-shot sugar only: run a single prompt through the default MoA preset, then + # restore the prior model. To *switch* to a MoA preset for the session, pick it from the + # model picker (MoA presets surface as a virtual "Mixture of Agents" provider). + from hermes_cli.moa_config import moa_usage, normalize_moa_config + from hermes_cli.config import load_config + + moa_payload = event.get_command_args().strip() + if not moa_payload: + return True, moa_usage() + try: + cfg = load_config() + moa_cfg = normalize_moa_config(cfg.get("moa") if isinstance(cfg, dict) else {}) + except Exception: + moa_cfg = normalize_moa_config({}) + preset = moa_cfg["default_preset"] + try: + event.text = moa_payload + _moa_state = self._session_state(_quick_key) + event._moa_restore_override = _moa_state.conversation.model_override + _moa_state.conversation.model_override = { + "provider": "moa", + "model": preset, + "base_url": "moa://local", + "api_key": "moa-virtual-provider", + "api_mode": "chat_completions", + } + self._evict_cached_agent(_quick_key) + event._moa_disable_after_turn = True + except Exception: + return True, "Failed to prepare MoA turn." + return False, None + + # Idle-path built-ins with bespoke flow (confirmations, prompt rewrites, one-shot MoA). Each + # returns ``(handled, result)``; ``(False, None)`` falls through to the agent. + _HM_CANONICAL_COMMANDS = { + "new": "_hm_cmd_new", + "start": "_hm_cmd_start", + "egress": "_hm_cmd_egress", + "learn": "_hm_cmd_learn", + "plan": "_hm_cmd_plan", + "init": "_hm_cmd_init", + "blueprint": "_hm_cmd_blueprint", + "undo": "_hm_cmd_undo", + "queue": "_hm_cmd_queue", + "steer": "_hm_cmd_steer", + "moa": "_hm_cmd_moa", + } + async def _hm_dispatch_canonical_command( self, event: "MessageEvent", @@ -941,183 +1111,36 @@ class GatewayInboundMixin: ) if plain_handler is not None: return True, await plain_handler(event) - - if canonical == "new": - if await asyncio.to_thread(self._is_telegram_topic_root_lobby, source): - return True, self._telegram_topic_root_new_message() - async def _do_reset(): - return await self._handle_reset_command(event) - return True, await self._maybe_confirm_destructive_slash( - event=event, - command="new", - title="/new", - detail=( - "This starts a fresh session and discards the current " - "conversation history." - ), - execute=_do_reset, - ) - - if canonical == "start": - logger.info("Ignoring /start platform ping for session %s", _quick_key) - return True, "" - - if canonical == "egress": - from hermes_cli.proxy_cli import format_status_text - - return True, format_status_text() - - if canonical == "learn": - # Open-ended: rewrite the turn to a standards-guided prompt and fall through to normal - # agent processing. Mirrors the /blueprint fall-through so role alternation is - # preserved. No engine, works on any backend. - from agent.learn_prompt import build_learn_prompt - - _learn_req = event.get_command_args().strip() - _ack = ( - "Learning a skill from what you described…" - if _learn_req - else "Learning a skill from this conversation…" - ) - await self._send_command_ack(source, _ack, "learn") - try: - event.text = build_learn_prompt(_learn_req) - # fall through to agent processing - except Exception: - return True, "Could not start /learn — please try again." - - if canonical == "plan": - # /plan: rewrite the turn to the plan-mode prompt and fall through to normal agent - # processing (the /learn fall-through keeps role alternation). Works on any backend. - from agent.plan_prompt import build_plan_prompt - - _plan_task = event.get_command_args().strip() - _ack = ( - f"Planning: {_plan_task[:80]}{'…' if len(_plan_task) > 80 else ''}" - if _plan_task - else "Planning from this conversation's context…" - ) - await self._send_command_ack(source, _ack, "plan") - try: - event.text = build_plan_prompt(_plan_task) - # fall through to agent processing - except Exception: - return True, "Could not start /plan — please try again." - - if canonical == "init": - # /init: rewrite the turn to a guidance-laden prompt and fall through to normal agent - # processing (the /learn fall-through keeps role alternation). Works on any backend. - from hermes_cli.init_command import build_init_prompt_for_cwd - - _init_notes = event.get_command_args().strip() - try: - _init_prompt = build_init_prompt_for_cwd(extra=_init_notes) - except Exception: - return True, "Could not start /init — please try again." - _ack = ( - "Updating AGENTS.md from a project scan…" - if "UPDATE the existing AGENTS.md" in _init_prompt - else "Generating AGENTS.md from a project scan…" - ) - await self._send_command_ack(source, _ack, "init") - event.text = _init_prompt - # fall through to agent processing - - if canonical == "blueprint": - _blueprint_result = await self._handle_blueprint_command(event) - _blueprint_seed = getattr(_blueprint_result, "agent_seed", None) - if _blueprint_seed: - # Blueprint matched — rewrite the turn to the seed and fall through to - # _handle_message_with_agent so the agent collects each slot value conversationally, - # then calls the cronjob tool (the /steer fall-through pattern). - _ack = getattr(_blueprint_result, "text", "") or "" - if _ack: - await self._send_command_ack(source, _ack, "blueprint") - try: - event.text = _blueprint_seed - except Exception: - return True, getattr(_blueprint_result, "text", "") or None - else: - return True, getattr(_blueprint_result, "text", "") or None - - if canonical == "undo": - async def _do_undo(): - return await self._handle_undo_command(event) - _undo_n = 1 - _undo_raw = event.get_command_args().strip() - if _undo_raw: - try: - _undo_n = max(1, int(_undo_raw.split()[0])) - except (ValueError, IndexError): - _undo_n = 1 - _undo_detail = ( - "This removes the last user/assistant exchange from history." - if _undo_n == 1 - else f"This removes the last {_undo_n} user turns from history." - ) - return True, await self._maybe_confirm_destructive_slash( - event=event, - command="undo", - title="/undo", - detail=_undo_detail, - execute=_do_undo, - ) - - if canonical == "queue": - queue_payload = event.get_command_args().strip() - if not queue_payload: - return True, "Usage: /queue " - with suppress(Exception): - event.text = queue_payload - - if canonical == "steer": - # No active agent — /steer has nothing to inject into. Strip the prefix so downstream - # treats it as a normal user message; an empty payload surfaces the usage hint. - steer_payload = event.get_command_args().strip() - if not steer_payload: - return True, "Usage: /steer (no agent is running; sending as a normal message)" - with suppress(Exception): - event.text = steer_payload - # Do NOT return — fall through to _handle_message_with_agent at the end of this function - # so the rewritten text is sent to the agent as a regular user turn. - - if canonical == "moa": - # /moa is one-shot sugar only: run a single prompt through the default MoA preset, then - # restore the prior model. To *switch* to a MoA preset for the session, pick it from the - # model picker (MoA presets surface as a virtual "Mixture of Agents" provider). - from hermes_cli.moa_config import ( - moa_usage, - normalize_moa_config, - ) - from hermes_cli.config import load_config - - moa_payload = event.get_command_args().strip() - if not moa_payload: - return True, moa_usage() - try: - cfg = load_config() - moa_cfg = normalize_moa_config(cfg.get("moa") if isinstance(cfg, dict) else {}) - except Exception: - moa_cfg = normalize_moa_config({}) - preset = moa_cfg["default_preset"] - try: - event.text = moa_payload - _moa_state = self._session_state(_quick_key) - event._moa_restore_override = _moa_state.conversation.model_override - _moa_state.conversation.model_override = { - "provider": "moa", - "model": preset, - "base_url": "moa://local", - "api_key": "moa-virtual-provider", - "api_mode": "chat_completions", - } - self._evict_cached_agent(_quick_key) - event._moa_disable_after_turn = True - except Exception: - return True, "Failed to prepare MoA turn." - + bespoke = self._HM_CANONICAL_COMMANDS.get(canonical) + if bespoke is not None: + return await getattr(self, bespoke)(event, source, _quick_key) return False, None + async def _hm_run_exec_quick_command(self, command: str, exec_cmd: str) -> str: + """Run a ``type: exec`` quick command in the gateway process (30 s cap, sanitized env).""" + try: + # Sanitize env to prevent credential leakage — the gateway process has all API keys + # in os.environ. + from tools.environments.local import build_subprocess_env + sanitized_env = build_subprocess_env() + proc = await asyncio.create_subprocess_shell( + exec_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=sanitized_env, + ) + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30) + output = (stdout or stderr).decode().strip() + # Redact any remaining sensitive patterns in output + if output: + from agent.redact import redact_sensitive_text + output = redact_sensitive_text(output) + return output if output else "Command returned no output." + except asyncio.TimeoutError: + return "Quick command timed out (30s)." + except Exception as e: + return f"Quick command error: {e}" + async def _hm_dispatch_quick_and_plugin_commands( self, event: "MessageEvent", source: SessionSource, command: Optional[str] ) -> Tuple[bool, Optional[str], Optional[str]]: @@ -1129,62 +1152,28 @@ class GatewayInboundMixin: return True, f"⏳ Gateway is {self._status_action_gerund()} and is not accepting new work right now.", command # User-defined quick commands (bypass agent loop, no LLM call) - if command: - if isinstance(self.config, dict): - quick_commands = self.config.get("quick_commands", {}) or {} + quick_commands = self._hm_quick_commands() + if command and command in quick_commands: + qcmd = quick_commands[command] + # Quick commands are slash capabilities too — and type:exec ones run a shell command in + # the gateway process. They are never in the registry, so the early gate never fires for + # them; apply the same admin/user policy to the raw typed name here. + _denied = self._check_slash_access(source, command) + if _denied is not None: + return True, _denied, command + qtype = qcmd.get("type") + if qtype == "exec": + exec_cmd = qcmd.get("command", "") + if not exec_cmd: + return True, f"Quick command '/{command}' has no command defined.", command + return True, await self._hm_run_exec_quick_command(command, exec_cmd), command + if qtype == "alias": + new_command = self._hm_expand_alias_quick_command(event, qcmd) + if new_command is None: + return True, f"Quick command '/{command}' has no target defined.", command + command = new_command # Fall through to normal command dispatch below else: - quick_commands = getattr(self.config, "quick_commands", {}) or {} - if not isinstance(quick_commands, dict): - quick_commands = {} - if command in quick_commands: - # Quick commands are slash capabilities too — and type:exec ones run a shell command - # in the gateway process. The early gate only fires for registry-known commands and - # quick commands are never in the registry, so apply the same admin/user policy to - # the raw typed name here so non-admins can't invoke admin-only quick commands. - _denied = self._check_slash_access(source, command) - if _denied is not None: - return True, _denied, command - qcmd = quick_commands[command] - if qcmd.get("type") == "exec": - exec_cmd = qcmd.get("command", "") - if exec_cmd: - try: - # Sanitize env to prevent credential leakage — quick commands run in the - # gateway process, which has all API keys in os.environ. - from tools.environments.local import build_subprocess_env - sanitized_env = build_subprocess_env() - proc = await asyncio.create_subprocess_shell( - exec_cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - env=sanitized_env, - ) - stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30) - output = (stdout or stderr).decode().strip() - # Redact any remaining sensitive patterns in output - if output: - from agent.redact import redact_sensitive_text - output = redact_sensitive_text(output) - return True, output if output else "Command returned no output.", command - except asyncio.TimeoutError: - return True, "Quick command timed out (30s).", command - except Exception as e: - return True, f"Quick command error: {e}", command - else: - return True, f"Quick command '/{command}' has no command defined.", command - elif qcmd.get("type") == "alias": - target = (qcmd.get("target") or "").strip() - if target: - target = target if target.startswith("/") else f"/{target}" - target_command = target.lstrip("/") - user_args = event.get_command_args().strip() - event.text = f"{target} {user_args}".strip() - command = target_command.split()[0] if target_command else target_command - # Fall through to normal command dispatch below - else: - return True, f"Quick command '/{command}' has no target defined.", command - else: - return True, f"Quick command '/{command}' has unsupported type (supported: 'exec', 'alias').", command + return True, f"Quick command '/{command}' has unsupported type (supported: 'exec', 'alias').", command # Plugin-registered slash commands if command: @@ -1204,6 +1193,66 @@ class GatewayInboundMixin: return False, None, command + def _hm_bundle_slash_rewrite( + self, event: "MessageEvent", source: SessionSource, _quick_key: str, command: str + ) -> bool: + """Rewrite ``/`` to the bundle invocation message; True when handled. + + Skill bundles take precedence over individual skill commands (mirrors CLI dispatch). + """ + try: + from agent.skill_bundles import ( + build_bundle_invocation_message, + resolve_bundle_command_key, + ) + bundle_key = resolve_bundle_command_key(command) + if bundle_key is None: + return False + user_instruction = event.get_command_args().strip() + # Pass the platform explicitly: bundle skill loading bypasses get_skill_commands()' + # scan-time disabled filter, and one gateway process serves several platforms, so + # env-var platform resolution can't be trusted here. + _bundle_plat = source.platform.value if source.platform else None + bundle_result = build_bundle_invocation_message( + bundle_key, user_instruction, task_id=_quick_key, platform=_bundle_plat, + ) + if not bundle_result: + return False + msg, _loaded, missing = bundle_result + event.text = msg + if missing: + logger.info("Bundle %s skipped missing skills: %s", bundle_key, ", ".join(missing)) + return True # Fall through to normal message processing with bundle content + except Exception as exc: + logger.warning("Bundle dispatch failed: %s", exc) + return False + + @staticmethod + def _hm_unknown_slash_reply(command: str, source: SessionSource) -> Optional[str]: + """Reply for a /command that is not built-in/plugin/skill; None when it is known.""" + from gateway.run import _check_unavailable_skill + from hermes_cli.commands import GATEWAY_KNOWN_COMMANDS + # Known-but-disabled or uninstalled skill → actionable guidance. + _unavail_msg = _check_unavailable_skill(command) + if _unavail_msg: + return _unavail_msg + # Genuinely unrecognized: warn instead of forwarding to the LLM as free text (it invents + # tool calls). Normalize to hyphenated form first: the quick-command block may have set an + # alias target, so the resolved def can be stale. + if command.replace("_", "-") in GATEWAY_KNOWN_COMMANDS: + return None + logger.warning( + "Unrecognized slash command /%s from %s — replying with unknown-command notice", + command, + source.platform.value if source.platform else "?", + ) + return ( + f"Unknown command `/{command}`. " + f"Type /commands to see what's available, " + f"or resend without the leading slash to send " + f"as a regular message." + ) + def _hm_skill_slash_rewrite( self, event: "MessageEvent", @@ -1214,145 +1263,75 @@ class GatewayInboundMixin: """Rewrite ``/`` / ``/`` invocations into the skill prompt on ``event.text``. Returns a reply string when the command is disabled/unknown/failed, else None. + resolve_skill_command_key() handles the Telegram underscore/hyphen round-trip so + /claude_code from Telegram autocomplete still resolves to the claude-code skill. """ - from gateway.run import _check_unavailable_skill - from hermes_cli.commands import GATEWAY_KNOWN_COMMANDS - - # Skill slash commands: /skill-name loads the skill and sends to agent. - # resolve_skill_command_key() handles the Telegram underscore/hyphen round-trip so - # /claude_code from Telegram autocomplete still resolves to the claude-code skill. - if command: - # Skill bundles take precedence over individual skill commands — - # / loads multiple skills at once. Mirrors CLI dispatch. - _bundle_handled = False - try: - from agent.skill_bundles import ( - build_bundle_invocation_message, - resolve_bundle_command_key, - ) - bundle_key = resolve_bundle_command_key(command) - if bundle_key is not None: - user_instruction = event.get_command_args().strip() - # Pass the platform explicitly: bundle skill loading bypasses - # get_skill_commands()' scan-time disabled filter, and the gateway serves - # multiple platforms in one process, so env-var platform resolution can't be - # trusted here. - _bundle_plat = source.platform.value if source.platform else None - bundle_result = build_bundle_invocation_message( - bundle_key, user_instruction, task_id=_quick_key, - platform=_bundle_plat, + if not command or self._hm_bundle_slash_rewrite(event, source, _quick_key, command): + return None + try: + from agent.skill_commands import ( + get_skill_commands, + build_skill_invocation_message, + resolve_skill_command_key, + ) + skill_cmds = get_skill_commands() + cmd_key = resolve_skill_command_key(command) + if cmd_key is None: + return self._hm_unknown_slash_reply(command, source) + # Per-platform disabled check: get_skill_commands() only applies the *global* disabled + # list at scan time and its cache is process-global across platforms. + _skill_name = skill_cmds[cmd_key].get("name", "") + _plat = source.platform.value if source.platform else None + if _plat and _skill_name: + from agent.skill_utils import get_disabled_skill_names as _get_plat_disabled + if _skill_name in _get_plat_disabled(platform=_plat): + return ( + f"The **{_skill_name}** skill is disabled for {_plat}.\n" + f"Enable it with: `hermes skills config`" ) - if bundle_result: - msg, _loaded, missing = bundle_result - event.text = msg - _bundle_handled = True - if missing: - logger.info( - "Bundle %s skipped missing skills: %s", - bundle_key, ", ".join(missing), - ) - # Fall through to normal message processing with bundle content - except Exception as exc: - logger.warning("Bundle dispatch failed: %s", exc) - - if command and not locals().get("_bundle_handled", False): + user_instruction = event.get_command_args().strip() + # Stacked slash-skill invocations: `/skill-a /skill-b do XYZ` loads every leading skill + # (up to 5), not just the first. Mirrors CLI. try: from agent.skill_commands import ( - get_skill_commands, - build_skill_invocation_message, - resolve_skill_command_key, + build_stacked_skill_invocation_message as _build_stacked, + split_stacked_skill_commands, ) - skill_cmds = get_skill_commands() - cmd_key = resolve_skill_command_key(command) - if cmd_key is not None: - # Check per-platform disabled status before executing. get_skill_commands() only - # applies the *global* disabled list at scan time; per-platform overrides need - # checking here because the cache is process-global across platforms. - _skill_name = skill_cmds[cmd_key].get("name", "") - _plat = source.platform.value if source.platform else None - if _plat and _skill_name: - from agent.skill_utils import get_disabled_skill_names as _get_plat_disabled - if _skill_name in _get_plat_disabled(platform=_plat): - return ( - f"The **{_skill_name}** skill is disabled for {_plat}.\n" - f"Enable it with: `hermes skills config`" - ) - user_instruction = event.get_command_args().strip() - # Stacked slash-skill invocations: `/skill-a /skill-b do XYZ` loads every - # leading skill (up to 5), not just the first. Mirrors CLI. - try: - from agent.skill_commands import ( - build_stacked_skill_invocation_message as _build_stacked, - split_stacked_skill_commands, - ) - extra_keys, stacked_instruction = ( - split_stacked_skill_commands(user_instruction) - ) - except Exception: - _build_stacked = None - extra_keys, stacked_instruction = [], user_instruction - if extra_keys and _plat: - # split_stacked_skill_commands() only resolves that each extra token is a - # KNOWN skill command — like get_skill_commands() itself, it has no per- - # platform view. Re-check every stacked skill against the same disabled list, - # or a skill disabled for this platform still gets loaded via the stack. - from agent.skill_utils import get_disabled_skill_names as _get_plat_disabled - _plat_disabled = _get_plat_disabled(platform=_plat) - _disabled_extra = [ - skill_cmds.get(k, {}).get("name", "") - for k in extra_keys - if skill_cmds.get(k, {}).get("name", "") in _plat_disabled - ] - if _disabled_extra: - return ( - f"The **{', '.join(_disabled_extra)}** skill(s) in this " - f"stacked invocation are disabled for {_plat}.\n" - f"Enable them with: `hermes skills config`" - ) - if extra_keys and _build_stacked is not None: - stacked_result = _build_stacked( - [cmd_key, *extra_keys], - stacked_instruction, - task_id=_quick_key, - ) - if stacked_result: - msg, _loaded, _missing = stacked_result - event.text = msg - # Fall through to normal message processing - else: - return f"Failed to load stacked skills for /{command}." - else: - msg = build_skill_invocation_message( - cmd_key, user_instruction, task_id=_quick_key - ) - if msg: - event.text = msg - # Fall through to normal message processing with skill content - else: - # Not an active skill — check if it's a known-but-disabled or - # uninstalled skill and give actionable guidance. - _unavail_msg = _check_unavailable_skill(command) - if _unavail_msg: - return _unavail_msg - # Genuinely unrecognized /command (not built-in/plugin/skill/known-inactive): - # warn instead of forwarding to the LLM as free text (it invents tool calls). - # Normalize to hyphenated form first: the quick-command block may have set an - # alias target, so _cmd_def can be stale. - if command.replace("_", "-") not in GATEWAY_KNOWN_COMMANDS: - logger.warning( - "Unrecognized slash command /%s from %s — " - "replying with unknown-command notice", - command, - source.platform.value if source.platform else "?", - ) - return ( - f"Unknown command `/{command}`. " - f"Type /commands to see what's available, " - f"or resend without the leading slash to send " - f"as a regular message." - ) - except Exception as e: - logger.debug("Skill command check failed (non-fatal): %s", e) + extra_keys, stacked_instruction = split_stacked_skill_commands(user_instruction) + except Exception: + _build_stacked = None + extra_keys, stacked_instruction = [], user_instruction + if extra_keys and _plat: + # split_stacked_skill_commands() only checks each extra token is a KNOWN skill; it + # has no per-platform view, so re-check every stacked skill against the disabled list. + from agent.skill_utils import get_disabled_skill_names as _get_plat_disabled + _plat_disabled = _get_plat_disabled(platform=_plat) + _disabled_extra = [ + skill_cmds.get(k, {}).get("name", "") + for k in extra_keys + if skill_cmds.get(k, {}).get("name", "") in _plat_disabled + ] + if _disabled_extra: + return ( + f"The **{', '.join(_disabled_extra)}** skill(s) in this " + f"stacked invocation are disabled for {_plat}.\n" + f"Enable them with: `hermes skills config`" + ) + if extra_keys and _build_stacked is not None: + stacked_result = _build_stacked( + [cmd_key, *extra_keys], stacked_instruction, task_id=_quick_key, + ) + if not stacked_result: + return f"Failed to load stacked skills for /{command}." + msg, _loaded, _missing = stacked_result + event.text = msg + else: + msg = build_skill_invocation_message(cmd_key, user_instruction, task_id=_quick_key) + if msg: + event.text = msg + # Fall through to normal message processing with skill content + except Exception as e: + logger.debug("Skill command check failed (non-fatal): %s", e) return None async def _handle_message(self, event: MessageEvent) -> Optional[str]: