src/delegation_mcp/adapters/base.py

"""Base adapter interface for CLI orchestrators."""

import os
import shutil
from abc import ABC, abstractmethod
from typing import Any


class CLIAdapter(ABC):
    """Base adapter for CLI orchestrators."""

    def __init__(self, name: str, config: dict[str, Any]):
        self.name = name
        self.config = config

    @abstractmethod
    async def execute(self, task: str, **kwargs: Any) -> tuple[str, str, int]:
        """
        Execute a task using the CLI.

        Args:
            task: Task description/query
            **kwargs: Additional CLI-specific arguments

        Returns:
            tuple: (stdout, stderr, return_code)
        """
        pass

    @abstractmethod
    def validate(self) -> bool:
        """Validate the CLI is installed and accessible."""
        pass

    @abstractmethod
    def format_task(self, task: str, **kwargs: Any) -> list[str]:
        """
        Format task into CLI command arguments.

        Args:
            task: Task description
            **kwargs: Additional formatting options

        Returns:
            list: Command arguments
        """
        pass

    def get_command(self) -> str | list[str]:
        """Get base command for this adapter."""
        return self.config.get("command", self.name)

    def get_args(self) -> list[str]:
        """Get default arguments for this adapter."""
        return self.config.get("args", [])

    def get_env(self) -> dict[str, str]:
        """Get environment variables for this adapter."""
        return self.config.get("env", {})

    def get_timeout(self) -> int:
        """Get timeout in seconds."""
        return self.config.get("timeout", 300)

    async def execute_streaming(
        self, 
        task: str,
        progress_callback: Any = None,
        timeout: int | None = None,
        **kwargs: Any
    ) -> tuple[str, str, int]:
        """
        Execute task with real-time output streaming.
        
        Args:
            task: Task description/query
            progress_callback: Optional async callback for progress updates
            timeout: Override timeout in seconds
            **kwargs: Additional CLI-specific arguments
        
        Returns:
            tuple: (stdout, stderr, return_code)
        """
        import asyncio
        
        # Get command and args
        cmd_args = self.format_task(task, **kwargs)
        resolved_cmd = self.resolve_command(cmd_args[0])
        
        if isinstance(resolved_cmd, str):
            full_cmd = [resolved_cmd] + cmd_args[1:]
        else:
            full_cmd = resolved_cmd + cmd_args[1:]
        
        # Merge environment
        env = os.environ.copy()
        env.update(self.get_env())
        
        # Start process with pipes for streaming
        process = await asyncio.create_subprocess_exec(
            *full_cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env=env
        )
        
        stdout_lines = []
        stderr_lines = []
        
        async def read_stream(stream, line_buffer, prefix=""):
            """Read stream line-by-line and call progress callback."""
            while True:
                line = await stream.readline()
                if not line:
                    break
                decoded = line.decode().strip()
                if decoded:
                    line_buffer.append(decoded)
                    if progress_callback:
                        try:
                            await progress_callback(f"{prefix}{decoded}")
                        except Exception:
                            pass  # Don't fail on callback errors
        
        try:
            # Read both streams concurrently with timeout
            effective_timeout = timeout or self.get_timeout()
            await asyncio.wait_for(
                asyncio.gather(
                    read_stream(process.stdout, stdout_lines),
                    read_stream(process.stderr, stderr_lines, "[stderr] "),
                ),
                timeout=effective_timeout
            )
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            raise TimeoutError(f"{self.name} timed out after {effective_timeout}s")
        
        # Wait for process to complete
        await process.wait()
        
        return (
            "\n".join(stdout_lines),
            "\n".join(stderr_lines),
            process.returncode
        )

    @staticmethod
    def resolve_command(cmd: str | list[str]) -> str | list[str]:
        """
        Resolve command to full path on Windows.

        On Windows, asyncio.create_subprocess_exec() doesn't reliably search PATH,
        so we need to resolve commands to their full paths using shutil.which().

        Args:
            cmd: Command string or list of command parts

        Returns:
            Resolved command (full path on Windows, original on Unix)
        """
        if os.name != "nt":
            # On Unix systems, PATH search works fine
            return cmd

        # On Windows, resolve the executable path
        if isinstance(cmd, list):
            if not cmd:
                return cmd
            # Resolve first element (the executable)
            resolved = shutil.which(cmd[0])
            if resolved:
                return [resolved] + cmd[1:]
            return cmd
        else:
            # Single string command
            resolved = shutil.which(cmd)
            return resolved if resolved else cmd