diff --git a/.changeset/workflow-capacity-scheduler.md b/.changeset/workflow-capacity-scheduler.md new file mode 100644 index 0000000000..f5838f3ef3 --- /dev/null +++ b/.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. diff --git a/packages/core/src/__tests__/store-movement.test.ts b/packages/core/src/__tests__/store-movement.test.ts index 3062bd3136..b8885f49fd 100644 --- a/packages/core/src/__tests__/store-movement.test.ts +++ b/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" }); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 03b98a1abc..c35cda22cb 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -3131,6 +3131,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} } private async writeTaskJsonFile(dir: string, task: Task): Promise { + 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 diff --git a/packages/engine/src/__tests__/scheduler.test.ts b/packages/engine/src/__tests__/scheduler.test.ts index 2a3341e8a7..67cb19e479 100644 --- a/packages/engine/src/__tests__/scheduler.test.ts +++ b/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" }), diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 4501e22e08..6d7e0f1801 100644 --- a/packages/engine/src/scheduler.ts +++ b/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", { - moveSource: "scheduler", - allocateWorktree: (reservedNames) => - this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings), - }); + try { + await this.store.moveTask(task.id, "in-progress", { + moveSource: "scheduler", + 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,