diff --git a/code_puppy/config.py b/code_puppy/config.py index 79b5c2088..10c1e63e9 100644 --- a/code_puppy/config.py +++ b/code_puppy/config.py @@ -55,6 +55,7 @@ def _get_xdg_dir(env_var: str, fallback: str) -> str: CHATGPT_MODELS_FILE = os.path.join(DATA_DIR, "chatgpt_models.json") CLAUDE_MODELS_FILE = os.path.join(DATA_DIR, "claude_models.json") ANTIGRAVITY_MODELS_FILE = os.path.join(DATA_DIR, "antigravity_models.json") +GITHUB_MODELS_FILE = os.path.join(DATA_DIR, "github_models.json") # Cache files (XDG_CACHE_HOME) AUTOSAVE_DIR = os.path.join(CACHE_DIR, "autosaves") diff --git a/code_puppy/model_factory.py b/code_puppy/model_factory.py index 12f716d7b..93aa94e16 100644 --- a/code_puppy/model_factory.py +++ b/code_puppy/model_factory.py @@ -312,6 +312,7 @@ def load_config() -> Dict[str, Any]: CHATGPT_MODELS_FILE, CLAUDE_MODELS_FILE, GEMINI_MODELS_FILE, + GITHUB_MODELS_FILE, ) # Build list of extra model sources @@ -321,6 +322,7 @@ def load_config() -> Dict[str, Any]: (pathlib.Path(CLAUDE_MODELS_FILE), "Claude Code OAuth models", True), (pathlib.Path(GEMINI_MODELS_FILE), "Gemini OAuth models", False), (pathlib.Path(ANTIGRAVITY_MODELS_FILE), "Antigravity OAuth models", False), + (pathlib.Path(GITHUB_MODELS_FILE), "GitHub Models OAuth models", False), ] for source_path, label, use_filtered in extra_sources: diff --git a/code_puppy/plugins/github_models_oauth/__init__.py b/code_puppy/plugins/github_models_oauth/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/code_puppy/plugins/github_models_oauth/config.py b/code_puppy/plugins/github_models_oauth/config.py new file mode 100644 index 000000000..e43f47ba5 --- /dev/null +++ b/code_puppy/plugins/github_models_oauth/config.py @@ -0,0 +1,61 @@ +"""Configuration for the GitHub Models OAuth plugin.""" + +from pathlib import Path +from typing import Any, Dict + +from code_puppy import config + +# GitHub Models OAuth configuration +GITHUB_MODELS_OAUTH_CONFIG: Dict[str, Any] = { + # GitHub OAuth Device Flow endpoints + "device_code_url": "https://github.com/login/device/code", + "access_token_url": "https://github.com/login/oauth/access_token", + "user_api_url": "https://api.github.com/user", + # GitHub Models Inference API (OpenAI-compatible) + "api_base_url": "https://models.github.ai/inference", + "api_version": "2026-03-10", + # GitHub Copilot API (OpenAI-compatible, has Claude/Gemini) + "copilot_api_base_url": "https://api.githubcopilot.com", + "copilot_integration_id": "vscode-chat", + "copilot_prefix": "copilot-", + # OAuth configuration β€” client_id from a registered GitHub OAuth App + # with device flow enabled. Set via GITHUB_MODELS_CLIENT_ID env var. + # No default β€” users must register their own app or use `gh` CLI auth. + "default_client_id": "", + "client_id_env_var": "GITHUB_MODELS_CLIENT_ID", + "scope": "read:user", + # Device flow polling + "poll_timeout": 900, # 15 minutes (matches GitHub's device_code expiry) + # Model configuration + "prefix": "github-", + "default_context_length": 200000, + # User-Agent for API calls + "user_agent": "code-puppy/github-models-oauth", +} + + +def get_client_id() -> str: + """Get the GitHub OAuth App client ID. + + Checks the environment variable first, then falls back to the default. + """ + import os + + return os.environ.get( + GITHUB_MODELS_OAUTH_CONFIG["client_id_env_var"], + GITHUB_MODELS_OAUTH_CONFIG["default_client_id"], + ) + + +def get_token_storage_path() -> Path: + """Get the path for storing GitHub OAuth tokens.""" + data_dir = Path(config.DATA_DIR) + data_dir.mkdir(parents=True, exist_ok=True, mode=0o700) + return data_dir / "github_models_oauth.json" + + +def get_github_models_path() -> Path: + """Get the path to the github_models.json model config file.""" + data_dir = Path(config.DATA_DIR) + data_dir.mkdir(parents=True, exist_ok=True, mode=0o700) + return data_dir / "github_models.json" diff --git a/code_puppy/plugins/github_models_oauth/device_flow.py b/code_puppy/plugins/github_models_oauth/device_flow.py new file mode 100644 index 000000000..b732effd8 --- /dev/null +++ b/code_puppy/plugins/github_models_oauth/device_flow.py @@ -0,0 +1,178 @@ +"""GitHub OAuth device flow implementation. + +Follows the GitHub device flow (RFC 8628) as used by copilot-sdk. +No local HTTP server required β€” the user visits github.com/login/device +and enters a short code. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Optional + +import requests + +from code_puppy.messaging import emit_error, emit_info, emit_success, emit_warning + +from .config import GITHUB_MODELS_OAUTH_CONFIG, get_client_id + +logger = logging.getLogger(__name__) + + +@dataclass +class DeviceFlowResponse: + """Response from GitHub's device code endpoint.""" + + device_code: str + user_code: str + verification_uri: str + expires_in: int + interval: int + + +def start_device_flow() -> Optional[DeviceFlowResponse]: + """Initiate the GitHub OAuth device flow. + + POST to ``github.com/login/device/code`` to obtain a device code and + user code that the user enters at ``github.com/login/device``. + """ + client_id = get_client_id() + url = GITHUB_MODELS_OAUTH_CONFIG["device_code_url"] + scope = GITHUB_MODELS_OAUTH_CONFIG["scope"] + + try: + response = requests.post( + url, + data={"client_id": client_id, "scope": scope}, + headers={ + "Accept": "application/json", + }, + timeout=30, + ) + response.raise_for_status() + data = response.json() + + return DeviceFlowResponse( + device_code=data["device_code"], + user_code=data["user_code"], + verification_uri=data["verification_uri"], + expires_in=int(data.get("expires_in", 900)), + interval=int(data.get("interval", 5)), + ) + except requests.RequestException as exc: + logger.error("Failed to start device flow: %s", exc) + emit_error(f"Failed to start GitHub device flow: {exc}") + return None + except (KeyError, ValueError) as exc: + logger.error("Unexpected device flow response: %s", exc) + emit_error(f"Unexpected response from GitHub: {exc}") + return None + + +def poll_for_access_token(device_code: str, interval: int) -> Optional[str]: + """Poll GitHub for an access token after the user authorises. + + Handles ``authorization_pending`` (keep trying), ``slow_down`` + (increase interval), and ``expired_token`` (give up). + """ + client_id = get_client_id() + url = GITHUB_MODELS_OAUTH_CONFIG["access_token_url"] + timeout = GITHUB_MODELS_OAUTH_CONFIG["poll_timeout"] + + delay = max(1, int(interval)) + deadline = time.monotonic() + timeout + + while time.monotonic() < deadline: + time.sleep(delay) + + try: + response = requests.post( + url, + data={ + "client_id": client_id, + "device_code": device_code, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + }, + headers={ + "Accept": "application/json", + }, + timeout=30, + ) + data = response.json() + except requests.RequestException as exc: + logger.warning("Token poll request failed: %s", exc) + continue + except ValueError: + logger.warning("Token poll returned non-JSON response") + continue + + # Success + if data.get("access_token"): + return data["access_token"] + + error = data.get("error", "") + + if error == "authorization_pending": + continue + + if error == "slow_down": + delay = int(data.get("interval", delay + 5)) + continue + + if error == "expired_token": + logger.warning("Device code expired before user authorized") + return None + + # Unknown error β€” abort + desc = data.get("error_description", error) + logger.error("OAuth polling error: %s", desc) + emit_error(f"GitHub OAuth error: {desc}") + return None + + logger.warning("Device flow polling timed out after %ds", timeout) + return None + + +def run_device_flow() -> Optional[str]: + """Run the full GitHub OAuth device flow. + + Returns the access token on success, or ``None`` on failure. + """ + emit_info("πŸ” Starting GitHub OAuth device flow…") + + device = start_device_flow() + if not device: + return None + + emit_info(f"\nπŸ“‹ Open: {device.verification_uri}") + emit_info(f"πŸ“‹ Enter code: {device.user_code}\n") + + # Try to open the browser automatically + try: + import webbrowser + + from code_puppy.tools.common import should_suppress_browser + + if should_suppress_browser(): + emit_info(f"[HEADLESS MODE] Would normally open: {device.verification_uri}") + else: + webbrowser.open(device.verification_uri) + except Exception as exc: # noqa: BLE001 + logger.debug("Could not open browser: %s", exc) + + emit_info("⏳ Waiting for authorization (press Ctrl+C to cancel)…") + + try: + token = poll_for_access_token(device.device_code, device.interval) + except KeyboardInterrupt: + emit_warning("GitHub authentication cancelled by user.") + return None + + if token: + emit_success("βœ… GitHub authentication successful!") + return token + + emit_error("❌ GitHub authentication failed or timed out.") + return None diff --git a/code_puppy/plugins/github_models_oauth/register_callbacks.py b/code_puppy/plugins/github_models_oauth/register_callbacks.py new file mode 100644 index 000000000..dfe667982 --- /dev/null +++ b/code_puppy/plugins/github_models_oauth/register_callbacks.py @@ -0,0 +1,246 @@ +"""GitHub Models OAuth plugin β€” authentication and model type handlers.""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional, Tuple + +from code_puppy.callbacks import register_callback +from code_puppy.messaging import emit_error, emit_info, emit_success, emit_warning +from code_puppy.model_switching import set_model_and_reload_agent + +from .config import GITHUB_MODELS_OAUTH_CONFIG, get_client_id, get_token_storage_path +from .device_flow import run_device_flow +from .utils import ( + add_copilot_models_to_config, + add_models_to_config, + fetch_copilot_models, + fetch_github_models, + get_env_token, + get_gh_cli_token, + get_github_username, + load_github_models_config, + load_stored_tokens, + prompt_for_token, + remove_github_models, + save_tokens, +) + +logger = logging.getLogger(__name__) + + +def _custom_help() -> List[Tuple[str, str]]: + return [ + ("github-auth", "Authenticate with GitHub. Use '/github-auth token' to paste a PAT"), + ("github-status", "Show GitHub auth status and configured models"), + ("github-logout", "Remove GitHub OAuth tokens and imported models"), + ] + + +def _handle_auth(force_prompt: bool = False) -> bool: + """Authenticate via gh CLI β†’ env var β†’ PAT paste β†’ device flow. + + *force_prompt* skips auto-detect and goes straight to the PAT prompt. + Returns ``True`` on success. + """ + tokens = load_stored_tokens() + if tokens and tokens.get("access_token"): + emit_warning("Existing GitHub tokens found. This will overwrite them.") + + access_token: Optional[str] = None + + if not force_prompt: + emit_info("πŸ” Checking for gh CLI authentication…") + access_token = get_gh_cli_token() + if access_token: + emit_success("βœ… Found token from gh CLI") + + if not access_token: + access_token = get_env_token() + if access_token: + emit_success("βœ… Found token from environment variable") + + if not access_token: + access_token = prompt_for_token() + if access_token: + emit_success("βœ… Token received") + + if not access_token: + client_id = get_client_id() + if client_id: + access_token = run_device_flow() + + if not access_token: + emit_error( + "❌ Authentication failed. Options:\n" + " β€’ Install GitHub CLI: brew install gh && gh auth login\n" + " β€’ Set GITHUB_TOKEN env var with a Personal Access Token\n" + " β€’ Run /github-auth token to paste a PAT directly" + ) + return False + + username = get_github_username(access_token) + if not username: + emit_error("❌ Token validation failed β€” could not fetch GitHub user.") + return False + + emit_info(f"πŸ‘€ Logged in as: {username}") + if not save_tokens({"access_token": access_token, "username": username}): + emit_error("Failed to save tokens. Check file permissions.") + return False + + # Discover and register models from both APIs + total = 0 + emit_info("πŸ“¦ Fetching available GitHub Models…") + models = fetch_github_models(access_token) + if models and add_models_to_config(models): + total += len(models) + + emit_info("πŸ“¦ Fetching available Copilot models (Claude, Gemini, etc.)…") + copilot_models = fetch_copilot_models(access_token) + if copilot_models and add_copilot_models_to_config(copilot_models): + total += len(copilot_models) + + if total: + emit_success( + f"βœ… {total} models registered!\n" + " github-* β†’ GitHub Models (OpenAI, Meta, Mistral, DeepSeek…)\n" + " copilot-* β†’ Copilot API (Claude, Gemini, GPT…)\n" + " Run /github-status to see all available models" + ) + else: + emit_warning("No models discovered. You can still try models manually.") + + return True + + +def _handle_status() -> None: + tokens = load_stored_tokens() + if not tokens or not tokens.get("access_token"): + emit_warning("πŸ”“ Not authenticated. Run /github-auth to sign in.") + return + + username = tokens.get("username", "unknown") + emit_success(f"πŸ” GitHub: Authenticated as {username}") + + config = load_github_models_config() + gh = sorted(n for n, c in config.items() if c.get("type") == "github_models") + cp = sorted(n for n, c in config.items() if c.get("type") == "github_copilot") + + if gh: + emit_info(f"\n🌐 GitHub Models ({len(gh)}) β€” prefix: github-") + for name in gh[:8]: + emit_info(f" β€’ {name}") + if len(gh) > 8: + emit_info(f" … and {len(gh) - 8} more") + if cp: + emit_info(f"\nπŸ€– Copilot Models ({len(cp)}) β€” prefix: copilot-") + for name in cp: + emit_info(f" β€’ {name}") + if not gh and not cp: + emit_warning("No models configured. Run /github-auth.") + + +def _handle_logout() -> None: + token_path = get_token_storage_path() + try: + if token_path.exists(): + token_path.unlink() + emit_info("βœ“ Removed GitHub OAuth tokens") + except OSError as exc: + logger.error("Failed to remove token file: %s", exc) + emit_error(f"Failed to remove token file: {exc}") + + removed = remove_github_models() + if removed: + emit_info(f"βœ“ Removed {removed} GitHub models from configuration") + + emit_success("πŸ‘‹ GitHub Models logout complete") + + +def _handle_custom_command(command: str, name: str) -> Optional[bool]: + if not name: + return None + + if name == "github-auth": + force = "token" in command.lower().split()[1:] if len(command.split()) > 1 else False + if _handle_auth(force_prompt=force): + model = "github-openai-gpt-4.1" + if load_github_models_config().get(model): + set_model_and_reload_agent(model) + emit_success(f"πŸ”„ Switched to model: {model}") + else: + emit_warning("Authenticated, but no default model was registered; skipping auto-switch.") + try: + from code_puppy.config import get_global_model_name + current = get_global_model_name() + if current and "copilot" in current: + emit_warning(f"⚠️ Agent pinned to '{current}'. Run: /model {model}") + except Exception: + pass + return True + + handlers = {"github-status": _handle_status, "github-logout": _handle_logout} + handler = handlers.get(name) + if handler: + handler() + return True + return None + + +def _create_model_with_token( + model_config: Dict, *, default_base_url: str, extra_headers: Dict[str, str], +) -> Any: + """Create an OpenAI-compatible model backed by a stored GitHub token.""" + from pydantic_ai.models.openai import OpenAIChatModel + from pydantic_ai.providers.openai import OpenAIProvider + + from code_puppy.http_utils import create_async_client, get_cert_bundle_path + + tokens = load_stored_tokens() + if not tokens or not tokens.get("access_token"): + emit_warning( + f"GitHub token not found for '{model_config.get('name')}'. Run /github-auth." + ) + return None + + base_url = model_config.get("custom_endpoint", {}).get("url", default_base_url) + client = create_async_client(headers=extra_headers, verify=get_cert_bundle_path()) + provider = OpenAIProvider( + api_key=tokens["access_token"], base_url=base_url, http_client=client, + ) + return OpenAIChatModel(model_name=model_config["name"], provider=provider) + + +def _create_github_models_model(model_name: str, model_config: Dict, config: Dict) -> Any: + return _create_model_with_token( + model_config, + default_base_url=GITHUB_MODELS_OAUTH_CONFIG["api_base_url"], + extra_headers={ + "X-GitHub-Api-Version": GITHUB_MODELS_OAUTH_CONFIG["api_version"], + "User-Agent": GITHUB_MODELS_OAUTH_CONFIG["user_agent"], + }, + ) + + +def _create_copilot_model(model_name: str, model_config: Dict, config: Dict) -> Any: + return _create_model_with_token( + model_config, + default_base_url=GITHUB_MODELS_OAUTH_CONFIG["copilot_api_base_url"], + extra_headers={ + "Copilot-Integration-Id": GITHUB_MODELS_OAUTH_CONFIG["copilot_integration_id"], + "User-Agent": GITHUB_MODELS_OAUTH_CONFIG["user_agent"], + }, + ) + + +def _register_model_types() -> List[Dict[str, Any]]: + return [ + {"type": "github_models", "handler": _create_github_models_model}, + {"type": "github_copilot", "handler": _create_copilot_model}, + ] + + +register_callback("custom_command_help", _custom_help) +register_callback("custom_command", _handle_custom_command) +register_callback("register_model_type", _register_model_types) diff --git a/code_puppy/plugins/github_models_oauth/test_plugin.py b/code_puppy/plugins/github_models_oauth/test_plugin.py new file mode 100644 index 000000000..29b36ceab --- /dev/null +++ b/code_puppy/plugins/github_models_oauth/test_plugin.py @@ -0,0 +1,832 @@ +"""Tests for the GitHub Models OAuth plugin.""" + +from __future__ import annotations + +import json +import os +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from code_puppy.plugins.github_models_oauth import config, utils +from code_puppy.plugins.github_models_oauth.device_flow import ( + DeviceFlowResponse, + poll_for_access_token, + start_device_flow, +) + + +# --------------------------------------------------------------------------- +# config.py tests +# --------------------------------------------------------------------------- + + +class TestConfig: + """Test configuration helpers.""" + + def test_config_has_required_keys(self): + assert "device_code_url" in config.GITHUB_MODELS_OAUTH_CONFIG + assert "access_token_url" in config.GITHUB_MODELS_OAUTH_CONFIG + assert "api_base_url" in config.GITHUB_MODELS_OAUTH_CONFIG + assert "prefix" in config.GITHUB_MODELS_OAUTH_CONFIG + assert "scope" in config.GITHUB_MODELS_OAUTH_CONFIG + + def test_config_values(self): + cfg = config.GITHUB_MODELS_OAUTH_CONFIG + assert cfg["device_code_url"] == "https://github.com/login/device/code" + assert cfg["access_token_url"] == "https://github.com/login/oauth/access_token" + assert cfg["api_base_url"] == "https://models.github.ai/inference" + assert cfg["prefix"] == "github-" + assert cfg["scope"] == "read:user" + + def test_token_storage_path(self): + token_path = config.get_token_storage_path() + assert token_path.name == "github_models_oauth.json" + assert "code_puppy" in str(token_path) + + def test_github_models_path(self): + models_path = config.get_github_models_path() + assert models_path.name == "github_models.json" + assert "code_puppy" in str(models_path) + + def test_get_client_id_default_is_empty(self): + with patch.dict(os.environ, {}, clear=False): + env = dict(os.environ) + env.pop("GITHUB_MODELS_CLIENT_ID", None) + with patch.dict(os.environ, env, clear=True): + client_id = config.get_client_id() + assert client_id == "" + + def test_get_client_id_from_env(self): + with patch.dict(os.environ, {"GITHUB_MODELS_CLIENT_ID": "custom_id_123"}): + client_id = config.get_client_id() + assert client_id == "custom_id_123" + + +# --------------------------------------------------------------------------- +# utils.py tests +# --------------------------------------------------------------------------- + + +class TestGhCliToken: + """Test gh CLI token detection.""" + + def test_get_gh_cli_token_success(self): + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = "gho_abc123_from_cli\n" + + with patch( + "code_puppy.plugins.github_models_oauth.utils.subprocess.run", + return_value=mock_result, + ): + token = utils.get_gh_cli_token() + assert token == "gho_abc123_from_cli" + + def test_get_gh_cli_token_not_installed(self): + with patch( + "code_puppy.plugins.github_models_oauth.utils.subprocess.run", + side_effect=FileNotFoundError("gh not found"), + ): + token = utils.get_gh_cli_token() + assert token is None + + def test_get_gh_cli_token_not_logged_in(self): + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stdout = "" + + with patch( + "code_puppy.plugins.github_models_oauth.utils.subprocess.run", + return_value=mock_result, + ): + token = utils.get_gh_cli_token() + assert token is None + + def test_get_gh_cli_token_empty_output(self): + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = "" + + with patch( + "code_puppy.plugins.github_models_oauth.utils.subprocess.run", + return_value=mock_result, + ): + token = utils.get_gh_cli_token() + assert token is None + + +class TestEnvToken: + """Test GITHUB_TOKEN / GH_TOKEN env var detection.""" + + def test_get_env_token_github_token(self): + with patch.dict(os.environ, {"GITHUB_TOKEN": "ghp_abc123"}, clear=False): + token = utils.get_env_token() + assert token == "ghp_abc123" + + def test_get_env_token_gh_token(self): + env = dict(os.environ) + env.pop("GITHUB_TOKEN", None) + env["GH_TOKEN"] = "ghp_from_gh" + with patch.dict(os.environ, env, clear=True): + token = utils.get_env_token() + assert token == "ghp_from_gh" + + def test_get_env_token_prefers_github_token(self): + with patch.dict( + os.environ, + {"GITHUB_TOKEN": "ghp_first", "GH_TOKEN": "ghp_second"}, + clear=False, + ): + token = utils.get_env_token() + assert token == "ghp_first" + + def test_get_env_token_returns_none_when_unset(self): + env = dict(os.environ) + env.pop("GITHUB_TOKEN", None) + env.pop("GH_TOKEN", None) + with patch.dict(os.environ, env, clear=True): + token = utils.get_env_token() + assert token is None + + def test_get_env_token_ignores_empty(self): + with patch.dict( + os.environ, {"GITHUB_TOKEN": "", "GH_TOKEN": ""}, clear=False + ): + # Remove any pre-existing values + env = dict(os.environ) + env["GITHUB_TOKEN"] = "" + env["GH_TOKEN"] = "" + with patch.dict(os.environ, env, clear=True): + token = utils.get_env_token() + assert token is None + + + +class TestPromptForToken: + """Test interactive PAT prompt.""" + + def test_prompt_returns_valid_token(self): + with patch( + "code_puppy.plugins.github_models_oauth.utils.getpass.getpass", + return_value="ghp_validtoken123", + ): + token = utils.prompt_for_token() + assert token == "ghp_validtoken123" + + def test_prompt_returns_none_on_empty(self): + with patch( + "code_puppy.plugins.github_models_oauth.utils.getpass.getpass", + return_value="", + ): + token = utils.prompt_for_token() + assert token is None + + def test_prompt_returns_none_on_keyboard_interrupt(self): + with patch( + "code_puppy.plugins.github_models_oauth.utils.getpass.getpass", + side_effect=KeyboardInterrupt, + ): + token = utils.prompt_for_token() + assert token is None + + +class TestTokenPersistence: + """Test token save/load operations.""" + + def test_save_and_load_tokens(self, tmp_path): + token_file = tmp_path / "github_models_oauth.json" + tokens = {"access_token": "gho_test123", "username": "testuser"} + + with patch( + "code_puppy.plugins.github_models_oauth.utils.get_token_storage_path", + return_value=token_file, + ): + assert utils.save_tokens(tokens) is True + loaded = utils.load_stored_tokens() + assert loaded is not None + assert loaded["access_token"] == "gho_test123" + assert loaded["username"] == "testuser" + + def test_save_tokens_rejects_none(self): + with pytest.raises(TypeError, match="cannot be None"): + utils.save_tokens(None) + + def test_load_tokens_returns_none_when_missing(self, tmp_path): + token_file = tmp_path / "nonexistent.json" + with patch( + "code_puppy.plugins.github_models_oauth.utils.get_token_storage_path", + return_value=token_file, + ): + assert utils.load_stored_tokens() is None + + def test_save_tokens_sets_permissions(self, tmp_path): + token_file = tmp_path / "github_models_oauth.json" + tokens = {"access_token": "gho_test"} + + with patch( + "code_puppy.plugins.github_models_oauth.utils.get_token_storage_path", + return_value=token_file, + ): + utils.save_tokens(tokens) + # Check file permissions (owner read/write only) + mode = oct(token_file.stat().st_mode)[-3:] + assert mode == "600" + + +class TestGitHubUsername: + """Test GitHub user API calls.""" + + def test_get_username_success(self): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"login": "octocat", "name": "Octo Cat"} + + with patch("code_puppy.plugins.github_models_oauth.utils.requests.get", return_value=mock_response): + username = utils.get_github_username("gho_test") + assert username == "octocat" + + def test_get_username_failure(self): + mock_response = MagicMock() + mock_response.status_code = 401 + + with patch("code_puppy.plugins.github_models_oauth.utils.requests.get", return_value=mock_response): + username = utils.get_github_username("bad_token") + assert username is None + + +class TestModelConfig: + """Test model configuration management.""" + + def test_add_and_load_models(self, tmp_path): + models_file = tmp_path / "github_models.json" + + with patch( + "code_puppy.plugins.github_models_oauth.utils.get_github_models_path", + return_value=models_file, + ): + result = utils.add_models_to_config(["openai/gpt-4.1", "meta/llama-4-scout"]) + assert result is True + + config_data = utils.load_github_models_config() + assert "github-openai-gpt-4.1" in config_data + assert "github-meta-llama-4-scout" in config_data + assert config_data["github-openai-gpt-4.1"]["type"] == "github_models" + assert config_data["github-openai-gpt-4.1"]["name"] == "openai/gpt-4.1" + assert config_data["github-openai-gpt-4.1"]["oauth_source"] == "github-models-plugin" + + def test_remove_models(self, tmp_path): + models_file = tmp_path / "github_models.json" + + with patch( + "code_puppy.plugins.github_models_oauth.utils.get_github_models_path", + return_value=models_file, + ): + utils.add_models_to_config(["openai/gpt-4.1"]) + removed = utils.remove_github_models() + assert removed == 1 + + config_data = utils.load_github_models_config() + assert len(config_data) == 0 + + def test_load_empty_config(self, tmp_path): + models_file = tmp_path / "nonexistent.json" + with patch( + "code_puppy.plugins.github_models_oauth.utils.get_github_models_path", + return_value=models_file, + ): + assert utils.load_github_models_config() == {} + + +class TestFetchModels: + """Test model catalog fetching.""" + + def test_fetch_models_success(self): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = [ + {"id": "openai/gpt-4.1"}, + {"id": "meta/llama-4-scout"}, + ] + + with patch("code_puppy.plugins.github_models_oauth.utils.requests.get", return_value=mock_response): + models = utils.fetch_github_models("gho_test") + assert "openai/gpt-4.1" in models + assert "meta/llama-4-scout" in models + + def test_fetch_models_fallback_on_failure(self): + mock_response = MagicMock() + mock_response.status_code = 404 + + with patch("code_puppy.plugins.github_models_oauth.utils.requests.get", return_value=mock_response): + models = utils.fetch_github_models("gho_test") + assert models == utils.DEFAULT_GITHUB_MODELS + + def test_fetch_models_fallback_on_timeout(self): + import requests as req_lib + + with patch( + "code_puppy.plugins.github_models_oauth.utils.requests.get", + side_effect=req_lib.exceptions.Timeout("timed out"), + ): + models = utils.fetch_github_models("gho_test") + assert models == utils.DEFAULT_GITHUB_MODELS + + +# --------------------------------------------------------------------------- +# device_flow.py tests +# --------------------------------------------------------------------------- + + +class TestDeviceFlow: + """Test the GitHub OAuth device flow.""" + + def test_start_device_flow_success(self): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.raise_for_status = MagicMock() + mock_response.json.return_value = { + "device_code": "dc_abc123", + "user_code": "ABCD-1234", + "verification_uri": "https://github.com/login/device", + "expires_in": 900, + "interval": 5, + } + + with patch("code_puppy.plugins.github_models_oauth.device_flow.requests.post", return_value=mock_response): + result = start_device_flow() + assert result is not None + assert result.device_code == "dc_abc123" + assert result.user_code == "ABCD-1234" + assert result.verification_uri == "https://github.com/login/device" + assert result.interval == 5 + + def test_start_device_flow_network_error(self): + import requests as req_lib + + with patch( + "code_puppy.plugins.github_models_oauth.device_flow.requests.post", + side_effect=req_lib.exceptions.ConnectionError("no network"), + ): + result = start_device_flow() + assert result is None + + def test_poll_success_immediate(self): + mock_response = MagicMock() + mock_response.json.return_value = {"access_token": "gho_success_token"} + + with patch("code_puppy.plugins.github_models_oauth.device_flow.requests.post", return_value=mock_response): + with patch("code_puppy.plugins.github_models_oauth.device_flow.time.sleep"): + token = poll_for_access_token("dc_test", interval=1) + assert token == "gho_success_token" + + def test_poll_handles_authorization_pending(self): + pending_response = MagicMock() + pending_response.json.return_value = {"error": "authorization_pending"} + + success_response = MagicMock() + success_response.json.return_value = {"access_token": "gho_after_wait"} + + with patch( + "code_puppy.plugins.github_models_oauth.device_flow.requests.post", + side_effect=[pending_response, pending_response, success_response], + ): + with patch("code_puppy.plugins.github_models_oauth.device_flow.time.sleep"): + token = poll_for_access_token("dc_test", interval=1) + assert token == "gho_after_wait" + + def test_poll_handles_slow_down(self): + slow_response = MagicMock() + slow_response.json.return_value = {"error": "slow_down", "interval": 10} + + success_response = MagicMock() + success_response.json.return_value = {"access_token": "gho_slowed"} + + with patch( + "code_puppy.plugins.github_models_oauth.device_flow.requests.post", + side_effect=[slow_response, success_response], + ): + with patch("code_puppy.plugins.github_models_oauth.device_flow.time.sleep"): + token = poll_for_access_token("dc_test", interval=1) + assert token == "gho_slowed" + + def test_poll_handles_expired_token(self): + expired_response = MagicMock() + expired_response.json.return_value = {"error": "expired_token"} + + with patch( + "code_puppy.plugins.github_models_oauth.device_flow.requests.post", + return_value=expired_response, + ): + with patch("code_puppy.plugins.github_models_oauth.device_flow.time.sleep"): + token = poll_for_access_token("dc_test", interval=1) + assert token is None + + def test_poll_handles_unknown_error(self): + error_response = MagicMock() + error_response.json.return_value = { + "error": "access_denied", + "error_description": "User denied access", + } + + with patch( + "code_puppy.plugins.github_models_oauth.device_flow.requests.post", + return_value=error_response, + ): + with patch("code_puppy.plugins.github_models_oauth.device_flow.time.sleep"): + token = poll_for_access_token("dc_test", interval=1) + assert token is None + + +# --------------------------------------------------------------------------- +# register_callbacks.py tests +# --------------------------------------------------------------------------- + + +class TestCustomCommands: + """Test slash command routing.""" + + def test_github_auth_returns_true(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_custom_command, + ) + + with patch( + "code_puppy.plugins.github_models_oauth.register_callbacks._handle_auth" + ): + with patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.set_model_and_reload_agent" + ): + result = _handle_custom_command("/github-auth", "github-auth") + assert result is True + + def test_github_status_returns_true(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_custom_command, + ) + + with patch( + "code_puppy.plugins.github_models_oauth.register_callbacks._handle_status" + ): + result = _handle_custom_command("/github-status", "github-status") + assert result is True + + def test_github_logout_returns_true(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_custom_command, + ) + + with patch( + "code_puppy.plugins.github_models_oauth.register_callbacks._handle_logout" + ): + result = _handle_custom_command("/github-logout", "github-logout") + assert result is True + + def test_unknown_command_returns_none(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_custom_command, + ) + + result = _handle_custom_command("/something-else", "something-else") + assert result is None + + def test_empty_name_returns_none(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_custom_command, + ) + + result = _handle_custom_command("", "") + assert result is None + + +class TestModelTypeHandler: + """Test the model type registration.""" + + def test_register_returns_github_models_type(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _register_model_types, + ) + + result = _register_model_types() + assert len(result) == 2 + types = {r["type"] for r in result} + assert "github_models" in types + assert "github_copilot" in types + for r in result: + assert callable(r["handler"]) + + def test_model_handler_returns_none_without_token(self, tmp_path): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _create_github_models_model, + ) + + token_file = tmp_path / "no_tokens.json" + with patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.load_stored_tokens", + return_value=None, + ): + model = _create_github_models_model( + "github-openai-gpt-4.1", + {"name": "openai/gpt-4.1", "custom_endpoint": {"url": "https://models.github.ai/inference"}}, + {}, + ) + assert model is None + + +class TestCustomHelp: + """Test the help entries.""" + + def test_help_entries(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _custom_help, + ) + + entries = _custom_help() + names = [e[0] for e in entries] + assert "github-auth" in names + assert "github-status" in names + assert "github-logout" in names + + +class TestAuthFlow: + """Test the multi-auth _handle_auth flow.""" + + def test_auth_uses_gh_cli_token_when_available(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_auth, + ) + + with ( + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_gh_cli_token", + return_value="gho_from_cli", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.load_stored_tokens", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_github_username", + return_value="octocat", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.save_tokens", + return_value=True, + ) as mock_save, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_github_models", + return_value=["openai/gpt-4.1"], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_models_to_config", + return_value=True, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.run_device_flow" + ) as mock_device_flow, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_copilot_models", + return_value=[], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_copilot_models_to_config", + return_value=True, + ), + ): + _handle_auth() + mock_device_flow.assert_not_called() + mock_save.assert_called_once() + saved_data = mock_save.call_args[0][0] + assert saved_data["access_token"] == "gho_from_cli" + assert saved_data["username"] == "octocat" + + def test_auth_uses_env_token_when_gh_cli_unavailable(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_auth, + ) + + with ( + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_gh_cli_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_env_token", + return_value="ghp_from_env", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.load_stored_tokens", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_github_username", + return_value="envuser", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.save_tokens", + return_value=True, + ) as mock_save, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_github_models", + return_value=["openai/gpt-4.1"], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_models_to_config", + return_value=True, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.prompt_for_token", + ) as mock_prompt, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_copilot_models", + return_value=[], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_copilot_models_to_config", + return_value=True, + ), + ): + _handle_auth() + mock_prompt.assert_not_called() + saved_data = mock_save.call_args[0][0] + assert saved_data["access_token"] == "ghp_from_env" + + def test_auth_prompts_for_pat_when_auto_methods_fail(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_auth, + ) + + with ( + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_gh_cli_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_env_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.prompt_for_token", + return_value="ghp_pasted", + ) as mock_prompt, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.load_stored_tokens", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_github_username", + return_value="pasteuser", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.save_tokens", + return_value=True, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_github_models", + return_value=["openai/gpt-4.1"], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_models_to_config", + return_value=True, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_copilot_models", + return_value=[], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_copilot_models_to_config", + return_value=True, + ), + ): + _handle_auth() + mock_prompt.assert_called_once() + + def test_auth_falls_back_to_device_flow(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_auth, + ) + + with ( + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_gh_cli_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_env_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.prompt_for_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_client_id", + return_value="Iv1.real_client_id", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.load_stored_tokens", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.run_device_flow", + return_value="gho_from_device_flow", + ) as mock_device_flow, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_github_username", + return_value="devuser", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.save_tokens", + return_value=True, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_github_models", + return_value=["openai/gpt-4.1"], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_models_to_config", + return_value=True, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.fetch_copilot_models", + return_value=[], + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.add_copilot_models_to_config", + return_value=True, + ), + ): + _handle_auth() + mock_device_flow.assert_called_once() + + def test_auth_shows_error_when_all_methods_fail(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_auth, + ) + + with ( + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_gh_cli_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_env_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.prompt_for_token", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_client_id", + return_value="", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.load_stored_tokens", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.emit_error", + ) as mock_emit_error, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.save_tokens", + ) as mock_save, + ): + _handle_auth() + mock_emit_error.assert_called() + error_msg = mock_emit_error.call_args[0][0] + assert "Authentication failed" in error_msg + mock_save.assert_not_called() + + def test_auth_rejects_invalid_token(self): + from code_puppy.plugins.github_models_oauth.register_callbacks import ( + _handle_auth, + ) + + with ( + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_gh_cli_token", + return_value="bad_token", + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.load_stored_tokens", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.get_github_username", + return_value=None, + ), + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.emit_error", + ) as mock_emit_error, + patch( + "code_puppy.plugins.github_models_oauth.register_callbacks.save_tokens", + ) as mock_save, + ): + _handle_auth() + mock_emit_error.assert_called() + error_msg = mock_emit_error.call_args[0][0] + assert "Token validation failed" in error_msg + mock_save.assert_not_called() diff --git a/code_puppy/plugins/github_models_oauth/utils.py b/code_puppy/plugins/github_models_oauth/utils.py new file mode 100644 index 000000000..aaeb51ff6 --- /dev/null +++ b/code_puppy/plugins/github_models_oauth/utils.py @@ -0,0 +1,315 @@ +"""Utility helpers for the GitHub Models OAuth plugin.""" + +from __future__ import annotations + +import getpass +import json +import logging +import os +import subprocess +from typing import Any, Dict, List, Optional + +import requests + +from .config import ( + GITHUB_MODELS_OAUTH_CONFIG, + get_github_models_path, + get_token_storage_path, +) + +logger = logging.getLogger(__name__) + + +def get_gh_cli_token() -> Optional[str]: + """Get a GitHub token from the ``gh`` CLI if installed and authenticated.""" + try: + result = subprocess.run( + ["gh", "auth", "token"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0 and result.stdout.strip(): + return result.stdout.strip() + except FileNotFoundError: + logger.debug("gh CLI not found in PATH") + except Exception as exc: + logger.debug("Failed to get gh CLI token: %s", exc) + return None + + +def get_env_token() -> Optional[str]: + """Get a GitHub token from ``GITHUB_TOKEN`` or ``GH_TOKEN``.""" + for var in ("GITHUB_TOKEN", "GH_TOKEN"): + token = os.environ.get(var, "").strip() + if token: + return token + return None + + +def prompt_for_token() -> Optional[str]: + """Prompt the user to paste a GitHub PAT (hidden input via ``getpass``).""" + from code_puppy.command_line.utils import _reset_windows_console + from code_puppy.messaging import emit_info + + emit_info( + "πŸ”‘ Create a Personal Access Token at:\n" + " https://github.com/settings/tokens\n" + " Fine-grained PAT with 'models:read' permission recommended.\n" + " Classic PATs also work without specific scopes." + ) + try: + _reset_windows_console() + token = getpass.getpass("Paste your GitHub token (hidden) or press Enter to skip: ") + if token and len(token) >= 4: + return token.strip() + except (EOFError, KeyboardInterrupt): + pass + return None + + +# Fallback model lists used when the catalog/API is unreachable. +DEFAULT_GITHUB_MODELS: List[str] = [ + "openai/gpt-5.4", + "openai/gpt-5.4-mini", + "openai/gpt-4.1", + "openai/gpt-4.1-mini", + "openai/gpt-4.1-nano", + "openai/o3", + "openai/o4-mini", + "meta/llama-4-scout-17b-16e-instruct", + "meta/llama-4-maverick-17b-128e-instruct-fp8", + "mistral-ai/mistral-medium-2505", + "deepseek/deepseek-r1", + "deepseek/deepseek-v3-0324", + "xai/grok-3", +] + +DEFAULT_COPILOT_MODELS: List[str] = [ + "claude-sonnet-4.6", + "claude-opus-4.6", + "claude-haiku-4.5", + "gpt-5.4", + "gemini-3-pro", + "gemini-3-flash-preview", +] + + +def save_tokens(tokens: Dict[str, Any]) -> bool: + """Save OAuth tokens to disk with restrictive permissions.""" + if tokens is None: + raise TypeError("tokens cannot be None") + try: + token_path = get_token_storage_path() + with open(token_path, "w", encoding="utf-8") as fh: + json.dump(tokens, fh, indent=2) + token_path.chmod(0o600) + return True + except Exception as exc: + logger.error("Failed to save GitHub tokens: %s", exc) + return False + + +def load_stored_tokens() -> Optional[Dict[str, Any]]: + """Load previously stored OAuth tokens from disk.""" + try: + token_path = get_token_storage_path() + if token_path.exists(): + with open(token_path, "r", encoding="utf-8") as fh: + return json.load(fh) + except Exception as exc: + logger.error("Failed to load GitHub tokens: %s", exc) + return None + + +def get_github_username(access_token: str) -> Optional[str]: + """Fetch the authenticated user's GitHub login.""" + try: + response = requests.get( + GITHUB_MODELS_OAUTH_CONFIG["user_api_url"], + headers={ + "Authorization": f"Bearer {access_token}", + "Accept": "application/json", + "User-Agent": GITHUB_MODELS_OAUTH_CONFIG["user_agent"], + }, + timeout=15, + ) + if response.status_code == 200: + return response.json().get("login") + except Exception as exc: + logger.warning("Failed to fetch GitHub user: %s", exc) + return None + + +def _parse_model_list(data: Any) -> List[str]: + """Extract model IDs from a catalog or API response.""" + items = data if isinstance(data, list) else data.get("models", data.get("data", [])) + models: List[str] = [] + for item in items: + if isinstance(item, str): + models.append(item) + elif isinstance(item, dict): + mid = item.get("id") or item.get("name") or item.get("model") + if mid and "embedding" not in mid and "accounts/" not in mid: + models.append(mid) + return models + + +def fetch_github_models(access_token: str) -> List[str]: + """Fetch models from the GitHub Models catalog, falling back to defaults.""" + from code_puppy.messaging import emit_info, emit_warning + + catalog_url = ( + GITHUB_MODELS_OAUTH_CONFIG["api_base_url"].rstrip("/").replace("/inference", "") + + "/catalog/models" + ) + try: + response = requests.get( + catalog_url, + headers={ + "Authorization": f"Bearer {access_token}", + "Accept": "application/json", + "User-Agent": GITHUB_MODELS_OAUTH_CONFIG["user_agent"], + }, + timeout=30, + ) + if response.status_code in (401, 403): + emit_warning( + f" Catalog returned HTTP {response.status_code} β€” token lacks access.\n" + " πŸ’‘ Fine-grained PATs need 'models:read' permission.\n" + " Skipping GitHub Models registration." + ) + return [] + if response.status_code == 200: + models = _parse_model_list(response.json()) + if models: + emit_info(f" πŸ“‘ Fetched {len(models)} models from GitHub catalog") + return models + emit_warning( + f" Catalog returned HTTP {response.status_code}; using {len(DEFAULT_GITHUB_MODELS)} built-in models.\n" + " πŸ’‘ For the full list, use a PAT or re-run: gh auth login -s read:user" + ) + except Exception as exc: + logger.warning("Error fetching GitHub Models catalog: %s", exc) + emit_warning(f" Catalog unavailable; using {len(DEFAULT_GITHUB_MODELS)} built-in models") + + return list(DEFAULT_GITHUB_MODELS) + + +def fetch_copilot_models(access_token: str) -> List[str]: + """Fetch models from the GitHub Copilot API, falling back to defaults.""" + from code_puppy.messaging import emit_info, emit_warning + + copilot_url = GITHUB_MODELS_OAUTH_CONFIG["copilot_api_base_url"] + try: + response = requests.get( + f"{copilot_url}/models", + headers={ + "Authorization": f"Bearer {access_token}", + "Accept": "application/json", + "Copilot-Integration-Id": GITHUB_MODELS_OAUTH_CONFIG["copilot_integration_id"], + "User-Agent": GITHUB_MODELS_OAUTH_CONFIG["user_agent"], + }, + timeout=30, + ) + if response.status_code in (401, 403): + emit_warning( + f" Copilot API returned HTTP {response.status_code} β€” token lacks access.\n" + " Skipping Copilot models registration." + ) + return [] + if response.status_code == 200: + models = _parse_model_list(response.json()) + if models: + emit_info(f" πŸ“‘ Fetched {len(models)} models from Copilot API") + return models + emit_warning(f" Copilot API returned HTTP {response.status_code}; using {len(DEFAULT_COPILOT_MODELS)} built-in models") + except Exception as exc: + logger.warning("Error fetching Copilot models: %s", exc) + emit_warning(f" Copilot API unavailable; using {len(DEFAULT_COPILOT_MODELS)} built-in models") + + return list(DEFAULT_COPILOT_MODELS) + + +def add_models_to_config(model_ids: List[str]) -> bool: + """Register GitHub Models (``github-`` prefix, ``github_models`` type).""" + return _add_models_to_config( + model_ids, + model_type="github_models", + prefix=GITHUB_MODELS_OAUTH_CONFIG["prefix"], + base_url=GITHUB_MODELS_OAUTH_CONFIG["api_base_url"], + ) + + +def add_copilot_models_to_config(model_ids: List[str]) -> bool: + """Register Copilot API models (``copilot-`` prefix, ``github_copilot`` type).""" + return _add_models_to_config( + model_ids, + model_type="github_copilot", + prefix=GITHUB_MODELS_OAUTH_CONFIG["copilot_prefix"], + base_url=GITHUB_MODELS_OAUTH_CONFIG["copilot_api_base_url"], + ) + + +def _add_models_to_config( + model_ids: List[str], *, model_type: str, prefix: str, base_url: str, +) -> bool: + """Register models in the local config file with the given type and prefix.""" + default_ctx = GITHUB_MODELS_OAUTH_CONFIG["default_context_length"] + try: + config = load_github_models_config() + for model_id in model_ids: + config[f"{prefix}{model_id.replace('/', '-')}"] = { + "type": model_type, + "name": model_id, + "custom_endpoint": {"url": base_url}, + "context_length": default_ctx, + "oauth_source": "github-models-plugin", + "supported_settings": ["temperature", "top_p"], + } + if save_github_models_config(config): + return True + except Exception as exc: + logger.error("Error adding %s models to config: %s", model_type, exc) + return False + + +def load_github_models_config() -> Dict[str, Any]: + """Load model configurations from the github_models.json file.""" + try: + models_path = get_github_models_path() + if models_path.exists(): + with open(models_path, "r", encoding="utf-8") as fh: + return json.load(fh) + except Exception as exc: + logger.error("Failed to load GitHub models config: %s", exc) + return {} + + +def save_github_models_config(models: Dict[str, Any]) -> bool: + """Save model configurations to disk.""" + try: + models_path = get_github_models_path() + with open(models_path, "w", encoding="utf-8") as fh: + json.dump(models, fh, indent=2) + return True + except Exception as exc: + logger.error("Failed to save GitHub models config: %s", exc) + return False + + +def remove_github_models() -> int: + """Remove all GitHub Models OAuth models from configuration.""" + try: + config = load_github_models_config() + to_remove = [ + name for name, cfg in config.items() + if cfg.get("oauth_source") == "github-models-plugin" + ] + for name in to_remove: + config.pop(name, None) + if save_github_models_config(config): + return len(to_remove) + except Exception as exc: + logger.error("Error removing GitHub models: %s", exc) + return 0