homelab/review-bot/review.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"github.com/failsafe-go/failsafe-go"
	"github.com/failsafe-go/failsafe-go/retrypolicy"
)

// Reviewer reviews one pull request: it picks the model, asks that model's backend, cleans the
// answer, derives the review event and posts the result as the review bot.
type Reviewer struct {
	cfg      Config
	gitea    *Gitea
	backends map[string]Backend
	selector *TypeSafe // nil: always the default model
	log      *slog.Logger
	metrics  *Metrics
	status   *Tracker

	mu    sync.RWMutex
	ready map[string]bool // backend -> prepared
}

func (r *Reviewer) setReady(backend string, ok bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.ready == nil {
		r.ready = map[string]bool{}
	}
	r.ready[backend] = ok
}

func (r *Reviewer) isReady(backend string) bool {
	r.mu.RLock()
	defer r.mu.RUnlock()
	return r.ready[backend]
}

// fallback is the first configured model whose backend is ready.
func (r *Reviewer) fallback() (Model, bool) {
	for _, m := range r.cfg.Models {
		if r.isReady(m.Backend) {
			return m, true
		}
	}
	return Model{}, false
}

func (r *Reviewer) Ready() bool { _, ok := r.fallback(); return ok }

// pick chooses the model for a pull request: the selector's choice when it is on and its choice
// is ready, else the first ready model.
func (r *Reviewer) pick(ctx context.Context, slug string, pr PR) (Model, error) {
	def, ok := r.fallback()
	if !ok {
		return Model{}, errors.New("no review backend is ready")
	}
	if r.selector == nil {
		return def, nil
	}
	files, err := r.gitea.Files(ctx, slug, pr.Number)
	if err != nil {
		r.log.Warn("selector: cannot list the changed files", "pr", key(slug, pr.Number), "err", err)
	}
	sctx, cancel := context.WithTimeout(ctx, time.Minute)
	defer cancel()
	c, err := r.selector.Select(sctx, slug, pr, files, r.cfg.Models)
	if err != nil {
		r.metrics.Inc("selections_total", "result", "failed")
		r.log.Warn("selector failed, using the default model", "pr", key(slug, pr.Number), "model", def.ID(), "err", err)
		return def, nil
	}
	for _, m := range r.cfg.Models {
		if m.ID() == c.Model && r.isReady(m.Backend) {
			r.metrics.Inc("selections_total", "result", "chosen")
			r.log.Info("model selected", "pr", key(slug, pr.Number), "model", m.ID(), "confidence", c.Confidence,
				"probabilities", c.Probabilities)
			return m, nil
		}
	}
	r.metrics.Inc("selections_total", "result", "unusable")
	r.log.Warn("selector chose a model that is not ready, using the default", "pr", key(slug, pr.Number),
		"chosen", c.Model, "model", def.ID())
	return def, nil
}

