ccproxy.plugins.copilot.plugin

ccproxy.plugins.copilot.plugin

GitHub Copilot plugin factory and runtime implementation.

CopilotPluginRuntime

CopilotPluginRuntime(manifest)

Bases: ProviderPluginRuntime, AuthProviderPluginRuntime

Runtime for GitHub Copilot plugin.

Source code in ccproxy/plugins/copilot/plugin.py

def __init__(self, manifest: PluginManifest):
    """Initialize runtime."""
    super().__init__(manifest)
    self.config: CopilotConfig | None = None
    self.adapter: CopilotAdapter | None = None
    self.credential_manager: CopilotTokenManager | None = None
    self.oauth_provider: CopilotOAuthProvider | None = None
    self.detection_service: CopilotDetectionService | None = None

cleanup async

cleanup()

Cleanup plugin resources.

Source code in ccproxy/plugins/copilot/plugin.py

async def cleanup(self) -> None:
    """Cleanup plugin resources."""
    errors = []

    # Cleanup adapter
    if self.adapter:
        try:
            await self.adapter.cleanup()
        except Exception as e:
            errors.append(f"Adapter cleanup failed: {e}")
        finally:
            self.adapter = None

    # Cleanup OAuth provider
    if self.oauth_provider:
        try:
            await self.oauth_provider.cleanup()
        except Exception as e:
            errors.append(f"OAuth provider cleanup failed: {e}")
        finally:
            self.oauth_provider = None

    if self.credential_manager:
        try:
            await self.credential_manager.aclose()
        except Exception as e:
            errors.append(f"Token manager cleanup failed: {e}")
        finally:
            self.credential_manager = None

    if errors:
        logger.error(
            "copilot_plugin_cleanup_failed",
            errors=errors,
        )
    else:
        logger.debug("copilot_plugin_cleanup_completed")

CopilotPluginFactory

CopilotPluginFactory()

Bases: BaseProviderPluginFactory, AuthProviderPluginFactory

Factory for GitHub Copilot plugin.

Source code in ccproxy/core/plugins/factories.py

def __init__(self) -> None:
    """Initialize factory with manifest built from class attributes."""
    # Validate required class attributes
    self._validate_class_attributes()

    # Validate runtime class is a proper subclass
    # Import locally to avoid circular import during module import
    from .runtime import ProviderPluginRuntime

    if not issubclass(self.runtime_class, ProviderPluginRuntime):
        raise TypeError(
            f"runtime_class {self.runtime_class.__name__} must be a subclass of ProviderPluginRuntime"
        )

    # Build routes from routers list
    routes = []
    for router_spec in self.routers:
        # Handle both router instances and router factory functions
        router_instance = router_spec.router
        if callable(router_spec.router) and not isinstance(
            router_spec.router, APIRouter
        ):
            # Router is a factory function, call it to get the actual router
            router_instance = router_spec.router()

        routes.append(
            RouteSpec(
                router=cast(APIRouter, router_instance),
                prefix=router_spec.prefix,
                tags=router_spec.tags or [],
                dependencies=router_spec.dependencies,
            )
        )

    # Create manifest from class attributes
    manifest = PluginManifest(
        name=self.plugin_name,
        version=self.plugin_version,
        description=self.plugin_description,
        is_provider=True,
        config_class=self.config_class,
        tool_accumulator_class=self.tool_accumulator_class,
        dependencies=self.dependencies.copy(),
        optional_requires=self.optional_requires.copy(),
        routes=routes,
        tasks=self.tasks.copy(),
        format_adapters=self.format_adapters.copy(),
        requires_format_adapters=self.requires_format_adapters.copy(),
        cli_commands=self.cli_commands.copy(),
        cli_arguments=self.cli_arguments.copy(),
    )

    # Format adapter specification validation is deferred to runtime
    # when settings are available via dependency injection

    # Store the manifest and runtime class directly
    # We don't call parent __init__ because ProviderPluginFactory
    # would override our runtime_class with ProviderPluginRuntime
    self.manifest = manifest
    self.runtime_class = self.__class__.runtime_class

create_context

create_context(core_services)

Create context with all plugin components.

Parameters:

Name	Type	Description	Default
core_services	Any	Core services container	required

Returns:

Type	Description
PluginContext	Plugin context with all components

Source code in ccproxy/plugins/copilot/plugin.py

def create_context(self, core_services: Any) -> PluginContext:
    """Create context with all plugin components.

    Args:
        core_services: Core services container

    Returns:
        Plugin context with all components
    """
    # Start with base context
    context = super().create_context(core_services)

    # Get or create configuration
    config = context.get("config")
    if not isinstance(config, CopilotConfig):
        config = CopilotConfig()
        context["config"] = config

    # Create OAuth provider
    oauth_provider = self.create_oauth_provider(context)
    context["oauth_provider"] = oauth_provider
    # Also set as auth_provider for AuthProviderPluginRuntime compatibility
    context["auth_provider"] = oauth_provider

    # Create detection service
    detection_service = self.create_detection_service(context)
    context["detection_service"] = detection_service

    # Note: adapter creation is handled asynchronously by create_runtime
    # in factories.py, so we don't create it here in the synchronous context creation

    return context

create_runtime

create_runtime()

Create runtime instance.

Source code in ccproxy/plugins/copilot/plugin.py

def create_runtime(self) -> CopilotPluginRuntime:
    """Create runtime instance."""
    return CopilotPluginRuntime(self.manifest)

