From 919420e11a7897f6e417034699338ad5c4197a6e Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 1 Jul 2026 10:18:32 -0700 Subject: [PATCH] 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) --- .changeset/fn-7373-max-worktrees-cap.md | 7 ++ docs/architecture.md | 2 +- .../core/src/__tests__/store-movement.test.ts | 80 ++++++++++++++++++- packages/core/src/store.ts | 50 ++++++++++++ .../engine/src/__tests__/hold-release.test.ts | 52 ++++++++++++ .../scheduler-workflow-cutover.test.ts | 53 ++++++++++++ 6 files changed, 242 insertions(+), 2 deletions(-) create mode 100644 .changeset/fn-7373-max-worktrees-cap.md diff --git a/.changeset/fn-7373-max-worktrees-cap.md b/.changeset/fn-7373-max-worktrees-cap.md new file mode 100644 index 0000000000..728c65e90f --- /dev/null +++ b/.changeset/fn-7373-max-worktrees-cap.md @@ -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. diff --git a/docs/architecture.md b/docs/architecture.md index 239e200755..6915c88c2d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -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. -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:** - `GET /api/agents/:id/runs/:runId/audit` — Returns audit trail for a specific agent run diff --git a/packages/core/src/__tests__/store-movement.test.ts b/packages/core/src/__tests__/store-movement.test.ts index 9432dc55f0..8757fd124c 100644 --- a/packages/core/src/__tests__/store-movement.test.ts +++ b/packages/core/src/__tests__/store-movement.test.ts @@ -6,7 +6,7 @@ import { existsSync } from "node:fs"; import * as projectMemory from "../project-memory.js"; import { AgentStore } from "../agent-store.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 { buildResearchDocumentKey, type Task } from "../types.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 { + 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); + + 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", () => { async function createInProgressTask(description: string): Promise { diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index b143d6da99..83f3804612 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -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.insertRunAuditEventRow({ taskId: id, @@ -15993,6 +16015,34 @@ ${stepsSection}`; 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 { const row = this.db .prepare("SELECT workflowId, stepIds FROM task_workflow_selection WHERE taskId = ?") diff --git a/packages/engine/src/__tests__/hold-release.test.ts b/packages/engine/src/__tests__/hold-release.test.ts index df3f87fe24..fd565eb3a6 100644 --- a/packages/engine/src/__tests__/hold-release.test.ts +++ b/packages/engine/src/__tests__/hold-release.test.ts @@ -370,6 +370,58 @@ describe("hold-release sweep (U6)", () => { 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[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[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 () => { await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); // Occupant has committed into in-progress AND is mid-transitionPending — it diff --git a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts index 5f1c8f179f..90e985b504 100644 --- a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts +++ b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts @@ -249,10 +249,63 @@ describe("Scheduler workflow cutover", () => { expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); 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(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 () => { const first = task({ id: "FN-001", status: "queued" }); const second = task({ id: "FN-002", status: "queued" });