From a1f134730fdec8fd715c98b3dc033fc8d5eec926 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:21:37 -0700 Subject: [PATCH 01/59] refactor(gateway): dedupe run.py regex tables, runtime-kwargs builders, media-kind predicates --- gateway/run.py | 289 ++++++++++++++++--------------------------------- 1 file changed, 94 insertions(+), 195 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index a478de1177..ad4fa654a1 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -47,9 +47,7 @@ from agent.conversation_compression import ( ) from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX from agent.interrupt_compat import request_hard_interrupt -from agent.turn_context import ( - compression_made_progress, -) +from agent.turn_context import compression_made_progress from hermes_cli.config import _is_ssh_remote_tilde_cwd, cfg_get from hermes_cli.fallback_config import get_fallback_chain @@ -126,6 +124,12 @@ _HYGIENE_COOLDOWN_MAX_SECONDS = 3600.0 _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, @@ -143,9 +147,7 @@ def _hygiene_cooldown_for_failure( state = gateway._session_state(session_key).persistent except Exception as exc: logger.debug("hygiene failure streak update failed: %s", exc) - session_db = getattr(gateway, "_session_db", None) - session_db = getattr(session_db, "_db", session_db) - increment = getattr(session_db, "increment_hygiene_failure_streak", None) + increment = getattr(_gateway_session_db_inner(gateway), "increment_hygiene_failure_streak", None) if callable(increment): try: streak = max(1, int(increment(session_key))) @@ -176,9 +178,7 @@ def _reset_hygiene_failure_streak(gateway, session_key: str) -> None: state.persistent.hygiene_failure_streak = 0 except Exception as exc: logger.debug("hygiene failure streak reset failed: %s", exc) - session_db = getattr(gateway, "_session_db", None) - session_db = getattr(session_db, "_db", session_db) - reset = getattr(session_db, "reset_hygiene_failure_streak", None) + reset = getattr(_gateway_session_db_inner(gateway), "reset_hygiene_failure_streak", None) if callable(reset): try: reset(session_key) @@ -376,16 +376,11 @@ def _record_hygiene_cooldown( Shares the in-conversation path's column/recorder so it survives restarts. ``error`` must be forwarded: the recorder writes ``compression_failure_error`` UNCONDITIONALLY (else NULL clobber). """ - import time as _time - session_db = getattr(gateway, "_session_db", None) - if session_db is None: - return - session_db = getattr(session_db, "_db", session_db) - recorder = getattr(session_db, "record_compression_failure_cooldown", None) + 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) + recorder(session_id, time.time() + cooldown_seconds, error) except Exception as exc: logger.debug("session hygiene cooldown persist failed: %s", exc) @@ -452,20 +447,6 @@ def _gateway_surface_passes_raw_text(platform: Any) -> bool: return _gateway_platform_value(platform) in _GATEWAY_RAW_TEXT_PLATFORMS -_GATEWAY_PROVIDER_ERROR_RE = re.compile( - r"(" # infrastructure/provider error preambles, not ordinary assistant prose - 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"|\bhttp\s*\d{3}\b" - r"|incorrect\s+api\s+key" - r"|invalid\s+api\s+key" - r")", - re.IGNORECASE, -) - _GATEWAY_PROVIDER_POLICY_RE = re.compile( r"(" # raw provider policy/safety bodies are noisy and may be sensitive r"cybersecurity\s+risk" @@ -491,24 +472,14 @@ _GATEWAY_RATE_LIMIT_RE = re.compile( re.IGNORECASE, ) -_GATEWAY_CONNECTION_ERROR_RE = re.compile( - r"(" - r"(?:\w+\.)?(?:api\s*)?connection\s*(?:error|timeout)" - r"|(?:\w+\.)?connect\s*(?:error|timeout)" - r"|connection\s+refused" - r"|connection\s+reset" - r"|connection\s+aborted" - r"|actively\s+refused" - r"|winerror\s+10061" - r"|errno\s+111" - r"|no\s+route\s+to\s+host" - r"|network\s+is\s+unreachable" - r"|cannot\s+connect" - r"|failed\s+to\s+establish" - r"|could\s+not\s+connect" - r")", - re.IGNORECASE, +# Connection-failure markers: the first 8 also anchor the provider-failure envelope shape below. +_CONNECTION_ERROR_MARKERS = ( + r"(?:\w+\.)?(?:api\s*)?connection\s*(?:error|timeout)", r"(?:\w+\.)?connect\s*(?:error|timeout)", + r"connection\s+refused", r"connection\s+reset", r"connection\s+aborted", r"actively\s+refused", + r"winerror\s+10061", r"errno\s+111", r"no\s+route\s+to\s+host", r"network\s+is\s+unreachable", + r"cannot\s+connect", r"failed\s+to\s+establish", r"could\s+not\s+connect", ) +_GATEWAY_CONNECTION_ERROR_RE = re.compile("(" + "|".join(_CONNECTION_ERROR_MARKERS) + ")", re.IGNORECASE) _GATEWAY_SECRET_PATTERNS = ( re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_\-]{12,}\b"), @@ -623,6 +594,13 @@ def _seed_hygiene_system_prompt( 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). @@ -631,29 +609,13 @@ def _is_transient_network_error(exc: BaseException) -> bool: seen: set[int] = set() cur: Optional[BaseException] = exc depth = 0 - transient_class_names = { - "TimedOut", - "NetworkError", - "ReadError", - "WriteError", - "ConnectError", - "ConnectTimeout", - "ReadTimeout", - "WriteTimeout", - "PoolTimeout", - "RemoteProtocolError", - "ServerDisconnectedError", - "ClientConnectorError", - "ClientOSError", - } while cur is not None and depth < 12: ident = id(cur) if ident in seen: break seen.add(ident) depth += 1 - name = type(cur).__name__ - if name in transient_class_names: + if type(cur).__name__ in _TRANSIENT_NETWORK_ERROR_CLASS_NAMES: return True cur = cur.__cause__ or cur.__context__ return False @@ -747,51 +709,39 @@ def _format_exec_approval_fallback( + ", ".join(choices[:-1]) + f", or {choices[-1]}." ) +# Ordered: auth beats policy beats rate-limit beats connection; first match wins. +_PROVIDER_ERROR_REPLIES = ( + (_GATEWAY_AUTH_ERROR_RE, "⚠️ Provider authentication failed. Check the configured credentials; " + "raw provider details are in the gateway logs."), + (_GATEWAY_PROVIDER_POLICY_RE, "⚠️ The model provider rejected the request. I kept the raw provider " + "error out of chat; check gateway logs for details or try rephrasing."), + (_GATEWAY_RATE_LIMIT_RE, "⏱️ The model provider is rate-limiting requests. Please wait a moment and try again."), + (_GATEWAY_CONNECTION_ERROR_RE, "⚠️ The model server is not responding — it looks like the configured " + "model endpoint is not running or is unreachable."), +) + + def _gateway_provider_error_reply(text: str) -> str: """Map raw provider/API errors to a short user-safe Telegram reply.""" - if _GATEWAY_AUTH_ERROR_RE.search(text): - return ( - "⚠️ Provider authentication failed. Check the configured credentials; " - "raw provider details are in the gateway logs." - ) - if _GATEWAY_PROVIDER_POLICY_RE.search(text): - return ( - "⚠️ The model provider rejected the request. I kept the raw provider " - "error out of chat; check gateway logs for details or try rephrasing." - ) - if _GATEWAY_RATE_LIMIT_RE.search(text): - return "⏱️ The model provider is rate-limiting requests. Please wait a moment and try again." - if _GATEWAY_CONNECTION_ERROR_RE.search(text): - return ( - "⚠️ The model server is not responding — it looks like the configured " - "model endpoint is not running or is unreachable." - ) + for pattern, reply in _PROVIDER_ERROR_REPLIES: + if pattern.search(text): + return reply return ( "⚠️ The model provider failed after retries. I kept raw provider details " "out of chat; check gateway logs for diagnostics." ) +# 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*)?(" - 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" - r"|(?:\w+\.)?(?:api\s*)?connection\s*(?:error|timeout)" - r"|(?:\w+\.)?connect\s*(?:error|timeout)" - r"|connection\s+refused" - r"|connection\s+reset" - r"|connection\s+aborted" - r"|actively\s+refused" - r"|winerror\s+10061" - r"|errno\s+111" - r"|all\s+connection\s+attempts\s+failed" - r")", + + "|".join(_PROVIDER_ERROR_MARKERS + _CONNECTION_ERROR_MARKERS[:8] + (r"all\s+connection\s+attempts\s+failed",)) + + ")", re.IGNORECASE, ) @@ -1846,7 +1796,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent)) # Resolve Hermes home directory (respects HERMES_HOME override) from hermes_constants import get_hermes_home, get_hermes_home_override -from utils import atomic_json_write, base_url_hostname, is_truthy_value # noqa: F401 (re-exported: run_* mixins + tests resolve gateway.run.) +from utils import atomic_json_write # noqa: F401 (re-exported: run_* mixins + tests resolve gateway.run.) _hermes_home = get_hermes_home() # Load environment variables from ~/.hermes/.env first. @@ -1934,11 +1884,7 @@ def _current_max_iterations() -> int: return _resolve_turn_limit(os.getenv("HERMES_MAX_ITERATIONS")) -from contextlib import ( - asynccontextmanager as _asynccontextmanager, - contextmanager as _contextmanager, -suppress, -) +from contextlib import asynccontextmanager as _asynccontextmanager, contextmanager as _contextmanager, suppress # Platforms that bind a host TCP port. In a profile multiplexer the default profile owns the single @@ -2566,11 +2512,6 @@ from gateway.restart import ( ) -from gateway.whatsapp_identity import ( - canonical_whatsapp_identifier as _canonical_whatsapp_identifier, # noqa: F401 - ) - - logger = logging.getLogger(__name__) @@ -2714,6 +2655,15 @@ def _resolve_runtime_agent_kwargs() -> dict: else {} ) + return {**_runtime_agent_kwargs(runtime), "max_tokens": max_tokens, "capabilities": capabilities} + + +def _runtime_agent_kwargs(runtime: dict) -> dict: + """AIAgent constructor kwargs shared by every runtime-provider resolution. + + ``request_overrides`` is passed through as resolved (custom_providers ``extra_body`` etc.) so + the provider's configured request body reaches the per-turn route on the gateway path. + """ return { "api_key": runtime.get("api_key"), "base_url": runtime.get("base_url"), @@ -2723,13 +2673,7 @@ def _resolve_runtime_agent_kwargs() -> dict: "command": runtime.get("command"), "args": list(runtime.get("args") or []), "credential_pool": runtime.get("credential_pool"), - "request_overrides": dict(runtime.get("request_overrides") or {}), - "max_tokens": max_tokens, - # Per-provider request_overrides (e.g. custom_providers ``extra_body`` with - # ``chat_template_kwargs``) from resolve_runtime_provider() must reach the per-turn route, - # else the provider's configured request body never reaches the model on the gateway path. "request_overrides": runtime.get("request_overrides"), - "capabilities": capabilities, } @@ -2858,14 +2802,7 @@ def _resolve_runtime_agent_kwargs_for_provider(provider: str) -> dict: except Exception as exc: raise RuntimeError(format_runtime_provider_error(exc)) from exc 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"), + **_runtime_agent_kwargs(runtime), "request_overrides": dict(runtime.get("request_overrides") or {}), "capabilities": dict(runtime.get("capabilities") or {}), "max_tokens": runtime.get("max_output_tokens"), @@ -2928,19 +2865,7 @@ def _try_resolve_fallback_provider() -> dict | None: entry.get("provider") or runtime.get("provider"), entry.get("model"), ) - 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": dict(runtime.get("request_overrides") or {}), - "model": entry.get("model"), - "request_overrides": runtime.get("request_overrides"), - } + return {**_runtime_agent_kwargs(runtime), "model": entry.get("model")} except Exception as fb_exc: logger.debug("Fallback entry %s failed: %s", entry.get("provider"), fb_exc) continue @@ -2955,24 +2880,23 @@ def _event_media_type_at(event, index: int) -> str: return media_types[index] if index < len(media_types) else "" -def _event_media_is_image(event, index: int) -> bool: - """True if the attachment at *index* is an image. +def _event_media_kind_is(event, index: int, mime_prefix: str, fallback_types: frozenset) -> bool: + """Trust the per-attachment MIME; fall back to the message-level type only when it is unknown. - Trust the per-attachment MIME; fall back to message-level ``PHOTO`` only when unknown, else a - document uploaded alongside an image is base64'd as vision and the provider 400s. + Otherwise 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("image/") - return getattr(event, "message_type", None) == MessageType.PHOTO + 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: - """True if the attachment at *index* is audio (per-attachment MIME first).""" - mtype = _event_media_type_at(event, index) - if mtype: - return mtype.startswith("audio/") - return getattr(event, "message_type", None) in {MessageType.VOICE, MessageType.AUDIO} + return _event_media_kind_is(event, index, "audio/", frozenset({MessageType.VOICE, MessageType.AUDIO})) def _event_media_is_stt_input(event, index: int) -> bool: @@ -2987,11 +2911,7 @@ def _event_media_is_stt_input(event, index: int) -> bool: def _event_media_is_video(event, index: int) -> bool: - """True if the attachment at *index* is video (per-attachment MIME first).""" - mtype = _event_media_type_at(event, index) - if mtype: - return mtype.startswith("video/") - return getattr(event, "message_type", None) == MessageType.VIDEO + return _event_media_kind_is(event, index, "video/", frozenset({MessageType.VIDEO})) def _build_media_placeholder(event) -> str: @@ -3713,12 +3633,9 @@ def _format_gateway_process_notification(evt: dict) -> "str | None": _sid = evt.get("session_id", "unknown") _cmd = evt.get("command", "unknown") - if evt_type == "watch_disabled": - return f"[IMPORTANT: {evt.get('message', '')}]" - - # Overflow events carry their human-readable summary in `message`, like watch_disabled + # watch_disabled and overflow events carry their human-readable summary in `message` # (shared formatter in tools/process_registry.py). - if evt_type in ("watch_overflow_tripped", "watch_overflow_released"): + if evt_type in ("watch_disabled", "watch_overflow_tripped", "watch_overflow_released"): return f"[IMPORTANT: {evt.get('message', '')}]" if evt_type == "watch_match": @@ -3856,14 +3773,9 @@ def _normalize_empty_agent_response( "This may be a transient error — try sending your message again." ) - # api_calls == 0, not failed, not interrupted: the agent never ran (post-/stop generation race). - # Without this the gateway silently drops the turn and the user sees no reply. - if ( - api_calls == 0 - and not agent_result.get("interrupted") - and not agent_result.get("failed") - and not agent_result.get("partial") - ): + # api_calls == 0, not failed, not interrupted (all returned above): the agent never ran + # (post-/stop generation race). Without this the turn is silently dropped with no reply. + if api_calls == 0 and not agent_result.get("partial"): return ( "⚠️ Your message wasn't processed (the previous turn was still " "being cleaned up). Please send it again." @@ -4740,21 +4652,10 @@ class GatewayRunner( _VOICE_MODE_PATH = _hermes_home / "gateway_voice_mode.json" - @property - def should_exit_cleanly(self) -> bool: - return self._exit_cleanly - - @property - def should_exit_with_failure(self) -> bool: - return self._exit_with_failure - - @property - def exit_reason(self) -> Optional[str]: - return self._exit_reason - - @property - def exit_code(self) -> Optional[int]: - return self._exit_code + 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.""" @@ -6026,22 +5927,20 @@ def _looks_like_profile_conflict_from_cmdline(command: str, our_home) -> bool: return tok[len(prefix):] return None - if profile_name is not None and profile_name != "default": - # Our home is a named profile: any explicit DIFFERENT named profile on the argv contradicts it; - # bare argv stays consistent (legacy default-gateway argv never carried profile flags). - for flag in ("--profile", "-p"): - value = _flag_value(flag) - if value is not None and value != profile_name: - return True - home_value = _flag_value("--hermes-home") or _env_home_value() - return bool(home_value is not None and os.path.normcase(os.path.normpath(home_value)) != os.path.normcase(os.path.normpath(str(our_home)))) + def _norm(path: str) -> str: + return os.path.normcase(os.path.normpath(path)) - # Our home is the default/root: ANY explicit named-profile flag on the - # argv contradicts it. - if _flag_value("--profile") is not None or _flag_value("-p") is not None: - return True + for flag in ("--profile", "-p"): + value = _flag_value(flag) + if value is None: + continue + # Named-profile home: a DIFFERENT explicit profile contradicts it (bare argv stays consistent; + # legacy default-gateway 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 os.path.normcase(os.path.normpath(home_value)) != os.path.normcase(os.path.normpath(str(our_home)))) + return bool(home_value is not None and _norm(home_value) != _norm(str(our_home))) async def _start_gateway_replace_existing_instance(existing_pid: int, replace: bool) -> bool: From 4e665e4e53203ab218d21e83209101859d4a4518 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:22:31 -0700 Subject: [PATCH 02/59] refactor(gateway): fold Slack/SMS env steps into _Cred rows, compact config dataclasses --- gateway/config.py | 732 +++++++++++++----------------------------- gateway/config_env.py | 222 ++++++------- 2 files changed, 316 insertions(+), 638 deletions(-) diff --git a/gateway/config.py b/gateway/config.py index 415450d1df..9923ccba4f 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -1,11 +1,5 @@ -""" -Gateway configuration management. - -Handles loading and validating configuration for: -- Connected platforms (Telegram, Discord, WhatsApp, Weixin, and more) -- Home channels for each platform -- Session reset policies -- Delivery preferences +"""Gateway configuration: connected platforms, home channels, session reset +policies and delivery preferences, loaded from config.yaml / gateway.json / env. """ import logging @@ -27,6 +21,9 @@ from utils import is_truthy_value logger = logging.getLogger(__name__) +_TRUTHY_STRINGS = frozenset({"1", "true", "yes", "on"}) +_FALSY_STRINGS = frozenset({"0", "false", "no", "off"}) + def _coerce_bool(value: Any, default: bool = True) -> bool: """Coerce bool-ish config values, preserving a caller-provided default.""" @@ -34,9 +31,9 @@ def _coerce_bool(value: Any, default: bool = True) -> bool: return default if isinstance(value, str): lowered = value.strip().lower() - if lowered in {"true", "1", "yes", "on"}: + if lowered in _TRUTHY_STRINGS: return True - if lowered in {"false", "0", "no", "off"}: + if lowered in _FALSY_STRINGS: return False return default return is_truthy_value(value, default=default) @@ -80,50 +77,38 @@ def _normalize_multiplex_profile_allowlist(value: Any) -> Optional[List[str]]: return normalized -# Recognized truthy / falsy tokens for the GATEWAY_MULTIPLEX_PROFILES operator -# override. Anything not in either set — and a blank/whitespace value — is -# treated as "unset" so it falls through to config.yaml rather than silently -# forcing the flag off. -_MULTIPLEX_TRUTHY_STRINGS = frozenset({"1", "true", "yes", "on"}) -_MULTIPLEX_FALSY_STRINGS = frozenset({"0", "false", "no", "off"}) - - def _env_multiplex_profiles_override() -> "bool | None": - """Resolve the GATEWAY_MULTIPLEX_PROFILES operator override. + """GATEWAY_MULTIPLEX_PROFILES operator override: True/False for a recognized token. - Returns ``True``/``False`` when the env var is set to a recognized truthy/ - falsy token, or ``None`` when it is unset, blank, or unrecognized — in which - case the caller keeps the config.yaml value (env > config > default). Blank - is deliberately ``None``, not ``False``: a provisioned-but-unpopulated Fly - secret arrives as ``""`` and must NOT shadow a config.yaml opt-in. + ``None`` when unset, blank, or unrecognized so the caller keeps the config.yaml + value (env > config > default). Blank is deliberately ``None``, not ``False``: + a provisioned-but-unpopulated Fly secret arrives as ``""`` and must NOT shadow + a config.yaml opt-in. """ raw = os.getenv("GATEWAY_MULTIPLEX_PROFILES") token = (raw or "").strip().lower() if not token: return None - if token in _MULTIPLEX_TRUTHY_STRINGS: + if token in _TRUTHY_STRINGS: return True - if token in _MULTIPLEX_FALSY_STRINGS: + if token in _FALSY_STRINGS: return False logger.warning( "Ignoring unrecognized GATEWAY_MULTIPLEX_PROFILES=%r " "(expected one of %s or %s); falling back to config.yaml.", raw, - sorted(_MULTIPLEX_TRUTHY_STRINGS), - sorted(_MULTIPLEX_FALSY_STRINGS), + sorted(_TRUTHY_STRINGS), + sorted(_FALSY_STRINGS), ) return None def _normalize_transport_token(value: Any) -> str: - """Normalize a streaming transport/mode value to a canonical token. + """Canonical streaming transport token. - Handles the YAML 1.1 boolean quirk where bare ``on`` / ``off`` parse to - Python ``True`` / ``False`` (see ``gateway/display_config.py`` ``_normalise``). - Without this, ``mode: off`` arrives as boolean ``False`` and stringifying it - yields ``"false"`` instead of the advertised ``"off"``, so streaming would be - enabled instead of disabled. Booleans map to ``"auto"`` (True) / ``"off"`` - (False); anything else is lower-cased, defaulting to ``"auto"``. + YAML 1.1 parses bare ``on``/``off`` as booleans, so ``mode: off`` arrives as + ``False``; stringifying would yield ``"false"`` and enable streaming. Booleans + map to ``"auto"`` / ``"off"``; anything else lower-cases, default ``"auto"``. """ if value is None: return "auto" @@ -133,7 +118,6 @@ def _normalize_transport_token(value: Any) -> str: def _coerce_float(value: Any, default: float) -> float: - """Coerce numeric config values, falling back on malformed input.""" if value is None: return default try: @@ -143,36 +127,23 @@ def _coerce_float(value: Any, default: float) -> float: def _coerce_int(value: Any, default: int) -> int: - """Coerce integer config values, falling back on malformed input.""" if value is None: return default try: return int(value) except (TypeError, ValueError, OverflowError): - # OverflowError: int(float("inf")) — a non-finite YAML value must - # degrade to the default, not abort gateway config loading. + # OverflowError: int(float("inf")) — non-finite YAML must degrade, not abort loading. return default def _coerce_optional_positive_int(value: Any, key: str) -> Optional[int]: - """Coerce an optional positive integer config value. - - ``None``/0/negative disable the setting. Malformed values are ignored with - a warning so a typo never prevents the gateway from starting. - """ + """``None``/0/negative disable; malformed values are ignored with a warning so a typo never blocks startup.""" if value is None: return None try: - if isinstance(value, bool): + if isinstance(value, bool) or (isinstance(value, float) and not value.is_integer()): raise ValueError(value) - if isinstance(value, float): - if not value.is_integer(): - raise ValueError(value) - parsed = int(value) - elif isinstance(value, str): - parsed = int(value.strip(), 10) - else: - parsed = int(value) + parsed = int(value.strip(), 10) if isinstance(value, str) else int(value) except (TypeError, ValueError): logger.warning( "Ignoring invalid %s=%r (expected a positive integer; 0/null disables)", key, value @@ -187,7 +158,7 @@ _SYSTEMD_WATCHDOG_MAX_SECONDS = 2_147_483_647 def coerce_systemd_watchdog_seconds( value: Any, key: str = "gateway.systemd_watchdog_seconds" ) -> int: - """Return a bounded positive watchdog interval or zero when disabled. + """Bounded positive watchdog interval, or zero when disabled/invalid. Runtime and service generation share this normalization so a value can never enable ``Type=notify`` while disabling application heartbeats. @@ -220,7 +191,6 @@ def coerce_systemd_watchdog_seconds( def _coerce_dict(value: Any) -> Dict[str, Any]: - """Return *value* when it is a mapping, otherwise an empty dict.""" return value if isinstance(value, dict) else {} @@ -245,11 +215,8 @@ def _dict_slot(container: dict, key: str) -> dict: def _getenv(name: str, default: Optional[str] = None) -> Optional[str]: """Read env vars through the active profile secret scope when present. - ``load_gateway_config()`` runs in many contexts, including multiplexed - profile startup where ``_profile_runtime_scope`` installs per-profile - secrets. In that scope we must prefer the scoped value; outside it we keep - legacy ``os.getenv`` behavior for single-profile callers and unscoped - gateway reads. + Multiplexed profile startup installs per-profile secrets, which must win; + outside a scope keep legacy ``os.getenv`` behavior. """ if current_secret_scope() is not None: scope_val = _get_secret(name, None) @@ -262,18 +229,16 @@ def _getenv_str(name: str, default: str = "") -> str: return val if val is not None else default -# Module-level cache for bundled platform plugin names (lives outside the -# enum so it doesn't become an accidental enum member). +# Bundled platform plugin names, cached outside the enum so it never becomes a member. _Platform__bundled_plugin_names: Optional[set] = None class Platform(Enum): """Supported messaging platforms. - Built-in platforms have explicit members. Plugin platforms use dynamic - members created on-demand by ``_missing_()`` so that - ``Platform("irc")`` works without modifying this enum. Dynamic members - are cached in ``_value2member_map_`` for identity-stable comparisons. + Built-ins are explicit members. Plugin platforms are dynamic members created + on demand by ``_missing_`` and cached in ``_value2member_map_`` so + ``Platform("irc") is Platform("irc")`` holds. """ LOCAL = "local" TELEGRAM = "telegram" @@ -299,39 +264,27 @@ class Platform(Enum): QQBOT = "qqbot" YUANBAO = "yuanbao" RELAY = "relay" # generic relay adapter fronted by the connector (EXPERIMENTAL) + @classmethod def _missing_(cls, value): - """Accept unknown platform names only for known plugin adapters. - - Creates a pseudo-member cached in ``_value2member_map_`` so that - ``Platform("irc") is Platform("irc")`` holds True (identity-stable). - Arbitrary strings are rejected to prevent enum pollution. - """ + """Accept unknown names only for bundled or runtime-registered plugin adapters (no enum pollution).""" if not isinstance(value, str) or not value.strip(): return None - # Normalise to lowercase to avoid case mismatches in config value = value.strip().lower() - # Check cache first (another call may have created it already) if value in cls._value2member_map_: return cls._value2member_map_[value] - # Only create pseudo-members for bundled plugin platforms (discovered - # via filesystem scan) or runtime-registered plugin platforms. global _Platform__bundled_plugin_names if _Platform__bundled_plugin_names is None: _Platform__bundled_plugin_names = cls._scan_bundled_plugin_platforms() if value in _Platform__bundled_plugin_names: return cls._add_pseudo_member(value) - - # Runtime-registered plugins (e.g. user-installed, discovered after - # the enum was defined). try: from gateway.platform_registry import platform_registry if platform_registry.is_registered(value): return cls._add_pseudo_member(value) except Exception: pass - return None @classmethod @@ -345,7 +298,7 @@ class Platform(Enum): @classmethod def _scan_bundled_plugin_platforms(cls) -> set: - """Return names of bundled platform plugins under ``plugins/platforms/``.""" + """Names of bundled platform plugins under ``plugins/platforms/``.""" names: set = set() try: platforms_dir = Path(__file__).parent.parent / "plugins" / "platforms" @@ -360,47 +313,29 @@ class Platform(Enum): return names -# Snapshot of built-in platform values before any dynamic _missing_ lookups. -# Used to distinguish real platforms from arbitrary strings. +# Built-in values snapshotted before any dynamic _missing_ lookup. _BUILTIN_PLATFORM_VALUES = frozenset(m.value for m in Platform.__members__.values()) -# Platforms that bind a host TCP port (HTTP/webhook listeners). In a profile -# multiplexer the default profile owns the single shared listener and serves -# every profile through the /p// URL prefix, so a SECONDARY profile -# enabling one of these is always a misconfiguration: it would try to bind a -# port already held by the default's listener. Single source of truth for -# both the gateway's fail-fast startup validation (gateway/run.py) and the -# dashboard's pre-write mutation validation (hermes_cli/web_server.py) so -# the two policies cannot drift. Stored as platform .value strings. +# Platforms that bind a host TCP port. In a profile multiplexer only the default +# profile owns the shared listener (served via /p//), so a SECONDARY +# profile enabling one of these is always a misconfiguration. Single source of +# truth for gateway/run.py startup validation and the dashboard's pre-write +# validation (hermes_cli/web_server.py). Platform .value strings. PORT_BINDING_PLATFORM_VALUES = frozenset({ - "webhook", - "api_server", - "msgraph_webhook", - "feishu", - "wecom_callback", - "bluebubbles", - "sms", - "whatsapp_cloud", - "line", - "teams", + "webhook", "api_server", "msgraph_webhook", "feishu", "wecom_callback", + "bluebubbles", "sms", "whatsapp_cloud", "line", "teams", }) -# Platforms whose port-binding status depends on connection mode. Feishu in -# websocket mode (its default) uses an outbound long connection — no listener. -# Only webhook/callback mode binds a port. Maps platform value → the mode -# value that actually binds (#52563). +# Platforms that only bind in one connection mode: Feishu's default websocket +# mode is an outbound long connection. platform value → the mode that binds. PORT_BINDING_CONDITIONAL_MODES: dict[str, str] = { "feishu": "webhook", } def platform_binds_port(platform_value: str, extra: Optional[dict] = None) -> bool: - """Return True when *platform_value* actually binds a port for *extra* config. - - Mode-conditional platforms (Feishu) only bind in their listener mode; - everything else in ``PORT_BINDING_PLATFORM_VALUES`` always binds. - """ + """True when *platform_value* actually binds a port for *extra* config.""" if platform_value not in PORT_BINDING_PLATFORM_VALUES: return False expected_mode = PORT_BINDING_CONDITIONAL_MODES.get(platform_value) @@ -412,24 +347,21 @@ def platform_binds_port(platform_value: str, extra: Optional[dict] = None) -> bo @dataclass class HomeChannel: - """ - Default destination for a platform. - - When a cron job specifies deliver="telegram" without a specific chat ID, - messages are sent to this home channel. Thread-aware platforms may also - store a thread/topic ID so the bare platform target routes to the exact - conversation where /sethome was run. + """Default destination for a platform (``deliver="telegram"`` without a chat ID). + + Thread-aware platforms may store a thread/topic ID so the bare platform + target routes to the conversation where /sethome was run. """ platform: Platform chat_id: str - name: str # Human-readable name for display + name: str thread_id: Optional[str] = None # Authenticated logical-target provenance observed by a platform adapter. - # Relay egress re-attaches these values, but the connector remains the - # authorization boundary and resolves them against its authoritative stores. + # Relay egress re-attaches these; the connector remains the authorization + # boundary and resolves them against its authoritative stores. user_id: Optional[str] = None scope_id: Optional[str] = None - + def to_dict(self) -> Dict[str, Any]: result = { "platform": self.platform.value, @@ -440,16 +372,15 @@ class HomeChannel: if getattr(self, key): result[key] = getattr(self, key) return result - + @classmethod def from_dict(cls, data: Dict[str, Any]) -> "HomeChannel": + optional = {k: str(data[k]) if data.get(k) else None for k in ("thread_id", "user_id", "scope_id")} return cls( platform=Platform(data["platform"]), chat_id=str(data["chat_id"]), name=data.get("name", "Home"), - thread_id=str(data["thread_id"]) if data.get("thread_id") else None, - user_id=str(data["user_id"]) if data.get("user_id") else None, - scope_id=str(data["scope_id"]) if data.get("scope_id") else None, + **optional, ) @@ -467,48 +398,32 @@ def persist_home_channel(home: HomeChannel, *, enabled_if_new: bool = False) -> @dataclass class SessionResetPolicy: - """ - Controls when sessions reset (lose context). - - Modes: - - "daily": Reset at a specific hour each day - - "idle": Reset after N minutes of inactivity - - "both": Whichever triggers first (daily boundary OR idle timeout) - - "none": Never auto-reset (context managed only by compression) + """Controls when sessions reset (lose context). - Default is "none" — sessions never auto-reset unless the user opts in - via the `session_reset` section in config.yaml (or gateway.json - overrides). Changed July 2026 from "both" (24h idle + daily 4am), which - surprised users who expected their conversations to persist. + Modes: "daily" (at ``at_hour``), "idle" (after ``idle_minutes``), "both" + (whichever first), "none" (never; context managed only by compression). + Default "none": sessions never auto-reset unless the user opts in via + ``session_reset`` in config.yaml (or gateway.json). """ - mode: str = "none" # "daily", "idle", "both", or "none" - at_hour: int = 4 # Hour for daily reset (0-23, local time) - idle_minutes: int = 1440 # Minutes of inactivity before reset (24 hours) - notify: bool = True # Send a notification to the user when auto-reset occurs - notify_exclude_platforms: tuple = ("api_server", "webhook") # Platforms that don't get reset notifications - # A background process this many hours old (or older) no longer blocks - # session idle/daily reset. A forgotten preview server should not keep a - # session alive forever (#29177). The process is NOT killed — only ignored - # by the reset guard. Raise this if you run legitimate multi-day jobs whose - # liveness should pin the conversation open. + mode: str = "none" + at_hour: int = 4 # 0-23, local time + idle_minutes: int = 1440 + notify: bool = True # Notify the user when auto-reset occurs + notify_exclude_platforms: tuple = ("api_server", "webhook") + # A background process this many hours old no longer blocks idle/daily reset + # (a forgotten preview server must not pin a session forever). The process + # is NOT killed — only ignored by the reset guard. bg_process_max_age_hours: int = 24 def to_dict(self) -> Dict[str, Any]: - return { - "mode": self.mode, - "at_hour": self.at_hour, - "idle_minutes": self.idle_minutes, - "notify": self.notify, - "notify_exclude_platforms": list(self.notify_exclude_platforms), - "bg_process_max_age_hours": self.bg_process_max_age_hours, - } - + return {**asdict(self), "notify_exclude_platforms": list(self.notify_exclude_platforms)} + @classmethod def from_dict(cls, data: Dict[str, Any]) -> "SessionResetPolicy": data = _coerce_dict(data) def val(key: str, default: Any) -> Any: - # Handle both missing keys and explicit null values (YAML null → None) + # Missing keys and explicit YAML nulls both take the default. value = data.get(key) return default if value is None else value @@ -525,13 +440,7 @@ class SessionResetPolicy: @dataclass class ChannelOverride: - """ - Per-channel override for model, provider, and system prompt. - - Used in config under platforms..channel_overrides[channel_id]. - Enables different channels (e.g. Discord #daily vs #dev) to use different - models and personas without running separate gateway instances. - """ + """Per-channel model/provider/system_prompt override (``platforms..channel_overrides[channel_id]``).""" model: Optional[str] = None provider: Optional[str] = None system_prompt: Optional[str] = None @@ -550,12 +459,11 @@ class ChannelOverride: ) -# Canonical map of platforms whose primary credential is ``PlatformConfig.token`` -# and the env var it loads from. Used for empty-token warnings at config -# validation and by the multiplex primary-startup credential gate in -# ``gateway.run`` (#64674). Platforms absent from this map authenticate some -# other way (session files, port-bound webhooks, api_key-only) and must never -# be skipped for a missing token. +# Platforms whose primary credential is ``PlatformConfig.token`` and the env var it +# loads from: drives empty-token warnings at validation and the multiplex +# primary-startup credential gate in ``gateway.run``. Platforms absent here +# authenticate another way (session files, webhooks, api_key) and must never be +# skipped for a missing token. PLATFORM_TOKEN_ENV_NAMES: dict["Platform", str] = { Platform.TELEGRAM: "TELEGRAM_BOT_TOKEN", Platform.DISCORD: "DISCORD_BOT_TOKEN", @@ -570,46 +478,24 @@ PLATFORM_TOKEN_ENV_NAMES: dict["Platform", str] = { class PlatformConfig: """Configuration for a single messaging platform.""" enabled: bool = False - token: Optional[str] = None # Bot token (Telegram, Discord) + token: Optional[str] = None api_key: Optional[str] = None # API key if different from token home_channel: Optional[HomeChannel] = None - - # Reply threading mode (Telegram/Slack) - # - "off": Never thread replies to original message - # - "first": Only first chunk threads to user's message (default) - # - "all": All chunks in multi-part replies thread to user's message + # Reply threading: "off" never threads, "first" threads only the first chunk, "all" every chunk. reply_to_mode: str = "first" - - # Whether the gateway is allowed to send "♻️ Gateway online" / - # "♻ Gateway restarted" lifecycle notifications on this platform. - # Default True preserves prior behavior. Set False on platforms used - # by end users (e.g. Slack) where operator-flavored restart pings are - # noise; keep True for back-channels where the operator wants them. + # "♻️ Gateway online/restarted" lifecycle pings. Set False on end-user + # platforms (e.g. Slack) where operator-flavored notices are noise. gateway_restart_notification: bool = True - - # Whether the gateway shows a "typing…" / "is thinking…" status indicator - # while the agent processes a message on this platform. Default True - # preserves prior behavior. Set False on platforms where the indicator is - # unwanted (e.g. Slack's assistant.threads.setStatus "is thinking…", which - # disables the compose box, or any platform where users find the bubble - # noisy). Drives the per-message _keep_typing refresh loop in - # gateway/platforms/base.py. + # "typing…" / "is thinking…" indicator while the agent works (drives the + # _keep_typing loop in gateway/platforms/base.py). Set False where it is + # unwanted, e.g. Slack's setStatus disables the compose box. typing_indicator: bool = True - - # Custom text for the working-state line on platforms whose typing - # indicator renders text rather than a native bubble: Slack's - # assistant.threads.setStatus line (shown next to the bot name; needs the - # assistant:write scope to render) and Google Chat's visible marker - # message. None keeps each platform's built-in default ("is thinking..." / - # "Hermes is thinking…"). Platforms with textless indicators (Discord, - # Telegram, Matrix, …) ignore it. + # Custom working-state text for platforms whose indicator renders text (Slack + # assistant status — needs assistant:write; Google Chat marker message). None + # keeps each platform's built-in default; textless indicators ignore it. typing_status_text: Optional[str] = None - - # Per-channel model/provider/system_prompt overrides (channel_id -> ChannelOverride) channel_overrides: Dict[str, ChannelOverride] = field(default_factory=dict) - - # Platform-specific settings - extra: Dict[str, Any] = field(default_factory=dict) + extra: Dict[str, Any] = field(default_factory=dict) # Platform-specific settings def to_dict(self) -> Dict[str, Any]: result = { @@ -640,22 +526,20 @@ class PlatformConfig: if isinstance(data.get("home_channel"), dict): home_channel = HomeChannel.from_dict(data["home_channel"]) - # gateway_restart_notification / typing_indicator / typing_status_text may - # arrive top-level or bridged into ``extra`` by the shared-key loop in - # load_gateway_config(), so YAML ``discord: gateway_restart_notification: false`` - # works without a separate platforms: block. Check both (top-level wins). + # gateway_restart_notification / typing_indicator / typing_status_text may be + # top-level or bridged into ``extra`` by the shared-key loop; top-level wins. extra = _coerce_dict(data.get("extra", {})) def toplevel_or_extra(key: str) -> Any: value = data.get(key) return extra.get(key) if value is None else value - channel_overrides: Dict[str, ChannelOverride] = {} raw_overrides = data.get("channel_overrides") or {} - if isinstance(raw_overrides, dict): - for cid, ov_data in raw_overrides.items(): - if isinstance(ov_data, dict): - channel_overrides[str(cid)] = ChannelOverride.from_dict(ov_data) + channel_overrides = { + str(cid): ChannelOverride.from_dict(ov_data) + for cid, ov_data in raw_overrides.items() + if isinstance(ov_data, dict) + } if isinstance(raw_overrides, dict) else {} return cls( enabled=_coerce_bool(data.get("enabled"), False), @@ -671,11 +555,9 @@ class PlatformConfig: ) -# Streaming defaults — single source of truth so both StreamingConfig and -# StreamConsumerConfig agree on the out-of-the-box edit rhythm. Tuned for -# Telegram's ~1 edit/s flood envelope: a touch under 1s lets the cadence -# breathe without bumping into rate limits, and a smaller buffer threshold -# makes short replies feel near-instant in DMs. +# Streaming defaults shared by StreamingConfig and StreamConsumerConfig. Tuned for +# Telegram's ~1 edit/s flood envelope: a touch under 1s breathes without hitting +# rate limits; the small buffer threshold makes short DM replies feel instant. DEFAULT_STREAMING_EDIT_INTERVAL: float = 0.8 DEFAULT_STREAMING_BUFFER_THRESHOLD: int = 24 DEFAULT_STREAMING_CURSOR: str = " ▉" @@ -683,79 +565,41 @@ DEFAULT_STREAMING_CURSOR: str = " ▉" @dataclass class StreamingConfig: - """Configuration for real-time token streaming to messaging platforms.""" + """Real-time token streaming to messaging platforms.""" enabled: bool = False - # Transport selection: - # "auto" — prefer native streaming-draft updates when the platform - # supports them (Telegram sendMessageDraft, Bot API 9.5+); - # fall back to edit-based when not. - # "draft" — explicitly request native drafts; falls back to edit when - # the platform/chat doesn't support them. - # "edit" — progressive editMessageText only (legacy behaviour). - # "off" — disable streaming entirely. - # - # Default is "auto": prefer native draft streaming on platforms that - # support it (Telegram DMs via sendMessageDraft, Bot API 9.5+) and fall - # back to edit-based streaming everywhere else. This is safe as a global - # default because adapters without draft support (Discord, Slack, Matrix, - # …) report supports_draft_streaming() == False and transparently use the - # edit path — so "auto" never regresses non-Telegram platforms, it only - # upgrades the chats that can render the smoother native preview. + # Transport: "auto" prefers native draft updates (Telegram sendMessageDraft, + # Bot API 9.5+) and falls back to edit-based; "draft" requests drafts with + # edit fallback; "edit" is progressive editMessageText only; "off" disables. + # "auto" is safe globally: adapters without draft support report + # supports_draft_streaming() == False and use the edit path unchanged. transport: str = "auto" edit_interval: float = DEFAULT_STREAMING_EDIT_INTERVAL buffer_threshold: int = DEFAULT_STREAMING_BUFFER_THRESHOLD cursor: str = DEFAULT_STREAMING_CURSOR - # Ported from openclaw/openclaw#72038. When >0, the final edit for - # a long-running streamed response is delivered as a fresh message - # if the original preview has been visible for at least this many - # seconds, so the platform's visible timestamp reflects completion - # time instead of the preview creation time. Currently applied to - # Telegram only (other platforms ignore the setting). Default 0 disables - # the fresh-message replacement path; set >0 to opt in. + # When >0, the final edit of a long stream is delivered as a fresh message if + # the preview has been visible at least this many seconds, so the visible + # timestamp reflects completion. Telegram only; 0 disables. fresh_final_after_seconds: float = 0.0 def to_dict(self) -> Dict[str, Any]: - return { - "enabled": self.enabled, - "transport": self.transport, - "edit_interval": self.edit_interval, - "buffer_threshold": self.buffer_threshold, - "cursor": self.cursor, - "fresh_final_after_seconds": self.fresh_final_after_seconds, - } + return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> "StreamingConfig": if not isinstance(data, dict) or not data: return cls() - # ``mode`` is an ergonomic alias for the transport that ALSO implies - # ``enabled``. A config like ``streaming: {mode: auto}`` reads as - # "turn streaming on, transport=auto" — matching the natural intent - # of someone enabling streaming without also spelling out - # ``enabled: true``. Without this, ``mode`` was silently ignored and - # streaming stayed disabled (``enabled`` defaults to False), which is - # a surprising footgun: the whole reply buffers and sends at once. - # ``mode: off`` disables streaming; an explicit ``enabled`` key always - # wins so callers can force either state. - # - # ``transport`` alone does NOT imply ``enabled``: ``streaming.enabled`` - # is the documented master switch (see website/docs/user-guide/ - # configuration.md), so a bare ``transport`` only selects HOW to stream - # once streaming is on. Only the ``mode`` alias flips ``enabled``. + # ``mode`` is a transport alias that ALSO implies ``enabled`` (``mode: off`` + # disables); an explicit ``enabled`` key always wins. A bare ``transport`` + # does NOT imply enabled: ``streaming.enabled`` is the documented master + # switch, so ``transport`` only selects HOW to stream once it is on. raw_transport = data.get("transport") raw_mode = data.get("mode") - # Normalize both through the same helper so YAML's bare ``off``/``on`` - # (parsed as bool False/True) become canonical tokens rather than - # ``"false"``/``"true"``. - picked = raw_transport if raw_transport is not None else raw_mode - transport = _normalize_transport_token(picked) + transport = _normalize_transport_token(raw_transport if raw_transport is not None else raw_mode) if "enabled" in data: enabled = _coerce_bool(data.get("enabled"), False) elif raw_mode is not None: - # The ``mode`` alias (and only ``mode``) infers enabled: - # ``off`` disables, anything else enables. enabled = _normalize_transport_token(raw_mode) != "off" else: enabled = False @@ -763,32 +607,18 @@ class StreamingConfig: return cls( enabled=enabled, transport=transport, - edit_interval=_coerce_float( - data.get("edit_interval"), DEFAULT_STREAMING_EDIT_INTERVAL, - ), - buffer_threshold=_coerce_int( - data.get("buffer_threshold"), DEFAULT_STREAMING_BUFFER_THRESHOLD, - ), + edit_interval=_coerce_float(data.get("edit_interval"), DEFAULT_STREAMING_EDIT_INTERVAL), + buffer_threshold=_coerce_int(data.get("buffer_threshold"), DEFAULT_STREAMING_BUFFER_THRESHOLD), cursor=data.get("cursor", DEFAULT_STREAMING_CURSOR), - fresh_final_after_seconds=_coerce_float( - data.get("fresh_final_after_seconds"), 0.0 - ), + fresh_final_after_seconds=_coerce_float(data.get("fresh_final_after_seconds"), 0.0), ) -# ----------------------------------------------------------------------------- -# Built-in platform connection checkers -# ----------------------------------------------------------------------------- -# Each callable receives a ``PlatformConfig`` and returns ``True`` when the -# platform is sufficiently configured to be considered "connected". Platforms -# that rely on the generic ``token or api_key`` check (Telegram, Discord, -# Slack, Matrix, Mattermost, HomeAssistant) do not need an entry here. def _has_usable_api_server_key(key: object) -> bool: - """True when API_SERVER_KEY is present and strong enough to be usable. + """True when API_SERVER_KEY is strong enough for the adapter to start. Mirrors the startup guard in ``gateway/platforms/api_server.py`` - (``has_usable_secret`` with ``min_length=16``) so the platform is only - enrolled at load time when the adapter would actually agree to start. + (``has_usable_secret`` with ``min_length=16``). """ if not key: return False @@ -799,152 +629,96 @@ def _has_usable_api_server_key(key: object) -> bool: return has_usable_secret(key, min_length=16) +def _needs_extra(*keys: str) -> Callable[[PlatformConfig], bool]: + return lambda cfg: all(cfg.extra.get(k) for k in keys) + + +# Built-in "is this platform sufficiently configured?" checks by PlatformConfig. +# Platforms covered by the generic ``token or api_key`` check (Telegram, Discord, +# Slack, Matrix, Mattermost, HomeAssistant) need no entry. _PLATFORM_CONNECTED_CHECKERS: dict[Platform, Callable[[PlatformConfig], bool]] = { - Platform.WEIXIN: lambda cfg: bool( - cfg.extra.get("account_id") and (cfg.token or cfg.extra.get("token")) - ), - Platform.WHATSAPP_CLOUD: lambda cfg: bool( - cfg.extra.get("phone_number_id") and cfg.extra.get("access_token") - ), - Platform.SIGNAL: lambda cfg: bool(cfg.extra.get("http_url")), - Platform.API_SERVER: lambda cfg: _has_usable_api_server_key( - cfg.extra.get("key") if cfg else None - ), + Platform.WEIXIN: lambda cfg: bool(cfg.extra.get("account_id") and (cfg.token or cfg.extra.get("token"))), + Platform.WHATSAPP_CLOUD: _needs_extra("phone_number_id", "access_token"), + Platform.SIGNAL: _needs_extra("http_url"), + Platform.API_SERVER: lambda cfg: _has_usable_api_server_key(cfg.extra.get("key") if cfg else None), Platform.WEBHOOK: lambda cfg: True, - Platform.MSGRAPH_WEBHOOK: lambda cfg: bool( - str(cfg.extra.get("client_state") or "").strip() - ), - Platform.BLUEBUBBLES: lambda cfg: bool( - cfg.extra.get("server_url") and cfg.extra.get("password") - ), - Platform.QQBOT: lambda cfg: bool( - cfg.extra.get("app_id") and cfg.extra.get("client_secret") - ), - Platform.YUANBAO: lambda cfg: bool( - cfg.extra.get("app_id") and cfg.extra.get("app_secret") - ), - # Relay dials OUT to a connector; it is "connected" once an endpoint URL is - # configured (extra["relay_url"] or extra["url"]). The capability descriptor - # is negotiated at handshake time, so the URL is the only config-level - # signal in the experimental phase. EXPERIMENTAL — may change. - Platform.RELAY: lambda cfg: bool( - cfg.extra.get("relay_url") or cfg.extra.get("url") - ), + Platform.MSGRAPH_WEBHOOK: lambda cfg: bool(str(cfg.extra.get("client_state") or "").strip()), + Platform.BLUEBUBBLES: _needs_extra("server_url", "password"), + Platform.QQBOT: _needs_extra("app_id", "client_secret"), + Platform.YUANBAO: _needs_extra("app_id", "app_secret"), + # Relay dials OUT to a connector: "connected" once an endpoint URL is configured + # (capabilities are negotiated at handshake). EXPERIMENTAL. + Platform.RELAY: lambda cfg: bool(cfg.extra.get("relay_url") or cfg.extra.get("url")), } @dataclass class GatewayConfig: - """ - Main gateway configuration. - - Manages all platform connections, session policies, and delivery settings. - """ - # Platform configurations + """Main gateway configuration: platform connections, session policies, delivery settings.""" platforms: Dict[Platform, PlatformConfig] = field(default_factory=dict) - - # Session reset policies by type default_reset_policy: SessionResetPolicy = field(default_factory=SessionResetPolicy) reset_by_type: Dict[str, SessionResetPolicy] = field(default_factory=dict) reset_by_platform: Dict[Platform, SessionResetPolicy] = field(default_factory=dict) - - # Reset trigger commands reset_triggers: List[str] = field(default_factory=lambda: ["/new", "/reset"]) - - # User-defined quick commands (slash commands that bypass the agent loop) + # Slash commands that bypass the agent loop. quick_commands: Dict[str, Any] = field(default_factory=dict) - - # Storage paths sessions_dir: Path = field(default_factory=lambda: get_hermes_home() / "sessions") - - # Whether to keep writing the legacy sessions.json mirror of the gateway - # routing index. The primary copy lives in state.db (gateway_routing - # table, #9006). Default True for backward compatibility with external - # tooling and downgrade safety; set gateway.write_sessions_json: false in - # config.yaml to stop producing the file. + # Keep writing the legacy sessions.json mirror of the routing index (primary + # copy: state.db gateway_routing). Default True for external tooling and + # downgrade safety. write_sessions_json: bool = True - - # Delivery settings always_log_local: bool = True # Always save cron outputs to local files - # Drop outbound "silence narration" messages (e.g. *(silent)*, 🔇, a bare - # ".") pre-send. These are model hallucinations emitted when a persona has - # nothing actionable to say; in bot-to-bot channels they mirror back and - # forth, burning tokens and crashing models. Substrate-level guard that - # survives SOUL.md/prompt drift across providers. Opt out with False for - # raw passthrough. + # Drop outbound "silence narration" (*(silent)*, 🔇, a bare ".") pre-send: + # model hallucinations that ping-pong in bot-to-bot channels. Substrate-level + # guard that survives SOUL.md/prompt drift; False = raw passthrough. filter_silence_narration: bool = True - - # STT settings - stt_enabled: bool = True # Whether to auto-transcribe inbound voice messages - stt_echo_transcripts: bool = True # Whether to echo raw STT transcripts back to the user - - # Session isolation in shared chats - group_sessions_per_user: bool = True # Isolate group/channel sessions per participant when user IDs are available - thread_sessions_per_user: bool = False # When False (default), threads are shared across all participants - max_concurrent_sessions: Optional[int] = None # Positive int caps simultaneous active chat sessions - - # Multi-profile multiplexing (opt-in; default off preserves one-gateway-per-profile). - # When True, the default profile's gateway serves inbound messages for every - # profile on the host: profiles are stamped into session keys and (in later - # phases) per-profile adapters/credentials are resolved. When False, the - # gateway behaves exactly as before — single HERMES_HOME, no profile stamping. + stt_enabled: bool = True # Auto-transcribe inbound voice messages + stt_echo_transcripts: bool = True # Echo raw STT transcripts back to the user + group_sessions_per_user: bool = True # Isolate group sessions per participant when user IDs exist + thread_sessions_per_user: bool = False # False = threads shared across participants + max_concurrent_sessions: Optional[int] = None # Positive int caps simultaneous active sessions + # Opt-in: the default profile's gateway serves inbound messages for every + # profile on the host (profiles stamped into session keys, per-profile + # adapters/credentials). False = single HERMES_HOME, no profile stamping. multiplex_profiles: bool = False - # Optional named-profile allowlist for multiplex mode. None preserves the - # historical serve-all behavior; [] serves only the default profile. + # None = historical serve-all; [] = default profile only. multiplex_profile_allowlist: Optional[List[str]] = None - - # Public HTTPS endpoint another gateway may use for scoped RoomLink calls. - # Disabled by default: setting an API key alone must never expose or - # advertise a route. HERMES_ROOM_LINK_URL remains the operator override. + # Public HTTPS endpoint for scoped RoomLink calls. Disabled by default: an API + # key alone must never advertise a route. HERMES_ROOM_LINK_URL overrides. room_link_url: Optional[str] = None - - # Opt-in systemd event-loop watchdog. Zero preserves Type=simple and - # disables sd_notify at runtime. + # Opt-in systemd event-loop watchdog; zero keeps Type=simple and disables sd_notify. systemd_watchdog_seconds: int = 0 - - # In-process event-loop liveness watchdog (#69089). A daemon OS thread - # probes the gateway loop with call_soon_threadsafe; after consecutive - # missed probes it dumps all-thread stacks and hard-exits with the - # service-restart code so the supervisor can revive the process. On by - # default; set gateway.loop_watchdog: false in config.yaml to disable. - # - # Tuning knobs (all seconds unless noted) make the watchdog tolerate - # *transient, self-recovering* event-loop stalls — e.g. Telegram/Discord - # reconnect doing synchronous socket I/O during a network blip — so a - # short block does not force exit code 75 and trigger a restart churn - # that stalls cron dispatch (recurring fleet incidents on 2026-08-17, - # kanban t_0f76430f/t_70483f23). A genuine wedge (event loop frozen for - # the full tolerance window) still escalates to a supervised restart. + # In-process event-loop liveness watchdog: a daemon thread probes the loop + # with call_soon_threadsafe; after consecutive misses it dumps all-thread + # stacks and hard-exits with the service-restart code. The knobs (seconds) + # tolerate transient self-recovering stalls (adapter reconnect doing sync + # socket I/O) so a short block does not force exit code 75 and restart + # churn that stalls cron; a genuine wedge still escalates. loop_watchdog: bool = True - # Seconds the watchdog waits between liveness probes. loop_watchdog_probe_interval_s: float = DEFAULT_LOOP_WATCHDOG_INTERVAL_S - # Seconds a single probe may go unprocessed before it counts as a miss. loop_watchdog_probe_timeout_s: float = DEFAULT_LOOP_WATCHDOG_TIMEOUT_S - # Consecutive missed probes allowed before the watchdog hard-exits. - # Default stays at 3 (~90-120s of sustained loop block): the transient - # false-positive class (the watchdog's own on-loop heartbeat fsync) - # is fixed at the root by the off-loop write + two-witness probe, so - # raising this fleet-wide would only delay genuine-wedge recovery. + # Default 3 (~90-120s sustained block): the false-positive class (the + # watchdog's own on-loop heartbeat fsync) is fixed at the root by the off-loop + # write + two-witness probe, so raising this would only delay recovery. loop_watchdog_max_strikes: int = DEFAULT_LOOP_WATCHDOG_MAX_STRIKES - - # Unauthorized DM policy unauthorized_dm_behavior: str = "pair" # "pair" or "ignore" - - # Streaming configuration streaming: StreamingConfig = field(default_factory=StreamingConfig) - - # Session store pruning: drop SessionEntry records older than this many - # days from the in-memory dict and sessions.json. Keeps the store from - # growing unbounded in gateways serving many chats/threads/users over - # months. Pruning is invisible to users — if they resume, they get a - # fresh session exactly as if the reset policy had fired. 0 = disabled. + # Drop SessionEntry records older than this from the store and sessions.json. + # Invisible to users (a resumed chat gets a fresh session). 0 = disabled. session_store_max_age_days: int = 90 - - # Profile-based routing: route specific guilds/channels/threads to - # different profiles. See gateway/profile_routing.py. Each entry is a - # dict with: name, platform, profile, and optional guild_id/chat_id/thread_id. + # Route guilds/channels/threads to profiles (gateway/profile_routing.py). profile_routes: list = field(default_factory=list) + # Scalar fields serialized verbatim by ``to_dict`` (in output order). + _SCALAR_DICT_FIELDS = ( + "write_sessions_json", "always_log_local", "filter_silence_narration", "stt_enabled", + "stt_echo_transcripts", "group_sessions_per_user", "thread_sessions_per_user", + "max_concurrent_sessions", "multiplex_profiles", "multiplex_profile_allowlist", + "room_link_url", "systemd_watchdog_seconds", "loop_watchdog", + "loop_watchdog_probe_interval_s", "loop_watchdog_probe_timeout_s", + "loop_watchdog_max_strikes", "unauthorized_dm_behavior", + ) + def __post_init__(self) -> None: self.multiplex_profile_allowlist = _normalize_multiplex_profile_allowlist( self.multiplex_profile_allowlist @@ -954,13 +728,10 @@ class GatewayConfig: ) def get_connected_platforms(self) -> List[Platform]: - """Return list of platforms that are enabled and configured. + """Enabled + configured platforms, sorted by value. - Sorted by platform value so the rendered "Connected Platforms" list - (and the home-channel blocks derived from it) is byte-stable across - gateway restarts and mid-process platform registration — dict - insertion order is not a stable contract and a reorder busts the - prompt cache without any semantic change. + Sorted so the rendered "Connected Platforms" list is byte-stable across + restarts and mid-process registration: a reorder busts the prompt cache. """ connected = [ platform @@ -970,25 +741,17 @@ class GatewayConfig: return sorted(connected, key=lambda p: str(p.value)) def _is_platform_connected(self, platform: Platform, config: PlatformConfig) -> bool: - """Check whether a single platform is sufficiently configured.""" checker = _PLATFORM_CONNECTED_CHECKERS.get(platform) - # Weixin requires both a token and an account_id (checked first so - # the generic token branch doesn't let it through without account_id). + # Weixin needs token AND account_id, so it must bypass the generic token branch. if platform == Platform.WEIXIN: return checker(config) - - # Generic token/api_key auth covers Telegram, Discord, Slack, etc. if config.token or config.api_key: return True - - # Platform-specific check if checker is not None: return checker(config) - # Plugin-registered platforms. Force plugin discovery first so this - # works even when GatewayConfig is constructed directly (e.g. in tests - # or callers that bypass load_gateway_config(), which is what triggers - # discovery in the normal path). discover_plugins() is idempotent. + # Plugin-registered platforms. Force (idempotent) plugin discovery so this + # works when GatewayConfig is constructed directly, bypassing load_gateway_config(). try: from gateway.platform_registry import platform_registry try: @@ -1005,74 +768,44 @@ class GatewayConfig: return True except Exception: pass # Registry not yet initialised during early import - return False - + def get_home_channel(self, platform: Platform) -> Optional[HomeChannel]: - """Get the home channel for a platform.""" config = self.platforms.get(platform) return config.home_channel if config else None - + def get_reset_policy( - self, + self, platform: Optional[Platform] = None, session_type: Optional[str] = None ) -> SessionResetPolicy: - """ - Get the appropriate reset policy for a session. - - Priority: platform override > type override > default - """ - # Platform-specific override takes precedence + """Priority: platform override > type override > default.""" if platform and platform in self.reset_by_platform: return self.reset_by_platform[platform] - - # Type-specific override (dm, group, thread) if session_type and session_type in self.reset_by_type: return self.reset_by_type[session_type] - return self.default_reset_policy - + def to_dict(self) -> Dict[str, Any]: - return { - "platforms": { - p.value: c.to_dict() for p, c in self.platforms.items() - }, + result: Dict[str, Any] = { + "platforms": {p.value: c.to_dict() for p, c in self.platforms.items()}, "default_reset_policy": self.default_reset_policy.to_dict(), - "reset_by_type": { - k: v.to_dict() for k, v in self.reset_by_type.items() - }, - "reset_by_platform": { - p.value: v.to_dict() for p, v in self.reset_by_platform.items() - }, + "reset_by_type": {k: v.to_dict() for k, v in self.reset_by_type.items()}, + "reset_by_platform": {p.value: v.to_dict() for p, v in self.reset_by_platform.items()}, "reset_triggers": self.reset_triggers, "quick_commands": self.quick_commands, "sessions_dir": str(self.sessions_dir), - "write_sessions_json": self.write_sessions_json, - "always_log_local": self.always_log_local, - "filter_silence_narration": self.filter_silence_narration, - "stt_enabled": self.stt_enabled, - "stt_echo_transcripts": self.stt_echo_transcripts, - "group_sessions_per_user": self.group_sessions_per_user, - "thread_sessions_per_user": self.thread_sessions_per_user, - "max_concurrent_sessions": self.max_concurrent_sessions, - "multiplex_profiles": self.multiplex_profiles, - "multiplex_profile_allowlist": self.multiplex_profile_allowlist, - "room_link_url": self.room_link_url, - "systemd_watchdog_seconds": self.systemd_watchdog_seconds, - "loop_watchdog": self.loop_watchdog, - "loop_watchdog_probe_interval_s": self.loop_watchdog_probe_interval_s, - "loop_watchdog_probe_timeout_s": self.loop_watchdog_probe_timeout_s, - "loop_watchdog_max_strikes": self.loop_watchdog_max_strikes, - "unauthorized_dm_behavior": self.unauthorized_dm_behavior, - "streaming": self.streaming.to_dict(), - "session_store_max_age_days": self.session_store_max_age_days, - "profile_routes": [ - asdict(r) if is_dataclass(r) and not isinstance(r, type) else r - for r in self.profile_routes - ], } - + for name in self._SCALAR_DICT_FIELDS: + result[name] = getattr(self, name) + result["streaming"] = self.streaming.to_dict() + result["session_store_max_age_days"] = self.session_store_max_age_days + result["profile_routes"] = [ + asdict(r) if is_dataclass(r) and not isinstance(r, type) else r + for r in self.profile_routes + ] + return result + @classmethod def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig": data = _coerce_dict(data) @@ -1082,6 +815,10 @@ class GatewayConfig: """Top-level key wins by presence; else the nested ``gateway.`` form.""" return data[key] if key in data else nested_gateway.get(key) + def key_label(key: str) -> str: + """Warning key prefix: "gateway." when the nested form was the one consulted.""" + return key if key in data else f"gateway.{key}" + platforms = {} for platform_name, platform_data in _coerce_dict(data.get("platforms", {})).items(): if not isinstance(platform_data, dict): @@ -1110,10 +847,6 @@ class GatewayConfig: if not isinstance(room_link_url, str): room_link_url = None - # Key prefix for the warning: "gateway." when the nested form was the one consulted. - def key_label(key: str) -> str: - return key if key in data else f"gateway.{key}" - # Watchdog knobs: out-of-range / non-finite values fall back to the shipped defaults. probe_interval = _coerce_float(pick("loop_watchdog_probe_interval_s"), DEFAULT_LOOP_WATCHDOG_INTERVAL_S) if not math.isfinite(probe_interval) or not 1.0 <= probe_interval <= 3600.0: @@ -1129,14 +862,10 @@ class GatewayConfig: pick("systemd_watchdog_seconds"), key_label("systemd_watchdog_seconds") ) - # Multiplexing is a genuine 3-tier chain: env > config.yaml > default False. The - # GATEWAY_MULTIPLEX_PROFILES operator override wins when set to a recognized value - # (hosted deployments stamp it on the container so the single multiplexed gateway the - # connector depends on is forced on at every boot regardless of the image's config.yaml); - # a blank or unrecognized env value falls through to config — a provisioned-but- - # unpopulated Fly secret must not shadow a config.yaml opt-in. Config side: the - # top-level VALUE wins when not None, else ``gateway.multiplex_profiles`` (written by - # ``hermes config set gateway.multiplex_profiles true``). + # Multiplexing is env > config.yaml > default False. The GATEWAY_MULTIPLEX_PROFILES + # operator override wins when set to a recognized value (hosted deployments stamp + # it on the container); blank/unrecognized falls through to config. Config side: + # the top-level VALUE wins when not None, else ``gateway.multiplex_profiles``. multiplex_profiles = data.get("multiplex_profiles") if multiplex_profiles is None: multiplex_profiles = nested_gateway.get("multiplex_profiles") @@ -1153,9 +882,7 @@ class GatewayConfig: except (TypeError, ValueError): session_store_max_age_days = 90 - # Parse profile routes (validated by gateway.profile_routing) from gateway.profile_routing import parse_profile_routes - profile_routes = parse_profile_routes(data.get("profile_routes") or []) return cls( platforms=platforms, @@ -1191,15 +918,15 @@ class GatewayConfig: ), streaming=StreamingConfig.from_dict(data.get("streaming", {})), session_store_max_age_days=session_store_max_age_days, - profile_routes=profile_routes, + profile_routes=parse_profile_routes(data.get("profile_routes") or []), ) def get_unauthorized_dm_behavior(self, platform: Optional[Platform] = None) -> str: - """Return the effective unauthorized-DM behavior for a platform. + """Effective unauthorized-DM behavior for a platform. Email is inbox-shaped, not chat-shaped, so it defaults to ``"ignore"`` - unless ``platforms.email.unauthorized_dm_behavior`` explicitly opts - into pairing. A global default does not opt email into pairing. + unless ``platforms.email.unauthorized_dm_behavior`` explicitly opts in; + a global default does not opt email into pairing. """ if platform: platform_cfg = self.platforms.get(platform) @@ -1214,7 +941,7 @@ class GatewayConfig: return self.unauthorized_dm_behavior def get_notice_delivery(self, platform: Optional[Platform] = None) -> str: - """Return the effective notice-delivery mode for a platform.""" + """Effective notice-delivery mode ("public"/"private") for a platform.""" if platform: platform_cfg = self.platforms.get(platform) if platform_cfg and "notice_delivery" in platform_cfg.extra: @@ -1225,15 +952,7 @@ class GatewayConfig: def load_gateway_config() -> GatewayConfig: - """ - Load gateway configuration from multiple sources. - - Priority (highest to lowest): - 1. Environment variables - 2. ~/.hermes/config.yaml (primary user-facing config) - 3. ~/.hermes/gateway.json (legacy — provides defaults under config.yaml) - 4. Built-in defaults - """ + """Load gateway configuration. Priority: env > ~/.hermes/config.yaml > legacy gateway.json > defaults.""" from gateway import config_loader _home = get_hermes_home() @@ -1255,11 +974,7 @@ def load_gateway_config() -> GatewayConfig: def _validate_gateway_config(config: "GatewayConfig") -> None: - """Validate and sanitize a loaded GatewayConfig in place. - - Called by ``load_gateway_config()`` after all config sources are merged. - Extracted as a separate function for testability. - """ + """Validate and sanitize a loaded GatewayConfig in place (after all sources are merged).""" policy = config.default_reset_policy if not (0 <= policy.at_hour <= 23): @@ -1275,8 +990,7 @@ def _validate_gateway_config(config: "GatewayConfig") -> None: ) policy.idle_minutes = 1440 - # Warn about empty bot tokens — platforms that loaded an empty string - # won't connect and the cause can be confusing without a log line. + # An empty token won't connect and the cause is confusing without a log line. for platform, pconfig in config.platforms.items(): if not pconfig.enabled: continue @@ -1288,10 +1002,8 @@ def _validate_gateway_config(config: "GatewayConfig") -> None: platform.value, env_name, ) - # Reject known-weak placeholder tokens. - # Ported from openclaw/openclaw#64586: users who copy .env.example - # without changing placeholder values get a clear startup error instead - # of a confusing "auth failed" from the platform API. + # Reject known-weak placeholder tokens (copied .env.example) with a clear + # startup error instead of a confusing "auth failed" from the platform API. try: from hermes_cli.auth import has_usable_secret except ImportError: diff --git a/gateway/config_env.py b/gateway/config_env.py index fc56881975..0b90e0980c 100644 --- a/gateway/config_env.py +++ b/gateway/config_env.py @@ -2,9 +2,9 @@ Runs after ``GatewayConfig.from_dict`` so env always wins over config.yaml / gateway.json. Most platforms follow one shape — "credentials present in env -⇒ enable the platform and copy the values into ``extra``" — and are declared -as ``_Cred`` rows in ``_ENV_STEPS`` (source order = application order). -Platforms with unique gating live as small functions in the same table. +⇒ enable the platform and copy the values into ``extra``" — declared as +``_Cred`` rows in ``_ENV_STEPS`` (source order = application order). +Platforms with unique gating are small functions in the same table. """ import contextlib @@ -13,7 +13,7 @@ import logging import os from dataclasses import dataclass from functools import partial -from typing import Any, Callable, Dict, List, Optional +from typing import Any, Callable, Dict, Optional from gateway.config import ( GatewayConfig, @@ -30,15 +30,13 @@ logger = logging.getLogger("gateway.config") getenv = _getenv_str -# Platforms for which the "explicitly disabled in config.yaml, but credentials -# are present in the environment" WARNING has already been emitted in this -# process. Config reloads on every turn, so the notice is one-time per platform -# per process — loud once at startup, never a per-turn drumbeat. +# Platforms already warned about "explicitly disabled in config.yaml, but +# credentials present in env". Config reloads every turn, so the notice is +# one-time per platform per process. _EXPLICIT_DISABLE_WARNED: set = set() - # Env var(s) whose presence drives each platform's env-enable branch, named in -# the explicit-disable WARNING below. +# the explicit-disable WARNING. _ENV_ENABLE_CREDENTIALS: dict = { Platform.TELEGRAM: ("TELEGRAM_BOT_TOKEN",), Platform.DISCORD: ("DISCORD_BOT_TOKEN",), @@ -65,10 +63,9 @@ _ENV_ENABLE_CREDENTIALS: dict = { def _warn_explicit_disable_beats_env(platform: Platform) -> None: """One-time WARNING: ``platforms..enabled: false`` wins over env creds. - Until #48820 credential presence force-enabled twelve platforms regardless - of an explicit ``enabled: false``; users relying on "creds in .env = - platform on" need to be told why it went dark. Names the platform, the - winning config key, and the env var(s) that used to override it. + Credential presence used to force-enable platforms regardless of an explicit + ``enabled: false``; users relying on "creds in .env = platform on" must be + told why it went dark. """ if platform in _EXPLICIT_DISABLE_WARNED: return @@ -88,12 +85,12 @@ def _warn_explicit_disable_beats_env(platform: Platform) -> None: # --- small value parsers ----------------------------------------------------- -def _csv_list(value: str) -> List[str]: +def _csv_list(value: str) -> list: return [part.strip() for part in value.split(",") if part.strip()] def _int_or(default: int) -> Callable[[str], int]: - """``int(raw.strip(), 10)`` with *default* on malformed/blank input (``_getenv_int`` semantics).""" + """``int(raw.strip(), 10)`` with *default* on malformed/blank input.""" def parse(raw: str) -> int: try: return int(str(raw).strip(), 10) @@ -121,11 +118,7 @@ def _env_first(envs) -> str: """First truthy value among *envs* (a single name or a tuple of alternatives).""" if isinstance(envs, str): return getenv(envs) - for env in envs: - value = getenv(env) - if value: - return value - return "" + return next((v for v in map(getenv, envs) if v), "") # --- reusable steps ----------------------------------------------------------- @@ -136,8 +129,8 @@ _INT = object() # spec marker: ``int(value)``, silently skipped when malformed def _env_extras(extra: Dict[str, Any], spec, *, strip: bool = False) -> None: """``extra[key] = fn(value)`` for each ``(key, env[, fn])`` whose env value is truthy. - With ``strip=True`` the value is stripped BEFORE the truthiness check. ``fn=_INT`` parses an - int and silently ignores a non-integer value. + ``strip=True`` strips BEFORE the truthiness check. ``fn=_INT`` parses an int + and silently ignores a non-integer value. """ for key, env, *fn in spec: value = getenv(env) @@ -175,8 +168,8 @@ def _env_reply_mode(config: GatewayConfig, platform: Platform, env: str) -> None def _enable_from_env(config: GatewayConfig, platform: Platform) -> PlatformConfig: """Enable *platform* because its env credentials are present — unless config.yaml explicitly disabled it. - READS (does not pop) the ``_enabled_explicit`` marker: the registry-driven plugin-enable pass later - also needs it. The marker is scrubbed for all platforms at the end of ``_apply_env_overrides``. + READS (does not pop) the ``_enabled_explicit`` marker: the registry-driven + plugin-enable pass also needs it; ``_scrub_explicit_markers`` removes it last. """ if platform not in config.platforms: config.platforms[platform] = PlatformConfig(enabled=True) @@ -184,7 +177,6 @@ def _enable_from_env(config: GatewayConfig, platform: Platform) -> PlatformConfi platform_config = config.platforms[platform] if not platform_config.enabled: if platform_config.extra.get("_enabled_explicit", False): - # Credentials are present (that is why we are here) but the user said no in config.yaml. _warn_explicit_disable_beats_env(platform) else: platform_config.enabled = True @@ -194,10 +186,10 @@ def _enable_from_env(config: GatewayConfig, platform: Platform) -> PlatformConfi def _enable_port_bound_from_env(config: GatewayConfig, platform: Platform) -> PlatformConfig: """Enable a port-binding platform (api_server/webhook) unless config.yaml explicitly disabled it. - In multiplex mode a secondary profile pins ``platforms..enabled: false`` so it shares the - default profile's listener instead of binding its own port, yet still inherits the process env; - without this guard the env presence would force-enable the listener and trip MultiplexConfigError. - POPs the ``_enabled_explicit`` marker: these branches are terminal (no later registry pass). + In multiplex mode a secondary profile pins ``platforms..enabled: false`` so + it shares the default profile's listener yet still inherits the process env; + without this guard env presence would force-enable the listener and trip + MultiplexConfigError. POPs the marker: these branches are terminal. """ platform_config = config.platforms.setdefault(platform, PlatformConfig()) explicit = platform_config.extra.pop("_enabled_explicit", False) @@ -211,11 +203,13 @@ class _Cred: """Credential-gated platform enable, applied as ``step(config)``. ``creds``: env names that must ALL be truthy; an inner tuple lists alternatives (ANY). - ``token``: env whose truthy value becomes ``PlatformConfig.token``. - ``fixed``: ``(extra_key, env[, default[, fn]])`` always written once enabled (``env`` may be a - tuple of alternatives). ``optional`` / ``optional_stripped``: ``_env_extras`` specs, written only - when truthy. ``warn_missing``: ``(env, msg)`` logged BEFORE enabling when *env* is blank. ``then``: unique tail ``fn(config, platform_config)``. - ``home``: ``_env_home_channel`` env base applied only when the gate passed. + ``token``: env whose truthy value becomes ``PlatformConfig.token`` (stored even when + yaml disables the adapter, so sending skills can use it). + ``fixed``: ``(extra_key, env[, default[, fn]])`` always written once enabled (``env`` may + be a tuple of alternatives). ``optional`` / ``optional_stripped``: ``_env_extras`` specs. + ``warn_missing``: ``(env, msg)`` logged BEFORE enabling when *env* is blank. + ``then``: unique tail ``fn(config, platform_config)``. ``home``: ``_env_home_channel`` env + base applied only when the gate passed. """ platform: Platform creds: tuple @@ -268,14 +262,12 @@ def _telegram_fallback_ips(config: GatewayConfig) -> None: def _whatsapp(config: GatewayConfig) -> None: - """WhatsApp (Baileys bridge) uses a flag, not credentials.""" + """WhatsApp (Baileys bridge) uses a flag, not credentials; an explicit false overrides YAML.""" raw = getenv("WHATSAPP_ENABLED") enabled = is_truthy_value(raw) - disabled_explicitly = raw.lower() in {"false", "0", "no"} - if Platform.WHATSAPP in config.platforms: - # YAML config exists — respect explicit disable; otherwise keep whatever the YAML set. - wa_cfg = config.platforms[Platform.WHATSAPP] - if disabled_explicitly: + wa_cfg = config.platforms.get(Platform.WHATSAPP) + if wa_cfg is not None: + if raw.lower() in {"false", "0", "no"}: wa_cfg.enabled = False elif enabled: wa_cfg.enabled = True @@ -283,34 +275,23 @@ def _whatsapp(config: GatewayConfig) -> None: config.platforms[Platform.WHATSAPP] = PlatformConfig(enabled=True) -def _slack(config: GatewayConfig) -> None: - slack_token = getenv("SLACK_BOT_TOKEN") - if slack_token: - if Platform.SLACK not in config.platforms: - config.platforms[Platform.SLACK] = PlatformConfig(enabled=True) - else: - slack_config = config.platforms[Platform.SLACK] - if not slack_config.enabled and not slack_config.extra.get("_enabled_explicit", False): - # Top-level Slack settings such as channel prompts must not turn an env-token setup - # into a disabled platform; only an explicit enabled: false should. - slack_config.enabled = True - elif not slack_config.enabled: - _warn_explicit_disable_beats_env(Platform.SLACK) - # Token is stored even when yaml disables the adapter so Slack-sending skills can use it. - config.platforms[Platform.SLACK].token = slack_token +def _slack_home(config: GatewayConfig) -> None: + """SLACK_HOME_CHANNEL creates a disabled Slack entry if needed and keeps + user_id/scope_id provenance when the chat_id is unchanged.""" slack_home = getenv("SLACK_HOME_CHANNEL") - if slack_home: - slack_config = config.platforms.setdefault(Platform.SLACK, PlatformConfig(enabled=False)) - existing_home = slack_config.home_channel - same_home = existing_home is not None and existing_home.chat_id == slack_home - slack_config.home_channel = HomeChannel( - platform=Platform.SLACK, - chat_id=slack_home, - name=getenv("SLACK_HOME_CHANNEL_NAME"), - thread_id=getenv("SLACK_HOME_CHANNEL_THREAD_ID") or None, - user_id=existing_home.user_id if same_home else None, - scope_id=existing_home.scope_id if same_home else None, - ) + if not slack_home: + return + slack_config = config.platforms.setdefault(Platform.SLACK, PlatformConfig(enabled=False)) + existing_home = slack_config.home_channel + same_home = existing_home is not None and existing_home.chat_id == slack_home + slack_config.home_channel = HomeChannel( + platform=Platform.SLACK, + chat_id=slack_home, + name=getenv("SLACK_HOME_CHANNEL_NAME"), + thread_id=getenv("SLACK_HOME_CHANNEL_THREAD_ID") or None, + user_id=existing_home.user_id if same_home else None, + scope_id=existing_home.scope_id if same_home else None, + ) def _matrix_e2ee(config: GatewayConfig, matrix_config: PlatformConfig) -> None: @@ -324,15 +305,13 @@ def _matrix_e2ee(config: GatewayConfig, matrix_config: PlatformConfig) -> None: _env_extras(matrix_config.extra, (("device_id", "MATRIX_DEVICE_ID"),)) -def _sms(config: GatewayConfig) -> None: - if getenv("TWILIO_ACCOUNT_SID"): - _enable_from_env(config, Platform.SMS).api_key = getenv("TWILIO_AUTH_TOKEN") +def _sms_api_key(config: GatewayConfig, sms_config: PlatformConfig) -> None: + sms_config.api_key = getenv("TWILIO_AUTH_TOKEN") def _api_server(config: GatewayConfig) -> None: - """Require a usable key: API_SERVER_ENABLED alone would load an unauthenticated platform whose - adapter refuses to start at connect() anyway, leaving the reconnect watcher spinning forever. - Same strength bar as the startup guard (has_usable_secret, min_length=16).""" + """Require a usable key: API_SERVER_ENABLED alone would load an unauthenticated + platform whose adapter refuses to start, leaving the reconnect watcher spinning.""" key = getenv("API_SERVER_KEY") if not _has_usable_api_server_key(key): return @@ -365,8 +344,8 @@ def _msgraph_webhook(config: GatewayConfig) -> None: ): return msgraph_cfg = config.platforms.setdefault(Platform.MSGRAPH_WEBHOOK, PlatformConfig()) - # Same explicit-disable guard as the webhook branch, but READ (don't pop) the marker: - # the relay-exclusive pass below still consults it; the end-of-function scrub removes it. + # Same explicit-disable guard as webhook, but READ (don't pop) the marker: the + # relay-exclusive pass still consults it; the end scrub removes it. if enabled and (not msgraph_cfg.extra.get("_enabled_explicit", False) or msgraph_cfg.enabled): msgraph_cfg.enabled = True _env_extras(msgraph_cfg.extra, (("port", "MSGRAPH_WEBHOOK_PORT", _INT),)) @@ -383,9 +362,8 @@ def _qq_home(config: GatewayConfig, qq_config: PlatformConfig) -> None: name_env = "QQBOT_HOME_CHANNEL_NAME" if not qq_home: # Back-compat: accept the pre-rename name and log a one-time warning. - legacy_home = getenv("QQ_HOME_CHANNEL").strip() - if legacy_home: - qq_home = legacy_home + qq_home = getenv("QQ_HOME_CHANNEL").strip() + if qq_home: name_env = "QQ_HOME_CHANNEL_NAME" logger.warning( "QQ_HOME_CHANNEL is deprecated; rename to QQBOT_HOME_CHANNEL " @@ -411,12 +389,12 @@ def _session_settings(config: GatewayConfig) -> None: def _enable_plugin_platforms_from_env(config: GatewayConfig) -> None: """Registry-driven enable for plugin platforms (built-ins have explicit rows in ``_ENV_STEPS``). - A plugin platform is enabled when its credentials are configured (``is_connected``) and its deps - are present (passive ``check_fn``) or installable on demand (``ensure_deps_fn`` — run later by - ``create_adapter()``, never here: the active installer used to be wired as ``check_fn`` and this - sweep pip-installed SDKs on every ``load_gateway_config()`` call, boot-looping the desktop app). - ``is_connected`` MUST gate enablement: ``check_fn`` alone ("is the SDK importable?") would enable - platforms the user never configured and the gateway would retry-connect forever with no token. + A plugin platform is enabled when its credentials are configured (``is_connected``) + and its deps are present (passive ``check_fn``) or installable on demand + (``ensure_deps_fn`` — run later by ``create_adapter()``, never here: an active + installer in this sweep pip-installed SDKs on every load and boot-looped the + desktop app). ``is_connected`` MUST gate enablement: ``check_fn`` alone would + enable unconfigured platforms that then retry-connect forever with no token. """ try: from hermes_cli.plugins import discover_plugins @@ -429,43 +407,31 @@ def _enable_plugin_platforms_from_env(config: GatewayConfig) -> None: logger.debug("unknown platform name %r: %s", entry.name, e) continue existing_cfg = config.platforms.get(platform) + already_enabled = existing_cfg is not None and existing_cfg.enabled + existing_extra = dict(existing_cfg.extra or {}) if existing_cfg is not None else {} # Never re-enable a platform the user explicitly disabled (marker set by the YAML loader). - if ( - existing_cfg is not None - and not existing_cfg.enabled - and bool((existing_cfg.extra or {}).get("_enabled_explicit", False)) - ): + if existing_cfg is not None and not already_enabled and existing_extra.get("_enabled_explicit", False): continue - # Seed candidate extras so plugins whose ``is_connected`` reads ``config.extra`` (Google Chat) - # see the same state they will after enablement. + # Seed candidate extras so plugins whose ``is_connected`` reads ``config.extra`` + # (Google Chat) see the same state they will after enablement. seed_for_probe = None if entry.env_enablement_fn is not None: try: seed_for_probe = entry.env_enablement_fn() except Exception as e: logger.debug("env_enablement_fn for %s raised: %s", entry.name, e) - seed_for_probe = None has_seed = isinstance(seed_for_probe, dict) and bool(seed_for_probe) - # Only consult is_connected for platforms not already enabled by YAML/env (keep that decision). - if (existing_cfg is None or not existing_cfg.enabled) and entry.is_connected is not None: + # Only consult is_connected for platforms not already enabled by YAML/env. + if not already_enabled and entry.is_connected is not None: try: - # Probe with ``enabled=True``: we ask "would this be configured if enabled?" — - # some ``is_connected`` short-circuit on ``config.enabled`` being False. - probe_cfg = existing_cfg - if probe_cfg is None or not probe_cfg.enabled: - probe_cfg = PlatformConfig( - enabled=True, - extra=dict(existing_cfg.extra or {}) if existing_cfg is not None else {}, - ) + # Probe a transient ``enabled=True`` view (some ``is_connected`` + # short-circuit on ``config.enabled``); never mutate ``existing_cfg``. if has_seed: - # Transient view; never mutate ``existing_cfg`` for the probe. - probe_extra = dict(probe_cfg.extra or {}) for k, v in seed_for_probe.items(): if k != "home_channel": - probe_extra.setdefault(k, v) - probe_cfg = PlatformConfig(enabled=True, extra=probe_extra) - configured = bool(entry.is_connected(probe_cfg)) + existing_extra.setdefault(k, v) + configured = bool(entry.is_connected(PlatformConfig(enabled=True, extra=existing_extra))) except Exception as exc: logger.debug("is_connected for %s raised: %s — skipping enablement", entry.name, exc) configured = False @@ -486,8 +452,8 @@ def _enable_plugin_platforms_from_env(config: GatewayConfig) -> None: continue platform_config = config.platforms.setdefault(platform, PlatformConfig()) platform_config.enabled = True - # Commit the env-seeded extras (reuse the probe result; don't call env_enablement_fn twice). if has_seed: + # Commit the env-seeded extras (reuse the probe result; don't call env_enablement_fn twice). seed = dict(seed_for_probe) home = seed.pop("home_channel", None) platform_config.extra.update(seed) @@ -503,19 +469,18 @@ def _enable_plugin_platforms_from_env(config: GatewayConfig) -> None: def _relay(config: GatewayConfig) -> None: - """Relay (generic connector-fronted platform, EXPERIMENTAL). Enabled by GATEWAY_RELAY_URL (env) - or gateway.relay_url (config.yaml). The adapter dials OUT to the connector (no inbound port), - so it only needs Platform.RELAY present+enabled for start_gateway()'s connect loop. The - connected-checker keys on extra["relay_url"], so the URL is mirrored into extra here. + """Relay (connector-fronted platform, EXPERIMENTAL): enabled by GATEWAY_RELAY_URL or + gateway.relay_url. The adapter dials OUT (no inbound port); the connected-checker + keys on extra["relay_url"], so the URL is mirrored into extra. - Relay-exclusive: a GATEWAY_RELAY_URL env stamp marks a connector-fronted deployment where the - connector owns every platform connection; a directly-connected adapter in the same process - would be a second unmanaged ingress (duplicate deliveries, split sessions, a live socket that - disarms scale-to-zero). So the env stamp disables all other messaging platforms — even ones - explicitly enabled in config.yaml. Non-messaging surfaces (local, api_server, webhook — same - exclusion set as the scale-to-zero arm gate) are untouched; relay via gateway.relay_url only - keeps the old additive behavior. Opt-out GATEWAY_RELAY_ALLOW_DIRECT_PLATFORMS=true is likewise - a deploy-stamp env var read through the profile-scope-aware getenv. + Relay-exclusive: the GATEWAY_RELAY_URL env stamp marks a deployment where the + connector owns every platform connection; a directly-connected adapter in the same + process would be a second unmanaged ingress (duplicate deliveries, split sessions, + a socket that disarms scale-to-zero). So the env stamp disables all other messaging + platforms — even ones explicitly enabled in config.yaml. Non-messaging surfaces + (local, api_server, webhook — same set as the scale-to-zero arm gate) are untouched; + relay via gateway.relay_url only keeps the old additive behavior. Opt out with + GATEWAY_RELAY_ALLOW_DIRECT_PLATFORMS=true (also read through the scoped getenv). """ relay_url_env = getenv("GATEWAY_RELAY_URL").strip() existing_relay = config.platforms.get(Platform.RELAY) @@ -554,10 +519,10 @@ def _scrub_explicit_markers(config: GatewayConfig) -> None: platform_config.extra.pop("_enabled_explicit", None) -# Application order is significant: e.g. Telegram's reply mode may create the (disabled) platform -# entry BEFORE its home channel is read, while Discord reads home first; a home channel is only -# attached to a platform that already exists. Relay-exclusive disabling runs after the plugin -# pass and before the marker scrub, which must be last. +# Application order is significant: e.g. Telegram's reply mode may create the (disabled) +# platform entry BEFORE its home channel is read, while Discord reads home first; a home +# channel is only attached to a platform that already exists. Relay-exclusive disabling +# runs after the plugin pass and before the marker scrub, which must be last. _ENV_STEPS: tuple = ( _Cred(Platform.TELEGRAM, ("TELEGRAM_BOT_TOKEN",), token="TELEGRAM_BOT_TOKEN"), _ReplyMode(Platform.TELEGRAM, "TELEGRAM_REPLY_TO_MODE"), @@ -568,7 +533,7 @@ _ENV_STEPS: tuple = ( _ReplyMode(Platform.DISCORD, "DISCORD_REPLY_TO_MODE"), _whatsapp, _Home(Platform.WHATSAPP, "WHATSAPP_HOME_CHANNEL"), - # WhatsApp Cloud API (official Business Platform via Meta). Distinct from the Baileys bridge; + # WhatsApp Cloud API (Meta Business Platform). Distinct from the Baileys bridge; # both adapters can run in parallel against different phone numbers. _Cred( Platform.WHATSAPP_CLOUD, ("WHATSAPP_CLOUD_PHONE_NUMBER_ID", "WHATSAPP_CLOUD_ACCESS_TOKEN"), @@ -585,7 +550,8 @@ _ENV_STEPS: tuple = ( ), ), _Home(Platform.WHATSAPP_CLOUD, "WHATSAPP_CLOUD_HOME_CHANNEL"), - _slack, + _Cred(Platform.SLACK, ("SLACK_BOT_TOKEN",), token="SLACK_BOT_TOKEN"), + _slack_home, _Cred( Platform.SIGNAL, ("SIGNAL_HTTP_URL", "SIGNAL_ACCOUNT"), fixed=( @@ -615,7 +581,7 @@ _ENV_STEPS: tuple = ( fixed=(("address", "EMAIL_ADDRESS"), ("imap_host", "EMAIL_IMAP_HOST"), ("smtp_host", "EMAIL_SMTP_HOST")), ), _Home(Platform.EMAIL, "EMAIL_HOME_ADDRESS"), - _sms, + _Cred(Platform.SMS, ("TWILIO_ACCOUNT_SID",), then=_sms_api_key), _Home(Platform.SMS, "SMS_HOME_CHANNEL"), _api_server, _webhook, From 98dccb4ca860f25c97b01fec66b7461e3ba0c7a6 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:23:35 -0700 Subject: [PATCH 03/59] refactor(gateway): unify /proc + coercer helpers, merge timestamp regexes, collapse defensive layers in lifecycle/monitoring/media modules --- gateway/agent_cache_pressure.py | 109 ++++++---------- gateway/cgroup_cleanup.py | 8 +- gateway/code_skew.py | 15 +-- gateway/cwd_placeholder.py | 9 +- gateway/disk_status.py | 56 +++----- gateway/drain_control.py | 113 ++++++---------- gateway/hooks.py | 111 ++++++++-------- gateway/lifecycle_ledger.py | 221 ++++++++++++-------------------- gateway/media_policy.py | 15 +-- gateway/media_repair.py | 77 +++++------ gateway/memory_monitor.py | 36 ++---- gateway/memory_status.py | 62 +++------ gateway/message_timestamps.py | 51 ++++---- gateway/mirror.py | 37 +++--- gateway/readiness.py | 19 ++- gateway/response_filters.py | 32 ++--- gateway/restart.py | 106 +++++++-------- gateway/restart_loop_guard.py | 52 ++++---- 18 files changed, 447 insertions(+), 682 deletions(-) diff --git a/gateway/agent_cache_pressure.py b/gateway/agent_cache_pressure.py index 1fc62dbadb..d7d862c24f 100644 --- a/gateway/agent_cache_pressure.py +++ b/gateway/agent_cache_pressure.py @@ -1,15 +1,12 @@ """Memory-pressure bounds for the gateway's per-session AIAgent cache. Each cached ``AIAgent`` pins ``_session_messages`` (the full live transcript, -tool outputs included — tens of MB on a tool-heavy session). The cache's LRU -cap counts entries, not bytes, and the idle TTL defers eviction for busy -sessions, so neither sees actual memory use. This module supplies that signal: -the process's own anonymous RSS against a budget derived from the cgroup limit. -``GatewayRunner._sweep_agent_cache_under_pressure`` uses it to shed LRU -transcripts via the soft-eviction path (rebuilt from the persisted session on -the next turn). - -Everything here is pure or read-only. Config lives under ``agent.agent_cache``. +tens of MB on a tool-heavy session). The cache's LRU cap counts entries, not +bytes, and the idle TTL defers eviction for busy sessions, so neither sees +actual memory use. This module supplies that signal: own anonymous RSS against +a budget derived from the cgroup limit; ``GatewayRunner`` sheds LRU transcripts +via soft eviction (rebuilt from the persisted session on the next turn). +Everything here is pure or read-only. Config lives under ``agent.agent_cache``. """ from __future__ import annotations @@ -33,6 +30,7 @@ _DEFAULT_MAX_EVICTIONS_PER_PASS = 16 _DEFAULT_PROTECT_RECENT = 8 _BYTES_PER_MB = 1024 * 1024 +_OFF_WORDS = frozenset({"", "off", "none", "false", "disabled"}) @dataclass(frozen=True) @@ -51,7 +49,11 @@ class AgentCacheBounds: protect_recent: int = _DEFAULT_PROTECT_RECENT -def _positive(value: Any, cast: Callable[[Any], Any]) -> Any: +def _is_int(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _positive(value: Any, cast: Callable[[Any], Any] = int) -> Any: """``cast(value)`` if it is a positive number (bools rejected), else None.""" if isinstance(value, bool) or value is None: return None @@ -62,21 +64,13 @@ def _positive(value: Any, cast: Callable[[Any], Any]) -> Any: return parsed if parsed > 0 else None -def _positive_int(value: Any) -> Optional[int]: - return _positive(value, int) - - -def _positive_float(value: Any) -> Optional[float]: - return _positive(value, float) - - def _cgroup_limit_bytes() -> Optional[int]: - """Return the memory limit this process runs under, if cgroup-capped. + """Memory limit this process runs under, if cgroup-capped. Prefers cgroup v2 ``memory.high`` (the throttling point) over ``memory.max``, - then cgroup v1. Checks the process's *own* cgroup first (where a systemd + then cgroup v1. Checks the process's *own* cgroup first (where a systemd unit's ``MemoryHigh=``/``MemoryMax=`` lands — the root files read ``max`` - there), then the root for container-style limits. ``max`` and the v1 + there), then the root for container-style limits. ``max`` and the v1 near-2^63 sentinel mean unlimited. """ if sys.platform != "linux": @@ -97,18 +91,11 @@ def _cgroup_limit_bytes() -> Optional[int]: ] for candidate in candidates: try: - raw = Path(candidate).read_text(encoding="utf-8").strip() - except OSError: + limit = int(Path(candidate).read_text(encoding="utf-8").strip()) + except (OSError, ValueError): # unreadable, empty, or "max" continue - if not raw or raw == "max": - continue - try: - limit = int(raw) - except ValueError: - continue - if limit <= 0 or limit >= (1 << 62): - continue - return limit + if 0 < limit < (1 << 62): + return limit return None @@ -134,16 +121,12 @@ def resolve_memory_high_mb(setting: Any) -> Optional[int]: if isinstance(setting, str): normalized = setting.strip().lower() if normalized != "auto": - return ( - None - if normalized in ("", "off", "none", "false", "disabled") - else _positive_int(normalized) - ) + return None if normalized in _OFF_WORDS else _positive(normalized) elif isinstance(setting, bool): if not setting: return None else: - return _positive_int(setting) + return _positive(setting) limit = _cgroup_limit_bytes() or _total_memory_bytes() if not limit: @@ -158,46 +141,32 @@ def resolve_agent_cache_bounds(config: Any) -> AgentCacheBounds: The gateway's loader does not deep-merge ``DEFAULT_CONFIG``, so an absent key stays absent and callers can tell "operator chose 128" from "unset". """ - section: Any = None - if isinstance(config, dict): - agent_cfg = config.get("agent") - if isinstance(agent_cfg, dict): - section = agent_cfg.get("agent_cache") + section = (config.get("agent") or {}).get("agent_cache") if isinstance(config, dict) else None if not isinstance(section, dict): section = {} - max_evictions = _positive_int(section.get("max_evictions_per_pass")) protect_recent = section.get("protect_recent") - protect_parsed = _positive_int(protect_recent) - # 0 means "shed anything" — distinct from unset. The bool guard keeps + protect_parsed = _positive(protect_recent) + # 0 means "shed anything" — distinct from unset. The bool guard keeps # `protect_recent: false` (False == 0) on the default instead of silently # disabling MRU protection. - if ( - protect_parsed is None - and isinstance(protect_recent, int) - and not isinstance(protect_recent, bool) - and protect_recent == 0 - ): + if protect_parsed is None and _is_int(protect_recent) and protect_recent == 0: protect_parsed = 0 return AgentCacheBounds( - max_size=_positive_int(section.get("max_size")), - idle_ttl_secs=_positive_float(section.get("idle_ttl_secs")), + max_size=_positive(section.get("max_size")), + idle_ttl_secs=_positive(section.get("idle_ttl_secs"), float), memory_high_mb=resolve_memory_high_mb(section.get("memory_high_mb", "auto")), - max_evictions_per_pass=( - max_evictions if max_evictions is not None else _DEFAULT_MAX_EVICTIONS_PER_PASS - ), - protect_recent=( - protect_parsed if protect_parsed is not None else _DEFAULT_PROTECT_RECENT - ), + max_evictions_per_pass=_positive(section.get("max_evictions_per_pass")) or _DEFAULT_MAX_EVICTIONS_PER_PASS, + protect_recent=_DEFAULT_PROTECT_RECENT if protect_parsed is None else protect_parsed, ) def read_anon_rss_mb() -> Optional[int]: - """Return the process's anonymous resident memory in MB, or None. + """Process anonymous resident memory in MB, or None. Anonymous pages are where cached transcripts live; file-backed pages are - noise. ``collect_memory_snapshot`` reads ``/proc/self/status`` without a + noise. ``collect_memory_snapshot`` reads ``/proc/self/status`` without a dependency; psutil covers other platforms (total RSS only). """ try: @@ -224,17 +193,13 @@ def transcript_persistence_caught_up(agent: Any) -> bool: Soft eviction drops ``_session_messages`` and rebuilds from the persisted session, so it is only safe once ``_last_flushed_db_idx`` (advanced to - ``len(messages)`` by ``AIAgent._flush_messages_to_session_db`` only on a - fully successful write) has caught up. Unknown shapes are *not* caught up: - a skipped eviction costs memory, a wrong one costs the conversation. + ``len(messages)`` only on a fully successful write) has caught up. Unknown + shapes are *not* caught up: a skipped eviction costs memory, a wrong one + costs the conversation. """ messages = getattr(agent, "_session_messages", None) - if not isinstance(messages, list): - return False flushed = getattr(agent, "_last_flushed_db_idx", None) - if not isinstance(flushed, int) or isinstance(flushed, bool): - return False - return flushed >= len(messages) + return isinstance(messages, list) and _is_int(flushed) and flushed >= len(messages) def plan_pressure_evictions( @@ -247,8 +212,8 @@ def plan_pressure_evictions( """Choose which cached sessions to shed, least-recently-used first. ``ordered_entries`` must be LRU→MRU (the cache OrderedDict is kept that way - by ``move_to_end`` on every hit). The batch is capped so one pass cannot - stall the gateway. ``protect_recent`` is clamped to half the cache: a few + by ``move_to_end`` on every hit). The batch is capped so one pass cannot + stall the gateway. ``protect_recent`` is clamped to half the cache: a few huge transcripts can exhaust the budget alone, and a fixed guard would then protect the whole cache with nothing left to shed. """ diff --git a/gateway/cgroup_cleanup.py b/gateway/cgroup_cleanup.py index a68a971c37..3e425696ca 100644 --- a/gateway/cgroup_cleanup.py +++ b/gateway/cgroup_cleanup.py @@ -3,11 +3,9 @@ Runs as ``ExecStopPost=`` after the gateway's main process has exited: the safety net for long-lived helpers the gateway doesn't track (``adb``, platform bridges) that would otherwise be orphaned in the cgroup and block -``Restart=always``. - -Per-PID SIGKILLs over ``cgroup.procs`` are used deliberately instead of writing -``1`` to ``cgroup.kill``: the kernel has returned ``EINVAL`` on the cgroup-wide -kill while per-PID signal delivery still works. +``Restart=always``. Per-PID SIGKILLs over ``cgroup.procs`` are used instead of +writing ``1`` to ``cgroup.kill``: the kernel has returned ``EINVAL`` on the +cgroup-wide kill while per-PID signal delivery still works. """ from __future__ import annotations diff --git a/gateway/code_skew.py b/gateway/code_skew.py index 9b77237c3a..3245aa158c 100644 --- a/gateway/code_skew.py +++ b/gateway/code_skew.py @@ -1,15 +1,11 @@ """Detect when the gateway is running stale code after a hot ``git pull``. The gateway's ``sys.modules`` is frozen at boot. If the checkout is updated -underneath it (manual ``git pull``, or the window before ``hermes update``'s -graceful restart), a first-time lazy import can resolve a freshly-pulled module -against a stale cached dependency -> ImportError -(``tests/test_stale_utils_module_import.py``). We snapshot the revision at +underneath it, a first-time lazy import can resolve a freshly-pulled module +against a stale cached dependency -> ImportError. We snapshot the revision at startup so risky callers (e.g. ``/model`` switching) can refuse with a clear -"restart the gateway" message instead. - -If the revision can't be read (non-git install, IO error) the boot snapshot -stays ``None`` and detection no-ops — never a false positive. +"restart the gateway" message. If the revision can't be read (non-git install, +IO error) the boot snapshot stays ``None`` and detection no-ops — never a false positive. """ from __future__ import annotations @@ -47,8 +43,7 @@ def _short(fingerprint: str) -> str: def detect_code_skew() -> tuple[str, str] | None: - """Return ``(boot_rev, disk_rev)`` short labels if the checkout drifted - since boot, else ``None``.""" + """``(boot_rev, disk_rev)`` short labels if the checkout drifted since boot, else ``None``.""" if _boot_fingerprint is None: return None current = _fingerprint() diff --git a/gateway/cwd_placeholder.py b/gateway/cwd_placeholder.py index b064104c6a..3d9d9dd8db 100644 --- a/gateway/cwd_placeholder.py +++ b/gateway/cwd_placeholder.py @@ -22,12 +22,9 @@ def resolve_placeholder_terminal_cwd( ) -> str | None: """Return the ``TERMINAL_CWD`` value to set, or ``None`` to leave it unset. - Cases: - - **local** + placeholder → ``MESSAGING_CWD`` or ``home_fallback`` - - **docker** + placeholder + mount on + host ``MESSAGING_CWD`` → host path - (for ``terminal_tool`` ``/workspace`` mapping) - - **docker** + placeholder + mount off → ``None`` (sandbox default) - - other non-local backends + placeholder → ``None`` + local + placeholder → ``MESSAGING_CWD`` or ``home_fallback``; docker + + placeholder + mount on + host ``MESSAGING_CWD`` → that host path (for the + ``/workspace`` mapping); any other non-local backend → ``None`` (sandbox default). """ if configured_cwd and configured_cwd not in CWD_PLACEHOLDERS: return configured_cwd diff --git a/gateway/disk_status.py b/gateway/disk_status.py index 22d0102b57..f76280a84d 100644 --- a/gateway/disk_status.py +++ b/gateway/disk_status.py @@ -1,20 +1,10 @@ """Disk-usage rollup for ``/api/status``. Companion to :mod:`gateway.memory_status`: a hosted agent can fill its data -volume (SQLite writes failing, sessions and config saves lost) while its -dashboard and the NAS agent card look healthy. ``gateway/readiness.py`` -already probes disk, but readiness is a component verdict, not user-facing -telemetry — this module produces the public block the dashboard SPA and the -NAS availability sweep consume. - -Disk is sampled live via :func:`shutil.disk_usage` (one ``statvfs`` call), so -there is no staleness dimension and no ``sampled_at``. - -Public-safety: ``/api/status`` is unauthenticated (``PUBLIC_API_PATHS``); this -block carries only coarse numbers (MB, one-decimal percent) and an enum — the same -disclosure class as the ``memory`` block. - -Best-effort and read-only: an unreadable filesystem degrades to +volume (SQLite writes failing, config saves lost) while its dashboard looks +healthy. Sampled live via one ``statvfs`` call, so there is no ``sampled_at``. +``/api/status`` is unauthenticated: only coarse numbers (MB, one-decimal +percent) and an enum. Best-effort: an unreadable filesystem degrades to ``pressure="unknown"`` rather than raising into the status endpoint. """ @@ -25,10 +15,12 @@ import shutil from pathlib import Path from typing import Any, Dict, Optional +from gateway.memory_status import _nonneg_int + logger = logging.getLogger(__name__) # Percent alone misleads both ways: 90% used on 100 GB leaves 10 GB, while 50% -# on a tiny volume is one download from write failures. So percent triggers are +# on a tiny volume is one download from write failures. Percent triggers are # gated on absolute headroom also being low, and a hard absolute floor applies # regardless of size (below it SQLite journaling / config writes are at risk). _CRITICAL_FREE_MB = 256 # < 256 MB free: critical on any volume @@ -41,21 +33,15 @@ _ELEVATED_HEADROOM_MB = 4096 _BYTES_PER_MB = 1024 * 1024 -def _coerce_mb(value: Any) -> Optional[int]: - if isinstance(value, bool) or not isinstance(value, int) or value < 0: - return None - return value - - def classify_disk_pressure(free_mb: Any, total_mb: Any) -> str: """Map free/total MB to ``ok``/``elevated``/``critical``. ``unknown`` when the sample is missing or malformed — the caller must not treat "we could not read it" as "disk is fine". """ - free = _coerce_mb(free_mb) - total = _coerce_mb(total_mb) - if free is None or total is None or total <= 0: + free = _nonneg_int(free_mb) + total = _nonneg_int(total_mb) + if free is None or not total: return "unknown" used_percent = (1 - free / total) * 100.0 for level, free_floor, percent_floor, headroom in ( @@ -71,16 +57,10 @@ def collect_disk_status(home: Optional[Path] = None) -> Dict[str, Any]: """Build the ``disk`` block for ``/api/status``. ``home`` scopes the sample to a profile's HERMES_HOME (same contract as the - ``memory`` block; on hosted images every profile shares one volume). - Always returns a dict and never raises — an unreadable/unmounted filesystem - yields ``{"pressure": "unknown", ...}``. + ``memory`` block). Always returns a dict and never raises — an unreadable + or unmounted filesystem yields ``{"pressure": "unknown", ...}``. """ - status: Dict[str, Any] = { - "pressure": "unknown", - "total_mb": None, - "free_mb": None, - "used_percent": None, - } + status: Dict[str, Any] = {"pressure": "unknown", "total_mb": None, "free_mb": None, "used_percent": None} try: if home is None: from hermes_constants import get_hermes_home @@ -93,8 +73,10 @@ def collect_disk_status(home: Optional[Path] = None) -> Dict[str, Any]: return status total_mb = usage.total // _BYTES_PER_MB free_mb = usage.free // _BYTES_PER_MB - status["total_mb"] = total_mb - status["free_mb"] = free_mb - status["used_percent"] = round((usage.used / usage.total) * 100, 1) - status["pressure"] = classify_disk_pressure(free_mb, total_mb) + status.update( + total_mb=total_mb, + free_mb=free_mb, + used_percent=round((usage.used / usage.total) * 100, 1), + pressure=classify_disk_pressure(free_mb, total_mb), + ) return status diff --git a/gateway/drain_control.py b/gateway/drain_control.py index e1e323368a..9816f7aa90 100644 --- a/gateway/drain_control.py +++ b/gateway/drain_control.py @@ -1,38 +1,27 @@ """External drain-control marker contract (dashboard → gateway). -There is no control channel into a running gateway, so the dashboard's -begin/cancel-drain endpoint talks to it the same way ``.restart_notify.json`` -does: it writes (or removes) a marker file and a gateway watcher reacts. This -module owns the contract so writer and reader can never disagree. - -Contract (presence-based): - * begin-drain → write ``{HERMES_HOME}/.drain_request.json`` with - ``{"action": "drain", "requested_at": , "principal": , - "epoch": , "suppress_notification": }``. - * cancel-drain → remove the marker. - * The watcher treats presence of an ACTIVE marker as "external drain": flip - ``gateway_state -> "draining"`` and stop accepting new turns. Absence or a - stale marker means "not draining" (revert to ``running``). +There is no control channel into a running gateway, so begin/cancel-drain +writes (or removes) ``{HERMES_HOME}/.drain_request.json`` and a gateway watcher +reacts; this module owns the contract so writer and reader never disagree. +Presence of an ACTIVE marker means "external drain" (``gateway_state -> +"draining"``); absence or a stale marker means "not draining". Staleness — two independent, individually-lenient signals (either suffices): * epoch mismatch: HERMES_HOME is a durable volume on Hermes Cloud, so a - marker survives the machine restart that the drain-gated action (update, - migrate, env edit) ends in; honouring it would park the fresh gateway in - ``draining`` forever. A marker is stamped with this instantiation's epoch - and ignored only on a *definite* mismatch. - * expiry: a same-epoch orphan (action completed without a restart, writer - never cancelled) is ignored once ``requested_at`` is older than - :data:`DRAIN_REQUEST_MAX_AGE_SECONDS`. Re-calling :func:`write_drain_request` - refreshes the timestamp — the sanctioned keep-alive for a long drain. + marker survives the machine restart that a drain-gated action ends in; + honouring it would park the fresh gateway in ``draining`` forever. + * expiry: a same-epoch orphan (writer never cancelled) is ignored once + ``requested_at`` is older than :data:`DRAIN_REQUEST_MAX_AGE_SECONDS`; + re-calling :func:`write_drain_request` refreshes it (long-drain keep-alive). -Reading never raises: a malformed/half-written file reads as "present but -contentless" (``{}``), which still counts as drain-active — fail-safe toward -quiescing. Both staleness checks only reject on a *definite* verdict; a marker -with no epoch/timestamp, or a host without ``/proc``, degrades to the -presence-only behaviour and never fails closed. +Reading never raises: a malformed/half-written file reads as ``{}``, which still +counts as drain-active (fail-safe toward quiescing). Both staleness checks +reject only on a *definite* verdict — a marker with no epoch/timestamp, or a +host without ``/proc``, degrades to presence-only and never fails closed. """ from __future__ import annotations +import contextlib import functools import json import logging @@ -42,7 +31,6 @@ from typing import Any, Optional from hermes_constants import get_hermes_home from utils import atomic_json_write -import contextlib _log = logging.getLogger(__name__) @@ -52,9 +40,8 @@ _DRAIN_REQUEST_FILENAME = ".drain_request.json" # leaked marker can cause. Long drains refresh the marker instead of raising this. DRAIN_REQUEST_MAX_AGE_SECONDS = 3600.0 -# Dedup for the expired-marker warning: the watcher re-reads every second and an -# expired orphan sits on disk until removed. Keyed by ``requested_at`` so a -# keep-alive re-write that later expires again logs again. +# Dedup for the expired-marker warning (the watcher re-reads every second). +# Keyed by ``requested_at`` so a keep-alive re-write that later expires logs again. _expiry_logged_for: Optional[str] = None @@ -62,27 +49,22 @@ _expiry_logged_for: Optional[str] = None def current_instantiation_epoch() -> str: """Identity of THIS container / VM instantiation (":"). - Stable for the life of PID 1 (so an s6 respawn of just the gateway, or a - host ``hermes gateway restart`` under systemd/launchd, keeps honouring an - in-flight drain) but changes whenever the machine is recreated: boot_id - changes on a VM/microVM reboot, PID 1's start time on a plain - ``docker restart`` (same host kernel, brand-new ``/init``). - - Returns ``""`` when neither source is readable (non-Linux, no ``/proc``), - which disables the epoch check downstream — never fail-closed. + Stable for the life of PID 1 (an s6 respawn of just the gateway, or a host + ``hermes gateway restart``, keeps honouring an in-flight drain) but changes + whenever the machine is recreated: boot_id on a VM reboot, PID 1's start + time on a plain ``docker restart``. ``""`` when neither source is readable + (non-Linux, no ``/proc``), which disables the epoch check — never fail-closed. """ boot_id = "" with contextlib.suppress(OSError): boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="utf-8").strip() pid1_start = "" - try: + with contextlib.suppress(OSError, IndexError): # " () ...": comm may contain spaces/parens, so split # on the LAST ')'. starttime is field 22 (1-indexed) = tail index 19. stat = Path("/proc/1/stat").read_text(encoding="utf-8") pid1_start = stat.rsplit(")", 1)[1].split()[19] - except (OSError, IndexError): - pass if not boot_id and not pid1_start: return "" @@ -91,8 +73,7 @@ def current_instantiation_epoch() -> str: def drain_request_path(home: Optional[Path] = None) -> Path: """Absolute path to the drain-request marker, respecting HERMES_HOME.""" - base = home if home is not None else get_hermes_home() - return Path(base) / _DRAIN_REQUEST_FILENAME + return Path(home if home is not None else get_hermes_home()) / _DRAIN_REQUEST_FILENAME def write_drain_request( @@ -104,13 +85,11 @@ def write_drain_request( """Write the begin-drain marker atomically. Returns the payload written. Idempotent: re-writing refreshes ``requested_at`` (keep-alive past the - max-age). ``suppress_notification`` asks the shutdown that ends this drain - to skip ONLY the home-channel "gateway shutting down" broadcast (the - per-active-session interrupt ping is never suppressed); the policy of - which drains are quiet lives entirely in the caller. Defaults False so - legacy/operator drains behave exactly as before. The marker is stamped - with :func:`current_instantiation_epoch` so a copy surviving a machine - restart on the durable volume can be recognised as stale. + max-age). ``suppress_notification`` asks the shutdown ending this drain to + skip ONLY the home-channel "gateway shutting down" broadcast (the per-session + interrupt ping is never suppressed); which drains are quiet is the caller's + policy. Stamped with :func:`current_instantiation_epoch` so a copy surviving + a machine restart on the durable volume is recognised as stale. """ payload = { "action": "drain", @@ -136,20 +115,12 @@ def clear_drain_request(*, home: Optional[Path] = None) -> bool: return False -def _marker_epoch_is_stale(body: dict[str, Any]) -> bool: - """True iff ``body``'s epoch is a *definite* mismatch (both epochs known and differ).""" - current = current_instantiation_epoch() - marker_epoch = body.get("epoch") - return bool(current and marker_epoch and marker_epoch != current) - - def _marker_is_expired(body: dict[str, Any]) -> bool: """True iff ``requested_at`` parses AND is older than the max-age. - Missing/unparseable timestamps and future-dated ones (clock skew) are - honoured. The expiry is logged once per marker, not per poll: this path - only fires when a writer leaked a marker, and the warning is the - operator's breadcrumb. + Missing/unparseable and future-dated (clock skew) timestamps are honoured. + Logged once per marker, not per poll — the warning is the operator's + breadcrumb for a writer that leaked a marker. """ global _expiry_logged_for raw = body.get("requested_at") @@ -179,15 +150,14 @@ def _marker_is_expired(body: dict[str, Any]) -> bool: return True -def _marker_is_stale(body: dict[str, Any]) -> bool: - """True iff the marker is definitely from a drain that is already over.""" - return _marker_epoch_is_stale(body) or _marker_is_expired(body) - - def _active_drain_body(home: Optional[Path]) -> Optional[dict[str, Any]]: - """Marker body if present AND not stale, else None.""" + """Marker body if present AND not stale (definite epoch mismatch or expired), else None.""" body = read_drain_request(home=home) - if body is None or _marker_is_stale(body): + if body is None: + return None + current = current_instantiation_epoch() + marker_epoch = body.get("epoch") + if (current and marker_epoch and marker_epoch != current) or _marker_is_expired(body): return None return body @@ -200,10 +170,9 @@ def drain_requested(*, home: Optional[Path] = None) -> bool: def drain_notification_suppressed(*, home: Optional[Path] = None) -> bool: """True iff an ACTIVE drain marker explicitly asks to suppress the shutdown broadcast. - Uses the same activeness rule as :func:`drain_requested`, so an orphaned - marker can never silence a fresh gateway's legitimate broadcast. A legacy - marker without the field or a contentless ``{}`` body reads as False - (fail toward the louder behaviour). + Same activeness rule as :func:`drain_requested`, so an orphaned marker can + never silence a fresh gateway's broadcast. A legacy marker without the + field or a contentless ``{}`` reads as False (fail toward the louder behaviour). """ body = _active_drain_body(home) return bool(body and body.get("suppress_notification")) diff --git a/gateway/hooks.py b/gateway/hooks.py index dcf29962a5..b4f672d2e5 100644 --- a/gateway/hooks.py +++ b/gateway/hooks.py @@ -25,6 +25,7 @@ Handlers posting a follow-up into the same Telegram forum-topic should pass import asyncio import importlib.util import sys +from pathlib import Path from typing import Any, Callable, Dict, List, Optional import yaml @@ -35,6 +36,49 @@ from hermes_cli.config import get_hermes_home HOOKS_DIR = get_hermes_home() / "hooks" +def _load_hook_dir(hook_dir: Path) -> Optional[tuple]: + """``(name, events, handle_fn, description)`` for a valid hook dir, else None (reason printed).""" + manifest_path = hook_dir / "HOOK.yaml" + handler_path = hook_dir / "handler.py" + if not manifest_path.exists() or not handler_path.exists(): + return None + + manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) + if not manifest or not isinstance(manifest, dict): + print(f"[hooks] Skipping {hook_dir.name}: invalid HOOK.yaml", flush=True) + return None + + hook_name = manifest.get("name", hook_dir.name) + events = manifest.get("events", []) + if not events: + print(f"[hooks] Skipping {hook_name}: no events declared", flush=True) + return None + + # Register in sys.modules BEFORE exec_module so Pydantic/dataclass forward + # references (from ``from __future__ import annotations``) resolve; otherwise + # a handler declaring a BaseModel fails at first dispatch with + # "TypeAdapter ... is not fully defined". + module_name = f"hermes_hook_{hook_name}" + spec = importlib.util.spec_from_file_location(module_name, handler_path) + if spec is None or spec.loader is None: + print(f"[hooks] Skipping {hook_name}: could not load handler.py", flush=True) + return None + + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + try: + spec.loader.exec_module(module) + except Exception: + sys.modules.pop(module_name, None) + raise + + handle_fn = getattr(module, "handle", None) + if handle_fn is None: + print(f"[hooks] Skipping {hook_name}: no 'handle' function found", flush=True) + return None + return hook_name, events, handle_fn, manifest.get("description", "") + + class HookRegistry: """Discovers, loads, and fires event hooks.""" @@ -60,62 +104,23 @@ class HookRegistry: for hook_dir in sorted(HOOKS_DIR.iterdir()): if not hook_dir.is_dir(): continue - - manifest_path = hook_dir / "HOOK.yaml" - handler_path = hook_dir / "handler.py" - - if not manifest_path.exists() or not handler_path.exists(): - continue - try: - manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) - if not manifest or not isinstance(manifest, dict): - print(f"[hooks] Skipping {hook_dir.name}: invalid HOOK.yaml", flush=True) - continue - - hook_name = manifest.get("name", hook_dir.name) - events = manifest.get("events", []) - if not events: - print(f"[hooks] Skipping {hook_name}: no events declared", flush=True) - continue - - # Register in sys.modules BEFORE exec_module so Pydantic/dataclass - # forward references (from ``from __future__ import annotations``) - # resolve; otherwise a handler declaring a BaseModel fails at first - # dispatch with "TypeAdapter ... is not fully defined". - module_name = f"hermes_hook_{hook_name}" - spec = importlib.util.spec_from_file_location(module_name, handler_path) - if spec is None or spec.loader is None: - print(f"[hooks] Skipping {hook_name}: could not load handler.py", flush=True) - continue - - module = importlib.util.module_from_spec(spec) - sys.modules[module_name] = module - try: - spec.loader.exec_module(module) - except Exception: - sys.modules.pop(module_name, None) - raise - - handle_fn = getattr(module, "handle", None) - if handle_fn is None: - print(f"[hooks] Skipping {hook_name}: no 'handle' function found", flush=True) - continue - - for event in events: - self._handlers.setdefault(event, []).append(handle_fn) - - self._loaded_hooks.append({ - "name": hook_name, - "description": manifest.get("description", ""), - "events": events, - "path": str(hook_dir), - }) - - print(f"[hooks] Loaded hook '{hook_name}' for events: {events}", flush=True) - + loaded = _load_hook_dir(hook_dir) except Exception as e: print(f"[hooks] Error loading hook {hook_dir.name}: {e}", flush=True) + continue + if loaded is None: + continue + hook_name, events, handle_fn, description = loaded + for event in events: + self._handlers.setdefault(event, []).append(handle_fn) + self._loaded_hooks.append({ + "name": hook_name, + "description": description, + "events": events, + "path": str(hook_dir), + }) + print(f"[hooks] Loaded hook '{hook_name}' for events: {events}", flush=True) def _resolve_handlers(self, event_type: str) -> List[Callable]: """Exact-match handlers first, then ``:*`` wildcard handlers. diff --git a/gateway/lifecycle_ledger.py b/gateway/lifecycle_ledger.py index 3985013923..f63dff5d15 100644 --- a/gateway/lifecycle_ledger.py +++ b/gateway/lifecycle_ledger.py @@ -1,24 +1,12 @@ """Gateway lifecycle ledger — durable termination-reason evidence. -Graceful shutdowns already leave forensics (:mod:`gateway.shutdown_forensics`, -``gateway-exit-diag.log``); an **unclean death** (SIGKILL, kernel OOM kill, VM -death) runs no handler, so the next boot would otherwise have no record of it. - -A tiny state machine persisted to ``/state/gateway.lifecycle.json`` -closes that gap: - -* :func:`record_startup` reads the previous life's sentinel. ``phase == - "running"`` means no exit path ran → unclean death. The finding (with the - last heartbeat's memory sample) is appended to ``gateway-exit-diag.log`` as a - ``gateway.previous_unclean_exit`` record and logged at WARNING, then the - sentinel is rewritten as ``phase=running`` for the new life. -* :func:`mark_exited` rewrites the sentinel as ``phase=exited`` on every clean - exit path (``_exit_after_graceful_shutdown`` and the watchdog ``os._exit`` sites). -* :func:`sample_memory` is the cheap /proc snapshot the 30s loop heartbeat - embeds, so OOM crash cycles are classifiable from the volume alone. - -Everything here is best-effort: a forensics failure must never affect the -gateway lifecycle it is observing. +Graceful shutdowns leave forensics (``gateway-exit-diag.log``); an unclean +death (SIGKILL, kernel OOM, VM death) runs no handler. A sentinel at +``/state/gateway.lifecycle.json`` closes the gap: +:func:`record_startup` finds ``phase == "running"`` from the previous life → +unclean death, appended to the exit-diag log as ``gateway.previous_unclean_exit`` +and logged at WARNING; :func:`mark_exited` rewrites ``phase=exited`` on every +clean exit path. Best-effort: forensics must never affect the lifecycle. """ from __future__ import annotations @@ -28,6 +16,7 @@ import logging import os import sqlite3 import time +from contextlib import closing from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Optional @@ -38,11 +27,6 @@ _LIFECYCLE_RELATIVE = ("state", "gateway.lifecycle.json") _EXIT_DIAG_RELATIVE = ("logs", "gateway-exit-diag.log") _STATE_DB_RELATIVE = ("state.db",) -# Heuristic OOM-suspicion thresholds on the last heartbeat's memory sample. -# Deliberately conservative: only a hint; classification stays with the reader. -_LOW_MEM_AVAILABLE_KIB = 64 * 1024 # < 64 MiB available -_LOW_MEM_AVAILABLE_FRACTION = 0.05 # < 5% of MemTotal available - def _process_hermes_home() -> Path: """HERMES_HOME for process-level identity files (ignore task overrides).""" @@ -55,8 +39,7 @@ def _process_hermes_home() -> Path: def _home_path(home: Optional[Path], relative: tuple) -> Path: - base = home if home is not None else _process_hermes_home() - return base.joinpath(*relative) + return (home if home is not None else _process_hermes_home()).joinpath(*relative) def get_lifecycle_sentinel_path(home: Optional[Path] = None) -> Path: @@ -64,38 +47,41 @@ def get_lifecycle_sentinel_path(home: Optional[Path] = None) -> Path: return _home_path(home, _LIFECYCLE_RELATIVE) -def sample_memory() -> Dict[str, Any]: - """Cheap memory snapshot: own RSS + system availability + swap. +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() - Pure ``/proc`` reads, Linux-only (returns ``{}`` elsewhere), never - raises. Values in KiB to match the kernel's units. - """ - sample: Dict[str, Any] = {} + +def _proc_fields(path: str, wanted: Dict[str, str]) -> Dict[str, int]: + """``{dst: int}`` for each ``src: dst`` key found in a ``Key: value`` /proc file.""" + found: Dict[str, int] = {} try: - with open("/proc/self/status", encoding="utf-8") as fh: - for line in fh: - if line.startswith("VmRSS:"): - sample["rss_kib"] = int(line.split()[1]) - break - except (OSError, ValueError, IndexError): - pass - try: - meminfo: Dict[str, int] = {} - wanted = {"MemTotal", "MemAvailable", "SwapTotal", "SwapFree"} - with open("/proc/meminfo", encoding="utf-8") as fh: + with open(path, encoding="utf-8") as fh: for line in fh: key = line.split(":", 1)[0] if key in wanted: - meminfo[key] = int(line.split()[1]) - if len(meminfo) == len(wanted): + found[wanted[key]] = int(line.split()[1]) + if len(found) == len(wanted): break - for src, dst in (("MemTotal", "mem_total_kib"), ("MemAvailable", "mem_available_kib")): - if src in meminfo: - sample[dst] = meminfo[src] - if "SwapTotal" in meminfo and "SwapFree" in meminfo: - sample["swap_used_kib"] = meminfo["SwapTotal"] - meminfo["SwapFree"] except (OSError, ValueError, IndexError): - pass + return {} + return found + + +def sample_memory() -> Dict[str, Any]: + """Cheap /proc snapshot (KiB, kernel units): own RSS + MemTotal/MemAvailable + swap used. + + Linux-only (``{}`` elsewhere), never raises; embedded in the 30s loop + heartbeat so OOM crash cycles are classifiable from the volume alone. + """ + sample = _proc_fields("/proc/self/status", {"VmRSS": "rss_kib"}) + mem = _proc_fields("/proc/meminfo", { + "MemTotal": "mem_total_kib", "MemAvailable": "mem_available_kib", + "SwapTotal": "SwapTotal", "SwapFree": "SwapFree", + }) + swap_total, swap_free = mem.pop("SwapTotal", None), mem.pop("SwapFree", None) + sample.update(mem) + if swap_total is not None and swap_free is not None: + sample["swap_used_kib"] = swap_total - swap_free return sample @@ -119,8 +105,7 @@ def _write_sentinel(payload: Dict[str, Any], home: Optional[Path]) -> None: def _append_exit_diag(record: Dict[str, Any], home: Optional[Path]) -> None: - """Append a JSON line to gateway-exit-diag.log (same format as the CLI's - ``_exit_diag`` records so existing tooling greps both).""" + """Append a JSON line to gateway-exit-diag.log (same format as the CLI's ``_exit_diag``).""" path = _home_path(home, _EXIT_DIAG_RELATIVE) try: path.parent.mkdir(parents=True, exist_ok=True) @@ -133,9 +118,8 @@ def _append_exit_diag(record: Dict[str, Any], home: Optional[Path]) -> None: def _pid_alive_with_start_time(pid: Any, start_time: Any) -> bool: """True when ``pid`` is a live process matching ``start_time`` (±2s). - Guards the takeover race: during ``--replace`` the old gateway can still - be mid-teardown when the new one boots — a live matching owner is a - planned handover, not an unclean death. + Guards the ``--replace`` takeover race: a live matching owner mid-teardown + is a planned handover, not an unclean death. """ try: pid_int = int(pid) @@ -144,9 +128,8 @@ def _pid_alive_with_start_time(pid: Any, start_time: Any) -> bool: if pid_int <= 0: return False try: - # NOT os.kill(pid, 0): on Windows that sends CTRL_C_EVENT to the - # target's console group. _pid_exists is the canonical no-kill probe. - from gateway.status import _pid_exists + # NOT os.kill(pid, 0): on Windows that sends CTRL_C_EVENT to the target's console group. + from gateway.status import _pid_exists, get_process_start_time if not _pid_exists(pid_int): return False @@ -155,8 +138,6 @@ def _pid_alive_with_start_time(pid: Any, start_time: Any) -> bool: if start_time is None: return True # alive; can't disambiguate PID reuse — err on "alive" try: - from gateway.status import get_process_start_time - actual = get_process_start_time(pid_int) return actual is None or abs(float(actual) - float(start_time)) <= 2.0 except Exception: @@ -164,9 +145,7 @@ def _pid_alive_with_start_time(pid: Any, start_time: Any) -> bool: def detect_unclean_exit(home: Optional[Path] = None) -> Optional[Dict[str, Any]]: - """Inspect the previous life's sentinel; return an evidence dict when it - died uncleanly, else ``None``. Read-only — does not rewrite the sentinel. - """ + """Evidence dict when the previous life died uncleanly, else ``None``. Read-only.""" sentinel = _read_json(get_lifecycle_sentinel_path(home)) if not sentinel or sentinel.get("phase") != "running": return None @@ -178,7 +157,6 @@ def detect_unclean_exit(home: Optional[Path] = None) -> Optional[Dict[str, Any]] "prior_started_at": sentinel.get("started_at"), "prior_start_time": sentinel.get("start_time"), } - # Enrich with the last heartbeat: last proven liveness and memory at that moment. try: from gateway.shutdown_watchdog import get_loop_heartbeat_path @@ -191,15 +169,15 @@ def detect_unclean_exit(home: Optional[Path] = None) -> Optional[Dict[str, Any]] mem = hb.get("mem") if isinstance(mem, dict): evidence["last_heartbeat_mem"] = mem - total = mem.get("mem_total_kib") - avail = mem.get("mem_available_kib") + # OOM suspicion is only a hint (classification stays with the reader); + # thresholds are memory_status' "critical" tier so a live warning and a + # post-mortem verdict can never disagree. + from gateway.memory_status import _CRITICAL_AVAILABLE_FRACTION, _CRITICAL_AVAILABLE_KIB + + total, avail = mem.get("mem_total_kib"), mem.get("mem_available_kib") if isinstance(avail, int) and ( - avail < _LOW_MEM_AVAILABLE_KIB - or ( - isinstance(total, int) - and total > 0 - and avail / total < _LOW_MEM_AVAILABLE_FRACTION - ) + avail < _CRITICAL_AVAILABLE_KIB + or (isinstance(total, int) and total > 0 and avail / total < _CRITICAL_AVAILABLE_FRACTION) ): evidence["suspected_oom"] = True return evidence @@ -209,22 +187,18 @@ def check_state_db_integrity(home: Optional[Path] = None) -> str: """Return ``"ok"``, ``"absent"``, or the first ``quick_check`` complaint. Called only after an unclean death — a SIGKILL mid-WAL-checkpoint can leave - half-written b-tree pages (see ``_enforce_macos_synchronous_full`` in - :mod:`hermes_state`). ``quick_check(1)`` stops at the first problem (~2s on - a healthy 500MB store): cheap once per unclean boot, too costly every boot. - Opened normally, not read-only: a WAL store needs its -shm sidecar for a - read-only open, and the PRAGMA itself writes nothing. Never raises. + half-written b-tree pages. ``quick_check(1)`` stops at the first problem + (~2s on a healthy 500MB store): cheap once per unclean boot, too costly every + boot. Opened normally, not read-only: a WAL store needs its -shm sidecar for + a read-only open, and the PRAGMA writes nothing. Never raises. """ path = _home_path(home, _STATE_DB_RELATIVE) if not path.exists(): return "absent" try: - conn = sqlite3.connect(str(path)) - try: + with closing(sqlite3.connect(str(path))) as conn: row = conn.execute("PRAGMA quick_check(1)").fetchone() - finally: - conn.close() - except Exception as exc: # sqlite3.Error, OSError, anything + except Exception as exc: return f"check-failed: {exc}" if not row or row[0] is None: return "check-failed: no result" @@ -232,12 +206,10 @@ def check_state_db_integrity(home: Optional[Path] = None) -> str: def record_startup(home: Optional[Path] = None) -> Optional[Dict[str, Any]]: - """Boot-time entry point: report any unclean previous exit, then claim - the sentinel for the current life. + """Boot entry point: report any unclean previous exit, then claim the sentinel. - Returns the unclean-exit evidence dict (also persisted to - ``gateway-exit-diag.log`` and logged at WARNING) or ``None``. Never - raises. + Returns the evidence dict (also persisted to ``gateway-exit-diag.log`` and + logged at WARNING) or ``None``. Never raises. """ evidence: Optional[Dict[str, Any]] = None try: @@ -253,36 +225,24 @@ def record_startup(home: Optional[Path] = None) -> Optional[Dict[str, Any]]: "repaired. Run `hermes doctor`.", verdict, ) - _append_exit_diag({ - "ts": datetime.now(timezone.utc).isoformat(), - "tag": "gateway.previous_unclean_exit", - "pid": os.getpid(), - **evidence, - }, home) + _append_exit_diag( + {"ts": _now_iso(), "tag": "gateway.previous_unclean_exit", "pid": os.getpid(), **evidence}, home + ) logger.warning( "Previous gateway life (pid=%s, started_at=%s) exited UNCLEANLY " "(no exit path ran — SIGKILL / OOM / VM death). " "last_heartbeat_at=%s last_mem=%s suspected_oom=%s", - evidence.get("prior_pid"), - evidence.get("prior_started_at"), - evidence.get("last_heartbeat_at"), - evidence.get("last_heartbeat_mem"), - evidence.get("suspected_oom", False), + evidence.get("prior_pid"), evidence.get("prior_started_at"), evidence.get("last_heartbeat_at"), + evidence.get("last_heartbeat_mem"), evidence.get("suspected_oom", False), ) except Exception: logger.debug("Unclean-exit detection failed", exc_info=True) try: - claim: Dict[str, Any] = { - "phase": "running", - "pid": os.getpid(), - "start_time": time.time(), - "started_at": datetime.now(timezone.utc).isoformat(), - } - # Carry the verdict on the PREVIOUS life forward on the new sentinel: it - # is the only machine-readable copy (the exit-diag log is append-only - # prose) and /api/status reads it to report an OOM restart. Scoped to - # this life only — the next clean exit or boot rewrites the sentinel. + claim: Dict[str, Any] = {"phase": "running", "pid": os.getpid(), "start_time": time.time(), "started_at": _now_iso()} + # Carry the verdict on the PREVIOUS life on the new sentinel: it is the only + # machine-readable copy (/api/status reads it to report an OOM restart). + # Scoped to this life — the next clean exit or boot rewrites the sentinel. if evidence is not None: claim["prior_unclean_exit"] = True if evidence.get("suspected_oom"): @@ -293,49 +253,34 @@ def record_startup(home: Optional[Path] = None) -> Optional[Dict[str, Any]]: return evidence -def mark_exited( - exit_code: Optional[int] = None, - reason: str = "graceful_shutdown", - home: Optional[Path] = None, -) -> None: +def mark_exited(exit_code: Optional[int] = None, reason: str = "graceful_shutdown", home: Optional[Path] = None) -> None: """Mark the current life as cleanly exited. Idempotent, never raises. - Only rewrites the sentinel when it is provably owned by this process: - during a ``--replace`` takeover the replacement claims the sentinel before - the old process finishes teardown, and the old life must not clobber the - new owner's ``running`` phase. A sentinel with ``pid=None`` (or malformed) - has unknown ownership and is likewise left alone. + Only rewrites a sentinel provably owned by this process: during a + ``--replace`` takeover the replacement claims it before the old process + finishes teardown, and the old life must not clobber the new ``running`` + phase. ``pid=None`` / malformed sentinels have unknown ownership → left alone. """ try: sentinel = _read_json(get_lifecycle_sentinel_path(home)) if sentinel is not None and sentinel.get("pid") != os.getpid(): return _write_sentinel({ - "phase": "exited", - "pid": os.getpid(), - "exit_code": exit_code, - "exit_reason": reason, - "exited_at": datetime.now(timezone.utc).isoformat(), + "phase": "exited", "pid": os.getpid(), "exit_code": exit_code, + "exit_reason": reason, "exited_at": _now_iso(), }, home) except Exception: logger.debug("Failed to mark lifecycle sentinel exited", exc_info=True) def read_prior_exit_label(profile_home: Path) -> str: - """Container-boot helper: ``clean`` / ``unclean`` / ``unknown`` summary of - how the profile's last gateway life ended. Read-only, exception-free; - used by ``hermes_cli.container_boot`` to annotate ``container-boot.log``. + """``clean`` / ``unclean`` / ``unknown`` for how the profile's last gateway life ended. + + Read-only, exception-free; annotates ``container-boot.log``. ``running`` is + unclean because the old PID namespace is gone at container boot. """ try: - sentinel = _read_json(get_lifecycle_sentinel_path(profile_home)) - if not sentinel: - return "unknown" - phase = sentinel.get("phase") - if phase == "exited": - return "clean" - if phase == "running": - # Old PID namespace is gone at container boot — never exited cleanly. - return "unclean" + phase = (_read_json(get_lifecycle_sentinel_path(profile_home)) or {}).get("phase") + return {"exited": "clean", "running": "unclean"}.get(phase, "unknown") except Exception: - pass - return "unknown" + return "unknown" diff --git a/gateway/media_policy.py b/gateway/media_policy.py index 2d6019a34d..c7f0f98749 100644 --- a/gateway/media_policy.py +++ b/gateway/media_policy.py @@ -7,15 +7,12 @@ from environment variables: - ``HERMES_MEDIA_ALLOW_DIRS`` <- gateway.media_delivery_allow_dirs - ``HERMES_MEDIA_TRUST_RECENT_FILES`` <- gateway.trust_recent_files -The translation used to run only in gateway startup, so standalone delivery -paths (``hermes cron run``, ``hermes send``, a standalone cron tick) filtered -MEDIA paths under a different policy and silently dropped attachments in -strict/allowlisted deployments (text is unaffected -- only media goes through -path validation). ``apply_media_policy_env()`` is the shared, -idempotent helper every delivery entrypoint calls before filtering media paths. - -Precedence: an explicitly-set environment variable WINS over config.yaml, so a -shell-exported override (and gateway startup's own earlier run) survives. +Every delivery entrypoint (gateway startup, ``hermes cron run``, ``hermes send``) +calls :func:`apply_media_policy_env` before filtering media paths, so standalone +paths filter under the same policy as the gateway instead of silently dropping +attachments in strict/allowlisted deployments. An explicitly-set environment +variable WINS over config.yaml, so a shell override (and gateway startup's own +earlier run) survives. """ from __future__ import annotations diff --git a/gateway/media_repair.py b/gateway/media_repair.py index adfc3c3bdc..e0a4271eb0 100644 --- a/gateway/media_repair.py +++ b/gateway/media_repair.py @@ -1,19 +1,13 @@ """Repair model-mangled ``computer_use`` screenshot paths in final responses. -``computer_use`` persists a screenshot into the Hermes image cache and tells the -model its absolute path. Some models rewrite a Windows path into a POSIX-looking -one (``C:\\Users\\Alice\\...`` -> ``/Users/Alice/...``) inside an explicit -``MEDIA:`` directive, so delivery-path validation rejects it and drops the -attachment. - -The repair is deliberately narrow: it only rewrites paths inside a response that -*already* carries an explicit ``MEDIA:`` directive, and only when the directive's -generated ``computer_use_`` basename exactly matches a canonical screenshot -path returned by ``computer_use`` in the current turn. It never auto-attaches -captures; normal media path validation still runs afterwards. - -Own module (like ``gateway/media_policy.py``) so the gateway turn path, gateway -background tasks and cron delivery share one implementation. +``computer_use`` persists a screenshot into the image cache and tells the model +its absolute path. Some models rewrite a Windows path into a POSIX-looking one +(``C:\\Users\\Alice\\...`` -> ``/Users/Alice/...``) inside an explicit ``MEDIA:`` +directive, so delivery-path validation rejects it and drops the attachment. +Deliberately narrow: only rewrites paths inside a response that *already* +carries a ``MEDIA:`` directive, and only when its ``computer_use_`` +basename exactly matches a canonical path returned by ``computer_use`` this +turn. Never auto-attaches; normal media path validation still runs afterwards. """ from __future__ import annotations @@ -29,15 +23,12 @@ logger = logging.getLogger(__name__) # root, or UNC share. Shared string so the summary regex stays in sync. _ABS_PATH_PREFIX_PATTERN = r"(?:[A-Za-z]:[/\\]|/|\\\\)" _ABS_PATH_PREFIX_RE = re.compile(r"^" + _ABS_PATH_PREFIX_PATTERN) +_CAPTURE_BASENAME_PATTERN = r"computer_use_[0-9a-f]{32}\.(?:png|jpe?g)" -_COMPUTER_USE_CAPTURE_BASENAME_RE = re.compile( - r"^computer_use_[0-9a-f]{32}\.(?:png|jpe?g)$", - re.IGNORECASE, -) +_COMPUTER_USE_CAPTURE_BASENAME_RE = re.compile(r"^" + _CAPTURE_BASENAME_PATTERN + r"$", re.IGNORECASE) _COMPUTER_USE_CAPTURE_SUMMARY_RE = re.compile( r"\(shareable screenshot saved to " - r"(?P" + _ABS_PATH_PREFIX_PATTERN + r"[^\r\n]*?" - r"computer_use_[0-9a-f]{32}\.(?:png|jpe?g))\)", + r"(?P" + _ABS_PATH_PREFIX_PATTERN + r"[^\r\n]*?" + _CAPTURE_BASENAME_PATTERN + r")\)", re.IGNORECASE, ) @@ -86,29 +77,23 @@ def _iter_computer_use_capture_paths(content: Any) -> Iterator[str]: return for match in _COMPUTER_USE_CAPTURE_SUMMARY_RE.finditer(content): yield match.group("path").strip() - return - - if isinstance(content, list): + elif isinstance(content, list): for part in content: yield from _iter_computer_use_capture_paths(part) - return - - if not isinstance(content, dict): - return - - screenshot_path = content.get("screenshot_path") - if isinstance(screenshot_path, str): - yield screenshot_path - meta = content.get("meta") - if isinstance(meta, dict) and isinstance(meta.get("screenshot_path"), str): - yield meta["screenshot_path"] - # Producer shapes (tools/computer_use/tool.py::_capture_response): - # content/text = multimodal parts; text_summary/summary = the line carrying - # "(shareable screenshot saved to ...)". - for field in ("content", "text", "text_summary", "summary"): - nested = content.get(field) - if isinstance(nested, (str, dict, list)): - yield from _iter_computer_use_capture_paths(nested) + elif isinstance(content, dict): + screenshot_path = content.get("screenshot_path") + if isinstance(screenshot_path, str): + yield screenshot_path + meta = content.get("meta") + if isinstance(meta, dict) and isinstance(meta.get("screenshot_path"), str): + yield meta["screenshot_path"] + # Producer shapes (tools/computer_use/tool.py::_capture_response): + # content/text = multimodal parts; text_summary/summary = the line carrying + # "(shareable screenshot saved to ...)". + for field in ("content", "text", "text_summary", "summary"): + nested = content.get(field) + if isinstance(nested, (str, dict, list)): + yield from _iter_computer_use_capture_paths(nested) def _current_turn_messages(messages: List[Dict[str, Any]], history_offset: int) -> List[Dict[str, Any]]: @@ -135,13 +120,11 @@ def repair_explicit_computer_use_media_paths( Repairs only an already-explicit ``MEDIA:`` directive whose generated basename case-insensitively matches a canonical screenshot path from this - turn. Fail-open: the repair is cosmetic, so any unexpected error returns + turn. Fail-open: the repair is cosmetic, so any unexpected error returns the response unchanged rather than aborting delivery. """ try: - return _repair_explicit_computer_use_media_paths_inner( - response, messages, history_offset - ) + return _repair_explicit_computer_use_media_paths_inner(response, messages, history_offset) except Exception: logger.debug("computer_use media path repair failed", exc_info=True) return response @@ -163,9 +146,7 @@ def _repair_explicit_computer_use_media_paths_inner( if msg.get("role") not in {"tool", "function"}: continue call_id = str(msg.get("tool_call_id") or msg.get("call_id") or "") - tool_name = str( - msg.get("name") or msg.get("tool_name") or call_id_names.get(call_id) or "" - ) + tool_name = str(msg.get("name") or msg.get("tool_name") or call_id_names.get(call_id) or "") if tool_name != "computer_use": continue for path in _iter_computer_use_capture_paths(msg.get("content")): diff --git a/gateway/memory_monitor.py b/gateway/memory_monitor.py index 3ac51b531e..ddf7b502d3 100644 --- a/gateway/memory_monitor.py +++ b/gateway/memory_monitor.py @@ -1,15 +1,15 @@ """Periodic process memory usage logging for the gateway. -Ported from cline/cline#10343 (src/standalone/memory-monitor.ts). Emits one -grep-friendly ``[MEMORY] ...`` line every N seconds (default 300) from a daemon -thread so slow leaks in the long-lived gateway show up as an RSS time series in -``agent.log`` / ``gateway.log``. A baseline snapshot is logged on start and a -final one on stop. Uses stdlib ``resource`` first, ``psutil`` as fallback -(Windows); if neither works the monitor warns once and stays disabled. +Emits one grep-friendly ``[MEMORY] ...`` line every N seconds (default 300) +from a daemon thread so slow leaks show up as an RSS time series in the logs. +A baseline snapshot is logged on start and a final one on stop. Uses stdlib +``resource`` first, ``psutil`` as fallback (Windows); if neither works the +monitor warns once and stays disabled. """ from __future__ import annotations +import contextlib import gc import logging import os @@ -17,7 +17,6 @@ import sys import threading import time from typing import Optional -import contextlib logger = logging.getLogger(__name__) @@ -49,25 +48,16 @@ def _get_rss_mb() -> Optional[int]: def log_memory_usage(prefix: str = "") -> None: - """Log current memory usage as ``[MEMORY] [ ]rss=... gc=... threads=... uptime=...``. - - Safe to call on-demand from any thread at lifecycle moments. - """ + """Log ``[MEMORY] [ ]rss=... gc=... threads=... uptime=...``; safe from any thread.""" rss = _get_rss_mb() uptime = int(time.monotonic() - _start_time) if _start_time else 0 - try: - gc_counts = gc.get_count() # (gen0, gen1, gen2) - except Exception: - gc_counts = (0, 0, 0) - try: - thread_count = threading.active_count() - except Exception: - thread_count = 0 - - tag = f"{prefix} " if prefix else "" - rss_text = "unavailable" if rss is None else f"{rss}MB" logger.info( - "[MEMORY] %srss=%s gc=%s threads=%d uptime=%ds", tag, rss_text, gc_counts, thread_count, uptime + "[MEMORY] %srss=%s gc=%s threads=%d uptime=%ds", + f"{prefix} " if prefix else "", + "unavailable" if rss is None else f"{rss}MB", + gc.get_count(), # (gen0, gen1, gen2) + threading.active_count(), + uptime, ) diff --git a/gateway/memory_status.py b/gateway/memory_status.py index 9170083346..44c37df4da 100644 --- a/gateway/memory_status.py +++ b/gateway/memory_status.py @@ -1,17 +1,10 @@ """Memory status rollup for ``/api/status``. -Read side for memory-pressure signals the gateway already persists but only -logs: the 30s ``state/gateway.heartbeat`` (RSS + MemAvailable/MemTotal + swap, -from :func:`gateway.shutdown_watchdog.write_loop_heartbeat`) and the lifecycle -sentinel's ``suspected_oom`` flag (:func:`gateway.lifecycle_ledger.record_startup`). -Distills them into a compact block the dashboard SPA and the NAS availability -sweep consume — no new sampling, no IPC, two small file reads. - -Public-safety: ``/api/status`` is unauthenticated (``PUBLIC_API_PATHS``), so this -block carries only coarse numbers (MB granularity), enums, and booleans — the -same disclosure class as ``active_agents`` / ``nous_session_valid``. - -Best-effort and read-only: a missing/corrupt file degrades to +Read side for signals the gateway already persists: the 30s +``state/gateway.heartbeat`` (RSS + MemAvailable/MemTotal + swap) and the +lifecycle sentinel's ``suspected_oom`` flag. Two small file reads, no IPC. +``/api/status`` is unauthenticated, so the block carries only coarse numbers +(MB), enums and booleans. Best-effort: a missing/corrupt file degrades to ``pressure="unknown"`` rather than raising into the status endpoint. """ @@ -24,21 +17,18 @@ from typing import Any, Dict, Optional logger = logging.getLogger(__name__) -# Thresholds on system MemAvailable. ``critical`` mirrors the lifecycle ledger's -# OOM-suspicion heuristics (_LOW_MEM_AVAILABLE_KIB / _LOW_MEM_AVAILABLE_FRACTION): -# a level that would make a later unclean death "suspected OOM" should already -# warn while the process is alive. +# Thresholds on system MemAvailable. ``critical`` doubles as the lifecycle +# ledger's OOM-suspicion heuristic: a level that makes a later unclean death +# "suspected OOM" already warns while the process is alive. _CRITICAL_AVAILABLE_KIB = 64 * 1024 # < 64 MiB available _CRITICAL_AVAILABLE_FRACTION = 0.05 # < 5% of MemTotal _ELEVATED_AVAILABLE_KIB = 128 * 1024 # < 128 MiB available _ELEVATED_AVAILABLE_FRACTION = 0.15 # < 15% of MemTotal -# Writer cadence is 30s (DEFAULT_HEARTBEAT_INTERVAL_S); 150s tolerates a briefly -# stalled loop without letting a long-dead gateway's last sample pose as current. +# Writer cadence is 30s; 150s tolerates a briefly stalled loop without letting +# a long-dead gateway's last sample pose as current. _HEARTBEAT_FRESH_TTL_S = 150.0 -_KIB_PER_MB = 1024 - def _nonneg_int(value: Any) -> Optional[int]: """Return *value* if it is a non-negative int (bools rejected), else None.""" @@ -49,7 +39,7 @@ def _nonneg_int(value: Any) -> Optional[int]: def _mb(kib: Any) -> Optional[int]: kib = _nonneg_int(kib) - return None if kib is None else kib // _KIB_PER_MB + return None if kib is None else kib // 1024 def _parse_iso(value: Any) -> Optional[datetime]: @@ -82,23 +72,15 @@ def classify_pressure(available_kib: Any, total_kib: Any) -> str: return "ok" -def _read_heartbeat(home: Optional[Path]) -> Optional[Dict[str, Any]]: - try: - from gateway.lifecycle_ledger import _read_json - from gateway.shutdown_watchdog import get_loop_heartbeat_path - - return _read_json(get_loop_heartbeat_path(home)) - except Exception: - return None - - -def _read_sentinel(home: Optional[Path]) -> Optional[Dict[str, Any]]: +def _read_state_files(home: Optional[Path]) -> tuple: + """``(heartbeat, sentinel)`` dicts, each ``None`` when unreadable.""" try: from gateway.lifecycle_ledger import _read_json, get_lifecycle_sentinel_path + from gateway.shutdown_watchdog import get_loop_heartbeat_path - return _read_json(get_lifecycle_sentinel_path(home)) + return _read_json(get_loop_heartbeat_path(home)), _read_json(get_lifecycle_sentinel_path(home)) except Exception: - return None + return None, None def collect_memory_status( @@ -110,8 +92,8 @@ def collect_memory_status( ``home`` scopes the read to a profile's HERMES_HOME (``None`` = active profile); ``now`` is injectable for tests. Always returns a dict and never - raises — a down/never-started gateway or corrupt files yield - ``{"pressure": "unknown", ...}`` plus whatever fields could be recovered. + raises — a down gateway or corrupt files yield ``{"pressure": "unknown", ...}`` + plus whatever fields could be recovered. """ moment = now or datetime.now(timezone.utc) status: Dict[str, Any] = { @@ -123,13 +105,12 @@ def collect_memory_status( "sampled_at": None, "last_boot_unclean": False, "last_boot_suspected_oom": False, - # Identity of the CURRENT gateway life (sentinel started_at) — changes on - # every boot, so the dashboard can key banner dismissal on it and - # acknowledging one OOM restart does not mute the NEXT one. + # Identity of the CURRENT life (sentinel started_at): the dashboard keys + # banner dismissal on it so acknowledging one OOM restart does not mute the NEXT. "boot_id": None, } - heartbeat = _read_heartbeat(home) + heartbeat, sentinel = _read_state_files(home) if heartbeat: sampled_at = _parse_iso(heartbeat.get("updated_at")) mem = heartbeat.get("mem") @@ -146,7 +127,6 @@ def collect_memory_status( if 0 <= (moment - sampled_at).total_seconds() <= _HEARTBEAT_FRESH_TTL_S: status["pressure"] = classify_pressure(mem.get("mem_available_kib"), mem.get("mem_total_kib")) - sentinel = _read_sentinel(home) if sentinel: status["last_boot_unclean"] = bool(sentinel.get("prior_unclean_exit")) status["last_boot_suspected_oom"] = bool(sentinel.get("prior_suspected_oom")) diff --git a/gateway/message_timestamps.py b/gateway/message_timestamps.py index 4bb5fe4579..cb6315fc10 100644 --- a/gateway/message_timestamps.py +++ b/gateway/message_timestamps.py @@ -12,17 +12,19 @@ from datetime import datetime from typing import Any, Optional, Tuple -# Current gateway format: [Tue 2026-04-28 13:40:53 CEST] -_HUMAN_TIMESTAMP_RE = re.compile( - r"^\[(?P[A-Z][a-z]{2}) " +# Leading timestamp prefix, either the current human format +# ``[Tue 2026-04-28 13:40:53 CEST]`` or the older ISO one +# ``[2026-04-13T17:02:06+0200]`` / ``[...+02:00]`` (human tried first). +_TIMESTAMP_PREFIX_RE = re.compile( + r"^\[(?:" + r"(?P[A-Z][a-z]{2}) " r"(?P\d{4}-\d{2}-\d{2}) " r"(?P