nisse/nisse-review/src/nisse_review/reader.py

"""The repository, read-only, as the reviewer's executor (R3, L13).

Reads are bound to the item's repository by construction: no tool names one.
Code comes from Gitea's archive of the PR's head and of its base branch, one
API call each, unpacked into a scratch folder on the agent host and read in
Python: no git, no shell, nothing a model chose is ever run. Issues come
from the Gitea API, for the same repository.
"""

import io
import posixpath
import shutil
import tarfile
from collections.abc import Mapping
from pathlib import Path
from typing import Any

from nisse_core.gitea import GiteaClient, GiteaError
from nisse_core.loop.driver import ToolOutcome
from nisse_core.loop.tools import tool

MAX_FILE_CHARS = 20000
MAX_LISTED = 500
MAX_FOUND = 50  # matching lines from search_code
MAX_ISSUES = 20

REF = "head (the PR's code) or base (the branch it merges into)."
READ_FILE = tool(
    "read_file",
    "Read a file in the repository. Its lines are numbered, for comment.",
    path="The file's path from the repository root.",
    ref=REF,
)
LIST_FILES = tool(
    "list_files",
    "List the files under a folder in the repository.",
    path="The folder, from the repository root; empty for the whole tree.",
    ref=REF,
)
SEARCH_CODE = tool(
    "search_code",
    "Find lines in the repository's files containing some text (ignoring "
    "case): file, line number and the line.",
    text="What to look for.",
    ref=REF,
)
READ_ISSUE = tool(
    "read_issue",
    "Read one of the repository's issues (or PRs) by number, with its text.",
    number="The issue's number.",
)
LIST_ISSUES = tool(
    "list_issues",
    "Find the repository's issues, open or closed, whose text matches.",
    text="What to look for.",
)
READER_TOOLS = [READ_FILE, LIST_FILES, SEARCH_CODE, READ_ISSUE, LIST_ISSUES]


