Merge branch 'main' into feat/dashboard-skill-analytics

This commit is contained in:
Austin Pickett
2026-04-20 05:25:49 -07:00
committed by GitHub
1022 changed files with 157411 additions and 17823 deletions

View File

@@ -951,13 +951,21 @@ class TestBuildAnthropicKwargs:
max_tokens=4096,
reasoning_config={"enabled": True, "effort": "high"},
)
assert kwargs["thinking"] == {"type": "adaptive"}
# Adaptive thinking + display="summarized" keeps reasoning text
# populated in the response stream (Opus 4.7 default is "omitted").
assert kwargs["thinking"] == {"type": "adaptive", "display": "summarized"}
assert kwargs["output_config"] == {"effort": "high"}
assert "budget_tokens" not in kwargs["thinking"]
assert "temperature" not in kwargs
assert kwargs["max_tokens"] == 4096
def test_reasoning_config_maps_xhigh_to_max_effort_for_4_6_models(self):
def test_reasoning_config_downgrades_xhigh_to_max_for_4_6_models(self):
# Opus 4.7 added "xhigh" as a distinct effort level (low/medium/high/
# xhigh/max). Opus 4.6 only supports low/medium/high/max — sending
# "xhigh" there returns an API 400. Preserve the pre-migration
# behavior of aliasing xhigh→max on pre-4.7 adaptive models so users
# who prefer xhigh as their default don't 400 every request when
# switching back to 4.6.
kwargs = build_anthropic_kwargs(
model="claude-sonnet-4-6",
messages=[{"role": "user", "content": "think harder"}],
@@ -965,9 +973,53 @@ class TestBuildAnthropicKwargs:
max_tokens=4096,
reasoning_config={"enabled": True, "effort": "xhigh"},
)
assert kwargs["thinking"] == {"type": "adaptive"}
assert kwargs["thinking"] == {"type": "adaptive", "display": "summarized"}
assert kwargs["output_config"] == {"effort": "max"}
def test_reasoning_config_preserves_xhigh_for_4_7_models(self):
# On 4.7+ xhigh is a real level and the recommended default for
# coding/agentic work — keep it distinct from max.
kwargs = build_anthropic_kwargs(
model="claude-opus-4-7",
messages=[{"role": "user", "content": "think harder"}],
tools=None,
max_tokens=4096,
reasoning_config={"enabled": True, "effort": "xhigh"},
)
assert kwargs["thinking"] == {"type": "adaptive", "display": "summarized"}
assert kwargs["output_config"] == {"effort": "xhigh"}
def test_reasoning_config_maps_max_effort_for_4_7_models(self):
kwargs = build_anthropic_kwargs(
model="claude-opus-4-7",
messages=[{"role": "user", "content": "maximum reasoning please"}],
tools=None,
max_tokens=4096,
reasoning_config={"enabled": True, "effort": "max"},
)
assert kwargs["thinking"] == {"type": "adaptive", "display": "summarized"}
assert kwargs["output_config"] == {"effort": "max"}
def test_opus_4_7_strips_sampling_params(self):
# Opus 4.7 returns 400 on non-default temperature/top_p/top_k.
# build_anthropic_kwargs must strip them as a safety net even if an
# upstream caller injects them for older-model compatibility.
kwargs = build_anthropic_kwargs(
model="claude-opus-4-7",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
reasoning_config=None,
)
# Manually inject sampling params then re-run through the guard.
# Because build_anthropic_kwargs doesn't currently accept sampling
# params through its signature, we exercise the strip behavior by
# calling the internal predicate directly.
from agent.anthropic_adapter import _forbids_sampling_params
assert _forbids_sampling_params("claude-opus-4-7") is True
assert _forbids_sampling_params("claude-opus-4-6") is False
assert _forbids_sampling_params("claude-sonnet-4-5") is False
def test_reasoning_disabled(self):
kwargs = build_anthropic_kwargs(
model="claude-sonnet-4-20250514",
@@ -1248,6 +1300,21 @@ class TestNormalizeResponse:
assert r2 == "tool_calls"
assert r3 == "length"
def test_stop_reason_refusal_and_context_exceeded(self):
# Claude 4.5+ introduced two new stop_reason values the Messages API
# returns. We map both to OpenAI-style finish_reasons upstream
# handlers already understand, instead of silently collapsing to
# "stop" (old behavior).
block = SimpleNamespace(type="text", text="")
_, refusal_reason = normalize_anthropic_response(
self._make_response([block], "refusal")
)
_, overflow_reason = normalize_anthropic_response(
self._make_response([block], "model_context_window_exceeded")
)
assert refusal_reason == "content_filter"
assert overflow_reason == "length"
def test_no_text_content(self):
block = SimpleNamespace(
type="tool_use", id="tc_1", name="search", input={"q": "hi"}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,107 @@
"""Tests for agent.auxiliary_client._try_custom_endpoint's anthropic_messages branch.
When a user configures a custom endpoint with ``api_mode: anthropic_messages``
(e.g. MiniMax, Zhipu GLM, LiteLLM in Anthropic-proxy mode), auxiliary tasks
(compression, web_extract, session_search, title generation) must use the
native Anthropic transport rather than being silently downgraded to an
OpenAI-wire client that speaks the wrong protocol.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture(autouse=True)
def _clean_env(monkeypatch):
for key in (
"OPENAI_API_KEY", "OPENAI_BASE_URL",
"ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN",
):
monkeypatch.delenv(key, raising=False)
def _install_anthropic_adapter_mocks():
"""Patch build_anthropic_client so the test doesn't need the SDK."""
fake_client = MagicMock(name="anthropic_client")
return patch(
"agent.anthropic_adapter.build_anthropic_client",
return_value=fake_client,
), fake_client
def test_custom_endpoint_anthropic_messages_builds_anthropic_wrapper():
"""api_mode=anthropic_messages → returns AnthropicAuxiliaryClient, not OpenAI."""
from agent.auxiliary_client import _try_custom_endpoint, AnthropicAuxiliaryClient
with patch(
"agent.auxiliary_client._resolve_custom_runtime",
return_value=(
"https://api.minimax.io/anthropic",
"minimax-key",
"anthropic_messages",
),
), patch(
"agent.auxiliary_client._read_main_model",
return_value="claude-sonnet-4-6",
):
adapter_patch, fake_client = _install_anthropic_adapter_mocks()
with adapter_patch:
client, model = _try_custom_endpoint()
assert isinstance(client, AnthropicAuxiliaryClient), (
"Custom endpoint with api_mode=anthropic_messages must return the "
f"native Anthropic wrapper, got {type(client).__name__}"
)
assert model == "claude-sonnet-4-6"
# Wrapper should NOT be marked as OAuth — third-party endpoints are
# always API-key authenticated.
assert client.api_key == "minimax-key"
assert client.base_url == "https://api.minimax.io/anthropic"
def test_custom_endpoint_anthropic_messages_falls_back_when_sdk_missing():
"""Graceful degradation when anthropic SDK is unavailable."""
from agent.auxiliary_client import _try_custom_endpoint
import_error = ImportError("anthropic package not installed")
with patch(
"agent.auxiliary_client._resolve_custom_runtime",
return_value=("https://api.minimax.io/anthropic", "k", "anthropic_messages"),
), patch(
"agent.auxiliary_client._read_main_model",
return_value="claude-sonnet-4-6",
), patch(
"agent.anthropic_adapter.build_anthropic_client",
side_effect=import_error,
):
client, model = _try_custom_endpoint()
# Should fall back to an OpenAI-wire client rather than returning
# (None, None) — the tool still needs to do *something*.
assert client is not None
assert model == "claude-sonnet-4-6"
# OpenAI client, not AnthropicAuxiliaryClient.
from agent.auxiliary_client import AnthropicAuxiliaryClient
assert not isinstance(client, AnthropicAuxiliaryClient)
def test_custom_endpoint_chat_completions_still_uses_openai_wire():
"""Regression: default path (no api_mode) must remain OpenAI client."""
from agent.auxiliary_client import _try_custom_endpoint, AnthropicAuxiliaryClient
with patch(
"agent.auxiliary_client._resolve_custom_runtime",
return_value=("https://api.example.com/v1", "key", None),
), patch(
"agent.auxiliary_client._read_main_model",
return_value="my-model",
):
client, model = _try_custom_endpoint()
assert client is not None
assert model == "my-model"
assert not isinstance(client, AnthropicAuxiliaryClient)

View File

@@ -0,0 +1,311 @@
"""Regression tests for the ``auto`` → main-model-first policy.
Prior to this change, aggregator users (OpenRouter / Nous Portal) had aux
tasks routed through a cheap provider-side default (Gemini Flash) while
non-aggregator users got their main model. This made behavior inconsistent
and surprising — users picked Claude but got Gemini Flash summaries.
The current policy: ``auto`` means "use my main chat model" for every user,
regardless of provider type. Explicit per-task overrides in ``config.yaml``
(``auxiliary.<task>.provider``) still win. The cheap fallback chain only
runs when the main provider has no working client.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
# ── Text aux tasks — _resolve_auto ──────────────────────────────────────────
class TestResolveAutoMainFirst:
"""_resolve_auto() must prefer main provider + main model for every user."""
def test_openrouter_main_uses_main_model_for_aux(self, monkeypatch):
"""OpenRouter main user → aux uses their picked OR model, not Gemini Flash."""
monkeypatch.setenv("OPENROUTER_API_KEY", "or-test-key")
with patch(
"agent.auxiliary_client._read_main_provider",
return_value="openrouter",
), patch(
"agent.auxiliary_client._read_main_model",
return_value="anthropic/claude-sonnet-4.6",
), patch(
"agent.auxiliary_client.resolve_provider_client"
) as mock_resolve:
mock_client = MagicMock()
mock_resolve.return_value = (mock_client, "anthropic/claude-sonnet-4.6")
from agent.auxiliary_client import _resolve_auto
client, model = _resolve_auto()
assert client is mock_client
assert model == "anthropic/claude-sonnet-4.6"
# Verify it asked resolve_provider_client for the MAIN provider+model,
# not a fallback-chain provider
mock_resolve.assert_called_once()
assert mock_resolve.call_args.args[0] == "openrouter"
assert mock_resolve.call_args.args[1] == "anthropic/claude-sonnet-4.6"
def test_nous_main_uses_main_model_for_aux(self, monkeypatch):
"""Nous Portal main user → aux uses their picked Nous model, not free-tier MiMo."""
# No OPENROUTER_API_KEY → ensures if main failed we'd fall to chain
with patch(
"agent.auxiliary_client._read_main_provider", return_value="nous",
), patch(
"agent.auxiliary_client._read_main_model",
return_value="anthropic/claude-opus-4.6",
), patch(
"agent.auxiliary_client.resolve_provider_client"
) as mock_resolve:
mock_client = MagicMock()
mock_resolve.return_value = (mock_client, "anthropic/claude-opus-4.6")
from agent.auxiliary_client import _resolve_auto
client, model = _resolve_auto()
assert client is mock_client
assert model == "anthropic/claude-opus-4.6"
assert mock_resolve.call_args.args[0] == "nous"
def test_non_aggregator_main_still_uses_main(self, monkeypatch):
"""Non-aggregator main (DeepSeek) → unchanged behavior, main model used."""
monkeypatch.setenv("DEEPSEEK_API_KEY", "ds-test")
with patch(
"agent.auxiliary_client._read_main_provider", return_value="deepseek",
), patch(
"agent.auxiliary_client._read_main_model", return_value="deepseek-chat",
), patch(
"agent.auxiliary_client.resolve_provider_client"
) as mock_resolve:
mock_client = MagicMock()
mock_resolve.return_value = (mock_client, "deepseek-chat")
from agent.auxiliary_client import _resolve_auto
client, model = _resolve_auto()
assert client is mock_client
assert model == "deepseek-chat"
assert mock_resolve.call_args.args[0] == "deepseek"
def test_main_unavailable_falls_through_to_chain(self, monkeypatch):
"""Main provider with no working client → fall back to aux chain."""
monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")
chain_client = MagicMock()
with patch(
"agent.auxiliary_client._read_main_provider", return_value="anthropic",
), patch(
"agent.auxiliary_client._read_main_model", return_value="claude-opus",
), patch(
"agent.auxiliary_client.resolve_provider_client",
return_value=(None, None), # main provider has no client
), patch(
"agent.auxiliary_client._try_openrouter",
return_value=(chain_client, "google/gemini-3-flash-preview"),
):
from agent.auxiliary_client import _resolve_auto
client, model = _resolve_auto()
assert client is chain_client
assert model == "google/gemini-3-flash-preview"
def test_no_main_config_uses_chain_directly(self):
"""No main provider configured → skip step 1, use chain (no regression)."""
chain_client = MagicMock()
with patch(
"agent.auxiliary_client._read_main_provider", return_value="",
), patch(
"agent.auxiliary_client._read_main_model", return_value="",
), patch(
"agent.auxiliary_client._try_openrouter",
return_value=(chain_client, "google/gemini-3-flash-preview"),
):
from agent.auxiliary_client import _resolve_auto
client, model = _resolve_auto()
assert client is chain_client
def test_runtime_override_wins_over_config(self, monkeypatch):
"""main_runtime kwarg overrides config-read main provider/model."""
with patch(
"agent.auxiliary_client._read_main_provider",
return_value="openrouter",
), patch(
"agent.auxiliary_client._read_main_model", return_value="config-model",
), patch(
"agent.auxiliary_client.resolve_provider_client"
) as mock_resolve:
mock_resolve.return_value = (MagicMock(), "runtime-model")
from agent.auxiliary_client import _resolve_auto
_resolve_auto(main_runtime={
"provider": "anthropic",
"model": "runtime-model",
"base_url": "",
"api_key": "",
"api_mode": "",
})
# Runtime override wins
assert mock_resolve.call_args.args[0] == "anthropic"
assert mock_resolve.call_args.args[1] == "runtime-model"
# ── Vision — resolve_vision_provider_client ─────────────────────────────────
class TestResolveVisionMainFirst:
"""Vision auto-detection prefers main provider + main model first."""
def test_openrouter_main_vision_uses_main_model(self, monkeypatch):
"""OpenRouter main with vision-capable model → aux vision uses main model."""
monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")
with patch(
"agent.auxiliary_client._read_main_provider", return_value="openrouter",
), patch(
"agent.auxiliary_client._read_main_model",
return_value="anthropic/claude-sonnet-4.6",
), patch(
"agent.auxiliary_client.resolve_provider_client"
) as mock_resolve, patch(
"agent.auxiliary_client._resolve_task_provider_model",
return_value=("auto", None, None, None, None),
):
mock_client = MagicMock()
mock_resolve.return_value = (mock_client, "anthropic/claude-sonnet-4.6")
from agent.auxiliary_client import resolve_vision_provider_client
provider, client, model = resolve_vision_provider_client()
assert provider == "openrouter"
assert client is mock_client
assert model == "anthropic/claude-sonnet-4.6"
# Verify it did NOT call the strict vision backend for OpenRouter
# (which would have used a cheap gemini-flash-preview default)
mock_resolve.assert_called_once()
assert mock_resolve.call_args.args[0] == "openrouter"
assert mock_resolve.call_args.args[1] == "anthropic/claude-sonnet-4.6"
def test_nous_main_vision_uses_main_model(self):
"""Nous Portal main → aux vision uses main model, not free-tier MiMo-V2-Omni."""
with patch(
"agent.auxiliary_client._read_main_provider", return_value="nous",
), patch(
"agent.auxiliary_client._read_main_model",
return_value="openai/gpt-5",
), patch(
"agent.auxiliary_client.resolve_provider_client"
) as mock_resolve, patch(
"agent.auxiliary_client._resolve_task_provider_model",
return_value=("auto", None, None, None, None),
):
mock_client = MagicMock()
mock_resolve.return_value = (mock_client, "openai/gpt-5")
from agent.auxiliary_client import resolve_vision_provider_client
provider, client, model = resolve_vision_provider_client()
assert provider == "nous"
assert model == "openai/gpt-5"
def test_exotic_provider_with_vision_override_preserved(self):
"""xiaomi → mimo-v2-omni override still wins over main_model."""
with patch(
"agent.auxiliary_client._read_main_provider", return_value="xiaomi",
), patch(
"agent.auxiliary_client._read_main_model",
return_value="mimo-v2-pro", # text model
), patch(
"agent.auxiliary_client.resolve_provider_client"
) as mock_resolve, patch(
"agent.auxiliary_client._resolve_task_provider_model",
return_value=("auto", None, None, None, None),
):
mock_resolve.return_value = (MagicMock(), "mimo-v2-omni")
from agent.auxiliary_client import resolve_vision_provider_client
provider, client, model = resolve_vision_provider_client()
assert provider == "xiaomi"
# Should use mimo-v2-omni (vision override), not mimo-v2-pro (text main)
assert mock_resolve.call_args.args[1] == "mimo-v2-omni"
def test_main_unavailable_vision_falls_through_to_aggregators(self):
"""Main provider fails → fall back to OpenRouter/Nous strict backends."""
fallback_client = MagicMock()
with patch(
"agent.auxiliary_client._read_main_provider", return_value="deepseek",
), patch(
"agent.auxiliary_client._read_main_model", return_value="deepseek-chat",
), patch(
"agent.auxiliary_client.resolve_provider_client",
return_value=(None, None),
), patch(
"agent.auxiliary_client._resolve_strict_vision_backend",
return_value=(fallback_client, "google/gemini-3-flash-preview"),
), patch(
"agent.auxiliary_client._resolve_task_provider_model",
return_value=("auto", None, None, None, None),
):
from agent.auxiliary_client import resolve_vision_provider_client
provider, client, model = resolve_vision_provider_client()
assert client is fallback_client
assert provider in ("openrouter", "nous")
def test_explicit_provider_override_still_wins(self):
"""Explicit config override bypasses main-first policy."""
with patch(
"agent.auxiliary_client._read_main_provider", return_value="openrouter",
), patch(
"agent.auxiliary_client._read_main_model",
return_value="anthropic/claude-opus-4.6",
), patch(
"agent.auxiliary_client._resolve_task_provider_model",
return_value=("nous", None, None, None, None), # explicit override
), patch(
"agent.auxiliary_client._resolve_strict_vision_backend"
) as mock_strict:
mock_strict.return_value = (MagicMock(), "nous-default-model")
from agent.auxiliary_client import resolve_vision_provider_client
provider, client, model = resolve_vision_provider_client()
# Explicit "nous" override → uses strict backend, NOT main model path
assert provider == "nous"
mock_strict.assert_called_once_with("nous")
# ── Constant cleanup ────────────────────────────────────────────────────────
def test_aggregator_providers_constant_removed():
"""The dead _AGGREGATOR_PROVIDERS constant should no longer live in the module.
Removed when the main-first policy made the aggregator-skip guard obsolete.
"""
import agent.auxiliary_client as aux_mod
assert not hasattr(aux_mod, "_AGGREGATOR_PROVIDERS"), (
"_AGGREGATOR_PROVIDERS was removed when _resolve_auto stopped "
"treating aggregators specially. If you re-added it, the main-first "
"policy may have regressed."
)

View File

@@ -232,7 +232,7 @@ class TestResolveVisionProviderClientModelNormalization:
assert provider == "zai"
assert client is not None
assert model == "glm-5.1"
assert model == "glm-5v-turbo" # zai has dedicated vision model in _PROVIDER_VISION_MODELS
class TestVisionPathApiMode:

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,440 @@
"""Integration tests for the AWS Bedrock provider wiring.
Verifies that the Bedrock provider is correctly registered in the
provider registry, model catalog, and runtime resolution pipeline.
These tests do NOT require AWS credentials or boto3 — all AWS calls
are mocked.
Note: Tests that import ``hermes_cli.auth`` or ``hermes_cli.runtime_provider``
require Python 3.10+ due to ``str | None`` type syntax in the import chain.
"""
import os
from unittest.mock import MagicMock, patch
import pytest
class TestProviderRegistry:
"""Verify Bedrock is registered in PROVIDER_REGISTRY."""
def test_bedrock_in_registry(self):
from hermes_cli.auth import PROVIDER_REGISTRY
assert "bedrock" in PROVIDER_REGISTRY
def test_bedrock_auth_type_is_aws_sdk(self):
from hermes_cli.auth import PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY["bedrock"]
assert pconfig.auth_type == "aws_sdk"
def test_bedrock_has_no_api_key_env_vars(self):
"""Bedrock uses the AWS SDK credential chain, not API keys."""
from hermes_cli.auth import PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY["bedrock"]
assert pconfig.api_key_env_vars == ()
def test_bedrock_base_url_env_var(self):
from hermes_cli.auth import PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY["bedrock"]
assert pconfig.base_url_env_var == "BEDROCK_BASE_URL"
class TestProviderAliases:
"""Verify Bedrock aliases resolve correctly."""
def test_aws_alias(self):
from hermes_cli.models import _PROVIDER_ALIASES
assert _PROVIDER_ALIASES.get("aws") == "bedrock"
def test_aws_bedrock_alias(self):
from hermes_cli.models import _PROVIDER_ALIASES
assert _PROVIDER_ALIASES.get("aws-bedrock") == "bedrock"
def test_amazon_bedrock_alias(self):
from hermes_cli.models import _PROVIDER_ALIASES
assert _PROVIDER_ALIASES.get("amazon-bedrock") == "bedrock"
def test_amazon_alias(self):
from hermes_cli.models import _PROVIDER_ALIASES
assert _PROVIDER_ALIASES.get("amazon") == "bedrock"
class TestProviderLabels:
"""Verify Bedrock appears in provider labels."""
def test_bedrock_label(self):
from hermes_cli.models import _PROVIDER_LABELS
assert _PROVIDER_LABELS.get("bedrock") == "AWS Bedrock"
class TestModelCatalog:
"""Verify Bedrock has a static model fallback list."""
def test_bedrock_has_curated_models(self):
from hermes_cli.models import _PROVIDER_MODELS
models = _PROVIDER_MODELS.get("bedrock", [])
assert len(models) > 0
def test_bedrock_models_include_claude(self):
from hermes_cli.models import _PROVIDER_MODELS
models = _PROVIDER_MODELS.get("bedrock", [])
claude_models = [m for m in models if "anthropic.claude" in m]
assert len(claude_models) > 0
def test_bedrock_models_include_nova(self):
from hermes_cli.models import _PROVIDER_MODELS
models = _PROVIDER_MODELS.get("bedrock", [])
nova_models = [m for m in models if "amazon.nova" in m]
assert len(nova_models) > 0
class TestResolveProvider:
"""Verify resolve_provider() handles bedrock correctly."""
def test_explicit_bedrock_resolves(self, monkeypatch):
"""When user explicitly requests 'bedrock', it should resolve."""
from hermes_cli.auth import PROVIDER_REGISTRY
# bedrock is in the registry, so resolve_provider should return it
from hermes_cli.auth import resolve_provider
result = resolve_provider("bedrock")
assert result == "bedrock"
def test_aws_alias_resolves_to_bedrock(self):
from hermes_cli.auth import resolve_provider
result = resolve_provider("aws")
assert result == "bedrock"
def test_amazon_bedrock_alias_resolves(self):
from hermes_cli.auth import resolve_provider
result = resolve_provider("amazon-bedrock")
assert result == "bedrock"
def test_auto_detect_with_aws_credentials(self, monkeypatch):
"""When AWS credentials are present and no other provider is configured,
auto-detect should find bedrock."""
from hermes_cli.auth import resolve_provider
# Clear all other provider env vars
for var in ["OPENAI_API_KEY", "OPENROUTER_API_KEY", "ANTHROPIC_API_KEY",
"ANTHROPIC_TOKEN", "GOOGLE_API_KEY", "DEEPSEEK_API_KEY"]:
monkeypatch.delenv(var, raising=False)
# Set AWS credentials
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
# Mock the auth store to have no active provider
with patch("hermes_cli.auth._load_auth_store", return_value={}):
result = resolve_provider("auto")
assert result == "bedrock"
class TestRuntimeProvider:
"""Verify resolve_runtime_provider() handles bedrock correctly."""
def test_bedrock_runtime_resolution(self, monkeypatch):
from hermes_cli.runtime_provider import resolve_runtime_provider
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
monkeypatch.setenv("AWS_REGION", "eu-west-1")
# Mock resolve_provider to return bedrock
with patch("hermes_cli.runtime_provider.resolve_provider", return_value="bedrock"), \
patch("hermes_cli.runtime_provider._get_model_config", return_value={"provider": "bedrock"}):
result = resolve_runtime_provider(requested="bedrock")
assert result["provider"] == "bedrock"
assert result["api_mode"] == "bedrock_converse"
assert result["region"] == "eu-west-1"
assert "bedrock-runtime.eu-west-1.amazonaws.com" in result["base_url"]
assert result["api_key"] == "aws-sdk"
def test_bedrock_runtime_default_region(self, monkeypatch):
from hermes_cli.runtime_provider import resolve_runtime_provider
monkeypatch.setenv("AWS_PROFILE", "default")
monkeypatch.delenv("AWS_REGION", raising=False)
monkeypatch.delenv("AWS_DEFAULT_REGION", raising=False)
with patch("hermes_cli.runtime_provider.resolve_provider", return_value="bedrock"), \
patch("hermes_cli.runtime_provider._get_model_config", return_value={"provider": "bedrock"}):
result = resolve_runtime_provider(requested="bedrock")
assert result["region"] == "us-east-1"
def test_bedrock_runtime_no_credentials_raises_on_auto_detect(self, monkeypatch):
"""When bedrock is auto-detected (not explicitly requested) and no
credentials are found, runtime resolution should raise AuthError."""
from hermes_cli.runtime_provider import resolve_runtime_provider
from hermes_cli.auth import AuthError
# Clear all AWS env vars
for var in ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_PROFILE",
"AWS_BEARER_TOKEN_BEDROCK", "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI",
"AWS_WEB_IDENTITY_TOKEN_FILE"]:
monkeypatch.delenv(var, raising=False)
# Mock both the provider resolution and boto3's credential chain
mock_session = MagicMock()
mock_session.get_credentials.return_value = None
with patch("hermes_cli.runtime_provider.resolve_provider", return_value="bedrock"), \
patch("hermes_cli.runtime_provider._get_model_config", return_value={"provider": "bedrock"}), \
patch("hermes_cli.runtime_provider.resolve_requested_provider", return_value="auto"), \
patch.dict("sys.modules", {"botocore": MagicMock(), "botocore.session": MagicMock()}):
import botocore.session as _bs
_bs.get_session = MagicMock(return_value=mock_session)
with pytest.raises(AuthError, match="No AWS credentials"):
resolve_runtime_provider(requested="auto")
def test_bedrock_runtime_explicit_skips_credential_check(self, monkeypatch):
"""When user explicitly requests bedrock, trust boto3's credential chain
even if env-var detection finds nothing (covers IMDS, SSO, etc.)."""
from hermes_cli.runtime_provider import resolve_runtime_provider
# No AWS env vars set — but explicit bedrock request should not raise
for var in ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_PROFILE",
"AWS_BEARER_TOKEN_BEDROCK"]:
monkeypatch.delenv(var, raising=False)
with patch("hermes_cli.runtime_provider.resolve_provider", return_value="bedrock"), \
patch("hermes_cli.runtime_provider._get_model_config", return_value={"provider": "bedrock"}):
result = resolve_runtime_provider(requested="bedrock")
assert result["provider"] == "bedrock"
assert result["api_mode"] == "bedrock_converse"
# ---------------------------------------------------------------------------
# providers.py integration
# ---------------------------------------------------------------------------
class TestProvidersModule:
"""Verify bedrock is wired into hermes_cli/providers.py."""
def test_bedrock_alias_in_providers(self):
from hermes_cli.providers import ALIASES
assert ALIASES.get("bedrock") is None # "bedrock" IS the canonical name, not an alias
assert ALIASES.get("aws") == "bedrock"
assert ALIASES.get("aws-bedrock") == "bedrock"
def test_bedrock_transport_mapping(self):
from hermes_cli.providers import TRANSPORT_TO_API_MODE
assert TRANSPORT_TO_API_MODE.get("bedrock_converse") == "bedrock_converse"
def test_determine_api_mode_from_bedrock_url(self):
from hermes_cli.providers import determine_api_mode
assert determine_api_mode(
"unknown", "https://bedrock-runtime.us-east-1.amazonaws.com"
) == "bedrock_converse"
def test_label_override(self):
from hermes_cli.providers import _LABEL_OVERRIDES
assert _LABEL_OVERRIDES.get("bedrock") == "AWS Bedrock"
# ---------------------------------------------------------------------------
# Error classifier integration
# ---------------------------------------------------------------------------
class TestErrorClassifierBedrock:
"""Verify Bedrock error patterns are in the global error classifier."""
def test_throttling_in_rate_limit_patterns(self):
from agent.error_classifier import _RATE_LIMIT_PATTERNS
assert "throttlingexception" in _RATE_LIMIT_PATTERNS
def test_context_overflow_patterns(self):
from agent.error_classifier import _CONTEXT_OVERFLOW_PATTERNS
assert "input is too long" in _CONTEXT_OVERFLOW_PATTERNS
# ---------------------------------------------------------------------------
# pyproject.toml bedrock extra
# ---------------------------------------------------------------------------
class TestPackaging:
"""Verify bedrock optional dependency is declared."""
def test_bedrock_extra_exists(self):
import configparser
from pathlib import Path
# Read pyproject.toml to verify [bedrock] extra
toml_path = Path(__file__).parent.parent.parent / "pyproject.toml"
content = toml_path.read_text()
assert 'bedrock = ["boto3' in content
def test_bedrock_in_all_extra(self):
from pathlib import Path
content = (Path(__file__).parent.parent.parent / "pyproject.toml").read_text()
assert '"hermes-agent[bedrock]"' in content
# ---------------------------------------------------------------------------
# Model ID dot preservation — regression for #11976
# ---------------------------------------------------------------------------
# AWS Bedrock inference-profile model IDs embed structural dots:
#
# global.anthropic.claude-opus-4-7
# us.anthropic.claude-sonnet-4-5-20250929-v1:0
# apac.anthropic.claude-haiku-4-5
#
# ``agent.anthropic_adapter.normalize_model_name`` converts dots to hyphens
# unless the caller opts in via ``preserve_dots=True``. Before this fix,
# ``AIAgent._anthropic_preserve_dots`` returned False for the ``bedrock``
# provider, so Claude-on-Bedrock requests went out with
# ``global-anthropic-claude-opus-4-7`` (all dots mangled to hyphens) and
# Bedrock rejected them with:
#
# HTTP 400: The provided model identifier is invalid.
#
# The fix adds ``bedrock`` to the preserve-dots provider allowlist and
# ``bedrock-runtime.`` to the base-URL heuristic, mirroring the shape of
# the opencode-go fix for #5211 (commit f77be22c), which extended this
# same allowlist.
class TestBedrockPreserveDotsFlag:
"""``AIAgent._anthropic_preserve_dots`` must return True on Bedrock so
inference-profile IDs survive the normalize step intact."""
def test_bedrock_provider_preserves_dots(self):
from types import SimpleNamespace
agent = SimpleNamespace(provider="bedrock", base_url="")
from run_agent import AIAgent
assert AIAgent._anthropic_preserve_dots(agent) is True
def test_bedrock_runtime_us_east_1_url_preserves_dots(self):
"""Defense-in-depth: even without an explicit ``provider="bedrock"``,
a ``bedrock-runtime.us-east-1.amazonaws.com`` base URL must not
mangle dots."""
from types import SimpleNamespace
agent = SimpleNamespace(
provider="custom",
base_url="https://bedrock-runtime.us-east-1.amazonaws.com",
)
from run_agent import AIAgent
assert AIAgent._anthropic_preserve_dots(agent) is True
def test_bedrock_runtime_ap_northeast_2_url_preserves_dots(self):
"""Reporter-reported region (ap-northeast-2) exercises the same
base-URL heuristic."""
from types import SimpleNamespace
agent = SimpleNamespace(
provider="custom",
base_url="https://bedrock-runtime.ap-northeast-2.amazonaws.com",
)
from run_agent import AIAgent
assert AIAgent._anthropic_preserve_dots(agent) is True
def test_non_bedrock_aws_url_does_not_preserve_dots(self):
"""Unrelated AWS endpoints (e.g. ``s3.us-east-1.amazonaws.com``)
must not accidentally activate the dot-preservation heuristic —
the heuristic is scoped to the ``bedrock-runtime.`` substring
specifically."""
from types import SimpleNamespace
agent = SimpleNamespace(
provider="custom",
base_url="https://s3.us-east-1.amazonaws.com",
)
from run_agent import AIAgent
assert AIAgent._anthropic_preserve_dots(agent) is False
def test_anthropic_native_still_does_not_preserve_dots(self):
"""Canary: adding Bedrock to the allowlist must not weaken the
existing Anthropic native behaviour — ``claude-sonnet-4.6`` still
becomes ``claude-sonnet-4-6`` for the Anthropic API."""
from types import SimpleNamespace
agent = SimpleNamespace(provider="anthropic", base_url="https://api.anthropic.com")
from run_agent import AIAgent
assert AIAgent._anthropic_preserve_dots(agent) is False
class TestBedrockModelNameNormalization:
"""End-to-end: ``normalize_model_name`` + the preserve-dots flag
reproduce the exact production request shape for each Bedrock model
family, confirming the fix resolves the reporter's HTTP 400."""
def test_global_anthropic_inference_profile_preserved(self):
"""The reporter's exact model ID."""
from agent.anthropic_adapter import normalize_model_name
assert normalize_model_name(
"global.anthropic.claude-opus-4-7", preserve_dots=True
) == "global.anthropic.claude-opus-4-7"
def test_us_anthropic_dated_inference_profile_preserved(self):
"""Regional + dated Sonnet inference profile."""
from agent.anthropic_adapter import normalize_model_name
assert normalize_model_name(
"us.anthropic.claude-sonnet-4-5-20250929-v1:0",
preserve_dots=True,
) == "us.anthropic.claude-sonnet-4-5-20250929-v1:0"
def test_apac_anthropic_haiku_inference_profile_preserved(self):
"""APAC inference profile — same structural-dot shape."""
from agent.anthropic_adapter import normalize_model_name
assert normalize_model_name(
"apac.anthropic.claude-haiku-4-5", preserve_dots=True
) == "apac.anthropic.claude-haiku-4-5"
def test_preserve_false_mangles_as_documented(self):
"""Canary: with ``preserve_dots=False`` the function still
produces the broken all-hyphen form — this is the shape that
Bedrock rejected and that the fix avoids. Keeping this test
locks in the existing behaviour of ``normalize_model_name`` so a
future refactor doesn't accidentally decouple the knob from its
effect."""
from agent.anthropic_adapter import normalize_model_name
assert normalize_model_name(
"global.anthropic.claude-opus-4-7", preserve_dots=False
) == "global-anthropic-claude-opus-4-7"
def test_bare_foundation_model_id_preserved(self):
"""Non-inference-profile Bedrock IDs
(e.g. ``anthropic.claude-3-5-sonnet-20241022-v2:0``) use dots as
vendor separators and must also survive intact under
``preserve_dots=True``."""
from agent.anthropic_adapter import normalize_model_name
assert normalize_model_name(
"anthropic.claude-3-5-sonnet-20241022-v2:0",
preserve_dots=True,
) == "anthropic.claude-3-5-sonnet-20241022-v2:0"
class TestBedrockBuildAnthropicKwargsEndToEnd:
"""Integration: calling ``build_anthropic_kwargs`` with a Bedrock-
shaped model ID and ``preserve_dots=True`` produces the unmangled
model string in the outgoing kwargs — the exact body sent to the
``bedrock-runtime.`` endpoint. This is the integration-level
regression for the reporter's HTTP 400."""
def test_bedrock_inference_profile_survives_build_kwargs(self):
from agent.anthropic_adapter import build_anthropic_kwargs
kwargs = build_anthropic_kwargs(
model="global.anthropic.claude-opus-4-7",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
reasoning_config=None,
preserve_dots=True,
)
assert kwargs["model"] == "global.anthropic.claude-opus-4-7", (
"Bedrock inference-profile ID was mangled in build_anthropic_kwargs: "
f"{kwargs['model']!r}"
)
def test_bedrock_model_mangled_without_preserve_dots(self):
"""Inverse canary: without the flag, ``build_anthropic_kwargs``
still produces the broken form — so the fix in
``_anthropic_preserve_dots`` is the load-bearing piece that
wires ``preserve_dots=True`` through to this builder for the
Bedrock case."""
from agent.anthropic_adapter import build_anthropic_kwargs
kwargs = build_anthropic_kwargs(
model="global.anthropic.claude-opus-4-7",
messages=[{"role": "user", "content": "hi"}],
tools=None,
max_tokens=1024,
reasoning_config=None,
preserve_dots=False,
)
assert kwargs["model"] == "global-anthropic-claude-opus-4-7"

View File

@@ -0,0 +1,253 @@
"""Regression guard: Codex Cloudflare 403 mitigation headers.
The ``chatgpt.com/backend-api/codex`` endpoint sits behind a Cloudflare layer
that whitelists a small set of first-party originators (``codex_cli_rs``,
``codex_vscode``, ``codex_sdk_ts``, ``Codex*``). Requests from non-residential
IPs (VPS, always-on servers, some corporate egress) that don't advertise an
allowed originator are served 403 with ``cf-mitigated: challenge`` regardless
of auth correctness.
``_codex_cloudflare_headers`` in ``agent.auxiliary_client`` centralizes the
header set so the primary chat client (``run_agent.AIAgent.__init__`` +
``_apply_client_headers_for_base_url``) and the auxiliary client paths
(``_try_codex`` and the ``raw_codex`` branch of ``resolve_provider_client``)
all emit the same headers.
These tests pin:
- the originator value (must be ``codex_cli_rs`` — the whitelisted one)
- the User-Agent shape (codex_cli_rs-prefixed)
- ``ChatGPT-Account-ID`` extraction from the OAuth JWT (canonical casing,
from codex-rs ``auth.rs``)
- graceful handling of malformed tokens (drop the account-ID header, don't
raise)
- primary-client wiring at both entry points in ``run_agent.py``
- aux-client wiring at both entry points in ``agent/auxiliary_client.py``
"""
from __future__ import annotations
import base64
import json
from unittest.mock import MagicMock, patch
import pytest
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_codex_jwt(account_id: str = "acct-test-123") -> str:
"""Build a syntactically valid Codex-style JWT with the account_id claim."""
def b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
header = b64url(b'{"alg":"RS256","typ":"JWT"}')
claims = {
"sub": "user-xyz",
"exp": 9999999999,
"https://api.openai.com/auth": {
"chatgpt_account_id": account_id,
"chatgpt_plan_type": "plus",
},
}
payload = b64url(json.dumps(claims).encode())
sig = b64url(b"fake-sig")
return f"{header}.{payload}.{sig}"
# ---------------------------------------------------------------------------
# _codex_cloudflare_headers — the shared helper
# ---------------------------------------------------------------------------
class TestCodexCloudflareHeaders:
def test_originator_is_codex_cli_rs(self):
"""Cloudflare whitelists codex_cli_rs — any other value is 403'd."""
from agent.auxiliary_client import _codex_cloudflare_headers
headers = _codex_cloudflare_headers(_make_codex_jwt())
assert headers["originator"] == "codex_cli_rs"
def test_user_agent_advertises_codex_cli_rs(self):
from agent.auxiliary_client import _codex_cloudflare_headers
headers = _codex_cloudflare_headers(_make_codex_jwt())
assert headers["User-Agent"].startswith("codex_cli_rs/")
def test_account_id_extracted_from_jwt(self):
from agent.auxiliary_client import _codex_cloudflare_headers
headers = _codex_cloudflare_headers(_make_codex_jwt("acct-abc-999"))
# Canonical casing — matches codex-rs auth.rs
assert headers["ChatGPT-Account-ID"] == "acct-abc-999"
def test_canonical_header_casing(self):
"""Upstream codex-rs uses PascalCase with trailing -ID. Match exactly."""
from agent.auxiliary_client import _codex_cloudflare_headers
headers = _codex_cloudflare_headers(_make_codex_jwt())
assert "ChatGPT-Account-ID" in headers
# The lowercase/titlecase variants MUST NOT be used — pin to be explicit
assert "chatgpt-account-id" not in headers
assert "ChatGPT-Account-Id" not in headers
def test_malformed_token_drops_account_id_without_raising(self):
from agent.auxiliary_client import _codex_cloudflare_headers
for bad in ["not-a-jwt", "", "only.one", " ", "...."]:
headers = _codex_cloudflare_headers(bad)
# Still returns base headers — never raises
assert headers["originator"] == "codex_cli_rs"
assert "ChatGPT-Account-ID" not in headers
def test_non_string_token_handled(self):
from agent.auxiliary_client import _codex_cloudflare_headers
headers = _codex_cloudflare_headers(None) # type: ignore[arg-type]
assert headers["originator"] == "codex_cli_rs"
assert "ChatGPT-Account-ID" not in headers
def test_jwt_without_chatgpt_account_id_claim(self):
"""A valid JWT that lacks the account_id claim should still return headers."""
from agent.auxiliary_client import _codex_cloudflare_headers
import base64 as _b64, json as _json
def b64url(data: bytes) -> str:
return _b64.urlsafe_b64encode(data).rstrip(b"=").decode()
payload = b64url(_json.dumps({"sub": "user-xyz", "exp": 9999999999}).encode())
token = f"{b64url(b'{}')}.{payload}.{b64url(b'sig')}"
headers = _codex_cloudflare_headers(token)
assert headers["originator"] == "codex_cli_rs"
assert "ChatGPT-Account-ID" not in headers
# ---------------------------------------------------------------------------
# Primary chat client wiring (run_agent.AIAgent)
# ---------------------------------------------------------------------------
class TestPrimaryClientWiring:
def test_init_wires_codex_headers_for_chatgpt_base_url(self):
from run_agent import AIAgent
token = _make_codex_jwt("acct-primary-init")
with patch("run_agent.OpenAI") as mock_openai:
mock_openai.return_value = MagicMock()
AIAgent(
api_key=token,
base_url="https://chatgpt.com/backend-api/codex",
provider="openai-codex",
model="gpt-5.4",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
headers = mock_openai.call_args.kwargs.get("default_headers") or {}
assert headers.get("originator") == "codex_cli_rs"
assert headers.get("ChatGPT-Account-ID") == "acct-primary-init"
assert headers.get("User-Agent", "").startswith("codex_cli_rs/")
def test_apply_client_headers_on_base_url_change(self):
"""Credential-rotation / base-url change path must also emit codex headers."""
from run_agent import AIAgent
token = _make_codex_jwt("acct-rotation")
with patch("run_agent.OpenAI") as mock_openai:
mock_openai.return_value = MagicMock()
agent = AIAgent(
api_key="placeholder-openrouter-key",
base_url="https://openrouter.ai/api/v1",
provider="openrouter",
model="anthropic/claude-sonnet-4.6",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
# Simulate rotation into a Codex credential
agent._client_kwargs["api_key"] = token
agent._apply_client_headers_for_base_url(
"https://chatgpt.com/backend-api/codex"
)
headers = agent._client_kwargs.get("default_headers") or {}
assert headers.get("originator") == "codex_cli_rs"
assert headers.get("ChatGPT-Account-ID") == "acct-rotation"
assert headers.get("User-Agent", "").startswith("codex_cli_rs/")
def test_apply_client_headers_clears_codex_headers_off_chatgpt(self):
"""Switching AWAY from chatgpt.com must drop the codex headers."""
from run_agent import AIAgent
token = _make_codex_jwt()
with patch("run_agent.OpenAI") as mock_openai:
mock_openai.return_value = MagicMock()
agent = AIAgent(
api_key=token,
base_url="https://chatgpt.com/backend-api/codex",
provider="openai-codex",
model="gpt-5.4",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
# Sanity: headers are set initially
assert "originator" in (agent._client_kwargs.get("default_headers") or {})
agent._apply_client_headers_for_base_url(
"https://api.anthropic.com"
)
# default_headers should be popped for anthropic base
assert "default_headers" not in agent._client_kwargs
def test_openrouter_base_url_does_not_get_codex_headers(self):
from run_agent import AIAgent
with patch("run_agent.OpenAI") as mock_openai:
mock_openai.return_value = MagicMock()
AIAgent(
api_key="sk-or-test",
base_url="https://openrouter.ai/api/v1",
provider="openrouter",
model="anthropic/claude-sonnet-4.6",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
headers = mock_openai.call_args.kwargs.get("default_headers") or {}
assert headers.get("originator") != "codex_cli_rs"
# ---------------------------------------------------------------------------
# Auxiliary client wiring (agent.auxiliary_client)
# ---------------------------------------------------------------------------
class TestAuxiliaryClientWiring:
def test_try_codex_passes_codex_headers(self, monkeypatch):
"""_try_codex builds the OpenAI client used for compression / vision /
title generation when routed through Codex. Must emit codex headers."""
from agent import auxiliary_client
token = _make_codex_jwt("acct-aux-try-codex")
# Force _select_pool_entry to return "no pool" so we fall through to
# _read_codex_access_token.
monkeypatch.setattr(
auxiliary_client, "_select_pool_entry",
lambda provider: (False, None),
)
monkeypatch.setattr(
auxiliary_client, "_read_codex_access_token",
lambda: token,
)
with patch("agent.auxiliary_client.OpenAI") as mock_openai:
mock_openai.return_value = MagicMock()
client, model = auxiliary_client._try_codex()
assert client is not None
headers = mock_openai.call_args.kwargs.get("default_headers") or {}
assert headers.get("originator") == "codex_cli_rs"
assert headers.get("ChatGPT-Account-ID") == "acct-aux-try-codex"
assert headers.get("User-Agent", "").startswith("codex_cli_rs/")
def test_resolve_provider_client_raw_codex_passes_codex_headers(self, monkeypatch):
"""The ``raw_codex=True`` branch (used by the main agent loop for direct
responses.stream() access) must also emit codex headers."""
from agent import auxiliary_client
token = _make_codex_jwt("acct-aux-raw-codex")
monkeypatch.setattr(
auxiliary_client, "_read_codex_access_token",
lambda: token,
)
with patch("agent.auxiliary_client.OpenAI") as mock_openai:
mock_openai.return_value = MagicMock()
client, model = auxiliary_client.resolve_provider_client(
"openai-codex", raw_codex=True,
)
assert client is not None
headers = mock_openai.call_args.kwargs.get("default_headers") or {}
assert headers.get("originator") == "codex_cli_rs"
assert headers.get("ChatGPT-Account-ID") == "acct-aux-raw-codex"
assert headers.get("User-Agent", "").startswith("codex_cli_rs/")

View File

@@ -781,3 +781,127 @@ class TestTokenBudgetTailProtection:
# Tool at index 2 is outside the protected tail (last 3 = indices 2,3,4)
# so it might or might not be pruned depending on boundary
assert isinstance(pruned, int)
class TestTruncateToolCallArgsJson:
"""Regression tests for #11762.
The previous implementation produced invalid JSON by slicing
``function.arguments`` mid-string, which caused non-retryable 400s from
strict providers (observed on MiniMax) and stuck long sessions in a
re-send loop. The helper here must always emit parseable JSON whose
shape matches the original — shrunken, not corrupted.
"""
def _helper(self):
from agent.context_compressor import _truncate_tool_call_args_json
return _truncate_tool_call_args_json
def test_shrunken_args_remain_valid_json(self):
import json as _json
shrink = self._helper()
original = _json.dumps({
"path": "~/.hermes/skills/shopping/browser-setup-notes.md",
"content": "# Shopping Browser Setup Notes\n\n" + "abc " * 400,
})
assert len(original) > 500
shrunk = shrink(original)
parsed = _json.loads(shrunk) # must not raise
assert parsed["path"] == "~/.hermes/skills/shopping/browser-setup-notes.md"
assert parsed["content"].endswith("...[truncated]")
assert len(shrunk) < len(original)
def test_non_json_arguments_pass_through(self):
shrink = self._helper()
not_json = "this is not json at all, " * 50
assert shrink(not_json) == not_json
def test_short_string_leaves_unchanged(self):
import json as _json
shrink = self._helper()
payload = _json.dumps({"command": "ls -la", "cwd": "/tmp"})
assert _json.loads(shrink(payload)) == {"command": "ls -la", "cwd": "/tmp"}
def test_nested_structures_are_walked(self):
import json as _json
shrink = self._helper()
payload = _json.dumps({
"messages": [
{"role": "user", "content": "x" * 500},
{"role": "assistant", "content": "ok"},
],
"meta": {"note": "y" * 500},
})
parsed = _json.loads(shrink(payload))
assert parsed["messages"][0]["content"].endswith("...[truncated]")
assert parsed["messages"][1]["content"] == "ok"
assert parsed["meta"]["note"].endswith("...[truncated]")
def test_non_string_leaves_preserved(self):
import json as _json
shrink = self._helper()
payload = _json.dumps({
"retries": 3,
"enabled": True,
"timeout": None,
"items": [1, 2, 3],
"note": "z" * 500,
})
parsed = _json.loads(shrink(payload))
assert parsed["retries"] == 3
assert parsed["enabled"] is True
assert parsed["timeout"] is None
assert parsed["items"] == [1, 2, 3]
assert parsed["note"].endswith("...[truncated]")
def test_scalar_json_string_gets_shrunk(self):
import json as _json
shrink = self._helper()
payload = _json.dumps("q" * 500)
parsed = _json.loads(shrink(payload))
assert isinstance(parsed, str)
assert parsed.endswith("...[truncated]")
def test_unicode_preserved(self):
import json as _json
shrink = self._helper()
payload = _json.dumps({"content": "非德满" + ("a" * 500)})
out = shrink(payload)
# ensure_ascii=False keeps CJK intact rather than emitting \uXXXX
assert "非德满" in out
def test_pass3_emits_valid_json_for_downstream_provider(self):
"""End-to-end: Pass 3 must never produce the exact failure payload
that caused the 400 loop (unterminated string, missing brace)."""
import json as _json
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
c = ContextCompressor(
model="test/model",
threshold_percent=0.85,
protect_first_n=1,
protect_last_n=1,
quiet_mode=True,
)
huge_content = "# Shopping Browser Setup Notes\n\n## Overview\n" + "x " * 400
args_payload = _json.dumps({
"path": "~/.hermes/skills/shopping/browser-setup-notes.md",
"content": huge_content,
})
assert len(args_payload) > 500 # triggers the Pass-3 shrink
messages = [
{"role": "user", "content": "please write two files"},
{"role": "assistant", "content": None, "tool_calls": [
{"id": "call_1", "type": "function",
"function": {"name": "write_file", "arguments": args_payload}},
]},
{"role": "tool", "tool_call_id": "call_1",
"content": '{"bytes_written": 727}'},
{"role": "user", "content": "ok"},
{"role": "assistant", "content": "done"},
]
result, _ = c._prune_old_tool_results(messages, protect_tail_count=2)
shrunk = result[1]["tool_calls"][0]["function"]["arguments"]
# Must parse — otherwise downstream provider returns 400
parsed = _json.loads(shrunk)
assert parsed["path"] == "~/.hermes/skills/shopping/browser-setup-notes.md"
assert parsed["content"].endswith("...[truncated]")

View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import subprocess
from pathlib import Path
from unittest.mock import patch
import pytest
@@ -124,6 +125,31 @@ def test_expand_file_range_and_folder_listing(sample_repo: Path):
assert not result.warnings
def test_folder_listing_falls_back_when_rg_is_blocked(sample_repo: Path):
from agent.context_references import preprocess_context_references
real_run = subprocess.run
def blocked_rg(*args, **kwargs):
cmd = args[0] if args else kwargs.get("args")
if isinstance(cmd, list) and cmd and cmd[0] == "rg":
raise PermissionError("rg blocked by policy")
return real_run(*args, **kwargs)
with patch("agent.context_references.subprocess.run", side_effect=blocked_rg):
result = preprocess_context_references(
"Review @folder:src/",
cwd=sample_repo,
context_length=100_000,
)
assert result.expanded
assert "src/" in result.message
assert "main.py" in result.message
assert "helper.py" in result.message
assert not result.warnings
def test_expand_quoted_file_reference_with_spaces(tmp_path: Path):
from agent.context_references import preprocess_context_references

View File

@@ -252,6 +252,11 @@ def test_exhausted_402_entry_resets_after_one_hour(tmp_path, monkeypatch):
def test_explicit_reset_timestamp_overrides_default_429_ttl(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
# Prevent auto-seeding from Codex CLI tokens on the host
monkeypatch.setattr(
"hermes_cli.auth._import_codex_cli_tokens",
lambda: None,
)
_write_auth_store(
tmp_path,
{
@@ -1091,6 +1096,7 @@ def test_load_pool_seeds_copilot_via_gh_auth_token(tmp_path, monkeypatch):
assert len(entries) == 1
assert entries[0].source == "gh_cli"
assert entries[0].access_token == "gho_fake_token_abc123"
assert entries[0].base_url == "https://api.githubcopilot.com"
def test_load_pool_does_not_seed_copilot_when_no_token(tmp_path, monkeypatch):

View File

@@ -1,129 +1,25 @@
"""Tests for credential pool preservation through smart routing and 429 recovery.
"""Tests for credential pool preservation through turn config and 429 recovery.
Covers:
1. credential_pool flows through resolve_turn_route (no-route and fallback paths)
2. CLI _resolve_turn_agent_config passes credential_pool to primary dict
3. Gateway _resolve_turn_agent_config passes credential_pool to primary dict
4. Eager fallback deferred when credential pool has credentials
5. Eager fallback fires when no credential pool exists
6. Full 429 rotation cycle: retry-same → rotate → exhaust → fallback
1. CLI _resolve_turn_agent_config passes credential_pool to runtime dict
2. Gateway _resolve_turn_agent_config passes credential_pool to runtime dict
3. Eager fallback deferred when credential pool has credentials
4. Eager fallback fires when no credential pool exists
5. Full 429 rotation cycle: retry-same → rotate → exhaust → fallback
"""
import os
import time
from types import SimpleNamespace
from unittest.mock import MagicMock, patch, PropertyMock
import pytest
from unittest.mock import MagicMock, patch
# ---------------------------------------------------------------------------
# 1. smart_model_routing: credential_pool preserved in no-route path
# ---------------------------------------------------------------------------
class TestSmartRoutingPoolPreservation:
def test_no_route_preserves_credential_pool(self):
from agent.smart_model_routing import resolve_turn_route
fake_pool = MagicMock(name="CredentialPool")
primary = {
"model": "gpt-5.4",
"api_key": "sk-test",
"base_url": None,
"provider": "openai-codex",
"api_mode": "codex_responses",
"command": None,
"args": [],
"credential_pool": fake_pool,
}
# routing disabled
result = resolve_turn_route("hello", None, primary)
assert result["runtime"]["credential_pool"] is fake_pool
def test_no_route_none_pool(self):
from agent.smart_model_routing import resolve_turn_route
primary = {
"model": "gpt-5.4",
"api_key": "sk-test",
"base_url": None,
"provider": "openai-codex",
"api_mode": "codex_responses",
"command": None,
"args": [],
}
result = resolve_turn_route("hello", None, primary)
assert result["runtime"]["credential_pool"] is None
def test_routing_disabled_preserves_pool(self):
from agent.smart_model_routing import resolve_turn_route
fake_pool = MagicMock(name="CredentialPool")
primary = {
"model": "gpt-5.4",
"api_key": "sk-test",
"base_url": None,
"provider": "openai-codex",
"api_mode": "codex_responses",
"command": None,
"args": [],
"credential_pool": fake_pool,
}
# routing explicitly disabled
result = resolve_turn_route("hello", {"enabled": False}, primary)
assert result["runtime"]["credential_pool"] is fake_pool
def test_route_fallback_on_resolve_error_preserves_pool(self, monkeypatch):
"""When smart routing picks a cheap model but resolve_runtime_provider
fails, the fallback to primary must still include credential_pool."""
from agent.smart_model_routing import resolve_turn_route
fake_pool = MagicMock(name="CredentialPool")
primary = {
"model": "gpt-5.4",
"api_key": "sk-test",
"base_url": None,
"provider": "openai-codex",
"api_mode": "codex_responses",
"command": None,
"args": [],
"credential_pool": fake_pool,
}
routing_config = {
"enabled": True,
"cheap_model": "openai/gpt-4.1-mini",
"cheap_provider": "openrouter",
"max_tokens": 200,
"patterns": ["^(hi|hello|hey)"],
}
# Force resolve_runtime_provider to fail so it falls back to primary
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
MagicMock(side_effect=RuntimeError("no credentials")),
)
result = resolve_turn_route("hi", routing_config, primary)
assert result["runtime"]["credential_pool"] is fake_pool
# ---------------------------------------------------------------------------
# 2 & 3. CLI and Gateway _resolve_turn_agent_config include credential_pool
# 1. CLI _resolve_turn_agent_config includes credential_pool
# ---------------------------------------------------------------------------
class TestCliTurnRoutePool:
def test_resolve_turn_includes_pool(self, monkeypatch, tmp_path):
"""CLI's _resolve_turn_agent_config must pass credential_pool to primary."""
from agent.smart_model_routing import resolve_turn_route
captured = {}
def spy_resolve(user_message, routing_config, primary):
captured["primary"] = primary
return resolve_turn_route(user_message, routing_config, primary)
monkeypatch.setattr(
"agent.smart_model_routing.resolve_turn_route", spy_resolve
)
# Build a minimal HermesCLI-like object with the method
def test_resolve_turn_includes_pool(self):
"""CLI's _resolve_turn_agent_config must pass credential_pool in runtime."""
fake_pool = MagicMock(name="FakePool")
shell = SimpleNamespace(
model="gpt-5.4",
api_key="sk-test",
@@ -132,58 +28,46 @@ class TestCliTurnRoutePool:
api_mode="codex_responses",
acp_command=None,
acp_args=[],
_credential_pool=MagicMock(name="FakePool"),
_smart_model_routing={"enabled": False},
_credential_pool=fake_pool,
service_tier=None,
)
# Import and bind the real method
from cli import HermesCLI
bound = HermesCLI._resolve_turn_agent_config.__get__(shell)
bound("test message")
route = bound("test message")
assert "credential_pool" in captured["primary"]
assert captured["primary"]["credential_pool"] is shell._credential_pool
assert route["runtime"]["credential_pool"] is fake_pool
# ---------------------------------------------------------------------------
# 2. Gateway _resolve_turn_agent_config includes credential_pool
# ---------------------------------------------------------------------------
class TestGatewayTurnRoutePool:
def test_resolve_turn_includes_pool(self, monkeypatch):
def test_resolve_turn_includes_pool(self):
"""Gateway's _resolve_turn_agent_config must pass credential_pool."""
from agent.smart_model_routing import resolve_turn_route
captured = {}
def spy_resolve(user_message, routing_config, primary):
captured["primary"] = primary
return resolve_turn_route(user_message, routing_config, primary)
monkeypatch.setattr(
"agent.smart_model_routing.resolve_turn_route", spy_resolve
)
from gateway.run import GatewayRunner
runner = SimpleNamespace(
_smart_model_routing={"enabled": False},
)
fake_pool = MagicMock(name="FakePool")
runner = SimpleNamespace(_service_tier=None)
runtime_kwargs = {
"api_key": "sk-test",
"api_key": "***",
"base_url": None,
"provider": "openai-codex",
"api_mode": "codex_responses",
"command": None,
"args": [],
"credential_pool": MagicMock(name="FakePool"),
"credential_pool": fake_pool,
}
bound = GatewayRunner._resolve_turn_agent_config.__get__(runner)
bound("test message", "gpt-5.4", runtime_kwargs)
route = bound("test message", "gpt-5.4", runtime_kwargs)
assert "credential_pool" in captured["primary"]
assert captured["primary"]["credential_pool"] is runtime_kwargs["credential_pool"]
assert route["runtime"]["credential_pool"] is fake_pool
# ---------------------------------------------------------------------------
# 4 & 5. Eager fallback deferred/fires based on credential pool
# 3 & 4. Eager fallback deferred/fires based on credential pool
# ---------------------------------------------------------------------------
class TestEagerFallbackWithPool:
@@ -251,7 +135,7 @@ class TestEagerFallbackWithPool:
# ---------------------------------------------------------------------------
# 6. Full 429 rotation cycle via _recover_with_credential_pool
# 5. Full 429 rotation cycle via _recover_with_credential_pool
# ---------------------------------------------------------------------------
class TestPoolRotationCycle:

View File

@@ -83,6 +83,13 @@ class TestBuildToolPreview:
assert result is not None
assert "user" in result
def test_memory_replace_missing_old_text_marked(self):
# Avoid empty quotes "" in the preview when old_text is missing/None.
result = build_tool_preview("memory", {"action": "replace", "target": "memory"})
assert result == '~memory: "<missing old_text>"'
result = build_tool_preview("memory", {"action": "remove", "target": "memory", "old_text": None})
assert result == '-memory: "<missing old_text>"'
def test_session_search_preview(self):
result = build_tool_preview("session_search", {"query": "find something"})
assert result is not None

View File

@@ -849,3 +849,97 @@ class TestAdversarialEdgeCases:
)
result = classify_api_error(e, provider="openrouter")
assert result.reason == FailoverReason.model_not_found
# ── Regression: dict-typed message field (Issue #11233) ──
def test_pydantic_dict_message_no_crash(self):
"""Pydantic validation errors return message as dict, not string.
Regression: classify_api_error must not crash when body['message']
is a dict (e.g. {"detail": [...]} from FastAPI/Pydantic). The
'or ""' fallback only handles None/falsy values — a non-empty
dict is truthy and passed to .lower(), causing AttributeError.
"""
e = MockAPIError(
"Unprocessable Entity",
status_code=422,
body={
"object": "error",
"message": {
"detail": [
{
"type": "extra_forbidden",
"loc": ["body", "think"],
"msg": "Extra inputs are not permitted",
}
]
},
},
)
result = classify_api_error(e)
assert result.reason == FailoverReason.format_error
assert result.status_code == 422
assert result.retryable is False
def test_nested_error_dict_message_no_crash(self):
"""Nested body['error']['message'] as dict must not crash.
Some providers wrap Pydantic errors in an 'error' object.
"""
e = MockAPIError(
"Validation error",
status_code=400,
body={
"error": {
"message": {
"detail": [
{"type": "missing", "loc": ["body", "required"]}
]
}
}
},
)
result = classify_api_error(e, approx_tokens=1000)
assert result.reason == FailoverReason.format_error
assert result.status_code == 400
def test_metadata_raw_dict_message_no_crash(self):
"""OpenRouter metadata.raw with dict message must not crash."""
e = MockAPIError(
"Provider error",
status_code=400,
body={
"error": {
"message": "Provider error",
"metadata": {
"raw": '{"error":{"message":{"detail":[{"type":"invalid"}]}}}'
}
}
},
)
result = classify_api_error(e)
assert result.reason == FailoverReason.format_error
# Broader non-string type guards — defense against other provider quirks.
def test_list_message_no_crash(self):
"""Some providers return message as a list of error entries."""
e = MockAPIError(
"validation",
status_code=400,
body={"message": [{"msg": "field required"}]},
)
result = classify_api_error(e)
assert result is not None
def test_int_message_no_crash(self):
"""Any non-string type must be coerced safely."""
e = MockAPIError("server error", status_code=500, body={"message": 42})
result = classify_api_error(e)
assert result is not None
def test_none_message_still_works(self):
"""Regression: None fallback (the 'or \"\"' path) must still work."""
e = MockAPIError("server error", status_code=500, body={"message": None})
result = classify_api_error(e)
assert result is not None

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,315 @@
"""Tests for the native Google AI Studio Gemini adapter."""
from __future__ import annotations
import json
from types import SimpleNamespace
import pytest
class DummyResponse:
def __init__(self, status_code=200, payload=None, headers=None, text=None):
self.status_code = status_code
self._payload = payload or {}
self.headers = headers or {}
self.text = text if text is not None else json.dumps(self._payload)
def json(self):
return self._payload
def test_build_native_request_preserves_thought_signature_on_tool_replay():
from agent.gemini_native_adapter import build_gemini_request
request = build_gemini_request(
messages=[
{"role": "system", "content": "Be helpful."},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
"extra_content": {
"google": {"thought_signature": "sig-123"}
},
}
],
},
],
tools=[],
tool_choice=None,
)
parts = request["contents"][0]["parts"]
assert parts[0]["functionCall"]["name"] == "get_weather"
assert parts[0]["thoughtSignature"] == "sig-123"
def test_build_native_request_uses_original_function_name_for_tool_result():
from agent.gemini_native_adapter import build_gemini_request
request = build_gemini_request(
messages=[
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
}
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": '{"forecast": "sunny"}',
},
],
tools=[],
tool_choice=None,
)
tool_response = request["contents"][1]["parts"][0]["functionResponse"]
assert tool_response["name"] == "get_weather"
def test_build_native_request_strips_json_schema_only_fields_from_tool_parameters():
from agent.gemini_native_adapter import build_gemini_request
request = build_gemini_request(
messages=[{"role": "user", "content": "Hello"}],
tools=[
{
"type": "function",
"function": {
"name": "lookup_weather",
"description": "Weather lookup",
"parameters": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"additionalProperties": False,
"properties": {
"city": {
"type": "string",
"$schema": "ignored",
"description": "City name",
}
},
"required": ["city"],
},
},
}
],
tool_choice=None,
)
params = request["tools"][0]["functionDeclarations"][0]["parameters"]
assert "$schema" not in params
assert "additionalProperties" not in params
assert params["type"] == "object"
assert params["properties"]["city"] == {
"type": "string",
"description": "City name",
}
def test_translate_native_response_surfaces_reasoning_and_tool_calls():
from agent.gemini_native_adapter import translate_gemini_response
payload = {
"candidates": [
{
"content": {
"parts": [
{"thought": True, "text": "thinking..."},
{"functionCall": {"name": "search", "args": {"q": "hermes"}}},
]
},
"finishReason": "STOP",
}
],
"usageMetadata": {
"promptTokenCount": 10,
"candidatesTokenCount": 5,
"totalTokenCount": 15,
},
}
response = translate_gemini_response(payload, model="gemini-2.5-flash")
choice = response.choices[0]
assert choice.finish_reason == "tool_calls"
assert choice.message.reasoning == "thinking..."
assert choice.message.tool_calls[0].function.name == "search"
assert json.loads(choice.message.tool_calls[0].function.arguments) == {"q": "hermes"}
def test_native_client_uses_x_goog_api_key_and_native_models_endpoint(monkeypatch):
from agent.gemini_native_adapter import GeminiNativeClient
recorded = {}
class DummyHTTP:
def post(self, url, json=None, headers=None, timeout=None):
recorded["url"] = url
recorded["json"] = json
recorded["headers"] = headers
return DummyResponse(
payload={
"candidates": [
{
"content": {"parts": [{"text": "hello"}]},
"finishReason": "STOP",
}
],
"usageMetadata": {
"promptTokenCount": 1,
"candidatesTokenCount": 1,
"totalTokenCount": 2,
},
}
)
def close(self):
return None
monkeypatch.setattr("agent.gemini_native_adapter.httpx.Client", lambda *a, **k: DummyHTTP())
client = GeminiNativeClient(api_key="AIza-test", base_url="https://generativelanguage.googleapis.com/v1beta")
response = client.chat.completions.create(
model="gemini-2.5-flash",
messages=[{"role": "user", "content": "Hello"}],
)
assert recorded["url"] == "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent"
assert recorded["headers"]["x-goog-api-key"] == "AIza-test"
assert "Authorization" not in recorded["headers"]
assert response.choices[0].message.content == "hello"
def test_native_http_error_keeps_status_and_retry_after():
from agent.gemini_native_adapter import gemini_http_error
response = DummyResponse(
status_code=429,
headers={"Retry-After": "17"},
payload={
"error": {
"code": 429,
"message": "quota exhausted",
"status": "RESOURCE_EXHAUSTED",
"details": [
{
"@type": "type.googleapis.com/google.rpc.ErrorInfo",
"reason": "RESOURCE_EXHAUSTED",
"metadata": {"service": "generativelanguage.googleapis.com"},
}
],
}
},
)
err = gemini_http_error(response)
assert getattr(err, "status_code", None) == 429
assert getattr(err, "retry_after", None) == 17.0
assert "quota exhausted" in str(err)
def test_native_client_accepts_injected_http_client():
from agent.gemini_native_adapter import GeminiNativeClient
injected = SimpleNamespace(close=lambda: None)
client = GeminiNativeClient(api_key="AIza-test", http_client=injected)
assert client._http is injected
@pytest.mark.asyncio
async def test_async_native_client_streams_without_requiring_async_iterator_from_sync_client():
from agent.gemini_native_adapter import AsyncGeminiNativeClient
chunk = SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace(content="hi"), finish_reason=None)])
sync_stream = iter([chunk])
def _advance(iterator):
try:
return False, next(iterator)
except StopIteration:
return True, None
sync_client = SimpleNamespace(
api_key="AIza-test",
base_url="https://generativelanguage.googleapis.com/v1beta",
chat=SimpleNamespace(completions=SimpleNamespace(create=lambda **kwargs: sync_stream)),
_advance_stream_iterator=_advance,
close=lambda: None,
)
async_client = AsyncGeminiNativeClient(sync_client)
stream = await async_client.chat.completions.create(stream=True)
collected = []
async for item in stream:
collected.append(item)
assert collected == [chunk]
def test_stream_event_translation_emits_tool_call_delta_with_stable_index():
from agent.gemini_native_adapter import translate_stream_event
tool_call_indices = {}
event = {
"candidates": [
{
"content": {
"parts": [
{"functionCall": {"name": "search", "args": {"q": "abc"}}}
]
},
"finishReason": "STOP",
}
]
}
first = translate_stream_event(event, model="gemini-2.5-flash", tool_call_indices=tool_call_indices)
second = translate_stream_event(event, model="gemini-2.5-flash", tool_call_indices=tool_call_indices)
assert first[0].choices[0].delta.tool_calls[0].index == 0
assert second[0].choices[0].delta.tool_calls[0].index == 0
assert first[0].choices[0].delta.tool_calls[0].id == second[0].choices[0].delta.tool_calls[0].id
assert first[0].choices[0].delta.tool_calls[0].function.arguments == '{"q": "abc"}'
assert second[0].choices[0].delta.tool_calls[0].function.arguments == ""
assert first[-1].choices[0].finish_reason == "tool_calls"
def test_stream_event_translation_keeps_identical_calls_in_distinct_parts():
from agent.gemini_native_adapter import translate_stream_event
event = {
"candidates": [
{
"content": {
"parts": [
{"functionCall": {"name": "search", "args": {"q": "abc"}}},
{"functionCall": {"name": "search", "args": {"q": "abc"}}},
]
},
"finishReason": "STOP",
}
]
}
chunks = translate_stream_event(event, model="gemini-2.5-flash", tool_call_indices={})
tool_chunks = [chunk for chunk in chunks if chunk.choices[0].delta.tool_calls]
assert tool_chunks[0].choices[0].delta.tool_calls[0].index == 0
assert tool_chunks[1].choices[0].delta.tool_calls[0].index == 1
assert tool_chunks[0].choices[0].delta.tool_calls[0].id != tool_chunks[1].choices[0].delta.tool_calls[0].id

