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

#!/usr/bin/env python3
"""Hermes webhook filter for Gitea pull_request_comment deliveries.

Contract (gateway/platforms/webhook.py): the payload arrives on stdin as JSON.
STDOUT IS A PROTOCOL CHANNEL, not a log:

  - exactly "[SILENT]"  -> delivery ignored, no agent run, no tokens spent
  - a JSON object       -> REPLACES the payload used by the prompt template
  - any other text      -> delivery is ALLOWED THROUGH and the text is attached
                           as script_output

That last case is why every diagnostic here goes to stderr. A stray print()
would not drop an event, it would let one through.

Empty stdout, a nonzero exit, a missing script, or a timeout also count as
"ignored", so this script fails CLOSED: if it breaks, nothing reaches the
agent rather than everything. That is the right direction for a loop guard,
but it does mean a syntax error silently disables the whole integration --
run the test file next to this one after editing.

Two jobs:

1. Filter. Drop the deliveries that must never wake the agent -- above all
   luna's own comments, which would otherwise loop forever: the prompt tells
   her to reply on the PR, and her reply is itself a pull_request_comment.
2. Normalise. Guarantee changes.body.from always exists, so the prompt's
   {changes.body.from} renders as empty rather than as an unfilled
   placeholder on "created" events, where Gitea omits `changes` entirely.
"""
import json
import sys

# Comment authors whose comments must never wake the agent. luna is the agent
# herself (loop guard). Add "ci-bot" here if CI ever starts commenting on PRs
# and you do not want her reacting to build output.
IGNORED_AUTHORS = {"luna"}

# Gitea's HookIssueCommentAction values are created / edited / deleted.
# "deleted" is dropped: the payload still carries the comment body, so letting
# it through would have her act on a request that was explicitly withdrawn.
ALLOWED_ACTIONS = {"created", "edited"}


def ignore(reason: str) -> None:
    print(f"gitea-pr-comment-filter: ignoring delivery: {reason}", file=sys.stderr)
    print("[SILENT]")
    raise SystemExit(0)


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

    comment = payload.get("comment") or {}
    issue = payload.get("issue") or {}
    action = (payload.get("action") or "").strip().lower()
    author = ((comment.get("user") or {}).get("login") or "").strip()

    if action not in ALLOWED_ACTIONS:
        ignore(f"action={action or '<missing>'}")

    if author.lower() in IGNORED_AUTHORS:
        ignore(f"author={author} is the agent itself (loop guard)")

    # Belt and braces: the route already filters to pull_request_comment, but
    # if that filter is ever loosened this keeps issue comments out. Only
    # enforced when the key is actually present.
    if "is_pull" in payload and not payload.get("is_pull"):
        ignore("not a pull request comment (is_pull=false)")

    body = (comment.get("body") or "").strip()
    if not body:
        ignore("empty comment body")

    # Gitea omits `changes` on created events and populates changes.body.from
    # with the pre-edit text on edits. Normalise it to a plain string so the
    # prompt template always resolves, and drop no-op edits (a label or
    # attachment change can fire "edited" without touching the body).
    changes = payload.get("changes") or {}
    previous = ((changes.get("body") or {}).get("from") or "") if isinstance(changes, dict) else ""
    if action == "edited":
        if previous.strip() == body:
            ignore("edited but comment body is unchanged")
        if not previous.strip():
            print(
                "gitea-pr-comment-filter: edited delivery carries no previous body; "
                "passing through so the agent can reconcile from the PR thread",
                file=sys.stderr,
            )

    payload["changes"] = {"body": {"from": previous}}

    print(
        "gitea-pr-comment-filter: allowing comment id=%s action=%s author=%s pr=%s"
        % (comment.get("id"), action, author, issue.get("number")),
        file=sys.stderr,
    )
    json.dump(payload, sys.stdout)


if __name__ == "__main__":
    main()