fix(anthropic): strip styled Claude model ids (#651)

## Description

Fixes #626 by normalizing Anthropic/Claude model ids that contain ANSI
escape sequences or dangling style suffixes before provider lookups and
upstream forwarding. The branch has been updated onto current `main` and
the proxy handler conflicts have been resolved.

## Type of Change

- [x] Bug fix
- [ ] New feature
- [ ] Documentation
- [ ] Refactor
- [x] Tests only

## Changes Made

- Normalize Anthropic model ids before context/pricing lookup.
- Sanitize Anthropic `/v1/models` metadata and styled `/v1/models/{id}`
passthrough paths.
- Sanitize `/v1/messages` request body model ids before upstream
forwarding.
- Resolved current-main conflicts while preserving newer
`model_override` and streaming passthrough behavior.

## Testing

- [x] Unit tests
- [x] Route/proxy tests
- [x] Lint/static checks
- [ ] Manual testing

### Test Output

```text
UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --with pytest --with fastapi --with httpx --with uvicorn --with h2 python -m pytest tests/test_providers/test_anthropic.py tests/test_provider_proxy_routes.py::test_anthropic_model_metadata_strips_ansi_model_ids tests/test_provider_proxy_routes.py::test_anthropic_model_detail_path_strips_ansi_model_id tests/test_provider_proxy_routes.py::test_anthropic_messages_strips_ansi_model_id_before_upstream -q
17 passed, 2 warnings in 39.91s

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --with ruff ruff check headroom/providers/anthropic.py headroom/proxy/handlers/openai.py headroom/proxy/handlers/anthropic.py tests/test_providers/test_anthropic.py tests/test_provider_proxy_routes.py
All checks passed!
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.13.3, focused local worktree for PR
#651 after merging current `upstream/main`.
- Exact command / steps: Merged current main, resolved conflicts in
Anthropic/OpenAI proxy handlers, ran the PR's targeted provider/proxy
tests and ruff checks.
- Observed result: Styled Anthropic model metadata, model-detail path,
and messages upstream sanitization tests pass; ruff reports no issues.
- Not tested: Full repository mypy/pre-commit; existing unrelated
Windows `fcntl` typing errors block full hook execution locally.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review


<!-- headroom-maintainer-template-completion:start -->

## Description

This PR prepares `fix(anthropic): strip styled Claude model ids` for
review by documenting the intended change, validation evidence, and
remaining merge-readiness context.

Linked issues: #626

## Type of Change

- [x] Bug fix
- [ ] New feature
- [ ] Documentation
- [ ] Refactor
- [ ] Tests only

## Changes Made

- Commit: fix(anthropic): normalize styled model ids
- Commit: fix(proxy): strip styled Anthropic model ids
- Commit: fix: format anthropic model sanitization
- Commit: Merge remote-tracking branch 'upstream/main' into
review/pr-651
- Touches `headroom/cache/dynamic_detector.py`
- Touches `headroom/providers/anthropic.py`
- Touches `headroom/proxy/handlers/anthropic.py`
- Touches `headroom/proxy/handlers/openai.py`
- Touches `tests/test_provider_proxy_routes.py`
- Touches `tests/test_providers/test_anthropic.py`

## Testing

- [x] GitHub checks reviewed
- [x] Metadata/template validation
- [ ] Local functional testing

### Test Output

```text
gh pr view 651 --repo chopratejas/headroom --json statusCheckRollup
- PR Governance / template: FAILURE
- CI / changes: SUCCESS
- Init E2E / docker-init-e2e: SUCCESS
- Wrap E2E / docker-wrap-e2e: SUCCESS
- Wrap Native E2E / wrap-native (ubuntu-latest): SUCCESS
- Wrap Native E2E / wrap-native (macos-latest): SUCCESS
- CI / commitlint: SUCCESS
- PR Governance / label: SUCCESS
- CI / lint: SUCCESS
- CI / build-wheel: SUCCESS
- CI / prefetch-model: SUCCESS
- CI / build: SUCCESS
```

## Real Behavior Proof

- Environment: GitHub PR metadata and checks for `chopratejas/headroom`
PR #651.
- Exact command / steps: Reviewed PR title, commits, changed files,
linked issues, labels, and check rollup; appended this maintainer
template completion block without replacing the author's original
description.
- Observed result: PR body now contains all required governance
sections, checked readiness fields, and a non-placeholder validation
evidence block.
- Not tested: This pass updated PR metadata only; code validation
remains represented by the linked GitHub checks and any author-provided
evidence above.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

<!-- headroom-maintainer-template-completion:end -->

---------

Co-authored-by: Tejas Chopra <chopratejas@gmail.com>
Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
This commit is contained in:
Kumario 2026-06-13 13:46:21 -05:00 committed by GitHub
parent 9b7b436b04
commit 0c5c89d05c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 224 additions and 14 deletions

View file

@ -770,6 +770,8 @@ class SemanticDetector:
# Encode all sentences
if not sentences:
return [], None
if self._model is None or self._exemplar_embeddings is None:
return [], self._load_error or "semantic detector is not initialized"
try:
import numpy as np
@ -777,7 +779,7 @@ class SemanticDetector:
return [], "numpy not installed. Install with: pip install numpy"
sentence_texts = [s[0] for s in sentences]
sentence_embeddings = self._model.encode( # type: ignore[union-attr]
sentence_embeddings = self._model.encode(
sentence_texts,
convert_to_numpy=True,
)

View file

@ -19,6 +19,7 @@ import importlib.util
import json
import logging
import os
import re
import warnings
from typing import Any, cast
@ -51,17 +52,39 @@ logger = logging.getLogger(__name__)
# Warning flags
_FALLBACK_WARNING_SHOWN = False
_UNKNOWN_MODEL_WARNINGS: set[str] = set()
_ANSI_ESCAPE_RE = re.compile(r"\x1b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
_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
def sanitize_anthropic_model_metadata(value: Any) -> Any:
"""Strip model-id styling artifacts from Anthropic model metadata payloads."""
if isinstance(value, list):
return [sanitize_anthropic_model_metadata(item) for item in value]
if not isinstance(value, dict):
return value
cleaned: dict[str, Any] = {}
for key, item in value.items():
if key in {"id", "model"} and isinstance(item, str):
cleaned[key] = sanitize_anthropic_model_id(item)
else:
cleaned[key] = sanitize_anthropic_model_metadata(item)
return cleaned
# Anthropic model context limits
# All Claude 3+ models have 200K context
ANTHROPIC_CONTEXT_LIMITS: dict[str, int] = {
# Claude 4.7 (Opus 4.7) - 1M context. Claude Code sends the model
# name with a `[1m]` suffix to select the 1M tier; both forms are
# registered explicitly because the lookup chain does not strip the
# tier suffix and LiteLLM does not key on it either.
# Claude 4.7 (Opus 4.7) - 1M context
"claude-opus-4-7": 1000000,
"claude-opus-4-7[1m]": 1000000,
# Claude 4.6 (Opus 4.6) - 1M context
"claude-opus-4-6": 1000000,
# Claude 4.5 (Opus 4.5)
@ -89,10 +112,8 @@ ANTHROPIC_CONTEXT_LIMITS: dict[str, int] = {
# NOTE: These are ESTIMATES. Always verify against actual Anthropic billing.
# Last updated: 2025-01-14
ANTHROPIC_PRICING: dict[str, dict[str, float]] = {
# Claude 4.7 (Opus tier pricing) — registered for both the bare
# name and the `[1m]` tier-suffixed form Claude Code sends.
# Claude 4.7 (Opus tier pricing)
"claude-opus-4-7": {"input": 15.00, "output": 75.00, "cached_input": 1.50},
"claude-opus-4-7[1m]": {"input": 15.00, "output": 75.00, "cached_input": 1.50},
# Claude 4.6 (Opus tier pricing)
"claude-opus-4-6": {"input": 15.00, "output": 75.00, "cached_input": 1.50},
# Claude 4.5 (Opus tier pricing)
@ -494,6 +515,7 @@ class AnthropicProvider(Provider):
If a client was provided to the provider, uses the Token Count API.
Otherwise falls back to tiktoken approximation.
"""
model = sanitize_anthropic_model_id(model)
if model not in self._token_counters:
self._token_counters[model] = AnthropicTokenCounter(
model=model,
@ -516,6 +538,7 @@ class AnthropicProvider(Provider):
Never raises an exception - uses sensible defaults for unknown models.
"""
model = sanitize_anthropic_model_id(model)
# Check explicit and loaded limits
if model in self._context_limits:
return self._context_limits[model]
@ -577,6 +600,7 @@ class AnthropicProvider(Provider):
def supports_model(self, model: str) -> bool:
"""Check if this provider supports the given model."""
model = sanitize_anthropic_model_id(model)
if model in self._context_limits:
return True
# Check prefix matches - support all Claude models
@ -593,6 +617,7 @@ class AnthropicProvider(Provider):
Tries LiteLLM first for up-to-date pricing, falls back to manual pricing.
"""
model = sanitize_anthropic_model_id(model)
# Try LiteLLM first for cost estimation
litellm, litellm_get_model_info = _get_litellm_clients()
if litellm is not None:
@ -646,6 +671,7 @@ class AnthropicProvider(Provider):
def _get_pricing(self, model: str) -> dict[str, float] | None:
"""Get pricing for a model with fallback logic."""
model = sanitize_anthropic_model_id(model)
# Direct match
if model in self._pricing:
return self._pricing[model]

View file

@ -422,6 +422,7 @@ class AnthropicHandlerMixin:
from headroom.cache.compression_store import get_compression_store
from headroom.ccr import CCRToolInjector
from headroom.providers.anthropic import sanitize_anthropic_model_id
from headroom.proxy.helpers import (
MAX_MESSAGE_ARRAY_LENGTH,
MAX_REQUEST_BODY_SIZE,
@ -596,7 +597,14 @@ class AnthropicHandlerMixin:
},
},
)
model = body.get("model") or model_override or "unknown"
raw_model = body.get("model") or model_override or "unknown"
model = (
sanitize_anthropic_model_id(raw_model) if isinstance(raw_model, str) else raw_model
)
body_model = body.get("model")
if isinstance(body_model, str) and model != body_model:
body["model"] = model
body_mutation_tracker.mark_mutated("sanitize_model_id")
messages = body.get("messages", [])
pipeline_provider = provider_name
pipeline_path = request.url.path if upstream_base_url else "/v1/messages"

View file

@ -21,7 +21,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import replace
from datetime import datetime
from typing import TYPE_CHECKING, Any
from urllib.parse import urlparse
from urllib.parse import quote, unquote, urlparse
from headroom.proxy.helpers import (
COMPRESSION_TIMEOUT_SECONDS,
@ -477,10 +477,10 @@ def _extract_codex_handshake_headers(upstream: Any) -> list[tuple[str, str]]:
return []
out: list[tuple[str, str]] = []
for name, value in items:
name_str = name.decode("latin-1") if isinstance(name, (bytes, bytearray)) else str(name)
name_str = name.decode("latin-1") if isinstance(name, bytes | bytearray) else str(name)
if name_str.lower().startswith("x-codex-"):
value_str = (
value.decode("latin-1") if isinstance(value, (bytes, bytearray)) else str(value)
value.decode("latin-1") if isinstance(value, bytes | bytearray) else str(value)
)
out.append((name_str, value_str))
return out
@ -6013,6 +6013,13 @@ class OpenAIHandlerMixin:
start_time = time.time()
path = request.url.path
if provider == "anthropic" and endpoint_name == "models" and path.startswith("/v1/models/"):
from headroom.providers.anthropic import sanitize_anthropic_model_id
raw_model_id = path[len("/v1/models/") :]
clean_model_id = sanitize_anthropic_model_id(unquote(raw_model_id))
if clean_model_id != unquote(raw_model_id):
path = "/v1/models/" + quote(clean_model_id, safe="")
url = build_copilot_upstream_url(base_url, path)
# Preserve query string parameters
@ -6079,6 +6086,23 @@ class OpenAIHandlerMixin:
response_headers = dict(response.headers)
response_headers.pop("content-encoding", None)
response_headers.pop("content-length", None) # Length changed after decompression
response_content = response.content
if provider == "anthropic" and endpoint_name == "models":
from headroom.providers.anthropic import sanitize_anthropic_model_metadata
try:
payload = response.json()
sanitized_payload = sanitize_anthropic_model_metadata(payload)
except (TypeError, ValueError):
sanitized_payload = None
if sanitized_payload is not None and sanitized_payload != payload:
response_content = json.dumps(
sanitized_payload,
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")
response_headers["content-type"] = "application/json"
# Passthrough request: forwarded upstream with no transforms.
# Still recorded so dashboards see traffic on the passthrough
@ -6119,7 +6143,7 @@ class OpenAIHandlerMixin:
)
return Response(
content=response.content,
content=response_content,
status_code=response.status_code,
headers=response_headers,
)

View file

@ -1,6 +1,7 @@
from __future__ import annotations
import importlib
import json
from typing import Any
from unittest.mock import patch
@ -694,3 +695,107 @@ def test_v1_models_routes_claude_code_gateway_discovery_to_anthropic() -> None:
("/v1/models", "https://api.anthropic.test", "anthropic"),
("/v1/models/claude-opus-4-8", "https://api.anthropic.test", "anthropic"),
]
def test_anthropic_model_metadata_strips_ansi_model_ids() -> None:
class FakeAsyncClient:
def __init__(self) -> None:
self.calls: list[tuple[str, str]] = []
async def request(self, method, url, **kwargs): # type: ignore[no-untyped-def]
self.calls.append((method, url))
return httpx.Response(
200,
json={
"object": "list",
"data": [
{"id": "claude-opus-4-8\x1b[1m", "object": "model"},
{"id": "claude-sonnet-4-5[1m]", "object": "model"},
],
},
)
async def aclose(self) -> None:
return None
with TestClient(_app()) as client:
fake_http_client = FakeAsyncClient()
client.app.state.proxy.http_client = fake_http_client
response = client.get("/v1/models", headers={"x-api-key": "sk-ant-test"})
assert response.status_code == 200
assert response.json()["data"] == [
{"id": "claude-opus-4-8", "object": "model"},
{"id": "claude-sonnet-4-5", "object": "model"},
]
assert fake_http_client.calls == [("GET", "https://api.anthropic.test/v1/models")]
def test_anthropic_model_detail_path_strips_ansi_model_id() -> None:
class FakeAsyncClient:
def __init__(self) -> None:
self.calls: list[tuple[str, str]] = []
async def request(self, method, url, **kwargs): # type: ignore[no-untyped-def]
self.calls.append((method, url))
return httpx.Response(
200,
json={"id": "claude-opus-4-8\x1b[1m", "object": "model"},
)
async def aclose(self) -> None:
return None
with TestClient(_app()) as client:
fake_http_client = FakeAsyncClient()
client.app.state.proxy.http_client = fake_http_client
response = client.get(
"/v1/models/claude-opus-4-8%1B%5B1m",
headers={"x-api-key": "sk-ant-test"},
)
assert response.status_code == 200
assert response.json()["id"] == "claude-opus-4-8"
assert fake_http_client.calls == [
("GET", "https://api.anthropic.test/v1/models/claude-opus-4-8")
]
def test_anthropic_messages_strips_ansi_model_id_before_upstream() -> None:
class FakeAsyncClient:
def __init__(self) -> None:
self.bodies: list[dict[str, Any]] = []
async def post(self, url, **kwargs): # type: ignore[no-untyped-def]
self.bodies.append(json.loads(kwargs["content"]))
return httpx.Response(
200,
json={
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-opus-4-8",
"content": [],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
async def aclose(self) -> None:
return None
with TestClient(_app()) as client:
fake_http_client = FakeAsyncClient()
client.app.state.proxy.http_client = fake_http_client
response = client.post(
"/v1/messages",
headers={"x-api-key": "sk-ant-test"},
json={
"model": "claude-opus-4-8\x1b[1m",
"max_tokens": 16,
"messages": [{"role": "user", "content": "hello"}],
},
)
assert response.status_code == 200
assert fake_http_client.bodies[0]["model"] == "claude-opus-4-8"

View file

@ -3,6 +3,37 @@
import pytest
class TestAnthropicModelSanitization:
def test_sanitize_model_id_removes_ansi_escape_sequences(self):
from headroom.providers.anthropic import sanitize_anthropic_model_id
assert sanitize_anthropic_model_id("claude-opus-4-8\x1b[1m") == "claude-opus-4-8"
def test_sanitize_model_id_removes_displayed_style_suffix(self):
from headroom.providers.anthropic import sanitize_anthropic_model_id
assert sanitize_anthropic_model_id("claude-opus-4-8[1m]") == "claude-opus-4-8"
def test_sanitize_model_metadata_cleans_nested_model_ids(self):
from headroom.providers.anthropic import sanitize_anthropic_model_metadata
payload = {
"data": [
{"id": "claude-opus-4-8\x1b[1m", "display_name": "Claude Opus 4.8"},
{"id": "claude-sonnet-4-5[1m]"},
],
"model": "claude-opus-4-8[1m]",
}
assert sanitize_anthropic_model_metadata(payload) == {
"data": [
{"id": "claude-opus-4-8", "display_name": "Claude Opus 4.8"},
{"id": "claude-sonnet-4-5"},
],
"model": "claude-opus-4-8",
}
class TestAnthropicTokenCounting:
@pytest.fixture
def anthropic_provider(self):
@ -43,12 +74,21 @@ class TestAnthropicModelLimits:
limit = anthropic_provider.get_context_limit("claude-3-opus-20240229")
assert limit == 200000
def test_get_context_limit_strips_ansi_model_suffix(self, anthropic_provider):
assert anthropic_provider.get_context_limit("claude-opus-4-7[1m]") == 1000000
def test_supports_model_known(self, anthropic_provider):
assert anthropic_provider.supports_model("claude-3-5-sonnet-20241022")
def test_supports_model_prefix(self, anthropic_provider):
assert anthropic_provider.supports_model("claude-3-5-sonnet-latest")
def test_token_counter_cache_uses_sanitized_model_id(self, anthropic_provider):
plain = anthropic_provider.get_token_counter("claude-opus-4-7")
styled = anthropic_provider.get_token_counter("claude-opus-4-7\x1b[1m")
assert styled is plain
class TestAnthropicCostEstimation:
@pytest.fixture
@ -65,3 +105,8 @@ class TestAnthropicCostEstimation:
)
# $3.00 per 1M input
assert cost == pytest.approx(3.00, rel=0.1)
def test_pricing_lookup_strips_ansi_model_suffix(self, anthropic_provider):
assert anthropic_provider._get_pricing("claude-opus-4-7[1m]") == (
anthropic_provider._get_pricing("claude-opus-4-7")
)