View File

@@ -462,8 +462,10 @@ class TestTerminalFormatting:
assert "Input tokens" in text
assert "Output tokens" in text
assert "Est. cost" in text
assert "$" in text
# Cost and cache metrics are intentionally hidden (pricing was unreliable).
assert "Est. cost" not in text
assert "Cache read" not in text
assert "Cache write" not in text
def test_terminal_format_shows_platforms(self, populated_db):
engine = InsightsEngine(populated_db)
@@ -482,8 +484,8 @@ class TestTerminalFormatting:
assert "" in text # Bar chart characters
def test_terminal_format_shows_na_for_custom_models(self, db):
"""Custom models should show N/A instead of fake cost."""
def test_terminal_format_hides_cost_for_custom_models(self, db):
"""Cost display is hidden entirely — custom models no longer show 'N/A' either."""
db.create_session(session_id="s1", source="cli", model="my-custom-model")
db.update_token_counts("s1", input_tokens=1000, output_tokens=500)
db._conn.commit()
@@ -492,8 +494,9 @@ class TestTerminalFormatting:
report = engine.generate(days=30)
text = engine.format_terminal(report)
assert "N/A" in text
assert "custom/self-hosted" in text
assert "N/A" not in text
assert "custom/self-hosted" not in text
assert "Cost" not in text
class TestGatewayFormatting:
@@ -512,7 +515,7 @@ class TestGatewayFormatting:
assert "**" in text # Markdown bold
def test_gateway_format_shows_cost(self, populated_db):
def test_gateway_format_hides_cost(self, populated_db):
engine = InsightsEngine(populated_db)
report = engine.generate(days=30)
text = engine.format_gateway(report)
@@ -520,6 +523,7 @@ class TestGatewayFormatting:
assert "$" in text
assert "Top Skills" in text
assert "Est. cost" in text
assert "cache" not in text.lower()
def test_gateway_format_shows_models(self, populated_db):
engine = InsightsEngine(populated_db)

