From 4dbeb72563901e6d27b4d69893220840782fafdb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Wed, 23 Sep 2026 17:39:51 +0100 Subject: [PATCH] fix: bound scheduled GitHub review cleanup polling (#3884) --- .../internal/backendapp/turn_adapters.go | 11 +- .../internal/backendapp/turn_adapters_test.go | 21 +- apps/backend/internal/github/metrics_vars.go | 60 +- .../internal/github/metrics_vars_test.go | 43 + apps/backend/internal/github/poller.go | 34 +- .../github/poller_cleanup_rate_limit_test.go | 6 +- .../internal/github/poller_review_cleanup.go | 271 +++++++ .../github/poller_review_cleanup_circuit.go | 119 +++ .../internal/github/service_cleanup.go | 354 ++++++-- .../github/service_cleanup_policy_test.go | 1 + .../service_review_cleanup_scheduled_test.go | 760 ++++++++++++++++++ .../github/service_review_cleanup_test.go | 40 +- apps/backend/internal/github/store.go | 46 +- .../github/store_task_cleanup_test.go | 8 +- .../plans/github-review-cleanup-quota/plan.md | 134 +++ .../task-01-archive-eligibility.md | 100 +++ .../task-02-cleanup-backoff.md | 122 +++ docs/public/integrations.md | 2 +- .../github-review-task-cleanup.md | 63 ++ .../github-review-task-cleanup.md | 81 ++ .../pr-watch-and-bounded-storage.md | 35 + 21 files changed, 2158 insertions(+), 153 deletions(-) create mode 100644 apps/backend/internal/github/poller_review_cleanup.go create mode 100644 apps/backend/internal/github/poller_review_cleanup_circuit.go create mode 100644 apps/backend/internal/github/service_review_cleanup_scheduled_test.go create mode 100644 docs/plans/github-review-cleanup-quota/plan.md create mode 100644 docs/plans/github-review-cleanup-quota/task-01-archive-eligibility.md create mode 100644 docs/plans/github-review-cleanup-quota/task-02-cleanup-backoff.md create mode 100644 docs/specs/integrations/requirements/github-review-task-cleanup.md create mode 100644 docs/specs/integrations/system-design/github-review-task-cleanup.md diff --git a/apps/backend/internal/backendapp/turn_adapters.go b/apps/backend/internal/backendapp/turn_adapters.go index 28d8466c1a..1ef365957b 100644 --- a/apps/backend/internal/backendapp/turn_adapters.go +++ b/apps/backend/internal/backendapp/turn_adapters.go @@ -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 { diff --git a/apps/backend/internal/backendapp/turn_adapters_test.go b/apps/backend/internal/backendapp/turn_adapters_test.go index bbedabf6a4..6040db67a0 100644 --- a/apps/backend/internal/backendapp/turn_adapters_test.go +++ b/apps/backend/internal/backendapp/turn_adapters_test.go @@ -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) { diff --git a/apps/backend/internal/github/metrics_vars.go b/apps/backend/internal/github/metrics_vars.go index ce4117d141..2035f4ac41 100644 --- a/apps/backend/internal/github/metrics_vars.go +++ b/apps/backend/internal/github/metrics_vars.go @@ -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) diff --git a/apps/backend/internal/github/metrics_vars_test.go b/apps/backend/internal/github/metrics_vars_test.go index 8be25172f7..5c8851eb4c 100644 --- a/apps/backend/internal/github/metrics_vars_test.go +++ b/apps/backend/internal/github/metrics_vars_test.go @@ -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 { diff --git a/apps/backend/internal/github/poller.go b/apps/backend/internal/github/poller.go index 41d6ff8125..d41fbd093d 100644 --- a/apps/backend/internal/github/poller.go +++ b/apps/backend/internal/github/poller.go @@ -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 diff --git a/apps/backend/internal/github/poller_cleanup_rate_limit_test.go b/apps/backend/internal/github/poller_cleanup_rate_limit_test.go index f26891f615..f7b93a260e 100644 --- a/apps/backend/internal/github/poller_cleanup_rate_limit_test.go +++ b/apps/backend/internal/github/poller_cleanup_rate_limit_test.go @@ -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) diff --git a/apps/backend/internal/github/poller_review_cleanup.go b/apps/backend/internal/github/poller_review_cleanup.go new file mode 100644 index 0000000000..64a1978cb4 --- /dev/null +++ b/apps/backend/internal/github/poller_review_cleanup.go @@ -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) +} diff --git a/apps/backend/internal/github/poller_review_cleanup_circuit.go b/apps/backend/internal/github/poller_review_cleanup_circuit.go new file mode 100644 index 0000000000..7e47b2b789 --- /dev/null +++ b/apps/backend/internal/github/poller_review_cleanup_circuit.go @@ -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) + } + } +} diff --git a/apps/backend/internal/github/service_cleanup.go b/apps/backend/internal/github/service_cleanup.go index 82d77a8093..ee8d71527e 100644 --- a/apps/backend/internal/github/service_cleanup.go +++ b/apps/backend/internal/github/service_cleanup.go @@ -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 } diff --git a/apps/backend/internal/github/service_cleanup_policy_test.go b/apps/backend/internal/github/service_cleanup_policy_test.go index fec591b52b..6afbf1e34a 100644 --- a/apps/backend/internal/github/service_cleanup_policy_test.go +++ b/apps/backend/internal/github/service_cleanup_policy_test.go @@ -460,6 +460,7 @@ func TestCheckReviewWatches_RunsOrphanSweepWhenNoEnabledWatches(t *testing.T) { rec := &recordingTaskDeleter{} svc.SetTaskDeleter(rec) + svc.SetTaskSessionChecker(&recordingSessionChecker{}) poller.checkReviewWatches(ctx) diff --git a/apps/backend/internal/github/service_review_cleanup_scheduled_test.go b/apps/backend/internal/github/service_review_cleanup_scheduled_test.go new file mode 100644 index 0000000000..c5f5e6bc7c --- /dev/null +++ b/apps/backend/internal/github/service_review_cleanup_scheduled_test.go @@ -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) + } +} diff --git a/apps/backend/internal/github/service_review_cleanup_test.go b/apps/backend/internal/github/service_review_cleanup_test.go index e7e22ed609..48ccbd58d2 100644 --- a/apps/backend/internal/github/service_review_cleanup_test.go +++ b/apps/backend/internal/github/service_review_cleanup_test.go @@ -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) } } diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index a5003404d2..b54505e6e6 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -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 } diff --git a/apps/backend/internal/github/store_task_cleanup_test.go b/apps/backend/internal/github/store_task_cleanup_test.go index 1cb64a2981..52018447a8 100644 --- a/apps/backend/internal/github/store_task_cleanup_test.go +++ b/apps/backend/internal/github/store_task_cleanup_test.go @@ -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) { diff --git a/docs/plans/github-review-cleanup-quota/plan.md b/docs/plans/github-review-cleanup-quota/plan.md new file mode 100644 index 0000000000..d359ab705c --- /dev/null +++ b/docs/plans/github-review-cleanup-quota/plan.md @@ -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. diff --git a/docs/plans/github-review-cleanup-quota/task-01-archive-eligibility.md b/docs/plans/github-review-cleanup-quota/task-01-archive-eligibility.md new file mode 100644 index 0000000000..01990cea9b --- /dev/null +++ b/docs/plans/github-review-cleanup-quota/task-01-archive-eligibility.md @@ -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`. diff --git a/docs/plans/github-review-cleanup-quota/task-02-cleanup-backoff.md b/docs/plans/github-review-cleanup-quota/task-02-cleanup-backoff.md new file mode 100644 index 0000000000..bf13c5f844 --- /dev/null +++ b/docs/plans/github-review-cleanup-quota/task-02-cleanup-backoff.md @@ -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. diff --git a/docs/public/integrations.md b/docs/public/integrations.md index a41747e0be..586d7988ac 100644 --- a/docs/public/integrations.md +++ b/docs/public/integrations.md @@ -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. diff --git a/docs/specs/integrations/requirements/github-review-task-cleanup.md b/docs/specs/integrations/requirements/github-review-task-cleanup.md new file mode 100644 index 0000000000..dafd28ebbe --- /dev/null +++ b/docs/specs/integrations/requirements/github-review-task-cleanup.md @@ -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) diff --git a/docs/specs/integrations/system-design/github-review-task-cleanup.md b/docs/specs/integrations/system-design/github-review-task-cleanup.md new file mode 100644 index 0000000000..5b9c3a7eb0 --- /dev/null +++ b/docs/specs/integrations/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. diff --git a/docs/specs/platform/system-design/pr-watch-and-bounded-storage.md b/docs/specs/platform/system-design/pr-watch-and-bounded-storage.md index 053f3f4192..b4ba88fde0 100644 --- a/docs/specs/platform/system-design/pr-watch-and-bounded-storage.md +++ b/docs/specs/platform/system-design/pr-watch-and-bounded-storage.md @@ -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)