fix(github): drain piled-up PR/issue review tasks via cleanup policy + manual sweep (#929)

apps/backend/internal/github/controller.go

@@ -60,6 +60,9 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) {
 	api.POST("/watches/issue/:id/trigger", c.httpTriggerIssueWatch)
 	api.POST("/watches/issue/trigger-all", c.httpTriggerAllIssueChecks)
 
+	api.POST("/cleanup/review-tasks", c.httpCleanupReviewTasks)
+	api.POST("/cleanup/issue-tasks", c.httpCleanupIssueTasks)
+
 	api.GET("/orgs", c.httpListUserOrgs)
 	api.GET("/repos/search", c.httpSearchRepos)
 	api.GET("/repos/:owner/:repo/branches", c.httpListRepoBranches)
@@ -779,3 +782,33 @@ func (c *Controller) httpResetActionPresets(ctx *gin.Context) {
 	}
 	ctx.JSON(http.StatusOK, presets)
 }
+
+// httpCleanupReviewTasks runs a sweep over every review-PR dedup row
+// (including rows under enabled watches) applying each watch's cleanup
+// policy. Manual trigger for users to drain a pile of merged-PR tasks
+// without waiting for the next 5-minute poller cycle.
+//
+// SCOPE: install-wide — walks dedup rows across all workspaces. The
+// trigger-all endpoints (httpTriggerAllReviewChecks /
+// httpTriggerAllIssueChecks) take a required workspace_id, but cleanup
+// is install-wide on purpose so a user can drain orphans whose original
+// watch lived in a workspace they no longer have open. A future
+// multi-tenant rollout would add an optional workspace_id query param.
+func (c *Controller) httpCleanupReviewTasks(ctx *gin.Context) {
+	deleted, err := c.service.CleanupAllReviewTasks(ctx.Request.Context())
+	if err != nil {
+		ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	ctx.JSON(http.StatusOK, gin.H{"deleted": deleted})
+}
+
+// httpCleanupIssueTasks mirrors httpCleanupReviewTasks for issue watches.
+func (c *Controller) httpCleanupIssueTasks(ctx *gin.Context) {
+	deleted, err := c.service.CleanupAllIssueTasks(ctx.Request.Context())
+	if err != nil {
+		ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	ctx.JSON(http.StatusOK, gin.H{"deleted": deleted})
+}

apps/backend/internal/github/handlers.go

@@ -14,6 +14,7 @@ import (
 const (
 	errMsgInvalidPayload = "invalid payload"
 	errMsgIDRequired     = "id required"
+	respKeyDeleted       = "deleted"
 )
 
 // RegisterRoutes registers HTTP and WebSocket routes for GitHub integration.
@@ -65,6 +66,34 @@ func registerWSHandlers(dispatcher *ws.Dispatcher, svc *Service, log *logger.Log
 	dispatcher.RegisterFunc(ws.ActionGitHubActionPresetsList, wsListActionPresets(svc))
 	dispatcher.RegisterFunc(ws.ActionGitHubActionPresetsUpdate, wsUpdateActionPresets(svc))
 	dispatcher.RegisterFunc(ws.ActionGitHubActionPresetsReset, wsResetActionPresets(svc))
+
+	// Manual cleanup sweeps
+	dispatcher.RegisterFunc(ws.ActionGitHubCleanupReviewTasks, wsCleanupReviewTasks(svc))
+	dispatcher.RegisterFunc(ws.ActionGitHubCleanupIssueTasks, wsCleanupIssueTasks(svc))
+}
+
+// wsCleanupReviewTasks runs the manual full sweep and returns the count
+// deleted. Manual users want everything drained now, so this skips the
+// poller's "orphans only" optimization.
+func wsCleanupReviewTasks(svc *Service) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
+	return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
+		deleted, err := svc.CleanupAllReviewTasks(ctx)
+		if err != nil {
+			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil)
+		}
+		return ws.NewResponse(msg.ID, msg.Action, map[string]int{respKeyDeleted: deleted})
+	}
+}
+
+// wsCleanupIssueTasks mirrors wsCleanupReviewTasks for issue watches.
+func wsCleanupIssueTasks(svc *Service) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
+	return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
+		deleted, err := svc.CleanupAllIssueTasks(ctx)
+		if err != nil {
+			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil)
+		}
+		return ws.NewResponse(msg.ID, msg.Action, map[string]int{respKeyDeleted: deleted})
+	}
 }
 
 // parseMap parses the WS message payload into a map for simple field lookups.
@@ -113,7 +142,7 @@ func wsDeleteByID(deleteFn func(ctx context.Context, id string) error) func(ctx
 		if err := deleteFn(ctx, id); err != nil {
 			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil)
 		}
-		return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"deleted": true})
+		return ws.NewResponse(msg.ID, msg.Action, map[string]bool{respKeyDeleted: true})
 	}
 }
 

apps/backend/internal/github/models.go

@@ -207,6 +207,42 @@ const (
 	ReviewScopeUserAndTeams = "user_and_teams"
 )
 
+// CleanupPolicy controls how a review or issue watch handles its auto-created
+// tasks once the underlying PR / issue reaches a terminal state.
+const (
+	// CleanupPolicyAuto deletes the task once the PR/issue is merged or closed
+	// UNLESS the user authored at least one message in the task (the agent's
+	// auto-start prompt does not count).
+	CleanupPolicyAuto = "auto"
+	// CleanupPolicyAlways deletes the task on terminal state regardless of
+	// user interaction. Use when the watch is purely informational and the
+	// user never expects a banner / history for merged PRs.
+	CleanupPolicyAlways = "always"
+	// CleanupPolicyNever disables automatic cleanup. Tasks pile up until the
+	// user invokes manual cleanup or deletes them by hand.
+	CleanupPolicyNever = "never"
+)
+
+// IsValidCleanupPolicy reports whether s is one of the recognized policies.
+// Empty string is treated as valid so legacy rows (pre-migration) and zero
+// values default to "auto" downstream.
+func IsValidCleanupPolicy(s string) bool {
+	switch s {
+	case "", CleanupPolicyAuto, CleanupPolicyAlways, CleanupPolicyNever:
+		return true
+	}
+	return false
+}
+
+// NormalizeCleanupPolicy maps the empty string to CleanupPolicyAuto. Unknown
+// values are returned unchanged so the caller can surface a validation error.
+func NormalizeCleanupPolicy(s string) string {
+	if s == "" {
+		return CleanupPolicyAuto
+	}
+	return s
+}
+
 // ReviewWatch configures periodic polling for PRs needing the user's review.
 // Repos holds the list of repositories to monitor. An empty list means all repos.
 type ReviewWatch struct {
@@ -223,6 +259,7 @@ type ReviewWatch struct {
 	CustomQuery         string       `json:"custom_query" db:"custom_query"`
 	Enabled             bool         `json:"enabled" db:"enabled"`
 	PollIntervalSeconds int          `json:"poll_interval_seconds" db:"poll_interval_seconds"`
+	CleanupPolicy       string       `json:"cleanup_policy" db:"cleanup_policy"`
 	LastPolledAt        *time.Time   `json:"last_polled_at,omitempty" db:"last_polled_at"`
 	CreatedAt           time.Time    `json:"created_at" db:"created_at"`
 	UpdatedAt           time.Time    `json:"updated_at" db:"updated_at"`
@@ -303,6 +340,7 @@ type CreateReviewWatchRequest struct {
 	ReviewScope         string       `json:"review_scope"`
 	CustomQuery         string       `json:"custom_query"`
 	PollIntervalSeconds int          `json:"poll_interval_seconds"`
+	CleanupPolicy       string       `json:"cleanup_policy"`
 }
 
 // UpdateReviewWatchRequest is the request body for updating a review watch.
@@ -317,6 +355,7 @@ type UpdateReviewWatchRequest struct {
 	CustomQuery         *string       `json:"custom_query,omitempty"`
 	Enabled             *bool         `json:"enabled,omitempty"`
 	PollIntervalSeconds *int          `json:"poll_interval_seconds,omitempty"`
+	CleanupPolicy       *string       `json:"cleanup_policy,omitempty"`
 }
 
 // PRFeedbackEvent is published to the event bus when a PR has new feedback.
@@ -424,6 +463,7 @@ type IssueWatch struct {
 	CustomQuery         string       `json:"custom_query" db:"custom_query"`
 	Enabled             bool         `json:"enabled" db:"enabled"`
 	PollIntervalSeconds int          `json:"poll_interval_seconds" db:"poll_interval_seconds"`
+	CleanupPolicy       string       `json:"cleanup_policy" db:"cleanup_policy"`
 	LastPolledAt        *time.Time   `json:"last_polled_at,omitempty" db:"last_polled_at"`
 	CreatedAt           time.Time    `json:"created_at" db:"created_at"`
 	UpdatedAt           time.Time    `json:"updated_at" db:"updated_at"`
@@ -465,6 +505,7 @@ type CreateIssueWatchRequest struct {
 	Labels              []string     `json:"labels"`
 	CustomQuery         string       `json:"custom_query"`
 	PollIntervalSeconds int          `json:"poll_interval_seconds"`
+	CleanupPolicy       string       `json:"cleanup_policy"`
 }
 
 // UpdateIssueWatchRequest is the request body for updating an issue watch.
@@ -479,6 +520,7 @@ type UpdateIssueWatchRequest struct {
 	CustomQuery         *string       `json:"custom_query,omitempty"`
 	Enabled             *bool         `json:"enabled,omitempty"`
 	PollIntervalSeconds *int          `json:"poll_interval_seconds,omitempty"`
+	CleanupPolicy       *string       `json:"cleanup_policy,omitempty"`
 }
 
 // --- Action presets (quick-launch prompts on the /github page) ---

apps/backend/internal/github/poller.go

@@ -538,16 +538,17 @@ func (p *Poller) checkReviewWatches(ctx context.Context) {
 		p.logger.Error("failed to list review watches", zap.Error(err))
 		return
 	}
-	if len(watches) == 0 {
-		return
-	}
+	// Note: do NOT early-return on len(watches) == 0 — the global orphan
+	// 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)))
 	for _, watch := range watches {
 		// A previous iteration in this same cycle may have exhausted the
-		// search bucket. Stop early instead of issuing another doomed search
-		// that would only deepen the secondary-limit penalty.
+		// search bucket. Skip remaining per-watch checks instead of issuing
+		// another doomed search, but fall through to the orphan sweep below
+		// since it doesn't touch the search API.
 		if p.searchBucketExhausted("review_watch") {
-			return
+			break
 		}
 		p.logger.Debug("polling review watch",
 			zap.String("watch_id", watch.ID),
@@ -582,6 +583,15 @@ func (p *Poller) checkReviewWatches(ctx context.Context) {
 				zap.String("watch_id", watch.ID), zap.Int("deleted", cleaned))
 		}
 	}
+	// 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 {
+		p.logger.Warn("failed to sweep orphaned review tasks", zap.Error(err))
+	} else if cleaned > 0 {
+		p.logger.Info("swept orphaned review tasks", zap.Int("deleted", cleaned))
+	}
 }
 
 // issueWatchLoop polls issue watches for new GitHub issues.
@@ -614,13 +624,12 @@ func (p *Poller) checkIssueWatches(ctx context.Context) {
 		p.logger.Error("failed to list issue watches", zap.Error(err))
 		return
 	}
-	if len(watches) == 0 {
-		return
-	}
+	// Note: do NOT early-return on len(watches) == 0 — the orphan sweep
+	// below is what reaps tasks for disabled/deleted watches. Fall through.
 	p.logger.Debug("checking issue watches", zap.Int("count", len(watches)))
 	for _, watch := range watches {
 		if p.searchBucketExhausted("issue_watch") {
-			return
+			break
 		}
 		p.logger.Debug("polling issue watch",
 			zap.String("watch_id", watch.ID),
@@ -654,4 +663,9 @@ func (p *Poller) checkIssueWatches(ctx context.Context) {
 				zap.String("watch_id", watch.ID), zap.Int("deleted", cleaned))
 		}
 	}
+	if cleaned, err := p.service.CleanupAllOrphanedIssueTasks(ctx); err != nil {
+		p.logger.Warn("failed to sweep orphaned issue tasks", zap.Error(err))
+	} else if cleaned > 0 {
+		p.logger.Info("swept orphaned issue tasks", zap.Int("deleted", cleaned))
+	}
 }

