feat(tasks): multi-branch tasks — N (repo, branch) pairs per task (#1226)

* feat(tasks): multi-branch tasks — N (repo, branch) pairs per task

Extends multi-repo into multi-branch so a single task can hold multiple
worktrees on the same repo (one per branch). Sibling worktrees layout at
<task>/<repo>/ + <task>/<repo>-<slug>/, dedicated PRWatch per branch,
live agentctl tracker rescan + frontend store updates so the UI splits
PRs / files / commits per branch without a session restart. MCP
add_branch_to_task_kandev tool is shielded to the worktree executor.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(web): unify multi-branch repo label as <repo> · <branch> across panels

PR Changes and Commits panels rendered repo labels differently for multi-branch
tasks — Commits showed "kandev · feature/x" while PR Changes leaked the raw
"kandev-feature-x" agentctl tracker tag. Reformat tracker-tagged repo names via
formatRepoLabel using the longest known-repo prefix so kandev-cli-* doesn't
collapse to kandev · cli-*.

* fix(tasks): address multi-branch PR review — data races, slug collisions, MCP error codes

Backend
- agentctl process.Manager: add repoTrackersMu RWMutex guarding repoTrackers
  + workspaceTracker. Reads snapshot under RLock; RescanRepositories builds
  trackers first, takes Lock for swap/append. Validates newWorkDir via
  os.Stat before publishing cfg.WorkDir.
- worktree.Manager: reject invalid explicit BranchSlug up-front; cache key
  uses sanitized wt.BranchSlug so read/write paths are symmetric.
- worktree.store: narrow isDuplicateColumnError to "duplicate column name".
- task service AddBranchToTask: slug-collision guard rejects feat/a vs feat-a.
- orchestrator resolvePushRepo: two-pass exact-name match before subdir
  prefix fallback so "backend" doesn't swallow pushes for "backend-admin".
- executor buildRepoSpecs: duplicate repos always get a non-empty BranchSlug
  (falls back to branch-N when both branch fields sanitize to empty).
- MCP handler: classifyAddBranchError maps service errors to NOT_FOUND /
  CONFLICT / VALIDATION_ERROR; add ErrorCodeConflict constant.
- MCP server: extract registerAddBranchToTaskTool, register in task mode
  only (external mode contract: config tools + create_task_kandev).
- workspace_rescan handler: reject malformed JSON with 400 instead of
  silently rescanning the current root.
- branch_materializer: split MaterializeBranch into prepare + finalize
  helpers to satisfy the 80-line cap.

Frontend
- existingPrUrl returns undefined when a task has >1 PR so per-repo View PR
  buttons take over (no silent linking to a sibling).
- formatRepoLabel checks exact known-repo match before prefix splitting so
  "kandev-cli" isn't mislabeled as "kandev · cli".
- Repo chip key now includes base_branch so siblings differing only on
  base render as distinct chips.

Docs
- multi-branch spec uniqueness wording aligned with the canonical 4-column
  key (task_id, repository_id, base_branch, checkout_branch).
- worktree.CreateRequest.BranchSlug doc describes the sibling layout
  (<repo>-<slug>) instead of the obsolete nested form.

* fix(tasks): scope slug-collision check to repo only + use errors.Is for ErrTaskNotFound

- service_branches: drop base_branch from the slug-collision check. Worktree
  path is (repo, slug); two rows on the same repo cut from different bases
  still collide on disk.
- service_branches: prepareBranchAdd wraps taskrepo.ErrTaskNotFound via
  fmt.Errorf with %w so classifiers don't have to substring-match the
  formatted task UUID.
- mcp handlers: classifyAddBranchError uses errors.Is(err, ErrTaskNotFound).

* fix(agentctl): sanitize rescan work_dir before os.Stat to clear CodeQL path-injection

CodeQL flagged the rescan endpoint as a path-injection sink (the
HTTP-supplied work_dir flowed straight into os.Stat and downstream
scanRepositorySubdirs). Add validateRescanPath which:
- requires the input to be a clean absolute path with no ".." segments
- rejects anything that isn't equal to or an ancestor of the current
  cfg.WorkDir — the only legitimate use case is the branch materializer
  promoting <task>/<repo>/ to <task>/

Defense in depth: agentctl binds to localhost, but any colocated process
could otherwise direct it at /etc or the user's home by POSTing
/api/v1/workspace/rescan.

* fix(agentctl): tighten validateRescanPath, close TOCTOU on existingTrackers, add tests

- Drop the strings.Contains(clean, "..") guard. filepath.Clean already
  resolves traversal; the substring check rejects legitimate names like
  /home/u/..hidden without adding defense (the ancestor check via
  filepath.Rel is the real boundary).
- Drop the rel == "." branch in the ancestor check — early return for
  clean == currentClean already covers identity.
- Read existingTrackers inside the write-lock section (not under RLock
  then re-Lock) so two concurrent rescans can't both see 0 trackers and
  double-bootstrap the multi-repo set.
- Add table-driven TestValidateRescanPath covering rejection paths
  (relative, descendant, sibling, traversal-cleaned, empty input),
  legitimate ancestor promotion, exact match, no-anchor first-launch,
  and a /home/u/..hidden dotted-segment case.

* fix(agentctl): break taint flow into os.Stat — derive rescan path from cfg, not HTTP body

CodeQL's go/path-injection trace still flagged manager_rescan.go after the
prior sanitizer because the user-supplied newWorkDir flowed (after
validation) into os.Stat and then into cfg.WorkDir → tracker paths →
workspace_monitor os.Stat sinks.

Replace validateRescanPath with resolveRescanPath: it returns a path
DERIVED from currentWorkDir (no-op = currentClean, promotion =
filepath.Dir(currentClean)) and uses newPath only for comparison, never
as a source for the returned string. os.Stat now receives a path whose
provenance is trusted manager config.

Functionally the legitimate caller (branch materializer promoting
<task>/<repo>/ to <task>/) is unchanged.

* test(agentctl): make TestResolveRescanPath windows-portable

The Linux Backend Tests job passed but the Backend (windows) job rejected
every "absolute" case because filepath.IsAbs("/task/repo") is false on
Windows (no volume letter).

Build inputs via filepath.Join against absRoot() (filepath.Abs of "/" /
"\\") so the same table covers POSIX (`/...`) and Windows (`C:\...`)
without per-OS branches. Keep the traversal-after-Clean case POSIX-only
since filepath.Clean is the shared primitive and that input form needs a
volume to be considered absolute on Windows anyway.

* fix(agentctl): serialize RescanRepositories + re-check trackers under write lock

Two concurrent RescanRepositories calls could both observe an empty
tracker set (between the snapshot read and the bootstrap branch) and both
enter transitionToMultiRepoMode, leaking a duplicate bare-root tracker
and duplicate per-repo trackers. Same race in appendNewRepoTrackers would
double-append the same new child across both callers.

Fix:
- Add rescanMu sync.Mutex on Manager. Acquired for the full body of
  RescanRepositories so the snapshot → branch → mutate sequence is
  serialized. repoTrackersMu still guards individual field reads from
  unrelated goroutines (subscribe/unsubscribe/poll-mode).
- appendNewRepoTrackers re-checks membership inside the write-lock
  before appending and drops already-present trackers (Stop + detach
  outside the lock). Defense in depth.
- Test: add `dotted segment is legal as promotion` case so the
  ..hidden-segment promotion path stays exercised.

---------

Co-authored-by: Kandev Agent <agent@kandev.dev>
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
João Salavisa
2026-06-01 19:36:20 +01:00
committed by GitHub
parent 5591cdc0bf
commit fd79d6170f
64 changed files with 3949 additions and 252 deletions
@@ -15,17 +15,18 @@ import (
// --- PR Watch operations ---
// CreatePRWatch creates a new PR watch for a (session, repository) pair.
// `repositoryID` may be empty for legacy single-repo callers; multi-repo
// callers must pass the per-task repository_id so each repo gets its own
// watch row.
// CreatePRWatch creates a new PR watch for a (session, repository, branch)
// triple. `repositoryID` may be empty for legacy single-repo callers;
// multi-repo callers must pass the per-task repository_id so each repo gets
// its own watch row. Multi-branch tasks store one watch per branch so a
// secondary branch's push isn't lost behind the primary's existing watch.
func (s *Service) CreatePRWatch(ctx context.Context, sessionID, taskID, repositoryID, owner, repo string, prNumber int, branch string) (*PRWatch, error) {
existing, err := s.store.GetPRWatchBySessionAndRepo(ctx, sessionID, repositoryID)
existing, err := s.store.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, repositoryID, branch)
if err != nil {
return nil, err
}
if existing != nil {
return existing, nil // already watching this (session, repo)
return existing, nil // already watching this (session, repo, branch)
}
w := &PRWatch{
SessionID: sessionID,
@@ -42,6 +43,7 @@ func (s *Service) CreatePRWatch(ctx context.Context, sessionID, taskID, reposito
s.logger.Info("created PR watch",
zap.String("session_id", sessionID),
zap.String("repository_id", repositoryID),
zap.String("branch", branch),
zap.Int("pr_number", prNumber))
return w, nil
}
@@ -58,6 +60,13 @@ func (s *Service) GetPRWatchBySessionAndRepo(ctx context.Context, sessionID, rep
return s.store.GetPRWatchBySessionAndRepo(ctx, sessionID, repositoryID)
}
// GetPRWatchBySessionRepoAndBranch returns the PR watch for a precise
// (session, repository, branch) triple — used by push detection in
// multi-branch tasks so each branch's push lands on its own watch row.
func (s *Service) GetPRWatchBySessionRepoAndBranch(ctx context.Context, sessionID, repositoryID, branch string) (*PRWatch, error) {
return s.store.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, repositoryID, branch)
}
// ListPRWatchesBySession returns every PR watch for a session.
func (s *Service) ListPRWatchesBySession(ctx context.Context, sessionID string) ([]*PRWatch, error) {
return s.store.ListPRWatchesBySession(ctx, sessionID)
@@ -116,14 +125,15 @@ func (s *Service) CheckPRWatch(ctx context.Context, watch *PRWatch) (*PRStatus,
return status, hasNew, nil
}
// EnsurePRWatch creates a PRWatch with pr_number=0 for a (session, repo) pair
// if one doesn't already exist. The poller will detect the PR by searching
// for the branch on GitHub. `repositoryID` is empty for legacy single-repo
// callers; multi-repo callers MUST pass the per-task repository_id so each
// repo gets its own watch (the table's UNIQUE(session_id, repository_id) used
// to be UNIQUE(session_id), which silently dropped second-repo watches).
// EnsurePRWatch creates a PRWatch with pr_number=0 for a
// (session, repo, branch) triple if one doesn't already exist. The poller
// will detect the PR by searching for the branch on GitHub. `repositoryID`
// is empty for legacy single-repo callers; multi-repo / multi-branch
// callers MUST pass the per-task repository_id and the worktree's branch
// so each branch gets its own watch — keying on (session, repo) alone
// drops secondary branches' watches behind the primary's existing row.
func (s *Service) EnsurePRWatch(ctx context.Context, sessionID, taskID, repositoryID, owner, repo, branch string) (*PRWatch, error) {
existing, err := s.store.GetPRWatchBySessionAndRepo(ctx, sessionID, repositoryID)
existing, err := s.store.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, repositoryID, branch)
if err != nil {
return nil, err
}
@@ -157,14 +167,18 @@ func (s *Service) EnsurePRWatch(ctx context.Context, sessionID, taskID, reposito
// callers MUST pass it — empty causes ReplaceTaskPR to wipe the entire task's
// PR rows (legacy "delete all" branch), which is what older code relied on.
func (s *Service) AssociatePRWithTask(ctx context.Context, taskID, repositoryID string, pr *PR) (*TaskPR, error) {
// Check for an existing PR for this exact (task, repo). Multi-repo callers
// must scope by repository_id so the same PR number in two repos doesn't
// short-circuit the second association.
existing, err := s.store.GetTaskPRByRepository(ctx, taskID, repositoryID)
// Multi-branch: scope the "already-current" short-circuit by exact
// pr_number too. A task can hold multiple PR rows per (task, repo) on
// different branches; the legacy by-repo lookup returns whichever row
// was most recently updated, which would make Associate think the
// secondary PR is already there (wrong PR number) and skip the insert
// — or worse, fall through to ReplaceTaskPR which used to delete the
// sibling row.
existing, err := s.store.GetTaskPRByRepoAndNumber(ctx, taskID, repositoryID, pr.Number)
if err != nil {
return nil, err
}
if existing != nil && existing.PRNumber == pr.Number {
if existing != nil {
return existing, nil
}
tp := &TaskPR{
@@ -185,20 +199,16 @@ func (s *Service) AssociatePRWithTask(ctx context.Context, taskID, repositoryID
MergedAt: pr.MergedAt,
ClosedAt: pr.ClosedAt,
}
// ReplaceTaskPR atomically deletes any existing association for the
// (task, repository) pair and inserts the new row inside one transaction.
// Scoping by repository_id keeps multi-repo tasks intact; legacy callers
// (repositoryID == "") still get the "delete all" semantics.
// ReplaceTaskPR upserts the row matching (task, repository, pr_number).
// Multi-branch tasks may already hold sibling rows for the SAME
// (task, repository) on different PR numbers — ReplaceTaskPR no longer
// touches them. The early-return above guarantees we only reach this
// line when no row for the exact pr_number exists yet, so this is a
// straight insert in steady state; the delete-then-insert form is
// retained so a retry that races a partial write resolves cleanly.
if err := s.store.ReplaceTaskPR(ctx, tp); err != nil {
return nil, fmt.Errorf("replace task PR: %w", err)
}
if existing != nil {
s.logger.Info("replaced stale task PR association",
zap.String("task_id", taskID),
zap.String("repository_id", repositoryID),
zap.Int("old_pr_number", existing.PRNumber),
zap.Int("new_pr_number", pr.Number))
}
// Publish event for UI
if s.eventBus != nil {
@@ -0,0 +1,112 @@
package github
import (
"context"
"testing"
)
// TestCreatePRWatch_AllowsMultipleBranchesPerRepo locks the multi-branch
// invariant: the same (session, repository) pair can hold N watches, one
// per branch. Previously the unique constraint was UNIQUE(session_id,
// repository_id) and the dedup in CreatePRWatch returned the existing row
// — secondary branches' pushes silently re-attached to the primary's watch
// and the secondary PR never landed in github_task_prs.
func TestCreatePRWatch_AllowsMultipleBranchesPerRepo(t *testing.T) {
_, svc, _, store := setupPollerTest(t)
ctx := context.Background()
seedTask(t, store, "task-1", false)
w1, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/primary")
if err != nil {
t.Fatalf("CreatePRWatch primary: %v", err)
}
if w1 == nil {
t.Fatal("primary watch must not be nil")
}
w2, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/branch-2")
if err != nil {
t.Fatalf("CreatePRWatch branch-2: %v", err)
}
if w2 == nil {
t.Fatal("branch-2 watch must not be nil")
}
if w1.ID == w2.ID {
t.Errorf("expected distinct watches for two branches, got same id %q", w1.ID)
}
all, err := store.ListPRWatchesBySession(ctx, "session-1")
if err != nil {
t.Fatalf("list watches: %v", err)
}
if len(all) != 2 {
t.Fatalf("expected 2 watches for (session, repo) across branches, got %d", len(all))
}
}
// TestCreatePRWatch_IdempotentPerBranch verifies that re-creating a watch
// for the same (session, repo, branch) triple is a no-op returning the
// existing row — push detection retries that race the original create
// must not duplicate watches.
func TestCreatePRWatch_IdempotentPerBranch(t *testing.T) {
_, svc, _, store := setupPollerTest(t)
ctx := context.Background()
seedTask(t, store, "task-1", false)
first, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/x")
if err != nil {
t.Fatalf("first CreatePRWatch: %v", err)
}
second, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 0, "feature/x")
if err != nil {
t.Fatalf("second CreatePRWatch: %v", err)
}
if first.ID != second.ID {
t.Errorf("expected idempotent return of same watch id, got %q vs %q", first.ID, second.ID)
}
all, err := store.ListPRWatchesBySession(ctx, "session-1")
if err != nil {
t.Fatalf("list: %v", err)
}
if len(all) != 1 {
t.Errorf("expected 1 watch after idempotent retry, got %d", len(all))
}
}
// TestGetPRWatchBySessionRepoAndBranch_FindsRightRow verifies the
// branch-aware lookup used by push detection. Two watches on the same
// repo, different branches: the lookup must return the matching branch's
// watch and not the other.
func TestGetPRWatchBySessionRepoAndBranch_FindsRightRow(t *testing.T) {
_, svc, _, _ := setupPollerTest(t)
ctx := context.Background()
seedTask(t, svc.store, "task-1", false)
primary, err := svc.CreatePRWatch(ctx, "s1", "task-1", "repo-1", "owner", "repo", 1217, "feature/primary")
if err != nil {
t.Fatalf("primary: %v", err)
}
secondary, err := svc.CreatePRWatch(ctx, "s1", "task-1", "repo-1", "owner", "repo", 0, "feature/secondary")
if err != nil {
t.Fatalf("secondary: %v", err)
}
got, err := svc.GetPRWatchBySessionRepoAndBranch(ctx, "s1", "repo-1", "feature/secondary")
if err != nil {
t.Fatalf("GetPRWatchBySessionRepoAndBranch: %v", err)
}
if got == nil || got.ID != secondary.ID {
t.Fatalf("expected secondary watch id=%q, got %+v", secondary.ID, got)
}
gotPrimary, err := svc.GetPRWatchBySessionRepoAndBranch(ctx, "s1", "repo-1", "feature/primary")
if err != nil {
t.Fatalf("GetPRWatchBySessionRepoAndBranch primary: %v", err)
}
if gotPrimary == nil || gotPrimary.ID != primary.ID {
t.Fatalf("expected primary watch id=%q, got %+v", primary.ID, gotPrimary)
}
}
@@ -74,11 +74,14 @@ func TestCleanupMergedReviewTasks_TaskAlreadyDeleted(t *testing.T) {
}
}
// Regression: when a task already has a TaskPR row pointing to an old PR
// (e.g. the first PR was closed and a second one opened on the same or a new
// branch), AssociatePRWithTask must replace the stale row so downstream
// consumers (UI, GetTaskPR) observe the current PR rather than the old one.
func TestAssociatePRWithTask_ReplacesStaleAssociation(t *testing.T) {
// Regression: when a task already has a TaskPR row pointing to PR #1 and
// AssociatePRWithTask is called with a different PR #2 (e.g. first PR
// closed, new PR opened — or a multi-branch task gaining a second PR on a
// different branch), the new row must be inserted as a sibling without
// touching the existing #1 row. Multi-branch tasks rely on this so two
// PRs on the same (task, repo) coexist; the old "delete-then-insert"
// behavior collapsed multi-branch tasks down to one PR row.
func TestAssociatePRWithTask_AddsSecondPRAsSibling(t *testing.T) {
svc, store, _ := setupSyncTest(t)
ctx := context.Background()
@@ -97,7 +100,7 @@ func TestAssociatePRWithTask_ReplacesStaleAssociation(t *testing.T) {
t.Fatalf("seed TaskPR: %v", err)
}
// Associate a new PR #2 (could be on same or different branch).
// Associate a new PR #2 on a different branch.
newPR := &PR{
Number: 2,
Title: "Second",
@@ -117,11 +120,18 @@ func TestAssociatePRWithTask_ReplacesStaleAssociation(t *testing.T) {
t.Errorf("returned TaskPR.PRNumber=%d, want 2", tp.PRNumber)
}
got, err := store.GetTaskPR(ctx, "t1")
all, err := store.ListTaskPRsByTask(ctx, "t1")
if err != nil {
t.Fatalf("GetTaskPR: %v", err)
t.Fatalf("ListTaskPRsByTask: %v", err)
}
if got == nil || got.PRNumber != 2 {
t.Errorf("GetTaskPR after replace = %+v, want PRNumber=2", got)
if len(all) != 2 {
t.Fatalf("expected 2 PR rows after associating second PR, got %d", len(all))
}
nums := map[int]bool{}
for _, r := range all {
nums[r.PRNumber] = true
}
if !nums[1] || !nums[2] {
t.Errorf("missing expected PR numbers: %v", nums)
}
}
+108 -52
View File
@@ -53,7 +53,7 @@ const createTablesSQL = `
last_review_state TEXT DEFAULT '',
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL,
UNIQUE(session_id, repository_id)
UNIQUE(session_id, repository_id, branch)
);
CREATE TABLE IF NOT EXISTS github_task_prs (
@@ -321,42 +321,50 @@ func (s *Store) tableExists(name string) bool {
// migratePRTablesForMultiRepo rebuilds `github_pr_watches` and
// `github_task_prs` to drop the legacy single-repo unique constraints
// (`UNIQUE(session_id)` and `UNIQUE(task_id, pr_number)`) and replace them
// with the multi-repo variants. SQLite can't ALTER TABLE DROP CONSTRAINT, so
// each table is rebuilt via the recommended copy-and-rename pattern. The
// migration is idempotent: it inspects `sqlite_master.sql` for the legacy
// constraint string and only runs the rebuild when found.
// with the multi-repo / multi-branch variants. SQLite can't ALTER TABLE
// DROP CONSTRAINT, so each table is rebuilt via the recommended
// copy-and-rename pattern. The migration is idempotent: it inspects
// `sqlite_master.sql` for the legacy constraint string and only runs the
// rebuild when found. The watch rebuild fires twice — once for the original
// single-repo shape, once for the interim multi-repo shape — so DBs caught
// in either state upgrade cleanly to the multi-branch shape.
func (s *Store) migratePRTablesForMultiRepo() error {
if err := s.rebuildIfHasLegacyConstraint(
"github_pr_watches",
for _, trigger := range []string{
"session_id TEXT NOT NULL UNIQUE",
`CREATE TABLE github_pr_watches_new (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
task_id TEXT NOT NULL,
repository_id TEXT NOT NULL DEFAULT '',
owner TEXT NOT NULL,
repo TEXT NOT NULL,
pr_number INTEGER NOT NULL,
branch TEXT NOT NULL,
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,
UNIQUE(session_id, repository_id)
)`,
`INSERT INTO github_pr_watches_new (
id, session_id, task_id, repository_id, owner, repo, pr_number, branch,
last_checked_at, last_comment_at, last_check_status, last_review_state,
created_at, updated_at
) SELECT
id, session_id, task_id, COALESCE(repository_id, ''), owner, repo, pr_number, branch,
last_checked_at, last_comment_at, last_check_status, last_review_state,
created_at, updated_at
FROM github_pr_watches`,
); err != nil {
return err
"UNIQUE(session_id, repository_id)\n",
} {
if err := s.rebuildIfHasLegacyConstraint(
"github_pr_watches",
trigger,
`CREATE TABLE github_pr_watches_new (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
task_id TEXT NOT NULL,
repository_id TEXT NOT NULL DEFAULT '',
owner TEXT NOT NULL,
repo TEXT NOT NULL,
pr_number INTEGER NOT NULL,
branch TEXT NOT NULL,
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,
UNIQUE(session_id, repository_id, branch)
)`,
`INSERT INTO github_pr_watches_new (
id, session_id, task_id, repository_id, owner, repo, pr_number, branch,
last_checked_at, last_comment_at, last_check_status, last_review_state,
created_at, updated_at
) SELECT
id, session_id, task_id, COALESCE(repository_id, ''), owner, repo, pr_number, branch,
last_checked_at, last_comment_at, last_check_status, last_review_state,
created_at, updated_at
FROM github_pr_watches`,
); err != nil {
return err
}
}
return s.rebuildIfHasLegacyConstraint(
"github_task_prs",
@@ -477,10 +485,16 @@ func (s *Store) GetPRWatchBySession(ctx context.Context, sessionID string) (*PRW
// GetPRWatchBySessionAndRepo returns the PR watch for a (session, repository)
// pair, or nil. Used by per-repo branch-switch / commit handlers so each
// repo's watch is reset independently.
//
// Multi-branch caveat: a task can hold multiple watches for the same
// (session, repository) on different branches. This lookup returns the
// most-recently-updated row — callers that need branch-specific lookup
// must use GetPRWatchBySessionRepoAndBranch.
func (s *Store) GetPRWatchBySessionAndRepo(ctx context.Context, sessionID, repositoryID string) (*PRWatch, error) {
var w PRWatch
err := s.ro.GetContext(ctx, &w,
`SELECT * FROM github_pr_watches WHERE session_id = ? AND repository_id = ? LIMIT 1`,
`SELECT * FROM github_pr_watches WHERE session_id = ? AND repository_id = ?
ORDER BY updated_at DESC LIMIT 1`,
sessionID, repositoryID)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
@@ -488,6 +502,23 @@ func (s *Store) GetPRWatchBySessionAndRepo(ctx context.Context, sessionID, repos
return &w, err
}
// GetPRWatchBySessionRepoAndBranch returns the PR watch for the precise
// (session, repository, branch) triple. Required for multi-branch tasks
// where each branch needs its own watch — querying by (session, repo)
// alone would collapse the secondary branch's push detection onto the
// primary's watch and the secondary PR would never land in github_task_prs.
func (s *Store) GetPRWatchBySessionRepoAndBranch(ctx context.Context, sessionID, repositoryID, branch string) (*PRWatch, error) {
var w PRWatch
err := s.ro.GetContext(ctx, &w,
`SELECT * FROM github_pr_watches
WHERE session_id = ? AND repository_id = ? AND branch = ? LIMIT 1`,
sessionID, repositoryID, branch)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
return &w, err
}
// ListPRWatchesBySession returns every PR watch for a session (one per repo
// in multi-repo workspaces). Empty slice when no watches exist.
func (s *Store) ListPRWatchesBySession(ctx context.Context, sessionID string) ([]*PRWatch, error) {
@@ -627,10 +658,16 @@ func (s *Store) GetTaskPR(ctx context.Context, taskID string) (*TaskPR, error) {
// GetTaskPRByRepository returns the PR association for a (task, repository)
// pair, or nil if none. Use this for multi-repo tasks.
//
// Multi-branch caveat: a task can hold N rows per (task, repo) — one per
// PR number. This lookup returns the most-recently-updated row so callers
// that need a deterministic single value still get one. Callers that need
// the row for a specific PR number must use GetTaskPRByRepoAndNumber.
func (s *Store) GetTaskPRByRepository(ctx context.Context, taskID, repositoryID string) (*TaskPR, error) {
var tp TaskPR
err := s.ro.GetContext(ctx, &tp,
`SELECT * FROM github_task_prs WHERE task_id = ? AND repository_id = ? LIMIT 1`,
`SELECT * FROM github_task_prs WHERE task_id = ? AND repository_id = ?
ORDER BY updated_at DESC LIMIT 1`,
taskID, repositoryID)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
@@ -638,6 +675,23 @@ func (s *Store) GetTaskPRByRepository(ctx context.Context, taskID, repositoryID
return &tp, err
}
// GetTaskPRByRepoAndNumber returns the exact PR row matching the
// (task, repository, pr_number) triple. Required for multi-branch tasks
// where AssociatePRWithTask's "already-current" short-circuit must check
// the same PR number, not a sibling PR that happens to be the first
// row returned by the legacy by-repo query.
func (s *Store) GetTaskPRByRepoAndNumber(ctx context.Context, taskID, repositoryID string, prNumber int) (*TaskPR, error) {
var tp TaskPR
err := s.ro.GetContext(ctx, &tp,
`SELECT * FROM github_task_prs
WHERE task_id = ? AND repository_id = ? AND pr_number = ? LIMIT 1`,
taskID, repositoryID, prNumber)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
return &tp, err
}
// ListTaskPRsByTask returns every PR association for a task (one per repo
// when multi-repo). Empty slice when no PRs exist.
func (s *Store) ListTaskPRsByTask(ctx context.Context, taskID string) ([]*TaskPR, error) {
@@ -699,11 +753,17 @@ func groupTaskPRsByTask(prs []TaskPR) map[string][]*TaskPR {
return result
}
// ReplaceTaskPR atomically replaces the task→PR association for a task: any
// existing rows are deleted and the new row is inserted inside a single
// transaction. The delete is scoped to (task_id, repository_id) so multi-repo
// tasks only replace the row for the affected repo; for single-repo tasks
// (RepositoryID == "") the legacy semantics are preserved (delete all).
// ReplaceTaskPR atomically associates a PR with a task, replacing only the
// row that matches the exact (task_id, repository_id, pr_number) triple.
// Multi-branch tasks may hold multiple PR rows per (task, repo) — one per
// branch — so the delete MUST NOT wipe sibling PR rows. Single-repo
// callers (RepositoryID == "") only delete legacy untagged rows for the
// same PR number.
//
// The DELETE+INSERT pair inside one transaction is the upsert form; an
// ON CONFLICT would also work but the per-row delete pattern matches the
// existing migration layout (rebuilds are easier to reason about) and
// avoids leaking SQLite-specific syntax into the service layer.
func (s *Store) ReplaceTaskPR(ctx context.Context, tp *TaskPR) error {
if tp.ID == "" {
tp.ID = uuid.New().String()
@@ -717,22 +777,18 @@ func (s *Store) ReplaceTaskPR(ctx context.Context, tp *TaskPR) error {
}
defer func() { _ = tx.Rollback() }()
// Multi-repo: only delete the row for the same (task, repo) pair, leaving
// other repos' rows alone. Single-repo callers (RepositoryID == "")
// only delete legacy untagged rows so a multi-repo task's per-repo rows
// don't get wiped if a single-repo callsite slips in. The poller and
// service.AssociatePR* paths now always pass RepositoryID explicitly;
// any future caller that forgets does the safer thing (preserve tagged
// rows) instead of the old "nuke everything" behavior.
if tp.RepositoryID != "" {
if _, err := tx.ExecContext(ctx,
`DELETE FROM github_task_prs WHERE task_id = ? AND repository_id = ?`,
tp.TaskID, tp.RepositoryID); err != nil {
`DELETE FROM github_task_prs
WHERE task_id = ? AND repository_id = ? AND pr_number = ?`,
tp.TaskID, tp.RepositoryID, tp.PRNumber); err != nil {
return err
}
} else {
if _, err := tx.ExecContext(ctx,
`DELETE FROM github_task_prs WHERE task_id = ? AND repository_id = ''`, tp.TaskID); err != nil {
`DELETE FROM github_task_prs
WHERE task_id = ? AND repository_id = '' AND pr_number = ?`,
tp.TaskID, tp.PRNumber); err != nil {
return err
}
}
@@ -83,7 +83,17 @@ func TestTaskPR_PerRepoStorage(t *testing.T) {
}
}
func TestTaskPR_ReplaceTaskPR_ScopedByRepository(t *testing.T) {
// TestTaskPR_ReplaceTaskPR_PreservesSiblingsAndUpserts locks the
// multi-branch upsert semantics of ReplaceTaskPR. The function now
// upserts on (task_id, repository_id, pr_number) instead of wiping all
// rows for (task, repo). Effects:
//
// - Different repo (repo-b) stays untouched.
// - Different PR number on the same repo (repo-a, pr=1) stays as a
// sibling — multi-branch tasks need this so two PRs on the same repo
// coexist instead of one clobbering the other.
// - Same (task, repo, pr_number) is overwritten (the upsert).
func TestTaskPR_ReplaceTaskPR_PreservesSiblingsAndUpserts(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
@@ -92,36 +102,40 @@ func TestTaskPR_ReplaceTaskPR_ScopedByRepository(t *testing.T) {
TaskID: "task-2", RepositoryID: "repo-a",
Owner: "o", Repo: "r", PRNumber: 1, CreatedAt: now,
}); err != nil {
t.Fatalf("create A: %v", err)
t.Fatalf("create A#1: %v", err)
}
if err := store.CreateTaskPR(ctx, &TaskPR{
TaskID: "task-2", RepositoryID: "repo-b",
Owner: "o", Repo: "r2", PRNumber: 2, CreatedAt: now,
}); err != nil {
t.Fatalf("create B: %v", err)
t.Fatalf("create B#2: %v", err)
}
// Replace only repo-a's PR — repo-b must survive.
// Add a new PR #99 on repo-a — must NOT delete repo-a's PR #1
// (multi-branch) nor repo-b's #2 (multi-repo).
if err := store.ReplaceTaskPR(ctx, &TaskPR{
TaskID: "task-2", RepositoryID: "repo-a",
Owner: "o", Repo: "r", PRNumber: 99, CreatedAt: now,
}); err != nil {
t.Fatalf("replace A: %v", err)
t.Fatalf("replace A#99: %v", err)
}
all, _ := store.ListTaskPRsByTask(ctx, "task-2")
if len(all) != 2 {
t.Fatalf("expected 2 PRs after scoped replace, got %d", len(all))
if len(all) != 3 {
t.Fatalf("expected 3 PR rows (multi-branch siblings + multi-repo), got %d", len(all))
}
bySpec := map[string]int{}
for _, p := range all {
bySpec[p.RepositoryID] = p.PRNumber
// Now overwrite repo-a #99 by calling ReplaceTaskPR with the same
// (task, repo, pr_number) and a different field — upsert path.
if err := store.ReplaceTaskPR(ctx, &TaskPR{
TaskID: "task-2", RepositoryID: "repo-a",
Owner: "o", Repo: "r", PRNumber: 99, PRTitle: "updated", CreatedAt: now,
}); err != nil {
t.Fatalf("upsert A#99: %v", err)
}
if bySpec["repo-a"] != 99 {
t.Errorf("expected repo-a updated to 99, got %d", bySpec["repo-a"])
}
if bySpec["repo-b"] != 2 {
t.Errorf("expected repo-b unchanged at 2, got %d", bySpec["repo-b"])
all, _ = store.ListTaskPRsByTask(ctx, "task-2")
if len(all) != 3 {
t.Fatalf("expected 3 rows after upsert, got %d", len(all))
}
}