nisse/nisse-core/src/nisse_core/gitea.py

"""Gitea client (architecture D4, D5).

Outbound: open PRs and post comments through a dedicated service account with
a scoped token. Inbound is deliberately *polled* rather than webhook-driven,
so the agent VM stays outbound-only (D7), and intake is structured JSON —
never routed through a chat bridge.

``find_pull_request`` is the external-truth query that makes opening a PR
idempotent across a crash: boot reconciliation (L8) asks Gitea whether the
effect already happened rather than repeating it.
"""

import base64
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any

import httpx

from nisse_core.state.reconcile import Reconciler
from nisse_core.state.store import Intent

PR_PAGE_SIZE = 50
MAX_PR_PAGES = 40  # 2000 PRs; beyond that, say so rather than guess


class GiteaError(RuntimeError):
    pass


class GiteaNotFound(GiteaError):
    """A 404: no such thing, or nothing this account may see."""


@dataclass(frozen=True)
class PullRequest:
    number: int
    state: str
    head: str
    url: str
    merged: bool = False
    head_sha: str = ""  # the commit the PR is at now
    mergeable: bool | None = None  # None: Gitea didn't say
    title: str = ""
    body: str = ""
    author: str = ""
    base: str = ""  # the branch it would merge into
    draft: bool = False


@dataclass(frozen=True)
class Review:
    id: int
    author: str
    state: str  # COMMENT, APPROVED, REQUEST_CHANGES, ...
    commit: str  # the commit it reviewed
    body: str


@dataclass(frozen=True)
class LineComment:
    """A review comment on one line of a file, as posted: ``line`` in the
    file at the PR's head."""

    path: str
    line: int
    body: str


@dataclass(frozen=True)
class Approval:
    """An approving review, and the commit it approved (L3: approvals bind
    to exactly what was approved)."""

    author: str
    commit: str
    stale: bool = False  # Gitea's mark: commits were pushed since


@dataclass(frozen=True)
class ReviewNote:
    """Something a reviewer said in a review: its body, or a comment on a
    line (#69). ``key`` is unique across both kinds."""

    key: str
    author: str
    body: str
    where: str  # "path:line" for a line comment, "" for a review body
    resolved: bool = False  # a line comment's thread, marked resolved


# A reviewer's reply that settles its thread: nothing more is needed from
# it there. It can't resolve the thread itself (read-only, R3), so the PR's
# author does.
SETTLED_MARKER = "<!-- nisse-review: settled -->"


def to_resolve(notes: list[ReviewNote]) -> list[int]:
    """The ids of the unresolved line comments in threads a settled reply
    closed: every comment on that path:line, as Gitea groups a thread."""
    settled = {
        n.where for n in notes if n.key.startswith("line-") and SETTLED_MARKER in n.body
    }
    return [
        int(n.key.removeprefix("line-"))
        for n in notes
        if n.key.startswith("line-") and n.where in settled and not n.resolved
    ]


@dataclass(frozen=True)
class Comment:
    id: int
    author: str
    body: str
    created_at: str


@dataclass(frozen=True)
class Issue:
    number: int
    title: str
    body: str
    author: str
    labels: tuple[str, ...]


