diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 2d853062f5..561ff5faeb 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -319,6 +319,7 @@ func startAgentInfrastructure( // Start GitHub background poller ghPoller := githubpkg.NewPoller(services.GitHub, eventBus, log) + ghPoller.SetTaskBranchProvider(orchestratorSvc) ghPoller.Start(ctx) addCleanup(func() error { ghPoller.Stop(); return nil }) log.Info("GitHub poller started") diff --git a/apps/backend/internal/github/client.go b/apps/backend/internal/github/client.go index b3eb92ece3..e423fc6c62 100644 --- a/apps/backend/internal/github/client.go +++ b/apps/backend/internal/github/client.go @@ -48,6 +48,10 @@ type Client interface { // GetPRFeedback fetches aggregated feedback (reviews, comments, checks) for a PR. GetPRFeedback(ctx context.Context, owner, repo string, number int) (*PRFeedback, error) + // GetPRStatus fetches lightweight PR status (state, review summary, checks summary). + // Unlike GetPRFeedback, it skips comments for efficiency. + GetPRStatus(ctx context.Context, owner, repo string, number int) (*PRStatus, error) + // ListPRFiles lists files changed in a pull request. ListPRFiles(ctx context.Context, owner, repo string, number int) ([]PRFile, error) diff --git a/apps/backend/internal/github/client_helpers.go b/apps/backend/internal/github/client_helpers.go index 649e56bcd2..b2eff70713 100644 --- a/apps/backend/internal/github/client_helpers.go +++ b/apps/backend/internal/github/client_helpers.go @@ -67,6 +67,37 @@ func getPRFeedback(ctx context.Context, c Client, owner, repo string, number int }, nil } +// getPRStatus fetches lightweight PR status using any Client implementation. +// Unlike getPRFeedback, it skips comments to reduce API calls. +func getPRStatus(ctx context.Context, c Client, owner, repo string, number int) (*PRStatus, error) { + pr, err := c.GetPR(ctx, owner, repo, number) + if err != nil { + return nil, err + } + reviews, err := c.ListPRReviews(ctx, owner, repo, number) + if err != nil { + return nil, err + } + checks, err := c.ListCheckRuns(ctx, owner, repo, pr.HeadSHA) + if err != nil { + return nil, err + } + if reviews == nil { + reviews = []PRReview{} + } + if checks == nil { + checks = []CheckRun{} + } + reviewState, pendingReviewCount := deriveReviewSyncState(pr, reviews) + return &PRStatus{ + PR: pr, + ReviewState: reviewState, + ChecksState: computeOverallCheckStatus(checks), + ReviewCount: len(reviews), + PendingReviewCount: pendingReviewCount, + }, nil +} + // convertRawCheckRuns converts raw ghCheckRun structs into the domain CheckRun type. func convertRawCheckRuns(raw []ghCheckRun) []CheckRun { checks := make([]CheckRun, len(raw)) diff --git a/apps/backend/internal/github/controller_test.go b/apps/backend/internal/github/controller_test.go index f5c5cb9faf..6939910197 100644 --- a/apps/backend/internal/github/controller_test.go +++ b/apps/backend/internal/github/controller_test.go @@ -66,6 +66,9 @@ func (s *stubClient) SubmitReview(context.Context, string, string, int, string, func (s *stubClient) ListRepoBranches(context.Context, string, string) ([]RepoBranch, error) { return nil, nil } +func (s *stubClient) GetPRStatus(context.Context, string, string, int) (*PRStatus, error) { + return nil, nil +} func newControllerTestLogger() *logger.Logger { log, _ := logger.NewLogger(logger.LoggingConfig{ diff --git a/apps/backend/internal/github/gh_client.go b/apps/backend/internal/github/gh_client.go index 17e5df3ff7..d2222a8e92 100644 --- a/apps/backend/internal/github/gh_client.go +++ b/apps/backend/internal/github/gh_client.go @@ -359,6 +359,10 @@ func (c *GHClient) GetPRFeedback(ctx context.Context, owner, repo string, number return getPRFeedback(ctx, c, owner, repo, number) } +func (c *GHClient) GetPRStatus(ctx context.Context, owner, repo string, number int) (*PRStatus, error) { + return getPRStatus(ctx, c, owner, repo, number) +} + func (c *GHClient) ListPRFiles(ctx context.Context, owner, repo string, number int) ([]PRFile, error) { out, err := c.run(ctx, "api", fmt.Sprintf("repos/%s/%s/pulls/%d/files", owner, repo, number), diff --git a/apps/backend/internal/github/handlers.go b/apps/backend/internal/github/handlers.go index 10be37e669..01b43fdaee 100644 --- a/apps/backend/internal/github/handlers.go +++ b/apps/backend/internal/github/handlers.go @@ -45,6 +45,7 @@ func registerWSHandlers(dispatcher *ws.Dispatcher, svc *Service, log *logger.Log dispatcher.RegisterFunc(ws.ActionGitHubPRWatchDelete, wsDeletePRWatch(svc, log)) dispatcher.RegisterFunc(ws.ActionGitHubPRFilesGet, wsGetPRFiles(svc, log)) dispatcher.RegisterFunc(ws.ActionGitHubPRCommitsGet, wsGetPRCommits(svc, log)) + dispatcher.RegisterFunc(ws.ActionGitHubTaskPRSync, wsSyncTaskPR(svc, log)) dispatcher.RegisterFunc(ws.ActionGitHubStats, wsGetStats(svc, log)) } @@ -59,6 +60,26 @@ func parseMap(msg *ws.Message) (map[string]interface{}, error) { return m, err } +// wsWithField returns a WS handler that parses a single named string field from +// the payload and passes it to serviceFn, returning the result as the response. +func wsWithField(field string, serviceFn func(ctx context.Context, val string) (interface{}, error)) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + payload, parseErr := parseMap(msg) + if parseErr != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload: "+parseErr.Error(), nil) + } + val, _ := payload[field].(string) + if val == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, field+" required", nil) + } + result, err := serviceFn(ctx, val) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, result) + } +} + // wsDeleteByID returns a WS handler that parses an "id" field from the payload // and calls deleteFn. Used by both wsDeletePRWatch and wsDeleteReviewWatch. func wsDeleteByID(deleteFn func(ctx context.Context, id string) error) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { @@ -129,6 +150,12 @@ func wsGetTaskPR(svc *Service, _ *logger.Logger) func(ctx context.Context, msg * } } +func wsSyncTaskPR(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return wsWithField("task_id", func(ctx context.Context, taskID string) (interface{}, error) { + return svc.TriggerPRSync(ctx, taskID) + }) +} + func wsGetPRFeedback(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { payload, parseErr := parseMap(msg) @@ -151,21 +178,10 @@ func wsGetPRFeedback(svc *Service, _ *logger.Logger) func(ctx context.Context, m } func wsListReviewWatches(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - payload, parseErr := parseMap(msg) - if parseErr != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload: "+parseErr.Error(), nil) - } - workspaceID, _ := payload["workspace_id"].(string) - if workspaceID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "workspace_id required", nil) - } + return wsWithField("workspace_id", func(ctx context.Context, workspaceID string) (interface{}, error) { watches, err := svc.ListReviewWatches(ctx, workspaceID) - if err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) - } - return ws.NewResponse(msg.ID, msg.Action, watches) - } + return watches, err + }) } func wsCreateReviewWatch(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { diff --git a/apps/backend/internal/github/mock_client.go b/apps/backend/internal/github/mock_client.go index 22214052a3..d2ef7a2517 100644 --- a/apps/backend/internal/github/mock_client.go +++ b/apps/backend/internal/github/mock_client.go @@ -190,6 +190,10 @@ func (m *MockClient) GetPRFeedback(ctx context.Context, owner, repo string, numb return getPRFeedback(ctx, m, owner, repo, number) } +func (m *MockClient) GetPRStatus(ctx context.Context, owner, repo string, number int) (*PRStatus, error) { + return getPRStatus(ctx, m, owner, repo, number) +} + func (m *MockClient) ListPRFiles(_ context.Context, owner, repo string, number int) ([]PRFile, error) { m.mu.RLock() defer m.mu.RUnlock() diff --git a/apps/backend/internal/github/models.go b/apps/backend/internal/github/models.go index da9ebee0d3..39b1c9b692 100644 --- a/apps/backend/internal/github/models.go +++ b/apps/backend/internal/github/models.go @@ -82,6 +82,16 @@ type PRFeedback struct { HasIssues bool `json:"has_issues"` } +// PRStatus contains lightweight PR state used by the background poller. +// Unlike PRFeedback, it skips comments to reduce API calls. +type PRStatus struct { + PR *PR `json:"pr"` + ReviewState string `json:"review_state"` // "approved", "changes_requested", "pending", "" + ChecksState string `json:"checks_state"` // "success", "failure", "pending", "" + ReviewCount int `json:"review_count"` + PendingReviewCount int `json:"pending_review_count"` +} + // PRWatch tracks active PR monitoring (session → PR). type PRWatch struct { ID string `json:"id" db:"id"` @@ -94,6 +104,7 @@ type PRWatch struct { LastCheckedAt *time.Time `json:"last_checked_at,omitempty" db:"last_checked_at"` LastCommentAt *time.Time `json:"last_comment_at,omitempty" db:"last_comment_at"` LastCheckStatus string `json:"last_check_status" db:"last_check_status"` + LastReviewState string `json:"last_review_state" db:"last_review_state"` CreatedAt time.Time `json:"created_at" db:"created_at"` UpdatedAt time.Time `json:"updated_at" db:"updated_at"` } @@ -243,8 +254,6 @@ type PRFeedbackEvent struct { PRNumber int `json:"pr_number"` Owner string `json:"owner"` Repo string `json:"repo"` - NewComments int `json:"new_comments"` - ChecksChanged bool `json:"checks_changed"` NewCheckStatus string `json:"new_check_status"` NewReviewState string `json:"new_review_state"` } diff --git a/apps/backend/internal/github/noop_client.go b/apps/backend/internal/github/noop_client.go index cf67c357f6..1184184fe5 100644 --- a/apps/backend/internal/github/noop_client.go +++ b/apps/backend/internal/github/noop_client.go @@ -61,6 +61,10 @@ func (c *NoopClient) GetPRFeedback(context.Context, string, string, int) (*PRFee return nil, ErrNoClient } +func (c *NoopClient) GetPRStatus(context.Context, string, string, int) (*PRStatus, error) { + return nil, ErrNoClient +} + func (c *NoopClient) ListPRFiles(context.Context, string, string, int) ([]PRFile, error) { return nil, ErrNoClient } diff --git a/apps/backend/internal/github/pat_client.go b/apps/backend/internal/github/pat_client.go index 7c918f644f..124469dace 100644 --- a/apps/backend/internal/github/pat_client.go +++ b/apps/backend/internal/github/pat_client.go @@ -256,6 +256,10 @@ func (c *PATClient) GetPRFeedback(ctx context.Context, owner, repo string, numbe return getPRFeedback(ctx, c, owner, repo, number) } +func (c *PATClient) GetPRStatus(ctx context.Context, owner, repo string, number int) (*PRStatus, error) { + return getPRStatus(ctx, c, owner, repo, number) +} + func (c *PATClient) ListPRFiles(ctx context.Context, owner, repo string, number int) ([]PRFile, error) { var raw []ghPRFile endpoint := fmt.Sprintf("/repos/%s/%s/pulls/%d/files?per_page=100", owner, repo, number) diff --git a/apps/backend/internal/github/poller.go b/apps/backend/internal/github/poller.go index 5b18fe59c2..c28f16a8dc 100644 --- a/apps/backend/internal/github/poller.go +++ b/apps/backend/internal/github/poller.go @@ -13,15 +13,30 @@ import ( ) const ( - defaultPRPollInterval = 1 * time.Minute + defaultPRPollInterval = 30 * time.Second defaultReviewPollInterval = 5 * time.Minute ) +// TaskBranchInfo describes a task+session that may need a PR watch. +type TaskBranchInfo struct { + TaskID string + SessionID string + Owner string + Repo string + Branch string +} + +// TaskBranchProvider lists tasks that should have PR watches. +type TaskBranchProvider interface { + ListTasksNeedingPRWatch(ctx context.Context) ([]TaskBranchInfo, error) +} + // Poller runs background loops for PR monitoring and review queue checking. type Poller struct { - service *Service - eventBus bus.EventBus - logger *logger.Logger + service *Service + eventBus bus.EventBus + logger *logger.Logger + taskBranchProvider TaskBranchProvider cancel context.CancelFunc wg sync.WaitGroup @@ -87,6 +102,8 @@ func (p *Poller) prMonitorLoop(ctx context.Context) { } func (p *Poller) checkPRWatches(ctx context.Context) { + p.reconcileWatches(ctx) + watches, err := p.service.ListActivePRWatches(ctx) if err != nil { p.logger.Error("failed to list PR watches", zap.Error(err)) @@ -104,25 +121,33 @@ func (p *Poller) checkSinglePRWatch(ctx context.Context, watch *PRWatch) { return } - feedback, hasNew, err := p.service.CheckPRWatch(ctx, watch) + status, hasNew, err := p.service.CheckPRWatch(ctx, watch) if err != nil { p.logger.Debug("failed to check PR watch", zap.String("id", watch.ID), zap.Error(err)) return } - if feedback == nil { + if status == nil { return } + // Always sync latest PR state to the task-PR record. + if syncErr := p.service.SyncTaskPR(ctx, watch.TaskID, status); syncErr != nil { + p.logger.Error("failed to sync task PR", + zap.String("task_id", watch.TaskID), zap.Error(syncErr)) + return // Keep watch so the next cycle can retry + } + // Auto-cleanup: remove watch when PR is merged or closed. - if feedback.PR != nil && (feedback.PR.State == prStateMerged || feedback.PR.State == prStateClosed) { + if status.PR != nil && (status.PR.State == prStateMerged || status.PR.State == prStateClosed) { + p.publishPRStatusEvent(ctx, watch, status) if delErr := p.service.DeletePRWatch(ctx, watch.ID); delErr != nil { p.logger.Error("failed to delete completed PR watch", zap.String("id", watch.ID), zap.Error(delErr)) } else { p.logger.Info("removed PR watch for completed PR", zap.String("id", watch.ID), - zap.String("state", feedback.PR.State), + zap.String("state", status.PR.State), zap.Int("pr_number", watch.PRNumber)) } return @@ -132,13 +157,7 @@ func (p *Poller) checkSinglePRWatch(ctx context.Context, watch *PRWatch) { return } - // Sync task-PR record with latest data - if syncErr := p.service.SyncTaskPR(ctx, watch.TaskID, feedback); syncErr != nil { - p.logger.Error("failed to sync task PR", zap.String("task_id", watch.TaskID), zap.Error(syncErr)) - } - - // Publish feedback event for UI notification - p.publishPRFeedbackEvent(ctx, watch, feedback) + p.publishPRStatusEvent(ctx, watch, status) } // detectPRForWatch searches GitHub for a PR on the watch's branch. @@ -159,7 +178,7 @@ func (p *Poller) detectPRForWatch(ctx context.Context, watch *PRWatch) { // Update last_checked_at regardless of result now := time.Now().UTC() - _ = p.service.store.UpdatePRWatchTimestamps(ctx, watch.ID, now, nil, "") + _ = p.service.store.UpdatePRWatchTimestamps(ctx, watch.ID, now, nil, "", "") if pr == nil { return @@ -188,7 +207,7 @@ func (p *Poller) detectPRForWatch(ctx context.Context, watch *PRWatch) { zap.Int("pr_number", pr.Number)) } -func (p *Poller) publishPRFeedbackEvent(ctx context.Context, watch *PRWatch, feedback *PRFeedback) { +func (p *Poller) publishPRStatusEvent(ctx context.Context, watch *PRWatch, status *PRStatus) { if p.eventBus == nil { return } @@ -198,10 +217,8 @@ func (p *Poller) publishPRFeedbackEvent(ctx context.Context, watch *PRWatch, fee PRNumber: watch.PRNumber, Owner: watch.Owner, Repo: watch.Repo, - NewComments: len(feedback.Comments), - ChecksChanged: true, - NewCheckStatus: computeOverallCheckStatus(feedback.Checks), - NewReviewState: computeOverallReviewState(feedback.Reviews), + NewCheckStatus: status.ChecksState, + NewReviewState: status.ReviewState, } event := bus.NewEvent(events.GitHubPRFeedback, "github_poller", evt) if err := p.eventBus.Publish(ctx, events.GitHubPRFeedback, event); err != nil { @@ -209,6 +226,31 @@ func (p *Poller) publishPRFeedbackEvent(ctx context.Context, watch *PRWatch, fee } } +// SetTaskBranchProvider sets the provider used for watch reconciliation. +func (p *Poller) SetTaskBranchProvider(provider TaskBranchProvider) { + p.taskBranchProvider = provider +} + +// reconcileWatches ensures PR watches exist for all tasks that need them. +func (p *Poller) reconcileWatches(ctx context.Context) { + if p.taskBranchProvider == nil { + return + } + tasks, err := p.taskBranchProvider.ListTasksNeedingPRWatch(ctx) + if err != nil { + p.logger.Error("failed to list tasks needing PR watch", zap.Error(err)) + return + } + for _, task := range tasks { + if _, ensureErr := p.service.EnsurePRWatch( + ctx, task.SessionID, task.TaskID, task.Owner, task.Repo, task.Branch, + ); ensureErr != nil { + p.logger.Error("failed to ensure PR watch", + zap.String("session_id", task.SessionID), zap.Error(ensureErr)) + } + } +} + // reviewQueueLoop polls review watches for new PRs. func (p *Poller) reviewQueueLoop(ctx context.Context) { defer p.wg.Done() diff --git a/apps/backend/internal/github/poller_test.go b/apps/backend/internal/github/poller_test.go new file mode 100644 index 0000000000..55871f0736 --- /dev/null +++ b/apps/backend/internal/github/poller_test.go @@ -0,0 +1,361 @@ +package github + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" +) + +// setupPollerTest creates a Poller backed by an in-memory SQLite store and MockClient. +func setupPollerTest(t *testing.T) (*Poller, *Service, *MockClient, *Store) { + t.Helper() + + rawDB, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + rawDB.SetMaxOpenConns(1) + rawDB.SetMaxIdleConns(1) + sqlxDB := sqlx.NewDb(rawDB, "sqlite3") + t.Cleanup(func() { _ = sqlxDB.Close() }) + + store, err := NewStore(sqlxDB, sqlxDB) + if err != nil { + t.Fatalf("new store: %v", err) + } + + mockClient := NewMockClient() + log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"}) + eventBus := bus.NewMemoryEventBus(log) + + svc := NewService(mockClient, "pat", nil, store, eventBus, log) + poller := NewPoller(svc, eventBus, log) + + return poller, svc, mockClient, store +} + +func TestCheckSinglePRWatch_MergedPR_SyncsThenDeletes(t *testing.T) { + poller, _, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Set up a merged PR in the mock client. + now := time.Now().UTC() + mergedAt := now.Add(-1 * time.Hour) + mockClient.AddPR(&PR{ + Number: 42, + Title: "Feature PR", + State: prStateMerged, + HeadSHA: "abc123", + HeadBranch: "feature-branch", + RepoOwner: "owner", + RepoName: "repo", + MergedAt: &mergedAt, + }) + + // Create a PRWatch in the DB. + watch := &PRWatch{ + SessionID: "sess-1", + TaskID: "task-1", + Owner: "owner", + Repo: "repo", + PRNumber: 42, + Branch: "feature-branch", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + + // Create a TaskPR record so SyncTaskPR has something to update. + taskPR := &TaskPR{ + TaskID: "task-1", + Owner: "owner", + Repo: "repo", + PRNumber: 42, + PRURL: "https://github.com/owner/repo/pull/42", + PRTitle: "Feature PR", + HeadBranch: "feature-branch", + BaseBranch: "main", + State: "open", + } + if err := store.CreateTaskPR(ctx, taskPR); err != nil { + t.Fatalf("create task PR: %v", err) + } + + // Act + poller.checkSinglePRWatch(ctx, watch) + + // Assert: TaskPR record should be updated with state="merged". + updatedTP, err := store.GetTaskPR(ctx, "task-1") + if err != nil { + t.Fatalf("get task PR: %v", err) + } + if updatedTP == nil { + t.Fatal("expected task PR to exist after sync") + } + if updatedTP.State != prStateMerged { + t.Errorf("expected task PR state=%q, got %q", prStateMerged, updatedTP.State) + } + if updatedTP.MergedAt == nil { + t.Error("expected task PR MergedAt to be set") + } + + // Assert: PRWatch should be deleted. + remainingWatch, err := store.GetPRWatchBySession(ctx, "sess-1") + if err != nil { + t.Fatalf("get PR watch: %v", err) + } + if remainingWatch != nil { + t.Error("expected PR watch to be deleted after merged PR") + } +} + +func TestCheckSinglePRWatch_OpenPR_SyncsOnChange(t *testing.T) { + poller, _, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Set up an open PR with a passing check. + mockClient.AddPR(&PR{ + Number: 10, + Title: "Open PR", + State: "open", + HeadSHA: "def456", + HeadBranch: "open-branch", + RepoOwner: "owner", + RepoName: "repo", + Additions: 5, + Deletions: 3, + }) + mockClient.AddCheckRuns("owner", "repo", "def456", []CheckRun{ + {Name: "ci", Status: "completed", Conclusion: "success"}, + }) + + // Create a PRWatch with a different last_check_status to trigger hasNew. + watch := &PRWatch{ + SessionID: "sess-2", + TaskID: "task-2", + Owner: "owner", + Repo: "repo", + PRNumber: 10, + Branch: "open-branch", + LastCheckStatus: "pending", // different from "success" -> hasNew=true + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + + // Create a TaskPR record. + taskPR := &TaskPR{ + TaskID: "task-2", + Owner: "owner", + Repo: "repo", + PRNumber: 10, + PRURL: "https://github.com/owner/repo/pull/10", + PRTitle: "Open PR", + HeadBranch: "open-branch", + BaseBranch: "main", + State: "open", + } + if err := store.CreateTaskPR(ctx, taskPR); err != nil { + t.Fatalf("create task PR: %v", err) + } + + // Act + poller.checkSinglePRWatch(ctx, watch) + + // Assert: TaskPR should be synced with latest data. + updatedTP, err := store.GetTaskPR(ctx, "task-2") + if err != nil { + t.Fatalf("get task PR: %v", err) + } + if updatedTP == nil { + t.Fatal("expected task PR to exist") + } + if updatedTP.State != "open" { + t.Errorf("expected state=open, got %q", updatedTP.State) + } + if updatedTP.ChecksState != "success" { + t.Errorf("expected checks_state=success, got %q", updatedTP.ChecksState) + } + if updatedTP.Additions != 5 { + t.Errorf("expected additions=5, got %d", updatedTP.Additions) + } + + // Assert: PRWatch should NOT be deleted (PR is still open). + remainingWatch, err := store.GetPRWatchBySession(ctx, "sess-2") + if err != nil { + t.Fatalf("get PR watch: %v", err) + } + if remainingWatch == nil { + t.Error("expected PR watch to still exist for open PR") + } +} + +// mockTaskBranchProvider implements TaskBranchProvider for testing. +type mockTaskBranchProvider struct { + tasks []TaskBranchInfo + err error +} + +func (m *mockTaskBranchProvider) ListTasksNeedingPRWatch(_ context.Context) ([]TaskBranchInfo, error) { + return m.tasks, m.err +} + +func TestReconcileWatches_CreatesWatchesForTasks(t *testing.T) { + poller, _, _, store := setupPollerTest(t) + ctx := context.Background() + + prov := &mockTaskBranchProvider{ + tasks: []TaskBranchInfo{ + {TaskID: "t1", SessionID: "s1", Owner: "myorg", Repo: "myrepo", Branch: "feature-a"}, + {TaskID: "t2", SessionID: "s2", Owner: "myorg", Repo: "myrepo", Branch: "feature-b"}, + }, + } + poller.SetTaskBranchProvider(prov) + + poller.reconcileWatches(ctx) + + // Verify watches were created for both sessions. + w1, err := store.GetPRWatchBySession(ctx, "s1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if w1 == nil { + t.Fatal("expected PR watch for session s1") + } + if w1.Branch != "feature-a" { + t.Errorf("expected branch %q, got %q", "feature-a", w1.Branch) + } + if w1.TaskID != "t1" { + t.Errorf("expected task_id %q, got %q", "t1", w1.TaskID) + } + + w2, err := store.GetPRWatchBySession(ctx, "s2") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if w2 == nil { + t.Fatal("expected PR watch for session s2") + } + if w2.Branch != "feature-b" { + t.Errorf("expected branch %q, got %q", "feature-b", w2.Branch) + } +} + +func TestReconcileWatches_NilProvider(t *testing.T) { + poller, _, _, _ := setupPollerTest(t) + ctx := context.Background() + + // Should not panic when provider is nil. + poller.reconcileWatches(ctx) +} + +func TestReconcileWatches_SkipsExistingWatches(t *testing.T) { + poller, _, _, store := setupPollerTest(t) + ctx := context.Background() + + // Pre-create a watch for s1. + existing := &PRWatch{ + SessionID: "s1", + TaskID: "t1", + Owner: "myorg", + Repo: "myrepo", + PRNumber: 0, + Branch: "feature-a", + } + if err := store.CreatePRWatch(ctx, existing); err != nil { + t.Fatalf("create PR watch: %v", err) + } + + prov := &mockTaskBranchProvider{ + tasks: []TaskBranchInfo{ + {TaskID: "t1", SessionID: "s1", Owner: "myorg", Repo: "myrepo", Branch: "feature-a"}, + {TaskID: "t2", SessionID: "s2", Owner: "myorg", Repo: "myrepo", Branch: "feature-b"}, + }, + } + poller.SetTaskBranchProvider(prov) + + poller.reconcileWatches(ctx) + + // s1 should still have its original watch (EnsurePRWatch is idempotent). + w1, _ := store.GetPRWatchBySession(ctx, "s1") + if w1 == nil { + t.Fatal("expected PR watch for session s1") + } + if w1.ID != existing.ID { + t.Errorf("expected original watch ID %q, got %q", existing.ID, w1.ID) + } + + // s2 should have a new watch. + w2, _ := store.GetPRWatchBySession(ctx, "s2") + if w2 == nil { + t.Fatal("expected PR watch for session s2") + } +} + +func TestCheckSinglePRWatch_OpenPR_NoChange_NoSync(t *testing.T) { + poller, _, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Set up an open PR with a passing check. + mockClient.AddPR(&PR{ + Number: 20, + Title: "Stable PR", + State: "open", + HeadSHA: "ghi789", + HeadBranch: "stable-branch", + RepoOwner: "owner", + RepoName: "repo", + }) + mockClient.AddCheckRuns("owner", "repo", "ghi789", []CheckRun{ + {Name: "ci", Status: "completed", Conclusion: "success"}, + }) + + // Create a PRWatch with matching last_check_status -> no change. + watch := &PRWatch{ + SessionID: "sess-3", + TaskID: "task-3", + Owner: "owner", + Repo: "repo", + PRNumber: 20, + Branch: "stable-branch", + LastCheckStatus: "success", // same -> hasNew=false + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + + // Create a TaskPR record. + taskPR := &TaskPR{ + TaskID: "task-3", + Owner: "owner", + Repo: "repo", + PRNumber: 20, + PRURL: "https://github.com/owner/repo/pull/20", + PRTitle: "Stable PR", + HeadBranch: "stable-branch", + BaseBranch: "main", + State: "open", + } + if err := store.CreateTaskPR(ctx, taskPR); err != nil { + t.Fatalf("create task PR: %v", err) + } + + // Act + poller.checkSinglePRWatch(ctx, watch) + + // Assert: PRWatch should NOT be deleted. + remainingWatch, err := store.GetPRWatchBySession(ctx, "sess-3") + if err != nil { + t.Fatalf("get PR watch: %v", err) + } + if remainingWatch == nil { + t.Error("expected PR watch to still exist") + } +} diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index 816582ec4b..d148f5baf8 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -213,37 +213,33 @@ func (s *Service) UpdatePRWatchPRNumber(ctx context.Context, id string, prNumber return s.store.UpdatePRWatchPRNumber(ctx, id, prNumber) } -// CheckPRWatch fetches latest feedback for a PR watch and determines if there are changes. -func (s *Service) CheckPRWatch(ctx context.Context, watch *PRWatch) (*PRFeedback, bool, error) { +// CheckPRWatch fetches lightweight PR status for a watch and determines if there are changes. +func (s *Service) CheckPRWatch(ctx context.Context, watch *PRWatch) (*PRStatus, bool, error) { if s.client == nil { return nil, false, fmt.Errorf("github client not available") } - feedback, err := s.client.GetPRFeedback(ctx, watch.Owner, watch.Repo, watch.PRNumber) + status, err := s.client.GetPRStatus(ctx, watch.Owner, watch.Repo, watch.PRNumber) if err != nil { return nil, false, err } hasNew := false - // Check for new comments - latestCommentAt := findLatestCommentTime(feedback.Comments) - if latestCommentAt != nil && (watch.LastCommentAt == nil || latestCommentAt.After(*watch.LastCommentAt)) { + // Check for check status or review state changes + if status.ChecksState != watch.LastCheckStatus { hasNew = true } - - // Check for check status changes - checkStatus := computeOverallCheckStatus(feedback.Checks) - if checkStatus != watch.LastCheckStatus { + if status.ReviewState != watch.LastReviewState { hasNew = true } // Update watch timestamps now := time.Now().UTC() - if err := s.store.UpdatePRWatchTimestamps(ctx, watch.ID, now, latestCommentAt, checkStatus); err != nil { + if err := s.store.UpdatePRWatchTimestamps(ctx, watch.ID, now, nil, status.ChecksState, status.ReviewState); err != nil { s.logger.Error("failed to update PR watch timestamps", zap.String("id", watch.ID), zap.Error(err)) } - return feedback, hasNew, nil + return status, hasNew, nil } // EnsurePRWatch creates a PRWatch with pr_number=0 for a session if one doesn't already exist. @@ -403,25 +399,27 @@ func (s *Service) ListTaskPRs(ctx context.Context, taskIDs []string) (map[string return s.store.ListTaskPRsByTaskIDs(ctx, taskIDs) } -// SyncTaskPR updates a TaskPR record with the latest PR data from feedback. -func (s *Service) SyncTaskPR(ctx context.Context, taskID string, feedback *PRFeedback) error { +// SyncTaskPR updates a TaskPR record with the latest PR status. +func (s *Service) SyncTaskPR(ctx context.Context, taskID string, status *PRStatus) error { + if status == nil || status.PR == nil { + return fmt.Errorf("sync task PR: missing PR data for task %s", taskID) + } tp, err := s.store.GetTaskPR(ctx, taskID) if err != nil || tp == nil { return err } - tp.State = feedback.PR.State - tp.PRTitle = feedback.PR.Title - tp.Additions = feedback.PR.Additions - tp.Deletions = feedback.PR.Deletions - tp.MergedAt = feedback.PR.MergedAt - tp.ClosedAt = feedback.PR.ClosedAt - tp.CommentCount = len(feedback.Comments) - tp.ReviewCount = len(feedback.Reviews) - reviewState, pendingReviewCount := deriveReviewSyncState(feedback.PR, feedback.Reviews) - tp.ReviewState = reviewState - tp.ChecksState = computeOverallCheckStatus(feedback.Checks) - tp.PendingReviewCount = pendingReviewCount + tp.State = status.PR.State + tp.PRTitle = status.PR.Title + tp.Additions = status.PR.Additions + tp.Deletions = status.PR.Deletions + tp.MergedAt = status.PR.MergedAt + tp.ClosedAt = status.PR.ClosedAt + tp.ReviewState = status.ReviewState + tp.ChecksState = status.ChecksState + tp.ReviewCount = status.ReviewCount + tp.PendingReviewCount = status.PendingReviewCount + // CommentCount is no longer updated from polling -- only refreshed on-demand now := time.Now().UTC() tp.LastSyncedAt = &now @@ -456,6 +454,64 @@ func (s *Service) GetPRFeedback(ctx context.Context, owner, repo string, number return s.client.GetPRFeedback(ctx, owner, repo, number) } +// TriggerPRSync performs an immediate PR status sync for a task. +// If the watch has a PR number, it fetches the latest status from GitHub +// and syncs it to the TaskPR record. If still searching (pr_number=0), +// it attempts to find the PR by branch. +func (s *Service) TriggerPRSync(ctx context.Context, taskID string) (*TaskPR, error) { + watch, err := s.store.GetPRWatchByTask(ctx, taskID) + if err != nil { + return nil, fmt.Errorf("get PR watch: %w", err) + } + if watch == nil { + // No watch — just return existing TaskPR if any + return s.store.GetTaskPR(ctx, taskID) + } + + if watch.PRNumber == 0 { + return s.triggerPRDetection(ctx, watch, taskID) + } + + return s.triggerPRStatusSync(ctx, watch, taskID) +} + +func (s *Service) triggerPRDetection(ctx context.Context, watch *PRWatch, taskID string) (*TaskPR, error) { + if s.client == nil { + return nil, nil + } + pr, err := s.client.FindPRByBranch(ctx, watch.Owner, watch.Repo, watch.Branch) + if err != nil || pr == nil { + return nil, err + } + if err := s.store.UpdatePRWatchPRNumber(ctx, watch.ID, pr.Number); err != nil { + s.logger.Error("failed to update PR watch number during sync", + zap.String("watch_id", watch.ID), zap.Int("pr_number", pr.Number), zap.Error(err)) + return nil, fmt.Errorf("update PR watch: %w", err) + } + if _, assocErr := s.AssociatePRWithTask(ctx, taskID, pr); assocErr != nil { + s.logger.Error("failed to associate PR with task during sync", + zap.String("task_id", taskID), zap.Int("pr_number", pr.Number), zap.Error(assocErr)) + return nil, fmt.Errorf("associate PR: %w", assocErr) + } + // Also fetch status so the first response includes review/check state + watch.PRNumber = pr.Number + return s.triggerPRStatusSync(ctx, watch, taskID) +} + +func (s *Service) triggerPRStatusSync(ctx context.Context, watch *PRWatch, taskID string) (*TaskPR, error) { + status, _, err := s.CheckPRWatch(ctx, watch) + if err != nil { + return nil, err + } + if status == nil { + return s.store.GetTaskPR(ctx, taskID) + } + if syncErr := s.SyncTaskPR(ctx, taskID, status); syncErr != nil { + return nil, syncErr + } + return s.store.GetTaskPR(ctx, taskID) +} + // --- PR files and commits (live) --- // GetPRFiles fetches files changed in a PR from GitHub. diff --git a/apps/backend/internal/github/service_test.go b/apps/backend/internal/github/service_test.go index 654ad182be..39a41a6a61 100644 --- a/apps/backend/internal/github/service_test.go +++ b/apps/backend/internal/github/service_test.go @@ -419,3 +419,107 @@ func TestFindLatestCommentTime(t *testing.T) { }) } } + +func TestTriggerPRSync_SyncsExistingWatch(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Set up a PR in mock client. + mockClient.AddPR(&PR{ + Number: 10, + Title: "Feature PR", + State: "open", + HeadSHA: "abc123", + HeadBranch: "feat/x", + RepoOwner: "org", + RepoName: "repo", + }) + + // Create a PR watch and TaskPR in DB. + watch := &PRWatch{ + SessionID: "s1", + TaskID: "t1", + Owner: "org", + Repo: "repo", + PRNumber: 10, + Branch: "feat/x", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatal(err) + } + tp := &TaskPR{ + TaskID: "t1", + Owner: "org", + Repo: "repo", + PRNumber: 10, + PRTitle: "Feature PR", + State: "open", + } + if err := store.CreateTaskPR(ctx, tp); err != nil { + t.Fatal(err) + } + + // Trigger sync. + result, err := svc.TriggerPRSync(ctx, "t1") + if err != nil { + t.Fatalf("TriggerPRSync: %v", err) + } + if result == nil { + t.Fatal("expected non-nil TaskPR") + } + if result.LastSyncedAt == nil { + t.Error("expected LastSyncedAt to be set after sync") + } +} + +func TestTriggerPRSync_DetectsPR(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Set up a PR findable by branch. + mockClient.AddPR(&PR{ + Number: 20, + Title: "New PR", + State: "open", + HeadBranch: "feat/y", + RepoOwner: "org", + RepoName: "repo", + }) + + // Create a watch with pr_number=0 (still searching). + watch := &PRWatch{ + SessionID: "s2", + TaskID: "t2", + Owner: "org", + Repo: "repo", + PRNumber: 0, + Branch: "feat/y", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatal(err) + } + + result, err := svc.TriggerPRSync(ctx, "t2") + if err != nil { + t.Fatalf("TriggerPRSync: %v", err) + } + if result == nil { + t.Fatal("expected TaskPR after detection") + } + if result.PRNumber != 20 { + t.Errorf("expected PR #20, got #%d", result.PRNumber) + } +} + +func TestTriggerPRSync_NoWatch(t *testing.T) { + _, svc, _, _ := setupPollerTest(t) + ctx := context.Background() + + result, err := svc.TriggerPRSync(ctx, "nonexistent") + if err != nil { + t.Fatalf("TriggerPRSync: %v", err) + } + if result != nil { + t.Errorf("expected nil TaskPR for task without watch, got %+v", result) + } +} diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index 0ef6795f11..3cbc563f5e 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -40,6 +40,7 @@ const createTablesSQL = ` last_checked_at DATETIME, last_comment_at DATETIME, last_check_status TEXT DEFAULT '', + last_review_state TEXT DEFAULT '', created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ); @@ -103,8 +104,12 @@ const createTablesSQL = ` ` func (s *Store) initSchema() error { - _, err := s.db.Exec(createTablesSQL) - return err + if _, err := s.db.Exec(createTablesSQL); err != nil { + return err + } + // Idempotent migrations for existing databases. + _, _ = s.db.Exec(`ALTER TABLE github_pr_watches ADD COLUMN last_review_state TEXT DEFAULT ''`) + return nil } // --- PR Watch operations --- @@ -134,6 +139,16 @@ func (s *Store) GetPRWatchBySession(ctx context.Context, sessionID string) (*PRW return &w, err } +// GetPRWatchByTask returns the PR watch for a task (first match). +func (s *Store) GetPRWatchByTask(ctx context.Context, taskID string) (*PRWatch, error) { + var w PRWatch + err := s.ro.GetContext(ctx, &w, `SELECT * FROM github_pr_watches WHERE task_id = ? ORDER BY updated_at DESC LIMIT 1`, taskID) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return &w, err +} + // ListActivePRWatches returns all active PR watches. func (s *Store) ListActivePRWatches(ctx context.Context) ([]*PRWatch, error) { var watches []*PRWatch @@ -141,12 +156,12 @@ func (s *Store) ListActivePRWatches(ctx context.Context) ([]*PRWatch, error) { return watches, err } -// UpdatePRWatchTimestamps updates the last checked timestamps. -func (s *Store) UpdatePRWatchTimestamps(ctx context.Context, id string, checkedAt time.Time, commentAt *time.Time, checkStatus string) error { +// UpdatePRWatchTimestamps updates the last checked timestamps and status fields. +func (s *Store) UpdatePRWatchTimestamps(ctx context.Context, id string, checkedAt time.Time, commentAt *time.Time, checkStatus, reviewState string) error { _, err := s.db.ExecContext(ctx, ` - UPDATE github_pr_watches SET last_checked_at = ?, last_comment_at = ?, last_check_status = ?, updated_at = ? + UPDATE github_pr_watches SET last_checked_at = ?, last_comment_at = ?, last_check_status = ?, last_review_state = ?, updated_at = ? WHERE id = ?`, - checkedAt, commentAt, checkStatus, time.Now().UTC(), id) + checkedAt, commentAt, checkStatus, reviewState, time.Now().UTC(), id) return err } diff --git a/apps/backend/internal/orchestrator/event_handlers_github.go b/apps/backend/internal/orchestrator/event_handlers_github.go index 612b767833..d6e60b17d4 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github.go +++ b/apps/backend/internal/orchestrator/event_handlers_github.go @@ -28,6 +28,7 @@ type GitHubService interface { UpdatePRWatchPRNumber(ctx context.Context, id string, prNumber int) error AssociatePRWithTask(ctx context.Context, taskID string, pr *github.PR) (*github.TaskPR, error) GetTaskPR(ctx context.Context, taskID string) (*github.TaskPR, error) + ListActivePRWatches(ctx context.Context) ([]*github.PRWatch, error) RecordReviewPRTask(ctx context.Context, watchID, repoOwner, repoName string, prNumber int, prURL, taskID string) error } @@ -557,6 +558,71 @@ func (s *Service) CheckSessionPR(ctx context.Context, taskID, sessionID string) return true, nil } +// ListTasksNeedingPRWatch returns task-session pairs that have worktree branches +// but no existing PR watch. This satisfies the github.TaskBranchProvider interface. +func (s *Service) ListTasksNeedingPRWatch(ctx context.Context) ([]github.TaskBranchInfo, error) { + if s.githubService == nil { + return nil, nil + } + store, ok := s.repo.(repoStore) + if !ok { + return nil, nil + } + return s.buildTaskBranchList(ctx, store) +} + +// buildTaskBranchList queries sessions with branches, filters out those with +// existing PR watches, and resolves GitHub owner/repo for the remainder. +func (s *Service) buildTaskBranchList(ctx context.Context, store repoStore) ([]github.TaskBranchInfo, error) { + sessions, err := store.ListSessionsWithBranches(ctx) + if err != nil { + return nil, fmt.Errorf("list sessions with branches: %w", err) + } + if len(sessions) == 0 { + return nil, nil + } + + // Batch fetch existing watches to avoid N+1 queries. + watchedSessions := s.buildWatchedSessionSet(ctx) + + var result []github.TaskBranchInfo + for _, sess := range sessions { + if watchedSessions[sess.SessionID] { + continue + } + owner, repo := s.resolveTaskRepo(ctx, sess.TaskID) + if owner == "" || repo == "" { + continue + } + branch := s.resolvePRWatchBranch(ctx, sess.TaskID, sess.SessionID, sess.Branch) + if branch == "" { + continue + } + result = append(result, github.TaskBranchInfo{ + TaskID: sess.TaskID, + SessionID: sess.SessionID, + Owner: owner, + Repo: repo, + Branch: branch, + }) + } + return result, nil +} + +// buildWatchedSessionSet returns a set of session IDs that already have PR watches. +func (s *Service) buildWatchedSessionSet(ctx context.Context) map[string]bool { + watches, err := s.githubService.ListActivePRWatches(ctx) + if err != nil { + s.logger.Debug("failed to list PR watches for reconciliation", zap.Error(err)) + return nil + } + set := make(map[string]bool, len(watches)) + for _, w := range watches { + set[w.SessionID] = true + } + return set +} + // subscribeGitHubEvents subscribes to GitHub-related events on the event bus. func (s *Service) subscribeGitHubEvents() { if s.eventBus == nil { diff --git a/apps/backend/internal/orchestrator/event_handlers_github_test.go b/apps/backend/internal/orchestrator/event_handlers_github_test.go index 1e5af849ba..064db3ee9a 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_github_test.go @@ -7,6 +7,8 @@ import ( "github.com/kandev/kandev/internal/github" "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + v1 "github.com/kandev/kandev/pkg/api/v1" ) // mockGitHubService implements GitHubService for testing. @@ -57,6 +59,9 @@ func (m *mockGitHubService) UpdatePRWatchPRNumber(_ context.Context, _ string, p m.updatedPRNumber = prNumber return nil } +func (m *mockGitHubService) ListActivePRWatches(context.Context) ([]*github.PRWatch, error) { + return nil, nil +} func (m *mockGitHubService) RecordReviewPRTask(context.Context, string, string, string, int, string, string) error { return nil } @@ -494,3 +499,133 @@ func TestDetectPushAndAssociatePR(t *testing.T) { } }) } + +func TestListTasksNeedingPRWatch(t *testing.T) { + ctx := context.Background() + now := time.Now().UTC() + + // seedFullSession creates a task, session, repository, task-repo, and worktree. + seedFullSession := func(t *testing.T, repo interface { + CreateRepository(ctx context.Context, r *models.Repository) error + CreateTaskRepository(ctx context.Context, tr *models.TaskRepository) error + CreateTaskSessionWorktree(ctx context.Context, wt *models.TaskSessionWorktree) error + }, taskID, sessionID, branch, repoID string) { + t.Helper() + rObj := &models.Repository{ + ID: repoID, WorkspaceID: "ws1", Name: "myrepo", + SourceType: "provider", Provider: "github", + ProviderOwner: "myorg", ProviderName: "myrepo", + CreatedAt: now, UpdatedAt: now, + } + // Ignore duplicate errors for shared repos. + _ = repo.CreateRepository(ctx, rObj) + + tr := &models.TaskRepository{ + ID: "tr-" + sessionID, TaskID: taskID, RepositoryID: repoID, + Position: 0, CreatedAt: now, UpdatedAt: now, + } + _ = repo.CreateTaskRepository(ctx, tr) + + wt := &models.TaskSessionWorktree{ + ID: "wt-" + sessionID, SessionID: sessionID, + WorktreeID: "wtree-" + sessionID, RepositoryID: repoID, + WorktreeBranch: branch, CreatedAt: now, + } + if err := repo.CreateTaskSessionWorktree(ctx, wt); err != nil { + t.Fatalf("failed to create worktree: %v", err) + } + } + + // seedTask creates a second task and session in an already-seeded repo. + seedTask := func(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID string) { + t.Helper() + task := &models.Task{ + ID: taskID, WorkflowID: "wf1", WorkflowStepID: "step1", + Title: "Test Task", Description: "Test", + State: v1.TaskStateInProgress, CreatedAt: now, UpdatedAt: now, + } + if err := repo.CreateTask(ctx, task); err != nil { + t.Fatalf("failed to create task: %v", err) + } + session := &models.TaskSession{ + ID: sessionID, TaskID: taskID, + State: models.TaskSessionStateRunning, StartedAt: now, UpdatedAt: now, + } + if err := repo.CreateTaskSession(ctx, session); err != nil { + t.Fatalf("failed to create session: %v", err) + } + } + + t.Run("returns sessions with branches but no watch", func(t *testing.T) { + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t1", "s1", "step1") + seedTask(t, testRepo, "t2", "s2") + seedFullSession(t, testRepo, "t1", "s1", "feature-a", "repo1") + seedFullSession(t, testRepo, "t2", "s2", "feature-b", "repo1") + + svc := createTestService(testRepo, newMockStepGetter(), newMockTaskRepo()) + ghSvc := &mockGitHubService{ + // s1 has a watch, s2 does not. + prWatch: nil, + } + svc.SetGitHubService(ghSvc) + + tasks, err := svc.ListTasksNeedingPRWatch(ctx) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // Both sessions have branches and no watches (mock returns nil for all). + if len(tasks) < 1 { + t.Fatalf("expected at least 1 task, got %d", len(tasks)) + } + + // Verify results contain expected fields. + found := map[string]bool{} + for _, ti := range tasks { + found[ti.SessionID] = true + if ti.Owner != "myorg" { + t.Errorf("expected owner %q, got %q", "myorg", ti.Owner) + } + } + if !found["s1"] && !found["s2"] { + t.Error("expected at least one of s1 or s2 in results") + } + }) + + t.Run("excludes sessions on archived tasks", func(t *testing.T) { + testRepo := setupTestRepo(t) + seedSession(t, testRepo, "t1", "s1", "step1") + seedFullSession(t, testRepo, "t1", "s1", "feature-a", "repo1") + + // Archive the task. + if err := testRepo.ArchiveTask(ctx, "t1"); err != nil { + t.Fatalf("failed to archive task: %v", err) + } + + svc := createTestService(testRepo, newMockStepGetter(), newMockTaskRepo()) + ghSvc := &mockGitHubService{} + svc.SetGitHubService(ghSvc) + + tasks, err := svc.ListTasksNeedingPRWatch(ctx) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(tasks) != 0 { + t.Errorf("expected no tasks for archived task, got %d", len(tasks)) + } + }) + + t.Run("returns nil when github service is nil", func(t *testing.T) { + testRepo := setupTestRepo(t) + svc := createTestService(testRepo, newMockStepGetter(), newMockTaskRepo()) + + tasks, err := svc.ListTasksNeedingPRWatch(ctx) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(tasks) != 0 { + t.Errorf("expected empty result when github service is nil, got %d", len(tasks)) + } + }) +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index e80b344487..3ec63b1c7c 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -113,6 +113,7 @@ type repoStore interface { ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) ListActiveTaskSessionsByTaskID(ctx context.Context, taskID string) ([]*models.TaskSession, error) CreateTaskSessionWorktree(ctx context.Context, sessionWorktree *models.TaskSessionWorktree) error + ListSessionsWithBranches(ctx context.Context) ([]models.SessionBranchInfo, error) GetRepository(ctx context.Context, id string) (*models.Repository, error) UpdateRepository(ctx context.Context, repository *models.Repository) error GetExecutorProfile(ctx context.Context, id string) (*models.ExecutorProfile, error) diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 2e29ee2d52..3b9597137a 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -145,7 +145,7 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr zap.Error(launchErr)) return } - if prepExec != nil && prepExec.WorktreeBranch != "" { + if prepExec != nil { s.ensureSessionPRWatch(bgCtx, taskID, prepExec.SessionID, prepExec.WorktreeBranch) } }() @@ -211,9 +211,7 @@ func (s *Service) StartTaskWithSession(ctx context.Context, taskID string, sessi } } } - if execution.WorktreeBranch != "" { - go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) - } + go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) return execution, nil } @@ -311,9 +309,7 @@ func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, ag // 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 // or hadn't completed, this guarantees coverage. - if execution.WorktreeBranch != "" { - go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) - } + go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) return execution, nil } @@ -452,9 +448,7 @@ func (s *Service) postLaunchStart(ctx context.Context, taskID string, execution } } } - if execution.WorktreeBranch != "" { - go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) - } + go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) } // applyWorkflowAndPlanMode applies workflow step configuration and plan mode injection to a prompt. @@ -591,9 +585,7 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin zap.String("task_id", taskID), zap.String("session_id", sessionID)) - if execution.WorktreeBranch != "" { - go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) - } + go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch) return execution, nil } diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 11b94bbe0d..3bda12cdae 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -240,6 +240,14 @@ type TaskSessionWorktree struct { WorktreeBranch string `json:"worktree_branch,omitempty"` } +// SessionBranchInfo is a lightweight projection of a session with its worktree branch. +// Used by the PR watch reconciler to find sessions that may need PR watches. +type SessionBranchInfo struct { + SessionID string + TaskID string + Branch string +} + // TaskSession represents a persistent agent execution session for a task. // This replaces the in-memory TaskExecution tracking and survives backend restarts. type TaskSession struct { diff --git a/apps/backend/internal/task/repository/sqlite/session_branch.go b/apps/backend/internal/task/repository/sqlite/session_branch.go new file mode 100644 index 0000000000..d1e52b0b11 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/session_branch.go @@ -0,0 +1,38 @@ +package sqlite + +import ( + "context" + + "github.com/kandev/kandev/internal/task/models" +) + +// ListSessionsWithBranches returns sessions that have worktree branches +// on non-archived tasks. Used by the PR watch reconciler. +func (r *Repository) ListSessionsWithBranches(ctx context.Context) ([]models.SessionBranchInfo, error) { + rows, err := r.ro.QueryContext(ctx, ` + SELECT ts.id, ts.task_id, tsw.worktree_branch + FROM task_sessions ts + INNER JOIN tasks t ON t.id = ts.task_id + INNER JOIN task_session_worktrees tsw ON tsw.session_id = ts.id + WHERE t.archived_at IS NULL + AND tsw.worktree_branch != '' + AND tsw.deleted_at IS NULL + AND tsw.status = 'active' + GROUP BY ts.id + ORDER BY ts.started_at DESC + `) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + var result []models.SessionBranchInfo + for rows.Next() { + var info models.SessionBranchInfo + if err := rows.Scan(&info.SessionID, &info.TaskID, &info.Branch); err != nil { + return nil, err + } + result = append(result, info) + } + return result, rows.Err() +} diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 281be52d62..f73993b6b3 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -358,6 +358,7 @@ const ( ActionGitHubTaskPRUpdated = "github.task_pr.updated" // Notification ActionGitHubPRFeedbackNotify = "github.pr_feedback.notify" // Notification ActionGitHubNewReviewPRNotify = "github.new_review_pr.notify" // Notification + ActionGitHubTaskPRSync = "github.task_pr.sync" ActionGitHubStats = "github.stats" ActionGitHubCheckSessionPR = "github.check_session_pr" ) diff --git a/apps/web/components/github/pr-task-icon.tsx b/apps/web/components/github/pr-task-icon.tsx index 4793ddb404..6e87d85666 100644 --- a/apps/web/components/github/pr-task-icon.tsx +++ b/apps/web/components/github/pr-task-icon.tsx @@ -6,7 +6,7 @@ import { cn } from "@/lib/utils"; import { useAppStore } from "@/components/state-provider"; import type { TaskPR } from "@/lib/types/github"; -function getPRStatusColor(pr: TaskPR): string { +export function getPRStatusColor(pr: TaskPR): string { if (pr.state === "merged") return "text-purple-500"; if (pr.state === "closed") return "text-red-500"; if (pr.review_state === "changes_requested" || pr.checks_state === "failure") { @@ -37,7 +37,11 @@ export function PRTaskIcon({ taskId }: { taskId: string }) { return ( - + diff --git a/apps/web/components/github/pr-topbar-button.tsx b/apps/web/components/github/pr-topbar-button.tsx index 99f7a5588e..a96ea192a6 100644 --- a/apps/web/components/github/pr-topbar-button.tsx +++ b/apps/web/components/github/pr-topbar-button.tsx @@ -6,15 +6,10 @@ import { Button } from "@kandev/ui/button"; import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip"; import { useDockviewStore } from "@/lib/state/dockview-store"; import { useActiveTaskPR, useTaskPR } from "@/hooks/domains/github/use-task-pr"; +import { getPRStatusColor } from "@/components/github/pr-task-icon"; import { useAppStore } from "@/components/state-provider"; import type { TaskPR } from "@/lib/types/github"; -function prIconColor(state: string): string { - if (state === "merged") return "text-purple-500"; - if (state === "closed") return "text-red-500"; - return "text-green-600"; -} - function PRStatusIcon({ pr }: { pr: TaskPR }) { // Terminal states take priority if (pr.state === "merged") { @@ -55,7 +50,7 @@ export const PRTopbarButton = memo(function PRTopbarButton() { className="cursor-pointer gap-1.5 px-2" onClick={addPRPanel} > - + #{pr.pr_number} diff --git a/apps/web/components/task/changes-panel.tsx b/apps/web/components/task/changes-panel.tsx index 90cd91d0cf..91a64845d6 100644 --- a/apps/web/components/task/changes-panel.tsx +++ b/apps/web/components/task/changes-panel.tsx @@ -22,7 +22,7 @@ import { } from "./changes-panel-timeline"; import type { PRChangedFile } from "./changes-panel-timeline"; import { useChangesGitHandlers, useChangesDialogHandlers } from "./changes-panel-hooks"; -import { useActiveTaskPR, useTaskPRDetection } from "@/hooks/domains/github/use-task-pr"; +import { useActiveTaskPR } from "@/hooks/domains/github/use-task-pr"; import { usePRDiff } from "@/hooks/domains/github/use-pr-diff"; import { usePRCommits } from "@/hooks/domains/github/use-pr-commits"; import type { PRDiffFile } from "@/lib/types/github"; @@ -512,7 +512,7 @@ const ChangesPanel = memo(function ChangesPanel({ onOpenReview, }: ChangesPanelProps) { const isArchived = useIsTaskArchived(); - const { activeTaskId, activeSessionId, baseBranch, existingPrUrl } = useChangesPanelStoreData(); + const { activeSessionId, baseBranch, existingPrUrl } = useChangesPanelStoreData(); const git = useSessionGit(activeSessionId); const { toast } = useToast(); @@ -520,13 +520,6 @@ const ChangesPanel = memo(function ChangesPanel({ const { prDiffFiles, prCommitsList, hasPRFiles, hasPRCommits, prFiles } = useChangesPanelPRData(); const vcsDialogs = useVcsDialogs(); - // Periodically check for a PR when the session has a branch but no PR yet - useTaskPRDetection( - isArchived ? null : (activeTaskId ?? null), - isArchived ? null : (activeSessionId ?? null), - isArchived ? null : (git.branch ?? null), - ); - const baseBranchDisplay = useMemo(() => getBaseBranchDisplay(baseBranch), [baseBranch]); const unstagedFiles = useMemo(() => mapToChangedFiles(git.unstagedFiles), [git.unstagedFiles]); const stagedFiles = useMemo(() => mapToChangedFiles(git.stagedFiles), [git.stagedFiles]); diff --git a/apps/web/e2e/helpers/api-client.ts b/apps/web/e2e/helpers/api-client.ts index bb46b59fcd..54ea325e09 100644 --- a/apps/web/e2e/helpers/api-client.ts +++ b/apps/web/e2e/helpers/api-client.ts @@ -25,6 +25,7 @@ export type MockPR = { draft?: boolean; additions?: number; deletions?: number; + merged_at?: string; requested_reviewers?: Array<{ login: string; type: string }>; }; diff --git a/apps/web/e2e/tests/pr/pr-detection.spec.ts b/apps/web/e2e/tests/pr/pr-detection.spec.ts index defa0ab801..66146f81f1 100644 --- a/apps/web/e2e/tests/pr/pr-detection.spec.ts +++ b/apps/web/e2e/tests/pr/pr-detection.spec.ts @@ -5,24 +5,24 @@ import { SessionPage } from "../../pages/session-page"; test.describe("PR auto-detection", () => { /** - * Tests the full PR auto-detection pipeline: the backend detects a PR on - * GitHub (mock) and associates it with the task — without manually calling - * mockGitHubAssociateTaskPR. + * Tests the full PR auto-detection pipeline: the backend poller detects + * a PR on GitHub (mock) and associates it with the task -- without + * manually calling mockGitHubAssociateTaskPR. * - * Detection flow exercised: - * useTaskPRDetection (frontend, every 30s) - * -> github.check_session_pr (WS action) - * -> CheckSessionPR (backend) - * -> resolveTaskRepo -> resolvePRWatchBranch -> FindPRByBranch (mock) - * -> AssociatePRWithTask -> github.task_pr.updated (WS event) - * -> PR button appears in topbar + * Detection flow: + * Poller (backend, every 30s) + * -> reconcileWatches -> EnsurePRWatch (if missing) + * -> checkPRWatches -> FindPRByBranch (mock) + * -> AssociatePRWithTask -> github.task_pr.updated (WS event) + * -> PR icon appears on kanban card * * Prerequisites for detection to work: * - Repository has provider=github with owner/name set * - TaskRepository has checkout_branch set * - Mock GitHub has a PR matching the checkout branch + * - Agent has started (creates PR watch via ensureSessionPRWatch) */ - test("detects PR automatically via check_session_pr", async ({ + test("detects PR automatically via backend poller", async ({ testPage, apiClient, seedData, @@ -85,6 +85,7 @@ test.describe("PR auto-detection", () => { await kanban.goto(); // Move task to Working -> auto_start -> mock agent completes -> Done + // This creates a PR watch via ensureSessionPRWatch during task start await apiClient.moveTask(task.id, workflow.id, workingStep.id); await expect(kanban.taskCardInColumn("Auto-Detect PR Task", doneStep.id)).toBeVisible({ @@ -107,14 +108,16 @@ test.describe("PR auto-detection", () => { }, ]); - // --- Open the task session --- + // --- Open the task to trigger on-demand sync (github.task_pr.sync) --- await kanban.taskCardInColumn("Auto-Detect PR Task", doneStep.id).click(); await expect(testPage).toHaveURL(/\/[st]\//, { timeout: 15_000 }); const session = new SessionPage(testPage); await session.waitForLoad(); - // --- Verify PR button appears in topbar --- + // The useTaskPR hook triggers github.task_pr.sync which calls TriggerPRSync. + // Since a PR watch was created during task start (ensureSessionPRWatch), + // TriggerPRSync finds the PR via FindPRByBranch and associates it. await expect(session.prTopbarButton()).toBeVisible({ timeout: 60_000 }); await expect(session.prTopbarButton()).toContainText("#99"); }); @@ -158,7 +161,7 @@ test.describe("PR auto-detection", () => { await apiClient.mockGitHubReset(); await apiClient.mockGitHubSetUser("test-user"); - // No PRs added to mock GitHub — detection should find nothing + // No PRs added to mock GitHub -- backend poller should find nothing const task = await apiClient.createTask(seedData.workspaceId, "No PR Task", { workflow_id: workflow.id, @@ -189,6 +192,128 @@ test.describe("PR auto-detection", () => { // Verify no PR button await expect(session.prTopbarButton()).not.toBeVisible({ timeout: 10_000 }); }); + + /** + * Verifies the sync-before-delete bug fix: when a PR is merged on GitHub, + * the backend poller syncs the merged state to the TaskPR record before + * deleting the watch. The PR icon on the kanban card should update to + * show the merged state. + */ + test("updates PR status to merged", async ({ testPage, apiClient, seedData, backend }) => { + test.setTimeout(180_000); + + // --- Seed workflow: Inbox -> Working (auto_start -> Done) -> Done --- + const workflow = await apiClient.createWorkflow( + seedData.workspaceId, + "PR Merged Status Workflow", + ); + + const inboxStep = await apiClient.createWorkflowStep(workflow.id, "Inbox", 0); + const workingStep = await apiClient.createWorkflowStep(workflow.id, "Working", 1); + const doneStep = await apiClient.createWorkflowStep(workflow.id, "Done", 2); + + await apiClient.updateWorkflowStep(workingStep.id, { + prompt: 'e2e:message("done")\n{{task_prompt}}', + events: { + on_enter: [{ type: "auto_start_agent" }], + on_turn_complete: [{ type: "move_to_step", config: { step_id: doneStep.id } }], + }, + }); + + await apiClient.saveUserSettings({ + workspace_id: seedData.workspaceId, + workflow_filter_id: workflow.id, + enable_preview_on_click: false, + }); + + // --- Create repository with GitHub provider info --- + const repoDir = path.join(backend.tmpDir, "repos", "e2e-repo"); + const githubRepo = await apiClient.createRepository(seedData.workspaceId, repoDir, "main", { + name: "GitHub Merged PR Repo", + provider: "github", + provider_owner: "testorg", + provider_name: "testrepo", + }); + + // --- Setup mock GitHub --- + await apiClient.mockGitHubReset(); + await apiClient.mockGitHubSetUser("test-user"); + + // --- Create task with checkout_branch --- + const task = await apiClient.createTask(seedData.workspaceId, "Merged PR Task", { + workflow_id: workflow.id, + workflow_step_id: inboxStep.id, + agent_profile_id: seedData.agentProfileId, + repositories: [ + { + repository_id: githubRepo.id, + checkout_branch: "main", + }, + ], + }); + + // Navigate to kanban and subscribe to WS events + const kanban = new KanbanPage(testPage); + await kanban.goto(); + + // Move task to Working -> auto_start -> mock agent completes -> Done + await apiClient.moveTask(task.id, workflow.id, workingStep.id); + + await expect(kanban.taskCardInColumn("Merged PR Task", doneStep.id)).toBeVisible({ + timeout: 45_000, + }); + + // --- Add OPEN PR to mock GitHub --- + await apiClient.mockGitHubAddPRs([ + { + number: 101, + title: "Feature branch PR", + state: "open", + head_branch: "main", + base_branch: "develop", + author_login: "test-user", + repo_owner: "testorg", + repo_name: "testrepo", + additions: 20, + deletions: 3, + }, + ]); + + // --- Open the task to trigger on-demand sync and detect the PR --- + await kanban.taskCardInColumn("Merged PR Task", doneStep.id).click(); + await expect(testPage).toHaveURL(/\/[st]\//, { timeout: 15_000 }); + + const session = new SessionPage(testPage); + await session.waitForLoad(); + + // useTaskPR triggers github.task_pr.sync -> TriggerPRSync -> FindPRByBranch + await expect(session.prTopbarButton()).toBeVisible({ timeout: 60_000 }); + await expect(session.prTopbarButton()).toContainText("#101"); + + // --- Update mock PR to MERGED state --- + await apiClient.mockGitHubAddPRs([ + { + number: 101, + title: "Feature branch PR", + state: "merged", + head_branch: "main", + base_branch: "develop", + author_login: "test-user", + repo_owner: "testorg", + repo_name: "testrepo", + additions: 20, + deletions: 3, + merged_at: "2026-03-30T12:00:00Z", + }, + ]); + + // --- Verify PR topbar button updates to merged state --- + // The backend poller (every 30s) syncs the merged state and broadcasts + // via WS. The PRTopbarButton re-renders with the purple merged icon. + await expect(session.prTopbarButton().locator(".text-purple-500").first()).toBeVisible({ + timeout: 90_000, + }); + }); }); test.describe("PR external detection", () => { @@ -196,6 +321,9 @@ test.describe("PR external detection", () => { * Verifies that when a PR is created externally (e.g. via `gh pr create`) * after a task session is already running, the PR button appears in the * topbar and the Changes panel shows the PR files and commits. + * + * This uses mockGitHubAssociateTaskPR which directly creates the TaskPR + * record and publishes the WS event -- no poller needed. */ test("shows PR button and changes after external PR creation", async ({ testPage, @@ -296,7 +424,7 @@ test.describe("PR external detection", () => { }, ]); - // Associate PR with task + // Associate PR with task directly via mock API await apiClient.mockGitHubAssociateTaskPR({ task_id: featureTask.id, owner: "testorg", diff --git a/apps/web/hooks/domains/github/use-task-pr.ts b/apps/web/hooks/domains/github/use-task-pr.ts index 1ca384e0ca..fdd17ef46c 100644 --- a/apps/web/hooks/domains/github/use-task-pr.ts +++ b/apps/web/hooks/domains/github/use-task-pr.ts @@ -1,7 +1,7 @@ "use client"; import { useEffect, useCallback, useRef } from "react"; -import { listTaskPRs, getTaskPR } from "@/lib/api/domains/github-api"; +import { listTaskPRs } from "@/lib/api/domains/github-api"; import { getWebSocketClient } from "@/lib/ws/connection"; import { useAppStore } from "@/components/state-provider"; import type { TaskPR } from "@/lib/types/github"; @@ -32,25 +32,57 @@ export function useTaskPRs(taskIds: string[]) { return { byTaskId, loading }; } -/** Fetch a single task's PR association. */ +const SYNC_RETRY_DELAY = 5_000; // 5 seconds +const SYNC_MAX_RETRIES = 6; // Up to 30 seconds of retries + +/** Fetch a single task's PR association, with on-demand sync via WS. */ export function useTaskPR(taskId: string | null) { const pr = useAppStore((state) => (taskId ? (state.taskPRs.byTaskId[taskId] ?? null) : null)); const setTaskPR = useAppStore((state) => state.setTaskPR); + const retryRef = useRef(0); const refresh = useCallback(() => { if (!taskId) return; - getTaskPR(taskId, { cache: "no-store" }) - .then((response) => { - if (response) setTaskPR(taskId, response); + const client = getWebSocketClient(); + if (!client) return; + + client + .request("github.task_pr.sync", { task_id: taskId }) + .then((result) => { + if (result?.task_id) { + setTaskPR(taskId, result); + retryRef.current = 0; + } }) .catch(() => { - // Ignore - PR may not exist for this task + // Ignore - sync may fail if no watch exists }); }, [taskId, setTaskPR]); + // Reset retry count when taskId changes useEffect(() => { - if (!taskId || pr) return; + retryRef.current = 0; + }, [taskId]); + + useEffect(() => { + if (!taskId) return; + + // Always sync once when the task becomes active (freshness) refresh(); + + // If no PR in store yet, retry periodically until found + if (pr) return; + + const interval = setInterval(() => { + if (retryRef.current >= SYNC_MAX_RETRIES) { + clearInterval(interval); + return; + } + retryRef.current++; + refresh(); + }, SYNC_RETRY_DELAY); + + return () => clearInterval(interval); }, [taskId, pr, refresh]); return { pr, refresh } as { pr: TaskPR | null; refresh: () => void }; @@ -63,60 +95,3 @@ export function useActiveTaskPR(): TaskPR | null { return taskId ? (s.taskPRs.byTaskId[taskId] ?? null) : null; }); } - -const PR_DETECTION_INTERVAL = 30_000; // 30 seconds - -/** - * Periodically checks for a PR on the session's branch when no PR is - * associated with the task yet. Triggers a backend check that ensures - * a PR watch exists and tries to find the PR immediately. - */ -export function useTaskPRDetection( - taskId: string | null, - sessionId: string | null, - branch: string | null, -) { - const pr = useAppStore((s) => (taskId ? (s.taskPRs.byTaskId[taskId] ?? null) : null)); - const setTaskPR = useAppStore((s) => s.setTaskPR); - const checkingRef = useRef(false); - - useEffect(() => { - if (!taskId || !sessionId || !branch || pr) return; - - const check = () => { - if (checkingRef.current) return; - checkingRef.current = true; - - const client = getWebSocketClient(); - if (!client) { - checkingRef.current = false; - return; - } - - client - .request<{ found?: boolean }>("github.check_session_pr", { - task_id: taskId, - session_id: sessionId, - }) - .then((resp) => { - if (resp?.found) { - // PR was found and associated — fetch the full TaskPR record - getTaskPR(taskId, { cache: "no-store" }) - .then((taskPr) => { - if (taskPr) setTaskPR(taskId, taskPr); - }) - .catch(() => {}); - } - }) - .catch(() => {}) - .finally(() => { - checkingRef.current = false; - }); - }; - - // Check immediately, then on interval - check(); - const interval = setInterval(check, PR_DETECTION_INTERVAL); - return () => clearInterval(interval); - }, [taskId, sessionId, branch, pr, setTaskPR]); -}