RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(compression): harden startup route scoping

This commit is contained in:
cucurigoo 2026-07-21 22:32:08 +00:00 committed by kshitij
parent 377244f7c8
commit 97499d702e
3 changed files with 540 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

174
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 urlparse, parse_qs, urlunparse
from urllib.parse import parse_qs, urlparse, urlsplit, urlunparse, urlunsplit
from agent.context_compressor import ContextCompressor
from agent.iteration_budget import IterationBudget
@ -68,6 +68,112 @@ def _ra():
return run_agent
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
def _provider_default_routes(provider: str) -> set[str]:
"""Return known exact default routes for a canonical provider id."""
routes: set[str] = set()
try:
from hermes_cli.providers import HERMES_OVERLAYS, get_provider
overlay = HERMES_OVERLAYS.get(provider)
provider_def = get_provider(provider)
for value in (
getattr(overlay, "base_url_override", ""),
getattr(provider_def, "base_url", ""),
):
route = _normalize_route_base_url(value)
if route:
routes.add(route)
except Exception:
pass
try:
from providers import get_provider_profile
profile = get_provider_profile(provider)
route = _normalize_route_base_url(
getattr(profile, "base_url", "")
)
if route:
routes.add(route)
except Exception:
pass
try:
from hermes_cli.auth import PROVIDER_REGISTRY
from hermes_cli.models import normalize_provider as normalize_model_provider
from hermes_cli.providers import normalize_provider as normalize_registry_provider
for provider_id, config in PROVIDER_REGISTRY.items():
canonical_id = normalize_registry_provider(
normalize_model_provider(provider_id)
)
if canonical_id != provider:
continue
route = _normalize_route_base_url(
getattr(config, "inference_base_url", "")
)
if route:
routes.add(route)
except Exception:
pass
if provider == "gemini":
routes.update(
f"{route.rstrip('/')}/openai"
for route in list(routes)
)
return routes
def _build_codex_gpt5_autoraise_notice(
autoraise: Dict[str, Any], context_length: Optional[int] = None
) -> str:
@ -1792,7 +1898,9 @@ def init_agent(
except Exception:
pass
_configured_provider = str(_model_cfg.get("provider") or "").strip()
_configured_base_url = str(_model_cfg.get("base_url") or "").rstrip("/")
_configured_base_url = _normalize_route_base_url(
_model_cfg.get("base_url")
)
if not _configured_base_url and _configured_provider.lower().startswith("custom:"):
_configured_custom_name = _configured_provider.split(":", 1)[1].lower()
for _provider_entry in _custom_providers:
@ -1800,11 +1908,25 @@ def init_agent(
continue
if str(_provider_entry.get("name") or "").strip().lower() != _configured_custom_name:
continue
_configured_base_url = str(
_provider_entry.get("base_url") or ""
).rstrip("/")
_configured_base_url = _normalize_route_base_url(
_provider_entry.get("base_url")
)
break
_active_base_url = str(agent.base_url or "").rstrip("/")
_active_route_url = str(agent.base_url or "")
_requested_route_url = str(base_url or "")
if "?" in _requested_route_url.split("#", 1)[0]:
try:
_requested_parts = urlparse(_requested_route_url)
_requested_without_query = urlunparse(
_requested_parts._replace(query="")
)
if _normalize_route_base_url(
_requested_without_query
) == _normalize_route_base_url(_active_route_url):
_active_route_url = _requested_route_url
except (TypeError, ValueError):
pass
_active_base_url = _normalize_route_base_url(_active_route_url)
_route_mismatch = bool(
_configured_base_url
and _active_base_url
@ -1812,19 +1934,43 @@ def init_agent(
)
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
from hermes_cli.models import normalize_provider as _normalize_provider_fn
_configured_provider = normalize_provider(_configured_provider)
_active_provider = normalize_provider(_active_provider)
_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()
_route_mismatch = bool(
_configured_provider
and _active_provider
and _configured_provider != _active_provider
)
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

4
tests/run_agent/test_invalid_context_length_warning.py
View File

@ -3,7 +3,7 @@
from unittest.mock import patch
def _build_agent(model_cfg, custom_providers=None, model="anthropic/claude-opus-4.6"):
def _build_agent(model_cfg, custom_providers=None, model=None):
"""Build an AIAgent with the given model config."""
cfg = {"model": model_cfg}
if custom_providers is not None:
@ -21,7 +21,7 @@ def _build_agent(model_cfg, custom_providers=None, model="anthropic/claude-opus-
from run_agent import AIAgent
agent = AIAgent(
model=model,
model=model or model_cfg.get("default") or "anthropic/claude-opus-4.6",
api_key="test-key-1234567890",
base_url=base_url,
quiet_mode=True,

378
tests/run_agent/test_switch_model_context.py
View File

@ -3,6 +3,7 @@
from unittest.mock import MagicMock, patch
from run_agent import AIAgent
from agent.agent_init import _normalize_route_base_url
from agent.context_compressor import ContextCompressor
@ -20,6 +21,34 @@ class _StubStartupCompressor:
return None
def test_route_url_normalization_preserves_path_slash_before_query():
"""A path slash before a query changes OpenAI SDK URL joining."""
assert _normalize_route_base_url(
"https://example.com/v1/?tenant=large"
) != _normalize_route_base_url("https://example.com/v1?tenant=large")
def test_route_url_normalization_preserves_trailing_whitespace():
"""Whitespace can alter the request target and must not collapse routes."""
assert _normalize_route_base_url(
"https://example.com/v1 "
) != _normalize_route_base_url("https://example.com/v1")
def test_route_url_normalization_preserves_bracketed_host_syntax():
"""Invalid bracketed host syntax must not collapse onto a valid DNS host."""
assert _normalize_route_base_url(
"http://[v1.Foo]/v1"
) != _normalize_route_base_url("http://v1.foo/v1")
def test_route_url_normalization_preserves_malformed_trailing_slash():
"""Malformed URLs are kept byte-exact rather than partially normalized."""
assert _normalize_route_base_url(
"http://[bad/v1/"
) != _normalize_route_base_url("http://[bad/v1")
def _make_direct_start_agent(
cfg: dict, *, model: str, provider: str, base_url: str
) -> AIAgent:
@ -202,6 +231,295 @@ def test_direct_start_preserves_context_for_bare_aggregator_model():
assert agent.context_compressor.config_context_length == 1_000_000
def test_direct_start_drops_context_for_same_provider_custom_base_url():
"""An explicit endpoint override changes the route even if provider matches."""
cfg = {
"model": {
"default": "gpt-5.4",
"provider": "openrouter",
"context_length": 1_000_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.4",
provider="openrouter",
base_url="https://small.example/v1",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_drops_context_for_provider_name_lookalike_host():
"""A hostname containing a provider domain is not that provider's route."""
cfg = {
"model": {
"default": "gpt-5.4",
"provider": "openrouter",
"context_length": 1_000_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.4",
provider="openrouter",
base_url="https://evil-openrouter.ai/v1",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_preserves_context_for_codex_default_endpoint():
"""ChatGPT's Codex endpoint belongs to the openai-codex route."""
cfg = {
"model": {
"default": "gpt-5.6-sol",
"provider": "openai-codex",
"context_length": 272_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.6-sol",
provider="openai-codex",
base_url="https://chatgpt.com/backend-api/codex",
)
assert agent.context_compressor.config_context_length == 272_000
def test_direct_start_drops_context_for_codex_wrong_path():
"""A known host with a different route path is not the Codex endpoint."""
cfg = {
"model": {
"default": "gpt-5.6-sol",
"provider": "openai-codex",
"context_length": 272_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.6-sol",
provider="openai-codex",
base_url="https://chatgpt.com/unrelated",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_drops_context_for_overridden_provider_wrong_path():
"""Providers with an explicit default route require that complete route."""
cfg = {
"model": {
"default": "grok-4",
"provider": "xai",
"context_length": 256_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="grok-4",
provider="xai",
base_url="https://api.x.ai/not-v1",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_preserves_context_for_equivalent_base_url_spellings():
"""Route identity ignores URL casing, default ports, and trailing slashes."""
cfg = {
"model": {
"default": "gpt-5.4",
"provider": "openrouter",
"base_url": "HTTPS://OPENROUTER.AI:443/api/v1/",
"context_length": 1_000_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.4",
provider="openrouter",
base_url="https://openrouter.ai/api/v1",
)
assert agent.context_compressor.config_context_length == 1_000_000
def test_direct_start_drops_context_when_path_parameter_segment_changes():
"""Trailing-slash normalization must not move params to another segment."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://example.com/v1/;tenant=large",
"context_length": 1_048_576,
}
}
agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="https://example.com/v1;tenant=large",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_drops_context_when_empty_path_parameter_changes():
"""An explicit empty path-parameter delimiter is not discarded."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://example.com/v1;",
"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_empty_query_delimiter_changes():
"""An explicit empty query changes OpenAI SDK base-URL joining semantics."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://example.com/v1?",
"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_active_query_changes():
"""Query parameters remain part of the effective route identity."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://example.com/v1",
"context_length": 1_048_576,
}
}
agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="https://example.com/v1?tenant=small",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_preserves_context_for_matching_query_route():
"""SDK query extraction must not hide an otherwise matching route."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://example.com/v1?tenant=large",
"context_length": 1_048_576,
}
}
agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="https://example.com/v1?tenant=large",
)
assert agent.context_compressor.config_context_length == 1_048_576
def test_direct_start_drops_context_when_extra_trailing_segment_changes():
"""Only one conventional trailing slash is ignored for route identity."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://example.com/v1//",
"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 = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "https://large-tenant:secret@example.com/v1",
"context_length": 1_048_576,
}
}
agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="https://small-tenant:secret@example.com/v1",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_drops_context_when_ipv6_zone_case_changes():
"""IPv6 address hex is case-insensitive, but its zone identifier is not."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom",
"base_url": "http://[FE80::1%25ETH0]/v1",
"context_length": 1_048_576,
}
}
agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="http://[fe80::1%25eth0]/v1",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_preserves_context_for_provider_alias():
"""Canonical provider aliases identify the same route when no URL is pinned."""
cfg = {
@ -222,6 +540,66 @@ def test_direct_start_preserves_context_for_provider_alias():
assert agent.context_compressor.config_context_length == 1_000_000
def test_direct_start_preserves_context_for_registry_provider_alias():
"""Legacy and models.dev provider IDs may identify the same route."""
cfg = {
"model": {
"default": "kimi-k3",
"provider": "kimi-for-coding",
"context_length": 1_048_576,
}
}
agent = _make_direct_start_agent(
cfg,
model="kimi-k3",
provider="kimi-coding",
base_url="https://api.kimi.com/coding",
)
assert agent.context_compressor.config_context_length == 1_048_576
def test_direct_start_preserves_context_for_profile_route_on_shared_host():
"""Exact provider-profile routes disambiguate providers sharing a hostname."""
cfg = {
"model": {
"default": "gpt-5.4",
"provider": "opencode-zen",
"context_length": 1_000_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.4",
provider="opencode",
base_url="https://opencode.ai/zen/v1",
)
assert agent.context_compressor.config_context_length == 1_000_000
def test_direct_start_drops_context_for_profile_wrong_path():
"""A shared hostname cannot substitute for a profile's complete route."""
cfg = {
"model": {
"default": "gpt-5.4",
"provider": "opencode-go",
"context_length": 1_000_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.4",
provider="opencode-go",
base_url="https://opencode.ai/unrelated",
)
assert agent.context_compressor.config_context_length is None
def test_direct_start_named_custom_route_resolves_configured_base_url():
"""Named custom providers must not collapse to one generic custom route."""
cfg = {