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

refactor: isolate custom health identity and trim redundant alias tests

Browse Source
This commit is contained in:
Teknium
2026-09-07 02:21:10 -07:00
parent 84bbf176a3
commit cb1a42d33b
3 changed files with 42 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
agent/auxiliary_client.py
View File

@@ -112,7 +112,7 @@ from agent.model_metadata import (
strip_codex_context_variant_suffix as _strip_codex_ctx_variant,
)
from hermes_cli.config import get_hermes_home
from hermes_cli.route_identity import normalize_route_base_url
from agent.auxiliary_health import _custom_health_base_url, _unhealthy_cache_key
from hermes_constants import OPENROUTER_BASE_URL
from utils import base_url_host_matches, base_url_hostname, env_float, is_truthy_value, model_forces_max_completion_tokens, normalize_proxy_env_vars
@@ -2901,33 +2901,6 @@ def _normalize_chain_label(provider: str) -> str:
return _AUX_UNHEALTHY_LABEL_ALIASES.get(p, p)
def _unhealthy_cache_key(provider: str, base_url: Optional[str] = None) -> Any:
"""Provider-wide key, or endpoint-specific key for an explicit custom endpoint."""
label = _normalize_chain_label(provider)
endpoint = normalize_route_base_url(_custom_health_base_url(provider, base_url))
if endpoint:
return "custom-endpoint", endpoint
return label
def _custom_health_base_url(provider: str, explicit_base_url: Optional[str] = None) -> str:
"""Return the concrete custom endpoint used to scope health and failed-route checks."""
explicit = str(explicit_base_url or "").strip()
label = _normalize_chain_label(provider)
if label == "local/custom":
return explicit or _current_custom_base_url()
if label.startswith("custom:") and explicit:
return explicit
with contextlib.suppress(ImportError):
from hermes_cli.runtime_provider import _get_named_custom_provider, _resolves_to_custom
if _resolves_to_custom(label):
return explicit or _current_custom_base_url()
entry = _get_named_custom_provider(provider)
if entry:
return explicit or str(entry.get("base_url") or "").strip()
return ""
def _mark_provider_unhealthy(
provider: str, ttl: Optional[float] = None, *, base_url: Optional[str] = None,
) -> None:
@@ -6867,7 +6840,11 @@ def _ladder_provider_fallback(first_err: Exception, route: _LadderRoute):
# separate custom URLs can carry separate credentials (or no billing relationship at all).
_chain_failed_model = None if reason in ("auth error", "payment error") else route.final_model
from agent.backend_identity import FailureScope
_chain_failure_scope = FailureScope.ENDPOINT if reason == "payment error" else None
_chain_failure_scope = (
FailureScope.ENDPOINT
if reason == "payment error" and _custom_health_base_url(resolved_provider, route.base_info)
else None
)
fb_client, fb_model, fb_label = _try_configured_fallback_chain(
task, resolved_provider or "auto", reason=reason, failed_model=_chain_failed_model,
failed_base_url=route.base_info, failure_scope=_chain_failure_scope)

36
agent/auxiliary_health.py Normal file
View File

@@ -0,0 +1,36 @@
"""Endpoint identity for auxiliary custom-provider health checks."""
import contextlib
from typing import Any, Optional
from hermes_cli.route_identity import normalize_route_base_url
def _unhealthy_cache_key(provider: str, base_url: Optional[str] = None) -> Any:
"""Provider-wide key, or endpoint-specific key for an explicit custom endpoint."""
from agent.auxiliary_client import _normalize_chain_label
label = _normalize_chain_label(provider)
endpoint = normalize_route_base_url(_custom_health_base_url(provider, base_url))
if endpoint:
return "custom-endpoint", endpoint
return label
def _custom_health_base_url(provider: str, explicit_base_url: Optional[str] = None) -> str:
"""Return the concrete custom endpoint used to scope health and failed-route checks."""
from agent.auxiliary_client import _current_custom_base_url
explicit = str(explicit_base_url or "").strip()
from agent.auxiliary_client import _normalize_chain_label
label = _normalize_chain_label(provider)
if label == "local/custom":
return explicit or _current_custom_base_url()
if label.startswith("custom:") and explicit:
return explicit
with contextlib.suppress(ImportError):
from hermes_cli.runtime_provider import _get_named_custom_provider, _resolves_to_custom
if _resolves_to_custom(label):
return explicit or _current_custom_base_url()
entry = _get_named_custom_provider(provider)
if entry:
return explicit or str(entry.get("base_url") or "").strip()
return ""

