refactor(copilot): enhance token snapshot with copilot_access detection and debug logging

Author: CaddyGlow

ccproxy/api/routes/health.py

@@ -6,7 +6,6 @@
 - /health: Detailed diagnostics (comprehensive status)
 
 Follows IETF Health Check Response Format draft standard.
-TODO: health endpoint Content-Type header to only return application/health+json per IETF spec
 """
 
 from datetime import UTC, datetime

ccproxy/cli/commands/auth.py

@@ -258,7 +258,7 @@ def _extract_token_snapshot_duck_typing(
         access_token=access_token,
         refresh_token=refresh_token,
         expires_at=expires_at,
-        extras=extras,
+        extras={},
     )
 
 
@@ -387,6 +387,11 @@ def _coerce_datetime(value: Any) -> datetime | None:
             ):
                 expires_at = _coerce_datetime(created_at + expires_in)
 
+    if provider_normalized == "copilot":
+        if "refresh_token_present" not in extras:
+            extras["refresh_token_present"] = bool(refresh_token)
+        extras.setdefault("id_token_present", bool(extras.get("has_copilot_token")))
+
     return TokenSnapshot(
         provider=provider_normalized,
         account_id=account_id,

ccproxy/plugins/copilot/manager.py

@@ -71,26 +71,23 @@ async def create(
     def _build_token_snapshot(self, credentials: CopilotCredentials) -> TokenSnapshot:
         """Construct a token snapshot for Copilot credentials."""
         access_token: str | None = None
-        if credentials.copilot_token and credentials.copilot_token.token:
-            access_token = credentials.copilot_token.token.get_secret_value()
-        elif credentials.oauth_token.access_token:
-            access_token = credentials.oauth_token.access_token.get_secret_value()
+        copilot_token = credentials.copilot_token
+        if copilot_token and copilot_token.token:
+            access_token = copilot_token.token.get_secret_value()
 
         refresh_token: str | None = None
-        if credentials.oauth_token.refresh_token:
-            refresh_token = credentials.oauth_token.refresh_token.get_secret_value()
+        oauth_token = credentials.oauth_token
+        if oauth_token.refresh_token:
+            refresh_token = oauth_token.refresh_token.get_secret_value()
 
         expires_at = None
-        if credentials.copilot_token and credentials.copilot_token.expires_at:
-            expires_at = credentials.copilot_token.expires_at
+        if copilot_token and copilot_token.expires_at:
+            expires_at = copilot_token.expires_at
         else:
-            if (
-                credentials.oauth_token.expires_in
-                and credentials.oauth_token.created_at
-            ):
-                expires_at = credentials.oauth_token.expires_at_datetime
+            if oauth_token.expires_in and oauth_token.created_at:
+                expires_at = oauth_token.expires_at_datetime
 
-        scope_value = credentials.oauth_token.scope or ""
+        scope_value = oauth_token.scope or ""
         scopes = tuple(
             scope
             for scope in (item.strip() for item in scope_value.split(" "))
@@ -102,6 +99,14 @@ def _build_token_snapshot(self, credentials: CopilotCredentials) -> TokenSnapsho
             "has_copilot_token": bool(credentials.copilot_token),
         }
 
+        logger.debug(
+            "copilot_token_snapshot",
+            scopes=scopes,
+            expires_at=expires_at,
+            credentials=credentials,
+            access_token=access_token,
+            refresh_token=refresh_token,
+        )
         return TokenSnapshot(
             provider="copilot",
             access_token=access_token,

ccproxy/plugins/copilot/oauth/provider.py

@@ -5,6 +5,7 @@
 
 import httpx
 
+from ccproxy.auth.managers.token_snapshot import TokenSnapshot
 from ccproxy.auth.oauth.protocol import ProfileLoggingMixin, StandardProfileFields
 from ccproxy.auth.oauth.registry import CliAuthConfig, FlowType, OAuthProviderInfo
 from ccproxy.core.logging import get_plugin_logger
@@ -311,14 +312,62 @@ async def get_token_info(self) -> CopilotTokenInfo | None:
         with contextlib.suppress(Exception):
             profile = await self.get_user_profile()
 
+        copilot_access = False
+        if profile is not None:
+            features = getattr(profile, "features", {}) or {}
+            copilot_access = bool(features.get("copilot_access"))
+            if not copilot_access and getattr(profile, "subscription_type", None):
+                copilot_access = True
+
+        if not copilot_access and credentials.copilot_token is not None:
+            token = credentials.copilot_token
+            indicative_flags = [
+                getattr(token, "chat_enabled", None),
+                getattr(token, "annotations_enabled", None),
+                getattr(token, "individual", None),
+            ]
+            if any(flag is True for flag in indicative_flags if flag is not None):
+                copilot_access = True
+            else:
+                copilot_access = (
+                    True  # Possession of a copilot token implies active access
+                )
+
+        if not copilot_access:
+            copilot_access = credentials.copilot_token is not None
+
         return CopilotTokenInfo(
             provider="copilot",
             oauth_expires_at=oauth_expires_at,
             copilot_expires_at=copilot_expires_at,
             account_type=credentials.account_type,
-            copilot_access=False,  # TODO: Get from profile or credentials
+            copilot_access=copilot_access,
         )
 
+    async def get_token_snapshot(self) -> TokenSnapshot | None:
+        """Return a token snapshot built from stored credentials."""
+
+        try:
+            manager = await self.create_token_manager(storage=self.storage)
+            snapshot = await manager.get_token_snapshot()
+            if snapshot:
+                return snapshot
+        except Exception as exc:  # pragma: no cover - defensive logging
+            logger.debug("copilot_snapshot_via_manager_failed", error=str(exc))
+
+        try:
+            credentials = await self.storage.load_credentials()
+            if not credentials:
+                return None
+
+            from ..manager import CopilotTokenManager
+
+            temp_manager = CopilotTokenManager(storage=self.storage)
+            return temp_manager._build_token_snapshot(credentials)
+        except Exception as exc:  # pragma: no cover - defensive logging
+            logger.debug("copilot_snapshot_from_credentials_failed", error=str(exc))
+            return None
+
     async def is_authenticated(self) -> bool:
         """Check if user is authenticated with valid tokens.
 

