diff --git a/.changeset/fix-retained-worktree-dispatch.md b/.changeset/fix-retained-worktree-dispatch.md new file mode 100644 index 0000000000..4828475c29 --- /dev/null +++ b/.changeset/fix-retained-worktree-dispatch.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Allow queued tasks to resume in their retained worktrees when worktree capacity is full. +category: fix +dev: Retained-worktree transfers bypass only the worktree allocation gate; agent limits still apply. diff --git a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts index 48646edc97..c0daead2f0 100644 --- a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts +++ b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts @@ -398,7 +398,7 @@ describe("Scheduler workflow cutover", () => { it("does not clear status or release work when maxConcurrent is full", async () => { const active = task({ id: "FN-001", column: "in-progress" }); - const ready = task({ id: "FN-002", status: "queued" }); + const ready = task({ id: "FN-002", status: "queued", worktree: "/tmp/project/.worktrees/fn-002" }); const store = storeWith([active, ready], { maxConcurrent: 1, maxWorktrees: 4 }); const onSchedule = vi.fn(); const scheduler = new Scheduler(store, { onSchedule }); @@ -460,6 +460,20 @@ describe("Scheduler workflow cutover", () => { expect(ready.column).toBe("todo"); }); + it("releases a retained-worktree task while already over maxWorktrees", async () => { + const active = task({ id: "FN-101", column: "in-progress", worktree: "/tmp/project/.worktrees/fn-101" }); + const ready = task({ id: "FN-200", status: "queued", worktree: "/tmp/project/.worktrees/fn-200" }); + const store = storeWith([active, ready], { maxConcurrent: 4, maxWorktrees: 1 }); + const onSchedule = vi.fn(); + const scheduler = new Scheduler(store, { onSchedule }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + expect(store.moveTaskIf).toHaveBeenCalledWith("FN-200", "in-progress", expect.any(Function), expect.anything()); + expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-200", column: "in-progress" })); + }); + /* FNXC:CapacityModel 2026-07-28-12:10: WORKTREES OFF → "limit via total agents only". These three cases are the proof @@ -600,8 +614,36 @@ describe("Scheduler workflow cutover", () => { expect(ready.status).toBe("queued"); }); + it("does not invent capacity when a retained-worktree transfer rejects", async () => { + const retained = task({ id: "FN-001", status: "queued", worktree: "/tmp/project/.worktrees/fn-001" }); + const fresh = task({ id: "FN-002", status: "queued" }); + const store = storeWith([retained, fresh], { maxConcurrent: 4, maxWorktrees: 1 }); + vi.mocked(store.moveTaskIf).mockRejectedValueOnce( + new TransitionRejectionError( + makeTransitionRejection( + "capacity-exhausted", + "transition.rejected.capacityExhausted", + true, + "Column is at capacity", + ), + "Column is at capacity", + ), + ); + const onSchedule = vi.fn(); + const scheduler = new Scheduler(store, { onSchedule }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + expect(store.moveTaskIf).toHaveBeenCalledTimes(1); + expect(store.moveTaskIf).toHaveBeenCalledWith("FN-001", "in-progress", expect.any(Function), expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything(), expect.anything()); + expect(onSchedule).not.toHaveBeenCalled(); + expect(fresh.column).toBe("todo"); + }); + it("does not release work when the shared semaphore is saturated", async () => { - const ready = task({ id: "FN-002", status: "queued" }); + const ready = task({ id: "FN-002", status: "queued", worktree: "/tmp/project/.worktrees/fn-002" }); const store = storeWith([ready], { maxConcurrent: 4, maxWorktrees: 4 }); const semaphore = new AgentSemaphore(1); await semaphore.acquire(); diff --git a/packages/engine/src/__tests__/scheduler-worktree-capacity-arithmetic.test.ts b/packages/engine/src/__tests__/scheduler-worktree-capacity-arithmetic.test.ts index 16b8911e78..b4797ebfb7 100644 --- a/packages/engine/src/__tests__/scheduler-worktree-capacity-arithmetic.test.ts +++ b/packages/engine/src/__tests__/scheduler-worktree-capacity-arithmetic.test.ts @@ -28,7 +28,7 @@ it here would make this file fail for that reason instead of this one. import { describe, expect, it } from "vitest"; import type { Task } from "@fusion/core"; -import { effectiveActiveWorktrees, nonWipWorktreeHolderIdsOf, releaseReservedSlot, reserveWorktreeOnDispatch } from "../scheduler.js"; +import { nonWipWorktreeHolderIdsOf, releaseReservedSlot, releaseWorktreeReservation, reserveWorktreeOnDispatch, resolveCandidateWorktreeCapacityLimit } from "../scheduler.js"; const task = (id: string, overrides: Partial = {}): Task => ({ id, @@ -103,25 +103,26 @@ describe("worktree-capacity holder set", () => { }); }); -describe("effectiveActiveWorktrees", () => { - it("subtracts a candidate's OWN retained worktree — the slot transfers, it does not add", () => { +describe("resolveCandidateWorktreeCapacityLimit", () => { + it("does not gate a candidate that reuses its retained worktree, even while the ledger is over cap", () => { /* - The self-deadlock fix. Without the subtraction a Ready card holding the very worktree it would - reuse is gated out by itself: 5 >= 4 blocks, where 4 >= 4 ... also blocks, but 5-1=4 is the - number the gate is supposed to compare. Pinned as the arithmetic, not as the comparison. + FNXC:WorktreeCapacity 2026-08-01-04:07: + Live regression: 12 non-terminal cards retained worktrees against maxWorktrees=9. A queued + candidate already owned one of those trees, so dispatch would transfer the existing slot and + leave the ledger at 12. Subtracting only the candidate (12 - 1 = 11) still wedged every queued + card forever. The worktree dimension must be absent for a zero-allocation transfer; the separate + maxConcurrent gate continues to arbitrate whether another agent may run. */ - expect(effectiveActiveWorktrees(5, true)).toBe(4); + expect(resolveCandidateWorktreeCapacityLimit(9, true)).toBeNull(); }); - it("leaves the total alone for a candidate that holds no worktree — it will ADD one", () => { - expect(effectiveActiveWorktrees(5, false)).toBe(5); + it("keeps the configured gate for a candidate that must allocate a worktree", () => { + expect(resolveCandidateWorktreeCapacityLimit(9, false)).toBe(9); }); - it("never invents capacity when nothing is reserved", () => { - expect(effectiveActiveWorktrees(0, false)).toBe(0); - /* A holder implies a reservation, so 0-with-holder cannot arise; assert it degrades rather than - silently handing out a negative slot count if a future caller gets the pairing wrong. */ - expect(effectiveActiveWorktrees(0, true)).toBeLessThanOrEqual(0); + it("preserves a disabled worktree gate for every candidate", () => { + expect(resolveCandidateWorktreeCapacityLimit(null, false)).toBeNull(); + expect(resolveCandidateWorktreeCapacityLimit(null, true)).toBeNull(); }); }); @@ -134,20 +135,12 @@ describe("ledger mutations", () => { expect(reserveWorktreeOnDispatch(4, false)).toBe(5); }); - it("the gate subtraction and the dispatch increment agree — the pairing invariant", () => { - /* - These two must move together. Subtracting the candidate's own slot for the gate check while - incrementing anyway on dispatch leaks one slot per dispatch, and the cap wedges after enough - Ready cards reuse their planning worktrees. Asserted as a round trip rather than two constants: - for a HOLDER the ledger must be unchanged across gate-then-dispatch. - */ - const reserved = 5; - for (const holds of [true, false]) { - const gated = effectiveActiveWorktrees(reserved, holds); - const after = reserveWorktreeOnDispatch(reserved, holds); - /* A holder occupies the slot it was gated against; a non-holder adds the one it was gated for. */ - expect(after - gated).toBe(1); - } + it("a failed retained transfer leaves the worktree ledger unchanged", () => { + expect(releaseWorktreeReservation(5, true)).toBe(5); + }); + + it("a failed fresh allocation gives its worktree slot back", () => { + expect(releaseWorktreeReservation(5, false)).toBe(4); }); it("a failed dispatch gives the slot back", () => { diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 21bfcd2e00..cc783c9207 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -925,30 +925,41 @@ export function nonWipWorktreeHolderIdsOf( } /** - * Slots a candidate must clear to dispatch. A candidate that ALREADY holds a worktree subtracts its - * own slot: on release the slot TRANSFERS (it executes in the same worktree) rather than adding. + * FNXC:WorktreeCapacity 2026-08-01-04:07: + * Resolve whether worktree capacity applies to this candidate. A task that already holds a + * worktree allocates no additional tree when it dispatches, so the worktree dimension must not + * gate that transfer. Agent and per-column capacity continue to apply independently. */ -export function effectiveActiveWorktrees(reservedWorktreeSlots: number, candidateHoldsWorktree: boolean): number { - return reservedWorktreeSlots - (candidateHoldsWorktree ? 1 : 0); +export function resolveCandidateWorktreeCapacityLimit( + maxWorktrees: number | null, + candidateHoldsWorktree: boolean, +): number | null { + return candidateHoldsWorktree ? null : maxWorktrees; } /** * FNXC:WorktreeCapacity 2026-08-01-01:05: - * The two ledger MUTATIONS, named so they can be pinned alongside the totals they modify. + * The ledger MUTATIONS, named so they can be pinned alongside the totals they modify. * - * `reserveWorktreeOnDispatch` is the exact counterpart of `effectiveActiveWorktrees`: a candidate - * that already holds a worktree reuses it, so dispatch TRANSFERS the slot rather than adding one. - * The two must agree — subtracting for the gate but incrementing anyway would leak a slot per - * dispatch until the cap wedged. + * `reserveWorktreeOnDispatch` is the ledger counterpart of + * `resolveCandidateWorktreeCapacityLimit`: a candidate that already holds a worktree reuses it, so + * dispatch TRANSFERS the slot rather than adding one. The two must agree — bypassing allocation + * capacity for a transfer but incrementing anyway would leak a slot per dispatch until the cap + * wedged. * - * `releaseReservedSlot` carries the floor. A failed dispatch gives its slot back, and the - * `Math.max(0, …)` is what stops a double-release from handing out capacity that does not exist — - * a negative reserved count reads as free slots to every later comparison in the loop. + * `releaseWorktreeReservation` gives back only a slot that this candidate added. + * `releaseReservedSlot` carries the floor; its `Math.max(0, …)` stops a double-release from handing + * out capacity that does not exist — a negative reserved count reads as free slots to every later + * comparison in the loop. */ export function reserveWorktreeOnDispatch(reservedWorktreeSlots: number, candidateHoldsWorktree: boolean): number { return candidateHoldsWorktree ? reservedWorktreeSlots : reservedWorktreeSlots + 1; } +export function releaseWorktreeReservation(reservedWorktreeSlots: number, candidateHoldsWorktree: boolean): number { + return candidateHoldsWorktree ? reservedWorktreeSlots : releaseReservedSlot(reservedWorktreeSlots); +} + export function releaseReservedSlot(reservedSlots: number): number { return Math.max(0, reservedSlots - 1); } @@ -2956,11 +2967,19 @@ export class Scheduler { tasks, }); const candidateHoldsWorktree = nonWipWorktreeHolderIdSet.has(task.id); + /* + FNXC:WorktreeCapacity 2026-08-01-03:55 (live over-cap transfer deadlock): + A retained-worktree candidate consumes zero NEW worktree slots. Subtracting only its own + slot still deadlocked an already-over-cap durable ledger (12 - 1 remained 11 against 9), + and even a ledger exactly at cap produced zero slack and blocked the transfer. Remove only + the worktree dimension for this candidate; maxConcurrent, semaphore, and column capacity + still decide whether another agent may run. Candidates without a tree keep the full gate. + */ const concurrencyDiagnostic = computeConcurrencyGateDiagnostic({ agentSlots: reservedConcurrentSlots, maxConcurrent, - activeWorktrees: effectiveActiveWorktrees(reservedWorktreeSlots, candidateHoldsWorktree), - maxWorktrees, + activeWorktrees: reservedWorktreeSlots, + maxWorktrees: resolveCandidateWorktreeCapacityLimit(maxWorktrees, candidateHoldsWorktree), worktreeHolderTaskIds: [...inProgressTaskIds, ...nonWipWorktreeHolderIds], semaphore: this.options.semaphore, inProgressTaskIds, @@ -3056,7 +3075,7 @@ export class Scheduler { activeScopes.delete(task.id); activeScopeColumns.delete(task.id); } - reservedWorktreeSlots = releaseReservedSlot(reservedWorktreeSlots); + reservedWorktreeSlots = releaseWorktreeReservation(reservedWorktreeSlots, candidateHoldsWorktree); reservedConcurrentSlots = releaseReservedSlot(reservedConcurrentSlots); dispatchPrepByTaskId.delete(task.id); if (dropPreHeldExecutorSlot(task.id)) sem?.release();