diff --git a/hermes_cli/codex_runtime_switch.py b/hermes_cli/codex_runtime_switch.py index c511c6cf38..dad696bff7 100644 --- a/hermes_cli/codex_runtime_switch.py +++ b/hermes_cli/codex_runtime_switch.py @@ -15,8 +15,7 @@ VALID_RUNTIMES = ("auto", "codex_app_server") # Human-friendly synonyms accepted by parse_args. _ARG_SYNONYMS = { "on": "codex_app_server", "codex": "codex_app_server", "enable": "codex_app_server", - "off": "auto", "default": "auto", "disable": "auto", "hermes": "auto", -} + "off": "auto", "default": "auto", "disable": "auto", "hermes": "auto"} _HERMES_TOOLS_CALLBACK_NOTE = ( "Hermes tool callback registered: codex can now use " @@ -25,8 +24,7 @@ _HERMES_TOOLS_CALLBACK_NOTE = ( "kanban_* (worker + orchestrator) via MCP.", " (delegate_task, memory, session_search, todo run " "only on the default Hermes runtime — they need the " - "agent loop context.)", -) + "agent loop context.)") @dataclass @@ -99,15 +97,13 @@ def _migration_lines(config: dict) -> list[str]: if mig_report.migrated_plugins: lines.append( f"Migrated {len(mig_report.migrated_plugins)} native " - f"Codex plugin(s): {', '.join(mig_report.migrated_plugins)}" - ) + f"Codex plugin(s): {', '.join(mig_report.migrated_plugins)}") elif mig_report.plugin_query_error: lines.append(f"Codex plugin discovery skipped: {mig_report.plugin_query_error}") if mig_report.wrote_permissions_default: lines.append( f"Default sandbox: {mig_report.wrote_permissions_default} " - f"(no approval prompt on every write)" - ) + f"(no approval prompt on every write)") if "hermes-tools" in mig_report.migrated: lines.extend(_HERMES_TOOLS_CALLBACK_NOTE) lines.append(f" (config: {mig_report.target_path})") @@ -119,11 +115,7 @@ def _migration_lines(config: dict) -> list[str]: def apply( - config: dict, - new_value: Optional[str], - *, - persist_callback=None, -) -> CodexRuntimeStatus: + config: dict, new_value: Optional[str], *, persist_callback=None) -> CodexRuntimeStatus: """Top-level entry point used by both CLI and gateway handlers. ``config`` is mutated in place when ``new_value`` is set (None means show current state only). @@ -151,8 +143,7 @@ def apply( if new_value == current and not reapplying_enable: return CodexRuntimeStatus( success=True, new_value=current, old_value=current, - message=f"openai_runtime already set to {current}", - ) + message=f"openai_runtime already set to {current}") # Switching ON: verify codex CLI before persisting — an opt-in toggle that silently fails on # the first turn is the worst possible UX. @@ -164,9 +155,7 @@ def apply( message=( "Cannot enable codex_app_server runtime: " f"{ver_or_msg or 'codex CLI not available'}\n" - "Install with: npm i -g @openai/codex" - ), - ) + "Install with: npm i -g @openai/codex")) if not reapplying_enable: set_runtime(config, new_value) @@ -177,14 +166,12 @@ def apply( logger.exception("failed to persist openai_runtime change") return CodexRuntimeStatus( success=False, new_value=new_value, old_value=current, - message=f"updated config in memory but persist failed: {exc}", - ) + message=f"updated config in memory but persist failed: {exc}") msg_lines = [ f"openai_runtime already set to {current} — re-applying migration" if reapplying_enable - else f"openai_runtime: {current} → {new_value}" - ] + else f"openai_runtime: {current} → {new_value}"] if new_value == "codex_app_server": ok, ver = _check_binary_cached() if ok: @@ -195,16 +182,13 @@ def apply( msg_lines.append( "OpenAI/Codex turns now run through `codex app-server` " "(terminal/file ops/patching inside Codex; " - "Hermes tools available via MCP callback)." - ) + "Hermes tools available via MCP callback).") msg_lines.append( "Effective on next session — current cached agent keeps " - "the prior runtime to preserve prompt cache." - ) + "the prior runtime to preserve prompt cache.") else: msg_lines.append("OpenAI/Codex turns will use the default Hermes runtime.") msg_lines.append("Effective on next session.") return CodexRuntimeStatus( success=True, new_value=new_value, old_value=current, - message="\n".join(msg_lines), requires_new_session=True, - ) + message="\n".join(msg_lines), requires_new_session=True) diff --git a/hermes_cli/config_migrations.py b/hermes_cli/config_migrations.py index ffc30e3ddc..f962c69584 100644 --- a/hermes_cli/config_migrations.py +++ b/hermes_cli/config_migrations.py @@ -28,8 +28,7 @@ def support_floor_message() -> str: "and can no longer be auto-migrated. Back up " f"{display_hermes_home()}/config.yaml and run `hermes setup` to " f"regenerate, or manually set _config_version: {SUPPORT_FLOOR_VERSION} " - "after reviewing the changelog." - ) + "after reviewing the changelog.") def _cfg(): @@ -58,8 +57,7 @@ def _commit( results: Dict[str, Any], quiet: bool, added: Optional[str], - message: Optional[str], -) -> None: + message: Optional[str]) -> None: """Persist *config*, record *added* under ``config_added`` and print *message* unless quiet.""" _persist_migration(config) if added: @@ -68,26 +66,30 @@ def _commit( print(message) -def _rewrite_stale_default( +def _rewrite_key( results: Dict[str, Any], quiet: bool, *, section: str, key: str, - old: Any, + match: Callable[[Any], bool], new: Any, added: str, message: str, extra_guard: Callable[[Dict[str, Any]], bool] = lambda _m: True, -) -> None: - """Rewrite ``
.`` only when it still equals the OLD default. + create_section: bool = False) -> None: + """Rewrite ``
.`` to *new* when ``match(current_value)`` holds. - Never clobbers a value the user deliberately customized; unset keys inherit the new default at - read time. ``new=None`` deletes the key instead of assigning. + ``new=None`` deletes the key. A missing/non-mapping section is skipped unless + *create_section* (then ``match(None)`` decides). """ config = read_raw_config() raw = config.get(section) - if isinstance(raw, dict) and raw.get(key) == old and extra_guard(raw): + if not isinstance(raw, dict): + if not create_section: + return + raw = {} + if match(raw.get(key)) and extra_guard(raw): if new is None: del raw[key] else: @@ -96,6 +98,19 @@ def _rewrite_stale_default( _commit(config, results, quiet, added, message) +def _rewrite_stale_default(results: Dict[str, Any], quiet: bool, *, old: Any, **kw: Any) -> None: + """Rewrite a key only while it still equals the OLD default. + + Never clobbers a value the user deliberately customized; unset keys inherit the new default at + read time. + """ + _rewrite_key(results, quiet, match=lambda cur: cur == old, **kw) + + +def _lower_is(word: str) -> Callable[[Any], bool]: + return lambda cur: isinstance(cur, str) and cur.strip().lower() == word + + def _migrate_to_12(results: Dict[str, Any], quiet: bool) -> None: # 11 → 12: custom_providers list → providers dict. _custom_provider_entry_to_provider_config = _cfg()._custom_provider_entry_to_provider_config @@ -173,8 +188,7 @@ _LOCAL_WHISPER_MODELS = frozenset({ "medium.en", "medium", "large-v1", "large-v2", "large-v3", "large", "distil-large-v2", "distil-medium.en", "distil-small.en", "distil-large-v3", "distil-large-v3.5", - "large-v3-turbo", "turbo", -}) + "large-v3-turbo", "turbo"}) def _migrate_to_14(results: Dict[str, Any], quiet: bool) -> None: @@ -203,8 +217,7 @@ def _migrate_to_14(results: Dict[str, Any], quiet: bool) -> None: _place(provider) config["stt"] = stt _commit( - config, results, quiet, None, " ✓ Migrated legacy stt.model to provider-specific config" - ) + config, results, quiet, None, " ✓ Migrated legacy stt.model to provider-specific config") def _migrate_to_16(results: Dict[str, Any], quiet: bool) -> None: @@ -226,8 +239,7 @@ def _migrate_to_16(results: Dict[str, Any], quiet: bool) -> None: _commit( config, results, quiet, "display.platforms (migrated from tool_progress_overrides)", - f" ✓ Migrated tool_progress_overrides → display.platforms: {migrated}", - ) + f" ✓ Migrated tool_progress_overrides → display.platforms: {migrated}") def _migrate_to_17(results: Dict[str, Any], quiet: bool) -> None: @@ -253,8 +265,7 @@ def _migrate_to_17(results: Dict[str, Any], quiet: bool) -> None: message = ( " ✓ Migrated compression.summary_* → auxiliary.compression: " f"{', '.join(migrated_keys)}" - if migrated_keys else " ✓ Removed unused compression.summary_* keys" - ) + if migrated_keys else " ✓ Removed unused compression.summary_* keys") _commit(config, results, quiet, None, message) @@ -302,12 +313,10 @@ def _migrate_to_21(results: Dict[str, Any], quiet: bool) -> None: f"{len(grandfathered)} existing plugin(s) into plugins.enabled" if grandfathered else " ✓ Plugins now opt-in: no existing plugins to grandfather. " - "Use `hermes plugins enable ` to activate." - ) + "Use `hermes plugins enable ` to activate.") _commit( config, results, quiet, - f"plugins.enabled (opt-in allow-list, {len(grandfathered)} grandfathered)", message, - ) + f"plugins.enabled (opt-in allow-list, {len(grandfathered)} grandfathered)", message) def _migrate_to_23(results: Dict[str, Any], quiet: bool) -> None: @@ -339,8 +348,7 @@ def _migrate_to_23(results: Dict[str, Any], quiet: bool) -> None: raw_aux = _dict_at(config, "auxiliary") raw_aux_curator = _dict_at(raw_aux, "curator") added_aux = _seed_missing( - raw_aux_curator, DEFAULT_CONFIG.get("auxiliary", {}).get("curator", {}) - ) + raw_aux_curator, DEFAULT_CONFIG.get("auxiliary", {}).get("curator", {})) if added_aux: raw_aux["curator"] = raw_aux_curator config["auxiliary"] = raw_aux @@ -354,8 +362,7 @@ def _migrate_to_23(results: Dict[str, Any], quiet: bool) -> None: if not quiet: print( f" ✓ {'Curator' if label == 'curator' else label} settings now available " - f"({', '.join(added)}) — edit via `hermes config set`" - ) + f"({', '.join(added)}) — edit via `hermes config set`") def _migrate_to_25(results: Dict[str, Any], quiet: bool) -> None: @@ -363,8 +370,7 @@ def _migrate_to_25(results: Dict[str, Any], quiet: bool) -> None: _rewrite_stale_default( results, quiet, section="model_catalog", key="ttl_hours", old=24, new=1, added="model_catalog.ttl_hours 24→1", - message=" ✓ Lowered model_catalog.ttl_hours to 1 (hourly picker refresh)", - ) + message=" ✓ Lowered model_catalog.ttl_hours to 1 (hourly picker refresh)") def _migrate_to_29(results: Dict[str, Any], quiet: bool) -> None: @@ -383,13 +389,10 @@ def _migrate_to_29(results: Dict[str, Any], quiet: bool) -> None: config[subsystem] = sub touched = True results["config_added"].append( - f"{subsystem}.write_mode → write_approval={sub['write_approval']}" - ) + f"{subsystem}.write_mode → write_approval={sub['write_approval']}") if touched: - _commit( - config, results, quiet, None, - " ✓ Renamed write_mode → write_approval (boolean gate)", - ) + _commit(config, results, quiet, None, + " ✓ Renamed write_mode → write_approval (boolean gate)") # 29 → 30 (curator.consolidate defaults to false) is schema-default-only: deep-merge supplies it @@ -399,18 +402,14 @@ def _migrate_to_29(results: Dict[str, Any], quiet: bool) -> None: def _migrate_to_31(results: Dict[str, Any], quiet: bool) -> None: # 30 → 31: verify_on_stop OFF (one-time). The "auto" sentinel was more noise than signal. # Rewrite only when missing or still "auto" — an explicit user true/false is preserved. - config = read_raw_config() - raw_agent = _dict_at(config, "agent") - cur = raw_agent.get("verify_on_stop") - if cur is None or (isinstance(cur, str) and cur.strip().lower() == "auto"): - raw_agent["verify_on_stop"] = False - config["agent"] = raw_agent - _commit( - config, results, quiet, "agent.verify_on_stop=false", + _rewrite_key( + results, quiet, section="agent", key="verify_on_stop", new=False, create_section=True, + match=lambda cur: cur is None or _lower_is("auto")(cur), + added="agent.verify_on_stop=false", + message=( " ✓ Turned off verify-on-stop (agent.verify_on_stop: false). " "Set it to true to re-enable, or \"auto\" for the legacy " - "surface-aware behavior.", - ) + "surface-aware behavior.")) def _migrate_to_32(results: Dict[str, Any], quiet: bool) -> None: @@ -425,10 +424,8 @@ def _migrate_to_32(results: Dict[str, Any], quiet: bool) -> None: " ✓ Turned off verify-on-stop (agent.verify_on_stop: false) — " "the old default was written into your config as a literal " "true. Set it to true again to re-enable, or \"auto\" for the " - "legacy surface-aware behavior." - ), - extra_guard=lambda raw: raw.get("verify_on_stop") is True, - ) + "legacy surface-aware behavior."), + extra_guard=lambda raw: raw.get("verify_on_stop") is True) def _migrate_to_33(results: Dict[str, Any], quiet: bool) -> None: @@ -452,15 +449,13 @@ def _migrate_to_33(results: Dict[str, Any], quiet: bool) -> None: raw_deleg["max_concurrent_children"] = old_async_i results["config_added"].append( f"delegation.max_concurrent_children={old_async_i} " - f"(folded from deprecated max_async_children)" - ) + f"(folded from deprecated max_async_children)") config["delegation"] = raw_deleg _commit( config, results, quiet, None, " ✓ Removed deprecated delegation.max_async_children — " "delegation.max_concurrent_children now caps background " - "delegations too.", - ) + "delegations too.") def _migrate_to_34(results: Dict[str, Any], quiet: bool) -> None: @@ -470,11 +465,7 @@ def _migrate_to_34(results: Dict[str, Any], quiet: bool) -> None: # had turned off. Reset display.personality → "" and scrub agent.system_prompt ONLY when it # verbatim-equals a known personality's rendered text; any other text is user-owned. from hermes_cli.personality import ( - available_personalities, - normalize_personality_name, - prompt_text, - render_personality_prompt, - ) + available_personalities, normalize_personality_name, prompt_text, render_personality_prompt) config = read_raw_config() touched = False @@ -494,8 +485,7 @@ def _migrate_to_34(results: Dict[str, Any], quiet: bool) -> None: manual = prompt_text(raw_agent.get("system_prompt", "")) if manual: rendered = { - render_personality_prompt(defn) - for defn in available_personalities(config).values() + render_personality_prompt(defn) for defn in available_personalities(config).values() } if manual in rendered: raw_agent["system_prompt"] = "" @@ -513,36 +503,27 @@ def _migrate_to_34(results: Dict[str, Any], quiet: bool) -> None: f" ✓ Personality reset to none (was '{old_name}'). Personality " "state was previously saved inconsistently across surfaces and " "could re-enable a personality you had turned off. " - f"Run /personality {old_name} to turn it back on." - ) + f"Run /personality {old_name} to turn it back on.") if scrubbed_text: print( " ✓ Removed personality text from agent.system_prompt (written " "by an older /personality). That field is now reserved for " - "manual system prompts; personalities live in display.personality." - ) + "manual system prompts; personalities live in display.personality.") def _migrate_to_35(results: Dict[str, Any], quiet: bool) -> None: # 34 → 35: background_process_notifications 'all' (old implicit default, rarely chosen on # purpose) → 'concise'. Explicit result/error/off choices are preserved. - config = read_raw_config() - raw_display = config.get("display") - if not isinstance(raw_display, dict): - return - raw_val = raw_display.get("background_process_notifications") - if isinstance(raw_val, str) and raw_val.strip().lower() == "all": - raw_display["background_process_notifications"] = "concise" - config["display"] = raw_display - _commit( - config, results, quiet, - "display.background_process_notifications=concise (was: all)", + _rewrite_key( + results, quiet, section="display", key="background_process_notifications", + match=_lower_is("all"), new="concise", + added="display.background_process_notifications=concise (was: all)", + message=( " ✓ Background process notifications switched from 'all' to " "'concise' — completions now show a one-line status message " "instead of the raw output dump. Set " "display.background_process_notifications: all to restore " - "the old behavior.", - ) + "the old behavior.")) def _migrate_to_36(results: Dict[str, Any], quiet: bool) -> None: @@ -554,9 +535,7 @@ def _migrate_to_36(results: Dict[str, Any], quiet: bool) -> None: " ✓ Raised delegation.max_iterations from 50 to 250 — subagents " "now get a larger per-child tool-call budget so delegated work " "finishes instead of truncating. Set delegation.max_iterations " - "back to 50 to restore the old cap." - ), - ) + "back to 50 to restore the old cap.")) def _migrate_to_37(results: Dict[str, Any], quiet: bool) -> None: @@ -568,9 +547,7 @@ def _migrate_to_37(results: Dict[str, Any], quiet: bool) -> None: " ✓ Raised delegation.max_concurrent_children from 3 to 10 — " "independent delegated children now fan out wider in parallel. " "Each child consumes API tokens independently; set " - "delegation.max_concurrent_children back to 3 to restore the old cap." - ), - ) + "delegation.max_concurrent_children back to 3 to restore the old cap.")) def _migrate_to_38(results: Dict[str, Any], quiet: bool) -> None: @@ -593,8 +570,7 @@ def _migrate_to_38(results: Dict[str, Any], quiet: bool) -> None: message = ( "Removed legacy Relay plugin from plugins.enabled: " f"{', '.join(removed)}. Configure native Relay plugins with " - "HERMES_NEMO_RELAY_PLUGINS_TOML." - ) + "HERMES_NEMO_RELAY_PLUGINS_TOML.") results["warnings"].append(message) if not quiet: print(f" ⚠ {message}") @@ -621,8 +597,7 @@ def _migrate_to_39(results: Dict[str, Any], quiet: bool) -> None: "removed retired 'bfl' toolset from saved toolset lists", " ✓ Removed the retired BFL FLUX 3 toolset from saved toolset " "lists — video generation now lives under `hermes tools` → " - "Video Generation (Nous Subscription or FAL).", - ) + "Video Generation (Nous Subscription or FAL).") def _migrate_to_40(results: Dict[str, Any], quiet: bool) -> None: @@ -632,8 +607,7 @@ def _migrate_to_40(results: Dict[str, Any], quiet: bool) -> None: results, quiet, section="model_catalog", key="ttl_hours", old=1, new=None, added="model_catalog.ttl_hours 1 → ttl_minutes 20 (default)", message=" ✓ Model catalog now refreshes every 20 minutes (model_catalog.ttl_minutes)", - extra_guard=lambda raw: "ttl_minutes" not in raw, - ) + extra_guard=lambda raw: "ttl_minutes" not in raw) #: Registry of (target_version, step), strictly ascending. Later steps observe earlier steps' @@ -660,8 +634,7 @@ MIGRATIONS: Tuple[Tuple[int, Callable[[Dict[str, Any], bool], None]], ...] = ( (37, _migrate_to_37), (38, _migrate_to_38), (39, _migrate_to_39), - (40, _migrate_to_40), -) + (40, _migrate_to_40)) def run_migrations(current_ver: int, results: Dict[str, Any], quiet: bool) -> None: diff --git a/hermes_cli/config_providers.py b/hermes_cli/config_providers.py index 12b4861505..5586153fa4 100644 --- a/hermes_cli/config_providers.py +++ b/hermes_cli/config_providers.py @@ -50,8 +50,7 @@ _API_MODE_ALIASES = { "anthropic-messages": "anthropic_messages", "messages": "anthropic_messages", "bedrock": "bedrock_converse", - "bedrock-converse": "bedrock_converse", -} + "bedrock-converse": "bedrock_converse"} _FALSE_WORDS = frozenset({"false", "0", "no", "off"}) _TRUE_WORDS = frozenset({"true", "1", "yes", "on"}) @@ -111,8 +110,7 @@ _CAMEL_ALIASES: Dict[str, str] = { "apiKeyEnv": "key_env", # OpenClaw-compatible + docs variant "defaultModel": "default_model", "contextLength": "context_length", - "rateLimitDelay": "rate_limit_delay", -} + "rateLimitDelay": "rate_limit_delay"} _KNOWN_PROVIDER_KEYS = { @@ -122,8 +120,7 @@ _KNOWN_PROVIDER_KEYS = { "name", "api", "url", "base_url", "api_key", "key_env", "api_key_env", "key_cmd", "api_mode", "transport", "model", "default_model", "models", "models_discovered", "context_length", "rate_limit_delay", "request_timeout_seconds", "stale_timeout_seconds", - "discover_models", "extra_body", "extra_headers", "capabilities", "ssl_ca_cert", "ssl_verify", -} + "discover_models", "extra_body", "extra_headers", "capabilities", "ssl_ca_cert", "ssl_verify"} def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str: @@ -146,8 +143,7 @@ def _pick_provider_base_url(entry: Dict[str, Any], provider_key: str) -> str: logger.warning( "providers.%s: '%s' value '%s' is not a valid URL " "(no scheme or host) — skipped", - provider_key or "?", url_key, candidate, - ) + provider_key or "?", url_key, candidate) return "" @@ -181,17 +177,13 @@ def _normalize_provider_models(models: Any) -> Tuple[Dict[str, Any], bool]: if not isinstance(model_id, str) or not model_id.strip(): continue normalized_models[model_id.strip()] = { - k: v for k, v in item.items() if k not in {"id", "name"} - } + k: v for k, v in item.items() if k not in {"id", "name"}} return normalized_models, discovered return {}, discovered def _normalize_custom_provider_entry( - entry: Any, - *, - provider_key: str = "", -) -> Optional[Dict[str, Any]]: + entry: Any, *, provider_key: str = "") -> Optional[Dict[str, Any]]: """Return a runtime-compatible custom provider entry or ``None``.""" if not isinstance(entry, dict): return None @@ -211,16 +203,14 @@ def _normalize_custom_provider_entry( provider_key, f"camel:{camel}", "providers.%s: camelCase key '%s' auto-mapped to '%s' " "(use snake_case to avoid this warning)", - provider_key or "?", camel, snake, - ) + provider_key or "?", camel, snake) entry[snake] = entry[camel] unknown = set(entry.keys()) - _KNOWN_PROVIDER_KEYS - set(_CAMEL_ALIASES.keys()) if unknown: _warn_once_per_provider( provider_key, "unknown:" + ",".join(sorted(unknown)), "providers.%s: unknown config keys ignored: %s", - provider_key or "?", ", ".join(sorted(unknown)), - ) + provider_key or "?", ", ".join(sorted(unknown))) base_url = _pick_provider_base_url(entry, provider_key) if not base_url: @@ -243,38 +233,30 @@ def _normalize_custom_provider_entry( break return val.strip() if isinstance(val, str) else "" - if _stripped("api_key"): - normalized["api_key"] = _stripped("api_key") + def _put(field: str, value: Any) -> None: + if value: + normalized[field] = value + _put("api_key", _stripped("api_key")) key_env = _stripped("key_env", "api_key_env") - if key_env: - normalized["key_env"] = key_env - if entry.get("api_key_env") and not entry.get("key_env"): - normalized["api_key_env"] = key_env - + _put("key_env", key_env) + if key_env and entry.get("api_key_env") and not entry.get("key_env"): + normalized["api_key_env"] = key_env api_mode = _stripped("api_mode", "transport") - if api_mode: - normalized["api_mode"] = _canonical_api_mode(api_mode) - - model_name = _stripped("model", "default_model") - if model_name: - normalized["model"] = model_name + _put("api_mode", _canonical_api_mode(api_mode) if api_mode else "") + _put("model", _stripped("model", "default_model")) # ``models_discovered`` marks a mapping auto-discovered by Hermes, not hand-curated. models_dict, discovered = _normalize_provider_models(entry.get("models")) - if models_dict: - normalized["models"] = models_dict + _put("models", models_dict) if entry.get("models_discovered") is True or discovered: normalized["models_discovered"] = True capabilities = entry.get("capabilities") if isinstance(capabilities, dict): - normalized_capabilities = { + _put("capabilities", { key: value for key, value in capabilities.items() - if isinstance(key, str) and isinstance(value, bool) - } - if normalized_capabilities: - normalized["capabilities"] = normalized_capabilities + if isinstance(key, str) and isinstance(value, bool)}) context_length = entry.get("context_length") if isinstance(context_length, int) and context_length > 0: @@ -291,13 +273,8 @@ def _normalize_custom_provider_entry( normalized["extra_body"] = dict(entry["extra_body"]) # Per-provider extra HTTP headers may carry credentials — never log them downstream. - normalized_headers = normalize_extra_headers(entry.get("extra_headers")) - if normalized_headers: - normalized["extra_headers"] = normalized_headers - - ssl_ca_cert = _stripped("ssl_ca_cert") - if ssl_ca_cert: - normalized["ssl_ca_cert"] = ssl_ca_cert + _put("extra_headers", normalize_extra_headers(entry.get("extra_headers"))) + _put("ssl_ca_cert", _stripped("ssl_ca_cert")) ssl_verify = entry.get("ssl_verify") if isinstance(ssl_verify, bool): @@ -309,10 +286,7 @@ def _normalize_custom_provider_entry( def _custom_provider_entry_to_provider_config( - entry: Any, - *, - provider_key: str = "", -) -> Optional[Dict[str, Any]]: + entry: Any, *, provider_key: str = "") -> Optional[Dict[str, Any]]: """Translate a legacy custom provider entry to the v12 providers shape.""" normalized = _normalize_custom_provider_entry(entry, provider_key=provider_key) if normalized is None: @@ -322,8 +296,7 @@ def _custom_provider_entry_to_provider_config( for field in ( "name", "api_key", "key_env", "models", "models_discovered", "context_length", "rate_limit_delay", "discover_models", "extra_body", "extra_headers", - "ssl_ca_cert", "ssl_verify", - ): + "ssl_ca_cert", "ssl_verify"): if field in normalized: provider_entry[field] = normalized[field] if "model" in normalized: @@ -348,8 +321,7 @@ def providers_dict_to_custom_providers(providers_dict: Any) -> List[Dict[str, An def get_compatible_custom_providers( - config: Optional[Dict[str, Any]] = None, -) -> List[Dict[str, Any]]: + config: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """Deduplicated list-shaped view over legacy ``custom_providers`` and v12+ ``providers``. Runtime and picker flows need one list; the compatibility layer is never materialised back @@ -359,48 +331,41 @@ def get_compatible_custom_providers( if config is None: config = load_config() + custom_providers = config.get("custom_providers") + if custom_providers is not None and not isinstance(custom_providers, list): + return [] + candidates = [_normalize_custom_provider_entry(e) for e in (custom_providers or [])] + candidates += providers_dict_to_custom_providers(config.get("providers")) + + def _norm(entry: Dict[str, Any], field: str) -> str: + return str(entry.get(field, "") or "").strip().lower() + compatible: List[Dict[str, Any]] = [] seen_provider_keys: set = set() seen_name_url_pairs: set = set() - - def _append_if_new(entry: Optional[Dict[str, Any]]) -> None: + for entry in candidates: if entry is None: - return - provider_key = str(entry.get("provider_key", "") or "").strip().lower() - name = str(entry.get("name", "") or "").strip().lower() + continue + provider_key = _norm(entry, "provider_key") + name = _norm(entry, "name") base_url = str(entry.get("base_url", "") or "").strip().rstrip("/").lower() - model = str(entry.get("model", "") or "").strip().lower() - pair = (name, base_url, model) - + pair = (name, base_url, _norm(entry, "model")) if provider_key and provider_key in seen_provider_keys: - return + continue if name and base_url and pair in seen_name_url_pairs: - return - + continue compatible.append(entry) if provider_key: seen_provider_keys.add(provider_key) if name and base_url: seen_name_url_pairs.add(pair) - - custom_providers = config.get("custom_providers") - if custom_providers is not None: - if not isinstance(custom_providers, list): - return [] - for entry in custom_providers: - _append_if_new(_normalize_custom_provider_entry(entry)) - - for entry in providers_dict_to_custom_providers(config.get("providers")): - _append_if_new(entry) - return compatible def _entries_for_route( base_url: str, custom_providers: Optional[List[Dict[str, Any]]], - config: Optional[Dict[str, Any]], -): + config: Optional[Dict[str, Any]]): """Yield custom-provider entries whose normalized route identity equals *base_url*. Loads ``get_compatible_custom_providers(config)`` when *custom_providers* is None (failures → @@ -438,8 +403,7 @@ def _route_model_cfgs( model: str, base_url: str, custom_providers: Optional[List[Dict[str, Any]]], - config: Optional[Dict[str, Any]], -) -> Iterator[Dict[str, Any]]: + config: Optional[Dict[str, Any]]) -> Iterator[Dict[str, Any]]: """Yield the ``models.`` mapping of every entry matching *base_url*.""" for entry in _entries_for_route(base_url, custom_providers, config): model_cfg = _route_model_cfg(entry, model) @@ -464,8 +428,7 @@ def _coerce_ssl_verify(value: Any) -> Optional[bool]: def get_custom_provider_tls_settings( base_url: str, custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Dict[str, Any]: + config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Return TLS settings from a matching ``custom_providers`` / ``providers`` entry.""" for entry in _entries_for_route(base_url, custom_providers, config): out: Dict[str, Any] = {} @@ -483,8 +446,7 @@ def apply_custom_provider_tls_to_client_kwargs( client_kwargs: Dict[str, Any], base_url: str, custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> None: + config: Optional[Dict[str, Any]] = None) -> None: """Attach per-provider TLS knobs to OpenAI client kwargs when matched.""" tls = get_custom_provider_tls_settings(base_url, custom_providers, config) if tls.get("ssl_ca_cert"): @@ -507,8 +469,7 @@ def normalize_extra_headers(extra_headers: Any) -> Dict[str, str]: def get_custom_provider_extra_headers( base_url: str, custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Dict[str, str]: + config: Optional[Dict[str, Any]] = None) -> Dict[str, str]: """Return ``extra_headers`` of the first route-matching entry that declares any, else ``{}``. SECURITY: values may carry credentials — callers must never log them. @@ -524,8 +485,7 @@ def apply_custom_provider_extra_headers_to_client_kwargs( client_kwargs: Dict[str, Any], base_url: str, custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> None: + config: Optional[Dict[str, Any]] = None) -> None: """Merge per-provider ``extra_headers`` onto OpenAI client ``default_headers``. Provider-specific headers win over SDK/provider defaults already in ``client_kwargs`` (they @@ -543,8 +503,7 @@ def get_custom_provider_context_length( model: str, base_url: str, custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Optional[int]: + config: Optional[Dict[str, Any]] = None) -> Optional[int]: """Per-model ``context_length`` override from a route-matching entry, or ``None``.""" from hermes_cli.config import get_compatible_custom_providers if not model or not base_url: @@ -576,8 +535,7 @@ def get_custom_provider_model_capability( base_url: str, capability: str, custom_providers: Optional[List[Dict[str, Any]]] = None, - config: Optional[Dict[str, Any]] = None, -) -> Optional[bool]: + config: Optional[Dict[str, Any]] = None) -> Optional[bool]: """Explicit boolean capability for one custom-provider model, or ``None``. Matching is scoped to the normalized route and exact runtime model id so aliases can declare diff --git a/hermes_cli/context_switch_guard.py b/hermes_cli/context_switch_guard.py index 50977a6baa..a7e1defba2 100644 --- a/hermes_cli/context_switch_guard.py +++ b/hermes_cli/context_switch_guard.py @@ -26,8 +26,7 @@ def _estimate_tokens(agent: Any, messages: Optional[List[dict]]) -> Optional[int if messages is not None: protect = ( - int(getattr(cc, "protect_first_n", 3)) + int(getattr(cc, "protect_last_n", 20)) + 1 - ) + int(getattr(cc, "protect_first_n", 3)) + int(getattr(cc, "protect_last_n", 20)) + 1) if len(messages) <= protect: return None try: @@ -37,9 +36,7 @@ def _estimate_tokens(agent: Any, messages: Optional[List[dict]]) -> Optional[int tools = getattr(agent, "tools", None) return int( estimate_request_tokens_rough( - messages, system_prompt=system_prompt, tools=tools or None - ) - ) + messages, system_prompt=system_prompt, tools=tools or None)) except Exception: pass @@ -59,8 +56,7 @@ def merge_preflight_compression_warning( config_context_length: int | None = None, configured_model: str | None = None, configured_provider: str | None = None, - configured_base_url: str | None = None, -) -> None: + configured_base_url: str | None = None) -> None: """If the next user message will likely preflight-compress, append a warning.""" if not result.success or agent is None: return @@ -90,8 +86,7 @@ def merge_preflight_compression_warning( config_context_length=config_context_length, configured_model=_or_agent(configured_model, "model"), configured_provider=_or_agent(configured_provider, "provider"), - configured_base_url=_or_agent(configured_base_url, "base_url"), - ) + configured_base_url=_or_agent(configured_base_url, "base_url")) if not new_ctx: return @@ -113,8 +108,7 @@ def merge_preflight_compression_warning( f"Session is ~{estimate:,} tokens; " f"{result.new_model} allows {new_ctx:,} " f"(auto-compress at ~{new_threshold:,}). " - f"Your next message will run preflight compression before the model replies." - ) + f"Your next message will run preflight compression before the model replies.") _append_warning(result, "".join(parts)) @@ -125,8 +119,7 @@ def enrich_model_switch_warnings_for_gateway( session_key: str, source: Any, custom_providers: list | None = None, - load_gateway_config: Callable[[], dict] | None = None, -) -> None: + load_gateway_config: Callable[[], dict] | None = None) -> None: """Gateway helper: cached agent + session DB messages.""" lock = getattr(runner, "_agent_cache_lock", None) cache = getattr(runner, "_agent_cache", None) @@ -139,19 +132,18 @@ def enrich_model_switch_warnings_for_gateway( if agent is None: return - cfg_ctx = None - configured_model = None - configured_provider = None - configured_base_url = None + configured: dict = dict.fromkeys( + ("config_context_length", "configured_model", "configured_provider", "configured_base_url")) if load_gateway_config is not None: try: cfg = load_gateway_config() model_cfg = cfg.get("model", {}) if isinstance(cfg, dict) else {} if isinstance(model_cfg, dict) and model_cfg.get("context_length") is not None: - cfg_ctx = int(model_cfg["context_length"]) - configured_model = model_cfg.get("default") or model_cfg.get("model") - configured_provider = model_cfg.get("provider") - configured_base_url = model_cfg.get("base_url") + configured.update( + config_context_length=int(model_cfg["context_length"]), + configured_model=model_cfg.get("default") or model_cfg.get("model"), + configured_provider=model_cfg.get("provider"), + configured_base_url=model_cfg.get("base_url")) except Exception: pass @@ -166,12 +158,4 @@ def enrich_model_switch_warnings_for_gateway( pass merge_preflight_compression_warning( - result, - agent=agent, - messages=messages, - custom_providers=custom_providers, - config_context_length=cfg_ctx, - configured_model=configured_model, - configured_provider=configured_provider, - configured_base_url=configured_base_url, - ) + result, agent=agent, messages=messages, custom_providers=custom_providers, **configured) diff --git a/hermes_cli/credential_lifecycle.py b/hermes_cli/credential_lifecycle.py index c5225598c8..06d50e8cd7 100644 --- a/hermes_cli/credential_lifecycle.py +++ b/hermes_cli/credential_lifecycle.py @@ -14,8 +14,7 @@ from typing import Any, Dict, List __all__ = [ "save_provider_env_credential", "remove_provider_env_credential", - "purge_env_credential_references", -] + "purge_env_credential_references"] def _providers_for_env_var(env_var: str) -> List[str]: @@ -153,8 +152,7 @@ def _scrub_config_yaml_mirrors(old_value: str, new_value: str | None) -> List[st def purge_env_credential_references( - env_var: str, *, clear_models_cache: bool = True -) -> Dict[str, Any]: + env_var: str, *, clear_models_cache: bool = True) -> Dict[str, Any]: """Remove non-.env references to an env-var credential. Prunes env-seeded pool entries and (optionally) the affected ``provider_models_cache.json`` rows @@ -223,5 +221,4 @@ def remove_provider_env_credential(env_var: str) -> Dict[str, Any]: "pool_pruned": refs["pool_pruned"], "providers": refs["providers"], "config_scrubbed": config_scrubbed, - "found": bool(removed_from_env or refs["pool_pruned"] or config_scrubbed), - } + "found": bool(removed_from_env or refs["pool_pruned"] or config_scrubbed)}