// Review reviews slug#n unless it needs none (closed, draft, WIP, already reviewed at this head).
func (r *Reviewer) Review(ctx context.Context, slug string, n int) error {
	if !r.Ready() {
		return errors.New("no review backend is ready")
	}
	pr, err := r.gitea.PR(ctx, slug, n)
	if err != nil {
		return err
	}
	reviews, err := r.gitea.Reviews(ctx, slug, n)
	if err != nil {
		return err
	}
	if why := eligible(pr, reviews, r.cfg.BotLogin); why != "ok" {
		r.log.Info("skipped", "pr", key(slug, n), "why", why)
		r.metrics.Inc("reviews_total", "result", "skipped")
		if rv, ok := headReview(pr, reviews, r.cfg.BotLogin); ok {
			r.status.Set(slug, n, pr.Head.SHA, PhasePosted, verdictWords(rv.State), rv.HTMLURL)
		} else {
			r.status.Set(slug, n, pr.Head.SHA, PhaseSkipped, why, pr.HTMLURL)
		}
		return nil
	}
	model, err := r.pick(ctx, slug, pr)
	if err != nil {
		return err
	}
	r.status.Set(slug, n, pr.Head.SHA, PhaseRunning, model.ID(), pr.HTMLURL)

	start := time.Now()
	task := fmt.Sprintf("Review pull request #%d of %s. Head commit: %s.", n, slug, pr.Head.SHA)
	ans, err := r.ask(ctx, model, task, func(detail string) {
		r.status.Set(slug, n, pr.Head.SHA, PhaseRetrying, detail, pr.HTMLURL)
	})
	if err != nil {
		r.metrics.Inc("reviews_total", "result", "failed")
		return err
	}
	r.log.Info("agent answered", "pr", key(slug, n), "model", model.ID(), "tool_calls", ans.ToolCalls,
		"seconds", int(time.Since(start).Seconds()))
	if len(ans.Violations) > 0 {
		r.metrics.Inc("reviews_total", "result", "violation")
		return fmt.Errorf("the agent called tools that are not read-only (%s); nothing is posted", strings.Join(ans.Violations, ", "))
	}
	text, verdict := splitVerdict(ans.Text)
	report := cleanReport(text)
	if report == "" {
		r.metrics.Inc("reviews_total", "result", "failed")
		return errors.New("the agent's answer holds no report")
	}

	// The review takes minutes; do not post into a PR that was closed meanwhile.
	if now, err := r.gitea.PR(ctx, slug, n); err == nil {
		if now.State != "open" {
			r.log.Info("closed during the review", "pr", key(slug, n))
			r.metrics.Inc("reviews_total", "result", "skipped")
			r.status.Set(slug, n, pr.Head.SHA, PhaseSkipped, "closed during the review", pr.HTMLURL)
			return nil
		}
		if now.Head.SHA != pr.Head.SHA {
			r.log.Info("new commits during the review, not posting the stale one", "pr", key(slug, n),
				"reviewed", pr.Head.SHA, "head", now.Head.SHA)
			r.metrics.Inc("reviews_total", "result", "stale")
			r.status.Set(slug, n, pr.Head.SHA, PhaseStale, "new head "+short(now.Head.SHA), pr.HTMLURL)
			return r.Review(ctx, slug, n)
		}
	}

	event := reviewEvent(verdict, report, pr.Additions+pr.Deletions+pr.ChangedFiles > 0)
	body := fmt.Sprintf("🤖 **Automated review** — model `%s`, commit `%s`.\n\n"+
		"> Automated review by the `review-bot`: it approves only when it found nothing red or yellow, "+
		"and a human still merges. Push new commits to get a fresh review.\n\n---\n\n%s\n\n<!-- %s -->\n",
		model.Name, pr.Head.SHA, report, marker(pr.Head.SHA))
	url, err := r.gitea.PostReview(ctx, slug, n, body, event, pr.Head.SHA)
	if err != nil {
		r.metrics.Inc("reviews_total", "result", "failed")
		return err
	}
	r.log.Info("posted", "pr", key(slug, n), "event", event)
	r.metrics.Inc("reviews_total", "result", "posted")
	if url == "" {
		url = pr.HTMLURL
	}
	r.status.Set(slug, n, pr.Head.SHA, PhasePosted, verdictWords(event), url)
	return nil
}

func short(sha string) string { return sha[:min(len(sha), 8)] }

// ask asks the model's backend, and asks again after a wait on a rate limit, a gateway failure or
// a broken response stream. Only the status code or the transport decides; the text of tool
// results is never inspected. onRetry hears of each scheduled retry.
func (r *Reviewer) ask(ctx context.Context, model Model, task string, onRetry func(detail string)) (Answer, error) {
	backend := r.backends[model.Backend]
	// One conversation's timeout plus the longest waits caps the whole review.
	ctx, cancelAll := context.WithTimeout(ctx, r.cfg.ReviewTimeout+time.Duration(r.cfg.RateLimitRetries)*maxLimitWait)
	defer cancelAll()
	policy := retrypolicy.NewBuilder[Answer]().
		HandleIf(func(_ Answer, err error) bool { return retryable(err) }).
		WithMaxRetries(r.cfg.RateLimitRetries).
		WithDelayFunc(func(e failsafe.ExecutionAttempt[Answer]) time.Duration {
			return limitDelay(r.cfg.RateLimitWait, e.Attempts(), e.LastError())
		}).
		OnRetryScheduled(func(e failsafe.ExecutionScheduledEvent[Answer]) {
			r.metrics.Inc("retries_total")
			var headers map[string]string
			reason := "broken stream"
			if api := asAPIError(e.LastError()); api != nil {
				headers = api.Limits
				reason = fmt.Sprintf("HTTP %d", api.Status)
			}
			r.log.Warn("asking again later", "wait", e.Delay.String(), "attempt", e.Attempts(),
				"limit_headers", headers, "err", e.LastError())
			onRetry(fmt.Sprintf("%s, retry %d/%d in %s", reason, e.Attempts(), r.cfg.RateLimitRetries,
				e.Delay.Round(time.Second)))
		}).
		ReturnLastFailure().
		Build()
	ans, err := failsafe.With(policy).WithContext(ctx).Get(func() (Answer, error) {
		// Every conversation gets the whole timeout; the waits between them do not eat into it.
		actx, cancel := context.WithTimeout(ctx, r.cfg.ReviewTimeout)
		defer cancel()
		return backend.Ask(actx, model.Name, task)
	})
	if retryable(err) {
		r.metrics.Inc("retry_giveups_total")
		return Answer{}, fmt.Errorf("still failing after %d retries: %w", r.cfg.RateLimitRetries, err)
	}
	return ans, err
}

// maxLimitWait is the longest wait before a retry, whether it comes from Retry-After or from the
// configured wait.
const maxLimitWait = 5 * time.Minute

