From 0818fc1da153fbfed0b25acb62e3807e1ef1f422 Mon Sep 17 00:00:00 2001 From: flexi767 <96955327+flexi767@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:10:34 +0200 Subject: [PATCH] fix(engine): respect user-paused dispatch stops (#2371) Re-lands #2337 directly on current main after its temporary base branch was merged and deleted.\n\n- excludes userPaused tasks from scheduler and remembered-owner selection\n- includes userPaused in candidacy fingerprints and unpause scheduling\n- keeps normal unpaused dispatch behavior\n- includes regressions and a release changeset\n\nValidation on current main: scheduler suites 50/50, @fusion/core typecheck, and @fusion/engine typecheck passed. The PostgreSQL routing file was discovered but skipped without a configured test database. ## Summary by CodeRabbit * **Bug Fixes** * Manually parked/paused tasks are no longer selected or dispatched while they remain paused. * A task only re-enters dispatch flow after it is explicitly unpaused. * Unpausing a task promptly refreshes scheduling and makes it eligible for dispatch. * Scheduler state updates now correctly react to pause status changes (including when pause is represented via `userPaused`). * **Tests** * Expanded scheduler and routing regression coverage for pause/unpause and dispatch invalidation behavior. --------- Co-authored-by: v Co-authored-by: gsxdsm --- .changeset/respect-user-paused-dispatch.md | 7 ++ .../agent-store-routing-policy.test.ts | 16 ++++ .../core/src/task-store/branch-group-ops.ts | 5 +- .../scheduler-auto-claim-invalidation.test.ts | 13 +++ .../scheduler-trait-dispatch.test.ts | 20 +++-- .../scheduler-workflow-cutover.test.ts | 88 +++++++++++++++---- packages/engine/src/hold-release.ts | 63 ++++--------- packages/engine/src/scheduler.ts | 54 ++++++++---- 8 files changed, 178 insertions(+), 88 deletions(-) create mode 100644 .changeset/respect-user-paused-dispatch.md diff --git a/.changeset/respect-user-paused-dispatch.md b/.changeset/respect-user-paused-dispatch.md new file mode 100644 index 0000000000..36974a48ea --- /dev/null +++ b/.changeset/respect-user-paused-dispatch.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep manually parked tasks out of scheduler and remembered-owner dispatch until explicitly unpaused. +category: fix +dev: Treats either paused flag as a dispatch stop and invalidates scheduler candidacy when userPaused changes. diff --git a/packages/core/src/__tests__/agent-store-routing-policy.test.ts b/packages/core/src/__tests__/agent-store-routing-policy.test.ts index ff4b16131b..2881b2aa1b 100644 --- a/packages/core/src/__tests__/agent-store-routing-policy.test.ts +++ b/packages/core/src/__tests__/agent-store-routing-policy.test.ts @@ -224,6 +224,22 @@ pgTest("task→agent routing policy (issue #2015)", () => { }); describe("selectNextTaskForAgent bind compatibility", () => { + it("does not select a remembered-owner todo task when only userPaused remains true", async () => { + const executor = await agentStore.createAgent({ name: "Exec", role: "executor" }); + const task = await h.store().createTask({ description: "manually parked work" }); + await h.store().updateTask(task.id, { assignedAgentId: executor.id }); + await h.store().moveTask(task.id, "todo"); + await h.store().moveTask(task.id, "in-progress"); + await h.store().moveTask(task.id, "todo", { moveSource: "user" }); + + const parked = await h.store().getTask(task.id); + expect(parked).toMatchObject({ assignedAgentId: executor.id, userPaused: true }); + expect(parked?.paused).not.toBe(true); + await expect( + h.store().selectNextTaskForAgent(executor.id, { id: executor.id, role: executor.role }), + ).resolves.toBeNull(); + }); + it("does not re-select a mis-bound in-progress implementation task for a role-incompatible agent", async () => { const liaison = await agentStore.createAgent({ name: "Liaison", role: "custom" }); const task = await h.store().createTask({ description: "mis-bound work" }); diff --git a/packages/core/src/task-store/branch-group-ops.ts b/packages/core/src/task-store/branch-group-ops.ts index 2d2f7e03d4..779fd2e85b 100644 --- a/packages/core/src/task-store/branch-group-ops.ts +++ b/packages/core/src/task-store/branch-group-ops.ts @@ -196,7 +196,10 @@ export async function selectNextTaskForAgentImpl(store: TaskStore, agentId: stri const roleCompatibleAssignedTasks = assignedTasks.filter(isBindCompatible); - const todoCandidates = roleCompatibleAssignedTasks.filter((task) => task.column === "todo" && task.paused !== true); + /** FNXC:TaskDispatch 2026-07-19-14:40: remembered ownership must not reselect an operator-parked task when `userPaused` remains true but legacy `paused` is false. */ + const todoCandidates = roleCompatibleAssignedTasks.filter( + (task) => task.column === "todo" && task.paused !== true && task.userPaused !== true, + ); const readyTodo = todoCandidates .filter((task) => { diff --git a/packages/engine/src/__tests__/scheduler-auto-claim-invalidation.test.ts b/packages/engine/src/__tests__/scheduler-auto-claim-invalidation.test.ts index 10b669e99c..c8054954dd 100644 --- a/packages/engine/src/__tests__/scheduler-auto-claim-invalidation.test.ts +++ b/packages/engine/src/__tests__/scheduler-auto-claim-invalidation.test.ts @@ -95,6 +95,7 @@ describe("Scheduler auto-claim snapshot invalidation", () => { it.each([ ["column", { column: "in-progress" }], ["paused", { paused: true }], + ["userPaused", { userPaused: true }], ["assignedAgentId", { assignedAgentId: "agent-1" }], ["checkedOutBy", { checkedOutBy: "agent-2" }], ["deletedAt", { deletedAt: "2026-01-02T00:00:00.000Z" }], @@ -113,6 +114,18 @@ describe("Scheduler auto-claim snapshot invalidation", () => { expect(invalidate).toHaveBeenNthCalledWith(2, "task:updated"); }); + it("triggers immediate scheduling when a userPaused-only task is unpaused", () => { + const { store, emit } = createStore(); + const scheduler = new Scheduler(store, {}); + const schedule = vi.spyOn(scheduler, "schedule").mockResolvedValue(undefined); + (scheduler as unknown as { running: boolean }).running = true; + + emit("task:updated", createTask({ userPaused: true })); + emit("task:updated", createTask({ userPaused: false })); + + expect(schedule).toHaveBeenCalledTimes(1); + }); + it("invalidates on first-sighting task:updated with no stored fingerprint", () => { const invalidate = vi.fn(); const { store, emit } = createStore(); diff --git a/packages/engine/src/__tests__/scheduler-trait-dispatch.test.ts b/packages/engine/src/__tests__/scheduler-trait-dispatch.test.ts index 49ca794d7a..e9eff4abd6 100644 --- a/packages/engine/src/__tests__/scheduler-trait-dispatch.test.ts +++ b/packages/engine/src/__tests__/scheduler-trait-dispatch.test.ts @@ -89,6 +89,12 @@ function storeWith(tasks: Task[], ir: WorkflowIr, settings: Record { const cur = byId.get(id); if (cur) cur.column = column; return cur as Task; }), + moveTaskIf: vi.fn(async (id: string, column: Task["column"], predicate: (live: Task) => boolean | Promise) => { + const cur = byId.get(id)!; + if (!await predicate(cur) || cur.column === column) return { task: cur, moved: false }; + cur.column = column; + return { task: cur, moved: true }; + }), logEntry: vi.fn(async () => undefined), recordRunAuditEvent: vi.fn(async () => undefined), getCompletionHandoffAcceptedMarker: vi.fn(async () => null), @@ -132,14 +138,14 @@ describe("hold/release sweep — capacity (U4)", () => { const first = await runHoldReleaseSweep(store, { now: () => Date.now() }); expect(first.released).toEqual([]); // saturated (1/1) expect(first.held.some((h) => h.taskId === "H" && h.reason === "downstream-full")).toBe(true); - expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.moveTaskIf).not.toHaveBeenCalled(); // Free the slot (occupant leaves in-progress) and sweep again. occupant.column = "done"; const second = await runHoldReleaseSweep(store, { now: () => Date.now() }); expect(second.released).toEqual(["H"]); - expect(store.moveTask).toHaveBeenCalledTimes(1); - expect(store.moveTask).toHaveBeenCalledWith("H", "in-progress", expect.anything()); + expect(store.moveTaskIf).toHaveBeenCalledTimes(1); + expect(store.moveTaskIf).toHaveBeenCalledWith("H", "in-progress", expect.any(Function), expect.anything()); }); it("counts a mid-transition (transitionPending) card toward the cap (scenario 2, countPending)", async () => { @@ -169,8 +175,8 @@ describe("hold/release sweep — capacity (U4)", () => { inB.column = "done"; const result2 = await runHoldReleaseSweep(store, { now: () => Date.now() }); expect(result2.released).toEqual(["H"]); - expect(store.moveTask).toHaveBeenCalledTimes(1); - expect(store.moveTask).toHaveBeenCalledWith("H", "wip-a", expect.anything()); + expect(store.moveTaskIf).toHaveBeenCalledTimes(1); + expect(store.moveTaskIf).toHaveBeenCalledWith("H", "wip-a", expect.any(Function), expect.anything()); }); it("never releases a paused or user-paused card (scenario 5)", async () => { @@ -180,7 +186,7 @@ describe("hold/release sweep — capacity (U4)", () => { const result = await runHoldReleaseSweep(store, { now: () => Date.now() }); expect(result.released).toEqual([]); - expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.moveTaskIf).not.toHaveBeenCalled(); }); }); @@ -212,7 +218,7 @@ describe("no-hold workflow saturation (scenario 4)", () => { const result = await runHoldReleaseSweep(store, { now: () => Date.now() }); // intake is not a hold column → not managed by the sweep at all. expect(result.released).toEqual([]); - expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.moveTaskIf).not.toHaveBeenCalled(); }); it("the in-txn capacity policy rejects a move into a saturated shared pool (the gate for no-hold moves)", () => { diff --git a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts index 778185b4bf..d14229b716 100644 --- a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts +++ b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts @@ -57,6 +57,12 @@ function storeWith( if (current) current.column = column; return current as Task; }), + moveTaskIf: vi.fn(async (id: string, column: Task["column"], predicate: (live: Task) => boolean | Promise) => { + const current = byId.get(id)!; + if (!await predicate(current) || current.column === column) return { task: current, moved: false }; + current.column = column; + return { task: current, moved: true }; + }), parseFileScopeFromPrompt: vi.fn(async () => []), logEntry: vi.fn(async () => undefined), getRootDir: vi.fn(() => "/tmp/project"), @@ -170,7 +176,7 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - expect(store.moveTask).toHaveBeenCalledWith("FN-100", "in-progress", expect.objectContaining({ + expect(store.moveTaskIf).toHaveBeenCalledWith("FN-100", "in-progress", expect.any(Function), expect.objectContaining({ moveSource: "scheduler", allocateWorktree: expect.any(Function), })); @@ -184,6 +190,54 @@ describe("Scheduler workflow cutover", () => { expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-100", column: "in-progress" })); }); + it("does not dispatch an operator-parked todo task when only userPaused remains true", async () => { + const parked = task({ id: "FN-PAUSED", paused: false, userPaused: true }); + const store = storeWith([parked]); + const onSchedule = vi.fn(); + const scheduler = new Scheduler(store, { onSchedule }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + expect(store.moveTaskIf).not.toHaveBeenCalled(); + expect(store.updateTask).not.toHaveBeenCalled(); + expect(onSchedule).not.toHaveBeenCalled(); + expect(parked.column).toBe("todo"); + }); + + it("does not dispatch when an operator sets userPaused after the queue snapshot", async () => { + const ready = task({ id: "FN-CONCURRENT-PAUSE", paused: false, userPaused: false }); + const store = storeWith([ready]); + vi.mocked(store.getTask).mockResolvedValue({ ...ready, userPaused: true }); + const onSchedule = vi.fn(); + const scheduler = new Scheduler(store, { onSchedule }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + expect(store.moveTaskIf).not.toHaveBeenCalled(); + expect(onSchedule).not.toHaveBeenCalled(); + expect(ready.column).toBe("todo"); + }); + + it("does not dispatch when userPaused wins the atomic move race", async () => { + const ready = task({ id: "FN-ATOMIC-PAUSE", paused: false, userPaused: false }); + const store = storeWith([ready]); + vi.mocked(store.moveTaskIf).mockImplementation(async (_id, _column, predicate) => { + ready.userPaused = true; + return { task: ready, moved: await predicate(ready) }; + }); + const onSchedule = vi.fn(); + const scheduler = new Scheduler(store, { onSchedule }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + expect(store.moveTaskIf).toHaveBeenCalled(); + expect(ready.column).toBe("todo"); + expect(onSchedule).not.toHaveBeenCalled(); + }); + /* FNXC:WorkflowScheduling 2026-07-07-00:00: FN-7648 regression: a custom workflow's intake column can be renamed away from @@ -223,7 +277,7 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - expect(store.moveTask).not.toHaveBeenCalledWith("FN-300", "in-progress", expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-300", "in-progress", expect.anything(), expect.anything()); expect(unplanned.column).toBe("ideas"); expect(onSchedule).not.toHaveBeenCalledWith(expect.objectContaining({ id: "FN-300" })); }); @@ -242,7 +296,7 @@ describe("Scheduler workflow cutover", () => { "FN-101", "queued — permanent executor selection unavailable (ephemeral agents disabled)", ); - expect(store.moveTask).not.toHaveBeenCalledWith("FN-101", "in-progress", expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-101", "in-progress", expect.anything(), expect.anything()); expect(onSchedule).not.toHaveBeenCalled(); expect(ready.column).toBe("todo"); }); @@ -258,7 +312,7 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - const moveOptions = vi.mocked(store.moveTask).mock.calls[0]?.[2] as { + const moveOptions = vi.mocked(store.moveTaskIf).mock.calls[0]?.[3] as { allocateWorktree?: (reservedNames: Set) => string | null; }; expect(moveOptions.allocateWorktree?.(new Set())).toBe("/tmp/project/custom-worktrees/fn-102"); @@ -322,7 +376,7 @@ describe("Scheduler workflow cutover", () => { status: "queued", blockedBy: "FN-001", }); - expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything(), expect.anything()); expect(onBlocked).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-002" }), ["FN-001"]); }); @@ -336,7 +390,7 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything(), expect.anything()); expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null })); expect(onSchedule).not.toHaveBeenCalledWith(expect.objectContaining({ id: "FN-002" })); expect(ready.column).toBe("todo"); @@ -352,7 +406,7 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything(), expect.anything()); expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null })); expect(store.logEntry).toHaveBeenCalledWith( "FN-002", @@ -376,7 +430,7 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - expect(store.moveTask).not.toHaveBeenCalledWith("FN-200", "in-progress", expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-200", "in-progress", expect.anything(), expect.anything()); expect(store.updateTask).toHaveBeenCalledWith("FN-200", { status: "queued" }); expect(store.logEntry).toHaveBeenCalledWith( "FN-200", @@ -401,9 +455,9 @@ describe("Scheduler workflow cutover", () => { 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.moveTaskIf).toHaveBeenCalledTimes(1); + expect(store.moveTaskIf).toHaveBeenCalledWith("FN-401", "in-progress", expect.any(Function), expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-402", "in-progress", expect.anything(), expect.anything()); expect(store.logEntry).toHaveBeenCalledWith( "FN-402", expect.stringContaining("gate=maxWorktrees; maxConcurrent used=4/10"), @@ -421,9 +475,9 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - expect(store.moveTask).toHaveBeenCalledTimes(1); - expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress", expect.anything()); - expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); + expect(store.moveTaskIf).toHaveBeenCalledTimes(1); + expect(store.moveTaskIf).toHaveBeenCalledWith("FN-001", "in-progress", expect.any(Function), expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything(), expect.anything()); expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null })); expect(onSchedule).toHaveBeenCalledTimes(1); expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-001", column: "in-progress" })); @@ -434,7 +488,7 @@ describe("Scheduler workflow cutover", () => { it("leaves a task queued when the authoritative release move rejects after reservation", async () => { const ready = task({ id: "FN-002", status: "queued" }); const store = storeWith([ready], { maxConcurrent: 4, maxWorktrees: 4 }); - vi.mocked(store.moveTask).mockRejectedValueOnce( + vi.mocked(store.moveTaskIf).mockRejectedValueOnce( new TransitionRejectionError( makeTransitionRejection( "capacity-exhausted", @@ -451,7 +505,7 @@ describe("Scheduler workflow cutover", () => { await scheduler.schedule(); - expect(store.moveTask).toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); + expect(store.moveTaskIf).toHaveBeenCalledWith("FN-002", "in-progress", expect.any(Function), expect.anything()); expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null })); expect(store.logEntry).not.toHaveBeenCalledWith( "FN-002", @@ -477,7 +531,7 @@ describe("Scheduler workflow cutover", () => { semaphore.release(); } - expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything()); + expect(store.moveTaskIf).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything(), expect.anything()); expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null })); expect(onSchedule).not.toHaveBeenCalled(); expect(ready.column).toBe("todo"); diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index ed38fc3ecb..6ccd960f20 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -554,54 +554,27 @@ async function issueRelease( } } - // A concurrent sweep (or explicit promote) can win the move for this same card - // while we hold a reservation. The store serializes the move under a per-task - // lock and resolves a redundant same-column move to a silent no-op: it returns - // the card already at the target WITHOUT re-allocating a slot or emitting a - // `task:moved`. A snapshot/pre-read can't tell winner from loser (both reads - // race ahead of either commit on the per-task lock). Instead we attribute the - // transition by OBJECT IDENTITY: a real move emits `task:moved` with the very - // Task object it then returns, whereas a no-op returns a freshly-read object - // and emits nothing. So the call whose `moveTask` result IS the emitted task is - // the real mover; any other call that reserved performed a redundant no-op and - // must release the slot it grabbed (FN-1415). - const movedTaskObjects = new Set(); - let sawMovedEventForTask = false; - const onMoved = (data: { task: Task; to: string }): void => { - if (data.to === target && data.task.id === task.id) { - sawMovedEventForTask = true; - movedTaskObjects.add(data.task); - } - }; - store.on?.("task:moved", onMoved); - try { const originalColumn = task.column; - const result = await store.moveTask(task.id, target, { - moveSource: "scheduler", - allocateWorktree: - targetIsProcessing && deps.allocateWorktree - ? (reservedNames) => deps.allocateWorktree!(task, reservedNames) - : undefined, - }); /* - FNXC:WorkflowScheduling 2026-06-23-21:57: - The cutover scheduler uses hold/release in tests and older embedded stores that may not expose task:moved events. Treat a returned task that clearly moved from the original column to the target as the committed release so minimal stores do not leak reservations or falsely report a racing same-column no-op. - - FNXC:WorkflowScheduling 2026-06-23-22:39: - Eventless-release fallback is scoped to the current task. Other cards moving to the same target column during the same sweep must not disable this task's fallback and leak its reservation. - - FNXC:WorkflowScheduling 2026-06-23-22:59: - Void-returning legacy stores are ambiguous: no event plus no returned task cannot prove the current task moved. Require a returned current-task row before keeping the reservation so same-column no-ops do not leak slots. + FNXC:UserPausedDispatch 2026-07-21-21:45: + Hold release must test the source column and both pause flags under the same task lock as the move. This makes an operator pause win atomically against scheduler dispatch and also replaces event-identity inference for concurrent release attempts. */ - const returnedMovedTask = !sawMovedEventForTask - && result?.id === task.id - && result.column === target - && originalColumn !== target; - if (reservation && !movedTaskObjects.has(result) && !returnedMovedTask) { - // Same-column no-op: a racing sweep already moved this card to the target. - reservation.release(); - schedulerLog.log(`Hold release for ${task.id} skipped — already at ${target} (racing sweep won)`); + const result = await store.moveTaskIf( + task.id, + target, + (live) => live.column === originalColumn && live.paused !== true && live.userPaused !== true, + { + moveSource: "scheduler", + allocateWorktree: + targetIsProcessing && deps.allocateWorktree + ? (reservedNames) => deps.allocateWorktree!(task, reservedNames) + : undefined, + }, + ); + if (!result.moved) { + reservation?.release(); + schedulerLog.log(`Hold release for ${task.id} skipped — task became paused or left ${originalColumn}`); return false; } return true; @@ -618,8 +591,6 @@ async function issueRelease( `Hold release for ${task.id} into ${target} failed: ${error instanceof Error ? error.message : String(error)}`, ); return false; - } finally { - store.off?.("task:moved", onMoved); } } diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 28a7e9a6f8..a9b5aaa40a 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -138,9 +138,11 @@ function isIgnoredOverlapPath(path: string, ignorePath: string): boolean { function computeAutoClaimFingerprint(task: Task): string { const dependencies = [...(task.dependencies ?? [])].sort().join(","); const sortAt = task.columnMovedAt ?? task.createdAt; + /** FNXC:TaskDispatch 2026-07-19-14:40: `userPaused` is a durable operator stop even when legacy `paused` is false; candidacy caching and every dispatch selector must treat either flag as parked. */ return [ task.column, task.paused === true ? "1" : "0", + task.userPaused === true ? "1" : "0", task.assignedAgentId ?? "", task.checkedOutBy ?? "", task.deletedAt ?? "", @@ -906,7 +908,7 @@ export class Scheduler { // When a previously-paused task is unpaused in a schedulable column, // trigger a scheduling pass immediately instead of waiting for the next // poll interval (up to 15 seconds). - if (task.paused) { + if (task.paused || task.userPaused) { this.pausedTaskIds.add(task.id); } else if (this.pausedTaskIds.has(task.id)) { // Task was paused, now unpaused — trigger scheduling @@ -1524,7 +1526,7 @@ export class Scheduler { const now = Date.now(); let todo = tasks.filter((t) => { - if (t.column !== "todo" || t.paused) return false; + if (t.column !== "todo" || t.paused || t.userPaused) return false; // Skip tasks with a recovery backoff that hasn't elapsed yet if (t.nextRecoveryAt && new Date(t.nextRecoveryAt).getTime() > now) return false; // FNXC:CodingIdeasWorkflow 2026-07-04-10:45: a todo task with status "planning" is being specified in place by the triage service (merged planner/capacity column in Coding (Ideas)); it must not be dispatched until planning finishes and the status clears. @@ -1982,7 +1984,11 @@ export class Scheduler { schedulerLog.log(`Task ${task.id} no longer in "todo" (column=${freshTask?.column ?? "N/A"}) — skipping dispatch`); continue; } - if (freshTask.paused) { + /* + FNXC:UserPausedDispatch 2026-07-21-21:30: + The final fresh-read dispatch gate must honor both pause representations because an operator can set userPaused after the scheduler's initial queue snapshot but before worktree allocation. + */ + if (freshTask.paused || freshTask.userPaused) { schedulerLog.log(`Task ${task.id} is paused — skipping dispatch`); continue; } @@ -2265,21 +2271,26 @@ export class Scheduler { continue; } - schedulerLog.log(`Starting ${task.id}: ${task.title || task.id} (deps satisfied)`); - await this.store.updateTask(task.id, { - status: null, - blockedBy: null, - executionStartBranch: baseBranch ?? undefined, - effectiveNodeId: effectiveNode.nodeId ?? null, - effectiveNodeSource: effectiveNode.source, - mergeRetries: 0, - }); try { - await this.store.moveTask(task.id, "in-progress", { - moveSource: "scheduler", - allocateWorktree: (reservedNames) => - this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings), - }); + /* + FNXC:UserPausedDispatch 2026-07-21-21:45: + Scheduler dispatch predicates the todo-to-in-progress transition on both pause flags under the task lock. No awaited routing, metadata, or worktree preparation gap may let a concurrent operator pause lose to stale scheduler state. + */ + const move = await this.store.moveTaskIf( + task.id, + "in-progress", + (live) => live.column === "todo" && live.paused !== true && live.userPaused !== true, + { + moveSource: "scheduler", + allocateWorktree: (reservedNames) => + this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings), + }, + ); + if (!move.moved) { + schedulerLog.log(`Task ${task.id} became paused or left todo before dispatch — skipping`); + continue; + } + Object.assign(task, move.task); } catch (error) { if (error instanceof TransitionRejectionError && error.rejection.code === "capacity-exhausted") { await this.store.updateTask(task.id, { status: "queued" }); @@ -2289,6 +2300,15 @@ export class Scheduler { } throw error; } + schedulerLog.log(`Starting ${task.id}: ${task.title || task.id} (deps satisfied)`); + await this.store.updateTask(task.id, { + status: null, + blockedBy: null, + executionStartBranch: baseBranch ?? undefined, + effectiveNodeId: effectiveNode.nodeId ?? null, + effectiveNodeSource: effectiveNode.source, + mergeRetries: 0, + }); await this.store.updateTask(task.id, { dispatchStormCount: nextDispatchStormCount, lastDispatchAt: dispatchTimestamp,