diff --git a/internal/db/project_path.go b/internal/db/project_path.go new file mode 100644 index 00000000..620a5ea4 --- /dev/null +++ b/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) +} diff --git a/internal/db/project_path_test.go b/internal/db/project_path_test.go new file mode 100644 index 00000000..2da70c17 --- /dev/null +++ b/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) + } +} diff --git a/internal/db/tasks.go b/internal/db/tasks.go index 726a70b1..74c23c0e 100644 --- a/internal/db/tasks.go +++ b/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 diff --git a/internal/executor/executor.go b/internal/executor/executor.go index dd639402..16d6fae2 100644 --- a/internal/executor/executor.go +++ b/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,13 +4559,22 @@ 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 @@ -4560,17 +4582,15 @@ func (e *Executor) setupWorktree(task *db.Task) (string, bool, error) { // 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 diff --git a/internal/executor/shared_branch.go b/internal/executor/shared_branch.go new file mode 100644 index 00000000..0796c24f --- /dev/null +++ b/internal/executor/shared_branch.go @@ -0,0 +1,259 @@ +package executor + +import ( + "errors" + "fmt" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/charmbracelet/log" + + "github.com/bborn/workflow/internal/db" + "github.com/bborn/workflow/internal/executorlock" +) + +// ErrBranchBusy means a step could not start because the branch it needs is +// checked out by a worktree whose task is still RUNNING. +// +// This is a wait, not a failure. The old behaviour — bubbling the raw git error +// out of setupWorktree — parked the step 'blocked' with started_at and +// completed_at one second apart: a step that reads as "ran and finished" on +// every surface, having never launched an agent. executeTask leaves an +// ErrBranchBusy task 'queued' instead, so the next daemon tick retries it once +// the holder is done. +var ErrBranchBusy = errors.New("branch is checked out by a running step") + +// gitWorktreeHolder returns the path of the worktree that currently has branch +// checked out, or "" if no worktree holds it. +// +// git allows a branch to be attached to at most ONE worktree. Everything in a +// workflow that shares a branch — a finished step whose worktree lingers for +// review, the next step trying to attach — is contending for that single slot, +// so the first question on failure is always "who has it?". +func gitWorktreeHolder(projectDir, branch string) string { + out, err := exec.Command("git", "-C", projectDir, "worktree", "list", "--porcelain").Output() + if err != nil { + return "" + } + want := "refs/heads/" + branch + current := "" + for _, line := range strings.Split(string(out), "\n") { + switch { + case strings.HasPrefix(line, "worktree "): + current = strings.TrimSpace(strings.TrimPrefix(line, "worktree ")) + case strings.HasPrefix(line, "branch "): + if strings.TrimSpace(strings.TrimPrefix(line, "branch ")) == want { + return current + } + } + } + return "" +} + +// gitDetachWorktree points a worktree's HEAD at its current commit instead of a +// branch, freeing that branch for another worktree. +// +// This is the cheapest possible release: `checkout --detach` with no ref moves +// nothing in the working tree, keeps any uncommitted files exactly as they are, +// and leaves the branch ref pointing at the same commit. The worktree stays on +// disk and stays readable — which matters, because a finished step's worktree is +// where a human goes to see what it did. +func gitDetachWorktree(worktreePath string) error { + cmd := exec.Command("git", "-C", worktreePath, "checkout", "--detach") + if out, err := cmd.CombinedOutput(); err != nil { + return fmt.Errorf("detach %s: %v: %s", worktreePath, err, strings.TrimSpace(string(out))) + } + return nil +} + +// releaseBranchFromFinishedHolder frees branch when the worktree holding it +// belongs to a step that is no longer running, and reports whether the branch is +// now available. +// +// A workflow's steps run one after another on one shared branch, but a step's +// worktree is not torn down when it completes — it is kept so its work can be +// inspected, and it goes on holding the branch forever. So the FIRST downstream +// step of every pipeline hit "fatal: '' is already checked out at ..." +// and died at spawn. Nothing about that is exceptional; it is the normal shape of +// a finished step, and the fix is to take the branch back rather than to refuse. +// +// Releasing is safe precisely because the holder is finished: its commits are on +// the branch ref (that is what "finished" means here — WorkflowStepFinished +// requires a commit that was pushed), and detaching cannot move or discard them. +// +// A holder that is still working keeps the branch. That returns (false, nil): +// not an error, just "not yet". +func (e *Executor) releaseBranchFromFinishedHolder(projectDir, branch string) (bool, error) { + holder := gitWorktreeHolder(projectDir, branch) + if holder == "" { + return true, nil // nobody has it + } + + task, err := e.taskForWorktree(holder) + if err != nil { + return false, fmt.Errorf("look up the task holding %s: %w", branch, err) + } + if task == nil { + // A worktree with no task behind it (hand-made, or its task was deleted). + // Leave it alone: we only ever reclaim a branch from a step we can prove + // has finished. + e.logger.Warn("branch held by a worktree with no task; not reclaiming", + "branch", branch, "worktree", holder) + return false, nil + } + + if e.taskIsLive(task) { + return false, nil + } + + if err := gitDetachWorktree(holder); err != nil { + return false, err + } + e.logger.Info("reclaimed shared branch from a finished step", + "branch", branch, "from_task", task.ID, "worktree", holder) + e.logLine(task.ID, "system", fmt.Sprintf( + "Detached this worktree's HEAD so branch %s could pass to the next step. Your commits are on the branch; the files here are untouched.", branch)) + return true, nil +} + +// taskForWorktree finds the task that owns a worktree git just named for us. +// +// git reports a worktree by its SYMLINK-RESOLVED path, while the DB holds the +// path the executor built when it created the worktree — on macOS those differ +// for anything under /tmp or /var ("/private/var/..." vs "/var/..."), and a +// plain string comparison then finds nothing. Reading "nothing" as "no task owns +// this" would make us refuse to reclaim a branch from a step that is long +// finished, which is the exact stall this code exists to clear. So compare the +// forms the two sides can legitimately disagree about. +func (e *Executor) taskForWorktree(holder string) (*db.Task, error) { + seen := make(map[string]bool, 4) + for _, candidate := range worktreePathForms(holder) { + if candidate == "" || seen[candidate] { + continue + } + seen[candidate] = true + task, err := e.db.GetTaskByWorktreePath(candidate) + if err != nil { + return nil, err + } + if task != nil { + return task, nil + } + } + return nil, nil +} + +// worktreePathForms returns the equivalent spellings of a path that the DB might +// hold: as given, symlink-resolved, and with macOS's /private prefix added or +// removed. +func worktreePathForms(path string) []string { + forms := []string{path} + if resolved, err := filepath.EvalSymlinks(path); err == nil { + forms = append(forms, resolved) + } + if strings.HasPrefix(path, "/private/") { + forms = append(forms, strings.TrimPrefix(path, "/private")) + } else { + forms = append(forms, filepath.Join("/private", path)) + } + return forms +} + +// taskIsLive reports whether a task is actively executing right now: mid-flight +// in this daemon, in a running status, or still owning a tmux window. Any of the +// three means "hands off". +func (e *Executor) taskIsLive(task *db.Task) bool { + if task == nil { + return false + } + e.mu.Lock() + running := e.runningTasks[task.ID] + e.mu.Unlock() + if running { + return true + } + if task.Status == db.StatusProcessing || task.Status == db.StatusQueued { + return true + } + return tmuxWindowExistsForTask(task.ID) +} + +// addStepBranchWorktree creates a worktree for a FAN-OUT step on its own branch, +// cut from the shared branch. +// +// Steps that run at the same time cannot share one branch — git attaches a +// branch to a single worktree, so the second sibling's `worktree add` fails +// outright. Each therefore gets `-`: the exact branch its +// composed instructions already tell it to push to, and the branch its dependent +// step is already told to read back with `git show origin/:`. +// Branching FROM the shared branch is unrestricted (only checking it out is), so +// no number of siblings can contend. +func (e *Executor) addStepBranchWorktree(projectDir, worktreePath, stepBranch, sharedBranch string) error { + // A retry after the step already ran once: keep its existing branch, which + // carries whatever it committed before, instead of re-cutting from the base. + if gitRefExists(projectDir, "refs/heads/"+stepBranch) { + if holder := gitWorktreeHolder(projectDir, stepBranch); holder != "" && holder != worktreePath { + if freed, err := e.releaseBranchFromFinishedHolder(projectDir, stepBranch); err != nil { + return err + } else if !freed { + return fmt.Errorf("%w: %s is checked out at %s", ErrBranchBusy, stepBranch, holder) + } + } + return runGitWorktreeAdd(projectDir, stepBranch, "worktree", "add", worktreePath, stepBranch) + } + + base := "origin/" + sharedBranch + if !gitRefExists(projectDir, "refs/remotes/"+base) { + // The shared branch may be local-only: a document root hands off through + // the artifact store and is told not to push. + base = sharedBranch + if !gitRefExists(projectDir, "refs/heads/"+base) { + return fmt.Errorf("shared branch %s not found on origin or locally", sharedBranch) + } + } + // --no-track: `worktree add -b X origin/` would set X's + // upstream to the SHARED branch, so a later bare `git push` from the step + // would push its commits onto the branch its instructions explicitly tell it + // not to touch. + return runGitWorktreeAdd(projectDir, stepBranch, "worktree", "add", "--no-track", "-b", stepBranch, worktreePath, base) +} + +// runGitWorktreeAdd runs a `git worktree add`, serialized against any other +// worktree creation in the same repository, and reports a failure with the git +// output — the only thing that explains what actually went wrong. +// +// The lock matters because `git worktree add` writes .git/config and fails +// outright rather than retrying when a concurrent add holds git's own config +// lock. Fan-out steps spawn in the same instant, so without this the second +// sibling dies with "could not lock config file .git/config: File exists". +func runGitWorktreeAdd(projectDir, branch string, args ...string) error { + out, err := runGitWorktreeAddOutput(projectDir, args...) + if err != nil { + return fmt.Errorf("create worktree on branch %s: %v\n%s", branch, err, string(out)) + } + return nil +} + +// runGitWorktreeAddOutput is runGitWorktreeAdd for callers that need to inspect +// git's output themselves (the ordinary-task path branches on "already exists" +// and "already checked out"). +func runGitWorktreeAddOutput(projectDir string, args ...string) ([]byte, error) { + if release, err := executorlock.AcquireRepo(executorSpawnLockDir(), projectDir, repoLockTimeout); err == nil { + defer release() + } else { + // Liveness over safety: a wedged holder must not stall every task + // forever. Worst case we are back to the unserialized behaviour, which + // fails loudly and is retried. + log.Warn("proceeding without the repo worktree lock", "repo", projectDir, "error", err) + } + cmd := exec.Command("git", args...) + cmd.Dir = projectDir + return cmd.CombinedOutput() +} + +// repoLockTimeout bounds the wait for another worktree creation in the same +// repo. A `git worktree add` on a large repo (checkout of every tracked file) +// can take a while, so this is generous compared with the spawn lock. +const repoLockTimeout = 120 * time.Second diff --git a/internal/executor/shared_branch_test.go b/internal/executor/shared_branch_test.go new file mode 100644 index 00000000..6d800381 --- /dev/null +++ b/internal/executor/shared_branch_test.go @@ -0,0 +1,268 @@ +package executor + +import ( + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + + "github.com/bborn/workflow/internal/config" + "github.com/bborn/workflow/internal/db" +) + +// sharedBranchRepo builds a real git repo with one commit on a shared pipeline +// branch. These tests are about what git actually permits (one worktree per +// branch), so a fake would prove nothing. +func sharedBranchRepo(t *testing.T) (repo, branch string) { + t.Helper() + repo = t.TempDir() + branch = "pipeline/1-demo" + run := func(args ...string) { + t.Helper() + cmd := exec.Command("git", args...) + cmd.Dir = repo + cmd.Env = append(os.Environ(), + "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@t", "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@t") + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("git %s: %v\n%s", strings.Join(args, " "), err, out) + } + } + run("init", "-b", "main") + if err := os.WriteFile(filepath.Join(repo, "README.md"), []byte("x\n"), 0o644); err != nil { + t.Fatal(err) + } + run("add", "-A") + run("commit", "-m", "base") + run("branch", branch) + return repo, branch +} + +func sharedBranchExecutor(t *testing.T, repo string) (*Executor, *db.DB) { + t.Helper() + database, err := db.Open(filepath.Join(t.TempDir(), "tasks.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { database.Close() }) + if err := database.CreateProject(&db.Project{Name: "p", Path: repo}); err != nil { + t.Fatal(err) + } + return New(database, &config.Config{}), database +} + +// holdBranch gives an existing task a worktree attached to branch, the way a +// step that has run leaves things behind. +func holdBranch(t *testing.T, e *Executor, database *db.DB, repo, branch, status string) *db.Task { + t.Helper() + holderPath := filepath.Join(t.TempDir(), "holder") + cmd := exec.Command("git", "worktree", "add", holderPath, branch) + cmd.Dir = repo + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("seed holder worktree: %v\n%s", err, out) + } + task := &db.Task{Title: "[Plan] x", Status: status, Project: "p", Tags: "pipeline", BranchName: branch} + if err := database.CreateTask(task); err != nil { + t.Fatal(err) + } + task.WorktreePath = holderPath + task.Status = status + if err := database.UpdateTask(task); err != nil { + t.Fatal(err) + } + if err := database.UpdateTaskStatus(task.ID, status); err != nil { + t.Fatal(err) + } + return task +} + +// The bug this whole change exists for: a pipeline's first step finishes, its +// worktree keeps the shared branch checked out, and the NEXT step can never +// attach — "fatal: '' is already checked out at ...". A finished holder +// must hand the branch over. +func TestSourceBranchWorktreeReclaimsBranchFromFinishedStep(t *testing.T) { + repo, branch := sharedBranchRepo(t) + e, database := sharedBranchExecutor(t, repo) + holder := holdBranch(t, e, database, repo, branch, db.StatusDone) + + next := filepath.Join(t.TempDir(), "next") + if err := e.addSourceBranchWorktree(repo, next, branch); err != nil { + t.Fatalf("next step could not take the shared branch: %v", err) + } + + got, err := gitCurrentBranch(next) + if err != nil { + t.Fatal(err) + } + if got != branch { + t.Fatalf("next step worktree is on %q, want it attached to %q", got, branch) + } + // The finished step keeps its files; only its HEAD moved off the branch. + if _, err := os.Stat(filepath.Join(holder.WorktreePath, "README.md")); err != nil { + t.Fatalf("finished step's worktree should stay readable: %v", err) + } + if head, err := gitCurrentBranch(holder.WorktreePath); err != nil || head != "HEAD" { + t.Fatalf("holder should be detached, got %q (err %v)", head, err) + } +} + +// A holder that is still RUNNING keeps the branch, and the waiting step must be +// told to come back later rather than being failed. A hard failure here is what +// stamped started_at/completed_at on a step that never launched an agent. +func TestSourceBranchWorktreeDefersWhileHolderIsRunning(t *testing.T) { + repo, branch := sharedBranchRepo(t) + e, database := sharedBranchExecutor(t, repo) + holdBranch(t, e, database, repo, branch, db.StatusProcessing) + + err := e.addSourceBranchWorktree(repo, filepath.Join(t.TempDir(), "next"), branch) + if !errors.Is(err, ErrBranchBusy) { + t.Fatalf("want ErrBranchBusy while the holder runs, got %v", err) + } +} + +// Fan-out: every sibling gets a worktree on its OWN branch, cut from the shared +// branch, and none of them contends with the others or with the root's worktree. +func TestStepBranchWorktreesDoNotContend(t *testing.T) { + repo, branch := sharedBranchRepo(t) + e, database := sharedBranchExecutor(t, repo) + holdBranch(t, e, database, repo, branch, db.StatusProcessing) // root still running + + for _, name := range []string{"planreviewa", "planreviewb", "planreviewc"} { + stepBranch := branch + "-" + name + path := filepath.Join(t.TempDir(), name) + if err := e.addStepBranchWorktree(repo, path, stepBranch, branch); err != nil { + t.Fatalf("%s could not get its own branch: %v", name, err) + } + got, err := gitCurrentBranch(path) + if err != nil { + t.Fatal(err) + } + if got != stepBranch { + t.Fatalf("%s is on %q, want %q", name, got, stepBranch) + } + } +} + +// A retried fan-out step reuses its own branch, keeping whatever it committed +// the first time rather than re-cutting from the shared branch. +func TestStepBranchWorktreeReusesExistingBranchOnRetry(t *testing.T) { + repo, branch := sharedBranchRepo(t) + e, _ := sharedBranchExecutor(t, repo) + stepBranch := branch + "-planreviewa" + + first := filepath.Join(t.TempDir(), "first") + if err := e.addStepBranchWorktree(repo, first, stepBranch, branch); err != nil { + t.Fatal(err) + } + // The step commits, then its worktree goes away (crash, cleanup, retry). + writeAndCommit(t, first, "REVIEW.md", "findings") + want := gitHeadCommit(first) + rm := exec.Command("git", "worktree", "remove", "--force", first) + rm.Dir = repo + if out, err := rm.CombinedOutput(); err != nil { + t.Fatalf("remove worktree: %v\n%s", err, out) + } + + second := filepath.Join(t.TempDir(), "second") + if err := e.addStepBranchWorktree(repo, second, stepBranch, branch); err != nil { + t.Fatalf("retry could not reattach: %v", err) + } + if got := gitHeadCommit(second); got != want { + t.Fatalf("retry started at %s, want the step's own commit %s", got, want) + } +} + +func writeAndCommit(t *testing.T, dir, name, content string) { + t.Helper() + if err := os.WriteFile(filepath.Join(dir, name), []byte(content+"\n"), 0o644); err != nil { + t.Fatal(err) + } + for _, args := range [][]string{{"add", "-A"}, {"commit", "-m", name}} { + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), + "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@t", "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@t") + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("git %s: %v\n%s", strings.Join(args, " "), err, out) + } + } +} + +// Fan-out siblings spawn in the same instant, and `git worktree add` is not safe +// to run concurrently in one repository: it writes .git/config under a lock of +// git's own and the loser fails outright with "could not lock config file +// .git/config: File exists". Creating every sibling's worktree at once must +// still produce every worktree. +func TestConcurrentStepBranchWorktreesAllSucceed(t *testing.T) { + repo, branch := sharedBranchRepo(t) + e, _ := sharedBranchExecutor(t, repo) + + const siblings = 6 + base := t.TempDir() + errs := make(chan error, siblings) + start := make(chan struct{}) + for i := 0; i < siblings; i++ { + go func(i int) { + <-start // release them together + name := fmt.Sprintf("review%d", i) + errs <- e.addStepBranchWorktree(repo, filepath.Join(base, name), branch+"-"+name, branch) + }(i) + } + close(start) + for i := 0; i < siblings; i++ { + if err := <-errs; err != nil { + t.Errorf("concurrent worktree add failed: %v", err) + } + } + + for i := 0; i < siblings; i++ { + name := fmt.Sprintf("review%d", i) + got, err := gitCurrentBranch(filepath.Join(base, name)) + if err != nil { + t.Fatalf("%s: %v", name, err) + } + if want := branch + "-" + name; got != want { + t.Errorf("%s is on %q, want %q", name, got, want) + } + } +} + +// A fan-out step's branch must NOT track the shared branch: `worktree add -b X +// origin/` sets that upstream by default, so a later bare +// `git push` would land the step's commits on the branch its instructions +// explicitly tell it not to touch. +func TestStepBranchDoesNotTrackTheSharedBranch(t *testing.T) { + repo, branch := sharedBranchRepo(t) + e, _ := sharedBranchExecutor(t, repo) + + // Give the repo an "origin" so the remote-ref path is the one exercised. + remote := t.TempDir() + mustGit(t, remote, "init", "--bare") + mustGit(t, repo, "remote", "add", "origin", remote) + mustGit(t, repo, "push", "origin", branch) + mustGit(t, repo, "fetch", "origin") + + path := filepath.Join(t.TempDir(), "reviewa") + stepBranch := branch + "-reviewa" + if err := e.addStepBranchWorktree(repo, path, stepBranch, branch); err != nil { + t.Fatal(err) + } + + out, err := exec.Command("git", "-C", repo, "config", "--get", "branch."+stepBranch+".merge").Output() + if upstream := strings.TrimSpace(string(out)); err == nil && upstream != "" { + t.Errorf("%s tracks %q; a bare `git push` would clobber the shared branch", stepBranch, upstream) + } +} + +func mustGit(t *testing.T, dir string, args ...string) { + t.Helper() + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), + "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@t", "GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@t") + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("git %s: %v\n%s", strings.Join(args, " "), err, out) + } +} diff --git a/internal/executorlock/executorlock.go b/internal/executorlock/executorlock.go index 8f1cfede..125104a9 100644 --- a/internal/executorlock/executorlock.go +++ b/internal/executorlock/executorlock.go @@ -14,6 +14,7 @@ package executorlock import ( + "crypto/sha256" "errors" "fmt" "os" @@ -26,6 +27,10 @@ import ( // taken before the timeout elapsed (another spawner is holding it). var ErrSpawnLockTimeout = errors.New("executorlock: timed out waiting for spawn lock") +// ErrRepoLockTimeout is returned by AcquireRepo when the repository lock could +// not be taken before the timeout elapsed. +var ErrRepoLockTimeout = errors.New("executorlock: timed out waiting for repo worktree lock") + // spawnPollInterval is how often AcquireSpawn retries the non-blocking flock // while waiting for a concurrent holder to release. const spawnPollInterval = 25 * time.Millisecond @@ -68,3 +73,47 @@ func AcquireSpawn(lockDir string, taskID int64, timeout time.Duration) (func(), time.Sleep(spawnPollInterval) } } + +// RepoLockPath returns the lock-file path serializing worktree creation in one +// repository. The repo path is hashed so the file name is filesystem-safe and +// bounded regardless of how deep the repo lives. +func RepoLockPath(lockDir, repoPath string) string { + sum := sha256.Sum256([]byte(filepath.Clean(repoPath))) + return filepath.Join(lockDir, fmt.Sprintf("git-worktree-%x.lock", sum[:8])) +} + +// AcquireRepo takes an exclusive, cross-process lock for a repository, blocking +// up to timeout, and returns a release func. +// +// `git worktree add` is not safe to run concurrently in the same repository: it +// writes .git/config (to record the new worktree's upstream) under a lock file +// of git's own, and the loser does not retry — it fails outright with +// +// error: could not lock config file .git/config: File exists +// +// Nothing serialized this before because nothing could reach it concurrently: +// steps sharing one branch were forced to run one at a time by git itself. Once +// a workflow's parallel steps each got their own branch they spawn together, and +// two `worktree add` calls land in the same repo in the same instant. +func AcquireRepo(lockDir, repoPath string, timeout time.Duration) (func(), error) { + path := RepoLockPath(lockDir, repoPath) + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0644) + if err != nil { + return nil, fmt.Errorf("open repo lock file: %w", err) + } + + deadline := time.Now().Add(timeout) + for { + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err == nil { + return func() { + _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) + _ = f.Close() + }, nil + } + if !time.Now().Before(deadline) { + _ = f.Close() + return nil, fmt.Errorf("%w: %s", ErrRepoLockTimeout, repoPath) + } + time.Sleep(spawnPollInterval) + } +} diff --git a/internal/pipeline/compose.go b/internal/pipeline/compose.go index a6d9fc0b..09980b3f 100644 --- a/internal/pipeline/compose.go +++ b/internal/pipeline/compose.go @@ -44,6 +44,23 @@ func (d Definition) dependents(name string) []Step { return out } +// StepBranch is the branch a fan-out step publishes to: the shared branch plus +// the step's slug. Create pins it on the task (so the step gets a worktree +// attached to it) and composeInstruction names it in the handoff (so the agent +// pushes there, and so the dependent step knows where to read its inputs). Both +// go through this one function: a step whose worktree and whose instructions +// disagreed about its branch is exactly how the fan-out broke before. +func StepBranch(shared, stepName string) string { + return shared + stepBranchSuffix(stepName) +} + +// stepBranchSuffix is the "-" tail of a fan-out step's branch. Kept +// separate because composeInstruction builds the name against the "{{branch}}" +// placeholder, before the shared branch is known. +func stepBranchSuffix(stepName string) string { + return "-" + slugify(stepName, 40) +} + // hasParallelPeer reports whether the step runs at the same time as another step, // so its output must go to its own branch to avoid clobbering the peer. Two steps // are parallel if they share a dependency; multiple root steps (no deps) are all @@ -96,16 +113,17 @@ func composeInstruction(def Definition, step Step) string { if len(parallelDeps) > 0 { b.WriteString("- Your inputs were produced in parallel and pushed to their own branches; read each:\n") for _, dep := range parallelDeps { - slug := slugify(dep.Name, 40) - fmt.Fprintf(&b, " - **%s** → `git fetch origin && git show origin/{{branch}}-%s:` (branch `{{branch}}-%s`)\n", dep.Name, slug, slug) + depBranch := StepBranch("{{branch}}", dep.Name) + fmt.Fprintf(&b, " - **%s** → `git fetch origin && git show origin/%s:` (branch `%s`)\n", dep.Name, depBranch, depBranch) } } // Output: own branch when parallel, shared branch otherwise. if def.hasParallelPeer(step) { slug := slugify(step.Name, 40) - b.WriteString("- You run in parallel with a sibling step, so push your output to YOUR OWN branch (one commit, one push — no rebase, no clobber):\n") - fmt.Fprintf(&b, " `git add && git commit -m \"%s\" && git push origin HEAD:{{branch}}-%s`\n", slug, slug) + ownBranch := StepBranch("{{branch}}", step.Name) + b.WriteString("- You run in parallel with a sibling step, so your worktree is on YOUR OWN branch; push your output there (one commit, one push — no rebase, no clobber):\n") + fmt.Fprintf(&b, " `git add && git commit -m \"%s\" && git push origin HEAD:%s`\n", slug, ownBranch) b.WriteString(" Do NOT push to `{{branch}}` itself.\n") } else { b.WriteString("- Commit your work and push the shared branch:\n") diff --git a/internal/pipeline/parallel_branch_test.go b/internal/pipeline/parallel_branch_test.go new file mode 100644 index 00000000..fa68b1a8 --- /dev/null +++ b/internal/pipeline/parallel_branch_test.go @@ -0,0 +1,86 @@ +package pipeline + +import ( + "strings" + "testing" + + "github.com/bborn/workflow/internal/db" +) + +// A fan-out step must be given its OWN branch, and it must be the same branch its +// composed handoff tells the agent to push to. +// +// This is the bug that killed every workflow with parallel steps: the +// instructions said "push to {{branch}}-planreviewa", but the task carried only +// SourceBranch, so the executor tried to attach the step to the SHARED branch — +// which a sibling (or the finished root) already had checked out. git allows one +// worktree per branch, so the step died at spawn, before its agent ever ran. +func TestParallelStepsGetTheirOwnBranch(t *testing.T) { + installWorkflow(t, "pcr", pcrYAML) + database := testDB(t) + res, err := Create(database, Options{Goal: "Add rate limiting to the API", Project: "test", Definition: "pcr"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + shared := res.Branch + + rvA := taskByStep(res, "Review A") + rvB := taskByStep(res, "Review B") + + for _, tk := range []*db.Task{rvA, rvB} { + if tk.SourceBranch != shared { + t.Errorf("%s SourceBranch = %q, want the shared branch %q to cut from", tk.Title, tk.SourceBranch, shared) + } + if tk.BranchName == "" || tk.BranchName == shared { + t.Errorf("%s BranchName = %q, want its own branch (siblings cannot share one branch)", tk.Title, tk.BranchName) + } + // The worktree it gets and the branch it is told to push to must agree. + if !strings.Contains(tk.Body, tk.BranchName) { + t.Errorf("%s is on branch %q but its handoff never names it:\n%s", tk.Title, tk.BranchName, tk.Body) + } + } + if rvA.BranchName == rvB.BranchName { + t.Errorf("both reviews got branch %q; they would contend", rvA.BranchName) + } + + // The step that consumes them is told to read those exact branches. + collect := taskByStep(res, "Collect") + for _, tk := range []*db.Task{rvA, rvB} { + if !strings.Contains(collect.Body, tk.BranchName) { + t.Errorf("Collect is not told to read %q:\n%s", tk.BranchName, collect.Body) + } + } +} + +// Sequential steps keep the proven behaviour: they attach to the shared branch +// itself, so work accumulates on it commit by commit. +func TestSequentialStepsStayOnTheSharedBranch(t *testing.T) { + installWorkflow(t, "pcr", pcrYAML) + database := testDB(t) + res, err := Create(database, Options{Goal: "Add rate limiting to the API", Project: "test", Definition: "pcr"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + + code := taskByStep(res, "Code") + if code.BranchName != "" { + t.Errorf("Code (sequential) got its own branch %q, want it on the shared branch", code.BranchName) + } + if code.SourceBranch != res.Branch { + t.Errorf("Code SourceBranch = %q, want %q", code.SourceBranch, res.Branch) + } + // Collect has two deps but no peer running beside it — also sequential. + collect := taskByStep(res, "Collect") + if collect.BranchName != "" { + t.Errorf("Collect got its own branch %q, want it on the shared branch", collect.BranchName) + } +} + +// StepBranch is the single source of truth for a fan-out step's branch name: the +// pinned branch and the name printed in the instructions come from it. +func TestStepBranchMatchesComposedInstructions(t *testing.T) { + shared := "pipeline/12-goal" + if got, want := StepBranch(shared, "Review A"), "pipeline/12-goal-review-a"; got != want { + t.Errorf("StepBranch = %q, want %q", got, want) + } +} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 72c25a17..f135bbcb 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -408,10 +408,23 @@ func Create(database *db.DB, opts Options) (*Result, error) { for _, s := range steps { task := byName[s.Name] task.Body = render(effectiveInstruction(def, s.Name), goal, branch, s.Name, reviewsList(s, branch)) - if rootNames[s.Name] && !multiRoot { + switch { + case rootNames[s.Name] && !multiRoot: task.BranchName = branch // Single root pins/creates the branch. - } else { - task.SourceBranch = branch // Checked out from the shared branch. + case def.hasParallelPeer(s): + // Fan-out. Git allows exactly ONE worktree per branch, so siblings that + // run at the same time cannot all sit on the shared branch — the second + // one's `git worktree add` dies with "is already checked out at ...". + // They therefore each get their own branch, cut from the shared branch: + // exactly what the composed handoff already tells them to push to (see + // composeInstruction). Pinning it here is what finally gives the step a + // worktree ON that branch; before this, the instructions named a branch + // the executor never created, and the whole fan-out failed at spawn. + // The dependent step reads its siblings' output off those branches. + task.SourceBranch = branch + task.BranchName = StepBranch(branch, s.Name) + default: + task.SourceBranch = branch // Sequential: checked out from the shared branch. } if err := database.UpdateTask(task); err != nil { return nil, fmt.Errorf("configure %s step: %w", s.Name, err)