diff --git a/apps/backend/internal/github/service_pr_watch_multi_branch_test.go b/apps/backend/internal/github/service_pr_watch_multi_branch_test.go index 17c2dc7fc1..56990aa39f 100644 --- a/apps/backend/internal/github/service_pr_watch_multi_branch_test.go +++ b/apps/backend/internal/github/service_pr_watch_multi_branch_test.go @@ -110,3 +110,149 @@ func TestGetPRWatchBySessionRepoAndBranch_FindsRightRow(t *testing.T) { t.Fatalf("expected primary watch id=%q, got %+v", primary.ID, gotPrimary) } } + +// TestUpdatePRWatchBranchIfSearching_NoCollision_UpdatesBranch covers the +// happy path: a single still-searching watch (pr_number=0) whose live +// branch changed gets its branch column rewritten. +func TestUpdatePRWatchBranchIfSearching_NoCollision_UpdatesBranch(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + + w, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/old") + if err != nil { + t.Fatalf("create watch: %v", err) + } + + if err := svc.UpdatePRWatchBranchIfSearching(ctx, w.ID, "feature/new"); err != nil { + t.Fatalf("UpdatePRWatchBranchIfSearching: %v", err) + } + + got, err := svc.GetPRWatchBySessionRepoAndBranch(ctx, "session-1", "repo-1", "feature/new") + if err != nil { + t.Fatalf("lookup updated branch: %v", err) + } + if got == nil || got.ID != w.ID { + t.Fatalf("expected watch %q on new branch, got %+v", w.ID, got) + } +} + +// TestUpdatePRWatchBranchIfSearching_CollidesWithSibling_DropsSource locks +// the fix for the UNIQUE-constraint collision: when a sibling watch already +// owns the destination (session, repo, branch) triple, the source row is +// deleted instead of triggering a UNIQUE constraint error. +func TestUpdatePRWatchBranchIfSearching_CollidesWithSibling_DropsSource(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + + source, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/A") + if err != nil { + t.Fatalf("create source watch: %v", err) + } + sibling, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/B") + if err != nil { + t.Fatalf("create sibling watch: %v", err) + } + + if err := svc.UpdatePRWatchBranchIfSearching(ctx, source.ID, "feature/B"); err != nil { + t.Fatalf("UpdatePRWatchBranchIfSearching must not error on sibling collision: %v", err) + } + + all, err := store.ListPRWatchesBySession(ctx, "session-1") + if err != nil { + t.Fatalf("list watches: %v", err) + } + if len(all) != 1 { + t.Fatalf("expected source watch dropped, got %d remaining", len(all)) + } + if all[0].ID != sibling.ID { + t.Fatalf("expected sibling watch %q to survive, got %q", sibling.ID, all[0].ID) + } + if all[0].Branch != "feature/B" { + t.Errorf("sibling branch must be untouched, got %q", all[0].Branch) + } +} + +// TestUpdatePRWatchBranchIfSearching_CollidesWithSiblingHasPR_DropsSource +// exercises the realistic race: the sibling has already discovered its PR +// (pr_number != 0) when the still-searching source tries to migrate onto +// the same branch. Source must be deleted, sibling row (including its PR +// number) must survive untouched. +func TestUpdatePRWatchBranchIfSearching_CollidesWithSiblingHasPR_DropsSource(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + + source, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/A") + if err != nil { + t.Fatalf("create source watch: %v", err) + } + sibling, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/B") + if err != nil { + t.Fatalf("create sibling watch: %v", err) + } + if err := store.UpdatePRWatchPRNumber(ctx, sibling.ID, 99); err != nil { + t.Fatalf("mark sibling as found: %v", err) + } + + if err := svc.UpdatePRWatchBranchIfSearching(ctx, source.ID, "feature/B"); err != nil { + t.Fatalf("UpdatePRWatchBranchIfSearching must not error when sibling owns branch with active PR: %v", err) + } + + all, err := store.ListPRWatchesBySession(ctx, "session-1") + if err != nil { + t.Fatalf("list watches: %v", err) + } + if len(all) != 1 { + t.Fatalf("expected source watch dropped, sibling preserved; got %d remaining", len(all)) + } + if all[0].ID != sibling.ID { + t.Fatalf("expected sibling %q to survive, got %q", sibling.ID, all[0].ID) + } + if all[0].Branch != "feature/B" { + t.Errorf("sibling branch must be untouched, got %q", all[0].Branch) + } + if all[0].PRNumber != 99 { + t.Errorf("sibling PR number must be preserved (99), got %d", all[0].PRNumber) + } +} + +// TestUpdatePRWatchBranchIfSearching_PRAlreadyFound_NoOp preserves the +// existing "searching" guard: a watch that already found its PR +// (pr_number != 0) must not have its branch overwritten. +func TestUpdatePRWatchBranchIfSearching_PRAlreadyFound_NoOp(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-1", false) + + w, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 42, "feature/found") + if err != nil { + t.Fatalf("create watch: %v", err) + } + + if err := svc.UpdatePRWatchBranchIfSearching(ctx, w.ID, "feature/other"); err != nil { + t.Fatalf("UpdatePRWatchBranchIfSearching: %v", err) + } + + got, err := svc.GetPRWatchBySessionRepoAndBranch(ctx, "session-1", "repo-1", "feature/found") + if err != nil { + t.Fatalf("lookup: %v", err) + } + if got == nil || got.ID != w.ID { + t.Fatalf("expected watch branch unchanged, got %+v", got) + } +} + +// TestUpdatePRWatchBranchIfSearching_MissingRow_NoOp locks the +// idempotency contract: calling with an ID that doesn't exist must be +// a silent no-op (matches the prior UPDATE-by-id semantics that +// affected 0 rows without error). +func TestUpdatePRWatchBranchIfSearching_MissingRow_NoOp(t *testing.T) { + _, svc, _, _ := setupPollerTest(t) + ctx := context.Background() + + if err := svc.UpdatePRWatchBranchIfSearching(ctx, "nonexistent-id", "feature/any"); err != nil { + t.Fatalf("must be a no-op for missing row, got: %v", err) + } +} diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index b49af0ee12..ddb5449ed8 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -688,11 +688,72 @@ func (s *Store) ResetPRWatch(ctx context.Context, id, branch string) error { // UpdatePRWatchBranchIfSearching atomically updates branch only when pr_number = 0, // preventing races with concurrent PR association. +// +// Collision semantics: a sibling watch may already own the destination +// (session_id, repository_id, branch) triple — e.g. multi-branch task where +// the agent's live branch collapsed onto a peer watch's branch. In that +// case the raw UPDATE would trip the UNIQUE constraint. We instead drop the +// source row (which is still searching, pr_number=0, so it owns no PR +// state) and let the sibling continue to track the branch. func (s *Store) UpdatePRWatchBranchIfSearching(ctx context.Context, id, branch string) error { - _, err := s.db.ExecContext(ctx, + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + + var sessionID, repositoryID string + var prNumber int + err = tx.QueryRowContext(ctx, + `SELECT session_id, repository_id, pr_number FROM github_pr_watches WHERE id = ?`, id). + Scan(&sessionID, &repositoryID, &prNumber) + if errors.Is(err, sql.ErrNoRows) { + return tx.Commit() + } + if err != nil { + return err + } + if prNumber != 0 { + return tx.Commit() + } + + var probe int // existence probe only; value unused + err = tx.QueryRowContext(ctx, + `SELECT 1 FROM github_pr_watches + WHERE session_id = ? AND repository_id = ? AND branch = ? AND id <> ?`, + sessionID, repositoryID, branch, id).Scan(&probe) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if err == nil { + return dropSourceAndCommit(ctx, tx, id) + } + + if _, err := tx.ExecContext(ctx, `UPDATE github_pr_watches SET branch = ?, updated_at = ? WHERE id = ? AND pr_number = 0`, - branch, time.Now().UTC(), id) - return err + branch, time.Now().UTC(), id); err != nil { + // Defensive belt-and-suspenders: the SQLite writer pool is + // SetMaxOpenConns(1), so an in-process CreatePRWatch cannot + // commit a sibling row between our probe and this UPDATE. But + // an external writer (separate process touching the same file, + // future pool reshuffle) could; if the UPDATE still trips + // UNIQUE, treat it identically to the probe-found path. + if strings.Contains(err.Error(), "UNIQUE constraint failed") { + return dropSourceAndCommit(ctx, tx, id) + } + return err + } + return tx.Commit() +} + +// dropSourceAndCommit removes a still-searching source watch (pr_number=0) +// whose destination branch is already owned by a sibling row, then commits. +func dropSourceAndCommit(ctx context.Context, tx *sql.Tx, id string) error { + if _, err := tx.ExecContext(ctx, + `DELETE FROM github_pr_watches WHERE id = ? AND pr_number = 0`, id); err != nil { + return err + } + return tx.Commit() } // --- TaskPR operations ---