pragent/pilot/entrypoints/webhook.py

#!/usr/bin/env python3
"""pragent pilot — central webhook receiver.

A stdlib-only HTTP server that Gitea posts user-webhook events to. It gates on
the PR's base ref having `.pr-review.json` with `"enabled": true`, then runs
the same review core (`ai_review.review_pr`) the CI-step pilot uses, posting
findings back as `pragent-bot`.

Per-owner setup: one Gitea **user-level webhook** per repo-owner fires for every
repo that owner has; this service filters to opted-in PRs. (Gitea 1.26.1 system
webhooks are broken — see pilot/README-webhook.md.) Onboarding a repo = add the
bot as a Write collaborator + commit a `.pr-review.json` with `"enabled": true`
on the base ref.

Stdlib only — no pip install, runs on python:3-slim with the scripts mounted.

Endpoints:
  POST /webhook   Gitea webhook delivery (HMAC-verified)
  GET  /health    liveness probe

Env:
  WEBHOOK_SECRET       shared secret used to register the Gitea webhook (HMAC)
  GITEA_API            in-cluster Gitea base URL
  PRAGENT_BOT_TOKEN    pragent-bot access token (non-admin; must be a Write
                       collaborator on each reviewed repo)
  OLLAMA_URL           headroom proxy URL, e.g. http://model-proxy.internal:8789
  OLLAMA_MODEL         model id, e.g. glm-5.2:cloud
  OLLAMA_MAX_TOKENS    (optional) output cap, default 6000
  DIFF_MAX_CHARS       (optional) diff truncation cap, default 150000
  WEBHOOK_PORT         (optional) listen port, default 8080
  PRAGENT_MAX_CONCURRENT_REVIEWS
                       (optional) how many reviews may run at once, default 2.
                       Each review forks an opencode process that checks out a
                       repo and runs linters, so this is the real resource knob.
  PRAGENT_MAX_BODY_BYTES
                       (optional) request-body cap, default 10 MiB
"""

import base64
import hashlib
import hmac
import json
import os
import threading
import urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from ai_review import gitea_get, review_pr
from review_config import repo_enabled

try:
    import feedback_harvest  # optional — absent in CI-step pod, present in
                              # central webhook service. Harvesting is the
                              # collection side of the feedback loop.
except ImportError:
    feedback_harvest = None

# Pull-request webhook `action` values. We fire on EVERY pull_request action
# except `closed` (no point reviewing a closed/merged PR) — the
# `.pr-review.json:enabled` gate + sha dedupe downstream make broadening safe:
# a same-sha re-fire (title edit, assignee, milestone, label toggle…) is
# skipped by `review_pr`'s dedupe. Gitea emits GitHub-style `action` names
# (`labeled`, `synchronize`) even though the `X-Gitea-Event-Type` header uses
# `label_updated` / `synchronized`.
SKIP_ACTIONS = {"closed"}

GITEA_API = os.environ.get("GITEA_API", "http://gitea-http.gitea.svc.cluster.local:3000")
BOT_TOKEN = os.environ.get("PRAGENT_BOT_TOKEN", "")
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://model-proxy.internal:8789")
OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "glm-5.2:cloud")
OLLAMA_MAX_TOKENS = int(os.environ.get("OLLAMA_MAX_TOKENS", "8000"))
DIFF_MAX_CHARS = int(os.environ.get("DIFF_MAX_CHARS", "150000"))
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "").encode()
PORT = int(os.environ.get("WEBHOOK_PORT", "8080"))
MAX_CONCURRENT = max(1, int(os.environ.get("PRAGENT_MAX_CONCURRENT_REVIEWS", "2")))
MAX_BODY_BYTES = int(os.environ.get("PRAGENT_MAX_BODY_BYTES", str(10 * 1024 * 1024)))
# Feedback DB — SQLite mounted at PRAGENT_FEEDBACK_DB. Empty / unset =
# feedback collection disabled (CI-step path doesn't have it).
FEEDBACK_DB = os.environ.get("PRAGENT_FEEDBACK_DB", "")

