""" Hermes Web UI -- Shared configuration, constants, and global state. Imported by all other api/* modules and by server.py. Discovery order for all paths: 1. Explicit environment variable 2. Filesystem heuristics (sibling checkout, parent dir, common install locations) 3. Hardened defaults relative to $HOME 4. Fail loudly with a human-readable fix-it message if required modules are missing """ import collections import json import os import sys import threading import time import traceback import uuid from pathlib import Path from urllib.parse import parse_qs, urlparse # ── Basic layout ────────────────────────────────────────────────────────────── HOME = Path.home() # REPO_ROOT is the directory that contains this file's parent (api/ -> repo root) REPO_ROOT = Path(__file__).parent.parent.resolve() # ── Network config (env-overridable) ───────────────────────────────────────── HOST = os.getenv('HERMES_WEBUI_HOST', '127.0.0.1') PORT = int(os.getenv('HERMES_WEBUI_PORT', '8787')) # ── State directory (env-overridable, never inside repo) ────────────────────── STATE_DIR = Path(os.getenv( 'HERMES_WEBUI_STATE_DIR', str(HOME / '.hermes' / 'webui') )).expanduser().resolve() SESSION_DIR = STATE_DIR / 'sessions' WORKSPACES_FILE = STATE_DIR / 'workspaces.json' SESSION_INDEX_FILE = SESSION_DIR / '_index.json' SETTINGS_FILE = STATE_DIR / 'settings.json' LAST_WORKSPACE_FILE = STATE_DIR / 'last_workspace.txt' PROJECTS_FILE = STATE_DIR / 'projects.json' # ── Hermes agent directory discovery ───────────────────────────────────────── def _discover_agent_dir() -> Path: """ Locate the hermes-agent checkout using a multi-strategy search. Priority: 1. HERMES_WEBUI_AGENT_DIR env var -- explicit override always wins 2. HERMES_HOME / hermes-agent -- e.g. ~/.hermes/hermes-agent 3. Sibling of this repo -- ../hermes-agent 4. Parent of this repo -- ../../hermes-agent (nested layout) 5. Common install paths -- ~/.hermes/hermes-agent (again as fallback) 6. HOME / hermes-agent -- ~/hermes-agent (simple flat layout) """ candidates = [] # 1. Explicit env var if os.getenv('HERMES_WEBUI_AGENT_DIR'): candidates.append(Path(os.getenv('HERMES_WEBUI_AGENT_DIR')).expanduser().resolve()) # 2. HERMES_HOME / hermes-agent hermes_home = os.getenv('HERMES_HOME', str(HOME / '.hermes')) candidates.append(Path(hermes_home).expanduser() / 'hermes-agent') # 3. Sibling: /../hermes-agent candidates.append(REPO_ROOT.parent / 'hermes-agent') # 4. Parent is the agent repo itself (repo cloned inside hermes-agent/) if (REPO_ROOT.parent / 'run_agent.py').exists(): candidates.append(REPO_ROOT.parent) # 5. ~/.hermes/hermes-agent (explicit common path) candidates.append(HOME / '.hermes' / 'hermes-agent') # 6. ~/hermes-agent candidates.append(HOME / 'hermes-agent') for path in candidates: if path.exists() and (path / 'run_agent.py').exists(): return path.resolve() return None def _discover_python(agent_dir: Path) -> str: """ Locate a Python executable that has the Hermes agent dependencies installed. Priority: 1. HERMES_WEBUI_PYTHON env var 2. Agent venv at /venv/bin/python 3. Local .venv inside this repo 4. System python3 """ if os.getenv('HERMES_WEBUI_PYTHON'): return os.getenv('HERMES_WEBUI_PYTHON') if agent_dir: venv_py = agent_dir / 'venv' / 'bin' / 'python' if venv_py.exists(): return str(venv_py) # Windows layout venv_py_win = agent_dir / 'venv' / 'Scripts' / 'python.exe' if venv_py_win.exists(): return str(venv_py_win) # Local .venv inside this repo local_venv = REPO_ROOT / '.venv' / 'bin' / 'python' if local_venv.exists(): return str(local_venv) # Fall back to system python3 import shutil for name in ('python3', 'python'): found = shutil.which(name) if found: return found return 'python3' # Run discovery _AGENT_DIR = _discover_agent_dir() PYTHON_EXE = _discover_python(_AGENT_DIR) # ── Inject agent dir into sys.path so Hermes modules are importable ────────── # When users (or CI builds) run `pip install --target .` or # `pip install -t .` inside the hermes-agent checkout, third-party # package directories (openai/, pydantic/, requests/, etc.) end up # alongside real Hermes source files. Putting _AGENT_DIR at the # FRONT of sys.path means Python resolves `import pydantic` from that # local directory — which breaks whenever the host platform differs # from the container (e.g. macOS .so files inside a Linux image). # # Fix: insert _AGENT_DIR at the END of sys.path. Python searches # entries in order, so site-packages resolves pip packages correctly, # and Hermes-specific modules (run_agent, hermes/, etc.) still # resolve because they do not exist in site-packages. if _AGENT_DIR is not None: if str(_AGENT_DIR) not in sys.path: sys.path.append(str(_AGENT_DIR)) _HERMES_FOUND = True else: _HERMES_FOUND = False try: from hermes_constants import VALID_REASONING_EFFORTS except Exception: VALID_REASONING_EFFORTS = ("xhigh", "high", "medium", "low", "minimal") # ── Config file (reloadable -- supports profile switching) ────────────────── _cfg_cache = {} _cfg_lock = threading.Lock() def _get_config_path() -> Path: """Return config.yaml path for the active profile.""" env_override = os.getenv('HERMES_CONFIG_PATH') if env_override: return Path(env_override).expanduser() try: from api.profiles import get_active_hermes_home return get_active_hermes_home() / 'config.yaml' except ImportError: return HOME / '.hermes' / 'config.yaml' def get_config() -> dict: """Return the cached config dict, loading from disk if needed.""" if not _cfg_cache: reload_config() return _cfg_cache def reload_config() -> None: """Reload config.yaml from the active profile's directory.""" with _cfg_lock: _cfg_cache.clear() config_path = _get_config_path() try: import yaml as _yaml if config_path.exists(): loaded = _yaml.safe_load(config_path.read_text()) if isinstance(loaded, dict): _cfg_cache.update(loaded) except Exception: pass # Initial load reload_config() cfg = _cfg_cache # alias for backward compat with existing references # ── Default workspace discovery ─────────────────────────────────────────────── def _discover_default_workspace() -> Path: """ Resolve the default workspace in order: 1. HERMES_WEBUI_DEFAULT_WORKSPACE env var 2. ~/workspace (common Hermes convention) 3. STATE_DIR / workspace (isolated fallback) """ if os.getenv('HERMES_WEBUI_DEFAULT_WORKSPACE'): return Path(os.getenv('HERMES_WEBUI_DEFAULT_WORKSPACE')).expanduser().resolve() common = HOME / 'workspace' if common.exists(): return common.resolve() return (STATE_DIR / 'workspace').resolve() DEFAULT_WORKSPACE = _discover_default_workspace() DEFAULT_MODEL = os.getenv('HERMES_WEBUI_DEFAULT_MODEL', 'openai/gpt-5.4-mini') # ── Startup diagnostics ─────────────────────────────────────────────────────── def print_startup_config() -> None: """Print detected configuration at startup so the user can verify what was found.""" ok = '\033[32m[ok]\033[0m' warn = '\033[33m[!!]\033[0m' err = '\033[31m[XX]\033[0m' lines = [ '', ' Hermes Web UI -- startup config', ' --------------------------------', f' repo root : {REPO_ROOT}', f' agent dir : {_AGENT_DIR if _AGENT_DIR else "NOT FOUND"} {ok if _AGENT_DIR else err}', f' python : {PYTHON_EXE}', f' state dir : {STATE_DIR}', f' workspace : {DEFAULT_WORKSPACE}', f' host:port : {HOST}:{PORT}', f' config file : {_get_config_path()} {"(found)" if _get_config_path().exists() else "(not found, using defaults)"}', '', ] print('\n'.join(lines), flush=True) if not _HERMES_FOUND: print( f'{err} Could not find the Hermes agent directory.\n' ' The server will start but agent features will not work.\n' '\n' ' To fix, set one of:\n' ' export HERMES_WEBUI_AGENT_DIR=/path/to/hermes-agent\n' ' export HERMES_HOME=/path/to/.hermes\n' '\n' ' Or clone hermes-agent as a sibling of this repo:\n' ' git clone ../hermes-agent\n', flush=True ) def verify_hermes_imports() -> tuple: """ Attempt to import the key Hermes modules. Returns (ok: bool, missing: list[str], errors: dict[str, str]). """ required = ['run_agent'] missing = [] errors = {} for mod in required: try: __import__(mod) except Exception as e: missing.append(mod) # Capture the full error message so startup logs show WHY # (e.g. pydantic_core .so mismatch) instead of just the name. errors[mod] = f"{type(e).__name__}: {e}" return (len(missing) == 0), missing, errors # ── Limits ─────────────────────────────────────────────────────────────────── MAX_FILE_BYTES = 200_000 MAX_UPLOAD_BYTES = 20 * 1024 * 1024 # ── File type maps ─────────────────────────────────────────────────────────── IMAGE_EXTS = {'.png', '.jpg', '.jpeg', '.gif', '.svg', '.webp', '.ico', '.bmp'} MD_EXTS = {'.md', '.markdown', '.mdown'} CODE_EXTS = {'.py', '.js', '.ts', '.jsx', '.tsx', '.css', '.html', '.json', '.yaml', '.yml', '.toml', '.sh', '.bash', '.txt', '.log', '.env', '.csv', '.xml', '.sql', '.rs', '.go', '.java', '.c', '.cpp', '.h'} MIME_MAP = { '.png':'image/png', '.jpg':'image/jpeg', '.jpeg':'image/jpeg', '.gif':'image/gif', '.svg':'image/svg+xml', '.webp':'image/webp', '.ico':'image/x-icon', '.bmp':'image/bmp', '.pdf':'application/pdf', '.json':'application/json', } # ── Toolsets (from config.yaml or hardcoded default) ───────────────────────── _DEFAULT_TOOLSETS = [ 'browser', 'clarify', 'code_execution', 'cronjob', 'delegation', 'file', 'image_gen', 'memory', 'session_search', 'skills', 'terminal', 'todo', 'web', 'webhook', ] CLI_TOOLSETS = get_config().get('platform_toolsets', {}).get('cli', _DEFAULT_TOOLSETS) # ── Model / provider discovery ─────────────────────────────────────────────── # Fallback OpenRouter list for the chat sidebar. # When the live catalog fetch succeeds we replace this with the current # free-text model list from OpenRouter's official models endpoint. _OPENROUTER_FREE_MODELS = [ {'provider': 'OpenRouter', 'id': 'openrouter/free', 'label': 'Auto (Free Router)'}, {'provider': 'OpenRouter', 'id': 'qwen/qwen3.6-plus:free', 'label': 'Qwen: Qwen3.6 Plus (free)'}, {'provider': 'OpenRouter', 'id': 'qwen/qwen3-coder:free', 'label': 'Qwen: Qwen3 Coder 480B A35B (free)'}, {'provider': 'OpenRouter', 'id': 'openai/gpt-oss-120b:free', 'label': 'OpenAI: gpt-oss-120b (free)'}, {'provider': 'OpenRouter', 'id': 'openai/gpt-oss-20b:free', 'label': 'OpenAI: gpt-oss-20b (free)'}, {'provider': 'OpenRouter', 'id': 'meta-llama/llama-3.3-70b-instruct:free', 'label': 'Meta: Llama 3.3 70B Instruct (free)'}, {'provider': 'OpenRouter', 'id': 'nousresearch/hermes-3-llama-3.1-405b:free', 'label': 'Nous: Hermes 3 405B Instruct (free)'}, ] _DYNAMIC_MODEL_CACHE: dict[str, dict] = {} _DYNAMIC_MODEL_CACHE_TTL_SECONDS = 300 # Provider display names for known Hermes provider IDs _PROVIDER_DISPLAY = { 'nous': 'Nous Portal', 'openrouter': 'OpenRouter', 'anthropic': 'Anthropic', 'openai': 'OpenAI', 'openai-codex': 'OpenAI Codex', 'copilot': 'GitHub Copilot', 'zai': 'Z.AI / GLM', 'kimi-coding': 'Kimi / Moonshot', 'deepseek': 'DeepSeek', 'mistral': 'Mistral', 'google-ai': 'Google AI', 'blockrun': 'Blockrun / ClawRouter', 'minimax': 'MiniMax', 'google': 'Google', 'meta-llama': 'Meta Llama', 'huggingface': 'HuggingFace', 'alibaba': 'Alibaba', 'ollama': 'Ollama', 'lmstudio': 'LM Studio', } _PROVIDER_SORT_ORDER = [ 'openai-codex', 'openai', 'anthropic', 'kimi-coding', 'deepseek', 'mistral', 'blockrun', 'openrouter', 'google', 'zai', 'minimax', 'nous', 'custom', 'google-ai', 'ollama', 'lmstudio', 'local', ] _PROVIDER_ENV_KEYS = { 'anthropic': ('ANTHROPIC_API_KEY', 'ANTHROPIC_TOKEN'), 'openai': ('OPENAI_API_KEY',), 'openrouter': ('OPENROUTER_API_KEY',), 'google': ('GOOGLE_API_KEY',), 'zai': ('GLM_API_KEY',), 'kimi-coding': ('KIMI_API_KEY', 'CUSTOM_API_MOONSHOT_AI_API_KEY'), 'deepseek': ('DEEPSEEK_API_KEY', 'CUSTOM_API_DEEPSEEK_COM_API_KEY'), 'mistral': ('MISTRAL_API_KEY',), 'blockrun': ('BLOCKRUN_API_KEY',), 'minimax': ('MINIMAX_API_KEY',), 'minimax-cn': ('MINIMAX_CN_API_KEY',), } # Well-known models per provider (used to populate dropdown for direct API providers) _PROVIDER_MODELS = { 'anthropic': [ {'id': 'claude-opus-4.6', 'label': 'Claude Opus 4.6'}, {'id': 'claude-sonnet-4.6', 'label': 'Claude Sonnet 4.6'}, {'id': 'claude-sonnet-4-5', 'label': 'Claude Sonnet 4.5'}, {'id': 'claude-haiku-4-5', 'label': 'Claude Haiku 4.5'}, ], 'openai': [ {'id': 'gpt-5.4-mini', 'label': 'GPT-5.4 Mini'}, {'id': 'gpt-4o', 'label': 'GPT-4o'}, {'id': 'o3', 'label': 'o3'}, {'id': 'o4-mini', 'label': 'o4-mini'}, ], 'openai-codex': [ {'id': 'gpt-5.3-codex', 'label': 'GPT-5.3 Codex'}, {'id': 'gpt-5.2-codex', 'label': 'GPT-5.2 Codex'}, {'id': 'gpt-5.1-codex-max', 'label': 'GPT-5.1 Codex Max'}, {'id': 'gpt-5.1-codex-mini', 'label': 'GPT-5.1 Codex Mini'}, ], 'google': [ {'id': 'gemini-2.5-pro', 'label': 'Gemini 2.5 Pro'}, ], 'deepseek': [ {'id': 'deepseek-chat', 'label': 'DeepSeek V3.2 Chat'}, {'id': 'deepseek-reasoner', 'label': 'DeepSeek V3.2 Thinking'}, ], 'nous': [ {'id': 'xiaomi/mimo-v2-pro', 'label': 'MiMo V2 Pro (free)'}, {'id': 'xiaomi/mimo-v2-omni', 'label': 'MiMo V2 Omni (free)'}, {'id': 'xiaomi/mimo-v2-flash', 'label': 'MiMo V2 Flash (free)'}, ], 'zai': [ {'id': 'glm-5.1', 'label': 'GLM-5.1'}, {'id': 'glm-5', 'label': 'GLM-5'}, {'id': 'glm-5-turbo', 'label': 'GLM-5 Turbo'}, {'id': 'glm-4.7', 'label': 'GLM-4.7'}, {'id': 'glm-4.5', 'label': 'GLM-4.5'}, {'id': 'glm-4.5-flash', 'label': 'GLM-4.5 Flash'}, ], 'kimi-coding': [ {'id': 'kimi-for-coding', 'label': 'Kimi for Coding'}, {'id': 'kimi-k2.5', 'label': 'Kimi K2.5'}, {'id': 'kimi-k2-thinking', 'label': 'Kimi K2 Thinking'}, {'id': 'kimi-k2-thinking-turbo', 'label': 'Kimi K2 Thinking Turbo'}, {'id': 'kimi-k2-turbo-preview', 'label': 'Kimi K2 Turbo Preview'}, {'id': 'kimi-k2-0905-preview', 'label': 'Kimi K2 0905'}, {'id': 'kimi-k2-0711-preview', 'label': 'Kimi K2 0711'}, {'id': 'moonshot-v1-128k', 'label': 'Moonshot v1 128k'}, {'id': 'moonshot-v1-32k', 'label': 'Moonshot v1 32k'}, {'id': 'moonshot-v1-8k', 'label': 'Moonshot v1 8k'}, {'id': 'moonshot-v1-auto', 'label': 'Moonshot v1 Auto'}, ], 'minimax': [ {'id': 'MiniMax-M2.7', 'label': 'MiniMax M2.7'}, {'id': 'MiniMax-M2.7-highspeed', 'label': 'MiniMax M2.7 Highspeed'}, {'id': 'MiniMax-M2.5', 'label': 'MiniMax M2.5'}, {'id': 'MiniMax-M2.5-highspeed', 'label': 'MiniMax M2.5 Highspeed'}, {'id': 'MiniMax-M2.1', 'label': 'MiniMax M2.1'}, ], # GitHub Copilot — model IDs served via the Copilot API 'copilot': [ {'id': 'gpt-5.4', 'label': 'GPT-5.4'}, {'id': 'gpt-5.4-mini', 'label': 'GPT-5.4 Mini'}, {'id': 'gpt-4o', 'label': 'GPT-4o'}, {'id': 'claude-opus-4.6', 'label': 'Claude Opus 4.6'}, {'id': 'claude-sonnet-4.6', 'label': 'Claude Sonnet 4.6'}, {'id': 'gemini-2.5-pro', 'label': 'Gemini 2.5 Pro'}, ], # 'gemini' is the hermes_cli provider ID for Google AI Studio 'gemini': [ {'id': 'gemini-2.5-pro', 'label': 'Gemini 2.5 Pro'}, {'id': 'gemini-2.0-flash', 'label': 'Gemini 2.0 Flash'}, ], } _REASONING_EFFORT_ORDER = tuple( effort for effort in ("minimal", "low", "medium", "high", "xhigh") if effort in VALID_REASONING_EFFORTS ) _MISTRAL_REASONING_MODEL_IDS = { 'mistral-small-2603', 'mistral-small-latest', 'mistral-vibe-cli-fast', } _MISTRAL_REASONING_MODEL_PREFIXES = ( 'magistral-', 'labs-leanstral-', ) def _normalize_provider_id(provider_id: str | None) -> str | None: provider_id = (provider_id or '').strip().lower() if not provider_id: return None aliases = { 'codex': 'openai-codex', 'moonshot': 'kimi-coding', 'kimi': 'kimi-coding', 'deep-seek': 'deepseek', 'clawrouter': 'blockrun', } return aliases.get(provider_id, provider_id) def _provider_sort_key(provider_id: str, active_provider: str | None = None) -> tuple: provider_id = _normalize_provider_id(provider_id) or '' active_provider = _normalize_provider_id(active_provider) if active_provider and provider_id == active_provider: return (-1, provider_id) try: return (_PROVIDER_SORT_ORDER.index(provider_id), provider_id) except ValueError: return (999, provider_id) def _load_profile_env() -> dict: """Read the active profile .env and merge in current process overrides. Reads the main ~/.hermes/.env first as a base layer, then overlays the profile-specific .env on top. This ensures API keys stored in the main file are visible even when the active profile doesn't duplicate them. """ main_env_path = HOME / '.hermes' / '.env' try: from api.profiles import get_active_hermes_home as _gah profile_env_path = _gah() / '.env' except ImportError: profile_env_path = main_env_path data: dict[str, str] = {} # Read main .env first (base layer), then profile overrides for path in (main_env_path, profile_env_path): if not path.exists(): continue try: for line in path.read_text().splitlines(): line = line.strip() if line and not line.startswith('#') and '=' in line: k, v = line.split('=', 1) data[k.strip()] = v.strip().strip('"').strip("'") except Exception: pass interesting = { key for keys in _PROVIDER_ENV_KEYS.values() for key in keys } | { 'HERMES_API_KEY', 'HERMES_OPENAI_API_KEY', 'LOCAL_API_KEY', 'API_KEY', } for key in interesting: val = os.getenv(key) if val: data[key] = val return data def _load_repo_env() -> dict: """Read repo-local .env overrides used by the WebUI service.""" env_path = REPO_ROOT / '.env' data: dict[str, str] = {} if not env_path.exists(): return data try: for line in env_path.read_text().splitlines(): line = line.strip() if line and not line.startswith('#') and '=' in line: k, v = line.split('=', 1) data[k.strip()] = v.strip().strip('"').strip("'") except Exception: pass return data def _get_allowed_providers() -> set[str] | None: raw = os.getenv('HERMES_WEBUI_ALLOWED_PROVIDERS') if not raw: raw = _load_repo_env().get('HERMES_WEBUI_ALLOWED_PROVIDERS', '') if not raw: return None allowed = { _normalize_provider_id(part) for part in raw.split(',') if str(part).strip() } return {provider_id for provider_id in allowed if provider_id} def _get_hidden_providers() -> set[str]: raw = os.getenv('HERMES_WEBUI_HIDDEN_PROVIDERS') if not raw: raw = _load_repo_env().get('HERMES_WEBUI_HIDDEN_PROVIDERS', '') hidden = { _normalize_provider_id(part) for part in raw.split(',') if str(part).strip() } return {provider_id for provider_id in hidden if provider_id} def _normalize_reasoning_options(options) -> list[str]: seen: set[str] = set() for raw in options or []: effort = str(raw or '').strip().lower() if effort in VALID_REASONING_EFFORTS: seen.add(effort) return [effort for effort in _REASONING_EFFORT_ORDER if effort in seen] def _reasoning_metadata(mode: str, *, options=None, note: str = '') -> dict: mode = str(mode or 'unsupported').strip().lower() or 'unsupported' normalized_options = _normalize_reasoning_options(options) values = [''] if mode == 'effort': values.append('none') values.extend(normalized_options) elif mode == 'toggle': values.append('none') elif mode not in {'model', 'unsupported'}: mode = 'unsupported' return { 'mode': mode, 'options': values, 'note': str(note or '').strip(), 'disabled': mode not in {'effort', 'toggle'}, } def _reasoning_metadata_from_supported_parameters(supported_parameters) -> dict | None: normalized = { str(item or '').strip().lower().replace('-', '_') for item in (supported_parameters or []) if str(item or '').strip() } if not normalized: return None if any('reasoning_effort' in item or 'reasoning.effort' in item for item in normalized): return _reasoning_metadata( 'effort', options=('low', 'medium', 'high'), note='This provider reports explicit reasoning-effort controls for the model.', ) if any( item == 'reasoning' or item == 'include_reasoning' or item.startswith('reasoning.') for item in normalized ): return _reasoning_metadata( 'toggle', note='This model exposes reasoning as an on/off toggle, not a low/medium/high ladder.', ) return None def _copilot_reasoning_efforts_for_model(model_id: str) -> list[str]: try: from hermes_cli.models import github_model_reasoning_efforts efforts = github_model_reasoning_efforts(model_id) except Exception: efforts = [] normalized = _normalize_reasoning_options(efforts) if normalized: return normalized raw = str(model_id or '').strip().lower() if raw.startswith(("openai/o1", "openai/o3", "openai/o4", "o1", "o3", "o4")): return _normalize_reasoning_options(("low", "medium", "high")) if raw.startswith("gpt-5") or "/gpt-5" in raw: return _normalize_reasoning_options(("minimal", "low", "medium", "high")) return [] def _reasoning_metadata_for_model(provider_id: str | None, model_id: str, raw_entry: dict | None = None) -> dict: provider_id = _normalize_provider_id(provider_id) or '' model_id = str(model_id or '').strip() raw_lower = model_id.lower() if provider_id == 'openrouter' and raw_lower == 'openrouter/free': return _reasoning_metadata( 'unsupported', note='OpenRouter Auto (Free) can route across multiple models, so reasoning controls are unavailable here.', ) if provider_id == 'deepseek' or raw_lower.startswith('deepseek-chat') or raw_lower.startswith('deepseek-reasoner'): return _reasoning_metadata( 'model', note='DeepSeek switches reasoning via separate Chat and Thinking models instead of an effort dropdown.', ) if provider_id == 'kimi-coding' and 'thinking' in raw_lower: return _reasoning_metadata( 'model', note='Moonshot exposes thinking via separate model variants instead of per-request effort levels.', ) efforts = _copilot_reasoning_efforts_for_model(model_id) if efforts: return _reasoning_metadata('effort', options=efforts) supported_meta = None capabilities = None if isinstance(raw_entry, dict): supported_meta = _reasoning_metadata_from_supported_parameters(raw_entry.get('supported_parameters')) capabilities = raw_entry.get('capabilities') if supported_meta is not None: return supported_meta if isinstance(capabilities, dict) and capabilities.get('reasoning') is True: return _reasoning_metadata( 'toggle', note='This model supports reasoning on or off, but does not expose low/medium/high effort levels.', ) if ( raw_lower in _MISTRAL_REASONING_MODEL_IDS or any(raw_lower.startswith(prefix) for prefix in _MISTRAL_REASONING_MODEL_PREFIXES) ): return _reasoning_metadata( 'toggle', note='This Mistral model supports reasoning as an on/off toggle.', ) return _reasoning_metadata('unsupported') def _annotate_reasoning_metadata(provider_id: str | None, raw_models: list[dict]) -> list[dict]: provider_id = _normalize_provider_id(provider_id) annotated: list[dict] = [] for raw in raw_models or []: if not isinstance(raw, dict): continue model_id = str(raw.get('id') or '').strip() if not model_id: continue entry = dict(raw) entry['reasoning'] = _reasoning_metadata_for_model(provider_id, model_id, raw_entry=entry) annotated.append(entry) return annotated def _lookup_model_catalog_entry( provider_id: str | None, model_id: str, *, all_env: dict | None = None, ) -> dict | None: provider_id = _normalize_provider_id(provider_id) model_id = str(model_id or '').strip() if not provider_id or not model_id: return None if provider_id == 'openrouter': raw_models = _discover_openrouter_free_models(all_env=all_env) elif provider_id == 'openai-codex': raw_models = _discover_codex_model_entries() or _PROVIDER_MODELS.get(provider_id, []) elif provider_id in _PROVIDER_MODELS: raw_models = _PROVIDER_MODELS.get(provider_id, []) else: raw_models = _discover_named_custom_provider_catalogs(all_env=all_env).get(provider_id, []) for raw in raw_models or []: if not isinstance(raw, dict): continue if str(raw.get('id') or '').strip() == model_id: return dict(raw) return None def get_model_reasoning_metadata( model_id: str, provider_id: str | None = None, *, all_env: dict | None = None, ) -> dict: selected_model = str(model_id or '').strip() resolved_model = selected_model resolved_provider = _normalize_provider_id(provider_id) if selected_model: try: _model, _provider, _ = resolve_model_provider(selected_model) if _model: resolved_model = _model if not resolved_provider and _provider: resolved_provider = _normalize_provider_id(_provider) except Exception: pass raw_entry = _lookup_model_catalog_entry( resolved_provider, resolved_model, all_env=all_env, ) return _reasoning_metadata_for_model(resolved_provider, resolved_model, raw_entry=raw_entry) def clamp_reasoning_effort( model_id: str, reasoning_effort: str, provider_id: str | None = None, *, all_env: dict | None = None, ) -> str: value = str(reasoning_effort or '').strip().lower() if not value: return '' meta = get_model_reasoning_metadata( model_id, provider_id=provider_id, all_env=all_env, ) if meta.get('disabled'): return '' allowed = { str(item or '').strip().lower() for item in (meta.get('options') or []) if str(item or '').strip() } return value if value in allowed else '' def _load_auth_store() -> dict: try: from api.profiles import get_active_hermes_home as _gah auth_store_path = _gah() / 'auth.json' except ImportError: auth_store_path = HOME / '.hermes' / 'auth.json' if not auth_store_path.exists(): return {} try: data = json.loads(auth_store_path.read_text()) return data if isinstance(data, dict) else {} except Exception: return {} def _auth_store_has_provider(auth_store: dict, provider_id: str) -> bool: provider_id = _normalize_provider_id(provider_id) or '' providers = auth_store.get('providers') or {} state = providers.get(provider_id) if isinstance(state, dict): if state.get('tokens') or state.get('api_key') or state.get('key') or state.get('api_keys'): return True if state.get('credential_pool') or state.get('oauth'): return True if state: return True elif state: return True pool = auth_store.get('credential_pool') or {} if provider_id in pool and pool.get(provider_id): return True return False def _infer_provider_from_base_url(base_url: str) -> str: base_url = (base_url or '').strip() if not base_url: return 'custom' parsed = urlparse(base_url if '://' in base_url else f'http://{base_url}') host = (parsed.hostname or parsed.netloc or parsed.path or '').lower() hosted_map = ( ('openrouter.ai', 'openrouter'), ('api.deepseek.com', 'deepseek'), ('moonshot.ai', 'kimi-coding'), ('api.kimi.com', 'kimi-coding'), ('api.openai.com', 'openai'), ('api.anthropic.com', 'anthropic'), ('generativelanguage.googleapis.com', 'google'), ('googleapis.com', 'google'), ) for needle, provider_id in hosted_map: if needle in host: return provider_id if parsed.hostname: try: import ipaddress addr = ipaddress.ip_address(parsed.hostname) if addr.is_private or addr.is_loopback or addr.is_link_local: if 'ollama' in host or parsed.hostname in {'127.0.0.1', 'localhost'}: return 'ollama' if 'lmstudio' in host or 'lm-studio' in host: return 'lmstudio' return 'local' except ValueError: if host.startswith('localhost') or host.endswith('.local'): return 'local' return 'custom' def _load_cached_model_catalog(cache_key: str, loader, ttl_seconds: int = _DYNAMIC_MODEL_CACHE_TTL_SECONDS) -> list[dict]: now = time.time() cached = _DYNAMIC_MODEL_CACHE.get(cache_key) if cached and (now - cached.get('ts', 0)) < ttl_seconds: return [dict(item) for item in cached.get('models', [])] try: fresh = loader() or [] except Exception: fresh = [] if fresh: models = [dict(item) for item in fresh] _DYNAMIC_MODEL_CACHE[cache_key] = { 'ts': now, 'models': models, } return [dict(item) for item in models] if cached: return [dict(item) for item in cached.get('models', [])] return [] def _looks_like_free_price(value) -> bool: return str(value).strip() in {'0', '0.0', '0.00', '0.000', '0.0000', ''} def _discover_openai_compatible_models( *, base_url: str, api_key: str = '', cache_key: str, ) -> list[dict]: base_url = (base_url or '').strip().rstrip('/') if not base_url: return [] def _load() -> list[dict]: import urllib.request endpoint_url = f'{base_url}/models' if base_url.endswith('/v1') else f'{base_url}/v1/models' headers = {'Accept': 'application/json'} token = (api_key or '').strip() if token and token != 'no-key-required': headers['Authorization'] = f'Bearer {token}' req = urllib.request.Request(endpoint_url, headers=headers, method='GET') with urllib.request.urlopen(req, timeout=10) as response: payload = json.loads(response.read().decode('utf-8')) models_list = [] if isinstance(payload, dict): if isinstance(payload.get('data'), list): models_list = payload.get('data', []) elif isinstance(payload.get('models'), list): models_list = payload.get('models', []) seen: set[str] = set() entries: list[dict] = [] for item in models_list: if isinstance(item, dict): model_id = str(item.get('id') or item.get('name') or item.get('model') or '').strip() label = str(item.get('name') or item.get('model') or model_id).strip() capabilities = item.get('capabilities') or {} supported_parameters = item.get('supported_parameters') or [] if isinstance(capabilities, dict) and capabilities.get('completion_chat') is False: continue else: model_id = str(item or '').strip() label = model_id capabilities = {} supported_parameters = [] if not model_id or model_id in seen: continue seen.add(model_id) entry = {'id': model_id, 'label': label or model_id} if isinstance(capabilities, dict) and capabilities: entry['capabilities'] = dict(capabilities) if isinstance(supported_parameters, list) and supported_parameters: entry['supported_parameters'] = list(supported_parameters) entries.append(entry) return entries return _load_cached_model_catalog(cache_key, _load) def _load_named_custom_providers(all_env: dict | None = None) -> list[dict]: env = all_env or _load_profile_env() custom_providers = cfg.get('custom_providers', []) if not isinstance(custom_providers, list): custom_providers = [] # Also read from the 'providers' dict section (hermes-agent format). # This section maps provider_id -> {api, name, transport, api_key, ...}. providers_dict = cfg.get('providers', {}) if isinstance(providers_dict, dict): for pid, raw in providers_dict.items(): if not isinstance(raw, dict): continue # Normalize to the same shape as custom_providers entries custom_providers.append({ 'name': raw.get('name', pid), 'base_url': raw.get('api', ''), 'api_key': raw.get('api_key', ''), 'api_mode': raw.get('transport', ''), }) entries: list[dict] = [] seen: set[str] = set() for raw_entry in custom_providers: if not isinstance(raw_entry, dict): continue raw_name = str(raw_entry.get('name') or '').strip() base_url = str(raw_entry.get('base_url') or '').strip().rstrip('/') if not raw_name or not base_url: continue provider_id = _normalize_provider_id(raw_name) if not provider_id: continue # Keep Moonshot/DeepSeek/OpenAI-style custom aliases folded into their # canonical built-in providers instead of surfacing duplicate groups. if raw_name.lower().startswith('custom-api-'): continue if provider_id in seen: continue seen.add(provider_id) api_key = str(raw_entry.get('api_key') or '').strip() if not api_key: for key in _PROVIDER_ENV_KEYS.get(provider_id, ()): candidate = str(env.get(key) or os.getenv(key) or '').strip() if candidate: api_key = candidate break entries.append({ 'provider_id': provider_id, 'name': raw_name, 'provider': _PROVIDER_DISPLAY.get(provider_id, raw_name.replace('-', ' ').replace('_', ' ').title()), 'base_url': base_url, 'api_key': api_key, 'api_mode': str(raw_entry.get('api_mode') or '').strip(), 'model': str(raw_entry.get('model') or '').strip(), }) return entries def _discover_named_custom_provider_catalogs(all_env: dict | None = None) -> dict[str, list[dict]]: catalogs: dict[str, list[dict]] = {} for entry in _load_named_custom_providers(all_env=all_env): models = _discover_openai_compatible_models( base_url=entry.get('base_url', ''), api_key=entry.get('api_key', ''), cache_key=f"named-custom:{entry.get('provider_id', '')}:{entry.get('base_url', '')}", ) configured_model = entry.get('model', '') if configured_model and configured_model not in {m['id'] for m in models}: models.append({'id': configured_model, 'label': configured_model}) if models: catalogs[entry['provider_id']] = models return catalogs def _provider_is_named_custom(provider_id: str | None) -> bool: provider_id = _normalize_provider_id(provider_id) if not provider_id: return False return any(entry.get('provider_id') == provider_id for entry in _load_named_custom_providers()) def _with_provider_hints(provider_id: str, raw_models: list[dict], active_provider: str | None) -> list[dict]: provider_id = _normalize_provider_id(provider_id) or '' active_provider = _normalize_provider_id(active_provider) models: list[dict] = [] for raw in raw_models or []: if not isinstance(raw, dict): continue model_id = str(raw.get('id') or '').strip() if not model_id: continue routed_id = model_id if provider_id and provider_id != active_provider and not model_id.startswith('@'): routed_id = f'@{provider_id}:{model_id}' entry = dict(raw) entry['id'] = routed_id entry['label'] = str(raw.get('label') or model_id).strip() or model_id models.append(entry) return models def _format_codex_model_label(model_id: str) -> str: raw = (model_id or '').strip() if not raw: return raw raw = raw.replace('_', '-') special = { 'gpt': 'GPT', 'oss': 'OSS', 'codex': 'Codex', 'mini': 'Mini', 'max': 'Max', 'spark': 'Spark', } parts = [] for token in raw.split('-'): low = token.lower() if low in special: parts.append(special[low]) elif low and low[0] == 'o' and low[1:].replace('.', '').isdigit(): parts.append(low) elif low.replace('.', '').isdigit(): parts.append(low) else: parts.append(token.upper() if token.isupper() else token.capitalize()) if len(parts) >= 2 and parts[0] == 'GPT' and parts[1].replace('.', '').isdigit(): parts = [f'GPT-{parts[1]}'] + parts[2:] return ' '.join(parts) def _discover_codex_model_entries() -> list[dict]: def _load() -> list[dict]: access_token = None try: from hermes_cli.auth import resolve_codex_runtime_credentials creds = resolve_codex_runtime_credentials() access_token = creds.get('api_key') except Exception: access_token = None try: from hermes_cli.codex_models import get_codex_model_ids model_ids = get_codex_model_ids(access_token=access_token) except Exception: model_ids = [] seen: set[str] = set() entries: list[dict] = [] for model_id in model_ids or []: if not isinstance(model_id, str): continue model_id = model_id.strip() if not model_id or model_id in seen: continue seen.add(model_id) entries.append({ 'id': model_id, 'label': _format_codex_model_label(model_id), }) return entries return _load_cached_model_catalog('openai-codex', _load) def _discover_openrouter_free_models(all_env: dict | None = None) -> list[dict]: env = all_env or _load_profile_env() def _load() -> list[dict]: import urllib.request headers = {'Accept': 'application/json'} api_key = env.get('OPENROUTER_API_KEY') or os.getenv('OPENROUTER_API_KEY') if api_key: headers['Authorization'] = f'Bearer {api_key}' req = urllib.request.Request('https://openrouter.ai/api/v1/models', headers=headers) with urllib.request.urlopen(req, timeout=15) as response: payload = json.loads(response.read().decode('utf-8')) entries = [] seen: set[str] = set() for item in payload.get('data', []) if isinstance(payload, dict) else []: if not isinstance(item, dict): continue model_id = str(item.get('id') or '').strip() if not model_id or model_id in seen: continue pricing = item.get('pricing') or {} if not (_looks_like_free_price(pricing.get('prompt')) and _looks_like_free_price(pricing.get('completion'))): continue arch = item.get('architecture') or {} inputs = {str(v).lower() for v in arch.get('input_modalities') or []} outputs = {str(v).lower() for v in arch.get('output_modalities') or []} if 'text' not in inputs or 'text' not in outputs or 'audio' in outputs: continue seen.add(model_id) name = str(item.get('name') or '').strip() if model_id == 'openrouter/free': label = 'Auto (Free Router)' else: label = name or model_id supported = {str(v).lower() for v in item.get('supported_parameters') or []} entries.append({ 'id': model_id, 'label': label, '_sort_tools': 0 if 'tools' in supported else 1, 'supported_parameters': sorted(supported), }) # Also include non-free models the user explicitly wants _wanted_nonfree = { 'z-ai/glm-5.1', 'qwen/qwen3.6-plus', } for item in payload.get('data', []) if isinstance(payload, dict) else []: if not isinstance(item, dict): continue model_id = str(item.get('id') or '').strip() if not model_id or model_id in seen or model_id not in _wanted_nonfree: continue seen.add(model_id) name = str(item.get('name') or '').strip() entries.append({ 'id': model_id, 'label': name or model_id, }) entries.sort(key=lambda item: ( 0 if item['id'] == 'openrouter/free' else 1, item.get('_sort_tools', 1), item['label'].lower(), )) for item in entries: item.pop('_sort_tools', None) return entries models = _load_cached_model_catalog('openrouter-free', _load) if models: return models return [{'id': m['id'], 'label': m['label']} for m in _OPENROUTER_FREE_MODELS] def resolve_model_provider(model_id: str) -> tuple: """Resolve model name, provider, and base_url for AIAgent. Model IDs from the dropdown can be in several formats: - 'claude-sonnet-4.6' (bare name, uses config default provider) - 'anthropic/claude-sonnet-4.6' (OpenRouter format, provider/model) - '@minimax:MiniMax-M2.7' (explicit provider hint from dropdown) The @provider:model format is used for models from non-default provider groups in the dropdown, so we can route them through the correct provider via resolve_runtime_provider(requested=provider) instead of the default. Returns (model, provider, base_url) where provider and base_url may be None. """ config_provider = None config_base_url = None model_cfg = cfg.get('model', {}) if isinstance(model_cfg, dict): config_provider = model_cfg.get('provider') config_base_url = model_cfg.get('base_url') config_provider = _normalize_provider_id(config_provider) config_provider_is_named_custom = _provider_is_named_custom(config_provider) model_id = (model_id or '').strip() if not model_id: return model_id, config_provider, config_base_url # @provider:model format — explicit provider hint from the dropdown. # Route through that provider directly (resolve_runtime_provider will # resolve credentials in streaming.py). if model_id.startswith('@') and ':' in model_id: provider_hint, bare_model = model_id[1:].split(':', 1) return bare_model, provider_hint, None if '/' in model_id: prefix, bare = model_id.split('/', 1) # OpenRouter always needs the full provider/model path (e.g. openrouter/free, # anthropic/claude-sonnet-4.6). Never strip the prefix for OpenRouter. if config_provider == 'openrouter': return model_id, 'openrouter', config_base_url # If prefix matches config provider exactly, strip it and use that provider directly. # e.g. config=anthropic, model=anthropic/claude-... → bare name to anthropic API if config_provider and prefix == config_provider: return bare, config_provider, config_base_url # If prefix does NOT match config provider, the user picked a cross-provider model # from the OpenRouter dropdown (e.g. config=anthropic but picked openai/gpt-5.4-mini). # In this case always route through openrouter with the full provider/model string. if prefix in _PROVIDER_MODELS and prefix != config_provider and not config_provider_is_named_custom: return model_id, 'openrouter', None return model_id, config_provider, config_base_url def get_available_models() -> dict: """ Return available models grouped by provider. Discovery order: 1. Read config.yaml 'model' section for active provider info 2. Check for known API keys in env or ~/.hermes/.env 3. Fetch models from custom endpoint if base_url is configured 4. Fall back to hardcoded model list (OpenRouter-style) Returns: { 'active_provider': str|None, 'default_model': str, 'groups': [{'provider': str, 'models': [{'id': str, 'label': str}]}] } """ active_provider = None default_model = DEFAULT_MODEL groups = [] all_env = _load_profile_env() auth_store = _load_auth_store() named_custom_providers = _load_named_custom_providers(all_env=all_env) named_custom_catalogs = _discover_named_custom_provider_catalogs(all_env=all_env) # 1. Read config.yaml model section cfg_base_url = '' # must be defined before conditional blocks (#117) model_cfg = cfg.get('model', {}) cfg_base_url = '' if isinstance(model_cfg, str): default_model = model_cfg elif isinstance(model_cfg, dict): active_provider = _normalize_provider_id(model_cfg.get('provider')) cfg_default = model_cfg.get('default', '') cfg_base_url = model_cfg.get('base_url', '') if cfg_default: default_model = cfg_default # 2. Also check env vars for model override env_model = os.getenv('HERMES_MODEL') or os.getenv('OPENAI_MODEL') or os.getenv('LLM_MODEL') if env_model: default_model = env_model.strip() # 3. Try to read auth store for active provider (if hermes is installed) if not active_provider: active_provider = _normalize_provider_id(auth_store.get('active_provider')) # 4. Detect available providers. # Primary: ask hermes-agent's auth layer — the authoritative source. Then # merge in auth.json and .env detection so custom local aliases remain visible. detected_providers = set() if active_provider: detected_providers.add(active_provider) try: from hermes_cli.models import list_available_providers as _lap from hermes_cli.auth import get_auth_status as _gas for _p in _lap(): if not _p.get('authenticated'): continue # Exclude providers whose credential came from an ambient token # (e.g. 'gh auth token' for Copilot on a machine with gh CLI auth). # Only include providers with an explicit, dedicated credential. try: _src = _gas(_p['id']).get('key_source', '') if _src == 'gh auth token': continue except Exception: pass _pid = _normalize_provider_id(_p['id']) if _pid: detected_providers.add(_pid) except Exception: pass # Supplement auth-layer discovery with the raw auth store. This keeps OAuth # providers visible even if the helper import fails or only partially loads. for _pid in (auth_store.get('providers') or {}).keys(): _norm_pid = _normalize_provider_id(_pid) if _norm_pid and _auth_store_has_provider(auth_store, _norm_pid): detected_providers.add(_norm_pid) for _pid in (auth_store.get('credential_pool') or {}).keys(): _norm_pid = _normalize_provider_id(_pid) if _norm_pid: detected_providers.add(_norm_pid) # Always merge direct env detection as well. Hermes' auth helpers do not # cover every custom env alias we accept in local .env files. for _pid, _keys in _PROVIDER_ENV_KEYS.items(): if any(all_env.get(k) for k in _keys): detected_providers.add(_pid) for _entry in named_custom_providers: _pid = _entry.get('provider_id') if _pid: detected_providers.add(_pid) allowed_providers = _get_allowed_providers() if allowed_providers: detected_providers = {pid for pid in detected_providers if pid in allowed_providers} if active_provider and active_provider not in allowed_providers: active_provider = None named_custom_providers = [ entry for entry in named_custom_providers if entry.get('provider_id') in allowed_providers ] named_custom_catalogs = { provider_id: models for provider_id, models in named_custom_catalogs.items() if provider_id in allowed_providers } hidden_providers = _get_hidden_providers() if hidden_providers: detected_providers = {pid for pid in detected_providers if pid not in hidden_providers} if active_provider and active_provider in hidden_providers: active_provider = None named_custom_providers = [ entry for entry in named_custom_providers if entry.get('provider_id') not in hidden_providers ] named_custom_catalogs = { provider_id: models for provider_id, models in named_custom_catalogs.items() if provider_id not in hidden_providers } # 3. Fetch models from custom endpoint if base_url is configured auto_detected_models = [] endpoint_provider = _infer_provider_from_base_url(cfg_base_url) if cfg_base_url else None if cfg_base_url and endpoint_provider in {'custom', 'ollama', 'lmstudio', 'local'}: try: import urllib.request # Normalize the base_url and build models endpoint base_url = cfg_base_url.strip() if base_url.endswith('/v1'): endpoint_url = base_url + '/models' # /v1/models else: endpoint_url = base_url.rstrip('/') + '/v1/models' # Resolve API key from environment (check profile .env keys too) headers = {} api_key_vars = ('HERMES_API_KEY', 'HERMES_OPENAI_API_KEY', 'OPENAI_API_KEY', 'LOCAL_API_KEY', 'OPENROUTER_API_KEY', 'API_KEY') for key in api_key_vars: api_key = all_env.get(key) or os.getenv(key) if api_key: headers['Authorization'] = f'Bearer {api_key}' break # Fetch model list from endpoint req = urllib.request.Request(endpoint_url, method='GET') for k, v in headers.items(): req.add_header(k, v) with urllib.request.urlopen(req, timeout=10) as response: data = json.loads(response.read().decode('utf-8')) # Handle both OpenAI-compatible and llama.cpp response formats models_list = [] if 'data' in data and isinstance(data['data'], list): models_list = data['data'] elif 'models' in data and isinstance(data['models'], list): models_list = data['models'] for model in models_list: if not isinstance(model, dict): continue model_id = model.get('id', '') or model.get('name', '') or model.get('model', '') model_name = model.get('name', '') or model.get('model', '') or model_id if model_id and model_name: auto_detected_models.append({'id': model_id, 'label': model_name}) detected_providers.add(endpoint_provider) except Exception: pass # custom endpoint unreachable or misconfigured -- fail silently # 3b. Include models from custom_providers config entries. # These are explicitly configured and should always appear even when the # /v1/models endpoint is unreachable or returns a subset. _custom_providers_cfg = cfg.get('custom_providers', []) if isinstance(_custom_providers_cfg, list): _seen_custom_ids = {m['id'] for m in auto_detected_models} for _cp in _custom_providers_cfg: if not isinstance(_cp, dict): continue _cp_model = _cp.get('model', '') if _cp_model and _cp_model not in _seen_custom_ids: _cp_label = _cp_model.split('/')[-1] if '/' in _cp_model else _cp_model auto_detected_models.append({'id': _cp_model, 'label': _cp_label}) _seen_custom_ids.add(_cp_model) detected_providers.add('custom') # 5. Build model groups if detected_providers: for pid in sorted(detected_providers, key=lambda x: _provider_sort_key(x, active_provider)): provider_name = _PROVIDER_DISPLAY.get(pid, pid.title()) if pid == 'openrouter': # Keep OpenRouter focused on free models in the chat sidebar. openrouter_models = _annotate_reasoning_metadata( pid, _discover_openrouter_free_models(all_env=all_env), ) groups.append({ 'provider_id': pid, 'provider': 'OpenRouter', 'models': _with_provider_hints(pid, openrouter_models, active_provider), }) elif pid in _PROVIDER_MODELS: # For non-default providers, prefix model IDs with @provider:model # so resolve_model_provider() routes through that specific provider # via resolve_runtime_provider(requested=provider). # The default provider's models keep bare names for direct API routing. raw_models = _PROVIDER_MODELS[pid] if pid == 'openai-codex': raw_models = _discover_codex_model_entries() or raw_models raw_models = _annotate_reasoning_metadata(pid, raw_models) models = _with_provider_hints(pid, raw_models, active_provider) groups.append({ 'provider_id': pid, 'provider': provider_name, 'models': models, }) elif pid in named_custom_catalogs: raw_models = _annotate_reasoning_metadata(pid, named_custom_catalogs.get(pid, [])) groups.append({ 'provider_id': pid, 'provider': provider_name, 'models': _with_provider_hints(pid, raw_models, active_provider), }) else: # Unknown provider -- use auto-detected models if available, # otherwise fall back to default model placeholder if auto_detected_models: raw_models = _annotate_reasoning_metadata(pid, auto_detected_models) groups.append({ 'provider_id': pid, 'provider': provider_name, 'models': _with_provider_hints(pid, raw_models, active_provider), }) elif pid.startswith('custom'): # Ignore auth-store custom stubs that have no discoverable # model catalog behind them. Showing a one-item placeholder # clutters the picker and is not actionable in the WebUI. continue else: fallback_model = { 'id': default_model, 'label': default_model.split('/')[-1], } fallback_model['reasoning'] = _reasoning_metadata_for_model(pid, default_model, raw_entry=fallback_model) groups.append({ 'provider_id': pid, 'provider': provider_name, 'models': [fallback_model], }) else: # No providers detected. Show only the configured default model so the user # can at least send messages with their current setting. Avoid showing a # generic multi-provider list — those models wouldn't be routable anyway. label = default_model.split('/')[-1] if '/' in default_model else default_model fallback_model = {'id': default_model, 'label': label} fallback_model['reasoning'] = _reasoning_metadata_for_model('default', default_model, raw_entry=fallback_model) groups.append({'provider_id': 'default', 'provider': 'Default', 'models': [fallback_model]}) # Ensure the user's configured default_model always appears in the dropdown. # It may be missing if the model isn't in any hardcoded list (e.g. openrouter/free, # a custom local model, or any model.default not in _OPENROUTER_FREE_MODELS). # Normalize before comparing: strip provider prefix so 'anthropic/claude-opus-4.6' # matches 'claude-opus-4.6' already in the list and avoids a duplicate entry. if default_model: def _norm(mid): mid = mid or '' if mid.startswith('@') and ':' in mid: mid = mid.split(':', 1)[1] if '/' in mid: mid = mid.split('/', 1)[-1] return mid all_ids_norm = {_norm(m['id']) for g in groups for m in g.get('models', [])} if _norm(default_model) not in all_ids_norm: # Determine which group to inject into label = default_model.split('/')[-1] if '/' in default_model else default_model injected = False for g in groups: if active_provider and g.get('provider_id') == active_provider: fallback_model = {'id': default_model, 'label': label} fallback_model['reasoning'] = _reasoning_metadata_for_model( g.get('provider_id'), default_model, raw_entry=fallback_model, ) g['models'].insert(0, fallback_model) injected = True break if not injected and groups: fallback_model = {'id': default_model, 'label': label} fallback_model['reasoning'] = _reasoning_metadata_for_model( groups[0].get('provider_id'), default_model, raw_entry=fallback_model, ) groups[0]['models'].insert(0, fallback_model) elif not groups: fallback_model = {'id': default_model, 'label': label} fallback_model['reasoning'] = _reasoning_metadata_for_model( active_provider or 'default', default_model, raw_entry=fallback_model, ) groups.append({ 'provider_id': active_provider or 'default', 'provider': _PROVIDER_DISPLAY.get(active_provider or 'default', active_provider or 'Default'), 'models': [fallback_model], }) return { 'active_provider': active_provider, 'default_model': default_model, 'groups': groups, } # ── Static file path ───────────────────────────────────────────────────────── _INDEX_HTML_PATH = REPO_ROOT / 'static' / 'index.html' # ── Thread synchronisation ─────────────────────────────────────────────────── LOCK = threading.Lock() SESSIONS_MAX = 100 CHAT_LOCK = threading.Lock() STREAMS: dict = {} STREAMS_LOCK = threading.Lock() CANCEL_FLAGS: dict = {} SERVER_START_TIME = time.time() # ── Thread-local env context ───────────────────────────────────────────────── _thread_ctx = threading.local() def _set_thread_env(**kwargs): _thread_ctx.env = kwargs def _clear_thread_env(): _thread_ctx.env = {} # ── Per-session agent locks ─────────────────────────────────────────────────── SESSION_AGENT_LOCKS: dict = {} SESSION_AGENT_LOCKS_LOCK = threading.Lock() def _get_session_agent_lock(session_id: str) -> threading.Lock: with SESSION_AGENT_LOCKS_LOCK: if session_id not in SESSION_AGENT_LOCKS: SESSION_AGENT_LOCKS[session_id] = threading.Lock() return SESSION_AGENT_LOCKS[session_id] # ── Settings persistence ───────────────────────────────────────────────────── _SETTINGS_DEFAULTS = { # default_model removed — Hermes config.yaml is the single source of truth 'default_workspace': str(DEFAULT_WORKSPACE), 'send_key': 'enter', # 'enter' or 'ctrl+enter' 'show_token_usage': False, # show input/output token badge below assistant messages 'show_cli_sessions': False, # merge CLI sessions from state.db into the sidebar 'sync_to_insights': False, # mirror WebUI token usage to state.db for /insights 'check_for_updates': True, # check if webui/agent repos are behind upstream 'theme': 'dark', # active UI theme name (no enum gate -- allows custom themes) 'bot_name': os.getenv('HERMES_WEBUI_BOT_NAME', 'Hermes'), # display name for the assistant 'password_hash': None, # SHA-256 hash; None = auth disabled } def _load_stored_settings() -> dict: """Load only persisted settings keys from disk. Older builds stored keys like ``default_model`` in settings.json. Those keys are now derived from config.yaml and must not leak back into API responses or be re-persisted forever. """ stored_settings = {} if SETTINGS_FILE.exists(): try: stored = json.loads(SETTINGS_FILE.read_text(encoding='utf-8')) if isinstance(stored, dict): for key in _SETTINGS_DEFAULTS: if key in stored: stored_settings[key] = stored[key] except Exception: pass return stored_settings def _current_default_model() -> str: """Return the effective default model from config.yaml.""" default_model = DEFAULT_MODEL try: cfg = get_config() model_cfg = cfg.get('model', {}) if isinstance(model_cfg, str) and model_cfg.strip(): default_model = model_cfg.strip() elif isinstance(model_cfg, dict): model_value = str(model_cfg.get('default', '') or '').strip() if model_value: default_model = model_value except Exception: pass return default_model def load_settings() -> dict: """Load settings from disk, merging with defaults for any missing keys. ``default_model`` is returned as a read-only compatibility field derived from config.yaml so every UI surface reports the same source of truth. """ settings = dict(_SETTINGS_DEFAULTS) settings.update(_load_stored_settings()) settings['default_model'] = _current_default_model() return settings _SETTINGS_ALLOWED_KEYS = set(_SETTINGS_DEFAULTS.keys()) - {'password_hash'} _SETTINGS_ENUM_VALUES = { 'send_key': {'enter', 'ctrl+enter'}, } _SETTINGS_BOOL_KEYS = {'show_token_usage', 'show_cli_sessions', 'sync_to_insights', 'check_for_updates'} def save_settings(settings: dict) -> dict: """Save settings to disk. Returns the merged settings. Ignores unknown keys.""" import hashlib as _hl current = dict(_SETTINGS_DEFAULTS) current.update(_load_stored_settings()) # Handle _set_password: hash and store as password_hash raw_pw = settings.pop('_set_password', None) if raw_pw and isinstance(raw_pw, str) and raw_pw.strip(): salt = str(STATE_DIR).encode() current['password_hash'] = _hl.sha256(salt + raw_pw.strip().encode()).hexdigest() # Handle _clear_password: explicitly disable auth if settings.pop('_clear_password', False): current['password_hash'] = None for k, v in settings.items(): if k in _SETTINGS_ALLOWED_KEYS: # Validate enum-constrained keys if k in _SETTINGS_ENUM_VALUES and v not in _SETTINGS_ENUM_VALUES[k]: continue # Coerce bool keys if k in _SETTINGS_BOOL_KEYS: v = bool(v) current[k] = v SETTINGS_FILE.write_text( json.dumps(current, ensure_ascii=False, indent=2), encoding='utf-8', ) # Update runtime defaults so new sessions use them immediately global DEFAULT_MODEL, DEFAULT_WORKSPACE if 'default_model' in current: DEFAULT_MODEL = current['default_model'] if 'default_workspace' in current: DEFAULT_WORKSPACE = Path(current['default_workspace']).expanduser().resolve() merged = load_settings() merged['password_hash'] = current.get('password_hash') return merged # Apply saved settings on startup (override env-derived defaults) _startup_settings = load_settings() if SETTINGS_FILE.exists(): if _startup_settings.get('default_workspace'): DEFAULT_WORKSPACE = Path(_startup_settings['default_workspace']).expanduser().resolve() # ── Override DEFAULT_MODEL from Hermes config.yaml (single source of truth) ─── # settings.json never overrides the model — only config.yaml / env vars do. try: _cfg_startup = get_config() _m = _cfg_startup.get('model', {}) if isinstance(_m, dict): _m = _m.get('default', '') if isinstance(_m, str) and _m.strip(): DEFAULT_MODEL = _m.strip() except Exception: pass # fall through to env-derived DEFAULT_MODEL # ── SESSIONS in-memory cache (LRU OrderedDict) ─────────────────────────────── SESSIONS: collections.OrderedDict = collections.OrderedDict() # ── Profile state initialisation ──────────────────────────────────────────── # Must run after all imports are resolved to correctly patch module-level caches try: from api.profiles import init_profile_state init_profile_state() except ImportError: pass # hermes_cli not available -- default profile only