98
tests/agent/test_auxiliary_client.py
View File

@@ -4285,104 +4285,6 @@ class TestAuxUnhealthyCache:
assert _is_provider_unhealthy("custom", hosted_url) is True
assert _is_provider_unhealthy("custom", local_url) is False
def test_named_custom_main_route_honors_endpoint_quarantine(self):
"""A named custom main route is skipped before its client is resolved."""
from agent.auxiliary_client import _mark_provider_unhealthy, _try_main_provider_route
hosted_url = "https://hosted.example/v1"
_mark_provider_unhealthy("custom:hosted", base_url=hosted_url)
with patch(
"hermes_cli.runtime_provider._get_named_custom_provider",
return_value={"name": "hosted", "base_url": hosted_url},
), patch("agent.auxiliary_client.resolve_provider_client") as resolver:
routed = _try_main_provider_route("custom:hosted", "hosted-model", "", "", "")
assert routed is None
resolver.assert_not_called()
def test_custom_aliases_share_endpoint_quarantine(self):
"""Aliases for one custom URL stay quarantined while a distinct endpoint remains eligible."""
from agent.auxiliary_client import _mark_provider_unhealthy, _try_configured_fallback_chain
hosted_url = "https://hosted.example/v1"
local_url = "http://127.0.0.1:8080/v1"
hosted_entry = {"provider": "custom:mirror", "model": "hosted-model"}
local_entry = {"provider": "custom", "model": "local-model", "base_url": local_url}
local_client = MagicMock()
_mark_provider_unhealthy("custom:primary", base_url=hosted_url)
def named_provider(name):
if name == "custom:mirror":
return {"name": "mirror", "base_url": hosted_url}
return None
with patch(
"agent.auxiliary_client._get_auxiliary_task_config",
return_value={"fallback_chain": [hosted_entry, local_entry]},
), patch(
"hermes_cli.runtime_provider._get_named_custom_provider",
side_effect=named_provider,
), patch(
"agent.auxiliary_client._resolve_fallback_entry",
return_value=(local_client, "local-model"),
) as resolver:
client, model, label = _try_configured_fallback_chain(
"compression", "openrouter", reason="payment error"
)
assert client is local_client
assert model == "local-model"
assert label == "fallback_chain[1](custom)"
resolver.assert_called_once_with(local_entry)
def test_bare_named_custom_aliases_share_endpoint_quarantine(self):
"""Bare registry aliases cannot retry a quarantined custom endpoint."""
from agent.auxiliary_client import _mark_provider_unhealthy, _try_configured_fallback_chain
hosted_url = "https://hosted.example/v1"
local_url = "http://127.0.0.1:8080/v1"
hosted_entry = {"provider": "mirror", "model": "hosted-model"}
local_entry = {"provider": "custom", "model": "local-model", "base_url": local_url}
local_client = MagicMock()
_mark_provider_unhealthy("custom:primary", base_url=hosted_url)
with patch(
"agent.auxiliary_client._get_auxiliary_task_config",
return_value={"fallback_chain": [hosted_entry, local_entry]},
), patch(
"hermes_cli.runtime_provider._get_named_custom_provider",
side_effect=lambda name: (
{"name": "mirror", "base_url": hosted_url} if name == "mirror" else None
),
), patch(
"agent.auxiliary_client._resolve_fallback_entry",
return_value=(local_client, "local-model"),
) as resolver:
client, model, label = _try_configured_fallback_chain(
"compression", "openrouter", reason="payment error"
)
assert (client, model, label) == (
local_client, "local-model", "fallback_chain[1](custom)",
)
resolver.assert_called_once_with(local_entry)
def test_bare_named_custom_main_route_honors_endpoint_quarantine(self):
"""A bare named custom main route is skipped before client resolution."""
from agent.auxiliary_client import _mark_provider_unhealthy, _try_main_provider_route
hosted_url = "https://hosted.example/v1"
_mark_provider_unhealthy("custom:primary", base_url=hosted_url)
with patch(
"hermes_cli.runtime_provider._get_named_custom_provider",
return_value={"name": "hosted", "base_url": hosted_url},
), patch("agent.auxiliary_client.resolve_provider_client") as resolver:
routed = _try_main_provider_route("hosted", "hosted-model", "", "", "")
assert routed is None
resolver.assert_not_called()
# ── auxiliary_max_tokens_param ──────────────────────────────────────────────