diff --git a/.env.example b/.env.example index ea74b3a..5e59cda 100644 --- a/.env.example +++ b/.env.example @@ -50,6 +50,13 @@ GITHUB_APP_ID= GITHUB_APP_PRIVATE_KEY= GITHUB_WEBHOOK_SECRET= PR_AF_BOT_MENTION=@pr-af +# Adding this label to a PR triggers a review via webhook (alt to @mention) +PR_AF_LABEL=pr-af +# Optional per-deployment caps for webhook-triggered reviews (unset = defaults). +# Useful to protect a small/shared host from resource spikes. +# PR_AF_MAX_CONCURRENT_REVIEWERS=1 +# PR_AF_MAX_REVIEW_DEPTH=0 +# PR_AF_MAX_COVERAGE_ITERATIONS=1 # --- Human-in-the-loop (optional) --- # Set HAX_API_KEY to enable HITL review gate diff --git a/agentfield-package.yaml b/agentfield-package.yaml index 95646ee..88f20b5 100644 --- a/agentfield-package.yaml +++ b/agentfield-package.yaml @@ -54,6 +54,15 @@ user_environment: - name: PR_AF_MODEL description: Model the harness uses default: openrouter/moonshotai/kimi-k2.5 + - name: PR_AF_LABEL + description: GitHub pull-request label that triggers a webhook review + default: pr-af + - name: PR_AF_MAX_CONCURRENT_REVIEWERS + description: Optional reviewer concurrency cap for webhook-triggered reviews (minimum 1) + - name: PR_AF_MAX_REVIEW_DEPTH + description: Optional sub-review depth cap for webhook-triggered reviews (minimum 0) + - name: PR_AF_MAX_COVERAGE_ITERATIONS + description: Optional coverage iteration cap for webhook-triggered reviews (minimum 1) - name: PR_AF_MAX_COST_USD description: Per-run cost ceiling (USD) default: "2.0" diff --git a/go/README.md b/go/README.md index 74b6b17..854716d 100644 --- a/go/README.md +++ b/go/README.md @@ -157,6 +157,10 @@ The node is configured entirely through the environment. | `PORT` | Listen port (default `8007`) | | `PR_AF_PROVIDER` | Harness provider (default `opencode`) | | `PR_AF_MODEL` | Harness model (default `openrouter/moonshotai/kimi-k2.5`) | +| `PR_AF_LABEL` | Pull-request label that triggers a webhook review (default `pr-af`) | +| `PR_AF_MAX_CONCURRENT_REVIEWERS` | Optional webhook review concurrency cap (minimum `1`) | +| `PR_AF_MAX_REVIEW_DEPTH` | Optional webhook sub-review depth cap (minimum `0`) | +| `PR_AF_MAX_COVERAGE_ITERATIONS` | Optional webhook coverage iteration cap (minimum `1`) | | `PR_AF_HARNESS_BIN` | Optional harness executable override for every provider; unset uses provider defaults | | `PR_AF_MAX_COST_USD` | Per-run cost ceiling in USD (default `2.0`) | | `PR_AF_MAX_DURATION_SECONDS`| Per-run wall-clock ceiling in seconds (default `3600`) | diff --git a/go/agentfield-package.yaml b/go/agentfield-package.yaml index 6eb127b..ed0b8de 100644 --- a/go/agentfield-package.yaml +++ b/go/agentfield-package.yaml @@ -46,6 +46,15 @@ user_environment: - name: PR_AF_MODEL description: harness model default: openrouter/moonshotai/kimi-k2.5 + - name: PR_AF_LABEL + description: GitHub pull-request label that triggers a webhook review + default: pr-af + - name: PR_AF_MAX_CONCURRENT_REVIEWERS + description: optional reviewer concurrency cap for webhook-triggered reviews (minimum 1) + - name: PR_AF_MAX_REVIEW_DEPTH + description: optional sub-review depth cap for webhook-triggered reviews (minimum 0) + - name: PR_AF_MAX_COVERAGE_ITERATIONS + description: optional coverage iteration cap for webhook-triggered reviews (minimum 1) - name: PR_AF_HARNESS_BIN description: optional executable override for every harness provider (unset uses provider defaults) - name: PR_AF_MAX_COST_USD diff --git a/go/internal/node/node.go b/go/internal/node/node.go index 31357aa..d4dcdc7 100644 --- a/go/internal/node/node.go +++ b/go/internal/node/node.go @@ -40,6 +40,14 @@ type Node struct { // field at it and Serve mounts App.Handler() as the fallback route. App *agent.Agent + // labelDedupe bounds duplicate label-triggered review dispatches. It is + // process-local by design; see webhookDedupe.claim for the limitation. + labelDedupe webhookDedupe + + // webhookClient is nil in production (fireReview uses a bounded default). + // Tests inject a transport so webhook dispatches need no listening socket. + webhookClient *http.Client + // NodeID is the resolved node id (NODE_ID env, or the pr-af default). NodeID string diff --git a/go/internal/node/webhook.go b/go/internal/node/webhook.go index 8de1080..6d7fda1 100644 --- a/go/internal/node/webhook.go +++ b/go/internal/node/webhook.go @@ -1,8 +1,7 @@ package node -// webhook.go ports the GitHub @mention webhook (app.py:250-367): an -// issue_comment listener that fires an async PR review at the control plane when -// someone comments "@pr-af …" on a PR. +// webhook.go ports the GitHub webhook (app.py:250-367): issue_comment mentions +// and pull_request labels fire an async PR review at the control plane. // // Env reads happen at REQUEST time (matching internal/config's call-time // convention) so the httptest table can drive GITHUB_WEBHOOK_SECRET / @@ -10,24 +9,80 @@ package node // behavior is identical for a fixed environment. import ( + "container/list" "context" "crypto/hmac" "crypto/sha256" "encoding/hex" "encoding/json" "io" + "log" "net/http" "os" + "strconv" "strings" + "sync" "time" ) -const defaultBotMention = "@pr-af" +const ( + defaultBotMention = "@pr-af" + defaultLabelTrigger = "pr-af" + deliveryCacheSize = 1024 + labelFireTTL = 10 * time.Minute +) + +type webhookDedupe struct { + mu sync.Mutex + deliveries *list.List + deliveryEntries map[string]*list.Element + recentPRs map[string]time.Time + now func() time.Time +} + +// claim guards label-triggered dispatches. The state is intentionally local to +// one node process and resets on restart; multi-process deployments need a +// shared store to dedupe across workers. +func (d *webhookDedupe) claim(deliveryID, prURL string) string { + d.mu.Lock() + defer d.mu.Unlock() + if d.deliveries == nil { + d.deliveries = list.New() + d.deliveryEntries = map[string]*list.Element{} + d.recentPRs = map[string]time.Time{} + } + if deliveryID != "" { + if elem := d.deliveryEntries[deliveryID]; elem != nil { + d.deliveries.MoveToBack(elem) + return "duplicate delivery" + } + delivery := d.deliveries.PushBack(deliveryID) + d.deliveryEntries[deliveryID] = delivery + if d.deliveries.Len() > deliveryCacheSize { + oldest := d.deliveries.Front() + delete(d.deliveryEntries, oldest.Value.(string)) + d.deliveries.Remove(oldest) + } + } + now := time.Now() + if d.now != nil { + now = d.now() + } + for url, firedAt := range d.recentPRs { + if now.Sub(firedAt) >= labelFireTTL { + delete(d.recentPRs, url) + } + } + if firedAt, ok := d.recentPRs[prURL]; ok && now.Sub(firedAt) < labelFireTTL { + return "recently dispatched" + } + d.recentPRs[prURL] = now + return "" +} // webhookGitHub handles POST /webhook/github. It mirrors app.py::webhook_github: -// verify the HMAC signature, answer ping with pong, ignore anything that is not -// a created issue_comment carrying the bot mention on a PR, then fire the async -// review and echo the execution id. +// verify the HMAC signature, answer ping with pong, dispatch matching label and +// mention triggers, and ignore all other events. func (n *Node) webhookGitHub(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(r.Body) if err != nil { @@ -48,7 +103,7 @@ func (n *Node) webhookGitHub(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]any{"status": "pong"}) return } - if event != "issue_comment" { + if event != "issue_comment" && event != "pull_request" { writeJSON(w, http.StatusOK, map[string]any{"status": "ignored", "reason": "event=" + event}) return } @@ -58,6 +113,10 @@ func (n *Node) webhookGitHub(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadRequest, map[string]any{"detail": "invalid JSON"}) return } + if event == "pull_request" { + n.handlePullRequestWebhook(w, r, payload) + return + } action := getStr(payload, "action") if action != "created" { @@ -91,6 +150,33 @@ func (n *Node) webhookGitHub(w http.ResponseWriter, r *http.Request) { }) } +func (n *Node) handlePullRequestWebhook(w http.ResponseWriter, r *http.Request, payload map[string]any) { + action := getStr(payload, "action") + if action != "labeled" { + writeJSON(w, http.StatusOK, map[string]any{"status": "ignored", "reason": "pr action=" + action}) + return + } + if nestedStr(payload, "label", "name") != envOr("PR_AF_LABEL", defaultLabelTrigger) { + writeJSON(w, http.StatusOK, map[string]any{"status": "ignored", "reason": "label not a trigger"}) + return + } + prURL := nestedStr(payload, "pull_request", "html_url") + if prURL == "" { + writeJSON(w, http.StatusOK, map[string]any{"status": "ignored", "reason": "no pr_url"}) + return + } + if reason := n.labelDedupe.claim(r.Header.Get("X-GitHub-Delivery"), prURL); reason != "" { + writeJSON(w, http.StatusOK, map[string]any{"status": "ignored", "reason": reason}) + return + } + execID := n.fireReview(r.Context(), prURL, nil) + writeJSON(w, http.StatusOK, map[string]any{ + "status": "review_dispatched", + "pr_url": prURL, + "execution_id": execID, + }) +} + // verifySignature ports app.py::_verify_signature. An empty secret means "no // secret configured — skip verification" (the caller already guards that). func verifySignature(payload []byte, signature, secret string) bool { @@ -116,6 +202,9 @@ func (n *Node) fireReview(ctx context.Context, prURL string, hints []string) any if len(hints) > 0 { inputPayload["hints"] = hints } + for key, value := range webhookReviewLimits() { + inputPayload[key] = value + } body, err := json.Marshal(map[string]any{"input": inputPayload}) if err != nil { return nil @@ -129,8 +218,14 @@ func (n *Node) fireReview(ctx context.Context, prURL string, hints []string) any return nil } req.Header.Set("Content-Type", "application/json") + if apiKey := os.Getenv("AGENTFIELD_API_KEY"); apiKey != "" { + req.Header.Set("X-API-Key", apiKey) + } - client := &http.Client{Timeout: 15 * time.Second} + client := n.webhookClient + if client == nil { + client = &http.Client{Timeout: 15 * time.Second} + } resp, err := client.Do(req) if err != nil { return nil @@ -151,6 +246,34 @@ func (n *Node) fireReview(ctx context.Context, prURL string, hints []string) any return nil } +// webhookReviewLimits reads optional per-deployment caps for each dispatch. +// Invalid values are deployment mistakes, but must never turn a webhook into a +// 500 or create an unusable zero-cap review, so they are logged and omitted. +func webhookReviewLimits() map[string]any { + limits := map[string]any{} + for _, cap := range []struct { + env string + key string + min int + }{ + {"PR_AF_MAX_CONCURRENT_REVIEWERS", "max_concurrent_reviewers", 1}, + {"PR_AF_MAX_REVIEW_DEPTH", "max_review_depth", 0}, + {"PR_AF_MAX_COVERAGE_ITERATIONS", "max_coverage_iterations", 1}, + } { + raw := os.Getenv(cap.env) + if raw == "" { + continue + } + value, err := strconv.Atoi(raw) + if err != nil || value < cap.min { + log.Printf("[PR-AF] Ignoring invalid %s=%q (must be an integer >= %d)", cap.env, raw, cap.min) + continue + } + limits[cap.key] = value + } + return limits +} + // extractHintsFromComment ports app.py::_extract_hints_from_comment: the text // after the (case-insensitive) bot mention, trimmed; [] when there is none. func extractHintsFromComment(commentBody, botMention string) []string { diff --git a/go/internal/node/webhook_test.go b/go/internal/node/webhook_test.go index b619a7d..e0405a4 100644 --- a/go/internal/node/webhook_test.go +++ b/go/internal/node/webhook_test.go @@ -10,32 +10,45 @@ import ( "net/http/httptest" "strings" "testing" + "time" ) // fakeCP records the async-execute request the webhook fires and returns a // canned execution id, standing in for the control plane. type fakeCP struct { - server *httptest.Server - gotPath string - gotBody map[string]any - hitCount int + url string + client *http.Client + gotPath string + gotBody map[string]any + gotAPIKey string + hitCount int } func newFakeCP(t *testing.T) *fakeCP { t.Helper() - cp := &fakeCP{} - cp.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cp := &fakeCP{url: "http://control-plane.test"} + cp.client = &http.Client{Transport: roundTripperFunc(func(r *http.Request) (*http.Response, error) { cp.hitCount++ cp.gotPath = r.URL.Path + cp.gotAPIKey = r.Header.Get("X-API-Key") body, _ := io.ReadAll(r.Body) _ = json.Unmarshal(body, &cp.gotBody) - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{"execution_id":"exec_abc123"}`)) - })) - t.Cleanup(cp.server.Close) + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"execution_id":"exec_abc123"}`)), + Request: r, + }, nil + })} return cp } +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (fn roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) { + return fn(r) +} + func sign(secret string, body []byte) string { mac := hmac.New(sha256.New, []byte(secret)) mac.Write(body) @@ -59,9 +72,24 @@ func prComment(action, commentBody, htmlURL string) []byte { return b } +func prLabeled(action, label, htmlURL string) []byte { + payload := map[string]any{ + "action": action, + "label": map[string]any{"name": label}, + "pull_request": map[string]any{"number": 42, "html_url": htmlURL}, + "repository": map[string]any{"full_name": "octo/repo"}, + } + b, _ := json.Marshal(payload) + return b +} + // doWebhook drives n.webhookGitHub with the given event/signature and returns // the recorded response plus the decoded JSON body. func doWebhook(t *testing.T, n *Node, event, signature string, body []byte) (*httptest.ResponseRecorder, map[string]any) { + return doWebhookDelivery(t, n, event, signature, "", body) +} + +func doWebhookDelivery(t *testing.T, n *Node, event, signature, delivery string, body []byte) (*httptest.ResponseRecorder, map[string]any) { t.Helper() req := httptest.NewRequest(http.MethodPost, "/webhook/github", strings.NewReader(string(body))) if event != "" { @@ -70,6 +98,9 @@ func doWebhook(t *testing.T, n *Node, event, signature string, body []byte) (*ht if signature != "" { req.Header.Set("X-Hub-Signature-256", signature) } + if delivery != "" { + req.Header.Set("X-GitHub-Delivery", delivery) + } rec := httptest.NewRecorder() n.webhookGitHub(rec, req) @@ -154,7 +185,7 @@ func TestWebhookIgnoreGates(t *testing.T) { func TestWebhookFiresAsyncReview(t *testing.T) { t.Setenv("GITHUB_WEBHOOK_SECRET", "") cp := newFakeCP(t) - n := &Node{NodeID: "pr-af", AgentFieldServer: cp.server.URL} + n := &Node{NodeID: "pr-af", AgentFieldServer: cp.url, webhookClient: cp.client} prURL := "https://github.com/octo/repo/pull/42" rec, resp := doWebhook(t, n, "issue_comment", "", @@ -199,11 +230,133 @@ func TestWebhookFiresAsyncReview(t *testing.T) { } } +func TestWebhookForwardsControlPlaneAPIKey(t *testing.T) { + t.Setenv("GITHUB_WEBHOOK_SECRET", "") + cp := newFakeCP(t) + n := &Node{NodeID: "pr-af", AgentFieldServer: cp.url, webhookClient: cp.client} + prURL := "https://github.com/octo/repo/pull/42" + + t.Setenv("AGENTFIELD_API_KEY", "cp-secret") + doWebhook(t, n, "issue_comment", "", prComment("created", "@pr-af", prURL)) + if cp.gotAPIKey != "cp-secret" { + t.Errorf("X-API-Key = %q, want cp-secret", cp.gotAPIKey) + } + + t.Setenv("AGENTFIELD_API_KEY", "") + doWebhook(t, n, "issue_comment", "", prComment("created", "@pr-af", prURL)) + if cp.gotAPIKey != "" { + t.Errorf("X-API-Key = %q, want absent", cp.gotAPIKey) + } +} + +func TestWebhookLabelTrigger(t *testing.T) { + t.Setenv("GITHUB_WEBHOOK_SECRET", "") + cp := newFakeCP(t) + n := &Node{NodeID: "custom-node", AgentFieldServer: cp.url, webhookClient: cp.client} + prURL := "https://github.com/octo/repo/pull/42" + + _, resp := doWebhook(t, n, "pull_request", "", prLabeled("opened", "pr-af", prURL)) + if resp["status"] != "ignored" || resp["reason"] != "pr action=opened" { + t.Fatalf("non-labeled action response = %v", resp) + } + _, resp = doWebhook(t, n, "pull_request", "", prLabeled("labeled", "other", prURL)) + if resp["status"] != "ignored" || resp["reason"] != "label not a trigger" { + t.Fatalf("wrong-label response = %v", resp) + } + _, resp = doWebhook(t, n, "pull_request", "", prLabeled("labeled", "pr-af", prURL)) + if resp["status"] != "review_dispatched" || cp.hitCount != 1 { + t.Fatalf("default-label response = %v, CP hits = %d", resp, cp.hitCount) + } + if cp.gotPath != "/api/v1/execute/async/custom-node.review" { + t.Errorf("fire path = %q, want custom node endpoint", cp.gotPath) + } + + t.Setenv("PR_AF_LABEL", "ready-for-ai") + overrideURL := "https://github.com/octo/repo/pull/43" + _, resp = doWebhook(t, n, "pull_request", "", prLabeled("labeled", "pr-af", overrideURL)) + if resp["status"] != "ignored" { + t.Fatalf("default label under override response = %v", resp) + } + _, resp = doWebhook(t, n, "pull_request", "", prLabeled("labeled", "ready-for-ai", overrideURL)) + if resp["status"] != "review_dispatched" || cp.hitCount != 2 { + t.Fatalf("override-label response = %v, CP hits = %d", resp, cp.hitCount) + } +} + +func TestWebhookReviewCaps(t *testing.T) { + t.Setenv("GITHUB_WEBHOOK_SECRET", "") + t.Setenv("PR_AF_MAX_CONCURRENT_REVIEWERS", "2") + t.Setenv("PR_AF_MAX_REVIEW_DEPTH", "0") + t.Setenv("PR_AF_MAX_COVERAGE_ITERATIONS", "3") + cp := newFakeCP(t) + n := &Node{NodeID: "pr-af", AgentFieldServer: cp.url, webhookClient: cp.client} + + doWebhook(t, n, "issue_comment", "", prComment("created", "@pr-af", "https://github.com/octo/repo/pull/42")) + input := cp.gotBody["input"].(map[string]any) + for key, want := range map[string]float64{ + "max_concurrent_reviewers": 2, + "max_review_depth": 0, + "max_coverage_iterations": 3, + } { + if input[key] != want { + t.Errorf("%s = %v, want %v", key, input[key], want) + } + } +} + +func TestWebhookInvalidReviewCapsAreIgnored(t *testing.T) { + t.Setenv("GITHUB_WEBHOOK_SECRET", "") + t.Setenv("PR_AF_MAX_CONCURRENT_REVIEWERS", "0") + t.Setenv("PR_AF_MAX_REVIEW_DEPTH", "-1") + t.Setenv("PR_AF_MAX_COVERAGE_ITERATIONS", "many") + cp := newFakeCP(t) + n := &Node{NodeID: "pr-af", AgentFieldServer: cp.url, webhookClient: cp.client} + + rec, resp := doWebhook(t, n, "issue_comment", "", prComment("created", "@pr-af", "https://github.com/octo/repo/pull/42")) + if rec.Code != http.StatusOK || resp["status"] != "review_dispatched" { + t.Fatalf("invalid caps response: code=%d body=%v", rec.Code, resp) + } + input := cp.gotBody["input"].(map[string]any) + for _, key := range []string{"max_concurrent_reviewers", "max_review_depth", "max_coverage_iterations"} { + if _, ok := input[key]; ok { + t.Errorf("invalid cap %s unexpectedly forwarded: %v", key, input[key]) + } + } +} + +func TestWebhookLabelDedupe(t *testing.T) { + t.Setenv("GITHUB_WEBHOOK_SECRET", "") + cp := newFakeCP(t) + n := &Node{NodeID: "pr-af", AgentFieldServer: cp.url, webhookClient: cp.client} + now := time.Unix(1000, 0) + n.labelDedupe.now = func() time.Time { return now } + payload := prLabeled("labeled", "pr-af", "https://github.com/octo/repo/pull/42") + + _, first := doWebhookDelivery(t, n, "pull_request", "", "delivery-1", payload) + _, duplicate := doWebhookDelivery(t, n, "pull_request", "", "delivery-1", payload) + _, recent := doWebhookDelivery(t, n, "pull_request", "", "delivery-2", payload) + now = now.Add(labelFireTTL + time.Second) + _, afterTTL := doWebhookDelivery(t, n, "pull_request", "", "delivery-3", payload) + + if first["status"] != "review_dispatched" || afterTTL["status"] != "review_dispatched" { + t.Errorf("first/after-TTL responses = %v / %v", first, afterTTL) + } + if duplicate["reason"] != "duplicate delivery" { + t.Errorf("duplicate response = %v", duplicate) + } + if recent["reason"] != "recently dispatched" { + t.Errorf("recent response = %v", recent) + } + if cp.hitCount != 2 { + t.Errorf("CP hit %d times, want 2", cp.hitCount) + } +} + func TestWebhookBotMentionOverride(t *testing.T) { t.Setenv("GITHUB_WEBHOOK_SECRET", "") t.Setenv("PR_AF_BOT_MENTION", "@reviewbot") cp := newFakeCP(t) - n := &Node{NodeID: "pr-af", AgentFieldServer: cp.server.URL} + n := &Node{NodeID: "pr-af", AgentFieldServer: cp.url, webhookClient: cp.client} prURL := "https://github.com/octo/repo/pull/7" // The default "@pr-af" no longer triggers; the configured "@reviewbot" does. diff --git a/src/pr_af/app.py b/src/pr_af/app.py index 66cfd6d..9c22bb5 100644 --- a/src/pr_af/app.py +++ b/src/pr_af/app.py @@ -6,6 +6,9 @@ import json import os import subprocess +import threading +import time +from collections import OrderedDict from pathlib import Path from typing import Any, cast @@ -257,11 +260,85 @@ async def review( # --------------------------------------------------------------------------- -# GitHub Webhook — @mention-triggered PR review +# GitHub Webhook — @mention- or label-triggered PR review # --------------------------------------------------------------------------- _BOT_MENTION = os.getenv("PR_AF_BOT_MENTION", "@pr-af") +# Adding this label to a PR triggers a review (alternative to the @mention). +_LABEL_TRIGGER = os.getenv("PR_AF_LABEL", "pr-af") _WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET", "") _CP_URL = os.getenv("AGENTFIELD_SERVER", "http://localhost:8080") +# Control-plane API key, forwarded on the dispatch call when the control plane +# has auth enabled (otherwise the webhook's review dispatch 401s). +_CP_API_KEY = os.getenv("AGENTFIELD_API_KEY", "") + +_DELIVERY_CACHE_SIZE = 1024 +_LABEL_FIRE_TTL_SECONDS = 10 * 60 +_webhook_dedupe_lock = threading.Lock() +_seen_deliveries: OrderedDict[str, None] = OrderedDict() +_recent_label_fires: OrderedDict[str, float] = OrderedDict() + + +def _claim_label_trigger(delivery_id: str, pr_url: str) -> str | None: + """Atomically claim one label-triggered dispatch, or return an ignore reason. + + These guards are intentionally in memory: they cover GitHub redeliveries and + bot-label loops only within one webhook process and reset on restart. A + multi-worker deployment needs a shared store for cross-process dedupe. + """ + now = time.monotonic() + with _webhook_dedupe_lock: + if delivery_id: + if delivery_id in _seen_deliveries: + _seen_deliveries.move_to_end(delivery_id) + return "duplicate delivery" + _seen_deliveries[delivery_id] = None + if len(_seen_deliveries) > _DELIVERY_CACHE_SIZE: + _seen_deliveries.popitem(last=False) + + while _recent_label_fires: + oldest_url, fired_at = next(iter(_recent_label_fires.items())) + if now - fired_at < _LABEL_FIRE_TTL_SECONDS: + break + del _recent_label_fires[oldest_url] + + last_fired = _recent_label_fires.get(pr_url) + if last_fired is not None and now - last_fired < _LABEL_FIRE_TTL_SECONDS: + return "recently dispatched" + _recent_label_fires[pr_url] = now + if len(_recent_label_fires) > _DELIVERY_CACHE_SIZE: + _recent_label_fires.popitem(last=False) + return None + + +def _webhook_review_limits() -> dict[str, object]: + """Optional per-deployment review limits for webhook-triggered runs. + + Only applied when the corresponding env var is set, so default behaviour is + unchanged. Lets a small/shared host cap resource use (e.g. concurrency=1, + review_depth=0) without a code change. + """ + limits: dict[str, object] = {} + for env_name, input_key, minimum in ( + ("PR_AF_MAX_CONCURRENT_REVIEWERS", "max_concurrent_reviewers", 1), + ("PR_AF_MAX_REVIEW_DEPTH", "max_review_depth", 0), + ("PR_AF_MAX_COVERAGE_ITERATIONS", "max_coverage_iterations", 1), + ): + raw = os.getenv(env_name) + if not raw: + continue + try: + value = int(raw) + except ValueError: + value = minimum - 1 + if value < minimum: + print( + f"[PR-AF] Ignoring invalid {env_name}={raw!r} " + f"(must be an integer >= {minimum})", + flush=True, + ) + continue + limits[input_key] = value + return limits def _verify_signature(payload: bytes, signature: str, secret: str) -> bool: @@ -281,16 +358,20 @@ async def _fire_review( "pr_url": pr_url, "depth": "standard", "dry_run": False, + **_webhook_review_limits(), } if hints: input_payload["hints"] = hints body = json.dumps({"input": input_payload}) + headers = {"Content-Type": "application/json"} + if _CP_API_KEY: + headers["X-API-Key"] = _CP_API_KEY try: async with httpx.AsyncClient(timeout=15.0) as client: resp = await client.post( - f"{_CP_URL}/api/v1/execute/async/pr-af.review", + f"{_CP_URL}/api/v1/execute/async/{NODE_ID}.review", content=body, - headers={"Content-Type": "application/json"}, + headers=headers, ) resp.raise_for_status() return resp.json().get("execution_id") @@ -320,11 +401,11 @@ def _get_pr_url_from_issue(payload: dict) -> str | None: async def webhook_github(request: Request) -> dict[str, object]: - """Handle GitHub webhook for @mention-triggered PR reviews. + """Handle GitHub webhook for @mention- or label-triggered PR reviews. - Listens for issue_comment events. When someone comments on a PR with - @pr-af (or the configured bot mention), fires an async review via the - Control Plane. Any text after the @mention is passed as review hints. + A matching pull_request label or an issue_comment containing @pr-af (or + the configured bot mention) fires an async review via the Control Plane. + Any text after an @mention is passed as review hints. Examples: "@pr-af" — standard review @@ -340,6 +421,30 @@ async def webhook_github(request: Request) -> dict[str, object]: if event == "ping": return {"status": "pong"} + # Label trigger: adding the configured label to a PR fires a review. + if event == "pull_request": + payload = json.loads(body) + if payload.get("action") != "labeled": + return {"status": "ignored", "reason": f"pr action={payload.get('action')}"} + if (payload.get("label") or {}).get("name", "") != _LABEL_TRIGGER: + return {"status": "ignored", "reason": "label not a trigger"} + pr_url = (payload.get("pull_request") or {}).get("html_url") + if not pr_url: + return {"status": "ignored", "reason": "no pr_url"} + if reason := _claim_label_trigger( + request.headers.get("x-github-delivery", ""), pr_url + ): + return {"status": "ignored", "reason": reason} + repo_name = payload.get("repository", {}).get("full_name", "") + number = (payload.get("pull_request") or {}).get("number") + print( + f"[PR-AF] Webhook: '{_LABEL_TRIGGER}' label on {repo_name}#{number}" + " — firing review", + flush=True, + ) + exec_id = await _fire_review(pr_url) + return {"status": "review_dispatched", "pr_url": pr_url, "execution_id": exec_id} + if event != "issue_comment": return {"status": "ignored", "reason": f"event={event}"} diff --git a/src/pr_af/orchestrator.py b/src/pr_af/orchestrator.py index 834b69b..5c2b405 100644 --- a/src/pr_af/orchestrator.py +++ b/src/pr_af/orchestrator.py @@ -731,14 +731,18 @@ async def run_dimension(dim: ReviewDimension, depth: int) -> None: await findings_queue.put(findings) sub_reviews = self._extract_sub_reviews(result_raw, dim) - if sub_reviews and depth < max_depth and not self._budget_or_timeout_exhausted("review"): - print( - f"[PR-AF] Dimension '{dim.name}' spawned {len(sub_reviews)} " - f"sub-review(s) at depth {depth + 1}/{max_depth}", - flush=True, - ) - sub_tasks = [run_dimension(sub_dim, depth + 1) for sub_dim in sub_reviews] - await asyncio.gather(*sub_tasks) + + # Children need their own semaphore permits. Await them only after + # the parent has released its permit, otherwise concurrency=1 + # deadlocks with the child waiting forever behind its parent. + if sub_reviews and depth < max_depth and not self._budget_or_timeout_exhausted("review"): + print( + f"[PR-AF] Dimension '{dim.name}' spawned {len(sub_reviews)} " + f"sub-review(s) at depth {depth + 1}/{max_depth}", + flush=True, + ) + sub_tasks = [run_dimension(sub_dim, depth + 1) for sub_dim in sub_reviews] + await asyncio.gather(*sub_tasks) try: tasks = [run_dimension(dim, current_depth) for dim in plan.dimensions] diff --git a/tests/test_orchestrator_concurrency.py b/tests/test_orchestrator_concurrency.py new file mode 100644 index 0000000..548d843 --- /dev/null +++ b/tests/test_orchestrator_concurrency.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import asyncio + +import pytest + +import pr_af.orchestrator as orchestrator_module +from pr_af.config import ReviewConfig +from pr_af.orchestrator import ReviewOrchestrator +from pr_af.schemas.input import ReviewInput +from pr_af.schemas.pipeline import ReviewDimension, ReviewPlan + + +@pytest.mark.asyncio +async def test_single_reviewer_permit_allows_spawned_sub_review( + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = ReviewConfig() + config.budget.max_concurrent_reviewers = 1 + config.budget.max_review_depth = 1 + orchestrator = ReviewOrchestrator(app=None, input=ReviewInput(diff_text="diff"), config=config) + calls: list[int] = [] + + async def fake_review_dimension(**kwargs: object) -> dict[str, object]: + depth = int(kwargs["current_depth"]) + calls.append(depth) + if depth == 0: + return { + "findings": [], + "sub_reviews": [ + { + "review_prompt": "inspect the child", + "target_files": ["src/child.py"], + "reason": "child path", + } + ], + } + return { + "findings": [ + { + "title": "child completed", + "file_path": "src/child.py", + "line_start": 1, + } + ], + "sub_reviews": [], + } + + monkeypatch.setattr(orchestrator_module, "review_dimension", fake_review_dimension) + plan = ReviewPlan( + dimensions=[ + ReviewDimension( + id="parent", + name="Parent", + review_prompt="inspect the parent", + target_files=["src/parent.py"], + ) + ] + ) + queue: asyncio.Queue = asyncio.Queue() + + await asyncio.wait_for(orchestrator._run_parallel_review(plan, queue), timeout=1) + + batches = [] + while (batch := await queue.get()) is not None: + batches.extend(batch) + assert calls == [0, 1] + assert [finding.title for finding in batches] == ["child completed"] diff --git a/tests/test_webhook.py b/tests/test_webhook.py new file mode 100644 index 0000000..e977446 --- /dev/null +++ b/tests/test_webhook.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import importlib +import json +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest +from fastapi import FastAPI + +app_module = importlib.import_module("pr_af.app") + + +class _FakeResponse: + def raise_for_status(self) -> None: + pass + + def json(self) -> dict[str, str]: + return {"execution_id": "exec_test"} + + +class _RecordingAsyncClient: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + async def __aenter__(self) -> _RecordingAsyncClient: + return self + + async def __aexit__(self, *args: object) -> None: + pass + + async def post(self, url: str, **kwargs: Any) -> _FakeResponse: + self.calls.append({"url": url, **kwargs}) + return _FakeResponse() + + +def _webhook_app() -> FastAPI: + webhook_app = FastAPI() + webhook_app.add_api_route("/webhook/github", app_module.webhook_github, methods=["POST"]) + return webhook_app + + +@pytest.fixture(autouse=True) +def clean_webhook_env(monkeypatch: pytest.MonkeyPatch) -> None: + for name in ( + "PR_AF_MAX_CONCURRENT_REVIEWERS", + "PR_AF_MAX_REVIEW_DEPTH", + "PR_AF_MAX_COVERAGE_ITERATIONS", + ): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(app_module, "_WEBHOOK_SECRET", "") + monkeypatch.setattr(app_module, "_LABEL_TRIGGER", "pr-af") + with app_module._webhook_dedupe_lock: + app_module._seen_deliveries.clear() + app_module._recent_label_fires.clear() + + +def _label_payload(label: str, *, action: str = "labeled") -> dict[str, Any]: + return { + "action": action, + "label": {"name": label}, + "pull_request": { + "number": 42, + "html_url": "https://github.com/octo/repo/pull/42", + }, + "repository": {"full_name": "octo/repo"}, + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api_key", ["cp-secret", ""]) +async def test_fire_review_uses_node_id_and_optional_auth( + monkeypatch: pytest.MonkeyPatch, api_key: str +) -> None: + client = _RecordingAsyncClient() + monkeypatch.setattr(app_module, "NODE_ID", "custom-node") + monkeypatch.setattr(app_module, "_CP_URL", "http://control-plane") + monkeypatch.setattr(app_module, "_CP_API_KEY", api_key) + monkeypatch.setattr(app_module.httpx, "AsyncClient", lambda **_kwargs: client) + + exec_id = await app_module._fire_review("https://github.com/octo/repo/pull/42") + + assert exec_id == "exec_test" + assert client.calls[0]["url"] == "http://control-plane/api/v1/execute/async/custom-node.review" + headers = client.calls[0]["headers"] + if api_key: + assert headers.get("X-API-Key") == api_key + else: + assert "X-API-Key" not in headers + + +@pytest.mark.asyncio +async def test_label_trigger_dispatches_and_wrong_label_is_ignored( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fire_review = AsyncMock(return_value="exec_label") + monkeypatch.setattr(app_module, "_fire_review", fire_review) + transport = httpx.ASGITransport(app=_webhook_app()) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as webhook_client: + wrong = await webhook_client.post( + "/webhook/github", + json=_label_payload("other"), + headers={"X-GitHub-Event": "pull_request"}, + ) + assert wrong.status_code == 200 + assert wrong.json() == {"status": "ignored", "reason": "label not a trigger"} + fire_review.assert_not_awaited() + + triggered = await webhook_client.post( + "/webhook/github", + json=_label_payload("pr-af"), + headers={"X-GitHub-Event": "pull_request"}, + ) + assert triggered.status_code == 200 + assert triggered.json()["status"] == "review_dispatched" + fire_review.assert_awaited_once_with("https://github.com/octo/repo/pull/42") + + +@pytest.mark.asyncio +async def test_webhook_caps_use_input_keys_and_ignore_invalid_values( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _RecordingAsyncClient() + webhook_transport = httpx.ASGITransport(app=_webhook_app()) + webhook_client = httpx.AsyncClient(transport=webhook_transport, base_url="http://test") + monkeypatch.setattr(app_module.httpx, "AsyncClient", lambda **_kwargs: client) + monkeypatch.setenv("PR_AF_MAX_CONCURRENT_REVIEWERS", "0") + monkeypatch.setenv("PR_AF_MAX_REVIEW_DEPTH", "-1") + monkeypatch.setenv("PR_AF_MAX_COVERAGE_ITERATIONS", "many") + + async with webhook_client: + response = await webhook_client.post( + "/webhook/github", + json=_label_payload("pr-af"), + headers={"X-GitHub-Event": "pull_request"}, + ) + assert response.status_code == 200 + assert response.json()["status"] == "review_dispatched" + invalid_input = json.loads(client.calls[-1]["content"])["input"] + assert "max_concurrent_reviewers" not in invalid_input + assert "max_review_depth" not in invalid_input + assert "max_coverage_iterations" not in invalid_input + + monkeypatch.setenv("PR_AF_MAX_CONCURRENT_REVIEWERS", "1") + monkeypatch.setenv("PR_AF_MAX_REVIEW_DEPTH", "0") + monkeypatch.setenv("PR_AF_MAX_COVERAGE_ITERATIONS", "2") + assert await app_module._fire_review("https://github.com/octo/repo/pull/42") == "exec_test" + valid_input = json.loads(client.calls[-1]["content"])["input"] + assert valid_input["max_concurrent_reviewers"] == 1 + assert valid_input["max_review_depth"] == 0 + assert valid_input["max_coverage_iterations"] == 2 + + +@pytest.mark.asyncio +async def test_label_trigger_dedupes_delivery_and_recent_pr( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fire_review = AsyncMock(return_value="exec_label") + monkeypatch.setattr(app_module, "_fire_review", fire_review) + now = 1000.0 + monkeypatch.setattr(app_module.time, "monotonic", lambda: now) + transport = httpx.ASGITransport(app=_webhook_app()) + payload = _label_payload("pr-af") + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + first = await client.post( + "/webhook/github", + json=payload, + headers={"X-GitHub-Event": "pull_request", "X-GitHub-Delivery": "delivery-1"}, + ) + duplicate = await client.post( + "/webhook/github", + json=payload, + headers={"X-GitHub-Event": "pull_request", "X-GitHub-Delivery": "delivery-1"}, + ) + recent = await client.post( + "/webhook/github", + json=payload, + headers={"X-GitHub-Event": "pull_request", "X-GitHub-Delivery": "delivery-2"}, + ) + now += app_module._LABEL_FIRE_TTL_SECONDS + 1 + after_ttl = await client.post( + "/webhook/github", + json=payload, + headers={"X-GitHub-Event": "pull_request", "X-GitHub-Delivery": "delivery-3"}, + ) + + assert first.json()["status"] == "review_dispatched" + assert duplicate.json() == {"status": "ignored", "reason": "duplicate delivery"} + assert recent.json() == {"status": "ignored", "reason": "recently dispatched"} + assert after_ttl.json()["status"] == "review_dispatched" + assert fire_review.await_count == 2