diff --git a/headroom/providers/anthropic.py b/headroom/providers/anthropic.py index 56a8d0790..107fc53c5 100644 --- a/headroom/providers/anthropic.py +++ b/headroom/providers/anthropic.py @@ -59,9 +59,7 @@ _DANGLING_ANSI_STYLE_SUFFIX_RE = re.compile(r"(?:\[[0-9;]*m\])+$") def sanitize_anthropic_model_id(model: str) -> str: """Return an Anthropic model id without terminal styling artifacts.""" cleaned = _ANSI_ESCAPE_RE.sub("", str(model)).strip() - if cleaned.startswith("claude-"): - cleaned = _DANGLING_ANSI_STYLE_SUFFIX_RE.sub("", cleaned) - return cleaned + return _DANGLING_ANSI_STYLE_SUFFIX_RE.sub("", cleaned) def sanitize_anthropic_model_metadata(value: Any) -> Any: diff --git a/tests/test_providers/test_anthropic.py b/tests/test_providers/test_anthropic.py index cbaf2accc..63fa1f73f 100644 --- a/tests/test_providers/test_anthropic.py +++ b/tests/test_providers/test_anthropic.py @@ -13,6 +13,7 @@ class TestAnthropicModelSanitization: from headroom.providers.anthropic import sanitize_anthropic_model_id assert sanitize_anthropic_model_id("claude-opus-4-8[1m]") == "claude-opus-4-8" + assert sanitize_anthropic_model_id("glm-5.2[1m]") == "glm-5.2" def test_sanitize_model_metadata_cleans_nested_model_ids(self): from headroom.providers.anthropic import sanitize_anthropic_model_metadata diff --git a/tests/test_proxy_anthropic_model_sanitization.py b/tests/test_proxy_anthropic_model_sanitization.py new file mode 100644 index 000000000..80880cab6 --- /dev/null +++ b/tests/test_proxy_anthropic_model_sanitization.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from typing import Any + +import httpx +import pytest + +pytest.importorskip("fastapi") + +from fastapi.testclient import TestClient + +from headroom.proxy.server import ProxyConfig, create_app + + +def test_anthropic_messages_strips_local_1m_model_suffix_before_forwarding() -> None: + config = ProxyConfig( + optimize=False, + cache_enabled=False, + rate_limit_enabled=False, + cost_tracking_enabled=False, + log_requests=False, + ccr_inject_tool=False, + ccr_handle_responses=False, + ccr_context_tracking=False, + image_optimize=False, + ) + app = create_app(config) + client = TestClient(app) + + captured: dict[str, Any] = {} + + async def _fake_retry( + method: str, # noqa: ARG001 + url: str, # noqa: ARG001 + headers: dict[str, str], # noqa: ARG001 + body: dict[str, Any], + body_mutated: bool, + mutation_reasons: list[str], + **kwargs: Any, + ) -> httpx.Response: + captured["body"] = dict(body) + captured["body_mutated"] = body_mutated + captured["mutation_reasons"] = list(mutation_reasons) + return httpx.Response( + 200, + json={ + "id": "msg_glm_1m", + "type": "message", + "role": "assistant", + "model": body["model"], + "content": [{"type": "text", "text": "ok"}], + "usage": { + "input_tokens": 10, + "output_tokens": 1, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + }, + }, + ) + + app.state.proxy._retry_request = _fake_retry # type: ignore[assignment] + + response = client.post( + "/v1/messages", + headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"}, + json={ + "model": "glm-5.2[1m]", + "max_tokens": 10, + "stream": False, + "messages": [{"role": "user", "content": "2+2"}], + }, + ) + + assert response.status_code == 200, response.text + assert captured["body"]["model"] == "glm-5.2" + assert captured["body_mutated"] is True + assert captured["mutation_reasons"] == ["sanitize_model_id"]