fix: pipelines died at spawn on shared-branch contention (and ~ in project paths)

internal/db/project_path.go

@@ -0,0 +1,48 @@
+package db
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+// ExpandHomePath turns a leading "~" into the user's home directory.
+//
+// A project path reaches the DB from several places — the settings form, the
+// project-detect flow, `ty project add`, a hand-edited row — and some of them
+// store the shell's own shorthand, "~/Projects/foo". Nothing downstream runs
+// through a shell: the executor and the pipeline hand these paths straight to
+// exec.Command("git", "-C", dir, ...), where a literal "~" is just a directory
+// name that does not exist. The failure is remote from its cause — a pipeline
+// aborting with
+//
+//	fatal: cannot change to '~/Projects/rails/offerlab': No such file or directory
+//
+// and a GetProjectByPath that silently never matches, so project detection and
+// project-local workflow dirs quietly do nothing.
+//
+// Callers used to be expected to remember config.GetProjectDir; the ones that
+// read Project.Path directly (three of them, at the time of writing) did not.
+// Normalizing on both write and read removes the chance to forget.
+//
+// A "~user/..." form is left alone: resolving another user's home is not
+// something we can do reliably, and it has never been a path we support.
+func ExpandHomePath(path string) string {
+	p := strings.TrimSpace(path)
+	if p != "~" && !strings.HasPrefix(p, "~/") {
+		return path
+	}
+	home, err := os.UserHomeDir()
+	if err != nil || home == "" {
+		return path
+	}
+	return filepath.Join(home, strings.TrimPrefix(p, "~"))
+}
+
+// normalizePath rewrites the project's path in place to its expanded form.
+func (p *Project) normalizePath() {
+	if p == nil {
+		return
+	}
+	p.Path = ExpandHomePath(p.Path)
+}

internal/db/project_path_test.go

