root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(FN-4068): add branch conflict detection and recovery for stale worktre

Browse Source 
Implements branch conflict detection and recovery across the Fusion engine, CLI, and dashboard — surfacing git worktree conflicts when tasks conflict with unrelated branch state, and providing a recovery workflow to resolve them. The executor and worktree pool now integrate typed branch conflict che

Fusion-Task-Id: FN-4068
This commit is contained in:
Fusion
2026-05-12 17:10:26 -07:00
committed by gsxdsm
parent fdd0af3b49
commit 75fe39d051
28 changed files with 1341 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
packages/engine/src/__tests__/branch-conflicts.test.ts Normal file
View File

@@ -0,0 +1,170 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import type { ExecException } from "node:child_process";
vi.mock("node:child_process", async () => {
const { promisify } = await import("node:util");
const execSyncFn = vi.fn();
const execFn: any = vi.fn((cmd: string, opts: any, cb: any) => {
const callback = typeof opts === "function" ? opts : cb;
const options = typeof opts === "function" ? {} : (opts ?? {});
try {
const out = execSyncFn(cmd, { ...options, stdio: ["pipe", "pipe", "pipe"] });
const stdout = out === undefined ? "" : out.toString();
if (typeof callback === "function") callback(null, stdout, "");
} catch (err) {
if (typeof callback === "function") {
const error = err as ExecException & { stdout?: string; stderr?: string };
callback(err, error?.stdout?.toString?.() ?? "", error?.stderr?.toString?.() ?? "");
}
}
});
execFn[promisify.custom] = (cmd: string, opts?: any) =>
new Promise((resolve, reject) => {
execFn(cmd, opts, (err: any, stdout: string, stderr: string) => {
if (err) {
(err as Record<string, unknown>).stdout = stdout;
(err as Record<string, unknown>).stderr = stderr;
reject(err);
} else {
resolve({ stdout, stderr });
}
});
});
return { exec: execFn, execSync: execSyncFn };
});
vi.mock("node:fs", () => ({
existsSync: vi.fn(),
}));
import { execSync } from "node:child_process";
import { existsSync } from "node:fs";
import { inspectBranchConflict, listBranchRecoveryCandidates, BranchConflictError } from "../branch-conflicts.js";
const mockedExecSync = vi.mocked(execSync);
const mockedExistsSync = vi.mocked(existsSync);
describe("branch-conflicts", () => {
beforeEach(() => {
vi.clearAllMocks();
mockedExistsSync.mockReturnValue(true);
});
it("classifies missing conflicting worktrees as stale", async () => {
mockedExistsSync.mockImplementation((value) => value !== "/tmp/missing-wt");
const result = await inspectBranchConflict({
repoDir: "/tmp/repo",
branchName: "fusion/fn-4068",
conflictingWorktreePath: "/tmp/missing-wt",
startPoint: "main",
});
expect(result).toEqual({ kind: "stale" });
expect(mockedExecSync).not.toHaveBeenCalled();
});
it("returns a typed live conflict with stranded commits", async () => {
mockedExecSync.mockImplementation((cmd: string | string[]) => {
const command = typeof cmd === "string" ? cmd : cmd[0];
if (command.includes("git rev-parse --verify 'fusion/fn-4068^{commit}'")) {
return Buffer.from("abc123def456\n");
}
if (command.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-4068'")) {
return Buffer.from("aaa111\tPreserve prior fix\nbbb222\tAdd regression coverage\n");
}
throw new Error(`Unexpected command: ${command}`);
});
const result = await inspectBranchConflict({
repoDir: "/tmp/repo",
branchName: "fusion/fn-4068",
conflictingWorktreePath: "/tmp/existing-wt",
startPoint: "main",
});
expect(result.kind).toBe("live");
if (result.kind !== "live") {
throw new Error("expected live conflict");
}
expect(result.error).toBeInstanceOf(BranchConflictError);
expect(result.error).toMatchObject({
branchName: "fusion/fn-4068",
conflictingWorktreePath: "/tmp/existing-wt",
existingTipSha: "abc123def456",
startPoint: "main",
});
expect(result.error.strandedCommits).toEqual([
{ sha: "aaa111", subject: "Preserve prior fix" },
{ sha: "bbb222", subject: "Add regression coverage" },
]);
expect(result.error.message).toContain("2 stranded commits since main");
});
it("lists canonical and sibling recovery candidates with worktrees and stranded commits", async () => {
mockedExecSync.mockImplementation((cmd: string | string[]) => {
const command = typeof cmd === "string" ? cmd : cmd[0];
if (command === "git for-each-ref --format='%(refname:short)' refs/heads/fusion/fn-4068 refs/heads/fusion/fn-4068-*") {
return Buffer.from("fusion/fn-4068\nfusion/fn-4068-2\n");
}
if (command === "git worktree list --porcelain") {
return Buffer.from([
"worktree /tmp/repo",
"HEAD 1111111",
"branch refs/heads/main",
"",
"worktree /tmp/fn-4068",
"HEAD 2222222",
"branch refs/heads/fusion/fn-4068",
"",
"worktree /tmp/fn-4068-2",
"HEAD 3333333",
"branch refs/heads/fusion/fn-4068-2",
"",
].join("\n"));
}
if (command.includes("git rev-parse --verify 'fusion/fn-4068^{commit}'")) {
return Buffer.from("abc123\n");
}
if (command.includes("git rev-parse --verify 'fusion/fn-4068-2^{commit}'")) {
return Buffer.from("def456\n");
}
if (command.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-4068'")) {
return Buffer.from("aaa111\tCanonical fix\n");
}
if (command.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-4068-2'")) {
return Buffer.from("bbb222\tSibling patch\nccc333\tMore work\n");
}
throw new Error(`Unexpected command: ${command}`);
});
const result = await listBranchRecoveryCandidates({
repoDir: "/tmp/repo",
branchName: "fusion/fn-4068",
startPoint: "main",
});
expect(result).toEqual([
{
branchName: "fusion/fn-4068",
tipSha: "abc123",
worktreePath: "/tmp/fn-4068",
strandedCommits: [{ sha: "aaa111", subject: "Canonical fix" }],
isCanonical: true,
},
{
branchName: "fusion/fn-4068-2",
tipSha: "def456",
worktreePath: "/tmp/fn-4068-2",
strandedCommits: [
{ sha: "bbb222", subject: "Sibling patch" },
{ sha: "ccc333", subject: "More work" },
],
isCanonical: false,
},
]);
});
});