class GiteaClient:
    def __init__(
        self,
        base_url: str,
        token: str,
        *,
        timeout_s: float = 30.0,
        transport: httpx.BaseTransport | None = None,
    ) -> None:
        self._http = httpx.Client(
            base_url=f"{base_url.rstrip('/')}/api/v1",
            headers={"Authorization": f"token {token}"},
            timeout=timeout_s,
            transport=transport,
        )

    def _request(self, method: str, path: str, **kwargs: Any) -> Any:
        try:
            response = self._http.request(method, path, **kwargs)
        except httpx.HTTPError as exc:
            raise GiteaError(f"gitea request failed: {exc}") from exc
        if response.status_code >= 400:
            error = GiteaNotFound if response.status_code == 404 else GiteaError
            raise error(
                f"gitea {method} {path} returned {response.status_code}: "
                f"{response.text[:300]}"
            )
        return response.json() if response.content else None

    def _raw(self, path: str, **kwargs: Any) -> bytes:
        """A GET whose answer isn't JSON: a diff."""
        try:
            response = self._http.get(path, **kwargs)
        except httpx.HTTPError as exc:
            raise GiteaError(f"gitea request failed: {exc}") from exc
        if response.status_code >= 400:
            raise GiteaError(
                f"gitea GET {path} returned {response.status_code}: "
                f"{response.text[:300]}"
            )
        return response.content

    def ssh_url(self, repo: str) -> str:
        """The repository's SSH clone URL, which carries no credential."""
        return str(self._request("GET", f"/repos/{repo}")["ssh_url"])

    def clone_url(self, repo: str) -> str:
        """The repository's HTTPS clone URL, which carries no credential."""
        return str(self._request("GET", f"/repos/{repo}")["clone_url"])

    def whoami(self) -> str:
        return str(self._request("GET", "/user")["login"])

    def open_pull_request(
        self, repo: str, *, head: str, base: str, title: str, body: str
    ) -> PullRequest:
        data = self._request(
            "POST",
            f"/repos/{repo}/pulls",
            json={"head": head, "base": base, "title": title, "body": body},
        )
        return _pull_request(data)

    def pull_request(self, repo: str, index: int) -> PullRequest:
        return _pull_request(self._request("GET", f"/repos/{repo}/pulls/{index}"))

    def review_notes(self, repo: str, index: int) -> list[ReviewNote]:
        """Review bodies and inline line comments on a PR (#69): the PR's
        conversation comments come from list_comments."""
        notes: list[ReviewNote] = []
        for review in (
            self._request("GET", f"/repos/{repo}/pulls/{index}/reviews") or []
        ):
            author = (review.get("user") or {}).get("login", "")
            body = str(review.get("body") or "").strip()
            if body:
                notes.append(ReviewNote(f"review-{review['id']}", author, body, ""))
            if not review.get("comments_count"):
                continue
            for comment in (
                self._request(
                    "GET",
                    f"/repos/{repo}/pulls/{index}/reviews/{review['id']}/comments",
                )
                or []
            ):
                line = comment.get("line") or comment.get("position") or ""
                notes.append(
                    ReviewNote(
                        key=f"line-{comment['id']}",
                        author=(comment.get("user") or {}).get("login", author),
                        body=str(comment.get("body") or ""),
                        where=f"{comment.get('path', '')}:{line}",
                        resolved=bool(comment.get("resolver")),
                    )
                )
        return notes

    def resolve_comment(self, repo: str, comment_id: int) -> None:
        """Mark a line comment's thread resolved: the PR's author or someone
        with write access may."""
        self._request("POST", f"/repos/{repo}/pulls/comments/{comment_id}/resolve")

    def find_pull_request(self, repo: str, head: str) -> PullRequest | None:
        """External truth for L8: did the PR for this branch already get
        opened? Includes closed PRs — a crash after opening must not be
        answered with "no PR is open, so open another".

        Pages to the end rather than sampling the first page: a "not found"
        that only means "not in the newest 50" would licence a duplicate PR.
        If the history is longer than this walk, raise instead of guessing —
        the reconciler turns that into "unverifiable" and gates to the human.
        """
        for page in range(1, MAX_PR_PAGES + 1):
            data = self._request(
                "GET",
                f"/repos/{repo}/pulls",
                params={"state": "all", "limit": PR_PAGE_SIZE, "page": page},
            )
            items = data or []
            for item in items:
                if (item.get("head") or {}).get("ref") == head:
                    return _pull_request(item)
            if len(items) < PR_PAGE_SIZE:
                return None  # reached the end: definitively absent
        raise GiteaError(
            f"searched {MAX_PR_PAGES * PR_PAGE_SIZE} pull requests in {repo} "
            f"without reaching the end; cannot confirm whether {head!r} was "
            "already opened"
        )

    def list_comments(
        self, repo: str, index: int, since: str | None = None
    ) -> list[Comment]:
        params: dict[str, str] = {}
        if since is not None:
            params["since"] = since
        data = self._request(
            "GET", f"/repos/{repo}/issues/{index}/comments", params=params
        )
        return [
            Comment(
                id=item["id"],
                author=(item.get("user") or {}).get("login", ""),
                body=item.get("body", ""),
                created_at=item.get("created_at", ""),
            )
            for item in data or []
        ]

    def post_comment(self, repo: str, index: int, body: str) -> Comment:
        data = self._request(
            "POST", f"/repos/{repo}/issues/{index}/comments", json={"body": body}
        )
        return Comment(
            id=data["id"],
            author=(data.get("user") or {}).get("login", ""),
            body=data.get("body", ""),
            created_at=data.get("created_at", ""),
        )

    def create_issue(self, repo: str, title: str, body: str) -> int:
        data = self._request(
            "POST", f"/repos/{repo}/issues", json={"title": title, "body": body}
        )
        return int(data["number"])

    def close_pull_request(self, repo: str, index: int) -> None:
        self._request("PATCH", f"/repos/{repo}/pulls/{index}", json={"state": "closed"})

    def assigned_issues(self, repo: str, assignee: str) -> list[Issue]:
        """Open issues (not PRs) assigned to ``assignee``: intake (D1)."""
        data = self._request(
            "GET",
            f"/repos/{repo}/issues",
            params={
                "type": "issues",
                "state": "open",
                "assigned_by": assignee,
                "limit": 50,
            },
        )
        return [
            Issue(
                number=int(item["number"]),
                title=str(item.get("title") or ""),
                body=str(item.get("body") or ""),
                author=str(item["user"]["login"]),
                labels=tuple(str(label["name"]) for label in item.get("labels") or []),
            )
            for item in data or []
            if not item.get("pull_request")
        ]

    def open_dependencies(self, repo: str, index: int) -> list[str]:
        """The issues this one depends on that are still open, as
        "owner/repo#N": what blocks it (L1)."""
        data = self._request("GET", f"/repos/{repo}/issues/{index}/dependencies")
        return [
            f"{(item.get('repository') or {}).get('full_name', repo)}#{item['number']}"
            for item in data or []
            if item.get("state") == "open"
        ]

    def assigned_by(self, repo: str, index: int, assignee: str) -> str | None:
        """Who last assigned the issue to ``assignee``, from its timeline;
        None if the timeline doesn't say."""
        events = self._request("GET", f"/repos/{repo}/issues/{index}/timeline") or []
        who = None
        for event in events:
            if (
                event.get("type") == "assignees"
                and not event.get("removed_assignee")
                and (event.get("assignee") or {}).get("login") == assignee
            ):
                who = (event.get("user") or {}).get("login")
        return who

    def assigned_at(self, repo: str, index: int, assignee: str) -> str | None:
        """When the issue was last assigned to ``assignee`` (ISO 8601), from
        its timeline, oldest first and paged to the end; None if the
        timeline doesn't say."""
        when = None
        for page in range(1, MAX_PR_PAGES + 1):
            events = (
                self._request(
                    "GET",
                    f"/repos/{repo}/issues/{index}/timeline",
                    params={"limit": PR_PAGE_SIZE, "page": page},
                )
                or []
            )
            for event in events:
                if (
                    event.get("type") == "assignees"
                    and not event.get("removed_assignee")
                    and (event.get("assignee") or {}).get("login") == assignee
                ):
                    when = event.get("created_at")
            if len(events) < PR_PAGE_SIZE:
                break
        return when

    def approvals(self, repo: str, index: int) -> list[Approval]:
        """The PR's approving reviews that still stand: not dismissed."""
        return [
            Approval(
                author=(review.get("user") or {}).get("login", ""),
                commit=str(review.get("commit_id") or ""),
                stale=bool(review.get("stale")),
            )
            for review in (
                self._request("GET", f"/repos/{repo}/pulls/{index}/reviews") or []
            )
            if review.get("state") == "APPROVED" and not review.get("dismissed")
        ]

    def checks(self, repo: str, sha: str) -> str:
        """A commit's combined check status: success, pending, failure or
        error; "none" when no check reported on it at all."""
        data = self._request("GET", f"/repos/{repo}/commits/{sha}/status") or {}
        if not data.get("total_count") and not data.get("statuses"):
            return "none"
        return str(data.get("state") or "pending")

    def branch_with_file(
        self, repo: str, *, base: str, branch: str, path: str, text: str, message: str
    ) -> None:
        """A new branch off ``base`` with one file added, made through the
        API: no clone, no workspace."""
        self._request(
            "POST",
            f"/repos/{repo}/contents/{path}",
            json={
                "content": base64.b64encode(text.encode()).decode(),
                "message": message,
                "branch": base,
                "new_branch": branch,
            },
        )

    def delete_branch(self, repo: str, name: str) -> None:
        self._request("DELETE", f"/repos/{repo}/branches/{name}")

    def repository(self, repo: str) -> dict[str, Any]:
        """The repository as this account sees it, permissions included."""
        return dict(self._request("GET", f"/repos/{repo}"))

    def branch(self, repo: str, name: str) -> dict[str, Any]:
        """A branch as this account sees it: protected, user_can_push."""
        return dict(self._request("GET", f"/repos/{repo}/branches/{name}"))

    def merge_pull_request(self, repo: str, index: int) -> None:
        self._request(
            "POST", f"/repos/{repo}/pulls/{index}/merge", json={"Do": "merge"}
        )

    def assign_issue(self, repo: str, index: int, assignee: str) -> None:
        self._request(
            "PATCH", f"/repos/{repo}/issues/{index}", json={"assignees": [assignee]}
        )

    def close_issue(self, repo: str, index: int) -> None:
        self._request(
            "PATCH", f"/repos/{repo}/issues/{index}", json={"state": "closed"}
        )

    def readable_repositories(self) -> list[str]:
        """Every repository this account can read, as "owner/name"."""
        found: list[str] = []
        for page in range(1, MAX_PR_PAGES + 1):
            items = (
                self._request(
                    "GET", "/user/repos", params={"limit": PR_PAGE_SIZE, "page": page}
                )
                or []
            )
            found += [str(item["full_name"]) for item in items]
            if len(items) < PR_PAGE_SIZE:
                return found
        raise GiteaError("too many repositories to list")

    def visible_repositories(self) -> list[str]:
        """Every repository this account can read, public ones included."""
        return self._paged_names("/repos/search", key="data")

    def _paged_names(self, path: str, key: str | None = None) -> list[str]:
        found: list[str] = []
        for page in range(1, MAX_PR_PAGES + 1):
            answer = self._request(
                "GET", path, params={"limit": PR_PAGE_SIZE, "page": page}
            )
            items = (answer or {}).get(key, []) if key else (answer or [])
            found += [str(item["full_name"]) for item in items]
            if len(items) < PR_PAGE_SIZE:
                return found
        raise GiteaError("too many repositories to list")

    def open_pull_requests(self, repo: str) -> list[PullRequest]:
        found: list[PullRequest] = []
        for page in range(1, MAX_PR_PAGES + 1):
            items = (
                self._request(
                    "GET",
                    f"/repos/{repo}/pulls",
                    params={"state": "open", "limit": PR_PAGE_SIZE, "page": page},
                )
                or []
            )
            found += [_pull_request(item) for item in items]
            if len(items) < PR_PAGE_SIZE:
                return found
        raise GiteaError(f"too many open pull requests in {repo} to list")

    def pull_request_diff(self, repo: str, index: int) -> str:
        return self._raw(f"/repos/{repo}/pulls/{index}.diff").decode("utf-8", "replace")

    def pull_request_commits(self, repo: str, index: int) -> list[str]:
        """The PR's commits, oldest first, as "sha7 first line"."""
        items = self._request("GET", f"/repos/{repo}/pulls/{index}/commits") or []
        found = []
        for item in items:
            message = str((item.get("commit") or {}).get("message") or "")
            first = message.splitlines()[0] if message.strip() else ""
            found.append(f"{str(item.get('sha', ''))[:7]} {first}".strip())
        return found

    def issue(self, repo: str, index: int) -> Issue:
        return _issue(self._request("GET", f"/repos/{repo}/issues/{index}"))

    def search_issues(self, repo: str, text: str) -> list[Issue]:
        """Issues (not PRs), open or closed, whose text matches."""
        items = self._request(
            "GET",
            f"/repos/{repo}/issues",
            params={"q": text, "type": "issues", "state": "all", "limit": 20},
        )
        return [_issue(item) for item in items or []]

    def reviews(self, repo: str, index: int) -> list[Review]:
        return [
            Review(
                id=int(item["id"]),
                author=(item.get("user") or {}).get("login", ""),
                state=str(item.get("state") or ""),
                commit=str(item.get("commit_id") or ""),
                body=str(item.get("body") or ""),
            )
            for item in self._request("GET", f"/repos/{repo}/pulls/{index}/reviews")
            or []
        ]

    def post_review(
        self,
        repo: str,
        index: int,
        *,
        body: str,
        commit: str,
        comments: Sequence[LineComment] = (),
    ) -> int:
        """A COMMENT review: never an approval or a request for changes."""
        data = self._request(
            "POST",
            f"/repos/{repo}/pulls/{index}/reviews",
            json={
                "event": "COMMENT",
                "body": body,
                "commit_id": commit,
                "comments": [
                    {"path": c.path, "new_position": c.line, "body": c.body}
                    for c in comments
                ],
            },
        )
        return int(data["id"])

    def commits_between(self, repo: str, base: str, head: str) -> list[str]:
        """The commits from ``base`` to ``head`` ("sha subject"), oldest
        first. After a force-push that drops ``base``, Gitea counts from the
        two heads' merge base."""
        data = self._request("GET", f"/repos/{repo}/compare/{base}...{head}") or {}
        found: list[str] = []
        for commit in data.get("commits") or []:
            message = str((commit.get("commit") or {}).get("message") or "")
            subject = message.splitlines()[0] if message.strip() else ""
            found.append(f"{commit['sha']} {subject}".rstrip())
        return found[::-1]  # Gitea lists the newest first

    def commit_diff(self, repo: str, sha: str) -> str:
        """One commit's diff. Through the API, so a token reads private
        repositories (the web's compare diff doesn't take one)."""
        return self._raw(f"/repos/{repo}/git/commits/{sha}.diff").decode(
            errors="replace"
        )

    def react(self, repo: str, comment_id: int, content: str) -> None:
        """A reaction (``+1``, ``eyes``, ...) on a PR's or issue's comment."""
        self._request(
            "POST",
            f"/repos/{repo}/issues/comments/{comment_id}/reactions",
            json={"content": content},
        )

    def close(self) -> None:
        self._http.close()