# Bound on reviews running at once. Every review forks an opencode process that
# untars a repo, reads files and shells out to linters, so an unbounded thread
# per delivery is a self-inflicted fork bomb the first time someone labels ten
# PRs (or Gitea retries a burst). Queued deliveries wait here rather than pile
# onto the box; the handler has already returned 202, so nothing times out.
_review_slots = threading.Semaphore(MAX_CONCURRENT)

# Reviews currently accepted or running, keyed (repo, index, sha). The
# sha-marker dedupe in `review_pr` reads Gitea *before* posting, so two
# deliveries for the same commit in flight together both see "not yet reviewed"
# and both post — the classic check-then-act race. Common triggers are Gitea
# retries after a slow 202 response and bursty re-fires from a rapid title /
# assign / label toggle. This set closes the window inside one process.
_inflight: set[tuple[str, str, str]] = set()
_inflight_lock = threading.Lock()


def is_repo_enabled(api: str, repo: str, ref: str, token: str) -> bool:
    """True iff `.pr-review.json` on `ref` has `"enabled": true`.

    Reads from the given ref (typically the PR's base ref). False on any
    failure: 404, parse error, missing file, missing `enabled`, wrong type.
    The bool-coerce of `.get("enabled") is True` rejects the common
    gotchas (`null`, `1`, `"yes"`, missing field all yield False).
    """
    return repo_enabled(gitea_get, api, repo, ref, token)


def _verify_signature(raw_body: bytes, headers) -> bool:
    if not WEBHOOK_SECRET:
        return False  # refuse to run without a configured secret
    sig_header = headers.get("X-Gitea-Signature") or headers.get("X-Forgejo-Signature")
    if not sig_header:
        return False
    mac = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(mac, sig_header)


def _handle_pull_request(payload: dict) -> tuple[int, str]:
    """Decide whether to review; if so, kick it off in a background thread.

    Returns (status, message) to Gitea immediately — the review itself runs
    async so Gitea's delivery timeout never fires and causes a retry.
    """
    action = payload.get("action", "")
    pr = payload.get("pull_request") or {}
    repo_obj = payload.get("repository") or {}
    repo = repo_obj.get("full_name") or ""

    if action in SKIP_ACTIONS:
        return 200, f"ignore action={action}"
    if not repo:
        return 400, "no repository.full_name"

    index = pr.get("number")
    if index is None:
        return 400, "no pull_request.number"
    title = pr.get("title", "") or ""
    body = pr.get("body", "") or ""
    head = pr.get("head") or {}
    sha = head.get("sha", "") or ""

    base_ref = (pr.get("base") or {}).get("ref", "") or ""

    if not is_repo_enabled(GITEA_API, repo, base_ref or "", BOT_TOKEN):
        return 200, f"skip (repo not opted in) action={action}"

    if not BOT_TOKEN:
        return 500, "PRAGENT_BOT_TOKEN not set"

    key = (repo, str(index), sha)
    if not _claim(key):
        return 200, f"ignore (already in flight) {repo}#{index} sha={sha[:8]}"

    threading.Thread(
        target=_run_review,
        args=(key, title, body, base_ref),
        daemon=True,
    ).start()
    return 202, f"reviewing {repo}#{index} action={action} sha={sha[:8]}"


def _claim(key: tuple[str, str, str]) -> bool:
    """Reserve (repo, index, sha) for review. False if already claimed."""
    with _inflight_lock:
        if key in _inflight:
            return False
        _inflight.add(key)
        return True


def _release(key: tuple[str, str, str]) -> None:
    with _inflight_lock:
        _inflight.discard(key)


