diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index 31e7b3da28..b279d4efa6 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -500,8 +500,8 @@ func (w *orchestratorWrapper) ResumeTaskSession(ctx context.Context, taskID, tas } // StartCreatedSession forwards to the orchestrator service, discarding the TaskExecution result. -func (w *orchestratorWrapper) StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode bool, attachments []v1.MessageAttachment) error { - _, err := w.svc.StartCreatedSession(ctx, taskID, sessionID, agentProfileID, prompt, skipMessageRecord, planMode, attachments) +func (w *orchestratorWrapper) StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode, autoStart bool, attachments []v1.MessageAttachment) error { + _, err := w.svc.StartCreatedSession(ctx, taskID, sessionID, agentProfileID, prompt, skipMessageRecord, planMode, autoStart, attachments) return err } diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 59b275f8e7..f39e50cbfd 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -874,7 +874,7 @@ func (a *schedulerTaskStarterAdapter) StartTask( ) error { _, err := a.orch.StartTask(ctx, taskID, agentProfileID, executorID, executorProfileID, priority, prompt, - workflowStepID, planMode, nil) + workflowStepID, planMode, false, nil) return err } @@ -1271,7 +1271,7 @@ func newOfficeTaskStarter(orchestratorSvc *orchestrator.Service) officeservice.T planMode bool, attachments []v1.MessageAttachment, env map[string]string) error { _, err := orchestratorSvc.StartTaskWithEnv(ctx, taskID, agentProfileID, executorID, executorProfileID, priority, prompt, - workflowStepID, planMode, attachments, env) + workflowStepID, planMode, false, attachments, env) return err }, ) diff --git a/apps/backend/cmd/kandev/turn_adapters.go b/apps/backend/cmd/kandev/turn_adapters.go index a5fa086156..51421563fb 100644 --- a/apps/backend/cmd/kandev/turn_adapters.go +++ b/apps/backend/cmd/kandev/turn_adapters.go @@ -36,13 +36,47 @@ func newTurnServiceAdapter(svc *taskservice.Service) *turnServiceAdapter { type taskSessionCheckerAdapter struct { repo interface { ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) + ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) } } -func (a *taskSessionCheckerAdapter) HasTaskSessions(ctx context.Context, taskID string) (bool, error) { +// HasUserAuthoredMessage reports whether the user has authored any message +// on this task that wasn't created by an automated trigger (workflow +// auto-start, PR/issue watch, Jira/Linear integration). Auto-start messages +// are tagged with metadata.auto_start = true; the check ignores them so a +// task whose only "user" message is the agent's auto-injected prompt counts +// as untouched and is eligible for cleanup when its PR/issue merges. +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 len(sessions) > 0, nil + for _, sess := range sessions { + messages, err := a.repo.ListMessages(ctx, sess.ID) + if err != nil { + return false, err + } + for _, m := range messages { + if m.AuthorType != models.MessageAuthorUser { + continue + } + // New code paths tag both auto_start and workflow_auto_start. + // Legacy rows (pre-cleanup-policy upgrade) carry only the + // workflow_auto_start tag from the old recordAutoStartMessage + // implementation — recognize it too so the install-wide + // cleanup button actually drains piled-up tasks after upgrade. + if metaFlag(m.Metadata, "auto_start") || metaFlag(m.Metadata, "workflow_auto_start") { + continue + } + return true, nil + } + } + return false, nil +} + +// 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 { + v, ok := meta[key].(bool) + return ok && v } diff --git a/apps/backend/cmd/kandev/turn_adapters_test.go b/apps/backend/cmd/kandev/turn_adapters_test.go new file mode 100644 index 0000000000..f39abf0eaa --- /dev/null +++ b/apps/backend/cmd/kandev/turn_adapters_test.go @@ -0,0 +1,147 @@ +package main + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/task/models" +) + +// 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 +} + +func (r *fakeSessionMsgRepo) ListTaskSessions(_ context.Context, taskID string) ([]*models.TaskSession, error) { + return r.sessions[taskID], nil +} + +func (r *fakeSessionMsgRepo) ListMessages(_ context.Context, sessionID string) ([]*models.Message, error) { + return r.messages[sessionID], nil +} + +func TestHasUserAuthoredMessage(t *testing.T) { + cases := []struct { + name string + messages []*models.Message + want bool + }{ + { + name: "agent-only messages → false", + messages: []*models.Message{ + {AuthorType: models.MessageAuthorAgent, Content: "hi"}, + }, + want: false, + }, + { + name: "user message tagged auto_start=true → false (PR/issue/Jira/Linear watch path)", + messages: []*models.Message{ + { + AuthorType: models.MessageAuthorUser, + Content: "Review this PR...", + Metadata: map[string]interface{}{"auto_start": true}, + }, + }, + want: false, + }, + { + // Regression for claude blocker on b89f880c: recordAutoStartMessage + // (workflow auto-start path) used to tag only workflow_auto_start + // without also setting auto_start, so messages slipped past + // HasUserAuthoredMessage and made the task look user-authored. + // The fix sets both flags; assert this filter recognizes either. + name: "workflow auto-start (auto_start=true, workflow_auto_start=true) → false", + messages: []*models.Message{ + { + AuthorType: models.MessageAuthorUser, + Content: "{{task_prompt}}\nDo the thing", + Metadata: map[string]interface{}{ + "auto_start": true, + "workflow_auto_start": true, + "plan_mode": true, + }, + }, + }, + want: false, + }, + { + // Legacy message from before the cleanup_policy work: + // recordAutoStartMessage only tagged workflow_auto_start. The + // filter must recognize that tag too so installs with piled-up + // pre-upgrade tasks can drain them via the manual button. + name: "legacy workflow auto-start (only workflow_auto_start=true) → false", + messages: []*models.Message{ + { + AuthorType: models.MessageAuthorUser, + Content: "legacy auto-start prompt", + Metadata: map[string]interface{}{"workflow_auto_start": true}, + }, + }, + want: false, + }, + { + name: "real user message (no auto_start tag) → true", + messages: []*models.Message{ + { + AuthorType: models.MessageAuthorUser, + Content: "looks good, ship it", + }, + }, + want: true, + }, + { + name: "auto-start message + later real user message → true", + messages: []*models.Message{ + { + AuthorType: models.MessageAuthorUser, + Content: "Review this PR...", + Metadata: map[string]interface{}{"auto_start": true}, + }, + { + AuthorType: models.MessageAuthorAgent, + Content: "Looking at the diff", + }, + { + AuthorType: models.MessageAuthorUser, + Content: "thanks", + }, + }, + want: true, + }, + { + name: "auto_start metadata value of false → still counted as user", + messages: []*models.Message{ + { + AuthorType: models.MessageAuthorUser, + Content: "real input", + Metadata: map[string]interface{}{"auto_start": false}, + }, + }, + want: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + adapter := &taskSessionCheckerAdapter{ + repo: &fakeSessionMsgRepo{ + sessions: map[string][]*models.TaskSession{ + "task-1": {{ID: "sess-1"}}, + }, + messages: map[string][]*models.Message{ + "sess-1": tc.messages, + }, + }, + } + got, err := adapter.HasUserAuthoredMessage(context.Background(), "task-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != tc.want { + t.Errorf("HasUserAuthoredMessage = %v, want %v", got, tc.want) + } + }) + } +} diff --git a/apps/backend/internal/github/controller.go b/apps/backend/internal/github/controller.go index feda28c5e5..9ecc96c01b 100644 --- a/apps/backend/internal/github/controller.go +++ b/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}) +} diff --git a/apps/backend/internal/github/handlers.go b/apps/backend/internal/github/handlers.go index 27b6d71c6b..7a73158323 100644 --- a/apps/backend/internal/github/handlers.go +++ b/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}) } } diff --git a/apps/backend/internal/github/models.go b/apps/backend/internal/github/models.go index 5b86740698..c90f2c40c5 100644 --- a/apps/backend/internal/github/models.go +++ b/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) --- diff --git a/apps/backend/internal/github/poller.go b/apps/backend/internal/github/poller.go index 2de082c1ee..eed9f1f88f 100644 --- a/apps/backend/internal/github/poller.go +++ b/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)) + } } diff --git a/apps/backend/internal/github/review_pr_task_test.go b/apps/backend/internal/github/review_pr_task_test.go index 1403a4a19d..61cf4f6275 100644 --- a/apps/backend/internal/github/review_pr_task_test.go +++ b/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) diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index d4c29fb36c..5954b957b1 100644 --- a/apps/backend/internal/github/service.go +++ b/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 diff --git a/apps/backend/internal/github/service_cleanup_policy_test.go b/apps/backend/internal/github/service_cleanup_policy_test.go new file mode 100644 index 0000000000..1f7e06cce5 --- /dev/null +++ b/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 +} diff --git a/apps/backend/internal/github/service_issue_test.go b/apps/backend/internal/github/service_issue_test.go index cfa89bf9b7..0e8b40860c 100644 --- a/apps/backend/internal/github/service_issue_test.go +++ b/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") } diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index 8991bf4afa..52b9839f2c 100644 --- a/apps/backend/internal/github/store.go +++ b/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) diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index f6342862df..8ed58b87b7 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -63,7 +63,7 @@ type EventBus interface { type SessionLauncher interface { LaunchSession(ctx context.Context, req *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) PromptTask(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool, attachments []v1.MessageAttachment, dispatchOnly bool) (*orchestrator.PromptResult, error) - StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode bool, attachments []v1.MessageAttachment) (*executor.TaskExecution, error) + StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode, autoStart bool, attachments []v1.MessageAttachment) (*executor.TaskExecution, error) ResumeTaskSession(ctx context.Context, taskID, sessionID string) (*executor.TaskExecution, error) GetMessageQueue() *messagequeue.Service } @@ -1014,7 +1014,7 @@ func (h *Handlers) dispatchTaskMessage(ctx context.Context, taskID string, sessi // StartCreatedSession fails (the previous order wrote the row up-front // regardless of launch outcome). skipMessageRecord=true keeps // postLaunchCreated from writing its own duplicate row. - if _, err := h.sessionLauncher.StartCreatedSession(ctx, taskID, session.ID, session.AgentProfileID, prompt, true, false, nil); err != nil { + if _, err := h.sessionLauncher.StartCreatedSession(ctx, taskID, session.ID, session.AgentProfileID, prompt, true, false, true, nil); err != nil { return "", fmt.Errorf("failed to start session: %w", err) } h.recordUserMessage(ctx, taskID, session.ID, prompt, metadata) diff --git a/apps/backend/internal/mcp/handlers/handlers_test.go b/apps/backend/internal/mcp/handlers/handlers_test.go index af3a9cad60..e901f42e4e 100644 --- a/apps/backend/internal/mcp/handlers/handlers_test.go +++ b/apps/backend/internal/mcp/handlers/handlers_test.go @@ -159,7 +159,7 @@ func (m *mockSessionLauncher) getRequest() *orchestrator.LaunchSessionRequest { func (m *mockSessionLauncher) PromptTask(context.Context, string, string, string, string, bool, []v1.MessageAttachment, bool) (*orchestrator.PromptResult, error) { return nil, nil } -func (m *mockSessionLauncher) StartCreatedSession(context.Context, string, string, string, string, bool, bool, []v1.MessageAttachment) (*executor.TaskExecution, error) { +func (m *mockSessionLauncher) StartCreatedSession(context.Context, string, string, string, string, bool, bool, bool, []v1.MessageAttachment) (*executor.TaskExecution, error) { return nil, nil } func (m *mockSessionLauncher) ResumeTaskSession(context.Context, string, string) (*executor.TaskExecution, error) { diff --git a/apps/backend/internal/mcp/handlers/message_task_test.go b/apps/backend/internal/mcp/handlers/message_task_test.go index 657e5a84d0..9957b2329c 100644 --- a/apps/backend/internal/mcp/handlers/message_task_test.go +++ b/apps/backend/internal/mcp/handlers/message_task_test.go @@ -60,7 +60,7 @@ func (f *fakeOrchestrator) PromptTask(_ context.Context, taskID, sessionID, prom return &orchestrator.PromptResult{}, nil } -func (f *fakeOrchestrator) StartCreatedSession(_ context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, _ bool, _ []v1.MessageAttachment) (*executor.TaskExecution, error) { +func (f *fakeOrchestrator) StartCreatedSession(_ context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, _, _ bool, _ []v1.MessageAttachment) (*executor.TaskExecution, error) { f.mu.Lock() defer f.mu.Unlock() f.startCreatedCalls = append(f.startCreatedCalls, startCreatedCall{ diff --git a/apps/backend/internal/orchestrator/event_handlers_github.go b/apps/backend/internal/orchestrator/event_handlers_github.go index 52eeaf49d9..5b982a4e15 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github.go +++ b/apps/backend/internal/orchestrator/event_handlers_github.go @@ -323,6 +323,7 @@ func (s *Service) autoStartReviewTask( task.Description, evt.WorkflowStepID, false, + true, nil, ) if err != nil { @@ -1132,6 +1133,7 @@ func (s *Service) autoStartIssueTask( task.Description, evt.WorkflowStepID, false, + true, nil, ) if err != nil { diff --git a/apps/backend/internal/orchestrator/event_handlers_jira.go b/apps/backend/internal/orchestrator/event_handlers_jira.go index 9223d04fb0..e3f3b92b09 100644 --- a/apps/backend/internal/orchestrator/event_handlers_jira.go +++ b/apps/backend/internal/orchestrator/event_handlers_jira.go @@ -109,7 +109,7 @@ func (s *Service) createJiraIssueTask(ctx context.Context, evt *jira.NewJiraIssu } if _, err := s.StartTask( ctx, task.ID, evt.AgentProfileID, "", evt.ExecutorProfileID, - "", task.Description, evt.WorkflowStepID, false, nil, + "", task.Description, evt.WorkflowStepID, false, true, nil, ); err != nil { s.logger.Error("failed to auto-start jira issue task", zap.String("task_id", task.ID), zap.Error(err)) diff --git a/apps/backend/internal/orchestrator/event_handlers_linear.go b/apps/backend/internal/orchestrator/event_handlers_linear.go index 5e227b9548..0ae5803ea5 100644 --- a/apps/backend/internal/orchestrator/event_handlers_linear.go +++ b/apps/backend/internal/orchestrator/event_handlers_linear.go @@ -112,7 +112,7 @@ func (s *Service) createLinearIssueTask(ctx context.Context, evt *linear.NewLine } if _, err := s.StartTask( ctx, task.ID, evt.AgentProfileID, "", evt.ExecutorProfileID, - "", task.Description, evt.WorkflowStepID, false, nil, + "", task.Description, evt.WorkflowStepID, false, true, nil, ); err != nil { s.logger.Error("failed to auto-start linear issue task", zap.String("task_id", task.ID), zap.Error(err)) diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 42991933a5..6fad53016e 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -349,7 +349,7 @@ func (s *Service) handleTaskMovedNoSession(ctx context.Context, data watcher.Tas // Async: event bus delivers synchronously; blocking here → HTTP timeout (see handleTaskMovedWithSession doc). go func() { asyncCtx := context.WithoutCancel(ctx) - _, err := s.StartTask(asyncCtx, task.ID, agentProfileID, "", executorProfileID, "", task.Description, data.ToStepID, planMode, nil) + _, err := s.StartTask(asyncCtx, task.ID, agentProfileID, "", executorProfileID, "", task.Description, data.ToStepID, planMode, true, nil) if err != nil { s.logger.Error("task.moved: failed to auto-start task", zap.String("task_id", data.TaskID), @@ -1151,7 +1151,7 @@ func (s *Service) autoStartStepPrompt( zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.String("step_name", stepName)) - _, err := s.StartCreatedSession(ctx, taskID, sessionID, session.AgentProfileID, recordedPrompt, true, planMode, attachments) + _, err := s.StartCreatedSession(ctx, taskID, sessionID, session.AgentProfileID, recordedPrompt, true, planMode, true, attachments) if err != nil { requeueTaken() } @@ -1265,7 +1265,7 @@ func (s *Service) fallbackFreshLaunchOnMissingExecution( return err } - if _, err := s.StartCreatedSession(ctx, taskID, sessionID, fresh.AgentProfileID, prompt, true, planMode, attachments); err != nil { + if _, err := s.StartCreatedSession(ctx, taskID, sessionID, fresh.AgentProfileID, prompt, true, planMode, true, attachments); err != nil { s.logger.Error("auto-start fallback: fresh launch failed", zap.String("session_id", sessionID), zap.Error(err)) requeue() @@ -1319,7 +1319,14 @@ func (s *Service) recordAutoStartMessage(ctx context.Context, taskID, sessionID, s.startTurnForSession(ctx, sessionID) turnID = s.getActiveTurnID(sessionID) } - meta := NewUserMessageMeta().WithPlanMode(planMode) + // auto_start tags this seed prompt as automation-originated so the + // github cleanup filter (HasUserAuthoredMessage) skips it — without + // this tag, a workflow auto-start fired on a PR-watch task makes the + // task look user-authored and the cleanup loop preserves it on merge, + // re-creating the exact pileup the cleanup_policy work fixes. + // workflow_auto_start is the original tag this function set; preserved + // for any consumer reading it directly. + meta := NewUserMessageMeta().WithPlanMode(planMode).WithAutoStart(true) metaMap := meta.ToMap() if metaMap == nil { metaMap = make(map[string]interface{}) diff --git a/apps/backend/internal/orchestrator/message_meta.go b/apps/backend/internal/orchestrator/message_meta.go index dffaa38d08..416d572057 100644 --- a/apps/backend/internal/orchestrator/message_meta.go +++ b/apps/backend/internal/orchestrator/message_meta.go @@ -7,6 +7,7 @@ import v1 "github.com/kandev/kandev/pkg/api/v1" type UserMessageMeta struct { PlanMode bool HasReviewComments bool + AutoStart bool Attachments []v1.MessageAttachment ContextFiles []v1.ContextFileMeta SenderTaskID string @@ -31,6 +32,15 @@ func (m *UserMessageMeta) WithReviewComments(has bool) *UserMessageMeta { return m } +// WithAutoStart marks the message as having been created by an automated +// trigger (workflow auto-start, PR/issue watch, Jira/Linear integration) +// rather than direct user input. Used by the GitHub cleanup logic to +// distinguish "the agent ran on its own" from "the user actually engaged". +func (m *UserMessageMeta) WithAutoStart(enabled bool) *UserMessageMeta { + m.AutoStart = enabled + return m +} + // WithAttachments sets the message attachments. func (m *UserMessageMeta) WithAttachments(attachments []v1.MessageAttachment) *UserMessageMeta { m.Attachments = attachments @@ -56,7 +66,7 @@ func (m *UserMessageMeta) WithSenderTask(taskID, taskTitle, sessionID string) *U // ToMap returns the metadata as a map suitable for message creation. // Returns nil if no metadata fields are set. func (m *UserMessageMeta) ToMap() map[string]interface{} { - if !m.PlanMode && !m.HasReviewComments && len(m.Attachments) == 0 && len(m.ContextFiles) == 0 && m.SenderTaskID == "" { + if !m.PlanMode && !m.HasReviewComments && !m.AutoStart && len(m.Attachments) == 0 && len(m.ContextFiles) == 0 && m.SenderTaskID == "" { return nil } meta := make(map[string]interface{}) @@ -66,6 +76,9 @@ func (m *UserMessageMeta) ToMap() map[string]interface{} { if m.HasReviewComments { meta["has_review_comments"] = true } + if m.AutoStart { + meta["auto_start"] = true + } if len(m.Attachments) > 0 { meta["attachments"] = m.Attachments } diff --git a/apps/backend/internal/orchestrator/session_launch.go b/apps/backend/internal/orchestrator/session_launch.go index cb22378b37..1815deb9a5 100644 --- a/apps/backend/internal/orchestrator/session_launch.go +++ b/apps/backend/internal/orchestrator/session_launch.go @@ -148,7 +148,7 @@ func (s *Service) launchStart(ctx context.Context, req *LaunchSessionRequest) (* execution, err := s.StartTask( ctx, req.TaskID, req.AgentProfileID, req.ExecutorID, req.ExecutorProfileID, req.Priority, req.Prompt, - req.WorkflowStepID, req.PlanMode, req.Attachments, + req.WorkflowStepID, req.PlanMode, req.AutoStart, req.Attachments, ) if err != nil { return nil, err @@ -190,7 +190,7 @@ func (s *Service) shouldBlockAutoStart(ctx context.Context, req *LaunchSessionRe func (s *Service) launchStartCreated(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { execution, err := s.StartCreatedSession( ctx, req.TaskID, req.SessionID, req.AgentProfileID, - req.Prompt, req.SkipMessageRecord, req.PlanMode, req.Attachments, + req.Prompt, req.SkipMessageRecord, req.PlanMode, req.AutoStart, req.Attachments, ) if err != nil { return nil, err diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 7c934e5df5..7c605b416f 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -216,7 +216,12 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr // and the user now wants to start the agent with a prompt (e.g., from the plan panel or chat). // When skipMessageRecord is true, only the session state is updated (the caller already stored the user message). // When planMode is true, plan mode instructions are injected into the prompt and session metadata is set. -func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode bool, attachments []v1.MessageAttachment) (*executor.TaskExecution, error) { +// autoStart marks the launch as having been triggered by an automated path +// (only consumed when skipMessageRecord is false — callers that store their +// own message control its metadata directly). +// +//nolint:cyclop,funlen // existing complexity inherited from main's session-lifecycle handling; signature touched here only to thread autoStart +func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode, autoStart bool, attachments []v1.MessageAttachment) (*executor.TaskExecution, error) { s.logger.Debug("starting created session", zap.String("task_id", taskID), zap.String("session_id", sessionID), @@ -365,7 +370,7 @@ func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, ag // Record the initial user message and set plan mode metadata after launch. // Note: we do NOT set session state here — the executor sets it to STARTING, // and event handlers (handleAgentReady) transition it to WAITING_FOR_INPUT. - s.postLaunchCreated(ctx, taskID, sessionID, effectivePrompt, skipMessageRecord, planModeActive, attachments) + s.postLaunchCreated(ctx, taskID, sessionID, effectivePrompt, skipMessageRecord, planModeActive, autoStart, attachments) // Ensure a PR watch exists so the poller can detect PRs created by the agent. // PrepareTaskSession may have already created one, but if that goroutine failed @@ -379,9 +384,13 @@ func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, ag // records the initial user message (unless skipped) and sets plan mode metadata. // It does NOT modify session state — the executor sets STARTING, and event handlers // (handleAgentReady) handle the transition to WAITING_FOR_INPUT. -func (s *Service) postLaunchCreated(ctx context.Context, taskID, sessionID, prompt string, skipMessage, planModeActive bool, attachments []v1.MessageAttachment) { +// autoStart marks the message as having been created by an automated trigger +// (preserved through to recordInitialMessage's auto_start metadata tag); the +// flag is only consumed when skipMessage is false (callers that store their +// own message control its metadata directly). +func (s *Service) postLaunchCreated(ctx context.Context, taskID, sessionID, prompt string, skipMessage, planModeActive, autoStart bool, attachments []v1.MessageAttachment) { if !skipMessage { - s.recordInitialMessage(ctx, taskID, sessionID, prompt, planModeActive, attachments) + s.recordInitialMessage(ctx, taskID, sessionID, prompt, planModeActive, autoStart, attachments) } if planModeActive { @@ -398,14 +407,18 @@ func (s *Service) postLaunchCreated(ctx context.Context, taskID, sessionID, prom // applied if the step has plan_mode enabled. // If planMode is true and the workflow step doesn't already apply plan mode, // default plan mode instructions are injected into the prompt. -func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority string, prompt string, workflowStepID string, planMode bool, attachments []v1.MessageAttachment) (*executor.TaskExecution, error) { - return s.startTask(ctx, taskID, agentProfileID, executorID, executorProfileID, priority, prompt, workflowStepID, planMode, attachments, nil, nil) +// autoStart marks the launch as having been triggered by an automated path +// (PR/issue/Jira/Linear watch, workflow auto-start) rather than direct user +// input — the seed prompt is tagged so the github cleanup loop can tell +// "agent ran on its own" from "user actually engaged". +func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority string, prompt string, workflowStepID string, planMode, autoStart bool, attachments []v1.MessageAttachment) (*executor.TaskExecution, error) { + return s.startTask(ctx, taskID, agentProfileID, executorID, executorProfileID, priority, prompt, workflowStepID, planMode, autoStart, attachments, nil, nil) } // StartTaskWithEnv starts a task and carries launch-scoped environment variables // through to the agent runtime. Existing StartTask callers keep the old behavior. -func (s *Service) StartTaskWithEnv(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority string, prompt string, workflowStepID string, planMode bool, attachments []v1.MessageAttachment, env map[string]string) (*executor.TaskExecution, error) { - return s.startTask(ctx, taskID, agentProfileID, executorID, executorProfileID, priority, prompt, workflowStepID, planMode, attachments, env, nil) +func (s *Service) StartTaskWithEnv(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority string, prompt string, workflowStepID string, planMode, autoStart bool, attachments []v1.MessageAttachment, env map[string]string) (*executor.TaskExecution, error) { + return s.startTask(ctx, taskID, agentProfileID, executorID, executorProfileID, priority, prompt, workflowStepID, planMode, autoStart, attachments, env, nil) } // StartTaskWithRoute launches a task with a fully resolved provider @@ -425,6 +438,9 @@ func (s *Service) StartTaskWithEnv(ctx context.Context, taskID string, agentProf // KANDEV_*); route.Env carries provider-scoped overrides. The route // env wins on key collisions because per-provider env is the more // specific authority. +// +// Office-routed launches use autoStart=false: the user kicked off the +// task; the office layer only chose the provider. func (s *Service) StartTaskWithRoute( ctx context.Context, taskID, agentProfileID string, launch executor.LaunchContext, route executor.RouteOverride, @@ -432,7 +448,7 @@ func (s *Service) StartTaskWithRoute( merged := mergeRouteEnv(launch.Env, route.Env) _, err := s.startTask(ctx, taskID, agentProfileID, launch.ExecutorID, launch.ExecutorProfileID, launch.Priority, - launch.Prompt, launch.WorkflowStepID, launch.PlanMode, + launch.Prompt, launch.WorkflowStepID, launch.PlanMode, false, launch.Attachments, merged, &route) return err } @@ -457,7 +473,8 @@ func mergeRouteEnv(launchEnv, routeEnv map[string]string) map[string]string { return out } -func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority string, prompt string, workflowStepID string, planMode bool, attachments []v1.MessageAttachment, env map[string]string, route *executor.RouteOverride) (*executor.TaskExecution, error) { +//nolint:cyclop,funlen // launch path threads many orthogonal concerns (workflow-step / agent-profile / office-task / config-mode / route / system-prompt wrapping); splitting it would require shared mutable state across helpers +func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority string, prompt string, workflowStepID string, planMode, autoStart bool, attachments []v1.MessageAttachment, env map[string]string, route *executor.RouteOverride) (*executor.TaskExecution, error) { s.logger.Debug("manually starting task", zap.String("task_id", taskID), zap.String("agent_profile_id", agentProfileID), @@ -466,6 +483,7 @@ func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID s zap.Int("prompt_length", len(prompt)), zap.String("workflow_step_id", workflowStepID), zap.Bool("plan_mode", planMode), + zap.Bool("auto_start", autoStart), zap.Int("attachments", len(attachments))) // Office tasks do NOT transition through SCHEDULING / IN_PROGRESS on @@ -577,7 +595,7 @@ func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID s return nil, err } - s.postLaunchStart(ctx, taskID, execution, effectivePrompt, planModeActive || configMode, planModeActive, attachments) + s.postLaunchStart(ctx, taskID, execution, effectivePrompt, planModeActive || configMode, planModeActive, autoStart, attachments) // Note: Task stays in SCHEDULING state until the agent is fully initialized. // The executor will transition to IN_PROGRESS after StartAgentProcess() succeeds. @@ -713,9 +731,9 @@ func (s *Service) resolveEffectiveAgentProfile(ctx context.Context, taskID, work } // postLaunchStart records the initial message and sets plan mode after a successful launch. -func (s *Service) postLaunchStart(ctx context.Context, taskID string, execution *executor.TaskExecution, prompt string, recordPlanMode, setPlanMode bool, attachments []v1.MessageAttachment) { +func (s *Service) postLaunchStart(ctx context.Context, taskID string, execution *executor.TaskExecution, prompt string, recordPlanMode, setPlanMode, autoStart bool, attachments []v1.MessageAttachment) { if execution.SessionID != "" { - s.recordInitialMessage(ctx, taskID, execution.SessionID, prompt, recordPlanMode, attachments) + s.recordInitialMessage(ctx, taskID, execution.SessionID, prompt, recordPlanMode, autoStart, attachments) if setPlanMode { session, err := s.repo.GetTaskSession(ctx, execution.SessionID) @@ -769,9 +787,12 @@ func (s *Service) applyWorkflowAndPlanMode(ctx context.Context, prompt string, t } // recordInitialMessage creates the initial user message and updates session state after launch. -func (s *Service) recordInitialMessage(ctx context.Context, taskID, sessionID, prompt string, planModeActive bool, attachments []v1.MessageAttachment) { +// autoStart marks the message as having been created by an automated trigger +// (workflow auto-start, PR/issue watch, Jira/Linear integration) so cleanup +// logic can distinguish "agent ran on its own" from "user actually engaged". +func (s *Service) recordInitialMessage(ctx context.Context, taskID, sessionID, prompt string, planModeActive, autoStart bool, attachments []v1.MessageAttachment) { if s.messageCreator != nil && (prompt != "" || len(attachments) > 0) { - meta := NewUserMessageMeta().WithPlanMode(planModeActive).WithAttachments(attachments) + meta := NewUserMessageMeta().WithPlanMode(planModeActive).WithAutoStart(autoStart).WithAttachments(attachments) if err := s.messageCreator.CreateUserMessage(ctx, taskID, prompt, sessionID, s.getActiveTurnID(sessionID), meta.ToMap()); err != nil { s.logger.Error("failed to create initial user message", zap.String("task_id", taskID), diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 5f6d85ba5b..ec6e6199d7 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -380,7 +380,7 @@ func TestStartCreatedSession_WrongTask(t *testing.T) { // Session belongs to "task-other", not "task1" seedTaskAndSession(t, repo, "task-other", "session1", models.TaskSessionStateCreated) - _, err := svc.StartCreatedSession(context.Background(), "task1", "session1", "profile1", "prompt", false, false, nil) + _, err := svc.StartCreatedSession(context.Background(), "task1", "session1", "profile1", "prompt", false, false, false, nil) if err == nil { t.Fatal("expected error when session does not belong to task") } @@ -392,7 +392,7 @@ func TestStartCreatedSession_NotInCreatedState(t *testing.T) { seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateRunning) - _, err := svc.StartCreatedSession(context.Background(), "task1", "session1", "profile1", "prompt", false, false, nil) + _, err := svc.StartCreatedSession(context.Background(), "task1", "session1", "profile1", "prompt", false, false, false, nil) if err == nil { t.Fatal("expected error when session is not in CREATED state") } @@ -482,7 +482,7 @@ func TestRecordInitialMessage_DoesNotChangeSessionState(t *testing.T) { svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) svc.messageCreator = mc - svc.recordInitialMessage(ctx, "task1", "session1", "hello world", false, nil) + svc.recordInitialMessage(ctx, "task1", "session1", "hello world", false, false, nil) // Session state must remain STARTING — recordInitialMessage should not modify state. session, err := repo.GetTaskSession(ctx, "session1") @@ -509,7 +509,7 @@ func TestPostLaunchCreated_SkipMessage_DoesNotChangeSessionState(t *testing.T) { svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) - svc.postLaunchCreated(ctx, "task1", "session1", "prompt", true, false, nil) + svc.postLaunchCreated(ctx, "task1", "session1", "prompt", true, false, false, nil) // Session state must remain STARTING when skipMessage=true. session, err := repo.GetTaskSession(ctx, "session1") @@ -530,7 +530,7 @@ func TestPostLaunchCreated_WithMessage_DoesNotChangeSessionState(t *testing.T) { svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) svc.messageCreator = mc - svc.postLaunchCreated(ctx, "task1", "session1", "hello", false, false, nil) + svc.postLaunchCreated(ctx, "task1", "session1", "hello", false, false, false, nil) // Session state must remain STARTING — postLaunchCreated delegates to // recordInitialMessage which only creates the message. @@ -547,6 +547,29 @@ func TestPostLaunchCreated_WithMessage_DoesNotChangeSessionState(t *testing.T) { } } +func TestPostLaunchCreated_AutoStart_SetsMetadata(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateStarting) + + mc := &mockMessageCreator{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.messageCreator = mc + + // autoStart=true should land an `auto_start: true` tag on the + // recorded user message so HasUserAuthoredMessage skips it. This + // asserts the metadata wiring in recordInitialMessage directly — + // the broader behavior is tested in cmd/kandev TestHasUserAuthoredMessage. + svc.postLaunchCreated(ctx, "task1", "session1", "auto-started by workflow", false, false, true, nil) + + if len(mc.userMessages) != 1 { + t.Fatalf("expected 1 user message, got %d", len(mc.userMessages)) + } + if mc.userMessages[0].metadata["auto_start"] != true { + t.Fatalf("expected auto_start=true in metadata, got %v", mc.userMessages[0].metadata) + } +} + func TestPostLaunchCreated_PlanMode_SetsMetadata(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) @@ -556,7 +579,7 @@ func TestPostLaunchCreated_PlanMode_SetsMetadata(t *testing.T) { svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) svc.messageCreator = mc - svc.postLaunchCreated(ctx, "task1", "session1", "plan this", false, true, nil) + svc.postLaunchCreated(ctx, "task1", "session1", "plan this", false, true, false, nil) // User message should have plan_mode metadata. if len(mc.userMessages) != 1 { @@ -608,7 +631,7 @@ func TestStartCreatedSession_WrapsFirstPromptWithKandevSystemBlock(t *testing.T) mc := &mockMessageCreator{} svc.messageCreator = mc - _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", "Build me a feature", false, false, nil) + _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", "Build me a feature", false, false, false, nil) if err != nil { t.Fatalf("StartCreatedSession failed: %v", err) } @@ -669,7 +692,7 @@ func TestStartCreatedSession_DoesNotDoubleWrapPreWrappedPrompt(t *testing.T) { // Simulate an upstream caller (e.g. wsAddMessage) that has already wrapped. preWrapped := sysprompt.InjectKandevContext("task1", "session1", "Build me a feature") - _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", preWrapped, false, false, nil) + _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", preWrapped, false, false, false, nil) if err != nil { t.Fatalf("StartCreatedSession failed: %v", err) } @@ -713,7 +736,7 @@ func TestStartCreatedSession_EmptyPromptSkipsWrap(t *testing.T) { mc := &mockMessageCreator{} svc.messageCreator = mc - _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", "", false, false, nil) + _, err := svc.StartCreatedSession(ctx, "task1", "session1", "profile1", "", false, false, false, nil) if err != nil { t.Fatalf("StartCreatedSession failed: %v", err) } diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index 3e19e17fe3..88a055314a 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -26,7 +26,7 @@ import ( type OrchestratorService interface { PromptTask(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool, attachments []v1.MessageAttachment, dispatchOnly bool) (*orchestrator.PromptResult, error) ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error - StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode bool, attachments []v1.MessageAttachment) error + StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string, skipMessageRecord, planMode, autoStart bool, attachments []v1.MessageAttachment) error ProcessOnTurnStart(ctx context.Context, taskID, sessionID string) error } @@ -363,7 +363,7 @@ func (h *MessageHandlers) forwardMessageAsPrompt( ) { // For CREATED sessions, start the agent with this message as the initial prompt if startCreated { - if err := h.orchestrator.StartCreatedSession(ctx, taskID, sessionID, agentProfileID, content, true, planMode, attachments); err != nil { + if err := h.orchestrator.StartCreatedSession(ctx, taskID, sessionID, agentProfileID, content, true, planMode, false, attachments); err != nil { h.logger.Warn("failed to start created session from message", zap.String("task_id", taskID), zap.String("session_id", sessionID), diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 77438de1e1..a7439ff7d1 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -398,6 +398,11 @@ const ( ActionGitHubActionPresetsList = "github.action_presets.list" ActionGitHubActionPresetsUpdate = "github.action_presets.update" ActionGitHubActionPresetsReset = "github.action_presets.reset" + + // Manual cleanup sweeps over all dedup rows (review/issue). Used by the + // settings-page button so users can drain piled-up tasks on demand. + ActionGitHubCleanupReviewTasks = "github.cleanup.review_tasks" + ActionGitHubCleanupIssueTasks = "github.cleanup.issue_tasks" ) // Jira integration actions diff --git a/apps/web/components/github/github-settings.tsx b/apps/web/components/github/github-settings.tsx index ba82f41dc4..d6abbbf974 100644 --- a/apps/web/components/github/github-settings.tsx +++ b/apps/web/components/github/github-settings.tsx @@ -1,7 +1,7 @@ "use client"; import { useState, useCallback } from "react"; -import { IconBrandGithub, IconPlus } from "@tabler/icons-react"; +import { IconBrandGithub, IconPlus, IconTrashX } from "@tabler/icons-react"; import { Button } from "@kandev/ui/button"; import { Card, CardContent } from "@kandev/ui/card"; import { Separator } from "@kandev/ui/separator"; @@ -19,8 +19,52 @@ import { PRStatsPanel } from "./pr-stats"; import { useReviewWatches } from "@/hooks/domains/github/use-review-watches"; import { useIssueWatches } from "@/hooks/domains/github/use-issue-watches"; import { WorkspaceScopedSection } from "@/components/integrations/workspace-scoped-section"; +import { cleanupMergedReviewTasks, cleanupClosedIssueTasks } from "@/lib/api/domains/github-api"; import type { ReviewWatch, IssueWatch } from "@/lib/types/github"; +// CleanupNowButton runs a manual global sweep over the dedup tables. Useful +// for users who upgraded with a pile of legacy merged-PR / closed-issue +// review tasks that pre-date the cleanup_policy work — clicking once drains +// them according to each watch's policy. +function CleanupNowButton({ + label, + run, +}: { + label: string; + run: () => Promise<{ deleted: number }>; +}) { + const { toast } = useToast(); + const [busy, setBusy] = useState(false); + return ( + + ); +} + function useWatchActions(workspaceId?: string | null) { const { items: watches, create, update, remove, trigger } = useReviewWatches(workspaceId); const { toast } = useToast(); @@ -205,17 +249,20 @@ function ReviewWatchSection() { title="Review Watches" description="Automatically create tasks for PRs that need your review." action={ - +