FN-7373: enforce active worktree cap

Enforce maxWorktrees at the task-store transition boundary so active execution worktrees cannot exceed the operator cap.

- Reject allocated moves into in-progress when existing active or pending holders already meet maxWorktrees.
- Count in-progress tasks and pending in-progress transitions while excluding the task being moved.
- Cover scheduler and hold-release paths with regression tests and document the hard-cap invariant.
- Add a patch changeset for the published CLI package.

Files changed:
 .changeset/fn-7373-max-worktrees-cap.md            |  7 ++
 docs/architecture.md                               |  2 +-
 packages/core/src/__tests__/store-movement.test.ts | 80 +++++++++++++++++++++-
 packages/core/src/store.ts                         | 50 ++++++++++++++
 packages/engine/src/__tests__/hold-release.test.ts | 52 ++++++++++++++
 .../__tests__/scheduler-workflow-cutover.test.ts   | 53 ++++++++++++++
 6 files changed, 242 insertions(+), 2 deletions(-)

Fusion-Task-Id: FN-7373
Fusion-Task-Lineage: be4a9649-f6ae-475f-9395-07433e98de5c
Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-07-01 10:18:32 -07:00
parent 17505231d6
commit 919420e11a
6 changed files with 242 additions and 2 deletions

View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Enforce maxWorktrees as a hard cap on active execution worktrees.
category: fix
dev: TaskStore rejects allocated in-progress moves once active holders reach maxWorktrees, independent of maxConcurrent.

View File

@@ -1117,7 +1117,7 @@ The run-audit system records every mutation performed by the engine across four
Events are tied to specific run IDs for end-to-end traceability. Events are tied to specific run IDs for end-to-end traceability.
For scheduler concurrency diagnostics, the queued reason now names the active limiter(s) and usage (for example `gate=maxConcurrent ...`). The reason includes the `bindingGates` (`maxConcurrent`/`maxWorktrees`/`semaphore`), per-gate `{ used, limit, slack }`, `holders`, and computed `available`. `holders.maxConcurrent` and `holders.maxWorktrees` are current `in-progress` task IDs; `holders.semaphore` mirrors that set but semaphore slots can also be consumed by triage/merge agents outside `in-progress`. So if `semaphore.used` exceeds the visible holder list, that usually indicates non-execution agents are legitimately consuming shared capacity (not stale accounting). These queued-reason logs are transition-only: a newly emitted line indicates the limiter signature changed or the condition cleared and later reappeared, not that a poll loop simply observed the same blocked state again. For scheduler concurrency diagnostics, the queued reason now names the active limiter(s) and usage (for example `gate=maxConcurrent ...`). The reason includes the `bindingGates` (`maxConcurrent`/`maxWorktrees`/`semaphore`), per-gate `{ used, limit, slack }`, `holders`, and computed `available`. `holders.maxConcurrent` and `holders.maxWorktrees` are current `in-progress` task IDs; `holders.semaphore` mirrors that set but semaphore slots can also be consumed by triage/merge agents outside `in-progress`. So if `semaphore.used` exceeds the visible holder list, that usually indicates non-execution agents are legitimately consuming shared capacity (not stale accounting). `maxWorktrees` is also enforced inside `TaskStore.moveTaskInternal` when committing an allocated move into `in-progress`, making it a hard active execution worktree cap even when workflow WIP/`maxConcurrent` would allow more tasks. These queued-reason logs are transition-only: a newly emitted line indicates the limiter signature changed or the condition cleared and later reappeared, not that a poll loop simply observed the same blocked state again.
**Run audit endpoints:** **Run audit endpoints:**
- `GET /api/agents/:id/runs/:runId/audit` — Returns audit trail for a specific agent run - `GET /api/agents/:id/runs/:runId/audit` — Returns audit trail for a specific agent run

View File