View File

@@ -396,6 +396,108 @@ class TestPluginMemoryDiscovery:
assert load_memory_provider("nonexistent_provider") is None
class TestUserInstalledProviderDiscovery:
"""Memory providers installed to $HERMES_HOME/plugins/ should be found.
Regression test for issues #4956 and #9099: load_memory_provider() and
discover_memory_providers() only scanned the bundled plugins/memory/
directory, ignoring user-installed plugins.
"""
def _make_user_memory_plugin(self, tmp_path, name="myprovider"):
"""Create a minimal user memory provider plugin."""
plugin_dir = tmp_path / "plugins" / name
plugin_dir.mkdir(parents=True)
(plugin_dir / "__init__.py").write_text(
"from agent.memory_provider import MemoryProvider\n"
"class MyProvider(MemoryProvider):\n"
f" @property\n"
f" def name(self): return {name!r}\n"
" def is_available(self): return True\n"
" def initialize(self, **kw): pass\n"
" def sync_turn(self, *a, **kw): pass\n"
" def get_tool_schemas(self): return []\n"
" def handle_tool_call(self, *a, **kw): return '{}'\n"
)
(plugin_dir / "plugin.yaml").write_text(
f"name: {name}\ndescription: Test user provider\n"
)
return plugin_dir
def test_discover_finds_user_plugins(self, tmp_path, monkeypatch):
"""discover_memory_providers() includes user-installed plugins."""
from plugins.memory import discover_memory_providers, _get_user_plugins_dir
self._make_user_memory_plugin(tmp_path, "myexternal")
monkeypatch.setattr(
"plugins.memory._get_user_plugins_dir",
lambda: tmp_path / "plugins",
)
providers = discover_memory_providers()
names = [n for n, _, _ in providers]
assert "myexternal" in names
assert "holographic" in names # bundled still found
def test_load_user_plugin(self, tmp_path, monkeypatch):
"""load_memory_provider() can load from $HERMES_HOME/plugins/."""
from plugins.memory import load_memory_provider
self._make_user_memory_plugin(tmp_path, "myexternal")
monkeypatch.setattr(
"plugins.memory._get_user_plugins_dir",
lambda: tmp_path / "plugins",
)
p = load_memory_provider("myexternal")
assert p is not None
assert p.name == "myexternal"
assert p.is_available()
def test_bundled_takes_precedence(self, tmp_path, monkeypatch):
"""Bundled provider wins when user plugin has the same name."""
from plugins.memory import load_memory_provider, discover_memory_providers
# Create user plugin named "holographic" (same as bundled)
plugin_dir = tmp_path / "plugins" / "holographic"
plugin_dir.mkdir(parents=True)
(plugin_dir / "__init__.py").write_text(
"from agent.memory_provider import MemoryProvider\n"
"class Fake(MemoryProvider):\n"
" @property\n"
" def name(self): return 'holographic-FAKE'\n"
" def is_available(self): return True\n"
" def initialize(self, **kw): pass\n"
" def sync_turn(self, *a, **kw): pass\n"
" def get_tool_schemas(self): return []\n"
" def handle_tool_call(self, *a, **kw): return '{}'\n"
)
monkeypatch.setattr(
"plugins.memory._get_user_plugins_dir",
lambda: tmp_path / "plugins",
)
# Load should return bundled (name "holographic"), not user (name "holographic-FAKE")
p = load_memory_provider("holographic")
assert p is not None
assert p.name == "holographic" # bundled wins
# discover should not duplicate
providers = discover_memory_providers()
holo_count = sum(1 for n, _, _ in providers if n == "holographic")
assert holo_count == 1
def test_non_memory_user_plugins_excluded(self, tmp_path, monkeypatch):
"""User plugins that don't reference MemoryProvider are skipped."""
from plugins.memory import discover_memory_providers
plugin_dir = tmp_path / "plugins" / "notmemory"
plugin_dir.mkdir(parents=True)
(plugin_dir / "__init__.py").write_text(
"def register(ctx):\n ctx.register_tool('foo', 'bar', {}, lambda: None)\n"
)
monkeypatch.setattr(
"plugins.memory._get_user_plugins_dir",
lambda: tmp_path / "plugins",
)
providers = discover_memory_providers()
names = [n for n, _, _ in providers]
assert "notmemory" not in names
# ---------------------------------------------------------------------------
# Sequential dispatch routing tests
# ---------------------------------------------------------------------------
@@ -695,3 +797,214 @@ class TestMemoryContextFencing:
fence_end = combined.index("</memory-context>")
assert "Alice" in combined[fence_start:fence_end]
assert combined.index("weather") < fence_start
# ---------------------------------------------------------------------------
# AIAgent.commit_memory_session — routes to MemoryManager.on_session_end
# ---------------------------------------------------------------------------
class _CommitRecorder(FakeMemoryProvider):
"""Provider that records on_session_end calls for assertions."""
def __init__(self, name="recorder"):
super().__init__(name)
self.end_calls = []
def on_session_end(self, messages):
self.end_calls.append(list(messages or []))
class TestCommitMemorySessionRouting:
def test_on_session_end_fans_out(self):
mgr = MemoryManager()
builtin = _CommitRecorder("builtin")
external = _CommitRecorder("openviking")
mgr.add_provider(builtin)
mgr.add_provider(external)
msgs = [{"role": "user", "content": "hi"}]
mgr.on_session_end(msgs)
assert builtin.end_calls == [msgs]
assert external.end_calls == [msgs]
def test_on_session_end_tolerates_failure(self):
mgr = MemoryManager()
builtin = FakeMemoryProvider("builtin")
bad = _CommitRecorder("bad-provider")
bad.on_session_end = lambda m: (_ for _ in ()).throw(RuntimeError("boom"))
mgr.add_provider(builtin)
mgr.add_provider(bad)
mgr.on_session_end([]) # must not raise
# ---------------------------------------------------------------------------
# on_memory_write bridge — must fire from both concurrent AND sequential paths
# ---------------------------------------------------------------------------
class TestOnMemoryWriteBridge:
"""Verify that MemoryManager.on_memory_write is called when built-in
memory writes happen. This is a regression test for #10174 where the
sequential tool execution path (_execute_tool_calls_sequential) was
missing the bridge call, so single memory tool calls never notified
external memory providers.
"""
def test_on_memory_write_add(self):
"""on_memory_write fires for 'add' actions."""
mgr = MemoryManager()
p = FakeMemoryProvider("ext")
mgr.add_provider(p)
mgr.on_memory_write("add", "memory", "new fact")
assert p.memory_writes == [("add", "memory", "new fact")]
def test_on_memory_write_replace(self):
"""on_memory_write fires for 'replace' actions."""
mgr = MemoryManager()
p = FakeMemoryProvider("ext")
mgr.add_provider(p)
mgr.on_memory_write("replace", "user", "updated pref")
assert p.memory_writes == [("replace", "user", "updated pref")]
def test_on_memory_write_remove_not_bridged(self):
"""The bridge intentionally skips 'remove' — only add/replace notify."""
# This tests the contract that run_agent.py checks:
# function_args.get("action") in ("add", "replace")
mgr = MemoryManager()
p = FakeMemoryProvider("ext")
mgr.add_provider(p)
# Manager itself doesn't filter — run_agent.py does.
# But providers should handle remove gracefully.
mgr.on_memory_write("remove", "memory", "old fact")
assert p.memory_writes == [("remove", "memory", "old fact")]
def test_memory_manager_tool_injection_deduplicates(self):
"""Memory manager tools already in self.tools (from plugin registry)
must not be appended again. Duplicate function names cause 400 errors
on providers that enforce unique names (e.g. Xiaomi MiMo via Nous Portal).
Regression test for: duplicate mnemosyne_recall / mnemosyne_remember /
mnemosyne_stats in tools array → 400 from Nous Portal.
"""
mgr = MemoryManager()
p = FakeMemoryProvider("ext", tools=[
{"name": "ext_recall", "description": "Recall", "parameters": {}},
{"name": "ext_remember", "description": "Remember", "parameters": {}},
])
mgr.add_provider(p)
# Simulate self.tools already containing one of the plugin tools
# (as if it was registered via ctx.register_tool → get_tool_definitions)
existing_tools = [
{"type": "function", "function": {"name": "ext_recall", "description": "Recall (from registry)", "parameters": {}}},
{"type": "function", "function": {"name": "web_search", "description": "Search", "parameters": {}}},
]
# Apply the same dedup logic from run_agent.py __init__
_existing_names = {
t.get("function", {}).get("name")
for t in existing_tools
if isinstance(t, dict)
}
for _schema in mgr.get_all_tool_schemas():
_tname = _schema.get("name", "")
if _tname and _tname in _existing_names:
continue
existing_tools.append({"type": "function", "function": _schema})
if _tname:
_existing_names.add(_tname)
# ext_recall should NOT be duplicated; ext_remember should be added
tool_names = [t["function"]["name"] for t in existing_tools]
assert tool_names.count("ext_recall") == 1, f"ext_recall duplicated: {tool_names}"
assert tool_names.count("ext_remember") == 1
assert tool_names.count("web_search") == 1
assert len(existing_tools) == 3 # web_search + ext_recall + ext_remember
def test_on_memory_write_tolerates_provider_failure(self):
"""If a provider's on_memory_write raises, others still get notified."""
mgr = MemoryManager()
bad = FakeMemoryProvider("builtin")
bad.on_memory_write = MagicMock(side_effect=RuntimeError("boom"))
good = FakeMemoryProvider("good")
mgr.add_provider(bad)
mgr.add_provider(good)
mgr.on_memory_write("add", "user", "test")
# Good provider still received the call despite bad provider crashing
assert good.memory_writes == [("add", "user", "test")]
class TestHonchoCadenceTracking:
"""Verify Honcho provider cadence gating depends on on_turn_start().
Bug: _turn_count was never updated because on_turn_start() was not called
from run_conversation(). This meant cadence checks always passed (every
turn fired both context refresh and dialectic). Fixed by calling
on_turn_start(self._user_turn_count, msg) before prefetch_all().
"""
def test_turn_count_updates_on_turn_start(self):
"""on_turn_start sets _turn_count, enabling cadence math."""
from plugins.memory.honcho import HonchoMemoryProvider
p = HonchoMemoryProvider()
assert p._turn_count == 0
p.on_turn_start(1, "hello")
assert p._turn_count == 1
p.on_turn_start(5, "world")
assert p._turn_count == 5
def test_queue_prefetch_respects_dialectic_cadence(self):
"""With dialecticCadence=3, dialectic should skip turns 2 and 3."""
from plugins.memory.honcho import HonchoMemoryProvider
p = HonchoMemoryProvider()
p._dialectic_cadence = 3
p._recall_mode = "context"
p._session_key = "test-session"
# Simulate a manager that records prefetch calls
class FakeManager:
def prefetch_context(self, key, query=None):
pass
p._manager = FakeManager()
# Simulate turn 1: last_dialectic_turn = -999, so (1 - (-999)) >= 3 -> fires
p.on_turn_start(1, "turn 1")
p._last_dialectic_turn = 1 # simulate it fired
p._last_context_turn = 1
# Simulate turn 2: (2 - 1) = 1 < 3 -> should NOT fire dialectic
p.on_turn_start(2, "turn 2")
assert (p._turn_count - p._last_dialectic_turn) < p._dialectic_cadence
# Simulate turn 3: (3 - 1) = 2 < 3 -> should NOT fire dialectic
p.on_turn_start(3, "turn 3")
assert (p._turn_count - p._last_dialectic_turn) < p._dialectic_cadence
# Simulate turn 4: (4 - 1) = 3 >= 3 -> should fire dialectic
p.on_turn_start(4, "turn 4")
assert (p._turn_count - p._last_dialectic_turn) >= p._dialectic_cadence
def test_injection_frequency_first_turn_with_1indexed(self):
"""injection_frequency='first-turn' must inject on turn 1 (1-indexed)."""
from plugins.memory.honcho import HonchoMemoryProvider
p = HonchoMemoryProvider()
p._injection_frequency = "first-turn"
# Turn 1 should inject (not skip)
p.on_turn_start(1, "first message")
assert p._turn_count == 1
# The guard is `_turn_count > 1`, so turn 1 passes through
should_skip = p._injection_frequency == "first-turn" and p._turn_count > 1
assert not should_skip, "First turn (turn 1) should NOT be skipped"
# Turn 2 should skip
p.on_turn_start(2, "second message")
should_skip = p._injection_frequency == "first-turn" and p._turn_count > 1
assert should_skip, "Second turn (turn 2) SHOULD be skipped"

