ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

refactor(hermes_cli): group B — _rewrite_key generalization (v31/v35 fold), _put in provider normalizer, dedup loop flatten, gateway configured-dict; AST-neutral layout hug/pack

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:47:14 -07:00
parent e3d78021f3
commit e4a149a64f
5 changed files with 142 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/codex_runtime_switch.py
+12 -28
View File
@@ -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)
hermes_cli/config_migrations.py
+64 -91
View File
@@ -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 ``<section>.<key>`` only when it still equals the OLD default.
create_section: bool = False) -> None:
"""Rewrite ``<section>.<key>`` 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 <name>` to activate."
)
"Use `hermes plugins enable <name>` 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:
hermes_cli/config_providers.py
+49 -91
View File
@@ -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.<model>`` 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
hermes_cli/context_switch_guard.py
+14 -30
View File
@@ -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)
hermes_cli/credential_lifecycle.py
+3 -6
View File
@@ -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)}