diff --git a/packages/dashboard/src/routes/register-task-workflow-routes.ts b/packages/dashboard/src/routes/register-task-workflow-routes.ts index 6e9ec29a98..822fd8967c 100644 --- a/packages/dashboard/src/routes/register-task-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-task-workflow-routes.ts @@ -1776,6 +1776,19 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork const result = await promoteHeldTask(scopedStore, req.params.id, { allocateWorktree }); if (!result.released) { + /* + FNXC:WorkflowScheduling 2026-07-21-22:31: + Pre-release Plan Review / unplanned holds are not capacity pressure. + Map them to a distinct API code so operators are not told the WIP column + is full when plan-review is still outstanding (FN-8471). + */ + if (result.rejection === "unplanned-for-execution") { + throw new ApiError(409, "Task is not ready for execution (plan review or planning still outstanding)", { + code: "unplanned-for-execution", + messageKey: "board.rejection.unplannedForExecution", + retryable: true, + }); + } if (result.rejection === "capacity-exhausted-or-no-slot") { throw new ApiError(409, "Downstream column is at capacity", { code: "capacity-exhausted", diff --git a/packages/engine/src/__tests__/workflow-continuation-selection.test.ts b/packages/engine/src/__tests__/workflow-continuation-selection.test.ts index c2ece26d1b..7f4ada85e1 100644 --- a/packages/engine/src/__tests__/workflow-continuation-selection.test.ts +++ b/packages/engine/src/__tests__/workflow-continuation-selection.test.ts @@ -1,17 +1,102 @@ import { describe, expect, it } from "vitest"; import type { Task, WorkflowWorkItem } from "@fusion/core"; -import { selectActionablePlanningContinuations } from "../runtimes/in-process-runtime.js"; +import { + isPlanningContinuationTaskDispatchable, + resolvePlanningContinuationCandidate, + selectActionablePlanningContinuations, +} from "../runtimes/in-process-runtime.js"; -function workItem(id: string, waitReason: WorkflowWorkItem["waitReason"]): WorkflowWorkItem { - return { id, waitReason } as WorkflowWorkItem; +function workItem( + id: string, + waitReason: WorkflowWorkItem["waitReason"], + patch: Partial = {}, +): WorkflowWorkItem { + return { id, taskId: `task-${id}`, waitReason, ...patch } as WorkflowWorkItem; } function task(id: string, patch: Partial = {}): Task { - return { id, paused: false, userPaused: false, ...patch } as Task; + return { id, column: "todo", paused: false, userPaused: false, ...patch } as Task; } +describe("isPlanningContinuationTaskDispatchable", () => { + it("rejects missing, paused, soft-deleted, archived, and done tasks", () => { + expect(isPlanningContinuationTaskDispatchable(undefined)).toBe(false); + expect(isPlanningContinuationTaskDispatchable(null)).toBe(false); + expect(isPlanningContinuationTaskDispatchable(task("T-1", { paused: true }))).toBe(false); + expect(isPlanningContinuationTaskDispatchable(task("T-2", { userPaused: true }))).toBe(false); + expect(isPlanningContinuationTaskDispatchable(task("T-3", { deletedAt: "2026-07-22T05:15:38.174Z" }))).toBe(false); + expect(isPlanningContinuationTaskDispatchable(task("T-4", { column: "archived" }))).toBe(false); + expect(isPlanningContinuationTaskDispatchable(task("T-5", { column: "done" }))).toBe(false); + expect(isPlanningContinuationTaskDispatchable(task("T-6", { column: "todo" }))).toBe(true); + }); +}); + +describe("resolvePlanningContinuationCandidate", () => { + it("marks lookup failures and missing tasks as orphans to cancel", () => { + const item = workItem("orphan-missing", "planning"); + expect(resolvePlanningContinuationCandidate(item, undefined, { taskLookupFailed: true })).toEqual({ + kind: "orphan", + item, + reason: "task-not-found", + }); + expect(resolvePlanningContinuationCandidate(item, null)).toEqual({ + kind: "orphan", + item, + reason: "task-not-found", + }); + }); + + it("marks terminal board tasks as orphans even when getTask returns an archive fallback", () => { + const item = workItem("orphan-terminal", "planning"); + expect( + resolvePlanningContinuationCandidate(item, task("FN-8470", { column: "archived" })), + ).toEqual({ kind: "orphan", item, reason: "task-terminal" }); + expect( + resolvePlanningContinuationCandidate(item, task("FN-8401", { column: "done" })), + ).toEqual({ kind: "orphan", item, reason: "task-terminal" }); + expect( + resolvePlanningContinuationCandidate( + item, + task("FN-soft", { deletedAt: "2026-07-22T05:15:38.174Z", column: "todo" }), + ), + ).toEqual({ kind: "orphan", item, reason: "task-terminal" }); + }); + + it("skips non-planning and paused planning items without cancelling", () => { + const capacity = workItem("cap", "capacity"); + expect(resolvePlanningContinuationCandidate(capacity, task("T-cap"))).toEqual({ + kind: "skip", + item: capacity, + reason: "not-planning", + }); + + const paused = workItem("paused", "planning"); + expect(resolvePlanningContinuationCandidate(paused, task("T-p", { paused: true }))).toEqual({ + kind: "skip", + item: paused, + reason: "paused", + }); + }); + + it("selects unpaused planning items on live non-terminal tasks", () => { + const item = workItem("eligible", "planning"); + const live = task("FN-8471", { column: "todo" }); + expect(resolvePlanningContinuationCandidate(item, live)).toEqual({ + kind: "actionable", + item, + task: live, + }); + }); +}); + describe("selectActionablePlanningContinuations", () => { - it("retains only planning items whose tasks are present and unpaused", () => { + it("retains only planning items whose tasks are present, unpaused, and non-terminal", () => { + /* + FNXC:WorkflowScheduling 2026-07-21-22:31: + Regression for the FN-8470→FN-8471 starvation class: a deleted/archived + earlier due row must not remain "actionable" and must not prevent a later + live planning continuation from being selected. + */ const selected = selectActionablePlanningContinuations([ { item: workItem("eligible", "planning"), task: task("T-1") }, { item: workItem("capacity", "capacity"), task: task("T-2") }, @@ -20,10 +105,15 @@ describe("selectActionablePlanningContinuations", () => { { item: workItem("no-wait-reason", null), task: task("T-5") }, { item: workItem("paused", "planning"), task: task("T-3", { paused: true }) }, { item: workItem("user-paused", "planning"), task: task("T-4", { userPaused: true }) }, + { item: workItem("archived", "planning"), task: task("FN-8470", { column: "archived" }) }, + { item: workItem("done", "planning"), task: task("FN-done", { column: "done" }) }, + { item: workItem("soft-deleted", "planning"), task: task("FN-soft", { deletedAt: "2026-07-22T05:15:38.174Z" }) }, + { item: workItem("later-live", "planning"), task: task("FN-8471", { column: "todo" }) }, ]); expect(selected.map(({ item, task: selectedTask }) => [item.id, selectedTask.id])).toEqual([ ["eligible", "T-1"], + ["later-live", "FN-8471"], ]); }); }); diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index c723f5a843..ed38fc3ecb 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -648,6 +648,20 @@ export async function promoteHeldTask( const target = resolveReleaseTarget(ir, task.column, true); if (!target) return { released: false, rejection: "no-release-target" }; + /* + FNXC:WorkflowScheduling 2026-07-21-22:31: + Surface pre-release Plan Review / unplanned holds distinctly from true WIP + capacity. issueRelease returns a bare false for both; without this check the + promote API mislabeled FN-8471-style plan-review waits as capacity-exhausted. + */ + const targetColumn = findColumn(ir, target); + const targetIsProcessing = targetColumn + ? resolveColumnFlags(targetColumn).countsTowardWip === true + : false; + if (targetIsProcessing && (await isUnplannedForExecution(store, task, ir))) { + return { released: false, rejection: "unplanned-for-execution", toColumn: target }; + } + const released = await issueRelease( store, { now: () => Date.now(), reserveSlot: deps.reserveSlot, allocateWorktree: deps.allocateWorktree }, diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 869f8b3648..f276209432 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -73,19 +73,77 @@ export interface PlanningContinuationCandidate { task: Task | null | undefined; } -/** FNXC:WorkflowScheduling 2026-07-21-12:30: - * Select due planning continuations whose task remains dispatchable. */ +/** + * FNXC:WorkflowScheduling 2026-07-21-22:31: + * A planning continuation is only dispatchable when its live task can still + * enter plan-review. Soft-deleted, archived, and done cards must be treated as + * non-dispatchable so their orphaned work items can be cancelled instead of + * blocking later due rows (FN-8470 tombstone starved FN-8471 plan-review). + */ +export function isPlanningContinuationTaskDispatchable( + task: Task | null | undefined, +): task is Task { + if (task == null) return false; + if (task.paused === true || task.userPaused === true) return false; + if (task.deletedAt) return false; + if (task.column === "archived" || task.column === "done") return false; + return true; +} + +/** Outcome of resolving one due work item for the planning-continuation drain. */ +export type PlanningContinuationResolution = + | { kind: "actionable"; item: WorkflowWorkItem; task: Task } + | { kind: "skip"; item: WorkflowWorkItem; reason: "not-planning" | "paused" } + | { + kind: "orphan"; + item: WorkflowWorkItem; + reason: "task-not-found" | "task-terminal"; + }; + +/** + * FNXC:WorkflowScheduling 2026-07-21-22:31: + * Classify a due work item after a per-item task load. Lookup failures and + * terminal/missing tasks become orphans (cancel); paused planning items stay + * held without cancel; non-planning due rows are skipped by this drain. + */ +export function resolvePlanningContinuationCandidate( + item: WorkflowWorkItem, + task: Task | null | undefined, + opts?: { taskLookupFailed?: boolean }, +): PlanningContinuationResolution { + if (opts?.taskLookupFailed === true || task == null) { + return { kind: "orphan", item, reason: "task-not-found" }; + } + if (task.deletedAt || task.column === "archived" || task.column === "done") { + return { kind: "orphan", item, reason: "task-terminal" }; + } + if (item.waitReason !== "planning") { + return { kind: "skip", item, reason: "not-planning" }; + } + if (task.paused === true || task.userPaused === true) { + return { kind: "skip", item, reason: "paused" }; + } + if (!isPlanningContinuationTaskDispatchable(task)) { + return { kind: "skip", item, reason: "paused" }; + } + return { kind: "actionable", item, task }; +} + +/** + * FNXC:WorkflowScheduling 2026-07-21-12:30: + * Select due planning continuations whose task remains dispatchable. + * + * FNXC:WorkflowScheduling 2026-07-21-22:31: + * Also exclude soft-deleted / archived / done tasks so archive-fallback rows + * returned by getTask cannot re-enter plan-review after the card left the board. + */ export function selectActionablePlanningContinuations( candidates: readonly PlanningContinuationCandidate[], ): Array<{ item: WorkflowWorkItem; task: Task }> { - return candidates.filter( - (candidate): candidate is { item: WorkflowWorkItem; task: Task } => - candidate.item.waitReason === "planning" - && candidate.task !== null - && candidate.task !== undefined - && candidate.task.paused !== true - && candidate.task.userPaused !== true, - ); + return candidates.flatMap((candidate) => { + const resolved = resolvePlanningContinuationCandidate(candidate.item, candidate.task); + return resolved.kind === "actionable" ? [{ item: resolved.item, task: resolved.task }] : []; + }); } export interface CliAgentAwaitingInputNotificationInfo { @@ -1893,6 +1951,12 @@ export class InProcessRuntime FNXC:WorkflowScheduling 2026-07-21-12:20: A single runtime drain owns selection at a time. Concurrent wakeups collapse behind this guard and the recurring processor supplies the next bounded pass. + + FNXC:WorkflowScheduling 2026-07-21-22:31: + Per-item task loads must not abort the pass. getTask throws for soft-deleted + rows without an archive snapshot; one orphan earlier in created_at FIFO used + to prevent every later planning continuation from dispatching (FN-8470 → FN-8471). + Cancel orphaned work items so they leave the due set and free the limit:20 window. */ if (this.workflowContinuationDrainActive || this.status !== "active") return; this.workflowContinuationDrainActive = true; @@ -1902,14 +1966,27 @@ export class InProcessRuntime states: ["runnable", "retrying"], limit: 20, }); - const candidates: PlanningContinuationCandidate[] = []; for (const item of items) { - const task = await this.taskStore.getTask(item.taskId); - candidates.push({ item, task }); - } - for (const { item, task } of selectActionablePlanningContinuations(candidates)) { - void this.executor.execute(task).catch((error) => { - runtimeLog.error(`Workflow continuation ${item.id} failed:`, error); + let task: Task | undefined; + let taskLookupFailed = false; + try { + task = await this.taskStore.getTask(item.taskId); + } catch (error) { + taskLookupFailed = true; + runtimeLog.warn( + `Workflow continuation ${item.id}: getTask(${item.taskId}) failed — treating as orphan: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + const resolved = resolvePlanningContinuationCandidate(item, task, { taskLookupFailed }); + if (resolved.kind === "orphan") { + await this.cancelOrphanedWorkflowWorkItem(resolved.item, resolved.reason); + continue; + } + if (resolved.kind !== "actionable") continue; + void this.executor.execute(resolved.task).catch((error) => { + runtimeLog.error(`Workflow continuation ${resolved.item.id} failed:`, error); }); } } finally { @@ -1917,6 +1994,36 @@ export class InProcessRuntime } } + /** + * FNXC:WorkflowScheduling 2026-07-21-22:31: + * Terminalize a due work item whose task can no longer host graph work so the + * FIFO due poll no longer revisits it every 2s. Fail-soft: a transition race + * must not abort the rest of the drain. + */ + private async cancelOrphanedWorkflowWorkItem( + item: WorkflowWorkItem, + reason: "task-not-found" | "task-terminal", + ): Promise { + if (typeof this.taskStore.transitionWorkflowWorkItem !== "function") return; + try { + await this.taskStore.transitionWorkflowWorkItem(item.id, "cancelled", { + leaseOwner: null, + leaseExpiresAt: null, + lastError: `orphaned-continuation:${reason}`, + blockedReason: reason, + }); + runtimeLog.log( + `Cancelled orphaned workflow work item ${item.id} (task=${item.taskId}, node=${item.nodeId}, reason=${reason})`, + ); + } catch (error) { + runtimeLog.warn( + `Failed to cancel orphaned workflow work item ${item.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + /** * Execute a heartbeat run for an agent. *