@@ -0,0 +1,102 @@
+package db
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestExpandHomePath(t *testing.T) {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		t.Skip("no home dir")
+	}
+	cases := []struct {
+		in, want string
+	}{
+		{"~/Projects/rails/offerlab", filepath.Join(home, "Projects/rails/offerlab")},
+		{"~", home},
+		{"/Users/someone/Projects/x", "/Users/someone/Projects/x"},
+		{"relative/path", "relative/path"},
+		{"", ""},
+		// Another user's home is not something we can resolve; leave it alone.
+		{"~someone/Projects", "~someone/Projects"},
+		// A tilde in the middle is a real (if odd) directory name.
+		{"/tmp/a~b", "/tmp/a~b"},
+	}
+	for _, c := range cases {
+		if got := ExpandHomePath(c.in); got != c.want {
+			t.Errorf("ExpandHomePath(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
+
+// A project path stored as the shell's "~/..." shorthand must never reach a
+// consumer that way: nothing downstream goes through a shell, so a literal "~"
+// becomes a directory that does not exist and git fails with
+// "cannot change to '~/Projects/rails/offerlab'" — which is how a pipeline lost
+// its shared branch on origin.
+func TestProjectPathIsExpandedOnWriteAndRead(t *testing.T) {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		t.Skip("no home dir")
+	}
+	database, err := Open(filepath.Join(t.TempDir(), "tasks.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer database.Close()
+
+	want := filepath.Join(home, "Projects/rails/offerlab")
+	p := &Project{Name: "offerlab", Path: "~/Projects/rails/offerlab"}
+	if err := database.CreateProject(p); err != nil {
+		t.Fatal(err)
+	}
+	if p.Path != want {
+		t.Errorf("CreateProject left the caller holding %q, want %q", p.Path, want)
+	}
+
+	got, err := database.GetProjectByName("offerlab")
+	if err != nil || got == nil {
+		t.Fatalf("GetProjectByName: %v", err)
+	}
+	if got.Path != want {
+		t.Errorf("GetProjectByName path = %q, want %q", got.Path, want)
+	}
+
+	list, err := database.ListProjects()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, lp := range list {
+		if lp.Name == "offerlab" && lp.Path != want {
+			t.Errorf("ListProjects path = %q, want %q", lp.Path, want)
+		}
+	}
+}
+
+// Rows written before the expansion existed (hand-edited, or by an older build)
+// must still come back usable.
+func TestLegacyTildeRowIsExpandedOnRead(t *testing.T) {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		t.Skip("no home dir")
+	}
+	database, err := Open(filepath.Join(t.TempDir(), "tasks.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer database.Close()
+
+	if _, err := database.Exec(`INSERT INTO projects (name, path) VALUES (?, ?)`,
+		"legacy", "~/Projects/rails/influencekit"); err != nil {
+		t.Fatal(err)
+	}
+	got, err := database.GetProjectByName("legacy")
+	if err != nil || got == nil {
+		t.Fatalf("GetProjectByName: %v", err)
+	}
+	if want := filepath.Join(home, "Projects/rails/influencekit"); got.Path != want {
+		t.Errorf("legacy row path = %q, want %q", got.Path, want)
+	}
+}

internal/db/tasks.go

@@ -428,6 +428,31 @@ func (db *DB) GetTask(id int64) (*Task, error) {
 	return t, nil
 }
 
+// GetTaskByWorktreePath returns the task that owns a worktree directory, or nil.
+//
+// Used when a branch turns out to be checked out somewhere: git names the
+// worktree holding it, and the only way to decide whether that branch can be
+// reclaimed is to ask what the owning task is doing. Trashed tasks are excluded
+// — a deleted task's worktree is nobody's live work.
+func (db *DB) GetTaskByWorktreePath(path string) (*Task, error) {
+	if strings.TrimSpace(path) == "" {
+		return nil, nil
+	}
+	var id int64
+	err := db.QueryRow(`
+		SELECT id FROM tasks
+		WHERE worktree_path = ? AND deleted_at IS NULL
+		ORDER BY id DESC LIMIT 1
+	`, path).Scan(&id)
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("query task by worktree: %w", err)
+	}
+	return db.GetTask(id)
+}
+
 // ListTasksOptions defines options for listing tasks.
 type ListTasksOptions struct {
 	Status         string
@@ -1712,6 +1737,7 @@ func boolToInt(b bool) int {
 
 // CreateProject creates a new project.
 func (db *DB) CreateProject(p *Project) error {
+	p.normalizePath()
 	actionsJSON, _ := json.Marshal(p.Actions)
 	result, err := db.Exec(`
 		INSERT INTO projects (name, path, aliases, instructions, actions, color, claude_config_dir, use_worktrees, default_permission_mode)
@@ -1727,6 +1753,7 @@ func (db *DB) CreateProject(p *Project) error {
 
 // UpdateProject updates a project.
 func (db *DB) UpdateProject(p *Project) error {
+	p.normalizePath()
 	actionsJSON, _ := json.Marshal(p.Actions)
 	_, err := db.Exec(`
 		UPDATE projects SET name = ?, path = ?, aliases = ?, instructions = ?, actions = ?, color = ?, claude_config_dir = ?, use_worktrees = ?, default_permission_mode = ?
@@ -1790,6 +1817,7 @@ func (db *DB) ListProjects() ([]*Project, error) {
 		}
 		json.Unmarshal([]byte(actionsJSON), &p.Actions)
 		p.UseWorktrees = useWorktrees != 0
+		p.normalizePath()
 		projects = append(projects, p)
 	}
 	return projects, nil
@@ -1808,6 +1836,7 @@ func (db *DB) GetProjectByName(name string) (*Project, error) {
 	if err == nil {
 		json.Unmarshal([]byte(actionsJSON), &p.Actions)
 		p.UseWorktrees = useWorktrees != 0
+		p.normalizePath()
 		return p, nil
 	}
 	if err != sql.ErrNoRows {
@@ -1828,6 +1857,7 @@ func (db *DB) GetProjectByName(name string) (*Project, error) {
 		}
 		json.Unmarshal([]byte(actionsJSON), &p.Actions)
 		p.UseWorktrees = useWorktrees != 0
+		p.normalizePath()
 		for _, alias := range splitAliases(p.Aliases) {
 			if alias == name {
 				return p, nil

internal/executor/executor.go

@@ -5,6 +5,7 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -1863,6 +1864,18 @@ func (e *Executor) executeTask(ctx context.Context, task *db.Task) {
 	// SECURITY: We must have a valid worktree - never fall back to project directory
 	// to prevent Claude from accidentally writing to the main repo
 	workDir, createdWorktree, err := e.setupWorktree(task)
+	if errors.Is(err, ErrBranchBusy) {
+		// Not a failure: the branch this step needs is held by a sibling that is
+		// still running. Leave the task QUEUED so the next tick retries it, and
+		// leave no completion timestamps behind — a step parked 'blocked' with a
+		// started_at/completed_at pair reads as a step that ran, which is how a
+		// pipeline silently loses a phase.
+		e.logger.Info("Deferring step until the branch it needs is free", "id", task.ID, "error", err)
+		if err := e.db.UpdateTaskStatus(task.ID, db.StatusQueued); err != nil {
+			e.logger.Error("Failed to requeue deferred step", "id", task.ID, "error", err)
+		}
+		return
+	}
 	if err != nil {
 		e.logger.Error("Failed to setup worktree", "error", err)
 		e.logLine(task.ID, "error", fmt.Sprintf("Failed to setup worktree: %v", err))
@@ -4546,31 +4559,38 @@ func (e *Executor) setupWorktree(task *db.Task) (string, bool, error) {
 		// next step's worktree, built from that branch, sees none of the work and the
 		// whole phase is silently lost. addSourceBranchWorktree picks the form of the
 		// command that guarantees attachment.
-		if err := e.addSourceBranchWorktree(projectDir, worktreePath, task.SourceBranch); err != nil {
-			return "", false, err
+		// A FAN-OUT step arrives with its own branch already pinned (BranchName),
+		// cut from the shared branch (SourceBranch), because siblings that run at
+		// the same time cannot all attach to one branch. A sequential step has no
+		// branch of its own and attaches to the shared branch itself.
+		if stepBranch := task.BranchName; stepBranch != "" && stepBranch != task.SourceBranch {
+			if err := e.addStepBranchWorktree(projectDir, worktreePath, stepBranch, task.SourceBranch); err != nil {
+				return "", false, err
+			}
+			branchName = stepBranch
+		} else {
+			if err := e.addSourceBranchWorktree(projectDir, worktreePath, task.SourceBranch); err != nil {
+				return "", false, err
+			}
+			branchName = task.SourceBranch
 		}
 
-		// Use the source branch name as the branch name for the task
-		branchName = task.SourceBranch
-
 		// Update task with worktree info
 		task.WorktreePath = worktreePath
 		task.BranchName = branchName
 	} else {
 		// Get default branch name
 		defaultBranch := e.getDefaultBranch(projectDir)
 
-		// Create new branch and worktree
-		cmd := exec.Command("git", "worktree", "add", "-b", branchName, worktreePath, defaultBranch)
-		cmd.Dir = projectDir
-		output, err := cmd.CombinedOutput()
+		// Create new branch and worktree. Serialized per repo: two ordinary tasks
+		// starting at the same moment would otherwise race on .git/config, and
+		// the loser fails outright rather than retrying.
+		output, err := runGitWorktreeAddOutput(projectDir, "worktree", "add", "-b", branchName, worktreePath, defaultBranch)
 		if err != nil {
 			// Check if branch already exists
 			if strings.Contains(string(output), "already exists") {
 				// Try using existing branch
-				cmd = exec.Command("git", "worktree", "add", worktreePath, branchName)
-				cmd.Dir = projectDir
-				output2, err2 := cmd.CombinedOutput()
+				output2, err2 := runGitWorktreeAddOutput(projectDir, "worktree", "add", worktreePath, branchName)
 				if err2 != nil {
 					// Check if worktree was created by another process
 					if strings.Contains(string(output2), "already checked out") {
@@ -5660,6 +5680,19 @@ func (e *Executor) addSourceBranchWorktree(projectDir, worktreePath, sourceBranc
 	remoteRef := "origin/" + sourceBranch
 	remoteExists := gitRefExists(projectDir, "refs/remotes/"+remoteRef)
 
+	// The shared branch is sequential by nature, and a finished step's worktree
+	// keeps holding it. Reclaim it from a step that is done; wait (retryably) for
+	// one that is still working.
+	if holder := gitWorktreeHolder(projectDir, sourceBranch); holder != "" {
+		freed, err := e.releaseBranchFromFinishedHolder(projectDir, sourceBranch)
+		if err != nil {
+			return err
+		}
+		if !freed {
+			return fmt.Errorf("%w: %s is checked out at %s", ErrBranchBusy, sourceBranch, holder)
+		}
+	}
+
 	var args []string
 	switch {
 	case localExists:
@@ -5683,10 +5716,8 @@ func (e *Executor) addSourceBranchWorktree(projectDir, worktreePath, sourceBranc
 		return fmt.Errorf("source branch %s not found on origin or locally", sourceBranch)
 	}
 
-	cmd := exec.Command("git", args...)
-	cmd.Dir = projectDir
-	if output, err := cmd.CombinedOutput(); err != nil {
-		return fmt.Errorf("create worktree on branch %s: %v\n%s", sourceBranch, err, string(output))
+	if err := runGitWorktreeAdd(projectDir, sourceBranch, args...); err != nil {
+		return err
 	}
 
 	// Belt and braces: if git still handed back a detached HEAD, fail loudly here