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

fix(engine): unblock retained worktree dispatch

Browse Source 
Allow queued tasks to reuse worktrees they already hold when the durable worktree ledger is at or above capacity. Preserve independent agent and semaphore limits, and avoid releasing a worktree slot that a rejected transfer never acquired.
This commit is contained in:
gsxdsm
2026-07-31 21:21:28 -07:00
parent 71576d9536
commit a3c0501822
4 changed files with 106 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fix-retained-worktree-dispatch.md Normal file
View File

@@ -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.

46
packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts
View File

@@ -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();

49
packages/engine/src/__tests__/scheduler-worktree-capacity-arithmetic.test.ts
View File

@@ -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> = {}): 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", () => {

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

@@ -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();