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>
2026-07-01 10:18:32 -07:00
parent 17505231d6
commit 919420e11a
6 changed files with 242 additions and 2 deletions
--- a/.changeset/fn-7373-max-worktrees-cap.md
+++ 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.
--- 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
--- 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<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", () => {
    async function createInProgressTask(description: string): Promise<Task> {
--- 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 = ?")
--- 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<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 () => {
    await store.updateSettings({ maxConcurrent: 1 } as Parameters<typeof store.updateSettings>[0]);
    // Occupant has committed into in-progress AND is mid-transitionPending — it
--- 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" });