fix: bound scheduled GitHub review cleanup polling (#3884)

This commit is contained in:
Carlos Florêncio
2026-09-23 17:39:51 +01:00
committed by GitHub
parent a9e069ada4
commit 4dbeb72563
21 changed files with 2158 additions and 153 deletions
@@ -104,12 +104,12 @@ type taskSessionCheckerAdapter struct {
func (a *taskSessionCheckerAdapter) HasUserAuthoredMessage(ctx context.Context, taskID string) (bool, error) {
sessions, err := a.repo.ListTaskSessions(ctx, taskID)
if err != nil {
return false, err
return false, wrapGitHubTaskSessionCheckerError(err)
}
for _, sess := range sessions {
messages, err := a.repo.ListMessages(ctx, sess.ID)
if err != nil {
return false, err
return false, wrapGitHubTaskSessionCheckerError(err)
}
for _, m := range messages {
if m.AuthorType != models.MessageAuthorUser {
@@ -129,6 +129,13 @@ func (a *taskSessionCheckerAdapter) HasUserAuthoredMessage(ctx context.Context,
return false, nil
}
func wrapGitHubTaskSessionCheckerError(err error) error {
if errors.Is(err, taskrepo.ErrTaskNotFound) {
return fmt.Errorf("%w: %w", github.ErrTaskNotFound, err)
}
return err
}
// metaFlag returns true when meta[key] is a bool with value true. Returns
// false for missing keys, nil maps, non-bool values, and false values.
func metaFlag(meta map[string]interface{}, key string) bool {
@@ -2,20 +2,35 @@ package backendapp
import (
"context"
"errors"
"testing"
githubsvc "github.com/kandev/kandev/internal/github"
"github.com/kandev/kandev/internal/task/models"
taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite"
)
// fakeSessionMsgRepo is a minimal stub for taskSessionCheckerAdapter.repo
// — returns canned sessions + per-session messages from in-memory maps.
type fakeSessionMsgRepo struct {
sessions map[string][]*models.TaskSession
messages map[string][]*models.Message
sessions map[string][]*models.TaskSession
messages map[string][]*models.Message
sessionsErr error
}
func (r *fakeSessionMsgRepo) ListTaskSessions(_ context.Context, taskID string) ([]*models.TaskSession, error) {
return r.sessions[taskID], nil
return r.sessions[taskID], r.sessionsErr
}
func TestTaskSessionCheckerWrapsMissingTask(t *testing.T) {
adapter := &taskSessionCheckerAdapter{repo: &fakeSessionMsgRepo{sessionsErr: taskrepo.ErrTaskNotFound}}
_, err := adapter.HasUserAuthoredMessage(context.Background(), "deleted-task")
if !errors.Is(err, githubsvc.ErrTaskNotFound) {
t.Fatalf("HasUserAuthoredMessage error = %v, want github ErrTaskNotFound", err)
}
if !errors.Is(err, taskrepo.ErrTaskNotFound) {
t.Fatalf("wrapped error = %v, want to preserve task repository ErrTaskNotFound", err)
}
}
func (r *fakeSessionMsgRepo) ListMessages(_ context.Context, sessionID string) ([]*models.Message, error) {
+53 -7
View File
@@ -5,6 +5,11 @@ import (
"strings"
)
const (
reviewCleanupMetricClassAuth = "auth"
reviewCleanupMetricClassOther = "other"
)
// expvar maps published at package init, exposed via stdlib's /debug/vars
// handler in dev mode (AC-38). Process-local and dev-mode-visible only; the
// durable, snapshot-checkable signals are the writer-health invariants
@@ -18,13 +23,17 @@ var (
// (see poller_circuit.go): skips confirm a broken credential/config
// stopped costing a GitHub call every cycle; resets confirm a
// credential change (rotate/reconnect) is promptly detected.
authCircuitSkipsTotal = expvar.NewInt("github_pr_watch_auth_circuit_skips_total")
authCircuitResetsTotal = expvar.NewInt("github_pr_watch_auth_circuit_resets_total")
prWatchActive = expvar.NewInt("github_pr_watch_active")
prWatchSearching = expvar.NewInt("github_pr_watch_searching")
prWatchDuplicates = expvar.NewInt("github_pr_watch_duplicates")
prWatchOrphans = expvar.NewInt("github_pr_watch_orphans")
canonicalPollRequests = expvar.NewInt("github_pr_watch_canonical_poll_requests_total")
authCircuitSkipsTotal = expvar.NewInt("github_pr_watch_auth_circuit_skips_total")
authCircuitResetsTotal = expvar.NewInt("github_pr_watch_auth_circuit_resets_total")
prWatchActive = expvar.NewInt("github_pr_watch_active")
prWatchSearching = expvar.NewInt("github_pr_watch_searching")
prWatchDuplicates = expvar.NewInt("github_pr_watch_duplicates")
prWatchOrphans = expvar.NewInt("github_pr_watch_orphans")
canonicalPollRequests = expvar.NewInt("github_pr_watch_canonical_poll_requests_total")
reviewCleanupCircuitSkipsTotal = expvar.NewMap("github_review_cleanup_circuit_skips_total")
reviewCleanupCircuitResetsTotal = expvar.NewMap("github_review_cleanup_circuit_resets_total")
reviewCleanupFailuresTotal = expvar.NewMap("github_review_cleanup_failures_total")
reviewCleanupCoreQuotaSkipsTotal = expvar.NewInt("github_review_cleanup_core_quota_skips_total")
)
// outcomeMetricLabel builds a "k1=v1;k2=v2;..." label string for an expvar
@@ -66,6 +75,43 @@ func incAuthCircuitReset() {
authCircuitResetsTotal.Add(1)
}
func incReviewCleanupCircuitSkip(scope, class string) {
if !validReviewCleanupMetricScope(scope) {
return
}
reviewCleanupCircuitSkipsTotal.Add(outcomeMetricLabel("scope", scope, "class", boundedReviewCleanupMetricClass(class)), 1)
}
func incReviewCleanupCircuitReset(scope string) {
if validReviewCleanupMetricScope(scope) {
reviewCleanupCircuitResetsTotal.Add(outcomeMetricLabel("scope", scope), 1)
}
}
func incReviewCleanupFailure(scope, class string) {
if !validReviewCleanupMetricScope(scope) {
return
}
reviewCleanupFailuresTotal.Add(outcomeMetricLabel("scope", scope, "class", boundedReviewCleanupMetricClass(class)), 1)
}
func incReviewCleanupCoreQuotaSkip() {
reviewCleanupCoreQuotaSkipsTotal.Add(1)
}
func validReviewCleanupMetricScope(scope string) bool {
return scope == "workspace" || scope == "record"
}
func boundedReviewCleanupMetricClass(class string) string {
switch class {
case reviewCleanupMetricClassAuth, "config", "transient", "rate_limit":
return class
default:
return reviewCleanupMetricClassOther
}
}
func recordPRWatchCardinality(cardinality PRWatchCardinality) {
prWatchActive.Set(cardinality.Active)
prWatchSearching.Set(cardinality.Searching)
@@ -97,6 +97,45 @@ func TestBoolLabel(t *testing.T) {
}
}
func TestReviewCleanupMetrics(t *testing.T) {
poller, _, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41, 42)
client.feedbackErrors[41] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}
client.feedbackErrors[42] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/42"}
failuresBefore := readOutcomeCounter(t, reviewCleanupFailuresTotal, "scope=workspace;class=auth")
skipsBefore := readOutcomeCounter(t, reviewCleanupCircuitSkipsTotal, "scope=workspace;class=auth")
poller.checkReviewWatches(context.Background())
poller.checkReviewWatches(context.Background())
if got := readOutcomeCounter(t, reviewCleanupFailuresTotal, "scope=workspace;class=auth"); got <= failuresBefore {
t.Fatalf("workspace auth failure metric = %d, want an increment", got)
}
if got := readOutcomeCounter(t, reviewCleanupCircuitSkipsTotal, "scope=workspace;class=auth"); got <= skipsBefore {
t.Fatalf("workspace auth circuit skip metric = %d, want an increment", got)
}
incReviewCleanupFailure("workspace-id-must-not-be-a-label", reviewCleanupMetricClassAuth)
assertBoundedLabels := func(name string, metric *expvar.Map, allowed map[string]bool) {
t.Helper()
metric.Do(func(kv expvar.KeyValue) {
if !allowed[kv.Key] {
t.Errorf("%s contains unbounded label key %q", name, kv.Key)
}
})
}
classLabels := []string{reviewCleanupMetricClassAuth, "config", "transient", "rate_limit", reviewCleanupMetricClassOther}
var outcomeKeys = make(map[string]bool)
for _, scope := range []string{"workspace", "record"} {
for _, class := range classLabels {
outcomeKeys[outcomeMetricLabel("scope", scope, "class", class)] = true
}
}
assertBoundedLabels("review cleanup failures", reviewCleanupFailuresTotal, outcomeKeys)
assertBoundedLabels("review cleanup circuit skips", reviewCleanupCircuitSkipsTotal, outcomeKeys)
assertBoundedLabels("review cleanup circuit resets", reviewCleanupCircuitResetsTotal, map[string]bool{
outcomeMetricLabel("scope", "workspace"): true,
outcomeMetricLabel("scope", "record"): true,
})
}
// TestIncTaskPROutcomeSyncRecordsOnExpvarMap covers AC-38: a sync's populated
// state is recorded on the outcome-syncs expvar map under the label
// incTaskPROutcomeSync builds, so the "did the writer stop populating
@@ -130,6 +169,10 @@ func TestOutcomeExpvarMapsPublishedAtKnownNames(t *testing.T) {
"github_pr_watch_duplicates",
"github_pr_watch_orphans",
"github_pr_watch_canonical_poll_requests_total",
"github_review_cleanup_circuit_skips_total",
"github_review_cleanup_circuit_resets_total",
"github_review_cleanup_failures_total",
"github_review_cleanup_core_quota_skips_total",
}
for _, name := range expected {
if expvar.Get(name) == nil {
+24 -10
View File
@@ -112,7 +112,8 @@ type Poller struct {
// circuits tracks per-workspace auth/config backoff for the PR-monitor
// loop only (see poller_circuit.go). Safe for concurrent use; never nil.
circuits *pollerCircuits
circuits *pollerCircuits
reviewCleanupCircuits *reviewCleanupCircuits
fallbackMu sync.Mutex
fallbackWorkspaceCursor int
@@ -130,6 +131,7 @@ func NewPoller(svc *Service, eventBus bus.EventBus, log *logger.Logger) *Poller
eventBus: eventBus,
logger: log,
circuits: newPollerCircuits(),
reviewCleanupCircuits: newReviewCleanupCircuits(),
fallbackTargetCursors: make(map[string]int),
clock: time.Now,
}
@@ -926,7 +928,13 @@ func (p *Poller) checkReviewWatches(ctx context.Context) {
// sweep below is precisely what handles the "user disabled / deleted
// every watch" case. Fall through to the no-op loop and run the sweep.
p.logger.Debug("checking review watches", zap.Int("count", len(watches)))
cleanupCycle := newReviewCleanupCycle(ctx, p)
cleanupAdmission := scheduledReviewCleanupAdmission{
workspace: cleanupCycle.admitWorkspace,
record: cleanupCycle.admit,
}
for _, watch := range watches {
cleanupCycle.registerWatch(watch)
// A previous iteration in this same cycle may have exhausted the
// search bucket. Skip remaining per-watch checks instead of issuing
// another doomed search, but fall through to the orphan sweep below
@@ -959,26 +967,32 @@ func (p *Poller) checkReviewWatches(ctx context.Context) {
p.service.publishNewReviewPREvent(ctx, watch, pr)
}
// Clean up tasks for merged/closed PRs that the user hasn't opened.
if cleaned, err := p.service.CleanupMergedReviewTasks(ctx, watch); err != nil {
cleanup, err := p.service.cleanupScheduledMergedReviewTasks(ctx, watch, cleanupAdmission)
if err != nil {
p.logCleanupError("failed to cleanup merged review tasks", err,
zap.String("watch_id", watch.ID))
if cleanupBatchShouldStop(err) {
return
}
} else if cleaned > 0 {
} else {
cleanupCycle.apply(cleanup, watch.WorkspaceID, watch)
}
if cleanup.Deleted > 0 {
p.logger.Info("cleaned up merged review tasks",
zap.String("watch_id", watch.ID), zap.Int("deleted", cleaned))
zap.String("watch_id", watch.ID), zap.Int("deleted", cleanup.Deleted))
}
}
// Global orphan sweep: catches dedup rows whose watch was deleted or
// disabled. Without this pass those rows (and the tasks they reference)
// would never be re-examined, since the per-watch loop only iterates
// enabled watches.
if cleaned, err := p.service.CleanupAllOrphanedReviewTasks(ctx); err != nil {
cleanup, err := p.service.cleanupScheduledOrphanedReviewTasks(ctx, cleanupAdmission)
if err != nil {
p.logCleanupError("failed to sweep orphaned review tasks", err)
} else if cleaned > 0 {
p.logger.Info("swept orphaned review tasks", zap.Int("deleted", cleaned))
} else {
cleanupCycle.apply(cleanup, "", nil)
if cleanup.Deleted > 0 {
p.logger.Info("swept orphaned review tasks", zap.Int("deleted", cleanup.Deleted))
}
}
p.pruneReviewCleanupRecordCircuits(ctx)
}
// logCleanupError logs a poller cleanup failure at WARN, except when the error
@@ -16,9 +16,9 @@ func TestCleanupPollCycleSkipsRemainingCleanupAfterRateLimit(t *testing.T) {
configureCleanupRateLimitClient(t, svc, client, nil)
ctx := context.Background()
first := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true}
second := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true}
orphan := &ReviewWatch{WorkspaceID: "ws-1", Enabled: false}
first := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyAlways}
second := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyAlways}
orphan := &ReviewWatch{WorkspaceID: "ws-1", Enabled: false, CleanupPolicy: CleanupPolicyAlways}
for _, watch := range []*ReviewWatch{first, second, orphan} {
if err := store.CreateReviewWatch(ctx, watch); err != nil {
t.Fatalf("CreateReviewWatch: %v", err)
@@ -0,0 +1,271 @@
package github
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"strings"
"time"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/authcircuit"
)
const (
reviewCleanupCircuitWorkspace = "workspace"
reviewCleanupCircuitRecord = "record"
)
type reviewCleanupCycle struct {
poller *Poller
ctx context.Context
watches map[string]*ReviewWatch
watchChecked map[string]bool
workspaceFingerprints map[string]string
workspaceFPChecked map[string]bool
workspaceSkipRecorded map[string]bool
coreQuotaSkipRecorded bool
}
func newReviewCleanupCycle(ctx context.Context, poller *Poller) *reviewCleanupCycle {
return &reviewCleanupCycle{
poller: poller,
ctx: ctx,
watches: make(map[string]*ReviewWatch),
watchChecked: make(map[string]bool),
workspaceFingerprints: make(map[string]string),
workspaceFPChecked: make(map[string]bool),
workspaceSkipRecorded: make(map[string]bool),
}
}
func (c *reviewCleanupCycle) registerWatch(watch *ReviewWatch) {
if watch != nil && watch.ID != "" {
c.watches[watch.ID] = watch
c.watchChecked[watch.ID] = true
}
}
func (c *reviewCleanupCycle) admit(task *ReviewPRTask, tracker *RateTracker) (bool, bool) {
if task == nil {
return true, false
}
watch := c.reviewWatch(task.ReviewWatchID)
if watch == nil || watch.WorkspaceID == "" {
return false, false
}
connectionFingerprint := c.workspaceFingerprint(watch.WorkspaceID)
recordFingerprint := reviewCleanupRecordFingerprint(connectionFingerprint, watch)
if c.poller.reviewCleanupCircuits.refreshRecordFingerprint(task.ID, recordFingerprint) {
incReviewCleanupCircuitReset(reviewCleanupCircuitRecord)
}
if allowed, stopWorkspace := c.admitWorkspace(watch.WorkspaceID); !allowed {
return false, stopWorkspace
}
serviceTracker := c.poller.service.rateTracker
if tracker != nil && tracker != serviceTracker && tracker.WaitDuration(ResourceCore) > 0 {
c.recordCoreQuotaSkip()
return false, true
}
now := time.Now().UTC()
if open, class := c.poller.reviewCleanupCircuits.recordOpen(task.ID, now); open {
incReviewCleanupCircuitSkip(reviewCleanupCircuitRecord, string(class))
return false, false
}
return true, false
}
func (c *reviewCleanupCycle) admitWorkspace(workspaceID string) (bool, bool) {
if workspaceID == "" {
return false, false
}
c.workspaceFingerprint(workspaceID)
if c.poller.service != nil && c.poller.service.rateTracker != nil &&
c.poller.service.rateTracker.WaitDuration(ResourceCore) > 0 {
c.recordCoreQuotaSkip()
return false, true
}
now := time.Now().UTC()
if open, class := c.poller.reviewCleanupCircuits.workspaceOpen(workspaceID, now); open {
if !c.workspaceSkipRecorded[workspaceID] {
incReviewCleanupCircuitSkip(reviewCleanupCircuitWorkspace, string(class))
c.workspaceSkipRecorded[workspaceID] = true
}
return false, true
}
return true, false
}
func (c *reviewCleanupCycle) recordCoreQuotaSkip() {
if c.coreQuotaSkipRecorded {
return
}
incReviewCleanupCoreQuotaSkip()
c.coreQuotaSkipRecorded = true
}
func (c *reviewCleanupCycle) apply(
result scheduledReviewCleanupResult,
fallbackWorkspace string,
fallbackWatch *ReviewWatch,
) {
now := time.Now().UTC()
for _, success := range result.Successes {
workspaceID := success.WorkspaceID
if workspaceID == "" {
workspaceID = fallbackWorkspace
}
fingerprint := c.workspaceFingerprint(workspaceID)
c.poller.reviewCleanupCircuits.recordWorkspaceOutcome(
workspaceID, fingerprint, authcircuit.FailureClassNone, now,
)
if success.Task != nil {
watch := c.reviewWatch(success.Task.ReviewWatchID)
c.poller.reviewCleanupCircuits.recordRecordOutcome(
success.Task.ID, reviewCleanupRecordFingerprint(fingerprint, watch),
authcircuit.FailureClassNone, now,
)
}
}
for _, failure := range result.Failures {
workspaceID := failure.WorkspaceID
if workspaceID == "" {
workspaceID = fallbackWorkspace
}
watch := fallbackWatch
if failure.Task != nil {
if cached := c.reviewWatch(failure.Task.ReviewWatchID); cached != nil {
watch = cached
}
}
fingerprint := c.workspaceFingerprint(workspaceID)
class := classifyPollErr(failure.Err)
metricClass := reviewCleanupFailureMetricClass(failure.Err, class)
if failure.RecordScoped && failure.Task != nil {
incReviewCleanupFailure(reviewCleanupCircuitRecord, metricClass)
c.poller.reviewCleanupCircuits.recordRecordOutcome(
failure.Task.ID, reviewCleanupRecordFingerprint(fingerprint, watch), class, now,
)
} else {
incReviewCleanupFailure(reviewCleanupCircuitWorkspace, metricClass)
c.poller.reviewCleanupCircuits.recordWorkspaceOutcome(workspaceID, fingerprint, class, now)
}
c.poller.logger.Debug("scheduled GitHub review cleanup failed",
zap.String("workspace_id", workspaceID), zap.Error(failure.Err))
}
}
func (c *reviewCleanupCycle) reviewWatch(watchID string) *ReviewWatch {
if watchID == "" || c.watchChecked[watchID] {
return c.watches[watchID]
}
c.watchChecked[watchID] = true
if c.poller.service == nil || c.poller.service.store == nil {
return nil
}
watch, err := c.poller.service.store.GetReviewWatch(c.ctx, watchID)
if err != nil {
c.poller.logger.Debug("failed to load review watch for cleanup circuit",
zap.String("watch_id", watchID), zap.Error(err))
return nil
}
if watch != nil {
c.watches[watchID] = watch
}
return watch
}
func (c *reviewCleanupCycle) workspaceFingerprint(workspaceID string) string {
if workspaceID == "" || c.workspaceFPChecked[workspaceID] {
return c.workspaceFingerprints[workspaceID]
}
c.workspaceFPChecked[workspaceID] = true
if c.poller.service == nil {
return ""
}
fingerprint, err := c.poller.service.WorkspaceConnectionFingerprint(c.ctx, workspaceID)
if err != nil {
c.poller.logger.Debug("failed to refresh review cleanup connection fingerprint",
zap.String("workspace_id", workspaceID), zap.Error(err))
return ""
}
c.workspaceFingerprints[workspaceID] = fingerprint
if c.poller.reviewCleanupCircuits.refreshWorkspaceFingerprint(workspaceID, fingerprint) {
incReviewCleanupCircuitReset(reviewCleanupCircuitWorkspace)
}
return fingerprint
}
// reviewCleanupRecordFingerprint follows watch configuration; UpdatedAt and
// LastPolledAt change during routine polling and must not reset a record circuit.
func reviewCleanupRecordFingerprint(connectionFingerprint string, watch *ReviewWatch) string {
parts := make([]string, 0, 2)
if connectionFingerprint != "" {
parts = append(parts, "connection="+connectionFingerprint)
}
if watch != nil {
config, err := json.Marshal(struct {
WorkspaceID string
WorkflowID string
WorkflowStepID string
Repos []RepoFilter
AgentProfileID string
ExecutorProfileID string
Prompt string
ReviewScope string
CustomQuery string
TargetLogin string
Enabled bool
PollIntervalSeconds int
CleanupPolicy string
}{
WorkspaceID: watch.WorkspaceID,
WorkflowID: watch.WorkflowID,
WorkflowStepID: watch.WorkflowStepID,
Repos: watch.Repos,
AgentProfileID: watch.AgentProfileID,
ExecutorProfileID: watch.ExecutorProfileID,
Prompt: watch.Prompt,
ReviewScope: watch.ReviewScope,
CustomQuery: watch.CustomQuery,
TargetLogin: watch.TargetLogin,
Enabled: watch.Enabled,
PollIntervalSeconds: watch.PollIntervalSeconds,
CleanupPolicy: watch.CleanupPolicy,
})
if err == nil {
digest := sha256.Sum256(config)
parts = append(parts, "watch_config="+hex.EncodeToString(digest[:]))
}
}
return strings.Join(parts, ";")
}
func reviewCleanupFailureMetricClass(err error, class authcircuit.FailureClass) string {
var apiErr *GitHubAPIError
if errors.As(err, &apiErr) && isGitHubRateLimitAPIError(apiErr) {
return "rate_limit"
}
return string(class)
}
func (p *Poller) pruneReviewCleanupRecordCircuits(ctx context.Context) {
if p.service == nil || p.service.store == nil || p.reviewCleanupCircuits == nil {
return
}
rows, err := p.service.store.ListAllReviewPRTasks(ctx)
if err != nil {
p.logger.Debug("failed to prune review cleanup record circuits", zap.Error(err))
return
}
active := make(map[string]struct{}, len(rows))
for _, row := range rows {
if row != nil && row.ID != "" {
active[row.ID] = struct{}{}
}
}
p.reviewCleanupCircuits.pruneRecords(active)
}
@@ -0,0 +1,119 @@
package github
import (
"sync"
"time"
"github.com/kandev/kandev/internal/common/authcircuit"
)
type reviewCleanupCircuits struct {
mu sync.Mutex
workspaces map[string]authcircuit.State
records map[string]authcircuit.State
}
func newReviewCleanupCircuits() *reviewCleanupCircuits {
return &reviewCleanupCircuits{
workspaces: make(map[string]authcircuit.State),
records: make(map[string]authcircuit.State),
}
}
func (c *reviewCleanupCircuits) workspaceOpen(workspaceID string, now time.Time) (bool, authcircuit.FailureClass) {
c.mu.Lock()
defer c.mu.Unlock()
state := c.workspaces[workspaceID]
return state.Open(now), state.FailureClass
}
func (c *reviewCleanupCircuits) recordOpen(recordID string, now time.Time) (bool, authcircuit.FailureClass) {
c.mu.Lock()
defer c.mu.Unlock()
state := c.records[recordID]
return state.Open(now), state.FailureClass
}
func (c *reviewCleanupCircuits) refreshWorkspaceFingerprint(workspaceID, fingerprint string) bool {
return refreshReviewCleanupFingerprint(c.workspaces, c, workspaceID, fingerprint)
}
func (c *reviewCleanupCircuits) refreshRecordFingerprint(recordID, fingerprint string) bool {
return refreshReviewCleanupFingerprint(c.records, c, recordID, fingerprint)
}
func refreshReviewCleanupFingerprint(
states map[string]authcircuit.State,
circuits *reviewCleanupCircuits,
key, fingerprint string,
) bool {
if key == "" || fingerprint == "" {
return false
}
circuits.mu.Lock()
defer circuits.mu.Unlock()
state, exists := states[key]
if !exists {
return false
}
if state.ResetIfFingerprintChanged(fingerprint) {
delete(states, key)
return true
}
states[key] = state
return false
}
func (c *reviewCleanupCircuits) recordWorkspaceOutcome(
workspaceID, fingerprint string,
class authcircuit.FailureClass,
now time.Time,
) {
if workspaceID == "" {
return
}
c.mu.Lock()
defer c.mu.Unlock()
if class == authcircuit.FailureClassNone {
delete(c.workspaces, workspaceID)
return
}
state := c.workspaces[workspaceID]
if fingerprint != "" {
state.Fingerprint = fingerprint
}
state.RecordFailure(now, class, nil)
c.workspaces[workspaceID] = state
}
func (c *reviewCleanupCircuits) recordRecordOutcome(
recordID, fingerprint string,
class authcircuit.FailureClass,
now time.Time,
) {
if recordID == "" {
return
}
c.mu.Lock()
defer c.mu.Unlock()
if class == authcircuit.FailureClassNone {
delete(c.records, recordID)
return
}
state := c.records[recordID]
if fingerprint != "" {
state.Fingerprint = fingerprint
}
state.RecordFailure(now, class, nil)
c.records[recordID] = state
}
func (c *reviewCleanupCircuits) pruneRecords(active map[string]struct{}) {
c.mu.Lock()
defer c.mu.Unlock()
for recordID := range c.records {
if _, exists := active[recordID]; !exists {
delete(c.records, recordID)
}
}
}
+265 -89
View File
@@ -6,9 +6,41 @@ import (
"fmt"
"net/http"
"github.com/kandev/kandev/internal/common/authcircuit"
"go.uber.org/zap"
)
type scheduledReviewCleanupFailure struct {
Task *ReviewPRTask
WorkspaceID string
RecordScoped bool
Err error
}
type scheduledReviewCleanupSuccess struct {
Task *ReviewPRTask
WorkspaceID string
}
type scheduledReviewCleanupResult struct {
Deleted int
Failures []scheduledReviewCleanupFailure
Successes []scheduledReviewCleanupSuccess
}
type scheduledReviewCleanupAdmission struct {
workspace func(string) (allow, stopWorkspace bool)
record func(*ReviewPRTask, *RateTracker) (allow, stopWorkspace bool)
}
func (a scheduledReviewCleanupAdmission) allowWorkspace(workspaceID string) bool {
if a.workspace == nil {
return true
}
allowed, _ := a.workspace(workspaceID)
return allowed
}
func cleanupBatchAdmission(ctx context.Context, client Client, tracker *RateTracker, policy string) error {
if err := ctx.Err(); err != nil {
return err
@@ -81,13 +113,51 @@ func (s *Service) CleanupMergedReviewTasks(ctx context.Context, watch *ReviewWat
func(_ *ReviewPRTask) string { return policy })
}
func (s *Service) cleanupScheduledMergedReviewTasks(
ctx context.Context, watch *ReviewWatch, admission scheduledReviewCleanupAdmission,
) (scheduledReviewCleanupResult, error) {
if s.taskDeleter == nil {
return scheduledReviewCleanupResult{}, nil
}
if watch == nil || watch.WorkspaceID == "" {
return scheduledReviewCleanupResult{}, ErrGitHubWorkspaceRequired
}
prTasks, err := s.store.ListScheduledReviewPRTasksByWatch(ctx, watch.ID)
if err != nil {
return scheduledReviewCleanupResult{}, fmt.Errorf("list scheduled review PR tasks: %w", err)
}
if len(prTasks) == 0 {
return scheduledReviewCleanupResult{}, nil
}
if !admission.allowWorkspace(watch.WorkspaceID) {
return scheduledReviewCleanupResult{}, nil
}
resolved, err := s.resolveAutomationClient(ctx, watch.WorkspaceID, "", "")
if err != nil {
return scheduledReviewCleanupResult{Failures: []scheduledReviewCleanupFailure{{
Task: prTasks[0], WorkspaceID: watch.WorkspaceID, Err: err,
}}}, nil
}
policy := NormalizeCleanupPolicy(watch.CleanupPolicy)
result := s.cleanupScheduledReviewPRTaskBatch(ctx, resolved.Client, resolved.RateTracker,
prTasks, watch.WorkspaceID, func(_ *ReviewPRTask) string { return policy }, admission)
return result, nil
}
// CleanupAllOrphanedReviewTasks sweeps dedup rows whose watch is deleted or
// disabled. The per-watch poller loop already processes enabled-watch rows
// in the same cycle, so re-walking them here would double GitHub API
// consumption (the GetPRFeedback path bypasses prStatusCache). Returns the
// number of tasks deleted.
func (s *Service) CleanupAllOrphanedReviewTasks(ctx context.Context) (int, error) {
return s.cleanupAllReviewTasks(ctx, true)
result, err := s.cleanupAllReviewTasks(ctx, true, false, scheduledReviewCleanupAdmission{})
return result.Deleted, err
}
func (s *Service) cleanupScheduledOrphanedReviewTasks(
ctx context.Context, admission scheduledReviewCleanupAdmission,
) (scheduledReviewCleanupResult, error) {
return s.cleanupAllReviewTasks(ctx, true, true, admission)
}
// CleanupAllReviewTasks sweeps every dedup row across all review watches,
@@ -95,7 +165,8 @@ func (s *Service) CleanupAllOrphanedReviewTasks(ctx context.Context) (int, error
// settings-page cleanup button so the user can drain everything on demand
// without waiting for the next 5-minute poll cycle.
func (s *Service) CleanupAllReviewTasks(ctx context.Context) (int, error) {
return s.cleanupAllReviewTasks(ctx, false)
result, err := s.cleanupAllReviewTasks(ctx, false, false, scheduledReviewCleanupAdmission{})
return result.Deleted, err
}
// CleanupReviewTasksForWorkspace sweeps review dedup rows owned by watches in
@@ -142,16 +213,29 @@ func (s *Service) CleanupReviewTasksForWorkspace(ctx context.Context, workspaceI
// handle them in the same cycle).
//
//nolint:dupl // mirrors cleanupAllIssueTasks — different types, same orchestration
func (s *Service) cleanupAllReviewTasks(ctx context.Context, orphansOnly bool) (int, error) {
func (s *Service) cleanupAllReviewTasks(
ctx context.Context,
orphansOnly bool,
scheduled bool,
admission scheduledReviewCleanupAdmission,
) (scheduledReviewCleanupResult, error) {
if s.taskDeleter == nil {
return 0, nil
return scheduledReviewCleanupResult{}, nil
}
var (
prTasks []*ReviewPRTask
err error
)
if scheduled {
prTasks, err = s.store.ListScheduledReviewPRTasks(ctx)
} else {
prTasks, err = s.store.ListAllReviewPRTasks(ctx)
}
prTasks, err := s.store.ListAllReviewPRTasks(ctx)
if err != nil {
return 0, fmt.Errorf("list all review PR tasks: %w", err)
return scheduledReviewCleanupResult{}, fmt.Errorf("list all review PR tasks: %w", err)
}
if len(prTasks) == 0 {
return 0, nil
return scheduledReviewCleanupResult{}, nil
}
policyCache, enabledCache, unknownCache, workspaceCache := s.buildReviewWatchCaches(ctx, prTasks)
// Allocate a fresh slice rather than reusing prTasks' backing array
@@ -176,23 +260,53 @@ func (s *Service) cleanupAllReviewTasks(ctx context.Context, orphansOnly bool) (
}
candidatesByWorkspace[workspaceID] = append(candidatesByWorkspace[workspaceID], rpt)
}
deleted := 0
result := scheduledReviewCleanupResult{}
for workspaceID, candidates := range candidatesByWorkspace {
resolved, resolveErr := s.resolveAutomationClient(ctx, workspaceID, "", "")
if resolveErr != nil {
s.logger.Warn("skip review cleanup for unavailable workspace connection",
zap.String("workspace_id", workspaceID), zap.Error(resolveErr))
continue
}
batchDeleted, batchErr := s.cleanupReviewPRTaskBatch(ctx, resolved.Client, resolved.RateTracker, candidates, func(rpt *ReviewPRTask) string {
return policyCache[rpt.ReviewWatchID]
})
deleted += batchDeleted
if batchErr != nil {
return deleted, batchErr
workspaceResult, workspaceErr := s.cleanupReviewTasksForWorkspace(
ctx, workspaceID, candidates, scheduled, admission,
func(rpt *ReviewPRTask) string { return policyCache[rpt.ReviewWatchID] },
)
result.Deleted += workspaceResult.Deleted
result.Failures = append(result.Failures, workspaceResult.Failures...)
result.Successes = append(result.Successes, workspaceResult.Successes...)
if workspaceErr != nil {
return result, workspaceErr
}
}
return deleted, nil
return result, nil
}
func (s *Service) cleanupReviewTasksForWorkspace(
ctx context.Context,
workspaceID string,
candidates []*ReviewPRTask,
scheduled bool,
admission scheduledReviewCleanupAdmission,
resolvePolicy func(*ReviewPRTask) string,
) (scheduledReviewCleanupResult, error) {
if scheduled && !admission.allowWorkspace(workspaceID) {
return scheduledReviewCleanupResult{}, nil
}
resolved, err := s.resolveAutomationClient(ctx, workspaceID, "", "")
if err != nil {
s.logger.Warn("skip review cleanup for unavailable workspace connection",
zap.String("workspace_id", workspaceID), zap.Error(err))
if scheduled {
return scheduledReviewCleanupResult{Failures: []scheduledReviewCleanupFailure{{
Task: candidates[0], WorkspaceID: workspaceID, Err: err,
}}}, nil
}
return scheduledReviewCleanupResult{}, nil
}
if scheduled {
return s.cleanupScheduledReviewPRTaskBatch(
ctx, resolved.Client, resolved.RateTracker, candidates, workspaceID, resolvePolicy, admission,
), nil
}
deleted, err := s.cleanupReviewPRTaskBatch(
ctx, resolved.Client, resolved.RateTracker, candidates, resolvePolicy,
)
return scheduledReviewCleanupResult{Deleted: deleted}, err
}
// buildReviewWatchCaches loads the cleanup policy + enabled flag for each
@@ -261,31 +375,6 @@ func (s *Service) cleanupReviewPRTaskBatch(
if err := cleanupBatchAdmission(ctx, client, tracker, policy); err != nil {
return deleted, err
}
// Orphan reservation: process was killed after ReserveReviewPRTask
// succeeded but before AssignReviewPRTaskID ran, so task_id is empty
// and there is no task to delete. Clean up the dedup row once the PR
// reaches a terminal state, same gating as the normal path.
if rpt.TaskID == "" {
// Orphan reservation row — no task was ever created (process
// crashed between Reserve and Assign). Clean it up but DON'T
// increment `deleted`: the count is reported back to the
// settings-page toast as "Deleted N tasks", and these rows
// never had an associated task.
should, _, err := s.shouldDeleteReviewTaskWithClient(ctx, client, rpt, policy)
if err != nil {
if cleanupBatchShouldStop(err) {
return deleted, err
}
continue
}
if should {
if err := s.store.DeleteReviewPRTask(ctx, rpt.ID); err != nil {
s.logger.Warn("failed to delete orphan reservation row",
zap.String("dedup_id", rpt.ID), zap.Error(err))
}
}
continue
}
shouldDelete, reason, err := s.shouldDeleteReviewTaskWithClient(ctx, client, rpt, policy)
if err != nil {
if cleanupBatchShouldStop(err) {
@@ -296,41 +385,126 @@ func (s *Service) cleanupReviewPRTaskBatch(
if !shouldDelete {
continue
}
if err := s.deleteTaskWithReason(ctx, rpt.TaskID, reason); err != nil {
if isTaskNotFound(err) {
// Task already deleted; clean up the orphaned dedup record.
if err := s.store.DeleteReviewPRTask(ctx, rpt.ID); err != nil {
s.logger.Warn("failed to delete orphan dedup row after task-not-found",
zap.String("dedup_id", rpt.ID), zap.Error(err))
continue
}
deleted++
continue
}
s.logger.Warn("failed to delete review PR task",
zap.String("task_id", rpt.TaskID), zap.Error(err))
continue
}
if err := s.store.DeleteReviewPRTask(ctx, rpt.ID); err != nil {
// Task is gone but dedup row survived: log Warn and DON'T
// increment deleted (the next sweep cycle will retry and the
// settings-page toast stays accurate).
s.logger.Warn("deleted task but failed to remove dedup row",
zap.String("task_id", rpt.TaskID),
zap.String("dedup_id", rpt.ID), zap.Error(err))
continue
}
s.logger.Info("deleted review task",
zap.String("task_id", rpt.TaskID),
zap.String("reason", reason),
zap.String("policy", policy),
zap.Int("pr_number", rpt.PRNumber),
zap.String("repo", rpt.RepoOwner+"/"+rpt.RepoName))
deleted++
deleted += s.applyReviewPRTaskDeletion(ctx, rpt, policy, reason)
}
return deleted, nil
}
func (s *Service) cleanupScheduledReviewPRTaskBatch(
ctx context.Context,
client Client,
tracker *RateTracker,
prTasks []*ReviewPRTask,
workspaceID string,
resolvePolicy func(*ReviewPRTask) string,
admission scheduledReviewCleanupAdmission,
) scheduledReviewCleanupResult {
result := scheduledReviewCleanupResult{}
for _, rpt := range prTasks {
policy := resolvePolicy(rpt)
if policy == CleanupPolicyNever || s.retainAutoReviewTaskBeforeFeedback(ctx, rpt, policy) {
continue
}
if admission.record != nil {
allowed, stopWorkspace := admission.record(rpt, tracker)
if !allowed {
if stopWorkspace {
break
}
continue
}
}
if err := cleanupBatchAdmission(ctx, client, tracker, policy); err != nil {
result.Failures = append(result.Failures, scheduledReviewCleanupFailure{
Task: rpt, WorkspaceID: workspaceID, Err: err,
})
break
}
shouldDelete, reason, err := s.shouldDeleteReviewTaskWithClient(ctx, client, rpt, policy)
if err != nil {
class := classifyPollErr(err)
result.Failures = append(result.Failures, scheduledReviewCleanupFailure{
Task: rpt, WorkspaceID: workspaceID,
RecordScoped: class == authcircuit.FailureClassConfig, Err: err,
})
if class != authcircuit.FailureClassConfig {
break
}
continue
}
result.Successes = append(result.Successes, scheduledReviewCleanupSuccess{
Task: rpt, WorkspaceID: workspaceID,
})
if shouldDelete {
result.Deleted += s.applyReviewPRTaskDeletion(ctx, rpt, policy, reason)
}
}
return result
}
func (s *Service) applyReviewPRTaskDeletion(ctx context.Context, rpt *ReviewPRTask, policy, reason string) int {
if rpt.TaskID == "" {
if err := s.store.DeleteReviewPRTask(ctx, rpt.ID); err != nil {
s.logger.Warn("failed to delete orphan reservation row",
zap.String("dedup_id", rpt.ID), zap.Error(err))
}
return 0
}
if err := s.deleteTaskWithReason(ctx, rpt.TaskID, reason); err != nil {
if isTaskNotFound(err) {
if err := s.store.DeleteReviewPRTask(ctx, rpt.ID); err != nil {
s.logger.Warn("failed to delete orphan dedup row after task-not-found",
zap.String("dedup_id", rpt.ID), zap.Error(err))
return 0
}
return 1
}
s.logger.Warn("failed to delete review PR task",
zap.String("task_id", rpt.TaskID), zap.Error(err))
return 0
}
if err := s.store.DeleteReviewPRTask(ctx, rpt.ID); err != nil {
s.logger.Warn("deleted task but failed to remove dedup row",
zap.String("task_id", rpt.TaskID), zap.String("dedup_id", rpt.ID), zap.Error(err))
return 0
}
s.logger.Info("deleted review task",
zap.String("task_id", rpt.TaskID), zap.String("reason", reason),
zap.String("policy", policy), zap.Int("pr_number", rpt.PRNumber),
zap.String("repo", rpt.RepoOwner+"/"+rpt.RepoName))
return 1
}
func (s *Service) retainAutoReviewTaskBeforeFeedback(ctx context.Context, rpt *ReviewPRTask, policy string) bool {
if policy != CleanupPolicyAuto || rpt == nil || rpt.TaskID == "" {
return false
}
if s.store == nil {
return true
}
if s.taskSessionChecker != nil {
hasUserMsg, err := s.taskSessionChecker.HasUserAuthoredMessage(ctx, rpt.TaskID)
if err != nil {
if errors.Is(err, ErrTaskNotFound) {
return false
}
s.logger.Debug("failed to check task user messages before scheduled cleanup",
zap.String("task_id", rpt.TaskID), zap.Error(err))
return true
}
if hasUserMsg {
return true
}
}
enabled, err := s.HasEnabledTaskPRAgentPrompts(ctx, rpt.TaskID)
if err != nil {
s.logger.Debug("failed to check task PR agent prompt options before scheduled cleanup",
zap.String("task_id", rpt.TaskID), zap.Error(err))
return true
}
return enabled || s.taskSessionChecker == nil
}
// shouldDeleteReviewTask checks whether a review PR task is eligible for
// cleanup under the supplied policy. Returns true + a short reason on hit.
// - CleanupPolicyNever → always false.
@@ -376,6 +550,20 @@ func (s *Service) shouldDeleteReviewTaskWithClient(
if policy == CleanupPolicyAlways || rpt.TaskID == "" {
return true, reason, nil
}
if s.taskSessionChecker != nil {
hasUserMsg, err := s.taskSessionChecker.HasUserAuthoredMessage(ctx, rpt.TaskID)
if err != nil {
if errors.Is(err, ErrTaskNotFound) {
return true, reason, nil
}
s.logger.Debug("failed to check task user messages",
zap.String("task_id", rpt.TaskID), zap.Error(err))
return false, "", nil
}
if hasUserMsg {
return false, "", nil
}
}
if s.store != nil {
enabled, err := s.HasEnabledTaskPRAgentPrompts(ctx, rpt.TaskID)
if err != nil {
@@ -387,18 +575,6 @@ func (s *Service) shouldDeleteReviewTaskWithClient(
return false, "", nil
}
}
if s.taskSessionChecker == nil {
return true, reason, nil
}
hasUserMsg, err := s.taskSessionChecker.HasUserAuthoredMessage(ctx, rpt.TaskID)
if err != nil {
s.logger.Debug("failed to check task user messages",
zap.String("task_id", rpt.TaskID), zap.Error(err))
return false, "", nil
}
if hasUserMsg {
return false, "", nil
}
return true, reason, nil
}
@@ -460,6 +460,7 @@ func TestCheckReviewWatches_RunsOrphanSweepWhenNoEnabledWatches(t *testing.T) {
rec := &recordingTaskDeleter{}
svc.SetTaskDeleter(rec)
svc.SetTaskSessionChecker(&recordingSessionChecker{})
poller.checkReviewWatches(ctx)
@@ -0,0 +1,760 @@
package github
import (
"context"
"errors"
"fmt"
"testing"
"time"
)
type countingReviewCleanupClient struct {
Client
feedbackCalls int
feedbackNumbers []int
feedbackErrors map[int]error
reviewPRNumbers map[int]bool
}
type missingTaskReviewDeleter struct {
calls []string
}
type taskNotFoundReviewDeleter struct{}
func (taskNotFoundReviewDeleter) DeleteTask(context.Context, string) error {
return ErrTaskNotFound
}
type countingReviewCleanupAutomationProvider struct {
client Client
calls int
}
func (p *countingReviewCleanupAutomationProvider) ResolveAutomation(
_ context.Context,
connection *WorkspaceConnection,
_ ResolveCredentialRequest,
) (*ResolvedCredential, error) {
p.calls++
return &ResolvedCredential{
Client: p.client,
Capabilities: allTokenCapabilities(),
Principal: AuthPrincipal{
Kind: AuthPrincipalHuman,
Source: ConnectionSourcePAT,
Login: connection.Login,
},
}, nil
}
func (d *missingTaskReviewDeleter) DeleteTask(_ context.Context, taskID string) error {
d.calls = append(d.calls, taskID)
if taskID == "hard-deleted-task-enabled" || taskID == "hard-deleted-task-disabled" {
return ErrTaskNotFound
}
return nil
}
func (c *countingReviewCleanupClient) GetPRFeedback(
ctx context.Context, owner, repo string, number int,
) (*PRFeedback, error) {
c.recordFeedbackCall(number)
if err := c.feedbackErrors[number]; err != nil {
return nil, err
}
return c.Client.GetPRFeedback(ctx, owner, repo, number)
}
func (c *countingReviewCleanupClient) GetPR(ctx context.Context, owner, repo string, number int) (*PR, error) {
c.recordFeedbackCall(number)
if err := c.feedbackErrors[number]; err != nil {
return nil, err
}
return c.Client.GetPR(ctx, owner, repo, number)
}
func (c *countingReviewCleanupClient) recordFeedbackCall(number int) {
if c.reviewPRNumbers != nil && !c.reviewPRNumbers[number] {
return
}
c.feedbackCalls++
c.feedbackNumbers = append(c.feedbackNumbers, number)
}
func setupArchivedReviewCleanup(t *testing.T, watchEnabled bool) (*Poller, *Service, *Store, *ReviewWatch, *countingReviewCleanupClient) {
t.Helper()
poller, service, mockClient, store := setupPollerTest(t)
ctx := context.Background()
watch := &ReviewWatch{
ID: "review-watch",
WorkspaceID: "ws-1",
Enabled: watchEnabled,
CleanupPolicy: CleanupPolicyAuto,
}
if err := store.CreateReviewWatch(ctx, watch); err != nil {
t.Fatalf("create review watch: %v", err)
}
seedTask(t, store, "archived-review-task", true)
if err := store.CreateReviewPRTask(ctx, &ReviewPRTask{
ReviewWatchID: watch.ID,
RepoOwner: "acme",
RepoName: "widget",
PRNumber: 42,
PRURL: "https://github.com/acme/widget/pull/42",
TaskID: "archived-review-task",
}); err != nil {
t.Fatalf("create review PR task: %v", err)
}
mockClient.AddPR(&PR{
Number: 42,
State: prStateMerged,
RepoOwner: "acme",
RepoName: "widget",
})
service.SetTaskSessionChecker(&recordingSessionChecker{hasUserMsg: true})
service.SetTaskDeleter(&recordingTaskDeleter{})
countingClient := &countingReviewCleanupClient{
Client: mockClient, reviewPRNumbers: map[int]bool{42: true},
}
configureTestWorkspaceAuth(t, service, countingClient, testWorkspaceID, "ws-1", "ws1")
return poller, service, store, watch, countingClient
}
func assertReviewTaskIsPreserved(t *testing.T, store *Store, watch *ReviewWatch, taskID string) {
t.Helper()
rows, err := store.ListReviewPRTasksByWatch(context.Background(), watch.ID)
if err != nil {
t.Fatalf("list review PR tasks: %v", err)
}
if len(rows) != 1 || rows[0].TaskID != taskID {
t.Fatalf("review task rows = %+v, want task %q's dedup row retained", rows, taskID)
}
}
func TestCleanupMergedReviewTasks_ArchivedEnabledWatch(t *testing.T) {
poller, _, store, watch, client := setupArchivedReviewCleanup(t, true)
for range 2 {
poller.checkReviewWatches(context.Background())
}
if client.feedbackCalls != 0 {
t.Errorf("feedback calls = %d, want 0 for an archived task", client.feedbackCalls)
}
assertReviewTaskIsPreserved(t, store, watch, "archived-review-task")
}
func TestCleanupAllOrphanedReviewTasks_ArchivedDisabledWatch(t *testing.T) {
poller, _, store, watch, client := setupArchivedReviewCleanup(t, false)
for range 2 {
poller.checkReviewWatches(context.Background())
}
if client.feedbackCalls != 0 {
t.Errorf("feedback calls = %d, want 0 for an archived task", client.feedbackCalls)
}
assertReviewTaskIsPreserved(t, store, watch, "archived-review-task")
}
func setupActiveReviewCleanup(
t *testing.T, checker *recordingSessionChecker, lifecyclePrompt bool,
) (*Poller, *Service, *Store, *ReviewWatch, *countingReviewCleanupClient) {
t.Helper()
poller, service, mockClient, store := setupPollerTest(t)
ctx := context.Background()
watch := &ReviewWatch{
ID: "review-watch", WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyAuto,
}
if err := store.CreateReviewWatch(ctx, watch); err != nil {
t.Fatalf("create review watch: %v", err)
}
seedTask(t, store, "active-review-task", false)
if err := store.CreateReviewPRTask(ctx, &ReviewPRTask{
ReviewWatchID: watch.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: 42,
PRURL: "https://github.com/acme/widget/pull/42", TaskID: "active-review-task",
}); err != nil {
t.Fatalf("create review PR task: %v", err)
}
mockClient.AddPR(&PR{
Number: 42, State: prStateMerged, RepoOwner: "acme", RepoName: "widget",
})
if lifecyclePrompt {
enabled := true
if _, err := store.UpdateTaskPRAutomationOptions(ctx, "active-review-task", "", 42,
TaskPRAutomationOptionsPatch{PromptOnMerged: &enabled}, false); err != nil {
t.Fatalf("enable merged lifecycle prompt: %v", err)
}
}
service.SetTaskSessionChecker(checker)
service.SetTaskDeleter(&recordingTaskDeleter{})
countingClient := &countingReviewCleanupClient{
Client: mockClient, reviewPRNumbers: map[int]bool{42: true},
}
configureTestWorkspaceAuth(t, service, countingClient, testWorkspaceID, "ws-1", "ws1")
return poller, service, store, watch, countingClient
}
func setupOpenReviewCleanupRecords(
t *testing.T, policy string, prNumbers ...int,
) (*Poller, *Service, *Store, *ReviewWatch, *countingReviewCleanupClient) {
t.Helper()
poller, service, mockClient, store := setupPollerTest(t)
ctx := context.Background()
watch := &ReviewWatch{
ID: "disabled-review-watch", WorkspaceID: "ws-1", Enabled: false, CleanupPolicy: policy,
}
if err := store.CreateReviewWatch(ctx, watch); err != nil {
t.Fatalf("create review watch: %v", err)
}
for _, number := range prNumbers {
taskID := fmt.Sprintf("review-task-%d", number)
seedTask(t, store, taskID, false)
if err := store.CreateReviewPRTask(ctx, &ReviewPRTask{
ReviewWatchID: watch.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: number, TaskID: taskID,
}); err != nil {
t.Fatalf("create review task #%d: %v", number, err)
}
mockClient.AddPR(&PR{
Number: number, State: "open", RepoOwner: "acme", RepoName: "widget",
})
}
service.SetTaskSessionChecker(&recordingSessionChecker{})
service.SetTaskDeleter(&recordingTaskDeleter{})
countingClient := &countingReviewCleanupClient{
Client: mockClient, feedbackErrors: make(map[int]error), reviewPRNumbers: make(map[int]bool),
}
for _, number := range prNumbers {
countingClient.reviewPRNumbers[number] = true
}
configureTestWorkspaceAuth(t, service, countingClient, testWorkspaceID, "ws-1", "ws1")
return poller, service, store, watch, countingClient
}
func TestCleanupReviewTasks_AutoRetentionSkipsFeedback(t *testing.T) {
cases := []struct {
name string
checker *recordingSessionChecker
lifecyclePrompt bool
}{
{name: "user message", checker: &recordingSessionChecker{hasUserMsg: true}},
{name: "lifecycle prompt", checker: &recordingSessionChecker{}, lifecyclePrompt: true},
{name: "failed user message check", checker: &recordingSessionChecker{err: errors.New("message lookup failed")}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
poller, _, _, _, client := setupActiveReviewCleanup(t, tc.checker, tc.lifecyclePrompt)
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 0 {
t.Errorf("feedback calls = %d, want 0 when Auto cleanup retains the task", client.feedbackCalls)
}
})
}
}
func TestCleanupReviewTasks_AutoTaskNotFoundRemainsEligible(t *testing.T) {
poller, service, store, watch, client := setupActiveReviewCleanup(
t, &recordingSessionChecker{err: ErrTaskNotFound}, true,
)
service.SetTaskDeleter(taskNotFoundReviewDeleter{})
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("feedback calls for missing task = %d, want one", client.feedbackCalls)
}
rows, err := store.ListReviewPRTasksByWatch(context.Background(), watch.ID)
if err != nil {
t.Fatalf("list review rows after missing-task cleanup: %v", err)
}
if len(rows) != 0 {
t.Fatalf("review rows after missing-task cleanup = %+v, want none", rows)
}
}
func TestListScheduledReviewPRTasks(t *testing.T) {
_, _, _, store := setupPollerTest(t)
ctx := context.Background()
watch := &ReviewWatch{ID: "review-watch", WorkspaceID: "ws-1", Enabled: true}
if err := store.CreateReviewWatch(ctx, watch); err != nil {
t.Fatalf("create review watch: %v", err)
}
seedTask(t, store, "archived-task", true)
seedTask(t, store, "active-task", false)
rows := []struct {
taskID string
prNumber int
}{
{taskID: "archived-task", prNumber: 41},
{taskID: "active-task", prNumber: 42},
{taskID: "", prNumber: 43},
{taskID: "hard-deleted-task", prNumber: 44},
}
for _, row := range rows {
if err := store.CreateReviewPRTask(ctx, &ReviewPRTask{
ReviewWatchID: watch.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: row.prNumber,
TaskID: row.taskID,
}); err != nil {
t.Fatalf("create review PR task #%d: %v", row.prNumber, err)
}
}
perWatch, err := store.ListScheduledReviewPRTasksByWatch(ctx, watch.ID)
if err != nil {
t.Fatalf("list scheduled review tasks by watch: %v", err)
}
all, err := store.ListScheduledReviewPRTasks(ctx)
if err != nil {
t.Fatalf("list all scheduled review tasks: %v", err)
}
if len(perWatch) != 3 || len(all) != 3 {
t.Fatalf("scheduled rows: per_watch=%d all=%d, want 3 each", len(perWatch), len(all))
}
for _, candidates := range [][]*ReviewPRTask{perWatch, all} {
eligible := make(map[string]bool, len(candidates))
for _, row := range candidates {
eligible[row.TaskID] = true
}
if len(eligible) != 3 || !eligible["active-task"] || !eligible[""] || !eligible["hard-deleted-task"] {
t.Errorf("scheduled candidates = %v, want active, reservation, and hard-deleted task", eligible)
}
}
complete, err := store.ListAllReviewPRTasks(ctx)
if err != nil {
t.Fatalf("list complete review task inventory: %v", err)
}
if len(complete) != 4 {
t.Errorf("complete inventory rows = %d, want archived rows to remain visible", len(complete))
}
}
func TestCleanupReviewTasks_ScheduledEligibilityAndOrphanRecovery(t *testing.T) {
poller, service, mockClient, store := setupPollerTest(t)
ctx := context.Background()
enabledWatch := &ReviewWatch{
ID: "enabled-watch", WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyAuto,
}
disabledWatch := &ReviewWatch{
ID: "disabled-watch", WorkspaceID: "ws-1", Enabled: false, CleanupPolicy: CleanupPolicyAuto,
}
for _, watch := range []*ReviewWatch{enabledWatch, disabledWatch} {
if err := store.CreateReviewWatch(ctx, watch); err != nil {
t.Fatalf("create review watch %q: %v", watch.ID, err)
}
}
client := &countingReviewCleanupClient{
Client: mockClient, reviewPRNumbers: make(map[int]bool),
}
deleter := &missingTaskReviewDeleter{}
service.SetTaskSessionChecker(&recordingSessionChecker{})
service.SetTaskDeleter(deleter)
configureTestWorkspaceAuth(t, service, client, testWorkspaceID, "ws-1", "ws1")
rows := []struct {
watch *ReviewWatch
taskID string
archived bool
missing bool
prNumber int
}{
{watch: enabledWatch, taskID: "archived-task-enabled", archived: true, prNumber: 41},
{watch: enabledWatch, taskID: "active-task-enabled", prNumber: 42},
{watch: enabledWatch, taskID: "", prNumber: 43},
{watch: enabledWatch, taskID: "hard-deleted-task-enabled", missing: true, prNumber: 44},
{watch: disabledWatch, taskID: "archived-task-disabled", archived: true, prNumber: 51},
{watch: disabledWatch, taskID: "active-task-disabled", prNumber: 52},
{watch: disabledWatch, taskID: "", prNumber: 53},
{watch: disabledWatch, taskID: "hard-deleted-task-disabled", missing: true, prNumber: 54},
}
for _, row := range rows {
client.reviewPRNumbers[row.prNumber] = true
if row.taskID != "" && !row.missing {
seedTask(t, store, row.taskID, row.archived)
}
if err := store.CreateReviewPRTask(ctx, &ReviewPRTask{
ReviewWatchID: row.watch.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: row.prNumber,
PRURL: "https://github.com/acme/widget/pull/" + fmt.Sprint(row.prNumber), TaskID: row.taskID,
}); err != nil {
t.Fatalf("create review PR task #%d: %v", row.prNumber, err)
}
mockClient.AddPR(&PR{
Number: row.prNumber, State: prStateMerged, RepoOwner: "acme", RepoName: "widget",
})
}
poller.checkReviewWatches(ctx)
poller.checkReviewWatches(ctx)
if client.feedbackCalls != 6 {
t.Fatalf("feedback calls = %d, want 6 for active tasks, reservations, and hard-deleted tasks", client.feedbackCalls)
}
if len(deleter.calls) != 4 {
t.Fatalf("task deletion calls = %v, want only the 2 active and 2 hard-deleted tasks", deleter.calls)
}
remaining, err := store.ListAllReviewPRTasks(ctx)
if err != nil {
t.Fatalf("list review PR tasks: %v", err)
}
remainingTasks := make(map[string]bool, len(remaining))
for _, row := range remaining {
remainingTasks[row.TaskID] = true
}
if len(remainingTasks) != 2 || !remainingTasks["archived-task-enabled"] || !remainingTasks["archived-task-disabled"] {
t.Fatalf("remaining review task IDs = %v, want both archived rows only", remainingTasks)
}
}
func TestCleanupReviewTasks_UnarchiveMakesExistingRecordEligible(t *testing.T) {
poller, service, store, watch, client := setupArchivedReviewCleanup(t, true)
rows, err := store.ListReviewPRTasksByWatch(context.Background(), watch.ID)
if err != nil || len(rows) != 1 {
t.Fatalf("initial review task rows = %+v, err = %v", rows, err)
}
recordID := rows[0].ID
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 0 {
t.Fatalf("archived task feedback calls = %d, want 0", client.feedbackCalls)
}
if _, err := store.db.Exec(`UPDATE tasks SET archived_at = NULL WHERE id = ?`, "archived-review-task"); err != nil {
t.Fatalf("unarchive task: %v", err)
}
service.SetTaskSessionChecker(&recordingSessionChecker{})
mockClient := client.Client.(*MockClient)
mockClient.AddPR(&PR{Number: 42, State: "open", RepoOwner: "acme", RepoName: "widget"})
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("feedback calls after unarchive = %d, want 1", client.feedbackCalls)
}
rows, err = store.ListReviewPRTasksByWatch(context.Background(), watch.ID)
if err != nil || len(rows) != 1 || rows[0].ID != recordID {
t.Fatalf("unarchived review task rows = %+v, err = %v; want the original dedup record", rows, err)
}
}
func TestCleanupAllReviewTasks_ArchivedTaskRemainsExplicitlyEligible(t *testing.T) {
_, service, store, watch, client := setupArchivedReviewCleanup(t, false)
deleted, err := service.CleanupAllReviewTasks(context.Background())
if err != nil {
t.Fatalf("manual cleanup: %v", err)
}
if deleted != 0 || client.feedbackCalls != 1 {
t.Fatalf("manual cleanup deleted=%d feedback_calls=%d, want 0 and 1", deleted, client.feedbackCalls)
}
assertReviewTaskIsPreserved(t, store, watch, "archived-review-task")
}
func TestResetReviewWatch_ArchivedTask(t *testing.T) {
_, service, store, watch, _ := setupArchivedReviewCleanup(t, false)
service.SetCascadeTaskDeleter(noopCascadeTaskDeleter{})
deleted, err := service.ResetReviewWatch(context.Background(), watch.ID)
if err != nil {
t.Fatalf("reset review watch: %v", err)
}
if deleted != 1 {
t.Fatalf("reset deleted = %d, want the archived task", deleted)
}
rows, err := store.ListReviewPRTasksByWatch(context.Background(), watch.ID)
if err != nil || len(rows) != 0 {
t.Fatalf("review task rows after reset = %+v, err = %v, want none", rows, err)
}
}
func TestReviewCleanupCircuit_StopsAfterSharedFailure(t *testing.T) {
cases := []struct {
name string
err error
}{
{name: "authentication", err: &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}},
{name: "rate limit", err: &GitHubAPIError{StatusCode: 403, Endpoint: "/pulls/41", Body: "API rate limit exceeded"}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
poller, _, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41, 42)
client.feedbackErrors[41] = tc.err
client.feedbackErrors[42] = tc.err
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("feedback calls after shared failure = %d, want 1", client.feedbackCalls)
}
poller.checkPRWatches(context.Background())
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("feedback calls while circuit is open = %d, want 1", client.feedbackCalls)
}
})
}
}
func TestReviewCleanupCoreQuota(t *testing.T) {
poller, service, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
provider := &countingReviewCleanupAutomationProvider{client: client}
service.resolver.SetAutomationProvider(provider)
quotaSkipsBefore := reviewCleanupCoreQuotaSkipsTotal.Value()
service.RateTracker().Record(RateSnapshot{
Resource: ResourceCore, Remaining: 0, ResetAt: time.Now().Add(time.Minute), UpdatedAt: time.Now(),
})
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 0 {
t.Fatalf("feedback calls with exhausted Core quota = %d, want 0", client.feedbackCalls)
}
if provider.calls != 0 {
t.Fatalf("credential resolutions with exhausted Core quota = %d, want 0", provider.calls)
}
if got := reviewCleanupCoreQuotaSkipsTotal.Value(); got <= quotaSkipsBefore {
t.Fatalf("Core quota skip metric = %d, want an increment", got)
}
}
func TestReviewCleanupWorkspaceCircuitSkipsCredentialResolution(t *testing.T) {
poller, service, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
provider := &countingReviewCleanupAutomationProvider{client: client}
service.resolver.SetAutomationProvider(provider)
client.feedbackErrors[41] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}
poller.checkReviewWatches(context.Background())
if provider.calls != 1 || client.feedbackCalls != 1 {
t.Fatalf("first cleanup resolutions=%d feedback=%d, want one each", provider.calls, client.feedbackCalls)
}
service.resolver.InvalidateWorkspace("ws-1")
poller.checkReviewWatches(context.Background())
if provider.calls != 1 || client.feedbackCalls != 1 {
t.Fatalf("open workspace circuit resolutions=%d feedback=%d, want no additional calls", provider.calls, client.feedbackCalls)
}
}
func TestReviewCleanupRecordCircuitPrunesDeletedDedupRow(t *testing.T) {
poller, _, store, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
client.feedbackErrors[41] = ErrRepoNotResolvable
poller.checkReviewWatches(context.Background())
if got := reviewCleanupRecordCircuitCount(poller); got != 1 {
t.Fatalf("record circuit count after config failure = %d, want 1", got)
}
rows, err := store.ListAllReviewPRTasks(context.Background())
if err != nil || len(rows) != 1 {
t.Fatalf("review rows before deleting dedup row = %+v, err=%v", rows, err)
}
if err := store.DeleteReviewPRTask(context.Background(), rows[0].ID); err != nil {
t.Fatalf("delete review dedup row: %v", err)
}
poller.checkReviewWatches(context.Background())
if got := reviewCleanupRecordCircuitCount(poller); got != 0 {
t.Fatalf("record circuit count after dedup row deletion = %d, want 0", got)
}
}
func reviewCleanupRecordCircuitCount(poller *Poller) int {
poller.reviewCleanupCircuits.mu.Lock()
defer poller.reviewCleanupCircuits.mu.Unlock()
return len(poller.reviewCleanupCircuits.records)
}
func TestReviewCleanupCoreQuota_ExpiredSnapshotAllowsFeedback(t *testing.T) {
poller, service, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
tracker := service.RateTracker()
tracker.Record(RateSnapshot{
Resource: ResourceCore, Remaining: 0, ResetAt: time.Now().Add(time.Minute), UpdatedAt: time.Now(),
})
tracker.mu.Lock()
snapshot := tracker.snapshots[ResourceCore]
snapshot.ResetAt = time.Now().Add(-time.Minute)
tracker.snapshots[ResourceCore] = snapshot
tracker.mu.Unlock()
if !tracker.IsExhausted(ResourceCore) || tracker.WaitDuration(ResourceCore) > 0 {
t.Fatal("test setup must retain the stale exhausted flag after the reset time")
}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("feedback calls after Core reset = %d, want one scheduled probe", client.feedbackCalls)
}
}
func TestReviewCleanupFailureScope(t *testing.T) {
poller, _, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41, 42)
client.feedbackErrors[41] = ErrRepoNotResolvable
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 2 {
t.Fatalf("feedback calls after PR-specific config failure = %d, want 2", client.feedbackCalls)
}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 3 {
t.Fatalf("feedback calls after record circuit opens = %d, want only the healthy sibling", client.feedbackCalls)
}
if client.feedbackNumbers[len(client.feedbackNumbers)-1] != 42 {
t.Fatalf("last feedback PR = %d, want healthy sibling PR 42", client.feedbackNumbers[len(client.feedbackNumbers)-1])
}
}
func TestReviewCleanupManualBypass(t *testing.T) {
poller, service, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
client.feedbackErrors[41] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("scheduled feedback calls = %d, want 1 before opening the circuit", client.feedbackCalls)
}
delete(client.feedbackErrors, 41)
if _, err := service.CleanupAllReviewTasks(context.Background()); err != nil {
t.Fatalf("manual cleanup: %v", err)
}
if client.feedbackCalls != 2 {
t.Fatalf("manual feedback calls with scheduled circuit open = %d, want 2", client.feedbackCalls)
}
}
func seedReviewCleanupConnection(t *testing.T, store *Store, workspaceID string, generation int64) {
t.Helper()
err := store.UpsertWorkspaceConnection(context.Background(), &WorkspaceConnection{
WorkspaceID: workspaceID, Source: ConnectionSourcePAT, GitHubHost: defaultGitHubHost,
Login: "test-user", Status: ConnectionStatusActive, CredentialGeneration: generation,
})
if err != nil {
t.Fatalf("upsert workspace connection: %v", err)
}
}
func TestReviewCleanupCircuit_FingerprintReset(t *testing.T) {
poller, _, store, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
seedReviewCleanupConnection(t, store, "ws-1", 1)
client.feedbackErrors[41] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("initial feedback calls = %d, want 1", client.feedbackCalls)
}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("feedback calls before credential change = %d, want 1", client.feedbackCalls)
}
delete(client.feedbackErrors, 41)
seedReviewCleanupConnection(t, store, "ws-1", 2)
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 2 {
t.Fatalf("feedback calls after credential change = %d, want one immediate probe", client.feedbackCalls)
}
}
func TestReviewCleanupCircuit_WatchPollDoesNotResetRecordButConfigEditDoes(t *testing.T) {
poller, _, store, watch, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
watch.Enabled = true
if err := store.UpdateReviewWatch(context.Background(), watch); err != nil {
t.Fatalf("enable review watch: %v", err)
}
client.feedbackErrors[41] = ErrRepoNotResolvable
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("initial feedback calls = %d, want one failing record", client.feedbackCalls)
}
// An enabled watch successfully polls before cleanup runs again. This
// advances last_polled_at and updated_at without changing its configuration.
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("feedback calls after successful watch poll = %d, want the record circuit to stay open", client.feedbackCalls)
}
updatedWatch, err := store.GetReviewWatch(context.Background(), watch.ID)
if err != nil || updatedWatch == nil {
t.Fatalf("load review watch for config edit: watch=%v err=%v", updatedWatch, err)
}
updatedWatch.CustomQuery = "is:pr is:open"
if err := store.UpdateReviewWatch(context.Background(), updatedWatch); err != nil {
t.Fatalf("change review watch config: %v", err)
}
delete(client.feedbackErrors, 41)
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 2 {
t.Fatalf("feedback calls after config edit = %d, want the record circuit to reset", client.feedbackCalls)
}
}
func TestReviewCleanupCircuit_DueProbe(t *testing.T) {
poller, _, _, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
client.feedbackErrors[41] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("initial feedback calls = %d, want 1", client.feedbackCalls)
}
poller.reviewCleanupCircuits.mu.Lock()
state := poller.reviewCleanupCircuits.workspaces["ws-1"]
due := time.Now().UTC().Add(-time.Second)
state.NextRetryAt = &due
poller.reviewCleanupCircuits.workspaces["ws-1"] = state
poller.reviewCleanupCircuits.mu.Unlock()
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 2 {
t.Fatalf("feedback calls after due probe = %d, want 2", client.feedbackCalls)
}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 2 {
t.Fatalf("feedback calls after failed due probe = %d, want circuit reopened", client.feedbackCalls)
}
}
func TestReviewCleanupCircuit_SeparateWorkspaces(t *testing.T) {
poller, service, mockClient, store := setupPollerTest(t)
ctx := context.Background()
client := &countingReviewCleanupClient{
Client: mockClient, feedbackErrors: make(map[int]error), reviewPRNumbers: make(map[int]bool),
}
service.SetTaskDeleter(&recordingTaskDeleter{})
configureTestWorkspaceAuth(t, service, client, testWorkspaceID, "ws-1", "ws1", "ws-2")
for _, fixture := range []struct {
watchID string
workspaceID string
taskID string
prNumber int
}{
{watchID: "watch-one", workspaceID: "ws-1", taskID: "task-one", prNumber: 41},
{watchID: "watch-two", workspaceID: "ws-2", taskID: "task-two", prNumber: 51},
} {
client.reviewPRNumbers[fixture.prNumber] = true
watch := &ReviewWatch{
ID: fixture.watchID, WorkspaceID: fixture.workspaceID,
Enabled: false, CleanupPolicy: CleanupPolicyAlways,
}
if err := store.CreateReviewWatch(ctx, watch); err != nil {
t.Fatalf("create review watch: %v", err)
}
seedTask(t, store, fixture.taskID, false)
if err := store.CreateReviewPRTask(ctx, &ReviewPRTask{
ReviewWatchID: watch.ID, RepoOwner: "acme", RepoName: "widget",
PRNumber: fixture.prNumber, TaskID: fixture.taskID,
}); err != nil {
t.Fatalf("create review task: %v", err)
}
mockClient.AddPR(&PR{
Number: fixture.prNumber, State: "open", RepoOwner: "acme", RepoName: "widget",
})
}
client.feedbackErrors[41] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}
poller.checkReviewWatches(ctx)
poller.checkReviewWatches(ctx)
counts := make(map[int]int)
for _, number := range client.feedbackNumbers {
counts[number]++
}
if counts[41] != 1 || counts[51] != 2 {
t.Fatalf("feedback call counts by PR = %v, want one for failed ws-1 and two for healthy ws-2", counts)
}
}
func TestReviewCleanupCircuit_PRMonitorSuccessDoesNotClear(t *testing.T) {
poller, _, store, _, client := setupOpenReviewCleanupRecords(t, CleanupPolicyAlways, 41)
client.feedbackErrors[41] = &GitHubAPIError{StatusCode: 401, Endpoint: "/pulls/41"}
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("initial review feedback calls = %d, want 1", client.feedbackCalls)
}
seedTask(t, store, "pr-monitor-task", false)
prWatch := &PRWatch{
ID: "pr-watch", WorkspaceID: "ws-1", SessionID: "session-1", TaskID: "pr-monitor-task",
Owner: "acme", Repo: "widget", PRNumber: 99, Branch: "feature",
}
if err := store.CreatePRWatch(context.Background(), prWatch); err != nil {
t.Fatalf("create PR watch: %v", err)
}
client.Client.(*MockClient).AddPR(&PR{
Number: 99, State: "open", RepoOwner: "acme", RepoName: "widget", HeadBranch: "feature",
})
poller.checkSinglePRWatch(context.Background(), prWatch)
poller.checkReviewWatches(context.Background())
if client.feedbackCalls != 1 {
t.Fatalf("review feedback calls after PR-monitor success = %d, want the cleanup circuit to remain open", client.feedbackCalls)
}
}
@@ -28,7 +28,7 @@ func (c *countingCleanupClient) ListPRReviews(ctx context.Context, owner, repo s
return c.MockClient.ListPRReviews(ctx, owner, repo, number)
}
func TestCleanupMergedReviewTasksSkipsHistoricalTasks(t *testing.T) {
func TestCleanupMergedReviewTasksIncludesHistoricalTasks(t *testing.T) {
_, svc, mockClient, store := setupPollerTest(t)
ctx := context.Background()
@@ -67,17 +67,18 @@ func TestCleanupMergedReviewTasksSkipsHistoricalTasks(t *testing.T) {
if err != nil {
t.Fatalf("CleanupMergedReviewTasks: %v", err)
}
if deleted != 1 {
t.Fatalf("deleted = %d, want 1 active task", deleted)
if deleted != 3 {
t.Fatalf("deleted = %d, want active, archived, and hard-deleted tasks", deleted)
}
if len(recorder.calls) != 1 || recorder.calls[0] != "active-review" {
t.Fatalf("DeleteTask calls = %v, want [active-review]", recorder.calls)
if len(recorder.calls) != 3 || recorder.calls[0] != "active-review" ||
recorder.calls[1] != "archived-review" || recorder.calls[2] != "missing-review" {
t.Fatalf("DeleteTask calls = %v, want active, archived, and hard-deleted task IDs", recorder.calls)
}
if client.prCalls != 3 {
t.Fatalf("GetPR calls = %d, want 3 eligible rows only", client.prCalls)
if client.prCalls != 5 {
t.Fatalf("GetPR calls = %d, want all five inventory rows", client.prCalls)
}
if client.reviewCalls != 1 {
t.Fatalf("ListPRReviews calls = %d, want 1 for the open eligible row", client.reviewCalls)
t.Fatalf("ListPRReviews calls = %d, want 1 for the open reservation", client.reviewCalls)
}
remaining, err := store.ListReviewPRTaskIDsByWatch(ctx, watch.ID)
@@ -88,9 +89,8 @@ func TestCleanupMergedReviewTasksSkipsHistoricalTasks(t *testing.T) {
for _, taskID := range remaining {
got[taskID] = true
}
want := map[string]bool{"archived-review": true, "missing-review": true, "": true}
if len(got) != len(want) || !got["archived-review"] || !got["missing-review"] || !got[""] {
t.Fatalf("remaining task IDs = %#v, want historical rows and open empty reservation", got)
if len(got) != 1 || !got[""] {
t.Fatalf("remaining task IDs = %#v, want only the open empty reservation", got)
}
}
@@ -220,8 +220,8 @@ func (s *stubTaskDeleter) DeleteTask(_ context.Context, _ string) error {
return s.err
}
// TestCleanupMergedReviewTasks_TaskAlreadyDeleted verifies that a dedup row for
// a missing task is ignored without consuming provider quota.
// TestCleanupMergedReviewTasks_TaskAlreadyDeleted verifies that cleanup removes
// a stale dedup row after the task service reports it was already deleted.
func TestCleanupMergedReviewTasks_TaskAlreadyDeleted(t *testing.T) {
_, svc, mockClient, store := setupPollerTest(t)
ctx := context.Background()
@@ -254,8 +254,8 @@ func TestCleanupMergedReviewTasks_TaskAlreadyDeleted(t *testing.T) {
RepoName: "widget",
})
// A missing task is not a cleanup candidate. The task deleter remains wired
// to prove the cleanup path does not try to delete it.
// The task deleter reports the prior hard deletion so cleanup can release
// the stale reservation row without counting a newly deleted task.
svc.SetTaskDeleter(&stubTaskDeleter{
err: fmt.Errorf("%w: %s", ErrTaskNotFound, taskID),
})
@@ -264,18 +264,16 @@ func TestCleanupMergedReviewTasks_TaskAlreadyDeleted(t *testing.T) {
if err != nil {
t.Fatalf("CleanupMergedReviewTasks returned error: %v", err)
}
if deleted != 0 {
t.Errorf("expected 0 deleted for a missing task, got %d", deleted)
if deleted != 1 {
t.Errorf("deleted = %d, want stale dedup row cleanup to count as one", deleted)
}
// The historical dedup record remains available for deduplication, but is
// excluded from cleanup candidate queries.
remaining, err := store.ListReviewPRTaskIDsByWatch(ctx, watch.ID)
if err != nil {
t.Fatalf("ListReviewPRTaskIDsByWatch: %v", err)
}
if len(remaining) != 1 || remaining[0] != taskID {
t.Errorf("remaining task IDs = %v, want [%s]", remaining, taskID)
if len(remaining) != 0 {
t.Errorf("remaining task IDs = %v, want stale dedup row removed", remaining)
}
}
+33 -13
View File
@@ -4912,13 +4912,8 @@ func (s *Store) ReleaseReviewPRTask(ctx context.Context, reviewWatchID, repoOwne
func (s *Store) ListReviewPRTasksByWatch(ctx context.Context, watchID string) ([]*ReviewPRTask, error) {
var tasks []*ReviewPRTask
err := s.ro.SelectContext(ctx, &tasks, s.ro.Rebind(
`SELECT d.id, d.review_watch_id, d.repo_owner, d.repo_name, d.pr_number, d.pr_url, d.task_id, d.created_at
FROM github_review_pr_tasks d
WHERE d.review_watch_id = ?
AND (d.task_id = '' OR EXISTS (
SELECT 1 FROM tasks t
WHERE t.id = d.task_id AND t.archived_at IS NULL
))`), watchID)
`SELECT id, review_watch_id, repo_owner, repo_name, pr_number, pr_url, task_id, created_at
FROM github_review_pr_tasks WHERE review_watch_id = ?`), watchID)
return tasks, err
}
@@ -4928,12 +4923,37 @@ func (s *Store) ListReviewPRTasksByWatch(ctx context.Context, watchID string) ([
func (s *Store) ListAllReviewPRTasks(ctx context.Context) ([]*ReviewPRTask, error) {
var tasks []*ReviewPRTask
err := s.ro.SelectContext(ctx, &tasks,
`SELECT d.id, d.review_watch_id, d.repo_owner, d.repo_name, d.pr_number, d.pr_url, d.task_id, d.created_at
FROM github_review_pr_tasks d
WHERE d.task_id = '' OR EXISTS (
SELECT 1 FROM tasks t
WHERE t.id = d.task_id AND t.archived_at IS NULL
)`)
`SELECT id, review_watch_id, repo_owner, repo_name, pr_number, pr_url, task_id, created_at
FROM github_review_pr_tasks`)
return tasks, err
}
// ListScheduledReviewPRTasksByWatch lists review-task records that routine
// cleanup may inspect. Archived tasks stay in the complete inventory but do
// not spend provider requests on each poll. Empty reservations and records for
// hard-deleted tasks remain eligible for recovery.
func (s *Store) ListScheduledReviewPRTasksByWatch(ctx context.Context, watchID string) ([]*ReviewPRTask, error) {
var tasks []*ReviewPRTask
err := s.ro.SelectContext(ctx, &tasks, s.ro.Rebind(`
SELECT rpt.id, rpt.review_watch_id, rpt.repo_owner, rpt.repo_name, rpt.pr_number,
rpt.pr_url, rpt.task_id, rpt.created_at
FROM github_review_pr_tasks rpt
LEFT JOIN tasks t ON t.id = rpt.task_id
WHERE rpt.review_watch_id = ?
AND (rpt.task_id = '' OR t.id IS NULL OR t.archived_at IS NULL)`), watchID)
return tasks, err
}
// ListScheduledReviewPRTasks lists routine-cleanup candidates across all
// watches, preserving reservations and records for hard-deleted tasks.
func (s *Store) ListScheduledReviewPRTasks(ctx context.Context) ([]*ReviewPRTask, error) {
var tasks []*ReviewPRTask
err := s.ro.SelectContext(ctx, &tasks, `
SELECT rpt.id, rpt.review_watch_id, rpt.repo_owner, rpt.repo_name, rpt.pr_number,
rpt.pr_url, rpt.task_id, rpt.created_at
FROM github_review_pr_tasks rpt
LEFT JOIN tasks t ON t.id = rpt.task_id
WHERE rpt.task_id = '' OR t.id IS NULL OR t.archived_at IS NULL`)
return tasks, err
}
@@ -35,7 +35,7 @@ func TestDeleteTaskPRsByTaskIDRemovesOnlyTargetRows(t *testing.T) {
}
}
func TestCleanupCandidateListsExcludeHistoricalTasks(t *testing.T) {
func TestReviewCleanupInventoryRetainsHistoricalTasks(t *testing.T) {
_, _, _, store := setupPollerTest(t)
ctx := context.Background()
@@ -86,9 +86,9 @@ func TestCleanupCandidateListsExcludeHistoricalTasks(t *testing.T) {
for _, candidate := range candidates {
got[candidate.TaskID] = true
}
want := map[string]bool{"active-review": true, "": true}
if len(got) != len(want) || !got["active-review"] || !got[""] {
t.Errorf("%s task IDs = %#v, want active-review and empty reservation", name, got)
if len(got) != 4 || !got["active-review"] || !got["archived-review"] ||
!got["missing-review"] || !got[""] {
t.Errorf("%s task IDs = %#v, want active, archived, missing, and empty reservation rows", name, got)
}
}
assertIssueCandidates := func(name string, candidates []*IssueWatchTask) {
@@ -0,0 +1,134 @@
---
created: 2026-09-23
status: complete
requirements:
- REQ-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001
- REQ-PLATFORM-PROVIDER-BACKOFF-001
system_design:
- ../../specs/integrations/system-design/github-review-task-cleanup.md
- ../../specs/platform/system-design/pr-watch-and-bounded-storage.md
legacy_specs: []
---
# Implementation plan: Bound GitHub review cleanup polling
## Overview
Exclude archived review tasks from scheduled cleanup, then bound feedback
requests from the remaining eligible records. The first work order removes
the recurring load that caused the reported quota incident. The second closes
the review-cleanup backoff gap left by PR #3319. Work orders are sequential
because both change the scheduled cleanup path.
## Evidence and requirement conformance
The five-minute `reviewQueueLoop` calls `CleanupMergedReviewTasks` for enabled
watches and `CleanupAllOrphanedReviewTasks` for disabled watches. Both paths
select historical rows without archive filtering. `shouldDeleteReviewTaskWithClient`
fetches full PR feedback before it checks for user-authored messages or enabled
lifecycle prompts. Under Auto, those conditions retain the row, so the next
cycle repeats the request. A temporary SQLite-backed Go repro confirmed two
feedback calls across two cleanup runs for one archived, merged review task
with a user-authored message. The repro file was removed after the test passed.
The PR-watch circuit applies to task-owned PR watches only. Review cleanup
does not consult it, and its feedback failures are swallowed inside the
cleanup batch. The review loop waits for Search capacity, while feedback
primarily consumes Core capacity. A successful PAT feedback request for a
merged PR normally makes six REST reads before pagination. The reported 89
archived and eight active rows are user-provided counts; production request
headers were not available to verify them.
The archive rule is missing from Integrations requirements. Platform's active
`REQ-PLATFORM-PROVIDER-BACKOFF-001` already requires classified backoff for
background integration loops. This package adds the Integrations rule and
extends the existing Platform design; it does not create a second backoff
requirement or change the established ownership boundary.
## Scope
### In scope
- Skip feedback and deletion for archived review tasks in scheduled cleanup.
- Keep incomplete reservations and missing-task records eligible for recovery.
- Avoid scheduled Auto feedback when local task state already requires retention.
- Preserve explicit cleanup, watch deletion, and reset behavior.
- Admit scheduled feedback by Core quota and a separate review-cleanup circuit.
- Stop further requests in the affected scope after a classified failure.
- Add focused backend regression tests and clarify public integration guidance.
### Out of scope
- Live database edits, deletion of historical rows, or a schema migration.
- Changes to issue-watch cleanup, PR-watch identity, or search query matching.
- New UI controls, copy, settings, or credential sources.
- Changes to manual cleanup admission or explicit destructive watch actions.
## Technical approach
Task 01 adds scheduled-only eligibility selectors in
`apps/backend/internal/github/store.go` and uses them in the scheduled review
cleanup service paths. Keep complete inventory selectors for manual cleanup,
watch deletion, and reset. Apply Auto retention checks before scheduled
feedback. Update `docs/public/integrations.md` to state that archived review
tasks pause routine cleanup while explicit reset can still delete them.
Task 02 keeps a review-cleanup circuit separate from the PR-monitor circuit.
Add a poller-facing cleanup result that reports classified feedback failures
without changing public cleanup responses. Apply Core quota admission before
feedback. Stop the remaining workspace batch on shared auth or rate failure;
isolate PR-specific configuration failures to their record. Credential or
watch changes reset the relevant circuit. A successful search or PR-monitor
call cannot clear a review-cleanup failure. Keep metrics bounded and free of
identifiers.
## Tests
| Acceptance criterion | Planned evidence |
| --- | --- |
| `AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.1` | SQLite-backed enabled and disabled watch tests assert no archived-row feedback across two scheduled cycles. |
| `AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.2` | Unarchive test asserts one retained dedup row becomes eligible without a duplicate task. |
| `AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.3` | Mixed active, archived, empty-reservation, and missing-task fixture checks eligible rows and orphan recovery. |
| `AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.4` | Auto tests assert zero feedback for user messages and lifecycle prompts, including a failed local check. |
| `AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.5` | Manual cleanup and reset tests assert archived records remain in complete inventories. |
| `AC-PLATFORM-PROVIDER-BACKOFF-001.1` through `.4` | Poller tests assert same-cycle stop, separate scopes, fingerprint reset, and one due probe. |
Metrics tests also assert bounded labels with no task, watch, PR, repository,
or workspace IDs. Existing credential and quota health output stays intact.
## End-to-end evidence
The backend poller test drives the scheduled entry point through the real
SQLite store and a counting GitHub client. It proves the user-visible task and
record remain intact without provider feedback. No rendered UI changes, so a
browser E2E test would not exercise the defect more faithfully.
## Work orders
- [x] [Task 01: Make scheduled cleanup archive-aware](task-01-archive-eligibility.md)
- [x] [Task 02: Bound eligible cleanup requests](task-02-cleanup-backoff.md)
## Verification results
Task 01 targeted cleanup tests and public-doc validators passed. Task 02
focused cleanup/circuit tests and the complete `internal/github` package
tests passed. The backend build passed, including a post-review-fix build of
`cmd/kandev`. Regression coverage confirms routine watch polling keeps a
record circuit open, a real configuration edit resets it, and an expired Core
quota snapshot allows feedback again. Public-doc validators and `git diff
--check` passed. Backend-wide `make test` completed with two failures in the
unchanged `internal/agentctl/server/process/probe` package; config-discovery
tests that picked up task-injected home config passed when rerun with isolated
HOME and config paths unset.
## Risks
- Filtering shared inventory methods would hide archived tasks from explicit
deletion and reset. Scheduled cleanup needs its own selectors.
- An inner join would hide empty reservations and hard-deleted-task records.
- Feedback makes several concurrent upstream reads; one failed fetch may have
multiple in-flight requests. The circuit must stop later records in the
same scope.
- A success from search or the PR monitor must not clear a cleanup failure.
- Core quota is tracked across credentials in the current tracker, so a Core
exhaustion pause is conservative across workspaces.
@@ -0,0 +1,100 @@
---
id: "01-archive-eligibility"
title: "Make scheduled cleanup archive-aware"
status: done
wave: 1
depends_on: []
plan: "plan.md"
requirements:
- REQ-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001
acceptance_criteria:
- AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.1
- AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.2
- AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.3
- AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.4
- AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.5
system_design:
- ../../specs/integrations/system-design/github-review-task-cleanup.md
---
# Task 01: Make scheduled cleanup archive-aware
## Summary
Scheduled review cleanup stops polling archived tasks while keeping their
dedup records. Active tasks and orphan recovery still work, and explicit
watch actions retain their current archived-task behavior.
## In scope
- Add scheduled-only review-task selectors. Preserve the complete inventory
selectors used by manual cleanup, watch deletion, and reset.
- Apply the selectors to enabled-watch cleanup and the global disabled-watch
sweep. Keep the existing no-double-fetch rule for enabled watches.
- Check local Auto retention conditions before scheduled feedback. Fail closed
when a local check fails.
- Update the review-watch cleanup paragraph in public integration docs.
## Out of scope
- Circuit admission, rate tracking, and issue-watch cleanup.
- Database migration or deletion of historical records.
## Acceptance
1. A mixed SQLite fixture with archived, active, empty-reservation, and
missing-task records calls feedback only for eligible rows. The archived
row remains after two enabled or disabled scheduled passes.
2. After unarchive, the same dedup row is eligible without a second task.
Auto retention skips feedback when a user message or lifecycle prompt
requires the task to remain, and a failed check keeps the row.
3. Manual cleanup, watch deletion, and reset continue to enumerate archived
rows. Public docs distinguish routine cleanup from reset.
## Verification
```bash
(cd apps/backend && go test -tags fts5 ./internal/github -run 'Test(ListScheduledReviewPRTasks|CleanupMergedReviewTasks_Archived|CleanupAllOrphanedReviewTasks_Archived|CleanupReviewTasks_Unarchive|CleanupReviewTasks_AutoRetention|CleanupAllReviewTasks_Archived|ResetReviewWatch_Archived)' -count=1)
node scripts/validate-public-docs.mjs
git diff --check
```
## Files likely touched
- `apps/backend/internal/github/store.go`
- `apps/backend/internal/github/service_cleanup.go`
- `apps/backend/internal/github/service_cleanup_policy_test.go`
- `apps/backend/internal/github/review_pr_task_test.go`
- `apps/backend/internal/github/poller_test.go`
- `docs/public/integrations.md`
## Dependencies
None.
## Risks
Shared inventory filtering could change explicit destructive actions. Keep
scheduled eligibility separate from complete inventory.
## Parallelism
`sequential`
## Inputs
- [Requirement](../../specs/integrations/requirements/github-review-task-cleanup.md)
- [System design](../../specs/integrations/system-design/github-review-task-cleanup.md)
- Existing cleanup policy and reset tests in `internal/github`.
## Results
Added scheduled-only review-task selectors that exclude archived tasks while
retaining empty reservations and rows for hard-deleted tasks. Scheduled Auto
cleanup checks lifecycle prompts and user-authored messages before feedback;
explicit cleanup and reset continue to use complete inventories. Public
integration guidance now describes archive, unarchive, cleanup, and reset
behavior.
Verification passed: the targeted GitHub cleanup Go tests, both public-doc
validators, and `git diff --check`.
@@ -0,0 +1,122 @@
---
id: "02-cleanup-backoff"
title: "Bound eligible cleanup requests"
status: done
wave: 2
depends_on:
- "01-archive-eligibility"
plan: "plan.md"
requirements:
- REQ-PLATFORM-PROVIDER-BACKOFF-001
- REQ-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001
acceptance_criteria:
- AC-PLATFORM-PROVIDER-BACKOFF-001.1
- AC-PLATFORM-PROVIDER-BACKOFF-001.2
- AC-PLATFORM-PROVIDER-BACKOFF-001.3
- AC-PLATFORM-PROVIDER-BACKOFF-001.4
- AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.1
- AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.3
system_design:
- ../../specs/platform/system-design/pr-watch-and-bounded-storage.md
- ../../specs/integrations/system-design/github-review-task-cleanup.md
---
# Task 02: Bound eligible cleanup requests
## Summary
Scheduled feedback obeys Core quota and classified backoff. A provider
failure stops later requests in its scope, while healthy workspaces and
explicit user actions retain their existing behavior.
## In scope
- Add a review-cleanup circuit distinct from the PR-monitor circuit.
- Report feedback failures to the poller without changing public cleanup
response signatures or using a successful search as cleanup success.
- Stop the affected workspace batch after shared auth or rate failure.
Isolate record-specific configuration failures.
- Check Core exhaustion before scheduled feedback. Apply credential and watch
fingerprint resets and bounded-label metrics.
## Out of scope
- Frontend status or settings changes, credential storage, and manual cleanup
admission.
- New general-purpose rate limiter or changes to task-owned PR-watch polling.
## Acceptance
1. An auth or rate-limit failure on the first eligible row prevents later
feedback calls in that workspace during the cycle. An open circuit makes
zero calls until a due probe or matching fingerprint change.
2. A PR-specific configuration failure does not block a healthy sibling.
A healthy search or PR-monitor call does not clear the cleanup circuit.
An exhausted Core snapshot prevents scheduled feedback.
3. Separate workspace credentials keep separate circuits. Metrics expose
skips, resets, and failure classes without identifiers. Manual cleanup
retains its current path.
## Verification
```bash
(cd apps/backend && go test -tags fts5 ./internal/github -run 'Test(ReviewCleanupCircuit|ReviewCleanupCoreQuota|ReviewCleanupFailureScope|ReviewCleanupManualBypass|PollerCircuits|ClassifyPollErr|ReviewCleanupMetrics)' -count=1)
(cd apps/backend && go test -tags fts5 ./internal/github -count=1)
git diff --check
```
## Files likely touched
- `apps/backend/internal/github/poller.go`
- `apps/backend/internal/github/poller_circuit.go`
- `apps/backend/internal/github/service_cleanup.go`
- `apps/backend/internal/github/metrics_vars.go`
- `apps/backend/internal/github/poller_circuit_test.go`
- `apps/backend/internal/github/service_cleanup_policy_test.go`
- `apps/backend/internal/github/metrics_vars_test.go`
## Dependencies
Task 01 must supply the scheduled-only cleanup path and its eligible-row tests.
## Risks
The current cleanup helper swallows feedback errors. Preserve manual response
behavior while exposing a typed poller outcome. Do not let one record's 404
block every PR in its workspace.
## Parallelism
`sequential`
## Inputs
- [Platform requirement](../../specs/platform/requirements/pr-watch-and-bounded-storage.md)
- [Platform design](../../specs/platform/system-design/pr-watch-and-bounded-storage.md)
- [Review cleanup design](../../specs/integrations/system-design/github-review-task-cleanup.md)
- Existing `authcircuit`, Core rate tracker, and PR-monitor circuit tests.
## Results
Added a scheduled cleanup circuit separate from PR-monitor circuits. Scheduled
feedback checks Core quota, stops the affected workspace after shared provider
failures, and isolates record-specific configuration failures. Credential and
actual watch-configuration changes reset record circuit state without routine
poll timestamps doing so. Core admission uses the remaining reset wait, so an
expired snapshot cannot block cleanup indefinitely. Core quota and workspace
circuit admission run before scheduled credential resolution. Auto retention
treats task-not-found as an already missing task while preserving fail-closed
behavior for other errors. Bounded expvar metrics report skips, resets,
failures, and Core-quota skips. Explicit cleanup remains outside the scheduled
admission path.
Verification passed: focused cleanup/circuit tests, full `internal/github` and
`internal/backendapp` package tests, the backend binary build, changed-package
lint, specification validators, and `git diff --check`. Regression tests cover
credential admission before resolution, missing-task cleanup, record-circuit
pruning, metric publication, an enabled watch poll between record failures, a
real watch-config edit, and recovery after a Core snapshot reset time expires.
The backend-wide `make test` reached all packages but reported two failures in
the unchanged `internal/agentctl/server/process/probe` package. Its
config-discovery failures passed when rerun with isolated HOME and without
task-injected config paths.
+1 -1
View File
@@ -344,7 +344,7 @@ When a review watch is created, Kandev saves its verified target GitHub login. A
An **Issue Watch** behaves similarly for issues. Its default search is `type:issue state:open`. Choose labels or provide a custom GitHub query; the custom query takes precedence over label selection.
Both watch types default to the **Auto** cleanup policy: delete merged/closed tasks only when the user has not typed a message. For a GitHub Review Watch task with any PR lifecycle prompt enabled, **Auto** instead retains the terminal task so that lifecycle delivery can finish. **Always delete** is the explicit override and deletes even after user engagement or enabled lifecycle prompts; **Never** retains every task. You can pause a watch, poll immediately, or clean completed work. Deleting a GitHub review or issue watch best-effort cascade-deletes the tasks it owns. **Reset** is also destructive: after its preview, it permanently cascade-deletes every watch-created task, including archived tasks, and clears cursor/deduplication state so current matches become eligible again. Review-watch reset schedules a re-import; issue-watch reset re-imports on its next poll. Reset is not a way to keep old tasks and rerun a query.
Both watch types default to the **Auto** cleanup policy: delete merged/closed tasks only when the user has not typed a message. For a GitHub Review Watch task with any PR lifecycle prompt enabled, **Auto** instead retains the terminal task so that lifecycle delivery can finish. **Always delete** is the explicit override and deletes even after user engagement or enabled lifecycle prompts; **Never** retains every task. Routine cleanup pauses while a review task is archived and resumes when you unarchive it. The task and its review-watch record stay in place while archived. You can pause a watch, poll immediately, or clean completed work. Explicit cleanup still applies the watch policy to archived tasks. Deleting a GitHub review or issue watch best-effort cascade-deletes the tasks it owns. **Reset** is also destructive: after its preview, it permanently cascade-deletes every watch-created task, including archived tasks, and clears cursor/deduplication state so current matches become eligible again. Review-watch reset schedules a re-import; issue-watch reset re-imports on its next poll. Reset is not a way to keep old tasks and rerun a query.
Repository scope, authentication, and watch filters are workspace-specific. Repository scope constrains Kandev operations in addition to the repositories allowed by the selected credential; it cannot grant access the credential lacks. Explicit executor profile tokens remain a separate override and should be scoped independently. GitHub workspace configuration can be copied, but credentials, App installation bindings, personal identities, and watches are deliberately not copied.
@@ -0,0 +1,63 @@
---
status: draft
system: integrations
created: 2026-09-23
owners:
- kandev
---
# GitHub review task cleanup requirements
## Overview
A GitHub Review Watch retains a record for each task that it creates. Routine
cleanup checks the pull request and applies the watch's cleanup policy. An
archived task remains in that record so an unarchive or an explicit watch action
can still find it. Routine cleanup must not spend GitHub requests on archived
tasks or delete them.
Integrations owns this contract because it owns GitHub Review Watches, their
cleanup policies, and their provider requests. The task system owns the archive
state. Platform owns shared provider backoff.
## Requirements
### REQ-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001: Archive-aware routine cleanup
**Intent:** Routine review cleanup shall evaluate eligible tasks without
rechecking archived tasks or changing explicit watch actions.
#### Acceptance criteria
- **AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.1:** When a review task is
archived, scheduled cleanup shall make no pull-request feedback request for
that task. It shall retain the task and its review-watch record under an
enabled or disabled watch.
- **AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.2:** When an archived review task
is unarchived, the next scheduled cleanup shall evaluate it under the watch's
current policy. Unarchiving shall not create a second task for the same
watch and pull request.
- **AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.3:** Scheduled cleanup shall
continue to evaluate active review tasks, incomplete task reservations, and
records whose task was hard-deleted. A missing task shall not hide a record
from orphan recovery.
- **AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.4:** Under Auto cleanup, a task
retained by a user-authored message or an enabled pull-request lifecycle
prompt shall not trigger a feedback request on each scheduled cleanup.
A failed retention check shall leave the task and record intact.
- **AC-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001.5:** Explicit cleanup, watch
deletion, and watch reset shall keep their existing policy and deletion
behavior for archived tasks. Routine filtering shall not hide archived tasks
from these actions.
## Out of scope
- Changing the meaning of Auto, Always delete, or Never.
- Deleting archived task history or review-watch records during migration.
- Changing GitHub issue-watch cleanup or task-owned PR-watch identity.
- New settings controls or rendered UI states.
## Related contracts
- [Provider failure backoff](../../platform/requirements/pr-watch-and-bounded-storage.md#req-platform-provider-backoff-001-generation-aware-provider-failure-backoff)
- [System design](../system-design/github-review-task-cleanup.md)
@@ -0,0 +1,81 @@
---
status: draft
system: integrations
requirements:
- REQ-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001
created: 2026-09-23
owners:
- kandev
---
# GitHub review task cleanup system design
## Purpose and boundaries
`internal/github` owns review-watch deduplication, cleanup policy, and GitHub
feedback requests. The task repository owns `tasks.archived_at`. The cleanup
selector reads that state; it does not change task archive state. Shared
provider backoff remains a Platform contract.
## Requirement mapping
| Requirement | Design section |
| --- | --- |
| `REQ-INTEGRATIONS-GITHUB-REVIEW-CLEANUP-001` | [Eligibility](#eligibility); [Cleanup flow](#cleanup-flow); [Explicit actions](#explicit-actions) |
## Eligibility
Add selectors for **scheduled** review cleanup in `internal/github/store.go`.
Select a review-task record when its `task_id` is empty, its task row is
missing, or its task row exists with `archived_at IS NULL`. Exclude a record
only when a matching task row is archived. Use a left join or an equivalent
predicate so missing tasks and incomplete reservations remain visible.
Keep `ListReviewPRTasksByWatch` and `ListAllReviewPRTasks` as complete inventory
methods. Watch deletion, reset, and explicit cleanup use those inventories.
Scheduled per-watch cleanup and the scheduled global orphan sweep use the new
eligible selectors. The global sweep still skips enabled-watch records that
the per-watch pass handled in the same cycle. A disabled watch remains eligible
through the global sweep.
The dedup row stays in storage while the task is archived. Unarchive makes the
same row eligible again, so the watch cannot create a duplicate task for that
pull request. No schema migration or backfill is required.
## Cleanup flow
Before a scheduled Auto feedback request, check the conditions that retain a
task regardless of PR state: an enabled lifecycle prompt and a user-authored
message. If either is present, skip feedback and keep the task and dedup row.
If either local check fails, fail closed and keep both. Keep Always delete and
Never behavior unchanged. Incomplete reservations have no task to inspect and
retain their terminal-PR cleanup path.
Eligibility comes from the database before client resolution where practical.
An empty eligible set does not resolve a credential or call GitHub. The poller
must check eligibility again on each scheduled pass; archive and unarchive
take effect without a service restart. A task archived after selection may
finish an already-started request, but later passes exclude it.
## Explicit actions
The manual all-workspace and workspace cleanup endpoints continue to inspect
archived records under each watch's policy. Watch deletion and reset continue
to enumerate archived tasks and apply their existing destructive behavior.
Do not route these actions through a scheduled-only selector.
## Failure and recovery
A missing task row remains visible to the cleanup path that removes stale
dedup records. A task-not-found result from the session checker counts as no
user-authored message so cleanup can remove its stale dedup record. Other
task-retention check failures leave both task and record intact. A watch read
failure leaves its records untouched for that cycle.
Provider failures follow the [Platform backoff design](../../platform/system-design/pr-watch-and-bounded-storage.md#provider-failure-backoff-and-health).
## Verification
SQLite-backed service and poller tests cover enabled and disabled watches,
archived and active tasks in one batch, incomplete and missing-task records,
unarchive, Auto retention, and explicit actions. A counting GitHub client
asserts feedback call counts over repeated scheduled cycles.
@@ -128,6 +128,41 @@ configs. `apps/backend/internal/health` exposes aggregate health (counts by
class, no workspace identifiers or secrets), and bounded-label expvar counters
report skips, resets, and failures.
### Review-cleanup extension
The [review task cleanup package](../../integrations/system-design/github-review-task-cleanup.md)
extends this existing backoff contract to scheduled review-task feedback. It
does not change task-owned PR-watch identity or the manual cleanup endpoints.
Keep review-cleanup circuit state separate from the PR-monitor circuit. A
successful PR-monitor request must not clear a failed review-cleanup target.
Key shared authentication and rate-limit failures by workspace automation
connection. Keep a record-specific configuration failure isolated to that
review record. One inaccessible PR must not suspend unrelated work in the
workspace. Use non-secret connection generation and status as the workspace
reset fingerprint. Reset a record-specific circuit when its watch
configuration changes, excluding routine polling timestamps. Circuit state
remains in memory and is discarded on restart. Remove record-specific state
when its dedup row disappears.
Before resolving automation credentials for scheduled cleanup, inspect the
Core rate tracker and workspace circuit. If Core is exhausted or the workspace
circuit is open, skip credential resolution. After resolving credentials,
inspect record-specific admission before each feedback request. A failed
feedback fetch must reach the poller as a classified outcome so it can open
the relevant circuit.
After a shared authentication or rate-limit failure, stop the remaining
feedback requests for that workspace in the same cycle. Continue other
workspaces unless the shared Core tracker reports exhaustion. A record-specific
configuration failure stops only that record. Search success does not count as
feedback success and does not reset the cleanup circuit.
Use the existing `authcircuit.State`, `classifyPollErr`, and connection
fingerprint rules. Do not change the public cleanup response shape. Keep
manual cleanup and explicit watch actions outside scheduled circuit admission.
Aggregate review-cleanup skips and failures into bounded-label health metrics;
never use task, watch, PR, repository, or workspace IDs as metric labels.
## Related decisions
- [Keep PR watches task owned](../../../decisions/2026-08-31-task-owned-pr-watch-identity.md)