apps/backend/internal/github/review_pr_task_test.go

@@ -233,14 +233,17 @@ func TestCleanupMergedReviewTasks_OrphanReservation(t *testing.T) {
 	if err != nil {
 		t.Fatalf("CleanupMergedReviewTasks: %v", err)
 	}
-	if deleted != 1 {
-		t.Errorf("expected 1 deleted, got %d", deleted)
+	// Returned count tracks deleted tasks specifically — orphan reservation
+	// rows have no associated task, so cleaning them shouldn't bump the
+	// count (the settings-page toast reports "Deleted N tasks").
+	if deleted != 0 {
+		t.Errorf("expected deleted=0 (orphan reservation has no task), got %d", deleted)
 	}
 	if len(rec.calls) != 0 {
 		t.Errorf("expected DeleteTask NOT to be called for orphan reservation, got calls=%v", rec.calls)
 	}
 
-	// The orphan dedup row must be gone.
+	// The orphan dedup row must still be gone.
 	remaining, err := store.ListReviewPRTasksByWatch(ctx, watch.ID)
 	if err != nil {
 		t.Fatalf("ListReviewPRTasksByWatch: %v", err)

apps/backend/internal/github/service.go

@@ -40,14 +40,39 @@ const (
 	defaultBranchMaster = "master"
 )
 
+// ErrTaskNotFound is the sentinel cleanup paths check for to distinguish
+// "the task is already gone — fine, mop up the dedup row" from a real
+// upstream failure. Adapter implementations of TaskDeleter wrap this when
+// the task domain reports a missing row so the github layer can recognize
+// the case without string-matching the underlying error message.
+var ErrTaskNotFound = errors.New("github: task not found for cleanup")
+
 // TaskDeleter deletes tasks by ID. Used for cleaning up merged PR tasks.
+// Implementations should return errors wrapping ErrTaskNotFound when the
+// task is already gone.
 type TaskDeleter interface {
 	DeleteTask(ctx context.Context, taskID string) error
 }
 
-// TaskSessionChecker checks if a task has any sessions (user interacted with it).
+// isTaskNotFound recognizes both the typed sentinel and the legacy
+// "not found" substring used by older adapters that haven't migrated yet.
+// String matching stays as a fallback so this PR doesn't regress installs
+// where the adapter wasn't updated in lockstep.
+func isTaskNotFound(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, ErrTaskNotFound) {
+		return true
+	}
+	return strings.Contains(err.Error(), "not found")
+}
+
+// TaskSessionChecker checks whether the user genuinely engaged with a task
+// (authored at least one non-auto-start message). Used by cleanup logic to
+// preserve tasks the user touched while sweeping auto-started-only ones.
 type TaskSessionChecker interface {
-	HasTaskSessions(ctx context.Context, taskID string) (bool, error)
+	HasUserAuthoredMessage(ctx context.Context, taskID string) (bool, error)
 }
 
 // SecretManager handles secret creation, update, and deletion.
@@ -60,6 +85,13 @@ type SecretManager interface {
 // prSyncFreshnessWindow is how long PR data is considered fresh (skip GitHub API).
 const prSyncFreshnessWindow = 30 * time.Second
 
+// cleanupFetchFailureThreshold is the number of consecutive GetPRFeedback /
+// GetIssueState errors a single dedup row may accumulate before the cleanup
+// loop logs a Warn. The previous behavior swallowed the error at Debug level
+// so transient outages — auth-token expiry, rate-limit exhaustion — silently
+// blocked cleanup forever.
+const cleanupFetchFailureThreshold = 5
+
 // Service coordinates GitHub integration operations.
 type Service struct {
 	mu                 sync.Mutex
@@ -78,21 +110,28 @@ type Service struct {
 	prStatusCache      *ttlCache
 	protectionCache    *branchProtectionCache
 	rateTracker        *RateTracker
+
+	// cleanupFailureMu guards cleanupFailureCounts; the cleanup loop is the
+	// only writer but the global sweep + per-watch sweep can run concurrently
+	// in different goroutines, and the map is shared between them.
+	cleanupFailureMu     sync.Mutex
+	cleanupFailureCounts map[string]int
 }
 
 // NewService creates a new GitHub service.
 func NewService(client Client, authMethod string, secrets SecretProvider, store *Store, eventBus bus.EventBus, log *logger.Logger) *Service {
 	return &Service{
-		client:          client,
-		authMethod:      authMethod,
-		secrets:         secrets,
-		store:           store,
-		eventBus:        eventBus,
-		logger:          log,
-		searchCache:     newTTLCache(),
-		prStatusCache:   newTTLCache(),
-		protectionCache: newBranchProtectionCache(),
-		rateTracker:     NewRateTracker(eventBus, log),
+		client:               client,
+		authMethod:           authMethod,
+		secrets:              secrets,
+		store:                store,
+		eventBus:             eventBus,
+		logger:               log,
+		searchCache:          newTTLCache(),
+		prStatusCache:        newTTLCache(),
+		protectionCache:      newBranchProtectionCache(),
+		rateTracker:          NewRateTracker(eventBus, log),
+		cleanupFailureCounts: make(map[string]int),
 	}
 }
 
@@ -1196,6 +1235,9 @@ func (s *Service) CreateReviewWatch(ctx context.Context, req *CreateReviewWatchR
 	if reviewScope == "" {
 		reviewScope = ReviewScopeUserAndTeams
 	}
+	if !IsValidCleanupPolicy(req.CleanupPolicy) {
+		return nil, fmt.Errorf("invalid cleanup_policy: %q", req.CleanupPolicy)
+	}
 	rw := &ReviewWatch{
 		WorkspaceID:         req.WorkspaceID,
 		WorkflowID:          req.WorkflowID,
@@ -1208,6 +1250,7 @@ func (s *Service) CreateReviewWatch(ctx context.Context, req *CreateReviewWatchR
 		CustomQuery:         req.CustomQuery,
 		Enabled:             true,
 		PollIntervalSeconds: req.PollIntervalSeconds,
+		CleanupPolicy:       NormalizeCleanupPolicy(req.CleanupPolicy),
 	}
 	if err := s.store.CreateReviewWatch(ctx, rw); err != nil {
 		return nil, fmt.Errorf("create review watch: %w", err)
@@ -1291,11 +1334,45 @@ func (s *Service) UpdateReviewWatch(ctx context.Context, id string, req *UpdateR
 	if req.PollIntervalSeconds != nil {
 		rw.PollIntervalSeconds = *req.PollIntervalSeconds
 	}
+	if req.CleanupPolicy != nil {
+		if !IsValidCleanupPolicy(*req.CleanupPolicy) {
+			return fmt.Errorf("invalid cleanup_policy: %q", *req.CleanupPolicy)
+		}
+		rw.CleanupPolicy = NormalizeCleanupPolicy(*req.CleanupPolicy)
+	}
 	return s.store.UpdateReviewWatch(ctx, rw)
 }
 
-// DeleteReviewWatch deletes a review watch.
+// DeleteReviewWatch deletes a review watch and best-effort reaps any tasks
+// it owned. The store layer drops the dedup rows transactionally with the
+// watch row, but tasks live in a separate domain and would leak forever
+// without this pre-pass (the global sweep can no longer find them after the
+// dedup rows are gone). True best-effort: a list error logs Warn and lets
+// the watch delete proceed so the user's primary action isn't blocked by
+// transient task-domain failures.
+//
+//nolint:nestif // straight-line list → for-loop → conditional delete; readable as-is
 func (s *Service) DeleteReviewWatch(ctx context.Context, id string) error {
+	if s.taskDeleter != nil {
+		prTasks, err := s.store.ListReviewPRTasksByWatch(ctx, id)
+		if err != nil {
+			s.logger.Warn("failed to list review PR tasks for pre-delete sweep",
+				zap.String("watch_id", id), zap.Error(err))
+		} else {
+			for _, rpt := range prTasks {
+				if rpt.TaskID == "" {
+					continue
+				}
+				if err := s.taskDeleter.DeleteTask(ctx, rpt.TaskID); err != nil &&
+					!isTaskNotFound(err) {
+					s.logger.Warn("failed to delete review task during watch cleanup",
+						zap.String("watch_id", id),
+						zap.String("task_id", rpt.TaskID),
+						zap.Error(err))
+				}
+			}
+		}
+	}
 	return s.store.DeleteReviewWatch(ctx, id)
 }
 
@@ -1731,27 +1808,149 @@ func (s *Service) CleanupMergedReviewTasks(ctx context.Context, watch *ReviewWat
 	if err != nil {
 		return 0, fmt.Errorf("list review PR tasks: %w", err)
 	}
+	policy := NormalizeCleanupPolicy(watch.CleanupPolicy)
+	return s.cleanupReviewPRTaskBatch(ctx, prTasks, func(_ *ReviewPRTask) string { return policy }), 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)
+}
+
+// CleanupAllReviewTasks sweeps every dedup row across all review watches,
+// including rows owned by currently-enabled watches. Used by the manual
+// 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)
+}
+
+// cleanupAllReviewTasks is the shared body. When orphansOnly is true, rows
+// whose watch is currently enabled are skipped (the per-watch poller will
+// 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) {
+	if s.client == nil || s.taskDeleter == nil {
+		return 0, nil
+	}
+	prTasks, err := s.store.ListAllReviewPRTasks(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("list all review PR tasks: %w", err)
+	}
+	if len(prTasks) == 0 {
+		return 0, nil
+	}
+	policyCache, enabledCache, unknownCache := s.buildReviewWatchCaches(ctx, prTasks)
+	// Allocate a fresh slice rather than reusing prTasks' backing array
+	// via prTasks[:0]; the in-place version is safe today (write index
+	// always trails read index) but silently breaks if a future caller
+	// reads prTasks after the filter.
+	candidates := make([]*ReviewPRTask, 0, len(prTasks))
+	for _, rpt := range prTasks {
+		// Watch fetch failed — skip this cycle so we don't fail-open and
+		// reap an enabled-watch row under the wrong policy.
+		if unknownCache[rpt.ReviewWatchID] {
+			continue
+		}
+		// Orphan-only path skips rows owned by an enabled watch; the
+		// per-watch loop already handled them in the same cycle.
+		if orphansOnly && enabledCache[rpt.ReviewWatchID] {
+			continue
+		}
+		candidates = append(candidates, rpt)
+	}
+	if len(candidates) == 0 {
+		return 0, nil
+	}
+	return s.cleanupReviewPRTaskBatch(ctx, candidates, func(rpt *ReviewPRTask) string {
+		if p, ok := policyCache[rpt.ReviewWatchID]; ok {
+			return p
+		}
+		// Watch was deleted: fall back to the documented default so the row
+		// (and any task it points at) can still be reaped.
+		return CleanupPolicyAuto
+	}), nil
+}
+
+// buildReviewWatchCaches loads the cleanup policy + enabled flag for each
+// distinct watch ID referenced by prTasks. A per-row fetch error logs Warn
+// and adds the watch ID to the returned `unknown` set — the caller MUST
+// skip those rows this cycle. Without that signal a transient DB hiccup
+// would silently fail-open: rows for the failed watch would be treated as
+// orphaned and reaped under the fallback auto policy, potentially losing
+// tasks the user wanted preserved under `never`. The next sweep cycle
+// retries the fetch and recovers naturally. Missing watches (deleted) are
+// distinct: they're absent from both `policy` and `enabled` but NOT in
+// `unknown`, so callers treat them as legitimate orphans.
+func (s *Service) buildReviewWatchCaches(ctx context.Context, prTasks []*ReviewPRTask) (policy map[string]string, enabled map[string]bool, unknown map[string]bool) {
+	seen := make(map[string]struct{})
+	for _, rpt := range prTasks {
+		seen[rpt.ReviewWatchID] = struct{}{}
+	}
+	policy = make(map[string]string, len(seen))
+	enabled = make(map[string]bool, len(seen))
+	unknown = make(map[string]bool)
+	for watchID := range seen {
+		watch, err := s.store.GetReviewWatch(ctx, watchID)
+		if err != nil {
+			s.logger.Warn("failed to fetch review watch during orphan sweep",
+				zap.String("watch_id", watchID), zap.Error(err))
+			unknown[watchID] = true
+			continue
+		}
+		if watch == nil {
+			continue
+		}
+		policy[watchID] = NormalizeCleanupPolicy(watch.CleanupPolicy)
+		enabled[watchID] = watch.Enabled
+	}
+	return policy, enabled, unknown
+}
+
+// cleanupReviewPRTaskBatch runs the deletion gate over a slice of dedup rows.
+// resolvePolicy returns the effective cleanup policy for each row; callers
+// supply it so per-watch and global-sweep paths can share this body.
+//
+//nolint:dupl // mirrors cleanupIssueTaskBatch — different types, same structure
+func (s *Service) cleanupReviewPRTaskBatch(ctx context.Context, prTasks []*ReviewPRTask, resolvePolicy func(*ReviewPRTask) string) int {
 	deleted := 0
 	for _, rpt := range prTasks {
+		policy := resolvePolicy(rpt)
 		// 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 == "" {
-			if should, _ := s.shouldDeleteReviewTask(ctx, rpt); should {
-				_ = s.store.DeleteReviewPRTask(ctx, rpt.ID)
-				deleted++
+			// 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.
+			if should, _ := s.shouldDeleteReviewTask(ctx, rpt, policy); 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 := s.shouldDeleteReviewTask(ctx, rpt)
+		shouldDelete, reason := s.shouldDeleteReviewTask(ctx, rpt, policy)
 		if !shouldDelete {
 			continue
 		}
 		if err := s.taskDeleter.DeleteTask(ctx, rpt.TaskID); err != nil {
-			if strings.Contains(err.Error(), "not found") {
+			if isTaskNotFound(err) {
 				// Task already deleted; clean up the orphaned dedup record.
-				_ = s.store.DeleteReviewPRTask(ctx, rpt.ID)
+				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
 			}
@@ -1759,28 +1958,45 @@ func (s *Service) CleanupMergedReviewTasks(ctx context.Context, watch *ReviewWat
 				zap.String("task_id", rpt.TaskID), zap.Error(err))
 			continue
 		}
-		_ = s.store.DeleteReviewPRTask(ctx, rpt.ID)
+		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++
 	}
-	return deleted, nil
+	return deleted
 }
 
-// shouldDeleteReviewTask checks if a review PR task should be cleaned up.
-// Returns true + reason if the PR is done (merged/closed/approved) AND the user
-// hasn't interacted with the task yet (no sessions). Tasks with sessions are
-// preserved so the user can see the PR merged banner and their work history.
-func (s *Service) shouldDeleteReviewTask(ctx context.Context, rpt *ReviewPRTask) (bool, string) {
-	feedback, err := s.client.GetPRFeedback(ctx, rpt.RepoOwner, rpt.RepoName, rpt.PRNumber)
-	if err != nil {
-		s.logger.Debug("failed to fetch PR feedback for cleanup",
-			zap.Int("pr_number", rpt.PRNumber), zap.Error(err))
+// 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.
+//   - CleanupPolicyAlways → terminal state alone is enough.
+//   - CleanupPolicyAuto   → terminal state + no user-authored messages.
+//
+// Terminal state covers: PR merged/closed, OR the authenticated user already
+// approved the PR on GitHub (so it's effectively done from their POV).
+func (s *Service) shouldDeleteReviewTask(ctx context.Context, rpt *ReviewPRTask, policy string) (bool, string) {
+	if policy == CleanupPolicyNever {
 		return false, ""
 	}
+	failureKey := reviewFailureKey(rpt)
+	feedback, err := s.client.GetPRFeedback(ctx, rpt.RepoOwner, rpt.RepoName, rpt.PRNumber)
+	if err != nil {
+		s.trackCleanupFailure(failureKey, "review", rpt.RepoOwner+"/"+rpt.RepoName, rpt.PRNumber, err)
+		return false, ""
+	}
+	s.resetCleanupFailure(failureKey)
 	if feedback.PR == nil {
 		return false, ""
 	}
@@ -1800,22 +2016,63 @@ func (s *Service) shouldDeleteReviewTask(ctx context.Context, rpt *ReviewPRTask)
 	if reason == "" {
 		return false, ""
 	}
-	// Don't delete tasks the user has interacted with (has sessions).
-	// Those show the PR merged banner instead.
-	if s.taskSessionChecker != nil {
-		hasSessions, err := s.taskSessionChecker.HasTaskSessions(ctx, rpt.TaskID)
-		if err != nil {
-			s.logger.Debug("failed to check task sessions",
-				zap.String("task_id", rpt.TaskID), zap.Error(err))
-			return false, ""
-		}
-		if hasSessions {
-			return false, ""
-		}
+	if policy == CleanupPolicyAlways || rpt.TaskID == "" || s.taskSessionChecker == nil {
+		return true, reason
+	}
+	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, ""
+	}
+	if hasUserMsg {
+		return false, ""
 	}
 	return true, reason
 }
 
+// reviewFailureKey builds the stable per-row identifier used for failure
+// tracking. Stable across polls so consecutive errors increment the same
+// counter and a recovery resets it cleanly. Includes the watch ID so two
+// watches monitoring the same (owner, repo, PR) don't collide and reset
+// each other's failure counters, suppressing the threshold-crossing Warn.
+func reviewFailureKey(rpt *ReviewPRTask) string {
+	return fmt.Sprintf("review:%s:%s/%s#%d", rpt.ReviewWatchID, rpt.RepoOwner, rpt.RepoName, rpt.PRNumber)
+}
+
+// trackCleanupFailure increments the failure counter for key and emits a
+// Warn log once the consecutive-failure threshold is crossed. Below the
+// threshold the failure is recorded at Debug so the normal log isn't flooded.
+func (s *Service) trackCleanupFailure(key, kind, repo string, number int, cause error) {
+	s.cleanupFailureMu.Lock()
+	s.cleanupFailureCounts[key]++
+	n := s.cleanupFailureCounts[key]
+	s.cleanupFailureMu.Unlock()
+	if n < cleanupFetchFailureThreshold {
+		s.logger.Debug("cleanup state fetch failed",
+			zap.String("kind", kind),
+			zap.String("repo", repo),
+			zap.Int("number", number),
+			zap.Int("consecutive_failures", n),
+			zap.Error(cause))
+		return
+	}
+	s.logger.Warn("cleanup state fetch failing repeatedly — blocked task deletion",
+		zap.String("kind", kind),
+		zap.String("repo", repo),
+		zap.Int("number", number),
+		zap.Int("consecutive_failures", n),
+		zap.Error(cause))
+}
+
+// resetCleanupFailure drops the counter for key once the upstream fetch
+// recovers, so a future flap doesn't cross the threshold prematurely.
+func (s *Service) resetCleanupFailure(key string) {
+	s.cleanupFailureMu.Lock()
+	delete(s.cleanupFailureCounts, key)
+	s.cleanupFailureMu.Unlock()
+}
+
 // TriggerAllReviewChecks triggers all review watches for a workspace.
 func (s *Service) TriggerAllReviewChecks(ctx context.Context, workspaceID string) (int, error) {
 	watches, err := s.store.ListReviewWatches(ctx, workspaceID)
@@ -1897,6 +2154,9 @@ func (s *Service) CreateIssueWatch(ctx context.Context, req *CreateIssueWatchReq
 	if labels == nil {
 		labels = []string{}
 	}
+	if !IsValidCleanupPolicy(req.CleanupPolicy) {
+		return nil, fmt.Errorf("invalid cleanup_policy: %q", req.CleanupPolicy)
+	}
 	iw := &IssueWatch{
 		WorkspaceID:         req.WorkspaceID,
 		WorkflowID:          req.WorkflowID,
@@ -1909,6 +2169,7 @@ func (s *Service) CreateIssueWatch(ctx context.Context, req *CreateIssueWatchReq
 		CustomQuery:         req.CustomQuery,
 		Enabled:             true,
 		PollIntervalSeconds: req.PollIntervalSeconds,
+		CleanupPolicy:       NormalizeCleanupPolicy(req.CleanupPolicy),
 	}
 	if err := s.store.CreateIssueWatch(ctx, iw); err != nil {
 		return nil, fmt.Errorf("create issue watch: %w", err)
@@ -1998,11 +2259,41 @@ func (s *Service) UpdateIssueWatch(ctx context.Context, id string, req *UpdateIs
 		}
 		iw.PollIntervalSeconds = v
 	}
+	if req.CleanupPolicy != nil {
+		if !IsValidCleanupPolicy(*req.CleanupPolicy) {
+			return fmt.Errorf("invalid cleanup_policy: %q", *req.CleanupPolicy)
+		}
+		iw.CleanupPolicy = NormalizeCleanupPolicy(*req.CleanupPolicy)
+	}
 	return s.store.UpdateIssueWatch(ctx, iw)
 }
 
-// DeleteIssueWatch deletes an issue watch.
+// DeleteIssueWatch deletes an issue watch and best-effort reaps any tasks
+// it owned (mirrors DeleteReviewWatch — list errors log Warn and let the
+// watch delete proceed).
+//
+//nolint:nestif // mirrors DeleteReviewWatch shape
 func (s *Service) DeleteIssueWatch(ctx context.Context, id string) error {
+	if s.taskDeleter != nil {
+		issueTasks, err := s.store.ListIssueWatchTasksByWatch(ctx, id)
+		if err != nil {
+			s.logger.Warn("failed to list issue tasks for pre-delete sweep",
+				zap.String("watch_id", id), zap.Error(err))
+		} else {
+			for _, it := range issueTasks {
+				if it.TaskID == "" {
+					continue
+				}
+				if err := s.taskDeleter.DeleteTask(ctx, it.TaskID); err != nil &&
+					!isTaskNotFound(err) {
+					s.logger.Warn("failed to delete issue task during watch cleanup",
+						zap.String("watch_id", id),
+						zap.String("task_id", it.TaskID),
+						zap.Error(err))
+				}
+			}
+		}
+	}
 	return s.store.DeleteIssueWatch(ctx, id)
 }
 
@@ -2175,35 +2466,119 @@ func (s *Service) TriggerAllIssueChecks(ctx context.Context, workspaceID string)
 }
 
 // CleanupClosedIssueTasks checks issues tracked by a watch and deletes
-// tasks whose issues are closed and the user hasn't interacted with.
+// tasks whose issues are closed under the watch's cleanup policy.
 //
 //nolint:dupl // mirrors CleanupMergedReviewTasks — different types, same structure
 func (s *Service) CleanupClosedIssueTasks(ctx context.Context, watch *IssueWatch) (int, error) {
-	if s.client == nil || s.taskDeleter == nil || s.taskSessionChecker == nil {
+	if s.client == nil || s.taskDeleter == nil {
 		return 0, nil
 	}
 	issueTasks, err := s.store.ListIssueWatchTasksByWatch(ctx, watch.ID)
 	if err != nil {
 		return 0, fmt.Errorf("list issue watch tasks: %w", err)
 	}
+	policy := NormalizeCleanupPolicy(watch.CleanupPolicy)
+	return s.cleanupIssueTaskBatch(ctx, issueTasks, func(_ *IssueWatchTask) string { return policy }), nil
+}
+
+// CleanupAllOrphanedIssueTasks sweeps dedup rows whose watch is deleted or
+// disabled (mirrors CleanupAllOrphanedReviewTasks).
+func (s *Service) CleanupAllOrphanedIssueTasks(ctx context.Context) (int, error) {
+	return s.cleanupAllIssueTasks(ctx, true)
+}
+
+// CleanupAllIssueTasks sweeps every dedup row across all issue watches for
+// the manual settings-page button.
+func (s *Service) CleanupAllIssueTasks(ctx context.Context) (int, error) {
+	return s.cleanupAllIssueTasks(ctx, false)
+}
+
+//nolint:dupl // mirrors cleanupAllReviewTasks — different types, same orchestration
+func (s *Service) cleanupAllIssueTasks(ctx context.Context, orphansOnly bool) (int, error) {
+	if s.client == nil || s.taskDeleter == nil {
+		return 0, nil
+	}
+	issueTasks, err := s.store.ListAllIssueWatchTasks(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("list all issue watch tasks: %w", err)
+	}
+	if len(issueTasks) == 0 {
+		return 0, nil
+	}
+	policyCache, enabledCache, unknownCache := s.buildIssueWatchCaches(ctx, issueTasks)
+	candidates := make([]*IssueWatchTask, 0, len(issueTasks))
+	for _, it := range issueTasks {
+		if unknownCache[it.IssueWatchID] {
+			continue
+		}
+		if orphansOnly && enabledCache[it.IssueWatchID] {
+			continue
+		}
+		candidates = append(candidates, it)
+	}
+	if len(candidates) == 0 {
+		return 0, nil
+	}
+	return s.cleanupIssueTaskBatch(ctx, candidates, func(it *IssueWatchTask) string {
+		if p, ok := policyCache[it.IssueWatchID]; ok {
+			return p
+		}
+		return CleanupPolicyAuto
+	}), nil
+}
+
+func (s *Service) buildIssueWatchCaches(ctx context.Context, issueTasks []*IssueWatchTask) (policy map[string]string, enabled map[string]bool, unknown map[string]bool) {
+	seen := make(map[string]struct{})
+	for _, it := range issueTasks {
+		seen[it.IssueWatchID] = struct{}{}
+	}
+	policy = make(map[string]string, len(seen))
+	enabled = make(map[string]bool, len(seen))
+	unknown = make(map[string]bool)
+	for watchID := range seen {
+		watch, err := s.store.GetIssueWatch(ctx, watchID)
+		if err != nil {
+			s.logger.Warn("failed to fetch issue watch during orphan sweep",
+				zap.String("watch_id", watchID), zap.Error(err))
+			unknown[watchID] = true
+			continue
+		}
+		if watch == nil {
+			continue
+		}
+		policy[watchID] = NormalizeCleanupPolicy(watch.CleanupPolicy)
+		enabled[watchID] = watch.Enabled
+	}
+	return policy, enabled, unknown
+}
+
+//nolint:dupl // mirrors cleanupReviewPRTaskBatch — different types, same structure
+func (s *Service) cleanupIssueTaskBatch(ctx context.Context, issueTasks []*IssueWatchTask, resolvePolicy func(*IssueWatchTask) string) int {
 	deleted := 0
 	for _, it := range issueTasks {
-		// Orphan reservation: task_id is empty because the process was
-		// killed between Reserve and Assign. Clean up once the issue is closed.
+		policy := resolvePolicy(it)
 		if it.TaskID == "" {
-			if should, _ := s.shouldDeleteIssueTask(ctx, it); should {
-				_ = s.store.DeleteIssueWatchTask(ctx, it.ID)
-				deleted++
+			// Orphan reservation row — no task was created. Clean it up
+			// but don't count it as a deleted task.
+			if should, _ := s.shouldDeleteIssueTask(ctx, it, policy); should {
+				if err := s.store.DeleteIssueWatchTask(ctx, it.ID); err != nil {
+					s.logger.Warn("failed to delete orphan reservation row",
+						zap.String("dedup_id", it.ID), zap.Error(err))
+				}
 			}
 			continue
 		}
-		shouldDelete, reason := s.shouldDeleteIssueTask(ctx, it)
+		shouldDelete, reason := s.shouldDeleteIssueTask(ctx, it, policy)
 		if !shouldDelete {
 			continue
 		}
 		if err := s.taskDeleter.DeleteTask(ctx, it.TaskID); err != nil {
-			if strings.Contains(err.Error(), "not found") {
-				_ = s.store.DeleteIssueWatchTask(ctx, it.ID)
+			if isTaskNotFound(err) {
+				if err := s.store.DeleteIssueWatchTask(ctx, it.ID); err != nil {
+					s.logger.Warn("failed to delete orphan dedup row after task-not-found",
+						zap.String("dedup_id", it.ID), zap.Error(err))
+					continue
+				}
 				deleted++
 				continue
 			}
@@ -2211,46 +2586,59 @@ func (s *Service) CleanupClosedIssueTasks(ctx context.Context, watch *IssueWatch
 				zap.String("task_id", it.TaskID), zap.Error(err))
 			continue
 		}
-		_ = s.store.DeleteIssueWatchTask(ctx, it.ID)
+		if err := s.store.DeleteIssueWatchTask(ctx, it.ID); err != nil {
+			s.logger.Warn("deleted task but failed to remove dedup row",
+				zap.String("task_id", it.TaskID),
+				zap.String("dedup_id", it.ID), zap.Error(err))
+			continue
+		}
 		s.logger.Info("deleted issue task",
 			zap.String("task_id", it.TaskID),
 			zap.String("reason", reason),
+			zap.String("policy", policy),
 			zap.Int("issue_number", it.IssueNumber),
 			zap.String("repo", it.RepoOwner+"/"+it.RepoName))
 		deleted++
 	}
-	return deleted, nil
+	return deleted
 }
 
-// shouldDeleteIssueTask checks if an issue task should be cleaned up.
-// Returns true + reason if the issue is closed AND the user hasn't interacted
-// with the task yet (no sessions).
-func (s *Service) shouldDeleteIssueTask(ctx context.Context, it *IssueWatchTask) (bool, string) {
-	state, err := s.client.GetIssueState(ctx, it.RepoOwner, it.RepoName, it.IssueNumber)
-	if err != nil {
-		s.logger.Debug("failed to fetch issue state for cleanup",
-			zap.Int("issue_number", it.IssueNumber), zap.Error(err))
+// shouldDeleteIssueTask checks whether an issue task is eligible for cleanup.
+// Policy gating mirrors shouldDeleteReviewTask.
+func (s *Service) shouldDeleteIssueTask(ctx context.Context, it *IssueWatchTask, policy string) (bool, string) {
+	if policy == CleanupPolicyNever {
 		return false, ""
 	}
+	failureKey := issueFailureKey(it)
+	state, err := s.client.GetIssueState(ctx, it.RepoOwner, it.RepoName, it.IssueNumber)
+	if err != nil {
+		s.trackCleanupFailure(failureKey, "issue", it.RepoOwner+"/"+it.RepoName, it.IssueNumber, err)
+		return false, ""
+	}
+	s.resetCleanupFailure(failureKey)
 	if state != "closed" {
 		return false, ""
 	}
 	reason := "issue_closed"
-	// Don't delete tasks the user has interacted with (has sessions).
-	if it.TaskID != "" {
-		hasSessions, err := s.taskSessionChecker.HasTaskSessions(ctx, it.TaskID)
-		if err != nil {
-			s.logger.Debug("failed to check task sessions",
-				zap.String("task_id", it.TaskID), zap.Error(err))
-			return false, ""
-		}
-		if hasSessions {
-			return false, ""
-		}
+	if policy == CleanupPolicyAlways || it.TaskID == "" || s.taskSessionChecker == nil {
+		return true, reason
+	}
+	hasUserMsg, err := s.taskSessionChecker.HasUserAuthoredMessage(ctx, it.TaskID)
+	if err != nil {
+		s.logger.Debug("failed to check task user messages",
+			zap.String("task_id", it.TaskID), zap.Error(err))
+		return false, ""
+	}
+	if hasUserMsg {
+		return false, ""
 	}
 	return true, reason
 }
 
+func issueFailureKey(it *IssueWatchTask) string {
+	return fmt.Sprintf("issue:%s:%s/%s#%d", it.IssueWatchID, it.RepoOwner, it.RepoName, it.IssueNumber)
+}
+
 func (s *Service) publishNewIssueEvent(ctx context.Context, watch *IssueWatch, issue *Issue) {
 	if s.eventBus == nil {
 		return

apps/backend/internal/github/service_cleanup_policy_test.go

@@ -0,0 +1,732 @@
+package github
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/kandev/kandev/internal/common/logger"
+)
+
+func testLogger(t *testing.T) *logger.Logger {
+	t.Helper()
+	log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"})
+	if err != nil {
+		t.Fatalf("create test logger: %v", err)
+	}
+	return log
+}
+
+// recordingSessionChecker satisfies TaskSessionChecker for tests and lets the
+// caller dial in "did the user type something" without spinning up a real
+// task repository.
+type recordingSessionChecker struct {
+	hasUserMsg bool
+	err        error
+	calls      int
+}
+
+func (s *recordingSessionChecker) HasUserAuthoredMessage(_ context.Context, _ string) (bool, error) {
+	s.calls++
+	return s.hasUserMsg, s.err
+}
+
+// prFeedbackStub returns a canned PRFeedback for shouldDeleteReviewTask tests.
+// Unrelated client methods come from NoopClient.
+type prFeedbackStub struct {
+	NoopClient
+	state string
+	err   error
+}
+
+func (c *prFeedbackStub) GetPRFeedback(_ context.Context, _, _ string, _ int) (*PRFeedback, error) {
+	if c.err != nil {
+		return nil, c.err
+	}
+	return &PRFeedback{PR: &PR{State: c.state}}, nil
+}
+
+func reviewTaskFixture() *ReviewPRTask {
+	return &ReviewPRTask{
+		ID:            "rpt-1",
+		ReviewWatchID: "rw-1",
+		RepoOwner:     "acme",
+		RepoName:      "widget",
+		PRNumber:      99,
+		TaskID:        "task-99",
+	}
+}
+
+func TestShouldDeleteReviewTask_AutoPolicy_NoUserMessages_Deletes(t *testing.T) {
+	log := testLogger(t)
+	checker := &recordingSessionChecker{hasUserMsg: false}
+	svc := newCleanupTestService(&prFeedbackStub{state: prStateMerged}, log, checker)
+
+	del, reason := svc.shouldDeleteReviewTask(context.Background(), reviewTaskFixture(), CleanupPolicyAuto)
+	if !del {
+		t.Fatalf("expected delete=true under auto with no user messages")
+	}
+	if reason != "pr_merged_or_closed" {
+		t.Fatalf("reason=%q, want pr_merged_or_closed", reason)
+	}
+	if checker.calls != 1 {
+		t.Fatalf("HasUserAuthoredMessage calls=%d, want 1", checker.calls)
+	}
+}
+
+func TestShouldDeleteReviewTask_AutoPolicy_UserMessage_Preserves(t *testing.T) {
+	log := testLogger(t)
+	checker := &recordingSessionChecker{hasUserMsg: true}
+	svc := newCleanupTestService(&prFeedbackStub{state: prStateClosed}, log, checker)
+
+	del, _ := svc.shouldDeleteReviewTask(context.Background(), reviewTaskFixture(), CleanupPolicyAuto)
+	if del {
+		t.Fatalf("expected delete=false when user authored a message under auto")
+	}
+}
+
+func TestShouldDeleteReviewTask_AlwaysPolicy_IgnoresUserMessages(t *testing.T) {
+	log := testLogger(t)
+	checker := &recordingSessionChecker{hasUserMsg: true}
+	svc := newCleanupTestService(&prFeedbackStub{state: prStateMerged}, log, checker)
+
+	del, reason := svc.shouldDeleteReviewTask(context.Background(), reviewTaskFixture(), CleanupPolicyAlways)
+	if !del {
+		t.Fatalf("expected delete=true under always-policy even with user messages")
+	}
+	if reason != "pr_merged_or_closed" {
+		t.Fatalf("reason=%q, want pr_merged_or_closed", reason)
+	}
+	if checker.calls != 0 {
+		t.Fatalf("HasUserAuthoredMessage calls=%d, want 0 (always-policy short-circuits)", checker.calls)
+	}
+}
+
+func TestShouldDeleteReviewTask_NeverPolicy_KeepsEverything(t *testing.T) {
+	log := testLogger(t)
+	checker := &recordingSessionChecker{hasUserMsg: false}
+	svc := newCleanupTestService(&prFeedbackStub{state: prStateMerged}, log, checker)
+
+	del, _ := svc.shouldDeleteReviewTask(context.Background(), reviewTaskFixture(), CleanupPolicyNever)
+	if del {
+		t.Fatalf("expected delete=false under never-policy")
+	}
+	// Never-policy must skip the upstream call too — no point burning quota.
+	if checker.calls != 0 {
+		t.Fatalf("HasUserAuthoredMessage calls=%d, want 0 (never-policy short-circuits)", checker.calls)
+	}
+}
+
+func TestShouldDeleteReviewTask_OpenPR_KeepsTask(t *testing.T) {
+	log := testLogger(t)
+	checker := &recordingSessionChecker{hasUserMsg: false}
+	svc := newCleanupTestService(&prFeedbackStub{state: "open"}, log, checker)
+
+	del, _ := svc.shouldDeleteReviewTask(context.Background(), reviewTaskFixture(), CleanupPolicyAuto)
+	if del {
+		t.Fatalf("expected delete=false for open PR")
+	}
+}
+
+func TestShouldDeleteReviewTask_GitHubError_TracksFailureCount(t *testing.T) {
+	log := testLogger(t)
+	svc := newCleanupTestService(&prFeedbackStub{err: fmt.Errorf("rate limit hit")}, log, &recordingSessionChecker{})
+
+	rpt := reviewTaskFixture()
+	for i := 0; i < cleanupFetchFailureThreshold+1; i++ {
+		if del, _ := svc.shouldDeleteReviewTask(context.Background(), rpt, CleanupPolicyAuto); del {
+			t.Fatalf("iteration %d: expected delete=false under fetch error", i)
+		}
+	}
+	// The counter is keyed per-row; one error → one increment.
+	key := reviewFailureKey(rpt)
+	svc.cleanupFailureMu.Lock()
+	got := svc.cleanupFailureCounts[key]
+	svc.cleanupFailureMu.Unlock()
+	if got != cleanupFetchFailureThreshold+1 {
+		t.Fatalf("failure count = %d, want %d", got, cleanupFetchFailureThreshold+1)
+	}
+}
+
+func TestShouldDeleteReviewTask_GitHubErrorThenSuccess_ResetsCounter(t *testing.T) {
+	log := testLogger(t)
+	client := &switchingFeedbackClient{err: fmt.Errorf("transient")}
+	svc := newCleanupTestService(client, log, &recordingSessionChecker{})
+
+	rpt := reviewTaskFixture()
+	// First call fails — only the side effect (counter inc) matters here.
+	_, _ = svc.shouldDeleteReviewTask(context.Background(), rpt, CleanupPolicyAuto)
+	// Now switch the client to a healthy response — the next call must reset.
+	client.err = nil
+	client.state = prStateMerged
+	_, _ = svc.shouldDeleteReviewTask(context.Background(), rpt, CleanupPolicyAuto)
+	key := reviewFailureKey(rpt)
+	svc.cleanupFailureMu.Lock()
+	_, present := svc.cleanupFailureCounts[key]
+	svc.cleanupFailureMu.Unlock()
+	if present {
+		t.Fatalf("failure counter should be cleared after a successful fetch")
+	}
+}
+
+func TestCleanupAllOrphanedReviewTasks_DeletedWatchOrphans(t *testing.T) {
+	_, svc, mockClient, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	// Watch + dedup row + merged PR.
+	watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true}
+	if err := store.CreateReviewWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateReviewWatch: %v", err)
+	}
+	rpt := &ReviewPRTask{
+		ReviewWatchID: watch.ID,
+		RepoOwner:     "acme",
+		RepoName:      "widget",
+		PRNumber:      77,
+		PRURL:         "https://github.com/acme/widget/pull/77",
+		TaskID:        "task-77",
+	}
+	if err := store.CreateReviewPRTask(ctx, rpt); err != nil {
+		t.Fatalf("CreateReviewPRTask: %v", err)
+	}
+	mockClient.AddPR(&PR{Number: 77, State: prStateMerged, RepoOwner: "acme", RepoName: "widget"})
+
+	// Simulate the watch having been deleted out from under the dedup row by
+	// removing only the watch (bypassing the cascade so the orphan survives).
+	if _, err := store.db.Exec(`DELETE FROM github_review_watches WHERE id = ?`, watch.ID); err != nil {
+		t.Fatalf("manual watch delete: %v", err)
+	}
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedReviewTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedReviewTasks: %v", err)
+	}
+	if deleted != 1 {
+		t.Fatalf("deleted=%d, want 1 (orphan with missing watch should fall back to auto policy and reap)", deleted)
+	}
+	if len(rec.calls) != 1 || rec.calls[0] != "task-77" {
+		t.Fatalf("DeleteTask calls=%v, want [task-77]", rec.calls)
+	}
+}
+
+func TestCleanupAllOrphanedReviewTasks_DisabledWatch_StillCleansUp(t *testing.T) {
+	_, svc, mockClient, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: false}
+	if err := store.CreateReviewWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateReviewWatch: %v", err)
+	}
+	rpt := &ReviewPRTask{
+		ReviewWatchID: watch.ID,
+		RepoOwner:     "acme",
+		RepoName:      "widget",
+		PRNumber:      88,
+		TaskID:        "task-88",
+	}
+	if err := store.CreateReviewPRTask(ctx, rpt); err != nil {
+		t.Fatalf("CreateReviewPRTask: %v", err)
+	}
+	mockClient.AddPR(&PR{Number: 88, State: prStateMerged, RepoOwner: "acme", RepoName: "widget"})
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedReviewTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedReviewTasks: %v", err)
+	}
+	if deleted != 1 {
+		t.Fatalf("deleted=%d, want 1 (disabled watch's PR task should still be reaped by global sweep)", deleted)
+	}
+}
+
+// Regression for cubic P1: when GetReviewWatch fails for a watch, the row
+// MUST be skipped this cycle. Treating it as orphan would fail-open under
+// the auto policy, reaping enabled-watch rows (potentially even rows whose
+// real watch is configured as `never`) on a transient DB hiccup.
+func TestCleanupAllOrphanedReviewTasks_UnknownWatchSkipped(t *testing.T) {
+	_, svc, mockClient, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyNever}
+	if err := store.CreateReviewWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateReviewWatch: %v", err)
+	}
+	rpt := &ReviewPRTask{
+		ReviewWatchID: watch.ID,
+		RepoOwner:     "acme",
+		RepoName:      "widget",
+		PRNumber:      222,
+		TaskID:        "task-222",
+	}
+	if err := store.CreateReviewPRTask(ctx, rpt); err != nil {
+		t.Fatalf("CreateReviewPRTask: %v", err)
+	}
+	mockClient.AddPR(&PR{Number: 222, State: prStateMerged, RepoOwner: "acme", RepoName: "widget"})
+
+	// Wedge the watch row so GetReviewWatch errors. Setting a bad enum value
+	// would still parse; instead, drop the table briefly. We simulate the
+	// store error by closing the read DB connection.
+	if _, err := store.db.Exec(`ALTER TABLE github_review_watches RENAME TO github_review_watches_x`); err != nil {
+		t.Fatalf("rename watches table: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = store.db.Exec(`ALTER TABLE github_review_watches_x RENAME TO github_review_watches`)
+	})
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedReviewTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedReviewTasks: %v", err)
+	}
+	if deleted != 0 {
+		t.Fatalf("deleted=%d, want 0 (row whose watch fetch errored must be skipped, not fail-open as orphan)", deleted)
+	}
+	if len(rec.calls) != 0 {
+		t.Fatalf("DeleteTask called %d times, want 0", len(rec.calls))
+	}
+}
+
+// Regression for greptile P2: the global sweep must not re-hit GitHub for
+// dedup rows the per-watch loop already processed in the same cycle.
+// CleanupAllOrphanedReviewTasks skips rows whose watch is currently
+// enabled — the per-watch loop is responsible for those.
+func TestCleanupAllOrphanedReviewTasks_SkipsEnabledWatchRows(t *testing.T) {
+	_, svc, mockClient, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true}
+	if err := store.CreateReviewWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateReviewWatch: %v", err)
+	}
+	rpt := &ReviewPRTask{
+		ReviewWatchID: watch.ID,
+		RepoOwner:     "acme",
+		RepoName:      "widget",
+		PRNumber:      111,
+		TaskID:        "task-111",
+	}
+	if err := store.CreateReviewPRTask(ctx, rpt); err != nil {
+		t.Fatalf("CreateReviewPRTask: %v", err)
+	}
+	// PR is merged — without the skip, the orphan sweep would happily
+	// delete it on top of the per-watch loop.
+	mockClient.AddPR(&PR{Number: 111, State: prStateMerged, RepoOwner: "acme", RepoName: "widget"})
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedReviewTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedReviewTasks: %v", err)
+	}
+	if deleted != 0 {
+		t.Fatalf("deleted=%d, want 0 (enabled-watch rows must be skipped by the orphan sweep)", deleted)
+	}
+	if len(rec.calls) != 0 {
+		t.Fatalf("DeleteTask called %d times, want 0", len(rec.calls))
+	}
+
+	// And CleanupAllReviewTasks (manual button path) reaps the same row.
+	deleted, err = svc.CleanupAllReviewTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllReviewTasks: %v", err)
+	}
+	if deleted != 1 {
+		t.Fatalf("manual sweep deleted=%d, want 1", deleted)
+	}
+}
+
+func TestCleanupAllOrphanedReviewTasks_RespectsNeverPolicy(t *testing.T) {
+	_, svc, mockClient, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyNever}
+	if err := store.CreateReviewWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateReviewWatch: %v", err)
+	}
+	rpt := &ReviewPRTask{
+		ReviewWatchID: watch.ID,
+		RepoOwner:     "acme",
+		RepoName:      "widget",
+		PRNumber:      55,
+		TaskID:        "task-55",
+	}
+	if err := store.CreateReviewPRTask(ctx, rpt); err != nil {
+		t.Fatalf("CreateReviewPRTask: %v", err)
+	}
+	mockClient.AddPR(&PR{Number: 55, State: prStateMerged, RepoOwner: "acme", RepoName: "widget"})
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedReviewTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedReviewTasks: %v", err)
+	}
+	if deleted != 0 {
+		t.Fatalf("deleted=%d, want 0 (never-policy must keep tasks even when PR merges)", deleted)
+	}
+	if len(rec.calls) != 0 {
+		t.Fatalf("DeleteTask calls=%v, want none", rec.calls)
+	}
+}
+
+// Regression for claude blocker on c4d0a678: when ListEnabledReviewWatches
+// returns an empty slice (user disabled or deleted every watch), the
+// poller used to early-return BEFORE running the orphan sweep — so the
+// sweep that exists precisely for this scenario never fired. Verify the
+// poller's checkReviewWatches now runs the sweep through to completion
+// and reaps the orphan task.
+func TestCheckReviewWatches_RunsOrphanSweepWhenNoEnabledWatches(t *testing.T) {
+	poller, svc, mockClient, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	// Create a watch + dedup row, then disable the watch so
+	// ListEnabledReviewWatches returns empty. The dedup row + task survive.
+	watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: false}
+	if err := store.CreateReviewWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateReviewWatch: %v", err)
+	}
+	rpt := &ReviewPRTask{
+		ReviewWatchID: watch.ID,
+		RepoOwner:     "acme",
+		RepoName:      "widget",
+		PRNumber:      9001,
+		TaskID:        "task-9001",
+	}
+	if err := store.CreateReviewPRTask(ctx, rpt); err != nil {
+		t.Fatalf("CreateReviewPRTask: %v", err)
+	}
+	mockClient.AddPR(&PR{Number: 9001, State: prStateMerged, RepoOwner: "acme", RepoName: "widget"})
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	poller.checkReviewWatches(ctx)
+
+	if len(rec.calls) != 1 || rec.calls[0] != "task-9001" {
+		t.Fatalf("DeleteTask calls=%v, want [task-9001] (orphan sweep must run even with zero enabled watches)", rec.calls)
+	}
+}
+
+// Same regression on the issue side.
+func TestCheckIssueWatches_RunsOrphanSweepWhenNoEnabledWatches(t *testing.T) {
+	poller, svc, _, store := setupPollerTest(t)
+	ctx := context.Background()
+	svc.client = newIssueStateMockClient(map[int]string{9002: "closed"})
+
+	watch := &IssueWatch{WorkspaceID: "ws-1", Enabled: false}
+	if err := store.CreateIssueWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateIssueWatch: %v", err)
+	}
+	_, _ = store.ReserveIssueWatchTask(ctx, watch.ID, "acme", "widget", 9002, "https://example/9002")
+	if err := store.AssignIssueWatchTaskID(ctx, watch.ID, "acme", "widget", 9002, "task-9002"); err != nil {
+		t.Fatalf("AssignIssueWatchTaskID: %v", err)
+	}
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	poller.checkIssueWatches(ctx)
+
+	if len(rec.calls) != 1 || rec.calls[0] != "task-9002" {
+		t.Fatalf("DeleteTask calls=%v, want [task-9002] (orphan sweep must run even with zero enabled watches)", rec.calls)
+	}
+}
+
+// --- Issue-side mirrors ---
+// These tests mirror the review-side coverage above for the symmetric
+// issue path. Without them a copy-paste divergence between
+// cleanupAllReviewTasks and cleanupAllIssueTasks — wrong cache key,
+// inverted predicate, forgotten unknown-watch guard — would only surface
+// in a production incident.
+
+// issueStateMockClient extends the default MockClient with per-issue state
+// lookup. MockClient.GetIssueState always returns "open"; cleanup tests
+// need to flip individual issues to "closed".
+type issueStateMockClient struct {
+	*MockClient
+	states map[int]string // issue number -> state
+}
+
+func newIssueStateMockClient(states map[int]string) *issueStateMockClient {
+	return &issueStateMockClient{MockClient: NewMockClient(), states: states}
+}
+
+func (c *issueStateMockClient) GetIssueState(_ context.Context, _, _ string, number int) (string, error) {
+	if s, ok := c.states[number]; ok {
+		return s, nil
+	}
+	return "open", nil
+}
+
+func TestCleanupAllOrphanedIssueTasks_DeletedWatchOrphans(t *testing.T) {
+	_, svc, _, store := setupPollerTest(t)
+	ctx := context.Background()
+	svc.client = newIssueStateMockClient(map[int]string{77: "closed"})
+
+	watch := &IssueWatch{WorkspaceID: "ws-1", Enabled: true}
+	if err := store.CreateIssueWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateIssueWatch: %v", err)
+	}
+	reserved, err := store.ReserveIssueWatchTask(ctx, watch.ID, "acme", "widget", 77, "https://example/77")
+	if err != nil || !reserved {
+		t.Fatalf("ReserveIssueWatchTask: ok=%v err=%v", reserved, err)
+	}
+	if err := store.AssignIssueWatchTaskID(ctx, watch.ID, "acme", "widget", 77, "task-77"); err != nil {
+		t.Fatalf("AssignIssueWatchTaskID: %v", err)
+	}
+
+	if _, err := store.db.Exec(`DELETE FROM github_issue_watches WHERE id = ?`, watch.ID); err != nil {
+		t.Fatalf("manual watch delete: %v", err)
+	}
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedIssueTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedIssueTasks: %v", err)
+	}
+	if deleted != 1 {
+		t.Fatalf("deleted=%d, want 1 (orphan with missing watch should fall back to auto policy and reap)", deleted)
+	}
+	if len(rec.calls) != 1 || rec.calls[0] != "task-77" {
+		t.Fatalf("DeleteTask calls=%v, want [task-77]", rec.calls)
+	}
+}
+
+func TestCleanupAllOrphanedIssueTasks_DisabledWatch_StillCleansUp(t *testing.T) {
+	_, svc, _, store := setupPollerTest(t)
+	ctx := context.Background()
+	svc.client = newIssueStateMockClient(map[int]string{88: "closed"})
+
+	watch := &IssueWatch{WorkspaceID: "ws-1", Enabled: false}
+	if err := store.CreateIssueWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateIssueWatch: %v", err)
+	}
+	_, _ = store.ReserveIssueWatchTask(ctx, watch.ID, "acme", "widget", 88, "https://example/88")
+	if err := store.AssignIssueWatchTaskID(ctx, watch.ID, "acme", "widget", 88, "task-88"); err != nil {
+		t.Fatalf("AssignIssueWatchTaskID: %v", err)
+	}
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedIssueTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedIssueTasks: %v", err)
+	}
+	if deleted != 1 {
+		t.Fatalf("deleted=%d, want 1 (disabled watch's issue task should still be reaped by global sweep)", deleted)
+	}
+}
+
+func TestCleanupAllOrphanedIssueTasks_UnknownWatchSkipped(t *testing.T) {
+	_, svc, _, store := setupPollerTest(t)
+	ctx := context.Background()
+	svc.client = newIssueStateMockClient(map[int]string{222: "closed"})
+
+	watch := &IssueWatch{WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyNever}
+	if err := store.CreateIssueWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateIssueWatch: %v", err)
+	}
+	_, _ = store.ReserveIssueWatchTask(ctx, watch.ID, "acme", "widget", 222, "https://example/222")
+	if err := store.AssignIssueWatchTaskID(ctx, watch.ID, "acme", "widget", 222, "task-222"); err != nil {
+		t.Fatalf("AssignIssueWatchTaskID: %v", err)
+	}
+
+	if _, err := store.db.Exec(`ALTER TABLE github_issue_watches RENAME TO github_issue_watches_x`); err != nil {
+		t.Fatalf("rename watches table: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = store.db.Exec(`ALTER TABLE github_issue_watches_x RENAME TO github_issue_watches`)
+	})
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedIssueTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedIssueTasks: %v", err)
+	}
+	if deleted != 0 {
+		t.Fatalf("deleted=%d, want 0 (row whose watch fetch errored must be skipped, not fail-open as orphan)", deleted)
+	}
+	if len(rec.calls) != 0 {
+		t.Fatalf("DeleteTask called %d times, want 0", len(rec.calls))
+	}
+}
+
+func TestCleanupAllOrphanedIssueTasks_SkipsEnabledWatchRows(t *testing.T) {
+	_, svc, _, store := setupPollerTest(t)
+	ctx := context.Background()
+	svc.client = newIssueStateMockClient(map[int]string{111: "closed"})
+
+	watch := &IssueWatch{WorkspaceID: "ws-1", Enabled: true}
+	if err := store.CreateIssueWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateIssueWatch: %v", err)
+	}
+	_, _ = store.ReserveIssueWatchTask(ctx, watch.ID, "acme", "widget", 111, "https://example/111")
+	if err := store.AssignIssueWatchTaskID(ctx, watch.ID, "acme", "widget", 111, "task-111"); err != nil {
+		t.Fatalf("AssignIssueWatchTaskID: %v", err)
+	}
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedIssueTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedIssueTasks: %v", err)
+	}
+	if deleted != 0 {
+		t.Fatalf("deleted=%d, want 0 (enabled-watch rows must be skipped by the orphan sweep)", deleted)
+	}
+	if len(rec.calls) != 0 {
+		t.Fatalf("DeleteTask called %d times, want 0", len(rec.calls))
+	}
+
+	// And CleanupAllIssueTasks (manual button path) reaps the same row.
+	deleted, err = svc.CleanupAllIssueTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllIssueTasks: %v", err)
+	}
+	if deleted != 1 {
+		t.Fatalf("manual sweep deleted=%d, want 1", deleted)
+	}
+}
+
+func TestCleanupAllOrphanedIssueTasks_RespectsNeverPolicy(t *testing.T) {
+	_, svc, _, store := setupPollerTest(t)
+	ctx := context.Background()
+	svc.client = newIssueStateMockClient(map[int]string{55: "closed"})
+
+	watch := &IssueWatch{WorkspaceID: "ws-1", Enabled: true, CleanupPolicy: CleanupPolicyNever}
+	if err := store.CreateIssueWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateIssueWatch: %v", err)
+	}
+	_, _ = store.ReserveIssueWatchTask(ctx, watch.ID, "acme", "widget", 55, "https://example/55")
+	if err := store.AssignIssueWatchTaskID(ctx, watch.ID, "acme", "widget", 55, "task-55"); err != nil {
+		t.Fatalf("AssignIssueWatchTaskID: %v", err)
+	}
+
+	rec := &recordingTaskDeleter{}
+	svc.SetTaskDeleter(rec)
+
+	deleted, err := svc.CleanupAllOrphanedIssueTasks(ctx)
+	if err != nil {
+		t.Fatalf("CleanupAllOrphanedIssueTasks: %v", err)
+	}
+	if deleted != 0 {
+		t.Fatalf("deleted=%d, want 0 (never-policy must keep tasks even when issue closes)", deleted)
+	}
+	if len(rec.calls) != 0 {
+		t.Fatalf("DeleteTask calls=%v, want none", rec.calls)
+	}
+}
+
+func TestDeleteIssueWatch_CascadesDedupRows(t *testing.T) {
+	_, _, _, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	watch := &IssueWatch{WorkspaceID: "ws-1", Enabled: true}
+	if err := store.CreateIssueWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateIssueWatch: %v", err)
+	}
+	for _, n := range []int{1, 2, 3} {
+		_, _ = store.ReserveIssueWatchTask(ctx, watch.ID, "acme", "widget", n, fmt.Sprintf("https://example/%d", n))
+		if err := store.AssignIssueWatchTaskID(ctx, watch.ID, "acme", "widget", n, fmt.Sprintf("task-%d", n)); err != nil {
+			t.Fatalf("AssignIssueWatchTaskID: %v", err)
+		}
+	}
+
+	if err := store.DeleteIssueWatch(ctx, watch.ID); err != nil {
+		t.Fatalf("DeleteIssueWatch: %v", err)
+	}
+
+	remaining, err := store.ListIssueWatchTasksByWatch(ctx, watch.ID)
+	if err != nil {
+		t.Fatalf("ListIssueWatchTasksByWatch: %v", err)
+	}
+	if len(remaining) != 0 {
+		t.Fatalf("dedup rows leaked after watch delete: %d remaining", len(remaining))
+	}
+}
+
+func TestDeleteReviewWatch_CascadesDedupRows(t *testing.T) {
+	_, _, _, store := setupPollerTest(t)
+	ctx := context.Background()
+
+	watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true}
+	if err := store.CreateReviewWatch(ctx, watch); err != nil {
+		t.Fatalf("CreateReviewWatch: %v", err)
+	}
+	for _, n := range []int{1, 2, 3} {
+		rpt := &ReviewPRTask{
+			ReviewWatchID: watch.ID,
+			RepoOwner:     "acme",
+			RepoName:      "widget",
+			PRNumber:      n,
+			TaskID:        fmt.Sprintf("task-%d", n),
+		}
+		if err := store.CreateReviewPRTask(ctx, rpt); err != nil {
+			t.Fatalf("CreateReviewPRTask: %v", err)
+		}
+	}
+
+	if err := store.DeleteReviewWatch(ctx, watch.ID); err != nil {
+		t.Fatalf("DeleteReviewWatch: %v", err)
+	}
+
+	remaining, err := store.ListReviewPRTasksByWatch(ctx, watch.ID)
+	if err != nil {
+		t.Fatalf("ListReviewPRTasksByWatch: %v", err)
+	}
+	if len(remaining) != 0 {
+		t.Fatalf("dedup rows leaked after watch delete: %d remaining", len(remaining))
+	}
+}
+
+func TestNormalizeCleanupPolicy(t *testing.T) {
+	if got := NormalizeCleanupPolicy(""); got != CleanupPolicyAuto {
+		t.Fatalf("empty → %q, want auto", got)
+	}
+	if got := NormalizeCleanupPolicy(CleanupPolicyAlways); got != CleanupPolicyAlways {
+		t.Fatalf("always → %q, want always", got)
+	}
+}
+
+func TestIsValidCleanupPolicy(t *testing.T) {
+	for _, ok := range []string{"", CleanupPolicyAuto, CleanupPolicyAlways, CleanupPolicyNever} {
+		if !IsValidCleanupPolicy(ok) {
+			t.Errorf("expected %q valid", ok)
+		}
+	}
+	for _, bad := range []string{"sometimes", "AUTO", "1"} {
+		if IsValidCleanupPolicy(bad) {
+			t.Errorf("expected %q invalid", bad)
+		}
+	}
+}
+
+// switchingFeedbackClient flips between failure and success on each call so
+// tests can simulate "transient outage clears".
+type switchingFeedbackClient struct {
+	NoopClient
+	state string
+	err   error
+}
+
+func (c *switchingFeedbackClient) GetPRFeedback(_ context.Context, _, _ string, _ int) (*PRFeedback, error) {
+	if c.err != nil {
+		return nil, c.err
+	}
+	return &PRFeedback{PR: &PR{State: c.state}}, nil
+}

