diff --git a/agent/anthropic_adapter.py b/agent/anthropic_adapter.py index bfe54c585f..1e6a2a240a 100644 --- a/agent/anthropic_adapter.py +++ b/agent/anthropic_adapter.py @@ -46,7 +46,10 @@ _COMMON_BETAS = [ ] # Additional beta headers required for OAuth/subscription auth +# Both clawdbot and OpenCode include claude-code-20250219 alongside oauth-2025-04-20. +# Without claude-code-20250219, Anthropic's API rejects OAuth tokens with 401. _OAUTH_ONLY_BETAS = [ + "claude-code-20250219", "oauth-2025-04-20", ] @@ -157,13 +160,91 @@ def is_claude_code_token_valid(creds: Dict[str, Any]) -> bool: return now_ms < (expires_at - 60_000) +def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: + """Attempt to refresh an expired Claude Code OAuth token. + + Uses the same token endpoint and client_id as Claude Code / OpenCode. + Only works for credentials that have a refresh token (from claude /login + or claude setup-token with OAuth flow). + + Returns the new access token, or None if refresh fails. + """ + import urllib.parse + import urllib.request + + refresh_token = creds.get("refreshToken", "") + if not refresh_token: + logger.debug("No refresh token available — cannot refresh") + return None + + # Client ID used by Claude Code's OAuth flow + CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" + + data = urllib.parse.urlencode({ + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": CLIENT_ID, + }).encode() + + req = urllib.request.Request( + "https://console.anthropic.com/v1/oauth/token", + data=data, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + method="POST", + ) + + try: + with urllib.request.urlopen(req, timeout=10) as resp: + result = json.loads(resp.read().decode()) + new_access = result.get("access_token", "") + new_refresh = result.get("refresh_token", refresh_token) + expires_in = result.get("expires_in", 3600) # seconds + + if new_access: + import time + new_expires_ms = int(time.time() * 1000) + (expires_in * 1000) + # Write refreshed credentials back to ~/.claude/.credentials.json + _write_claude_code_credentials(new_access, new_refresh, new_expires_ms) + logger.debug("Successfully refreshed Claude Code OAuth token") + return new_access + except Exception as e: + logger.debug("Failed to refresh Claude Code token: %s", e) + + return None + + +def _write_claude_code_credentials(access_token: str, refresh_token: str, expires_at_ms: int) -> None: + """Write refreshed credentials back to ~/.claude/.credentials.json.""" + cred_path = Path.home() / ".claude" / ".credentials.json" + try: + # Read existing file to preserve other fields + existing = {} + if cred_path.exists(): + existing = json.loads(cred_path.read_text(encoding="utf-8")) + + existing["claudeAiOauth"] = { + "accessToken": access_token, + "refreshToken": refresh_token, + "expiresAt": expires_at_ms, + } + + cred_path.parent.mkdir(parents=True, exist_ok=True) + cred_path.write_text(json.dumps(existing, indent=2), encoding="utf-8") + # Restrict permissions (credentials file) + cred_path.chmod(0o600) + except (OSError, IOError) as e: + logger.debug("Failed to write refreshed credentials: %s", e) + + def resolve_anthropic_token() -> Optional[str]: """Resolve an Anthropic token from all available sources. Priority: 1. ANTHROPIC_API_KEY env var (regular API key) 2. ANTHROPIC_TOKEN env var (OAuth/setup token) - 3. Claude Code credentials (~/.claude.json or ~/.claude/.credentials.json) + 3. CLAUDE_CODE_OAUTH_TOKEN env var + 4. Claude Code credentials (~/.claude.json or ~/.claude/.credentials.json) + — with automatic refresh if expired and a refresh token is available Returns the token string or None. """ @@ -177,18 +258,63 @@ def resolve_anthropic_token() -> Optional[str]: if token: return token - # Also check CLAUDE_CODE_OAUTH_TOKEN (used by Claude Code for setup-tokens) + # 3. CLAUDE_CODE_OAUTH_TOKEN (used by Claude Code for setup-tokens) cc_token = os.getenv("CLAUDE_CODE_OAUTH_TOKEN", "").strip() if cc_token: return cc_token - # 3. Claude Code credential file + # 4. Claude Code credential file creds = read_claude_code_credentials() if creds and is_claude_code_token_valid(creds): logger.debug("Using Claude Code credentials (auto-detected)") return creds["accessToken"] elif creds: - logger.debug("Claude Code credentials expired — run 'claude' to refresh") + # Token expired — attempt to refresh + logger.debug("Claude Code credentials expired — attempting refresh") + refreshed = _refresh_oauth_token(creds) + if refreshed: + return refreshed + logger.debug("Token refresh failed — re-run 'claude setup-token' to reauthenticate") + + return None + + +def run_oauth_setup_token() -> Optional[str]: + """Run 'claude setup-token' interactively and return the resulting token. + + Checks multiple sources after the subprocess completes: + 1. Claude Code credential files (may be written by the subprocess) + 2. CLAUDE_CODE_OAUTH_TOKEN / ANTHROPIC_TOKEN env vars + + Returns the token string, or None if no credentials were obtained. + Raises FileNotFoundError if the 'claude' CLI is not installed. + """ + import shutil + import subprocess + + claude_path = shutil.which("claude") + if not claude_path: + raise FileNotFoundError( + "The 'claude' CLI is not installed. " + "Install it with: npm install -g @anthropic-ai/claude-code" + ) + + # Run interactively — stdin/stdout/stderr inherited so user can interact + try: + subprocess.run([claude_path, "setup-token"]) + except (KeyboardInterrupt, EOFError): + return None + + # Check if credentials were saved to Claude Code's config files + creds = read_claude_code_credentials() + if creds and is_claude_code_token_valid(creds): + return creds["accessToken"] + + # Check env vars that may have been set + for env_var in ("CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_TOKEN"): + val = os.getenv(env_var, "").strip() + if val: + return val return None diff --git a/hermes_cli/auth.py b/hermes_cli/auth.py index c1b0834844..3eadd5d701 100644 --- a/hermes_cli/auth.py +++ b/hermes_cli/auth.py @@ -1571,7 +1571,11 @@ def _update_config_for_provider(provider_id: str, inference_base_url: str) -> Pa model_cfg = {} model_cfg["provider"] = provider_id - model_cfg["base_url"] = inference_base_url.rstrip("/") + if inference_base_url and inference_base_url.strip(): + model_cfg["base_url"] = inference_base_url.rstrip("/") + else: + # Clear stale base_url to prevent contamination when switching providers + model_cfg.pop("base_url", None) config["model"] = model_cfg config_path.write_text(yaml.safe_dump(config, sort_keys=False)) diff --git a/hermes_cli/main.py b/hermes_cli/main.py index c98e6f290b..cad11531e6 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -1590,8 +1590,67 @@ def _model_flow_api_key_provider(config, provider_id, current_model=""): print("No change.") +def _run_anthropic_oauth_flow(save_env_value): + """Run the Claude OAuth setup-token flow. Returns True if credentials were saved.""" + from agent.anthropic_adapter import run_oauth_setup_token + + try: + print() + print(" Running 'claude setup-token' — follow the prompts below.") + print(" A browser window will open for you to authorize access.") + print() + token = run_oauth_setup_token() + if token: + save_env_value("ANTHROPIC_API_KEY", token) + print(" ✓ OAuth credentials saved.") + return True + + # Subprocess completed but no token auto-detected — ask user to paste + print() + print(" If the setup-token was displayed above, paste it here:") + print() + try: + manual_token = input(" Paste setup-token (or Enter to cancel): ").strip() + except (KeyboardInterrupt, EOFError): + print() + return False + if manual_token: + save_env_value("ANTHROPIC_API_KEY", manual_token) + print(" ✓ Setup-token saved.") + return True + + print(" ⚠ Could not detect saved credentials.") + return False + + except FileNotFoundError: + # Claude CLI not installed — guide user through manual setup + print() + print(" The 'claude' CLI is required for OAuth login.") + print() + print(" To install and authenticate:") + print() + print(" 1. Install Claude Code: npm install -g @anthropic-ai/claude-code") + print(" 2. Run: claude setup-token") + print(" 3. Follow the browser prompts to authorize") + print(" 4. Re-run: hermes model") + print() + print(" Or paste an existing setup-token now (sk-ant-oat-...):") + print() + try: + token = input(" Setup-token (or Enter to cancel): ").strip() + except (KeyboardInterrupt, EOFError): + print() + return False + if token: + save_env_value("ANTHROPIC_API_KEY", token) + print(" ✓ Setup-token saved.") + return True + print(" Cancelled — install Claude Code and try again.") + return False + + def _model_flow_anthropic(config, current_model=""): - """Flow for Anthropic provider — setup-token, API key, or Claude Code creds.""" + """Flow for Anthropic provider — OAuth subscription, API key, or Claude Code creds.""" import os from hermes_cli.auth import ( PROVIDER_REGISTRY, _prompt_model_selection, _save_model_choice, @@ -1602,12 +1661,13 @@ def _model_flow_anthropic(config, current_model=""): pconfig = PROVIDER_REGISTRY["anthropic"] - # Check for existing credentials + # Check ALL credential sources existing_key = ( get_env_value("ANTHROPIC_API_KEY") or os.getenv("ANTHROPIC_API_KEY", "") or get_env_value("ANTHROPIC_TOKEN") or os.getenv("ANTHROPIC_TOKEN", "") + or os.getenv("CLAUDE_CODE_OAUTH_TOKEN", "") ) cc_available = False try: @@ -1618,27 +1678,37 @@ def _model_flow_anthropic(config, current_model=""): except Exception: pass - if existing_key: - print(f" Anthropic credentials: {existing_key[:12]}... ✓") + has_creds = bool(existing_key) or cc_available + needs_auth = not has_creds + + if has_creds: + # Show what we found + if existing_key: + print(f" Anthropic credentials: {existing_key[:12]}... ✓") + elif cc_available: + print(" Claude Code credentials: ✓ (auto-detected)") + print() + print(" 1. Use existing credentials") + print(" 2. Reauthenticate (new OAuth login)") + print(" 3. Cancel") print() try: - update = input("Update credentials? [y/N]: ").strip().lower() + choice = input(" Choice [1/2/3]: ").strip() except (KeyboardInterrupt, EOFError): - update = "" - if update != "y": - pass # skip to model selection - else: - existing_key = "" # fall through to auth choice below - elif cc_available: - print(" Claude Code credentials: ✓ (auto-detected)") - print() - - if not existing_key and not cc_available: - # No credentials — show auth method choice + choice = "1" + + if choice == "2": + needs_auth = True + elif choice == "3": + return + # choice == "1" or default: use existing, proceed to model selection + + if needs_auth: + # Show auth method choice print() print(" Choose authentication method:") print() - print(" 1. Claude Pro/Max subscription (setup-token)") + print(" 1. Claude Pro/Max subscription (OAuth login)") print(" 2. Anthropic API key (pay-per-token)") print(" 3. Cancel") print() @@ -1649,33 +1719,15 @@ def _model_flow_anthropic(config, current_model=""): return if choice == "1": - print() - print(" To get a setup-token from your Claude subscription:") - print() - print(" 1. Install Claude Code: npm install -g @anthropic-ai/claude-code") - print(" 2. Run: claude setup-token") - print(" 3. Open the URL it prints in your browser") - print(" 4. Log in and click \"Authorize\"") - print(" 5. Paste the auth code back into Claude Code") - print(" 6. Copy the resulting sk-ant-oat01-... token") - print() - try: - token = input(" Paste setup-token here: ").strip() - except (KeyboardInterrupt, EOFError): - print() + if not _run_anthropic_oauth_flow(save_env_value): return - if not token: - print(" Cancelled.") - return - save_env_value("ANTHROPIC_API_KEY", token) - print(" ✓ Setup-token saved.") elif choice == "2": print() print(" Get an API key at: https://console.anthropic.com/settings/keys") print() try: - api_key = input(" API key (sk-ant-api03-...): ").strip() + api_key = input(" API key (sk-ant-...): ").strip() except (KeyboardInterrupt, EOFError): print() return @@ -1708,14 +1760,17 @@ def _model_flow_anthropic(config, current_model=""): _save_model_choice(selected) - # Update config with provider + # Update config with provider — clear base_url since + # resolve_runtime_provider() always hardcodes Anthropic's URL. + # Leaving a stale base_url in config can contaminate other + # providers if the user switches without running 'hermes model'. cfg = load_config() model = cfg.get("model") if not isinstance(model, dict): model = {"default": model} if model else {} cfg["model"] = model model["provider"] = "anthropic" - model["base_url"] = pconfig.inference_base_url + model.pop("base_url", None) save_config(cfg) deactivate_provider() diff --git a/hermes_cli/models.py b/hermes_cli/models.py index ff26a9d16c..3b3d0ab4db 100644 --- a/hermes_cli/models.py +++ b/hermes_cli/models.py @@ -271,7 +271,8 @@ def _fetch_anthropic_models(timeout: float = 5.0) -> Optional[list[str]]: headers: dict[str, str] = {"anthropic-version": "2023-06-01"} if _is_oauth_token(token): headers["Authorization"] = f"Bearer {token}" - headers["anthropic-beta"] = "oauth-2025-04-20" + from agent.anthropic_adapter import _COMMON_BETAS, _OAUTH_ONLY_BETAS + headers["anthropic-beta"] = ",".join(_COMMON_BETAS + _OAUTH_ONLY_BETAS) else: headers["x-api-key"] = token diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index d2582c6778..f217540bf0 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -1076,65 +1076,101 @@ def setup_model_provider(config: dict): from hermes_cli.auth import PROVIDER_REGISTRY pconfig = PROVIDER_REGISTRY["anthropic"] - # Check for Claude Code credential auto-discovery - from agent.anthropic_adapter import read_claude_code_credentials, is_claude_code_token_valid + # Check ALL credential sources + import os as _os + from agent.anthropic_adapter import ( + read_claude_code_credentials, is_claude_code_token_valid, + run_oauth_setup_token, + ) cc_creds = read_claude_code_credentials() - if cc_creds and is_claude_code_token_valid(cc_creds): - print_success("Found valid Claude Code credentials (~/.claude/.credentials.json)") - if prompt_yes_no("Use these credentials?", True): - print_success("Using Claude Code subscription credentials") - else: - cc_creds = None + cc_valid = bool(cc_creds and is_claude_code_token_valid(cc_creds)) - existing_key = get_env_value("ANTHROPIC_API_KEY") or get_env_value("ANTHROPIC_TOKEN") + existing_key = ( + get_env_value("ANTHROPIC_API_KEY") + or get_env_value("ANTHROPIC_TOKEN") + or _os.getenv("CLAUDE_CODE_OAUTH_TOKEN", "") + ) - if not (cc_creds and is_claude_code_token_valid(cc_creds)): + has_creds = bool(existing_key) or cc_valid + needs_auth = not has_creds + + if has_creds: if existing_key: print_info(f"Current credentials: {existing_key[:12]}...") - if not prompt_yes_no("Update credentials?", False): - # User wants to keep existing — skip auth prompt entirely - existing_key = "KEEP" # truthy sentinel to skip auth choice + elif cc_valid: + print_success("Found valid Claude Code credentials (auto-detected)") - if not existing_key and not (cc_creds and is_claude_code_token_valid(cc_creds)): - auth_choices = [ - "Claude Pro/Max subscription (setup-token)", - "Anthropic API key (pay-per-token)", - ] - auth_idx = prompt_choice("Choose authentication method:", auth_choices, 0) + auth_choices = [ + "Use existing credentials", + "Reauthenticate (new OAuth login)", + "Cancel", + ] + choice_idx = prompt_choice("What would you like to do?", auth_choices, 0) + if choice_idx == 1: + needs_auth = True + elif choice_idx == 2: + pass # fall through to provider config - if auth_idx == 0: + if needs_auth: + auth_choices = [ + "Claude Pro/Max subscription (OAuth login)", + "Anthropic API key (pay-per-token)", + ] + auth_idx = prompt_choice("Choose authentication method:", auth_choices, 0) + + if auth_idx == 0: + # OAuth setup-token flow + try: print() - print_info("To get a setup-token from your Claude subscription:") - print_info(" 1. Install Claude Code: npm install -g @anthropic-ai/claude-code") - print_info(" 2. Run: claude setup-token") - print_info(" 3. Open the URL it prints in your browser") - print_info(" 4. Log in and click \"Authorize\"") - print_info(" 5. Paste the auth code back into Claude Code") - print_info(" 6. Copy the resulting sk-ant-oat01-... token") + print_info("Running 'claude setup-token' — follow the prompts below.") + print_info("A browser window will open for you to authorize access.") print() - token = prompt("Paste setup-token here", password=True) + token = run_oauth_setup_token() + if token: + save_env_value("ANTHROPIC_API_KEY", token) + print_success("OAuth credentials saved") + else: + # Subprocess completed but no token auto-detected + print() + token = prompt("Paste setup-token here (if displayed above)", password=True) + if token: + save_env_value("ANTHROPIC_API_KEY", token) + print_success("Setup-token saved") + else: + print_warning("Skipped — agent won't work without credentials") + except FileNotFoundError: + print() + print_info("The 'claude' CLI is required for OAuth login.") + print() + print_info("To install: npm install -g @anthropic-ai/claude-code") + print_info("Then run: claude setup-token") + print_info("Or paste an existing setup-token below:") + print() + token = prompt("Setup-token (sk-ant-oat-...)", password=True) if token: save_env_value("ANTHROPIC_API_KEY", token) print_success("Setup-token saved") else: - print_warning("Skipped — agent won't work without credentials") + print_warning("Skipped — install Claude Code and re-run setup") + else: + print() + print_info("Get an API key at: https://console.anthropic.com/settings/keys") + print() + api_key = prompt("API key (sk-ant-...)", password=True) + if api_key: + save_env_value("ANTHROPIC_API_KEY", api_key) + print_success("API key saved") else: - print() - print_info("Get an API key at: https://console.anthropic.com/settings/keys") - print() - api_key = prompt("API key (sk-ant-api03-...)", password=True) - if api_key: - save_env_value("ANTHROPIC_API_KEY", api_key) - print_success("API key saved") - else: - print_warning("Skipped — agent won't work without credentials") + print_warning("Skipped — agent won't work without credentials") # Clear custom endpoint vars if switching if existing_custom: save_env_value("OPENAI_BASE_URL", "") save_env_value("OPENAI_API_KEY", "") - _update_config_for_provider("anthropic", pconfig.inference_base_url) - _set_model_provider(config, "anthropic", pconfig.inference_base_url) + # Don't save base_url for Anthropic — resolve_runtime_provider() + # always hardcodes it. Stale base_urls contaminate other providers. + _update_config_for_provider("anthropic", "") + _set_model_provider(config, "anthropic") # else: provider_idx == 9 (Keep current) — only shown when a provider already exists diff --git a/tests/test_anthropic_adapter.py b/tests/test_anthropic_adapter.py index 2d36cdd7c5..1f4006d518 100644 --- a/tests/test_anthropic_adapter.py +++ b/tests/test_anthropic_adapter.py @@ -9,6 +9,8 @@ import pytest 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, @@ -18,6 +20,7 @@ from agent.anthropic_adapter import ( normalize_model_name, read_claude_code_credentials, resolve_anthropic_token, + run_oauth_setup_token, ) @@ -53,6 +56,7 @@ class TestBuildAnthropicClient: 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 @@ -67,6 +71,7 @@ class TestBuildAnthropicClient: 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: @@ -145,6 +150,194 @@ class TestResolveAnthropicToken: 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" + + +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" + + +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