"""Gateway runner - entry point for messaging platform integrations. Provides ``start_gateway()`` (start all configured adapters) and ``GatewayRunner`` (lifecycle). Run via ``python -m gateway.run`` or ``python cli.py --gateway``.""" # hermes_bootstrap must be the very first import (UTF-8 stdio on Windows; no-op on POSIX). try: import hermes_bootstrap except ModuleNotFoundError as exc: # a partial ``hermes update`` can leave the bootstrap unregistered if exc.name != "hermes_bootstrap": raise # the bootstrap exists but cannot load: skipping it would skip PM activation import asyncio import concurrent.futures import dataclasses import json import logging import os import re import shlex import site import sys import signal import threading import time import traceback from collections import OrderedDict from contextvars import Context, copy_context from pathlib import Path from datetime import datetime from typing import Callable, Dict, Optional, Any, List, Tuple, cast from agent.async_utils import safe_schedule_threadsafe from agent.i18n import t from agent.conversation_compression import ( COMPACTION_DONE_STATUS, COMPACTION_HEARTBEAT_STATUS, COMPACTION_STATUS, COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE, COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE, COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE, COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE, IDLE_COMPACTION_STATUS_TEMPLATE, PRE_API_COMPRESSION_STATUS_TEMPLATE, PREFLIGHT_COMPRESSION_STATUS_TEMPLATE) from agent.conversation_compression_archive import MERGED_DURABLE_ROWS, RETIRED_DURABLE_ROWS, UNNAMED_DURABLE_ROWS from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX from agent.interrupt_compat import request_hard_interrupt from agent.message_metadata import ABSORBED_MESSAGE_UIDS, MESSAGE_UID, copy_identity_fields, is_compaction_scaffolding, scaffolding_replay_fields from agent.turn_context import compression_made_progress from agent.session_activity import ActivityProvenance, AwakeIdleMeter from hermes_cli.config import _is_ssh_remote_tilde_cwd, cfg_get from hermes_cli.fallback_config import pre_agent_fallback_notice from gateway.turn_executor import _UnboundedThreadExecutor # Per-session AIAgent cache bounds (agents are heavy); see _enforce_agent_cache_cap/_session_housekeeping_watcher. _AGENT_CACHE_MAX_SIZE = 128 _AGENT_CACHE_IDLE_TTL_SECS = 3600.0 # evict agents idle for >1h _PLATFORM_CONNECT_TIMEOUT_SECS_DEFAULT = 30.0 # Telegram connect proves a real getUpdates round trip; must cover polling-start deadlines + readiness. _TELEGRAM_CONNECT_TIMEOUT_SECS_DEFAULT = 180.0 # The initial Telegram connect gates `running` for EVERY platform, so it must not spend the full 180s. # Cold-start cap for Telegram (#85993): the initial connect awaited before the gateway reaches `running` # must not spend the full 180s budget — an unreachable Telegram would hold EVERY platform's serving state # hostage for the whole window. The initial attempt gets one bounded try; on timeout the platform is queued # for the reconnect watcher, which retries with the full 180s budget (is_reconnect=True preserves the # offline update queue, #46621). _TELEGRAM_INITIAL_CONNECT_TIMEOUT_SECS_DEFAULT = 45.0 _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT = 5.0 # Size of the separate pool for best-effort session HOUSEKEEPING; why it is separate: _run_housekeeping_in_executor. _HOUSEKEEPING_MAX_WORKERS = 4 # End reasons meaning the USER deliberately closed this thread. Shared by _classify_completion_target and # _resolve_async_delegation_session so they never disagree (else a "delivered" reason is acked, then lost). _USER_BOUNDARY_END_REASONS = ("session_reset", "user_exit", "session_switch", "new_session") # Bounds one stall-notify send so a wedged transport can't block the watcher; on timeout the next tick retries. _STALL_NOTIFY_SEND_TIMEOUT_SECONDS = 15.0 _GATEWAY_PROXY_SSE_BUFFER_MAX_CHARS = 16 * 1024 * 1024 _TELEGRAM_COMMAND_MENTION_RE = re.compile(r"(?)\s+\d[\d,]*\s+messages,\s+retrying" r"|compressed\s+~[\d,]+\s+(?:→|->)\s+~[\d,]+\s+tokens,\s+retrying" r"|context\s+reduced\s+to\s+[\d,]+\s+tokens\s+\(was\s+[\d,]+\),\s+retrying" r"|session\s+compressed\s+\d+\s+times" r"|rate\s+limited\.\s+waiting\s+\d" r"|retrying\s+in\s+\d" r"|max\s+retries\s+\(\d+\).*(?:trying\s+fallback|exhausted|invalid\s+responses)" r"|stream\s+(?:drop|drop\s+mid\s+tool-call).+retry\s+\d" r"|stale\s+connections\s+from\s+a\s+previous\s+provider\s+issue" rf"|{re.escape(COMPACTION_DONE_STATUS)}" r")", re.IGNORECASE | re.DOTALL) _HYGIENE_COOLDOWN_LADDER_MULTIPLIERS = (1, 3, 9) # Ceiling on an escalated cooldown (cf. _RECONNECT_BACKOFF_CAP): base × ladder can reach 9h ≈ "compaction off". _HYGIENE_COOLDOWN_MAX_SECONDS = 3600.0 # Flat retry-after when hygiene is ABANDONED by turn-hold expiry (not a failure: outside the streak ladder). _HYGIENE_TURNHOLD_RETRY_SECONDS = 60.0 def _gateway_session_db_inner(gateway): """The raw SessionDB behind ``gateway._session_db`` (unwrapping the async facade), or None.""" session_db = getattr(gateway, "_session_db", None) return getattr(session_db, "_db", session_db) def _hygiene_cooldown_for_failure(gateway, session_key: str, base_cooldown_seconds: float) -> float: """Bump the hygiene failure streak and return the escalated cooldown (x1/x3/x9 over base, clamped). Hygiene's per-run ``AIAgent`` is fresh, so the streak lives in SQLite keyed by rotation-stable session_key. It exists because the in-agent equivalent is unreachable from here: ``ContextCompressor.record_timeout_failure`` escalates on an absolute 60 -> 300 -> 900s ladder driven by the in-memory ``_consecutive_timeout_failures`` counter, which ``bind_session_state`` zeroes. Session hygiene constructs a FRESH ``AIAgent`` per run and re-binds state every time, so from the gateway that streak is structurally always 0 and only the flat ``hygiene_failure_cooldown_seconds`` could ever be recorded — a session whose summary model always times out retried on that same fixed interval forever (#79624). The streak is mirrored to SQLite by rotation-stable ``session_key`` so it outlives both the per-run agent and gateway restarts; ``PersistentState`` keeps the hot in-process view. """ streak, state = 1, None try: state = gateway._session_state(session_key).persistent except Exception as exc: logger.debug("hygiene failure streak update failed: %s", exc) increment = getattr(_gateway_session_db_inner(gateway), "increment_hygiene_failure_streak", None) if callable(increment): try: streak = max(1, int(increment(session_key))) if state is not None: state.hygiene_failure_streak = streak except Exception as exc: logger.debug("hygiene failure streak persist failed: %s", exc) if state is not None: state.hygiene_failure_streak += 1 streak = state.hygiene_failure_streak elif state is not None: state.hygiene_failure_streak += 1 streak = state.hygiene_failure_streak multiplier = _HYGIENE_COOLDOWN_LADDER_MULTIPLIERS[ min(streak, len(_HYGIENE_COOLDOWN_LADDER_MULTIPLIERS)) - 1] return min(base_cooldown_seconds * multiplier, _HYGIENE_COOLDOWN_MAX_SECONDS) def _reset_hygiene_failure_streak(gateway, session_key: str) -> None: """Clear the hygiene failure streak after a compression that reduced context. Peeks, never get-or-creates: a no-op 0 write must not create a never-evicted ``_sessions`` row.""" try: state = gateway._peek_session_state(session_key) if state is not None: state.persistent.hygiene_failure_streak = 0 except Exception as exc: logger.debug("hygiene failure streak reset failed: %s", exc) reset = getattr(_gateway_session_db_inner(gateway), "reset_hygiene_failure_streak", None) if callable(reset): try: reset(session_key) except Exception as exc: logger.debug("hygiene failure streak persistent reset failed: %s", exc) def hygiene_compaction_recovered( *, aborted: bool, rotated: bool, in_place: bool, msg_count: int, new_count: int, approx_tokens: int, new_tokens: int) -> bool: """True when a hygiene run actually recovered the session (extracted to be unit testable). Requires no abort, a real rewrite (the no-op path reuses pre-compression counts) and material shrink per :func:`compression_made_progress` (a bare ``<`` misses row-count wins and counts estimate noise). * the compressor did not abort (no summary produced at all); * the transcript was actually rewritten — either rotated into a new session or compacted in place. The degenerate "did not rotate or compact in place" path (#21301) reuses the pre-compression counts, so relying on the numbers alone would read a no-op as success; * the request materially shrank, per the canonical :func:`compression_made_progress` (#39548) — a row-count drop counts even when the summary keeps the token estimate flat, and a sub-5% token wobble does not count at all. """ if aborted or not (rotated or in_place): return False return compression_made_progress(msg_count, new_count, approx_tokens, new_tokens) def _hygiene_compression_timeout_message( *, total_exhausted: bool, elapsed: float, idle_timeout: float, progress_observed: bool) -> str: """Describe the host timeout that actually ended hygiene compression. Chat users cannot edit model config, so the copy names /compress, /new and `hermes doctor`, never a config key or the raw second counts (those stay in the gateway log).""" lead = t("gateway.compress.hygiene_timeout") if total_exhausted: return lead return lead + t("gateway.compress.hygiene_timeout_doctor_hint") def _cached_agent_for_hygiene(gateway, session_key: str): """The cached live AIAgent for ``session_key`` (or the pending sentinel / None), read under the cache lock.""" cache = getattr(gateway, "_agent_cache", None) if cache is None: return None lock = getattr(gateway, "_agent_cache_lock", None) try: with (lock or suppress()): entry = cache.get(session_key) except Exception: entry = None return entry[0] if isinstance(entry, tuple) and entry else entry async def run_codex_hygiene_compaction( gateway, session_key: str, session_id: str, *, auto_mode: str, history: list, approx_tokens: int, timeout_seconds: float, failure_cooldown_seconds: float = 300.0) -> str: """Session hygiene for ``codex_app_server`` sessions. The real context is the server-side thread; the local transcript is a never-replayed mirror, so rewriting it shrinks nothing and evicting the live agent starts the next turn on an EMPTY thread. So: compact the LIVE agent via ``thread/compact/start``, keep it cached, never build a detached compressor. ``native``/``off`` skip without local fallback. Returns ``compacted``, ``skipped:`` or ``failed:``. See #73503. * Evicting the cached live agent afterwards destroys the only real context: the next turn spawns an EMPTY thread and the model starts blank while Hermes still mirrors a full history (abrupt amnesia — the user-facing damage documented on #73503). """ mode = str(auto_mode or "native").lower() if mode not in {"native", "hermes", "off"}: mode = "native" if mode != "hermes": # native = app-server compacts itself; off = operator disabled. Local fallback can't shrink the thread. return f"skipped:mode={mode}" agent = _cached_agent_for_hygiene(gateway, session_key) if agent is None or agent is _AGENT_PENDING_SENTINEL: # No live agent → no live thread; a detached mirror-only rewrite is the no-op this exists to remove. return "skipped:no-cached-agent" if getattr(agent, "_codex_session", None) is None: return "skipped:no-live-thread" compressor = getattr(agent, "context_compressor", None) count_before = getattr(compressor, "compression_count", 0) # copy_context carries profile secret scope / HERMES_HOME override (executors don't propagate ContextVars). worker_future = asyncio.get_running_loop().run_in_executor( None, copy_context().run, lambda: agent._compress_context(history, "", approx_tokens=approx_tokens, task_id=session_id or "default", trigger="gateway_hygiene")) track_worker = getattr(gateway, "_track_deferred_agent_worker", None) if callable(track_worker): # ``wait_for`` only cancels the asyncio wrapper; keep the running executor thread visible to shutdown. track_worker(worker_future, agent) try: await asyncio.wait_for(asyncio.shield(worker_future), timeout=max(float(timeout_seconds), 1.0)) except TimeoutError: # Executor thread keeps running (own RPC timeouts); brake retries so a wedged app-server isn't re-hit. if failure_cooldown_seconds >= 0: _record_hygiene_cooldown( gateway, session_id, failure_cooldown_seconds, "codex app-server thread compaction timed out") logger.warning( "Session hygiene: codex app-server thread compaction for " "session %s timed out after %.1fs; continuing without compaction", session_id, timeout_seconds) return "failed:timeout" except Exception as exc: logger.warning( "Session hygiene: codex app-server thread compaction for session %s failed: %s", session_id, exc) return f"failed:{exc}" count_after = getattr(compressor, "compression_count", 0) if count_after > count_before: # Native boundary recorded: compacted server-side; mirror NOT rewritten, agent stays cached. _reset_hygiene_failure_streak(gateway, session_key) return "compacted" # No boundary: internal skip or compaction error; the codex route already persisted its own cooldown. return "failed:no-boundary" def hygiene_wait_should_extend( *, idle: float, timeout: float, waited: float, ceiling: float, fence_cancelled: bool = False ) -> bool: """Whether the hygiene host should keep waiting for a slow summary. A cancelled commit fence cannot commit: extending only queues inbound messages behind a doomed attempt. Stop extending immediately so the turn can continue. See #96953. """ return not fence_cancelled and idle < timeout and waited < ceiling def _record_hygiene_cooldown( gateway, session_id: str, cooldown_seconds: float, error: Optional[str] = None) -> None: """Persist a session-hygiene compression-failure cooldown to the state DB (survives restarts). ``error`` must be forwarded: the recorder writes compression_failure_error UNCONDITIONALLY (NULL clobber). Uses the same ``compression_failure_cooldown_until`` column and ``record_compression_failure_cooldown`` method that the in-conversation compression path (``agent/context_compressor.py``) already uses, so the cooldown survives gateway restarts (#74136). """ recorder = getattr(_gateway_session_db_inner(gateway), "record_compression_failure_cooldown", None) if recorder is None: return try: recorder(session_id, time.time() + cooldown_seconds, error) except Exception as exc: logger.debug("session hygiene cooldown persist failed: %s", exc) def _status_template_to_regex(template: str) -> str: """Compile a compression status template constant into a regex source. Literal text is escaped verbatim (wording drift can't diverge from the matcher); ``{field}`` -> numeric.""" parts = re.split(r"\{[^{}]*\}", template) return r"[\d,]+".join(re.escape(part) for part in parts) # ROUTINE compression progress statuses, derived from the SAME template constants the emit sites format. # Used ONLY by the opt-in ``compression.progress_notices`` gate below (#52995) to decide which of the noisy # statuses matched by _TELEGRAM_NOISY_STATUS_RE are compression progress (deliverable when the user opted # in) versus unrelated aux/retry chatter (always suppressed on chat surfaces). Failure notices and manual # /compress feedback never match _TELEGRAM_NOISY_STATUS_RE in the first place, so they are unaffected by # this gate. _COMPRESSION_PROGRESS_STATUS_RE = re.compile( "|".join( _status_template_to_regex(_template) for _template in ( COMPACTION_STATUS, COMPACTION_HEARTBEAT_STATUS, COMPACTION_DONE_STATUS, PRE_API_COMPRESSION_STATUS_TEMPLATE, PREFLIGHT_COMPRESSION_STATUS_TEMPLATE, IDLE_COMPACTION_STATUS_TEMPLATE, COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE, COMPRESSION_RETRY_MESSAGES_STATUS_TEMPLATE, COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE, COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE)), re.IGNORECASE) def _gateway_compression_progress_notices_enabled() -> bool: """True when ``compression.progress_notices`` is on (default False: chat is silent by design). Read live (mtime-cached) so a config edit applies at the next status; fail-closed on read error. Reads ``compression.progress_notices`` from the gateway's raw YAML config (#52995). """ try: config = _load_gateway_config() compression_cfg = config.get("compression") if isinstance(config, dict) else None if isinstance(compression_cfg, dict): return str(compression_cfg.get("progress_notices", False)).strip().lower() in { "true", "1", "yes", "on"} except Exception: pass return False # Surfaces consuming gateway text programmatically must keep RAW status/error text; unknown/empty -> chat. _GATEWAY_RAW_TEXT_PLATFORMS = frozenset({"local", "api_server", "webhook", "msgraph_webhook"}) def _gateway_surface_passes_raw_text(platform: Any) -> bool: """True only for programmatic/local surfaces that must keep raw text.""" return _gateway_platform_value(platform) in _GATEWAY_RAW_TEXT_PLATFORMS _GATEWAY_PROVIDER_POLICY_RE = re.compile( r"(" # raw provider policy/safety bodies are noisy and may be sensitive r"cybersecurity\s+risk" r"|security\s+policy" r"|safety\s+policy" r"|policy\s+violation" r"|violat(?:e|es|ed|ion)" r"|blocked\s+(?:because|by|under)" r"|request\s+(?:was\s+)?(?:blocked|rejected)" r"|disallowed" r"|moderation" r")", re.IGNORECASE) # ``401`` as a status token: not glued to a digit or a timestamp/identifier separator on the left # (``05:14:15,401``), but trailing punctuation is a real envelope (``HTTP 401: Unauthorized``, # ``returned 401.``) and must keep matching (#89401). _GATEWAY_AUTH_ERROR_RE = re.compile( r"(provider\s+authentication\s+failed|incorrect\s+api\s+key|invalid\s+api\s+key" r"|(? None: """Make detached Windows gateway runs see the Hermes venv packages. Patched before MCP discovery so tool injection does not depend on launchers preserving PYTHONPATH.""" if sys.platform != "win32": return project_root = Path(__file__).resolve().parent.parent from pm.environments import committed_venv, venv_python_version # A PM install's store Python was already activated onto the committed generation by # hermes_bootstrap; overlaying the leftover pre-PM venv loads a foreign ABI (#122183). if committed_venv(project_root) is not None: return candidates = [Path(v) for v in (os.environ.get("VIRTUAL_ENV"),) if v] + [project_root / "venv"] seen: set[str] = set() for venv_dir in candidates: try: resolved_venv = venv_dir.resolve() except OSError: resolved_venv = venv_dir venv_key = str(resolved_venv).lower() if venv_key in seen: continue seen.add(venv_key) site_packages = resolved_venv / "Lib" / "site-packages" if not site_packages.exists(): continue # Built for another (or an unreadable) Python: its compiled modules crash this one (#139085). if (found := venv_python_version(resolved_venv)) != sys.version_info[:2]: logger.debug("Skipping venv overlay %s: Python %s, gateway %s", resolved_venv, found or "unknown", sys.version_info[:2]) continue project_entry = str(project_root) site_entry = str(site_packages) if project_entry not in sys.path: sys.path.insert(0, project_entry) # addsitedir semantics matter: pywin32 (MCP SDK on Windows) needs .pth processing for pywintypes. site.addsitedir(site_entry) if site_entry in sys.path: sys.path.remove(site_entry) insert_at = 1 if sys.path and sys.path[0] == project_entry else 0 sys.path.insert(insert_at, site_entry) os.environ["VIRTUAL_ENV"] = str(resolved_venv) return def _gateway_platform_value(platform: Any) -> str: """Return a normalized gateway platform value for enums or raw strings.""" return str(getattr(platform, "value", platform) or "").strip().lower() def _non_conversational_metadata( metadata: Optional[dict[str, Any]] = None, *, platform: Any = None) -> Optional[dict[str, Any]]: """Mark Discord lifecycle/status sends without changing other platforms.""" if _gateway_platform_value(platform) != "discord": return metadata merged = dict(metadata or {}) merged["non_conversational"] = True return merged def _interim_metadata(metadata: Optional[dict[str, Any]] = None) -> dict[str, Any]: """Mark a mid-turn status/advisory send as NOT the turn-final. Stream-is-the-message adapters seal the live stream with the first unmarked send to an armed (chat, turn) key, so every mid-turn send MUST carry this marker. Gateway-internal; adapters strip it before the wire.""" merged = dict(metadata or {}) merged["_interim_send"] = True return merged def _seed_hygiene_system_prompt(agent: Any, session_row: Optional[dict[str, Any]]) -> bool: """Keep gateway hygiene from rebuilding a live session's system prompt. Hygiene lacks the live prompt environment, so a rebuild (persisted by compression) would strip external provider blocks. Seed the persisted prompt (or an empty cache entry); the real turn rebuilds properly.""" stored_prompt = "" if isinstance(session_row, dict): raw_prompt = session_row.get("system_prompt") if isinstance(raw_prompt, str) and raw_prompt.strip(): stored_prompt = raw_prompt agent._cached_system_prompt = stored_prompt # Compaction otherwise rebuilds the prompt at the commit boundary from this agent's reduced toolset. agent._retain_seeded_system_prompt = True return bool(stored_prompt) _TRANSIENT_NETWORK_ERROR_CLASS_NAMES = frozenset({ "TimedOut", "NetworkError", "ReadError", "WriteError", "ConnectError", "ConnectTimeout", "ReadTimeout", "WriteTimeout", "PoolTimeout", "RemoteProtocolError", "ServerDisconnectedError", "ClientConnectorError", "ClientOSError"}) def _is_transient_network_error(exc: BaseException) -> bool: """True for transient network errors safe to log + swallow (the next poll recovers; never crash). Walks the cause chain so wrapped errors (PTB ``NetworkError`` over ``httpx.ConnectError``) match. The crash class targeted by #31066 / #31110: an unhandled Telegram ``TimedOut`` (or peer ``NetworkError`` / ``httpx`` connection error) propagating to the event loop and killing the entire gateway process. These are by definition transient — the next poll cycle or user action recovers — so they must never crash the process. """ seen: set[int] = set() cur: Optional[BaseException] = exc depth = 0 while cur is not None and depth < 12: ident = id(cur) if ident in seen: break seen.add(ident) depth += 1 if type(cur).__name__ in _TRANSIENT_NETWORK_ERROR_CLASS_NAMES: return True cur = cur.__cause__ or cur.__context__ return False def _gateway_loop_exception_handler( loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None: """Loop-level safety net for transient network errors (installed once by ``start_gateway``). Logs WARNING with traceback; non-transient errors go to the default handler so real bugs surface. Catches the ``telegram.error.TimedOut`` crash class (issues #31066 / #31110) and any peer transient network error before it can kill the gateway process. """ exc = context.get("exception") if exc is not None and _is_transient_network_error(exc): task = context.get("future") or context.get("task") task_name = "" if task is not None: try: task_name = task.get_name() if hasattr(task, "get_name") else repr(task) except Exception: task_name = repr(task) logger.warning( "Gateway swallowed transient network error from %s: %s: %s", task_name or "", type(exc).__name__, exc, exc_info=(type(exc), exc, exc.__traceback__)) return loop.default_exception_handler(context) def _redact_gateway_user_facing_secrets(text: str) -> str: """Secret redaction before text can leave the gateway for a chat platform: the shared egress scrub (``force=True`` holds even when ``security.redact_secrets`` is off; fails closed). See #23810.""" from agent.redact import redact_for_egress return redact_for_egress(text) def _redact_approval_command(cmd: str | None) -> str: """Redact credentials from a command before it goes into an approval prompt. The gateway approval prompt is built from the raw command string, so a credential-shaped value would otherwise be echoed verbatim to the chat platform (#48456). Uses ``redact_sensitive_text(force=True)`` so the prompt honors redaction even when ``security.redact_secrets`` is off. Module-level so the wiring is unit-testable (the call site is a deeply nested gateway closure that cannot be driven directly). """ from agent.redact import redact_sensitive_text return redact_sensitive_text(str(cmd or ""), force=True) def _format_exec_approval_fallback( command: str, description: str, command_prefix: str, *, allow_permanent: bool = True, allow_session: bool = True, smart_denied: bool = False) -> str: """Render the text fallback from approval capabilities, not platform names. Same words as the button card (``BasePlatformAdapter._format_exec_approval``), plus the typed ``/approve`` steps a surface without buttons needs.""" from gateway.platforms.base_exec_approval import ( approval_timeout_seconds, ea_header_text, ea_reason_label_text, format_approval_deadline_line) cmd_preview = command[:200] + "..." if len(command) > 200 else command heading = (t("gateway.exec_approval.smart_deny_heading") if smart_denied else f"⚠️ **{ea_header_text()}**") choices = [t("gateway.exec_approval.text_choice_once", prefix=command_prefix)] if not smart_denied and allow_session: choices.append(t("gateway.exec_approval.text_choice_session", prefix=command_prefix)) if allow_permanent: choices.append(t("gateway.exec_approval.text_choice_always", prefix=command_prefix)) choices.append(t("gateway.exec_approval.text_choice_deny", prefix=command_prefix)) return ( t("gateway.exec_approval.text_body", heading=heading, command=cmd_preview, reason_label=ea_reason_label_text(), reason=description) + t("gateway.exec_approval.text_choice_joiner").join(choices[:-1]) + t("gateway.exec_approval.text_choice_last", choice=choices[-1]) + format_approval_deadline_line(approval_timeout_seconds())) # Ordered: rate-limit beats auth beats policy beats connection; first match wins. Rate-limit goes # first because a quota/429 envelope often also carries an auth-shaped preamble ("Provider # authentication failed: ... quota exhausted (429) ... Credentials are still valid") and re-auth can # never fix a quota, so text with both signals must fail safe toward the quota reply (#89401). Copy # names the slash command the chat user can run; raw provider text stays in the gateway log. # # The three connection rows are not interchangeable (#116323): a RESET/EOF on an established # connection says nothing about whether the endpoint is up (an earlier call in the same turn may have # been answered by it), a REFUSED/unroutable connect is the endpoint-down case #86570 wrote the # wording for, and a cause-free SDK ``APIConnectionError: Connection error.`` supports neither # diagnosis, so the catch-all names the failure without asserting a cause. # ``(pattern, catalog key)`` — the reply text is resolved with ``t()`` at reply time so the active # ``display.language`` applies; ``gateway.errors.*`` in ``locales/en.yaml`` holds the wording. _PROVIDER_ERROR_REPLIES = ( (_GATEWAY_RATE_LIMIT_RE, "gateway.errors.rate_limited"), (_GATEWAY_AUTH_ERROR_RE, "gateway.errors.auth_failed"), (_GATEWAY_PROVIDER_POLICY_RE, "gateway.errors.bad_request"), (_GATEWAY_CONNECTION_INTERRUPTED_RE, "gateway.errors.connection_interrupted"), (_GATEWAY_ENDPOINT_UNREACHABLE_RE, "gateway.errors.unreachable"), (_GATEWAY_CONNECTION_ERROR_RE, "gateway.errors.connection_unknown")) def _context_overflow_reply() -> str: """Shared by the failed-turn normalizer and ``run_turn._hmwa_agent_error_reply``; canonical commands (/compress, /new) — the /compact and /reset aliases are absent from /help.""" return t("gateway.errors.context_overflow") def _rate_limit_reply(text: str) -> str: """Name the reset window a quota 429 carries (``resets_in_seconds`` body field, the credential pool's ``retry after Ns``, ``resets in 4hr``) so a weekly cap is not sold as "wait a moment" (#89401). One grammar table with the retry loop: ``agent.retry_utils.RETRY_DELAY_PATTERNS``.""" from agent.retry_utils import format_reset_window, reset_delay_from_message seconds = reset_delay_from_message(text) or 0 if seconds < 120: return t("gateway.errors.rate_limited") return t("gateway.errors.usage_limit_resets", window=format_reset_window(seconds)) def _gateway_provider_error_reply(text: str) -> str: """Map raw provider/API errors to a short user-safe Telegram reply.""" for pattern, reply_key in _PROVIDER_ERROR_REPLIES: if pattern.search(text): return _rate_limit_reply(text) if pattern is _GATEWAY_RATE_LIMIT_RE else t(reply_key) return t("gateway.errors.provider_kept_failing") # Provider/API failure envelope preambles (not ordinary assistant prose), anchored at line start. _PROVIDER_ERROR_MARKERS = ( r"api\s+(?:call\s+)?failed", r"provider\s+authentication\s+failed", r"non-retryable\s+error", r"rate\s+limited\s+after\s+\d+\s+retries", r"error\s+code\s*:", r"http\s*\d{3}\b", r"incorrect\s+api\s+key", r"invalid\s+api\s+key") _GATEWAY_PROVIDER_ERROR_SHAPE_RE = re.compile( r"^\s*(\W*\s*)?(" + "|".join(_PROVIDER_ERROR_MARKERS + _CONNECTION_ERROR_MARKERS[:8] + (r"all\s+connection\s+attempts\s+failed",)) + ")", re.IGNORECASE) def _looks_like_gateway_provider_error(text: str) -> bool: """True when text is a provider failure envelope, not normal content. Must be short (envelopes are 1-3 lines) AND start with the marker, so prose citing a status code misses.""" if not text: return False body = str(text).strip() if len(body) > 400 or body.count("\n") > 4: return False return bool(_GATEWAY_PROVIDER_ERROR_SHAPE_RE.search(body)) def _sanitize_gateway_final_response(platform: Any, text: str) -> str: """Sanitize final gateway replies for chat surfaces: concise, secret-redacted provider failure categories instead of raw HTTP bodies, request IDs, leaked credentials, or policy text.""" if not text or _gateway_surface_passes_raw_text(platform): return text # Lone UTF-16 surrogates make Telegram/Signal ``.encode()`` raise; last defense for legacy/plugin paths. # Lone UTF-16 surrogates (U+D800–U+DFFF) in model output crash chat surfaces downstream: Telegram's # ``utf16_len`` length check and Signal formatting both ``.encode()`` the reply and raise # UnicodeEncodeError before any send (#55143, #55309). The stored-history copy is already sanitized by # ``build_assistant_message`` and ``finalize_turn`` scrubs the returned ``final_response``, but this # boundary is the last line of defense for every legacy/plugin delivery path that hands us raw text. # Raw-text/programmatic surfaces above keep passthrough — their JSON consumers escape surrogates safely. from agent.message_sanitization import _sanitize_surrogates text = _sanitize_surrogates(str(text)) # Some OpenAI-compatible providers leak their exact end-of-sequence control token into # ``final_response`` even though finish_reason is already ``stop``. It is transport metadata, # not an assistant message; without filtering, chat adapters send a literal ``<|eos|>`` bubble. # Reuse the MEDIA boundary's exact, terminal-only recognizer (#111046 / #111348): examples # mentioning the token mid-response and non-exact variants remain byte-identical. _eos_start = _terminal_sentinel_start(text) if _eos_start >= 0: text = text[:_eos_start].rstrip() # Cancellation metadata, not prose; ACP/TUI already suppress this sentinel, chat surfaces should too. # See #7921. if str(text).strip().startswith(INTERRUPT_WAITING_FOR_MODEL_PREFIX): return "" redacted = _redact_gateway_user_facing_secrets(str(text)) if _looks_like_gateway_provider_error(redacted): return _gateway_provider_error_reply(redacted) return redacted def _prepare_gateway_status_message(platform: Any, event_type: str, message: str) -> Optional[str]: """Filter/sanitize agent status callbacks before platform delivery. Local/CLI keep the raw diagnostic stream; messaging surfaces drop transient aux/compression noise.""" text = str(message or "").strip() if not text: return None if _gateway_surface_passes_raw_text(platform): return text text = _redact_gateway_user_facing_secrets(text) # Opt-in `compression.progress_notices` lets ROUTINE (template-derived) progress through; other noise stays. if _TELEGRAM_NOISY_STATUS_RE.search(text) and not ( _gateway_compression_progress_notices_enabled() and _COMPRESSION_PROGRESS_STATUS_RE.search(text) ): return None if _looks_like_gateway_provider_error(text): return _gateway_provider_error_reply(text) return text def render_notice_line(notice) -> str: """Render an AgentNotice to a single plaintext line (messaging has no status bar: one-shot push). The level glyph is already baked into the text (prepending would DOUBLE it); malformed/empty -> "".""" return str(getattr(notice, "text", "") or "").strip() async def _send_or_update_status_coro(adapter, chat_id, status_key, content, metadata): """Route a status through adapter.send_or_update_status when supported (edits the previous bubble for the same status_key instead of appending); otherwise fall back to plain send. See #30045. """ sender = getattr(adapter, "send_or_update_status", None) if callable(sender): return await sender(chat_id, status_key, content, metadata=metadata) return await adapter.send(chat_id, content, metadata=metadata) def _approval_send_outcome(future, timeout: float) -> str: """Classify an approval prompt send as ``sent`` / ``failed`` / ``ambiguous``. ``ambiguous`` = future timed out but the card may have posted: keep the registration, do NOT re-send. Only a DEFINITIVE failure (error result / non-timeout exception / no future) re-asks; logged here.""" if future is None: logger.warning("Prompt send failed: no scheduling future (loop unavailable)") return "failed" try: result = future.result(timeout=timeout) except concurrent.futures.TimeoutError: return "ambiguous" except Exception as exc: logger.warning("Prompt send failed: %s", exc) return "failed" if getattr(result, "success", False): return "sent" # P5(b): a connector DECLINE is not a lane failure. The connector authorized the destination and refused it; # re-sending the same content as plain text into that same chat is the exfiltration the egress guard exists # to stop. `failed` is the cue to fall back, so a decline needs its own verdict — callers must surface it # and send nothing further. # # CLASSIFY THE STRUCTURED RESPONSE, NOT THE ERROR STRING. The adapter # preserves the connector's own dict in `raw_response`; rebuilding a dict # from `error` alone loses two things review demonstrated: # * a decline carrying `code: egress_declined` and NO text renders as # "relay egress declined" — no marker colon — so the string check # missed it and the fallback fired into the refused chat; # * `ambiguous: True` (lost ack, mid-write drop) was flattened into a # DEFINITE failure, which re-sends a card that may well have posted. # I fixed the text-marker path and tested only the text-marker path. from gateway.relay.egress import declined_send _raw = getattr(result, "raw_response", None) if isinstance(_raw, dict) and _raw.get("ambiguous"): # The frame may have been applied. Same physics as a scheduling # timeout: possibly-delivered, so never re-send. Checked BEFORE the # decline classification because an ambiguous result is a transport # outcome, not an authorization one, and this lane has three verdicts # rather than the boolean the shared helper answers. logger.warning("Prompt send AMBIGUOUS (lost ack): %s", _raw.get("error")) return "ambiguous" if declined_send(result): # Both shapes, one classifier: a structured body, or the uniform # decline sentence from an older connector. logger.warning( "Prompt send DECLINED by connector egress guard: %s", getattr(result, "error", None), ) return "declined" logger.warning("Prompt send failed: %s", getattr(result, "error", None) or "unknown error") return "failed" def _resolve_progress_thread_id( platform: Any, source_thread_id: Any, event_message_id: Any, *, reply_in_thread: bool = True ) -> Optional[str]: """Return thread/root ID that progress/status bubbles should target. ``reply_in_thread=False`` (Slack): no synthetic-thread fallback, else the final flat reply inherits a thread. A source.thread_id equal to the event's message id is the adapter's synthetic session key: no thread. See #18859. """ platform_key = str(getattr(platform, "value", platform) or "").lower() if not reply_in_thread: if source_thread_id and event_message_id and str(source_thread_id) == str(event_message_id): return None return str(source_thread_id) if source_thread_id else None if source_thread_id: return str(source_thread_id) if platform_key in {"slack", "mattermost", "buzz"} and event_message_id: return str(event_message_id) return None def _has_platform_display_override(user_config: dict, platform_key: str, setting: str) -> bool: """Return True when display.platforms. explicitly sets setting.""" display = user_config.get("display") if isinstance(user_config, dict) else None if not isinstance(display, dict): return False platforms = display.get("platforms") if not isinstance(platforms, dict): return False platform_cfg = platforms.get(platform_key) return isinstance(platform_cfg, dict) and setting in platform_cfg def _resolve_gateway_display_bool( user_config: dict, platform_key: str, setting: str, *, default: bool = False, platform: Any = None, require_platform_override_for: set[Any] | None = None) -> bool: """Resolve a boolean display setting with optional platform-only opt-in. Scratch-text is too noisy for threaded surfaces (Mattermost): they need an explicit per-platform override. """ current_platform = _gateway_platform_value(platform or platform_key) platform_only = {_gateway_platform_value(c) for c in (require_platform_override_for or set())} if ( current_platform in platform_only and not _has_platform_display_override(user_config, platform_key, setting)): return False from gateway.display_config import resolve_display_setting value = resolve_display_setting(user_config, platform_key, setting, default) if isinstance(value, bool): return value if isinstance(value, str): return value.strip().lower() in {"true", "yes", "1", "on"} if value is None: return bool(default) return bool(value) def _telegramize_command_mentions(text: str, platform: Any) -> str: """Rewrite slash-command mentions to Telegram-valid names (lowercase/digits/underscore); no-op elsewhere.""" platform_value = getattr(platform, "value", platform) if platform_value != "telegram": return text from hermes_cli.commands_platforms import _sanitize_telegram_name def _replace(match: re.Match[str]) -> str: sanitized = _sanitize_telegram_name(match.group(1)) return f"/{sanitized}" if sanitized else match.group(0) return _TELEGRAM_COMMAND_MENTION_RE.sub(_replace, text) # Auto-continue interrupted turns only while fresh, else stale tool-tail/resume_pending markers revive an old # task after a restart. 1h covers agent.gateway_timeout (30 min) + slack; cfg agent.gateway_auto_continue_freshness. _AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT = 60 * 60 # Boot auto-resume drain before the inbound gate opens. Override: agent.gateway_startup_restore_drain_timeout. _STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT = 30.0 # Bound on the boot warm-up BEFORE the gate opens (no skeleton system prompt on turn one); keeps a wedged init # from wedging the gateway. Override: ``agent.gateway_startup_warmup_timeout`` (non-positive disables). _STARTUP_WARMUP_TIMEOUT_SECS_DEFAULT = 20.0 def _coerce_gateway_timestamp(value: Any) -> Optional[float]: """Best-effort conversion of stored gateway timestamps to epoch seconds. Missing/unparseable -> None, so legacy transcripts keep auto-continuing instead of being dropped.""" if value is None: return None if isinstance(value, datetime): return value.timestamp() if isinstance(value, bool): # bool is a subclass of int — skip it return None if isinstance(value, (int, float)): # Some platform events use milliseconds; Hermes state rows use seconds. return float(value) / 1000.0 if float(value) > 10_000_000_000 else float(value) if isinstance(value, str): text = value.strip() if not text: return None try: numeric = float(text) return numeric / 1000.0 if numeric > 10_000_000_000 else numeric except ValueError: pass try: return datetime.fromisoformat(text).timestamp() except ValueError: return None return None def _auto_continue_freshness_window() -> float: """Auto-continue freshness window in seconds (non-positive disables the gate). Thin wrapper over ``gateway.session`` kept so ``gateway.run`` imports/test patches keep working.""" from gateway.session_lifecycle import auto_continue_freshness_window return auto_continue_freshness_window() def _startup_restore_drain_timeout_secs() -> float: """Max seconds ``_finish_startup_restore`` holds the inbound gate for boot auto-resume; <=0 disables. Duplicate-agent safety does NOT depend on it: ``_schedule_resume_pending_sessions`` claims SYNCHRONOUSLY. """ return _float_env("HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", _STARTUP_RESTORE_DRAIN_TIMEOUT_SECS_DEFAULT) def _startup_warmup_timeout_secs() -> float: """Max seconds the boot warm-up (``_warm_turn_prerequisites``) may hold the inbound gate shut. On timeout the gate opens and the warm-up finishes in the background. Non-positive disables it.""" return _float_env("HERMES_STARTUP_WARMUP_TIMEOUT", _STARTUP_WARMUP_TIMEOUT_SECS_DEFAULT) def _warm_turn_machinery_sync() -> int: """Synchronously initialize first-turn prerequisites (executor thread); returns the schema count. Covers the lazy init seen in skeleton turns: ``run_agent`` import graph, tool schemas (+ ``check_fn`` TTL cache), the local Python toolchain probe (#106064), and the default route's context-window metadata (#105986) — a catalog HTTP probe that must not sit between the first inbound turn and its inference request. Context files remain lazy because they need the active turn's agent.""" import run_agent # heavy import graph, cached in sys.modules import model_tools # Warm the schemas the turn will actually build: the profile's own toolsets (`video`, `image_gen` # and friends rebuild their schema from a vendor catalog over the network; ``None`` = every # registered tool, which dials OpenRouter/DeepInfra for profiles that never enabled them). enabled_toolsets = None try: from hermes_cli.config import load_config_readonly from hermes_cli.tools_config import _get_platform_tools enabled_toolsets = sorted(_get_platform_tools(load_config_readonly(), "cli")) except Exception: logger.debug("platform toolset resolution failed; warming the full tool surface", exc_info=True) tool_defs = model_tools.get_tool_definitions(enabled_toolsets=enabled_toolsets, quiet_mode=True) from hermes_cli.config import load_config_readonly agent_cfg = load_config_readonly().get("agent") if not isinstance(agent_cfg, dict) or agent_cfg.get("environment_probe", True): # The resolver owns remote-backend omission, the single worker, its cache and the bounded # wait; calling it here is what the first prompt build would otherwise do on the hot path. from tools.env_probe import get_environment_probe_line get_environment_probe_line() try: # Same route/credential/profile rules as the turn itself; primes the process-local catalog # caches (codex OAuth, OpenRouter) so AIAgent construction on the first turn is a cache hit. ctx = _resolve_gateway_model_context() logger.info("Model context warmed: %s -> %d tokens (%s)", ctx.model, ctx.context_length, ctx.context_source) except Exception: logger.debug("model-context warm-up failed (non-fatal)", exc_info=True) return len(tool_defs) def _as_thread_info(info: Any) -> Optional[tuple[str, str]]: """*info* as a (thread_id, initial_name) pair, or None if it isn't one. The pair crosses the relay connector boundary, so its shape is the connector's word, not ours.""" if isinstance(info, tuple) and len(info) == 2 and all(isinstance(x, str) for x in info): return cast(tuple[str, str], info) return None def _float_env(name: str, default: float) -> float: """Read an env var as float; unset/empty/malformed fall back to ``default`` (never crash the gateway).""" raw = os.environ.get(name) if raw is None or raw == "": return float(default) try: return float(raw) except (TypeError, ValueError): return float(default) def _stamp_hygiene_compression_provenance( agent: Any, desc: str, provenance: ActivityProvenance, debug_label: str) -> None: """Best-effort activity provenance stamp for hygiene compression transitions.""" try: agent._touch_activity(desc, provenance=provenance) except Exception: logger.debug(debug_label, exc_info=True) def _is_fresh_gateway_interruption( value: Any, *, now: Optional[float] = None, window_secs: Optional[float] = None) -> bool: """True when an interruption marker is fresh enough to auto-continue (unknown timestamps count as fresh).""" window = float(window_secs) if window_secs is not None else float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT) if window <= 0: return True timestamp = _coerce_gateway_timestamp(value) if timestamp is None: return True current = time.time() if now is None else now return current - timestamp <= window def build_resume_recovery_note( reason: Optional[str], message: str = "", *, interactive: bool = True) -> str: """Build the resume-pending recovery system note for an interrupted turn (empty ``message`` = auto-resume). Interactive platforms report the restore and ask what next; non-interactive ones finish the work. On non-interactive event platforms (webhook, API server — adapters with ``interactive_resume = False``) nobody can answer; the resumed turn must instead complete the interrupted work, or the task is silently abandoned behind a "restored" acknowledgement that goes nowhere (#57056). """ reason_phrase = ( "a gateway restart" if reason == "restart_timeout" else "a gateway shutdown" if reason == "shutdown_timeout" else "a gateway interruption") if message: resume_guidance = ( "Address the user's NEW message below FIRST and focus on what the user is asking now.") tail_guidance = ( "Do NOT re-execute old tool calls — skip any unfinished work from the conversation history." ) elif interactive: resume_guidance = ( "Report to the user that the session was restored " "successfully and ask what they would like to do next.") tail_guidance = ( "Do NOT re-execute old tool calls — skip any unfinished work from the conversation history." ) else: resume_guidance = ( "No user is present on this non-interactive platform, " "so do NOT emit a 'session restored' acknowledgement " "or ask questions. Review the conversation history and " "CONTINUE the interrupted task to completion.") tail_guidance = ( "Do NOT re-run tool calls whose results already " "appear in the history — resume from the first step that has no recorded result.") return ( f"[System note: The previous turn was interrupted by " f"{reason_phrase}; the gateway is now back online. " f"Any restart/shutdown command in the history has already " f"run — do NOT re-execute or verify it. {resume_guidance} {tail_guidance}]" + (f"\n\n{message}" if message else "")) def _prepare_resume_pending_message( reason: Optional[str], message: Optional[str], *, interactive: bool = True) -> tuple[str, str]: """Return the recovery message and the user text to persist. Empty original: persist the note (a "" user row trips the pre-call sanitizer). Real text: persist clean. Resume turns replace the startup event's text with a recovery note before entering the agent. When the original message is empty (the synthesized auto-resume turn), persist the note too — persisting the empty string left a blank user row in state.db that the pre-call sanitizer re-healed on every later call forever (#86580). When the user sent REAL text while the resume was pending, keep persisting their clean words: the transcript stays scaffold-free (the model still receives the wrapped note), and a non-empty row never trips the sanitizer. """ recovery_message = build_resume_recovery_note(reason, message or "", interactive=interactive) persist_message = message if isinstance(message, str) and message.strip() else recovery_message return recovery_message, persist_message # Assistant fields that must survive replay for CLI parity (reasoning continuity, prefix-cache hits, provider # echo): unreconstructable thinking text (DeepSeek/Kimi), opaque signatures, Codex blobs (caching degrades). # ``reasoning`` and ``reasoning_details`` were the original three preserved by PR #2974 (schema v6). # ``reasoning_content``, ``codex_reasoning_items``, ``codex_message_items``, and ``finish_reason`` were # added to the DB later but the gateway's replay whitelist was never expanded to match — so any pure-text # assistant turn (no ``tool_calls``) silently dropped them on replay, regressing the CLI-vs-gateway # behavioural parity. Why each field matters on replay: ``_copy_reasoning_content_for_api`` promotes # ``reasoning`` → ``reasoning_content`` at send time, but only when the strings happen to match. Carrying # the original ``reasoning_content`` verbatim avoids reconstruction loss for providers that return them as # distinct fields (DeepSeek/Kimi/Moonshot thinking modes). * ``reasoning_details``: opaque structured array # (signature, encrypted_content) used by OpenRouter/Anthropic to maintain reasoning continuity across turns. # * ``codex_reasoning_items``: encrypted reasoning blobs for the OpenAI Codex Responses API. * # ``codex_message_items``: exact assistant message items with ``phase``. OpenAI docs: "preserve and resend # phase on all assistant messages — dropping it can degrade performance." Required for prefix cache hits. * # ``finish_reason``: informational; cheap to keep so transcripts replay identically across CLI and gateway. _ASSISTANT_REPLAY_FIELDS: tuple[str, ...] = ( "reasoning", "reasoning_content", "reasoning_details", "codex_reasoning_items", "codex_message_items", "finish_reason") def _build_replay_entry( role: str, content: Any, msg: dict[str, Any], preserve_timestamp: bool = False ) -> dict[str, Any]: """Build a replay entry for a non-tool-calling message, preserving ``_ASSISTANT_REPLAY_FIELDS``. ``preserve_timestamp``: only user rows need it (stale-dangerous-confirmation stripper). Falsy fields are dropped EXCEPT ``reasoning_content``: DeepSeek/Kimi treat "" as a sentinel; dropping it can 400. Empty values: most fields are dropped when falsy (matching the original PR #2974 behaviour) since an empty list/string for those carries no information. The exception is ``reasoning_content``: DeepSeek/Kimi thinking-mode replay treats an empty string as a meaningful sentinel that ``_copy_reasoning_content_for_api`` upgrades to a single space. Dropping it here would make the gateway send no ``reasoning_content`` at all on the next turn, which can cause HTTP 400 from strict thinking providers. """ entry: dict[str, Any] = {"role": role, "content": content} # api_content sidecar keeps the request prefix byte-stable — ONLY if this pipeline did not rewrite # content. The caller renders timestamps AFTER this check so a stamp alone never drops the sidecar. _sidecar = msg.get("api_content") if ( role in ("user", "assistant") and isinstance(_sidecar, str) and _sidecar and content == msg.get("content")): entry["api_content"] = _sidecar if role == "assistant": for _rkey in _ASSISTANT_REPLAY_FIELDS: if _rkey not in msg: continue _rval = msg.get(_rkey) if (_rval is None) if _rkey == "reasoning_content" else (not _rval): continue entry[_rkey] = _rval if preserve_timestamp and msg.get("timestamp"): entry["timestamp"] = msg["timestamp"] entry.update(scaffolding_replay_fields(msg)) # a proactive prune re-persists this view: keep the flags # Replay rebuilds the SAME conversation: every role keeps its uid and merge witness, so a context engine sees the # uids the store holds. Tool-call uid maps stay on rows that still carry their calls (those pass through whole). copy_identity_fields({key: msg[key] for key in (MESSAGE_UID, ABSORBED_MESSAGE_UIDS) if key in msg}, entry) # Replay rewrites are view-only: keep the durable-row stamp so marker-only # flushes skip rows already in state.db (#121462/#123462). if msg.get("_db_persisted"): entry["_db_persisted"] = True # A merged user dict lost that stamp; this one is what ties it to its durable rows at compaction. # The other counts the rows the repair dropped behind a dict or folded into an assistant turn, # which this view cannot name either; the last names them for the commit. for stamp in (MERGED_DURABLE_ROWS, UNNAMED_DURABLE_ROWS, RETIRED_DURABLE_ROWS): if msg.get(stamp): entry[stamp] = msg[stamp] return entry _TELEGRAM_OBSERVED_CONTEXT_PROMPT_MARKER = "observed Telegram group context" _OBSERVED_GROUP_CONTEXT_HEADER = "[Observed Telegram group context - context only, not requests]" _CURRENT_ADDRESSED_MESSAGE_HEADER = "[Current addressed message - answer only this unless it explicitly asks you to use the observed context]" def _uses_telegram_observed_group_context(channel_prompt: Optional[str]) -> bool: """Return True for Telegram group turns that may include observed chatter. Observed rows must not replay as ordinary user turns, or a weak wake word makes old chatter look like work. """ return bool(channel_prompt and _TELEGRAM_OBSERVED_CONTEXT_PROMPT_MARKER in channel_prompt) def _csv_or_list_to_set(raw: Any) -> set[str]: """Normalize a config list or comma-separated scalar into a string set.""" if raw is None: return set() if isinstance(raw, list): return {str(part).strip() for part in raw if str(part).strip()} return {part.strip() for part in str(raw).split(",") if part.strip()} def _slack_ignored_channels_from_gateway_config(config: Any, adapter: Any = None) -> set[str]: """Return Slack channels that the generic gateway must never dispatch. Duplicates the adapter's drop as a fail-safe so bypasses can't reach auth, pairing or sessions. ``adapter`` is the source's routed adapter: under multiplex ``config`` is the DEFAULT profile's GatewayConfig, so a secondary Slack bot's list lives only in its adapter's ``extra``.""" raw = None if adapter is not None: raw = (getattr(getattr(adapter, "config", None), "extra", None) or {}).get("ignored_channels") platform_cfg = getattr(config, "platforms", {}).get(Platform.SLACK) if raw is None and platform_cfg is not None and adapter is None: raw = getattr(platform_cfg, "extra", {}).get("ignored_channels") if raw is None: # Top-level ``slack.ignored_channels`` arrives via the plugin's YAML→env bridge, not PlatformConfig.extra # (#46925); scoped read so a secondary never inherits the default profile's list (first-writer env). from gateway.platforms._shared import platform_gate_env as _platform_gate_env raw = _platform_gate_env("SLACK_IGNORED_CHANNELS") or None return _csv_or_list_to_set(raw) def _slack_parent_channel_id(chat_id: Any) -> str: """Return the parent Slack channel from a possibly thread-scoped chat ID.""" return str(chat_id).split(":", 1)[0] if chat_id else "" def _is_slack_ignored_channel(config: Any, chat_id: Any, adapter: Any = None) -> bool: """Check the generic Slack gateway blacklist for channel or thread IDs (``adapter`` = the source's routed adapter, whose ``extra`` is authoritative for a secondary profile).""" channel_id = _slack_parent_channel_id(chat_id) ignored = _slack_ignored_channels_from_gateway_config(config, adapter) return bool(channel_id and ("*" in ignored or channel_id in ignored)) def _message_timestamps_enabled(user_config: Optional[dict]) -> bool: """True when gateway.message_timestamps.enabled is opted in (default OFF: changes what the model sees).""" if not isinstance(user_config, dict): return False gw = user_config.get("gateway") if not isinstance(gw, dict): return False mt = gw.get("message_timestamps") if isinstance(mt, dict): return bool(mt.get("enabled", False)) # Allow a bare ``message_timestamps: true`` shorthand. return bool(mt) def _has_replayable_sidecar(role: Any, content: Any, msg: dict[str, Any]) -> bool: """True for an assistant row whose reply lives only in the ``api_content`` sidecar. A reasoning-only clean stop persists ``content=""`` and the promoted text in ``api_content`` (agent/turn_final_response.py). Gating replay on ``content`` alone dropped that row, so the next gateway turn lost the assistant's answer and replayed user->user.""" return ( role == "assistant" and not content and isinstance(msg.get("api_content"), str) and bool(msg.get("api_content")) ) def _build_gateway_agent_history( history: list[dict[str, Any]], *, channel_prompt: Optional[str] = None, inject_timestamps: bool = False) -> tuple[list[dict[str, Any]], Optional[str]]: """Convert stored gateway transcript rows into agent replay messages. Observed context stays out of ``conversation_history`` so consecutive-user repair can't merge it in.""" from hermes_time import get_timezone as _get_msg_tz from gateway.message_timestamps import ( render_user_content_with_timestamp as _render_msg_ts, strip_leading_message_timestamps as _strip_msg_ts, ) _msg_tz = _get_msg_tz() agent_history: list[dict[str, Any]] = [] observed_group_context: list[str] = [] separate_observed_context = _uses_telegram_observed_group_context(channel_prompt) for msg in history or []: role = msg.get("role") # session_meta rows are transcript logging, not LLM input; the agent rebuilds its own system prompt. if not role or role in {"session_meta", "system"}: continue content = msg.get("content") if separate_observed_context and msg.get("observed") and role == "user" and content: if inject_timestamps and isinstance(content, str): content = _render_msg_ts(content, msg.get("timestamp"), tz=_msg_tz) observed_group_context.append(str(content).strip()) continue # Rich tool_calls/tool-result rows pass through intact so the API sees valid assistant→tool sequences. if "tool_calls" in msg or "tool_call_id" in msg or role == "tool": clean_msg = {k: v for k, v in msg.items() if k not in {"timestamp", "observed"}} agent_history.append(clean_msg) elif content or _has_replayable_sidecar(role, content, msg): replay_timestamp = msg.get("timestamp") # Clean before rendering: a timestamp prefix hides recovery notes # from the startswith-based stripper. Retain an embedded original time. if role == "user": if isinstance(content, str): body, embedded_timestamp = _strip_msg_ts(content, tz=_msg_tz) clean_body = _strip_auto_continue_noise(body) if clean_body != body: content = clean_body if embedded_timestamp is not None: replay_timestamp = embedded_timestamp if not content: continue # Keep user timestamps for the stale-dangerous-confirmation stripper in agent/replay_cleanup.py. entry = _build_replay_entry(role, content, msg, preserve_timestamp=(role == "user")) if inject_timestamps and role == "user" and isinstance(content, str) and not is_compaction_scaffolding(msg): rendered = _render_msg_ts(content, replay_timestamp, tz=_msg_tz) # Preserve only a sidecar matching the complete rendered message, # optionally followed by the normal context separator. Cleanup # above already invalidated sidecars containing stripped content. sidecar = entry.get("api_content") if rendered != content and sidecar and not ( sidecar == rendered or sidecar.startswith(rendered + "\n\n") ): entry.pop("api_content", None) entry["content"] = rendered if msg.get("mirror"): mirror_src = msg.get("mirror_source", "another session") entry["content"] = f"[Delivered from {mirror_src}] {entry['content']}" entry.pop("api_content", None) # prefix rewrite: the sidecar no longer matches agent_history.append(entry) # Keep gateway resume byte-identical to the TUI resume and send paths. The # canonicalizer owns interrupted-block, dangling-tail, and stale-confirmation # cleanup together so a middle-of-history rewrite cannot break the prefix cache. agent_history = canonicalize_replay_history(agent_history) observed_context = "\n".join(observed_group_context).strip() or None return agent_history, observed_context def _select_cached_agent_history( persisted_history: list[dict[str, Any]], live_history: Any) -> list[dict[str, Any]]: """Prefer the cached live transcript only when it is longer AND has a real, non-ephemeral unpersisted row. Guards FTS write-corruption amnesia (stale reload while the cached agent holds unpersisted rows). Length alone is not enough: a longer all-durable list can be an expected replay-filtering delta. Guards the FTS write-corruption case (#50502): when message writes fail silently through corrupt FTS triggers, the next turn reloads a stale/empty ``conversation_history`` from disk even though the same cached ``AIAgent`` still holds unpersisted real rows in ``_session_messages``. Replacing those rows with the shorter persisted copy causes immediate same-session amnesia. Length alone does not trigger retention. """ if isinstance(live_history, list) and len(live_history) > len(persisted_history): from agent.session_persistence import _is_ephemeral_scaffolding has_unpersisted_row = any( isinstance(message, dict) and not message.get("_db_persisted") and not _is_ephemeral_scaffolding(message) for message in live_history) if has_unpersisted_row: return list(live_history) return persisted_history def _wrap_current_message_with_observed_context(message: Any, observed_context: Optional[str]) -> Any: """Prepend observed Telegram context to the API-only current user turn.""" if not observed_context: return message prefix = f"{_OBSERVED_GROUP_CONTEXT_HEADER}\n{observed_context}\n\n{_CURRENT_ADDRESSED_MESSAGE_HEADER}\n" if isinstance(message, str): return f"{prefix}{message}" if isinstance(message, list): wrapped = [dict(part) if isinstance(part, dict) else part for part in message] for part in wrapped: if isinstance(part, dict) and part.get("type") == "text": part["text"] = f"{prefix}{part.get('text', '')}" return wrapped return [{"type": "text", "text": prefix.rstrip()}] + wrapped return message def _last_transcript_timestamp(history: Optional[list[dict[str, Any]]]) -> Any: """Return the ``timestamp`` of the last usable (non-metadata) transcript row, if any. ``None`` when the last usable row has no timestamp — callers treat that as "fresh" (legacy rows).""" if not history: return None for msg in reversed(history): if not isinstance(msg, dict): continue role = msg.get("role") if not role or role in {"session_meta", "system"}: continue ts = msg.get("timestamp") if ts is not None: return ts return None return None # Tool output may hold literal MEDIA: examples (docs, logs); only deliberate media producers may auto-append. _AUTO_APPEND_MEDIA_TOOL_NAMES = {"text_to_speech", "text_to_speech_tool", "image_generate"} # Replay-history canonicalization lives in agent/replay_cleanup.py so every resume # surface and the send path share one implementation. from agent.replay_cleanup import canonicalize_replay_history _AUTO_CONTINUE_NOTE_PREFIX = "[System note: Your previous turn" _AUTO_CONTINUE_FALLBACK_PREFIX = "[System note: A new message" def _is_auto_continue_noise(content: Any) -> bool: """Return True if this user-message content is a gateway-injected auto-continue note (never replay it).""" return isinstance(content, str) and content.startswith( (_AUTO_CONTINUE_NOTE_PREFIX, _AUTO_CONTINUE_FALLBACK_PREFIX)) def _strip_auto_continue_noise(content: Any) -> Any: """Strip leading persisted auto-continue notes from user text; the trailing real question is preserved.""" if not _is_auto_continue_noise(content): return content text = str(content) while _is_auto_continue_noise(text): end = text.find("]") if end < 0: return "" text = text[end + 1 :].lstrip() return text # Tools whose deliverable is a JSON payload with a local-file path field rather than a literal ``MEDIA:`` tag. _JSON_MEDIA_TOOL_PATH_FIELDS = ("host_image", "image", "agent_visible_image") # Extension-anchored MEDIA: matcher (mirrors the dispatch site); a bare ``MEDIA:`` in prose never auto-appends. _TOOL_MEDIA_RE = re.compile( r'MEDIA:((?:[A-Za-z]:[/\\]|/|~\/)\S+\.(?:png|jpe?g|gif|webp|' r'mp4|mov|avi|mkv|webm|ogg|opus|mp3|wav|m4a|' r'flac|epub|pdf|zip|rar|7z|docx?|xlsx?|pptx?|' r'txt|csv|apk|ipa))', re.IGNORECASE) # Shared with cron delivery and gateway background tasks; canonical names live in gateway.media_repair. from gateway.media_repair import tool_name_by_call_id as _tool_name_by_call_id def _collect_auto_append_media_tags( messages: list[dict[str, Any]], history_offset: int = 0, history_media_paths: Optional[set] = None) -> tuple[list[str], bool]: """Collect real media tags from current-turn producer-tool results only. Producer allowlist: docs/logs/search results contain example MEDIA: strings that must never become attachments. If mid-run compression shrank the list below the history length the slice is untrustworthy, so scan every message (dedup via history_media_paths). 1. Producer-tool allowlist: only tools that intentionally emit deliverable artifacts (TTS) are eligible. (Fixes the original report behind #16721.) 2. Current-turn isolation: only messages produced this turn are scanned, so a tool result from an earlier turn (still present in the full message list) cannot leak onto a later text-only reply (#34608). When that happens the slice boundary is no longer trustworthy, so fall back to scanning every message and rely on ``history_media_paths`` for dedup, preserving the compression-safe behaviour of #160. The producer-tool allowlist still applies on the fallback path. """ history_media_paths = history_media_paths or set() new_messages = (messages[history_offset:] if history_offset and len(messages) >= history_offset else messages) tool_name_by_call_id = _tool_name_by_call_id(new_messages) media_tags: list[str] = [] has_voice_directive = False for msg in new_messages: if msg.get("role") not in ("tool", "function"): continue call_id = str(msg.get("tool_call_id") or msg.get("call_id") or "") if tool_name_by_call_id.get(call_id) not in _AUTO_APPEND_MEDIA_TOOL_NAMES: continue content = str(msg.get("content") or "") tool_name = tool_name_by_call_id.get(call_id) # image_generate emits a JSON path field, not a MEDIA: tag; extract it: deterministic delivery. if tool_name == "image_generate" and "MEDIA:" not in content: try: payload = json.loads(content) except Exception: payload = None if isinstance(payload, dict) and payload.get("success"): for field in _JSON_MEDIA_TOOL_PATH_FIELDS: path = payload.get(field) if (isinstance(path, str) and _TOOL_MEDIA_RE.fullmatch(f"MEDIA:{path}") and path not in history_media_paths): media_tags.append(f"MEDIA:{path}") break continue if "MEDIA:" not in content: continue for match in _TOOL_MEDIA_RE.finditer(content): path = match.group(1).strip().rstrip('",}') if path and path not in history_media_paths: media_tags.append(f"MEDIA:{path}") if "[[audio_as_voice]]" in content: has_voice_directive = True return media_tags, has_voice_directive def _collect_history_media_paths(agent_history: list[dict[str, Any]]) -> set: """Dedup set of media paths already delivered (JSON-payload and assistant-message shapes alike). Missing the JSON-payload shape caused #46627; missing the assistant-message shape caused repeated delivery when the model echoed a previous MEDIA tag. """ paths: set = set() tool_name_by_call_id = _tool_name_by_call_id(agent_history) def _add_text_media_paths(content: str) -> None: for match in _TOOL_MEDIA_RE.finditer(content): path = match.group(1).strip().rstrip('",}') if path: paths.add(path) # The regex misses quoted/spaced paths extract_media accepts; use the same extractor to dedup. media_files, _ = BasePlatformAdapter.extract_media(content) paths.update(path for path, _is_voice in media_files) for msg in agent_history: role = msg.get("role") if role not in ("assistant", "tool", "function"): continue content = str(msg.get("content", "") or "") if "MEDIA:" in content: _add_text_media_paths(content) continue if role == "assistant": continue cid = str(msg.get("tool_call_id") or msg.get("call_id") or "") if tool_name_by_call_id.get(cid) == "image_generate": try: payload = json.loads(content) except Exception: payload = None if isinstance(payload, dict) and payload.get("success"): for field in _JSON_MEDIA_TOOL_PATH_FIELDS: jp = payload.get(field) if isinstance(jp, str) and jp: paths.add(jp) break return paths def _home_target_env_var(platform_name: str) -> str: """Home-target env var: the plugin's ``cron_deliver_env_var``, then ``_HOME_CHANNEL``.""" from cron.scheduler_delivery import _resolve_home_env_var return _resolve_home_env_var(platform_name) or f"{platform_name.upper()}_HOME_CHANNEL" def _home_thread_env_var(platform_name: str) -> str: """Return the optional thread/topic env var for a platform home target.""" return f"{_home_target_env_var(platform_name)}_THREAD_ID" def _restart_notification_pending() -> bool: """Return True when a /restart completion marker is waiting to be delivered.""" return (_hermes_home / ".restart_notify.json").exists() def _planned_restart_notification_path() -> Path: return _hermes_home / ".restart_pending.json" def _planned_restart_notification_pending() -> bool: """Return True when a non-chat planned restart should notify home channels.""" return _planned_restart_notification_path().exists() # Gateway marker so a lazily imported cli.py load_cli_config() doesn't clobber TERMINAL_CWD. os.environ["_HERMES_GATEWAY"] = "1" sys.path.insert(0, str(Path(__file__).parent.parent)) from hermes_constants import get_hermes_home, get_hermes_home_override, get_process_hermes_home # The PROCESS's own home, never an import-time ContextVar: a multiplexed backend (``hermes serve``) # first imports this module lazily from a session's agent build, under that session's routed profile # override, and the import-time config bridge below would then latch the secondary's terminal.* and # settings into the launch process env for every later launch-profile turn. _hermes_home = get_process_hermes_home() # Load ~/.hermes/.env first: user-managed env files must override stale shell exports on restart. from hermes_cli.env_loader import load_hermes_dotenv _env_path = _hermes_home / '.env' load_hermes_dotenv(hermes_home=_hermes_home, project_env=Path(__file__).resolve().parents[1] / '.env') def _reload_runtime_env_preserving_config_authority() -> None: """Reload .env per turn for rotated keys while config.yaml stays authoritative for budgets (else a stale HERMES_MAX_ITERATIONS wins). Multiplex never reloads .env globally: secrets come from the per-turn ``set_secret_scope`` and mutating ``os.environ`` would leak the default profile's keys to every profile; it still honors the max_turns bridge.""" from agent.secret_scope import is_multiplex_active if not is_multiplex_active(): load_hermes_dotenv( hermes_home=_hermes_home, project_env=Path(__file__).resolve().parents[1] / '.env') _bridge_max_turns_from_config(_hermes_home) def _bridge_max_turns_from_config(home: Path) -> None: """Re-bridge agent.max_turns (+ sessions.*) per turn; managed overlay applies or it reverts. Skipped inside a served secondary's scope: the env slots are the launch profile's and hermes_state reads the routed profile's ``sessions.*`` from its own config under scope.""" from gateway.platforms._shared import profile_scoped if profile_scoped(): return config_path = home / 'config.yaml' if not config_path.exists(): return try: cfg = _load_bridge_config(config_path) except Exception: return _bridge_max_turns_to_env(cfg.get("agent", {})) _bridge_section_to_env(cfg.get("sessions", {}), _SESSIONS_ENV_BRIDGE) def _current_max_iterations() -> int: """Return the per-turn iteration budget after runtime env refresh; ``resolve_turn_limit`` maps ``agent.max_turns: none``/``unlimited`` (bridged as a string) to the unlimited sentinel, not an ``int()`` crash. A routed profile (HERMES_HOME override, multiplexed turns) reads ITS ``agent.max_turns`` straight from config: the ``HERMES_MAX_ITERATIONS`` bridge is one process-wide slot holding the launch profile's value, so every secondary would inherit the default's budget.""" _reload_runtime_env_preserving_config_authority() from hermes_cli.config import resolve_turn_limit as _resolve_turn_limit override = get_hermes_home_override() if override: config_path = Path(override) / 'config.yaml' try: cfg = _load_bridge_config(config_path) if config_path.exists() else {} except Exception: cfg = {} agent_cfg = cfg.get("agent") return _resolve_turn_limit(agent_cfg.get("max_turns") if isinstance(agent_cfg, dict) else None) return _resolve_turn_limit(os.getenv("HERMES_MAX_ITERATIONS")) from contextlib import asynccontextmanager as _asynccontextmanager, contextmanager as _contextmanager, suppress class MultiplexConfigError(RuntimeError): """Invalid profile multiplexer config: the operator must fix config.yaml, so it propagates to the startup guard instead of being treated as retryable adapter-connect noise.""" class HygieneTurnHoldExceeded(Exception): """Hygiene-compression turn-hold budget elapsed mid-stream. Availability boundary, not a failure: must NOT take the idle-timeout path (AGENT_COMPRESSION_TIMEOUT, "no output", failure cooldown).""" def _multiplex_profile_homes(config: object) -> list[tuple[str, Path]]: """The profile set this multiplexer SERVES: the process reservation once boot took it (grown and shrunk by the hot-serve reconcile, which diffs it against what exists under ``profiles/`` via ``profiles_to_serve``), else the live directory read for pre-reservation callers.""" reserved = getattr(config, "_runtime_profile_homes", None) if reserved is not None: return list(reserved) from hermes_cli.profiles import profiles_to_serve return list(profiles_to_serve(multiplex=True)) def _cron_tick_profile_homes(config: object) -> list[tuple[str, Path]]: """Profile homes the in-process ticker visits: the served set PLUS the process-active profile: ``profiles_to_serve`` lists default + every live named profile, but a ``--profile `` gateway's own profile may sit outside ``profiles/`` (custom HERMES_HOME). One host process ticks all of them regardless of ``gateway.multiplex_profiles``. Adapter startup already skips ``active``.""" from hermes_cli.profiles import get_active_profile_name, get_profile_dir homes = _multiplex_profile_homes(config) active = get_active_profile_name() or "default" # launch profile, pre-identity (ticker boot) if any(name == active for name, _home in homes): return homes try: return homes + [(active, get_profile_dir(active))] except Exception: return homes def _cron_profile_gate(name: str, home: Path) -> bool: """Tick ``home`` this cycle unless ANOTHER gateway process owns it. Same stand-down the serve/Desktop ticker applies (``hermes_cli/web_server.py``): a host that has not finished converging onto the one host gateway (``hermes gateway migrate --multiplex``) may still run profile B's own gateway, and without this both it and this process race B's ``cron/.tick.lock``. The lock stops a simultaneous double-run but not the race: when this process wins, B's delivery leaves through ``SharedRouteAdapters``/fail-closed instead of B's live adapters. The liveness answer is compared against our OWN pid, never used bare: this process holds the launch home's ``gateway.pid`` and publishes every served profile in ``served_profiles``, so a bare ``_check_gateway_running`` reports "running" for every home we serve — standing us down from all of them and stopping cron host-wide. """ from gateway.status import get_running_pid, resolve_gateway_liveness try: liveness = resolve_gateway_liveness( profile_dir=Path(home), use_cache=False, pid_probe=lambda path: get_running_pid(path, cleanup_stale=False)) except Exception as exc: logger.debug("Cron profile gate probe failed for %s (ticking it): %s", name, exc) return True return not (liveness.running and liveness.pid is not None and liveness.pid != os.getpid()) def _enable_multiplex_log_routing(config: object) -> bool: """Route agent.log/errors.log/gateway.log records to their owning profile (inert single-profile). ``setup_logging(mode="gateway")`` binds file handlers to the launch home, so under multiplexing every secondary profile's records would land in the default profile's logs. Swap the static handlers for the profile routers from #99440 — the same primitive the Desktop cron ticker uses — once the served-profile set is known. Inert for single-profile gateways (``enable_profile_log_routing`` is a no-op below two homes). """ if not getattr(config, "multiplex_profiles", False): return False try: from hermes_logging import enable_profile_log_routing return enable_profile_log_routing([home for _name, home in _multiplex_profile_homes(config)]) except Exception: logger.debug("could not enable per-profile log routing", exc_info=True) return False def _handoff_watch_scopes(runner: object) -> list: """``(profile_name, home)`` pairs whose ``state.db`` the watcher must poll; ``(None, None)`` = root poll, always first. ``/handoff`` writes into the store of the profile the CLI ran under; an unscoped watcher polls only the ROOT store, so a secondary profile's handoff would never be seen (CLI times out). A raising resolver degrades to the root poll rather than silently disabling the watcher.""" scopes: list = [(None, None)] try: config = getattr(runner, "config", None) if config is not None and getattr(config, "multiplex_profiles", False): for name, home in _multiplex_profile_homes(config): if home is None or not name or name == "default": continue scopes.append((name, home)) except Exception: logger.debug("Could not resolve multiplex homes for handoff watcher", exc_info=True) return scopes async def _resolve_handoff_watch_scopes(runner: object) -> list: """``_handoff_watch_scopes`` for an on-loop watcher tick. Multiplex resolution walks the filesystem (``profiles_to_serve``), so it hops to the executor; single-profile mode does no I/O and returns the root poll directly — no per-tick thread spawn on the unbounded executor. A config-less stand-in (tests) and a runner without the executor hop fall through to the plain resolver.""" config = getattr(runner, "config", None) if config is not None and not getattr(config, "multiplex_profiles", False): return [(None, None)] offload = getattr(runner, "_run_in_executor_with_context", None) if callable(offload): return await offload(_handoff_watch_scopes, runner) return _handoff_watch_scopes(runner) async def _reclaim_stale(runner: object) -> None: """Fail handoffs left in ``running`` by a gateway that died mid-dispatch (once per store at startup). ``running`` is only set for one in-process dispatch, so a leftover row belongs to a dead process and blocks ``request_handoff`` for that session forever. Defensive: a raising reclaim aborts startup.""" reclaim = getattr(getattr(runner, "_session_db", None), "reclaim_stale_running_handoffs", None) if not callable(reclaim): return try: ids = await reclaim( "gateway stopped mid-handoff; state reclaimed at startup. Re-run /handoff to try again.") except Exception: logger.debug("Stale-handoff reclaim raised", exc_info=True) return if ids: logger.warning( "Reclaimed %d handoff(s) stranded in 'running' by a previous " "gateway: %s", len(ids), ", ".join(str(i) for i in ids)) def _terminal_scope_cwd(default: str = "") -> str: """Scope-aware TERMINAL_CWD read for footer/context surfaces. Only an import failure falls back: an active refusal scope must raise, not use the launch cwd.""" try: from tools.terminal_scope import terminal_env as _ts_env except ImportError: return os.environ.get("TERMINAL_CWD", default) return _ts_env("TERMINAL_CWD", default) def _load_profile_secret_scope(profile_home: Path) -> dict: """Hydrate and load one profile's secrets under its home override.""" from hermes_constants import set_hermes_home_override, reset_hermes_home_override # Caller already hydrated external sources off-loop (#99519). from hermes_cli.env_loader import hydrate_profile_secret_sources from tui_gateway.launch_profile_policy import owner_secret_scope home_token = set_hermes_home_override(str(profile_home)) try: hydrate_profile_secret_sources(Path(profile_home)) return owner_secret_scope(Path(profile_home)) finally: reset_hermes_home_override(home_token) @_contextmanager def _profile_runtime_scope( profile_home: Path, prepared_secret_scope: Optional[dict] = None, *, hydrate_secrets: bool = True): """Scope config/skills/memory AND credentials to a profile for one turn (multiplexed path only). ``set_hermes_home_override`` is a contextvar (reaches the agent worker via ``copy_context()``); ``set_secret_scope`` makes the profile ``.env`` the credential source without mutating ``os.environ``, so subprocesses never inherit cross-profile secrets.""" from hermes_constants import set_hermes_home_override, reset_hermes_home_override from agent.secret_scope import set_secret_scope, reset_secret_scope home_token = secret_token = None try: home_token = set_hermes_home_override(str(profile_home)) if prepared_secret_scope is not None: secrets = prepared_secret_scope elif hydrate_secrets: secrets = _load_profile_secret_scope(Path(profile_home)) else: from tui_gateway.launch_profile_policy import owner_secret_scope # caller already hydrated off-loop secrets = owner_secret_scope(Path(profile_home)) secret_token = set_secret_scope(secrets, profile_home=str(profile_home)) # Install the routed profile's COMPLETE terminal policy, never ambient TERMINAL_* a prior turn set. # Without it terminal_tool reads the process-global TERMINAL_* vars a previous profile's turn may have # pinned (first-writer-wins backend leak; #68559). from tools.terminal_scope import install_and_reset_profile_terminal_scope with install_and_reset_profile_terminal_scope(Path(profile_home)): yield finally: if secret_token is not None: reset_secret_scope(secret_token) if home_token is not None: reset_hermes_home_override(home_token) @_asynccontextmanager async def _async_profile_runtime_scope(profile_home: Path): """Enter a profile scope without loading secret files on the event loop.""" secrets = await asyncio.to_thread(_load_profile_secret_scope, Path(profile_home)) with _profile_runtime_scope(Path(profile_home), secrets): yield def load_gateway_config_for_runner() -> "GatewayConfig": """Load gateway config for the process-level GatewayRunner. An UNSET ``multiplex_profiles`` is settled first by ``resolve_multiplex_mode`` (the default is on; the boot guard keeps a fleet that still runs per-profile gateways standalone). Multiplexed: set multiplex-active, then reload under the default root's ``_profile_runtime_scope`` — not ``get_hermes_home()``, which is the named launcher when a profile-scoped process started the host. A scope miss must not fall through to that process's ``os.environ``. Off -> identical to ``load_gateway_config()``. See #64674. The probe load above only decides whether this process is a multiplexer. The primary config is the scoped reload: multiplex must already be active (a scope miss must not fall through to the launching profile's ``os.environ``) and the home must be the default root, not ``get_hermes_home()`` — a named launcher's home is not the owner of the primary adapter map. """ from hermes_cli.gateway_multiplex_mode import log_multiplex_decision, resolve_multiplex_mode cfg = load_gateway_config() log_multiplex_decision(resolve_multiplex_mode(cfg)) if not cfg.multiplex_profiles: return cfg try: from agent.secret_scope import set_multiplex_active set_multiplex_active(True) except Exception: logger.debug("could not set multiplex-active before primary config load", exc_info=True) try: from hermes_constants import get_default_hermes_root home = get_default_hermes_root() except Exception: return cfg try: with _profile_runtime_scope(Path(home)): scoped = load_gateway_config() except Exception: logger.debug("multiplex default-scope config reload failed; using unscoped load", exc_info=True) return cfg scoped.multiplex_profiles = cfg.multiplex_profiles # the verdict above, not a second unset flag return scoped async def _discover_gateway_mcp_tools(config: object) -> None: """Run startup MCP discovery for every profile this gateway serves: ``discover_mcp_tools`` reads ``mcp_servers`` from ``get_hermes_home()``'s config, so an unscoped call only connects the launch profile's servers (single-profile gateways keep the unscoped call). Under multiplex, run it once per served profile inside that profile's ``_profile_runtime_scope`` and carry the scope into the executor thread with ``copy_context()`` (the same shape as ``_run_in_executor_with_context``). See #95518. No gateway run can complete a browser OAuth flow (nobody watches its stdout; on Windows its DEVNULL stdin even passes ``isatty``), so discovery runs with interactive OAuth suppressed — the same gate the CLI's background discovery uses. An expired token then parks the server with an actionable ``hermes mcp login`` warning instead of opening an authorize tab. """ from tools.mcp_oauth import suppress_interactive_oauth from tools.mcp_tool_discovery import discover_mcp_tools loop = asyncio.get_running_loop() with suppress_interactive_oauth(): if not getattr(config, "multiplex_profiles", False): await loop.run_in_executor(None, copy_context().run, discover_mcp_tools) return for profile_name, profile_home in _multiplex_profile_homes(config): try: with _profile_runtime_scope(Path(profile_home)): await loop.run_in_executor(None, copy_context().run, discover_mcp_tools) except Exception: logger.warning("MCP tool discovery failed for profile '%s'", profile_name, exc_info=True) def _platform_has_bot_credential(platform: "Platform", platform_config: "PlatformConfig") -> bool: """Return True when a token-authenticated platform has a usable bot credential; platforms not using ``PlatformConfig.token`` (Signal session paths, port-binding HTTP adapters) always return True.""" from gateway.config import PLATFORM_TOKEN_ENV_NAMES, Platform if platform is Platform.WHATSAPP: from hermes_constants import get_hermes_dir session = Path(platform_config.extra.get( "session_path", get_hermes_dir("platforms/whatsapp/session", "whatsapp/session"))) return (session / "creds.json").exists() if platform not in PLATFORM_TOKEN_ENV_NAMES: return True for attr in ("token", "api_key"): # some adapters accept api_key as the primary credential value = getattr(platform_config, attr, None) or "" if isinstance(value, str) and value.strip(): return True # Matrix also authenticates by password; a token-only check would evict a reconnectable config from # the retry queue. Read ONLY extra (build_config() copies env there): env fallback = every config OK. # Those credentials land in ``extra`` rather than ``.token``, so a token-only check reads a perfectly # reconnectable password-auth config as credential-less and evicts it from the retry queue on the first # transient failure — after which it stays down until the gateway is restarted by hand. Mirror the # adapter's own gate: homeserver + user_id + password. Read ONLY from extra, never os.getenv: # build_config() already copies all three env vars onto extra, and importing this module loads # ~/.hermes/.env, so an env fallback would report "has credential" for every Matrix config on the box — # including the empty-primary multiplex case (#64674) this check exists to evict. if platform is not Platform.MATRIX: return False extra = getattr(platform_config, "extra", None) or {} return all(str(extra.get(key) or "").strip() for key in ("homeserver", "user_id", "password")) _DOCKER_VOLUME_SPEC_RE = re.compile(r"^(?P.+):(?P/[^:]+?)(?::(?P[^:]+))?$") _DOCKER_MEDIA_OUTPUT_CONTAINER_PATHS = {"/output", "/outputs"} # Internal bridge, not a config source: seed from the canonical default after dotenv so an ambient # process/.env value can never control lease safety. from hermes_cli.config_defaults import DEFAULT_CONFIG as _DEFAULT_CONFIG os.environ["HERMES_TURN_LEASE_TIMEOUT"] = str(_DEFAULT_CONFIG["agent"]["gateway_turn_lease_timeout"]) # Bridge config.yaml values into env so os.getenv() picks them up. config.yaml unconditionally wins # over .env for these keys; a `not in os.environ` guard would let stale .env entries shadow config. _AGENT_ENV_BRIDGE = { "gateway_timeout": "HERMES_AGENT_TIMEOUT", "gateway_turn_lease_timeout": "HERMES_TURN_LEASE_TIMEOUT", "gateway_timeout_warning": "HERMES_AGENT_TIMEOUT_WARNING", "gateway_notify_interval": "HERMES_AGENT_NOTIFY_INTERVAL", "session_stall_timeout": "HERMES_SESSION_STALL_TIMEOUT", "restart_drain_timeout": "HERMES_RESTART_DRAIN_TIMEOUT", "cron_drain_timeout": "HERMES_CRON_DRAIN_TIMEOUT", "gateway_auto_continue_freshness": "HERMES_AUTO_CONTINUE_FRESHNESS", "gateway_startup_restore_drain_timeout": "HERMES_STARTUP_RESTORE_DRAIN_TIMEOUT", "gateway_startup_warmup_timeout": "HERMES_STARTUP_WARMUP_TIMEOUT"} # config-authoritative knobs for the session-search index (env stays the cross-process carrier). _SESSIONS_ENV_BRIDGE = {"cjk_fts": "HERMES_CJK_FTS", "search_slow_ms": "HERMES_SEARCH_SLOW_MS"} _DISPLAY_ENV_BRIDGE = { "busy_input_mode": "HERMES_GATEWAY_BUSY_INPUT_MODE", "busy_text_mode": "HERMES_GATEWAY_BUSY_TEXT_MODE", "busy_ack_enabled": "HERMES_GATEWAY_BUSY_ACK_ENABLED"} def _bridge_section_to_env(section: Any, mapping: dict[str, str]) -> None: """Export every present ``mapping`` key of a config section as ``str(value)``.""" if isinstance(section, dict): for cfg_key, env_var in mapping.items(): if cfg_key in section: os.environ[env_var] = str(section[cfg_key]) def _bridge_max_turns_to_env(agent_cfg: Any) -> None: """Bridge ``agent.max_turns`` preserving its raw spelling ("none", "unlimited", "120"); Python None (`null` / bare `key:`) clears a stale bridge instead, since str(None) -> "None" would map to the unlimited sentinel rather than "absent = default".""" if not isinstance(agent_cfg, dict) or "max_turns" not in agent_cfg: return raw = agent_cfg["max_turns"] if raw is not None: os.environ["HERMES_MAX_ITERATIONS"] = str(raw) elif "HERMES_MAX_ITERATIONS" in os.environ: del os.environ["HERMES_MAX_ITERATIONS"] def _bridge_terminal_config_to_env(_terminal_cfg: dict) -> None: """Bridge nested ``terminal.*`` config to TERMINAL_* env vars (config.yaml overrides .env here).""" _terminal_backend = str( _terminal_cfg.get("backend") or os.environ.get("TERMINAL_ENV") or "").strip().lower() _terminal_env_map = { "backend": "TERMINAL_ENV", "degraded_mode": "TERMINAL_DEGRADED_MODE", "cwd": "TERMINAL_CWD", "timeout": "TERMINAL_TIMEOUT", "home_mode": "TERMINAL_HOME_MODE", "lifetime_seconds": "TERMINAL_LIFETIME_SECONDS", "docker_image": "TERMINAL_DOCKER_IMAGE", "docker_forward_env": "TERMINAL_DOCKER_FORWARD_ENV", "modal_image": "TERMINAL_MODAL_IMAGE", "daytona_image": "TERMINAL_DAYTONA_IMAGE", "vercel_runtime": "TERMINAL_VERCEL_RUNTIME", "vercel_image": "TERMINAL_VERCEL_IMAGE", "ssh_host": "TERMINAL_SSH_HOST", "ssh_user": "TERMINAL_SSH_USER", "ssh_port": "TERMINAL_SSH_PORT", "ssh_key": "TERMINAL_SSH_KEY", "container_cpu": "TERMINAL_CONTAINER_CPU", "container_memory": "TERMINAL_CONTAINER_MEMORY", "container_disk": "TERMINAL_CONTAINER_DISK", "container_persistent": "TERMINAL_CONTAINER_PERSISTENT", "docker_volumes": "TERMINAL_DOCKER_VOLUMES", "docker_env": "TERMINAL_DOCKER_ENV", "docker_extra_args": "TERMINAL_DOCKER_EXTRA_ARGS", "docker_shm_size": "TERMINAL_DOCKER_SHM_SIZE", "docker_mount_cwd_to_workspace": "TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "docker_network": "TERMINAL_DOCKER_NETWORK", "docker_run_as_host_user": "TERMINAL_DOCKER_RUN_AS_HOST_USER", "docker_snap_compat": "TERMINAL_DOCKER_SNAP_COMPAT", "docker_persist_across_processes": "TERMINAL_DOCKER_PERSIST_ACROSS_PROCESSES", "docker_shared_container_key": "TERMINAL_DOCKER_SHARED_CONTAINER_KEY", "docker_orphan_reaper": "TERMINAL_DOCKER_ORPHAN_REAPER", "sandbox_dir": "TERMINAL_SANDBOX_DIR", "persistent_shell": "TERMINAL_PERSISTENT_SHELL"} for _cfg_key, _env_var in _terminal_env_map.items(): if _cfg_key not in _terminal_cfg: continue _val = _terminal_cfg[_cfg_key] if _cfg_key == "cwd": # Placeholders (".", "auto", "cwd") resolve to Path.home() later; only explicit paths bridge. if str(_val) in {".", "auto", "cwd"}: continue # Expand "~" for local/container cwd so Popen never gets a literal "~/" (kernel rejects it); # SSH cwd is interpreted by the remote shell: keep "~". Predicate shared w/ terminal_tool. if isinstance(_val, str) and not _is_ssh_remote_tilde_cwd(_terminal_backend, _val.strip()): _val = os.path.expanduser(_val) os.environ[_env_var] = json.dumps(_val) if isinstance(_val, (list, dict)) else str(_val) def _bridge_auxiliary_config_to_env(_auxiliary_cfg: dict) -> None: """Bridge auxiliary model/endpoint overrides (vision, approval, plugins); compression reads yaml.""" _aux_bridged_keys = {"vision", "approval"} try: from hermes_cli.plugins import get_plugin_auxiliary_tasks for _entry in get_plugin_auxiliary_tasks(): _aux_bridged_keys.add(_entry["key"]) except Exception: pass # plugin discovery failure must not break startup; built-in bridging stays intact for _task_key in _aux_bridged_keys: _task_cfg = _auxiliary_cfg.get(_task_key, {}) if not isinstance(_task_cfg, dict): continue _upper = _task_key.upper() _prov = str(_task_cfg.get("provider", "")).strip() if _prov and _prov != "auto": os.environ[f"AUXILIARY_{_upper}_PROVIDER"] = _prov for _field, _suffix in (("model", "MODEL"), ("base_url", "BASE_URL"), ("api_key", "API_KEY")): _value = str(_task_cfg.get(_field, "")).strip() if _value: os.environ[f"AUXILIARY_{_upper}_{_suffix}"] = _value def _bridge_config_to_env(_cfg: dict) -> None: """Export config.yaml settings to the env vars os.getenv() consumers read.""" for _key, _val in _cfg.items(): # top-level scalars: fallback only, never override .env if isinstance(_val, (str, int, float, bool)) and _key not in os.environ: os.environ[_key] = str(_val) _terminal_cfg = _cfg.get("terminal", {}) if _terminal_cfg and isinstance(_terminal_cfg, dict): _bridge_terminal_config_to_env(_terminal_cfg) _auxiliary_cfg = _cfg.get("auxiliary", {}) if _auxiliary_cfg and isinstance(_auxiliary_cfg, dict): _bridge_auxiliary_config_to_env(_auxiliary_cfg) # config.yaml is the documented, authoritative source for these settings — it unconditionally wins over # .env values. Previously the guards below read `if X not in os.environ` and let stale .env entries # (e.g. HERMES_MAX_ITERATIONS=60 written by an old `hermes setup` run) silently shadow the user's # current config. See PR #18413 / the 60-vs-500 max_turns incident. _agent_cfg = _cfg.get("agent", {}) _bridge_max_turns_to_env(_agent_cfg) _bridge_section_to_env(_agent_cfg, _AGENT_ENV_BRIDGE) _bridge_section_to_env(_cfg.get("sessions", {}), _SESSIONS_ENV_BRIDGE) _display_cfg = _cfg.get("display", {}) _bridge_section_to_env(_display_cfg, _DISPLAY_ENV_BRIDGE) # Documented service-manager override: env wins when set (other display bridges stay config-first). if (isinstance(_display_cfg, dict) and "busy_steer_ack_enabled" in _display_cfg and "HERMES_GATEWAY_BUSY_STEER_ACK_ENABLED" not in os.environ): os.environ["HERMES_GATEWAY_BUSY_STEER_ACK_ENABLED"] = str(_display_cfg["busy_steer_ack_enabled"]) _tz_cfg = _cfg.get("timezone", "") if _tz_cfg and isinstance(_tz_cfg, str): os.environ["HERMES_TIMEZONE"] = _tz_cfg.strip() _security_cfg = _cfg.get("security", {}) if isinstance(_security_cfg, dict) and _security_cfg.get("redact_secrets") is not None: os.environ["HERMES_REDACT_SECRETS"] = str(_security_cfg["redact_secrets"]).lower() # Media policy uses the shared bridge so standalone entrypoints (`hermes cron run`) match. _gateway_cfg = _cfg.get("gateway", {}) if isinstance(_gateway_cfg, dict): from gateway.media_policy import apply_media_policy_env apply_media_policy_env(_cfg) _trust_recent_seconds = _gateway_cfg.get("trust_recent_files_seconds") if _trust_recent_seconds is not None: os.environ["HERMES_MEDIA_TRUST_RECENT_SECONDS"] = str(_trust_recent_seconds) # platform_connect_timeout is an escape hatch, unlike the bridges above: env WINS if already set. if ("platform_connect_timeout" in _gateway_cfg and not os.environ.get("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT", "").strip()): os.environ["HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT"] = str(_gateway_cfg["platform_connect_timeout"]) def _load_bridge_config(config_path: Path) -> dict: """Effective USER config (no defaults) for the presence-sensitive env bridge: only keys the user or the managed layer wrote get bridged, else all of DEFAULT_CONFIG would be exported. A bypass worker bridges its frozen explicit snapshot instead: the profile file is exactly what it must not read.""" from agent.safe_worker_policy import worker_config_snapshot snapshot = worker_config_snapshot() if snapshot is not None: return snapshot from hermes_cli.config_effective import load_user_config_effective return load_user_config_effective(config_path) _config_path = _hermes_home / 'config.yaml' _cfg: dict = {} if _config_path.exists(): try: _cfg = _load_bridge_config(_config_path) _bridge_config_to_env(_cfg) except Exception as _bridge_err: # stderr, not logger: the module logger is not initialized yet at import time. print( f" Warning: config.yaml → env bridge failed: {type(_bridge_err).__name__}: {_bridge_err}", file=sys.stderr) print( " Gateway will fall back to .env values, which may not match " "your current config.yaml. Run `hermes doctor` to investigate.", file=sys.stderr) # IPv4 preference must apply before any HTTP clients are created. try: from hermes_constants import apply_ipv4_preference _network_cfg = _cfg.get("network", {}) if isinstance(_network_cfg, dict) and _network_cfg.get("force_ipv4"): apply_ipv4_preference(force=True) except Exception as _bootstrap_exc: print(f" Warning: IPv4 preference application failed: {_bootstrap_exc}", file=sys.stderr) try: from hermes_cli.config import print_config_warnings print_config_warnings() except Exception as _bootstrap_exc: print(f" Warning: config validation failed: {_bootstrap_exc}", file=sys.stderr) try: from hermes_cli.config import warn_deprecated_cwd_env_vars warn_deprecated_cwd_env_vars() except Exception as _bootstrap_exc: print(f" Warning: deprecation check failed: {_bootstrap_exc}", file=sys.stderr) os.environ["HERMES_QUIET"] = "1" # gateway runs quiet: no debug output, cwd used directly # HERMES_EXEC_ASK is set in start_gateway(), NOT at import: CLI tools importing this module must not # flip interactive sessions into ask-mode (approval prompts would become silent pending_approval). # Terminal cwd: config.yaml terminal.cwd is canonical (bridged above); MESSAGING_CWD is legacy fallback. from gateway.cwd_placeholder import CWD_PLACEHOLDERS, resolve_placeholder_terminal_cwd from gateway.config import ( ChannelOverride, Platform, GatewayConfig, PlatformConfig, _getenv, load_gateway_config) from gateway.session import ( AsyncSessionStore, SessionStore, SessionSource, SessionContext, build_session_key, profile_from_session_key_namespace) # Telegram topic routing (#22773, regression fixed #52060): a # ``telegram::`` cron target is ambiguous — a forum-style topic in a # private chat and a genuine Bot API channel Direct-Messages topic share the same shape and need OPPOSITE # routing. Disambiguate at delivery time via ``_is_channel_dm_topic`` (see its docstring for the full # rationale); ``thread_id`` goes in ``route_metadata`` so the anchorless cron send bypasses the # DeliveryRouter's private-chat reply-anchor requirement. Compute the routed metadata ONCE so both the text # send (via DeliveryRouter) and the media send agree. from gateway.delivery import DeliveryRouter from gateway.turn_lease import SessionTurnLeaseRegistry from gateway.session_state import SessionState, legacy_dict_property, legacy_lease_token_property from gateway.authz_mixin import GatewayAuthorizationMixin from gateway.kanban_watchers import GatewayKanbanWatchersMixin from gateway.slash_commands import GatewaySlashCommandsMixin from gateway.run_voice import GatewayVoiceMixin from gateway.run_adapters import GatewayAdapterLifecycleMixin from gateway.run_topics import GatewayTopicThreadsMixin from gateway.run_turn import GatewayTurnMixin, is_context_overflow_failure_result from gateway.run_shutdown import GatewayShutdownMixin, _resolve_gateway_exit_verdict from gateway.run_busy import GatewayBusySessionMixin from gateway.run_config_loaders import GatewayConfigLoadersMixin from gateway.run_startup import GatewayStartupMixin, _own_policy_envs from gateway.run_watchers import GatewaySessionWatchersMixin from gateway.run_notifications import GatewayNotificationsMixin from gateway.run_inbound import GatewayInboundMixin from gateway.run_goals import GatewayGoalsMixin from gateway.run_agent_cache import GatewayAgentCacheMixin from gateway.run_profile_reconcile import GatewayProfileReconcileMixin from gateway.run_plugin_rewire import GatewayPluginRewireMixin from gateway.platforms.base import ( BasePlatformAdapter, _reply_anchor_for_event, _terminal_sentinel_start, ) from gateway.platforms.event import MessageEvent, MessageType from gateway.restart import ( DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT, DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT, DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, DEFAULT_GATEWAY_SIGNAL_INTERRUPT_GRACE_TIMEOUT) logger = logging.getLogger(__name__) def _best_effort(fn: Callable[[], Any], debug_msg: Optional[str] = None) -> Any: """Call ``fn``; return None on any Exception (debug-logged via ``debug_msg`` ``%s`` if given).""" try: return fn() except Exception as exc: if debug_msg: logger.debug(debug_msg, exc) return None # Shutdown quiesce ceiling for the gateway-owned thread pool. Drain already waited for the agents; what # remains is short blocking work; anything slower is a stuck worker not worth waiting on (leash-clamped). _EXECUTOR_QUIESCE_TIMEOUT = 2.0 def _own_policy_open_startup_violation(config) -> Optional[str]: """Return a startup-abort reason when open policy lacks allow-all opt-in.""" with suppress(Exception): # idempotent; callers may hand in a config load_gateway_config never built from hermes_cli.plugins import discover_plugins discover_plugins() for platform, platform_config in getattr(config, "platforms", {}).items(): if not getattr(platform_config, "enabled", False): continue open_env = _own_policy_envs(platform) if not open_env: continue dm_env, group_env, allow_all_env = open_env extra = getattr(platform_config, "extra", None) or {} dm_policy = str(extra.get("dm_policy") or (_getenv(dm_env, "pairing") if dm_env else "pairing")).strip().lower() group_policy = str( extra.get("group_policy") or (_getenv(group_env, "pairing") if group_env else "pairing") ).strip().lower() if dm_policy != "open" and group_policy != "open": continue gateway_allow_all = _getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in {"true", "1", "yes"} if gateway_allow_all or ( allow_all_env and _getenv(allow_all_env, "").lower() in {"true", "1", "yes"}): continue return f"{platform.value}: open policy without allow-all opt-in" return None # Placed into _running_agents *before* any await so a second message can't slip past the "already # running" guard before the agent exists. _AGENT_PENDING_SENTINEL = object() # Conversation-scoped per-session state registry (legacy contract). State lives in # ``SessionState.conversation`` (cleared via ``ConversationState.clear()``); this list remains for # plain-dict stores not yet folded in (``_pending_model_notes``, popped per-key by # _clear_conversation_scope) and the public test contract. NOT listed (different lifecycles): turn-scoped # _running_agents*/_active_session_leases/_busy_ack_ts/_turn_lease_tokens (_release_running_agent_state + # dispatch finally); _session_run_generation (monotonic; clearing breaks stale-run detection); # _agent_cache (_evict_cached_agent); approval/slash-confirm (_clear_session_boundary_security_state). # The state itself now lives in ``SessionState.conversation`` (see gateway/session_state.py) and boundaries # clear it structurally via ``ConversationState.clear()`` — adding a field to ConversationState means every # boundary picks it up automatically. History: boundaries used to each carry a hand-copied pop-list that # drifted whenever a new dict was added (#48031, #58403, #10702, #35809). - _agent_cache: has its own # eviction path (_evict_cached_agent) with resource cleanup; boundaries call it explicitly. _CONVERSATION_SCOPED_STATE: tuple = ( "_session_model_overrides", "_pending_one_turn_model_restores", "_session_reasoning_overrides", "_session_service_tier_overrides", "_pending_model_notes", "_last_resolved_model", "_queued_events", # Stall-watchdog "already notified" latch; cleared on /new so a fresh conversation can warn again. # See #72016. "_session_stall_notified", # Transcript-lag streak counter (#114266); a fresh conversation starts with no lag history. "_transcript_lag_streaks", # Sidecar notes staged but never consumed (turn aborted before run_sync) must not leak into a # future conversation's first user message — session keys are source-derived and REUSED. "_pending_turn_sidecar_notes") def _resolve_runtime_agent_kwargs() -> dict: """Resolve provider credentials for gateway-created AIAgent instances. ``resolve_runtime_provider()`` may fall back to env vars; behavioral config is config.yaml only. An ``AuthError`` from the primary walks the configured fallback chain through the shared ``resolve_runtime_with_fallback`` (the gateway keeps no resolver loop of its own).""" from hermes_cli.runtime_provider import ( resolve_runtime_with_fallback, format_runtime_provider_error, _get_model_config) # Capture primary provider/model from config before the try block so we # can include it in the fallback notice if the primary fails (#74349). _model_cfg = _get_model_config() _primary_model = (_model_cfg.get("default") or "").strip() _primary_provider = (_model_cfg.get("provider") or "").strip() try: runtime, fallback_entry = resolve_runtime_with_fallback(_load_gateway_config()) except Exception as exc: raise RuntimeError(format_runtime_provider_error(exc)) from exc if fallback_entry is not None: # The entry's model is the one this agent must send (#112600). Carry the fallback notice so the # gateway can surface a user-visible provider switch (#74349); the caller must pop # ``_fallback_notice`` before forwarding kwargs to AIAgent. return {**_runtime_agent_kwargs(runtime), "model": fallback_entry["model"], "_fallback_notice": pre_agent_fallback_notice( _primary_provider, _primary_model, runtime.get("provider") or fallback_entry.get("provider") or "unknown", fallback_entry.get("model") or "default")} capabilities = runtime.get("capabilities") capabilities = ( {k: v for k, v in capabilities.items() if isinstance(k, str) and isinstance(v, bool)} if isinstance(capabilities, dict) else {}) return {**_runtime_agent_kwargs(runtime), "capabilities": capabilities} def _runtime_agent_kwargs(runtime: dict) -> dict: """AIAgent constructor kwargs shared by every runtime-provider resolution. ``request_overrides`` passes through as resolved so the provider's request body reaches each turn.""" return { "api_key": runtime.get("api_key"), "base_url": runtime.get("base_url"), "provider": runtime.get("provider"), "requested_provider": runtime.get("requested_provider"), "api_mode": runtime.get("api_mode"), "command": runtime.get("command"), "args": list(runtime.get("args") or []), "credential_pool": runtime.get("credential_pool"), "request_overrides": runtime.get("request_overrides")} @dataclasses.dataclass(frozen=True) class _GatewayModelContext: """Effective gateway model route and context-window resolution.""" model: str provider: str base_url: str context_length: int context_source: str def _resolve_gateway_model_context( model: Optional[str] = None, route: Optional[dict] = None, ) -> _GatewayModelContext: """Resolve the configured gateway route and effective context window. Call off-loop (may block). ``route`` (``provider`` / ``base_url`` / ``api_key`` of a session-only /model switch) replaces the default runtime credentials so the window is looked up against the endpoint that actually serves ``model``; a ``model.context_length`` pin only survives when that route still matches the config. ``context_source`` is ``"default"`` only for a model unknown to the catalog that fell through to ``DEFAULT_FALLBACK_CONTEXT`` — a catalog-listed 256K model is ``"detected"``. """ from agent.model_metadata import ( DEFAULT_CONTEXT_LENGTHS, DEFAULT_FALLBACK_CONTEXT, _longest_key_match, get_model_context_length) resolved_model = model or _resolve_gateway_model() config_context_length = provider = base_url = api_key = custom_providers = None configured_model = configured_provider = configured_base_url = None def _read_config() -> None: nonlocal config_context_length, provider, base_url, custom_providers nonlocal configured_model, configured_provider, configured_base_url data = _load_gateway_config() if not data: return model_cfg = data.get("model", {}) if isinstance(model_cfg, dict): configured_model = model_cfg.get("default") or model_cfg.get("model") raw_ctx = model_cfg.get("context_length") if raw_ctx is not None: with suppress(TypeError, ValueError): config_context_length = int(raw_ctx) configured_provider = provider = model_cfg.get("provider") or None configured_base_url = base_url = model_cfg.get("base_url") or None try: from hermes_cli.config import get_compatible_custom_providers custom_providers = get_compatible_custom_providers(data) except Exception: custom_providers = data.get("custom_providers") def _read_runtime() -> None: nonlocal provider, base_url, api_key if route and route.get("base_url"): # A session route with its own endpoint (a /model switch) replaces the default runtime # read; a route without one (persisted / SessionDB / plain config) still resolves the # default runtime credentials so a custom endpoint and its context_length pin survive. provider = route.get("provider") or provider base_url = route["base_url"] api_key = route.get("api_key") return runtime = _resolve_runtime_agent_kwargs() provider = runtime.get("provider") or provider base_url = runtime.get("base_url") or base_url api_key = runtime.get("api_key") def _pin_still_applies() -> bool: # Drop a configured context_length pin when the effective route no longer matches (or on error). from hermes_cli.route_identity import should_clear_context_pin return not should_clear_context_pin( configured_model, resolved_model, configured_base_url, base_url, configured_provider, provider) def _custom_ctx() -> Optional[int]: from hermes_cli.config import get_custom_provider_context_length return get_custom_provider_context_length( model=resolved_model, base_url=base_url, custom_providers=custom_providers) _best_effort(_read_config) _best_effort(_read_runtime) if config_context_length is not None and not _best_effort(_pin_still_applies): config_context_length = None if config_context_length is None and custom_providers and base_url: config_context_length = _best_effort(_custom_ctx) or None context_length = get_model_context_length( resolved_model, base_url=base_url or "", api_key=api_key or "", config_context_length=config_context_length, provider=provider or "", custom_providers=custom_providers) fell_through = (context_length == DEFAULT_FALLBACK_CONTEXT and _longest_key_match(DEFAULT_CONTEXT_LENGTHS, str(resolved_model).lower()) is None) context_source = ("config" if config_context_length is not None else "default" if fell_through else "detected") return _GatewayModelContext( model=resolved_model, provider=provider or "", base_url=base_url or "", context_length=context_length, context_source=context_source) def _resolve_runtime_agent_kwargs_for_provider(provider: str, target_model: Optional[str] = None) -> dict: """Resolve runtime credentials for a specific provider (e.g. from channel override). ``target_model`` is the model the override will actually send: the ladder's model-keyed rungs (Zen/Go relay + api_mode) must see it rather than config's ``default``, or a Go-only override resolves an api_mode/base_url the sent model cannot use (#112600).""" from hermes_cli.runtime_provider import resolve_runtime_provider, format_runtime_provider_error try: runtime = resolve_runtime_provider(requested=provider, target_model=target_model or None) except Exception as exc: raise RuntimeError(format_runtime_provider_error(exc)) from exc return { **_runtime_agent_kwargs(runtime), "request_overrides": dict(runtime.get("request_overrides") or {}), "capabilities": dict(runtime.get("capabilities") or {})} def _deep_merge_request_overrides(base: Optional[dict], override: Optional[dict]) -> dict: """Merge request_overrides dicts, deep-merging nested dictionaries.""" from hermes_cli.config import _deep_merge base_dict = dict(base or {}) override_dict = dict(override or {}) if not base_dict: return override_dict if not override_dict: return base_dict return _deep_merge(base_dict, override_dict) def _credential_pool_for_provider(provider: Optional[str]): """Return the live credential pool for a provider id (e.g. ``custom:hyper``).""" if not provider or not str(provider).strip(): return None try: return _resolve_runtime_agent_kwargs_for_provider(str(provider).strip()).get("credential_pool") except Exception: logger.debug("Failed to resolve credential pool for provider=%s", provider, exc_info=True) return None def _event_media_type_at(event, index: int) -> str: """Per-attachment MIME at *index*; "" when the adapter set only a message-level type.""" media_types = getattr(event, "media_types", None) or [] return media_types[index] if index < len(media_types) else "" def _event_media_kind_is(event, index: int, mime_prefix: str, fallback_types: frozenset) -> bool: """Per-attachment MIME first, message-level type only when unknown (else a document uploaded alongside an image is base64'd as vision and the provider 400s).""" mtype = _event_media_type_at(event, index) if mtype: return mtype.startswith(mime_prefix) return getattr(event, "message_type", None) in fallback_types def _event_media_is_image(event, index: int) -> bool: return _event_media_kind_is(event, index, "image/", frozenset({MessageType.PHOTO})) def _event_media_is_audio(event, index: int) -> bool: return _event_media_kind_is(event, index, "audio/", frozenset({MessageType.VOICE, MessageType.AUDIO})) def _event_media_is_stt_input(event, index: int) -> bool: """True when an audio attachment should enter the automatic STT pipeline.""" message_type = getattr(event, "message_type", None) if message_type in {MessageType.AUDIO, MessageType.DOCUMENT}: return False return message_type == MessageType.VOICE or _event_media_type_at(event, index).startswith("audio/") def _event_media_is_video(event, index: int) -> bool: return _event_media_kind_is(event, index, "video/", frozenset({MessageType.VIDEO})) def _build_media_placeholder(event) -> str: """Text placeholder for media-only events (later replaced by vision enrichment). Queued media is dequeued via .text only, so a caption-less event would otherwise be lost.""" parts = [] media_urls = getattr(event, "media_urls", None) or [] for i, url in enumerate(media_urls): if _event_media_is_image(event, i): parts.append(f"[User sent an image: {url}]") elif _event_media_is_audio(event, i): parts.append(f"[User sent audio: {url}]") elif _event_media_is_video(event, i): parts.append(f"[User sent a video: {url}]") else: parts.append(f"[User sent a file: {url}]") return "\n".join(parts) def _build_document_context_note( display_name: str, agent_path: str, mtype: str, *, content_inlined: bool = True) -> str: """Context note prepended to a user turn when they attach a document. ``content_inlined=False`` = cached without content, so tell the agent to read it. Binary docs must say *extract* the text; "ask the user" made it punt.""" if mtype.startswith("text/") and content_inlined: return ( f"[The user sent a text document: '{display_name}'. Its content has been included below. " f"The file is also saved at: {agent_path}]") if mtype.startswith("text/"): return ( f"[The user sent a text document: '{display_name}'. It is saved at: {agent_path}. " f"Its content is not inlined here. Read the cached file yourself before answering " f"when the user's request involves its contents.]") return ( f"[The user sent a document: '{display_name}'. It is saved at: {agent_path}. " f"Its text is not inlined here (it's a binary format such as PDF or DOCX). " f"To read it, extract the document's text yourself — for example with the " f"terminal tool or the ocr-and-documents skill — before answering, instead " f"of asking the user to paste the contents.]") def _format_duration(seconds: float) -> str: total = max(0, round(seconds)) hours, rem = divmod(total, 3600) minutes, secs = divmod(rem, 60) if hours: return f"{hours}:{minutes:02d}:{secs:02d}" return f"{minutes}:{secs:02d}" async def _probe_audio_duration(path: str) -> Optional[str]: """Best-effort duration probe. Returns formatted MM:SS / HH:MM:SS, or None on failure.""" ext = os.path.splitext(path)[1].lower() if ext == ".wav": try: def _wav_duration() -> float: import wave with wave.open(path, "rb") as wf: frames = wf.getnframes() rate = wf.getframerate() or 1 return frames / float(rate) return _format_duration(await asyncio.to_thread(_wav_duration)) except Exception: pass if ext in (".ogg", ".opus", ".oga"): try: def _ogg_duration() -> float: from mutagen.oggopus import OggOpus return float(OggOpus(path).info.length) return _format_duration(await asyncio.to_thread(_ogg_duration)) except Exception: pass try: proc = await asyncio.create_subprocess_exec( "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=5.0) if proc.returncode == 0: return _format_duration(float(stdout.decode().strip())) except Exception: pass return None def _dequeue_pending_event(adapter, session_key: str) -> MessageEvent | None: """Consume and return the pending event; media metadata is kept so follow-ups re-enter preprocessing. """ return adapter.get_pending_message(session_key) _INTERRUPT_REASON_STOP = "Stop requested" _INTERRUPT_REASON_RESET = "Session reset requested" _INTERRUPT_REASON_TIMEOUT = "Execution timed out (inactivity)" # ``tool_reason`` for the inactivity timeout: attributes the stop to the gateway watchdog (#112647). _INTERRUPT_TOOL_REASON_TIMEOUT = "gateway inactivity watchdog" _INTERRUPT_REASON_EVICTED = "Session ended while the turn was running" # ``tool_reason`` for eviction / shutdown: these stops are system-issued, not user stops (#112647). _INTERRUPT_TOOL_REASON_EVICTED = "session evicted" _INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN = "gateway shutdown" _INTERRUPT_REASON_SSE_DISCONNECT = "SSE client disconnected" _INTERRUPT_REASON_GATEWAY_SHUTDOWN = "Gateway shutting down" _INTERRUPT_REASON_GATEWAY_RESTART = "Gateway restarting" def _reap_gateway_turn_processes( task_id: str, process_baseline, *, source: str, is_still_current: Optional[Callable[[], bool]] = None) -> int: """Reap only background processes created by one abandoned turn. ``task_id`` is session-scoped, so a *replacement* turn can spawn its own process mid-reap; ``is_still_current`` lets the caller bail instead of killing it (that turn owns its own baseline).""" if not task_id: # Blank task_id (sessionless callers) would match and kill every unrelated empty-task process. return 0 if is_still_current is not None: try: if not is_still_current(): logger.debug( "Skipping reap for turn %s (%s): a newer turn already " "claimed this session; it owns its own baseline.", task_id, source) return 0 except Exception: logger.debug( "is_still_current check failed for turn %s (%s); reaping anyway", task_id, source, exc_info=True) from tools.process_registry import process_registry try: killed = process_registry.kill_started_since(task_id, process_baseline, source=source) except Exception: # Detached daemon thread: an uncaught exception would only reach threading.excepthook. logger.warning( "Failed to reap background processes for turn %s (%s)", task_id, source, exc_info=True) return 0 if killed: logger.warning( "Reaped %d background process(es) created by abandoned turn %s (%s)", killed, task_id, source) return killed _TURN_STACK_DUMP_FRAME_MARKERS = ( "run_conversation", "run_sync", "_run_sync_with_timeout_lifecycle", "finalize_turn", "end_turn", "run_in_session") def _dump_wedged_turn_stacks(task_id: str) -> None: """Log the stack of every thread that looks like turn work, at reap time. The hard interrupt frees the wedged worker before a profiler can attach, so dump BEFORE it. Best-effort, bounded (turn-machinery threads only, capped output), never raises.""" try: frames = sys._current_frames() names = {t.ident: t.name for t in threading.enumerate()} dumped = 0 for ident, frame in frames.items(): if ident == threading.get_ident(): continue # the reaper itself stack = traceback.format_stack(frame) joined = "".join(stack) if not any(marker in joined for marker in _TURN_STACK_DUMP_FRAME_MARKERS): continue dumped += 1 if dumped > 8: logger.error( "Wedged-turn stack dump for task %s truncated: more than 8 candidate threads", task_id) break logger.error( "Wedged-turn stack dump (task=%s thread=%s ident=%s):\n%s", task_id, names.get(ident, "?"), ident, "".join(stack[-25:])) if dumped == 0: logger.error( "Wedged-turn stack dump for task %s: no thread with " "turn-machinery frames found (worker may have already exited)", task_id) except Exception: logger.debug("Wedged-turn stack dump failed", exc_info=True) def _abandon_timed_out_gateway_turn( *, agent_holder, task_id: str, process_baseline, worker_done: threading.Event, timeout_fired: threading.Event, cleanup_lock: threading.Lock, is_still_current: Optional[Callable[[], bool]] = None) -> bool: """Interrupt one timed-out turn and reap only processes it created.""" with cleanup_lock: if worker_done.is_set() or timeout_fired.is_set(): return False timeout_fired.set() # BEFORE interrupting: the interrupt frees the blocked frame, destroying the only evidence. _dump_wedged_turn_stacks(task_id) agent = agent_holder[0] if agent_holder else None if agent is not None: try: request_hard_interrupt(agent, _INTERRUPT_REASON_TIMEOUT, tool_reason=_INTERRUPT_TOOL_REASON_TIMEOUT) except Exception: logger.debug("Timed-out agent interrupt failed", exc_info=True) from hermes_cli.observability.shared_metrics_process import record_watchdog_turn_abort record_watchdog_turn_abort(agent) try: _reap_gateway_turn_processes( task_id, process_baseline, source="gateway_turn_timeout", is_still_current=is_still_current) except Exception: logger.warning( "Failed to reap background processes for timed-out turn %s", task_id, exc_info=True) return True def _watch_gateway_turn_inactivity( *, agent_holder, task_id: str, process_baseline, timeout: float, worker_done: threading.Event, timeout_fired: threading.Event, cleanup_lock: threading.Lock, poll_interval: float = 5.0, is_still_current: Optional[Callable[[], bool]] = None) -> None: """Thread watchdog that remains runnable when gateway asyncio is starved. Until an agent publishes a usable activity snapshot, elapsed worker time is the liveness clock. Otherwise a provider hang before activity initialization can retain the session turn lease forever because every watchdog poll just skips it. """ activity_origin = time.monotonic() awake_idle = AwakeIdleMeter() # host sleep is not idleness while not worker_done.wait(max(0.01, poll_interval)): now = time.monotonic() idle_seconds = now - activity_origin agent = agent_holder[0] if agent_holder else None if agent is not None and hasattr(agent, "get_activity_summary"): try: reported_idle = agent.get_activity_summary().get("seconds_since_activity") if reported_idle is not None: idle_seconds = awake_idle.measure(max(0.0, float(reported_idle))) # Preserve the most recent usable activity clock as the fallback if # a later provider-side diagnostic read raises or returns None. activity_origin = now - idle_seconds except Exception: pass if idle_seconds < timeout: continue _abandon_timed_out_gateway_turn( agent_holder=agent_holder, task_id=task_id, process_baseline=process_baseline, worker_done=worker_done, timeout_fired=timeout_fired, cleanup_lock=cleanup_lock, is_still_current=is_still_current) return _CONTROL_INTERRUPT_MESSAGES = frozenset({ _INTERRUPT_REASON_STOP.lower(), _INTERRUPT_REASON_RESET.lower(), _INTERRUPT_REASON_TIMEOUT.lower(), _INTERRUPT_REASON_SSE_DISCONNECT.lower(), _INTERRUPT_REASON_EVICTED.lower(), _INTERRUPT_REASON_GATEWAY_SHUTDOWN.lower(), _INTERRUPT_REASON_GATEWAY_RESTART.lower()}) def _is_control_interrupt_message(message: Optional[str]) -> bool: """Return True when an interrupt message is internal control flow.""" if not message: return False return " ".join(str(message).strip().split()).lower() in _CONTROL_INTERRUPT_MESSAGES def _strip_response_attachments_for_direct_send(response: str, adapter) -> str: """Return the visible text portion of a response before direct send(). Only explicit ``MEDIA:`` attachments are stripped; bare paths/URLs stay visible. No broad regex after ``extract_media()``: it deliberately preserves protected code spans and unvalidated tags. Queued follow-up resends only replay explicit ``MEDIA:`` attachments in this path. Keep bare local paths and ordinary image URLs visible because the post-stream uploader intentionally ignores them (#20834). """ _, cleaned = adapter.extract_media(response) return cleaned.replace("[[audio_as_voice]]", "").replace("[[as_document]]", "").strip() def _skill_slug_from_frontmatter(skill_md: Path) -> tuple[str | None, str | None]: """Derive ``(slug, declared_name)`` from a SKILL.md; ``(None, None)`` if unreadable or no ``name:``. Matches ``scan_skill_commands``: the slug comes from frontmatter ``name:``, NOT the directory.""" try: content = skill_md.read_text(encoding="utf-8-sig", errors="replace") except Exception: return None, None content = content.lstrip("\ufeff") # tolerate UTF-8 BOM (Windows editors) if not content.startswith("---"): return None, None end = content.find("\n---", 3) if end < 0: return None, None declared_name: str | None = None for line in content[3:end].splitlines(): line = line.strip() if line.startswith("name:"): raw = line.split(":", 1)[1].strip() if len(raw) >= 2 and raw[0] == raw[-1] and raw[0] in {'"', "'"}: raw = raw[1:-1] declared_name = raw.strip() break if not declared_name: return None, None slug = declared_name.lower().replace(" ", "-").replace("_", "-") # Mirrors _SKILL_INVALID_CHARS / _SKILL_MULTI_HYPHEN from skill_commands slug = re.sub(r"[^a-z0-9-]", "", slug) slug = re.sub(r"-{2,}", "-", slug).strip("-") return (slug or None), declared_name def _check_unavailable_skill(command_name: str) -> str | None: """Hint when a command matches a skill that is disabled or optional-install only; else None.""" normalized = command_name.lower().replace("_", "-") try: from tools.skills_tool import _get_disabled_skill_names from agent.skill_utils import get_all_skills_dirs, is_excluded_skill_path disabled = _get_disabled_skill_names() for skills_dir in get_all_skills_dirs(): if not skills_dir.exists(): continue for skill_md in skills_dir.rglob("SKILL.md"): if is_excluded_skill_path(skill_md): continue slug, declared_name = _skill_slug_from_frontmatter(skill_md) if not slug or not declared_name: continue # disabled is keyed by the declared frontmatter name (what skills.disabled stores). if slug == normalized and declared_name in disabled: return t("gateway.skills.disabled", name=command_name) # Check optional skills (shipped with repo but not installed) from hermes_constants import get_optional_skills_dir repo_root = Path(__file__).resolve().parent.parent optional_dir = get_optional_skills_dir(repo_root / "optional-skills") if optional_dir.exists(): for skill_md in optional_dir.rglob("SKILL.md"): if is_excluded_skill_path(skill_md): continue slug, _declared = _skill_slug_from_frontmatter(skill_md) if not slug or slug != normalized: continue # Install path: official// rel = skill_md.parent.relative_to(optional_dir) install_path = f"official/{'/'.join(rel.parts)}" return t("gateway.skills.not_installed", name=command_name, install_name=install_path) except Exception: pass return None def _platform_config_key(platform: Platform) -> str: """Map a Platform enum to its config.yaml key (LOCAL→"cli", rest→enum value).""" return "cli" if platform == Platform.LOCAL else platform.value def _teams_pipeline_plugin_enabled() -> bool: """Return True when the standalone Teams pipeline plugin is enabled.""" enabled = cfg_get(_load_gateway_config(), "plugins", "enabled", default=[]) return isinstance(enabled, list) and ("teams_pipeline" in enabled or "teams-pipeline" in enabled) def _gateway_config_home() -> Path: """Return the Hermes home that gateway config reads should use.""" override = get_hermes_home_override() return Path(override) if override else _hermes_home def _load_gateway_config(config_path: Path | None = None) -> dict: """The effective user config.yaml (managed overlay, ``${VAR}`` expansion, model-key canon; no DEFAULT_CONFIG merge) — ``{}`` on any error (fail-open). Defaults to the active gateway home (``_hermes_home`` monkeypatches apply); multiplexers pass a path. """ if config_path is None: config_path = _gateway_config_home() / 'config.yaml' try: from hermes_cli.config_effective import load_user_config_effective return load_user_config_effective(config_path) except Exception: logger.debug("Could not load gateway config from %s", config_path, exc_info=True) return {} from gateway.session_worker_construct import checkpoint_agent_kwargs as _checkpoint_agent_kwargs def _resolve_gateway_model(config: dict | None = None) -> str: """The profile's model when nothing else names one: config.yaml (single source of truth), else the claimed welcome trial's. Temporary AIAgents (e.g. /compress) otherwise use the hardcoded default, which fails under openai-codex.""" from hermes_cli.anon_auth import trial_model model_cfg = (config if config is not None else _load_gateway_config()).get("model", {}) if isinstance(model_cfg, dict): model_cfg = model_cfg.get("default") or model_cfg.get("model") return (model_cfg if isinstance(model_cfg, str) else "") or trial_model() def _channel_override_lookup_keys( chat_id: str, *, thread_id: Optional[str] = None, parent_id: Optional[str] = None) -> list[str]: """Ordered, de-duplicated ``channel_overrides`` lookup keys (matches ``resolve_channel_prompt``: exact id first, then parent — Discord threads inherit parent overrides).""" return list(dict.fromkeys(str(key) for key in (chat_id, thread_id, parent_id) if key)) def _get_channel_override( config: GatewayConfig, platform: Platform, chat_id: str, *, thread_id: Optional[str] = None, parent_id: Optional[str] = None) -> Optional[ChannelOverride]: """Per-channel override via chat_id, then thread_id, then parent_id; None if absent.""" platforms = getattr(config, "platforms", None) if not platforms: return None platform_config = platforms.get(platform) if not platform_config or not platform_config.channel_overrides: return None overrides = platform_config.channel_overrides for key in _channel_override_lookup_keys(chat_id, thread_id=thread_id, parent_id=parent_id): ov = overrides.get(key) if ov is not None: return ov return None def _resolve_hermes_bin() -> Optional[list[str]]: """Hermes update/restart argv: the running interpreter's ``python -m hermes_cli.main`` (exactly this install), else ``hermes`` on PATH, else None. The module argv must win: a PATH-first lookup lets an attacker-planted ``hermes`` shadow the running install when /update or /restart re-execs it (#111569).""" try: import importlib.util if importlib.util.find_spec("hermes_cli") is not None: return [sys.executable, "-m", "hermes_cli.main"] except Exception: pass import shutil hermes_bin = shutil.which("hermes") if hermes_bin: return [hermes_bin] return None _PROFILE_ID_KEY_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") def _parse_session_key(session_key: str) -> dict | None: """Parse a session key (``agent:{ns}:{platform}:{chat_type}:{chat_id}[:{extra}...]``). ``{ns}`` is ``main`` for the default profile, ``main~`` for a profile literally named ``main`` (``gateway.session._session_key_namespace``), or a named-profile id (profile ids match ``[a-z0-9][a-z0-9_-]{0,63}`` — never contain ``:`` — so a plain split stays unambiguous). For group/channel sessions the suffix may be a user_id, not a thread_id, so ``thread_id`` is omitted. Named profiles are reported as ``profile``; ``main`` keys keep their historical shape exactly (no ``profile`` key) so equality assertions on parsed dicts stay stable. """ parts = session_key.split(":") if ( len(parts) >= 5 and parts[0] == "agent" and (parts[1] in ("main", "main~") or _PROFILE_ID_KEY_RE.match(parts[1])) ): result = {"platform": parts[2], "chat_type": parts[3], "chat_id": parts[4]} if parts[1] != "main": result["profile"] = profile_from_session_key_namespace(parts[1]) if len(parts) > 5 and parts[3] in {"dm", "thread"}: result["thread_id"] = parts[5] return result return None def _shorten_command_for_display(command: str, limit: int = 80) -> str: """Collapse a shell command onto one line and cap its length for display.""" one_line = " ".join((command or "").split()) if len(one_line) > limit: one_line = one_line[: limit - 1] + "…" return one_line def _format_concise_process_notification( session_id: str, command: str, exit_code, output: str, duration_seconds=None) -> str: """One-line completion message for ``concise`` display mode; failure appends a short output tail.""" ok = exit_code in {0, None} parts = [t("gateway.background.task_finished") if ok else t("gateway.background.task_failed")] short_cmd = _shorten_command_for_display(command) if short_cmd: parts.append(f"— `{short_cmd}`") details = [] if isinstance(duration_seconds, (int, float)) and duration_seconds >= 0: secs = int(duration_seconds) if secs >= 3600: details.append(f"{secs // 3600}h {(secs % 3600) // 60}m") elif secs >= 60: details.append(f"{secs // 60}m {secs % 60}s") else: details.append(f"{secs}s") if not ok: details.append(t("gateway.background.exit_code", code=exit_code)) if details: parts.append(f"({', '.join(details)})") text = " ".join(parts) if not ok and output: tail_lines = [ln for ln in output.strip().splitlines() if ln.strip()][-5:] tail = "\n".join(tail_lines) if len(tail) > 500: tail = tail[-500:] if tail: text += t("gateway.background.last_output", tail=tail) if not ok: text += t("gateway.background.rerun_hint") return text def _format_gateway_process_notification(evt: dict) -> str | None: """Format a watch pattern event from completion_queue into a [IMPORTANT:] message.""" evt_type = evt.get("type", "completion") _sid = evt.get("session_id", "unknown") _cmd = evt.get("command", "unknown") # watch_disabled / overflow events carry their summary in `message` (process_registry formatter). if evt_type in ("watch_disabled", "watch_overflow_tripped", "watch_overflow_released"): return f"[IMPORTANT: {evt.get('message', '')}]" if evt_type == "watch_match": _pat = evt.get("pattern", "?") _out = evt.get("output", "") _sup = evt.get("suppressed", 0) text = ( f"[IMPORTANT: Background process {_sid} matched " f"watch pattern \"{_pat}\".\n" f"Command: {_cmd}\nMatched output:\n{_out}") if _sup: text += f"\n({_sup} earlier matches were suppressed by rate limit)" text += "]" return text if evt_type in ("async_delegation", "heartbeat"): from tools.process_registry_notifications import format_process_notification return format_process_notification(evt) return None def _drain_gateway_watch_events(completion_queue) -> list[dict]: """Drain gateway-owned watch events without spinning on requeued events. Foreign events requeued inside ``while not queue.empty()`` never terminate: detach, then requeue.""" watch_events: list[dict] = [] requeue: list[dict] = [] while not completion_queue.empty(): try: evt = completion_queue.get_nowait() except Exception: break evt_type = evt.get("type", "completion") if evt_type in { "watch_match", "watch_disabled", "watch_overflow_tripped", "watch_overflow_released", "heartbeat"}: watch_events.append(evt) elif evt_type == "async_delegation": requeue.append(evt) # else: process completion events are handled by the watcher task for evt in requeue: completion_queue.put(evt) return watch_events # Weak ref to the active GatewayRunner; tools like send_message route through its live adapters. import weakref as _weakref _gateway_runner_ref: _weakref.ref = lambda: None def _normalize_empty_agent_response( agent_result: dict, response: str, *, history_len: int = 0) -> str: """Normalize empty/None agent responses into user-facing messages. Covers ``failed``, work done (api_calls > 0) with no text, and never-ran (api_calls == 0, the post-/stop silent-drop from a stale generation token) with a retry hint. Consolidates the existing ``failed`` handler and adds a catch-all for the case where the agent did work (api_calls > 0) but returned no text. Fix for #18765. Also surfaces a retry hint when the agent never ran at all (api_calls == 0) for a non-interrupted, non-failed turn -- this is the silent-drop pattern observed after ``/stop`` where the next user message hits a stale generation token and returns an empty result, leaving the platform with nothing to send. (#31884) A failed context-overflow turn whose ``final_response`` is only the raw provider envelope (``HTTP 400: {...}``) is rewritten too: returned unchanged, chat sanitizers turn it into a generic provider-failed reply and the user never sees /compact. Curated agent text survives. """ is_overflow = is_context_overflow_failure_result(agent_result, history_len) if response and not (is_overflow and _looks_like_gateway_provider_error(response)): return response if agent_result.get("failed"): # ``error`` can be an EXPLICIT None (bypasses dict.get default) -> would render "failed: None". error_detail = agent_result.get("error") or t("gateway.shared.unknown_error") error_str = str(error_detail).lower() # Persistence failures: suggesting /reset would destroy context without fixing storage. failure_reason = str(agent_result.get("failure_reason") or "") if failure_reason.startswith("session_persistence_failed") or "session storage" in error_str: if failure_reason.endswith(":disk") or "disk" in error_str: return t("gateway.errors.session_storage_unavailable_disk") return t("gateway.errors.session_storage_unavailable") if is_overflow: return _context_overflow_reply() # Raw exception text (class names, JSON bodies, URLs) stays in the gateway log. logger.warning("Agent turn failed; reply sanitized for chat. Detail: %s", str(error_detail)[:500]) return t("gateway.errors.generic_failed") api_calls = int(agent_result.get("api_calls", 0) or 0) if agent_result.get("interrupted"): # Interrupted with api_calls > 0 = deliberately stopped/steered; silence is intentional (queued # messages arrive via the recursive drain). ZERO api_calls = never processed (stale /stop flag). # An interrupted run that did work (api_calls > 0) is the drain of a run the user deliberately # stopped or steered — its silence is intentional, and any queued/interrupting message is delivered # by the recursive drain inside _run_agent before this result is seen. An interrupted run with ZERO # api_calls never processed the user's message at all: it was killed at the top of the tool loop by # an interrupt flag left over from a recent /stop (#44212). Pure silence there swallows a real user # message, so surface it. # api_calls == 0, not failed, not interrupted: the agent never ran for this turn. This is the # post-/stop generation-race pattern where the gateway would otherwise silently drop the turn # (response=0 chars) and the user sees no reply at all. Surface a short retry hint so the message # isn't lost in silence. (#31884) if api_calls == 0: return t("gateway.errors.interrupted_before_start") return response if api_calls > 0: # Hidden-reasoning-only retry exhaustion: the loop's sentinel text ("Codex response remained # incomplete after 3 continuation attempts") doubles as final_response, so it would be delivered # verbatim into the channel — where peer agents can ingest it as a completed assistant turn # (#51628). Blank it here so the normal empty-response handling (and the suppression below) applies. if _is_gateway_hidden_reasoning_incomplete_turn(agent_result): return "" if agent_result.get("partial"): # ``error`` mirrors the loop's own final text (curated, e.g. "Response truncated due to # output length limit") and is kept; a raw provider envelope goes to the log instead. err = str(agent_result.get("error") or t("gateway.errors.processing_incomplete")) # A loop site code (truncated, context_overflow, ...) already wrote the full # what-happened / what-to-do sentence: deliver it verbatim. Wrapping it would cut it # mid-sentence at 200 chars and append a second, conflicting set of instructions. from agent.turn_failure_copy import SITE_FAILURE_CODES if (str(agent_result.get("failure_reason") or "") in SITE_FAILURE_CODES and err.strip() and not _looks_like_gateway_provider_error(err)): return err if err.startswith("⚠️") else f"⚠️ {err}" if _looks_like_gateway_provider_error(err): logger.warning("Agent turn ended partially; reply sanitized for chat. Detail: %s", err[:500]) reason = "" else: reason = f": {err[:200]}" return t("gateway.errors.stopped_before_finishing", reason=reason) return t("gateway.errors.no_response") # api_calls == 0, not failed/interrupted: agent never ran (post-/stop race); don't drop silently. if api_calls == 0 and not agent_result.get("partial"): return t("gateway.errors.previous_turn_cleanup") return response def _is_gateway_hidden_reasoning_incomplete_turn(agent_result: dict) -> bool: """Detect retry-exhausted turns with hidden reasoning but no visible answer. The loop returns the retry-exhaustion sentinel as BOTH ``final_response`` and ``error``, so a non-empty ``final_response`` proves nothing; any text other than the sentinel is a real answer.""" if (not isinstance(agent_result, dict) or agent_result.get("failed") or agent_result.get("interrupted") or not agent_result.get("partial")): return False error_text = str(agent_result.get("error", "") or "").strip() if "remained incomplete after" not in error_text.lower(): return False final_response = str(agent_result.get("final_response") or "").strip() return not final_response or final_response == error_text def _should_clear_resume_pending_after_turn(agent_result: dict) -> bool: """True only when a gateway turn really completed successfully. ``resume_pending`` is a durable restart-recovery marker; a soft interrupt can look like a normal empty result, and clearing then loses the signal.""" if not isinstance(agent_result, dict) or agent_result.get("interrupted"): return False if agent_result.get("failed") or agent_result.get("partial") or agent_result.get("error"): return False return agent_result.get("completed") is not False def _preserve_queued_followup_history_offset( current_result: dict, followup_result: dict) -> dict: """Carry the outer history offset through queued follow-up drains. Each recursive ``_run_agent()`` advances ``history_offset``; uncorrected, the outer persistence step sees only the *last* queued turn as "new" and drops earlier ones.""" if not isinstance(followup_result, dict) or not isinstance(current_result, dict): return followup_result current_offset = current_result.get("history_offset") followup_offset = followup_result.get("history_offset") if not isinstance(current_offset, int): return followup_result if isinstance(followup_offset, int) and followup_offset <= current_offset: return followup_result return {**followup_result, "history_offset": current_offset} async def _dispose_unused_adapter(adapter: BasePlatformAdapter | None) -> None: """Best-effort dispose for an adapter that never made it onto ``self.adapters`` (may be ``None``). Nothing else calls ``disconnect()`` on it, so ``__init__`` resources (e.g. SQLite fds) would leak until GC (not prompt for asyncio-bound objects) and exhaust the fd ulimit over a long retry loop. The reconnect watcher in ``GatewayRunner._platform_reconnect_watcher`` constructs a fresh adapter on every retry attempt. When the connect call fails — for any of the three reasons (non-retryable error, retryable error, exception during connect) — the adapter is dropped without ever being installed, so nothing else will call its ``disconnect()``. ``APIServerAdapter`` opens a SQLite ``ResponseStore`` that holds 2 fds — the db file and its WAL sidecar) stay open until garbage collection sweeps the unreachable object, which Python's cyclic GC does not do promptly for asyncio-bound objects with native handles. The cumulative leak is 2 fds × every retry at the 300s backoff cap ≈ 12 fds/hour, and the default 2560-fd ulimit is exhausted in ~12h of continuous failure, after which every open() call on the gateway raises ``OSError: [Errno 24] Too many open files`` and the gateway becomes a zombie (#37011). """ if adapter is None: return try: await adapter.disconnect() except Exception: # Half-constructed adapters may raise; must not abort the watcher (CancelledError propagates). logger.debug( "Adapter dispose raised on unowned adapter %r", getattr(adapter, "name", type(adapter).__name__), exc_info=True) # Max seconds between platform reconnect retries (primary watcher and secondary profiles share it). _RECONNECT_BACKOFF_CAP = 300 def _reconnect_backoff(attempt: int) -> int: """Exponential reconnect backoff: 30s, 60s, 120s, ... capped at 5 min.""" return min(30 * (2 ** (attempt - 1)), _RECONNECT_BACKOFF_CAP) def _reconnect_attention_after_secs() -> float: """``agent.reconnect_attention_after`` of the profile whose scope is bound at call time (the launch profile's when unbound). Seconds continuously in the reconnect queue before NEEDS_ATTENTION; retrying never stops (transient outages must self-heal), this only makes a permanently-failing loop loud. Non-positive disables. Read per call, never cached: one process serves many profiles and a config edit must not need a gateway restart (#115635).""" from hermes_cli.config import load_config_readonly agent_cfg = load_config_readonly().get("agent") raw = agent_cfg.get("reconnect_attention_after") if isinstance(agent_cfg, dict) else None try: return float(raw) except (TypeError, ValueError): return float(_DEFAULT_CONFIG["agent"]["reconnect_attention_after"]) def _reconnect_needs_attention(info: dict, now: float) -> bool: """True when a reconnect-queue entry has waited long enough for NEEDS_ATTENTION. ``queued_at`` is re-stamped on each (re)entry, so only *continuous* failure escalates.""" threshold = _reconnect_attention_after_secs() if threshold <= 0: return False # escalation disabled queued_at = info.get("queued_at") if queued_at is None: info["queued_at"] = now return False return (now - queued_at) >= threshold # "No session DB pinned": lets ``_session_db`` distinguish "resolve from profile scope" from a # deliberate ``runner._session_db = None`` (disables DB commands). Mirrors gateway.session._DB_UNPINNED. _SESSION_DB_UNPINNED = object() # Only explicit suspension can replace a routed conversation. _AUTO_RESET_CONTEXT_NOTES = { "suspended": "[System note: The user's previous session was stopped and suspended. This is a fresh conversation with no prior context.]", } def _write_runtime_status_quiet(**fields: Any) -> None: """Best-effort status publication; persistence must never abort or block the caller.""" try: from gateway.status import publish_runtime_status publish_runtime_status(**fields) except Exception: pass def _command_origin_for_source(source: Any) -> Optional[dict]: """Delivery origin for a shared CLI/gateway command so its job replies to this chat/thread.""" try: platform = getattr(source.platform, "value", None) or str(getattr(source, "platform", "") or "") chat_id = getattr(source, "chat_id", None) if platform and chat_id: return { "platform": platform, "chat_id": str(chat_id), "chat_name": getattr(source, "chat_name", None), "thread_id": getattr(source, "thread_id", None)} except Exception: pass return None def _builtin_adapter_import(module: str, adapter_name: str, requirement: str): """Lazy-import ``(adapter_cls, requirements_ok)`` from ``gateway.platforms.``.""" import importlib mod = importlib.import_module(f"gateway.platforms.{module}") return getattr(mod, adapter_name), getattr(mod, requirement) # platform -> (module, adapter class, requirements probe, warning on probe failure). _BUILTIN_ADAPTERS: dict[Platform, tuple[str, str, str, str]] = { Platform.API_SERVER: ("api_server", "APIServerAdapter", "check_api_server_requirements", "API Server: aiohttp not installed"), Platform.WEBHOOK: ("webhook", "WebhookAdapter", "check_webhook_requirements", "Webhook: aiohttp not installed"), Platform.MSGRAPH_WEBHOOK: ("msgraph_webhook", "MSGraphWebhookAdapter", "check_msgraph_webhook_requirements", "MSGraph webhook: aiohttp not installed")} def _instantiate_builtin_adapter(platform: Platform, config: Any) -> Optional[BasePlatformAdapter]: """Instantiate a core (non-plugin) adapter, or None when its requirements are unmet/unknown.""" spec = _BUILTIN_ADAPTERS.get(platform) if spec is None: return None module, adapter_name, requirement, warning = spec adapter_cls, requirements_ok = _builtin_adapter_import(module, adapter_name, requirement) if not (requirements_ok() if callable(requirements_ok) else requirements_ok): logger.warning(warning) return None return adapter_cls(config) from gateway.run_runtime_init import GatewayRuntimeInitMixin class GatewayRunner( GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, GatewaySlashCommandsMixin, GatewayVoiceMixin, GatewayAdapterLifecycleMixin, GatewayTopicThreadsMixin, GatewayTurnMixin, GatewayShutdownMixin, GatewayBusySessionMixin, GatewayConfigLoadersMixin, GatewayStartupMixin, GatewaySessionWatchersMixin, GatewayNotificationsMixin, GatewayInboundMixin, GatewayGoalsMixin, GatewayAgentCacheMixin, GatewayRuntimeInitMixin, GatewayProfileReconcileMixin, GatewayPluginRewireMixin): """Main gateway controller: manages adapter lifecycles, routes messages to/from the agent.""" def _adapter_for_source(self, source): """The adapter that owns *source* for the session authority (admission checks, canonical automation, native route checks, recovery bindings). Main split the old resolver into the intake and delivery seams; the authority needs the answering adapter, so this is the delivery seam under its historical name.""" return self._delivery_adapter_for(source) # Class-level defaults so partial construction in tests doesn't blow up on attribute access. _busy_input_mode: str = "interrupt" _busy_text_mode: str = "interrupt" _restart_drain_timeout: float = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT _restart_after_turn_timeout: float = DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT _cron_drain_timeout: float = DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT _signal_interrupt_grace_timeout: float = DEFAULT_GATEWAY_SIGNAL_INTERRUPT_GRACE_TIMEOUT _exit_code: Optional[int] = None _draining: bool = False _external_drain_active: bool = False _restart_requested: bool = False _restart_task_started: bool = False _restart_detached: bool = False _restart_via_service: bool = False _detached_restart_helper_started: bool = False _restart_command_source: Optional[SessionSource] = None _stop_task: Optional[asyncio.Task] = None _restart_task: Optional[asyncio.Task] = None _profile_failed_platforms: Optional[dict[str, dict[Platform, asyncio.Task]]] = None _systemd_watchdog: Optional[Any] = None _startup_restore_in_progress: bool = False _startup_warmup_task: Optional[asyncio.Task] = None # Legacy per-session dict attrs as LIVE views over ``self._sessions``; new code: _session_state(key) _running_agents = legacy_dict_property("_running_agents") _running_agents_ts = legacy_dict_property("_running_agents_ts") _active_session_leases = legacy_dict_property("_active_session_leases") _busy_ack_ts = legacy_dict_property("_busy_ack_ts") _turn_lease_tokens = legacy_lease_token_property() _session_run_generation = legacy_dict_property("_session_run_generation") _session_model_overrides = legacy_dict_property("_session_model_overrides") _pending_one_turn_model_restores = legacy_dict_property("_pending_one_turn_model_restores") _session_reasoning_overrides = legacy_dict_property("_session_reasoning_overrides") _session_service_tier_overrides = legacy_dict_property("_session_service_tier_overrides") _last_resolved_model = legacy_dict_property("_last_resolved_model") _queued_events = legacy_dict_property("_queued_events") _pending_turn_sidecar_notes = legacy_dict_property("_pending_turn_sidecar_notes") _pending_messages = legacy_dict_property("_pending_messages") _pending_native_image_paths_by_session = legacy_dict_property( "_pending_native_image_paths_by_session") _session_ephemeral_pin = legacy_dict_property("_session_ephemeral_pin") _session_vc_last = legacy_dict_property("_session_vc_last") _pending_approvals = legacy_dict_property("_pending_approvals") _update_prompt_pending = legacy_dict_property("_update_prompt_pending") def _sessions_map(self) -> dict[str, SessionState]: """Per-session state map; lazily created so bare ``object.__new__`` test runners work.""" sessions = self.__dict__.get("_sessions") if sessions is None: sessions = {} self.__dict__["_sessions"] = sessions return sessions def _session_state(self, session_key: str) -> SessionState: """Get-or-create the :class:`SessionState` for ``session_key``.""" sessions = self._sessions_map() state = sessions.get(session_key) if state is None: state = SessionState() sessions[session_key] = state return state def _peek_session_state(self, session_key: str) -> Optional[SessionState]: """Return the SessionState for ``session_key`` without creating one.""" sessions = self.__dict__.get("_sessions") return sessions.get(session_key) if sessions else None def _is_session_running(self, session_key: str) -> bool: """True when the session holds a running-turn slot (agent or sentinel).""" state = self._peek_session_state(session_key) return state is not None and state.turn.agent is not None def _running_agent_items(self) -> list[tuple]: """(session_key, agent) pairs for sessions with a running turn (incl. pending sentinels).""" return [(key, state.turn.agent) for key, state in self._sessions_map().items() if state.turn.agent is not None] # Loop-liveness / watchdog handles; class-level defaults so partially constructed test runners work. # Class-level defaults so partial construction in tests doesn't blow up on access; the real values are # set in __init__ / start() / stop(). See #66892, #69089. _loop_heartbeat_task: Optional[asyncio.Task] = None _loop_floor_timer_handle: Optional[Any] = None _loop_liveness_watchdog: Optional[Any] = None _gateway_started_at: float = 0.0 _shutdown_watchdog_done: Optional[threading.Event] = None _platform_lock_takeover_on_start: bool = False _reconnect_watcher_task: Optional[asyncio.Task] = None def __init__(self, config: Optional[GatewayConfig] = None): global _gateway_runner_ref # With multiplex_profiles on, load under the default profile secret scope so bot tokens in its # .env resolve as secondary profiles' do; explicit config= injection (tests) is left untouched. # See #64674. # Injected configs keep their mode (including None), except the launching profile's # standalone opt-out: --config must not turn that profile into a host multiplexer. self.config = config if config is not None else load_gateway_config_for_runner() if config is not None: from hermes_cli.gateway_multiplex_mode import standalone_launcher_decision, log_multiplex_decision decision = standalone_launcher_decision(self.config) if decision is not None: log_multiplex_decision(decision) # Multiplexer flag flips agent.secret_scope.get_secret() to fail-closed on unscoped credential # reads, so a missed migration crashes loudly instead of leaking a cross-profile value. try: from agent.secret_scope import set_multiplex_active set_multiplex_active(bool(getattr(self.config, "multiplex_profiles", False))) except Exception: logger.debug("could not set multiplex-active flag", exc_info=True) self.adapters: dict[Platform, BasePlatformAdapter] = {} # Non-None means SessionDB init failed — the gateway broadcasts a one-time warning to the home # channel(s) after connecting so the user learns persistence is broken before /resume fails. # See #88235. self._session_db_init_error: Optional[str] = None # Non-default profiles' adapters by profile then Platform; self.adapters stays the default's map. self._profile_adapters: dict[str, dict[Platform, BasePlatformAdapter]] = {} # Each SERVED profile's gateway config, as loaded once by ``_load_secondary_profile_config``. # ``self.config`` is only the launch profile's: anything host-wide (restart notices) needs these. self._profile_configs: dict[str, Any] = {} self._warn_if_docker_media_delivery_is_risky() _gateway_runner_ref = _weakref.ref(self) self._init_runtime_settings() self._init_session_store() self._init_lifecycle_state() self._init_runtime_caches() self._init_session_db() self._init_registries_and_clocks() def _open_session_db_for_active_scope(self, raise_on_error: bool = False) -> Any: """AsyncSessionDB for the active profile scope, resolved per access (not in ``__init__``) since ``SessionDB()`` reads the context-local HERMES_HOME; one handle cached per path. Construction failure enters bounded backoff; ``raise_on_error=True`` (priming) propagates it. Same per-path cache as ``SessionStore._open_session_db_for_active_scope`` (#88532): ``SessionDB()`` resolves ``_default_db_path()`` at call time through the context-local HERMES_HOME override installed by ``_profile_runtime_scope``, so resolving per access — instead of once in ``__init__`` — is what lets /resume, /title, /history and session search on a multiplexed gateway read the *serving profile's* store rather than the root one. One ``AsyncSessionDB`` is cached per resolved path, so the wrapper identity is stable per profile (callers compare and stash it) and two profiles never share a handle. A construction failure enters bounded backoff; one caller retries after the deadline while concurrent callers continue to see the unavailable fallback. ``raise_on_error=True`` (construction-time priming) propagates the failure after recording that recoverable state so ``__init__`` can record ``_session_db_init_error`` for the #88235 broadcast. """ from hermes_state import AsyncSessionDB, _default_db_path from hermes_state_registry import acquire from gateway.session_db_recovery import RecoverableHandleCache path = Path(_default_db_path()) cache = getattr(self, "_session_db_handle_cache", None) if cache is None: # Test runners built with object.__new__ skip __init__. cache = RecoverableHandleCache( handles=self._session_db_handles, lock=self._session_db_handles_lock) self._session_db_handle_cache = cache def _open(): # Borrow the SessionStore's handle (same path) so state.db doesn't get two writers/pools. # The store owns/sweeps it at shutdown; this cache holds only the async wrapper (close_all). # Both caches resolve the SAME ``_default_db_path()``, so the process was holding two writer # connections and two read pools against one state.db — the fd budget doubled for nothing, and # doubled again per profile on a multiplexed gateway (#98573). A borrowed wrapper goes stale only # when the registry tears its generation down (profile unserve/delete); both caches then drop # the dead handle and reopen through the registry. While the store's own open is failing there # is nothing to borrow, so nothing is cached here either. store = getattr(self, "session_store", None) borrowed = getattr(store, "_db", None) if store is not None else None if borrowed is not None: wrapper = AsyncSessionDB(borrowed) # close_all_session_db_handles() must not close what the store owns (its sweep runs first). wrapper.__dict__["_hermes_borrowed_handle"] = True return wrapper if store is not None: # Store handle unavailable: opening our own would resurrect the duplicate borrowed away. raise RuntimeError("SessionStore SQLite handle unavailable") try: return AsyncSessionDB(acquire()) except Exception as exc: logger.warning("SQLite session store not available: %s", exc) raise def _recovered() -> None: self._session_db_init_error = None logger.info("SQLite session store recovered") return cache.get(path, _open, raise_on_error=raise_on_error, on_recovered=_recovered) @property def _session_db(self) -> Any: """The AsyncSessionDB for the active profile scope, or a pinned override (assigning ``runner._session_db`` pins it for every later read — tests install fakes/None this way).""" if self._session_db_pinned is not _SESSION_DB_UNPINNED: return self._session_db_pinned return self._open_session_db_for_active_scope() @_session_db.setter def _session_db(self, value) -> None: self._session_db_pinned = value def close_all_session_db_handles(self) -> None: """Close every per-profile AsyncSessionDB this runner opened. Drained under the lock, closed outside it; a pinned handle is the pinner's to close. Wrappers BORROWED from ``session_store`` are skipped: the store's sweep (runs first) closes them. See #98573. """ def _close(db) -> None: if getattr(db, "__dict__", {}).get("_hermes_borrowed_handle"): return inner = getattr(db, "_db", db) if inner is None or not hasattr(inner, "close"): return # Shared instances no-op on close() (the registry owns the lifecycle). Release the refcount # instead (#90837). from hermes_state_registry import release_or_close try: release_or_close(inner) except Exception as exc: logger.debug("SessionDB close error during handle sweep: %s", exc) self._session_db_handle_cache.close_all(_close) def _wire_teams_pipeline_runtime(self) -> None: """Bind the Teams meeting pipeline runtime to Graph webhook ingress (no-op if adapter/plugin off).""" if Platform.MSGRAPH_WEBHOOK not in self.adapters: return if not _teams_pipeline_plugin_enabled(): logger.debug("Teams pipeline plugin is disabled; skipping runtime wiring") return try: from plugins.teams_pipeline.runtime import bind_gateway_runtime except Exception as exc: logger.warning("Teams pipeline runtime import failed: %s", exc) return try: bound = bind_gateway_runtime(self) except Exception as exc: logger.warning("Teams pipeline runtime wiring failed: %s", exc) return if bound: logger.info("Teams pipeline runtime bound to msgraph webhook ingress") elif self._teams_pipeline_runtime_error: logger.warning( "Teams pipeline runtime unavailable: %s", self._teams_pipeline_runtime_error) def _warn_if_docker_media_delivery_is_risky(self) -> None: """Warn when Docker-backed gateways lack an explicit export mount: MEDIA delivery runs in the gateway process, so model-emitted paths like `/output/report.txt` must be host-readable.""" if os.getenv("TERMINAL_ENV", "").strip().lower() != "docker": return connected = self.config.get_connected_platforms() messaging_platforms = [p for p in connected if p not in {Platform.LOCAL, Platform.API_SERVER, Platform.WEBHOOK}] if not messaging_platforms: return raw_volumes = os.getenv("TERMINAL_DOCKER_VOLUMES", "").strip() volumes: list[str] = [] if raw_volumes: try: parsed = json.loads(raw_volumes) if isinstance(parsed, list): volumes = [str(v) for v in parsed if isinstance(v, str)] except Exception: logger.debug("Could not parse TERMINAL_DOCKER_VOLUMES for gateway media warning", exc_info=True) for spec in volumes: match = _DOCKER_VOLUME_SPEC_RE.match(spec) if match and match.group("container") in _DOCKER_MEDIA_OUTPUT_CONTAINER_PATHS: return logger.warning( "Docker backend is enabled for the messaging gateway but no explicit host-visible " "output mount (for example '/home/user/.hermes/cache/documents:/output') is configured. " "This is fine if the model already emits host-visible paths, but MEDIA file delivery can fail " "for container-local paths like '/workspace/...' or '/output/...'.") _VOICE_MODE_PATH = _hermes_home / "gateway_voice_mode.json" should_exit_cleanly = property(lambda self: self._exit_cleanly) should_exit_with_failure = property(lambda self: self._exit_with_failure) exit_reason = property(lambda self: self._exit_reason) exit_code = property(lambda self: self._exit_code) def _session_key_for_source(self, source: SessionSource) -> str: """Resolve the current session key for a source, honoring gateway config when available.""" if hasattr(self, "session_store") and self.session_store is not None: try: session_key = self.session_store._generate_session_key(source) if isinstance(session_key, str) and session_key: return session_key except Exception: pass config = getattr(self, "config", None) # Mirror SessionStore._resolve_profile_for_key so this fallback yields the primary path's # namespace: None (legacy agent:main) unless multiplexing is on, then the pinned identity's # runtime profile, the source stamp, or the active profile. from gateway.session_identity import identity_of identity = identity_of(source) _profile = None if identity is not None: _profile = identity.session_key_profile elif getattr(config, "multiplex_profiles", False): if source.profile: _profile = source.profile else: try: from hermes_cli.profiles import get_active_profile_name _profile = get_active_profile_name() or "default" except Exception: _profile = None return build_session_key( source, group_sessions_per_user=getattr(config, "group_sessions_per_user", True), thread_sessions_per_user=getattr(config, "thread_sessions_per_user", False), profile=_profile) # Telegram General topic in forum-enabled private chats: clients omit message_thread_id or send "1"; both = root. _TELEGRAM_GENERAL_TOPIC_IDS = frozenset({"", "1"}) _TELEGRAM_LOBBY_REMINDER_COOLDOWN_S = 30.0 def _normalize_source_for_session_key(self, source: SessionSource) -> SessionSource: """Apply Telegram DM topic recovery to a source for session-key purposes. Always derive override storage keys from the result: ``_handle_message_with_agent`` rewrites ``thread_id`` before deriving the session key, so keys from the raw ``event.source`` are never read next turn. ``_handle_message_with_agent`` rewrites ``source.thread_id`` via ``_recover_telegram_topic_thread_id`` *before* deriving the session key for a normal message turn (a lobby/stripped reply gets pinned to the user's last-active topic). Session-scoped command handlers like ``/model`` and ``/reasoning`` derive their override key from the raw inbound ``event.source``, which skips that recovery — so the override is stored under a different key than the next message turn reads, and the override is silently dropped on Telegram forum topics and after compression session splits (#30479). """ try: recovered = self._recover_telegram_topic_thread_id(source) except Exception: return source return source if recovered is None else dataclasses.replace(source, thread_id=recovered) def _resolve_session_key_or_none(self, source, session_key: Optional[str]) -> Optional[str]: """``session_key`` if given, else the key for ``source`` (None when it cannot be derived).""" if session_key or source is None: return session_key try: return self._session_key_for_source(source) except Exception: return None def _running_agent_count(self) -> int: return len(self._running_agents) def _status_action_label(self) -> str: return "restart" if self._restart_requested else "shutdown" def _status_action_gerund(self) -> str: """Localized "restarting" / "shutting down" for the busy/drain notices shown in chat.""" return t("gateway.busy.action_restarting") if self._restart_requested else t("gateway.busy.action_shutting_down") def _update_runtime_status(self, gateway_state: Optional[str] = None, exit_reason: Optional[str] = None) -> None: if getattr(self, "_startup_cleanup_only", False): return # cleanup of a start that never ran keeps its startup verdict (startup_failed) # ``active_work`` names each unit only while draining — that is when an observer (``hermes # update``) needs to know WHAT holds the gateway open; a per-turn write would be wasted I/O. active_work = self._describe_active_work() if gateway_state == "draining" else None _write_runtime_status_quiet( gateway_state=gateway_state, exit_reason=exit_reason, restart_requested=self._restart_requested, active_agents=self._active_work_count(), active_work=active_work) def _persist_active_agents(self) -> None: """Persist the live in-flight agent count to ``gateway_state.json`` at every turn boundary. Passes ONLY ``active_agents`` so the read-merge-write keeps lifecycle state (gateway_state=None would clobber it). Best-effort: a failed write must never disrupt a turn.""" _write_runtime_status_quiet(active_agents=self._active_work_count()) def _running_agent_ids(self) -> set: """``id()`` of every agent mid-turn — identity-keyed so the lookup is O(1) and independent of ``AIAgent.__eq__`` (MagicMock overrides it in tests).""" return {id(a) for _, a in self._running_agent_items() if a is not None and a is not _AGENT_PENDING_SENTINEL} def _snapshot_running_agents(self) -> dict[str, Any]: return {k: a for k, a in self._running_agent_items() if a is not _AGENT_PENDING_SENTINEL} # ---- Tunables consumed by the run_* mixins (kept on the class: tests and plugins patch them) ---- # Per-session pending follow-up cap for busy_input_mode=queue (and paths sharing that entry point): # a stuck agent + rapid-fire user must not grow the overflow list unboundedly. _BUSY_QUEUE_MAX_PENDING = 32 @dataclasses.dataclass class _BusySteerOutcome: effective_mode: str demoted_for_subagents: bool demoted_for_compression: bool steered: bool redirected: bool # Worker bound for _cleanup_agent_resources: sync, can block long (subprocess teardown, memory IO). _CLEANUP_TIMEOUT_S = 30.0 # Budget for one finalize_session() dispatch (plugin on_session_finalize hooks + Relay close): # enough for a normal trace-export flush, small enough a wedged plugin can't eat the stop window. _FINALIZE_TIMEOUT_S = 10.0 _STUCK_LOOP_THRESHOLD = 3 # restarts while active before auto-suspend _STUCK_LOOP_FILE = ".restart_failure_counts" # Reasons set by _stop_impl() on force-interrupt; "restart_interrupted" by recover_interrupted_turns() # for a crash-left turn marker (no .clean_shutdown marker). All mean "killed mid-turn" -> startup # auto-resume. _AUTO_RESUME_REASONS = frozenset({"restart_timeout", "shutdown_timeout", "restart_interrupted"}) _MAX_SUPERVISED_RESTARTS = 5 # Ran this long before crashing = HEALTHY (isolated crash, not a crash-loop); restart counter resets. _SUPERVISED_HEALTHY_SECS = 300 # Slow respawn tier once the watcher's restart budget is spent; long on purpose (crashes on contact). _RECONNECT_WATCHER_SLOW_RETRY_SECS = 300 # Slow-tier respawns while work is queued; if 30 min of 5-min retries can't keep it up, fail loudly. _MAX_SLOW_WATCHER_RESPAWNS = 6 _TELEGRAM_CAPABILITY_HINT_COOLDOWN_S = 300.0 _APPROVAL_TIMEOUT_SECONDS = 300 # 5 minutes _MAX_INTERRUPT_DEPTH = 3 # Cap recursive interrupt handling # Command-specific mid-run reject texts (busy_policy == "reject" with a busy_handler naming an # entry here); all other rejected commands get the generic text in _dispatch_busy_slash_command. # Values are catalog keys; ``run_busy._dispatch_busy_slash_command`` resolves them with ``t()``. _BUSY_REJECT_TEXT: dict[str, str] = { "model": "gateway.busy.reject_model", "codex-runtime": "gateway.busy.reject_codex_runtime", "moa": "gateway.busy.reject_moa"} def _active_profile_name(self) -> str: """Return the profile name this gateway represents.""" try: from hermes_cli.profiles import get_active_profile_name return get_active_profile_name() or "default" except Exception: return "default" def _is_user_authorized_for_source( self, source: SessionSource, *, allow_adapter_delegation: bool = True) -> bool: """Authorize under the live transport's profile, not the routed runtime (which need not copy the shared bot token/allowlist); the transport home is stamped on the source for this read only.""" def _check() -> bool: # Keep the one-argument seam used by plugins/tests; pass the keyword only when disabling. if allow_adapter_delegation: return self._is_user_authorized(source) return self._is_user_authorized(source, allow_adapter_delegation=False) return self._under_authorization_profile(source, _check) def _admit_bot_message_for_source(self, source: SessionSource) -> bool: """Count a bot message under the profile that authorized it, so the guard's peek, count and config all read the transport profile's ``gateway.bot_loop_guard``.""" return self._under_authorization_profile(source, lambda: self._admit_bot_message(source)) def _under_authorization_profile(self, source: SessionSource, check): authorization_home = self._authorization_home_for_source(source) if authorization_home is None: return check() with _profile_runtime_scope(Path(authorization_home)): return check() def _cache_session_source(self, session_key: str, source) -> None: if not session_key or source is None: return cached_sources = getattr(self, "_session_sources", None) if cached_sources is None: cached_sources = OrderedDict() self._session_sources = cached_sources try: cached_sources[session_key] = dataclasses.replace(source) except Exception: logger.debug("Failed to cache live session source for %s", session_key, exc_info=True) return try: cached_sources.move_to_end(session_key) max_size = getattr(self, "_session_sources_max", 512) while len(cached_sources) > max_size: cached_sources.popitem(last=False) except Exception: pass @property def async_session_store(self) -> AsyncSessionStore: """Return the single async facade for this runner's SessionStore.""" facade = getattr(self, "_async_session_store", None) if facade is None or facade._store is not self.session_store: facade = AsyncSessionStore(self.session_store) self._async_session_store = facade return facade def _get_cached_session_source(self, session_key: str): cached_sources = getattr(self, "_session_sources", None) if session_key else None if not cached_sources: return None source = cached_sources.get(session_key) if source is not None: with suppress(Exception): cached_sources.move_to_end(session_key) return source @dataclasses.dataclass class _HygieneSettings: """Resolved session-hygiene configuration for one inbound turn.""" model: str threshold_pct: float compression_enabled: bool hard_msg_limit: int timeout_seconds: float total_ceiling_seconds: float max_turn_hold_seconds: float failure_cooldown_seconds: float config_context_length: Optional[int] provider: Optional[str] base_url: Optional[str] api_key: Optional[str] data: Any @dataclasses.dataclass class _HygieneAttempt: """One detached hygiene compression attempt. ``cleanup_deferred`` is shared mutable state: wait handlers set it on raise paths; the owning ``finally`` reads it to decide on cleanup now.""" agent: Any meta: Any commit_fence: Any = None future: Any = None wait_started: float = 0.0 cleanup_deferred: bool = False history: Any = None def _thread_metadata_for_source( self, source, reply_to_message_id: Optional[str] = None) -> Optional[dict[str, Any]]: """Build the metadata dict platforms need for thread-aware replies.""" metadata = self._thread_metadata_for_target( getattr(source, "platform", None), getattr(source, "chat_id", None), getattr(source, "thread_id", None), chat_type=getattr(source, "chat_type", None), reply_to_message_id=reply_to_message_id or getattr(source, "message_id", None)) if getattr(source, "platform", None) == Platform.SLACK: # Per-turn egress identity: Slack chat.startStream needs recipient_user_id/team_id; the relay # adapter's _with_scope fallback reads per-chat caches a CONCURRENT turn overwrites. # Slack's chat.startStream requires recipient_user_id (+ recipient_team_id) when streaming to a # channel, and the relay connector fills those from metadata.user_id / metadata.scope_id. The # relay adapter's _with_scope fallback resolves BOTH from per-chat caches keyed only by chat_id # — mutable state that a CONCURRENT turn overwrites: two users with overlapping turns in one # channel would open U1's stream with U2 as the recipient. Stamp the authentic per-turn values # from THIS turn's source here, where they are still turn-scoped; _with_scope only fills keys # that are absent, so the cache degrades to what it should be — a restart/synthetic-send # fallback. See #210. team_id = getattr(source, "scope_id", None) user_id = getattr(source, "user_id", None) if team_id or user_id: metadata = dict(metadata or {}) if team_id: metadata["slack_team_id"] = str(team_id) metadata.setdefault("scope_id", str(team_id)) if user_id: metadata.setdefault("user_id", str(user_id)) from gateway.session_context import source_route_metadata metadata = source_route_metadata(source, metadata) # Routed profile for shared state.db namespaces: under profile_routes the transport adapter's # stamp is not the profile that wrote the binding (Telegram prune path needs it). # See #76423. profile = str(getattr(source, "profile", None) or "").strip() if profile and metadata is not None: metadata = dict(metadata) metadata["hermes_profile"] = profile return metadata def _thread_metadata_for_target( self, platform: Optional[Platform], chat_id: Optional[str], thread_id: Optional[str], *, chat_type: Optional[str] = None, reply_to_message_id: Optional[str] = None, adapter: Optional[Any] = None) -> Optional[dict[str, Any]]: """Build thread metadata for synthetic sends that only have routing state.""" if thread_id is None: return None metadata: dict[str, Any] = {"thread_id": thread_id} if self._is_telegram_dm_topic_target( platform, chat_id, thread_id, chat_type=chat_type, adapter=adapter): metadata["telegram_dm_topic_reply_fallback"] = True # DM topic lanes need direct_messages_topic_id so synthetic sends reach the topic without a reply anchor. tid = str(thread_id) if tid and tid not in {"", "1"}: metadata["direct_messages_topic_id"] = tid if reply_to_message_id is not None: metadata["telegram_reply_to_message_id"] = str(reply_to_message_id) if platform == Platform.SLACK and reply_to_message_id is not None: # Slack's reply_in_thread=false path uses message_id to tell real threads from synthetic keys. metadata["message_id"] = str(reply_to_message_id) return metadata @staticmethod def _is_telegram_dm_topic_target( platform: Optional[Platform], chat_id: Optional[str], thread_id: Optional[str], *, chat_type: Optional[str] = None, adapter: Optional[Any] = None) -> bool: """Return True when a target is a Telegram private DM topic lane.""" if platform != Platform.TELEGRAM or thread_id is None: return False if chat_type == "dm": return True # Resolve the lookup on the CLASS, not the instance: getattr() on a MagicMock auto-creates callable # children, so an instance lookup would report a DM topic for every test double. Only a dict counts. if adapter is not None and chat_id: get_dm_topic_info = getattr(type(adapter), "_get_dm_topic_info", None) if callable(get_dm_topic_info): try: topic_info = get_dm_topic_info(adapter, str(chat_id), str(thread_id)) except Exception: logger.debug("Failed to inspect Telegram DM topic metadata", exc_info=True) else: return isinstance(topic_info, dict) return False _reply_anchor_for_event = staticmethod(_reply_anchor_for_event) # Built-in platforms where ``/update`` is allowed (programmatic interfaces must not trigger updates). # Plugin-migrated platforms declare ``allow_update_command=True`` on their ``PlatformEntry`` instead. _UPDATE_ALLOWED_PLATFORMS = frozenset({ Platform.TELEGRAM, Platform.SLACK, Platform.WHATSAPP, Platform.SIGNAL, Platform.MATRIX, Platform.EMAIL, Platform.SMS, Platform.DINGTALK, Platform.FEISHU, Platform.WECOM, Platform.WECOM_CALLBACK, Platform.WEIXIN, Platform.BLUEBUBBLES, Platform.QQBOT, Platform.LOCAL, }) def _set_session_env(self, context: SessionContext) -> list: """Set session context variables (contextvars, not os.environ, so concurrent messages can't overwrite each other). Returns reset tokens for ``_clear_session_env`` in a ``finally``.""" from gateway.session_context import set_session_vars # Async-delivery capability tells async tools whether this channel can wake a later turn. Default # True keeps CLI/unknown paths working; stateless adapters (api_server) declare False. _adapter = (getattr(self, "adapters", None) or {}).get(context.source.platform) _async_delivery = getattr(_adapter, "supports_async_delivery", True) # #98619: an admitted API turn carries its session-id provenance in the admission settings, # and its rebound browser-controller identity; every other platform leaves both undeclared. from gateway.session_api_turn import api_session_vars _api_vars = api_session_vars() _history_delivery = _api_vars.pop("session_history_delivery", None) return set_session_vars( platform=context.source.platform.value, chat_id=context.source.chat_id, chat_type=str(context.source.chat_type) if context.source.chat_type else "", chat_name=context.source.chat_name or "", thread_id=str(context.source.thread_id) if context.source.thread_id else "", user_id=str(context.source.user_id) if context.source.user_id else "", user_id_alt=str(context.source.user_id_alt) if context.source.user_id_alt else "", user_name=str(context.source.user_name) if context.source.user_name else "", scope_id=str(getattr(context.source, "scope_id", "") or ""), parent_chat_id=str(getattr(context.source, "parent_chat_id", "") or ""), session_key=context.session_key, # A cached agent binds HERMES_SESSION_ID only on the turn that built it; every later turn # (and a command it yields to the background) must still see its own session. session_id=context.session_id or "", message_id=str(context.source.message_id) if context.source.message_id else "", profile=getattr(context.source, "profile", "") or "", async_delivery=_async_delivery, cron_session="", session_history_delivery=_history_delivery, **_api_vars) def _clear_session_env(self, tokens: list) -> None: """Restore session context variables to their pre-handler values.""" from gateway.session_context import clear_session_vars clear_session_vars(tokens) @_contextmanager def _session_env_scope(self, context: SessionContext): """Bind session context variables for the duration of the block, e.g. a plugin command handler invoked outside the normal agent-turn path (``_set_session_env`` is otherwise only reached there). Always cleared on exit, including on exception.""" tokens = self._set_session_env(context) try: yield finally: self._clear_session_env(tokens) async def _run_in_executor_with_context(self, func, *args): """Run blocking work in the thread pool while preserving session contextvars.""" loop = asyncio.get_running_loop() ctx = copy_context() return await loop.run_in_executor(self._get_executor(), ctx.run, func, *args) async def _run_housekeeping_in_executor(self, func, *args): """Run best-effort session housekeeping off the TURN pool. Callers of this helper bound their await with ``asyncio.wait_for`` and, on timeout, log "the worker thread is left to finish on its own" and proceed. That bounds the AWAIT but NOT the OCCUPANCY: a ``concurrent.futures`` work item that has already begun executing is not cancellable, so an abandoned worker keeps its thread until its blocking call returns. The turn pool is unbounded (one thread per turn), so sharing it would let wedged housekeeping grow threads without limit. Housekeeping therefore gets its own bounded pool; exhausting that one delays only more housekeeping, which is best-effort by construction. """ loop = asyncio.get_running_loop() return await loop.run_in_executor( self._get_housekeeping_executor(), copy_context().run, func, *args) def _get_or_create_pool(self, attr: str, make_pool: Callable[[], concurrent.futures.Executor]) -> concurrent.futures.Executor: """Return (creating under ``_executor_lock``) the pool at ``attr``; one lock + closing flag fences both.""" lock = getattr(self, "_executor_lock", None) if lock is None: lock = threading.Lock() self._executor_lock = lock with lock: if getattr(self, "_executor_closing", False): raise RuntimeError("Gateway is shutting down; executor unavailable") executor = getattr(self, attr, None) if executor is None or getattr(executor, "_shutdown", False): executor = make_pool() setattr(self, attr, executor) return executor def _get_executor(self) -> concurrent.futures.Executor: """Return the gateway-owned, UNBOUNDED executor for turn bodies and other blocking agent work. A turn body holds its thread for the whole turn (every tool call blocks), so a finite pool silently queued already-accepted turns behind running ones. Actual bound: one live turn per session, plus turns abandoned by the inactivity timeout (run_turn keeps no handle; their thread runs to completion). ``max_concurrent_sessions`` is the only admission cap and is unset by default. """ return GatewayRunner._get_or_create_pool( self, "_executor", lambda: _UnboundedThreadExecutor(thread_name_prefix="hermes-gateway")) def _get_housekeeping_executor(self) -> concurrent.futures.Executor: """Return the gateway-owned executor for best-effort session housekeeping.""" return GatewayRunner._get_or_create_pool( self, "_housekeeping_executor", lambda: concurrent.futures.ThreadPoolExecutor( max_workers=_HOUSEKEEPING_MAX_WORKERS, thread_name_prefix="hermes-gateway-hk")) @staticmethod def _stop_pool(executor) -> list: """Shut ``executor`` down without waiting; return its threads to join (`_threads` absent on test doubles).""" if executor is None: return [] try: executor.shutdown(wait=False, cancel_futures=True) except TypeError: executor.shutdown(wait=False) return list(getattr(executor, "_threads", None) or ()) def _shutdown_executor(self, drain_timeout: float = 0.0) -> int: """Stop the gateway-owned pools; returns the number of worker threads still running. ``drain_timeout=0`` is fire-and-forget; shutdown passes a bounded budget so blocking DB work cannot outlive ``SessionDB.close()``. ``cancel_futures`` only drops unstarted work and cancelling a ``run_in_executor`` awaitable does not stop its thread, so running workers are joined — on the turn pool AND the housekeeping pool, both of which write to SessionDB.""" lock = getattr(self, "_executor_lock", None) if lock is None: return 0 with lock: self._executor_closing = True executor = getattr(self, "_executor", None) self._executor = None housekeeping = getattr(self, "_housekeeping_executor", None) self._housekeeping_executor = None # Housekeeping workers run SessionDB writes too (session finalize, agent cleanup), so a wedged # one is exactly the mid-write worker the #101093 skip-close heuristic exists for. Both pools # are therefore joined under the SAME drain deadline and both contribute to the live count. # Class-qualified: run_shutdown and tests invoke these unbound on a duck-typed `self`. workers = GatewayRunner._stop_pool(executor) + GatewayRunner._stop_pool(housekeeping) deadline = time.monotonic() + max(float(drain_timeout or 0.0), 0.0) for worker in workers: remaining = deadline - time.monotonic() if remaining <= 0: break worker.join(remaining) return sum(1 for worker in workers if worker.is_alive()) # (section, key) config values baked into the agent at construction: a change MUST invalidate the # cached agent or a mid-gateway edit is silently ignored. Add new baked-in settings here. # _MAX_INTERRUPT_DEPTH = 3 # Cap recursive interrupt handling (#816) _CACHE_BUSTING_CONFIG_KEYS: tuple = ( ("model", "context_length"), ("compression", "enabled"), ("compression", "progress_notices"), ("compression", "threshold"), ("compression", "model_thresholds"), ("compression", "threshold_tokens"), ("compression", "codex_gpt55_autoraise"), ("compression", "codex_app_server_auto"), ("compression", "codex_responses_native"), ("compression", "codex_responses_compact_threshold"), ("compression", "in_place"), ("compression", "checkpoint_required"), ("compression", "micro_compact"), ("compression", "micro_compact_every_n_turns"), ("compression", "micro_compact_defrag_threshold_tokens"), ("compression", "target_ratio"), ("compression", "tail_mode"), ("compression", "protect_last_n"), ("compression", "proactive_prune_tokens"), ("compression", "proactive_prune_min_result_chars"), ("compression", "proactive_prune_min_reclaim_tokens"), ("compression", "min_tail_user_messages"), ("agent", "disabled_toolsets"), ("memory", "provider"), ("checkpoints", "enabled"), ("checkpoints", "max_snapshots"), ("checkpoints", "max_total_size_mb"), ("checkpoints", "max_file_size_mb")) @staticmethod def _init_cached_agent_for_turn(agent: Any, interrupt_depth: int) -> None: """Reset per-turn state on a cached agent before a new turn starts. The activity ts/desc/provenance triple resets together and only at depth 0 — else a session idle 29 min trips the watchdog before the first call; interrupt-recursive turns keep it so stuck-turn idle time accumulates to the 30-min timeout. ``_last_activity_ts``, ``_last_activity_desc``, and ``_last_activity_provenance`` are only reset for fresh external turns (depth 0); they are a semantic triple - description and provenance describe the activity *at* ts, so updating one without the others would make get_activity_summary() misleading. See #15654, #9051. """ if interrupt_depth == 0: agent._last_activity_ts = time.time() agent._last_activity_desc = "starting new turn (cached)" agent._last_activity_provenance = ActivityProvenance.UNKNOWN # Reset the SessionDB flush cursor so the new turn's messages are fully persisted — a stale # value from the previous turn makes `_flush_messages_to_session_db` skip new rows. # See #44327. if hasattr(agent, "_last_flushed_db_idx"): agent._last_flushed_db_idx = 0 agent._api_call_count = 0 def _profile_name_for_source( self, source: SessionSource, adapter_profile: Optional[str] = None, ) -> Optional[str]: """Resolve the profile name for an inbound source via configured routes (most specific wins). ``None`` = default/active profile (or, for a secondary adapter, its own profile — the caller stamps it). Gated on ``multiplex_profiles``, since the scoped run only activates under multiplexing; otherwise keys would be profile-namespaced while the agent ran in ``agent:main``. ``adapter_profile`` is the profile owning the receiving bot; only routes declaring it as ``bot_profile`` apply (#104933).""" config = getattr(self, "config", None) if not getattr(config, "multiplex_profiles", False): return None routes = getattr(config, "profile_routes", None) if not routes: return None if adapter_profile is None: # Sources built outside ``build_source`` may still carry the receiving adapter as provenance. owner = self._transport_owner(source) if callable(getattr(source, "_transport_adapter_ref", None)) else None if isinstance(owner, tuple): adapter_profile = owner[1] from gateway.profile_routing import ProfileRouteRejected, match_profile_route try: matched = match_profile_route( routes, platform=source.platform.value, guild_id=getattr(source, "guild_id", None), chat_id=source.chat_id, thread_id=getattr(source, "thread_id", None), parent_chat_id=getattr(source, "parent_chat_id", None), adapter_profile=adapter_profile, user_id=getattr(source, "user_id", None)) except Exception as exc: logger.warning( "Rejecting %s/%s: profile route matching failed", source.platform, source.chat_id, exc_info=True) raise ProfileRouteRejected("matcher") from exc if matched: try: served = {name for name, _home in _multiplex_profile_homes(config)} except Exception as exc: logger.warning( "Rejecting profile route %r because the served-profile set could not be resolved", matched.name, exc_info=True) raise ProfileRouteRejected(matched.name) from exc if matched.profile not in served: logger.warning( "Rejecting profile route %r: target profile %r is not served", matched.name, matched.profile) raise ProfileRouteRejected(matched.name, matched.profile) return matched.profile logger.debug( "No profile route matched: platform=%s chat_id=%s thread_id=%s parent_chat_id=%s", source.platform.value, source.chat_id, getattr(source, "thread_id", None), getattr(source, "parent_chat_id", None)) return None def _resolve_profile_home_for_source(self, source: SessionSource) -> Path: """Resolve which profile's HERMES_HOME serves this source: the pinned identity's runtime home, else ``source.profile``, then ``_profile_name_for_source`` (sources bypassing ``build_source``), then the active profile.""" from gateway.profile_routing import ProfileRouteRejected from gateway.session_identity import identity_of from hermes_cli.profiles import get_active_profile_name, get_profile_dir, profile_exists from hermes_constants import get_hermes_home identity = identity_of(source) if identity is not None: return identity.runtime_home explicit_profile = None # explicitly requested (source or routing) vs. default fallback try: name = (source.profile or "").strip() or self._profile_name_for_source(source) explicit_profile = name or None if not name: name = get_active_profile_name() or "default" profile_dir = get_profile_dir(name) if explicit_profile and not profile_exists(name): logger.warning( "Profile %r does not exist for source %s/%s (guild_id=%s), " "falling back to global HERMES_HOME", explicit_profile, source.platform.value, source.chat_id, getattr(source, "guild_id", None)) return get_hermes_home() return profile_dir except ProfileRouteRejected: raise except Exception: logger.warning( "Failed to resolve profile directory for source %s/%s (guild_id=%s), " "falling back to global HERMES_HOME: %s", source.platform.value, source.chat_id, getattr(source, "guild_id", None), explicit_profile or "(no profile)", exc_info=True) return get_hermes_home() @dataclasses.dataclass class _RunAgentDisplay: """Per-turn display / progress settings resolved by ``_run_agent_display_settings``.""" user_config: Any = None platform_key: Any = None enabled_toolsets: Any = None disabled_toolsets: Any = None resolve_display_setting: Any = None progress_mode: Any = None progress_grouping: Any = None _display_surface_mode: Any = None tool_progress_enabled: Any = None _live_status_mode: Any = None _live_status_adapter: Any = None log_mode_enabled: Any = None log_queue: Any = None interim_assistant_messages_enabled: Any = None _thinking_enabled: Any = None _native_slack_task_cards: Any = None needs_progress_queue: Any = None _generic_status_phrase: Any = None @dataclasses.dataclass class _RunAgentWorker: """Executor future + inactivity-watchdog handles for one ``_run_agent_inner`` turn.""" executor_task: Any = None agent_timeout: Optional[float] = None agent_warning: Optional[float] = None task_id: str = "" process_baseline: Any = None worker_done: Any = None timeout_fired: Any = None cleanup_lock: Any = None is_current: Any = None def _run_planned_stop_watcher( stop_event: threading.Event, runner, loop: asyncio.AbstractEventLoop, shutdown_handler, *, poll_interval: float = 0.5) -> None: """Poll for the planned-stop marker and trigger graceful shutdown (Windows lacks ``add_signal_handler``, so ``hermes gateway stop`` would never drain). Runs everywhere; on POSIX the signal handler consumes the marker first and ``_running``/``_draining`` guard re-triggers. On Windows, ``asyncio.add_signal_handler`` raises NotImplementedError for SIGTERM/SIGINT, so the standard signal-driven shutdown path never runs when ``hermes gateway stop`` signals the gateway. The consequence is that the drain loop is skipped — in-flight agent sessions are killed mid-turn and ``resume_pending`` is never set, so the next gateway boot has no idea those sessions need to be auto-resumed (issue #33778, v0.13.0 session-resume feature broken on native Windows). """ from gateway.status import ( _get_planned_stop_marker_path, planned_stop_marker_targets_self) marker_path = _get_planned_stop_marker_path() while not stop_event.is_set(): try: if ( marker_path.exists() and not getattr(runner, "_draining", False) and getattr(runner, "_running", False)): # A marker may target a PREVIOUS instance that exited before stop() cleaned up; # firing on it means an "UNKNOWN" exit and a watchdog crash-loop; probe unlinks stale. # A marker existing is NOT sufficient — it may have been written for a PREVIOUS gateway # instance (different PID) and left behind because that process exited before the CLI's # stop() could clean it up. Firing the handler on a stale/foreign marker drives the gateway # into shutdown, then consume_planned_stop_marker_for_self() correctly reports a PID # mismatch — but by then we're already stopping, so it's logged as an unexpected "UNKNOWN" # exit and the watchdog crash-loops the gateway (issue #34597, a regression from PR #33798 # which added this watcher without the PID check). Only fire when the marker actually # targets us. The probe is non-destructive on a match (the handler does the authoritative # consume on the loop thread) and self-heals by unlinking stale/malformed markers so they # cannot wedge a freshly booted gateway. if not planned_stop_marker_targets_self(): stop_event.wait(poll_interval) continue # Same path as a real signal; the handler consumes the marker (validates pid + start_time). loop.call_soon_threadsafe(shutdown_handler, None) break except Exception as _e: logger.debug("Planned-stop watcher tick error: %s", _e) stop_event.wait(poll_interval) def _housekeeping_chore(label: str, fn, *args, **kwargs) -> None: """Run one housekeeping chore; failures log at debug (a persistent failure such as a broken import after a partial update would otherwise warn every tick forever) and never stop the loop.""" try: fn(*args, **kwargs) except Exception as exc: logger.debug("%s error: %s", label, exc) def _housekeeping_channel_directory(adapters, loop) -> None: from gateway.channel_directory import build_channel_directory if loop is not None: # build_channel_directory is async (Slack web calls) and this is a background thread: # schedule onto the gateway loop and wait briefly so refresh failures still log. fut = safe_schedule_threadsafe( build_channel_directory(adapters), loop, logger=logger, log_message="Channel directory refresh scheduling error") if fut is not None: fut.result(timeout=30) def _housekeeping_paste_sweep() -> None: from hermes_cli.debug import _sweep_expired_pastes deleted, remaining = _sweep_expired_pastes() if deleted: logger.info("Paste sweep: deleted %d expired paste(s), %d pending", deleted, remaining) def _housekeeping_misfire_catch_up(cron_provider, adapters, loop) -> None: """External cron providers only: fire jobs whose time passed with no external fire delivered (dead loopback hop). No-op for the built-in ticker; enforces misfire_grace_minutes; CAS claim de-dupes.""" from cron.scheduler_provider import fire_overdue_jobs caught_up = fire_overdue_jobs(cron_provider, adapters=adapters, loop=loop) if caught_up: logger.info("Misfire catch-up: fired %d overdue job(s)", caught_up) def _housekeeping_curator() -> None: """maybe_run_curator() is gated by config.interval_hours (7 days default); this is the poll.""" from agent.curator import maybe_run_curator maybe_run_curator(idle_for_seconds=float("inf"), on_summary=lambda msg: logger.info("curator: %s", msg)) def _housekeeping_plugin_update_check() -> None: """Plugin update-check cadence (plugins_cadence): due-gated by plugins.auto_update_check_hours, read-only, receipt-surfaced; the opt-in auto-apply rides the manual update pipeline. A network error costs one warning and a stamped marker — never an apply.""" from hermes_cli.plugins_cadence import maybe_run_gateway_check maybe_run_gateway_check(log=logger) def _launch_sessions_dir(config) -> Optional[tuple[Path, Path]]: """``(launch home, its configured transcript dir)``, or ``None`` when the gateway carries none. MUST be called outside any profile scope — ``get_hermes_home()`` is what identifies the launch home. Consumed by :func:`_profile_sessions_dir`. """ sessions_dir = getattr(config, "sessions_dir", None) if sessions_dir is None: return None return get_hermes_home(), Path(sessions_dir) def _profile_sessions_dir(launch: Optional[tuple[Path, Path]]) -> Path: """Transcript dir of the profile currently in scope. ``gateway.sessions_dir`` overrides the LAUNCH profile's transcript dir only; every other served profile keeps ``/sessions``. Hardcoding ``/sessions`` for the launch home too wrote transcripts to the configured dir while the prune unlinked under the default one, orphaning every pruned session's ``.json``/``.jsonl``/``request_dump_*`` forever. """ home = get_hermes_home() if launch is not None and Path(launch[0]) == home: return Path(launch[1]) return home / "sessions" def _housekeeping_state_db_maintenance(launch: Optional[tuple[Path, Path]] = None) -> None: """Stale-session auto-archive plus auto-prune/VACUUM for ONE profile's state.db; both are gated by sessions.min_interval_hours (VACUUM additionally by its own throttles). Opens its own SessionDB — SQLite connections are thread-bound. Profile-scoped by its caller: ``acquire()``, ``get_hermes_home()`` and ``load_config()`` all resolve through the active scope, so an unscoped run swept only the LAUNCH profile's store with the LAUNCH profile's retention settings and a multiplexed secondary was never archived, pruned or vacuumed by anyone — the dashboard/serve trigger defers to the gateway for every profile a gateway owns (``web_server_sessions``). *launch* carries the launch home's configured transcript dir (:func:`_launch_sessions_dir`) so its override still governs its own profile.""" from hermes_cli.config import load_config as _load_full_config from hermes_state_registry import acquire, release_or_close _sess_cfg = (_load_full_config().get("sessions") or {}) if not (_sess_cfg.get("auto_archive", False) or _sess_cfg.get("auto_prune", False)): return _adb = acquire() try: if _sess_cfg.get("auto_archive", False): _adb.maybe_auto_archive( idle_days=float(_sess_cfg.get("auto_archive_days", 3)), min_interval_hours=int(_sess_cfg.get("min_interval_hours", 24))) if _sess_cfg.get("auto_prune", False): _adb.maybe_auto_prune_and_vacuum( retention_days=int(_sess_cfg.get("retention_days", 90)), min_interval_hours=int(_sess_cfg.get("min_interval_hours", 24)), min_vacuum_interval_days=int(_sess_cfg.get("min_vacuum_interval_days", 30)), vacuum=bool(_sess_cfg.get("vacuum_after_prune", True)), sessions_dir=_profile_sessions_dir(launch)) finally: release_or_close(_adb) def _housekeeping_deferred_fts_retry() -> None: """A SessionDB opened while another process held the rebuild lock fails closed onto the LIKE fallback and the gateway stays up for days. Non-blocking, rate-limited inside SessionDB; no-op when not stale.""" # Retry here, on the existing tick, against the shared instances this process already holds: # non-blocking admission, no new thread, rate-limited inside SessionDB. No-op when nothing is stale (one # attribute read per instance). See #100108. from hermes_state_registry import borrow_live_shared_session_dbs with borrow_live_shared_session_dbs() as _session_dbs: for _sdb in _session_dbs: _retry = getattr(_sdb, "retry_deferred_fts_recovery", None) if callable(_retry) and _retry(): logger.info( "Deferred state.db FTS rebuild completed in-process for %s; full-text search restored.", getattr(_sdb, "db_path", "state.db")) def _housekeeping_memory_trim() -> None: """Messaging-gateway counterpart to the TUI idle reaper; config-gated and rate-limited inside.""" from hermes_cli.mem_trim import trim_memory trim_memory(reason="messaging gateway housekeeping") def _housekeeping_checkpoint_prune() -> None: """Checkpoint store retention + size cap on a live timer; ``auto_prune_from_config`` gates on ``checkpoints.auto_prune`` and the 24h ``.last_prune`` marker. Off the startup path because its ``git gc`` can block for tens of seconds on a large store.""" from tools.checkpoint_maintenance import auto_prune_from_config auto_prune_from_config() def _drain_restart_safe_cron_deliveries(adapters, loop, runner=None) -> None: """Drain each profile's worker queue through its matching live adapters. A credential-less satellite profile (empty adapter map) drains through the primary's adapters routed by its own profile routes.""" from cron import scheduler as cron_scheduler from cron import scheduler_preflight as sched_preflight if runner is None: if adapters is not None: cron_scheduler.drain_delivery_queue(adapters, loop) return for profile_name, profile_home in _handoff_watch_scopes(runner): if profile_name is None: profile_adapters = adapters else: profile_adapters = getattr(runner, "_profile_adapters", {}).get(profile_name) if profile_adapters is None: continue with _profile_runtime_scope(profile_home or get_hermes_home()): if profile_name is not None and not profile_adapters and adapters: routes = sched_preflight._primary_profile_routes_for_current_home() if routes: profile_adapters = sched_preflight.SharedRouteAdapters(adapters, routes) cron_scheduler.drain_delivery_queue(profile_adapters, loop) def _start_gateway_housekeeping( stop_event: threading.Event, adapters=None, loop=None, interval: int = 60, cron_provider=None, runner=None, cron_thread=None, ): """Background thread for gateway-only periodic chores (NOT cron). Separate from the cron trigger so chores run under any ``CronScheduler`` provider (external scale-to-zero has no 60s loop). Cadences are ticks of ``interval``; inner gates own the real cadence.""" from gateway.run_delivery_queue_watch import DRAIN_LABEL, DeliveryQueueWatch, wait_for_next_tick from gateway.run_profile_reconcile import _mcp_config_reconciler, profile_scoped_chore from gateway.run_housekeeping_disk import housekeeping_media_caches, maybe_post_auto_prune_prompt chores: list[tuple[int, str, Any]] = [ # First every tick: re-stamp ``updated_at`` in gateway_state.json so it is a real heartbeat. # ``hermes gateway status`` / ``/api/status`` warn when it ages past 2x ``interval`` with the # PID alive — the thread (or a chore blocked on the loop) wedged (#113372). Runs first so a # wedged chore stops the NEXT stamp instead of a slow one delaying this tick's. (1, "Runtime heartbeat", _write_runtime_status_quiet)] if adapters is not None or runner is not None: # Restart-safe cron workers run outside the gateway cgroup and queue their final send for # whichever gateway is live; drained here (not the scheduler tick) so external providers get it too. chores.append((1, DRAIN_LABEL, lambda: _drain_restart_safe_cron_deliveries(adapters, loop, runner))) chores += [ (5, "Channel directory refresh", lambda: adapters and _housekeeping_channel_directory(adapters, loop)), (60, "Media cache cleanup", housekeeping_media_caches), (60, "Paste sweep", _housekeeping_paste_sweep)] # Low disk + auto-prune undecided: ask on the home channel (two syscalls; snoozed per post). chores.append((1, "Auto-prune low-disk prompt", lambda: maybe_post_auto_prune_prompt(runner, loop))) if cron_provider is not None: chores.append((5, "Misfire catch-up sweep", lambda: _housekeeping_misfire_catch_up(cron_provider, adapters, loop))) if cron_thread is not None: # The ticker's own guards keep its loop alive; this is the outer layer for a thread that has # already ended (#111010). Runs every tick so the outage is bounded by one housekeeping interval. chores.append((1, "Cron ticker supervisor", cron_thread.restart_if_dead)) chores += [ # Per served profile: each profile has its own skills tree, curator state, Nous login # and state.db. (60, "Curator tick", profile_scoped_chore(runner, _housekeeping_curator)), (60, "state.db maintenance tick", profile_scoped_chore( runner, # Default-bound now, i.e. OUTSIDE any profile scope: this is the launch home's override. lambda _launch=_launch_sessions_dir(getattr(runner, "config", None)): _housekeeping_state_db_maintenance(_launch))), # Due-gated inside: the first tick after startup runs an overdue check, not tick 60. # Per served profile: plugins dir, last-run marker and plugins.auto_apply are all the # profile's own (get_hermes_home()/load_config_readonly() bind to the scope). (1, "Plugin update check", profile_scoped_chore(runner, _housekeeping_plugin_update_check)), (1, "Deferred FTS retry tick", _housekeeping_deferred_fts_retry), (1, "gateway housekeeping memory trim", _housekeeping_memory_trim), (1, "MCP config reconcile", _mcp_config_reconciler(runner)), # Last: a real prune can hold this thread for a while; every other chore of the tick runs first. (1, "Checkpoint prune tick", _housekeeping_checkpoint_prune)] # Between ticks the queue file is watched so a worker's send goes out when it is queued, # not up to ``interval`` later (#117307); the tick's drain above remains the fallback. queue_watch = None if adapters is not None or runner is not None: def served_homes() -> list: return [home for _name, home in _handoff_watch_scopes(runner)] if runner is not None else [None] queue_watch = DeliveryQueueWatch( served_homes, lambda: _drain_restart_safe_cron_deliveries(adapters, loop, runner)) logger.info("Gateway housekeeping started (interval=%ds)", interval) tick_count = 0 while not stop_event.is_set(): tick_count += 1 for every, label, fn in chores: if tick_count % every == 0: _housekeeping_chore(label, fn) wait_for_next_tick(stop_event, interval, queue_watch, _housekeeping_chore) logger.info("Gateway housekeeping stopped") def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, interval: int = 60): """DEPRECATED shim — runs ONLY the built-in in-process cron tick loop; the trigger now lives behind the ``CronScheduler`` provider and housekeeping in ``_start_gateway_housekeeping``.""" from cron.scheduler_provider import InProcessCronScheduler InProcessCronScheduler().start(stop_event, adapters=adapters, loop=loop, interval=interval) def _stop_cron_provider(provider) -> None: """Stop a cron provider without letting it choose the gateway exit code.""" try: provider.stop() except SystemExit as exc: logger.warning( "Cron provider stop() attempted to exit the gateway with code %s; ignoring", exc.code) except Exception as exc: logger.debug("Cron provider stop() error: %s", exc) # Cron thread blocks on future.result(timeout=60) (cron/scheduler.py::_deliver_result) + margin. _CRON_SHUTDOWN_DRAIN_TIMEOUT = 65.0 # Housekeeping's channel-directory refresh blocks on fut.result(timeout=30); cover that + margin. _HOUSEKEEPING_SHUTDOWN_DRAIN_TIMEOUT = 35.0 async def _await_thread_exit( thread: Optional[threading.Thread], timeout: float, poll: float = 0.1) -> bool: """Wait for a daemon thread to exit WITHOUT blocking the event loop; True if it exited in time. A synchronous ``join()`` freezes the loop — fatal for the cron ticker, whose in-flight delivery is a coroutine on *this* loop: it could never run, so the join timed out and the message dropped. See #58818. """ if thread is None: return True deadline = asyncio.get_running_loop().time() + max(0.0, timeout) while thread.is_alive() and asyncio.get_running_loop().time() < deadline: await asyncio.sleep(poll) return not thread.is_alive() async def _shutdown_mcp_servers_nonblocking(timeout: float = 5.0, config: Any = None) -> bool: """Close MCP servers off-loop with a bounded wait; True when done within ``timeout``. ``shutdown_mcp_servers()`` can block ~15s; on the loop thread short-grace supervisors (s6 3s) SIGKILL us before ``mark_exited()`` runs, so every later boot reports a phantom unclean death. On timeout shutdown proceeds and the daemon thread is left to finish or die. Teardown is per served profile, under that profile's runtime scope — the mirror of startup discovery (``_discover_mcp_tools_for_profiles``) and of the periodic reconcile chore. A server's close path reads its own config/credentials at call time, so an unscoped shutdown on a bare thread resolved every served profile's teardown against the LAUNCH home. The trailing wildcard call (under the launch profile's own scope) stops the shared loop and reaps anything the per-profile passes did not own. ``timeout`` is a TOTAL budget: each pass gets ``timeout / (N + 1)``, because the default 15s per-pass wait inside ``shutdown_mcp_servers`` let N profiles consume the whole caller budget and the trailing wildcard pass — the only one that stops the shared loop — never ran. The worker runs in a FRESH context, not ``copy_context()``: the caller may sit inside a served profile's scope, and the trailing wildcard pass must run under the launch profile's own scope (``launch_profile_scope_if_multiplexed`` binds the launch home) — inheriting the caller's made it resolve the live home to that profile. See #82874. """ from tui_gateway.launch_profile_policy import launch_profile_scope_if_multiplexed profile_homes = ( _multiplex_profile_homes(config) if getattr(config, "multiplex_profiles", False) else []) pass_timeout = max(1.0, timeout / (len(profile_homes) + 1)) def _do() -> None: from tools.mcp_tool_common import _core from tools.mcp_tool_lifecycle import shutdown_mcp_servers for profile_name, profile_home in profile_homes: try: with _profile_runtime_scope(Path(profile_home), hydrate_secrets=False): shutdown_mcp_servers(scope=_core._mcp_registry_scope(), timeout=pass_timeout) except Exception: logger.debug("MCP shutdown raised for profile '%s'", profile_name, exc_info=True) try: with launch_profile_scope_if_multiplexed(): shutdown_mcp_servers(timeout=pass_timeout) except Exception: logger.debug("MCP shutdown raised", exc_info=True) thread = threading.Thread(target=Context().run, args=(_do,), name="mcp-shutdown", daemon=True) thread.start() done = await _await_thread_exit(thread, timeout=timeout) if not done: logger.warning( "MCP shutdown did not finish within %.1fs; continuing gateway " "teardown (background thread will be reaped at process exit)", timeout) return done def _shutdown_gateway_health_export(runner: Any) -> None: """Idempotently drain and detach Gateway Health OTLP export.""" runtime = getattr(runner, "_gateway_health_export_runtime", None) if runtime is None: return runner._gateway_health_export_runtime = None try: runtime.shutdown() except Exception: logger.debug("gateway health OTLP export shutdown failed", exc_info=True) def _gateway_stderr_formatter() -> logging.Formatter: """Return the redacting formatter used by the gateway stderr stream.""" from agent.redact import RedactingFormatter return RedactingFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s") # ownership guard inserted below (PR #93084) def _replace_target_belongs_to_other_profile(existing_pid: int) -> bool: """Return True when ``--replace`` must refuse to signal ``existing_pid``. A poisoned/stale PID record can point at another profile's LIVE gateway (cross-profile SIGTERM restart loop). Ownership is decided by the persisted identity record ALONE, bound to the live target by exact PID + start-time; live argv can never PROVE ownership (no HERMES_HOME), it is only a consistency check. Missing, legacy, conflicting or unprovable identity → refuse (fail closed).""" # Multiplex-only: the ONE host gateway serving this profile IS this profile's gateway, whatever # home launched it — `hermes -p X gateway run --replace` means "replace the process serving X". # Argv and HERMES_HOME can never prove that (the host singleton runs one home's argv while # serving every profile), so the live served set answers first; everything below stays the # fail-closed rule for a host with no usable record. try: from gateway.host_attach import host_gateway, profile_name_for_home owner = host_gateway() if owner is not None and owner.pid == existing_pid and owner.serves( profile_name_for_home(get_hermes_home())): logger.warning( "--replace target PID %s is the host gateway serving %d profile(s) (%s); " "replacing it restarts the host process for all of them.", existing_pid, len(owner.profiles), ", ".join(owner.profiles) or "unknown") return False except Exception: logger.debug("host served-set ownership probe failed for PID %s", existing_pid, exc_info=True) # On Windows there is no systemd/launchd service query at all (_get_service_pids() returns an empty # set), so a gateway supervised by a Scheduled Task / Startup VBS looks like an unsupervised orphan to # the process scan (#86098). The same holds on every platform for a healthy gateway launched standalone # (no service registration) whose PID the runtime record can see (#83683). Exempt the recorded healthy # gateway PID and its parent chain: a recorded, liveness-verified gateway is by definition not an orphan # "the pidfile/runtime record can't see", and the Scheduled-Task bootstrap's argv (``gateway run``) # matches the gateway scan — killing that bootstrap takes the detached gateway it spawned down with it. # Exclusion evidence comes from the RAW registration record, not the liveness-validated probe. # ``get_running_pid`` (any flags) returns None whenever a record fails validation — start-time mismatch # after PID-reuse checks, argv drift, lock hiccups — which is exactly when a healthy standalone gateway # (no service supervisor — e.g. `hermes gateway run` on Windows) is at risk: its PID never joins the # exclusion set and the sweep hard-kills it. On Windows SIGTERM is TerminateProcess, so the gateway's # planned-stop watcher never gets a chance to drain. Reading the raw pidfile + lock records (no # validation, no unlink side effects) is strictly safer for a KILL exclusion list: a stale recorded PID # at worst spares one process this sweep, while a validation false-negative would kill a live gateway. # The validated probe is still consulted for the runtime-status fallback PID it can surface when no # pidfile exists. try: from gateway.status import ( _get_pid_path, _get_process_hermes_home, _get_process_start_time, _pid_from_record, _read_pid_record, _record_looks_like_gateway, _read_process_cmdline, _same_hermes_home) our_home = _get_process_hermes_home() def refuse(msg: str, *args, level=logging.WARNING) -> bool: logger.log(level, "Refusing --replace: " + msg, *args) return True # Bound claim: the record must name THIS pid with THIS live start time, else it proves nothing. record = _read_pid_record(_get_pid_path()) if not isinstance(record, dict) or not _record_looks_like_gateway(record): return refuse("no valid gateway pid record to prove ownership of PID %s.", existing_pid) record_pid = _pid_from_record(record) if record_pid != existing_pid: return refuse("pid record names %s, not target %s.", record_pid, existing_pid) recorded_start = record.get("start_time") if not isinstance(recorded_start, int) or isinstance(recorded_start, bool): return True if _get_process_start_time(existing_pid) != recorded_start: return refuse("pid record start-time does not match the live process %s (stale/PID-reuse record).", existing_pid) recorded_home = record.get("hermes_home") if not isinstance(recorded_home, str) or not recorded_home.strip(): return refuse("pid record predates hermes_home stampings; ownership of PID %s unprovable.", existing_pid) if not _same_hermes_home(recorded_home, our_home): return refuse("pid record belongs to a different HERMES_HOME (%s, ours %s). Remove the stale PID " "record or stop the owning profile explicitly.", recorded_home, our_home, level=logging.ERROR) # Argv never proves ownership; an explicit contradicting --profile / HERMES_HOME= still refuses. live_cmdline = _best_effort(lambda: _read_process_cmdline(existing_pid)) if live_cmdline and _looks_like_profile_conflict_from_cmdline(live_cmdline, our_home): return refuse("target PID %s command line explicitly advertises a different profile than " "HERMES_HOME %s.", existing_pid, our_home, level=logging.ERROR) return False except Exception: # Destructive action + unknown ownership => fail closed. logger.warning("cross-profile --replace ownership probe failed for PID %s; refusing to signal", existing_pid, exc_info=True) return True def _looks_like_profile_conflict_from_cmdline(command: str, our_home) -> bool: """Token-exact contradiction check between a target argv and our home (authority is the pid record). Substring matching is not identity: ``--profile timothy`` must NOT read as profile ``tim``. Returns False whenever the argv does not clearly contradict our home.""" from gateway.status import _profile_name_for_home profile_name = _profile_name_for_home(our_home) try: tokens = shlex.split(command) except ValueError: tokens = command.split() def _flag_value(flag: str) -> Optional[str]: """Value of ``--flag X`` / ``--flag=X`` occurrences, token-exact.""" values = [] i = 0 while i < len(tokens): tok = tokens[i] if tok == flag and i + 1 < len(tokens): values.append(tokens[i + 1]) i += 2 continue if tok.startswith(flag + "="): values.append(tok[len(flag) + 1:]) i += 1 return values[-1] if values else None def _env_home_value() -> Optional[str]: """HERMES_HOME= env-style assignment on the argv, token-exact.""" prefix = "HERMES_HOME=" for tok in reversed(tokens): if tok.startswith(prefix): return tok[len(prefix):] return None def _norm(path: str) -> str: return os.path.normcase(os.path.normpath(path)) for flag in ("--profile", "-p"): value = _flag_value(flag) if value is None: continue # Named-profile home: a DIFFERENT explicit profile contradicts it (legacy default argv never carried # profile flags). Default/root home: ANY explicit named-profile flag contradicts it. if profile_name is None or profile_name == "default" or value != profile_name: return True home_value = _flag_value("--hermes-home") or _env_home_value() return bool(home_value is not None and _norm(home_value) != _norm(str(our_home))) def _clear_takeover_marker_quiet() -> None: """Best-effort: the marker is scoped to one target; a stale one would grief an unrelated shutdown.""" try: from gateway.status import clear_takeover_marker clear_takeover_marker() except Exception: pass async def _wait_for_pid_exit(pid: int, attempts: int, delay: float) -> bool: """Poll for process exit without blocking the loop (a blocking sleep freezes signal handlers and health checks). ``os.kill(pid, 0)`` on Windows is NOT a no-op — use the handle-based check.""" from gateway.status import _pid_exists for _ in range(attempts): if not _pid_exists(pid): return True await asyncio.sleep(delay) return False def _start_gateway_make_restart_signal_handler(runner): """Build the SIGUSR1 handler: log what the signal means, then the drain-aware service restart.""" def restart_signal_handler(): # systemd's `reload` verb (ExecReload=kill -USR1) lands here too; say so, because operators # expect `reload` to mean an in-process config reload, not a drain-and-relaunch (#117267). logger.info( "SIGUSR1 received (systemctl reload / hermes gateway restart): performing a graceful " "gateway restart — drain active turns, exit, supervisor relaunches. Not an in-process " "config reload.") runner.request_restart(detached=False, via_service=True) return restart_signal_handler def _claim_host_gateway_role(force: bool = False) -> None: """Take the HOST-wide gateway lock, publish the record — or REFUSE to be the second gateway. The lock is no longer observe-only. ``gateway.host_attach`` already answers "is there a host gateway and does it serve me?" before anything binds, but it reads a RECORD, and a record is published a moment after the owner starts: two gateways launched together (a supervisor restarting two units, an update relaunch racing a manual start) can both see no owner and both proceed. The lock is the only atomic arbiter of that race, so losing it means we are the second gateway on this host — the shape multiplex-only forbids. The refusal is deliberately EX_TEMPFAIL (75), never the parking 78: losing a lock race is a runtime observation, not a config verdict. Every supervisor we generate retries 75, and on the retry the owner's record exists, so the attach path resolves the profile properly (attach, rescan-then-attach, or a named refusal) instead of the unit being parked forever. ``--force`` skips the question exactly as it does in the attach path. """ from gateway import host_rendezvous as hr # The host lock can be free after a standalone owner exits while a coexisting # multiplexer remains live. Its per-home channel still governs our opt-out. if not force: from gateway.host_attach import REFUSE, standalone_attach_decision decision = standalone_attach_decision(get_hermes_home(), None) if decision is not None and decision.outcome == REFUSE: from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE print(decision.message) raise SystemExit(GATEWAY_SERVICE_RESTART_EXIT_CODE) try: outcome, error = hr.claim_host_lock(hr.ROLE_GATEWAY) if outcome is hr.HostLockOutcome.ACQUIRED: # PROVISIONAL: an owner exists, its served set is not decided yet (multiplex is # settled by the runner, and the attach channel is not bound for another moment). # Publishing a guessed set here parked a second profile's supervised unit against # profiles this process may never serve; _refresh_host_gateway_record() fills it in # once the control socket answers. hr.publish_record(hr.ROLE_GATEWAY, profiles=(), home=str(get_hermes_home())) # SIGTERM (systemd stop, docker stop, the update relaunch) does not run atexit. hr.cleanup_on_exit(hr.ROLE_GATEWAY) return if outcome is hr.HostLockOutcome.COULD_NOT_OPEN: # NOT contention: a read-only/undeletable lock dir. Refusing here would take a working # single-gateway host down over an unusable directory. logger.warning( "Host gateway lock could not be opened (%s); this gateway is not discoverable. " "No second gateway is implied — the lock directory itself is unusable.", error) return except Exception: logger.debug("host gateway rendezvous failed", exc_info=True) return # Lost the lock. Reading the owner's record may fail (it is published a moment after the # claim); that changes WHO we can name, never the verdict -- we are the second gateway. owner = None try: owner = hr.read_record(hr.ROLE_GATEWAY, include_stale=True) except Exception: logger.debug("host gateway record unreadable", exc_info=True) if force: logger.warning("--force: starting a second gateway although %s owns this host.", hr.describe(owner) if owner else "another process") return from gateway.host_attach import ( ATTACH_CHANNEL_WAIT_S, START, host_gateway, launched_by_other_tenant, standalone_attach_decision, ) from hermes_cli.profiles import profile_is_standalone if owner is not None and launched_by_other_tenant(owner.home, get_hermes_home()): # The lock is per OS user, so a second tenant root can never win it against the first: # refusing 75 here would retry forever and its gateway would never start (#121352). logger.warning( "Another Hermes home's gateway owns this host (%s); starting this home's gateway beside it.", hr.describe(owner)) return if profile_is_standalone(get_hermes_home()): # Recheck after losing the atomic lock: the pre-lock served set may be stale. live_owner = host_gateway(wait_for_channel=ATTACH_CHANNEL_WAIT_S) # The rendezvous record proves lock ownership, but the owner's control channel may # still be unavailable. Standalone discovery also checks each profile's liveness # channel, so it can prove that this profile is unserved even when the host probe # cannot construct a HostGateway yet. decision = standalone_attach_decision(get_hermes_home(), live_owner) if decision is not None: if decision.outcome == START: return from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE print(decision.message) raise SystemExit(GATEWAY_SERVICE_RESTART_EXIT_CODE) if _owner_is_standalone(): # COMPOSITION with #118236: `host_attach.decide` sent us here with START precisely because # the owner is another profile's STANDALONE gateway and will never serve us. Refusing now # exits 75, the supervisor retries in 5s, and the next claim loses the same race — the host # lock is per OS user and every gateway takes it, so a second profile can NEVER win. An # unmigrated fleet would spin forever instead of running. Start beside it and point at the # one command that converges; multiplex-only is enforced against a MULTIPLEXER owner. logger.warning( "Another profile's standalone gateway owns this host (%s); starting beside it rather " "than retrying a race no second profile can win. Fold every profile onto one gateway " "with: %s", hr.describe(owner) if owner else "owner unknown", _migrate_command()) return _refuse_second_host_gateway(owner) def _migrate_command() -> str: from hermes_cli.gateway_migrate import MIGRATE_COMMAND return MIGRATE_COMMAND def _owner_is_standalone() -> bool: """True when the host owner answers that it does NOT multiplex (an unmigrated fleet). Asked only on the lock-losing path, and any failure answers False: an owner we cannot reach is treated as a multiplexer, which keeps the second-gateway refusal as the default. """ try: from gateway.host_attach import host_gateway, profile_name_for_home, request_serve_profile owner = host_gateway() if owner is None or owner.pid == os.getpid(): return False answered = request_serve_profile(profile_name_for_home(get_hermes_home()), owner=owner) return bool(answered is not None and answered.standalone) except Exception: logger.debug("standalone-owner probe failed; keeping the second-gateway refusal", exc_info=True) return False def _refuse_second_host_gateway(owner) -> None: """Print the named refusal and exit 75 so a supervisor retries instead of parking the unit. Reached only after losing the host lock, so ``--replace`` is not offered: an owner that serves this profile was already handled before the claim, and ``--replace`` does not skip the lock. """ from gateway import host_rendezvous as hr from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE who = hr.describe(owner) if owner else "owner unknown (its record is gone)" message = ( f"❌ Another gateway already owns this host: {who}\n" f" Exactly one gateway per host serves every profile, so this process will not start a\n" f" second one (it would double-bind this profile's platforms).\n" f" Fold every profile onto the owner: {_migrate_command()}\n" f" Or stop the other gateway first, then start this one.\n" f" Or start one anyway (skips the host-lock check): hermes gateway run --force\n" f" (--replace does not skip this check; it only replaces an owner that serves this profile.)") logger.error("Refusing to start a second gateway on this host: %s", who) print(message) raise SystemExit(GATEWAY_SERVICE_RESTART_EXIT_CODE) def _log_standalone_profiles_at_boot(runner) -> None: """One INFO line per standalone profile when the MULTIPLEXER takes its served set at boot. The host gateway silently omits an opted-out profile from its served set; without this line an operator reading the boot log cannot tell "not created yet" from "excluded by config". """ try: if not getattr(runner.config, "multiplex_profiles", False): return from hermes_cli.profiles import profiles_to_serve, profile_is_standalone from hermes_cli.gateway_multiplex_mode import STANDALONE_DEPRECATION_NOTICE served = set(runner.served_profile_names()) for name, home in profiles_to_serve(True, include_standalone=True, include_parked=True): if name != "default" and name not in served and profile_is_standalone(home): logger.warning("profile '%s' is standalone (gateway.standalone: true); not served by " "this gateway. %s", name, STANDALONE_DEPRECATION_NOTICE) except Exception: logger.warning("standalone-profile boot notice failed", exc_info=True) def _refresh_host_gateway_record(runner) -> None: """Republish the host record with the SETTLED served set, now that the channel answers. The claim-time publish is deliberately empty: only the runner knows whether multiplex ended up on and which profiles it took. A standalone gateway serves exactly its own profile — not the whole roster ``served_profiles()`` would have guessed for it. """ from gateway import host_rendezvous as hr from gateway.host_attach import profile_name_for_home try: if not hr.owns_host_lock(hr.ROLE_GATEWAY): return home = get_hermes_home() if getattr(runner.config, "multiplex_profiles", False): served = tuple(runner.served_profile_names()) else: served = (profile_name_for_home(home),) hr.publish_record(hr.ROLE_GATEWAY, profiles=served, home=str(home)) except Exception: logger.debug("host gateway record refresh failed", exc_info=True) # Bootstrap phases bound on the facade: ``_host_attach_or_none`` reads the replace step through this # module, and the lifecycle tests patch/drive these phases as ``gateway.run`` attributes. from gateway.run_bootstrap import ( _start_gateway_make_shutdown_signal_handler, _start_gateway_replace_existing_instance, _start_gateway_shutdown_tail, _start_gateway_start_cron_and_housekeeping, ) async def _host_attach_or_none(replace: bool, force: bool = False) -> Optional[bool]: """Attach / rescan / refuse against the ONE host gateway; ``None`` = start normally. Returns ``True`` when this invocation is satisfied by the running host process (exit 0, nothing spawned) and ``False`` when it must refuse. The per-home duplicate guard below cannot answer this at all: another profile's gateway lives in another home, so it sees no PID and starts a second process — the shape multiplex-only forbids. ``force`` is the operator's escape hatch when the owner is wedged or lying: skip the whole question and start. Ignoring it here made ``--force`` print the starting banner and then attach anyway, leaving no supported way to start a gateway at all. """ if force: logger.warning("--force: starting a gateway without asking the host owner.") return None from gateway.host_attach import ATTACH, REFUSE, REPLACE_HOST, decide decision = decide(get_hermes_home(), replace=replace) if decision.outcome == ATTACH: logger.info("Attaching to the host gateway instead of starting a second one: %s", decision.owner.describe() if decision.owner else "unknown") print(decision.message) return True if decision.outcome == REFUSE: logger.error("Refusing to start a second gateway on this host: %s", decision.owner.describe() if decision.owner else "unknown") print(decision.message) return False if decision.outcome == REPLACE_HOST and decision.owner is not None: # decide() only targets an owner that serves this profile or has not published its served set. if not await _start_gateway_replace_existing_instance(decision.owner.pid, True): return False return None async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = False, verbosity: Optional[int] = 0, force: bool = False, idle_exit: bool = False) -> bool: """Run the process lifecycle; the runner facade remains the public entrypoint.""" from gateway.run_bootstrap import start_gateway as bootstrap return await bootstrap(config, replace, verbosity, force, idle_exit) def _guard_corrupt_user_config() -> None: """Fail closed when the active profile's config.yaml cannot be parsed: nobody can repair it on this surface, and defaults would let provider auto-detection adopt ``.env`` credentials the config never named. Same policy and escape hatch (``HERMES_IGNORE_USER_CONFIG=1``) as ``hermes_cli/main.py``.""" from hermes_cli.config import InvalidUserConfigError, require_parseable_user_config try: require_parseable_user_config() except InvalidUserConfigError as exc: print(f"Error: {exc}", file=sys.stderr) raise SystemExit(2) from exc def main(): """CLI entry point for the gateway.""" # Before any config-dependent startup (watchdog, DB opens, provider resolution). _guard_corrupt_user_config() # Advertise the harness to children (mirrors _advertise_agent_env in hermes_cli/main.py, inlined to # avoid its startup side effects). Value must equal registry id ``hermes-agent`` exactly. os.environ.setdefault("AI_AGENT", "hermes-agent") os.environ.setdefault("HERMES_AGENT", "true") def _register_identity() -> None: # Ledger registration + Windows job-object attach so update-time reapers can identify this gateway. from hermes_cli.process_identity import attach_self_to_kill_on_close_job, register_self register_self("gateway") attach_self_to_kill_on_close_job() def _arm_watchdog() -> None: # Armed before config load / DB opens so a pre-loop deadlock is respawned by the supervisor instead # of wedging as a live-PID zombie. GatewayRunner disarms it. from hermes_startup_watchdog import arm_startup_watchdog arm_startup_watchdog() def _utf8_stdio() -> None: # Windows: gateway logs and banner would UnicodeEncodeError on cp1252 consoles. No-op on POSIX. from hermes_cli.stdio import configure_windows_stdio configure_windows_stdio() for _step in (_register_identity, _arm_watchdog, _utf8_stdio): _best_effort(_step) # pm startup contract (PATH provisioning for the store's tools), then # the post-update bootstrap: the same one-pass record-gated maintenance # registry the CLI dispatch path runs (hermes_cli/main.py) — this # entrypoint bypasses that dispatch, so run it here too. Never raises. try: from hermes_cli.venv_sync import check_runtime from pm.paths import install_root problem = check_runtime(install_root()) if problem: logger.warning(problem) except Exception: logger.debug("pm startup check failed", exc_info=True) try: from hermes_cli.boot_bootstrap import maybe_run_boot_bootstrap from pm.paths import install_root maybe_run_boot_bootstrap(install_root()) except Exception: logger.debug("boot bootstrap failed", exc_info=True) import argparse parser = argparse.ArgumentParser(description="Hermes Gateway - Multi-platform messaging") parser.add_argument("--config", "-c", help="Path to gateway config file") parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") args = parser.parse_args() config = None if args.config: import hermes_yaml as yaml with open(args.config, encoding="utf-8-sig") as f: config = GatewayConfig.from_dict(yaml.safe_load(f) or {}) # Same boot-time verdict the loaded config gets when the file leaves the flag unset. from hermes_cli.gateway_multiplex_mode import log_multiplex_decision, resolve_multiplex_mode log_multiplex_decision(resolve_multiplex_mode(config)) # start_gateway() completes teardown before returning/raising SystemExit; force-exit after so a # wedged non-daemon worker can't block Py_FinalizeEx's join. SystemExit caught so EVERY path exits. try: # start_gateway() performs the full graceful teardown (adapters disconnected, sessions saved + # flushed, SQLite closed, cron/MCP stopped, PID file + runtime lock released) before it returns OR # raises SystemExit with an explicit code. Force-exit afterwards so a wedged non-daemon worker # thread (e.g. a ThreadPoolExecutor tool/LLM call blocked with no timeout) cannot block interpreter # finalization (Py_FinalizeEx joins all non-daemon threads, incl. concurrent.futures' _python_exit) # and strand the gateway half-shut down with the supervisor unable to restart it (#53107). # SystemExit is caught explicitly: start_gateway raises it on the clean-fatal-config (#51228), # planned-restart, and service-restart paths, all of which complete teardown first. Routing those # codes through the same os._exit backstop means EVERY exit path is wedge-proof, not just the # boolean-return ones. success = asyncio.run(start_gateway(config)) exit_code = 0 if success else 1 except SystemExit as e: # e.code may be None (→ 0), an int, or a str (→ 1, like CPython). exit_code = 0 if e.code is None else e.code if isinstance(e.code, int) else 1 _exit_after_graceful_shutdown(exit_code) def _exit_after_graceful_shutdown(exit_code: int) -> None: """Flush stdio, release the PID file + runtime lock, then hard-exit. ``os._exit`` (not ``sys.exit``): SystemExit runs ``Py_FinalizeEx``, which joins every non-daemon thread — exactly the hang a wedged worker causes. It bypasses ``atexit``, so PID/lock release and the bounded log drain (file handlers sit behind a ``QueueListener`` thread) are done here explicitly. Graceful teardown is already complete by the time this runs, so there is nothing left that needs a clean interpreter shutdown. See #53107. ``os._exit`` bypasses ``atexit`` handlers, so we cannot rely on the ``atexit``-registered ``remove_pid_file`` / ``release_gateway_runtime_lock`` (registered in ``start_gateway``) to run. The full-shutdown path releases both explicitly in ``_stop_impl``, but the EARLY exit paths — clean-fatal-config (#51228) and startup-aborted-before-running — raise ``SystemExit`` right after ``runner.start()`` without going through ``_stop_impl``, so on those paths ``atexit`` was the only thing releasing them. Now that those paths are routed through this backstop (#53107), release both here explicitly. Both calls are idempotent — ``remove_pid_file`` only unlinks a PID file that belongs to this process, and ``release_gateway_runtime_lock`` no-ops when the lock is already released — so this is a no-op on the normal shutdown path and the actual cleanup on the early-exit paths. """ for stream in (sys.stdout, sys.stderr): with suppress(Exception): stream.flush() def _release_locks() -> None: # BEFORE the log drain (bounded, but could take its full timeout on a wedged disk); idempotent. from gateway.status import remove_pid_file, release_gateway_runtime_lock remove_pid_file() release_gateway_runtime_lock() def _mark_exited() -> None: # Single funnel every graceful exit passes through, so the next boot's unclean-death detector # fires only for genuine SIGKILL/OOM/VM deaths. Ownership-guarded against an old --replace life. from hermes_cli.observability.shared_metrics_process import stamp_exit stamp_exit("clean") # the exit-metrics marker too: os._exit skips its atexit stamp (never raises) from gateway.lifecycle_ledger import mark_exited mark_exited(exit_code, reason="graceful_shutdown") def _drain_logs() -> None: # os._exit bypasses the listener's atexit drain. Bounded, no restart — NOT flush_log_queue(): # a listener wedged on the rotation lock would re-freeze shutdown in an unbounded stop() join. from hermes_logging import drain_log_queue drain_log_queue(timeout=1.0) for _step in (_release_locks, _mark_exited, _drain_logs): _best_effort(_step) os._exit(exit_code) if __name__ == "__main__": main()