View File

@@ -208,34 +208,81 @@ class TestMem0UserIdScoping:
class TestHonchoUserIdScoping:
"""Verify Honcho plugin uses gateway user_id for peer_name when provided."""
"""Verify Honcho plugin keeps runtime user scoping separate from config peer_name."""
def test_gateway_user_id_overrides_peer_name(self):
"""When user_id is in kwargs and no explicit peer_name, user_id should be used."""
def test_gateway_user_id_is_passed_as_runtime_peer(self):
"""Gateway user_id should scope Honcho sessions without mutating config peer_name."""
from plugins.memory.honcho import HonchoMemoryProvider
provider = HonchoMemoryProvider()
# Create a mock config with NO explicit peer_name
mock_cfg = MagicMock()
mock_cfg.enabled = True
mock_cfg.api_key = "test-key"
mock_cfg.base_url = None
mock_cfg.peer_name = "" # No explicit peer_name — user_id should fill it
mock_cfg.recall_mode = "tools" # Use tools mode to defer session init
mock_cfg.peer_name = "static-user"
mock_cfg.recall_mode = "context"
mock_cfg.context_tokens = None
mock_cfg.raw = {}
mock_cfg.dialectic_depth = 1
mock_cfg.dialectic_depth_levels = None
mock_cfg.init_on_session_start = False
mock_cfg.ai_peer = "hermes"
mock_cfg.resolve_session_name.return_value = "test-sess"
mock_cfg.session_strategy = "shared"
with patch(
"plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
return_value=mock_cfg,
):
), patch(
"plugins.memory.honcho.client.get_honcho_client",
return_value=MagicMock(),
), patch(
"plugins.memory.honcho.session.HonchoSessionManager",
) as mock_manager_cls:
mock_manager = MagicMock()
mock_manager.get_or_create.return_value = MagicMock(messages=[])
mock_manager_cls.return_value = mock_manager
provider.initialize(
session_id="test-sess",
user_id="discord_user_789",
platform="discord",
)
# The config's peer_name should have been overridden with the user_id
assert mock_cfg.peer_name == "discord_user_789"
assert mock_cfg.peer_name == "static-user"
assert mock_manager_cls.call_args.kwargs["runtime_user_peer_name"] == "discord_user_789"
def test_session_manager_prefers_runtime_user_id_over_config_peer_name(self):
"""Session manager should isolate gateway users even when config peer_name is static."""
from plugins.memory.honcho.session import HonchoSessionManager
mock_cfg = MagicMock()
mock_cfg.peer_name = "static-user"
mock_cfg.ai_peer = "hermes"
mock_cfg.write_frequency = "sync"
mock_cfg.dialectic_reasoning_level = "low"
mock_cfg.dialectic_dynamic = True
mock_cfg.dialectic_max_chars = 600
mock_cfg.observation_mode = "directional"
mock_cfg.user_observe_me = True
mock_cfg.user_observe_others = True
mock_cfg.ai_observe_me = True
mock_cfg.ai_observe_others = True
manager = HonchoSessionManager(
honcho=MagicMock(),
config=mock_cfg,
runtime_user_peer_name="discord_user_789",
)
with patch.object(manager, "_get_or_create_peer", return_value=MagicMock()), patch.object(
manager,
"_get_or_create_honcho_session",
return_value=(MagicMock(), []),
):
session = manager.get_or_create("discord:channel-1")
assert session.user_peer_id == "discord_user_789"
def test_no_user_id_preserves_config_peer_name(self):
"""Without user_id, the config peer_name should be preserved."""

