From b1bc7f77991117f05483657b6e9fb9e6d6b3c6f5 Mon Sep 17 00:00:00 2001 From: "Andrex Ibiza, MBA" <84248988+andrexibiza@users.noreply.github.com> Date: Wed, 5 Aug 2026 08:09:39 -0500 Subject: [PATCH 1/2] refactor(gateway): extract session-config + telegram-topics mixins from run.py (shard s2) --- contributors/emails/andrexibiza@gmail.com | 1 + .../andrexibiza@users.noreply.github.com | 1 + gateway/run.py | 919 +----------------- gateway/session_config_mixin.py | 831 ++++++++++++++++ gateway/telegram_topics_mixin.py | 216 ++++ .../test_run_s2_session_config_and_topics.py | 293 ++++++ 6 files changed, 1345 insertions(+), 916 deletions(-) create mode 100644 contributors/emails/andrexibiza@gmail.com create mode 100644 contributors/emails/andrexibiza@users.noreply.github.com create mode 100644 gateway/session_config_mixin.py create mode 100644 gateway/telegram_topics_mixin.py create mode 100644 tests/gateway/test_run_s2_session_config_and_topics.py diff --git a/contributors/emails/andrexibiza@gmail.com b/contributors/emails/andrexibiza@gmail.com new file mode 100644 index 0000000000000..efa930813a29e --- /dev/null +++ b/contributors/emails/andrexibiza@gmail.com @@ -0,0 +1 @@ +andrexibiza diff --git a/contributors/emails/andrexibiza@users.noreply.github.com b/contributors/emails/andrexibiza@users.noreply.github.com new file mode 100644 index 0000000000000..efa930813a29e --- /dev/null +++ b/contributors/emails/andrexibiza@users.noreply.github.com @@ -0,0 +1 @@ +andrexibiza diff --git a/gateway/run.py b/gateway/run.py index 24d501b5b752f..71f992d0ffc55 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2257,6 +2257,8 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor from gateway.authz_mixin import GatewayAuthorizationMixin from gateway.kanban_watchers import GatewayKanbanWatchersMixin from gateway.slash_commands import GatewaySlashCommandsMixin +from gateway.session_config_mixin import SessionConfigMixin +from gateway.telegram_topics_mixin import TelegramTopicsMixin from gateway.turn_context import TurnContext from gateway.platforms.base import ( BasePlatformAdapter, @@ -5637,7 +5639,7 @@ def _title_failure_cb(task: str, exc: BaseException) -> None: -class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, GatewaySlashCommandsMixin): +class GatewayRunner(SessionConfigMixin, TelegramTopicsMixin, GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, GatewaySlashCommandsMixin): """ Main gateway controller. @@ -6587,201 +6589,22 @@ def _session_key_for_source(self, source: SessionSource) -> str: profile=_profile, ) - def _telegram_topic_mode_enabled(self, source: SessionSource) -> bool: - """Return whether Telegram DM topic mode is active for this chat.""" - if source.platform != Platform.TELEGRAM or source.chat_type != "dm": - return False - session_db = getattr(self, "_session_db", None) - if session_db is None: - return False - # Runs off-loop (always via asyncio.to_thread); use the sync handle. - session_db = getattr(session_db, "_db", session_db) - try: - raw = session_db.is_telegram_topic_mode_enabled( - chat_id=str(source.chat_id), - user_id=str(source.user_id), - ) - except Exception: - logger.debug("Failed to read Telegram topic mode state", exc_info=True) - return False - # Only honor a real True from the SessionDB. Any other value - # (including MagicMock instances from test fixtures that didn't - # opt into topic mode) means topic mode is off for this chat. - return raw is True # Telegram's General (pinned top) topic in forum-enabled private chats. # Bot API behavior varies: some clients omit message_thread_id for # General, others send "1". Treat both as "root" for lobby/lane purposes. _TELEGRAM_GENERAL_TOPIC_IDS = frozenset({"", "1"}) - def _is_telegram_topic_root_lobby(self, source: SessionSource) -> bool: - """True for the main Telegram DM (or General topic) when topic mode has made it a lobby.""" - if source.platform != Platform.TELEGRAM or source.chat_type != "dm": - return False - if not self._telegram_topic_mode_enabled(source): - return False - tid = str(source.thread_id or "") - return tid in self._TELEGRAM_GENERAL_TOPIC_IDS - def _is_telegram_topic_lane(self, source: SessionSource) -> bool: - """True for a user-created Telegram private-chat topic lane.""" - if source.platform != Platform.TELEGRAM or source.chat_type != "dm": - return False - if not self._telegram_topic_mode_enabled(source): - return False - tid = str(source.thread_id or "") - if not tid or tid in self._TELEGRAM_GENERAL_TOPIC_IDS: - return False - return True _TELEGRAM_LOBBY_REMINDER_COOLDOWN_S = 30.0 - def _should_send_telegram_lobby_reminder(self, source: SessionSource) -> bool: - """Rate-limit root-DM lobby reminders to one message per cooldown window. - A user who forgets multi-session mode is enabled and types several - prompts in the root DM would otherwise get a reminder for every - message. Cap it so the first one lands and the rest stay quiet. - """ - if not hasattr(self, "_telegram_lobby_reminder_ts"): - self._telegram_lobby_reminder_ts = {} - chat_id = str(source.chat_id or "") - if not chat_id: - return True - import time as _time - now = _time.monotonic() - last = self._telegram_lobby_reminder_ts.get(chat_id, 0.0) - if now - last < self._TELEGRAM_LOBBY_REMINDER_COOLDOWN_S: - return False - self._telegram_lobby_reminder_ts[chat_id] = now - return True - def _telegram_topic_root_lobby_message(self) -> str: - return ( - "This main chat is reserved for system commands.\n\n" - "To start a new Hermes chat, open the All Messages topic at the top " - "of this bot interface and send any message there. Telegram will " - "create a new topic for that message; each topic works as an " - "independent Hermes session." - ) - def _telegram_topic_root_new_message(self) -> str: - return ( - "To start a new parallel Hermes chat, open the All Messages topic " - "at the top of this bot interface and send any message there. " - "Telegram will create a new topic for it.\n\n" - "Each topic is an independent Hermes session. Use /new inside an " - "existing topic only if you want to replace that topic's current session." - ) - def _telegram_topic_new_header(self, source: SessionSource) -> Optional[str]: - if not self._is_telegram_topic_lane(source): - return None - return ( - "Started a new Hermes session in this topic.\n\n" - "Tip: for parallel work, open All Messages and send a message there " - "to create a separate topic instead of using /new here. /new replaces " - "the session attached to the current topic." - ) - def _record_telegram_topic_binding( - self, - source: SessionSource, - session_entry, - ) -> None: - """Persist the Telegram topic -> Hermes session binding for topic lanes.""" - session_db = getattr(self, "_session_db", None) - if session_db is None or not source.chat_id or not source.thread_id: - return - # Runs off-loop (always via asyncio.to_thread); use the sync handle. - session_db = getattr(session_db, "_db", session_db) - session_db.bind_telegram_topic( - chat_id=str(source.chat_id), - thread_id=str(source.thread_id), - user_id=str(source.user_id or ""), - session_key=session_entry.session_key, - session_id=session_entry.session_id, - ) - - def _sync_telegram_topic_binding( - self, - source: SessionSource, - session_entry, - *, - reason: str, - ) -> None: - """Update the topic binding to point at ``session_entry.session_id``. - - Telegram topic lanes persist a (chat_id, thread_id) -> session_id row - so reopening a topic in a fresh process resumes the right Hermes - session. When compression rotates ``session_entry.session_id`` mid-turn, - the binding goes stale and the next inbound message in that topic - reloads the oversized parent transcript instead of the compressed - child, retriggering preflight compression — sometimes in a loop - (#20470, #29712, #33414). - """ - if not self._is_telegram_topic_lane(source): - return - try: - self._record_telegram_topic_binding(source, session_entry) - except Exception: - logger.debug( - "telegram topic binding refresh failed (%s)", reason, exc_info=True, - ) - - def _recover_telegram_topic_thread_id( - self, - source: SessionSource, - ) -> Optional[str]: - """Pin DM-topic routing to the user's last-active topic. - - Telegram can omit ``message_thread_id`` or surface General (``1``) - for some topic-mode DM replies. In those lobby-shaped cases, keep the - conversation attached to the user's most-recent bound topic. - Do not rewrite a non-lobby, previously-unbound thread id: a newly - created Telegram DM topic is also "unknown" until the first inbound - message is recorded, and rewriting it would send that brand-new topic's - answer into an older lane. Returns None to leave the source alone. - """ - if ( - source.platform != Platform.TELEGRAM - or source.chat_type != "dm" - or not source.chat_id - or not source.user_id - or not self._telegram_topic_mode_enabled(source) - ): - return None - inbound = str(source.thread_id or "") - is_lobby = not inbound or inbound in self._TELEGRAM_GENERAL_TOPIC_IDS - if not is_lobby: - # A non-lobby, unknown thread_id is most likely the first message in - # a brand-new Telegram DM topic. Preserve it so it can be recorded - # as a new independent lane below instead of hijacking the latest - # existing topic binding. - return None - session_db = getattr(self, "_session_db", None) - if session_db is None: - return None - # Runs off-loop (always via asyncio.to_thread); use the sync handle. - session_db = getattr(session_db, "_db", session_db) - try: - bindings = session_db.list_telegram_topic_bindings_for_chat( - chat_id=str(source.chat_id), - ) - except Exception: - logger.debug("topic-recover: read failed", exc_info=True) - return None - if not bindings: - return None - user_id = str(source.user_id) - for b in bindings: # newest-first - if str(b.get("user_id") or "") == user_id: - recovered = str(b.get("thread_id") or "") - if recovered and recovered != inbound: - return recovered - return None - return None def _normalize_source_for_session_key( self, @@ -6811,268 +6634,8 @@ def _normalize_source_for_session_key( return source return dataclasses.replace(source, thread_id=recovered) - def _resolve_session_agent_runtime( - self, - *, - source: Optional[SessionSource] = None, - session_key: Optional[str] = None, - user_config: Optional[dict] = None, - ) -> tuple[str, dict]: - """Resolve model/runtime for a session. - - Priority (highest first): session ``/model`` → ``channel_overrides`` → - global config/env (``_resolve_gateway_model(user_config)`` and default - provider resolution). - """ - resolved_session_key = session_key - if not resolved_session_key and source is not None: - try: - resolved_session_key = self._session_key_for_source(source) - except Exception: - resolved_session_key = None - - model = _resolve_gateway_model(user_config) - if resolved_session_key: - self._rehydrate_session_model_override(resolved_session_key) - _override_state = ( - self._peek_session_state(resolved_session_key) - if resolved_session_key - else None - ) - override = ( - _override_state.conversation.model_override if _override_state else None - ) - if override: - override_model = override.get("model", model) - override_runtime = { - "provider": override.get("provider"), - "api_key": override.get("api_key"), - "base_url": override.get("base_url"), - "api_mode": override.get("api_mode"), - "max_tokens": override.get("max_tokens"), - "credential_pool": override.get("credential_pool"), - } - if override_runtime.get("api_key"): - if override_runtime.get("credential_pool") is None: - override_runtime["credential_pool"] = _credential_pool_for_provider( - override.get("provider") - ) - logger.debug( - "Session model override (fast): session=%s config_model=%s -> override_model=%s provider=%s", - resolved_session_key or "", model, override_model, - override_runtime.get("provider"), - ) - return override_model, override_runtime - # Override exists but has no api_key — fall through to env-based - # resolution and apply model/provider from the override on top. - logger.debug( - "Session model override (no api_key, fallback): session=%s config_model=%s override_model=%s", - resolved_session_key or "", model, override_model, - ) - else: - logger.debug( - "No session model override: session=%s config_model=%s override_keys=%s", - resolved_session_key or "", model, - [ - _key - for _key, _st in list(self._sessions_map().items()) - if _st.conversation.model_override is not None - ][:5] or "[]", - ) - - runtime_kwargs = _resolve_runtime_agent_kwargs() - runtime_model = runtime_kwargs.pop("model", None) - if runtime_model: - logger.info( - "Runtime provider supplied explicit model override: %s -> %s", - model, - runtime_model, - ) - model = runtime_model - - cfg = getattr(self, "config", None) - if cfg and source is not None: - chat_id = str(source.chat_id) if source.chat_id else "" - thread_id = ( - str(source.thread_id) if getattr(source, "thread_id", None) else None - ) - parent_id = ( - str(source.parent_chat_id) - if getattr(source, "parent_chat_id", None) - else None - ) - ch = _get_channel_override( - cfg, - source.platform, - chat_id, - thread_id=thread_id, - parent_id=parent_id, - ) - if ch: - if ch.model: - model = ch.model - if ch.provider: - runtime_kwargs = _resolve_runtime_agent_kwargs_for_provider( - ch.provider - ) - ch_runtime_model = runtime_kwargs.pop("model", None) - # Only adopt the provider's bundled model when the override - # did not specify an explicit model. - if ch_runtime_model and not ch.model: - model = ch_runtime_model - - if override and resolved_session_key: - model, runtime_kwargs = self._apply_session_model_override( - resolved_session_key, model, runtime_kwargs - ) - - # When the config has no model.default but a provider was resolved - # (e.g. user ran `hermes auth add openai-codex` without `hermes model`), - # fall back to the provider's first catalog model so the API call - # doesn't fail with "model must be a non-empty string". - if not model and runtime_kwargs.get("provider"): - try: - from hermes_cli.models import get_default_model_for_provider - model = get_default_model_for_provider(runtime_kwargs["provider"]) - if model: - logger.info( - "No model configured — defaulting to %s for provider %s", - model, runtime_kwargs["provider"], - ) - except Exception: - pass - - # Final safety net (#35314): if resolution still produced an empty - # model — e.g. a transient config-cache miss during a post-interrupt - # recovery turn returned an empty user_config — reuse the last model we - # successfully resolved for this session (or, failing that, the most - # recent one resolved process-wide). Building an agent with model="" - # makes every API call fail HTTP 400 "No models provided" and the - # session goes silent until the user manually re-sends. ``getattr`` - # guards against bare test runners built via ``object.__new__``. - if not model: - _lr_state = ( - self._peek_session_state(resolved_session_key) - if resolved_session_key - else None - ) - _lr_star = self._peek_session_state("*") - _recovered = ( - (_lr_state.conversation.last_resolved_model if _lr_state else "") - or (_lr_star.conversation.last_resolved_model if _lr_star else "") - ) - if _recovered: - logger.warning( - "Empty model resolved for session=%s — recovering " - "last-known-good model %s (config read likely returned " - "empty; see #35314)", - resolved_session_key or "", _recovered, - ) - model = _recovered - elif model: - # Cache the good resolution for future recovery turns. - if resolved_session_key: - self._session_state( - resolved_session_key - ).conversation.last_resolved_model = model - self._session_state("*").conversation.last_resolved_model = model - - return model, runtime_kwargs - - def _resolve_turn_agent_config(self, user_message: str, model: str, runtime_kwargs: dict) -> dict: - """Build the effective model/runtime config for a single turn. - - Always uses the session's primary model/provider. If `/fast` is - enabled and the model supports Priority Processing / Anthropic fast - mode, attach `request_overrides` so the API call is marked - accordingly. - """ - from hermes_cli.models import resolve_fast_mode_overrides - - runtime = { - "api_key": runtime_kwargs.get("api_key"), - "base_url": runtime_kwargs.get("base_url"), - "provider": runtime_kwargs.get("provider"), - "requested_provider": runtime_kwargs.get("requested_provider"), - "api_mode": runtime_kwargs.get("api_mode"), - "command": runtime_kwargs.get("command"), - "args": list(runtime_kwargs.get("args") or []), - "credential_pool": runtime_kwargs.get("credential_pool"), - "max_tokens": runtime_kwargs.get("max_tokens"), - } - route = { - "model": model, - "runtime": runtime, - "signature": ( - model, - runtime["provider"], - runtime["requested_provider"], - runtime["base_url"], - runtime["api_mode"], - runtime["command"], - tuple(runtime["args"]), - ), - } - - service_tier = getattr(self, "_service_tier", None) - if not service_tier: - route["request_overrides"] = {} - return route - - try: - overrides = resolve_fast_mode_overrides(route["model"]) - except Exception: - overrides = None - route["request_overrides"] = overrides or {} - return route - - def _sync_session_model_from_agent(self, session_id: str, agent: Any) -> None: - """Persist the runtime model/provider actually used by a gateway turn. - Provider fallback can switch ``agent.model``/``agent.provider`` after the - session row was created. Keep the session DB metadata in sync so session - lists, desktop/dashboard details, and follow-up session tooling report the - backend that actually answered the latest turn. - Called from the ``run_sync`` closure, which executes off the event loop - in the executor thread — so the synchronous ``SessionDB`` (``_db``) is - used directly rather than awaiting the AsyncSessionDB forwarder. - """ - if not session_id or agent is None or self._session_db is None: - return - model = getattr(agent, "model", None) - if not model: - return - runtime = { - "provider": getattr(agent, "provider", None), - "base_url": getattr(agent, "base_url", None), - "api_mode": getattr(agent, "api_mode", None), - "fallback_active": bool(getattr(agent, "_fallback_activated", False)), - } - runtime = {k: v for k, v in runtime.items() if v not in (None, "")} - - try: - db = self._session_db._db - row = db.get_session(session_id) - if not row: - return - current_model = row.get("model") - raw_config = row.get("model_config") - try: - config = json.loads(raw_config) if raw_config else {} - except Exception: - config = {} - if not isinstance(config, dict): - config = {} - gateway_runtime = dict(config.get("gateway_runtime") or {}) - if current_model == model and all( - gateway_runtime.get(k) == v for k, v in runtime.items() - ): - return - config["gateway_runtime"] = runtime - db.update_session_meta(session_id, json.dumps(config), model=model) - except Exception: - logger.debug("Failed to sync gateway session model metadata", exc_info=True) async def _handle_reaction_event(self, ctx: Dict[str, Any]) -> None: """Fan a normalised platform reaction event out to the HookRegistry. @@ -7879,502 +7442,26 @@ def _resume_paused_platform(self, platform) -> bool: logger.info("%s resumed — retrying on next watcher tick", platform.value) return True - @staticmethod - def _load_prefill_messages() -> List[Dict[str, Any]]: - """Load ephemeral prefill messages from config or env var. - - Checks HERMES_PREFILL_MESSAGES_FILE env var first, then falls back to - the top-level prefill_messages_file key in ~/.hermes/config.yaml. - agent.prefill_messages_file is accepted as a legacy fallback. - Relative paths are resolved from ~/.hermes/. - """ - file_path = os.getenv("HERMES_PREFILL_MESSAGES_FILE", "") - if not file_path: - cfg = _load_gateway_runtime_config() - file_path = str(cfg.get("prefill_messages_file", "") or "") - if not file_path: - file_path = str(cfg_get(cfg, "agent", "prefill_messages_file", default="") or "") - if not file_path: - return [] - path = Path(file_path).expanduser() - if not path.is_absolute(): - path = _hermes_home / path - if not path.exists(): - logger.warning("Prefill messages file not found: %s", path) - return [] - try: - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - if not isinstance(data, list): - logger.warning("Prefill messages file must contain a JSON array: %s", path) - return [] - return data - except Exception as e: - logger.warning("Failed to load prefill messages from %s: %s", path, e) - return [] - - @staticmethod - def _load_ephemeral_system_prompt() -> str: - """Load ephemeral system prompt from config or env var. - - Checks HERMES_EPHEMERAL_SYSTEM_PROMPT env var first, then falls back to - agent.system_prompt in ~/.hermes/config.yaml. - """ - prompt = os.getenv("HERMES_EPHEMERAL_SYSTEM_PROMPT", "") - if prompt: - return prompt - cfg = _load_gateway_runtime_config() - return str(cfg_get(cfg, "agent", "system_prompt", default="") or "").strip() - - def _resolve_model_for_channel( - self, - platform: Platform, - chat_id: str, - *, - user_config: Optional[dict] = None, - thread_id: Optional[str] = None, - parent_id: Optional[str] = None, - ) -> str: - """Resolve model for this channel: channel_overrides else global default. - - Delegates the precedence rule to - :func:`hermes_cli.model_switch.resolve_effective_model` (session - override > channel override > global default) — the single owner - shared with the API server, so the two surfaces cannot diverge - again (see 7dd00bb47d). This call site has no session tier: session - /model overrides are applied later by - ``_apply_session_model_override`` on the resolved runtime. - """ - from hermes_cli.model_switch import resolve_effective_model - - override = None - config = getattr(self, "config", None) - if config: - override = _get_channel_override( - config, - platform, - chat_id, - thread_id=thread_id, - parent_id=parent_id, - ) - return resolve_effective_model( - None, # session tier applied downstream (_apply_session_model_override) - override, - _resolve_gateway_model(user_config), - ) - - def _get_system_prompt_for_channel( - self, - platform: Platform, - chat_id: str, - *, - thread_id: Optional[str] = None, - parent_id: Optional[str] = None, - ) -> str: - """Ephemeral system prompt for this channel/thread. - - Uses ``channel_overrides`` when set, else the global gateway prompt. - Legacy ``channel_prompts`` are applied separately via ``event.channel_prompt`` - in ``run_sync`` (adapter ``resolve_channel_prompt``), so they are not - duplicated here. - """ - config = getattr(self, "config", None) - if config: - override = _get_channel_override( - config, - platform, - chat_id, - thread_id=thread_id, - parent_id=parent_id, - ) - if override and override.system_prompt: - return (override.system_prompt or "").strip() - return getattr(self, "_ephemeral_system_prompt", None) or "" - - @staticmethod - def _load_reasoning_config(model: str = "") -> dict | None: - """Load reasoning effort from config.yaml, respecting per-model overrides. - - Thin wrapper over the shared chokepoint - :func:`hermes_constants.resolve_reasoning_config` (per-model override > - global ``agent.reasoning_effort``; YAML boolean False = disabled). - Closes #21256. - - Args: - model: The effective model for the calling session. When empty, - the config's ``model.default`` is used. - """ - from hermes_constants import resolve_reasoning_config - cfg = _load_gateway_runtime_config() - return resolve_reasoning_config(cfg, model) - - @staticmethod - def _parse_reasoning_command_args(raw_args: str) -> tuple[str, bool]: - """Parse `/reasoning` args into `(value, persist_global)`. - - `/reasoning ` is session-scoped by default. `--global` may be - supplied in any position to persist the change to config.yaml. - """ - import shlex - - text = str(raw_args or "").strip().replace("—", "--") - if not text: - return "", False - try: - tokens = shlex.split(text) - except ValueError: - tokens = text.split() - - persist_global = False - value_tokens = [] - for token in tokens: - if token == "--global": - persist_global = True - else: - value_tokens.append(token) - return " ".join(value_tokens).strip().lower(), persist_global - - def _resolve_session_reasoning_config( - self, - *, - source: Optional[SessionSource] = None, - session_key: Optional[str] = None, - model: str = "", - ) -> dict | None: - """Resolve reasoning effort for a session, honoring session overrides. - - Priority: session-scoped ``/reasoning --session`` override > - per-model override (``agent.reasoning_overrides``) > global - ``agent.reasoning_effort``. ``model`` should be the session's - *effective* model (session ``/model`` override included) so - per-model overrides track what the session actually runs — when - empty, the config's ``model.default`` is used. - """ - resolved_session_key = session_key - if not resolved_session_key and source is not None: - try: - resolved_session_key = self._session_key_for_source(source) - except Exception: - resolved_session_key = None - - if resolved_session_key: - _r_state = self._peek_session_state(resolved_session_key) - if _r_state is not None and _r_state.conversation.reasoning_override is not None: - return _r_state.conversation.reasoning_override - return self._load_reasoning_config(model) - - def _set_session_reasoning_override( - self, - session_key: str, - reasoning_config: Optional[dict], - ) -> None: - """Set or clear the session-scoped reasoning override.""" - if not session_key: - return - # Per-session field write — the old lazy ``self._session_reasoning_overrides - # = {}`` init replaced the WHOLE dict, racing concurrent sessions' - # overrides; a SessionState field reset cannot cross sessions. - self._session_state(session_key).conversation.reasoning_override = ( - None if reasoning_config is None else dict(reasoning_config) - ) - - def _resolve_session_service_tier( - self, - source=None, - session_key: Optional[str] = None, - ) -> Optional[str]: - """Resolve the effective service tier for a session. - A session-scoped /fast override wins over the config default. The - override dict stores "priority" or None (explicit normal), so key - presence — not value truthiness — decides whether it applies. - """ - resolved_session_key = session_key - if not resolved_session_key and source is not None: - try: - resolved_session_key = self._session_key_for_source(source) - except Exception: - resolved_session_key = None - if resolved_session_key: - _t_state = self._peek_session_state(resolved_session_key) - if ( - _t_state is not None - and _t_state.conversation.service_tier_override - is not _SERVICE_TIER_UNSET - ): - return _t_state.conversation.service_tier_override - return self._load_service_tier() - def _set_session_service_tier_override( - self, - session_key: str, - service_tier, - clear: bool = False, - ) -> None: - """Set or clear the session-scoped /fast override. - ``service_tier`` is "priority" or None (explicit normal). Pass - ``clear=True`` to remove the override entirely (fall back to config). - """ - if not session_key: - return - # Presence-sensitive: "priority" or None (explicit normal) both count - # as an override; the sentinel means "no override". Old code - # wholesale-replaced the dict on lazy init (cross-session race) — - # per-session field writes eliminate that class of bug. - self._session_state(session_key).conversation.service_tier_override = ( - _SERVICE_TIER_UNSET if clear else service_tier - ) - @staticmethod - def _load_service_tier() -> str | None: - """Load Priority Processing setting from config.yaml. - Reads agent.service_tier from config.yaml. Accepted values mirror the CLI: - "fast"/"priority"/"on" => "priority", while "normal"/"off" disables it. - Returns None when unset or unsupported. - """ - cfg = _load_gateway_runtime_config() - raw = str(cfg_get(cfg, "agent", "service_tier", default="") or "").strip() - value = raw.lower() - if not value or value in {"normal", "default", "standard", "off", "none"}: - return None - if value in {"fast", "priority", "on"}: - return "priority" - logger.warning("Unknown service_tier '%s', ignoring", raw) - return None - @staticmethod - def _load_show_reasoning() -> bool: - """Load show_reasoning toggle from config.yaml display section.""" - cfg = _load_gateway_runtime_config() - return is_truthy_value( - cfg_get(cfg, "display", "show_reasoning"), - default=False, - ) - @staticmethod - def _load_busy_input_mode() -> str: - """Load gateway drain-time busy-input behavior from config/env.""" - mode = os.getenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "").strip().lower() - if not mode: - cfg = _load_gateway_runtime_config() - mode = str(cfg_get(cfg, "display", "busy_input_mode", default="") or "").strip().lower() - if mode == "queue": - return "queue" - if mode == "steer": - return "steer" - return "interrupt" - @staticmethod - def _load_busy_text_mode() -> str: - """Resolve normal busy TEXT follow-up behavior. - - ``busy_input_mode`` is the single source of truth (default - ``interrupt``). The legacy ``busy_text_mode`` knob is honored only - when a user explicitly set it, so existing queue setups keep - working; new installs follow ``busy_input_mode``. Returns one of - ``interrupt`` | ``queue`` (``steer`` is handled upstream by - ``busy_input_mode`` and maps to non-queue text handling here). - """ - # Legacy explicit override wins for backward compat. - legacy = os.getenv("HERMES_GATEWAY_BUSY_TEXT_MODE", "").strip().lower() - if not legacy: - cfg = _load_gateway_runtime_config() - legacy = str(cfg_get(cfg, "display", "busy_text_mode", default="") or "").strip().lower() - if legacy == "interrupt": - return "interrupt" - if legacy == "queue": - return "queue" - # No explicit legacy knob → follow busy_input_mode. - input_mode = GatewayRunner._load_busy_input_mode() - return "queue" if input_mode == "queue" else "interrupt" - @staticmethod - def _load_restart_drain_timeout() -> float: - """Load graceful gateway restart/stop drain timeout in seconds.""" - raw = os.getenv("HERMES_RESTART_DRAIN_TIMEOUT", "").strip() - if not raw: - cfg = _load_gateway_runtime_config() - raw = str(cfg_get(cfg, "agent", "restart_drain_timeout", default="") or "").strip() - value = parse_restart_drain_timeout(raw) - if raw and value == DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT: - try: - float(raw) - except (TypeError, ValueError): - logger.warning( - "Invalid restart_drain_timeout '%s', using default %.0fs", - raw, - DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, - ) - return value - @staticmethod - def _load_restart_after_turn_timeout() -> float: - """Load in-band restart wait-for-idle timeout in seconds (#77184).""" - env_raw = os.getenv("HERMES_RESTART_AFTER_TURN_TIMEOUT") - if env_raw is not None and str(env_raw).strip() != "": - raw: object = env_raw - else: - cfg = _load_gateway_runtime_config() - raw = cfg_get(cfg, "agent", "restart_after_turn_timeout", default=None) - value = parse_restart_after_turn_timeout(raw) - # Warn only when the user supplied a non-empty value that failed to - # parse (parser falls back to the default). ``0`` is valid. - if raw is not None and str(raw).strip() != "": - try: - float(raw) - except (TypeError, ValueError): - logger.warning( - "Invalid restart_after_turn_timeout '%s', using default %.0fs", - raw, - DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT, - ) - return value - @staticmethod - def _load_background_notifications_mode() -> str: - """Load background process notification mode from config or env var. - - Modes: - - ``all`` — push running-output updates *and* the final message (default) - - ``result`` — only the final completion message (regardless of exit code) - - ``error`` — only the final message when exit code is non-zero - - ``off`` — no watcher messages at all - """ - mode = os.getenv("HERMES_BACKGROUND_NOTIFICATIONS", "") - if not mode: - cfg = _load_gateway_runtime_config() - raw = cfg_get(cfg, "display", "background_process_notifications") - if raw is False: - mode = "off" - elif raw not in {None, ""}: - mode = str(raw) - mode = (mode or "all").strip().lower() - valid = {"all", "result", "error", "off"} - if mode not in valid: - logger.warning( - "Unknown background_process_notifications '%s', defaulting to 'all'", - mode, - ) - return "all" - return mode - @staticmethod - def _load_provider_routing() -> dict: - """Load OpenRouter provider routing preferences from config.yaml.""" - try: - # Canonical gateway loader (fail-open): managed overlay + ${VAR} - # expansion now apply to provider_routing too. - cfg = _load_gateway_runtime_config() - return cfg.get("provider_routing", {}) or {} - except Exception: - pass - return {} - @staticmethod - def _load_fallback_model() -> list | None: - """Load fallback provider chain from config.yaml. - Returns the merged effective chain from ``fallback_providers`` plus any - legacy ``fallback_model`` entries. ``fallback_providers`` stays first - when both keys are present. - """ - try: - # Canonical gateway loader (fail-open): managed overlay + ${VAR} - # expansion now apply to the fallback chain too. - cfg = _load_gateway_runtime_config() - fb = get_fallback_chain(cfg) - if fb: - return fb - except Exception: - pass - return None - def _refresh_fallback_model(self) -> list | None: - """Re-read fallback_providers from disk for the next agent create/reuse. - Cron already does this per job via ``get_fallback_chain``; the gateway - previously froze ``self._fallback_model`` at process start, so a chain - configured (or changed) after ``hermes gateway`` was running never - reached messaging sessions even though the same process's cron jobs - fell back correctly. Fixes #60955. - A TRANSIENT read/parse failure (user mid-edit of config.yaml with a - non-atomic write) keeps the last known-good chain instead of wiping a - cached agent's working fallback for that turn. Only a successful read - that genuinely lacks the key clears the chain. - """ - try: - from hermes_cli.config import read_user_config_raw - cfg_path = _hermes_home / "config.yaml" - if not cfg_path.exists(): - self._fallback_model = None - return self._fallback_model - # Raw primitive (raises on parse failure) is required here: the - # canonical fail-open loader would return {} on a torn mid-edit - # write and WIPE the last known-good chain. The overlay/expansion - # below fixes the managed-scope/${VAR} drift without losing that. - cfg = read_user_config_raw(cfg_path) - try: - from hermes_cli import managed_scope - cfg = managed_scope.apply_managed_overlay(cfg) - except Exception: - pass - try: - from hermes_cli.config import _expand_env_vars - expanded = _expand_env_vars(cfg) - if isinstance(expanded, dict): - cfg = expanded - except Exception: - pass - except Exception: - # Transient failure — keep last known-good chain. - logger.debug( - "fallback_providers refresh: config.yaml read failed; " - "keeping last known-good chain", exc_info=True, - ) - return self._fallback_model - self._fallback_model = get_fallback_chain(cfg) or None - return self._fallback_model - @staticmethod - def _apply_fallback_chain_to_agent(agent: Any, chain: list | None) -> None: - """Keep a cached agent's fallback chain aligned with current config. - - Skips rewrite while a cooldown is holding the agent on an already- - activated fallback provider — ``restore_primary_runtime`` owns that - turn-scoped lifecycle. When primary is active (or cooldown expired), - replace the chain so mid-uptime ``fallback_providers`` edits take - effect without requiring a gateway restart (#60955). - """ - if agent is None: - return - new_chain = list(chain or []) - rate_limited_until = getattr(agent, "_rate_limited_until", 0) or 0 - if ( - getattr(agent, "_fallback_activated", False) - and rate_limited_until > time.monotonic() - ): - return - old_chain = list(getattr(agent, "_fallback_chain", []) or []) - agent._fallback_chain = new_chain - agent._fallback_model = new_chain[0] if new_chain else None - if not getattr(agent, "_fallback_activated", False): - agent._fallback_index = 0 - # A config edit signals the user changed something — drop the - # session-scoped unavailability memo so re-configured entries - # (e.g. credentials added mid-uptime for a previously-failing - # provider) get retried instead of staying suppressed for the - # cached agent's lifetime. Only on actual content change, so - # the per-message no-op refresh keeps the memo's rate-limiting - # benefit (#60955). - if new_chain != old_chain: - unavailable = getattr(agent, "_unavailable_fallback_keys", None) - if unavailable: - unavailable.clear() def _snapshot_running_agents(self) -> Dict[str, Any]: return { diff --git a/gateway/session_config_mixin.py b/gateway/session_config_mixin.py new file mode 100644 index 0000000000000..840acb04a2b00 --- /dev/null +++ b/gateway/session_config_mixin.py @@ -0,0 +1,831 @@ +"""Session-config resolution methods for ``GatewayRunner``. + +Extracted from ``gateway/run.py`` (god-file decomposition campaign, Wave 1, +shard s2, cluster c12). This mixin holds the session-config cluster: runtime +session agent kwargs resolution, per-channel model/system-prompt resolution, +reasoning-config and service-tier loaders, busy-input/text-mode knobs, restart +drain timeouts, provider routing, and the fallback-chain refresh path. + +Behavior-neutral: every method is lifted verbatim from ``GatewayRunner``. +``self.*`` calls resolve unchanged via the MRO. The module-level ``logger`` is +``logging.getLogger("gateway.run")`` so log records keep the exact name +(``"gateway.run"``), matching the sibling mixins' convention. run.py module +helpers/constants that stay behind (``_load_gateway_runtime_config``, +``_get_channel_override``, ``_resolve_gateway_model``, +``_resolve_runtime_agent_kwargs``, ``_resolve_runtime_agent_kwargs_for_provider``, +``_credential_pool_for_provider``, ``_hermes_home``, ``GatewayRunner``) are +imported lazily inside the method that uses them — a deferred +``from gateway.run import ...`` resolves at call time (run.py fully loaded by +then), so this module never imports ``gateway.run`` at import time -> no import +cycle. +""" + +from __future__ import annotations + +import json +import logging +import os +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +from gateway.config import Platform +from gateway.restart import ( + DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT, + DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, + parse_restart_after_turn_timeout, + parse_restart_drain_timeout, +) +from gateway.session import SessionSource +from gateway.session_state import SERVICE_TIER_UNSET as _SERVICE_TIER_UNSET +from hermes_cli.config import cfg_get +from hermes_cli.fallback_config import get_fallback_chain +from utils import is_truthy_value + +logger = logging.getLogger("gateway.run") + + +class SessionConfigMixin: + + def _resolve_session_agent_runtime( + self, + *, + source: Optional[SessionSource] = None, + session_key: Optional[str] = None, + user_config: Optional[dict] = None, + ) -> tuple[str, dict]: + """Resolve model/runtime for a session. + + Priority (highest first): session ``/model`` → ``channel_overrides`` → + global config/env (``_resolve_gateway_model(user_config)`` and default + provider resolution). + """ + from gateway.run import _credential_pool_for_provider + from gateway.run import _get_channel_override + from gateway.run import _resolve_gateway_model + from gateway.run import _resolve_runtime_agent_kwargs + from gateway.run import _resolve_runtime_agent_kwargs_for_provider + resolved_session_key = session_key + if not resolved_session_key and source is not None: + try: + resolved_session_key = self._session_key_for_source(source) + except Exception: + resolved_session_key = None + + model = _resolve_gateway_model(user_config) + if resolved_session_key: + self._rehydrate_session_model_override(resolved_session_key) + _override_state = ( + self._peek_session_state(resolved_session_key) + if resolved_session_key + else None + ) + override = ( + _override_state.conversation.model_override if _override_state else None + ) + if override: + override_model = override.get("model", model) + override_runtime = { + "provider": override.get("provider"), + "api_key": override.get("api_key"), + "base_url": override.get("base_url"), + "api_mode": override.get("api_mode"), + "max_tokens": override.get("max_tokens"), + "credential_pool": override.get("credential_pool"), + } + if override_runtime.get("api_key"): + if override_runtime.get("credential_pool") is None: + override_runtime["credential_pool"] = _credential_pool_for_provider( + override.get("provider") + ) + logger.debug( + "Session model override (fast): session=%s config_model=%s -> override_model=%s provider=%s", + resolved_session_key or "", model, override_model, + override_runtime.get("provider"), + ) + return override_model, override_runtime + # Override exists but has no api_key — fall through to env-based + # resolution and apply model/provider from the override on top. + logger.debug( + "Session model override (no api_key, fallback): session=%s config_model=%s override_model=%s", + resolved_session_key or "", model, override_model, + ) + else: + logger.debug( + "No session model override: session=%s config_model=%s override_keys=%s", + resolved_session_key or "", model, + [ + _key + for _key, _st in list(self._sessions_map().items()) + if _st.conversation.model_override is not None + ][:5] or "[]", + ) + + runtime_kwargs = _resolve_runtime_agent_kwargs() + runtime_model = runtime_kwargs.pop("model", None) + if runtime_model: + logger.info( + "Runtime provider supplied explicit model override: %s -> %s", + model, + runtime_model, + ) + model = runtime_model + + cfg = getattr(self, "config", None) + if cfg and source is not None: + chat_id = str(source.chat_id) if source.chat_id else "" + thread_id = ( + str(source.thread_id) if getattr(source, "thread_id", None) else None + ) + parent_id = ( + str(source.parent_chat_id) + if getattr(source, "parent_chat_id", None) + else None + ) + ch = _get_channel_override( + cfg, + source.platform, + chat_id, + thread_id=thread_id, + parent_id=parent_id, + ) + if ch: + if ch.model: + model = ch.model + if ch.provider: + runtime_kwargs = _resolve_runtime_agent_kwargs_for_provider( + ch.provider + ) + ch_runtime_model = runtime_kwargs.pop("model", None) + # Only adopt the provider's bundled model when the override + # did not specify an explicit model. + if ch_runtime_model and not ch.model: + model = ch_runtime_model + + if override and resolved_session_key: + model, runtime_kwargs = self._apply_session_model_override( + resolved_session_key, model, runtime_kwargs + ) + + # When the config has no model.default but a provider was resolved + # (e.g. user ran `hermes auth add openai-codex` without `hermes model`), + # fall back to the provider's first catalog model so the API call + # doesn't fail with "model must be a non-empty string". + if not model and runtime_kwargs.get("provider"): + try: + from hermes_cli.models import get_default_model_for_provider + model = get_default_model_for_provider(runtime_kwargs["provider"]) + if model: + logger.info( + "No model configured — defaulting to %s for provider %s", + model, runtime_kwargs["provider"], + ) + except Exception: + pass + + # Final safety net (#35314): if resolution still produced an empty + # model — e.g. a transient config-cache miss during a post-interrupt + # recovery turn returned an empty user_config — reuse the last model we + # successfully resolved for this session (or, failing that, the most + # recent one resolved process-wide). Building an agent with model="" + # makes every API call fail HTTP 400 "No models provided" and the + # session goes silent until the user manually re-sends. ``getattr`` + # guards against bare test runners built via ``object.__new__``. + if not model: + _lr_state = ( + self._peek_session_state(resolved_session_key) + if resolved_session_key + else None + ) + _lr_star = self._peek_session_state("*") + _recovered = ( + (_lr_state.conversation.last_resolved_model if _lr_state else "") + or (_lr_star.conversation.last_resolved_model if _lr_star else "") + ) + if _recovered: + logger.warning( + "Empty model resolved for session=%s — recovering " + "last-known-good model %s (config read likely returned " + "empty; see #35314)", + resolved_session_key or "", _recovered, + ) + model = _recovered + elif model: + # Cache the good resolution for future recovery turns. + if resolved_session_key: + self._session_state( + resolved_session_key + ).conversation.last_resolved_model = model + self._session_state("*").conversation.last_resolved_model = model + + return model, runtime_kwargs + + def _resolve_turn_agent_config(self, user_message: str, model: str, runtime_kwargs: dict) -> dict: + """Build the effective model/runtime config for a single turn. + + Always uses the session's primary model/provider. If `/fast` is + enabled and the model supports Priority Processing / Anthropic fast + mode, attach `request_overrides` so the API call is marked + accordingly. + """ + from hermes_cli.models import resolve_fast_mode_overrides + + runtime = { + "api_key": runtime_kwargs.get("api_key"), + "base_url": runtime_kwargs.get("base_url"), + "provider": runtime_kwargs.get("provider"), + "requested_provider": runtime_kwargs.get("requested_provider"), + "api_mode": runtime_kwargs.get("api_mode"), + "command": runtime_kwargs.get("command"), + "args": list(runtime_kwargs.get("args") or []), + "credential_pool": runtime_kwargs.get("credential_pool"), + "max_tokens": runtime_kwargs.get("max_tokens"), + } + route = { + "model": model, + "runtime": runtime, + "signature": ( + model, + runtime["provider"], + runtime["requested_provider"], + runtime["base_url"], + runtime["api_mode"], + runtime["command"], + tuple(runtime["args"]), + ), + } + + service_tier = getattr(self, "_service_tier", None) + if not service_tier: + route["request_overrides"] = {} + return route + + try: + overrides = resolve_fast_mode_overrides(route["model"]) + except Exception: + overrides = None + route["request_overrides"] = overrides or {} + return route + + def _sync_session_model_from_agent(self, session_id: str, agent: Any) -> None: + """Persist the runtime model/provider actually used by a gateway turn. + + Provider fallback can switch ``agent.model``/``agent.provider`` after the + session row was created. Keep the session DB metadata in sync so session + lists, desktop/dashboard details, and follow-up session tooling report the + backend that actually answered the latest turn. + + Called from the ``run_sync`` closure, which executes off the event loop + in the executor thread — so the synchronous ``SessionDB`` (``_db``) is + used directly rather than awaiting the AsyncSessionDB forwarder. + """ + if not session_id or agent is None or self._session_db is None: + return + model = getattr(agent, "model", None) + if not model: + return + runtime = { + "provider": getattr(agent, "provider", None), + "base_url": getattr(agent, "base_url", None), + "api_mode": getattr(agent, "api_mode", None), + "fallback_active": bool(getattr(agent, "_fallback_activated", False)), + } + runtime = {k: v for k, v in runtime.items() if v not in (None, "")} + + try: + db = self._session_db._db + row = db.get_session(session_id) + if not row: + return + current_model = row.get("model") + raw_config = row.get("model_config") + try: + config = json.loads(raw_config) if raw_config else {} + except Exception: + config = {} + if not isinstance(config, dict): + config = {} + gateway_runtime = dict(config.get("gateway_runtime") or {}) + if current_model == model and all( + gateway_runtime.get(k) == v for k, v in runtime.items() + ): + return + config["gateway_runtime"] = runtime + db.update_session_meta(session_id, json.dumps(config), model=model) + except Exception: + logger.debug("Failed to sync gateway session model metadata", exc_info=True) + + @staticmethod + def _load_prefill_messages() -> List[Dict[str, Any]]: + """Load ephemeral prefill messages from config or env var. + + Checks HERMES_PREFILL_MESSAGES_FILE env var first, then falls back to + the top-level prefill_messages_file key in ~/.hermes/config.yaml. + agent.prefill_messages_file is accepted as a legacy fallback. + Relative paths are resolved from ~/.hermes/. + """ + from gateway.run import _hermes_home + from gateway.run import _load_gateway_runtime_config + file_path = os.getenv("HERMES_PREFILL_MESSAGES_FILE", "") + if not file_path: + cfg = _load_gateway_runtime_config() + file_path = str(cfg.get("prefill_messages_file", "") or "") + if not file_path: + file_path = str(cfg_get(cfg, "agent", "prefill_messages_file", default="") or "") + if not file_path: + return [] + path = Path(file_path).expanduser() + if not path.is_absolute(): + path = _hermes_home / path + if not path.exists(): + logger.warning("Prefill messages file not found: %s", path) + return [] + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, list): + logger.warning("Prefill messages file must contain a JSON array: %s", path) + return [] + return data + except Exception as e: + logger.warning("Failed to load prefill messages from %s: %s", path, e) + return [] + + @staticmethod + def _load_ephemeral_system_prompt() -> str: + """Load ephemeral system prompt from config or env var. + + Checks HERMES_EPHEMERAL_SYSTEM_PROMPT env var first, then falls back to + agent.system_prompt in ~/.hermes/config.yaml. + """ + from gateway.run import _load_gateway_runtime_config + prompt = os.getenv("HERMES_EPHEMERAL_SYSTEM_PROMPT", "") + if prompt: + return prompt + cfg = _load_gateway_runtime_config() + return str(cfg_get(cfg, "agent", "system_prompt", default="") or "").strip() + + def _resolve_model_for_channel( + self, + platform: Platform, + chat_id: str, + *, + user_config: Optional[dict] = None, + thread_id: Optional[str] = None, + parent_id: Optional[str] = None, + ) -> str: + """Resolve model for this channel: channel_overrides else global default. + + Delegates the precedence rule to + :func:`hermes_cli.model_switch.resolve_effective_model` (session + override > channel override > global default) — the single owner + shared with the API server, so the two surfaces cannot diverge + again (see 7dd00bb47d). This call site has no session tier: session + /model overrides are applied later by + ``_apply_session_model_override`` on the resolved runtime. + """ + from gateway.run import _get_channel_override + from gateway.run import _resolve_gateway_model + from hermes_cli.model_switch import resolve_effective_model + + override = None + config = getattr(self, "config", None) + if config: + override = _get_channel_override( + config, + platform, + chat_id, + thread_id=thread_id, + parent_id=parent_id, + ) + return resolve_effective_model( + None, # session tier applied downstream (_apply_session_model_override) + override, + _resolve_gateway_model(user_config), + ) + + def _get_system_prompt_for_channel( + self, + platform: Platform, + chat_id: str, + *, + thread_id: Optional[str] = None, + parent_id: Optional[str] = None, + ) -> str: + """Ephemeral system prompt for this channel/thread. + + Uses ``channel_overrides`` when set, else the global gateway prompt. + Legacy ``channel_prompts`` are applied separately via ``event.channel_prompt`` + in ``run_sync`` (adapter ``resolve_channel_prompt``), so they are not + duplicated here. + """ + from gateway.run import _get_channel_override + config = getattr(self, "config", None) + if config: + override = _get_channel_override( + config, + platform, + chat_id, + thread_id=thread_id, + parent_id=parent_id, + ) + if override and override.system_prompt: + return (override.system_prompt or "").strip() + return getattr(self, "_ephemeral_system_prompt", None) or "" + + @staticmethod + def _load_reasoning_config(model: str = "") -> dict | None: + """Load reasoning effort from config.yaml, respecting per-model overrides. + + Thin wrapper over the shared chokepoint + :func:`hermes_constants.resolve_reasoning_config` (per-model override > + global ``agent.reasoning_effort``; YAML boolean False = disabled). + Closes #21256. + + Args: + model: The effective model for the calling session. When empty, + the config's ``model.default`` is used. + """ + from gateway.run import _load_gateway_runtime_config + from hermes_constants import resolve_reasoning_config + cfg = _load_gateway_runtime_config() + return resolve_reasoning_config(cfg, model) + + @staticmethod + def _parse_reasoning_command_args(raw_args: str) -> tuple[str, bool]: + """Parse `/reasoning` args into `(value, persist_global)`. + + `/reasoning ` is session-scoped by default. `--global` may be + supplied in any position to persist the change to config.yaml. + """ + import shlex + + text = str(raw_args or "").strip().replace("—", "--") + if not text: + return "", False + try: + tokens = shlex.split(text) + except ValueError: + tokens = text.split() + + persist_global = False + value_tokens = [] + for token in tokens: + if token == "--global": + persist_global = True + else: + value_tokens.append(token) + return " ".join(value_tokens).strip().lower(), persist_global + + def _resolve_session_reasoning_config( + self, + *, + source: Optional[SessionSource] = None, + session_key: Optional[str] = None, + model: str = "", + ) -> dict | None: + """Resolve reasoning effort for a session, honoring session overrides. + + Priority: session-scoped ``/reasoning --session`` override > + per-model override (``agent.reasoning_overrides``) > global + ``agent.reasoning_effort``. ``model`` should be the session's + *effective* model (session ``/model`` override included) so + per-model overrides track what the session actually runs — when + empty, the config's ``model.default`` is used. + """ + resolved_session_key = session_key + if not resolved_session_key and source is not None: + try: + resolved_session_key = self._session_key_for_source(source) + except Exception: + resolved_session_key = None + + if resolved_session_key: + _r_state = self._peek_session_state(resolved_session_key) + if _r_state is not None and _r_state.conversation.reasoning_override is not None: + return _r_state.conversation.reasoning_override + return self._load_reasoning_config(model) + + def _set_session_reasoning_override( + self, + session_key: str, + reasoning_config: Optional[dict], + ) -> None: + """Set or clear the session-scoped reasoning override.""" + if not session_key: + return + # Per-session field write — the old lazy ``self._session_reasoning_overrides + # = {}`` init replaced the WHOLE dict, racing concurrent sessions' + # overrides; a SessionState field reset cannot cross sessions. + self._session_state(session_key).conversation.reasoning_override = ( + None if reasoning_config is None else dict(reasoning_config) + ) + + def _resolve_session_service_tier( + self, + source=None, + session_key: Optional[str] = None, + ) -> Optional[str]: + """Resolve the effective service tier for a session. + + A session-scoped /fast override wins over the config default. The + override dict stores "priority" or None (explicit normal), so key + presence — not value truthiness — decides whether it applies. + """ + resolved_session_key = session_key + if not resolved_session_key and source is not None: + try: + resolved_session_key = self._session_key_for_source(source) + except Exception: + resolved_session_key = None + + if resolved_session_key: + _t_state = self._peek_session_state(resolved_session_key) + if ( + _t_state is not None + and _t_state.conversation.service_tier_override + is not _SERVICE_TIER_UNSET + ): + return _t_state.conversation.service_tier_override + return self._load_service_tier() + + def _set_session_service_tier_override( + self, + session_key: str, + service_tier, + clear: bool = False, + ) -> None: + """Set or clear the session-scoped /fast override. + + ``service_tier`` is "priority" or None (explicit normal). Pass + ``clear=True`` to remove the override entirely (fall back to config). + """ + if not session_key: + return + # Presence-sensitive: "priority" or None (explicit normal) both count + # as an override; the sentinel means "no override". Old code + # wholesale-replaced the dict on lazy init (cross-session race) — + # per-session field writes eliminate that class of bug. + self._session_state(session_key).conversation.service_tier_override = ( + _SERVICE_TIER_UNSET if clear else service_tier + ) + + @staticmethod + def _load_service_tier() -> str | None: + """Load Priority Processing setting from config.yaml. + + Reads agent.service_tier from config.yaml. Accepted values mirror the CLI: + "fast"/"priority"/"on" => "priority", while "normal"/"off" disables it. + Returns None when unset or unsupported. + """ + from gateway.run import _load_gateway_runtime_config + cfg = _load_gateway_runtime_config() + raw = str(cfg_get(cfg, "agent", "service_tier", default="") or "").strip() + + value = raw.lower() + if not value or value in {"normal", "default", "standard", "off", "none"}: + return None + if value in {"fast", "priority", "on"}: + return "priority" + logger.warning("Unknown service_tier '%s', ignoring", raw) + return None + + @staticmethod + def _load_show_reasoning() -> bool: + """Load show_reasoning toggle from config.yaml display section.""" + from gateway.run import _load_gateway_runtime_config + cfg = _load_gateway_runtime_config() + return is_truthy_value( + cfg_get(cfg, "display", "show_reasoning"), + default=False, + ) + + @staticmethod + def _load_busy_input_mode() -> str: + """Load gateway drain-time busy-input behavior from config/env.""" + from gateway.run import _load_gateway_runtime_config + mode = os.getenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "").strip().lower() + if not mode: + cfg = _load_gateway_runtime_config() + mode = str(cfg_get(cfg, "display", "busy_input_mode", default="") or "").strip().lower() + if mode == "queue": + return "queue" + if mode == "steer": + return "steer" + return "interrupt" + + @staticmethod + def _load_busy_text_mode() -> str: + """Resolve normal busy TEXT follow-up behavior. + + ``busy_input_mode`` is the single source of truth (default + ``interrupt``). The legacy ``busy_text_mode`` knob is honored only + when a user explicitly set it, so existing queue setups keep + working; new installs follow ``busy_input_mode``. Returns one of + ``interrupt`` | ``queue`` (``steer`` is handled upstream by + ``busy_input_mode`` and maps to non-queue text handling here). + """ + from gateway.run import _load_gateway_runtime_config + from gateway.run import GatewayRunner + # Legacy explicit override wins for backward compat. + legacy = os.getenv("HERMES_GATEWAY_BUSY_TEXT_MODE", "").strip().lower() + if not legacy: + cfg = _load_gateway_runtime_config() + legacy = str(cfg_get(cfg, "display", "busy_text_mode", default="") or "").strip().lower() + if legacy == "interrupt": + return "interrupt" + if legacy == "queue": + return "queue" + # No explicit legacy knob → follow busy_input_mode. + input_mode = GatewayRunner._load_busy_input_mode() + return "queue" if input_mode == "queue" else "interrupt" + + @staticmethod + def _load_restart_drain_timeout() -> float: + """Load graceful gateway restart/stop drain timeout in seconds.""" + from gateway.run import _load_gateway_runtime_config + raw = os.getenv("HERMES_RESTART_DRAIN_TIMEOUT", "").strip() + if not raw: + cfg = _load_gateway_runtime_config() + raw = str(cfg_get(cfg, "agent", "restart_drain_timeout", default="") or "").strip() + value = parse_restart_drain_timeout(raw) + if raw and value == DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT: + try: + float(raw) + except (TypeError, ValueError): + logger.warning( + "Invalid restart_drain_timeout '%s', using default %.0fs", + raw, + DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, + ) + return value + + @staticmethod + def _load_restart_after_turn_timeout() -> float: + """Load in-band restart wait-for-idle timeout in seconds (#77184).""" + from gateway.run import _load_gateway_runtime_config + env_raw = os.getenv("HERMES_RESTART_AFTER_TURN_TIMEOUT") + if env_raw is not None and str(env_raw).strip() != "": + raw: object = env_raw + else: + cfg = _load_gateway_runtime_config() + raw = cfg_get(cfg, "agent", "restart_after_turn_timeout", default=None) + value = parse_restart_after_turn_timeout(raw) + # Warn only when the user supplied a non-empty value that failed to + # parse (parser falls back to the default). ``0`` is valid. + if raw is not None and str(raw).strip() != "": + try: + float(raw) + except (TypeError, ValueError): + logger.warning( + "Invalid restart_after_turn_timeout '%s', using default %.0fs", + raw, + DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT, + ) + return value + + @staticmethod + def _load_background_notifications_mode() -> str: + """Load background process notification mode from config or env var. + + Modes: + - ``all`` — push running-output updates *and* the final message (default) + - ``result`` — only the final completion message (regardless of exit code) + - ``error`` — only the final message when exit code is non-zero + - ``off`` — no watcher messages at all + """ + from gateway.run import _load_gateway_runtime_config + mode = os.getenv("HERMES_BACKGROUND_NOTIFICATIONS", "") + if not mode: + cfg = _load_gateway_runtime_config() + raw = cfg_get(cfg, "display", "background_process_notifications") + if raw is False: + mode = "off" + elif raw not in {None, ""}: + mode = str(raw) + mode = (mode or "all").strip().lower() + valid = {"all", "result", "error", "off"} + if mode not in valid: + logger.warning( + "Unknown background_process_notifications '%s', defaulting to 'all'", + mode, + ) + return "all" + return mode + + @staticmethod + def _load_provider_routing() -> dict: + """Load OpenRouter provider routing preferences from config.yaml.""" + from gateway.run import _load_gateway_runtime_config + try: + # Canonical gateway loader (fail-open): managed overlay + ${VAR} + # expansion now apply to provider_routing too. + cfg = _load_gateway_runtime_config() + return cfg.get("provider_routing", {}) or {} + except Exception: + pass + return {} + + @staticmethod + def _load_fallback_model() -> list | None: + """Load fallback provider chain from config.yaml. + + Returns the merged effective chain from ``fallback_providers`` plus any + legacy ``fallback_model`` entries. ``fallback_providers`` stays first + when both keys are present. + """ + from gateway.run import _load_gateway_runtime_config + try: + # Canonical gateway loader (fail-open): managed overlay + ${VAR} + # expansion now apply to the fallback chain too. + cfg = _load_gateway_runtime_config() + fb = get_fallback_chain(cfg) + if fb: + return fb + except Exception: + pass + return None + + def _refresh_fallback_model(self) -> list | None: + """Re-read fallback_providers from disk for the next agent create/reuse. + + Cron already does this per job via ``get_fallback_chain``; the gateway + previously froze ``self._fallback_model`` at process start, so a chain + configured (or changed) after ``hermes gateway`` was running never + reached messaging sessions even though the same process's cron jobs + fell back correctly. Fixes #60955. + + A TRANSIENT read/parse failure (user mid-edit of config.yaml with a + non-atomic write) keeps the last known-good chain instead of wiping a + cached agent's working fallback for that turn. Only a successful read + that genuinely lacks the key clears the chain. + """ + from gateway.run import _hermes_home + try: + from hermes_cli.config import read_user_config_raw + cfg_path = _hermes_home / "config.yaml" + if not cfg_path.exists(): + self._fallback_model = None + return self._fallback_model + # Raw primitive (raises on parse failure) is required here: the + # canonical fail-open loader would return {} on a torn mid-edit + # write and WIPE the last known-good chain. The overlay/expansion + # below fixes the managed-scope/${VAR} drift without losing that. + cfg = read_user_config_raw(cfg_path) + try: + from hermes_cli import managed_scope + cfg = managed_scope.apply_managed_overlay(cfg) + except Exception: + pass + try: + from hermes_cli.config import _expand_env_vars + expanded = _expand_env_vars(cfg) + if isinstance(expanded, dict): + cfg = expanded + except Exception: + pass + except Exception: + # Transient failure — keep last known-good chain. + logger.debug( + "fallback_providers refresh: config.yaml read failed; " + "keeping last known-good chain", exc_info=True, + ) + return self._fallback_model + self._fallback_model = get_fallback_chain(cfg) or None + return self._fallback_model + + @staticmethod + def _apply_fallback_chain_to_agent(agent: Any, chain: list | None) -> None: + """Keep a cached agent's fallback chain aligned with current config. + + Skips rewrite while a cooldown is holding the agent on an already- + activated fallback provider — ``restore_primary_runtime`` owns that + turn-scoped lifecycle. When primary is active (or cooldown expired), + replace the chain so mid-uptime ``fallback_providers`` edits take + effect without requiring a gateway restart (#60955). + """ + if agent is None: + return + new_chain = list(chain or []) + rate_limited_until = getattr(agent, "_rate_limited_until", 0) or 0 + if ( + getattr(agent, "_fallback_activated", False) + and rate_limited_until > time.monotonic() + ): + return + old_chain = list(getattr(agent, "_fallback_chain", []) or []) + agent._fallback_chain = new_chain + agent._fallback_model = new_chain[0] if new_chain else None + if not getattr(agent, "_fallback_activated", False): + agent._fallback_index = 0 + # A config edit signals the user changed something — drop the + # session-scoped unavailability memo so re-configured entries + # (e.g. credentials added mid-uptime for a previously-failing + # provider) get retried instead of staying suppressed for the + # cached agent's lifetime. Only on actual content change, so + # the per-message no-op refresh keeps the memo's rate-limiting + # benefit (#60955). + if new_chain != old_chain: + unavailable = getattr(agent, "_unavailable_fallback_keys", None) + if unavailable: + unavailable.clear() diff --git a/gateway/telegram_topics_mixin.py b/gateway/telegram_topics_mixin.py new file mode 100644 index 0000000000000..52add22bba3e0 --- /dev/null +++ b/gateway/telegram_topics_mixin.py @@ -0,0 +1,216 @@ +"""Telegram topic-mode routing methods for ``GatewayRunner``. + +Extracted from ``gateway/run.py`` (god-file decomposition campaign, Wave 1, +shard s2, cluster c11). This mixin holds the Telegram DM topic cluster: +topic-mode enablement, lobby/lane classification, lobby reminders, topic +binding persistence and thread-id recovery. + +Behavior-neutral: every method is lifted verbatim from ``GatewayRunner``. +``self.*`` calls resolve unchanged via the MRO. The class attributes +``_TELEGRAM_GENERAL_TOPIC_IDS`` and ``_TELEGRAM_LOBBY_REMINDER_COOLDOWN_S`` +stay on ``GatewayRunner`` and are resolved through the MRO. The module-level +``logger`` is ``logging.getLogger("gateway.run")`` so log records keep the +exact name (``"gateway.run"``), matching the sibling mixins' convention. +""" + +from __future__ import annotations + +import logging +from typing import Optional + +from gateway.config import Platform +from gateway.session import SessionSource + +logger = logging.getLogger("gateway.run") + + +class TelegramTopicsMixin: + + def _telegram_topic_mode_enabled(self, source: SessionSource) -> bool: + """Return whether Telegram DM topic mode is active for this chat.""" + if source.platform != Platform.TELEGRAM or source.chat_type != "dm": + return False + session_db = getattr(self, "_session_db", None) + if session_db is None: + return False + # Runs off-loop (always via asyncio.to_thread); use the sync handle. + session_db = getattr(session_db, "_db", session_db) + try: + raw = session_db.is_telegram_topic_mode_enabled( + chat_id=str(source.chat_id), + user_id=str(source.user_id), + ) + except Exception: + logger.debug("Failed to read Telegram topic mode state", exc_info=True) + return False + # Only honor a real True from the SessionDB. Any other value + # (including MagicMock instances from test fixtures that didn't + # opt into topic mode) means topic mode is off for this chat. + return raw is True + + def _is_telegram_topic_root_lobby(self, source: SessionSource) -> bool: + """True for the main Telegram DM (or General topic) when topic mode has made it a lobby.""" + if source.platform != Platform.TELEGRAM or source.chat_type != "dm": + return False + if not self._telegram_topic_mode_enabled(source): + return False + tid = str(source.thread_id or "") + return tid in self._TELEGRAM_GENERAL_TOPIC_IDS + + def _is_telegram_topic_lane(self, source: SessionSource) -> bool: + """True for a user-created Telegram private-chat topic lane.""" + if source.platform != Platform.TELEGRAM or source.chat_type != "dm": + return False + if not self._telegram_topic_mode_enabled(source): + return False + tid = str(source.thread_id or "") + if not tid or tid in self._TELEGRAM_GENERAL_TOPIC_IDS: + return False + return True + + def _should_send_telegram_lobby_reminder(self, source: SessionSource) -> bool: + """Rate-limit root-DM lobby reminders to one message per cooldown window. + + A user who forgets multi-session mode is enabled and types several + prompts in the root DM would otherwise get a reminder for every + message. Cap it so the first one lands and the rest stay quiet. + """ + if not hasattr(self, "_telegram_lobby_reminder_ts"): + self._telegram_lobby_reminder_ts = {} + chat_id = str(source.chat_id or "") + if not chat_id: + return True + import time as _time + now = _time.monotonic() + last = self._telegram_lobby_reminder_ts.get(chat_id, 0.0) + if now - last < self._TELEGRAM_LOBBY_REMINDER_COOLDOWN_S: + return False + self._telegram_lobby_reminder_ts[chat_id] = now + return True + + def _telegram_topic_root_lobby_message(self) -> str: + return ( + "This main chat is reserved for system commands.\n\n" + "To start a new Hermes chat, open the All Messages topic at the top " + "of this bot interface and send any message there. Telegram will " + "create a new topic for that message; each topic works as an " + "independent Hermes session." + ) + + def _telegram_topic_root_new_message(self) -> str: + return ( + "To start a new parallel Hermes chat, open the All Messages topic " + "at the top of this bot interface and send any message there. " + "Telegram will create a new topic for it.\n\n" + "Each topic is an independent Hermes session. Use /new inside an " + "existing topic only if you want to replace that topic's current session." + ) + + def _telegram_topic_new_header(self, source: SessionSource) -> Optional[str]: + if not self._is_telegram_topic_lane(source): + return None + return ( + "Started a new Hermes session in this topic.\n\n" + "Tip: for parallel work, open All Messages and send a message there " + "to create a separate topic instead of using /new here. /new replaces " + "the session attached to the current topic." + ) + + def _record_telegram_topic_binding( + self, + source: SessionSource, + session_entry, + ) -> None: + """Persist the Telegram topic -> Hermes session binding for topic lanes.""" + session_db = getattr(self, "_session_db", None) + if session_db is None or not source.chat_id or not source.thread_id: + return + # Runs off-loop (always via asyncio.to_thread); use the sync handle. + session_db = getattr(session_db, "_db", session_db) + session_db.bind_telegram_topic( + chat_id=str(source.chat_id), + thread_id=str(source.thread_id), + user_id=str(source.user_id or ""), + session_key=session_entry.session_key, + session_id=session_entry.session_id, + ) + + def _sync_telegram_topic_binding( + self, + source: SessionSource, + session_entry, + *, + reason: str, + ) -> None: + """Update the topic binding to point at ``session_entry.session_id``. + + Telegram topic lanes persist a (chat_id, thread_id) -> session_id row + so reopening a topic in a fresh process resumes the right Hermes + session. When compression rotates ``session_entry.session_id`` mid-turn, + the binding goes stale and the next inbound message in that topic + reloads the oversized parent transcript instead of the compressed + child, retriggering preflight compression — sometimes in a loop + (#20470, #29712, #33414). + """ + if not self._is_telegram_topic_lane(source): + return + try: + self._record_telegram_topic_binding(source, session_entry) + except Exception: + logger.debug( + "telegram topic binding refresh failed (%s)", reason, exc_info=True, + ) + + def _recover_telegram_topic_thread_id( + self, + source: SessionSource, + ) -> Optional[str]: + """Pin DM-topic routing to the user's last-active topic. + + Telegram can omit ``message_thread_id`` or surface General (``1``) + for some topic-mode DM replies. In those lobby-shaped cases, keep the + conversation attached to the user's most-recent bound topic. + + Do not rewrite a non-lobby, previously-unbound thread id: a newly + created Telegram DM topic is also "unknown" until the first inbound + message is recorded, and rewriting it would send that brand-new topic's + answer into an older lane. Returns None to leave the source alone. + """ + if ( + source.platform != Platform.TELEGRAM + or source.chat_type != "dm" + or not source.chat_id + or not source.user_id + or not self._telegram_topic_mode_enabled(source) + ): + return None + inbound = str(source.thread_id or "") + is_lobby = not inbound or inbound in self._TELEGRAM_GENERAL_TOPIC_IDS + if not is_lobby: + # A non-lobby, unknown thread_id is most likely the first message in + # a brand-new Telegram DM topic. Preserve it so it can be recorded + # as a new independent lane below instead of hijacking the latest + # existing topic binding. + return None + session_db = getattr(self, "_session_db", None) + if session_db is None: + return None + # Runs off-loop (always via asyncio.to_thread); use the sync handle. + session_db = getattr(session_db, "_db", session_db) + try: + bindings = session_db.list_telegram_topic_bindings_for_chat( + chat_id=str(source.chat_id), + ) + except Exception: + logger.debug("topic-recover: read failed", exc_info=True) + return None + if not bindings: + return None + user_id = str(source.user_id) + for b in bindings: # newest-first + if str(b.get("user_id") or "") == user_id: + recovered = str(b.get("thread_id") or "") + if recovered and recovered != inbound: + return recovered + return None + return None diff --git a/tests/gateway/test_run_s2_session_config_and_topics.py b/tests/gateway/test_run_s2_session_config_and_topics.py new file mode 100644 index 0000000000000..a238c4ed3fe6d --- /dev/null +++ b/tests/gateway/test_run_s2_session_config_and_topics.py @@ -0,0 +1,293 @@ +"""Regression tests for the Wave-1 s2 mixin extraction (clusters c11 + c12). + +Methods moved verbatim from ``GatewayRunner`` (``gateway/run.py``) into: + +- ``gateway/telegram_topics_mixin.py`` (``TelegramTopicsMixin``, cluster c11): + Telegram DM topic-mode classification, lobby/lane routing, topic header and + lobby reminder message builders, reminder rate-limiting. +- ``gateway/session_config_mixin.py`` (``SessionConfigMixin``, cluster c12): + ``/reasoning`` arg parsing and the config loaders (busy input/text modes, + service tier, show_reasoning, restart drain/after-turn timeouts). + +These tests pin the PURE behavior of the moved methods so the extraction can +never silently change semantics: every assertion encodes the pre-extraction +behavior observed in ``gateway/run.py``. + +Test seam: bare mixin instances are built with ``object.__new__`` and the +class attributes they read (``_TELEGRAM_GENERAL_TOPIC_IDS``, +``_TELEGRAM_LOBBY_REMINDER_COOLDOWN_S``) attached as instance attributes — +those attributes intentionally stay on ``GatewayRunner`` and resolve via the +MRO in production. The lazy ``from gateway.run import +_load_gateway_runtime_config`` seam is monkeypatched through the +``gateway.run`` module attribute, and the mixin's ``cfg_get`` binding is +patched to read the fixture dict, making every loader deterministic. +""" + +from types import SimpleNamespace + +import pytest + +import gateway.run as gateway_run +import gateway.session_config_mixin as scc_mixin +from gateway.config import Platform +from gateway.restart import ( + DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT, + DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, +) +from gateway.session_config_mixin import SessionConfigMixin +from gateway.telegram_topics_mixin import TelegramTopicsMixin + +# Class attributes that stay on GatewayRunner (MRO-resolved in production) and +# are attached to bare mixin instances here — mirroring the real values. +_TELEGRAM_GENERAL_TOPIC_IDS = frozenset({"", "1"}) +_TELEGRAM_LOBBY_REMINDER_COOLDOWN_S = 300.0 + + +def _bare(cls, **attrs): + """Build a bare mixin instance with the given instance attributes.""" + inst = object.__new__(cls) + for key, value in attrs.items(): + setattr(inst, key, value) + return inst + + +def _telegram_mixin(*, topic_mode=True): + inst = _bare( + TelegramTopicsMixin, + _TELEGRAM_GENERAL_TOPIC_IDS=_TELEGRAM_GENERAL_TOPIC_IDS, + _TELEGRAM_LOBBY_REMINDER_COOLDOWN_S=_TELEGRAM_LOBBY_REMINDER_COOLDOWN_S, + ) + inst._telegram_topic_mode_enabled = lambda source: topic_mode + return inst + + +def _source(platform=Platform.TELEGRAM, chat_type="dm", chat_id="123", + user_id="456", thread_id=None): + return SimpleNamespace( + platform=platform, + chat_type=chat_type, + chat_id=chat_id, + user_id=user_id, + thread_id=thread_id, + ) + + +def _patch_config(monkeypatch, cfg): + """Point _load_gateway_runtime_config + the mixin's cfg_get at a fixture dict.""" + monkeypatch.setattr(gateway_run, "_load_gateway_runtime_config", lambda: cfg) + + def fake_cfg_get(cfg_dict, section, key, default=None): + return cfg_dict.get(section, {}).get(key, default) + + monkeypatch.setattr(scc_mixin, "cfg_get", fake_cfg_get) + + +class TestTelegramTopicClassification: + """Lobby/lane classification and topic message builders (cluster c11).""" + + def test_root_lobby_general_thread_ids(self): + inst = _telegram_mixin() + # General topic arrives as empty thread_id or "1" in some clients. + assert inst._is_telegram_topic_root_lobby(_source(thread_id=None)) is True + assert inst._is_telegram_topic_root_lobby(_source(thread_id="1")) is True + + def test_root_lobby_plain_dm_without_topic_mode(self): + # Topic mode off -> the DM is a normal chat, not a lobby. + inst = _telegram_mixin(topic_mode=False) + assert inst._is_telegram_topic_root_lobby(_source(thread_id=None)) is False + + def test_root_lobby_non_telegram_or_non_dm(self): + inst = _telegram_mixin() + assert inst._is_telegram_topic_root_lobby( + _source(platform=Platform.DISCORD, thread_id=None) + ) is False + assert inst._is_telegram_topic_root_lobby( + _source(chat_type="group", thread_id=None) + ) is False + + def test_root_lobby_unknown_thread_is_not_lobby(self): + inst = _telegram_mixin() + assert inst._is_telegram_topic_root_lobby(_source(thread_id="42")) is False + + def test_lane_user_created_topic(self): + inst = _telegram_mixin() + assert inst._is_telegram_topic_lane(_source(thread_id="42")) is True + + def test_lane_rejects_general_and_unknown(self): + inst = _telegram_mixin() + assert inst._is_telegram_topic_lane(_source(thread_id=None)) is False + assert inst._is_telegram_topic_lane(_source(thread_id="1")) is False + + def test_lane_requires_topic_mode(self): + inst = _telegram_mixin(topic_mode=False) + assert inst._is_telegram_topic_lane(_source(thread_id="42")) is False + + def test_new_header_only_for_lanes(self): + inst = _telegram_mixin() + header = inst._telegram_topic_new_header(_source(thread_id="42")) + assert header is not None + assert "Started a new Hermes session" in header + assert inst._telegram_topic_new_header(_source(thread_id="1")) is None + + def test_lobby_message_texts_mention_all_messages_topic(self): + inst = _telegram_mixin() + assert "All Messages topic" in inst._telegram_topic_root_lobby_message() + assert "All Messages topic" in inst._telegram_topic_root_new_message() + + def test_lobby_reminder_rate_limited(self): + inst = _telegram_mixin() + assert inst._should_send_telegram_lobby_reminder(_source(chat_id="777")) is True + # Immediately after, still inside the cooldown window -> suppressed. + assert inst._should_send_telegram_lobby_reminder(_source(chat_id="777")) is False + # A different chat is not affected by the first chat's cooldown. + assert inst._should_send_telegram_lobby_reminder(_source(chat_id="888")) is True + + def test_lobby_reminder_without_chat_id(self): + inst = _telegram_mixin() + assert inst._should_send_telegram_lobby_reminder( + _source(chat_id="") + ) is True + + +class TestParseReasoningCommandArgs: + """/reasoning arg parsing (cluster c12) — pure, static.""" + + def test_empty_input(self): + assert SessionConfigMixin._parse_reasoning_command_args(None) == ("", False) + assert SessionConfigMixin._parse_reasoning_command_args("") == ("", False) + + def test_simple_value(self): + assert SessionConfigMixin._parse_reasoning_command_args("fast") == ("fast", False) + + def test_value_lowercased(self): + assert SessionConfigMixin._parse_reasoning_command_args("FAST") == ("fast", False) + + def test_global_any_position(self): + assert SessionConfigMixin._parse_reasoning_command_args("--global fast") == ("fast", True) + assert SessionConfigMixin._parse_reasoning_command_args("fast --global") == ("fast", True) + + def test_quoted_value(self): + assert SessionConfigMixin._parse_reasoning_command_args( + '"high detail"' + ) == ("high detail", False) + + def test_em_dash_normalized(self): + # Unicode em-dash is normalized to -- so it never becomes a value token. + value, persist = SessionConfigMixin._parse_reasoning_command_args("auto —off") + assert persist is False + assert "--off" in value + + def test_extra_tokens_kept(self): + value, persist = SessionConfigMixin._parse_reasoning_command_args( + "fast --global extra" + ) + assert value == "fast extra" + assert persist is True + + +class TestSessionConfigLoaders: + """Deterministic config loaders via the monkeypatched runtime-config seam.""" + + def test_busy_input_mode_env_wins(self, monkeypatch): + _patch_config(monkeypatch, {"display": {}}) + monkeypatch.setenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "queue") + assert SessionConfigMixin._load_busy_input_mode() == "queue" + monkeypatch.setenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "steer") + assert SessionConfigMixin._load_busy_input_mode() == "steer" + monkeypatch.setenv("HERMES_GATEWAY_BUSY_INPUT_MODE", "bogus") + assert SessionConfigMixin._load_busy_input_mode() == "interrupt" + + def test_busy_input_mode_from_config(self, monkeypatch): + monkeypatch.delenv("HERMES_GATEWAY_BUSY_INPUT_MODE", raising=False) + _patch_config(monkeypatch, {"display": {"busy_input_mode": "steer"}}) + assert SessionConfigMixin._load_busy_input_mode() == "steer" + + def test_busy_input_mode_default(self, monkeypatch): + monkeypatch.delenv("HERMES_GATEWAY_BUSY_INPUT_MODE", raising=False) + _patch_config(monkeypatch, {"display": {}}) + assert SessionConfigMixin._load_busy_input_mode() == "interrupt" + + def test_busy_text_mode_config(self, monkeypatch): + monkeypatch.delenv("HERMES_GATEWAY_BUSY_TEXT_MODE", raising=False) + _patch_config(monkeypatch, {"display": {"busy_text_mode": "queue"}}) + assert SessionConfigMixin._load_busy_text_mode() == "queue" + + def test_busy_text_mode_falls_back_to_busy_input_mode(self, monkeypatch): + monkeypatch.delenv("HERMES_GATEWAY_BUSY_TEXT_MODE", raising=False) + _patch_config(monkeypatch, {"display": {}}) + # busy_input_mode resolves to "interrupt" -> text mode follows. + assert SessionConfigMixin._load_busy_text_mode() == "interrupt" + + def test_show_reasoning(self, monkeypatch): + _patch_config(monkeypatch, {"display": {"show_reasoning": "true"}}) + assert SessionConfigMixin._load_show_reasoning() is True + _patch_config(monkeypatch, {"display": {"show_reasoning": "false"}}) + assert SessionConfigMixin._load_show_reasoning() is False + _patch_config(monkeypatch, {"display": {}}) + assert SessionConfigMixin._load_show_reasoning() is False + + def test_service_tier_mapping(self, monkeypatch): + for raw, expected in [ + ("fast", "priority"), + ("priority", "priority"), + ("on", "priority"), + ("normal", None), + ("off", None), + ("", None), + ("bogus", None), + ]: + _patch_config(monkeypatch, {"agent": {"service_tier": raw}}) + assert SessionConfigMixin._load_service_tier() == expected, raw + + def test_restart_drain_timeout_env_and_default(self, monkeypatch): + monkeypatch.setenv("HERMES_RESTART_DRAIN_TIMEOUT", "42") + assert SessionConfigMixin._load_restart_drain_timeout() == 42.0 + monkeypatch.delenv("HERMES_RESTART_DRAIN_TIMEOUT", raising=False) + _patch_config(monkeypatch, {"agent": {"restart_drain_timeout": "30"}}) + assert SessionConfigMixin._load_restart_drain_timeout() == 30.0 + _patch_config(monkeypatch, {"agent": {}}) + assert ( + SessionConfigMixin._load_restart_drain_timeout() + == DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT + ) + + def test_restart_after_turn_timeout_env_and_default(self, monkeypatch): + monkeypatch.setenv("HERMES_RESTART_AFTER_TURN_TIMEOUT", "60") + assert SessionConfigMixin._load_restart_after_turn_timeout() == 60.0 + monkeypatch.delenv("HERMES_RESTART_AFTER_TURN_TIMEOUT", raising=False) + _patch_config(monkeypatch, {"agent": {"restart_after_turn_timeout": 15}}) + assert SessionConfigMixin._load_restart_after_turn_timeout() == 15.0 + _patch_config(monkeypatch, {"agent": {}}) + assert ( + SessionConfigMixin._load_restart_after_turn_timeout() + == DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT + ) + + +class TestWiringSmoke: + """The extraction must leave GatewayRunner fully wired via the mixins.""" + + def test_gateway_runner_still_exposes_moved_methods(self): + assert hasattr(gateway_run.GatewayRunner, "_load_busy_input_mode") + assert hasattr(gateway_run.GatewayRunner, "_parse_reasoning_command_args") + assert hasattr(gateway_run.GatewayRunner, "_is_telegram_topic_lane") + assert hasattr(gateway_run.GatewayRunner, "_recover_telegram_topic_thread_id") + + def test_mixins_in_mro_and_class_attrs_stay(self): + mro = gateway_run.GatewayRunner.__mro__ + assert TelegramTopicsMixin in mro + assert SessionConfigMixin in mro + assert gateway_run.GatewayRunner._TELEGRAM_GENERAL_TOPIC_IDS == frozenset({"", "1"}) + + def test_mixin_bare_instance_behavior_matches_gateway_runner(self): + # Same inputs -> same outputs whether dispatched on GatewayRunner or the + # bare mixin: proves the MRO move is behavior-neutral for pure methods. + runner = object.__new__(gateway_run.GatewayRunner) + runner._telegram_topic_mode_enabled = lambda source: True + runner._TELEGRAM_GENERAL_TOPIC_IDS = frozenset({"", "1"}) + runner._TELEGRAM_LOBBY_REMINDER_COOLDOWN_S = 300.0 + source = _source(thread_id="42") + assert runner._is_telegram_topic_lane(source) is True + assert TelegramTopicsMixin._is_telegram_topic_lane( + _telegram_mixin(), source + ) is True From bf2f7bd69000eeaece0d0a673726eecc827c32e6 Mon Sep 17 00:00:00 2001 From: "Andrex Ibiza, MBA" <84248988+andrexibiza@users.noreply.github.com> Date: Wed, 5 Aug 2026 11:51:58 -0500 Subject: [PATCH 2/2] test(gateway): make lobby-reminder rate-limit test boot-time independent (monotonic seed) --- tests/gateway/test_run_s2_session_config_and_topics.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/gateway/test_run_s2_session_config_and_topics.py b/tests/gateway/test_run_s2_session_config_and_topics.py index a238c4ed3fe6d..15fca8c3815a4 100644 --- a/tests/gateway/test_run_s2_session_config_and_topics.py +++ b/tests/gateway/test_run_s2_session_config_and_topics.py @@ -136,6 +136,14 @@ def test_lobby_message_texts_mention_all_messages_topic(self): def test_lobby_reminder_rate_limited(self): inst = _telegram_mixin() + import time as _time + # monotonic() is boot-relative (fresh CI containers start near 0), so + # seed an old baseline per chat to make the "first reminder allowed" + # assertion independent of how long the machine has been up. + inst._telegram_lobby_reminder_ts = { + "777": _time.monotonic() - 1000.0, + "888": _time.monotonic() - 1000.0, + } assert inst._should_send_telegram_lobby_reminder(_source(chat_id="777")) is True # Immediately after, still inside the cooldown window -> suppressed. assert inst._should_send_telegram_lobby_reminder(_source(chat_id="777")) is False