github_issues.py

"""
GitHub Issues Tool - Read issues from a repository.
"""

import subprocess

from ..base import BaseTool, ToolResult
from ..registry import register_tool
from .base import validate_project_root


@register_tool
class GitHubIssuesReadTool(BaseTool):
    """List and view GitHub issues."""

    @property
    def name(self) -> str:
        return "github_issues_read"

    @property
    def description(self) -> str:
        return (
            "Read GitHub issues. List issues with filters or view a specific issue. "
            "Can filter by state, labels, assignee, and search terms."
        )

    @property
    def input_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "issue_number": {
                    "type": "integer",
                    "description": "View a specific issue by number",
                },
                "state": {
                    "type": "string",
                    "enum": ["open", "closed", "all"],
                    "description": "Filter by state (default: open)",
                },
                "label": {
                    "type": "string",
                    "description": "Filter by label",
                },
                "assignee": {
                    "type": "string",
                    "description": "Filter by assignee username",
                },
                "search": {
                    "type": "string",
                    "description": "Search in title and body",
                },
                "limit": {
                    "type": "integer",
                    "description": "Number of issues to list (default: 30)",
                },
            },
        }

    @property
    def requires_grant_metadata(self) -> list[str]:
        return ["project_root"]

    def credential_keys(self) -> list[str]:
        return []

    async def execute(
        self,
        issue_number: int = None,
        state: str = "open",
        label: str = None,
        assignee: str = None,
        search: str = None,
        limit: int = 30,
        **kwargs
    ) -> ToolResult:
        project_root = self.get_grant_metadata("project_root")

        valid, error = validate_project_root(project_root)
        if not valid:
            return ToolResult.fail(error)

        try:
            if issue_number:
                return await self._view_issue(project_root, issue_number)
            else:
                return await self._list_issues(
                    project_root, state, label, assignee, search, limit
                )

        except FileNotFoundError:
            return ToolResult.fail("GitHub CLI (gh) not installed")
        except subprocess.TimeoutExpired:
            return ToolResult.fail("Command timed out")
        except Exception as e:
            return ToolResult.fail(f"GitHub error: {str(e)}")

    async def _list_issues(
        self,
        project_root: str,
        state: str,
        label: str = None,
        assignee: str = None,
        search: str = None,
        limit: int = 30,
    ) -> ToolResult:
        """List issues with filters."""
        cmd = ["gh", "issue", "list", f"--state={state}", f"--limit={limit}"]

        if label:
            cmd.extend(["--label", label])
        if assignee:
            cmd.extend(["--assignee", assignee])
        if search:
            cmd.extend(["--search", search])

        result = subprocess.run(
            cmd,
            cwd=project_root,
            capture_output=True,
            text=True,
            timeout=30,
        )

        if result.returncode != 0:
            stderr = result.stderr.lower()
            if "not a git repository" in stderr:
                return ToolResult.fail("Not a git repository")
            if "no issues" in stderr or "no results" in stderr:
                return ToolResult.ok({
                    "issues": [],
                    "message": "No issues found",
                })
            return ToolResult.fail(f"gh error: {result.stderr}")

        output = result.stdout.strip()

        if not output:
            return ToolResult.ok({
                "issues": [],
                "message": "No issues found",
            })

        lines = output.split('\n')
        return ToolResult.ok({
            "state": state,
            "count": len(lines),
            "issues": output,
        })

    async def _view_issue(self, project_root: str, issue_number: int) -> ToolResult:
        """View a specific issue with comments."""
        result = subprocess.run(
            ["gh", "issue", "view", str(issue_number), "--comments"],
            cwd=project_root,
            capture_output=True,
            text=True,
            timeout=30,
        )

        if result.returncode != 0:
            stderr = result.stderr.lower()
            if "not found" in stderr or "could not find" in stderr:
                return ToolResult.fail(f"Issue #{issue_number} not found")
            return ToolResult.fail(f"gh error: {result.stderr}")

        return ToolResult.ok({
            "issue_number": issue_number,
            "content": result.stdout.strip(),
        })