apps/backend/internal/github/service_issue_test.go

@@ -27,10 +27,19 @@ type stubSessionChecker struct {
 	err error
 }
 
-func (s *stubSessionChecker) HasTaskSessions(_ context.Context, _ string) (bool, error) {
+func (s *stubSessionChecker) HasUserAuthoredMessage(_ context.Context, _ string) (bool, error) {
 	return s.has, s.err
 }
 
+func newCleanupTestService(client Client, log *logger.Logger, checker TaskSessionChecker) *Service {
+	return &Service{
+		client:               client,
+		logger:               log,
+		taskSessionChecker:   checker,
+		cleanupFailureCounts: make(map[string]int),
+	}
+}
+
 func TestShouldDeleteIssueTask(t *testing.T) {
 	log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"})
 
@@ -44,46 +53,51 @@ func TestShouldDeleteIssueTask(t *testing.T) {
 	}
 
 	t.Run("open issue returns false", func(t *testing.T) {
-		svc := &Service{client: &issueStateClient{state: "open"}, logger: log}
-		del, _ := svc.shouldDeleteIssueTask(context.Background(), task)
+		svc := newCleanupTestService(&issueStateClient{state: "open"}, log, nil)
+		del, _ := svc.shouldDeleteIssueTask(context.Background(), task, CleanupPolicyAuto)
 		if del {
 			t.Error("expected false for open issue")
 		}
 	})
 
-	t.Run("closed issue without sessions returns true", func(t *testing.T) {
-		svc := &Service{
-			client:             &issueStateClient{state: "closed"},
-			logger:             log,
-			taskSessionChecker: &stubSessionChecker{has: false},
-		}
-		del, reason := svc.shouldDeleteIssueTask(context.Background(), task)
+	t.Run("closed issue with no user messages returns true", func(t *testing.T) {
+		svc := newCleanupTestService(&issueStateClient{state: "closed"}, log, &stubSessionChecker{has: false})
+		del, reason := svc.shouldDeleteIssueTask(context.Background(), task, CleanupPolicyAuto)
 		if !del {
-			t.Error("expected true for closed issue without sessions")
+			t.Error("expected true for closed issue without user messages")
 		}
 		if reason != "issue_closed" {
 			t.Errorf("reason = %q, want %q", reason, "issue_closed")
 		}
 	})
 
-	t.Run("closed issue with sessions returns false", func(t *testing.T) {
-		svc := &Service{
-			client:             &issueStateClient{state: "closed"},
-			logger:             log,
-			taskSessionChecker: &stubSessionChecker{has: true},
-		}
-		del, _ := svc.shouldDeleteIssueTask(context.Background(), task)
+	t.Run("closed issue with user message returns false under auto", func(t *testing.T) {
+		svc := newCleanupTestService(&issueStateClient{state: "closed"}, log, &stubSessionChecker{has: true})
+		del, _ := svc.shouldDeleteIssueTask(context.Background(), task, CleanupPolicyAuto)
 		if del {
-			t.Error("expected false for closed issue with active sessions")
+			t.Error("expected false for closed issue when user engaged")
+		}
+	})
+
+	t.Run("closed issue under always deletes even with user message", func(t *testing.T) {
+		svc := newCleanupTestService(&issueStateClient{state: "closed"}, log, &stubSessionChecker{has: true})
+		del, _ := svc.shouldDeleteIssueTask(context.Background(), task, CleanupPolicyAlways)
+		if !del {
+			t.Error("expected true for closed issue under CleanupPolicyAlways")
+		}
+	})
+
+	t.Run("closed issue under never keeps task", func(t *testing.T) {
+		svc := newCleanupTestService(&issueStateClient{state: "closed"}, log, &stubSessionChecker{has: false})
+		del, _ := svc.shouldDeleteIssueTask(context.Background(), task, CleanupPolicyNever)
+		if del {
+			t.Error("expected false under CleanupPolicyNever")
 		}
 	})
 
 	t.Run("API error returns false", func(t *testing.T) {
-		svc := &Service{
-			client: &issueStateClient{err: fmt.Errorf("api error")},
-			logger: log,
-		}
-		del, _ := svc.shouldDeleteIssueTask(context.Background(), task)
+		svc := newCleanupTestService(&issueStateClient{err: fmt.Errorf("api error")}, log, nil)
+		del, _ := svc.shouldDeleteIssueTask(context.Background(), task, CleanupPolicyAuto)
 		if del {
 			t.Error("expected false when API returns error")
 		}

apps/backend/internal/github/store.go

@@ -102,6 +102,7 @@ const createTablesSQL = `
 		custom_query TEXT NOT NULL DEFAULT '',
 		enabled BOOLEAN DEFAULT 1,
 		poll_interval_seconds INTEGER DEFAULT 300,
+		cleanup_policy TEXT NOT NULL DEFAULT 'auto',
 		last_polled_at DATETIME,
 		created_at DATETIME NOT NULL,
 		updated_at DATETIME NOT NULL
@@ -132,6 +133,7 @@ const createTablesSQL = `
 		custom_query TEXT NOT NULL DEFAULT '',
 		enabled BOOLEAN DEFAULT 1,
 		poll_interval_seconds INTEGER DEFAULT 300,
+		cleanup_policy TEXT NOT NULL DEFAULT 'auto',
 		last_polled_at DATETIME,
 		created_at DATETIME NOT NULL,
 		updated_at DATETIME NOT NULL
@@ -174,6 +176,12 @@ func (s *Store) initSchema() error {
 	_, _ = s.db.Exec(`ALTER TABLE github_task_prs ADD COLUMN unresolved_review_threads INTEGER DEFAULT 0`)
 	_, _ = s.db.Exec(`ALTER TABLE github_task_prs ADD COLUMN checks_total INTEGER DEFAULT 0`)
 	_, _ = s.db.Exec(`ALTER TABLE github_task_prs ADD COLUMN checks_passing INTEGER DEFAULT 0`)
+	// Per-watch cleanup policy for review/issue watches: controls whether the
+	// poller deletes auto-created tasks when the underlying PR/issue reaches
+	// a terminal state. Values: 'auto' (default — preserve only when user
+	// engaged), 'always' (delete on terminal state), 'never' (manual only).
+	_, _ = s.db.Exec(`ALTER TABLE github_review_watches ADD COLUMN cleanup_policy TEXT NOT NULL DEFAULT 'auto'`)
+	_, _ = s.db.Exec(`ALTER TABLE github_issue_watches ADD COLUMN cleanup_policy TEXT NOT NULL DEFAULT 'auto'`)
 	if err := s.migratePRTablesForMultiRepo(); err != nil {
 		return fmt.Errorf("migrate PR tables for multi-repo: %w", err)
 	}
@@ -771,6 +779,7 @@ func (s *Store) CreateReviewWatch(ctx context.Context, rw *ReviewWatch) error {
 	now := time.Now().UTC()
 	rw.CreatedAt = now
 	rw.UpdatedAt = now
+	rw.CleanupPolicy = NormalizeCleanupPolicy(rw.CleanupPolicy)
 	reposJSON, err := json.Marshal(rw.Repos)
 	if err != nil {
 		return fmt.Errorf("marshal repos: %w", err)
@@ -779,15 +788,17 @@ func (s *Store) CreateReviewWatch(ctx context.Context, rw *ReviewWatch) error {
 	_, err = s.db.ExecContext(ctx, `
 		INSERT INTO github_review_watches (id, workspace_id, workflow_id, workflow_step_id, repos,
 			agent_profile_id, executor_profile_id, prompt, review_scope, custom_query,
-			enabled, poll_interval_seconds, created_at, updated_at)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+			enabled, poll_interval_seconds, cleanup_policy, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		rw.ID, rw.WorkspaceID, rw.WorkflowID, rw.WorkflowStepID, rw.ReposJSON,
 		rw.AgentProfileID, rw.ExecutorProfileID, rw.Prompt, rw.ReviewScope, rw.CustomQuery,
-		rw.Enabled, rw.PollIntervalSeconds, rw.CreatedAt, rw.UpdatedAt)
+		rw.Enabled, rw.PollIntervalSeconds, rw.CleanupPolicy, rw.CreatedAt, rw.UpdatedAt)
 	return err
 }
 
-// hydrateReviewWatchRepos unmarshals the ReposJSON field into the Repos slice.
+// hydrateReviewWatchRepos unmarshals the ReposJSON field into the Repos slice
+// and normalizes the cleanup policy so legacy rows (or zero values) surface
+// as the documented default.
 func hydrateReviewWatchRepos(rw *ReviewWatch) {
 	if rw.ReposJSON != "" {
 		if err := json.Unmarshal([]byte(rw.ReposJSON), &rw.Repos); err != nil {
@@ -798,6 +809,7 @@ func hydrateReviewWatchRepos(rw *ReviewWatch) {
 	if rw.Repos == nil {
 		rw.Repos = []RepoFilter{}
 	}
+	rw.CleanupPolicy = NormalizeCleanupPolicy(rw.CleanupPolicy)
 }
 
 // GetReviewWatch returns a review watch by ID.
@@ -860,6 +872,7 @@ func (s *Store) ListEnabledReviewWatches(ctx context.Context) ([]*ReviewWatch, e
 // UpdateReviewWatch updates a review watch.
 func (s *Store) UpdateReviewWatch(ctx context.Context, rw *ReviewWatch) error {
 	rw.UpdatedAt = time.Now().UTC()
+	rw.CleanupPolicy = NormalizeCleanupPolicy(rw.CleanupPolicy)
 	reposJSON, err := json.Marshal(rw.Repos)
 	if err != nil {
 		return fmt.Errorf("marshal repos: %w", err)
@@ -869,19 +882,33 @@ func (s *Store) UpdateReviewWatch(ctx context.Context, rw *ReviewWatch) error {
 		UPDATE github_review_watches SET workflow_id = ?, workflow_step_id = ?, repos = ?,
 			agent_profile_id = ?, executor_profile_id = ?,
 			prompt = ?, review_scope = ?, custom_query = ?,
-			enabled = ?, poll_interval_seconds = ?, last_polled_at = ?, updated_at = ?
+			enabled = ?, poll_interval_seconds = ?, cleanup_policy = ?, last_polled_at = ?, updated_at = ?
 		WHERE id = ?`,
 		rw.WorkflowID, rw.WorkflowStepID, rw.ReposJSON,
 		rw.AgentProfileID, rw.ExecutorProfileID,
 		rw.Prompt, rw.ReviewScope, rw.CustomQuery,
-		rw.Enabled, rw.PollIntervalSeconds, rw.LastPolledAt, rw.UpdatedAt, rw.ID)
+		rw.Enabled, rw.PollIntervalSeconds, rw.CleanupPolicy, rw.LastPolledAt, rw.UpdatedAt, rw.ID)
 	return err
 }
 
-// DeleteReviewWatch deletes a review watch.
+// DeleteReviewWatch deletes a review watch and all its associated dedup rows
+// in one transaction. Dedup rows have no foreign key (SQLite never enforced
+// one for this table), so the explicit cascade is required — otherwise the
+// rows survive after the watch is gone, become invisible to the per-watch
+// poller, and the tasks they reference leak forever.
 func (s *Store) DeleteReviewWatch(ctx context.Context, id string) error {
-	_, err := s.db.ExecContext(ctx, `DELETE FROM github_review_watches WHERE id = ?`, id)
-	return err
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback() }()
+	if _, err := tx.ExecContext(ctx, `DELETE FROM github_review_pr_tasks WHERE review_watch_id = ?`, id); err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `DELETE FROM github_review_watches WHERE id = ?`, id); err != nil {
+		return err
+	}
+	return tx.Commit()
 }
 
 // --- Review PR Task deduplication ---
@@ -972,6 +999,17 @@ func (s *Store) ListReviewPRTasksByWatch(ctx context.Context, watchID string) ([
 	return tasks, err
 }
 
+// ListAllReviewPRTasks lists every dedup record across all watches. Used by
+// the global cleanup sweep so orphaned rows (whose watch was deleted or
+// disabled) still get evaluated for terminal-state cleanup.
+func (s *Store) ListAllReviewPRTasks(ctx context.Context) ([]*ReviewPRTask, error) {
+	var tasks []*ReviewPRTask
+	err := s.ro.SelectContext(ctx, &tasks,
+		`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
+}
+
 // DeleteReviewPRTask deletes a dedup record by ID.
 func (s *Store) DeleteReviewPRTask(ctx context.Context, id string) error {
 	_, err := s.db.ExecContext(ctx, `DELETE FROM github_review_pr_tasks WHERE id = ?`, id)
@@ -1090,7 +1128,9 @@ func (s *Store) fetchApprovalRate(ctx context.Context, q *prStatsQuery, stats *P
 
 // --- Issue Watch operations ---
 
-// hydrateIssueWatch unmarshals JSON fields into their Go slices.
+// hydrateIssueWatch unmarshals JSON fields into their Go slices and
+// normalizes the cleanup policy so legacy rows (or zero values) surface as
+// the documented default.
 func hydrateIssueWatch(iw *IssueWatch) {
 	if iw.ReposJSON != "" {
 		if err := json.Unmarshal([]byte(iw.ReposJSON), &iw.Repos); err != nil {
@@ -1108,6 +1148,7 @@ func hydrateIssueWatch(iw *IssueWatch) {
 	if iw.Labels == nil {
 		iw.Labels = []string{}
 	}
+	iw.CleanupPolicy = NormalizeCleanupPolicy(iw.CleanupPolicy)
 }
 
 // CreateIssueWatch creates a new issue watch configuration.
@@ -1118,6 +1159,7 @@ func (s *Store) CreateIssueWatch(ctx context.Context, iw *IssueWatch) error {
 	now := time.Now().UTC()
 	iw.CreatedAt = now
 	iw.UpdatedAt = now
+	iw.CleanupPolicy = NormalizeCleanupPolicy(iw.CleanupPolicy)
 	reposJSON, err := json.Marshal(iw.Repos)
 	if err != nil {
 		return fmt.Errorf("marshal repos: %w", err)
@@ -1131,11 +1173,11 @@ func (s *Store) CreateIssueWatch(ctx context.Context, iw *IssueWatch) error {
 	_, err = s.db.ExecContext(ctx, `
 		INSERT INTO github_issue_watches (id, workspace_id, workflow_id, workflow_step_id, repos,
 			agent_profile_id, executor_profile_id, prompt, labels, custom_query,
-			enabled, poll_interval_seconds, created_at, updated_at)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+			enabled, poll_interval_seconds, cleanup_policy, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		iw.ID, iw.WorkspaceID, iw.WorkflowID, iw.WorkflowStepID, iw.ReposJSON,
 		iw.AgentProfileID, iw.ExecutorProfileID, iw.Prompt, iw.LabelsJSON, iw.CustomQuery,
-		iw.Enabled, iw.PollIntervalSeconds, iw.CreatedAt, iw.UpdatedAt)
+		iw.Enabled, iw.PollIntervalSeconds, iw.CleanupPolicy, iw.CreatedAt, iw.UpdatedAt)
 	return err
 }
 
@@ -1198,6 +1240,7 @@ func (s *Store) ListEnabledIssueWatches(ctx context.Context) ([]*IssueWatch, err
 // UpdateIssueWatch updates an issue watch.
 func (s *Store) UpdateIssueWatch(ctx context.Context, iw *IssueWatch) error {
 	iw.UpdatedAt = time.Now().UTC()
+	iw.CleanupPolicy = NormalizeCleanupPolicy(iw.CleanupPolicy)
 	reposJSON, err := json.Marshal(iw.Repos)
 	if err != nil {
 		return fmt.Errorf("marshal repos: %w", err)
@@ -1212,12 +1255,12 @@ func (s *Store) UpdateIssueWatch(ctx context.Context, iw *IssueWatch) error {
 		UPDATE github_issue_watches SET workflow_id = ?, workflow_step_id = ?, repos = ?,
 			agent_profile_id = ?, executor_profile_id = ?,
 			prompt = ?, labels = ?, custom_query = ?,
-			enabled = ?, poll_interval_seconds = ?, last_polled_at = ?, updated_at = ?
+			enabled = ?, poll_interval_seconds = ?, cleanup_policy = ?, last_polled_at = ?, updated_at = ?
 		WHERE id = ?`,
 		iw.WorkflowID, iw.WorkflowStepID, iw.ReposJSON,
 		iw.AgentProfileID, iw.ExecutorProfileID,
 		iw.Prompt, iw.LabelsJSON, iw.CustomQuery,
-		iw.Enabled, iw.PollIntervalSeconds, iw.LastPolledAt, iw.UpdatedAt, iw.ID)
+		iw.Enabled, iw.PollIntervalSeconds, iw.CleanupPolicy, iw.LastPolledAt, iw.UpdatedAt, iw.ID)
 	return err
 }
 
@@ -1302,6 +1345,16 @@ func (s *Store) ListIssueWatchTasksByWatch(ctx context.Context, watchID string)
 	return tasks, err
 }
 
+// ListAllIssueWatchTasks lists every dedup record across all watches. Used by
+// the global cleanup sweep so orphaned rows still get evaluated.
+func (s *Store) ListAllIssueWatchTasks(ctx context.Context) ([]*IssueWatchTask, error) {
+	var tasks []*IssueWatchTask
+	err := s.ro.SelectContext(ctx, &tasks,
+		`SELECT id, issue_watch_id, repo_owner, repo_name, issue_number, issue_url, task_id, created_at
+		 FROM github_issue_watch_tasks`)
+	return tasks, err
+}
+
 // DeleteIssueWatchTask deletes a dedup record by ID.
 func (s *Store) DeleteIssueWatchTask(ctx context.Context, id string) error {
 	_, err := s.db.ExecContext(ctx, `DELETE FROM github_issue_watch_tasks WHERE id = ?`, id)