mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-22 16:25:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(providers): scope route-owned runtime settings

Browse source
This commit is contained in:
cucurigoo 2026-07-22 03:42:44 +00:00 • committed by kshitij
parent f3f0135154
commit 63dd651b3d
21 changed files with 794 additions and 167 deletions
Download patch file Download diff file Expand all files Collapse all files

149
agent/agent_init.py
View file

@ -28,7 +28,7 @@ import time
import uuid
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional
from urllib.parse import parse_qs, urlparse, urlsplit, urlunparse, urlunsplit
from urllib.parse import parse_qs, urlparse, urlunparse
from agent.context_compressor import ContextCompressor
from agent.iteration_budget import IterationBudget
@ -48,6 +48,7 @@ from agent.tool_guardrails import (
ToolGuardrailDecision,
)
from hermes_cli.config import cfg_get
from hermes_cli.route_identity import normalize_route_base_url
from hermes_cli.timeouts import get_provider_request_timeout
from hermes_constants import get_hermes_home
from utils import base_url_host_matches, is_truthy_value
@ -70,51 +71,7 @@ def _ra():
def _normalize_route_base_url(base_url: Any) -> str:
"""Canonicalize an endpoint URL for model-route identity comparisons."""
raw = str(base_url or "")
if not raw:
return ""
if any(ord(char) <= 0x20 for char in raw):
return raw
had_query_delimiter = "?" in raw.split("#", 1)[0]
try:
parsed = urlsplit(raw)
hostname = parsed.hostname
if not parsed.scheme or not hostname:
return raw
scheme = parsed.scheme.lower()
if "%" in hostname:
address, zone = hostname.split("%", 1)
host = f"{address.lower()}%{zone}"
else:
host = hostname.lower()
port = parsed.port
except (TypeError, ValueError):
return raw
route_host = parsed.netloc.rsplit("@", 1)[-1]
if route_host.startswith("[") or ":" in host:
host = f"[{host}]"
if port is not None and (scheme, port) not in {("http", 80), ("https", 443)}:
host = f"{host}:{port}"
if "@" in parsed.netloc:
host = f"{parsed.netloc.rsplit('@', 1)[0]}@{host}"
path = parsed.path
if path.endswith("/") and not had_query_delimiter:
path = path[:-1]
normalized = urlunsplit(
(
scheme,
host,
path,
parsed.query,
"",
)
)
if had_query_delimiter and not parsed.query:
normalized += "?"
return normalized
return normalize_route_base_url(base_url)
def _provider_default_routes(provider: str) -> set[str]:
@ -174,6 +131,54 @@ def _provider_default_routes(provider: str) -> set[str]:
return routes
def _context_route_mismatch(
configured_base_url: Any,
active_base_url: Any,
configured_provider: Any,
active_provider: Any,
*,
already_normalized: bool = False,
) -> bool:
"""Return whether a context pin's configured route differs from runtime."""
if already_normalized:
configured_route = str(configured_base_url or "")
active_route = str(active_base_url or "")
else:
configured_route = _normalize_route_base_url(configured_base_url)
active_route = _normalize_route_base_url(active_base_url)
if configured_route:
return configured_route != active_route
configured_provider = str(configured_provider or "").strip()
active_provider = str(active_provider or "").strip()
if not configured_provider:
return False
try:
from hermes_cli.models import normalize_provider as normalize_model_provider
configured_provider = normalize_model_provider(configured_provider)
active_provider = normalize_model_provider(active_provider)
except Exception:
configured_provider = configured_provider.lower()
active_provider = active_provider.lower()
try:
from hermes_cli.providers import normalize_provider as normalize_registry_provider
configured_provider = normalize_registry_provider(configured_provider)
active_provider = normalize_registry_provider(active_provider)
except Exception:
pass
if active_route:
configured_routes = _provider_default_routes(configured_provider)
return not configured_routes or active_route not in configured_routes
return bool(
configured_provider
and active_provider
and configured_provider != active_provider
)
def _normalize_custom_provider_name(value: Any) -> str:
"""Mirror runtime normalization for a requested custom-provider identity."""
return str(value or "").strip().lower().replace(" ", "-")
@ -2021,49 +2026,13 @@ def init_agent(
except (TypeError, ValueError):
pass
_active_base_url = _normalize_route_base_url(_active_route_url)
_route_mismatch = bool(
_configured_base_url
and _configured_base_url != _active_base_url
_route_mismatch = _context_route_mismatch(
_configured_base_url,
_active_base_url,
_configured_provider,
agent.provider,
already_normalized=True,
)
if not _configured_base_url:
_active_provider = str(agent.provider or "").strip()
_normalize_provider_fn = None
_normalize_registry_provider_fn = None
try:
from hermes_cli.models import normalize_provider as _normalize_provider_fn
_configured_provider = _normalize_provider_fn(_configured_provider)
_active_provider = _normalize_provider_fn(_active_provider)
except Exception:
_configured_provider = _configured_provider.lower()
_active_provider = _active_provider.lower()
try:
from hermes_cli.providers import (
normalize_provider as _normalize_registry_provider_fn,
)
_configured_provider = _normalize_registry_provider_fn(
_configured_provider
)
_active_provider = _normalize_registry_provider_fn(
_active_provider
)
except Exception:
pass
if _active_base_url:
_configured_routes = _provider_default_routes(
_configured_provider
)
_route_mismatch = bool(
not _configured_routes
or _active_base_url not in _configured_routes
)
else:
_route_mismatch = bool(
_configured_provider
and _active_provider
and _configured_provider != _active_provider
)
_model_mismatch = bool(
_configured_default_runtime_model
and _configured_default_runtime_model != _active_runtime_model
@ -2102,11 +2071,11 @@ def init_agent(
# Surface a clear warning if the user set a context_length but it
# wasn't a valid positive int — the helper silently skips those.
if _config_context_length is None:
_target = agent.base_url.rstrip("/") if agent.base_url else ""
_target = _normalize_route_base_url(agent.base_url)
for _cp_entry in _custom_providers:
if not isinstance(_cp_entry, dict):
continue
_cp_url = (_cp_entry.get("base_url") or "").rstrip("/")
_cp_url = _normalize_route_base_url(_cp_entry.get("base_url"))
if _target and _cp_url == _target:
_cp_models = _cp_entry.get("models", {})
if isinstance(_cp_models, dict):

25
cli.py
View file

@ -8171,6 +8171,29 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
scroll_offset = max(0, min(scroll_offset, n - visible))
return scroll_offset, visible
def _clear_persisted_context_for_model_switch(self, result) -> None:
"""Drop a global context pin when its configured owner changes."""
try:
from agent.agent_init import _context_route_mismatch
from hermes_cli.config import load_config_readonly
config = load_config_readonly()
model_cfg = config.get("model", {}) if isinstance(config, dict) else {}
if not isinstance(model_cfg, dict) or "context_length" not in model_cfg:
return
configured_model = model_cfg.get("default") or model_cfg.get("model")
if (
configured_model and configured_model != result.new_model
) or _context_route_mismatch(
model_cfg.get("base_url"),
result.base_url,
model_cfg.get("provider"),
result.target_provider,
):
save_config_value("model.context_length", None)
except Exception:
save_config_value("model.context_length", None)
def _apply_model_switch_result(self, result, persist_global: bool) -> None:
if not result.success:
_cprint(f" ✗ {result.error_message}")
@ -8289,6 +8312,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
if result.warning_message:
_cprint(f" ⚠ {result.warning_message}")
if persist_global:
HermesCLI._clear_persisted_context_for_model_switch(self, result)
save_config_value("model.default", result.new_model)
if result.provider_changed:
save_config_value("model.provider", result.target_provider)
@ -8635,6 +8659,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
# Persistence
if persist_global:
HermesCLI._clear_persisted_context_for_model_switch(self, result)
save_config_value("model.default", result.new_model)
if result.provider_changed:
save_config_value("model.provider", result.target_provider)

137
gateway/run.py
View file

@ -11957,6 +11957,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
)
if _msg_model != _msg_configured_model:
_msg_config_ctx = None
if _msg_config_ctx is not None and isinstance(_msg_model_cfg, dict):
try:
from agent.agent_init import _context_route_mismatch
if _context_route_mismatch(
_msg_model_cfg.get("base_url"),
_msg_base_url,
_msg_model_cfg.get("provider"),
_msg_runtime.get("provider"),
):
_msg_config_ctx = None
except Exception:
_msg_config_ctx = None
if _msg_custom_providers and _msg_base_url:
try:
from hermes_cli.config import get_custom_provider_context_length
@ -12451,6 +12464,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_hyg_provider = None
_hyg_base_url = None
_hyg_api_key = None
_hyg_configured_model = None
_hyg_configured_provider = None
_hyg_configured_base_url = None
_hyg_data = {}
try:
_hyg_data = _load_gateway_config()
@ -12490,6 +12506,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except (TypeError, ValueError):
pass
_hyg_configured_model = _hyg_model
_hyg_configured_provider = _hyg_provider
_hyg_configured_base_url = _hyg_base_url
try:
_hyg_model, _hyg_runtime = self._resolve_session_agent_runtime(
source=source,
@ -12502,31 +12522,45 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except Exception:
pass
if _hyg_config_context_length is not None:
try:
from agent.agent_init import _context_route_mismatch
if (
_hyg_configured_model
and _hyg_model != _hyg_configured_model
) or _context_route_mismatch(
_hyg_configured_base_url,
_hyg_base_url,
_hyg_configured_provider,
_hyg_provider,
):
_hyg_config_context_length = None
except Exception:
_hyg_config_context_length = None
# Check custom_providers per-model context_length
# (same fallback as run_agent.py lines 1171-1189).
# Must run after runtime resolution so _hyg_base_url is set.
if _hyg_config_context_length is None and _hyg_base_url:
try:
try:
from hermes_cli.config import get_compatible_custom_providers as _gw_gcp
from hermes_cli.config import (
get_compatible_custom_providers as _gw_gcp,
get_custom_provider_context_length as _gw_gccl,
)
_hyg_custom_providers = _gw_gcp(_hyg_data)
except Exception:
_hyg_custom_providers = _hyg_data.get("custom_providers")
if not isinstance(_hyg_custom_providers, list):
_hyg_custom_providers = []
for _cp in _hyg_custom_providers:
if not isinstance(_cp, dict):
continue
_cp_url = (_cp.get("base_url") or "").rstrip("/")
if _cp_url and _cp_url == _hyg_base_url.rstrip("/"):
_cp_models = _cp.get("models", {})
if isinstance(_cp_models, dict):
_cp_model_cfg = _cp_models.get(_hyg_model, {})
if isinstance(_cp_model_cfg, dict):
_cp_ctx = _cp_model_cfg.get("context_length")
if _cp_ctx is not None:
_hyg_config_context_length = int(_cp_ctx)
break
_hyg_custom_ctx = _gw_gccl(
model=_hyg_model,
base_url=_hyg_base_url,
custom_providers=_hyg_custom_providers,
)
if _hyg_custom_ctx:
_hyg_config_context_length = int(_hyg_custom_ctx)
except (TypeError, ValueError):
pass
except Exception:
@ -13786,12 +13820,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
api_key = None
custom_provs = None
data = None
configured_model = None
configured_provider = None
configured_base_url = None
try:
data = _load_gateway_config()
if data:
model_cfg = data.get("model", {})
if isinstance(model_cfg, dict):
configured_model = model_cfg.get("default") or model_cfg.get("model")
raw_ctx = model_cfg.get("context_length")
if raw_ctx is not None:
try:
@ -13800,6 +13838,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
pass
provider = model_cfg.get("provider") or None
base_url = model_cfg.get("base_url") or None
configured_provider = provider
configured_base_url = base_url
try:
from hermes_cli.config import get_compatible_custom_providers
custom_provs = get_compatible_custom_providers(data)
@ -13808,50 +13848,45 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
except Exception:
pass
# Also check custom_providers for context_length when top-level model.context_length is not set
if config_context_length is None and data:
try:
custom_providers = data.get("custom_providers", [])
if custom_providers:
for cp in custom_providers:
if not isinstance(cp, dict):
continue
cp_model = cp.get("model") or ""
cp_models = cp.get("models") or {}
# Match provider model to current model
if cp_model and cp_model == model:
raw_cp_ctx = cp.get("context_length")
if raw_cp_ctx is not None:
try:
config_context_length = int(raw_cp_ctx)
break
except (TypeError, ValueError):
pass
# Also check per-model context_length
if isinstance(cp_models, dict):
model_entry = cp_models.get(model)
if isinstance(model_entry, dict):
model_ctx = model_entry.get("context_length")
else:
model_ctx = model_entry
if model_ctx is not None and isinstance(model_ctx, (int, float)):
try:
config_context_length = int(model_ctx)
break
except (TypeError, ValueError):
pass
except Exception:
pass
# Resolve runtime credentials for probing
try:
runtime = _resolve_runtime_agent_kwargs()
provider = provider or runtime.get("provider")
base_url = base_url or runtime.get("base_url")
provider = runtime.get("provider") or provider
base_url = runtime.get("base_url") or base_url
api_key = runtime.get("api_key")
except Exception:
pass
if config_context_length is not None:
try:
from agent.agent_init import _context_route_mismatch
if (
configured_model and model != configured_model
) or _context_route_mismatch(
configured_base_url,
base_url,
configured_provider,
provider,
):
config_context_length = None
except Exception:
config_context_length = None
if config_context_length is None and custom_provs and base_url:
try:
from hermes_cli.config import get_custom_provider_context_length
custom_ctx = get_custom_provider_context_length(
model=model,
base_url=base_url,
custom_providers=custom_provs,
)
if custom_ctx:
config_context_length = custom_ctx
except Exception:
pass
context_length = get_model_context_length(
model,
base_url=base_url or "",

52
gateway/slash_commands.py
View file

@ -1729,6 +1729,25 @@ class GatewaySlashCommandsMixin:
else:
_persist_model_cfg = {}
_persist_cfg["model"] = _persist_model_cfg
try:
from agent.agent_init import _context_route_mismatch
_old_default = (
_persist_model_cfg.get("default")
or _persist_model_cfg.get("model")
)
if (
_old_default
and _old_default != result.new_model
) or _context_route_mismatch(
_persist_model_cfg.get("base_url"),
result.base_url,
_persist_model_cfg.get("provider"),
result.target_provider,
):
_persist_model_cfg.pop("context_length", None)
except Exception:
_persist_model_cfg.pop("context_length", None)
_persist_model_cfg["default"] = result.new_model
_persist_model_cfg["provider"] = result.target_provider
# Named providers always resolve base_url/api_mode fresh,
@ -1764,6 +1783,7 @@ class GatewaySlashCommandsMixin:
mi = result.model_info
from hermes_cli.model_switch import resolve_display_context_length
_sw_config_ctx = None
_sw_model_cfg = {}
try:
_sw_cfg = _load_gateway_config()
_sw_model_cfg = _sw_cfg.get("model", {})
@ -1773,6 +1793,8 @@ class GatewaySlashCommandsMixin:
_sw_config_ctx = int(_sw_raw)
except Exception:
pass
if not isinstance(_sw_model_cfg, dict):
_sw_model_cfg = {}
ctx = resolve_display_context_length(
result.new_model,
result.target_provider,
@ -1781,6 +1803,12 @@ class GatewaySlashCommandsMixin:
model_info=mi,
custom_providers=custom_provs,
config_context_length=_sw_config_ctx,
configured_model=(
_sw_model_cfg.get("default")
or _sw_model_cfg.get("model")
),
configured_provider=_sw_model_cfg.get("provider"),
configured_base_url=_sw_model_cfg.get("base_url"),
)
if ctx:
lines.append(t("gateway.model.context_label", tokens=f"{ctx:,}"))
@ -2033,6 +2061,21 @@ class GatewaySlashCommandsMixin:
else:
model_cfg = {}
cfg["model"] = model_cfg
try:
from agent.agent_init import _context_route_mismatch
old_default = model_cfg.get("default") or model_cfg.get("model")
if (
old_default and old_default != result.new_model
) or _context_route_mismatch(
model_cfg.get("base_url"),
result.base_url,
model_cfg.get("provider"),
result.target_provider,
):
model_cfg.pop("context_length", None)
except Exception:
model_cfg.pop("context_length", None)
model_cfg["default"] = result.new_model
model_cfg["provider"] = result.target_provider
# See the picker handler above for why custom providers need an
@ -2064,6 +2107,7 @@ class GatewaySlashCommandsMixin:
mi = result.model_info
from hermes_cli.model_switch import resolve_display_context_length
_sw2_config_ctx = None
_sw2_model_cfg = {}
try:
_sw2_cfg = _load_gateway_config()
_sw2_model_cfg = _sw2_cfg.get("model", {})
@ -2073,6 +2117,8 @@ class GatewaySlashCommandsMixin:
_sw2_config_ctx = int(_sw2_raw)
except Exception:
pass
if not isinstance(_sw2_model_cfg, dict):
_sw2_model_cfg = {}
ctx = resolve_display_context_length(
result.new_model,
result.target_provider,
@ -2081,6 +2127,12 @@ class GatewaySlashCommandsMixin:
model_info=mi,
custom_providers=custom_provs,
config_context_length=_sw2_config_ctx,
configured_model=(
_sw2_model_cfg.get("default")
or _sw2_model_cfg.get("model")
),
configured_provider=_sw2_model_cfg.get("provider"),
configured_base_url=_sw2_model_cfg.get("base_url"),
)
if ctx:
lines.append(t("gateway.model.context_label", tokens=f"{ctx:,}"))

31
hermes_cli/config.py
View file

@ -31,6 +31,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Any, Optional, List, Tuple, Set
from hermes_cli.route_identity import normalize_route_base_url
from hermes_cli.secret_prompt import masked_secret_prompt
logger = logging.getLogger(__name__)
@ -5315,16 +5316,11 @@ def get_custom_provider_tls_settings(
if not base_url or not isinstance(custom_providers, list):
return {}
# Case-insensitive compare: elsewhere custom_providers are keyed on a
# lowercased base_url (see get_compatible_custom_providers dedup), and
# scheme/host are case-insensitive anyway — so a config entry written as
# https://Ollama.Example.com/v1 must still match a lowercased runtime
# base_url. Exact match after rstrip('/') + lower() (no prefix/substring).
target_url = (base_url or "").rstrip("/").lower()
target_url = normalize_route_base_url(base_url)
for entry in custom_providers:
if not isinstance(entry, dict):
continue
entry_url = (entry.get("base_url") or "").rstrip("/").lower()
entry_url = normalize_route_base_url(entry.get("base_url"))
if not entry_url or entry_url != target_url:
continue
out: Dict[str, Any] = {}
@ -5376,10 +5372,9 @@ def get_custom_provider_extra_headers(
) -> Dict[str, str]:
"""Return ``extra_headers`` from a matching ``providers`` / ``custom_providers`` entry.
Matches the entry whose ``base_url`` equals *base_url* (trailing-slash and
case insensitive, mirroring :func:`get_custom_provider_tls_settings`) and
returns its ``extra_headers`` dict, or ``{}`` when no entry matches or the
entry declares none.
Matches the entry whose normalized route identity equals *base_url*,
mirroring :func:`get_custom_provider_tls_settings`, and returns its
``extra_headers`` dict, or ``{}`` when no entry matches or declares none.
SECURITY: header values routinely carry credentials (Cloudflare Access
service tokens, proxy auth, custom bearer schemes). Callers must never
@ -5393,11 +5388,11 @@ def get_custom_provider_extra_headers(
if not base_url or not isinstance(custom_providers, list):
return {}
target_url = (base_url or "").rstrip("/").lower()
target_url = normalize_route_base_url(base_url)
for entry in custom_providers:
if not isinstance(entry, dict):
continue
entry_url = (entry.get("base_url") or "").rstrip("/").lower()
entry_url = normalize_route_base_url(entry.get("base_url"))
if not entry_url or entry_url != target_url:
continue
return normalize_extra_headers(entry.get("extra_headers"))
@ -5435,9 +5430,9 @@ def get_custom_provider_context_length(
) -> Optional[int]:
"""Look up a per-model ``context_length`` override from ``custom_providers``.
Matches any entry whose ``base_url`` equals ``base_url`` (trailing-slash
insensitive) and returns ``custom_providers[i].models.<model>.context_length``
if present and valid. Returns ``None`` when no override applies.
Matches any entry whose normalized route identity equals ``base_url`` and
returns ``custom_providers[i].models.<model>.context_length`` if present and
valid. Returns ``None`` when no override applies.
This is the single source of truth for custom-provider context overrides,
used by:
@ -5464,14 +5459,14 @@ def get_custom_provider_context_length(
if not isinstance(custom_providers, list):
return None
target_url = (base_url or "").rstrip("/")
target_url = normalize_route_base_url(base_url)
if not target_url:
return None
for entry in custom_providers:
if not isinstance(entry, dict):
continue
entry_url = (entry.get("base_url") or "").rstrip("/")
entry_url = normalize_route_base_url(entry.get("base_url"))
if not entry_url or entry_url != target_url:
continue
models = entry.get("models")

27
hermes_cli/context_switch_guard.py
View file

@ -69,6 +69,9 @@ def merge_preflight_compression_warning(
messages: Optional[List[dict]] = None,
custom_providers: list | None = None,
config_context_length: int | None = None,
configured_model: str | None = None,
configured_provider: str | 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:
@ -89,6 +92,21 @@ def merge_preflight_compression_warning(
model_info=result.model_info,
custom_providers=custom_providers,
config_context_length=config_context_length,
configured_model=(
configured_model
if configured_model is not None
else getattr(agent, "model", None)
),
configured_provider=(
configured_provider
if configured_provider is not None
else getattr(agent, "provider", None)
),
configured_base_url=(
configured_base_url
if configured_base_url is not None
else getattr(agent, "base_url", None)
),
)
if not new_ctx:
return
@ -141,12 +159,18 @@ def enrich_model_switch_warnings_for_gateway(
return
cfg_ctx = None
configured_model = None
configured_provider = None
configured_base_url = None
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")
except Exception:
pass
@ -166,4 +190,7 @@ def enrich_model_switch_warnings_for_gateway(
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,
)

21
hermes_cli/model_switch.py
View file

@ -838,6 +838,9 @@ def resolve_display_context_length(
model_info: Optional[ModelInfo] = None,
custom_providers: list | None = None,
config_context_length: int | None = None,
configured_model: str | None = None,
configured_provider: str | None = None,
configured_base_url: str | None = None,
) -> Optional[int]:
"""Resolve the context length to show in /model output.
@ -856,6 +859,24 @@ def resolve_display_context_length(
Prefer the provider-aware value; fall back to ``model_info.context_window``
only if the resolver returns nothing.
"""
if config_context_length is not None and (
configured_model or configured_provider or configured_base_url
):
try:
from agent.agent_init import _context_route_mismatch
if (
configured_model and configured_model != model
) or _context_route_mismatch(
configured_base_url,
base_url,
configured_provider,
provider,
):
config_context_length = None
except Exception:
config_context_length = None
try:
from agent.model_metadata import get_model_context_length
ctx = get_model_context_length(

47
hermes_cli/route_identity.py Normal file
View file

@ -0,0 +1,47 @@
"""Fail-closed URL identity normalization for model/provider routes."""
from __future__ import annotations
from typing import Any
from urllib.parse import urlsplit, urlunsplit
def normalize_route_base_url(base_url: Any) -> str:
"""Canonicalize only proven-equivalent endpoint URL components."""
raw = str(base_url or "")
if not raw:
return ""
if any(ord(char) <= 0x20 for char in raw):
return raw
had_query_delimiter = "?" in raw.split("#", 1)[0]
try:
parsed = urlsplit(raw)
hostname = parsed.hostname
if not parsed.scheme or not hostname:
return raw
scheme = parsed.scheme.lower()
if "%" in hostname:
address, zone = hostname.split("%", 1)
host = f"{address.lower()}%{zone}"
else:
host = hostname.lower()
port = parsed.port
except (TypeError, ValueError):
return raw
route_host = parsed.netloc.rsplit("@", 1)[-1]
if route_host.startswith("[") or ":" in host:
host = f"[{host}]"
if port is not None and (scheme, port) not in {("http", 80), ("https", 443)}:
host = f"{host}:{port}"
if "@" in parsed.netloc:
host = f"{parsed.netloc.rsplit('@', 1)[0]}@{host}"
path = parsed.path
if path.endswith("/") and not had_query_delimiter:
path = path[:-1]
normalized = urlunsplit((scheme, host, path, parsed.query, ""))
if had_query_delimiter and not parsed.query:
normalized += "?"
return normalized

46
run_agent.py
View file

@ -4656,7 +4656,12 @@ class AIAgent:
self._is_anthropic_oauth = _is_oauth_token(new_token) if self.provider == "anthropic" else False
return True
def _apply_client_headers_for_base_url(self, base_url: str) -> None:
def _apply_client_headers_for_base_url(
self,
base_url: str,
*,
apply_user_headers: bool = True,
) -> None:
from agent.auxiliary_client import (
build_nvidia_nim_headers,
build_or_headers,
@ -4696,10 +4701,10 @@ class AIAgent:
else:
self._client_kwargs.pop("default_headers", None)
# User-configured overrides win over URL/profile defaults — keep them
# applied across credential swaps and client rebuilds, not just at
# first construction.
self._apply_user_default_headers()
# User-configured overrides win over URL/profile defaults for the same
# route. A credential swap to another endpoint must not inherit them.
if apply_user_headers:
self._apply_user_default_headers()
# Per-provider extra HTTP headers (providers.<name>.extra_headers /
# custom_providers[].extra_headers) — applied last so the most
@ -4750,6 +4755,11 @@ class AIAgent:
def _swap_credential(self, entry) -> None:
runtime_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
runtime_base = getattr(entry, "runtime_base_url", None) or getattr(entry, "base_url", None) or self.base_url
from hermes_cli.route_identity import normalize_route_base_url
route_changed = normalize_route_base_url(self.base_url) != normalize_route_base_url(
runtime_base
)
if self.api_mode == "anthropic_messages":
from agent.anthropic_adapter import build_anthropic_client, _is_oauth_token
@ -4771,10 +4781,32 @@ class AIAgent:
return
self.api_key = runtime_key
self.base_url = runtime_base.rstrip("/") if isinstance(runtime_base, str) else runtime_base
self.base_url = runtime_base
self._client_kwargs["api_key"] = self.api_key
self._client_kwargs["base_url"] = self.base_url
self._apply_client_headers_for_base_url(self.base_url)
self._client_kwargs.pop("ssl_verify", None)
self._client_kwargs.pop("ssl_ca_cert", None)
try:
from hermes_cli.config import (
apply_custom_provider_tls_to_client_kwargs,
get_compatible_custom_providers,
load_config_readonly,
)
apply_custom_provider_tls_to_client_kwargs(
self._client_kwargs,
str(self.base_url or ""),
get_compatible_custom_providers(load_config_readonly()),
)
except Exception:
logger.debug(
"custom-provider TLS resolution skipped on credential rotation",
exc_info=True,
)
self._apply_client_headers_for_base_url(
self.base_url,
apply_user_headers=not route_changed,
)
self._replace_primary_openai_client(reason="credential_rotation")
def _recover_with_credential_pool(

51
tests/gateway/test_context_ref_expansion_runtime.py
View file

@ -359,3 +359,54 @@ async def test_at_reference_ignores_global_context_for_session_model_override(mo
event=MessageEvent(text="@file:note", source=source), source=source, history=[]
)
assert captured["config_context_length"] is None
@pytest.mark.asyncio
async def test_at_reference_ignores_global_context_for_runtime_route_override(monkeypatch):
"""Context expansion must not inherit a global pin from another route."""
runner = _make_runner()
source = _source()
captured = {}
monkeypatch.setattr(
gateway_run,
"_load_gateway_config",
lambda: {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://large.example/v1",
"context_length": 1_048_576,
}
},
)
monkeypatch.setattr(
runner,
"_resolve_session_agent_runtime",
lambda **_kwargs: (
"shared-model",
{
"provider": "custom",
"api_key": "test",
"base_url": "https://small.example/v1",
},
),
)
import agent.context_references as ctx_mod
import agent.model_metadata as model_meta_mod
async def _fake_get_context(_model, **kwargs):
captured["config_context_length"] = kwargs["config_context_length"]
return 32_768
async def _passthrough(message, **_kwargs):
return ContextReferenceResult(message=message, original_message=message)
monkeypatch.setattr(model_meta_mod, "get_model_context_length_async", _fake_get_context)
monkeypatch.setattr(ctx_mod, "preprocess_context_references_async", _passthrough)
await runner._prepare_inbound_message_text(
event=MessageEvent(text="@file:note", source=source), source=source, history=[]
)
assert captured["config_context_length"] is None

7
tests/gateway/test_model_command_flat_string_config.py
View file

@ -145,7 +145,11 @@ async def test_model_global_persists_when_config_has_proper_dict_model(tmp_path,
cfg_path = _setup_isolated_home(
tmp_path,
monkeypatch,
{"default": "old-model", "provider": "openai-codex"},
{
"default": "old-model",
"provider": "openai-codex",
"context_length": 1_048_576,
},
)
result = await _make_runner()._handle_model_command(
@ -156,6 +160,7 @@ async def test_model_global_persists_when_config_has_proper_dict_model(tmp_path,
written = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
assert written["model"]["default"] == "gpt-5.5"
assert written["model"]["provider"] == "openrouter"
assert "context_length" not in written["model"]
@pytest.mark.asyncio

2
tests/gateway/test_model_picker_persist.py
View file

@ -167,6 +167,7 @@ async def _drive_picker(runner, event):
"default": "old-model",
"provider": "custom",
"base_url": "https://api.custom.example/v1",
"context_length": 1_048_576,
"api_key": "sk-stale",
"api_mode": "anthropic_messages",
},
@ -197,6 +198,7 @@ async def test_picker_tap_global_flag_persists(tmp_path, monkeypatch, seed_model
assert "base_url" not in written["model"]
assert "api_key" not in written["model"]
assert "api_mode" not in written["model"]
assert "context_length" not in written["model"]
@pytest.mark.asyncio

54
tests/gateway/test_session_info.py
View file

@ -87,6 +87,60 @@ class TestFormatSessionInfo:
info = runner._format_session_info()
assert "1.0M" in info
def test_custom_context_is_scoped_to_active_runtime_route(self, runner, tmp_path):
config = """
model:
default: shared-model
provider: custom
custom_providers:
- name: large-route
base_url: https://example.com/v1//
models:
shared-model:
context_length: 1048576
"""
p1, p2, p3 = _patch_info(
tmp_path,
config,
"shared-model",
{
"provider": "custom",
"base_url": "https://example.com/v1",
"api_key": "k",
},
)
with p1, p2, p3:
info = runner._format_session_info()
assert "1.0M" not in info
assert "(config)" not in info
def test_global_context_is_scoped_to_active_runtime_route(self, runner, tmp_path):
config = """
model:
default: shared-model
provider: custom
base_url: https://large.example/v1
context_length: 1048576
"""
p1, p2, p3 = _patch_info(
tmp_path,
config,
"shared-model",
{
"provider": "custom",
"base_url": "https://small.example/v1",
"api_key": "k",
},
)
with p1, p2, p3:
info = runner._format_session_info()
assert "1.0M" not in info
assert "(config)" not in info
def test_missing_config(self, runner, tmp_path):
"""No config.yaml should not crash."""
p1, p2, p3 = _patch_info(tmp_path, None, # don't create config

51
tests/hermes_cli/test_apply_model_switch_result_context.py
View file

@ -150,3 +150,54 @@ def test_picker_path_falls_back_to_model_info_when_resolver_empty(monkeypatch):
assert "1,050,000" in ctx_line, (
f"resolver-empty path should fall back to ModelInfo, got: {ctx_line!r}"
)
def test_global_switch_clears_context_pin_owned_by_previous_route(monkeypatch):
import cli as cli_mod
writes = []
monkeypatch.setattr(cli_mod, "_cprint", lambda *_a, **_k: None)
monkeypatch.setattr(
cli_mod,
"save_config_value",
lambda key, value: writes.append((key, value)),
)
cli = _StubCLI()
cli.model = "shared-model"
cli.provider = "custom"
# Runtime may already diverge from persisted config through a session override.
cli.base_url = "https://small.example/v1"
result = ModelSwitchResult(
success=True,
new_model="shared-model",
target_provider="custom",
provider_changed=False,
api_key="",
base_url="https://small.example/v1",
api_mode="chat_completions",
warning_message="",
provider_label="Custom",
resolved_via_alias=False,
capabilities=None,
model_info=_FakeModelInfo(),
is_global=True,
)
configured = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://large.example/v1",
"context_length": 1_048_576,
}
}
with (
patch(
"agent.model_metadata.get_model_context_length",
return_value=256_000,
),
patch("hermes_cli.config.load_config_readonly", return_value=configured),
):
cli_mod.HermesCLI._apply_model_switch_result(cli, result, True)
assert ("model.context_length", None) in writes

33
tests/hermes_cli/test_context_switch_guard.py
View file

@ -103,3 +103,36 @@ def test_merge_appends_to_existing_warning(monkeypatch):
merge_preflight_compression_warning(result, agent=agent)
assert "expensive" in result.warning_message
assert "preflight compression" in result.warning_message
def test_cross_route_switch_does_not_inherit_current_context_pin(monkeypatch):
def _resolve_metadata(*_args, **kwargs):
return kwargs["config_context_length"] or 32_000
monkeypatch.setattr(
"agent.model_metadata.get_model_context_length",
_resolve_metadata,
)
monkeypatch.setattr(
"hermes_cli.context_switch_guard._estimate_tokens",
lambda *a, **k: 90_000,
)
cc = _compressor(monkeypatch, context_length=1_048_576)
agent = SimpleNamespace(
model="shared-model",
provider="custom",
context_compressor=cc,
compression_enabled=True,
conversation_history=[],
base_url="https://large.example/v1",
api_key="",
)
result = _result(model="shared-model")
merge_preflight_compression_warning(
result,
agent=agent,
config_context_length=1_048_576,
)
assert "preflight compression" in result.warning_message

15
tests/hermes_cli/test_custom_provider_context_length.py
View file

@ -55,6 +55,21 @@ class TestGetCustomProviderContextLength:
== 500_000
)
def test_extra_trailing_segment_is_route_significant(self):
custom = [
{
"base_url": "https://example.invalid/v1//",
"models": {"m": {"context_length": 500_000}},
}
]
assert (
get_custom_provider_context_length(
"m", "https://example.invalid/v1", custom
)
is None
)
def test_returns_none_when_url_does_not_match(self):
custom = [
{

14
tests/hermes_cli/test_custom_provider_extra_headers.py
View file

@ -101,6 +101,20 @@ def test_get_custom_provider_extra_headers_no_match_returns_empty():
) == {}
def test_get_custom_provider_extra_headers_preserves_extra_path_segment():
providers = [
{
"base_url": "https://llm.internal.example.com/v1//",
"extra_headers": {"Authorization": "secret"},
}
]
assert get_custom_provider_extra_headers(
"https://llm.internal.example.com/v1",
custom_providers=providers,
) == {}
def test_apply_extra_headers_merges_onto_existing_defaults():
client_kwargs = {
"api_key": "x",

14
tests/hermes_cli/test_custom_provider_tls.py
View file

@ -70,3 +70,17 @@ def test_get_custom_provider_tls_settings_no_substring_bypass():
"https://ollama.example.com.attacker.test/v1",
custom_providers=providers,
) == {}
def test_get_custom_provider_tls_settings_preserves_extra_path_segment():
providers = [
{
"base_url": "https://ollama.example.com/v1//",
"ssl_verify": False,
}
]
assert get_custom_provider_tls_settings(
"https://ollama.example.com/v1",
custom_providers=providers,
) == {}

18
tests/hermes_cli/test_model_switch_context_display.py
View file

@ -173,3 +173,21 @@ class TestResolveDisplayContextLength:
"The fix ensures custom_providers is passed so per-model overrides "
"are honored."
)
def test_global_context_is_scoped_to_configured_route(self):
with patch(
"agent.model_metadata.get_model_context_length",
return_value=256_000,
) as resolver:
ctx = resolve_display_context_length(
"shared-model",
"custom",
base_url="https://small.example/v1",
config_context_length=1_048_576,
configured_model="shared-model",
configured_provider="custom",
configured_base_url="https://large.example/v1",
)
assert ctx == 256_000
assert resolver.call_args.kwargs["config_context_length"] is None

131
tests/run_agent/test_credential_rotation_route_settings.py Normal file
View file

@ -0,0 +1,131 @@
"""Credential rotation must not carry route-scoped TLS policy."""
from types import MethodType, SimpleNamespace
from unittest.mock import MagicMock, patch
from run_agent import AIAgent
def test_credential_rotation_replaces_route_scoped_tls_settings():
agent = SimpleNamespace(
api_mode="chat_completions",
provider="custom",
model="shared-model",
api_key="old",
base_url="https://a.example/v1",
_client_kwargs={
"api_key": "old",
"base_url": "https://a.example/v1",
"ssl_verify": False,
"ssl_ca_cert": "/a.pem",
},
_apply_client_headers_for_base_url=MagicMock(),
_replace_primary_openai_client=MagicMock(),
)
entry = SimpleNamespace(
runtime_api_key="new",
access_token="",
runtime_base_url="https://b.example/v1",
base_url="https://b.example/v1",
)
config = {
"custom_providers": [
{
"name": "b",
"base_url": "https://b.example/v1",
"ssl_verify": True,
}
]
}
with patch("hermes_cli.config.load_config_readonly", return_value=config):
AIAgent._swap_credential(agent, entry)
assert agent._client_kwargs["ssl_verify"] is True
assert "ssl_ca_cert" not in agent._client_kwargs
agent._replace_primary_openai_client.assert_called_once_with(
reason="credential_rotation"
)
def test_credential_rotation_does_not_carry_global_headers_across_routes():
agent = SimpleNamespace(
api_mode="chat_completions",
provider="custom",
model="shared-model",
api_key="old",
base_url="https://a.example/v1",
_client_kwargs={
"api_key": "old",
"base_url": "https://a.example/v1",
"default_headers": {"Authorization": "old-secret"},
},
_replace_primary_openai_client=MagicMock(),
)
agent._apply_client_headers_for_base_url = MethodType(
AIAgent._apply_client_headers_for_base_url,
agent,
)
agent._apply_user_default_headers = MethodType(
AIAgent._apply_user_default_headers,
agent,
)
entry = SimpleNamespace(
runtime_api_key="new",
access_token="",
runtime_base_url="https://b.example/v1",
base_url="https://b.example/v1",
)
config = {
"model": {
"default_headers": {"Authorization": "global-secret"},
},
"custom_providers": [
{
"name": "b",
"base_url": "https://b.example/v1",
"extra_headers": {"X-Route": "b"},
}
],
}
with (
patch("hermes_cli.config.load_config_readonly", return_value=config),
patch(
"hermes_cli.config.get_compatible_custom_providers",
return_value=config["custom_providers"],
),
):
AIAgent._swap_credential(agent, entry)
headers = agent._client_kwargs["default_headers"]
assert "Authorization" not in headers
assert headers["X-Route"] == "b"
def test_credential_rotation_preserves_route_significant_trailing_segments():
agent = SimpleNamespace(
api_mode="chat_completions",
provider="custom",
model="shared-model",
api_key="old",
base_url="https://a.example/v1",
_client_kwargs={
"api_key": "old",
"base_url": "https://a.example/v1",
},
_apply_client_headers_for_base_url=MagicMock(),
_replace_primary_openai_client=MagicMock(),
)
entry = SimpleNamespace(
runtime_api_key="new",
access_token="",
runtime_base_url="https://b.example/v1//",
base_url="https://b.example/v1//",
)
with patch("hermes_cli.config.load_config_readonly", return_value={}):
AIAgent._swap_credential(agent, entry)
assert agent.base_url == "https://b.example/v1//"
assert agent._client_kwargs["base_url"] == "https://b.example/v1//"

36
tests/run_agent/test_switch_model_context.py
View file

@ -51,6 +51,14 @@ def test_route_url_normalization_preserves_malformed_trailing_slash():
) != _normalize_route_base_url("http://[bad/v1")
@pytest.mark.parametrize(
"raw",
["http://[bad/v1/", "example.com/v1/", "https:///v1/"],
)
def test_route_url_normalization_keeps_unparseable_routes_byte_exact(raw):
assert _normalize_route_base_url(raw) == raw
@pytest.mark.parametrize(
("configured", "active"),
[
@ -583,6 +591,34 @@ def test_direct_start_drops_context_when_extra_trailing_segment_changes():
assert agent.context_compressor.config_context_length is None
def test_direct_start_does_not_reapply_custom_context_across_extra_slash():
"""Per-model custom overrides use the same fail-closed route identity."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
},
"custom_providers": [
{
"name": "large-route",
"base_url": "https://example.com/v1//",
"models": {
"shared-model": {"context_length": 1_048_576}
},
}
],
}
agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="https://example.com/v1",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_drops_context_when_url_userinfo_changes():
"""Credentials embedded in a URL remain part of route identity."""
cfg = {