tests/plugins/copilot/unit/oauth/test_provider.py

@@ -7,6 +7,7 @@
 import pytest
 from pydantic import SecretStr
 
+from ccproxy.auth.managers.token_snapshot import TokenSnapshot
 from ccproxy.auth.oauth.protocol import StandardProfileFields
 from ccproxy.plugins.copilot.config import CopilotOAuthConfig
 from ccproxy.plugins.copilot.oauth.models import (
@@ -87,6 +88,7 @@ def mock_oauth_token(self) -> CopilotOAuthToken:
         return CopilotOAuthToken(
             access_token=SecretStr("gho_test_token"),
             token_type="bearer",
+            refresh_token=SecretStr("gho_refresh_token"),
             expires_in=28800,  # 8 hours
             created_at=now,
             scope="read:user",
@@ -285,6 +287,7 @@ async def test_get_user_profile_success(
             provider_type="copilot",
             email="test@example.com",
             display_name="Test User",
+            features={"copilot_access": True},
         )
 
         with patch.object(
@@ -367,6 +370,25 @@ async def test_get_token_info_success(
             assert result.account_type == "individual"
             assert result.oauth_expires_at is not None
             assert result.copilot_expires_at is not None
+            assert result.copilot_access is True
+
+    async def test_get_token_info_falls_back_to_credentials(
+        self,
+        oauth_provider: CopilotOAuthProvider,
+        mock_credentials: CopilotCredentials,
+    ) -> None:
+        """copilot_access derived from credentials when profile unavailable."""
+        oauth_provider.storage.load_credentials.return_value = mock_credentials  # type: ignore[attr-defined]
+
+        with patch.object(
+            oauth_provider, "get_user_profile", new_callable=AsyncMock
+        ) as mock_get_profile:
+            mock_get_profile.side_effect = RuntimeError("profile not available")
+
+            result = await oauth_provider.get_token_info()
+
+            assert isinstance(result, CopilotTokenInfo)
+            assert result.copilot_access is True
 
     async def test_get_token_info_no_credentials(
         self, oauth_provider: CopilotOAuthProvider
@@ -378,6 +400,40 @@ async def test_get_token_info_no_credentials(
 
         assert result is None
 
+    async def test_get_token_snapshot_uses_manager(
+        self, oauth_provider: CopilotOAuthProvider
+    ) -> None:
+        manager = AsyncMock()
+        manager.get_token_snapshot.return_value = TokenSnapshot(provider="copilot")  # type: ignore[attr-defined]
+
+        with patch.object(
+            oauth_provider, "create_token_manager", AsyncMock(return_value=manager)
+        ) as mock_create:
+            snapshot = await oauth_provider.get_token_snapshot()
+
+        assert isinstance(snapshot, TokenSnapshot)
+        assert snapshot.provider == "copilot"
+        mock_create.assert_awaited_once()
+        manager.get_token_snapshot.assert_awaited_once()
+
+    async def test_get_token_snapshot_fallback_to_credentials(
+        self,
+        oauth_provider: CopilotOAuthProvider,
+        mock_credentials: CopilotCredentials,
+    ) -> None:
+        oauth_provider.storage.load_credentials.return_value = mock_credentials  # type: ignore[attr-defined]
+
+        with patch.object(
+            oauth_provider,
+            "create_token_manager",
+            AsyncMock(side_effect=RuntimeError("boom")),
+        ):
+            snapshot = await oauth_provider.get_token_snapshot()
+
+        assert isinstance(snapshot, TokenSnapshot)
+        assert snapshot.provider == "copilot"
+        assert snapshot.has_refresh_token() is True
+
     async def test_is_authenticated_with_valid_tokens(
         self,
         oauth_provider: CopilotOAuthProvider,

tests/plugins/copilot/unit/test_manager.py

@@ -14,7 +14,7 @@
 from ccproxy.plugins.copilot.oauth.storage import CopilotOAuthStorage
 
 
-def _build_oauth_token(now: datetime) -> CopilotOAuthToken:
+def _build_oauth_token(now: datetime, with_refresh: bool = True) -> CopilotOAuthToken:
     """Helper to create a valid OAuth token scoped for tests."""
 
     return CopilotOAuthToken(
@@ -23,6 +23,7 @@ def _build_oauth_token(now: datetime) -> CopilotOAuthToken:
         scope="read:user",
         created_at=int(now.timestamp()),
         expires_in=3600,
+        refresh_token=SecretStr("refresh-token") if with_refresh else None,
     )
 
 
@@ -50,6 +51,24 @@ def _build_credentials(
     )
 
 
+def test_token_snapshot_flags(tmp_path: Path) -> None:
+    """Token snapshot should flag refresh/id token presence."""
+
+    storage = CopilotOAuthStorage(credentials_path=tmp_path / "credentials.json")
+    manager = CopilotTokenManager(storage=storage)
+
+    now = datetime.now(UTC)
+    credentials = _build_credentials(
+        now=now,
+        refresh_in=1200,
+        updated_offset=timedelta(seconds=30),
+    )
+
+    snapshot = manager._build_token_snapshot(credentials)
+
+    assert snapshot.has_refresh_token() is True
+
+
 def test_is_expired_uses_refresh_window(tmp_path: Path) -> None:
     """Manager should treat refresh window crossing as expiration trigger."""