Gradle MCP Server

"""Gradle wrapper interface for executing Gradle commands."""

import subprocess
import json
import re
import asyncio
from pathlib import Path
from typing import Optional, TYPE_CHECKING
from dataclasses import dataclass

if TYPE_CHECKING:
    from fastmcp import Context


@dataclass
class GradleProject:
    """Represents a Gradle project."""
    name: str
    path: str
    description: Optional[str] = None


@dataclass
class GradleTask:
    """Represents a Gradle task."""
    name: str
    project: str
    description: Optional[str] = None
    group: Optional[str] = None


class GradleWrapper:
    """Interface for executing Gradle commands using the Gradle wrapper."""

    # Cleaning task patterns that should not be allowed in run_task
    CLEANING_TASK_PATTERNS = [
        r"^clean.*",
        r".*clean$",
        r"^cleanBuild.*",
        r"^cleanTest.*",
    ]

    def __init__(self, project_root: Optional[str] = None) -> None:
        """Initialize Gradle wrapper.

        Args:
            project_root: Root directory of the Gradle project. Defaults to current directory.
        """
        self.project_root = Path(project_root or ".")
        self.wrapper_script = self._find_wrapper_script()

    def _find_wrapper_script(self) -> Path:
        """Find the Gradle wrapper script.

        Returns:
            Path to the gradlew script.

        Raises:
            FileNotFoundError: If Gradle wrapper is not found.
        """
        gradle_wrapper = self.project_root / "gradlew"
        if not gradle_wrapper.exists():
            raise FileNotFoundError(
                f"Gradle wrapper not found at {gradle_wrapper}. "
                "Please ensure gradlew script exists in the project root."
            )
        return gradle_wrapper

    def _is_cleaning_task(self, task: str) -> bool:
        """Check if a task is a cleaning task.

        Args:
            task: Task name to check.

        Returns:
            True if the task is a cleaning task, False otherwise.
        """
        task_lower = task.lower()
        for pattern in self.CLEANING_TASK_PATTERNS:
            if re.match(pattern, task_lower):
                return True
        return False

    def _extract_error_message(self, stdout: str, stderr: str, default_message: str = "Task failed") -> str:
        """Extract comprehensive error message from Gradle output.
        
        This method searches for failed tasks and captures all error details
        by searching backwards from FAILURE: or BUILD FAILED markers.
        
        Args:
            stdout: Standard output from Gradle.
            stderr: Standard error from Gradle.
            default_message: Default message if no error markers found.
            
        Returns:
            Extracted error message with full context.
        """
        # Combine stdout and stderr since Gradle splits output between them
        # Task failures and error details go to stdout
        # FAILURE: summary goes to stderr
        combined_output = stdout + "\n" + stderr if stdout and stderr else (stdout or stderr or default_message)
        error_lines = combined_output.strip().split('\n')
        
        # Strategy: Find where actual errors start by searching backwards
        # Gradle output structure for failures:
        # 1. Failed tasks with their errors appear first (in stdout)
        # 2. Then "FAILURE:" section with summaries (in stderr)
        # 3. Finally "BUILD FAILED" summary (in stderr)
        
        # We want to capture from the first failed task onwards
        
        first_failure_idx = -1
        failure_marker_idx = -1
        build_failed_idx = -1
        
        # Find key markers
        for i, line in enumerate(error_lines):
            if 'FAILED' in line and '> Task' in line:
                # Track first failed task
                if first_failure_idx == -1:
                    first_failure_idx = i
            if 'FAILURE:' in line or '* What went wrong:' in line:
                if failure_marker_idx == -1:
                    failure_marker_idx = i
            if 'BUILD FAILED' in line:
                build_failed_idx = i
        
        # If we found FAILURE: or BUILD FAILED, search backwards for the first task failure
        if failure_marker_idx >= 0 or build_failed_idx >= 0:
            marker_idx = failure_marker_idx if failure_marker_idx >= 0 else build_failed_idx
            
            # Search backwards from the marker to find ALL failed tasks
            # Keep updating first_failure_idx to get the earliest one
            for i in range(marker_idx - 1, -1, -1):
                line = error_lines[i]
                if 'FAILED' in line and '> Task' in line:
                    # Update to capture the earliest failed task
                    first_failure_idx = i
                # Stop if we hit successful/skipped tasks (but not failed ones)
                elif '> Task' in line and 'FAILED' not in line:
                    # Hit a non-failed task (UP-TO-DATE, NO-SOURCE, FROM-CACHE, etc.)
                    # Stop searching backwards
                    break
                elif any(marker in line for marker in ['Configuration cache', 'BUILD SUCCESSFUL']):
                    # Hit build start indicators (but NOT "Reusing configuration" which appears at the top)
                    break
        
        # Use the first failure we found
        if first_failure_idx >= 0:
            return '\n'.join(error_lines[first_failure_idx:])
        # Fallback: include substantial context before BUILD FAILED or from the end
        elif build_failed_idx >= 0:
            start_idx = max(0, build_failed_idx - 100)
            return '\n'.join(error_lines[start_idx:])
        else:
            # Last resort: include last 50 lines
            return '\n'.join(error_lines[-50:]) if len(error_lines) > 50 else combined_output

    def list_projects(self) -> list[GradleProject]:
        """List all Gradle projects in the workspace.

        Returns:
            List of GradleProject objects.

        Raises:
            subprocess.CalledProcessError: If Gradle command fails.
        """
        try:
            result = subprocess.run(
                [str(self.wrapper_script), "projects", "-q"],
                cwd=str(self.project_root),
                capture_output=True,
                text=True,
                check=True,
            )

            projects = []
            root_added = False
            
            for line in result.stdout.split("\n"):
                line = line.strip()
                
                # Add root project (only once)
                if "Root project" in line and not root_added:
                    projects.append(
                        GradleProject(
                            name=":",
                            path=str(self.project_root),
                            description="Root project"
                        )
                    )
                    root_added = True
                    continue
                
                # Look for subproject lines like "+--- Project ':app'" or "Project ':app'"
                # But skip if it's the root project line we already handled
                if "Project '" in line and "Root project" not in line:
                    # Extract project name from various formats
                    match = re.search(r"Project '([^']+)'", line)
                    if match:
                        project_name = match.group(1)
                        # Skip root project if it appears again
                        if project_name != ":":
                            projects.append(
                                GradleProject(
                                    name=project_name,
                                    path=str(self.project_root),
                                )
                            )

            return projects
        except subprocess.CalledProcessError as e:
            raise RuntimeError(
                f"Failed to list projects: {e.stderr}"
            ) from e

    def list_tasks(self, project: str = ":") -> list[GradleTask]:
        """List all available tasks for a specific Gradle project.

        Args:
            project: Project name (e.g., ':app'). Use ':' or empty string for root project.

        Returns:
            List of GradleTask objects.

        Raises:
            subprocess.CalledProcessError: If Gradle command fails.
        """
        try:
            # Use tasks --all to get all tasks including inherited ones
            # For root project (: or empty), use just "tasks", for subprojects use "project:tasks"
            is_root = project == ":" or project == "" or project is None
            task_cmd = "tasks" if is_root else f"{project}:tasks"
            result = subprocess.run(
                [str(self.wrapper_script), task_cmd, "--all"],
                cwd=str(self.project_root),
                capture_output=True,
                text=True,
                check=True,
            )

            tasks = []
            in_task_section = False
            current_group = None

            for line in result.stdout.split("\n"):
                line_stripped = line.strip()
                
                # Skip empty lines
                if not line_stripped:
                    continue
                
                # Look for task group headers (end with "tasks")
                if line_stripped.endswith(" tasks") and line_stripped[0].isupper():
                    in_task_section = True
                    current_group = line_stripped.replace(" tasks", "").strip()
                    continue
                
                # Skip separators and rules
                if line_stripped.startswith("-") or "Pattern:" in line_stripped:
                    continue
                
                # Stop at help text
                if "To see all tasks" in line_stripped or line_stripped.startswith("BUILD"):
                    break
                
                # Parse task lines when in a task section
                if in_task_section:
                    # Task lines format: "taskName - description"
                    task_match = re.match(r"^(\w+)\s+-\s+(.+)$", line_stripped)
                    if task_match:
                        task_name = task_match.group(1)
                        description = task_match.group(2)
                        
                        tasks.append(
                            GradleTask(
                                name=task_name,
                                project=project,
                                description=description,
                                group=current_group,
                            )
                        )
                    # Also handle tasks without description
                    elif re.match(r"^(\w+)$", line_stripped):
                        task_name = line_stripped
                        tasks.append(
                            GradleTask(
                                name=task_name,
                                project=project,
                                description="",
                                group=current_group,
                            )
                        )

            return tasks
        except subprocess.CalledProcessError as e:
            raise RuntimeError(
                f"Failed to list tasks for project {project}: {e.stderr}"
            ) from e

    async def run_task(self, task: str, args: Optional[list[str]] = None, ctx: Optional['Context'] = None) -> dict:
        """Run a Gradle task with real-time progress reporting.

        Args:
            task: Task name to run. Can be simple (e.g., 'build') for root project
                  or qualified with project path (e.g., ':app:build', ':core:test').
            args: Additional arguments to pass to Gradle.
            ctx: Optional FastMCP Context for progress reporting.

        Returns:
            Dictionary with 'success', 'error'
            - success (bool): True if task completed successfully
            - error (str or None): Error message if task failed, None otherwise

        Raises:
            ValueError: If task is a cleaning task.
        """
        if self._is_cleaning_task(task):
            raise ValueError(
                f"Task '{task}' is a cleaning task and cannot be run via run_task. "
                "Please use the clean tool instead."
            )

        # Remove -q flag to get progress output
        cmd = [str(self.wrapper_script), task, "--no-build-cache"]

        if args:
            cmd.extend(args)

        try:
            process = subprocess.Popen(
                cmd,
                cwd=str(self.project_root),
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,
            )
            
            stdout_lines = []
            stderr_lines = []
            
            # Pattern to match Gradle progress: <============-> 93% EXECUTING [19s]
            progress_pattern = re.compile(r'(\d+)%')
            
            # Read output in real-time
            while True:
                # Check if process has finished
                if process.poll() is not None:
                    # Read any remaining output
                    remaining_out = process.stdout.read()
                    remaining_err = process.stderr.read()
                    if remaining_out:
                        stdout_lines.append(remaining_out)
                    if remaining_err:
                        stderr_lines.append(remaining_err)
                    break
                
                # Read available output
                out_line = process.stdout.readline()
                if out_line:
                    stdout_lines.append(out_line)
                    if ctx:
                        ctx.info(f"{out_line}")
                    
                    # Parse progress: look for patterns like "93% EXECUTING" or "<======> 93%"
                    if ctx and '%' in out_line:
                        match = progress_pattern.search(out_line)
                        if match:
                            progress = int(match.group(1))
                            await ctx.report_progress(progress=progress, total=100)
                
                err_line = process.stderr.readline()
                if err_line:
                    stderr_lines.append(err_line)
                
                # Small async sleep to not block event loop
                await asyncio.sleep(0.01)
            
            stdout = ''.join(stdout_lines)
            stderr = ''.join(stderr_lines)
            
            if process.returncode == 0:
                return {
                    "success": True,
                    "error": None
                }
            else:
                # Extract comprehensive error message using helper method
                error_message = self._extract_error_message(stdout, stderr, "Task failed")
                
                return {
                    "success": False,
                    "error": error_message
                }
                
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

    async def clean(self, project: Optional[str] = None, ctx: Optional['Context'] = None) -> dict:
        """Run the clean task for a project.

        Args:
            project: Project path (e.g., ':app'). Use ':' or empty string or None for root project.
            ctx: FastMCP context for progress reporting and logging.

        Returns:
            Dictionary with 'success', 'error', 'stdout', and 'stderr' keys.
            - success (bool): True if clean completed successfully
            - error (str or None): Error message if clean failed, None otherwise
            - stdout (str): Standard output from Gradle
            - stderr (str): Standard error from Gradle

        Raises:
            subprocess.CalledProcessError: If Gradle command fails.
        """
        # Root project if project is None, empty, or ":"
        is_root = project is None or project == "" or project == ":"
        project_arg = "" if is_root else f"{project}:"
        # Remove -q to get progress output
        cmd = [str(self.wrapper_script), f"{project_arg}clean", "--no-build-cache"]

        try:
            progress_pattern = re.compile(r'(\d+)%')
            
            process = subprocess.Popen(
                cmd,
                cwd=str(self.project_root),
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
            )
            
            stdout_lines = []
            stderr_lines = []
            
            # Read output line by line in real-time
            while True:
                # Check if process has finished
                if process.poll() is not None:
                    # Read any remaining output
                    remaining_out = process.stdout.read()
                    remaining_err = process.stderr.read()
                    if remaining_out:
                        stdout_lines.append(remaining_out)
                    if remaining_err:
                        stderr_lines.append(remaining_err)
                    break
                
                # Read available output
                out_line = process.stdout.readline()
                if out_line:
                    stdout_lines.append(out_line)
                    
                    # Parse progress
                    if ctx and '%' in out_line:
                        match = progress_pattern.search(out_line)
                        if match:
                            progress = int(match.group(1))
                            await ctx.report_progress(progress=progress, total=100)
                
                err_line = process.stderr.readline()
                if err_line:
                    stderr_lines.append(err_line)
                
                # Small async sleep to not block event loop
                await asyncio.sleep(0.01)
            
            stdout = ''.join(stdout_lines)
            stderr = ''.join(stderr_lines)
            
            if process.returncode == 0:
                return {
                    "success": True,
                    "error": None
                }
            else:
                # Extract comprehensive error message using helper method
                error_message = self._extract_error_message(stdout, stderr, "Clean failed")
                
                return {
                    "success": False,
                    "error": error_message
                }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }