pragent/pilot/webhook_server.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 `AI-REVIEW` PR label, 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 labeled PRs. (Gitea 1.26.1 system
webhooks are broken — see pilot/README-webhook.md.) Onboarding a repo = add the
bot as a Write collaborator + create the label + label a PR.

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://100.74.17.70: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
"""

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

from ai_review import review_pr

# Pull-request webhook `action` values. We fire on EVERY pull_request action
# except `closed` (no point reviewing a closed/merged PR) — the AI-REVIEW label
# gate + sha dedupe downstream make broadening safe: a same-sha re-fire (title
# edit, assignee, milestone, label toggle of another label…) is skipped by
# `review_pr`'s dedupe, and an `unlabeled` event that removed AI-REVIEW fails
# the label gate (payload `labels` reflect current state). Gitea emits
# GitHub-style `action` names (`labeled`, `synchronize`) even though the
# `X-Gitea-Event-Type` header uses `label_updated` / `synchronized`.
SKIP_ACTIONS = {"closed"}
AI_REVIEW_LABEL = "AI-REVIEW"
AI_USAGE_LABEL = "AI-USAGE"

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://100.74.17.70: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"))


def _labels_have(labels, name: str) -> bool:
    """True if the Gitea PR `labels` list (dicts with `name`, or bare strings)
    contains `name`."""
    if not isinstance(labels, list):
        return False
    for lab in labels:
        if isinstance(lab, dict) and lab.get("name") == name:
            return True
        if isinstance(lab, str) and lab == name:
            return True
    return False


def _labels_have_ai_review(labels) -> bool:
    return _labels_have(labels, AI_REVIEW_LABEL)


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"

    labels = pr.get("labels")
    if not _labels_have_ai_review(labels):
        return 200, f"ignore (no {AI_REVIEW_LABEL} label) action={action}"

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

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

    # AI-USAGE label (opt-in) → append the token-usage section + per-comment 🪙
    # lines to the review. PRAGENT_USAGE_ALWAYS forces it on for testing / a
    # future default-on.
    report_usage = _labels_have(labels, AI_USAGE_LABEL) or bool(
        os.environ.get("PRAGENT_USAGE_ALWAYS")
    )

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


def _run_review(repo: str, index: str, title: str, body: str, sha: str, report_usage: bool) -> None:
    try:
        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,
            report_usage=report_usage,
        )
        print(f"pragent-webhook: reviewed {repo}#{index} sha={sha[:8]} ok={ok} usage={report_usage}", 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)


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":
            self._send(200, "ok")
        else:
            self._send(404, "not found")

    def do_POST(self):
        if self.path != "/webhook":
            self._send(404, "not found")
            return
        length = int(self.headers.get("Content-Length", "0") or "0")
        raw = self.rfile.read(length) if length else b""

        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

        pr0 = payload.get("pull_request") or {}
        print(
            f"pragent-webhook: pull_request action={payload.get('action')} "
            f"repo={(payload.get('repository') or {}).get('full_name')} "
            f"ai_review={_labels_have_ai_review(pr0.get('labels'))}",
            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())