Cortex-Inara/cortex/tools/git.py

"""
Git inspection tools — project-scoped, read-only.

  git_status  — working tree status (staged, unstaged, untracked changes)
  git_log     — recent commit history with optional path filter
  git_diff    — diff between commits, branches, or working tree vs HEAD
"""

import asyncio
import logging
from pathlib import Path

from google.genai import types

logger = logging.getLogger(__name__)

_PROJECT_ROOT: Path = Path(__file__).parent.parent.parent.resolve()
_MAX_OUTPUT = 50_000


async def _git(*args: str, timeout: int = 15) -> tuple[int, str]:
    """Run a git command in the project root. Returns (returncode, output)."""
    proc = await asyncio.create_subprocess_exec(
        "git", "-C", str(_PROJECT_ROOT), *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        return 1, "git command timed out"
    out = (stdout or b"").decode(errors="replace").strip()
    err = (stderr or b"").decode(errors="replace").strip()
    combined = out if out else err
    return proc.returncode, combined


def _cap(text: str) -> str:
    if len(text) > _MAX_OUTPUT:
        return text[:_MAX_OUTPUT] + "\n… [truncated]"
    return text


async def git_status() -> str:
    """Return the current git working tree status."""
    rc, out = await _git("status")
    if rc != 0:
        return f"git status failed: {out}"
    return out or "Working tree clean — nothing to report."


async def git_log(n: int = 20, path: str = "", oneline: bool = True) -> str:
    """Return recent git commit history."""
    args = ["log"]
    if oneline:
        args += ["--oneline"]
    else:
        args += ["--format=%H %as %an%n  %s", "--date=short"]
    args += [f"-{max(1, min(n, 200))}"]
    if path:
        args += ["--", path]
    rc, out = await _git(*args)
    if rc != 0:
        return f"git log failed: {out}"
    return _cap(out) or "No commits found."


async def git_diff(ref_a: str = "", ref_b: str = "", path: str = "", stat_only: bool = False) -> str:
    """Show a git diff. Defaults to working tree vs HEAD (unstaged changes)."""
    args = ["diff"]
    if stat_only:
        args += ["--stat"]
    if ref_a and ref_b:
        args += [f"{ref_a}..{ref_b}"]
    elif ref_a:
        args += [ref_a]
    if path:
        args += ["--", path]
    rc, out = await _git(*args)
    # diff exits 1 when there are differences — that's normal
    if rc not in (0, 1):
        return f"git diff failed: {out}"
    return _cap(out) or "No differences found."


# ── Declarations ──────────────────────────────────────────────────────────────

DECLARATIONS = [
    types.FunctionDeclaration(
        name="git_status",
        description=(
            "Show the current git working tree status for the Cortex project: "
            "staged changes, unstaged modifications, and untracked files. "
            "Use to check whether there are uncommitted changes before restarting or deploying."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={},
        ),
    ),
    types.FunctionDeclaration(
        name="git_log",
        description=(
            "Show recent git commit history for the Cortex project. "
            "Returns commit hashes, dates, and messages. "
            "Optionally filter to a specific file or directory path."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "n": types.Schema(
                    type=types.Type.INTEGER,
                    description="Number of commits to return (default 20, max 200)",
                ),
                "path": types.Schema(
                    type=types.Type.STRING,
                    description="Optional file or directory path to filter commits by",
                ),
                "oneline": types.Schema(
                    type=types.Type.BOOLEAN,
                    description="Use compact one-line format (default true). Set false for more detail.",
                ),
            },
        ),
    ),
    types.FunctionDeclaration(
        name="git_diff",
        description=(
            "Show a git diff for the Cortex project. "
            "With no arguments: shows unstaged working tree changes vs HEAD. "
            "With ref_a only: shows changes between that ref and HEAD. "
            "With ref_a and ref_b: shows changes between the two refs (commits, branches, or tags). "
            "Use stat_only to get a summary of changed files instead of full patch output."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "ref_a": types.Schema(
                    type=types.Type.STRING,
                    description="First ref (commit hash, branch name, or tag). Omit for working tree diff.",
                ),
                "ref_b": types.Schema(
                    type=types.Type.STRING,
                    description="Second ref. When provided with ref_a, shows diff between the two.",
                ),
                "path": types.Schema(
                    type=types.Type.STRING,
                    description="Optional file or directory path to restrict the diff to",
                ),
                "stat_only": types.Schema(
                    type=types.Type.BOOLEAN,
                    description="Return only a file-change summary (--stat) instead of the full diff",
                ),
            },
        ),
    ),
]