"""Tests for agent/anthropic_adapter.py — Anthropic Messages API adapter.""" import json import time from types import SimpleNamespace from unittest.mock import patch, MagicMock import pytest from agent.prompt_caching import apply_anthropic_cache_control from agent.anthropic_adapter import ( _is_oauth_token, _refresh_oauth_token, _write_claude_code_credentials, build_anthropic_client, build_anthropic_kwargs, convert_messages_to_anthropic, convert_tools_to_anthropic, get_anthropic_token_source, is_claude_code_token_valid, normalize_anthropic_response, normalize_model_name, read_claude_code_credentials, resolve_anthropic_token, run_oauth_setup_token, ) # --------------------------------------------------------------------------- # Auth helpers # --------------------------------------------------------------------------- class TestIsOAuthToken: def test_setup_token(self): assert _is_oauth_token("sk-ant-oat01-abcdef1234567890") is True def test_api_key(self): assert _is_oauth_token("sk-ant-api03-abcdef1234567890") is False def test_managed_key(self): # Managed keys from ~/.claude.json are NOT regular API keys assert _is_oauth_token("ou1R1z-ft0A-bDeZ9wAA") is True def test_jwt_token(self): # JWTs from OAuth flow assert _is_oauth_token("eyJhbGciOiJSUzI1NiJ9.test") is True def test_empty(self): assert _is_oauth_token("") is False class TestBuildAnthropicClient: def test_setup_token_uses_auth_token(self): with patch("agent.anthropic_adapter._anthropic_sdk") as mock_sdk: build_anthropic_client("sk-ant-oat01-" + "x" * 60) kwargs = mock_sdk.Anthropic.call_args[1] assert "auth_token" in kwargs betas = kwargs["default_headers"]["anthropic-beta"] assert "oauth-2025-04-20" in betas assert "claude-code-20250219" in betas assert "interleaved-thinking-2025-05-14" in betas assert "fine-grained-tool-streaming-2025-05-14" in betas assert "api_key" not in kwargs def test_api_key_uses_api_key(self): with patch("agent.anthropic_adapter._anthropic_sdk") as mock_sdk: build_anthropic_client("sk-ant-api03-something") kwargs = mock_sdk.Anthropic.call_args[1] assert kwargs["api_key"] == "sk-ant-api03-something" assert "auth_token" not in kwargs # API key auth should still get common betas betas = kwargs["default_headers"]["anthropic-beta"] assert "interleaved-thinking-2025-05-14" in betas assert "oauth-2025-04-20" not in betas # OAuth-only beta NOT present assert "claude-code-20250219" not in betas # OAuth-only beta NOT present def test_custom_base_url(self): with patch("agent.anthropic_adapter._anthropic_sdk") as mock_sdk: build_anthropic_client("sk-ant-api03-x", base_url="https://custom.api.com") kwargs = mock_sdk.Anthropic.call_args[1] assert kwargs["base_url"] == "https://custom.api.com" class TestReadClaudeCodeCredentials: def test_reads_valid_credentials(self, tmp_path, monkeypatch): cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({ "claudeAiOauth": { "accessToken": "sk-ant-oat01-token", "refreshToken": "sk-ant-oat01-refresh", "expiresAt": int(time.time() * 1000) + 3600_000, } })) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) creds = read_claude_code_credentials() assert creds is not None assert creds["accessToken"] == "sk-ant-oat01-token" assert creds["refreshToken"] == "sk-ant-oat01-refresh" assert creds["source"] == "claude_code_credentials_file" def test_ignores_primary_api_key_for_native_anthropic_resolution(self, tmp_path, monkeypatch): claude_json = tmp_path / ".claude.json" claude_json.write_text(json.dumps({"primaryApiKey": "sk-ant-api03-primary"})) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) creds = read_claude_code_credentials() assert creds is None def test_returns_none_for_missing_file(self, tmp_path, monkeypatch): monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert read_claude_code_credentials() is None def test_returns_none_for_missing_oauth_key(self, tmp_path, monkeypatch): cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({"someOtherKey": {}})) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert read_claude_code_credentials() is None def test_returns_none_for_empty_access_token(self, tmp_path, monkeypatch): cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({ "claudeAiOauth": {"accessToken": "", "refreshToken": "x"} })) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert read_claude_code_credentials() is None class TestIsClaudeCodeTokenValid: def test_valid_token(self): creds = {"accessToken": "tok", "expiresAt": int(time.time() * 1000) + 3600_000} assert is_claude_code_token_valid(creds) is True def test_expired_token(self): creds = {"accessToken": "tok", "expiresAt": int(time.time() * 1000) - 3600_000} assert is_claude_code_token_valid(creds) is False def test_no_expiry_but_has_token(self): creds = {"accessToken": "tok", "expiresAt": 0} assert is_claude_code_token_valid(creds) is True class TestResolveAnthropicToken: def test_prefers_oauth_token_over_api_key(self, monkeypatch, tmp_path): monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-api03-mykey") monkeypatch.setenv("ANTHROPIC_TOKEN", "sk-ant-oat01-mytoken") monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() == "sk-ant-oat01-mytoken" def test_reports_claude_json_primary_key_source(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) (tmp_path / ".claude.json").write_text(json.dumps({"primaryApiKey": "sk-ant-api03-primary"})) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert get_anthropic_token_source("sk-ant-api03-primary") == "claude_json_primary_api_key" def test_does_not_resolve_primary_api_key_as_native_anthropic_token(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) (tmp_path / ".claude.json").write_text(json.dumps({"primaryApiKey": "sk-ant-api03-primary"})) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() is None def test_falls_back_to_api_key_when_no_oauth_sources_exist(self, monkeypatch, tmp_path): monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-api03-mykey") monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() == "sk-ant-api03-mykey" def test_falls_back_to_token(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.setenv("ANTHROPIC_TOKEN", "sk-ant-oat01-mytoken") monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() == "sk-ant-oat01-mytoken" def test_returns_none_with_no_creds(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() is None def test_falls_back_to_claude_code_oauth_token(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "sk-ant-oat01-test-token") monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() == "sk-ant-oat01-test-token" def test_falls_back_to_claude_code_credentials(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({ "claudeAiOauth": { "accessToken": "cc-auto-token", "refreshToken": "refresh", "expiresAt": int(time.time() * 1000) + 3600_000, } })) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() == "cc-auto-token" def test_prefers_refreshable_claude_code_credentials_over_static_anthropic_token(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.setenv("ANTHROPIC_TOKEN", "sk-ant-oat01-static-token") monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({ "claudeAiOauth": { "accessToken": "cc-auto-token", "refreshToken": "refresh-token", "expiresAt": int(time.time() * 1000) + 3600_000, } })) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() == "cc-auto-token" def test_keeps_static_anthropic_token_when_only_non_refreshable_claude_key_exists(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.setenv("ANTHROPIC_TOKEN", "sk-ant-oat01-static-token") monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) claude_json = tmp_path / ".claude.json" claude_json.write_text(json.dumps({"primaryApiKey": "sk-ant-api03-managed-key"})) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) assert resolve_anthropic_token() == "sk-ant-oat01-static-token" class TestRefreshOauthToken: def test_returns_none_without_refresh_token(self): creds = {"accessToken": "expired", "refreshToken": "", "expiresAt": 0} assert _refresh_oauth_token(creds) is None def test_successful_refresh(self, tmp_path, monkeypatch): monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) creds = { "accessToken": "old-token", "refreshToken": "refresh-123", "expiresAt": int(time.time() * 1000) - 3600_000, } mock_response = json.dumps({ "access_token": "new-token-abc", "refresh_token": "new-refresh-456", "expires_in": 7200, }).encode() with patch("urllib.request.urlopen") as mock_urlopen: mock_ctx = MagicMock() mock_ctx.__enter__ = MagicMock(return_value=MagicMock( read=MagicMock(return_value=mock_response) )) mock_ctx.__exit__ = MagicMock(return_value=False) mock_urlopen.return_value = mock_ctx result = _refresh_oauth_token(creds) assert result == "new-token-abc" # Verify credentials were written back cred_file = tmp_path / ".claude" / ".credentials.json" assert cred_file.exists() written = json.loads(cred_file.read_text()) assert written["claudeAiOauth"]["accessToken"] == "new-token-abc" assert written["claudeAiOauth"]["refreshToken"] == "new-refresh-456" def test_failed_refresh_returns_none(self): creds = { "accessToken": "old", "refreshToken": "refresh-123", "expiresAt": 0, } with patch("urllib.request.urlopen", side_effect=Exception("network error")): assert _refresh_oauth_token(creds) is None class TestWriteClaudeCodeCredentials: def test_writes_new_file(self, tmp_path, monkeypatch): monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) _write_claude_code_credentials("tok", "ref", 12345) cred_file = tmp_path / ".claude" / ".credentials.json" assert cred_file.exists() data = json.loads(cred_file.read_text()) assert data["claudeAiOauth"]["accessToken"] == "tok" assert data["claudeAiOauth"]["refreshToken"] == "ref" assert data["claudeAiOauth"]["expiresAt"] == 12345 def test_preserves_existing_fields(self, tmp_path, monkeypatch): monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) cred_dir = tmp_path / ".claude" cred_dir.mkdir() cred_file = cred_dir / ".credentials.json" cred_file.write_text(json.dumps({"otherField": "keep-me"})) _write_claude_code_credentials("new-tok", "new-ref", 99999) data = json.loads(cred_file.read_text()) assert data["otherField"] == "keep-me" assert data["claudeAiOauth"]["accessToken"] == "new-tok" class TestResolveWithRefresh: def test_auto_refresh_on_expired_creds(self, monkeypatch, tmp_path): """When cred file has expired token + refresh token, auto-refresh is attempted.""" monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) # Set up expired creds with a refresh token cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({ "claudeAiOauth": { "accessToken": "expired-tok", "refreshToken": "valid-refresh", "expiresAt": int(time.time() * 1000) - 3600_000, } })) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) # Mock refresh to succeed with patch("agent.anthropic_adapter._refresh_oauth_token", return_value="refreshed-token"): result = resolve_anthropic_token() assert result == "refreshed-token" def test_static_env_oauth_token_does_not_block_refreshable_claude_creds(self, monkeypatch, tmp_path): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.setenv("ANTHROPIC_TOKEN", "sk-ant-oat01-expired-env-token") monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({ "claudeAiOauth": { "accessToken": "expired-claude-creds-token", "refreshToken": "valid-refresh", "expiresAt": int(time.time() * 1000) - 3600_000, } })) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) with patch("agent.anthropic_adapter._refresh_oauth_token", return_value="refreshed-token"): result = resolve_anthropic_token() assert result == "refreshed-token" class TestRunOauthSetupToken: def test_raises_when_claude_not_installed(self, monkeypatch): monkeypatch.setattr("shutil.which", lambda _: None) with pytest.raises(FileNotFoundError, match="claude.*CLI.*not installed"): run_oauth_setup_token() def test_returns_token_from_credential_files(self, monkeypatch, tmp_path): """After subprocess completes, reads credentials from Claude Code files.""" monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude") monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) # Pre-create credential files that will be found after subprocess cred_file = tmp_path / ".claude" / ".credentials.json" cred_file.parent.mkdir(parents=True) cred_file.write_text(json.dumps({ "claudeAiOauth": { "accessToken": "from-cred-file", "refreshToken": "refresh", "expiresAt": int(time.time() * 1000) + 3600_000, } })) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) token = run_oauth_setup_token() assert token == "from-cred-file" mock_run.assert_called_once() def test_returns_token_from_env_var(self, monkeypatch, tmp_path): """Falls back to CLAUDE_CODE_OAUTH_TOKEN env var when no cred files.""" monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude") monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "from-env-var") monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) token = run_oauth_setup_token() assert token == "from-env-var" def test_returns_none_when_no_creds_found(self, monkeypatch, tmp_path): """Returns None when subprocess completes but no credentials are found.""" monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude") monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False) monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path) with patch("subprocess.run") as mock_run: mock_run.return_value = MagicMock(returncode=0) token = run_oauth_setup_token() assert token is None def test_returns_none_on_keyboard_interrupt(self, monkeypatch): """Returns None gracefully when user interrupts the flow.""" monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude") with patch("subprocess.run", side_effect=KeyboardInterrupt): token = run_oauth_setup_token() assert token is None # --------------------------------------------------------------------------- # Model name normalization # --------------------------------------------------------------------------- class TestNormalizeModelName: def test_strips_anthropic_prefix(self): assert normalize_model_name("anthropic/claude-sonnet-4-20250514") == "claude-sonnet-4-20250514" def test_leaves_bare_name(self): assert normalize_model_name("claude-sonnet-4-20250514") == "claude-sonnet-4-20250514" def test_converts_dots_to_hyphens(self): """OpenRouter uses dots (4.6), Anthropic uses hyphens (4-6).""" assert normalize_model_name("anthropic/claude-opus-4.6") == "claude-opus-4-6" assert normalize_model_name("anthropic/claude-sonnet-4.5") == "claude-sonnet-4-5" assert normalize_model_name("claude-opus-4.6") == "claude-opus-4-6" def test_already_hyphenated_unchanged(self): """Names already in Anthropic format should pass through.""" assert normalize_model_name("claude-opus-4-6") == "claude-opus-4-6" assert normalize_model_name("claude-opus-4-5-20251101") == "claude-opus-4-5-20251101" def test_preserve_dots_for_alibaba_dashscope(self): """Alibaba/DashScope use dots in model names (e.g. qwen3.5-plus). Fixes #1739.""" assert normalize_model_name("qwen3.5-plus", preserve_dots=True) == "qwen3.5-plus" assert normalize_model_name("anthropic/qwen3.5-plus", preserve_dots=True) == "qwen3.5-plus" assert normalize_model_name("qwen3.5-flash", preserve_dots=True) == "qwen3.5-flash" # --------------------------------------------------------------------------- # Tool conversion # --------------------------------------------------------------------------- class TestConvertTools: def test_converts_openai_to_anthropic_format(self): tools = [ { "type": "function", "function": { "name": "search", "description": "Search the web", "parameters": { "type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"], }, }, } ] result = convert_tools_to_anthropic(tools) assert len(result) == 1 assert result[0]["name"] == "search" assert result[0]["description"] == "Search the web" assert result[0]["input_schema"]["properties"]["query"]["type"] == "string" def test_empty_tools(self): assert convert_tools_to_anthropic([]) == [] assert convert_tools_to_anthropic(None) == [] # --------------------------------------------------------------------------- # Message conversion # --------------------------------------------------------------------------- class TestConvertMessages: def test_extracts_system_prompt(self): messages = [ {"role": "system", "content": "You are helpful."}, {"role": "user", "content": "Hello"}, ] system, result = convert_messages_to_anthropic(messages) assert system == "You are helpful." assert len(result) == 1 assert result[0]["role"] == "user" def test_converts_user_image_url_blocks_to_anthropic_image_blocks(self): messages = [ { "role": "user", "content": [ {"type": "text", "text": "Can you see this?"}, {"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}}, ], } ] _, result = convert_messages_to_anthropic(messages) assert result == [ { "role": "user", "content": [ {"type": "text", "text": "Can you see this?"}, {"type": "image", "source": {"type": "url", "url": "https://example.com/cat.png"}}, ], } ] def test_converts_data_url_image_blocks_to_base64_anthropic_image_blocks(self): messages = [ { "role": "user", "content": [ {"type": "input_text", "text": "What is in this screenshot?"}, {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}, ], } ] _, result = convert_messages_to_anthropic(messages) assert result == [ { "role": "user", "content": [ {"type": "text", "text": "What is in this screenshot?"}, { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": "AAAA", }, }, ], } ] def test_converts_tool_calls(self): messages = [ { "role": "assistant", "content": "Let me search.", "tool_calls": [ { "id": "tc_1", "function": { "name": "search", "arguments": '{"query": "test"}', }, } ], }, {"role": "tool", "tool_call_id": "tc_1", "content": "search results"}, ] _, result = convert_messages_to_anthropic(messages) blocks = result[0]["content"] assert blocks[0] == {"type": "text", "text": "Let me search."} assert blocks[1]["type"] == "tool_use" assert blocks[1]["id"] == "tc_1" assert blocks[1]["input"] == {"query": "test"} def test_converts_tool_results(self): messages = [ { "role": "assistant", "content": "", "tool_calls": [ {"id": "tc_1", "function": {"name": "test_tool", "arguments": "{}"}}, ], }, {"role": "tool", "tool_call_id": "tc_1", "content": "result data"}, ] _, result = convert_messages_to_anthropic(messages) # tool result is in the second message (user role) user_msg = [m for m in result if m["role"] == "user"][0] assert user_msg["content"][0]["type"] == "tool_result" assert user_msg["content"][0]["tool_use_id"] == "tc_1" def test_merges_consecutive_tool_results(self): messages = [ { "role": "assistant", "content": "", "tool_calls": [ {"id": "tc_1", "function": {"name": "tool_a", "arguments": "{}"}}, {"id": "tc_2", "function": {"name": "tool_b", "arguments": "{}"}}, ], }, {"role": "tool", "tool_call_id": "tc_1", "content": "result 1"}, {"role": "tool", "tool_call_id": "tc_2", "content": "result 2"}, ] _, result = convert_messages_to_anthropic(messages) # assistant + merged user (with 2 tool_results) user_msgs = [m for m in result if m["role"] == "user"] assert len(user_msgs) == 1 assert len(user_msgs[0]["content"]) == 2 def test_strips_orphaned_tool_use(self): messages = [ { "role": "assistant", "content": "", "tool_calls": [ {"id": "tc_orphan", "function": {"name": "x", "arguments": "{}"}} ], }, {"role": "user", "content": "never mind"}, ] _, result = convert_messages_to_anthropic(messages) # tc_orphan has no matching tool_result, should be stripped assistant_blocks = result[0]["content"] assert all(b.get("type") != "tool_use" for b in assistant_blocks) def test_strips_orphaned_tool_result(self): """tool_result with no matching tool_use should be stripped. This happens when context compression removes the assistant message containing the tool_use but leaves the subsequent tool_result intact. Anthropic rejects orphaned tool_results with a 400. """ messages = [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there"}, # The assistant tool_use message was removed by compression, # but the tool_result survived: {"role": "tool", "tool_call_id": "tc_gone", "content": "stale result"}, {"role": "user", "content": "Thanks"}, ] _, result = convert_messages_to_anthropic(messages) # tc_gone has no matching tool_use — its tool_result should be stripped for m in result: if m["role"] == "user" and isinstance(m["content"], list): assert all( b.get("type") != "tool_result" for b in m["content"] ), "Orphaned tool_result should have been stripped" def test_strips_orphaned_tool_result_preserves_valid(self): """Orphaned tool_results are stripped while valid ones survive.""" messages = [ { "role": "assistant", "content": "", "tool_calls": [ {"id": "tc_valid", "function": {"name": "search", "arguments": "{}"}}, ], }, {"role": "tool", "tool_call_id": "tc_valid", "content": "good result"}, {"role": "tool", "tool_call_id": "tc_orphan", "content": "stale result"}, ] _, result = convert_messages_to_anthropic(messages) user_msg = [m for m in result if m["role"] == "user"][0] tool_results = [ b for b in user_msg["content"] if b.get("type") == "tool_result" ] assert len(tool_results) == 1 assert tool_results[0]["tool_use_id"] == "tc_valid" def test_system_with_cache_control(self): messages = [ { "role": "system", "content": [ {"type": "text", "text": "System prompt", "cache_control": {"type": "ephemeral"}}, ], }, {"role": "user", "content": "Hi"}, ] system, result = convert_messages_to_anthropic(messages) # When cache_control is present, system should be a list of blocks assert isinstance(system, list) assert system[0]["cache_control"] == {"type": "ephemeral"} def test_assistant_cache_control_blocks_are_preserved(self): messages = apply_anthropic_cache_control([ {"role": "system", "content": "System prompt"}, {"role": "assistant", "content": "Hello from assistant"}, ]) _, result = convert_messages_to_anthropic(messages) assistant_blocks = result[0]["content"] assert assistant_blocks[0]["type"] == "text" assert assistant_blocks[0]["text"] == "Hello from assistant" assert assistant_blocks[0]["cache_control"] == {"type": "ephemeral"} def test_tool_cache_control_is_preserved_on_tool_result_block(self): messages = apply_anthropic_cache_control([ {"role": "system", "content": "System prompt"}, { "role": "assistant", "content": "", "tool_calls": [ {"id": "tc_1", "function": {"name": "test_tool", "arguments": "{}"}}, ], }, {"role": "tool", "tool_call_id": "tc_1", "content": "result"}, ], native_anthropic=True) _, result = convert_messages_to_anthropic(messages) user_msg = [m for m in result if m["role"] == "user"][0] tool_block = user_msg["content"][0] assert tool_block["type"] == "tool_result" assert tool_block["tool_use_id"] == "tc_1" assert tool_block["content"] == "result" assert tool_block["cache_control"] == {"type": "ephemeral"} def test_converts_data_url_image_to_anthropic_image_block(self): messages = [ { "role": "user", "content": [ {"type": "text", "text": "Describe this image"}, { "type": "image_url", "image_url": {"url": "data:image/png;base64,ZmFrZQ=="}, }, ], } ] _, result = convert_messages_to_anthropic(messages) blocks = result[0]["content"] assert blocks[0] == {"type": "text", "text": "Describe this image"} assert blocks[1] == { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": "ZmFrZQ==", }, } def test_converts_remote_image_url_to_anthropic_image_block(self): messages = [ { "role": "user", "content": [ {"type": "text", "text": "Describe this image"}, { "type": "image_url", "image_url": {"url": "https://example.com/cat.png"}, }, ], } ] _, result = convert_messages_to_anthropic(messages) blocks = result[0]["content"] assert blocks[1] == { "type": "image", "source": { "type": "url", "url": "https://example.com/cat.png", }, } def test_empty_cached_assistant_tool_turn_converts_without_empty_text_block(self): messages = apply_anthropic_cache_control([ {"role": "system", "content": "System prompt"}, {"role": "user", "content": "Find the skill"}, { "role": "assistant", "content": "", "tool_calls": [ {"id": "tc_1", "function": {"name": "skill_view", "arguments": "{}"}}, ], }, {"role": "tool", "tool_call_id": "tc_1", "content": "result"}, ]) _, result = convert_messages_to_anthropic(messages) assistant_turn = next(msg for msg in result if msg["role"] == "assistant") assistant_blocks = assistant_turn["content"] assert all(not (b.get("type") == "text" and b.get("text") == "") for b in assistant_blocks) assert any(b.get("type") == "tool_use" for b in assistant_blocks) def test_empty_user_message_string_gets_placeholder(self): """Empty user message strings should get '(empty message)' placeholder. Anthropic rejects requests with empty user message content. Regression test for #3143 — Discord @mention-only messages. """ messages = [ {"role": "user", "content": ""}, ] _, result = convert_messages_to_anthropic(messages) assert result[0]["role"] == "user" assert result[0]["content"] == "(empty message)" def test_whitespace_only_user_message_gets_placeholder(self): """Whitespace-only user messages should also get placeholder.""" messages = [ {"role": "user", "content": " \n\t "}, ] _, result = convert_messages_to_anthropic(messages) assert result[0]["content"] == "(empty message)" def test_empty_user_message_list_gets_placeholder(self): """Empty content list for user messages should get placeholder block.""" messages = [ {"role": "user", "content": []}, ] _, result = convert_messages_to_anthropic(messages) assert result[0]["role"] == "user" assert isinstance(result[0]["content"], list) assert len(result[0]["content"]) == 1 assert result[0]["content"][0] == {"type": "text", "text": "(empty message)"} def test_user_message_with_empty_text_blocks_gets_placeholder(self): """User message with only empty text blocks should get placeholder.""" messages = [ {"role": "user", "content": [{"type": "text", "text": ""}, {"type": "text", "text": " "}]}, ] _, result = convert_messages_to_anthropic(messages) assert result[0]["role"] == "user" assert isinstance(result[0]["content"], list) assert result[0]["content"] == [{"type": "text", "text": "(empty message)"}] # --------------------------------------------------------------------------- # Build kwargs # --------------------------------------------------------------------------- class TestBuildAnthropicKwargs: def test_basic_kwargs(self): messages = [ {"role": "system", "content": "Be helpful."}, {"role": "user", "content": "Hi"}, ] kwargs = build_anthropic_kwargs( model="claude-sonnet-4-20250514", messages=messages, tools=None, max_tokens=4096, reasoning_config=None, ) assert kwargs["model"] == "claude-sonnet-4-20250514" assert kwargs["system"] == "Be helpful." assert kwargs["max_tokens"] == 4096 assert "tools" not in kwargs def test_strips_anthropic_prefix(self): kwargs = build_anthropic_kwargs( model="anthropic/claude-sonnet-4-20250514", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=4096, reasoning_config=None, ) assert kwargs["model"] == "claude-sonnet-4-20250514" def test_reasoning_config_maps_to_manual_thinking_for_pre_4_6_models(self): kwargs = build_anthropic_kwargs( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "think hard"}], tools=None, max_tokens=4096, reasoning_config={"enabled": True, "effort": "high"}, ) assert kwargs["thinking"]["type"] == "enabled" assert kwargs["thinking"]["budget_tokens"] == 16000 assert kwargs["temperature"] == 1 assert kwargs["max_tokens"] >= 16000 + 4096 assert "output_config" not in kwargs def test_reasoning_config_maps_to_adaptive_thinking_for_4_6_models(self): kwargs = build_anthropic_kwargs( model="claude-opus-4-6", messages=[{"role": "user", "content": "think hard"}], tools=None, max_tokens=4096, reasoning_config={"enabled": True, "effort": "high"}, ) assert kwargs["thinking"] == {"type": "adaptive"} 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): kwargs = build_anthropic_kwargs( model="claude-sonnet-4-6", messages=[{"role": "user", "content": "think harder"}], tools=None, max_tokens=4096, reasoning_config={"enabled": True, "effort": "xhigh"}, ) assert kwargs["thinking"] == {"type": "adaptive"} assert kwargs["output_config"] == {"effort": "max"} def test_reasoning_disabled(self): kwargs = build_anthropic_kwargs( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "quick"}], tools=None, max_tokens=4096, reasoning_config={"enabled": False}, ) assert "thinking" not in kwargs def test_default_max_tokens_uses_model_output_limit(self): """When max_tokens is None, use the model's native output limit.""" kwargs = build_anthropic_kwargs( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, ) assert kwargs["max_tokens"] == 64_000 # Sonnet 4 output limit def test_default_max_tokens_opus_4_6(self): kwargs = build_anthropic_kwargs( model="claude-opus-4-6", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, ) assert kwargs["max_tokens"] == 128_000 def test_default_max_tokens_sonnet_4_6(self): kwargs = build_anthropic_kwargs( model="claude-sonnet-4-6", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, ) assert kwargs["max_tokens"] == 64_000 def test_default_max_tokens_date_stamped_model(self): """Date-stamped model IDs should resolve via substring match.""" kwargs = build_anthropic_kwargs( model="claude-sonnet-4-5-20250929", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, ) assert kwargs["max_tokens"] == 64_000 def test_default_max_tokens_older_model(self): kwargs = build_anthropic_kwargs( model="claude-3-5-sonnet-20241022", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, ) assert kwargs["max_tokens"] == 8_192 def test_default_max_tokens_unknown_model_uses_highest(self): """Unknown future models should get the highest known limit.""" kwargs = build_anthropic_kwargs( model="claude-ultra-5-20260101", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, ) assert kwargs["max_tokens"] == 128_000 def test_explicit_max_tokens_overrides_default(self): """User-specified max_tokens should be respected.""" kwargs = build_anthropic_kwargs( model="claude-opus-4-6", messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=4096, reasoning_config=None, ) assert kwargs["max_tokens"] == 4096 def test_context_length_clamp(self): """max_tokens should be clamped to context_length if it's smaller.""" kwargs = build_anthropic_kwargs( model="claude-opus-4-6", # 128K output messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, context_length=50000, ) assert kwargs["max_tokens"] == 49999 # context_length - 1 def test_context_length_no_clamp_when_larger(self): """No clamping when context_length exceeds output limit.""" kwargs = build_anthropic_kwargs( model="claude-sonnet-4-6", # 64K output messages=[{"role": "user", "content": "Hi"}], tools=None, max_tokens=None, reasoning_config=None, context_length=200000, ) assert kwargs["max_tokens"] == 64_000 # --------------------------------------------------------------------------- # Model output limit lookup # --------------------------------------------------------------------------- class TestGetAnthropicMaxOutput: def test_opus_4_6(self): from agent.anthropic_adapter import _get_anthropic_max_output assert _get_anthropic_max_output("claude-opus-4-6") == 128_000 def test_opus_4_6_variant(self): from agent.anthropic_adapter import _get_anthropic_max_output assert _get_anthropic_max_output("claude-opus-4-6:1m:fast") == 128_000 def test_sonnet_4_6(self): from agent.anthropic_adapter import _get_anthropic_max_output assert _get_anthropic_max_output("claude-sonnet-4-6") == 64_000 def test_sonnet_4_date_stamped(self): from agent.anthropic_adapter import _get_anthropic_max_output assert _get_anthropic_max_output("claude-sonnet-4-20250514") == 64_000 def test_claude_3_5_sonnet(self): from agent.anthropic_adapter import _get_anthropic_max_output assert _get_anthropic_max_output("claude-3-5-sonnet-20241022") == 8_192 def test_claude_3_opus(self): from agent.anthropic_adapter import _get_anthropic_max_output assert _get_anthropic_max_output("claude-3-opus-20240229") == 4_096 def test_unknown_future_model(self): from agent.anthropic_adapter import _get_anthropic_max_output assert _get_anthropic_max_output("claude-ultra-5-20260101") == 128_000 def test_longest_prefix_wins(self): """'claude-3-5-sonnet' should match before 'claude-3-5'.""" from agent.anthropic_adapter import _get_anthropic_max_output # claude-3-5-sonnet (8192) should win over a hypothetical shorter match assert _get_anthropic_max_output("claude-3-5-sonnet-20241022") == 8_192 # --------------------------------------------------------------------------- # Response normalization # --------------------------------------------------------------------------- class TestNormalizeResponse: def _make_response(self, content_blocks, stop_reason="end_turn"): resp = SimpleNamespace() resp.content = content_blocks resp.stop_reason = stop_reason resp.usage = SimpleNamespace(input_tokens=100, output_tokens=50) return resp def test_text_response(self): block = SimpleNamespace(type="text", text="Hello world") msg, reason = normalize_anthropic_response(self._make_response([block])) assert msg.content == "Hello world" assert reason == "stop" assert msg.tool_calls is None def test_tool_use_response(self): blocks = [ SimpleNamespace(type="text", text="Searching..."), SimpleNamespace( type="tool_use", id="tc_1", name="search", input={"query": "test"}, ), ] msg, reason = normalize_anthropic_response( self._make_response(blocks, "tool_use") ) assert msg.content == "Searching..." assert reason == "tool_calls" assert len(msg.tool_calls) == 1 assert msg.tool_calls[0].function.name == "search" assert json.loads(msg.tool_calls[0].function.arguments) == {"query": "test"} def test_thinking_response(self): blocks = [ SimpleNamespace(type="thinking", thinking="Let me reason about this..."), SimpleNamespace(type="text", text="The answer is 42."), ] msg, reason = normalize_anthropic_response(self._make_response(blocks)) assert msg.content == "The answer is 42." assert msg.reasoning == "Let me reason about this..." def test_stop_reason_mapping(self): block = SimpleNamespace(type="text", text="x") _, r1 = normalize_anthropic_response( self._make_response([block], "end_turn") ) _, r2 = normalize_anthropic_response( self._make_response([block], "tool_use") ) _, r3 = normalize_anthropic_response( self._make_response([block], "max_tokens") ) assert r1 == "stop" assert r2 == "tool_calls" assert r3 == "length" def test_no_text_content(self): block = SimpleNamespace( type="tool_use", id="tc_1", name="search", input={"q": "hi"} ) msg, reason = normalize_anthropic_response( self._make_response([block], "tool_use") ) assert msg.content is None assert len(msg.tool_calls) == 1 # --------------------------------------------------------------------------- # Role alternation # --------------------------------------------------------------------------- class TestRoleAlternation: def test_merges_consecutive_user_messages(self): messages = [ {"role": "user", "content": "Hello"}, {"role": "user", "content": "World"}, ] _, result = convert_messages_to_anthropic(messages) assert len(result) == 1 assert result[0]["role"] == "user" assert "Hello" in result[0]["content"] assert "World" in result[0]["content"] def test_preserves_proper_alternation(self): messages = [ {"role": "user", "content": "Hi"}, {"role": "assistant", "content": "Hello!"}, {"role": "user", "content": "How are you?"}, ] _, result = convert_messages_to_anthropic(messages) assert len(result) == 3 assert [m["role"] for m in result] == ["user", "assistant", "user"] # --------------------------------------------------------------------------- # Tool choice # --------------------------------------------------------------------------- class TestToolChoice: _DUMMY_TOOL = [ { "type": "function", "function": { "name": "test", "description": "x", "parameters": {"type": "object", "properties": {}}, }, } ] def test_auto_tool_choice(self): kwargs = build_anthropic_kwargs( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "Hi"}], tools=self._DUMMY_TOOL, max_tokens=4096, reasoning_config=None, tool_choice="auto", ) assert kwargs["tool_choice"] == {"type": "auto"} def test_required_tool_choice(self): kwargs = build_anthropic_kwargs( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "Hi"}], tools=self._DUMMY_TOOL, max_tokens=4096, reasoning_config=None, tool_choice="required", ) assert kwargs["tool_choice"] == {"type": "any"} def test_specific_tool_choice(self): kwargs = build_anthropic_kwargs( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "Hi"}], tools=self._DUMMY_TOOL, max_tokens=4096, reasoning_config=None, tool_choice="search", ) assert kwargs["tool_choice"] == {"type": "tool", "name": "search"}