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.
"""

import os
from collections.abc import Mapping
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,
        )

    @classmethod
    def from_env(cls, env: Mapping[str, str] = os.environ) -> "GiteaClient":
        base_url = env.get("NISSE_GITEA_URL")
        token = env.get("NISSE_GITEA_TOKEN")
        if not base_url or not token:
            raise GiteaError("set NISSE_GITEA_URL and NISSE_GITEA_TOKEN")
        return cls(base_url, token)

    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