nisse/nisse-dev/src/nisse_dev/gitea.py

"""Gitea client (spec B2).

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 (B3), 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 (A3) asks Gitea whether the
effect already happened rather than repeating it.
"""

from dataclasses import dataclass
from typing import Any

import httpx

from nisse_dev.reconcile import Reconciler
from nisse_dev.state import Intent

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


class GiteaError(RuntimeError):
    pass


@dataclass(frozen=True)
class PullRequest:
    number: int
    state: str
    head: str
    url: str


@dataclass(frozen=True)
class Comment:
    id: int
    author: str
    body: str
    created_at: 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:
            raise GiteaError(
                f"gitea {method} {path} returned {response.status_code}: "
                f"{response.text[:300]}"
            )
        return response.json() if response.content else None

    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 find_pull_request(self, repo: str, head: str) -> PullRequest | None:
        """External truth for A3: 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 close(self) -> None:
        self._http.close()


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", ""),
    )


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

    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