agent-tools/internal/agent/watch.go

package agent

import (
	"errors"
	"fmt"
	"time"
)

// errPRGone marks a 404 from the PR lookup itself. A 404 from any other endpoint
// can be a proxy or ingress blip and is left to the ordinary failure cap.
var errPRGone = errors.New("PR no longer visible")

// IsPRGone reports whether err is a 404 from the PR lookup, meaning the PR is no
// longer visible rather than one endpoint being briefly unreachable.
func IsPRGone(err error) bool {
	return errors.Is(err, errPRGone)
}

// PRState is a point-in-time snapshot of the PR attributes watchpr tracks.
type PRState struct {
	Ref              PRRef  `json:"ref"`
	State            string `json:"state"` // open / closed
	Merged           bool   `json:"merged"`
	HeadSHA          string `json:"head_sha"`
	Mergeable        bool   `json:"mergeable"`
	CIStatus         string `json:"ci_status"` // success / pending / failure / error / ""
	NonAgentComments int    `json:"non_agent_comments"`
	Title            string `json:"title"`
	URL              string `json:"url"`
}

// FetchState builds a PRState for the given ref. agentLogin's comments are
// excluded from the non-agent comment count.
func FetchState(c *GiteaClient, ref PRRef, agentLogin string) (PRState, error) {
	pr, err := c.GetPR(ref.RepoPath(), ref.Number)
	if err != nil {
		if IsNotFound(err) {
			return PRState{}, fmt.Errorf("%w: %w", errPRGone, err)
		}
		return PRState{}, err
	}
	// A 404 here means the head commit is gone (branch deleted after a squash/
	// rebase merge); the PR object is still authoritative, so treat CI as absent
	// rather than discarding the merge signal and hanging the watch loop.
	ci, err := c.CommitStatus(ref.RepoPath(), pr.Head.Sha)
	if err != nil && !IsNotFound(err) {
		return PRState{}, err
	}
	comments, err := c.ListComments(ref.RepoPath(), ref.Number)
	if err != nil {
		return PRState{}, err
	}
	return PRState{
		Ref:              ref,
		State:            pr.State,
		Merged:           pr.Merged,
		HeadSHA:          pr.Head.Sha,
		Mergeable:        pr.Mergeable,
		CIStatus:         ci,
		NonAgentComments: countNonAgentComments(comments, agentLogin),
		Title:            pr.Title,
		URL:              pr.HTMLURL,
	}, nil
}

// StateFetcher fetches the current PRState for a ref. *GiteaClient satisfies it
// via its FetchState method; tests inject fakes.
type StateFetcher interface {
	FetchState(ref PRRef, agentLogin string) (PRState, error)
}

// FetchState makes *GiteaClient a StateFetcher.
func (c *GiteaClient) FetchState(ref PRRef, agentLogin string) (PRState, error) {
	return FetchState(c, ref, agentLogin)
}

// WatchResult is the change that ended a watch.
type WatchResult struct {
	Ref    PRRef
	Reason string
	State  PRState
}

// terminalState reports whether a PR has reached a final state from which no
// further meaningful change is possible, with a human-readable reason. Unlike a
// transition (see MeaningfulChange) this holds for a single snapshot, so it also
// catches a PR that is already merged/closed the moment watchpr starts.
func terminalState(st PRState) (bool, string) {
	if st.Merged {
		return true, "PR merged"
	}
	if st.State == "closed" {
		return true, "PR closed without merging"
	}
	return false, ""
}

// MaxPollFailures is how many consecutive failed polls of the same PR are
// tolerated before Watch gives up. The abort fires on the 20th failed tick, so
// at watchpr's default 60s interval a watch rides out ~19 minutes of failure.
const MaxPollFailures = 20

// Watch establishes a baseline for each ref, then polls on every tick until a
// tracked PR changes meaningfully, returning the first such change. A PR that is
// already terminal (merged/closed) at baseline is reported immediately rather
// than polled forever. Transient poll errors are handed to onError and the loop
// continues, but never blindly: a baseline fetch error, an authentication
// failure surviving a token re-mint, a 404 from the PR lookup itself (the repo
// is gone, renamed, or no longer visible), and MaxPollFailures consecutive
// failures of one PR all abort, because a watcher that sees nothing must not
// look healthy.
// onBaseline, if set, fires once after all baselines are captured and before the
// first tick.
func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Time, onBaseline func(), onError func(PRRef, error)) (WatchResult, error) {
	prev := make(map[string]PRState, len(refs))
	for _, ref := range refs {
		st, err := f.FetchState(ref, agentLogin)
		if err != nil {
			return WatchResult{}, err
		}
		if terminal, reason := terminalState(st); terminal {
			return WatchResult{Ref: ref, Reason: reason, State: st}, nil
		}
		prev[ref.String()] = st
	}
	if onBaseline != nil {
		onBaseline()
	}
	fails := make(map[string]int, len(refs))
	for range ticks {
		for _, ref := range refs {
			key := ref.String()
			cur, err := f.FetchState(ref, agentLogin)
			if err != nil {
				if IsAuthError(err) || IsPRGone(err) {
					return WatchResult{}, fmt.Errorf("polling %s: %w", key, err)
				}
				fails[key]++
				if onError != nil {
					onError(ref, err)
				}
				if fails[key] >= MaxPollFailures {
					return WatchResult{}, fmt.Errorf("polling %s: giving up after %d consecutive failures: %w", key, fails[key], err)
				}
				continue
			}
			fails[key] = 0
			if changed, reason := MeaningfulChange(prev[key], cur); changed {
				return WatchResult{Ref: ref, Reason: reason, State: cur}, nil
			}
			prev[key] = cur
		}
	}
	return WatchResult{}, nil
}

// countNonAgentComments counts comments authored by anyone other than agentLogin.
func countNonAgentComments(comments []Comment, agentLogin string) int {
	n := 0
	for _, cm := range comments {
		if cm.User.Login != agentLogin {
			n++
		}
	}
	return n
}

// isFailedCI reports whether a combined CI state is a terminal failure.
func isFailedCI(state string) bool {
	return state == "failure" || state == "error"
}

// MeaningfulChange compares a previous state to the current one and reports
// whether a change warrants alerting the operator, with a human-readable
// reason. Benign transitions (CI pending→success, the agent's own comments, an
// unchanged snapshot) return false.
//
// Alerting conditions:
//   - the PR merged
//   - the PR closed without merging
//   - a new comment from someone other than the agent
//   - CI transitioned into failure/error
//   - the PR lost mergeability (a conflict appeared) for two consecutive polls
func MeaningfulChange(prev, cur PRState) (bool, string) {
	if !prev.Merged && cur.Merged {
		return true, "PR merged"
	}
	// Closed (not merged): only alert on the open→closed edge.
	if prev.State == "open" && cur.State == "closed" && !cur.Merged {
		return true, "PR closed without merging"
	}
	if cur.NonAgentComments > prev.NonAgentComments {
		return true, "new comment from a non-agent user"
	}
	if isFailedCI(cur.CIStatus) && !isFailedCI(prev.CIStatus) {
		return true, "CI failed (" + cur.CIStatus + ")"
	}
	// Gitea computes mergeability asynchronously, so a PR can briefly report
	// mergeable=false right after a push. Require the loss to persist across two
	// consecutive polls (both prev and cur false, still open) before treating it
	// as a real conflict; a single false poll is debounced.
	if !prev.Mergeable && !cur.Mergeable && cur.State == "open" {
		return true, "PR lost mergeability (conflict)"
	}
	return false, ""
}