def _run_review(
    key: tuple[str, str, str], title: str, body: str, base_ref: str
) -> None:
    repo, index, sha = key
    # Harvest reactions on PRIOR bot comments on this PR (best-effort —
    # piggy-backs the webhook path so we don't need a separate cron).
    # Disabled if feedback_harvest isn't importable (CI-step image) or
    # FEEDBACK_DB isn't set.
    if FEEDBACK_DB and feedback_harvest is not None:
        try:
            hstats = feedback_harvest.harvest_for_pr(
                api=GITEA_API, token=BOT_TOKEN,
                repo=repo, pr_index=int(index), db_path=FEEDBACK_DB,
            )
            print(
                f"pragent-webhook: harvested {repo}#{index} "
                f"reviews={hstats['reviews_seen']} "
                f"findings={hstats['findings_seen']} "
                f"reactions={hstats['reactions_recorded']}",
                flush=True,
            )
        except Exception as e:
            # Harvest must never abort a review.
            print(f"pragent-webhook: harvest failed for {repo}#{index}: {e}", flush=True)

    try:
        with _review_slots:
            ok = review_pr(
                api=GITEA_API,
                repo=repo,
                index=index,
                title=title,
                body=body,
                sha=sha,
                token=BOT_TOKEN,
                ollama_url=OLLAMA_URL,
                model=OLLAMA_MODEL,
                max_tokens=OLLAMA_MAX_TOKENS,
                max_chars=DIFF_MAX_CHARS,
                base_ref=base_ref,
            )
        print(f"pragent-webhook: reviewed {repo}#{index} sha={sha[:8]} ok={ok}", flush=True)
    except Exception as e:  # review_pr is fail-open, but guard the thread anyway
        print(f"pragent-webhook: thread crashed for {repo}#{index}: {e}", flush=True)
    finally:
        _release(key)


class Handler(BaseHTTPRequestHandler):
    def _send(self, status: int, body: str) -> None:
        data = body.encode()
        self.send_response(status)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        if self.path == "/health":
            with _inflight_lock:
                n = len(_inflight)
            self._send(200, f"ok inflight={n} max_concurrent={MAX_CONCURRENT}")
        else:
            self._send(404, "not found")

    def do_POST(self):
        if self.path != "/webhook":
            self._send(404, "not found")
            return
        try:
            length = int(self.headers.get("Content-Length", "0") or "0")
        except ValueError:
            self._send(400, "bad content-length")
            return
        # Cap before reading: the body is read whole into memory, so an
        # unbounded Content-Length is a one-request OOM.
        if length < 0 or length > MAX_BODY_BYTES:
            self._send(413, "payload too large")
            return
        raw = self.rfile.read(length) if length else b""
        if len(raw) != length:
            self._send(400, "truncated body")
            return

        if not _verify_signature(raw, self.headers):
            self._send(401, "invalid signature")
            return

        try:
            payload = json.loads(raw)
        except json.JSONDecodeError:
            self._send(400, "invalid json")
            return

        event = self.headers.get("X-Gitea-Event") or payload.get("action") or ""
        if event != "pull_request":
            self._send(200, f"ignore event={event}")
            return

        repo_full = (payload.get("repository") or {}).get("full_name")
        print(
            f"pragent-webhook: pull_request action={payload.get('action')} repo={repo_full}",
            flush=True,
        )
        status, msg = _handle_pull_request(payload)
        self._send(status, msg)

    def log_message(self, fmt, *args):
        # Keep k8s logs to our own lines (see _run_review / _send paths).
        print(f"pragent-webhook: {self.address_string()} {fmt % args}", flush=True)


def main() -> int:
    if not WEBHOOK_SECRET:
        print("pragent-webhook: FATAL: WEBHOOK_SECRET not set", flush=True)
        return 1
    if not BOT_TOKEN:
        print("pragent-webhook: FATAL: PRAGENT_BOT_TOKEN not set", flush=True)
        return 1
    server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    print(f"pragent-webhook: listening on :{PORT} (model={OLLAMA_MODEL})", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    raise SystemExit(main())