diff --git a/.changeset/worktree-capacity-hold-release.md b/.changeset/worktree-capacity-hold-release.md new file mode 100644 index 0000000000..207a20aa3c --- /dev/null +++ b/.changeset/worktree-capacity-hold-release.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Honor the worktree execution limit when workflow-column hold releases dispatch tasks. diff --git a/packages/engine/src/__tests__/scheduler.test.ts b/packages/engine/src/__tests__/scheduler.test.ts index 67cb19e479..d39885ccaa 100644 --- a/packages/engine/src/__tests__/scheduler.test.ts +++ b/packages/engine/src/__tests__/scheduler.test.ts @@ -630,6 +630,55 @@ describe("Scheduler", () => { expect(vi.mocked(store.listTasks).mock.calls.some(([options]) => options?.startupMemo === false)).toBe(true); }); + it("holds workflow-column releases when maxWorktrees is exhausted", async () => { + vi.mocked(existsSync).mockReturnValue(true); + vi.mocked(readFile).mockResolvedValue("# Task\nDo something"); + + const tasks = new Map([ + ["FN-001", createMockTask({ id: "FN-001", column: "in-progress", dependencies: [] })], + ["FN-002", createMockTask({ id: "FN-002", column: "in-progress", dependencies: [] })], + ["FN-003", createMockTask({ id: "FN-003", column: "todo", dependencies: [] })], + ["FN-004", createMockTask({ id: "FN-004", column: "todo", dependencies: [] })], + ["FN-005", createMockTask({ id: "FN-005", column: "todo", dependencies: [] })], + ]); + const movedListeners = new Set<(data: { task: object; to: string }) => void>(); + const moveTask = vi.fn(async (taskId: string, column: Task["column"]) => { + const current = tasks.get(taskId); + if (!current) throw new Error(`missing task ${taskId}`); + const updated = { ...current, column } as Task; + tasks.set(taskId, updated); + for (const listener of movedListeners) { + listener({ task: updated, to: column }); + } + return updated; + }); + const store = createMockStore({ + listTasks: vi.fn(async () => [...tasks.values()]), + getTask: vi.fn(async (taskId: string) => tasks.get(taskId) ?? null), + getSettings: vi.fn().mockResolvedValue({ + maxConcurrent: 15, + maxWorktrees: 3, + experimentalFeatures: { workflowColumns: true }, + }), + moveTask, + on: vi.fn((event: string, listener: (data: { task: object; to: string }) => void) => { + if (event === "task:moved") movedListeners.add(listener); + }), + off: vi.fn((event: string, listener: (data: { task: object; to: string }) => void) => { + if (event === "task:moved") movedListeners.delete(listener); + }), + }); + + const scheduler = new Scheduler(store); + (scheduler as unknown as { running: boolean }).running = true; + await scheduler.schedule(); + + const inProgress = [...tasks.values()].filter((task) => task.column === "in-progress"); + expect(inProgress.map((task) => task.id)).toEqual(["FN-001", "FN-002", "FN-003"]); + expect(moveTask.mock.calls.filter((call) => call[1] === "in-progress").map((call) => call[0])).toEqual(["FN-003"]); + expect(schedulerLog.log).toHaveBeenCalledWith(expect.stringContaining("no reservable slot")); + }); + it("flag-OFF: todo dispatch is tagged as scheduler-sourced for redispatch guards", async () => { const off = setupTodoStore(false); await off.scheduler.schedule(); diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 6d7e0f1801..fc31981913 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -1273,7 +1273,7 @@ export class Scheduler { // still drives todo→in-progress pickup (parity); the sweep adds custom- // workflow hold handling and the generalized capacity-release path. if (isWorkflowColumnsEnabled(settings)) { - await this.runHoldReleaseSweepPass(); + await this.runHoldReleaseSweepPass(tasks, settings); tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false }); settings = await this.store.getSettings(); } @@ -2122,24 +2122,31 @@ export class Scheduler { * worktree allocation into the reservation-first ordering (KTD-10). Failures * are isolated so a sweep error never breaks the scheduling pass. */ - private async runHoldReleaseSweepPass(): Promise { + private async runHoldReleaseSweepPass(tasks: Task[], settings: Settings): Promise { try { + const maxWorktrees = settings.maxWorktrees ?? this.options.maxWorktrees ?? 4; + let reservedWorktreeSlots = tasks.filter((task) => task.column === "in-progress").length; await runHoldReleaseSweep(this.store, { now: () => Date.now(), - reserveSlot: this.options.semaphore - ? (): SlotReservation | null => { - const sem = this.options.semaphore!; - if (!sem.tryAcquire()) return null; - let released = false; - return { - release: () => { - if (released) return; - released = true; - sem.release(); - }, - }; + reserveSlot: (): SlotReservation | null => { + if (Number.isFinite(maxWorktrees) && reservedWorktreeSlots >= maxWorktrees) { + return null; } - : undefined, + + const sem = this.options.semaphore; + if (sem && !sem.tryAcquire()) return null; + + reservedWorktreeSlots += 1; + let released = false; + return { + release: () => { + if (released) return; + released = true; + reservedWorktreeSlots = Math.max(0, reservedWorktreeSlots - 1); + sem?.release(); + }, + }; + }, allocateWorktree: (task, reservedNames) => planTaskWorktreePath(task, this.store.getRootDir(), undefined, reservedNames, {}), });