Cortex-Inara/cortex/tools/git.py

"""
Git tools — project-scoped.

Read-only inspection:
  git_status  — working tree status
  git_log     — recent commit history
  git_diff    — diff between refs or working tree vs HEAD

Write operations (admin-only, confirm-required):
  git_commit  — stage files and create a commit
  git_push    — push current branch to remote

All tools accept an optional `project` parameter using the same aliases as aider_run:
  "cortex" (default), "aether_api", "aether_frontend", "aether_container"
Or pass an absolute path directly.
"""

import asyncio
import logging
import os
from pathlib import Path

from google.genai import types

logger = logging.getLogger(__name__)

_CORTEX_ROOT: Path = Path(__file__).parent.parent.parent.resolve()

_PROJECT_ALIASES: dict[str, str] = {
    "cortex":           str(_CORTEX_ROOT),
    "aether_api":       "~/OSIT_dev/aether_api_fastapi",
    "aether_frontend":  "~/OSIT_dev/aether_app_sveltekit",
    "aether_container": "~/OSIT_dev/aether_container_env",
}

_MAX_OUTPUT = 50_000

_PROJECT_PARAM = types.Schema(
    type=types.Type.STRING,
    description=(
        "Project to run git in. Known aliases: 'cortex' (default), 'aether_api', "
        "'aether_frontend', 'aether_container'. Or an absolute path. "
        "Omit to use the Cortex project."
    ),
)


def _resolve_project(project: str) -> Path:
    """Resolve a project alias or path string to an absolute Path."""
    if not project:
        return _CORTEX_ROOT
    resolved = _PROJECT_ALIASES.get(project, project)
    return Path(os.path.expanduser(resolved))


async def _git(*args: str, cwd: Path, timeout: int = 15) -> tuple[int, str]:
    """Run a git command in cwd. Returns (returncode, combined output)."""
    proc = await asyncio.create_subprocess_exec(
        "git", "-C", str(cwd), *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()
    return proc.returncode, (out if out else err)


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


# ── Read-only tools ────────────────────────────────────────────────────────────

async def git_status(project: str = "") -> str:
    """Return the working tree status for a project."""
    cwd = _resolve_project(project)
    if not cwd.is_dir():
        return f"Error: project directory not found: {cwd}"
    rc, out = await _git("status", cwd=cwd)
    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, project: str = "") -> str:
    """Return recent commit history for a project."""
    cwd = _resolve_project(project)
    if not cwd.is_dir():
        return f"Error: project directory not found: {cwd}"
    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, cwd=cwd)
    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, project: str = "") -> str:
    """Show a diff for a project. Defaults to working tree vs HEAD."""
    cwd = _resolve_project(project)
    if not cwd.is_dir():
        return f"Error: project directory not found: {cwd}"
    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, cwd=cwd)
    # diff exits 1 when differences exist — normal
    if rc not in (0, 1):
        return f"git diff failed: {out}"
    return _cap(out) or "No differences found."


# ── Write tools (admin-only, confirm-required) ─────────────────────────────────

async def git_commit(message: str, project: str = "", files: list[str] | None = None) -> str:
    """Stage files and create a commit in a project."""
    cwd = _resolve_project(project)
    if not cwd.is_dir():
        return f"Error: project directory not found: {cwd}"
    if not message.strip():
        return "Error: commit message is required."

    # Stage specified files or all changes
    if files:
        for f in files:
            rc, out = await _git("add", "--", f, cwd=cwd)
            if rc != 0:
                return f"git add '{f}' failed: {out}"
    else:
        rc, out = await _git("add", "-A", cwd=cwd)
        if rc != 0:
            return f"git add -A failed: {out}"

    # Check that something is actually staged
    rc, staged = await _git("diff", "--cached", "--stat", cwd=cwd)
    if not staged.strip():
        return "Nothing staged to commit — working tree already clean."

    rc, out = await _git("commit", "-m", message, cwd=cwd)
    if rc != 0:
        return f"git commit failed: {out}"
    return out or "Committed successfully."


async def git_push(project: str = "", remote: str = "origin", branch: str = "") -> str:
    """Push the current branch to a remote."""
    cwd = _resolve_project(project)
    if not cwd.is_dir():
        return f"Error: project directory not found: {cwd}"

    args = ["push", remote]
    if branch:
        args.append(branch)

    rc, out = await _git(*args, cwd=cwd, timeout=30)
    if rc != 0:
        return f"git push failed: {out}"
    return out or f"Pushed to {remote} successfully."


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

DECLARATIONS = [
    types.FunctionDeclaration(
        name="git_status",
        description=(
            "Show the working tree status for a project: staged changes, unstaged "
            "modifications, and untracked files. Use before committing to see what "
            "will be included. Defaults to the Cortex project."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={"project": _PROJECT_PARAM},
        ),
    ),
    types.FunctionDeclaration(
        name="git_log",
        description=(
            "Show recent commit history for a project. Returns commit hashes, dates, "
            "and messages. Use after aider_run completes to see what was committed. "
            "Defaults to the Cortex project."
        ),
        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="Compact one-line format (default true). False for more detail.",
                ),
                "project": _PROJECT_PARAM,
            },
        ),
    ),
    types.FunctionDeclaration(
        name="git_diff",
        description=(
            "Show a git diff for a project. "
            "With no arguments: unstaged working tree changes vs HEAD. "
            "With ref_a only: changes between that ref and HEAD. "
            "With ref_a and ref_b: changes between the two refs. "
            "Use after aider_run (auto_commit=False) to review changes before committing. "
            "Defaults to the Cortex project."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "ref_a": types.Schema(
                    type=types.Type.STRING,
                    description="First ref (commit hash, branch, or tag). Omit for working tree diff.",
                ),
                "ref_b": types.Schema(
                    type=types.Type.STRING,
                    description="Second ref. 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 patch",
                ),
                "project": _PROJECT_PARAM,
            },
        ),
    ),
    types.FunctionDeclaration(
        name="git_commit",
        description=(
            "Stage files and create a git commit in a project. "
            "Use after reviewing changes with git_diff — especially when aider_run ran "
            "with auto_commit=False. Stages all changes by default (files=None). "
            "ADMIN ONLY. Requires confirmation."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "message": types.Schema(
                    type=types.Type.STRING,
                    description="Commit message. Follow the project's commit style (e.g. 'feat: ...').",
                ),
                "project": _PROJECT_PARAM,
                "files": types.Schema(
                    type=types.Type.ARRAY,
                    items=types.Schema(type=types.Type.STRING),
                    description=(
                        "Specific files to stage (paths relative to project root). "
                        "Omit to stage all changes (git add -A)."
                    ),
                ),
            },
            required=["message"],
        ),
    ),
    types.FunctionDeclaration(
        name="git_push",
        description=(
            "Push the current branch to a remote. "
            "Use after git_commit or after aider_run commits to share the changes. "
            "ADMIN ONLY. Requires confirmation."
        ),
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={
                "project": _PROJECT_PARAM,
                "remote": types.Schema(
                    type=types.Type.STRING,
                    description="Remote name (default: 'origin')",
                ),
                "branch": types.Schema(
                    type=types.Type.STRING,
                    description="Branch to push. Omit to push the current tracking branch.",
                ),
            },
        ),
    ),
]