diff --git a/headroom/cli/wrap.py b/headroom/cli/wrap.py index 50f84b1e5..f25a1368b 100644 --- a/headroom/cli/wrap.py +++ b/headroom/cli/wrap.py @@ -58,6 +58,12 @@ from headroom.agent_savings import ( apply_agent_savings_env_defaults, ) from headroom.copilot_auth import ( + _API_TOKEN_ENV_VARS, + _API_TOKEN_EXPIRES_AT_ENV_VAR, + _COPILOT_OAUTH_TOKEN_ENV_VARS, + _GENERIC_GITHUB_TOKEN_ENV_VARS, + _REFRESH_OAUTH_TOKEN_ENV_VAR, + CopilotSubscriptionTokenResolution, has_oauth_auth, resolve_client_bearer_token, resolve_copilot_api_url, @@ -167,8 +173,15 @@ from .main import main _COPILOT_PROXY_SEED_ENV_VARS = ( "GITHUB_COPILOT_API_TOKEN", - "GITHUB_COPILOT_REFRESH_OAUTH_TOKEN", - "GITHUB_COPILOT_API_TOKEN_EXPIRES_AT", + _REFRESH_OAUTH_TOKEN_ENV_VAR, + _API_TOKEN_EXPIRES_AT_ENV_VAR, +) +_COPILOT_SUBSCRIPTION_LAUNCH_SECRET_ENV_VARS = ( + *_API_TOKEN_ENV_VARS, + _REFRESH_OAUTH_TOKEN_ENV_VAR, + _API_TOKEN_EXPIRES_AT_ENV_VAR, + *_COPILOT_OAUTH_TOKEN_ENV_VARS, + *_GENERIC_GITHUB_TOKEN_ENV_VARS, ) @@ -177,6 +190,11 @@ def _scrub_copilot_proxy_seed_env(env: dict[str, str]) -> None: env.pop(key, None) +def _scrub_copilot_subscription_launch_env(env: dict[str, str]) -> None: + for key in _COPILOT_SUBSCRIPTION_LAUNCH_SECRET_ENV_VARS: + env.pop(key, None) + + def _read_text(path: Path) -> str: """Read a text file as UTF-8, falling back to the system locale encoding.""" return fsutil.read_text(path) @@ -3982,8 +4000,10 @@ def _ensure_proxy( ) -> tuple[subprocess.Popen | None, int]: """Start or verify proxy. Returns (process_handle, actual_port).""" helpers = _live_wrap_module() - copilot_subscription_seed_requested = bool(copilot_refresh_oauth_token) or ( - copilot_api_token_expires_at is not None + copilot_subscription_seed_requested = ( + bool(copilot_api_token) + or bool(copilot_refresh_oauth_token) + or copilot_api_token_expires_at is not None ) # --no-proxy reuses an already-running proxy, so backend/region/provider # flags (which only apply when we start one) would be silently dropped. @@ -3999,7 +4019,7 @@ def _ensure_proxy( ) if isolated_copilot_subscription_proxy: click.echo( - " Copilot subscription refresh seeds are session-specific; " + " Copilot subscription seeds are session-specific; " "starting a dedicated local proxy instance for this wrap session." ) if not isolated_copilot_subscription_proxy and manifest is not None: @@ -5401,6 +5421,17 @@ def unwrap_claude( # ============================================================================= +def _require_copilot_subscription_resolution() -> CopilotSubscriptionTokenResolution: + resolution = resolve_subscription_bearer_token_details() + if resolution is None: + raise click.ClickException( + "GitHub Copilot subscription mode requires a reusable GitHub/Copilot bearer " + "token, but none could be resolved. Run `headroom copilot-auth login` first, or set " + "GITHUB_COPILOT_TOKEN / GITHUB_COPILOT_GITHUB_TOKEN." + ) + return resolution + + @wrap.command(context_settings={"ignore_unknown_options": True}) @_rtk_option @click.option( @@ -5543,7 +5574,8 @@ def copilot( copilot_proxy_token: str | None = None copilot_refresh_oauth_token: str | None = None copilot_api_token_expires_at: float | None = None - subscription_resolution = None + client_bearer: str | None = None + subscription_resolution: CopilotSubscriptionTokenResolution | None = None if _should_use_copilot_oauth( backend=effective_backend, provider_type=provider_type, @@ -5551,10 +5583,8 @@ def copilot( force_subscription=subscription, ): if subscription: - subscription_resolution = resolve_subscription_bearer_token_details() - client_bearer = ( - subscription_resolution.token if subscription_resolution is not None else None - ) + subscription_resolution = _require_copilot_subscription_resolution() + client_bearer = subscription_resolution.token else: client_bearer = resolve_client_bearer_token() if not client_bearer: @@ -7646,6 +7676,11 @@ def openclaw( help="Enable code graph indexing via codebase-memory-mcp (optional)", ) @click.option("--no-proxy", is_flag=True, help="Skip proxy startup (use existing proxy)") +@click.option( + "--copilot-subscription", + is_flag=True, + help="Route headroom/* models through the authenticated GitHub Copilot subscription", +) @click.option("--learn", is_flag=True, help="Enable live traffic learning") @click.option("--memory", is_flag=True, help="Enable persistent cross-session memory") @click.option( @@ -7664,6 +7699,7 @@ def opencode( no_serena: bool, code_graph: bool, no_proxy: bool, + copilot_subscription: bool, learn: bool, memory: bool, backend: str | None, @@ -7690,7 +7726,28 @@ def opencode( headroom wrap opencode --no-serena # Skip Serena MCP registration headroom wrap opencode --port 9999 # Custom proxy port headroom wrap opencode --backend anyllm --anyllm-provider groq + headroom wrap opencode --copilot-subscription # Use a GitHub Copilot subscription """ + subscription_resolution = None + if copilot_subscription: + effective_backend = backend or os.environ.get("HEADROOM_BACKEND") + if effective_backend not in (None, "", "anthropic"): + raise click.ClickException( + "--copilot-subscription cannot be combined with translated backends " + "such as anyllm or litellm-*; use the anthropic backend." + ) + if no_proxy: + raise click.ClickException( + "--copilot-subscription cannot be combined with --no-proxy because " + "it requires a private seeded proxy." + ) + if prepare_only: + raise click.ClickException( + "--copilot-subscription cannot be combined with --prepare-only because " + "it requires a running private seeded proxy." + ) + subscription_resolution = _require_copilot_subscription_resolution() + # Snapshot OpenCode config.json BEFORE any wrap-time mutation so # `headroom unwrap opencode` can restore the user's pre-wrap state. _opencode_config_file, _opencode_backup_file = opencode_config_paths() @@ -7770,33 +7827,44 @@ def opencode( backend=backend, anyllm_provider=anyllm_provider, region=region, + openai_api_url=(subscription_resolution.api_url if subscription_resolution else None), + copilot_api_token=(subscription_resolution.token if subscription_resolution else None), + copilot_refresh_oauth_token=( + subscription_resolution.refresh_oauth_token if subscription_resolution else None + ), + copilot_api_token_expires_at=( + subscription_resolution.api_token_expires_at if subscription_resolution else None + ), ) - # If the proxy fell back to a different port, move our marker so - # cleanup tracking stays accurate and update MCP config. - if actual_port != port: - _unregister_proxy_client(port) - _register_proxy_client(actual_port) - if not no_mcp: - from headroom.mcp_registry import OpencodeRegistrar + try: + # If the proxy fell back to a different port, move our marker so + # cleanup tracking stays accurate and update MCP config. + if actual_port != port: + _unregister_proxy_client(port) + _register_proxy_client(actual_port) + if not no_mcp: + from headroom.mcp_registry import OpencodeRegistrar - _setup_headroom_mcp(OpencodeRegistrar(), actual_port, verbose=verbose, force=True) + _setup_headroom_mcp(OpencodeRegistrar(), actual_port, verbose=verbose, force=True) - env, env_vars_display = _build_opencode_launch_env( - actual_port, os.environ, project=_project_name_from_cwd(), include_mcp=not no_mcp - ) - - # Inject Headroom provider into OpenCode config so traffic routes through proxy. - inject_opencode_provider_config(actual_port) - if memory: - mem_dir = Path.cwd() / ".headroom" - _inject_memory_mcp_config( - os.environ.get("USER", os.environ.get("USERNAME", "default")), + launch_environ = os.environ.copy() + if subscription_resolution is not None: + _scrub_copilot_subscription_launch_env(launch_environ) + env, env_vars_display = _build_opencode_launch_env( + actual_port, launch_environ, project=_project_name_from_cwd(), include_mcp=not no_mcp ) - # Proxy already started by _ensure_proxy above; tell _launch_tool to - # skip duplicate startup. - try: + # Inject Headroom provider into OpenCode config so traffic routes through proxy. + inject_opencode_provider_config(actual_port) + if memory: + mem_dir = Path.cwd() / ".headroom" + _inject_memory_mcp_config( + os.environ.get("USER", os.environ.get("USERNAME", "default")), + ) + + # Proxy already started by _ensure_proxy above; tell _launch_tool to + # skip duplicate startup. _launch_tool( binary=opencode_bin, args=opencode_args, diff --git a/tests/test_cli/test_wrap_opencode.py b/tests/test_cli/test_wrap_opencode.py index 6eaf8d550..a7cb816de 100644 --- a/tests/test_cli/test_wrap_opencode.py +++ b/tests/test_cli/test_wrap_opencode.py @@ -11,6 +11,7 @@ from click.testing import CliRunner from headroom.cli import wrap as wrap_mod from headroom.cli.main import main +from headroom.copilot_auth import CopilotSubscriptionTokenResolution @pytest.fixture(autouse=True) @@ -32,11 +33,225 @@ def _set_test_home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: monkeypatch.delenv("OPENCODE_CONFIG", raising=False) +def _subscription_resolution() -> CopilotSubscriptionTokenResolution: + return CopilotSubscriptionTokenResolution( + token="copilot-api-secret", + source="test", + confidence="test", + api_url="https://api.githubcopilot.com", + token_fingerprint="sha256:test", + refresh_oauth_token="copilot-refresh-secret", + api_token_expires_at=123.5, + ) + + # --------------------------------------------------------------------------- # Wrap opencode # --------------------------------------------------------------------------- +def test_wrap_opencode_copilot_subscription_handoffs_seed_after_actual_port( + runner: CliRunner, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + _set_test_home(monkeypatch, tmp_path) + monkeypatch.setenv("GITHUB_COPILOT_API_TOKEN", "inherited-api-secret") + monkeypatch.setenv("GITHUB_COPILOT_REFRESH_OAUTH_TOKEN", "inherited-refresh-secret") + monkeypatch.setenv("GITHUB_COPILOT_API_TOKEN_EXPIRES_AT", "999.0") + monkeypatch.setenv("GITHUB_COPILOT_TOKEN", "inherited-seat-token") + monkeypatch.setenv("GITHUB_COPILOT_GITHUB_TOKEN", "inherited-github-token") + monkeypatch.setenv("COPILOT_GITHUB_TOKEN", "inherited-alt-github-token") + monkeypatch.setenv("COPILOT_PROVIDER_BEARER_TOKEN", "inherited-provider-bearer") + monkeypatch.setenv("GH_TOKEN", "inherited-gh-token") + monkeypatch.setenv("GITHUB_TOKEN", "inherited-github-pat") + captured: dict[str, object] = {} + + def fake_ensure_proxy(*args, **kwargs): # noqa: ANN002, ANN003 + captured["ensure"] = kwargs + return None, 9010 + + def fake_launch_tool(**kwargs): # noqa: ANN003 + captured["launch"] = kwargs + + with ( + patch.object(wrap_mod.shutil, "which", return_value="opencode"), + patch.object( + wrap_mod, + "_require_copilot_subscription_resolution", + return_value=_subscription_resolution(), + ), + patch.object(wrap_mod, "_ensure_proxy", side_effect=fake_ensure_proxy), + patch.object(wrap_mod, "_launch_tool", side_effect=fake_launch_tool), + ): + result = runner.invoke( + main, + [ + "wrap", + "opencode", + "--copilot-subscription", + "--no-rtk", + "--no-mcp", + "--no-serena", + ], + ) + + assert result.exit_code == 0, result.output + ensure = captured["ensure"] + assert ensure["openai_api_url"] == "https://api.githubcopilot.com" + assert ensure["copilot_api_token"] == "copilot-api-secret" + assert ensure["copilot_refresh_oauth_token"] == "copilot-refresh-secret" + assert ensure["copilot_api_token_expires_at"] == 123.5 + launch = captured["launch"] + assert launch["port"] == 9010 + assert "copilot-api-secret" not in result.output + assert "copilot-api-secret" not in str(launch["env"]) + assert "copilot-refresh-secret" not in str(launch["env"]) + assert "copilot-api-secret" not in launch["env"]["OPENCODE_CONFIG_CONTENT"] + assert "GITHUB_COPILOT_API_TOKEN" not in launch["env"] + assert "GITHUB_COPILOT_REFRESH_OAUTH_TOKEN" not in launch["env"] + assert "GITHUB_COPILOT_API_TOKEN_EXPIRES_AT" not in launch["env"] + assert "GITHUB_COPILOT_TOKEN" not in launch["env"] + assert "GITHUB_COPILOT_GITHUB_TOKEN" not in launch["env"] + assert "COPILOT_GITHUB_TOKEN" not in launch["env"] + assert "COPILOT_PROVIDER_BEARER_TOKEN" not in launch["env"] + assert "GH_TOKEN" not in launch["env"] + assert "GITHUB_TOKEN" not in launch["env"] + + +@pytest.mark.parametrize( + "extra_args, message", + [ + (["--no-proxy"], "--no-proxy"), + (["--prepare-only"], "--prepare-only"), + (["--backend", "anyllm"], "translated backends"), + ], +) +def test_wrap_opencode_copilot_subscription_rejects_incompatible_modes( + runner: CliRunner, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + extra_args: list[str], + message: str, +) -> None: + monkeypatch.chdir(tmp_path) + _set_test_home(monkeypatch, tmp_path) + config_file = tmp_path / ".config" / "opencode" / "opencode.json" + config_file.parent.mkdir(parents=True) + config_file.write_text("{}", encoding="utf-8") + with patch.object(wrap_mod, "_ensure_proxy", side_effect=AssertionError("proxy launched")): + result = runner.invoke( + main, + ["wrap", "opencode", "--copilot-subscription", "--no-rtk", "--no-mcp", *extra_args], + ) + assert result.exit_code == 1 + assert message in result.output + assert not config_file.with_name("opencode.json.headroom-backup").exists() + + +def test_wrap_opencode_copilot_subscription_rejects_headroom_backend_env( + runner: CliRunner, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + _set_test_home(monkeypatch, tmp_path) + monkeypatch.setenv("HEADROOM_BACKEND", "anyllm") + with patch.object(wrap_mod, "_ensure_proxy", side_effect=AssertionError("proxy launched")): + result = runner.invoke( + main, + ["wrap", "opencode", "--copilot-subscription", "--no-rtk", "--no-mcp"], + ) + assert result.exit_code == 1 + assert "translated backends" in result.output + + +def test_wrap_opencode_copilot_subscription_requires_login_before_launch( + runner: CliRunner, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + _set_test_home(monkeypatch, tmp_path) + with ( + patch.object( + wrap_mod, + "resolve_subscription_bearer_token_details", + return_value=None, + ), + patch.object(wrap_mod, "_ensure_proxy", side_effect=AssertionError("proxy launched")), + ): + result = runner.invoke( + main, + ["wrap", "opencode", "--copilot-subscription", "--no-rtk", "--no-mcp"], + ) + assert result.exit_code == 1 + assert "headroom copilot-auth login" in result.output + + +def test_wrap_opencode_copilot_subscription_cleans_up_proxy_on_config_failure( + runner: CliRunner, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + _set_test_home(monkeypatch, tmp_path) + + class _FakeProxy: + def __init__(self) -> None: + self.terminated = False + self.wait_timeout: float | None = None + + def poll(self) -> None: + return None + + def terminate(self) -> None: + self.terminated = True + + def wait(self, timeout: float | None = None) -> int: + self.wait_timeout = timeout + return 0 + + proxy = _FakeProxy() + + with ( + patch.object(wrap_mod.shutil, "which", return_value="opencode"), + patch.object( + wrap_mod, + "_require_copilot_subscription_resolution", + return_value=_subscription_resolution(), + ), + patch.object(wrap_mod, "_ensure_proxy", return_value=(proxy, 9010)), + patch.object(wrap_mod, "_register_proxy_client"), + patch.object(wrap_mod, "_unregister_proxy_client"), + patch.object(wrap_mod, "_live_proxy_clients", return_value=[]), + patch.object( + wrap_mod, + "inject_opencode_provider_config", + side_effect=RuntimeError("config write failed"), + ), + patch.object(wrap_mod, "_launch_tool", side_effect=AssertionError("launch should not run")), + ): + result = runner.invoke( + main, + [ + "wrap", + "opencode", + "--copilot-subscription", + "--no-rtk", + "--no-mcp", + "--no-serena", + ], + ) + + assert result.exit_code == 1 + assert isinstance(result.exception, RuntimeError) + assert str(result.exception) == "config write failed" + assert proxy.terminated is True + assert proxy.wait_timeout == 5 + + def test_wrap_opencode_sets_config_content_env( runner: CliRunner, tmp_path: Path, diff --git a/tests/test_cli/test_wrap_persistent.py b/tests/test_cli/test_wrap_persistent.py index 3a7793438..740b81afc 100644 --- a/tests/test_cli/test_wrap_persistent.py +++ b/tests/test_cli/test_wrap_persistent.py @@ -417,6 +417,35 @@ def test_ensure_proxy_starts_isolated_ephemeral_proxy_for_copilot_subscription_s assert calls[1][2]["copilot_api_token_expires_at"] == 456.5 +def test_ensure_proxy_isolates_copilot_subscription_seed_with_api_token_only(monkeypatch) -> None: + calls: list[object] = [] + + monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: None) + monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: port == 8787) + monkeypatch.setattr( + wrap_cli, + "_find_available_port", + lambda start_port, **kw: calls.append(("find_port", start_port)) or 8788, + ) + monkeypatch.setattr( + wrap_cli, + "_start_proxy", + lambda *args, **kwargs: calls.append(("start", args, kwargs)), + ) + + proc, actual_port = wrap_cli._ensure_proxy( + 8787, + False, + copilot_api_token="direct-token-only", + ) + + assert proc is None + assert actual_port == 8788 + assert calls[0] == ("find_port", 8788) + assert calls[1][0] == "start" + assert calls[1][2]["copilot_api_token"] == "direct-token-only" + + def test_ensure_proxy_starts_isolated_ephemeral_proxy_when_subscription_seed_targets_persistent_port( monkeypatch, ) -> None: