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

fix(engine): prevent worktree collisions on manual task moves

Browse Source 
Two related bugs let two in-progress tasks share a single
.worktrees/<name> directory:

1. The dashboard POST /tasks/:id/move route promoted tasks to
   in-progress without allocating a fresh worktree path, so a queued
   task carrying a stale worktree field from a prior preserveResumeState
   requeue could land in-progress on a directory already held by another
   active task.

2. moveTask({preserveResumeState:true}) kept the worktree pointer on
   requeue. When the on-disk checkout was later removed or reassigned,
   the next dispatch collided with a worktree the scheduler had handed
   to another task.

moveTask now releases the worktree pointer on every reopen-to-todo hop
(branch is kept so committed progress survives via git worktree add
<path> <branch>). A new preserveWorktree option opts internal bounces
out of the release. moveTask also accepts an allocateWorktree callback
that runs under a new cross-task allocation lock in TaskStore, so two
concurrent moves cannot pick the same name from a stale snapshot. Both
the manual-move route and the scheduler dispatch path flow through the
allocator and share the lock.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-05-06 07:23:38 -07:00
parent 9f2b96e235
commit 4866f341bd
11 changed files with 308 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
packages/core/src/__tests__/store.test.ts
View File

@@ -6350,15 +6350,17 @@ Task with acceptance criteria
expect(doneMoved.executionStartedAt).toBeUndefined();
});
it("preserveResumeState takes precedence when preserveProgress and preserveResumeState are both true", async () => {
it("preserveResumeState keeps step progress and timing but always releases the worktree", async () => {
const task = await createTaskWithSteps();
await store.moveTask(task.id, "todo");
await store.moveTask(task.id, "in-progress");
await setMixedStepStatuses(task.id);
const startedAt = new Date().toISOString();
await store.updateTask(task.id, {
currentStep: 2,
worktree: "/tmp/worktree",
executionStartedAt: new Date().toISOString(),
branch: "fusion/fn-test",
executionStartedAt: startedAt,
executionCompletedAt: new Date().toISOString(),
});
@@ -6370,9 +6372,68 @@ Task with acceptance criteria
expect(moved.steps[0].status).toBe("done");
expect(moved.steps[1].status).toBe("in-progress");
expect(moved.currentStep).toBe(2);
expect(moved.worktree).toBe("/tmp/worktree");
expect(moved.executionStartedAt).toBeDefined();
// Worktree is always released on requeue so the directory can be
// reused by another task; the branch stays so progress is kept.
expect(moved.worktree).toBeUndefined();
expect(moved.branch).toBe("fusion/fn-test");
expect(moved.executionStartedAt).toBe(startedAt);
expect(moved.executionCompletedAt).toBeUndefined();
// Round-trip: when the task is re-promoted to in-progress with a
// fresh allocator, the branch reference must survive the requeue
// so the executor can reattach to it via createFromExistingBranch
// and resume the in-flight changes. Guards against regressions in
// the in-review → todo full-reset path leaking into other paths.
const repromoted = await store.moveTask(task.id, "in-progress", {
allocateWorktree: () => "/tmp/worktree-fresh",
});
expect(repromoted.branch).toBe("fusion/fn-test");
expect(repromoted.worktree).toBe("/tmp/worktree-fresh");
});
it("preserveWorktree keeps the directory across an internal bounce", async () => {
const task = await createTaskWithSteps();
await store.moveTask(task.id, "todo");
await store.moveTask(task.id, "in-progress");
await store.updateTask(task.id, { worktree: "/tmp/wt-bounce" });
const moved = await store.moveTask(task.id, "todo", {
preserveResumeState: true,
preserveWorktree: true,
});
// The bounce path keeps the same checkout assigned so listeners
// never observe an interim worktree=null state and self-healing
// can't reclaim the directory as idle.
expect(moved.worktree).toBe("/tmp/wt-bounce");
});
it("allocateWorktree assigns a path under the cross-task lock and avoids names already in use", async () => {
const a = await createTaskWithSteps();
const b = await createTaskWithSteps();
await store.moveTask(a.id, "todo");
await store.moveTask(a.id, "in-progress");
await store.updateTask(a.id, { worktree: "/tmp/.worktrees/eager-daisy" });
await store.moveTask(b.id, "todo");
const seenReserved: Set<string>[] = [];
const moved = await store.moveTask(b.id, "in-progress", {
allocateWorktree: (reservedNames) => {
seenReserved.push(new Set(reservedNames));
// Caller picks a name; if it collides with reservedNames the
// caller is responsible for choosing a different one. Here we
// assert the reservedNames snapshot reflects task A's
// assignment, then return a non-colliding path.
return "/tmp/.worktrees/swift-falcon";
},
});
expect(seenReserved).toHaveLength(1);
expect(seenReserved[0].has("eager-daisy")).toBe(true);
// The allocator's task itself must not appear in reservedNames —
// a task should never be told to avoid its own current name.
expect(seenReserved[0].has("swift-falcon")).toBe(false);
expect(moved.worktree).toBe("/tmp/.worktrees/swift-falcon");
});
it("resets steps when moving from in-review to todo", async () => {

83
packages/core/src/store.ts
View File

@@ -482,6 +482,13 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
private debounceMs = 150;
/** Per-task promise chain for serializing writes */
private taskLocks: Map<string, Promise<void>> = new Map();
/**
* Cross-task lock for worktree path allocation. Serializes the
* read-tasks → pick-name → write-task sequence so two concurrent
* `moveTask` calls (or a moveTask vs. a scheduler dispatch) cannot
* pick the same name from a stale snapshot.
*/
private worktreeAllocationLock: Promise<void> = Promise.resolve();
/** Promise chain for serializing config.json read-modify-write cycles */
private configLock: Promise<void> = Promise.resolve();
/** Cached workflow steps — invalidated on create/update/delete */
@@ -1503,6 +1510,21 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* Serialize all mutations to a given task's task.json by chaining promises
* per task ID. Concurrent callers for the same ID will queue behind each other.
*/
private withWorktreeAllocationLock<T>(fn: () => Promise<T>): Promise<T> {
let resolve: () => void;
const next = new Promise<void>((r) => { resolve = r; });
const prev = this.worktreeAllocationLock;
this.worktreeAllocationLock = next;
return prev.then(async () => {
try {
return await fn();
} finally {
resolve!();
}
});
}
private withTaskLock<T>(id: string, fn: () => Promise<T>): Promise<T> {
const prev = this.taskLocks.get(id) ?? Promise.resolve();
let resolve: () => void;
@@ -2841,6 +2863,26 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* (worktree and wall-clock timing fields).
*/
preserveProgress?: boolean;
/**
* Skip the default "release worktree on requeue" behavior. Used by
* internal bounce paths (e.g. workflow-rerun) that immediately
* promote the task back to in-progress on the same checkout, where
* publishing an interim `worktree=null` state to listeners would be
* misleading. Has no effect on transitions that don't otherwise
* clear the worktree.
*/
preserveWorktree?: boolean;
/**
* When transitioning to in-progress on a task that has no worktree
* assigned, invoke this allocator to pick a path. The store calls
* the allocator with a fresh `reservedNames` set (built from every
* other task's current `worktree`) inside a cross-task allocation
* lock, so two concurrent moves cannot pick the same name. The
* allocator should return an absolute path or `null` to skip
* allocation. Provided by callers (the manual-move route, the
* scheduler) so the store stays free of worktree-naming policy.
*/
allocateWorktree?: (reservedNames: Set<string>) => string | null;
},
): Promise<Task> {
return this.withTaskLock(id, async () => {
@@ -2918,8 +2960,24 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
const preserveStepProgress =
options?.preserveResumeState || (options?.preserveProgress === true && hasNonPendingStepProgress);
if (!options?.preserveResumeState) {
// Default: release the on-disk worktree directory on requeue. The
// checkout may have been removed, may now collide with another
// task's allocation, or may simply be abandoned by the bounce.
// `task.branch` is intentionally left intact so the next run can
// reattach to the same line of work — the executor's worktree
// creation path falls back to `git worktree add <path> <branch>`
// when the branch already exists, so any committed progress is
// preserved even though a fresh directory is allocated.
//
// Opt-out: internal bounces that immediately re-promote the task
// to in-progress on the same checkout (e.g. workflow-rerun) pass
// `preserveWorktree: true` so listeners never observe an interim
// `worktree=null` state.
if (!options?.preserveWorktree) {
task.worktree = undefined;
}
if (!options?.preserveResumeState) {
// Reset wall-clock runtime so the next run gets a fresh timer.
task.executionStartedAt = undefined;
task.executionCompletedAt = undefined;
@@ -2962,6 +3020,29 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.nextRecoveryAt = undefined;
}
// Atomic worktree allocation on transition to in-progress.
// Wrapped in withWorktreeAllocationLock so the read-tasks → pick-name
// sequence cannot interleave with another concurrent moveTask. The
// caller supplies the naming policy via the `allocateWorktree`
// callback; the store builds `reservedNames` here so the snapshot
// is fresh under the global lock.
if (toColumn === "in-progress" && !task.worktree && options?.allocateWorktree) {
const allocator = options.allocateWorktree;
const allocated = await this.withWorktreeAllocationLock(async () => {
const others = await this.listTasks({ slim: true, includeArchived: false });
const reservedNames = new Set<string>();
for (const other of others) {
if (other.id === id || !other.worktree) continue;
const name = other.worktree.split("/").filter(Boolean).pop();
if (name) reservedNames.add(name);
}
return allocator(reservedNames);
});
if (allocated) {
task.worktree = allocated;
}
}
await this.atomicWriteTaskJson(dir, task);
if (toColumn === "done") {
this.clearLinkedAgentTaskIds(id, task.updatedAt);

22
packages/dashboard/src/routes/register-task-workflow-routes.ts
View File

@@ -8,6 +8,7 @@ import {
resolveTitleSummarizerSettingsModel,
validateNodeOverrideChange,
} from "@fusion/core";
import { planTaskWorktreePath } from "@fusion/engine";
import { ApiError, badRequest, conflict, notFound } from "../api-error.js";
import type { ApiRoutesContext } from "./types.js";
@@ -228,8 +229,29 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
if (preserveProgress != null && typeof preserveProgress !== "boolean") {
throw badRequest("preserveProgress must be a boolean");
}
// When manually promoting to in-progress, supply an allocator so
// moveTask assigns a worktree path under its cross-task allocation
// lock. This mirrors scheduler dispatch semantics — without it, a
// user-initiated move would land the task in-progress with a stale
// (or null) worktree and could collide with another active task.
// The executor's createWorktree path will reuse `task.branch` if it
// already exists, so any prior committed progress survives even
// though the on-disk worktree directory is freshly allocated.
let allocateWorktree: ((reservedNames: Set<string>) => string | null) | undefined;
if ((column as Column) === "in-progress") {
const existing = await scopedStore.getTask(req.params.id);
if (existing) {
const settings = await scopedStore.getSettings();
const rootDir = scopedStore.getRootDir();
allocateWorktree = (reservedNames) =>
planTaskWorktreePath(existing, rootDir, settings.worktreeNaming, reservedNames);
}
}
const task = await scopedStore.moveTask(req.params.id, column as Column, {
preserveProgress,
allocateWorktree,
});
res.json(task);
} catch (err: unknown) {

10
packages/engine/src/__tests__/executor.test.ts
View File

@@ -8649,7 +8649,7 @@ describe("Workflow Steps Execution", () => {
// todo must flag preserveResumeState so the workflow-rerun bounce keeps
// the worktree and accumulated step progress through the transient
// todo state on its way back to in-progress.
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveResumeState: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveResumeState: true, preserveWorktree: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
// onComplete should NOT be called (task is being retried, not completed)
@@ -8784,7 +8784,7 @@ describe("Workflow Steps Execution", () => {
// todo must flag preserveResumeState so the workflow-rerun bounce keeps
// the worktree and accumulated step progress through the transient
// todo state on its way back to in-progress.
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveResumeState: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveResumeState: true, preserveWorktree: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
// onComplete should NOT be called (task is being retried, not completed)
@@ -8925,7 +8925,7 @@ describe("Workflow Steps Execution", () => {
await new Promise<void>((resolve) => queueMicrotask(resolve));
// (2) bounce uses preserveResumeState so step progress + worktree survive
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveResumeState: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveResumeState: true, preserveWorktree: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review");
expect(onError).not.toHaveBeenCalled();
@@ -11880,7 +11880,7 @@ describe("TaskExecutor watchdogs", () => {
executionStartedAt: originalExecutionStartedAt,
});
expect(store.moveTask.mock.calls).toEqual([
["FN-WD-4", "todo", { preserveResumeState: true }],
["FN-WD-4", "todo", { preserveResumeState: true, preserveWorktree: true }],
["FN-WD-4", "in-progress"],
]);
});
@@ -12716,7 +12716,7 @@ describe("StepSessionExecutor integration", () => {
// Task should move to todo then in-progress (not in-review). The
// workflow-rerun bounce flags preserveResumeState so the worktree and
// accumulated step progress survive the transient todo state.
expect(store.moveTask).toHaveBeenCalledWith("FN-200", "todo", { preserveResumeState: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-200", "todo", { preserveResumeState: true, preserveWorktree: true });
expect(store.moveTask).toHaveBeenCalledWith("FN-200", "in-progress");
vi.useRealTimers();

18
packages/engine/src/__tests__/restart.integration.test.ts
View File

@@ -725,8 +725,14 @@ describe("In-progress task resume after restart", () => {
// Run any pending microtasks (the async code in setTimeout)
await vi.runAllTimersAsync();
// Task should move to todo then in-progress (not in-review)
expect(store.moveTask).toHaveBeenCalledWith("FN-963", "todo");
// Task should move to todo then in-progress (not in-review). The
// workflow-rerun bounce passes `preserveWorktree: true` so the
// checkout doesn't briefly disappear during the hop.
expect(store.moveTask).toHaveBeenCalledWith(
"FN-963",
"todo",
expect.objectContaining({ preserveWorktree: true }),
);
expect(store.moveTask).toHaveBeenCalledWith("FN-963", "in-progress");
vi.useRealTimers();
@@ -963,7 +969,7 @@ describe("Scheduler after restart", () => {
await new Promise((r) => setTimeout(r, 50));
scheduler.stop();
expect(store.moveTask).toHaveBeenCalledWith("FN-070", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-070", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
expect(store.updateTask).toHaveBeenCalledWith("FN-070", expect.objectContaining({ status: null, blockedBy: null }));
expect(onSchedule).toHaveBeenCalledWith(todoTask);
});
@@ -1033,7 +1039,7 @@ describe("Scheduler after restart", () => {
await new Promise((r) => setTimeout(r, 50));
scheduler.stop();
expect(store.moveTask).toHaveBeenCalledWith("FN-081", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-081", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
// 3. Executor resumes in-progress tasks
vi.clearAllMocks();
@@ -1549,7 +1555,7 @@ describe("Engine pause/unpause cycle", () => {
await new Promise((r) => setTimeout(r, 50));
// Scheduler should have moved todo task to in-progress
expect(store.moveTask).toHaveBeenCalledWith("FN-EP3", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-EP3", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
// Now simulate engine pause then unpause
store.moveTask.mockClear();
@@ -1573,7 +1579,7 @@ describe("Engine pause/unpause cycle", () => {
scheduler.stop();
// The new task should have been scheduled after unpause
expect(store.moveTask).toHaveBeenCalledWith("FN-EP4", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-EP4", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("concurrency slots freed after agent completes during enginePaused (soft pause)", async () => {

33
packages/engine/src/__tests__/scheduler.test.ts
View File

@@ -244,7 +244,7 @@ describe("Scheduler", () => {
await flushAsyncWork();
// Verify schedule() was called (moveTask should be called since task can start)
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("resets mergeRetries when dispatching a task to in-progress", async () => {
@@ -282,7 +282,7 @@ describe("Scheduler", () => {
"FN-001",
expect.objectContaining({ mergeRetries: 0 }),
);
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("registers task:moved event listener", () => {
@@ -335,7 +335,7 @@ describe("Scheduler", () => {
await flushAsyncWork();
// Verify schedule() was called - FN-002 should now be able to start
expect(store.moveTask).toHaveBeenCalledWith("FN-002", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-002", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("does not trigger scheduling for non-done task:moved events", async () => {
@@ -406,7 +406,7 @@ describe("Scheduler", () => {
await flushAsyncWork();
// Verify schedule() was called — task in todo should be scheduled
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
});
@@ -450,7 +450,7 @@ describe("Scheduler", () => {
await flushAsyncWork();
// Should have triggered scheduling and moved the task to in-progress
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("does not trigger scheduling on unpause if scheduler is not running", async () => {
@@ -710,7 +710,7 @@ describe("Scheduler", () => {
expect(moveTask).not.toHaveBeenCalledWith("FN-102", "in-progress");
// Lower-priority ready task still runs.
expect(moveTask).toHaveBeenCalledWith("FN-104", "in-progress");
expect(moveTask).toHaveBeenCalledWith("FN-104", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
// Overlap-blocked urgent task must not run.
expect(moveTask).not.toHaveBeenCalledWith("FN-103", "in-progress");
});
@@ -751,7 +751,7 @@ describe("Scheduler", () => {
(scheduler as any).running = true;
await scheduler.schedule();
expect(moveTask).toHaveBeenCalledWith("FN-002", "in-progress");
expect(moveTask).toHaveBeenCalledWith("FN-002", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
expect(updateTask).not.toHaveBeenCalledWith("FN-002", { status: "queued", blockedBy: "FN-001" });
});
@@ -789,7 +789,7 @@ describe("Scheduler", () => {
(scheduler as any).running = true;
await scheduler.schedule();
expect(moveTask).toHaveBeenCalledWith("FN-002", "in-progress");
expect(moveTask).toHaveBeenCalledWith("FN-002", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
expect(updateTask).not.toHaveBeenCalledWith("FN-002", { status: "queued", blockedBy: "FN-001" });
});
@@ -855,12 +855,11 @@ describe("Scheduler", () => {
status: null,
blockedBy: null,
executionStartBranch: undefined,
worktree: "/test/project/.worktrees/fn-010",
effectiveNodeId: null,
effectiveNodeSource: "local",
mergeRetries: 0,
});
expect(moveTask).toHaveBeenCalledWith("FN-010", "in-progress");
expect(moveTask).toHaveBeenCalledWith("FN-010", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
expect(updateTask.mock.invocationCallOrder[0]).toBeLessThan(moveTask.mock.invocationCallOrder[0]);
});
@@ -894,7 +893,6 @@ describe("Scheduler", () => {
status: null,
blockedBy: null,
executionStartBranch: undefined,
worktree: "/test/project/.worktrees/amber-aspen",
effectiveNodeId: null,
effectiveNodeSource: "local",
mergeRetries: 0,
@@ -903,7 +901,6 @@ describe("Scheduler", () => {
status: null,
blockedBy: null,
executionStartBranch: undefined,
worktree: "/test/project/.worktrees/amber-aspen-2",
effectiveNodeId: null,
effectiveNodeSource: "local",
mergeRetries: 0,
@@ -1035,7 +1032,7 @@ describe("Scheduler", () => {
// Flush any remaining microtasks
await new Promise(resolve => setTimeout(resolve, 0));
expect(moveTask).toHaveBeenCalledWith("FN-010", "in-progress");
expect(moveTask).toHaveBeenCalledWith("FN-010", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
expect(moveTask).not.toHaveBeenCalledWith("FN-010", "triage");
});
@@ -1217,7 +1214,7 @@ describe("Scheduler", () => {
expect.any(String)
);
// Should move to in-progress (since deps are satisfied and concurrency allows)
expect(moveTask).toHaveBeenCalledWith("FN-004", "in-progress");
expect(moveTask).toHaveBeenCalledWith("FN-004", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("does not validate filesystem for tasks with unmet dependencies", async () => {
@@ -1956,7 +1953,7 @@ describe("Scheduler", () => {
(scheduler as any).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledWith("FN-100", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-100", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("schedules tasks without sliceId regardless of mission state", async () => {
@@ -1981,7 +1978,7 @@ describe("Scheduler", () => {
(scheduler as any).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledWith("FN-100", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-100", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
});
@@ -2037,7 +2034,7 @@ describe("Scheduler", () => {
(scheduler as any).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledWith("FN-011", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-011", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
it("picks up todo tasks without nextRecoveryAt normally", async () => {
@@ -2063,7 +2060,7 @@ describe("Scheduler", () => {
(scheduler as any).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledWith("FN-012", "in-progress");
expect(store.moveTask).toHaveBeenCalledWith("FN-012", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
});
});

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

@@ -1427,11 +1427,25 @@ export class TaskExecutor {
// moveTask's default reopen-to-todo path resets every step to
// pending and rewrites PROMPT.md checkboxes, which would discard
// the partial progress this bounce is supposed to retry on top of.
// `preserveWorktree` keeps the same checkout assigned across the
// hop so listeners never observe an interim `worktree=null` state
// — this bounce immediately re-promotes the task on the same
// directory, so releasing it would publish a misleading snapshot
// and could let self-healing reclaim the worktree as idle.
if (preserveResumeState) {
await this.store.moveTask(taskId, "todo", { preserveResumeState: true });
await this.store.moveTask(taskId, "todo", {
preserveResumeState: true,
preserveWorktree: true,
});
} else {
await this.store.moveTask(taskId, "todo");
await this.store.moveTask(taskId, "todo", { preserveWorktree: true });
}
// Restore worktree + executionStartedAt unconditionally to match
// the original bounce contract: even with preserveWorktree the
// worktree pointer could have been cleared by an in-flight
// updateTask, and executionStartedAt is reset by moveTask when
// preserveResumeState is false. Keep the writes so callers and
// tests can observe the restoration deterministically.
await this.store.updateTask(taskId, {
worktree: worktreePath,
executionStartedAt: originalExecutionStartedAt ?? null,

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

@@ -66,6 +66,7 @@ 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 { generateReservedWorktreeName, generateWorktreeName, planTaskWorktreePath, slugify } from "./worktree-names.js";
export { createLogger, type Logger } from "./logger.js";
export { isUsageLimitError, UsageLimitPauser } from "./usage-limit-detector.js";
export { withRateLimitRetry } from "./rate-limit-retry.js";

48
packages/engine/src/scheduler.ts
View File

@@ -10,9 +10,9 @@ import {
} from "@fusion/core";
import { existsSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { basename, join } from "node:path";
import { join } from "node:path";
import type { AgentSemaphore } from "./concurrency.js";
import { generateReservedWorktreeName, slugify } from "./worktree-names.js";
import { planTaskWorktreePath } from "./worktree-names.js";
import { schedulerLog } from "./logger.js";
import { type PrMonitor, type PrComment } from "./pr-monitor.js";
import { reconcileMissionFeatureState } from "./mission-feature-sync.js";
@@ -495,28 +495,7 @@ export class Scheduler {
naming: string | undefined,
reservedNames: Set<string>,
): string {
if (task.worktree) {
const existingName = basename(task.worktree);
if (existingName) reservedNames.add(existingName);
return task.worktree;
}
let worktreeName: string;
switch (naming || "random") {
case "task-id":
worktreeName = task.id.toLowerCase();
break;
case "task-title":
worktreeName = slugify(task.title || task.description.slice(0, 60));
break;
case "random":
default:
worktreeName = generateReservedWorktreeName(this.store.getRootDir(), reservedNames);
break;
}
reservedNames.add(worktreeName);
return join(this.store.getRootDir(), ".worktrees", worktreeName);
return planTaskWorktreePath(task, this.store.getRootDir(), naming, reservedNames);
}
/**
@@ -687,11 +666,6 @@ export class Scheduler {
// Resolve dependency order among todo tasks
const ordered = resolveDependencyOrder(todo);
let started = 0;
const reservedWorktreeNames = new Set(
tasks
.map((task) => (task.worktree ? basename(task.worktree) : undefined))
.filter((name): name is string => Boolean(name)),
);
for (const taskId of ordered) {
const task = tasks.find((t) => t.id === taskId)!;
@@ -760,13 +734,11 @@ export class Scheduler {
continue;
}
// Dependencies met — resolve base branch from in-review deps
// Dependencies met — resolve base branch from in-review deps.
// Worktree allocation is deferred to moveTask below, where it
// runs under TaskStore's cross-task allocation lock so it can't
// race against a concurrent manual-move.
const baseBranch = this.resolveBaseBranch(task, tasks);
const plannedWorktree = this.planWorktreePath(
task,
settings.worktreeNaming,
reservedWorktreeNames,
);
// Compare-and-swap: re-read the task to verify it's still in "todo" before dispatching.
// This prevents dispatching a task twice if another schedule() call or user action
@@ -836,12 +808,14 @@ export class Scheduler {
status: null,
blockedBy: null,
executionStartBranch: baseBranch ?? undefined,
worktree: plannedWorktree,
effectiveNodeId: effectiveNode.nodeId ?? null,
effectiveNodeSource: effectiveNode.source,
mergeRetries: 0,
});
await this.store.moveTask(task.id, "in-progress");
await this.store.moveTask(task.id, "in-progress", {
allocateWorktree: (reservedNames) =>
this.planWorktreePath(task, settings.worktreeNaming, reservedNames),
});
this.wasNodeBlocked.delete(task.id);
await this.store.logEntry(task.id, `Node routing resolved: ${effectiveNode.nodeId ?? "local"} (source: ${effectiveNode.source})`);
this.options.onSchedule?.(task);

42
packages/engine/src/worktree-names.ts
View File

@@ -94,6 +94,48 @@ export function generateReservedWorktreeName(
return `${baseName}-${suffix}`;
}
/**
* Plan a worktree directory path for a task that is about to enter
* `in-progress`. Returns the absolute path under `<rootDir>/.worktrees/`.
*
* If the task already carries a `worktree` value, it is reused — the
* caller is responsible for ensuring it does not collide with another
* active task. Otherwise a name is generated according to `naming`,
* avoiding any names already in `reservedNames`.
*
* Shared by the scheduler dispatch path and the manual-move HTTP route
* so both allocate via the same collision rules.
*/
export function planTaskWorktreePath(
task: { id: string; title?: string | null; description: string; worktree?: string | null },
rootDir: string,
naming: string | undefined,
reservedNames: Set<string>,
): string {
if (task.worktree) {
const existingName = task.worktree.split("/").filter(Boolean).pop();
if (existingName) reservedNames.add(existingName);
return task.worktree;
}
let worktreeName: string;
switch (naming || "random") {
case "task-id":
worktreeName = task.id.toLowerCase();
break;
case "task-title":
worktreeName = slugify(task.title || task.description.slice(0, 60));
break;
case "random":
default:
worktreeName = generateReservedWorktreeName(rootDir, reservedNames);
break;
}
reservedNames.add(worktreeName);
return join(rootDir, ".worktrees", worktreeName);
}
function getExistingWorktreeNames(worktreesDir: string): Set<string> {
if (!existsSync(worktreesDir)) {
return new Set();