View File

@@ -113,8 +113,10 @@ class TestDefaultContextLengths:
for key, value in DEFAULT_CONTEXT_LENGTHS.items():
if "claude" not in key:
continue
# Claude 4.6 models have 1M context
if "4.6" in key or "4-6" in key:
# Claude 4.6+ models (4.6 and 4.7) have 1M context at standard
# API pricing (no long-context premium). Older Claude 4.x and
# 3.x models cap at 200k.
if any(tag in key for tag in ("4.6", "4-6", "4.7", "4-7")):
assert value == 1000000, f"{key} should be 1000000"
else:
assert value == 200000, f"{key} should be 200000"

View File

@@ -0,0 +1,253 @@
"""Tests for agent/nous_rate_guard.py — cross-session Nous Portal rate limit guard."""
import json
import os
import time
import pytest
@pytest.fixture
def rate_guard_env(tmp_path, monkeypatch):
"""Isolate rate guard state to a temp directory."""
hermes_home = str(tmp_path / ".hermes")
os.makedirs(hermes_home, exist_ok=True)
monkeypatch.setenv("HERMES_HOME", hermes_home)
# Clear any cached module-level imports
return hermes_home
class TestRecordNousRateLimit:
"""Test recording rate limit state."""
def test_records_with_header_reset(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
headers = {"x-ratelimit-reset-requests-1h": "1800"}
record_nous_rate_limit(headers=headers)
path = _state_path()
assert os.path.exists(path)
with open(path) as f:
state = json.load(f)
assert state["reset_seconds"] == pytest.approx(1800, abs=2)
assert state["reset_at"] > time.time()
def test_records_with_per_minute_header(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
headers = {"x-ratelimit-reset-requests": "45"}
record_nous_rate_limit(headers=headers)
with open(_state_path()) as f:
state = json.load(f)
assert state["reset_seconds"] == pytest.approx(45, abs=2)
def test_records_with_retry_after_header(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
headers = {"retry-after": "60"}
record_nous_rate_limit(headers=headers)
with open(_state_path()) as f:
state = json.load(f)
assert state["reset_seconds"] == pytest.approx(60, abs=2)
def test_prefers_hourly_over_per_minute(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
headers = {
"x-ratelimit-reset-requests-1h": "1800",
"x-ratelimit-reset-requests": "45",
}
record_nous_rate_limit(headers=headers)
with open(_state_path()) as f:
state = json.load(f)
# Should use the hourly value, not the per-minute one
assert state["reset_seconds"] == pytest.approx(1800, abs=2)
def test_falls_back_to_error_context_reset_at(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
future_reset = time.time() + 900
record_nous_rate_limit(
headers=None,
error_context={"reset_at": future_reset},
)
with open(_state_path()) as f:
state = json.load(f)
assert state["reset_at"] == pytest.approx(future_reset, abs=1)
def test_falls_back_to_default_cooldown(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
record_nous_rate_limit(headers=None)
with open(_state_path()) as f:
state = json.load(f)
# Default is 300 seconds (5 minutes)
assert state["reset_seconds"] == pytest.approx(300, abs=2)
def test_custom_default_cooldown(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
record_nous_rate_limit(headers=None, default_cooldown=120.0)
with open(_state_path()) as f:
state = json.load(f)
assert state["reset_seconds"] == pytest.approx(120, abs=2)
def test_creates_directory_if_missing(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, _state_path
record_nous_rate_limit(headers={"retry-after": "10"})
assert os.path.exists(_state_path())
class TestNousRateLimitRemaining:
"""Test checking remaining rate limit time."""
def test_returns_none_when_no_file(self, rate_guard_env):
from agent.nous_rate_guard import nous_rate_limit_remaining
assert nous_rate_limit_remaining() is None
def test_returns_remaining_seconds_when_active(self, rate_guard_env):
from agent.nous_rate_guard import record_nous_rate_limit, nous_rate_limit_remaining
record_nous_rate_limit(headers={"x-ratelimit-reset-requests-1h": "600"})
remaining = nous_rate_limit_remaining()
assert remaining is not None
assert 595 < remaining <= 605 # ~600 seconds, allowing for test execution time
def test_returns_none_when_expired(self, rate_guard_env):
from agent.nous_rate_guard import nous_rate_limit_remaining, _state_path
# Write an already-expired state
state_dir = os.path.dirname(_state_path())
os.makedirs(state_dir, exist_ok=True)
with open(_state_path(), "w") as f:
json.dump({"reset_at": time.time() - 10, "recorded_at": time.time() - 100}, f)
assert nous_rate_limit_remaining() is None
# File should be cleaned up
assert not os.path.exists(_state_path())
def test_handles_corrupt_file(self, rate_guard_env):
from agent.nous_rate_guard import nous_rate_limit_remaining, _state_path
state_dir = os.path.dirname(_state_path())
os.makedirs(state_dir, exist_ok=True)
with open(_state_path(), "w") as f:
f.write("not valid json{{{")
assert nous_rate_limit_remaining() is None
class TestClearNousRateLimit:
"""Test clearing rate limit state."""
def test_clears_existing_file(self, rate_guard_env):
from agent.nous_rate_guard import (
record_nous_rate_limit,
clear_nous_rate_limit,
nous_rate_limit_remaining,
_state_path,
)
record_nous_rate_limit(headers={"retry-after": "600"})
assert nous_rate_limit_remaining() is not None
clear_nous_rate_limit()
assert nous_rate_limit_remaining() is None
assert not os.path.exists(_state_path())
def test_clear_when_no_file(self, rate_guard_env):
from agent.nous_rate_guard import clear_nous_rate_limit
# Should not raise
clear_nous_rate_limit()
class TestFormatRemaining:
"""Test human-readable duration formatting."""
def test_seconds(self):
from agent.nous_rate_guard import format_remaining
assert format_remaining(30) == "30s"
def test_minutes(self):
from agent.nous_rate_guard import format_remaining
assert format_remaining(125) == "2m 5s"
def test_exact_minutes(self):
from agent.nous_rate_guard import format_remaining
assert format_remaining(120) == "2m"
def test_hours(self):
from agent.nous_rate_guard import format_remaining
assert format_remaining(3720) == "1h 2m"
class TestParseResetSeconds:
"""Test header parsing for reset times."""
def test_case_insensitive_headers(self, rate_guard_env):
from agent.nous_rate_guard import _parse_reset_seconds
headers = {"X-Ratelimit-Reset-Requests-1h": "1200"}
assert _parse_reset_seconds(headers) == 1200.0
def test_returns_none_for_empty_headers(self):
from agent.nous_rate_guard import _parse_reset_seconds
assert _parse_reset_seconds(None) is None
assert _parse_reset_seconds({}) is None
def test_ignores_zero_values(self):
from agent.nous_rate_guard import _parse_reset_seconds
headers = {"x-ratelimit-reset-requests-1h": "0"}
assert _parse_reset_seconds(headers) is None
def test_ignores_invalid_values(self):
from agent.nous_rate_guard import _parse_reset_seconds
headers = {"x-ratelimit-reset-requests-1h": "not-a-number"}
assert _parse_reset_seconds(headers) is None
class TestAuxiliaryClientIntegration:
"""Test that the auxiliary client respects the rate guard."""
def test_try_nous_skips_when_rate_limited(self, rate_guard_env, monkeypatch):
from agent.nous_rate_guard import record_nous_rate_limit
# Record a rate limit
record_nous_rate_limit(headers={"retry-after": "600"})
# Mock _read_nous_auth to return valid creds (would normally succeed)
import agent.auxiliary_client as aux
monkeypatch.setattr(aux, "_read_nous_auth", lambda: {
"access_token": "test-token",
"inference_base_url": "https://api.nous.test/v1",
})
result = aux._try_nous()
assert result == (None, None)
def test_try_nous_works_when_not_rate_limited(self, rate_guard_env, monkeypatch):
import agent.auxiliary_client as aux
# No rate limit recorded — _try_nous should proceed normally
# (will return None because no real creds, but won't be blocked
# by the rate guard)
monkeypatch.setattr(aux, "_read_nous_auth", lambda: None)
result = aux._try_nous()
assert result == (None, None)

View File

@@ -354,6 +354,24 @@ class TestBuildSkillsSystemPrompt:
assert "web-search" in result
assert "old-tool" not in result
def test_rebuilds_prompt_when_disabled_skills_change(self, monkeypatch, tmp_path):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
skill_dir = tmp_path / "skills" / "tools" / "cached-skill"
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text(
"---\nname: cached-skill\ndescription: Cached skill\n---\n"
)
first = build_skills_system_prompt()
assert "cached-skill" in first
(tmp_path / "config.yaml").write_text(
"skills:\n disabled: [cached-skill]\n"
)
second = build_skills_system_prompt()
assert "cached-skill" not in second
def test_includes_setup_needed_skills(self, monkeypatch, tmp_path):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("MISSING_API_KEY_XYZ", raising=False)
@@ -413,7 +431,7 @@ class TestBuildSkillsSystemPrompt:
class TestBuildNousSubscriptionPrompt:
def test_includes_active_subscription_features(self, monkeypatch):
monkeypatch.setenv("HERMES_ENABLE_NOUS_MANAGED_TOOLS", "1")
monkeypatch.setattr("tools.tool_backend_helpers.managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_subscription_features",
lambda config=None: NousSubscriptionFeatures(
@@ -437,7 +455,7 @@ class TestBuildNousSubscriptionPrompt:
assert "do not ask the user for Firecrawl, FAL, OpenAI TTS, or Browser-Use API keys" in prompt
def test_non_subscriber_prompt_includes_relevant_upgrade_guidance(self, monkeypatch):
monkeypatch.setenv("HERMES_ENABLE_NOUS_MANAGED_TOOLS", "1")
monkeypatch.setattr("tools.tool_backend_helpers.managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_subscription_features",
lambda config=None: NousSubscriptionFeatures(
@@ -460,7 +478,7 @@ class TestBuildNousSubscriptionPrompt:
assert "Do not mention subscription unless" in prompt
def test_feature_flag_off_returns_empty_prompt(self, monkeypatch):
monkeypatch.delenv("HERMES_ENABLE_NOUS_MANAGED_TOOLS", raising=False)
monkeypatch.setattr("tools.tool_backend_helpers.managed_nous_tools_enabled", lambda: False)
prompt = build_nous_subscription_prompt({"web_search"})

View File

@@ -0,0 +1,60 @@
"""Tests for malformed proxy env var and base URL validation.
Salvaged from PR #6403 by MestreY0d4-Uninter — validates that the agent
surfaces clear errors instead of cryptic httpx ``Invalid port`` exceptions
when proxy env vars or custom endpoint URLs are malformed.
"""
from __future__ import annotations
import pytest
from agent.auxiliary_client import _validate_base_url, _validate_proxy_env_urls
# -- proxy env validation ------------------------------------------------
def test_proxy_env_accepts_normal_values(monkeypatch):
monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:6153")
monkeypatch.setenv("HTTPS_PROXY", "https://proxy.example.com:8443")
monkeypatch.setenv("ALL_PROXY", "socks5://127.0.0.1:1080")
_validate_proxy_env_urls() # should not raise
def test_proxy_env_accepts_empty(monkeypatch):
monkeypatch.delenv("HTTP_PROXY", raising=False)
monkeypatch.delenv("HTTPS_PROXY", raising=False)
monkeypatch.delenv("ALL_PROXY", raising=False)
monkeypatch.delenv("http_proxy", raising=False)
monkeypatch.delenv("https_proxy", raising=False)
monkeypatch.delenv("all_proxy", raising=False)
_validate_proxy_env_urls() # should not raise
@pytest.mark.parametrize("key", [
"HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY",
"http_proxy", "https_proxy", "all_proxy",
])
def test_proxy_env_rejects_malformed_port(monkeypatch, key):
monkeypatch.setenv(key, "http://127.0.0.1:6153export")
with pytest.raises(RuntimeError, match=rf"Malformed proxy environment variable {key}=.*6153export"):
_validate_proxy_env_urls()
# -- base URL validation -------------------------------------------------
@pytest.mark.parametrize("url", [
"https://api.example.com/v1",
"http://127.0.0.1:6153/v1",
"acp://copilot",
"",
None,
])
def test_base_url_accepts_valid(url):
_validate_base_url(url) # should not raise
def test_base_url_rejects_malformed_port():
with pytest.raises(RuntimeError, match="Malformed custom endpoint URL"):
_validate_base_url("http://127.0.0.1:6153export")

View File

@@ -284,3 +284,95 @@ class TestElevenLabsTavilyExaKeys:
assert "XYZ789abcdef" not in result
assert "HOME=/home/user" in result
assert "SHELL=/bin/bash" in result
class TestJWTTokens:
"""JWT tokens start with eyJ (base64 for '{') and have dot-separated parts."""
def test_full_3part_jwt(self):
text = (
"Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"
".eyJpc3MiOiI0MjNiZDJkYjg4MjI0MDAwIn0"
".Gxgv0rru-_kS-I_60EJ7CENTnBh9UeuL3QhkMoQ-VnM"
)
result = redact_sensitive_text(text)
assert "Token:" in result
# Payload and signature must not survive
assert "eyJpc3Mi" not in result
assert "Gxgv0rru" not in result
def test_2part_jwt(self):
text = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0"
result = redact_sensitive_text(text)
assert "eyJzdWIi" not in result
def test_standalone_jwt_header(self):
text = "leaked header: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9 here"
result = redact_sensitive_text(text)
assert "IkpXVCJ9" not in result
assert "leaked header:" in result
def test_jwt_with_base64_padding(self):
text = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0=.abc123def456ghij"
result = redact_sensitive_text(text)
assert "abc123def456" not in result
def test_short_eyj_not_matched(self):
"""eyJ followed by fewer than 10 base64 chars should not match."""
text = "eyJust a normal word"
assert redact_sensitive_text(text) == text
def test_jwt_preserves_surrounding_text(self):
text = "before eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0 after"
result = redact_sensitive_text(text)
assert result.startswith("before ")
assert result.endswith(" after")
def test_home_assistant_jwt_in_memory(self):
"""Real-world pattern: HA token stored in agent memory block."""
text = (
"Home Assistant API Token: "
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"
".eyJpc3MiOiJhYmNkZWYiLCJleHAiOjE3NzQ5NTcxMDN9"
".Gxgv0rru-_kS-I_60EJ7CENTnBh9UeuL3QhkMoQ-VnM"
)
result = redact_sensitive_text(text)
assert "Home Assistant API Token:" in result
assert "Gxgv0rru" not in result
assert "..." in result
class TestDiscordMentions:
"""Discord snowflake IDs in <@ID> or <@!ID> format."""
def test_normal_mention(self):
result = redact_sensitive_text("Hello <@222589316709220353>")
assert "222589316709220353" not in result
assert "<@***>" in result
def test_nickname_mention(self):
result = redact_sensitive_text("Ping <@!1331549159177846844>")
assert "1331549159177846844" not in result
assert "<@!***>" in result
def test_multiple_mentions(self):
text = "<@111111111111111111> and <@222222222222222222>"
result = redact_sensitive_text(text)
assert "111111111111111111" not in result
assert "222222222222222222" not in result
def test_short_id_not_matched(self):
"""IDs shorter than 17 digits are not Discord snowflakes."""
text = "<@12345>"
assert redact_sensitive_text(text) == text
def test_slack_mention_not_matched(self):
"""Slack mentions use letters, not pure digits."""
text = "<@U024BE7LH>"
assert redact_sensitive_text(text) == text
def test_preserves_surrounding_text(self):
text = "User <@222589316709220353> said hello"
result = redact_sensitive_text(text)
assert result.startswith("User ")
assert result.endswith(" said hello")

View File

@@ -1,61 +0,0 @@
from agent.smart_model_routing import choose_cheap_model_route
_BASE_CONFIG = {
"enabled": True,
"cheap_model": {
"provider": "openrouter",
"model": "google/gemini-2.5-flash",
},
}
def test_returns_none_when_disabled():
cfg = {**_BASE_CONFIG, "enabled": False}
assert choose_cheap_model_route("what time is it in tokyo?", cfg) is None
def test_routes_short_simple_prompt():
result = choose_cheap_model_route("what time is it in tokyo?", _BASE_CONFIG)
assert result is not None
assert result["provider"] == "openrouter"
assert result["model"] == "google/gemini-2.5-flash"
assert result["routing_reason"] == "simple_turn"
def test_skips_long_prompt():
prompt = "please summarize this carefully " * 20
assert choose_cheap_model_route(prompt, _BASE_CONFIG) is None
def test_skips_code_like_prompt():
prompt = "debug this traceback: ```python\nraise ValueError('bad')\n```"
assert choose_cheap_model_route(prompt, _BASE_CONFIG) is None
def test_skips_tool_heavy_prompt_keywords():
prompt = "implement a patch for this docker error"
assert choose_cheap_model_route(prompt, _BASE_CONFIG) is None
def test_resolve_turn_route_falls_back_to_primary_when_route_runtime_cannot_be_resolved(monkeypatch):
from agent.smart_model_routing import resolve_turn_route
monkeypatch.setattr(
"hermes_cli.runtime_provider.resolve_runtime_provider",
lambda **kwargs: (_ for _ in ()).throw(RuntimeError("bad route")),
)
result = resolve_turn_route(
"what time is it in tokyo?",
_BASE_CONFIG,
{
"model": "anthropic/claude-sonnet-4",
"provider": "openrouter",
"base_url": "https://openrouter.ai/api/v1",
"api_mode": "chat_completions",
"api_key": "sk-primary",
},
)
assert result["model"] == "anthropic/claude-sonnet-4"
assert result["runtime"]["provider"] == "openrouter"
assert result["label"] is None

View File

@@ -79,7 +79,7 @@ class TestBuildChildProgressCallback:
parent._delegate_spinner = None
parent.tool_progress_callback = None
cb = _build_child_progress_callback(0, parent)
cb = _build_child_progress_callback(0, "test goal", parent)
assert cb is None
def test_cli_spinner_tool_event(self):
@@ -93,7 +93,7 @@ class TestBuildChildProgressCallback:
parent._delegate_spinner = spinner
parent.tool_progress_callback = None
cb = _build_child_progress_callback(0, parent)
cb = _build_child_progress_callback(0, "test goal", parent)
assert cb is not None
cb("tool.started", "web_search", "quantum computing", {})
@@ -113,7 +113,7 @@ class TestBuildChildProgressCallback:
parent._delegate_spinner = spinner
parent.tool_progress_callback = None
cb = _build_child_progress_callback(0, parent)
cb = _build_child_progress_callback(0, "test goal", parent)
cb("_thinking", "I'll search for papers first")
output = buf.getvalue()
@@ -121,54 +121,64 @@ class TestBuildChildProgressCallback:
assert "search for papers" in output
def test_gateway_batched_progress(self):
"""Gateway path should batch tool calls and flush at BATCH_SIZE."""
"""Gateway path: each tool.started relays a subagent.tool event, and a
subagent.progress summary fires once BATCH_SIZE tools accumulate."""
parent = MagicMock()
parent._delegate_spinner = None
parent_cb = MagicMock()
parent.tool_progress_callback = parent_cb
cb = _build_child_progress_callback(0, parent)
# Send 4 tool calls shouldn't flush yet (BATCH_SIZE = 5)
cb = _build_child_progress_callback(0, "test goal", parent)
# Each tool.started relays a subagent.tool event immediately (per-tool relay).
for i in range(4):
cb("tool.started", f"tool_{i}", f"arg_{i}", {})
parent_cb.assert_not_called()
# 5th call should trigger flush
cb("tool.started", "tool_4", "arg_4", {})
parent_cb.assert_called_once()
call_args = parent_cb.call_args
assert "tool_0" in call_args[0][1]
assert "tool_4" in call_args[0][1]
# 4 per-tool relays so far, no batch summary yet (BATCH_SIZE=5)
events = [c.args[0] for c in parent_cb.call_args_list]
assert events == ["subagent.tool"] * 4
def test_thinking_not_relayed_to_gateway(self):
"""Thinking events should NOT be sent to gateway (too noisy)."""
# 5th call triggers another per-tool relay PLUS the batch-size summary
cb("tool.started", "tool_4", "arg_4", {})
events = [c.args[0] for c in parent_cb.call_args_list]
assert events == ["subagent.tool"] * 5 + ["subagent.progress"]
summary_call = parent_cb.call_args_list[-1]
summary_text = summary_call.kwargs.get("preview") or summary_call.args[2]
assert "tool_0" in summary_text
assert "tool_4" in summary_text
def test_thinking_relayed_to_gateway(self):
"""Thinking events are relayed as subagent.thinking events."""
parent = MagicMock()
parent._delegate_spinner = None
parent_cb = MagicMock()
parent.tool_progress_callback = parent_cb
cb = _build_child_progress_callback(0, parent)
cb = _build_child_progress_callback(0, "test goal", parent)
cb("_thinking", "some reasoning text")
parent_cb.assert_not_called()
parent_cb.assert_called_once()
assert parent_cb.call_args.args[0] == "subagent.thinking"
assert parent_cb.call_args.args[2] == "some reasoning text"
def test_parallel_callbacks_independent(self):
"""Each child's callback should have independent batch state."""
"""Each child's callback batches tool names independently."""
parent = MagicMock()
parent._delegate_spinner = None
parent_cb = MagicMock()
parent.tool_progress_callback = parent_cb
cb0 = _build_child_progress_callback(0, parent)
cb1 = _build_child_progress_callback(1, parent)
# Send 3 calls to each — neither should flush (batch size = 5)
cb0 = _build_child_progress_callback(0, "goal a", parent)
cb1 = _build_child_progress_callback(1, "goal b", parent)
# 3 tool.started per child = 6 per-tool relays; neither should hit
# the batch-size summary (batch size = 5, counted per-child).
for i in range(3):
cb0(f"tool_{i}")
cb1(f"other_{i}")
parent_cb.assert_not_called()
cb0("tool.started", f"tool_{i}", f"a_{i}", {})
cb1("tool.started", f"other_{i}", f"b_{i}", {})
events = [c.args[0] for c in parent_cb.call_args_list]
assert events.count("subagent.tool") == 6
assert "subagent.progress" not in events
def test_task_index_prefix_in_batch_mode(self):
"""Batch mode (task_count > 1) should show 1-indexed prefix for all tasks."""
@@ -182,7 +192,7 @@ class TestBuildChildProgressCallback:
parent.tool_progress_callback = None
# task_index=0 in a batch of 3 → prefix "[1]"
cb0 = _build_child_progress_callback(0, parent, task_count=3)
cb0 = _build_child_progress_callback(0, "test goal", parent, task_count=3)
cb0("web_search", "test")
output = buf.getvalue()
assert "[1]" in output
@@ -190,7 +200,7 @@ class TestBuildChildProgressCallback:
# task_index=2 in a batch of 3 → prefix "[3]"
buf.truncate(0)
buf.seek(0)
cb2 = _build_child_progress_callback(2, parent, task_count=3)
cb2 = _build_child_progress_callback(2, "test goal", parent, task_count=3)
cb2("web_search", "test")
output = buf.getvalue()
assert "[3]" in output
@@ -206,7 +216,7 @@ class TestBuildChildProgressCallback:
parent._delegate_spinner = spinner
parent.tool_progress_callback = None
cb = _build_child_progress_callback(0, parent, task_count=1)
cb = _build_child_progress_callback(0, "test goal", parent, task_count=1)
cb("tool.started", "web_search", "test", {})
output = buf.getvalue()
@@ -321,26 +331,31 @@ class TestBatchFlush:
"""Tests for gateway batch flush on subagent completion."""
def test_flush_sends_remaining_batch(self):
"""_flush should send remaining tool names to gateway."""
"""_flush should send a final subagent.progress summary of any unsent
tool names in the batch (less than BATCH_SIZE)."""
parent = MagicMock()
parent._delegate_spinner = None
parent_cb = MagicMock()
parent.tool_progress_callback = parent_cb
cb = _build_child_progress_callback(0, parent)
cb = _build_child_progress_callback(0, "test goal", parent)
# Send 3 tools (below batch size of 5)
# Send 3 tools (below batch size of 5) — each relays subagent.tool
cb("tool.started", "web_search", "query1", {})
cb("tool.started", "read_file", "file.txt", {})
cb("tool.started", "write_file", "out.txt", {})
parent_cb.assert_not_called()
events = [c.args[0] for c in parent_cb.call_args_list]
assert events == ["subagent.tool"] * 3 # per-tool relays so far
assert "subagent.progress" not in events # no batch-size summary yet
# Flush should send the remaining 3
# Flush should send the remaining 3 as a summary
cb._flush()
parent_cb.assert_called_once()
summary = parent_cb.call_args[0][1]
assert "web_search" in summary
assert "write_file" in summary
events = [c.args[0] for c in parent_cb.call_args_list]
assert events[-1] == "subagent.progress"
summary_call = parent_cb.call_args_list[-1]
summary_text = summary_call.kwargs.get("preview") or summary_call.args[2]
assert "web_search" in summary_text
assert "write_file" in summary_text
def test_flush_noop_when_batch_empty(self):
"""_flush should not send anything when batch is empty."""
@@ -349,7 +364,7 @@ class TestBatchFlush:
parent_cb = MagicMock()
parent.tool_progress_callback = parent_cb
cb = _build_child_progress_callback(0, parent)
cb = _build_child_progress_callback(0, "test goal", parent)
cb._flush()
parent_cb.assert_not_called()
@@ -364,7 +379,7 @@ class TestBatchFlush:
parent._delegate_spinner = spinner
parent.tool_progress_callback = None
cb = _build_child_progress_callback(0, parent)
cb = _build_child_progress_callback(0, "test goal", parent)
cb("tool.started", "web_search", "test", {})
cb._flush() # Should not crash

View File

@@ -0,0 +1,40 @@
"""Test that call_llm vision path passes resolved provider args, not raw ones."""
from unittest.mock import patch, MagicMock
def test_vision_call_uses_resolved_provider_args():
"""Resolved provider/model/key/url from config must reach resolve_vision_provider_client."""
from agent.auxiliary_client import call_llm
fake_client = MagicMock()
fake_client.chat.completions.create.return_value = MagicMock(
choices=[MagicMock(message=MagicMock(content="description"))],
usage=MagicMock(prompt_tokens=10, completion_tokens=5),
)
with (
patch(
"agent.auxiliary_client._resolve_task_provider_model",
return_value=("my-resolved-provider", "my-resolved-model", "http://resolved", "resolved-key", "chat_completions"),
),
patch(
"agent.auxiliary_client.resolve_vision_provider_client",
return_value=("my-resolved-provider", fake_client, "my-resolved-model"),
) as mock_vision,
):
call_llm(
"vision",
provider="raw-provider",
model="raw-model",
base_url="http://raw",
api_key="raw-key",
messages=[{"role": "user", "content": "describe this"}],
)
# The resolved values must be passed, not the raw call_llm arguments
call_args = mock_vision.call_args
assert call_args.kwargs["provider"] == "my-resolved-provider"
assert call_args.kwargs["model"] == "my-resolved-model"
assert call_args.kwargs["base_url"] == "http://resolved"
assert call_args.kwargs["api_key"] == "resolved-key"