"""GitHub Copilot is a model provider over the Copilot API. GitHub's device sign-in stores an account, the daemon mints a short-lived Copilot token from the GitHub token, and each model family is served on the route Copilot publishes for it: Chat Completions, OpenAI Responses, or Anthropic Messages. The request carries GitHub's client identity, and the picker lists only the models the account may actually pick. The sign-in writes the daemon's own creds, so the class is exclusive. """ import json import time import unittest import urllib.parse from github_copilot_support import ACCESS_TOKEN, COPILOT_TOKEN, Github from harness import ( Albedo, daemon_resource, Provider, Reply, exclusive_class, exclusive, error, operation_id, text, ) CHAT_MODEL = "gemini-3.7-flash" RESPONSES_MODEL = "gpt-5.4-luna" MESSAGES_MODEL = "claude-haiku-5.5" # The Copilot API's own model list: what the account may pick, and what each # model is served on. API_MODELS = { "data": [ { "id": CHAT_MODEL, "name": "Gemini 3.7 Flash", "version": "2026-05-01", "model_picker_enabled": True, "supported_endpoints": ["/chat/completions"], "capabilities": { "family": "gemini", "limits": {"max_context_window_tokens": 1000000}, "supports": {"tool_calls": True, "streaming": True, "vision": True}, }, }, { "id": RESPONSES_MODEL, "name": "GPT-5.4 Luna", "version": "2026-06-01", "model_picker_enabled": True, "supported_endpoints": ["/responses"], "capabilities": { "family": "gpt", "limits": {"max_context_window_tokens": 400000}, "supports": {"tool_calls": True, "streaming": True}, }, }, { "id": MESSAGES_MODEL, "name": "Claude Haiku 5.5", "version": "2026-07-01", "model_picker_enabled": True, "supported_endpoints": ["/v1/messages"], "capabilities": { "family": "claude", "limits": {"max_context_window_tokens": 200000}, "supports": {"tool_calls": True, "streaming": True}, }, }, { "id": "gpt-4o-mini", "name": "GPT-4o mini", "version": "2024-07-18", "model_picker_enabled": False, "capabilities": {"family": "gpt", "supports": {"tool_calls": False}}, }, ] } # models.dev's `github-copilot` namespace. MODELS_DEV = { "github-copilot": { "id": "github-copilot", "name": "GitHub Copilot", "env": ["GITHUB_TOKEN"], "api": "https://api.githubcopilot.com", "models": { CHAT_MODEL: { "id": CHAT_MODEL, "limit": {"context": 1000000, "output": 64000}, }, RESPONSES_MODEL: { "id": RESPONSES_MODEL, "limit": {"context": 400000, "output": 128000}, }, MESSAGES_MODEL: { "id": MESSAGES_MODEL, "limit": {"context": 200000, "output": 64000}, }, }, } } def anthropic_events(answer, billing=None): """An Anthropic Messages stream that carries `answer` as text. `billing` is the charge that Copilot reports beside the final usage, in nano AI units: billionths of an AI credit, and one credit costs one cent. """ delta = { "type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, "usage": {"output_tokens": 5}, } if billing is not None: delta["copilot_usage"] = {"total_nano_aiu": billing} return [ { "type": "message_start", "message": { "id": "msg_fixture", "usage": {"input_tokens": 10, "output_tokens": 1}, }, }, { "type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}, }, { "type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": answer}, }, {"type": "content_block_stop", "index": 0}, delta, {"type": "message_stop"}, ] @exclusive_class class GithubCopilotTests(unittest.TestCase): def setUp(self) -> None: self.github = daemon_resource("github", lambda: Github().start()) with self.github.lock: self.github.seen.clear() self.github.authorized = False def start(self, model, replies, *, token="github-fixture-seed"): """A Copilot profile pinned at `model` over a scripted Copilot API.""" provider = Provider(scripting(list(replies)), catalog=API_MODELS) self.addCleanup(provider.close) app = Albedo( provider, providers={ "copilot": { "extension": "github-copilot", "baseUrl": provider.url, "model": model, "protocol": "chat_completions", } }, prepare=self.prepare, ) app.__enter__() self.addCleanup(app.__exit__, None, None, None) if token: app.store_secrets( "accounts", { "github-copilot": { "type": "key", "apiKey": token, "label": "copilot fixture token", } }, ) return app, provider def prepare(self, app): (app.home / "models.json").write_text(json.dumps(MODELS_DEV)) app.env["COPILOT_GITHUB_WEB"] = self.github.url app.env["COPILOT_GITHUB_API"] = self.github.url def ask(self, app, session, prompt): app.prompt(session, prompt).close() app.idle(session) def test_device_sign_in_stores_the_account_and_streams_chat_completions(self): app, provider = self.start(CHAT_MODEL, [text("from copilot")], token=None) login = operation_id() with app.api( f"/auth/logins/{login}", {"provider": "github-copilot", "flow": "browser", "values": {}}, method="PUT", ) as response: json.load(response) # The device code arrives once GitHub answered the authorization start. deadline = time.monotonic() + 30 while True: with app.api(f"/auth/logins/{login}") as response: state = json.load(response) if state.get("url") or state["state"] not in ("waiting", "exchanging"): break self.assertLess(time.monotonic(), deadline, state) time.sleep(0.05) if not state.get("url"): self.fail(state) self.assertTrue(state["url"].startswith(self.github.url), state) self.assertIn("ABCD-1234", json.dumps(state), state) self.github.authorize() while state["state"] in ("waiting", "exchanging"): with app.api(f"/auth/logins/{login}") as response: state = json.load(response) self.assertLess(time.monotonic(), deadline, state) time.sleep(0.05) if state["state"] != "complete": self.fail({"state": state, "seen": self.github.seen}) # The account is stored as an OAuth credential, not a token. stored = json.loads((app.home / "creds.json").read_text()) account = stored["accounts"]["github-copilot"] self.assertEqual(account["type"], "oauth", account) self.assertEqual(account["access"], ACCESS_TOKEN, account) self.assertEqual(account["accountId"], "octocat", account) session = app.session() self.ask(app, session, "hello") # The request carried a Copilot token GitHub traded for the access # token under its own client identity, never the GitHub token itself. exchanges = [ entry for entry in self.github.requests("/copilot_internal/v2/token") if entry["headers"].get("authorization") == f"token {ACCESS_TOKEN}" ] self.assertTrue(exchanges, self.github.seen) self.assertEqual(exchanges[0]["headers"]["editor-version"], "vscode/1.104.1") self.assertEqual( exchanges[0]["headers"]["copilot-integration-id"], "vscode-chat" ) [request] = provider.requests self.assertTrue(request["path"].endswith("/chat/completions"), request["path"]) self.assertEqual(request["authorization"], f"Bearer {COPILOT_TOKEN}") headers = request["headers"] self.assertEqual(headers["editor-version"], "vscode/1.104.1") self.assertEqual(headers["editor-plugin-version"], "copilot-chat/0.30.0") self.assertEqual(headers["copilot-integration-id"], "vscode-chat") self.assertEqual(headers["x-github-api-version"], "2026-08-01") self.assertEqual(headers["x-initiator"], "user") self.assertEqual(request["request"]["model"], CHAT_MODEL) self.assertEqual( request["request"]["messages"][-1], {"role": "user", "content": "hello"} ) answers = [ entry for entry in app.history(session)["items"] if entry["kind"] == "assistant" ] self.assertEqual( answers[-1]["content"], [{"kind": "text", "text": "from copilot"}] ) def test_gpt5_models_stream_on_the_responses_route(self): app, provider = self.start( RESPONSES_MODEL, [text("from responses")], token="github-fixture-responses" ) session = app.session() self.ask(app, session, "hello") [request] = provider.requests self.assertTrue(request["path"].endswith("/responses"), request["path"]) self.assertEqual(request["authorization"], f"Bearer {COPILOT_TOKEN}") self.assertEqual(request["request"]["model"], RESPONSES_MODEL) self.assertIn("input", request["request"]) self.assertNotIn("messages", request["request"]) # Strict schemas are validated literally by Copilot; albedo's tools carry # optional properties, so they are offered non-strict. tools = request["request"]["tools"] self.assertTrue(tools, request["request"]) self.assertTrue(all(tool["strict"] is False for tool in tools), tools) self.assertEqual(request["headers"]["x-initiator"], "user") answers = [ entry for entry in app.history(session)["items"] if entry["kind"] == "assistant" ] self.assertEqual( answers[-1]["content"], [{"kind": "text", "text": "from responses"}] ) def test_gemini_reasoning_text_with_usage_is_saved_and_replayed_to_compaction(self): thought = "compare the two routes" frame = { "id": "copilot-thinking", "choices": [ { "index": 0, "delta": { "role": "assistant", "content": "first answer", "reasoning_text": thought, }, "finish_reason": None, } ], "usage": {"prompt_tokens": 10, "completion_tokens": 5}, } app, provider = self.start( CHAT_MODEL, [ Reply( "raw", events=[ frame, { "id": "copilot-thinking", "choices": [ {"index": 0, "delta": {}, "finish_reason": "stop"} ], }, ], ), text("second answer"), ], ) session = app.session() self.ask(app, session, "first question") self.assertIn("first answer", json.dumps(app.history(session))) self.ask(app, session, "second question") provider.script = lambda _: text("older conversation summary") before = len(provider.requests) with app.api(f"/sessions/{session}/compaction", {}) as response: self.assertEqual(json.load(response)["state"], "compacted") self.assertTrue( any( message.get("reasoning_text") == thought for summary in provider.requests[before:] for message in summary["request"]["messages"] ) ) provider.script = lambda _: text("after compaction") self.ask(app, session, "keep going") history = app.history(session)["items"] self.assertTrue( any( entry["kind"] == "thinking" and thought in json.dumps(entry["content"]) for entry in history ) ) self.assertIn("after compaction", json.dumps(history)) def test_claude_models_stream_on_the_anthropic_messages_route(self): app, provider = self.start( MESSAGES_MODEL, [text("from messages")], token="github-fixture-messages", ) session = app.session() self.ask(app, session, "hello") [request] = provider.requests self.assertTrue(request["path"].endswith("/v1/messages"), request["path"]) self.assertEqual(request["authorization"], f"Bearer {COPILOT_TOKEN}") headers = request["headers"] self.assertEqual(headers["anthropic-version"], "2023-06-01") self.assertEqual( headers["anthropic-beta"], "interleaved-thinking-2025-05-14,extended-cache-ttl-2025-04-11", ) self.assertNotIn("x-api-key", headers) self.assertEqual(request["request"]["model"], MESSAGES_MODEL) # The catalogue's ceiling, not the wire's 8192 default: a long answer # must not be cut off for a budget the turn never named. self.assertEqual(request["request"]["max_tokens"], 64000) self.assertIn("messages", request["request"]) answers = [ entry for entry in app.history(session)["items"] if entry["kind"] == "assistant" ] self.assertEqual( answers[-1]["content"], [{"kind": "text", "text": "from messages"}] ) def test_claude_requests_mark_their_cache_and_record_the_charge(self): """The Messages route caches only what a request marks, and reports the charge. A Claude turn repeats a large prefix at every call, so without a breakpoint it pays the full price each time. Copilot names the charge of the call as `copilot_usage`, in nano AI units, beside the usage of its last frame. """ app, provider = self.start( MESSAGES_MODEL, [Reply("text", events=anthropic_events("marked", billing=420_000_000))], token="github-fixture-billing", ) session = app.session() self.ask(app, session, "hello") [request] = provider.requests body = request["request"] marked = [ block for block in body["messages"][-1]["content"] if "cache_control" in block ] self.assertTrue(marked, body["messages"][-1]) self.assertEqual( marked[-1]["cache_control"], {"type": "ephemeral", "ttl": "1h"} ) last_tool = body["tools"][-1] self.assertEqual(last_tool["cache_control"], {"type": "ephemeral", "ttl": "1h"}) self.assertTrue( all("cache_control" not in tool for tool in body["tools"][:-1]), body["tools"], ) # 420 million nano AI units is 0.42 of an AI credit, and one credit # costs one cent: $0.0042, counted against the subscription and not # charged to a balance. A client shows the total of the session. The # figure for one call stays in the ledger. with app.api(f"/sessions/{session}?tail=0") as response: usage = json.load(response)["usage"] self.assertEqual(usage["spent"]["basis"], "counted") self.assertAlmostEqual(usage["spent"]["usd"], 0.0042) def test_the_picker_lists_only_the_models_the_account_may_pick(self): app, provider = self.start(CHAT_MODEL, []) endpoint = urllib.parse.quote(provider.url, safe="") with app.api( f"/models?provider=github-copilot&endpoint={endpoint}" ) as response: listed = json.load(response) self.assertEqual( sorted(item["id"] for item in listed["items"]), sorted([CHAT_MODEL, RESPONSES_MODEL, MESSAGES_MODEL]), ) @exclusive def test_selected_oauth_reuses_its_minted_token_for_turns_and_catalog(self): app, provider = self.start( CHAT_MODEL, [text("first"), text("second")], token=None ) github_token = "github-fixture-selected-oauth-cache" app.store_secrets( "accounts", { "github-copilot": [ {"type": "key", "apiKey": "github-fixture-unselected-key"}, { "type": "oauth", "access": github_token, "accountId": "chosen", "login": "chosen", "selected": True, }, ] }, ) # Catalog and session creation can mint concurrently on a cold cache. # Complete one catalog request before checking sequential cache reuse. endpoint = urllib.parse.quote(provider.url, safe="") with app.api( f"/models?provider=github-copilot&endpoint={endpoint}" ) as response: self.assertTrue(json.load(response)["items"]) session = app.session() self.ask(app, session, "first") self.ask(app, session, "second") exchanges = [ entry for entry in self.github.requests("/copilot_internal/v2/token") if entry["headers"].get("authorization") == f"token {github_token}" ] self.assertEqual(len(exchanges), 1, exchanges) self.assertEqual(len(provider.requests), 2) stored = json.loads((app.home / "creds.json").read_text())["accounts"][ "github-copilot" ] self.assertEqual(stored[1]["access"], github_token) self.assertNotIn("token", stored[1]) @exclusive def test_rejected_session_token_invalidates_the_github_exchange_cache(self): github_token = "github-fixture-rejected-cache" app, provider = self.start( CHAT_MODEL, [error(401, "expired"), text("renewed")], token=github_token ) endpoint = urllib.parse.quote(provider.url, safe="") with app.api( f"/models?provider=github-copilot&endpoint={endpoint}" ) as response: self.assertTrue(json.load(response)["items"]) session = app.session() self.ask(app, session, "first") # The fake exchange returns the same session token. Rotation refuses to # resend it, but the exchange must still have been invalidated. exchanges = [ entry for entry in self.github.requests("/copilot_internal/v2/token") if entry["headers"].get("authorization") == f"token {github_token}" ] self.assertEqual(len(exchanges), 2, exchanges) self.ask(app, session, "second") self.assertEqual(len(provider.requests), 2) self.assertEqual( provider.requests[-1]["headers"]["authorization"], f"Bearer {COPILOT_TOKEN}" ) def test_a_pasted_github_token_is_saved_as_a_key_account_and_serves(self): """The other half of the sign-in: a pasted token enters as a key account. `/auth/keys` stores a key only for providers `albedo_key_accounts` lists, so a Copilot profile with no saved account and no GITHUB_TOKEN depends on that entry to be usable at all. """ app, provider = self.start( CHAT_MODEL, [text("from a pasted token")], token=None ) account = operation_id() with app.api( f"/auth/keys/{account}", { "provider": "github-copilot", "endpoint": provider.url, "api_key": "github-fixture-pasted", "label": "fixture", }, method="PUT", ) as response: saved = json.load(response) self.assertEqual(saved["kind"], "key") session = app.session() self.ask(app, session, "where do you run") answers = [ entry for entry in app.history(session)["items"] if entry["kind"] == "assistant" ] self.assertEqual( answers[-1]["content"], [{"kind": "text", "text": "from a pasted token"}] ) def scripting(replies): def reply(request): return replies.pop(0) if replies else text("done") return reply