diff --git a/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts b/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts index bb066772e3..60745f234e 100644 --- a/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts +++ b/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts @@ -110,6 +110,27 @@ describe("fast mode workflow/runtime invariants", () => { ); }); + it("falls back to the runner task when prepareWorktree cannot trust the live row", async () => { + const store = createMockStore(); + store.getTask.mockResolvedValue({ ...task({ id: "FN-OTHER", worktree: "/tmp/wrong" }) }); + const executor = new TaskExecutor(store, "/tmp/test"); + + const result = await (executor as any) + .createAuthoritativeWorkflowPrimitives({ experimentalFeatures: { workflowGraphExecutor: true } }) + .prepareWorktree( + { run: { taskId: "FN-6226" }, node: { node: { id: "execute" }, context: {} } }, + task({ id: "FN-6226", worktree: "/tmp/right", branch: "fusion/fn-6226" }), + ); + + expect(result).toMatchObject({ + outcome: "success", + data: { + worktreePath: "/tmp/right", + branchName: "fusion/fn-6226", + }, + }); + }); + it("graph executor with builtin:coding selection skips the workflow-step seam in fast mode", async () => { const { executor } = makeExecutorForTask(task({ executionMode: "fast", worktree: "/tmp/wt" })); const runWorkflowSteps = vi.spyOn(executor as any, "runWorkflowSteps").mockResolvedValue(workflowResult()); diff --git a/packages/engine/src/__tests__/hold-release.test.ts b/packages/engine/src/__tests__/hold-release.test.ts index 5b43d40789..f813f77b25 100644 --- a/packages/engine/src/__tests__/hold-release.test.ts +++ b/packages/engine/src/__tests__/hold-release.test.ts @@ -19,7 +19,7 @@ import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { execSync } from "node:child_process"; -import { TaskStore, type WorkflowIr } from "@fusion/core"; +import { TaskStore, type Task, type WorkflowIr } from "@fusion/core"; import { runHoldReleaseSweep, promoteHeldTask, @@ -104,6 +104,48 @@ describe("hold-release sweep (U6)", () => { expect((await store.getTask(id))?.column).toBe("in-progress"); }); + it("does not let unrelated moved events disable the current task's eventless release fallback", async () => { + const held = { + id: "FN-777", + title: "Held", + description: "", + column: "todo", + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + } as Task; + let onMoved: ((data: { task: Task; to: string }) => void) | undefined; + const release = vi.fn(); + const fakeStore = { + getSettings: vi.fn(async () => ({ + maxConcurrent: 4, + experimentalFeatures: { workflowColumns: true }, + })), + listTasks: vi.fn(async () => [held]), + moveTask: vi.fn(async () => { + onMoved?.({ task: { ...held, id: "FN-OTHER" }, to: "in-progress" }); + held.column = "in-progress"; + return held; + }), + getTaskWorkflowSelection: vi.fn(() => null), + on: vi.fn((_event: string, listener: (data: { task: Task; to: string }) => void) => { + onMoved = listener; + }), + off: vi.fn(), + } as unknown as TaskStore; + + const result = await runHoldReleaseSweep(fakeStore, { + now: () => Date.now(), + reserveSlot: () => ({ release }), + }); + + expect(result.released).toEqual(["FN-777"]); + expect(release).not.toHaveBeenCalled(); + }); + it("two holds, one slot: exactly one releases; the other releases next sweep after the slot frees", async () => { await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); const a = await seedTodoCard(); diff --git a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts index eb50700ba5..6db863864f 100644 --- a/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts +++ b/packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts @@ -96,6 +96,86 @@ describe("Scheduler workflow cutover", () => { expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-100", column: "in-progress" })); }); + it("queues without dispatch when ephemeral agents are disabled and no agent store is available", async () => { + const ready = task({ id: "FN-101" }); + const store = storeWith([ready], { ephemeralAgentsEnabled: false }); + const onSchedule = vi.fn(); + const scheduler = new Scheduler(store, { onSchedule }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + expect(store.updateTask).toHaveBeenCalledWith("FN-101", { status: "queued" }); + expect(store.logEntry).toHaveBeenCalledWith( + "FN-101", + "queued — permanent executor selection unavailable (ephemeral agents disabled)", + ); + expect(store.moveTask).not.toHaveBeenCalledWith("FN-101", "in-progress", expect.anything()); + expect(onSchedule).not.toHaveBeenCalled(); + expect(ready.column).toBe("todo"); + }); + + it("passes worktree naming and directory settings to the workflow release allocator", async () => { + const ready = task({ id: "FN-102" }); + const store = storeWith([ready], { + worktreeNaming: "task-id", + worktreesDir: "custom-worktrees", + }); + const scheduler = new Scheduler(store, { onSchedule: vi.fn() }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + const moveOptions = vi.mocked(store.moveTask).mock.calls[0]?.[2] as { + allocateWorktree?: (reservedNames: Set) => string | null; + }; + expect(moveOptions.allocateWorktree?.(new Set())).toBe("/tmp/project/custom-worktrees/fn-102"); + }); + + it("continues executor handoff for all released tasks when post-release metadata or logs fail", async () => { + const first = task({ id: "FN-201", status: "queued" }); + const second = task({ id: "FN-202", status: "queued" }); + const store = storeWith([first, second], { maxConcurrent: 4, maxWorktrees: 4 }); + const updateImpl = vi.mocked(store.updateTask).getMockImplementation()!; + vi.mocked(store.updateTask).mockImplementation(async (id, patch) => { + if (id === "FN-201" && "lastDispatchAt" in patch) { + throw new Error("metadata write failed"); + } + return updateImpl(id, patch); + }); + vi.mocked(store.logEntry).mockImplementation(async (id, message) => { + if (id === "FN-201" && message.startsWith("Node routing resolved")) { + throw new Error("log write failed"); + } + }); + const onSchedule = vi.fn(); + const scheduler = new Scheduler(store, { onSchedule }); + (scheduler as unknown as { running: boolean }).running = true; + + await scheduler.schedule(); + + expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ + id: "FN-201", + column: "in-progress", + status: undefined, + effectiveNodeSource: "local", + })); + expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ + id: "FN-202", + column: "in-progress", + status: undefined, + effectiveNodeSource: "local", + })); + expect(store.updateTask).toHaveBeenCalledWith("FN-202", expect.objectContaining({ + status: null, + effectiveNodeSource: "local", + })); + expect(store.logEntry).toHaveBeenCalledWith( + "FN-202", + "Node routing resolved: local (source: local)", + ); + }); + it("keeps dependency-blocked todo tasks queued on the workflow sweep path", async () => { const blocker = task({ id: "FN-001", column: "todo" }); const dependent = task({ id: "FN-002", dependencies: ["FN-001"] }); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index af1cf625e4..9a50ecd15f 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -5155,14 +5155,18 @@ export class TaskExecutor { return { prepareWorktree: async (_ctx, task) => { - const live = await this.store.getTask(task.id); + const live = await this.store.getTask(task.id).catch(() => null); + const liveTask = live?.id === task.id ? live : null; /* FNXC:WorkflowExecution 2026-06-23-11:49: The workflow execute node must not perform a second worktree acquisition ahead of the authoritative executor. Passing the repo root as a prepared worktree makes the inner execute() reject a valid fresh-worktree task as repo-root reuse; pass only an existing task worktree and let execute() acquire when none exists. + + FNXC:WorkflowExecution 2026-06-23-22:31: + Upgrade safety requires the graph primitive to tolerate older or minimal stores that return null or a mismatched row during startup/cutover. Only trust the live row when it is for the requested task; otherwise fall back to the runner snapshot. */ const prepared: PreparedWorktree = { - worktreePath: live.worktree || task.worktree || "", - branchName: live.branch || task.branch, + worktreePath: liveTask?.worktree || task.worktree || "", + branchName: liveTask?.branch || task.branch, }; return { outcome: "success", value: "worktree-ready", data: prepared }; }, diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index 076213cb7e..637166fb06 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -429,8 +429,12 @@ async function issueRelease( // 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(); - const onMoved = (data: { task: object; to: string }): void => { - if (data.to === target) movedTaskObjects.add(data.task); + 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); @@ -446,8 +450,11 @@ async function issueRelease( /* 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. */ - const returnedMovedTask = movedTaskObjects.size === 0 + const returnedMovedTask = !sawMovedEventForTask && ( result === undefined || (result.id === task.id && result.column === target && originalColumn !== target) diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 58adebea39..aa67f43071 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -2376,7 +2376,23 @@ export class Scheduler { } } - if (latestSettings.ephemeralAgentsEnabled === false && !freshTask.assignedAgentId && this.options.agentStore) { + if (latestSettings.ephemeralAgentsEnabled === false && !freshTask.assignedAgentId) { + /* + FNXC:WorkflowScheduling 2026-06-23-22:33: + The workflow cutover path must not silently dispatch unassigned work when ephemeral agents are disabled. Queue until permanent-agent selection is available so upgrades preserve the executor contract instead of falling through to local execution. + */ + if (!this.options.agentStore) { + await this.store.updateTask(task.id, { status: "queued" }); + if (!this.wasPermanentAgentUnavailable.has(task.id)) { + await this.logDispatchQueuedReason( + task.id, + "queued — permanent executor selection unavailable (ephemeral agents disabled)", + ); + this.wasPermanentAgentUnavailable.add(task.id); + } + return null; + } + const selectedAgent = await selectPermanentAgentForTask({ task: freshTask, agentStore: this.options.agentStore, @@ -2564,23 +2580,21 @@ export class Scheduler { }; }, allocateWorktree: (task, reservedNames) => - planTaskWorktreePath(task, this.store.getRootDir(), undefined, reservedNames, {}), + this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings), }); for (const taskId of result.released) { const prep = dispatchPrepByTaskId.get(taskId); if (!prep) continue; /* FNXC:WorkflowScheduling 2026-06-23-21:49: - A workflow hold release is not a committed dispatch until moveTask succeeds and appears in result.released. Only then may the scheduler emit "Starting" and clear queued state. Call onSchedule before best-effort metadata/log writes so a post-release store/log failure does not strand an in-progress task without executor handoff. + A workflow hold release is not a committed dispatch until moveTask succeeds and appears in result.released. Only then may the scheduler emit "Starting" and clear queued state. + + FNXC:WorkflowScheduling 2026-06-23-22:36: + Persist dispatch metadata before executor handoff when possible, but isolate update/log failures per task. A metadata failure must not block later released tasks or strand an already released task without onSchedule handoff. */ schedulerLog.log(`Starting ${taskId}: ${prep.task.title || taskId} (deps satisfied)`); const latest = await this.store.getTask(taskId).catch(() => null); - try { - this.options.onSchedule?.(latest ?? prep.task); - } catch (error) { - schedulerLog.error(`onSchedule failed for ${taskId}:`, error); - } - await this.store.updateTask(taskId, { + const dispatchUpdate = { status: null, blockedBy: null, executionStartBranch: prep.baseBranch ?? undefined, @@ -2589,13 +2603,36 @@ export class Scheduler { mergeRetries: 0, dispatchStormCount: prep.dispatchStormCount, lastDispatchAt: prep.dispatchTimestamp, - }); + }; + const scheduledTask = { + ...(latest?.id === taskId ? latest : prep.task), + ...dispatchUpdate, + status: undefined, + blockedBy: undefined, + effectiveNodeId: prep.effectiveNodeId ?? undefined, + effectiveNodeSource: prep.effectiveNodeSource as Task["effectiveNodeSource"], + column: "in-progress" as const, + }; + try { + await this.store.updateTask(taskId, dispatchUpdate); + } catch (error) { + schedulerLog.error(`Post-release dispatch metadata update failed for ${taskId}:`, error); + } + try { + this.options.onSchedule?.(scheduledTask); + } catch (error) { + schedulerLog.error(`onSchedule failed for ${taskId}:`, error); + } this.recentEngineTodoRequeues.delete(taskId); this.wasNodeBlocked.delete(taskId); this.wasNodeDispatchValidationBlocked.delete(taskId); this.wasPermanentAgentUnavailable.delete(taskId); this.clearDispatchQueuedReasonMemo(taskId); - await this.store.logEntry(taskId, `Node routing resolved: ${prep.effectiveNodeId ?? "local"} (source: ${prep.effectiveNodeSource})`); + try { + await this.store.logEntry(taskId, `Node routing resolved: ${prep.effectiveNodeId ?? "local"} (source: ${prep.effectiveNodeSource})`); + } catch (error) { + schedulerLog.error(`Post-release dispatch log failed for ${taskId}:`, error); + } } } catch (error) { schedulerLog.error("Hold/release sweep failed:", error);