def _issue(item: dict[str, Any]) -> Issue:
    return Issue(
        number=int(item["number"]),
        title=str(item.get("title") or ""),
        body=str(item.get("body") or ""),
        author=str((item.get("user") or {}).get("login", "")),
        labels=tuple(str(label["name"]) for label in item.get("labels") or []),
    )


def _pull_request(data: dict[str, Any]) -> PullRequest:
    return PullRequest(
        number=data["number"],
        state=data.get("state", ""),
        head=(data.get("head") or {}).get("ref", ""),
        url=data.get("html_url", ""),
        merged=bool(data.get("merged")),
        head_sha=(data.get("head") or {}).get("sha", ""),
        mergeable=data.get("mergeable"),
        title=str(data.get("title") or ""),
        body=str(data.get("body") or ""),
        author=str((data.get("user") or {}).get("login", "")),
        base=str((data.get("base") or {}).get("ref", "")),
        draft=bool(data.get("draft")),
    )


def pull_request_reconciler(client: GiteaClient, repo: str) -> Reconciler:
    """Reconciler for an ``open_pr`` intent (L8 boot step 2).

    An unreachable Gitea yields None — unverifiable, so the agent gates to
    the human rather than risking a duplicate PR. That is the conservative
    direction: a delayed decision costs nothing, a repeated side effect does.
    """

    def check(intent: Intent) -> bool | None:
        branch = intent.payload.get("branch")
        if not branch:
            return None
        try:
            return client.find_pull_request(repo, branch) is not None
        except GiteaError:
            return None

    return check