src/delegation_mcp/adapters/copilot.py

"""GitHub Copilot CLI adapter."""

import asyncio
import subprocess
from typing import Any

from .base import CLIAdapter


class CopilotAdapter(CLIAdapter):
    """Adapter for GitHub Copilot CLI."""

    async def execute(self, task: str, progress_callback: Any = None, timeout: int | None = None, **kwargs: Any) -> tuple[str, str, int]:
        """Execute task using Copilot CLI with optional streaming."""
        if progress_callback:
            return await self.execute_streaming(task, progress_callback, timeout, **kwargs)
        
        cmd = self.format_task(task, **kwargs)
        resolved_cmd = self.resolve_command(cmd)

        process = await asyncio.create_subprocess_exec(
            *resolved_cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env={**subprocess.os.environ, **self.get_env()},
        )

        try:
            effective_timeout = timeout or self.get_timeout()
            stdout, stderr = await asyncio.wait_for(
                process.communicate(), timeout=effective_timeout
            )
            return (
                stdout.decode("utf-8", errors="replace"),
                stderr.decode("utf-8", errors="replace"),
                process.returncode or 0,
            )
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            raise TimeoutError(f"Copilot CLI timed out after {effective_timeout}s")

    def validate(self) -> bool:
        """Validate Copilot CLI is available."""
        try:
            # Check for copilot CLI
            subprocess.run(
                ["which", "copilot"] if subprocess.os.name != "nt" else ["where", "copilot"],
                capture_output=True,
                check=True,
            )
            return True
        except subprocess.CalledProcessError:
            return False

    def format_task(self, task: str, **kwargs: Any) -> list[str]:
        """Format task for Copilot CLI."""
        cmd = ["copilot"]

        # Add subcommand (suggest, explain, etc.)
        subcommand = kwargs.get("subcommand", "suggest")
        cmd.append(subcommand)

        # Add any custom args from config
        cmd.extend(self.get_args())

        # Add task
        cmd.append(task)

        return cmd