create_oauth_provider

create_oauth_provider(context=None)

Create OAuth provider instance.

Parameters:

Name	Type	Description	Default
context	PluginContext | None	Plugin context containing shared resources	None

Returns:

Type	Description
CopilotOAuthProvider	CopilotOAuthProvider instance

Source code in ccproxy/plugins/copilot/plugin.py

def create_oauth_provider(
    self, context: PluginContext | None = None
) -> CopilotOAuthProvider:
    """Create OAuth provider instance.

    Args:
        context: Plugin context containing shared resources

    Returns:
        CopilotOAuthProvider instance
    """
    if context and isinstance(context.get("config"), CopilotConfig):
        cfg = cast(CopilotConfig, context.get("config"))
    else:
        cfg = CopilotConfig()

    config: CopilotConfig = cfg
    http_client = context.get("http_client") if context else None
    hook_manager = context.get("hook_manager") if context else None
    cli_detection_service = (
        context.get("cli_detection_service") if context else None
    )

    return CopilotOAuthProvider(
        config.oauth,
        http_client=http_client,
        hook_manager=hook_manager,
        detection_service=cli_detection_service,
    )

create_detection_service

create_detection_service(context)

Create detection service instance.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
DetectionServiceProtocol	CopilotDetectionService instance

Source code in ccproxy/plugins/copilot/plugin.py

def create_detection_service(
    self, context: PluginContext
) -> DetectionServiceProtocol:
    """Create detection service instance.

    Args:
        context: Plugin context

    Returns:
        CopilotDetectionService instance
    """
    settings = context.get("settings")
    cli_service = context.get("cli_detection_service")

    if not settings or not cli_service:
        raise ValueError("Settings and CLI detection service required")

    service = CopilotDetectionService(settings, cli_service)
    return cast(DetectionServiceProtocol, service)

create_adapter async

create_adapter(context)

Create main adapter instance.

Parameters:

Name	Type	Description	Default
context	PluginContext	Plugin context	required

Returns:

Type	Description
BaseAdapter	CopilotAdapter instance

Source code in ccproxy/plugins/copilot/plugin.py

async def create_adapter(self, context: PluginContext) -> BaseAdapter:
    """Create main adapter instance.

    Args:
        context: Plugin context

    Returns:
        CopilotAdapter instance
    """
    if not context:
        raise ValueError("Context required for adapter")

    config = context.get("config")
    if not isinstance(config, CopilotConfig):
        config = CopilotConfig()

    # Get required dependencies following BaseHTTPAdapter pattern
    oauth_provider = context.get("oauth_provider")
    detection_service = context.get("detection_service")
    http_pool_manager = context.get("http_pool_manager")
    auth_manager = context.get("credentials_manager")

    # Optional dependencies
    request_tracer = context.get("request_tracer") or NullRequestTracer()
    metrics = context.get("metrics") or NullMetricsCollector()
    streaming_handler = context.get("streaming_handler") or NullStreamingHandler()
    hook_manager = context.get("hook_manager")

    # Get format_registry from service container
    service_container = context.get("service_container")
    format_registry = None
    if service_container:
        format_registry = service_container.get_format_registry()

    # Debug: Log what we actually have in the context
    logger.debug(
        "copilot_adapter_dependencies_debug",
        context_keys=list(context.keys()) if context else [],
        has_auth_manager=bool(auth_manager),
        has_detection_service=bool(detection_service),
        has_http_pool_manager=bool(http_pool_manager),
        has_oauth_provider=bool(oauth_provider),
        has_format_registry=bool(format_registry),
    )

    if not all([detection_service, http_pool_manager, oauth_provider]):
        missing = []
        if not detection_service:
            missing.append("detection_service")
        if not http_pool_manager:
            missing.append("http_pool_manager")
        if not oauth_provider:
            missing.append("oauth_provider")

        raise ValueError(
            f"Required dependencies missing for CopilotAdapter: {missing}"
        )

    if auth_manager is None:
        configured_override = None
        if hasattr(context, "config") and context.config is not None:
            with contextlib.suppress(AttributeError):
                configured_override = getattr(context.config, "auth_manager", None)

        logger.debug(
            "copilot_adapter_missing_auth_manager",
            reason="unresolved_override",
            configured_override=configured_override,
        )

    return CopilotAdapter(
        config=config,
        auth_manager=auth_manager,
        detection_service=detection_service,
        http_pool_manager=http_pool_manager,
        oauth_provider=oauth_provider,
        request_tracer=request_tracer,
        metrics=metrics,
        streaming_handler=streaming_handler,
        hook_manager=hook_manager,
        format_registry=format_registry,
        context=context,
    )

create_auth_provider

create_auth_provider(context=None)

Create OAuth provider instance for AuthProviderPluginFactory interface.

Parameters:

Name	Type	Description	Default
context	PluginContext | None	Plugin context containing shared resources	None

Returns:

Type	Description
OAuthProviderProtocol	CopilotOAuthProvider instance

Source code in ccproxy/plugins/copilot/plugin.py

def create_auth_provider(
    self, context: PluginContext | None = None
) -> OAuthProviderProtocol:
    """Create OAuth provider instance for AuthProviderPluginFactory interface.

    Args:
        context: Plugin context containing shared resources

    Returns:
        CopilotOAuthProvider instance
    """
    provider = self.create_oauth_provider(context)
    return cast(OAuthProviderProtocol, provider)