class Repository:
    """``refs``: what ``head`` and ``base`` name (commits or branches).
    ``scratch``: a folder of this run's own, removed by ``close``."""

    def __init__(
        self,
        gitea: GiteaClient,
        repo: str,
        refs: Mapping[str, str],
        scratch: Path,
    ) -> None:
        self._gitea = gitea
        self.repo = repo
        self._refs = dict(refs)
        self._scratch = scratch
        self._trees: dict[str, Path] = {}

    def close(self) -> None:
        shutil.rmtree(self._scratch, ignore_errors=True)

    def tree(self, ref: str) -> Path:
        """The repository at ``ref``, unpacked on first use."""
        if ref not in self._refs:
            raise ValueError(f"ref is head or base, not {ref!r}")
        if ref not in self._trees:
            root = self._scratch / ref
            shutil.rmtree(root, ignore_errors=True)
            root.mkdir(parents=True)
            _unpack(self._gitea.archive(self.repo, self._refs[ref]), root)
            self._trees[ref] = root
        return self._trees[ref]

    def lines(self, path: str, ref: str = "head") -> list[str] | None:
        """A file's lines, or None if there's no such file."""
        found = self._file(path, ref)
        if found is None:
            return None
        return found.read_text(errors="replace").splitlines()

    def _file(self, path: str, ref: str) -> Path | None:
        clean = _clean(path)
        if clean is None:
            return None
        found = self.tree(ref) / clean
        return found if found.is_file() and not found.is_symlink() else None

    def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome:
        try:
            match tool:
                case "read_file":
                    return self._read(arguments)
                case "list_files":
                    return self._list(arguments)
                case "search_code":
                    return self._search(arguments)
                case "read_issue":
                    return self._issue(arguments)
                case "list_issues":
                    return self._issues(arguments)
        except (ValueError, KeyError) as exc:
            return ToolOutcome(str(exc), False)
        except GiteaError as exc:
            return ToolOutcome(f"Gitea didn't answer: {exc}", False)
        return ToolOutcome(f"no such tool: {tool}", False)

    def _ref(self, arguments: Mapping[str, Any]) -> str:
        return str(arguments.get("ref") or "head").strip().lower()

    def _read(self, arguments: Mapping[str, Any]) -> ToolOutcome:
        path = str(arguments.get("path", ""))
        ref = self._ref(arguments)
        lines = self.lines(path, ref)
        if lines is None:
            return ToolOutcome(f"{path}: no such file at {ref}", False)
        text = "\n".join(f"{n:>5}  {line}" for n, line in enumerate(lines, 1))
        if len(text) > MAX_FILE_CHARS:
            more = len(text) - MAX_FILE_CHARS
            text = text[:MAX_FILE_CHARS] + f"\n[{more} more characters not shown]"
        return ToolOutcome(f"[{self.repo}@{ref}: {path}]\n{text}", True)

    def _list(self, arguments: Mapping[str, Any]) -> ToolOutcome:
        folder = str(arguments.get("path", "")).strip().strip("/")
        ref = self._ref(arguments)
        root = self.tree(ref)
        start = root if not folder else root / (_clean(folder) or "\0")
        if not start.is_dir():
            return ToolOutcome(f"{folder}: no such folder at {ref}", False)
        found = sorted(
            str(p.relative_to(root))
            for p in start.rglob("*")
            if p.is_file() and not p.is_symlink()
        )
        shown = found[:MAX_LISTED]
        more = f"\n[{len(found) - len(shown)} more]" if len(found) > len(shown) else ""
        return ToolOutcome(
            f"[{self.repo}@{ref}: {folder or 'every file'}]\n"
            + ("\n".join(shown) or "no files")
            + more,
            True,
        )

    def _search(self, arguments: Mapping[str, Any]) -> ToolOutcome:
        text = str(arguments.get("text", ""))
        if not text.strip():
            return ToolOutcome("search for some text", False)
        ref = self._ref(arguments)
        root = self.tree(ref)
        needle = text.lower()
        hits: list[str] = []
        for path in sorted(root.rglob("*")):
            if not path.is_file() or path.is_symlink():
                continue
            try:
                content = path.read_text()
            except (UnicodeDecodeError, OSError):
                continue  # binary
            for n, line in enumerate(content.splitlines(), 1):
                if needle in line.lower():
                    hits.append(f"{path.relative_to(root)}:{n}: {line.strip()[:200]}")
                    if len(hits) > MAX_FOUND:
                        break
            if len(hits) > MAX_FOUND:
                break
        more = "\n[more not shown: narrow the search]" if len(hits) > MAX_FOUND else ""
        body = "\n".join(hits[:MAX_FOUND]) or "nothing"
        return ToolOutcome(f"[{self.repo}@{ref} search: {text}]\n{body}{more}", True)

    def _issue(self, arguments: Mapping[str, Any]) -> ToolOutcome:
        number = _number(arguments.get("number"))
        issue = self._gitea.issue(self.repo, number)
        return ToolOutcome(
            f"[{self.repo}#{issue.number}, by {issue.author}] {issue.title}\n\n"
            f"{issue.body}",
            True,
        )

    def _issues(self, arguments: Mapping[str, Any]) -> ToolOutcome:
        text = str(arguments.get("text", "")).strip()
        found = self._gitea.search_issues(self.repo, text)[:MAX_ISSUES]
        lines = [f"#{i.number}: {i.title}" for i in found]
        return ToolOutcome(
            f"[{self.repo} issues: {text}]\n" + ("\n".join(lines) or "nothing"), True
        )


def _number(raw: Any) -> int:
    try:
        number = int(str(raw).strip().lstrip("#"))
    except ValueError:
        raise ValueError(f"{raw!r}: an issue's number, like 14") from None
    if number <= 0:
        raise ValueError(f"{raw!r}: an issue's number, like 14")
    return number


def _clean(path: str) -> str | None:
    """A path inside the tree, or None: never absolute, never up and out."""
    path = path.strip()
    if not path or path.startswith("/") or "\\" in path:
        return None
    clean = posixpath.normpath(path)
    if clean == "." or clean.startswith(".."):
        return None
    return clean


def _unpack(archive: bytes, root: Path) -> None:
    """Gitea's archive, without its top folder; only plain files and
    folders, each inside ``root``."""
    with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as tar:
        for member in tar.getmembers():
            if not (member.isfile() or member.isdir()):
                continue  # links and devices are never followed
            parts = member.name.split("/", 1)
            if len(parts) < 2 or not parts[1]:
                continue
            inner = _clean(parts[1])
            if inner is None:
                continue
            target = root / inner
            if member.isdir():
                target.mkdir(parents=True, exist_ok=True)
                continue
            target.parent.mkdir(parents=True, exist_ok=True)
            source = tar.extractfile(member)
            if source is not None:
                target.write_bytes(source.read())