From 10194f63afc395e111d43f3e5e62651772dfef2c Mon Sep 17 00:00:00 2001 From: Bruno Bornsztein Date: Fri, 14 Aug 2026 12:42:12 -0500 Subject: [PATCH 1/4] fix(db): expand ~ in project paths on read and write MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A project path can reach the DB as the shell's own shorthand ("~/Projects/foo") from the settings form, project detect, or a hand-edited row. Nothing downstream runs through a shell — the executor and the pipeline hand these paths straight to exec.Command("git", "-C", dir, ...) — so a literal "~" is just a directory name that does not exist: pipeline: could not pre-seed shared branch "pipeline/5120-..." on origin; git push: fatal: cannot change to '~/Projects/rails/offerlab' config.GetProjectDir expanded it; the callers that read Project.Path directly (pipeline.projectDirFor, executor.lookupKindInstructions, completion.Complete) did not, so those projects silently lost their pre-seeded branch, their project-local .taskyou/workflows dir, and any GetProjectByPath match. Normalizing on both write and read removes the chance to forget. Also adds GetTaskByWorktreePath: given a worktree, which task owns it. Co-Authored-By: Claude Opus 5 (1M context) --- internal/db/project_path.go | 48 +++++++++++++++ internal/db/project_path_test.go | 102 +++++++++++++++++++++++++++++++ internal/db/tasks.go | 30 +++++++++ 3 files changed, 180 insertions(+) create mode 100644 internal/db/project_path.go create mode 100644 internal/db/project_path_test.go 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 From 6236b1ff1849c32538bca0d94e18212d8b0bae3b Mon Sep 17 00:00:00 2001 From: Bruno Bornsztein Date: Fri, 14 Aug 2026 12:42:30 -0500 Subject: [PATCH 2/4] fix(pipeline): give fan-out steps their own branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Steps that run at the same time cannot share one branch: git attaches a branch to at most ONE worktree, so the second sibling's `git worktree add` dies with fatal: 'pipeline/5120-...' is already checked out at ... before its agent ever starts. Every workflow with a fan-out was affected — PlanReviewA/B and the three CodeReviews of design-build-verify never ran. The design was already there in the prompts: composeInstruction has always told a parallel step to push to `{{branch}}-` and told its dependent to read that branch back with `git show origin/{{branch}}-:`. What was missing is that nothing ever gave the step a worktree on that branch — Create pinned only SourceBranch, so the executor tried to attach it to the shared branch that the instructions explicitly say NOT to push to. Create now pins BranchName = StepBranch(shared, step) for any step with a parallel peer, cut from the shared branch. StepBranch is the one source of truth for that name, used by both Create and composeInstruction, so the branch a step is given and the branch its instructions name cannot drift apart again. Sequential steps are unchanged: they attach to the shared branch itself. Co-Authored-By: Claude Opus 5 (1M context) --- internal/pipeline/compose.go | 26 +++++-- internal/pipeline/parallel_branch_test.go | 86 +++++++++++++++++++++++ internal/pipeline/pipeline.go | 19 ++++- 3 files changed, 124 insertions(+), 7 deletions(-) create mode 100644 internal/pipeline/parallel_branch_test.go 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) From 10dd8b3936851da87a6916d3c4ab525afd7d0b88 Mon Sep 17 00:00:00 2001 From: Bruno Bornsztein Date: Fri, 14 Aug 2026 12:42:30 -0500 Subject: [PATCH 3/4] fix(executor): reclaim a shared branch from a finished step, and wait instead of failing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two ways a pipeline died at spawn, both from the same fact: git allows one worktree per branch. 1. A finished step keeps holding the shared branch. Its worktree is deliberately kept for inspection, so it holds the branch forever and the NEXT step can never attach. This hit the first downstream step of every pipeline, not just fan-outs. A finished holder now hands the branch over: `git checkout --detach` moves no files, keeps uncommitted work, and leaves the branch ref where it is, so the worktree stays readable and nothing can be lost. A holder that is still RUNNING keeps the branch. 2. A step whose branch is busy was failed outright. setupWorktree's error parked it '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, which is how a pipeline silently loses a phase. A busy branch is now ErrBranchBusy: the task stays 'queued' and the next tick retries it. Fan-out steps arrive with their own branch pinned (see the previous commit) and get a worktree on it, cut from the shared branch. Cutting a branch FROM a branch is unrestricted, so no number of siblings can contend. Holder lookup compares symlink-resolved path forms: git reports a worktree as /private/var/... while the DB stores /var/..., and reading that mismatch as "no task owns this" would refuse to reclaim a branch from a long-finished step. Co-Authored-By: Claude Opus 5 (1M context) --- internal/executor/executor.go | 45 ++++- internal/executor/shared_branch.go | 224 ++++++++++++++++++++++++ internal/executor/shared_branch_test.go | 190 ++++++++++++++++++++ 3 files changed, 454 insertions(+), 5 deletions(-) create mode 100644 internal/executor/shared_branch.go create mode 100644 internal/executor/shared_branch_test.go diff --git a/internal/executor/executor.go b/internal/executor/executor.go index dd639402..716d3695 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 @@ -5660,6 +5682,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: diff --git a/internal/executor/shared_branch.go b/internal/executor/shared_branch.go new file mode 100644 index 00000000..bb980fd4 --- /dev/null +++ b/internal/executor/shared_branch.go @@ -0,0 +1,224 @@ +package executor + +import ( + "errors" + "fmt" + "os/exec" + "path/filepath" + "strings" + + "github.com/bborn/workflow/internal/db" +) + +// 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) + } + } + return runGitWorktreeAdd(projectDir, stepBranch, "worktree", "add", "-b", stepBranch, worktreePath, base) +} + +// runGitWorktreeAdd runs a `git worktree add` and reports a failure with the +// git output, which is the only thing that explains what actually went wrong. +func runGitWorktreeAdd(projectDir, branch string, args ...string) error { + cmd := exec.Command("git", args...) + cmd.Dir = projectDir + if out, err := cmd.CombinedOutput(); err != nil { + return fmt.Errorf("create worktree on branch %s: %v\n%s", branch, err, string(out)) + } + return nil +} diff --git a/internal/executor/shared_branch_test.go b/internal/executor/shared_branch_test.go new file mode 100644 index 00000000..385d0ef4 --- /dev/null +++ b/internal/executor/shared_branch_test.go @@ -0,0 +1,190 @@ +package executor + +import ( + "errors" + "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) + } + } +} From 1e093a60f920a93639fe8840406d20e1e59183ab Mon Sep 17 00:00:00 2001 From: Bruno Bornsztein Date: Fri, 14 Aug 2026 12:52:19 -0500 Subject: [PATCH 4/4] fix(executor): serialize git worktree creation per repo, and don't track the shared branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both bugs were unreachable until fan-out steps could actually spawn together, and both showed up on the first real run. `git worktree add` is not concurrency-safe within a repository: it writes .git/config to record the new worktree's upstream, and a loser of git's own config lock fails outright instead of retrying — error: could not lock config file .git/config: File exists error: unable to write upstream branch configuration which killed one of two sibling steps spawned in the same instant. Worktree creation now takes a per-repo flock (executorlock.AcquireRepo), covering the ordinary-task path too: two normal tasks starting at once raced the same way. `worktree add -b X origin/` also sets X's upstream to the SHARED branch, so a later bare `git push` from a fan-out step would land its commits on the branch its instructions explicitly tell it not to touch. Cut with --no-track. Co-Authored-By: Claude Opus 5 (1M context) --- internal/executor/executor.go | 18 +++--- internal/executor/shared_branch.go | 47 +++++++++++++-- internal/executor/shared_branch_test.go | 78 +++++++++++++++++++++++++ internal/executorlock/executorlock.go | 49 ++++++++++++++++ 4 files changed, 175 insertions(+), 17 deletions(-) diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 716d3695..16d6fae2 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -4582,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") { @@ -5718,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 index bb980fd4..0796c24f 100644 --- a/internal/executor/shared_branch.go +++ b/internal/executor/shared_branch.go @@ -6,8 +6,12 @@ import ( "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 @@ -209,16 +213,47 @@ func (e *Executor) addStepBranchWorktree(projectDir, worktreePath, stepBranch, s return fmt.Errorf("shared branch %s not found on origin or locally", sharedBranch) } } - return runGitWorktreeAdd(projectDir, stepBranch, "worktree", "add", "-b", stepBranch, worktreePath, base) + // --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` and reports a failure with the -// git output, which is the only thing that explains what actually went wrong. +// 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 { - cmd := exec.Command("git", args...) - cmd.Dir = projectDir - if out, err := cmd.CombinedOutput(); err != nil { + 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 index 385d0ef4..6d800381 100644 --- a/internal/executor/shared_branch_test.go +++ b/internal/executor/shared_branch_test.go @@ -2,6 +2,7 @@ package executor import ( "errors" + "fmt" "os" "os/exec" "path/filepath" @@ -188,3 +189,80 @@ func writeAndCommit(t *testing.T, dir, name, content string) { } } } + +// 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) + } +}