122
packages/engine/src/__tests__/executor-worktree.test.ts
View File

@@ -10,6 +10,7 @@ import { reviewStep as mockedReviewStepFn } from "../reviewer.js";
import { execSync } from "node:child_process";
import { findWorktreeUser, aiMergeTask } from "../merger.js";
import { WorktreePool } from "../worktree-pool.js";
import { BranchConflictError } from "../branch-conflicts.js";
import { generateWorktreeName, slugify } from "../worktree-names.js";
import type { Task, TaskDetail } from "@fusion/core";
import { SessionManager } from "@mariozechner/pi-coding-agent";
@@ -780,39 +781,49 @@ describe("TaskExecutor worktree recovery", () => {
);
});
it("recovers from worktree conflict and retries", async () => {
it("records recovery context when handling a branch conflict", async () => {
const store = createMockStore();
let callCount = 0;
const onError = vi.fn();
const executor = new TaskExecutor(store, "/tmp/test", { onError });
// First call fails with conflict, second succeeds
mockedExecSync.mockImplementation((cmd: string | string[]) => {
const command = typeof cmd === "string" ? cmd : cmd[0];
if (command.includes("git worktree add") && callCount++ === 0) {
const error: any = new Error(
"fatal: 'fusion/fn-050' is already used by worktree at '/tmp/test/.worktrees/green-sage'",
);
error.stderr = Buffer.from(
"fatal: 'fusion/fn-050' is already used by worktree at '/tmp/test/.worktrees/green-sage'",
);
throw error;
}
return Buffer.from("");
});
const executor = new TaskExecutor(store, "/tmp/test");
await executor.execute(makeTask());
// Should have logged cleanup and retry
expect(store.logEntry).toHaveBeenCalledWith(
"FN-050",
expect.stringContaining("Cleaned up conflicting worktree, retrying"),
"/tmp/test/.worktrees/swift-falcon",
await (executor as any).handleBranchConflict(
makeTask(),
new BranchConflictError({
branchName: "fusion/fn-050",
conflictingWorktreePath: "/tmp/test/.worktrees/green-sage",
existingTipSha: "abc123def456",
strandedCommits: [
{ sha: "aaa111", subject: "Preserve prior fix" },
{ sha: "bbb222", subject: "Add regression coverage" },
],
startPoint: "HEAD",
recommendedAction: "Reclaim the existing task branch/worktree or explicitly discard prior work before retrying.",
}),
);
// Should eventually succeed
expect(store.updateTask).toHaveBeenCalledWith(
"FN-050",
expect.objectContaining({ worktree: expect.any(String) }),
expect.objectContaining({
status: "failed",
branch: "fusion/fn-050",
worktree: "/tmp/test/.worktrees/green-sage",
}),
);
expect(store.moveTask).toHaveBeenCalledWith("FN-050", "todo", { preserveProgress: true });
expect(store.logEntry).toHaveBeenCalledWith(
"FN-050",
expect.stringContaining("Existing tip: abc123def456"),
undefined,
undefined,
);
expect(store.appendAgentLog).toHaveBeenCalledWith(
"FN-050",
"Branch conflict recovery required",
"tool_error",
expect.stringContaining("stranded=aaa111 Preserve prior fix"),
"executor",
);
expect(onError).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-050" }), expect.any(BranchConflictError));
});
it("falls back to default base and clears task.executionStartBranch when the configured base ref is missing (FN-2165)", async () => {
@@ -1060,8 +1071,16 @@ describe("TaskExecutor worktree recovery", () => {
);
});
it("generates new worktree name when conflicting worktree belongs to active task", async () => {
it("generates new worktree name when conflicting worktree belongs to active task in legacy rename mode", async () => {
const store = createMockStore();
store.getSettings.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
groupOverlappingFiles: false,
autoMerge: false,
executorAllowSiblingBranchRename: true,
});
store.listTasks.mockResolvedValue([
{
id: "FN-049",
@@ -1079,6 +1098,7 @@ describe("TaskExecutor worktree recovery", () => {
]);
mockedFindWorktreeUser.mockResolvedValue("FN-049");
mockedExistsSync.mockImplementation((path) => path === "/tmp/test/.worktrees/green-sage");
let callCount = 0;
mockedExecSync.mockImplementation((cmd: string | string[]) => {
@@ -1734,6 +1754,7 @@ describe("TaskExecutor dependency-based worktree creation", () => {
"/tmp/test/.worktrees/idle-wt",
"fusion/fn-064",
"fusion/fn-063",
{ allowSiblingBranchRename: false, repoDir: "/tmp/test" },
);
});
@@ -1766,6 +1787,7 @@ describe("TaskExecutor dependency-based worktree creation", () => {
"/tmp/test/.worktrees/idle-wt",
"fusion/fn-065",
undefined,
{ allowSiblingBranchRename: false, repoDir: "/tmp/test" },
);
});
@@ -1994,11 +2016,9 @@ describe("TaskExecutor worktree pool integration", () => {
it("falls through to fresh worktree when pool prepareForTask throws", async () => {
const pool = new WorktreePool();
pool.release("/tmp/test/.worktrees/bad-wt");
// Pool path must exist on disk for acquire() to return it
mockedExistsSync.mockImplementation(
(p) => p === "/tmp/test/.worktrees/bad-wt",
);
// Make prepareForTask throw
vi.spyOn(pool, "prepareForTask").mockImplementation(() => {
throw new Error("branch conflict unrecoverable");
});
@@ -2017,16 +2037,13 @@ describe("TaskExecutor worktree pool integration", () => {
const executor = new TaskExecutor(store, "/tmp/test", { pool });
await executor.execute(makeTask());
// Should have released the bad worktree back to pool
expect(releaseSpy).toHaveBeenCalledWith("/tmp/test/.worktrees/bad-wt");
// Should have fallen through to fresh worktree creation
const worktreeAddCalls = mockedExecSync.mock.calls.filter(
(c) => typeof c[0] === "string" && (c[0] as string).includes("worktree add"),
);
expect(worktreeAddCalls.length).toBeGreaterThan(0);
// Should log the pool failure
expect(store.logEntry).toHaveBeenCalledWith(
"FN-020",
expect.stringContaining("Pool worktree preparation failed"),
@@ -2034,6 +2051,45 @@ describe("TaskExecutor worktree pool integration", () => {
expect.objectContaining({ agentId: "executor" }),
);
});
it("does not fall through to a fresh worktree when pooled preparation hits a typed branch conflict", async () => {
const pool = new WorktreePool();
pool.release("/tmp/test/.worktrees/warm-wt");
mockedExistsSync.mockImplementation((p) => p === "/tmp/test/.worktrees/warm-wt");
vi.spyOn(pool, "prepareForTask").mockRejectedValue(
new BranchConflictError({
branchName: "fusion/fn-020",
conflictingWorktreePath: "/tmp/test/.worktrees/existing-fn-020",
existingTipSha: "abc123def456",
strandedCommits: [{ sha: "aaa111", subject: "Preserve prior fix" }],
startPoint: "main",
recommendedAction: "Reclaim the existing task branch/worktree or explicitly discard prior work before retrying.",
}),
);
const store = createMockStore();
store.getSettings.mockResolvedValue({
maxConcurrent: 2,
maxWorktrees: 4,
pollIntervalMs: 15000,
groupOverlappingFiles: false,
autoMerge: false,
recycleWorktrees: true,
});
const executor = new TaskExecutor(store, "/tmp/test", { pool });
await executor.execute(makeTask("FN-020"));
const worktreeAddCalls = mockedExecSync.mock.calls.filter(
(c) => typeof c[0] === "string" && (c[0] as string).includes("worktree add"),
);
expect(worktreeAddCalls).toHaveLength(0);
expect(store.moveTask).toHaveBeenCalledWith("FN-020", "todo", { preserveProgress: true });
expect(store.updateTask).toHaveBeenCalledWith(
"FN-020",
expect.objectContaining({ branch: "fusion/fn-020", worktree: "/tmp/test/.worktrees/existing-fn-020" }),
);
});
});
describe("WorktreePool capacity", () => {

93
packages/engine/src/__tests__/worktree-pool.test.ts
View File

@@ -54,6 +54,7 @@ import {
reapOrphanWorktrees,
scanOrphanedBranches,
} from "../worktree-pool.js";
import { BranchConflictError } from "../branch-conflicts.js";
import { execSync } from "node:child_process";
import { existsSync, lstatSync, readdirSync, rmSync } from "node:fs";
import type { Task, Column } from "@fusion/core";
@@ -260,9 +261,8 @@ describe("WorktreePool", () => {
);
});
it("uses suffixed branch name when original is in use by an active worktree", async () => {
it("throws a typed branch conflict when the canonical branch is already live elsewhere by default", async () => {
mockedExistsSync.mockImplementation((p) => {
// The conflicting worktree exists on disk
if (p === "/other/wt") return true;
return true;
});
@@ -276,20 +276,26 @@ describe("WorktreePool", () => {
);
throw err;
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042^{commit}'")) {
return Buffer.from("abc123def456\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042'")) {
return Buffer.from("aaa111\tPreserve prior fix\n");
}
return Buffer.from("");
});
const result = await pool.prepareForTask("/tmp/wt", "fusion/fn-042");
expect(result).toBe("fusion/fn-042-2");
await expect(
pool.prepareForTask("/tmp/wt", "fusion/fn-042", undefined, { repoDir: "/tmp/repo" })
).rejects.toBeInstanceOf(BranchConflictError);
// Verify the suffixed checkout was called
const checkoutCalls = mockedExecSync.mock.calls
.map((c) => c[0])
.filter((c) => typeof c === "string" && c.includes("checkout -B"));
expect(checkoutCalls).toContain('git checkout -B "fusion/fn-042-2" fusion/fn-042');
expect(checkoutCalls).not.toContain('git checkout -B "fusion/fn-042-2" fusion/fn-042');
});
it("seeds suffixed retry branches from the original branch instead of the generic base", async () => {
it("restores legacy suffixed branch behavior only when explicitly enabled", async () => {
mockedExistsSync.mockReturnValue(true);
mockedExecSync.mockImplementation((cmd: any) => {
@@ -301,10 +307,21 @@ describe("WorktreePool", () => {
);
throw err;
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042^{commit}'")) {
return Buffer.from("abc123def456\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'fusion/fn-041..fusion/fn-042'")) {
return Buffer.from("aaa111\tPreserve prior fix\n");
}
return Buffer.from("");
});
const result = await pool.prepareForTask("/tmp/wt", "fusion/fn-042", "fusion/fn-041");
const result = await pool.prepareForTask(
"/tmp/wt",
"fusion/fn-042",
"fusion/fn-041",
{ allowSiblingBranchRename: true, repoDir: "/tmp/repo" },
);
expect(result).toBe("fusion/fn-042-2");
const checkoutCalls = mockedExecSync.mock.calls
@@ -314,12 +331,11 @@ describe("WorktreePool", () => {
expect(checkoutCalls).not.toContain('git checkout -B "fusion/fn-042-2" fusion/fn-041');
});
it("increments suffix when lower suffixes are also in use", async () => {
it("increments suffix when lower suffixes are also in use in legacy rename mode", async () => {
mockedExistsSync.mockReturnValue(true);
mockedExecSync.mockImplementation((cmd: any) => {
const cmdStr = String(cmd);
// Original and -2 are both in use
if (cmdStr.startsWith('git checkout -B "fusion/fn-042" ') ||
cmdStr.startsWith('git checkout -B "fusion/fn-042-2" ')) {
const err: any = new Error("branch conflict");
@@ -328,10 +344,27 @@ describe("WorktreePool", () => {
);
throw err;
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042^{commit}'")) {
return Buffer.from("abc123def456\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042'")) {
return Buffer.from("aaa111\tPreserve prior fix\n");
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042-2^{commit}'")) {
return Buffer.from("bbb222ccc333\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042-2'")) {
return Buffer.from("bbb222\tFirst sibling\n");
}
return Buffer.from("");
});
const result = await pool.prepareForTask("/tmp/wt", "fusion/fn-042");
const result = await pool.prepareForTask(
"/tmp/wt",
"fusion/fn-042",
undefined,
{ allowSiblingBranchRename: true, repoDir: "/tmp/repo" },
);
expect(result).toBe("fusion/fn-042-3");
const checkoutCalls = mockedExecSync.mock.calls
@@ -386,7 +419,7 @@ describe("WorktreePool", () => {
);
});
it("throws when all suffixed names are exhausted", async () => {
it("throws when all suffixed names are exhausted in legacy rename mode", async () => {
mockedExistsSync.mockReturnValue(true);
mockedExecSync.mockImplementation((cmd: any) => {
@@ -398,12 +431,42 @@ describe("WorktreePool", () => {
);
throw err;
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042^{commit}'")) {
return Buffer.from("abc123def456\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042'")) {
return Buffer.from("aaa111\tPreserve prior fix\n");
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042-2^{commit}'")) {
return Buffer.from("bbb222ccc333\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042-2'")) {
return Buffer.from("bbb222\tFirst sibling\n");
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042-3^{commit}'")) {
return Buffer.from("ccc333ddd444\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042-3'")) {
return Buffer.from("ccc333\tSecond sibling\n");
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042-4^{commit}'")) {
return Buffer.from("ddd444eee555\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042-4'")) {
return Buffer.from("ddd444\tThird sibling\n");
}
if (cmdStr.includes("git rev-parse --verify 'fusion/fn-042-5^{commit}'")) {
return Buffer.from("eee555fff666\n");
}
if (cmdStr.includes("git log --reverse --format=%H%x09%s 'main..fusion/fn-042-5'")) {
return Buffer.from("eee555\tFourth sibling\n");
}
return Buffer.from("");
});
await expect(pool.prepareForTask("/tmp/wt", "fusion/fn-042")).rejects.toThrow(
/suffixes -2 through -6 are all in use/
);
await expect(
pool.prepareForTask("/tmp/wt", "fusion/fn-042", undefined, { allowSiblingBranchRename: true, repoDir: "/tmp/repo" })
).rejects.toThrow(/suffixes -2 through -6 are all in use/);
});
});

198
packages/engine/src/branch-conflicts.ts Normal file
View File

@@ -0,0 +1,198 @@
import { exec } from "node:child_process";
import { existsSync } from "node:fs";
import { promisify } from "node:util";
const execAsync = promisify(exec);
export interface BranchConflictCommit {
sha: string;
subject: string;
}
export interface BranchRecoveryCandidate {
branchName: string;
tipSha: string;
worktreePath: string | null;
strandedCommits: BranchConflictCommit[];
isCanonical: boolean;
}
export interface BranchConflictDetails {
branchName: string;
conflictingWorktreePath: string;
existingTipSha: string;
strandedCommits: BranchConflictCommit[];
startPoint: string;
recommendedAction: string;
}
export class BranchConflictError extends Error implements BranchConflictDetails {
readonly name = "BranchConflictError";
readonly branchName: string;
readonly conflictingWorktreePath: string;
readonly existingTipSha: string;
readonly strandedCommits: BranchConflictCommit[];
readonly startPoint: string;
readonly recommendedAction: string;
constructor(details: BranchConflictDetails) {
const commitSummary = details.strandedCommits.length > 0
? `${details.strandedCommits.length} stranded commit${details.strandedCommits.length === 1 ? "" : "s"}`
: "no stranded commits";
super(
`Branch ${details.branchName} is already checked out at ${details.conflictingWorktreePath} ` +
`(tip ${details.existingTipSha.slice(0, 12)}, ${commitSummary} since ${details.startPoint}). ` +
details.recommendedAction,
);
this.branchName = details.branchName;
this.conflictingWorktreePath = details.conflictingWorktreePath;
this.existingTipSha = details.existingTipSha;
this.strandedCommits = details.strandedCommits;
this.startPoint = details.startPoint;
this.recommendedAction = details.recommendedAction;
}
}
export function isBranchConflictError(error: unknown): error is BranchConflictError {
return error instanceof BranchConflictError;
}
export interface InspectBranchConflictInput {
repoDir: string;
branchName: string;
conflictingWorktreePath: string;
startPoint?: string;
}
export type BranchConflictInspectionResult =
| { kind: "stale" }
| { kind: "live"; error: BranchConflictError };
export interface ListBranchRecoveryCandidatesInput {
repoDir: string;
branchName: string;
startPoint?: string;
}
function quoteShellArg(value: string): string {
return `'${value.replace(/'/g, `'\\''`)}'`;
}
async function runGit(repoDir: string, command: string): Promise<string> {
const { stdout } = await execAsync(command, {
cwd: repoDir,
encoding: "utf-8",
});
return stdout.trim();
}
async function revParse(repoDir: string, ref: string): Promise<string> {
return runGit(repoDir, `git rev-parse --verify ${quoteShellArg(`${ref}^{commit}`)}`);
}
async function listStrandedCommits(repoDir: string, startPoint: string, branchName: string): Promise<BranchConflictCommit[]> {
try {
const output = await runGit(
repoDir,
`git log --reverse --format=%H%x09%s ${quoteShellArg(`${startPoint}..${branchName}`)}`,
);
if (!output) return [];
return output
.split("\n")
.map((line) => line.trim())
.filter(Boolean)
.map((line) => {
const [sha, ...subjectParts] = line.split("\t");
return { sha, subject: subjectParts.join("\t") };
});
} catch {
return [];
}
}
async function getWorktreeBranchMap(repoDir: string): Promise<Map<string, string>> {
const output = await runGit(repoDir, "git worktree list --porcelain");
const map = new Map<string, string>();
let currentWorktree: string | null = null;
for (const line of output.split("\n")) {
if (line.startsWith("worktree ")) {
currentWorktree = line.slice("worktree ".length).trim();
continue;
}
if (line.startsWith("branch refs/heads/") && currentWorktree) {
map.set(line.slice("branch refs/heads/".length).trim(), currentWorktree);
}
if (!line.trim()) {
currentWorktree = null;
}
}
return map;
}
function parseBranchNames(output: string): string[] {
return output
.split("\n")
.map((line) => line.trim())
.filter(Boolean);
}
export async function listBranchRecoveryCandidates(
input: ListBranchRecoveryCandidatesInput,
): Promise<BranchRecoveryCandidate[]> {
const { repoDir, branchName } = input;
const startPoint = input.startPoint ?? "HEAD";
const [branchListOutput, worktreeBranches] = await Promise.all([
runGit(
repoDir,
`git for-each-ref --format='%(refname:short)' refs/heads/${branchName} refs/heads/${branchName}-*`,
),
getWorktreeBranchMap(repoDir),
]);
const candidates: BranchRecoveryCandidate[] = [];
for (const candidateName of parseBranchNames(branchListOutput)) {
const tipSha = await revParse(repoDir, candidateName);
const strandedCommits = await listStrandedCommits(repoDir, startPoint, candidateName);
candidates.push({
branchName: candidateName,
tipSha,
worktreePath: worktreeBranches.get(candidateName) ?? null,
strandedCommits,
isCanonical: candidateName === branchName,
});
}
candidates.sort((left, right) => {
if (left.branchName === branchName) return -1;
if (right.branchName === branchName) return 1;
return left.branchName.localeCompare(right.branchName);
});
return candidates;
}
export async function inspectBranchConflict(
input: InspectBranchConflictInput,
): Promise<BranchConflictInspectionResult> {
const startPoint = input.startPoint ?? "HEAD";
if (!existsSync(input.conflictingWorktreePath)) {
return { kind: "stale" };
}
const existingTipSha = await revParse(input.repoDir, input.branchName);
const strandedCommits = await listStrandedCommits(input.repoDir, startPoint, input.branchName);
return {
kind: "live",
error: new BranchConflictError({
branchName: input.branchName,
conflictingWorktreePath: input.conflictingWorktreePath,
existingTipSha,
strandedCommits,
startPoint,
recommendedAction: "Reclaim the existing task branch/worktree or explicitly discard prior work before retrying.",
}),
};
}

140
packages/engine/src/executor.ts
View File

@@ -37,6 +37,7 @@ import { reviewStep, type ReviewVerdict } from "./reviewer.js";
import { ModelRegistry, SessionManager, type ToolDefinition, type AgentSession } from "@mariozechner/pi-coding-agent";
import { PRIORITY_EXECUTE, type AgentSemaphore } from "./concurrency.js";
import { getRegisteredWorktreePaths, isGitRepository, isRegisteredGitWorktree, isUsableTaskWorktree, type WorktreePool } from "./worktree-pool.js";
import { BranchConflictError, isBranchConflictError, inspectBranchConflict } from "./branch-conflicts.js";
import { AgentLogger } from "./agent-logger.js";
import { executorLog, reviewerLog, formatError } from "./logger.js";
import { TokenCapDetector } from "./token-cap-detector.js";
@@ -2482,6 +2483,7 @@ export class TaskExecutor {
// Resolve the base branch — set by the scheduler when a dep is in-review
const baseBranch = task.executionStartBranch || null;
const allowSiblingBranchRename = settings.executorAllowSiblingBranchRename === true;
if (task.worktree && isResume && !await isUsableTaskWorktree(this.rootDir, worktreePath)) {
const invalidWorktreePath = worktreePath;
@@ -2504,7 +2506,12 @@ export class TaskExecutor {
const pooled = this.options.pool.acquire();
if (pooled) {
try {
const actualBranch = await this.options.pool.prepareForTask(pooled, branchName, baseBranch ?? undefined);
const actualBranch = await this.options.pool.prepareForTask(
pooled,
branchName,
baseBranch ?? undefined,
{ allowSiblingBranchRename, repoDir: this.rootDir },
);
worktreePath = pooled;
acquiredFromPool = true;
executorLog.log(`Acquired worktree from pool: ${pooled}`);
@@ -2547,10 +2554,11 @@ export class TaskExecutor {
}
}
} catch (poolErr: unknown) {
// Pool preparation failed — release the worktree back and fall through
// to fresh worktree creation
const poolErrMessage = poolErr instanceof Error ? poolErr.message : String(poolErr);
this.options.pool.release(pooled);
if (isBranchConflictError(poolErr)) {
throw poolErr;
}
const poolErrMessage = poolErr instanceof Error ? poolErr.message : String(poolErr);
executorLog.log(`Pool prepareForTask failed, falling through to fresh worktree: ${poolErrMessage}`);
await this.store.logEntry(
task.id,
@@ -2564,7 +2572,7 @@ export class TaskExecutor {
// Fall through to fresh worktree creation if pool had nothing
if (!acquiredFromPool) {
const created = await this.createWorktree(branchName, worktreePath, task.id, baseBranch ?? undefined);
const created = await this.createWorktree(branchName, worktreePath, task.id, baseBranch ?? undefined, allowSiblingBranchRename);
worktreePath = created.path;
await this.store.updateTask(task.id, { worktree: created.path, branch: created.branch });
// Audit trail: record worktree creation and branch creation (FN-1404)
@@ -2701,7 +2709,7 @@ export class TaskExecutor {
}
} else {
// Directory exists at generated path but task has no worktree — create via normal flow
const created = await this.createWorktree(branchName, worktreePath, task.id);
const created = await this.createWorktree(branchName, worktreePath, task.id, undefined, allowSiblingBranchRename);
worktreePath = created.path;
await this.store.updateTask(task.id, { worktree: created.path, branch: created.branch });
// Audit trail: record worktree creation and branch creation (FN-1404)
@@ -4099,6 +4107,9 @@ export class TaskExecutor {
nextRecoveryAt: null,
});
// Fall through to terminal failure marking
} else if (isBranchConflictError(err)) {
await this.handleBranchConflict(task, err);
return;
} else if (this.options.usageLimitPauser && isUsageLimitError(errorMessage)) {
await this.options.usageLimitPauser.onUsageLimitHit("executor", task.id, errorMessage);
} else if (isTransientError(errorMessage)) {
@@ -6216,11 +6227,73 @@ and show an appropriate message to the user.\`
* @param startPoint - Optional base branch/commit for new branch
* @returns The actual worktree path (may differ if recovery generated new name)
*/
private formatBranchConflictLifecycleLog(taskId: string, error: BranchConflictError): string {
const strandedSummary = error.strandedCommits.length > 0
? error.strandedCommits.map((commit) => `${commit.sha.slice(0, 12)} ${commit.subject}`).join("; ")
: "none";
const recommendation = `Run \`fn task branch-recovery ${taskId}\` to inspect candidates, then reclaim the existing branch or discard prior work explicitly.`;
return [
`Branch conflict: ${error.branchName} is already checked out at ${error.conflictingWorktreePath}`,
`Existing tip: ${error.existingTipSha}`,
`Stranded commits since ${error.startPoint}: ${strandedSummary}`,
recommendation,
].join("\n");
}
private formatBranchConflictAgentLog(taskId: string, error: BranchConflictError): string {
const lines = [
`branch=${error.branchName}`,
`worktree=${error.conflictingWorktreePath}`,
`existingTipSha=${error.existingTipSha}`,
`startPoint=${error.startPoint}`,
];
if (error.strandedCommits.length > 0) {
lines.push(
...error.strandedCommits.map((commit) => `stranded=${commit.sha.slice(0, 12)} ${commit.subject}`),
);
} else {
lines.push("stranded=none");
}
lines.push(
`recommendation=Run 'fn task branch-recovery ${taskId}' to inspect candidates, then reclaim the existing branch or discard prior work explicitly.`,
);
return lines.join("\n");
}
private async handleBranchConflict(task: Task, error: BranchConflictError): Promise<void> {
const conflictMessage = `Task branch conflict: ${error.branchName} is already checked out at ${error.conflictingWorktreePath}. ` +
`Run 'fn task branch-recovery ${task.id}' to inspect candidates, then reclaim the existing branch or discard prior work explicitly.`;
await this.store.logEntry(
task.id,
this.formatBranchConflictLifecycleLog(task.id, error),
undefined,
this.currentRunContext,
);
await this.store.appendAgentLog(
task.id,
"Branch conflict recovery required",
"tool_error",
this.formatBranchConflictAgentLog(task.id, error),
"executor",
);
await this.store.updateTask(task.id, {
status: "failed",
error: conflictMessage,
branch: error.branchName,
worktree: error.conflictingWorktreePath,
});
await this.persistTokenUsage(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true });
executorLog.warn(`✗ ${task.id} branch conflict → todo: ${error.branchName} @ ${error.conflictingWorktreePath}`);
this.options.onError?.(task, error);
}
private async createWorktree(
branch: string,
path: string,
taskId: string,
startPoint?: string,
allowSiblingBranchRename = false,
): Promise<{ path: string; branch: string }> {
// Track the worktree path we're attempting to use (may change during recovery)
const currentPath = path;
@@ -6262,7 +6335,15 @@ and show an appropriate message to the user.\`
for (let attempt = 0; attempt < this.MAX_WORKTREE_RETRIES; attempt++) {
try {
const result = await this.tryCreateWorktree(branch, currentPath, taskId, initialStartPoint, attempt);
const result = await this.tryCreateWorktree(
branch,
currentPath,
taskId,
initialStartPoint,
attempt,
0,
allowSiblingBranchRename,
);
// Squash-import dep content into the freshly created worktree so the
// branch contains main's history + 1 import commit instead of the
// dep's raw commits.
@@ -6293,7 +6374,8 @@ and show an appropriate message to the user.\`
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : String(error);
const isLastAttempt = attempt === this.MAX_WORKTREE_RETRIES - 1;
const isTerminalWorktreeError = error instanceof NonRetryableWorktreeError;
const isBranchConflict = isBranchConflictError(error);
const isTerminalWorktreeError = error instanceof NonRetryableWorktreeError || isBranchConflict;
if (isLastAttempt || isTerminalWorktreeError) {
await this.store.logEntry(
@@ -6301,6 +6383,9 @@ and show an appropriate message to the user.\`
`Worktree creation failed after ${this.MAX_WORKTREE_RETRIES} attempts`,
errorMessage,
);
if (isBranchConflict) {
throw error;
}
throw new Error(
`Failed to create worktree after ${this.MAX_WORKTREE_RETRIES} attempts: ${errorMessage}`,
);
@@ -6631,6 +6716,7 @@ and show an appropriate message to the user.\`
startPoint?: string,
attemptNumber = 0,
recoveryDepth = 0,
allowSiblingBranchRename = false,
): Promise<{ path: string; branch: string }> {
// Guard: refuse to create a worktree nested inside another worktree.
// Nested worktrees happen when the executor is launched with rootDir pointed
@@ -6719,6 +6805,7 @@ and show an appropriate message to the user.\`
taskId,
startPoint,
attemptNumber,
allowSiblingBranchRename,
);
if (result) {
return result;
@@ -6738,7 +6825,7 @@ and show an appropriate message to the user.\`
const branchCleaned = await this.cleanupStaleBranch(branch, taskId);
if (branchCleaned) {
await this.store.logEntry(taskId, `Removed stale branch reference, retrying`);
return this.tryCreateWorktree(branch, path, taskId, startPoint, attemptNumber, recoveryDepth + 1);
return this.tryCreateWorktree(branch, path, taskId, startPoint, attemptNumber, recoveryDepth + 1, allowSiblingBranchRename);
}
throw new Error(
`Invalid reference for branch ${branch}: unable to clean up stale reference`,
@@ -6776,6 +6863,7 @@ and show an appropriate message to the user.\`
taskId,
startPoint,
attemptNumber,
allowSiblingBranchRename,
);
if (result) {
return result;
@@ -6795,7 +6883,7 @@ and show an appropriate message to the user.\`
const branchCleaned = await this.cleanupStaleBranch(branch, taskId);
if (branchCleaned) {
await this.store.logEntry(taskId, `Cleaned up stale reference in fallback, retrying`);
return this.tryCreateWorktree(branch, path, taskId, startPoint, attemptNumber, recoveryDepth + 1);
return this.tryCreateWorktree(branch, path, taskId, startPoint, attemptNumber, recoveryDepth + 1, allowSiblingBranchRename);
}
}
@@ -6818,6 +6906,7 @@ and show an appropriate message to the user.\`
taskId: string,
startPoint?: string,
attemptNumber?: number,
allowSiblingBranchRename = false,
): Promise<{ path: string; branch: string } | null> {
const shouldGenerateNewName = await this.shouldGenerateNewWorktreeName(
conflictPath,
@@ -6825,12 +6914,25 @@ and show an appropriate message to the user.\`
);
if (shouldGenerateNewName) {
// Conflicting worktree belongs to an active task — generate new path AND
// use a suffixed branch name so git doesn't conflict with the branch
// already checked out in the existing worktree. Branch conflicts here
// mean the original task branch already exists and is checked out
// elsewhere, so suffix retries must branch from that task branch tip
// rather than the stale base ref to preserve the task's commits.
const inspection = await inspectBranchConflict({
repoDir: this.rootDir,
branchName: branch,
conflictingWorktreePath: conflictPath,
startPoint,
});
if (inspection.kind === "stale") {
const cleanupSuccess = await this.cleanupConflictingWorktree(conflictPath, branch, taskId);
if (cleanupSuccess) {
await this.store.logEntry(taskId, `Cleaned up conflicting worktree, retrying`, path);
return this.tryCreateWorktree(branch, path, taskId, startPoint, attemptNumber, 0, allowSiblingBranchRename);
}
return null;
}
if (!allowSiblingBranchRename) {
throw inspection.error;
}
const conflictStartPoint = branch;
const newPath = join(this.rootDir, ".worktrees", generateWorktreeName(this.rootDir));
for (let suffix = 2; suffix <= 6; suffix++) {
@@ -6841,11 +6943,10 @@ and show an appropriate message to the user.\`
`Conflicting worktree in use by active task, trying new path with branch ${suffixedBranch}`,
newPath,
);
return await this.tryCreateWorktree(suffixedBranch, newPath, taskId, conflictStartPoint, attemptNumber);
return await this.tryCreateWorktree(suffixedBranch, newPath, taskId, conflictStartPoint, attemptNumber, 0, true);
} catch (suffixErr: unknown) {
const info = this.extractWorktreeConflictInfo(suffixErr);
if (info.type === "already-used") {
// This suffixed branch is also in use — try next suffix
continue;
}
throw suffixErr;
@@ -6856,11 +6957,10 @@ and show an appropriate message to the user.\`
);
}
// Safe to clean up - conflicting worktree is not in use
const cleanupSuccess = await this.cleanupConflictingWorktree(conflictPath, branch, taskId);
if (cleanupSuccess) {
await this.store.logEntry(taskId, `Cleaned up conflicting worktree, retrying`, path);
return this.tryCreateWorktree(branch, path, taskId, startPoint, attemptNumber);
return this.tryCreateWorktree(branch, path, taskId, startPoint, attemptNumber, 0, allowSiblingBranchRename);
}
return null;

12
packages/engine/src/index.ts
View File

@@ -90,6 +90,18 @@ export {
} from "./agent-instructions.js";
export { HEARTBEAT_PROCEDURE, HEARTBEAT_SYSTEM_PROMPT, HEARTBEAT_NO_TASK_SYSTEM_PROMPT } from "./agent-heartbeat.js";
export { WorktreePool, scanIdleWorktrees, cleanupOrphanedWorktrees, reapOrphanWorktrees } from "./worktree-pool.js";
export {
BranchConflictError,
isBranchConflictError,
inspectBranchConflict,
listBranchRecoveryCandidates,
type BranchConflictCommit,
type BranchConflictDetails,
type BranchRecoveryCandidate,
type BranchConflictInspectionResult,
type InspectBranchConflictInput,
type ListBranchRecoveryCandidatesInput,
} from "./branch-conflicts.js";
export { generateReservedWorktreeName, generateWorktreeName, planTaskWorktreePath, slugify } from "./worktree-names.js";
export { createLogger, type Logger } from "./logger.js";
export { fetchWebContent, assertSafeUrl, WebFetchError, type WebFetchOptions, type WebFetchResult, type WebFetchErrorCode } from "./web-fetch.js";

36
packages/engine/src/worktree-pool.ts
View File

@@ -3,6 +3,7 @@ import { promisify } from "node:util";
import { existsSync, lstatSync, readdirSync, rmSync } from "node:fs";
import { join, relative, resolve, isAbsolute } from "node:path";
import type { Column, TaskStore } from "@fusion/core";
import { inspectBranchConflict } from "./branch-conflicts.js";
import { worktreePoolLog } from "./logger.js";
const execAsync = promisify(exec);
@@ -181,14 +182,20 @@ export class WorktreePool {
* 4. `git checkout -B <branchName> <startPoint>` — create/reset branch from start point
*
* Returns the actual branch name used. This may differ from `branchName`
* when conflict recovery generates a suffixed name (e.g., `fusion/fn-042-2`).
* when legacy conflict recovery is explicitly enabled and generates a suffixed
* name (e.g., `fusion/fn-042-2`).
*
* @param worktreePath — Absolute path to the recycled worktree
* @param branchName — Branch name for the new task (e.g., `fusion/fn-042`)
* @param startPoint — Git ref to branch from (e.g., `fusion/fn-041`). Defaults to `main`.
* @returns The actual branch name checked out in the worktree
*/
async prepareForTask(worktreePath: string, branchName: string, startPoint?: string): Promise<string> {
async prepareForTask(
worktreePath: string,
branchName: string,
startPoint?: string,
options?: { allowSiblingBranchRename?: boolean; repoDir?: string },
): Promise<string> {
// Clean tracked modifications
try {
await execAsync("git checkout -- .", { cwd: worktreePath });
@@ -223,20 +230,27 @@ export class WorktreePool {
throw err;
}
// The branch is checked out in a different worktree.
// First check if the conflicting worktree still exists on disk.
// The branch is checked out in a different worktree. Keep stale-conflict
// cleanup behavior for missing paths; otherwise either surface a typed
// conflict or, when explicitly enabled, fall back to the legacy sibling
// suffix flow.
const conflictingPath = match[1];
if (!existsSync(conflictingPath)) {
// Conflicting worktree no longer exists — prune and retry with original name
const inspection = await inspectBranchConflict({
repoDir: options?.repoDir ?? worktreePath,
branchName,
conflictingWorktreePath: conflictingPath,
startPoint: base,
});
if (inspection.kind === "stale") {
await execAsync("git worktree prune", { cwd: worktreePath });
await execAsync(checkoutCmd, { cwd: worktreePath });
return branchName;
}
// Conflicting worktree exists and is active — use a suffixed branch name
// to avoid disrupting the other worktree. Seed the suffix from the
// original task branch tip rather than the generic base ref so retries
// preserve the task's commits instead of resetting to main/baseBranch.
if (!options?.allowSiblingBranchRename) {
throw inspection.error;
}
const conflictBase = branchName;
for (let suffix = 2; suffix <= 6; suffix++) {
const suffixedName = `${branchName}-${suffix}`;
@@ -252,11 +266,9 @@ export class WorktreePool {
if (!suffixStderr.includes("already used by worktree")) {
throw suffixErr;
}
// This suffixed name is also in use — try the next one
}
}
// All suffixed names exhausted — should not happen in practice
throw new Error(
`Cannot create branch for task: "${branchName}" and suffixes -2 through -6 are all in use by other worktrees`,
);