@@ -6,7 +6,7 @@ import { existsSync } from "node:fs";
import * as projectMemory from "../project-memory.js"; import * as projectMemory from "../project-memory.js";
import { AgentStore } from "../agent-store.js"; import { AgentStore } from "../agent-store.js";
import { CentralDatabase } from "../central-db.js"; import { CentralDatabase } from "../central-db.js";
import { TaskStore, TaskHasDependentsError } from "../store.js"; import { TaskStore, TaskHasDependentsError, TransitionRejectionError } from "../store.js";
import { TASK_DONE_BYPASS_BLOCKER_MESSAGE, allowsAutoMergeProcessing, resolveEffectiveAutoMerge } from "../task-merge.js"; import { TASK_DONE_BYPASS_BLOCKER_MESSAGE, allowsAutoMergeProcessing, resolveEffectiveAutoMerge } from "../task-merge.js";
import { buildResearchDocumentKey, type Task } from "../types.js"; import { buildResearchDocumentKey, type Task } from "../types.js";
import { createSharedTaskStoreTestHarness, makeTmpDir } from "./store-test-helpers.js"; import { createSharedTaskStoreTestHarness, makeTmpDir } from "./store-test-helpers.js";
@@ -63,6 +63,84 @@ describe("TaskStore", () => {
}); });
}); });
describe("moveTask — maxWorktrees hard active-worktree cap", () => {
async function createActiveHolder(index: number): Promise<Task> {
const task = await store.createTask({ description: `active holder ${index}` });
await store.moveTask(task.id, "todo");
return store.moveTask(task.id, "in-progress", {
moveSource: "scheduler",
allocateWorktree: () => `/tmp/fn-7373-holder-${index}`,
});
}
it("rejects a fifth allocated in-progress move when maxWorktrees is 4 and maxConcurrent is higher", async () => {
await store.updateSettings({ maxWorktrees: 4, maxConcurrent: 10 });
for (let i = 0; i < 4; i += 1) {
await createActiveHolder(i);
}
const fifth = await store.createTask({ description: "fifth holder" });
await store.moveTask(fifth.id, "todo");
await expect(store.moveTask(fifth.id, "in-progress", {
moveSource: "scheduler",
allocateWorktree: () => "/tmp/fn-7373-holder-5",
})).rejects.toMatchObject({
rejection: expect.objectContaining({ code: "capacity-exhausted", retryable: true }),
} satisfies Partial<TransitionRejectionError>);
const active = (await store.listTasks({ includeArchived: false })).filter((task) => task.column === "in-progress");
expect(active).toHaveLength(4);
expect((await store.getTask(fifth.id))?.column).toBe("todo");
});
it("serializes racing allocated moves so only one final maxWorktrees slot is committed", async () => {
await store.updateSettings({ maxWorktrees: 4, maxConcurrent: 10 });
for (let i = 0; i < 3; i += 1) {
await createActiveHolder(i);
}
const contenders = await Promise.all([
store.createTask({ description: "race contender a" }),
store.createTask({ description: "race contender b" }),
]);
for (const contender of contenders) {
await store.moveTask(contender.id, "todo");
}
const results = await Promise.allSettled(contenders.map((contender, index) =>
store.moveTask(contender.id, "in-progress", {
moveSource: "scheduler",
allocateWorktree: () => `/tmp/fn-7373-race-${index}`,
}),
));
expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(1);
expect(results.filter((result) => result.status === "rejected")).toHaveLength(1);
const active = (await store.listTasks({ includeArchived: false })).filter((task) => task.column === "in-progress");
expect(active).toHaveLength(4);
const activeContenders = contenders.filter((contender) => active.some((task) => task.id === contender.id));
expect(activeContenders).toHaveLength(1);
});
it("does not count idle in-review worktrees against maxWorktrees active allocation", async () => {
await store.updateSettings({ maxWorktrees: 1, maxConcurrent: 10 });
for (let i = 0; i < 3; i += 1) {
const reviewing = await createActiveHolder(i);
await store.moveTask(reviewing.id, "in-review", { moveSource: "engine", allowDirectInReviewMove: true });
await store.updateTask(reviewing.id, { worktree: `/tmp/fn-7373-review-${i}` });
}
const next = await store.createTask({ description: "next active holder" });
await store.moveTask(next.id, "todo");
const moved = await store.moveTask(next.id, "in-progress", {
moveSource: "scheduler",
allocateWorktree: () => "/tmp/fn-7373-next-active",
});
expect(moved.column).toBe("in-progress");
expect(moved.worktree).toBe("/tmp/fn-7373-next-active");
expect((await store.listTasks({ includeArchived: false })).filter((task) => task.column === "in-progress")).toHaveLength(1);
});
});
describe("moveTask — autoMerge follows live settings on in-review", () => { describe("moveTask — autoMerge follows live settings on in-review", () => {
async function createInProgressTask(description: string): Promise<Task> { async function createInProgressTask(description: string): Promise<Task> {

View File

@@ -7798,6 +7798,28 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
} }
} }
/*
FNXC:WorkflowCapacity 2026-07-01-00:00:
maxWorktrees is an operator resource cap for active execution checkouts, not a workflow WIP policy. Re-check it inside the move transaction before committing an allocated in-progress worktree so maxConcurrent, stale scheduler snapshots, or racing releases cannot create a fifth active holder.
*/
if (fromColumn !== toColumn && toColumn === "in-progress" && task.worktree) {
const maxWorktrees = typeof mergedSettingsForMove.maxWorktrees === "number" && Number.isFinite(mergedSettingsForMove.maxWorktrees)
? mergedSettingsForMove.maxWorktrees
: 4;
const activeHolders = this.countActiveExecutionWorktreeHoldersSync({ excludeTaskId: id });
if (activeHolders >= maxWorktrees) {
throw new TransitionRejectionError(
makeTransitionRejection(
"capacity-exhausted",
"transition.rejected.capacityExhausted",
true,
`Active worktree cap exhausted (${activeHolders}/${maxWorktrees})`,
),
`Cannot move ${id} to '${toColumn}': active worktree cap exhausted (${activeHolders}/${maxWorktrees})`,
);
}
}
this.upsertTaskWithFtsRecovery(task); this.upsertTaskWithFtsRecovery(task);
this.insertRunAuditEventRow({ this.insertRunAuditEventRow({
taskId: id, taskId: id,
@@ -15993,6 +16015,34 @@ ${stepsSection}`;
return count; return count;
} }
private countActiveExecutionWorktreeHoldersSync(params: { excludeTaskId: string }): number {
const rows = this.db
.prepare(
`SELECT id, "column" AS col, transitionPending AS tp
FROM tasks
WHERE deletedAt IS NULL
AND id != ?
AND ("column" = 'in-progress' OR (transitionPending IS NOT NULL AND transitionPending != ''))`,
)
.all(params.excludeTaskId) as Array<{ id: string; col: string; tp: string | null }>;
let count = 0;
for (const row of rows) {
if (row.col === "in-progress") {
count += 1;
continue;
}
if (!row.tp) continue;
try {
const parsed = JSON.parse(row.tp) as { toColumn?: unknown };
if (parsed.toColumn === "in-progress") count += 1;
} catch {
// Corrupt marker — do not inflate active worktree usage.
}
}
return count;
}
getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined { getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined {
const row = this.db const row = this.db
.prepare("SELECT workflowId, stepIds FROM task_workflow_selection WHERE taskId = ?") .prepare("SELECT workflowId, stepIds FROM task_workflow_selection WHERE taskId = ?")

View File

@@ -370,6 +370,58 @@ describe("hold-release sweep (U6)", () => {
expect((await store.getTask(held))?.column).toBe("todo"); expect((await store.getTask(held))?.column).toBe("todo");
}); });
it("does not release a fifth active worktree when maxWorktrees is 4 and maxConcurrent is higher", async () => {
await store.updateSettings({ maxWorktrees: 4, maxConcurrent: 10 } as Parameters<typeof store.updateSettings>[0]);
for (let i = 0; i < 4; i += 1) {
const occupant = await store.createTask({ description: `maxWorktrees occupant ${i}` });
setColumn(store, occupant.id, "in-progress");
await store.updateTask(occupant.id, { worktree: `/tmp/fn-7373-occupant-${i}` });
}
const held = await seedTodoCard();
const release = vi.fn();
const result = await runHoldReleaseSweep(store, {
now: () => Date.now(),
reserveSlot: () => ({ release }),
allocateWorktree: () => "/tmp/fn-7373-fifth",
});
expect(result.released).not.toContain(held);
expect((await store.getTask(held))?.column).toBe("todo");
expect((await store.listTasks({ includeArchived: false })).filter((task) => task.column === "in-progress")).toHaveLength(4);
expect(release).toHaveBeenCalledTimes(1);
});
it("racing held releases cannot commit more than one final maxWorktrees slot", async () => {
await store.updateSettings({ maxWorktrees: 4, maxConcurrent: 10 } as Parameters<typeof store.updateSettings>[0]);
for (let i = 0; i < 3; i += 1) {
const occupant = await store.createTask({ description: `race occupant ${i}` });
setColumn(store, occupant.id, "in-progress");
await store.updateTask(occupant.id, { worktree: `/tmp/fn-7373-race-occupant-${i}` });
}
const heldA = await seedTodoCard();
const heldB = await seedTodoCard();
let reserveCount = 0;
let releaseCount = 0;
const result = await runHoldReleaseSweep(store, {
now: () => Date.now(),
reserveSlot: () => {
reserveCount += 1;
return { release: () => { releaseCount += 1; } };
},
allocateWorktree: (_task, reservedNames) => `/tmp/fn-7373-race-${reservedNames.size}`,
});
const releasedHeldTasks = result.released.filter((taskId) => taskId === heldA || taskId === heldB);
expect(releasedHeldTasks).toHaveLength(1);
const active = (await store.listTasks({ includeArchived: false })).filter((task) => task.column === "in-progress");
expect(active).toHaveLength(4);
expect([heldA, heldB].filter((taskId) => active.some((task) => task.id === taskId))).toHaveLength(1);
expect(reserveCount).toBe(2);
expect(releaseCount).toBe(1);
});
it("capacity release respects cards mid-transitionPending (they hold the slot from commit time)", async () => { it("capacity release respects cards mid-transitionPending (they hold the slot from commit time)", async () => {
await store.updateSettings({ maxConcurrent: 1 } as Parameters<typeof store.updateSettings>[0]); await store.updateSettings({ maxConcurrent: 1 } as Parameters<typeof store.updateSettings>[0]);
// Occupant has committed into in-progress AND is mid-transitionPending — it // Occupant has committed into in-progress AND is mid-transitionPending — it

View File

@@ -249,10 +249,63 @@ describe("Scheduler workflow cutover", () => {
expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything());
expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null })); expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null }));
expect(store.logEntry).toHaveBeenCalledWith(
"FN-002",
expect.stringContaining("gate=maxWorktrees; maxConcurrent used=1/4"),
);
expect(store.logEntry).toHaveBeenCalledWith(
"FN-002",
expect.stringContaining("maxWorktrees used=1/1"),
);
expect(onSchedule).not.toHaveBeenCalledWith(expect.objectContaining({ id: "FN-002" })); expect(onSchedule).not.toHaveBeenCalledWith(expect.objectContaining({ id: "FN-002" }));
expect(ready.column).toBe("todo"); expect(ready.column).toBe("todo");
}); });
it("does not release work when already over maxWorktrees even if maxConcurrent has slack", async () => {
const active = Array.from({ length: 5 }, (_, index) => task({ id: `FN-10${index}`, column: "in-progress" }));
const ready = task({ id: "FN-200", status: "queued" });
const store = storeWith([...active, ready], { maxConcurrent: 10, maxWorktrees: 4 });
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.moveTask).not.toHaveBeenCalledWith("FN-200", "in-progress", expect.anything());
expect(store.updateTask).toHaveBeenCalledWith("FN-200", { status: "queued" });
expect(store.logEntry).toHaveBeenCalledWith(
"FN-200",
expect.stringContaining("gate=maxWorktrees; maxConcurrent used=5/10"),
);
expect(store.logEntry).toHaveBeenCalledWith(
"FN-200",
expect.stringContaining("maxWorktrees used=5/4"),
);
expect(onSchedule).not.toHaveBeenCalled();
expect(ready.column).toBe("todo");
});
it("releases one ready task when maxWorktrees has exactly one remaining slot and maxConcurrent is higher", async () => {
const active = Array.from({ length: 3 }, (_, index) => task({ id: `FN-30${index}`, column: "in-progress" }));
const first = task({ id: "FN-401", status: "queued" });
const second = task({ id: "FN-402", status: "queued" });
const store = storeWith([...active, first, second], { maxConcurrent: 10, maxWorktrees: 4 });
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledTimes(1);
expect(store.moveTask).toHaveBeenCalledWith("FN-401", "in-progress", expect.anything());
expect(store.moveTask).not.toHaveBeenCalledWith("FN-402", "in-progress", expect.anything());
expect(store.logEntry).toHaveBeenCalledWith(
"FN-402",
expect.stringContaining("gate=maxWorktrees; maxConcurrent used=4/10"),
);
expect(onSchedule).toHaveBeenCalledTimes(1);
});
it("reserves same-sweep capacity so only one ready task is released into one slot", async () => { it("reserves same-sweep capacity so only one ready task is released into one slot", async () => {
const first = task({ id: "FN-001", status: "queued" }); const first = task({ id: "FN-001", status: "queued" });
const second = task({ id: "FN-002", status: "queued" }); const second = task({ id: "FN-002", status: "queued" });