mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
unsupportedpastels
b51e4272ea
Platform.LOCAL is the canonical transport, not the owner: hosted rooms (bot_room) and A2A peers stage LOCAL sessions too, so the platform gate still sent Daybreak on their turns. The gateway now reads the route's frozen launch policy and allows only the owner's own creation labels (cli, tui, gui, oneshot; gateway.session_policy.DAYBREAK_LABELS). Third- party tool integrations, editor (ACP) clients and an unreadable policy get standard requests. With turn isolation the live agent is in the compute-host child, which read the switch once at build; the parent's live flip only reaches its own process. The child now re-reads it from its profile's config.yaml on every reused turn, with the same local-surface gate.
650 lines
34 KiB
Python
650 lines
34 KiB
Python
"""Config/env loaders (busy modes, reasoning, service tier, timeouts, fallback) for GatewayRunner.
|
|
|
|
Split out of ``gateway/run.py``; bound onto ``GatewayRunner`` via the MRO.
|
|
``gateway.run`` internals are imported lazily inside method bodies (import cycle),
|
|
so ``patch("gateway.run.X")`` keeps intercepting them at call time.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from contextlib import suppress
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any, Dict, List, Optional
|
|
|
|
from gateway.config import Platform
|
|
from gateway.restart import (
|
|
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT, DEFAULT_GATEWAY_POST_INTERRUPT_GRACE_TIMEOUT,
|
|
DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT, DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
|
|
DEFAULT_GATEWAY_SIGNAL_INTERRUPT_GRACE_TIMEOUT, parse_cron_drain_timeout,
|
|
parse_restart_after_turn_timeout, parse_restart_drain_timeout,
|
|
launchd_service_label, parse_signal_interrupt_grace_timeout, read_launchd_exit_timeout_s,
|
|
resolve_launchd_capped_drain,
|
|
)
|
|
from gateway.session import SessionSource
|
|
from gateway.session_state import SERVICE_TIER_UNSET as _SERVICE_TIER_UNSET
|
|
from hermes_cli.config import cfg_get, resolve_ephemeral_system_prompt_from_config
|
|
from hermes_cli.fallback_config import get_fallback_chain
|
|
from utils import is_truthy_value
|
|
|
|
if TYPE_CHECKING: # string annotations only; never imported at runtime (cycle)
|
|
from gateway.run import GatewayRunner
|
|
from gateway.run_turn_runner import TurnRunner
|
|
|
|
# Log-record parity with the origin module.
|
|
logger = logging.getLogger("gateway.run")
|
|
|
|
DEFAULT_HUMAN_DELAY_RANGE_MS: tuple[int, int] = (800, 2500)
|
|
|
|
_BUSY_INPUT_MODES = {"interrupt", "queue", "steer"}
|
|
|
|
|
|
class GatewayConfigLoadersMixin:
|
|
"""Config/env loaders (busy modes, reasoning, service tier, timeouts, fallback) for GatewayRunner."""
|
|
|
|
@staticmethod
|
|
def _cfg_str(section: str, key: str) -> str:
|
|
"""``<section>.<key>`` from the gateway runtime config as a stripped string ("" when unset)."""
|
|
from gateway.run import _load_gateway_config
|
|
return str(cfg_get(_load_gateway_config(), section, key, default="") or "").strip()
|
|
|
|
@classmethod
|
|
def _env_or_cfg_str(cls, env_var: str, section: str, key: str) -> str:
|
|
"""Non-empty stripped env var, else :meth:`_cfg_str`."""
|
|
return os.getenv(env_var, "").strip() or cls._cfg_str(section, key)
|
|
|
|
@staticmethod
|
|
def _load_prefill_messages() -> list[dict[str, Any]]:
|
|
"""Load ephemeral prefill messages from config or env var.
|
|
|
|
HERMES_PREFILL_MESSAGES_FILE env wins, then top-level prefill_messages_file in config.yaml,
|
|
then legacy agent.prefill_messages_file. Relative paths resolve from ~/.hermes/.
|
|
"""
|
|
from gateway.run import _gateway_config_home, _load_gateway_config
|
|
file_path = os.getenv("HERMES_PREFILL_MESSAGES_FILE", "")
|
|
if not file_path:
|
|
cfg = _load_gateway_config()
|
|
file_path = str(
|
|
cfg.get("prefill_messages_file", "") or 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 = _gateway_config_home() / path
|
|
if not path.exists():
|
|
logger.warning("Prefill messages file not found: %s", path)
|
|
return []
|
|
try:
|
|
with open(path, "r", encoding="utf-8-sig") 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:
|
|
"""HERMES_EPHEMERAL_SYSTEM_PROMPT env first (a local session's: its client's), then
|
|
``display.personality`` / ``agent.system_prompt``."""
|
|
from agent.session_client_env import session_env
|
|
from gateway.run import _load_gateway_config
|
|
prompt = session_env("HERMES_EPHEMERAL_SYSTEM_PROMPT")
|
|
if prompt:
|
|
return prompt
|
|
return resolve_ephemeral_system_prompt_from_config(_load_gateway_config())
|
|
|
|
def _channel_override(self, platform: Platform, chat_id: str, thread_id, parent_id):
|
|
"""``channel_overrides`` entry for this channel/thread, or None (also when no config is bound)."""
|
|
from gateway.run import _get_channel_override
|
|
config = getattr(self, "config", None)
|
|
if not config:
|
|
return None
|
|
return _get_channel_override(config, platform, chat_id, thread_id=thread_id, parent_id=parent_id)
|
|
|
|
def _silence_allowed_for_source(self, source, display_kind, reply_expected, agent_result) -> bool:
|
|
"""Evaluate existing exemptions before consulting the optional room permission."""
|
|
from gateway.response_filters import silence_allowed
|
|
return silence_allowed(display_kind, reply_expected) or silence_allowed(
|
|
display_kind, reply_expected, allow_silence=self._room_allow_silence(source),
|
|
agent_result=agent_result)
|
|
|
|
def _room_allow_silence(self, source: SessionSource) -> Optional[bool]:
|
|
"""Resolve the room policy in the serving profile, never the launch snapshot for a route."""
|
|
from gateway.config import GatewayConfig
|
|
from gateway.config_loader import merge_platform_sections, platform_section
|
|
from gateway.run import _get_channel_override, _load_gateway_config
|
|
from gateway.session_identity import identity_of
|
|
config = getattr(self, "config", None)
|
|
identity = identity_of(source)
|
|
runtime = identity.runtime_profile if identity is not None else (source.profile or None)
|
|
# ``self.config`` (including a ``gateway --config`` file) serves the primary profile; only a
|
|
# turn routed to another profile reads that profile's config.yaml.
|
|
routed = bool(runtime) and runtime != (getattr(self, "_primary_profile_name", None) or "default")
|
|
try:
|
|
if config is None or routed:
|
|
raw = _load_gateway_config(self._resolve_profile_home_for_source(source) / "config.yaml")
|
|
# The loader's own two steps: merge every nested platform block, then the overrides of
|
|
# the block platform_section picks (a top-level ``<platform>:`` block first) win.
|
|
# ``GatewayConfig.from_dict(raw)`` alone reads only the nested ``platforms`` form.
|
|
name, gateway_section = str(source.platform.value), raw.get("gateway")
|
|
overrides = (merge_platform_sections(raw, gateway_section, {}).get(name) or {}).get("channel_overrides")
|
|
section, _ = platform_section(raw, name, (
|
|
gateway_section.get("platforms") if isinstance(gateway_section, dict) else None))
|
|
if isinstance(section, dict) and isinstance(section.get("channel_overrides"), dict):
|
|
overrides = section["channel_overrides"]
|
|
config = GatewayConfig.from_dict({"platforms": {name: {
|
|
"channel_overrides": overrides if isinstance(overrides, dict) else {}}}})
|
|
override = _get_channel_override(
|
|
config, source.platform, source.chat_id,
|
|
thread_id=source.thread_id, parent_id=source.parent_chat_id)
|
|
except Exception:
|
|
# No permission means the default reply rule; a lookup failure must not become an error reply.
|
|
logger.warning("Room allow_silence lookup failed for %s; keeping the reply rule",
|
|
source.chat_id, exc_info=True)
|
|
return None
|
|
return override.allow_silence if override is not None else None
|
|
|
|
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.
|
|
|
|
Precedence lives in :func:`hermes_cli.model_switch.resolve_effective_model` (shared with the
|
|
API server so the surfaces cannot diverge). No session tier here: session /model overrides
|
|
are applied later by ``_apply_session_model_override``.
|
|
"""
|
|
from gateway.run import _resolve_gateway_model
|
|
from hermes_cli.model_switch import resolve_effective_model
|
|
return resolve_effective_model(
|
|
None, # session tier applied downstream (_apply_session_model_override)
|
|
self._channel_override(platform, chat_id, thread_id, parent_id),
|
|
_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.
|
|
|
|
``channel_overrides`` when set, else the gateway prompt resolved from the CURRENT profile's
|
|
config on every call (callers run inside ``_profile_runtime_scope``, so routed multiplex
|
|
profiles get their own personality/system_prompt and ``/personality`` edits apply next turn).
|
|
Legacy ``channel_prompts`` are applied separately via ``event.channel_prompt`` in ``run_sync``.
|
|
|
|
Callers run inside ``_profile_runtime_scope`` (``run_sync`` under ``_run_agent``), so a routed
|
|
multiplex profile gets its own ``display.personality`` / ``agent.system_prompt`` instead of a
|
|
boot-time snapshot of the launch profile's (#89161); ``/personality`` edits take effect on the next
|
|
turn for the same reason.
|
|
"""
|
|
override = self._channel_override(platform, chat_id, thread_id, parent_id)
|
|
if override and override.system_prompt:
|
|
return (override.system_prompt or "").strip()
|
|
return self._load_ephemeral_system_prompt()
|
|
|
|
@staticmethod
|
|
def _load_reasoning_config(model: str = "") -> dict | None:
|
|
"""Reasoning effort from config.yaml via :func:`hermes_constants.resolve_reasoning_config`.
|
|
|
|
Per-model override > global ``agent.reasoning_effort``; YAML False = disabled. Empty
|
|
``model`` uses ``model.default``.
|
|
|
|
Closes #21256.
|
|
"""
|
|
from gateway.run import _load_gateway_config
|
|
from hermes_constants import resolve_reasoning_config
|
|
return resolve_reasoning_config(_load_gateway_config(), model)
|
|
|
|
@staticmethod
|
|
def _parse_reasoning_command_args(raw_args: str) -> tuple[str, bool]:
|
|
"""Parse `/reasoning` args into `(value, persist_global)`; `--global` anywhere persists to config."""
|
|
import shlex
|
|
text = str(raw_args or "").strip().replace("—", "--")
|
|
if not text:
|
|
return "", False
|
|
try:
|
|
tokens = shlex.split(text)
|
|
except ValueError:
|
|
tokens = text.split()
|
|
value = " ".join(token for token in tokens if token != "--global")
|
|
return value.strip().lower(), "--global" in tokens
|
|
|
|
def _resolve_session_reasoning_config(
|
|
self, *, source: Optional[SessionSource] = None, session_key: Optional[str] = None,
|
|
model: str = "",
|
|
) -> dict | None:
|
|
"""Session ``/reasoning --session`` > per-model ``agent.reasoning_overrides`` > global.
|
|
|
|
``model`` must be the session's *effective* model (session ``/model`` override included);
|
|
empty uses ``model.default``.
|
|
"""
|
|
resolved_session_key = self._resolve_session_key_or_none(source, session_key)
|
|
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
|
|
from gateway.session_setting_commands import local_policy
|
|
if (policy := local_policy(self, source)) is not None:
|
|
# A local session runs its frozen (or followed) level, not the profile's.
|
|
return policy.current_reasoning_config(self, resolved_session_key)
|
|
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: a lazy ``_session_reasoning_overrides = {}`` init replaced the
|
|
# WHOLE dict, racing concurrent sessions; 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]:
|
|
"""Effective service tier: a session-scoped /fast override beats the config default.
|
|
|
|
The override stores "priority" or None (explicit normal), so presence — not truthiness — decides.
|
|
"""
|
|
resolved_session_key = self._resolve_session_key_or_none(source, session_key)
|
|
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
|
|
from gateway.session_setting_commands import local_live_for_route, local_policy
|
|
live = local_live_for_route(self, resolved_session_key)
|
|
if live is not None and (policy := local_policy(self, live.source)) is not None:
|
|
return policy.current_service_tier(self, resolved_session_key) # its frozen tier, not config.yaml's
|
|
return self._load_service_tier()
|
|
|
|
def _set_session_service_tier_override(self, session_key: str, service_tier, clear: bool = False) -> None:
|
|
"""Set ("priority" / None = explicit normal) or ``clear`` the session-scoped /fast override."""
|
|
if not session_key:
|
|
return
|
|
# Presence-sensitive: "priority" or None (explicit normal) both count as an override; the
|
|
# sentinel means "no override". Per-session field write: a lazy dict replace races sessions.
|
|
self._session_state(session_key).conversation.service_tier_override = (
|
|
_SERVICE_TIER_UNSET if clear else service_tier
|
|
)
|
|
|
|
@classmethod
|
|
def _load_service_tier(cls) -> str | None:
|
|
"""``agent.service_tier`` parsed like the CLI (``hermes_cli.cli_config_load``); None when unset/unknown."""
|
|
from hermes_cli.cli_config_load import _parse_service_tier_config
|
|
|
|
return _parse_service_tier_config(cls._cfg_str("agent", "service_tier"))
|
|
|
|
def _load_daybreak(self, source) -> bool:
|
|
"""``agent.daybreak`` (off unless explicitly true) for a turn from ``source``. Read per turn so the
|
|
Desktop switch reaches the next turn of a running chat without rebuilding its cached agent. Only the
|
|
owner's own chats carry it: Daybreak is the approved account's own access (OpenAI: no third-party
|
|
or downstream use), and a messaging platform's chats, a hosted room or an A2A peer can be other people.
|
|
LOCAL is the canonical transport, not the owner, so the route's frozen launch policy decides."""
|
|
from gateway.run import _load_gateway_config
|
|
from gateway.session_policy import DAYBREAK_LABELS, policy_for_source
|
|
if getattr(source, "platform", None) != Platform.LOCAL:
|
|
return False
|
|
try:
|
|
policy = policy_for_source(self, source)
|
|
except Exception: # health: allow BLE001 -- an unreadable policy fails closed to standard requests
|
|
return False
|
|
if getattr(policy, "source", None) not in DAYBREAK_LABELS:
|
|
return False
|
|
return cfg_get(_load_gateway_config(), "agent", "daybreak") is True
|
|
|
|
@staticmethod
|
|
def _load_show_reasoning() -> bool:
|
|
"""``display.show_reasoning`` toggle."""
|
|
from gateway.run import _load_gateway_config
|
|
return is_truthy_value(cfg_get(_load_gateway_config(), "display", "show_reasoning"), default=False)
|
|
|
|
@classmethod
|
|
def _load_busy_input_mode(cls) -> str:
|
|
"""Gateway drain-time busy-input behavior from env/config (default ``interrupt``)."""
|
|
mode = cls._env_or_cfg_str("HERMES_GATEWAY_BUSY_INPUT_MODE", "display", "busy_input_mode").lower()
|
|
return mode if mode in {"queue", "steer"} else "interrupt"
|
|
|
|
@classmethod
|
|
def _load_busy_text_mode(cls) -> str:
|
|
"""Normal busy TEXT follow-up behavior.
|
|
|
|
``busy_input_mode`` is the source of truth (default ``interrupt``); legacy ``busy_text_mode``
|
|
is honored only when explicitly set so existing queue setups keep working.
|
|
"""
|
|
from gateway.run import GatewayRunner
|
|
legacy = cls._env_or_cfg_str("HERMES_GATEWAY_BUSY_TEXT_MODE", "display", "busy_text_mode").lower()
|
|
if legacy in {"interrupt", "queue"}:
|
|
return legacy
|
|
return "queue" if GatewayRunner._load_busy_input_mode() == "queue" else "interrupt"
|
|
|
|
@staticmethod
|
|
def _busy_modes_from_config(config: dict, *, fallback_input: str, fallback_text: str) -> tuple[str, str]:
|
|
"""Resolve one profile's busy modes without consulting process env."""
|
|
raw_input = str(cfg_get(config, "display", "busy_input_mode", default="") or "").strip().lower()
|
|
input_mode = raw_input if raw_input in _BUSY_INPUT_MODES else fallback_input
|
|
raw_text = str(cfg_get(config, "display", "busy_text_mode", default="") or "").strip().lower()
|
|
if raw_text in {"interrupt", "queue"}:
|
|
text_mode = raw_text
|
|
elif raw_input in _BUSY_INPUT_MODES:
|
|
text_mode = "queue" if input_mode == "queue" else "interrupt"
|
|
else:
|
|
text_mode = fallback_text
|
|
return input_mode, text_mode
|
|
|
|
@staticmethod
|
|
def _busy_text_timing_from_config(config: dict) -> tuple[float, float]:
|
|
"""``display.busy_text_debounce_seconds`` / ``display.busy_text_hard_cap_seconds`` for one
|
|
profile, without consulting process env (#116893). A non-numeric or negative value is
|
|
rejected with a warning naming the key, never silently coerced."""
|
|
from gateway.platforms.base import DEFAULT_BUSY_TEXT_DEBOUNCE_SECONDS, DEFAULT_BUSY_TEXT_HARD_CAP_SECONDS
|
|
out = []
|
|
for key, default in (("busy_text_debounce_seconds", DEFAULT_BUSY_TEXT_DEBOUNCE_SECONDS),
|
|
("busy_text_hard_cap_seconds", DEFAULT_BUSY_TEXT_HARD_CAP_SECONDS)):
|
|
raw = cfg_get(config, "display", key, default=None)
|
|
if raw is None or raw == "":
|
|
out.append(default)
|
|
continue
|
|
try:
|
|
value = float(raw)
|
|
if value < 0 or value != value:
|
|
raise ValueError(raw)
|
|
except (TypeError, ValueError):
|
|
logger.warning("display.%s=%r is not a non-negative number; using %s", key, raw, default)
|
|
value = default
|
|
out.append(value)
|
|
return out[0], out[1]
|
|
|
|
@staticmethod
|
|
def _human_delay_from_config(config: dict) -> Optional[tuple[int, int]]:
|
|
"""``human_delay.{mode,min_ms,max_ms}`` for one profile as a ``(lo_ms, hi_ms)`` range, or
|
|
``None`` when off (#116895). ``natural`` is the fixed 800-2500 range; ``custom`` reads the
|
|
bounds and rejects non-integer, negative or inverted values with a warning naming the key,
|
|
falling back to the natural range instead of letting ``random.uniform`` raise at send."""
|
|
mode = str(cfg_get(config, "human_delay", "mode", default="off") or "off").strip().lower()
|
|
if mode == "off":
|
|
return None
|
|
lo, hi = DEFAULT_HUMAN_DELAY_RANGE_MS
|
|
if mode != "natural":
|
|
if mode != "custom":
|
|
logger.warning("human_delay.mode=%r is not off/natural/custom; using natural", mode)
|
|
return DEFAULT_HUMAN_DELAY_RANGE_MS
|
|
bounds = []
|
|
for key, default in (("min_ms", lo), ("max_ms", hi)):
|
|
raw = cfg_get(config, "human_delay", key, default=None)
|
|
try:
|
|
value = default if raw is None or raw == "" else int(raw)
|
|
if value < 0:
|
|
raise ValueError(raw)
|
|
except (TypeError, ValueError):
|
|
logger.warning("human_delay.%s=%r is not a non-negative integer; using %s", key, raw, default)
|
|
value = default
|
|
bounds.append(value)
|
|
lo, hi = bounds
|
|
if lo > hi:
|
|
logger.warning("human_delay.min_ms=%s exceeds human_delay.max_ms=%s; using natural range", lo, hi)
|
|
lo, hi = DEFAULT_HUMAN_DELAY_RANGE_MS
|
|
return lo, hi
|
|
|
|
def _snapshot_profile_busy_modes(self, profile_name: str, config: dict) -> None:
|
|
"""Cache a routed profile's busy policy and pacing for this gateway lifetime."""
|
|
input_mode, text_mode = self._busy_modes_from_config(
|
|
config, fallback_input=getattr(self, "_busy_input_mode", "interrupt"),
|
|
fallback_text=getattr(self, "_busy_text_mode", "interrupt"),
|
|
)
|
|
self.__dict__.setdefault("_busy_input_modes_by_profile", {})[profile_name] = input_mode
|
|
self.__dict__.setdefault("_busy_text_modes_by_profile", {})[profile_name] = text_mode
|
|
self.__dict__.setdefault("_busy_text_timing_by_profile", {})[profile_name] = (
|
|
self._busy_text_timing_from_config(config))
|
|
self.__dict__.setdefault("_human_delay_by_profile", {})[profile_name] = (
|
|
self._human_delay_from_config(config))
|
|
|
|
def _busy_profile_name_for_source(self, source: SessionSource) -> Optional[str]:
|
|
"""Return the routed profile whose busy policy applies, if any."""
|
|
if not getattr(getattr(self, "config", None), "multiplex_profiles", False):
|
|
return None
|
|
name = str(getattr(source, "profile", "") or "").strip()
|
|
if not name:
|
|
try:
|
|
name = str(self._profile_name_for_source(source) or "").strip()
|
|
except Exception:
|
|
name = ""
|
|
return name or None
|
|
|
|
def _effective_busy_mode(self, source: SessionSource, attr: str) -> str:
|
|
"""Busy mode from the routed profile snapshot (``attr``: ``_busy_input_mode`` / ``_busy_text_mode``)."""
|
|
fallback = getattr(self, attr, "interrupt")
|
|
profile_name = self._busy_profile_name_for_source(source)
|
|
if not profile_name:
|
|
return fallback
|
|
modes = getattr(self, attr + "s_by_profile", None)
|
|
return modes.get(profile_name, fallback) if isinstance(modes, dict) else fallback
|
|
|
|
def _effective_busy_input_mode(self, source: SessionSource) -> str:
|
|
"""Resolve busy input mode from the routed profile startup snapshot."""
|
|
return self._effective_busy_mode(source, "_busy_input_mode")
|
|
|
|
def _effective_busy_text_mode(self, source: SessionSource) -> str:
|
|
"""Resolve legacy busy text mode from the routed profile snapshot."""
|
|
return self._effective_busy_mode(source, "_busy_text_mode")
|
|
|
|
@staticmethod
|
|
def _warn_unparsable_timeout(cfg_key: str, raw: object, default: float) -> None:
|
|
"""Warn when a supplied timeout value is not a number (the parser already fell back to ``default``)."""
|
|
try:
|
|
float(raw)
|
|
except (TypeError, ValueError):
|
|
logger.warning("Invalid %s '%s', using default %.0fs", cfg_key, raw, default)
|
|
|
|
@classmethod
|
|
def _load_restart_drain_timeout(cls) -> float:
|
|
"""Graceful gateway restart/stop drain timeout in seconds."""
|
|
raw = cls._env_or_cfg_str("HERMES_RESTART_DRAIN_TIMEOUT", "agent", "restart_drain_timeout")
|
|
value = parse_restart_drain_timeout(raw)
|
|
if raw and value == DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT:
|
|
cls._warn_unparsable_timeout("restart_drain_timeout", raw, DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT)
|
|
return value
|
|
|
|
@staticmethod
|
|
def _load_launchd_exit_timeout(drain_timeout: float) -> Optional[float]:
|
|
"""Read the live launchd ``ExitTimeOut`` this job runs under, if any.
|
|
|
|
launchd is the one supervisor the gateway cannot size from config: the per-user (gui)
|
|
domain clamps ``ExitTimeOut`` (measured 60s on macOS 26), and any signal-driven stop that
|
|
drains past it is SIGKILLed mid-teardown — the unclean-exit half of the state.db
|
|
corruption class. Returns ``None`` (fail-open, drain unchanged) when not launchd-owned or
|
|
when ``launchctl print`` is unavailable. Logs a WARNING when the configured drain exceeds
|
|
the live budget so the misconfiguration is visible at boot, not at the next SIGKILL.
|
|
"""
|
|
label = launchd_service_label()
|
|
if label is None:
|
|
return None
|
|
# read_launchd_exit_timeout_s is already fail-open (returns None on any probe failure).
|
|
exit_timeout = read_launchd_exit_timeout_s(label)
|
|
if exit_timeout is None:
|
|
return None
|
|
effective = resolve_launchd_capped_drain(drain_timeout, exit_timeout)
|
|
if effective < drain_timeout:
|
|
logger.warning(
|
|
"restart_drain_timeout=%.0fs exceeds the live launchd exit timeout (%.0fs) for %s; "
|
|
"signal-driven stops will drain at most %.0fs so teardown finishes before launchd "
|
|
"SIGKILLs (launchd clamps ExitTimeOut in the per-user domain).",
|
|
drain_timeout, exit_timeout, label, effective,
|
|
)
|
|
else:
|
|
logger.info(
|
|
"launchd exit timeout for %s is %.0fs (drain %.0fs fits)",
|
|
label, exit_timeout, drain_timeout,
|
|
)
|
|
return exit_timeout
|
|
|
|
@classmethod
|
|
def _load_env_or_agent_cfg_timeout(cls, env_var: str, cfg_key: str, parse, default: float) -> float:
|
|
"""Env var (non-empty) else ``agent.<cfg_key>``; warn once when a supplied value fails to parse.
|
|
|
|
``0`` is a valid value; the parser falls back to ``default`` on garbage."""
|
|
from gateway.run import _load_gateway_config
|
|
env_raw = os.getenv(env_var)
|
|
if env_raw is not None and str(env_raw).strip() != "":
|
|
raw: object = env_raw
|
|
else:
|
|
raw = cfg_get(_load_gateway_config(), "agent", cfg_key, default=None)
|
|
value = parse(raw)
|
|
if raw is not None and str(raw).strip() != "":
|
|
cls._warn_unparsable_timeout(cfg_key, raw, default)
|
|
return value
|
|
|
|
@classmethod
|
|
def _load_restart_after_turn_timeout(cls) -> float:
|
|
"""In-band restart wait-for-idle timeout in seconds."""
|
|
return cls._load_env_or_agent_cfg_timeout(
|
|
"HERMES_RESTART_AFTER_TURN_TIMEOUT", "restart_after_turn_timeout",
|
|
parse_restart_after_turn_timeout, DEFAULT_GATEWAY_RESTART_AFTER_TURN_TIMEOUT,
|
|
)
|
|
|
|
@classmethod
|
|
def _load_cron_drain_timeout(cls) -> float:
|
|
"""The cron and api_server (/v1) floor under the stop()/drain wait.
|
|
|
|
See #82161.
|
|
"""
|
|
return cls._load_env_or_agent_cfg_timeout(
|
|
"HERMES_CRON_DRAIN_TIMEOUT", "cron_drain_timeout",
|
|
parse_cron_drain_timeout, DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT,
|
|
)
|
|
|
|
@classmethod
|
|
def _load_signal_interrupt_grace_timeout(cls) -> float:
|
|
"""``gateway.signal_interrupt_grace_timeout``: unexpected-signal post-interrupt grace in seconds."""
|
|
from gateway.run import _load_gateway_config
|
|
raw = cfg_get(_load_gateway_config(), "gateway", "signal_interrupt_grace_timeout", default=None)
|
|
value = parse_signal_interrupt_grace_timeout(raw)
|
|
if raw is not None and raw != "":
|
|
cls._warn_unparsable_timeout(
|
|
"signal_interrupt_grace_timeout", raw, DEFAULT_GATEWAY_SIGNAL_INTERRUPT_GRACE_TIMEOUT
|
|
)
|
|
return value
|
|
|
|
def _post_interrupt_grace_timeout(self) -> float:
|
|
"""Grace before teardown after forcibly interrupting agents (longer on an unexpected signal)."""
|
|
if getattr(self, "_signal_initiated_shutdown", False) and not getattr(self, "_restart_requested", False):
|
|
grace = getattr(self, "_signal_interrupt_grace_timeout", DEFAULT_GATEWAY_SIGNAL_INTERRUPT_GRACE_TIMEOUT)
|
|
return max(0.0, float(grace))
|
|
return DEFAULT_GATEWAY_POST_INTERRUPT_GRACE_TIMEOUT
|
|
|
|
@staticmethod
|
|
def _load_background_notifications_mode() -> str:
|
|
"""Background process notification mode from env/config (default ``concise``), resolved for
|
|
the AMBIENT profile — callers deciding for another profile's event enter its scope first
|
|
(``_completion_event_scope``). The env override reads through the secret scope so a served
|
|
secondary sees its own ``.env`` value, not the launch profile's ``os.environ``."""
|
|
from gateway.run import _load_gateway_config
|
|
from gateway.platforms._shared import platform_gate_env as _platform_gate_env
|
|
mode = _platform_gate_env("HERMES_BACKGROUND_NOTIFICATIONS")
|
|
if not mode:
|
|
raw = cfg_get(_load_gateway_config(), "display", "background_process_notifications")
|
|
if raw is False:
|
|
mode = "off"
|
|
elif raw not in {None, ""}:
|
|
mode = str(raw)
|
|
mode = (mode or "concise").strip().lower()
|
|
if mode not in {"concise", "all", "result", "error", "off"}:
|
|
logger.warning("Unknown background_process_notifications '%s', defaulting to 'concise'", mode)
|
|
return "concise"
|
|
return mode
|
|
|
|
@staticmethod
|
|
def _load_provider_routing() -> dict:
|
|
"""OpenRouter provider routing preferences (canonical fail-open loader: managed overlay + ${VAR})."""
|
|
from gateway.run import _load_gateway_config
|
|
try:
|
|
return _load_gateway_config().get("provider_routing", {}) or {}
|
|
except Exception:
|
|
return {}
|
|
|
|
@staticmethod
|
|
def _load_fallback_model() -> list | None:
|
|
"""Fallback chain: ``fallback_providers`` (kept first) merged with legacy ``fallback_model``."""
|
|
from gateway.run import _load_gateway_config
|
|
try:
|
|
# Canonical gateway loader (fail-open): managed overlay + ${VAR} expansion apply here too.
|
|
return get_fallback_chain(_load_gateway_config()) or None
|
|
except Exception:
|
|
return None
|
|
|
|
def _refresh_fallback_model(self) -> list | None:
|
|
"""Re-read fallback_providers from disk for the next agent create/reuse.
|
|
|
|
Lets a chain edited after startup reach messaging sessions (cron already re-reads per job).
|
|
A TRANSIENT read/parse failure (user mid-edit, non-atomic write) keeps the last known-good
|
|
chain; only a successful read that genuinely lacks the key clears it.
|
|
|
|
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.
|
|
|
|
Reads the ACTIVE gateway home (the routed profile under multiplexing, else the launch home)
|
|
and keeps one last-known-good chain per home: a single runner-wide slot filled from the launch
|
|
home handed every secondary profile the default profile's fallback chain.
|
|
"""
|
|
from hermes_cli.config_backend import config_exists
|
|
from gateway.run import _gateway_config_home
|
|
from hermes_constants import hermes_home_key
|
|
home = _gateway_config_home()
|
|
by_home = getattr(self, "_fallback_model_by_home", None)
|
|
if by_home is None:
|
|
by_home = self._fallback_model_by_home = {}
|
|
home_key = hermes_home_key(home)
|
|
try:
|
|
from hermes_cli.config_effective import load_user_config_effective
|
|
cfg_path = home / "config.yaml"
|
|
if not config_exists(cfg_path):
|
|
by_home[home_key] = self._fallback_model = None
|
|
return self._fallback_model
|
|
# fail_closed: a torn mid-edit write must raise so the per-home last known-good chain
|
|
# below survives, instead of being WIPED by an empty fail-open result.
|
|
cfg = load_user_config_effective(cfg_path, fail_closed=True)
|
|
except Exception:
|
|
logger.debug("fallback_providers refresh: config.yaml read failed; keeping last known-good chain", exc_info=True)
|
|
self._fallback_model = by_home.get(home_key, self._fallback_model)
|
|
return self._fallback_model
|
|
by_home[home_key] = 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 the rewrite while a cooldown holds the agent on an activated fallback provider
|
|
(``restore_primary_runtime`` owns that lifecycle); otherwise replaces the chain so
|
|
mid-uptime ``fallback_providers`` edits apply without a restart.
|
|
|
|
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 means the user changed something — drop the session-scoped unavailability
|
|
# memo so re-configured entries (e.g. credentials added mid-uptime) get retried. Only on real
|
|
# content change, so the per-message no-op refresh keeps the memo's rate-limiting benefit.
|
|
# See #60955.
|
|
if new_chain != old_chain:
|
|
unavailable = getattr(agent, "_unavailable_fallback_keys", None)
|
|
if unavailable:
|
|
unavailable.clear()
|