homelab/hosts/mars/gitea-pr-review-filter.py

#!/usr/bin/env python3
"""Hermes webhook filter for Gitea pull request REVIEW deliveries.

Same stdout contract as gitea-pr-comment-filter.py next to this file -- read
that docstring first; the protocol, the fail-closed direction and the reason
drops exit 3 instead of printing "[SILENT]" are all identical and are not
repeated here.

What is different is the payload. A review is NOT an IssueCommentPayload: it
arrives as a PullRequestPayload with action "reviewed" and a `review` object
that Gitea defines (modules/structs/hook.go) as exactly two fields:

    {"type": "<the HookEventType>", "content": "<the review's summary body>"}

There is no review id and no list of line comments, so this filter cannot see
what the review actually asks for -- the prompt has the agent fetch the
comments with `tea pulls review-comments`. `content` is routinely EMPTY (a
review whose substance is entirely in line comments has no summary body), so
an empty body is deliberately NOT a drop here, unlike in the comment filter.

review.type is the SUBSCRIPTION-namespace name, not the wire name, and the two
collide -- see the long comment in hermes-agent.nix. Both of the wire events
this route subscribes to map back to a review type here:

    wire (X-GitHub-Event)   review.type                     what it is
    ---------------------   -----------------------------   ------------------
    pull_request_comment    pull_request_review_comment     review with a body
    pull_request_rejected   pull_request_review_rejected    changes requested

Approvals DO reach the gitea hook: its api-level `pull_request_review` event
is a single switch for all three review types and cannot be narrowed (HasEvent
in models/webhook/webhook.go collapses them onto it). They get dropped one
step earlier than this script instead -- "pull_request_approved" is not in the
route's event list, so Hermes ignores those deliveries on the event match,
before the script runs. That is why pull_request_review_approved is absent
from ALLOWED_REVIEW_TYPES below: an approval is darman signing off, not asking
for work. Widening means adding it in both places.
"""
import json
import sys

# Reviewers whose reviews must never wake the agent. luna is the agent
# herself: she is told to reply with a PR comment rather than a review, so
# this is a backstop rather than the primary loop guard -- but she can post
# reviews via tea, and one self-review would otherwise recurse.
IGNORED_REVIEWERS = {"luna"}

# Exit code for a deliberate drop; see the comment filter's docstring.
DROP_EXIT_CODE = 3

# Reviews are the only thing this route should ever see. Every other
# PullRequestPayload action (opened, synchronized, label_updated, ...) means
# the hook was widened without widening the prompt.
ALLOWED_ACTIONS = {"reviewed"}

ALLOWED_REVIEW_TYPES = {
    "pull_request_review_comment",
    "pull_request_review_rejected",
}


def ignore(reason: str) -> None:
    """Drop the delivery, loudly enough to find in the gateway log."""
    print(f"gitea-pr-review-filter: ignoring delivery: {reason}", file=sys.stderr)
    raise SystemExit(DROP_EXIT_CODE)


def main() -> None:
    try:
        payload = json.loads(sys.stdin.read())
    except (ValueError, OSError) as exc:
        ignore(f"unparseable payload: {exc}")

    if not isinstance(payload, dict):
        ignore("payload is not a JSON object")

    action = (payload.get("action") or "").strip().lower()
    if action not in ALLOWED_ACTIONS:
        ignore(f"action={action or '<missing>'}")

    reviewer = ((payload.get("sender") or {}).get("login") or "").strip()
    if reviewer.lower() in IGNORED_REVIEWERS:
        ignore(f"reviewer={reviewer} is the agent itself (loop guard)")

    review = payload.get("review")
    if not isinstance(review, dict):
        ignore("payload carries no review object")

    review_type = (review.get("type") or "").strip().lower()
    if review_type not in ALLOWED_REVIEW_TYPES:
        ignore(f"review.type={review_type or '<missing>'}")

    pull_request = payload.get("pull_request")
    if not isinstance(pull_request, dict):
        ignore("payload carries no pull_request object")

    # Without a head branch there is nowhere to push, and the prompt would
    # render an unfilled {pull_request.head.ref} placeholder.
    head_ref = ((pull_request.get("head") or {}).get("ref") or "").strip()
    if not head_ref:
        ignore("pull_request.head.ref is missing")

    # A review on a merged or closed PR is history, not a request. Gitea marks
    # merged PRs closed too, so the state check covers both.
    if (pull_request.get("state") or "").strip().lower() != "open":
        ignore(f"pull request is {pull_request.get('state') or '<unknown>'}, not open")

    number = payload.get("number")
    repo = ((payload.get("repository") or {}).get("full_name") or "").strip()
    if not number or not repo:
        ignore(f"incomplete payload: number={number!r} repository.full_name={repo!r}")

    # Normalise the two review fields to plain strings so the prompt template
    # always resolves. Gitea omits neither in practice, but `content` being
    # null rather than "" would render as the literal string "None".
    payload["review"] = {
        "type": review.get("type") or "",
        "content": review.get("content") or "",
    }

    print(
        "gitea-pr-review-filter: allowing review type=%s reviewer=%s pr=%s head=%s"
        % (review_type, reviewer, number, head_ref),
        file=sys.stderr,
    )
    json.dump(payload, sys.stdout)


if __name__ == "__main__":
    main()