FN-6063: enforce workflow scheduling capacity after hold release

Keep scheduler capacity checks aligned with fresh task state during workflow dispatch.

- bypass the startup slim-list memo when scheduling and after workflow hold release sweeps
- clear the startup slim-list memo when task.json writes move tasks between columns
- treat capacity-exhausted moveTask rejections as queued dispatches, log the reason, and cover the behavior with scheduler/store tests
- add a patch changeset for the published CLI package

Files changed:
 .changeset/workflow-capacity-scheduler.md          |  5 +++
 packages/core/src/__tests__/store-movement.test.ts | 15 ++++++++
 packages/core/src/store.ts                         |  1 +
 packages/engine/src/__tests__/scheduler.test.ts    | 40 ++++++++++++++++++++--
 packages/engine/src/scheduler.ts                   | 25 ++++++++++----
 5 files changed, 77 insertions(+), 9 deletions(-)

Fusion-Task-Id: FN-6063

Fusion-Task-Lineage: 80ddc0e7-f78d-483c-886a-27c4c07bde7a

.changeset/workflow-capacity-scheduler.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Fix workflow scheduling so in-progress column limits are enforced from fresh task state after hold-advancing sweep dispatches.

packages/core/src/__tests__/store-movement.test.ts

@@ -33,6 +33,21 @@ describe("TaskStore", () => {
   const createSourceIssueFixture = () => harness.createSourceIssueFixture();
   const insertLogEntryWithTimestamp = (...args: any[]) => (harness as any).insertLogEntryWithTimestamp(...args);
 
+  describe("listTasks startup memo invalidation", () => {
+    it("returns fresh slim task state after moveTask writes task json", async () => {
+      const task = await store.createTask({ description: "memo invalidation after move" });
+      await store.moveTask(task.id, "todo");
+
+      const beforeMove = await store.listTasks({ slim: true, includeArchived: false, startupMemo: true });
+      expect(beforeMove.find((listed) => listed.id === task.id)?.column).toBe("todo");
+
+      await store.moveTask(task.id, "in-progress");
+
+      const afterMove = await store.listTasks({ slim: true, includeArchived: false, startupMemo: true });
+      expect(afterMove.find((listed) => listed.id === task.id)?.column).toBe("in-progress");
+    });
+  });
+
   describe("moveTask — in-progress to triage", () => {
     it("allows moving an in-progress task to triage", async () => {
       const task = await store.createTask({ description: "test in-progress to triage" });

packages/core/src/store.ts

@@ -3131,6 +3131,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
   }
 
   private async writeTaskJsonFile(dir: string, task: Task): Promise<void> {
+    this.clearStartupSlimListMemo();
     const taskJsonPath = join(dir, "task.json");
     // Use a unique tmp filename per write so concurrent writers to the same task
     // don't race on a shared `task.json.tmp` (one rename consumes it, the other

packages/engine/src/__tests__/scheduler.test.ts

@@ -11,7 +11,7 @@ import {
   getUnmetSchedulingDependencies,
 } from "../scheduler.js";
 import { AgentSemaphore } from "../concurrency.js";
-import type { TaskStore, Task, TaskDetail } from "@fusion/core";
+import { makeTransitionRejection, TransitionRejectionError, type TaskStore, type Task, type TaskDetail } from "@fusion/core";
 import { existsSync } from "node:fs";
 import { readFile } from "node:fs/promises";
 import { schedulerLog } from "../logger.js";
@@ -599,7 +599,10 @@ describe("Scheduler", () => {
         if (column === "in-progress") {
           const inProgressCount = [...tasks.values()].filter((task) => task.column === "in-progress").length;
           if (inProgressCount >= 3) {
-            throw new Error("capacity-exhausted");
+            throw new TransitionRejectionError(
+              makeTransitionRejection("capacity-exhausted", "transition.rejected.capacityExhausted", true),
+              "queued — in-progress column at capacity",
+            );
           }
         }
         const updated = { ...current, column } as Task;
@@ -624,6 +627,7 @@ describe("Scheduler", () => {
       expect([...tasks.values()].filter((task) => task.column === "in-progress")).toHaveLength(3);
       expect(moveTask.mock.calls.filter((call) => call[1] === "in-progress")).toHaveLength(6);
       expect(vi.mocked(store.listTasks).mock.calls.length).toBeGreaterThanOrEqual(2);
+      expect(vi.mocked(store.listTasks).mock.calls.some(([options]) => options?.startupMemo === false)).toBe(true);
     });
 
     it("flag-OFF: todo dispatch is tagged as scheduler-sourced for redispatch guards", async () => {
@@ -1481,6 +1485,38 @@ describe("Scheduler", () => {
       expect(store.moveTask).not.toHaveBeenCalled();
     });
 
+    it("queues capacity-exhausted dispatches and continues to later candidates", async () => {
+      vi.mocked(existsSync).mockReturnValue(true);
+      vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
+
+      const tasks = [
+        createMockTask({ id: "FN-001", column: "todo", dependencies: [] }),
+        createMockTask({ id: "FN-002", column: "todo", dependencies: [] }),
+      ];
+      const store = createMockStore({
+        listTasks: vi.fn().mockResolvedValue(tasks),
+        getTask: vi.fn(async (taskId: string) => tasks.find((task) => task.id === taskId) ?? null),
+        getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 5, maxWorktrees: 10 }),
+        moveTask: vi.fn().mockRejectedValue(
+          new TransitionRejectionError(
+            makeTransitionRejection("capacity-exhausted", "transition.rejected.capacityExhausted", true),
+            "queued — in-progress column at capacity",
+          ),
+        ),
+      });
+
+      const scheduler = new Scheduler(store);
+      (scheduler as unknown as { running: boolean }).running = true;
+      await scheduler.schedule();
+
+      expect(store.moveTask).toHaveBeenCalledTimes(2);
+      expect(vi.mocked(store.updateTask).mock.calls.filter((call) => call[1]?.status === "queued").map((call) => call[0])).toEqual([
+        "FN-001",
+        "FN-002",
+      ]);
+      expect(store.logEntry).toHaveBeenCalledWith("FN-001", expect.stringContaining("queued — in-progress column at capacity"));
+    });
+
     it("respects maxWorktrees limit", async () => {
       const tasks = [
         createMockTask({ id: "FN-001", column: "in-progress" }),

packages/engine/src/scheduler.ts

@@ -13,6 +13,7 @@ import {
   type PrInfo,
   type AgentStore,
   type Settings,
+  TransitionRejectionError,
 } from "@fusion/core";
 import { existsSync } from "node:fs";
 import { readFile } from "node:fs/promises";
@@ -1225,7 +1226,7 @@ export class Scheduler {
     this.scheduling = true;
 
     try {
-      let tasks = await this.store.listTasks({ slim: true, includeArchived: false });
+      let tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false });
       let settings = await this.store.getSettings();
       this.idleSemaphoreLeakCandidateSince = recoverIdleSemaphoreLeak(
         this.options.semaphore,
@@ -1273,7 +1274,7 @@ export class Scheduler {
       // workflow hold handling and the generalized capacity-release path.
       if (isWorkflowColumnsEnabled(settings)) {
         await this.runHoldReleaseSweepPass();
-        tasks = await this.store.listTasks({ slim: true, includeArchived: false });
+        tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false });
         settings = await this.store.getSettings();
       }
 
@@ -2039,11 +2040,21 @@ export class Scheduler {
           effectiveNodeSource: effectiveNode.source,
           mergeRetries: 0,
         });
-        await this.store.moveTask(task.id, "in-progress", {
+        try {
-          moveSource: "scheduler",
+          await this.store.moveTask(task.id, "in-progress", {
-          allocateWorktree: (reservedNames) =>
+            moveSource: "scheduler",
-            this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings),
+            allocateWorktree: (reservedNames) =>
-        });
+              this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings),
+          });
+        } catch (error) {
+          if (error instanceof TransitionRejectionError && error.rejection.code === "capacity-exhausted") {
+            await this.store.updateTask(task.id, { status: "queued" });
+            const reason = error.message || "queued — in-progress column at capacity";
+            await this.logDispatchQueuedReason(task.id, reason, `capacity-exhausted:${reason}`);
+            continue;
+          }
+          throw error;
+        }
         await this.store.updateTask(task.id, {
           dispatchStormCount: nextDispatchStormCount,
           lastDispatchAt: dispatchTimestamp,