diff --git a/apps/backend/AGENTS.md b/apps/backend/AGENTS.md index 1b625666a6..35fddf95b9 100644 --- a/apps/backend/AGENTS.md +++ b/apps/backend/AGENTS.md @@ -217,7 +217,7 @@ Client (WS) ← Orchestrator ← Lifecycle Manager ←──── stream update `internal/common/subproc`; choose `interactive`, `lifecycle`, or `background`. - **Event-bus wildcard parity:** New NATS wildcard subscriptions must verify equivalent `MemoryEventBus` semantics in `go test ./internal/events/bus`. - **Repository provider identity:** Provider-backed repositories are keyed by workspace, provider, normalized `provider_host` origin, full owner/namespace, and name. Persist `provider_host` when importing or resolving a remote; do not infer self-managed GitLab rows from owner/name alone. Legacy rows with an empty host have unknown identity and must fail closed for provider write/link operations. -- **PR status sync coverage:** A `github_pr_watches` row is unique per (session, repository, branch) and gets re-pointed when the session's branch moves on, so watches are not a complete index of a task's PRs; a watch-only sweep leaves every PR from an earlier branch frozen at its last-observed state, which the UI renders as an open, mergeable PR long after it merged. `service_pr_unwatched.go` reconciles the `github_task_prs` rows no watch points at, and deliberately writes **lifecycle fields only** (`state` / `merged_at` / `closed_at`) — check and review aggregates belong to the watch-driven sync that actually fetches reviews and check runs, and the per-PR REST status marks its counts populated even with nothing to count, so widening this path to aggregates persists 0/0 over good data (it broke six PR-popover e2e specs exactly that way). Terminal and detached rows are excluded so the set stays bounded. +- **PR status sync coverage:** A `github_pr_watches` row is unique per (session, repository, branch) and gets re-pointed when the session's branch moves on, so watches are not a complete index of a task's PRs; a watch-only sweep leaves every PR from an earlier branch frozen at its last-observed state, which the UI renders as an open, mergeable PR long after it merged. `service_pr_unwatched.go` reconciles the `github_task_prs` rows no watch points at, and deliberately writes **lifecycle fields only** (`state` / `merged_at` / `closed_at`) — check and review aggregates belong to the watch-driven sync that actually fetches reviews and check runs, and the per-PR REST status marks its counts populated even with nothing to count, so widening this path to aggregates persists 0/0 over good data (it broke six PR-popover e2e specs exactly that way). Terminal and detached rows are excluded so the set stays bounded. **The backend is the only writer of `github_task_prs`:** the PR-feedback read makes the same upstream calls the status sync does (`GetPR` + `ListPRReviews` + `ListCheckRuns`, plus comments), so it persists through `SyncTaskPR` (`service_pr_feedback_sync.go`) and lets `github.task_pr.updated` converge every surface. The frontend must never patch PR lifecycle fields into its own store copy — that write cannot reach the DB, so it desyncs the panel from the kanban card and evaporates on reload (it also masked the sync hole above for months). Both derivations share `newPRStatus` so they cannot drift; the persist rides *inside* the TTL cache's fetch closure, so a cache hit costs no write; failures are logged and swallowed because this is a read endpoint; and the row lookup is workspace-scoped (`ListTaskPRsByPRNumber`) to match the credential scope the read was authorized under. Finally, `merged` is the one state `SyncTaskPR` refuses to leave (`resolveTerminalMergeState`) since GitHub cannot un-merge a PR — do **not** generalize that into a `merged > closed > open` rank, because `closed -> open` is a real transition (a PR can be reopened) and ranking it as a regression pins reopened PRs closed forever. - **Execution access:** Workspace-oriented handlers (files, shell, inference, ports, vscode, LSP) MUST use `GetOrEnsureExecution(ctx, sessionID)` — it recovers from backend restarts by creating executions on-demand. Only use `GetExecutionBySessionID` for operations that require a running agent process (prompt, cancel, mode). - **Task lifecycle events:** Any code path that mutates a task row must publish via the event bus (`task.created` / `task.updated` / `task.deleted`) — either by going through `Service.CreateTask` / `UpdateTask` / `DeleteTask` / `ArchiveTask`, or by calling `publishTaskEvent` (or one of the `Publish*` helpers in `service_events.go`) directly. Walking `repository.TaskRepository` straight bypasses event publishing and breaks WS-driven UI like the All-Workflows kanban view. `HandoffService`'s cascade methods learned this the hard way — they now require a `TaskEventPublisher` wired via `SetTaskEventPublisher`. New cascade / bulk / cleanup paths must follow the same pattern. **Workflow steps** follow the same rule with their own publisher: step create/update/delete publish `workflow_step.created` / `.updated` / `.deleted` through `internal/workflow/stepevents.Publisher` (the WS gateway fans them out; the orchestrator re-evaluates queue admission off `.updated`). Both mutation surfaces — REST/WS in `internal/workflow/handlers`, MCP in `internal/mcp/handlers` — share that publisher and differ only in the source label they pass, so a new step-mutation path uses it rather than hand-rolling the payload; promoting a start step demotes the previous one, so also publish `.updated` for every entry in `DemotedStartSteps`. - **Testing:** Prefer `testing/synctest` (Go 1.24+) over `time.Sleep` for time-dependent tests. Use `synctest.Test` to wrap tests with tickers or timeouts — it advances fake time instantly when all goroutines are idle. When `synctest` is not feasible (e.g., tests spawning external processes like `git`), use channel-based synchronization (`<-started`, non-blocking `select`) instead of sleep-based waits. Reserve `time.Sleep` only for integration tests that need real subprocess execution time. diff --git a/apps/backend/internal/github/client_helpers.go b/apps/backend/internal/github/client_helpers.go index 1893e3d01e..3f9390b1e0 100644 --- a/apps/backend/internal/github/client_helpers.go +++ b/apps/backend/internal/github/client_helpers.go @@ -126,6 +126,20 @@ func getPRStatus(ctx context.Context, c Client, owner, repo string, number int) if checks == nil { checks = []CheckRun{} } + return newPRStatus(pr, reviews, checks), nil +} + +// newPRStatus derives the PRStatus rollup from the three upstream reads every +// REST-shaped path already makes: the PR itself, its reviews, and its check +// runs. getPRStatus and the PR-feedback path both land here so the two cannot +// drift — feedback fetches a strict superset (it adds comments), so the same +// derivation is valid for both. +// +// Both Populated flags are set because this path genuinely counted: a zero +// here is a real "no checks" / "no reviews" answer, not "I didn't look". +// UnresolvedReviewThreadsPopulated stays false — neither REST caller fetches +// review threads, so SyncTaskPR must preserve whatever the GraphQL path stored. +func newPRStatus(pr *PR, reviews []PRReview, checks []CheckRun) *PRStatus { reviewState, pendingReviewCount := deriveReviewSyncState(pr, reviews) total, passing := countCheckResults(checks) return &PRStatus{ @@ -146,7 +160,7 @@ func getPRStatus(ctx context.Context, c Client, owner, repo string, number int) // from the batched-GraphQL path that only carries rollup state. ChecksPopulated: true, ReviewCountsPopulated: true, - }, nil + } } // reviewSample is a normalized review row consumed by latestReviewStateByAuthor. diff --git a/apps/backend/internal/github/constants.go b/apps/backend/internal/github/constants.go index 02d7f8d57b..2a4badf0eb 100644 --- a/apps/backend/internal/github/constants.go +++ b/apps/backend/internal/github/constants.go @@ -4,6 +4,7 @@ package github const ( prStateMerged = "merged" prStateClosed = "closed" + prStateOpen = "open" ) // PR review states from the GitHub API. diff --git a/apps/backend/internal/github/poller_test.go b/apps/backend/internal/github/poller_test.go index 3f477bfde5..d5e6ef603a 100644 --- a/apps/backend/internal/github/poller_test.go +++ b/apps/backend/internal/github/poller_test.go @@ -33,10 +33,13 @@ func setupPollerTest(t *testing.T) (*Poller, *Service, *MockClient, *Store) { sqlxDB := sqlx.NewDb(rawDB, "sqlite3") t.Cleanup(func() { _ = sqlxDB.Close() }) - // Minimal tasks schema: ListActivePRWatches only joins on id and filters by archived_at. + // Minimal tasks schema: ListActivePRWatches only joins on id and filters by + // archived_at; workspace_id is carried because the workspace-scoped task-PR + // lookups (ListTaskIDsByPRNumber, ListTaskPRsByPRNumber) join on it. if _, err := sqlxDB.Exec(` CREATE TABLE tasks ( id TEXT PRIMARY KEY, + workspace_id TEXT, archived_at DATETIME )`); err != nil { t.Fatalf("create tasks table: %v", err) @@ -66,7 +69,8 @@ func seedTask(t *testing.T, store *Store, taskID string, archived bool) { if archived { archivedAt = time.Now().UTC() } - if _, err := store.db.Exec(`INSERT INTO tasks (id, archived_at) VALUES (?, ?)`, taskID, archivedAt); err != nil { + if _, err := store.db.Exec(`INSERT INTO tasks (id, workspace_id, archived_at) VALUES (?, ?, ?)`, + taskID, testWorkspaceID, archivedAt); err != nil { t.Fatalf("seed task %s: %v", taskID, err) } } diff --git a/apps/backend/internal/github/service_automation_auth.go b/apps/backend/internal/github/service_automation_auth.go index 3978b1d8ae..6da8daea31 100644 --- a/apps/backend/internal/github/service_automation_auth.go +++ b/apps/backend/internal/github/service_automation_auth.go @@ -38,7 +38,7 @@ func (s *Service) GetPRFeedbackForAutomation( if err != nil { return nil, err } - return s.getPRFeedback(ctx, resolved.Client, resolved.CacheScope, owner, repo, number) + return s.getPRFeedback(ctx, resolved.Client, resolved.CacheScope, workspaceID, owner, repo, number) } func (s *Service) MergePRForAutomation( diff --git a/apps/backend/internal/github/service_pr.go b/apps/backend/internal/github/service_pr.go index 249fabb4c3..f7d1ffa6e1 100644 --- a/apps/backend/internal/github/service_pr.go +++ b/apps/backend/internal/github/service_pr.go @@ -309,7 +309,8 @@ func (s *Service) GetPRFeedback(ctx context.Context, owner, repo string, number if s.client == nil { return nil, fmt.Errorf("github client not available") } - return s.getPRFeedback(ctx, s.client, "legacy", owner, repo, number) + // No workspace to scope a persist to; this legacy entry point reads only. + return s.getPRFeedback(ctx, s.client, "legacy", "", owner, repo, number) } func (s *Service) GetPRFeedbackForWorkspace( @@ -322,11 +323,11 @@ func (s *Service) GetPRFeedbackForWorkspace( if err != nil { return nil, err } - return s.getPRFeedback(ctx, resolved.Client, resolved.CacheScope, owner, repo, number) + return s.getPRFeedback(ctx, resolved.Client, resolved.CacheScope, workspaceID, owner, repo, number) } func (s *Service) getPRFeedback( - ctx context.Context, client Client, cacheScope, owner, repo string, number int, + ctx context.Context, client Client, cacheScope, workspaceID, owner, repo string, number int, ) (*PRFeedback, error) { // Detach the upstream call's context from the singleflight leader: a // cancelled leader (user closes the tab) would otherwise return its @@ -340,7 +341,17 @@ func (s *Service) getPRFeedback( defer cancelFetch() key := scopedCacheKey(cacheScope, prStatusCacheKey(owner, repo, number)) v, err := s.prFeedbackCache.doOrFetch(key, func() (any, error) { - return client.GetPRFeedback(fetchCtx, owner, repo, number) + feedback, fetchErr := client.GetPRFeedback(fetchCtx, owner, repo, number) + if fetchErr != nil { + return nil, fetchErr + } + // Persist inside the fetch closure, not around doOrFetch: this runs + // once per real upstream call (TTL miss, singleflight leader) rather + // than on every panel render served from cache, and it inherits + // fetchCtx so a client disconnecting mid-flight can't abort the write + // that its own fetch just earned. + s.persistPRFeedbackState(fetchCtx, workspaceID, feedback) + return feedback, nil }) if err != nil { return nil, err diff --git a/apps/backend/internal/github/service_pr_feedback_sync.go b/apps/backend/internal/github/service_pr_feedback_sync.go new file mode 100644 index 0000000000..425bcff245 --- /dev/null +++ b/apps/backend/internal/github/service_pr_feedback_sync.go @@ -0,0 +1,65 @@ +package github + +import ( + "context" + + "go.uber.org/zap" +) + +// persistPRFeedbackState reconciles the workspace's github_task_prs rows for a +// PR from the live data a feedback fetch just retrieved. +// +// The feedback read is the same server-side GitHub fetch the status sync makes +// — GetPR + ListPRReviews + ListCheckRuns, plus comments — so the authoritative +// PR state is already in hand here. It used to be shipped to the browser and +// re-derived there (useSyncLivePRState in pr-detail-panel.tsx), which patched +// only the Zustand copy: opening a PR's detail panel turned that one tab purple +// while the kanban card, the boot payload and every other surface kept reading +// state=open from the stale row, and a reload snapped it back. Writing through +// SyncTaskPR here makes the backend the single writer and lets the resulting +// github.task_pr.updated event converge every surface at once. +// +// Best-effort by contract: this hangs off a read endpoint, so every failure is +// logged at Debug and swallowed. A dead store must never turn a working PR +// panel into an error. +// +// Scoping: rows are looked up inside workspaceID only, matching the credential +// scope the caller already passed ensureRepositoryInWorkspaceScope with. The +// legacy workspace-less GetPRFeedback entry point passes "" and is skipped +// rather than being allowed to write across workspaces. +func (s *Service) persistPRFeedbackState(ctx context.Context, workspaceID string, feedback *PRFeedback) { + if s == nil || s.store == nil || workspaceID == "" { + return + } + if feedback == nil || feedback.PR == nil || feedback.PR.Number == 0 { + return + } + pr := feedback.PR + rows, err := s.store.ListTaskPRsByPRNumber(ctx, workspaceID, pr.RepoOwner, pr.RepoName, pr.Number) + if err != nil { + s.logger.Debug("list task PRs for feedback persist failed", + zap.String("workspace_id", workspaceID), zap.String("owner", pr.RepoOwner), + zap.String("repo", pr.RepoName), zap.Int("pr_number", pr.Number), zap.Error(err)) + return + } + if len(rows) == 0 { + return + } + status := newPRStatus(pr, feedback.Reviews, feedback.Checks) + // Several tasks in a workspace can link the same PR; each owns its own row. + // SyncTaskPR re-narrows by (task, owner, repo, number) internally. + seen := make(map[string]struct{}, len(rows)) + for _, tp := range rows { + if tp == nil || tp.TaskID == "" { + continue + } + if _, dup := seen[tp.TaskID]; dup { + continue + } + seen[tp.TaskID] = struct{}{} + if syncErr := s.SyncTaskPR(ctx, tp.TaskID, status); syncErr != nil { + s.logger.Debug("task PR sync from PR feedback failed", + zap.String("task_id", tp.TaskID), zap.Int("pr_number", pr.Number), zap.Error(syncErr)) + } + } +} diff --git a/apps/backend/internal/github/service_pr_feedback_sync_test.go b/apps/backend/internal/github/service_pr_feedback_sync_test.go new file mode 100644 index 0000000000..1cff1e92ea --- /dev/null +++ b/apps/backend/internal/github/service_pr_feedback_sync_test.go @@ -0,0 +1,359 @@ +package github + +import ( + "context" + "testing" + "time" +) + +const ( + feedbackSyncOwner = "owner" + feedbackSyncRepo = "repo" + feedbackSyncHead = "sha-feedback" +) + +// seedFeedbackTaskPR inserts a linked TaskPR row for the feedback-sync tests, +// applying `mutate` so each test can dirty the fields it cares about. +func seedFeedbackTaskPR(t *testing.T, store *Store, taskID string, number int, mutate func(*TaskPR)) { + t.Helper() + now := time.Now().UTC() + staleSync := now.Add(-1 * time.Hour) + tp := &TaskPR{ + TaskID: taskID, + WorkspaceID: testWorkspaceID, + RepositoryID: "repo-1", + Owner: feedbackSyncOwner, + Repo: feedbackSyncRepo, + PRNumber: number, + PRURL: "https://github.com/owner/repo/pull/2408", + PRTitle: "Collapse the dual writers", + HeadBranch: "feature/collapse", + BaseBranch: "main", + State: prStateOpen, + CreatedAt: now.Add(-3 * time.Hour), + LastSyncedAt: &staleSync, + } + if mutate != nil { + mutate(tp) + } + if err := store.CreateTaskPR(context.Background(), tp); err != nil { + t.Fatalf("seed feedback task PR: %v", err) + } +} + +// addMergedFeedbackPR registers a merged PR the workspace's credentials can +// reach, so GetPRFeedbackForWorkspace resolves a client and returns live data. +func addMergedFeedbackPR(mockClient *MockClient, number int, mergedAt time.Time) { + mockClient.AddRepos("test-user", []GitHubRepo{{ + Owner: feedbackSyncOwner, Name: feedbackSyncRepo, + FullName: feedbackSyncOwner + "/" + feedbackSyncRepo, + }}) + mockClient.AddPR(&PR{ + Number: number, Title: "Collapse the dual writers", State: prStateMerged, + HTMLURL: "https://github.com/owner/repo/pull/2408", + HeadBranch: "feature/collapse", BaseBranch: "main", HeadSHA: feedbackSyncHead, + RepoOwner: feedbackSyncOwner, RepoName: feedbackSyncRepo, + CreatedAt: mergedAt.Add(-3 * time.Hour), UpdatedAt: mergedAt, + MergedAt: &mergedAt, ClosedAt: &mergedAt, + }) +} + +// The bug this collapses: GetPRFeedbackForWorkspace already holds an +// authoritative live *PR server-side, but used to return it without persisting. +// The browser then patched only its in-memory Zustand copy, so opening a PR's +// detail panel turned that tab purple while github_task_prs still said +// state=open — every other surface disagreed, and a reload snapped it back. +// Confirmed on a live instance for kdlbs/kandev#2408. +// +// This covers the stale-row direction (stored open, live merged). The inverse +// — stored merged, live read stale at open — is covered one layer down by +// TestSyncTaskPR_DoesNotRegressAMergedRow, deliberately: persistPRFeedbackState +// has no state logic of its own, it hands every row to the real SyncTaskPR, so +// asserting the guard at its own seam is where a break would actually surface. +func TestGetPRFeedbackForWorkspace_PersistsMergedStateOverStaleRow(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + seedFeedbackTaskPR(t, store, "task-1", 2408, nil) + + mergedAt := time.Now().UTC().Add(-10 * time.Minute) + addMergedFeedbackPR(mockClient, 2408, mergedAt) + + // The published event is what converges every other surface, so subscribe + // before the fetch and use it as the join. + updated := subscribeTaskPRUpdated(t, svc) + + if _, err := svc.GetPRFeedbackForWorkspace( + ctx, testWorkspaceID, "", feedbackSyncOwner, feedbackSyncRepo, 2408, + ); err != nil { + t.Fatalf("GetPRFeedbackForWorkspace: %v", err) + } + + event := awaitTaskPRUpdated(t, updated) + if event == nil || event.State != prStateMerged { + t.Fatalf("published event state = %+v, want %q", event, prStateMerged) + } + + got, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 2408) + if err != nil { + t.Fatalf("GetTaskPRByRepoAndNumber: %v", err) + } + if got == nil { + t.Fatal("expected the task PR row to still exist") + } + if got.State != prStateMerged { + t.Errorf("persisted state = %q, want %q", got.State, prStateMerged) + } + if got.MergedAt == nil { + t.Error("persisted merged_at must be set once the feedback fetch observes the merge") + } +} + +// A feedback fetch computes checks and review counts (it makes the same +// GetPR + ListPRReviews + ListCheckRuns calls the REST status sync does) but +// never fetches review threads. SyncTaskPR's Populated flags must therefore +// carry unresolved_review_threads through untouched, while the two counts it +// did compute come from the real lists rather than being reset to zero. +func TestGetPRFeedbackForWorkspace_PartialSyncKeepsUncomputedAggregates(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + seedFeedbackTaskPR(t, store, "task-1", 2408, func(tp *TaskPR) { + tp.ReviewCount = 3 + tp.ChecksTotal = 27 + tp.ChecksPassing = 22 + // Only the GraphQL path can compute this one. + tp.UnresolvedReviewThreads = 4 + }) + + mergedAt := time.Now().UTC().Add(-10 * time.Minute) + addMergedFeedbackPR(mockClient, 2408, mergedAt) + mockClient.AddReviews(feedbackSyncOwner, feedbackSyncRepo, 2408, []PRReview{ + {ID: 1, Author: "octocat", State: reviewStateApproved, CreatedAt: mergedAt}, + }) + mockClient.AddCheckRuns(feedbackSyncOwner, feedbackSyncRepo, feedbackSyncHead, []CheckRun{ + {Name: "build", Status: checkStatusCompleted, Conclusion: checkConclusionSuccess}, + {Name: "lint", Status: checkStatusCompleted, Conclusion: checkConclusionSuccess}, + }) + + updated := subscribeTaskPRUpdated(t, svc) + if _, err := svc.GetPRFeedbackForWorkspace( + ctx, testWorkspaceID, "", feedbackSyncOwner, feedbackSyncRepo, 2408, + ); err != nil { + t.Fatalf("GetPRFeedbackForWorkspace: %v", err) + } + awaitTaskPRUpdated(t, updated) + + got, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 2408) + if err != nil || got == nil { + t.Fatalf("GetTaskPRByRepoAndNumber: %v (row %+v)", err, got) + } + if got.UnresolvedReviewThreads != 4 { + t.Errorf("unresolved_review_threads = %d, want the stored 4 preserved — "+ + "the feedback path never fetches review threads", got.UnresolvedReviewThreads) + } + // Counted from the real lists, so neither is allowed to land on zero. + if got.ReviewCount != 1 { + t.Errorf("review_count = %d, want 1 approving reviewer", got.ReviewCount) + } + if got.ChecksTotal != 2 || got.ChecksPassing != 2 { + t.Errorf("checks = %d/%d, want 2/2", got.ChecksPassing, got.ChecksTotal) + } +} + +// The feedback path is TTL-cached and singleflight-coalesced; the persist must +// ride the upstream fetch, not the read. Otherwise every render of an open +// detail panel writes the row and republishes an event, and the WS handler +// applying that event re-renders the panel — a write amplifier on a read path. +func TestGetPRFeedbackForWorkspace_CachedResponseDoesNotRewriteRow(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + seedFeedbackTaskPR(t, store, "task-1", 2408, nil) + + mergedAt := time.Now().UTC().Add(-10 * time.Minute) + addMergedFeedbackPR(mockClient, 2408, mergedAt) + + updated := subscribeTaskPRUpdated(t, svc) + for i := range 3 { + if _, err := svc.GetPRFeedbackForWorkspace( + ctx, testWorkspaceID, "", feedbackSyncOwner, feedbackSyncRepo, 2408, + ); err != nil { + t.Fatalf("GetPRFeedbackForWorkspace call %d: %v", i, err) + } + } + awaitTaskPRUpdated(t, updated) + + first, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 2408) + if err != nil || first == nil { + t.Fatalf("GetTaskPRByRepoAndNumber: %v (row %+v)", err, first) + } + writtenAt := first.LastSyncedAt + if writtenAt == nil { + t.Fatal("expected the first fetch to stamp last_synced_at") + } + + // Cache hits must not touch the row at all. + if _, err := svc.GetPRFeedbackForWorkspace( + ctx, testWorkspaceID, "", feedbackSyncOwner, feedbackSyncRepo, 2408, + ); err != nil { + t.Fatalf("GetPRFeedbackForWorkspace (cached): %v", err) + } + after, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 2408) + if err != nil || after == nil { + t.Fatalf("GetTaskPRByRepoAndNumber: %v (row %+v)", err, after) + } + if !timeEqual(after.LastSyncedAt, writtenAt) { + t.Errorf("cached feedback rewrote the row: last_synced_at %v -> %v", writtenAt, after.LastSyncedAt) + } + // ...and must not republish either, or the WS handler applying the event + // re-renders the panel that triggered it. MemoryEventBus delivers to + // subscribers synchronously (see MemoryEventBus.Publish), so any event the + // call above caused is already in the channel — a non-blocking read is + // exact here, with no sleep to go flaky under load. + // + // last_synced_at above is the assertion that actually catches a persist + // moved outside the fetch closure; this one is defence in depth. A + // redundant sync writes the row but re-derives identical field values, so + // SyncTaskPR's `changed` guard already suppresses the event — this pins + // that no-WS-traffic-on-a-cache-hit property against a future change that + // relaxes that guard. + select { + case extra := <-updated: + t.Errorf("cached feedback republished github.task_pr.updated: %+v", extra) + default: + } +} + +// Rows belonging to another workspace must stay untouched: the feedback read +// was authorized against one workspace's credentials only, so the persist +// cannot reach past that boundary just because a PR number matches. +func TestGetPRFeedbackForWorkspace_DoesNotWriteOtherWorkspacesRows(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + if _, err := store.db.Exec( + `INSERT INTO tasks (id, workspace_id, archived_at) VALUES (?, ?, NULL)`, + "task-other", "ws-other"); err != nil { + t.Fatalf("seed foreign task: %v", err) + } + seedFeedbackTaskPR(t, store, "task-1", 2408, nil) + seedFeedbackTaskPR(t, store, "task-other", 2408, func(tp *TaskPR) { + tp.WorkspaceID = "ws-other" + tp.RepositoryID = "repo-other" + }) + + mergedAt := time.Now().UTC().Add(-10 * time.Minute) + addMergedFeedbackPR(mockClient, 2408, mergedAt) + + updated := subscribeTaskPRUpdated(t, svc) + if _, err := svc.GetPRFeedbackForWorkspace( + ctx, testWorkspaceID, "", feedbackSyncOwner, feedbackSyncRepo, 2408, + ); err != nil { + t.Fatalf("GetPRFeedbackForWorkspace: %v", err) + } + awaitTaskPRUpdated(t, updated) + + foreign, err := store.GetTaskPRByRepoAndNumber(ctx, "task-other", "repo-other", 2408) + if err != nil || foreign == nil { + t.Fatalf("GetTaskPRByRepoAndNumber: %v (row %+v)", err, foreign) + } + if foreign.State != prStateOpen { + t.Errorf("foreign-workspace row state = %q, want it left at %q", foreign.State, prStateOpen) + } +} + +// A failing store must not turn a working PR detail panel into an error: this +// is a read endpoint that gained a write, so the write is best-effort. +func TestGetPRFeedbackForWorkspace_PersistFailureDoesNotFailTheRead(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + seedFeedbackTaskPR(t, store, "task-1", 2408, nil) + addMergedFeedbackPR(mockClient, 2408, time.Now().UTC().Add(-10*time.Minute)) + + // Drop the table the persist path reads to force a store error underneath it. + if _, err := store.db.Exec(`DROP TABLE tasks`); err != nil { + t.Fatalf("drop tasks table: %v", err) + } + + feedback, err := svc.GetPRFeedbackForWorkspace( + ctx, testWorkspaceID, "", feedbackSyncOwner, feedbackSyncRepo, 2408, + ) + if err != nil { + t.Fatalf("persist failure must not fail the feedback read: %v", err) + } + if feedback == nil || feedback.PR == nil || feedback.PR.State != prStateMerged { + t.Fatalf("expected the live feedback to still be returned, got %+v", feedback) + } +} + +// SyncTaskPR is now the only writer, so the one invariant the deleted +// client-side hook was protecting has to live here: GitHub cannot un-merge a +// PR, so a stale live read reporting "open" for a merged row must not regress +// it. Only "merged" is terminal — see the reopen case below. +func TestSyncTaskPR_DoesNotRegressAMergedRow(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + mergedAt := time.Now().UTC().Add(-1 * time.Hour) + seedFeedbackTaskPR(t, store, "task-1", 2408, func(tp *TaskPR) { + tp.State = prStateMerged + tp.MergedAt = &mergedAt + tp.ClosedAt = &mergedAt + }) + + // A stale read — an eventually-consistent REST reply, or a fetch that + // started before another writer observed the merge. + stale := &PRStatus{PR: &PR{ + Number: 2408, State: prStateOpen, Title: "Collapse the dual writers", + RepoOwner: feedbackSyncOwner, RepoName: feedbackSyncRepo, BaseBranch: "main", + }} + if err := svc.SyncTaskPR(ctx, "task-1", stale); err != nil { + t.Fatalf("SyncTaskPR: %v", err) + } + + got, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 2408) + if err != nil || got == nil { + t.Fatalf("GetTaskPRByRepoAndNumber: %v (row %+v)", err, got) + } + if got.State != prStateMerged { + t.Errorf("state = %q, want the terminal %q preserved", got.State, prStateMerged) + } + if got.MergedAt == nil { + t.Error("merged_at must survive a stale non-merged read") + } +} + +// The counterpart: closed -> open is a real GitHub transition (a PR can be +// reopened), so the guard must be merged-only. A blanket merged > closed > open +// rank — what the frontend hook applied — would pin every reopened PR closed. +func TestSyncTaskPR_AllowsReopeningAClosedRow(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + closedAt := time.Now().UTC().Add(-1 * time.Hour) + seedFeedbackTaskPR(t, store, "task-1", 2408, func(tp *TaskPR) { + tp.State = prStateClosed + tp.ClosedAt = &closedAt + }) + + reopened := &PRStatus{PR: &PR{ + Number: 2408, State: prStateOpen, Title: "Collapse the dual writers", + RepoOwner: feedbackSyncOwner, RepoName: feedbackSyncRepo, BaseBranch: "main", + }} + if err := svc.SyncTaskPR(ctx, "task-1", reopened); err != nil { + t.Fatalf("SyncTaskPR: %v", err) + } + + got, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 2408) + if err != nil || got == nil { + t.Fatalf("GetTaskPRByRepoAndNumber: %v (row %+v)", err, got) + } + if got.State != prStateOpen { + t.Errorf("state = %q, want %q — reopening must not be treated as a regression", got.State, prStateOpen) + } + if got.ClosedAt != nil { + t.Error("closed_at must clear when the PR reopens") + } +} diff --git a/apps/backend/internal/github/service_pr_watch.go b/apps/backend/internal/github/service_pr_watch.go index 20f81f1012..b261f8cf15 100644 --- a/apps/backend/internal/github/service_pr_watch.go +++ b/apps/backend/internal/github/service_pr_watch.go @@ -700,7 +700,25 @@ type taskPRSyncState struct { checksTotal, checksPassing int unresolved, reviewCount, pendingReviewCount int requiredReviews *int - baseBranch, mergeableState string + baseBranch, mergeableState, state string + mergedAt, closedAt *time.Time +} + +// resolveTerminalMergeState keeps a row that already reached "merged" there. +// GitHub cannot un-merge a PR, so a live read reporting anything else for a +// merged row is stale — an eventually-consistent REST reply, or a fetch that +// started before a concurrent writer observed the merge. Without this, two +// writers racing the same PR could flip its badge back to green. +// +// Only "merged" is guarded. The PR detail panel used to apply a full +// merged > closed > open rank client-side; that is deliberately not ported, +// because closed -> open is a real transition (a PR can be reopened on GitHub) +// and ranking it as a regression would pin every reopened PR closed forever. +func resolveTerminalMergeState(tp *TaskPR, pr *PR) (string, *time.Time, *time.Time) { + if tp.State == prStateMerged && pr.State != prStateMerged { + return tp.State, tp.MergedAt, tp.ClosedAt + } + return pr.State, pr.MergedAt, pr.ClosedAt } func (s *Service) prepareTaskPRSyncState(ctx context.Context, tp *TaskPR, status *PRStatus) taskPRSyncState { @@ -755,10 +773,12 @@ func (s *Service) prepareTaskPRSyncState(ctx context.Context, tp *TaskPR, status if status.PR.Draft { nextMergeableState = "draft" } + nextState, nextMergedAt, nextClosedAt := resolveTerminalMergeState(tp, status.PR) return taskPRSyncState{ checksTotal: nextChecksTotal, checksPassing: nextChecksPassing, unresolved: nextUnresolved, reviewCount: nextReviewCount, pendingReviewCount: nextPendingReviewCount, requiredReviews: nextRequiredReviews, baseBranch: nextBaseBranch, mergeableState: nextMergeableState, + state: nextState, mergedAt: nextMergedAt, closedAt: nextClosedAt, } } @@ -780,7 +800,7 @@ func (s *Service) SyncTaskPR(ctx context.Context, taskID string, status *PRStatu } next := s.prepareTaskPRSyncState(ctx, tp, status) - changed := tp.State != status.PR.State || + changed := tp.State != next.state || tp.PRTitle != status.PR.Title || tp.Additions != status.PR.Additions || tp.Deletions != status.PR.Deletions || @@ -794,15 +814,15 @@ func (s *Service) SyncTaskPR(ctx context.Context, taskID string, status *PRStatu tp.ChecksPassing != next.checksPassing || tp.UnresolvedReviewThreads != next.unresolved || tp.BaseBranch != next.baseBranch || - !timeEqual(tp.MergedAt, status.PR.MergedAt) || - !timeEqual(tp.ClosedAt, status.PR.ClosedAt) + !timeEqual(tp.MergedAt, next.mergedAt) || + !timeEqual(tp.ClosedAt, next.closedAt) - tp.State = status.PR.State + tp.State = next.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.MergedAt = next.mergedAt + tp.ClosedAt = next.closedAt tp.ReviewState = status.ReviewState tp.ChecksState = status.ChecksState tp.MergeableState = next.mergeableState diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index dace925909..589f969dc5 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -1690,6 +1690,31 @@ func (s *Store) ListTaskIDsByPRNumber(ctx context.Context, workspaceID string, p return ids, nil } +// ListTaskPRsByPRNumber returns the non-detached PR associations in a workspace +// that point at one exact (owner, repo, pr_number). Workspace-scoped via the +// JOIN on tasks — the same guard ListTaskIDsByPRNumber uses — so a PR number +// shared across workspaces never reaches a caller holding only one workspace's +// credentials. More than one row is normal: several tasks in a workspace can +// legitimately link the same PR. +func (s *Store) ListTaskPRsByPRNumber( + ctx context.Context, workspaceID, owner, repo string, prNumber int, +) ([]*TaskPR, error) { + var prs []TaskPR + if err := s.ro.SelectContext(ctx, &prs, + `SELECT `+taskPRColumnsQualified+` FROM github_task_prs gtp + INNER JOIN tasks t ON gtp.task_id = t.id + WHERE t.workspace_id = ? AND gtp.owner = ? AND gtp.repo = ? AND gtp.pr_number = ? + AND gtp.detached_at IS NULL + ORDER BY gtp.created_at ASC`, workspaceID, owner, repo, prNumber); err != nil { + return nil, err + } + out := make([]*TaskPR, 0, len(prs)) + for i := range prs { + out = append(out, &prs[i]) + } + return out, nil +} + func groupTaskPRsByTask(prs []TaskPR) map[string][]*TaskPR { result := make(map[string][]*TaskPR) for i := range prs { diff --git a/apps/web/components/github/pr-detail-panel.state-sync.test.ts b/apps/web/components/github/pr-detail-panel.state-sync.test.ts new file mode 100644 index 0000000000..c8705c1419 --- /dev/null +++ b/apps/web/components/github/pr-detail-panel.state-sync.test.ts @@ -0,0 +1,200 @@ +import { cleanup, render, screen, waitFor } from "@testing-library/react"; +import { createElement, useEffect } from "react"; +import { TooltipProvider } from "@kandev/ui/tooltip"; +import { StateProvider, useAppStore } from "@/components/state-provider"; +import type { AppState } from "@/lib/state/store"; +import { afterEach, describe, it, expect, vi } from "vitest"; +import { PRDetailContent } from "./pr-detail-panel"; +import type { TaskPR, PRFeedback, GitHubPR } from "@/lib/types/github"; + +const feedbackMocks = vi.hoisted(() => ({ + refresh: vi.fn(), + value: null as PRFeedback | null, +})); + +vi.mock("@/hooks/domains/github/use-pr-feedback", () => ({ + usePRFeedback: () => ({ + feedback: feedbackMocks.value, + loading: false, + refresh: feedbackMocks.refresh, + }), +})); +vi.mock("@/lib/state/slices/comments", () => ({ + isPRFeedbackComment: () => false, + useCommentsStore: ( + selector: (state: { + addComment: () => void; + pendingForChat: string[]; + byId: object; + }) => unknown, + ) => selector({ addComment: vi.fn(), pendingForChat: [], byId: {} }), +})); +vi.mock("@/components/toast-provider", () => ({ useToast: () => ({ toast: vi.fn() }) })); +vi.mock("@/hooks/domains/github/use-github-status", () => ({ + useGitHubStatus: () => ({ status: null }), +})); +vi.mock("./pr-merge-button", () => ({ PRMergeButton: () => null })); +vi.mock("./pr-mergeability-notice", () => ({ + PRMergeabilityNotice: () => null, + buildConflictResolutionMessage: () => "", +})); +vi.mock("./pr-checks-section", () => ({ ChecksSection: () => null })); +vi.mock("./pr-comments-section", () => ({ CommentsSection: () => null })); + +afterEach(() => { + cleanup(); + feedbackMocks.refresh.mockReset(); + feedbackMocks.value = null; +}); + +function makeTaskPR(overrides: Partial = {}): TaskPR { + return { + id: "id", + task_id: "task", + owner: "o", + repo: "r", + pr_number: 1, + pr_url: "", + pr_title: "Test PR", + head_branch: "feat", + base_branch: "main", + author_login: "alice", + state: "open", + review_state: "", + checks_state: "", + mergeable_state: "", + review_count: 0, + pending_review_count: 0, + comment_count: 0, + unresolved_review_threads: 0, + checks_total: 0, + checks_passing: 0, + additions: 0, + deletions: 0, + created_at: "", + merged_at: null, + closed_at: null, + last_synced_at: null, + updated_at: "", + ...overrides, + }; +} + +function makeFeedback(pr: Partial = {}): PRFeedback { + return { + pr: { + number: 1, + title: "Test PR", + url: "", + html_url: "", + state: "open", + head_branch: "feat", + base_branch: "main", + author_login: "alice", + repo_owner: "o", + repo_name: "r", + draft: false, + mergeable: true, + additions: 0, + deletions: 0, + ...pr, + } as GitHubPR, + reviews: [], + comments: [], + checks: [], + has_issues: false, + }; +} + +const TEST_INITIAL_STATE = { + workspaces: { activeId: "workspace-1" }, +} as unknown as Partial; + +// Seeds the store the way the boot payload / WS sync would, so the probe below +// starts from a real row rather than an empty map. +function SeedTaskPR({ taskPR }: { taskPR: TaskPR }) { + const setTaskPRs = useAppStore((s) => s.setTaskPRs); + useEffect(() => { + setTaskPRs({ [taskPR.task_id]: [taskPR] }); + }, [setTaskPRs, taskPR]); + return null; +} + +function renderPRDetail(taskPR: TaskPR) { + return render( + createElement(StateProvider, { + initialState: TEST_INITIAL_STATE, + children: createElement( + TooltipProvider, + undefined, + createElement(PRDetailContent, { taskPR, sessionId: "session-1" }), + ), + }), + ); +} + +// Renders the panel alongside a probe that reports the live store row, so a +// test can assert what the panel did (or did not) write back into it. +function renderPRDetailWithStoreProbe(taskPR: TaskPR) { + let read: () => TaskPR | undefined = () => undefined; + function Probe() { + const byTaskId = useAppStore((s) => s.taskPRs.byTaskId); + read = () => byTaskId[taskPR.task_id]?.[0]; + return null; + } + const view = render( + createElement(StateProvider, { + initialState: TEST_INITIAL_STATE, + children: createElement( + TooltipProvider, + undefined, + createElement(SeedTaskPR, { taskPR }), + createElement(Probe), + createElement(PRDetailContent, { taskPR, sessionId: "session-1" }), + ), + }), + ); + return { ...view, readStoreRow: () => read() }; +} + +describe("PRDetailContent merged state", () => { + // The panel reads live state straight off the feedback response, so a merge + // observed by the fetch paints immediately. + it("renders the merged badge from live feedback", () => { + feedbackMocks.value = makeFeedback({ state: "merged" }); + renderPRDetail(makeTaskPR({ state: "open" })); + expect(screen.getByText("merged")).toBeTruthy(); + }); + + // And it keeps rendering merged once the row itself has converged — which is + // what survives a reload, since the store rehydrates from github_task_prs. + it("renders the merged badge from a converged store row before feedback loads", () => { + feedbackMocks.value = null; + renderPRDetail(makeTaskPR({ state: "merged", merged_at: "2026-01-01T00:00:00Z" })); + expect(screen.getByText("merged")).toBeTruthy(); + }); + + // Regression guard for the collapse: the panel used to patch state / + // merged_at / closed_at / additions / deletions / mergeable_state into the + // Zustand row itself (useSyncLivePRState). That write never reached the DB, + // so this tab went purple while the kanban card stayed green and a reload + // undid it. The backend now persists during the feedback fetch and the + // github.task_pr.updated handler applies it — the panel must stay read-only. + it("does not write live PR state back into the store", async () => { + feedbackMocks.value = makeFeedback({ + state: "merged", + merged_at: "2026-01-01T00:00:00Z", + additions: 42, + deletions: 7, + }); + const stale = makeTaskPR({ state: "open", additions: 0, deletions: 0 }); + const { readStoreRow } = renderPRDetailWithStoreProbe(stale); + + await waitFor(() => expect(readStoreRow()).toBeTruthy()); + const row = readStoreRow(); + expect(row?.state).toBe("open"); + expect(row?.merged_at).toBeNull(); + expect(row?.additions).toBe(0); + expect(row?.deletions).toBe(0); + }); +}); diff --git a/apps/web/components/github/pr-detail-panel.tsx b/apps/web/components/github/pr-detail-panel.tsx index 00b04c9396..aba1104b7d 100644 --- a/apps/web/components/github/pr-detail-panel.tsx +++ b/apps/web/components/github/pr-detail-panel.tsx @@ -111,64 +111,14 @@ function useAddPRFeedbackAsContext(sessionId: string, prNumber: number) { return { addAsContext }; } -// Sync live feedback data back to the store so topbar/other consumers stay up to date. -// Use primitive deps to avoid re-render loops from object reference changes. -// Guard: never regress the store to a less-terminal state (e.g. merged → open) -// because the feedback fetch may return stale data from before a backend poll update. -function useSyncLivePRState(taskPR: TaskPR, feedback: PRFeedback | null) { - const setTaskPR = useAppStore((s) => s.setTaskPR); - const prState = taskPR.state; - const prMergedAt = taskPR.merged_at ?? null; - const prClosedAt = taskPR.closed_at ?? null; - const prAdditions = taskPR.additions; - const prDeletions = taskPR.deletions; - const prMergeableState = taskPR.mergeable_state; - const prTaskId = taskPR.task_id; - useEffect(() => { - if (!feedback) return; - const livePR = feedback.pr; - // State priority: merged > closed > open. Never regress to a less-terminal state. - const stateRank = (s: string) => { - if (s === "merged") return 2; - if (s === "closed") return 1; - return 0; - }; - const effectiveState = stateRank(livePR.state) >= stateRank(prState) ? livePR.state : prState; - const effectiveMergedAt = effectiveState === prState ? prMergedAt : (livePR.merged_at ?? null); - const effectiveClosedAt = effectiveState === prState ? prClosedAt : (livePR.closed_at ?? null); - // Live mergeable_state is authoritative when present; otherwise keep the stored value. - const effectiveMergeableState = livePR.mergeable_state ?? prMergeableState; - if ( - effectiveState !== prState || - effectiveMergedAt !== prMergedAt || - effectiveClosedAt !== prClosedAt || - livePR.additions !== prAdditions || - livePR.deletions !== prDeletions || - effectiveMergeableState !== prMergeableState - ) { - setTaskPR(prTaskId, { - ...taskPR, - state: effectiveState as TaskPR["state"], - additions: livePR.additions, - deletions: livePR.deletions, - merged_at: effectiveMergedAt, - closed_at: effectiveClosedAt, - mergeable_state: effectiveMergeableState, - }); - } - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [ - feedback, - prState, - prMergedAt, - prClosedAt, - prAdditions, - prDeletions, - prMergeableState, - prTaskId, - setTaskPR, - ]); -} +// The panel used to patch state / merged_at / closed_at / additions / deletions / +// mergeable_state from the feedback response straight into the Zustand taskPRs +// row (useSyncLivePRState). That write never reached github_task_prs, so opening +// this panel turned its own tab purple while the kanban card and boot payload +// kept reading state=open, and a reload snapped it back. The backend now writes +// the same live PR through SyncTaskPR during the feedback fetch and publishes +// github.task_pr.updated, which lib/ws/handlers/github.ts applies to the store — +// one writer, and it survives a reload. Do not reintroduce a client-side patch. type PRPanelMetrics = { reviewCount: number; @@ -359,8 +309,6 @@ export function PRDetailContent({ taskPR, sessionId }: { taskPR: TaskPR; session toast, }); - useSyncLivePRState(taskPR, feedback); - const metrics = derivePanelMetrics(taskPR, feedback); const conflictQueued = useConflictQueued(sessionId, taskPR.pr_number);