// limitDelay is the wait before retry number `retry` (1 for the first): the provider's own
// Retry-After when the last answer carried one, else the configured wait times the retry number,
// capped at maxLimitWait either way.
func limitDelay(wait time.Duration, retry int, last error) time.Duration {
	if api := asAPIError(last); api != nil && retryable(last) && api.HasRetryAfter {
		return min(api.RetryAfter, maxLimitWait)
	}
	return min(wait*time.Duration(retry), maxLimitWait)
}

func key(slug string, n int) string { return fmt.Sprintf("%s#%d", slug, n) }

type job struct {
	slug string
	num  int
}

// Queue runs reviews on a fixed number of workers. A PR that is already queued or running is not
// queued twice.
type Queue struct {
	jobs     chan job
	reviewer *Reviewer
	log      *slog.Logger
	metrics  *Metrics

	mu       sync.Mutex
	inflight map[string]bool
	running  map[string]bool // inflight and picked up by a worker
	again    map[string]bool // asked for again while running; reviewed once more when it ends
	closed   bool
	wg       sync.WaitGroup
}

func newQueue(r *Reviewer, size int, log *slog.Logger, m *Metrics) *Queue {
	return &Queue{jobs: make(chan job, size), reviewer: r, log: log, metrics: m,
		inflight: map[string]bool{}, running: map[string]bool{}, again: map[string]bool{}}
}

// Enqueue returns "queued", "duplicate" or "full". A request for a PR whose review is already
// running is a duplicate, but the PR is reviewed once more when that review ends, because the
// request may be for a newer head commit. head is the commit the request is for, "" when unknown.
func (q *Queue) Enqueue(slug string, n int, head string) string {
	q.mu.Lock()
	defer q.mu.Unlock()
	return q.enqueueLocked(slug, n, head)
}

func (q *Queue) enqueueLocked(slug string, n int, head string) string {
	k := key(slug, n)
	if q.closed {
		return "full"
	}
	if q.inflight[k] {
		if q.running[k] {
			q.again[k] = true
		}
		return "duplicate"
	}
	select {
	case q.jobs <- job{slug, n}:
		q.inflight[k] = true
		q.metrics.Set("queue_depth", float64(len(q.jobs)))
		q.reviewer.status.Set(slug, n, head, PhaseQueued, "", "")
		return "queued"
	default:
		return "full"
	}
}

func (q *Queue) Start(ctx context.Context, workers int) {
	for i := 0; i < workers; i++ {
		q.wg.Add(1)
		go func() {
			defer q.wg.Done()
			for j := range q.jobs {
				q.metrics.Add("in_flight", 1)
				q.mu.Lock()
				q.running[key(j.slug, j.num)] = true
				q.mu.Unlock()
				// A review that started is allowed to finish, so it runs on its own context.
				if err := q.reviewer.Review(context.WithoutCancel(ctx), j.slug, j.num); err != nil {
					q.log.Error("review failed", "pr", key(j.slug, j.num), "err", err)
					q.reviewer.status.Set(j.slug, j.num, "", PhaseFailed, err.Error(), "")
				}
				q.metrics.Add("in_flight", -1)
				q.mu.Lock()
				k := key(j.slug, j.num)
				delete(q.inflight, k)
				delete(q.running, k)
				if q.again[k] {
					delete(q.again, k)
					q.log.Info("asked for again during the review", "pr", k, "result", q.enqueueLocked(j.slug, j.num, ""))
				}
				q.metrics.Set("queue_depth", float64(len(q.jobs)))
				q.mu.Unlock()
			}
		}()
	}
}

// Stop stops taking jobs and waits for the running ones, up to wait.
func (q *Queue) Stop(wait time.Duration) {
	q.mu.Lock()
	q.closed = true
	close(q.jobs)
	q.mu.Unlock()
	done := make(chan struct{})
	go func() { q.wg.Wait(); close(done) }()
	select {
	case <-done:
	case <-time.After(wait):
		q.log.Warn("reviews still running at shutdown")
	}
}

// Sweep queues every open PR of the configured repositories that has no review at its head
// commit. It is the safety net for webhook deliveries that were lost.
func (r *Reviewer) Sweep(ctx context.Context, q *Queue) {
	for _, slug := range r.cfg.Repos {
		prs, err := r.gitea.OpenPRs(ctx, slug)
		if err != nil {
			r.log.Error("sweep: cannot list pull requests", "repo", slug, "err", err)
			continue
		}
		for _, pr := range prs {
			reviews, err := r.gitea.Reviews(ctx, slug, pr.Number)
			if err != nil {
				r.log.Error("sweep: cannot read reviews", "pr", key(slug, pr.Number), "err", err)
				continue
			}
			if eligible(pr, reviews, r.cfg.BotLogin) == "ok" {
				r.log.Info("sweep: queued", "pr", key(slug, pr.Number), "result", q.Enqueue(slug, pr.Number, pr.Head.SHA))
			}
		}
	}
	r.metrics.Inc("sweeps_total")
}