root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

FN-6063: enforce workflow scheduling capacity after hold release

Browse Source 
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
This commit is contained in:
gsxdsm
2026-06-09 02:03:22 -07:00
parent a566f498d8
commit 4b4c32dc18
5 changed files with 77 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
packages/core/src/__tests__/store-movement.test.ts
View File

@@ -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" });

1
packages/core/src/store.ts
View File

@@ -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

40
packages/engine/src/__tests__/scheduler.test.ts
View File

@@ -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" }),

25
packages/engine/src/scheduler.ts
View File

@@ -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,