From 30e2fd7b09480e9e2f321322be7b0eb770dd4ffb Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 4 Jun 2026 02:50:54 -0700 Subject: [PATCH] fix(review): apply autofix feedback --- .../migration-workflow-columns.test.ts | 32 ++++++++++++++ packages/core/src/index.ts | 3 +- packages/core/src/store.ts | 19 ++++---- packages/core/src/workflow-capacity.ts | 5 +++ packages/core/src/workflow-reconciliation.ts | 23 +++++++++- packages/dashboard/app/api/legacy.ts | 14 +++++- packages/dashboard/app/components/Board.tsx | 42 +++++++++++++----- .../app/components/TaskDetailModal.tsx | 13 ++++++ .../app/components/WorkflowColumnPanel.tsx | 12 ++++- .../app/components/WorkflowNodeEditor.tsx | 20 ++++++--- .../app/components/WorkflowResultsTab.tsx | 12 ++++- .../dashboard/src/routes/board-workflows.ts | 24 ++++++---- .../src/routes/register-workflow-routes.ts | 7 ++- packages/dashboard/vitest.config.ts | 5 ++- packages/engine/src/hold-release.ts | 44 ++++++++++++++----- packages/engine/src/merge-trait.ts | 3 +- packages/engine/src/scheduler.ts | 4 +- .../engine/src/workflow-graph-branches.ts | 27 ++++++++++-- 18 files changed, 248 insertions(+), 61 deletions(-) diff --git a/packages/core/src/__tests__/migration-workflow-columns.test.ts b/packages/core/src/__tests__/migration-workflow-columns.test.ts index 1035705577..6cedf60893 100644 --- a/packages/core/src/__tests__/migration-workflow-columns.test.ts +++ b/packages/core/src/__tests__/migration-workflow-columns.test.ts @@ -206,6 +206,38 @@ describe("U12 rollback safety — flag OFF after flag ON keeps legacy behavior", expect(caught).toBeInstanceOf(Error); expect((caught as Error).message).toMatch(/Invalid transition/); }); + + it("a card stranded in a custom column when the flag is toggled OFF degrades to a clean Invalid-transition error (no TypeError) and listTasks stays healthy", async () => { + // Flag ON: select a custom workflow whose entry column is custom, so the + // card is re-homed into a column that VALID_TRANSITIONS never keys. + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } }); + const wf = await store.createWorkflowDefinition({ + name: "stranded", + ir: customIr("stranded", ["intake", "build", "ship"], "intake"), + }); + const task = await store.createTask({ description: "stranded card" }); + await store.selectTaskWorkflowAndReconcile(task.id, wf.id); + expect((await store.getTask(task.id)).column).toBe("intake"); + + // Toggle the flag OFF — the card stays in the custom "intake" column. + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: false } }); + expect((await store.getTask(task.id)).column).toBe("intake"); + + // listTasks must not throw with a task sitting in an unknown column. + await expect(store.listTasks()).resolves.toBeDefined(); + + // A move attempt degrades to the legacy "Invalid transition" error rather + // than a TypeError on the undefined VALID_TRANSITIONS lookup. + let caught: unknown; + try { + await store.moveTask(task.id, "in-progress", { moveSource: "user" }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).toMatch(/Invalid transition/); + expect((caught as Error)).not.toBeInstanceOf(TypeError); + }); }); describe("Residual B: getBranchProgressByTask reads workflow_run_branches", () => { diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 5d540b150b..3a5a88962d 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -145,11 +145,12 @@ export { } from "./plugin-gate-verdict.js"; export type { PluginGateVerdict, ColumnPluginGate } from "./plugin-gate-verdict.js"; // ── U6: workflow capacity (WIP) resolution shared by store + sweep ─────────── -export { resolveColumnCapacity } from "./workflow-capacity.js"; +export { resolveColumnCapacity, DEFAULT_WORKFLOW_POOL_ID } from "./workflow-capacity.js"; export type { ColumnCapacity } from "./workflow-capacity.js"; // ── U5: workflow lifecycle reconciliation (switch / edit / delete) ─────────── export { OccupiedColumnsError, + InvalidRehomeTargetError, resolveEntryColumnId, resolveSwitchReconciliation, computeRemovedOccupiedColumns, diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index cdb73291bf..d519411fe6 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -16,7 +16,7 @@ import { resolveColumnPluginGates, } from "./plugin-gate-verdict.js"; import { getTraitRegistry, assertColumnTraitsValid } from "./trait-registry.js"; -import { resolveColumnCapacity } from "./workflow-capacity.js"; +import { resolveColumnCapacity, DEFAULT_WORKFLOW_POOL_ID } from "./workflow-capacity.js"; import { OccupiedColumnsError, assertRehomeTargetValid, @@ -1150,8 +1150,10 @@ export class TaskStore extends EventEmitter { private static readonly ACTIVE_TASKS_WHERE = '"deletedAt" IS NULL'; /** U6: sentinel effective-workflow id for default-workflow (null-selection) * tasks, so they all share one per-column capacity pool (KTD-10). It is not a - * real workflow row id (no `builtin:`/custom collision possible). */ - private static readonly DEFAULT_WORKFLOW_POOL_ID = "__default-workflow__"; + * real workflow row id (no `builtin:`/custom collision possible). Re-exposed + * as a static member for internal call sites; the canonical const lives in + * `workflow-capacity.ts` (`DEFAULT_WORKFLOW_POOL_ID`). */ + private static readonly DEFAULT_WORKFLOW_POOL_ID = DEFAULT_WORKFLOW_POOL_ID; static async getOrCreateForProject( projectId?: string, @@ -5050,12 +5052,6 @@ export class TaskStore extends EventEmitter { // Group by task; for each task keep only the branches of its most-recent // run (the runId of the row with the latest updatedAt). - const latestRunByTask = new Map(); - for (const row of rows) { - const known = latestRunByTask.get(row.taskId); - if (!known) latestRunByTask.set(row.taskId, row.runId); - } - // Re-derive the latest runId precisely from the max-updatedAt row. const maxByTask = new Map(); for (const row of rows) { const cur = maxByTask.get(row.taskId); @@ -5933,7 +5929,10 @@ export class TaskStore extends EventEmitter { } } else { // ── Flag-OFF legacy path (unchanged) ─────────────────────────────────── - const validTargets = VALID_TRANSITIONS[task.column]; + // A task can sit in a custom column when the flag was toggled ON→OFF; + // `VALID_TRANSITIONS` only keys the legacy columns, so a missing entry + // degrades to the legacy "Invalid transition" error instead of a TypeError. + const validTargets = VALID_TRANSITIONS[task.column as Column] ?? []; if (!validTargets.includes(toColumn)) { throw new Error( `Invalid transition: '${task.column}' → '${toColumn}'. ` + diff --git a/packages/core/src/workflow-capacity.ts b/packages/core/src/workflow-capacity.ts index 9ef0c40851..e34ae46eab 100644 --- a/packages/core/src/workflow-capacity.ts +++ b/packages/core/src/workflow-capacity.ts @@ -29,6 +29,11 @@ import { getTraitRegistry } from "./trait-registry.js"; * `settings.maxConcurrent` (the legacy "N agents in-progress" gate). */ const DEFAULT_WIP_COLUMN_ID = "in-progress"; +/** U6 (KTD-10): sentinel effective-workflow id for default-workflow + * (null-selection) tasks, so they all share one per-column capacity pool. It + * is not a real workflow row id (no `builtin:`/custom collision possible). */ +export const DEFAULT_WORKFLOW_POOL_ID = "__default-workflow__"; + /** Resolved capacity configuration for a single column. */ export interface ColumnCapacity { /** True when the column carries a capacity (`wip`/`countsTowardWip`) trait. */ diff --git a/packages/core/src/workflow-reconciliation.ts b/packages/core/src/workflow-reconciliation.ts index 5c0a176371..382d3645fa 100644 --- a/packages/core/src/workflow-reconciliation.ts +++ b/packages/core/src/workflow-reconciliation.ts @@ -148,6 +148,25 @@ export function computeRemovedOccupiedColumns( return removed; } +/** + * Thrown when a supplied `rehomeTo` names a column that does not exist in the + * post-edit workflow. Distinct from {@link OccupiedColumnsError} (which signals + * a conflict needing a re-home target) — this is a bad-request input error and + * the dashboard maps it to a 400, not a 409. + */ +export class InvalidRehomeTargetError extends Error { + readonly workflowId: string; + readonly rehomeTo: string; + constructor(workflowId: string, rehomeTo: string) { + super( + `Workflow '${workflowId}' has no column '${rehomeTo}' to re-home occupants into.`, + ); + this.name = "InvalidRehomeTargetError"; + this.workflowId = workflowId; + this.rehomeTo = rehomeTo; + } +} + /** * Validate that `rehomeTo` (when supplied for an edit that removes occupied * columns) names a column that survives in `nextIr`. Throws when it does not, so @@ -155,9 +174,9 @@ export function computeRemovedOccupiedColumns( */ export function assertRehomeTargetValid(nextIr: WorkflowIr, rehomeTo: string): void { if (!workflowHasColumn(nextIr, rehomeTo)) { - throw new OccupiedColumnsError( + throw new InvalidRehomeTargetError( (nextIr as WorkflowIrV2).name ?? "(unknown)", - [], + rehomeTo, ); } } diff --git a/packages/dashboard/app/api/legacy.ts b/packages/dashboard/app/api/legacy.ts index a0a0b477f4..719bfaa214 100644 --- a/packages/dashboard/app/api/legacy.ts +++ b/packages/dashboard/app/api/legacy.ts @@ -5075,8 +5075,18 @@ export function selectTaskWorkflow( taskId: string, workflowId: string | null, projectId?: string, -): Promise<{ workflowId: string | null; enabledWorkflowSteps: string[] }> { - return api<{ workflowId: string | null; enabledWorkflowSteps: string[] }>( +): Promise<{ + workflowId: string | null; + enabledWorkflowSteps: string[]; + // U5 (R20): present (flag ON) when the switch re-homed the card; `preserved` + // false means the card moved columns and the board needs a refresh. + reconciliation?: { preserved: boolean; fromColumn: string; toColumn: string }; +}> { + return api<{ + workflowId: string | null; + enabledWorkflowSteps: string[]; + reconciliation?: { preserved: boolean; fromColumn: string; toColumn: string }; + }>( withProjectId(`/tasks/${encodeURIComponent(taskId)}/workflow`, projectId), { method: "PUT", diff --git a/packages/dashboard/app/components/Board.tsx b/packages/dashboard/app/components/Board.tsx index d2fe8ca8fc..eb1aa81c2f 100644 --- a/packages/dashboard/app/components/Board.tsx +++ b/packages/dashboard/app/components/Board.tsx @@ -282,19 +282,39 @@ export function Board({ tasks, projectId, maxConcurrent, onMoveTask, onPauseTask return new Set(); }); + // Fetch board workflow lanes for the project. Deliberately NOT keyed on + // `tasks` — that refetched on every SSE tick. Instead we refetch on project + // change and when the tab regains visibility/focus. A stale-response guard + // (monotonic sequence ref) drops out-of-order responses. + // TODO: replace the visibility/focus staleness stopgap with a + // `workflow:updated` SSE event when one exists. + const boardWorkflowsFetchSeqRef = useRef(0); useEffect(() => { - let cancelled = false; - fetchBoardWorkflows(projectId) - .then((payload) => { - if (!cancelled) setBoardWorkflows(payload); - }) - .catch(() => { - if (!cancelled) setBoardWorkflows({ flagEnabled: false, defaultWorkflowId: "builtin:coding", workflows: [], taskWorkflowIds: {} }); - }); - return () => { - cancelled = true; + const runFetch = () => { + const seq = ++boardWorkflowsFetchSeqRef.current; + fetchBoardWorkflows(projectId) + .then((payload) => { + if (seq === boardWorkflowsFetchSeqRef.current) setBoardWorkflows(payload); + }) + .catch(() => { + if (seq === boardWorkflowsFetchSeqRef.current) { + setBoardWorkflows({ flagEnabled: false, defaultWorkflowId: "builtin:coding", workflows: [], taskWorkflowIds: {} }); + } + }); }; - }, [projectId, tasks]); + runFetch(); + const onVisible = () => { + if (typeof document === "undefined" || document.visibilityState === "visible") runFetch(); + }; + if (typeof document !== "undefined") document.addEventListener("visibilitychange", onVisible); + if (typeof window !== "undefined") window.addEventListener("focus", onVisible); + return () => { + // Advance the seq so any in-flight response is dropped on cleanup. + boardWorkflowsFetchSeqRef.current++; + if (typeof document !== "undefined") document.removeEventListener("visibilitychange", onVisible); + if (typeof window !== "undefined") window.removeEventListener("focus", onVisible); + }; + }, [projectId]); const handleToggleLaneCollapse = useCallback((workflowId: string) => { setCollapsedLanes((prev) => { diff --git a/packages/dashboard/app/components/TaskDetailModal.tsx b/packages/dashboard/app/components/TaskDetailModal.tsx index f92b153f67..bdc56eb52e 100644 --- a/packages/dashboard/app/components/TaskDetailModal.tsx +++ b/packages/dashboard/app/components/TaskDetailModal.tsx @@ -1971,6 +1971,18 @@ export function TaskDetailContent({ } }, [task.id, projectId, workflowEnabledSteps, onTaskUpdated, addToast]); + // U5 (R20): a workflow switch re-homed the card to a new column. Refetch the + // task and push it up so the board reflects the move before the SSE catch-up. + const handleWorkflowReconciled = useCallback(async () => { + try { + const detail = await fetchTaskDetail(task.id, projectId); + setFullDetail(detail); + onTaskUpdated?.(detail); + } catch { + // Best-effort refresh; the SSE stream will catch the board up regardless. + } + }, [task.id, projectId, onTaskUpdated]); + const loadAgents = useCallback(async () => { setAgentsLoading(true); try { @@ -2761,6 +2773,7 @@ export function TaskDetailContent({ && task.status !== "awaiting-cli-approval" } onWorkflowStepsChange={handleWorkflowStepsChange} + onWorkflowReconciled={handleWorkflowReconciled} taskStatus={task.status} taskPausedReason={task.pausedReason} /> diff --git a/packages/dashboard/app/components/WorkflowColumnPanel.tsx b/packages/dashboard/app/components/WorkflowColumnPanel.tsx index 64a7ac6115..10c3b3ca1a 100644 --- a/packages/dashboard/app/components/WorkflowColumnPanel.tsx +++ b/packages/dashboard/app/components/WorkflowColumnPanel.tsx @@ -36,9 +36,17 @@ export function WorkflowColumnPanel({ const [catalog, setCatalog] = useState([]); useEffect(() => { + let cancelled = false; fetchTraits(projectId) - .then(setCatalog) - .catch((err) => addToast(getErrorMessage(err) || t("workflowColumns.traitsLoadFailed", "Failed to load traits"), "error")); + .then((catalog) => { + if (!cancelled) setCatalog(catalog); + }) + .catch((err) => { + if (!cancelled) addToast(getErrorMessage(err) || t("workflowColumns.traitsLoadFailed", "Failed to load traits"), "error"); + }); + return () => { + cancelled = true; + }; }, [projectId, addToast, t]); const workflowWide = violations.filter((v) => v.columnId === null); diff --git a/packages/dashboard/app/components/WorkflowNodeEditor.tsx b/packages/dashboard/app/components/WorkflowNodeEditor.tsx index f440817993..38b08b9f4a 100644 --- a/packages/dashboard/app/components/WorkflowNodeEditor.tsx +++ b/packages/dashboard/app/components/WorkflowNodeEditor.tsx @@ -42,7 +42,7 @@ import { emptyWorkflowLayout, columnsOf, columnsToBandNodes, - columnForY, + strictColumnForY, validateColumnsClient, unplacedNodeIds, isColumnBandNode, @@ -120,9 +120,17 @@ function InnerEditor({ // Trait catalog (for client-side composition validation; the panel fetches its // own copy for the picker, but the editor needs the flags to validate). useEffect(() => { - fetchTraits(projectId).then(setTraitCatalog).catch(() => { - // Non-fatal: validation degrades to server-side parse on save. - }); + let cancelled = false; + fetchTraits(projectId) + .then((catalog) => { + if (!cancelled) setTraitCatalog(catalog); + }) + .catch(() => { + // Non-fatal: validation degrades to server-side parse on save. + }); + return () => { + cancelled = true; + }; }, [projectId]); // Composition violations (client mirror of validateColumnTraits). @@ -194,7 +202,9 @@ function InnerEditor({ const onNodeDragStop = useCallback( (_evt: unknown, node: FlowNode) => { if (isColumnBandNode(node.id) || columns.length === 0) return; - const column = columnForY(node.position.y, columns); + // strictColumnForY (not the clamping columnForY): a node dragged above or + // below all bands keeps no column rather than snapping to the nearest one. + const column = strictColumnForY(node.position.y, columns); if (!column) return; setNodes((ns) => ns.map((n) => (n.id === node.id ? { ...n, data: { ...n.data, column } } : n)), diff --git a/packages/dashboard/app/components/WorkflowResultsTab.tsx b/packages/dashboard/app/components/WorkflowResultsTab.tsx index 73a2cf4168..6ebe34bff8 100644 --- a/packages/dashboard/app/components/WorkflowResultsTab.tsx +++ b/packages/dashboard/app/components/WorkflowResultsTab.tsx @@ -46,6 +46,10 @@ interface WorkflowResultsTabProps { onWorkflowStepsChange?: (steps: string[]) => void; taskStatus?: string; taskPausedReason?: string; + /** U5 (R20): called after a workflow switch re-homed the card to a new column + * (reconciliation present and not preserved) so the board can refresh before + * the SSE catch-up arrives. */ + onWorkflowReconciled?: () => void; } /** Extract the user-facing question from a workflow-input paused reason. @@ -227,6 +231,7 @@ export function WorkflowResultsTab({ onWorkflowStepsChange, taskStatus, taskPausedReason, + onWorkflowReconciled, }: WorkflowResultsTabProps) { const { t } = useTranslation("app"); const [expandedOutputs, setExpandedOutputs] = useState>({}); @@ -270,8 +275,13 @@ export function WorkflowResultsTab({ const res = await selectTaskWorkflow(taskId, workflowId, projectId); setSelectedWorkflowId(res.workflowId); onWorkflowStepsChange?.(res.enabledWorkflowSteps); + // U5 (R20): the switch re-homed the card to a new column — refresh the + // board now rather than waiting for the SSE catch-up. + if (res.reconciliation && !res.reconciliation.preserved) { + onWorkflowReconciled?.(); + } }, - [taskId, projectId, onWorkflowStepsChange], + [taskId, projectId, onWorkflowStepsChange, onWorkflowReconciled], ); // Check if any result has pending status diff --git a/packages/dashboard/src/routes/board-workflows.ts b/packages/dashboard/src/routes/board-workflows.ts index 7536ebe46b..7ab6a4b637 100644 --- a/packages/dashboard/src/routes/board-workflows.ts +++ b/packages/dashboard/src/routes/board-workflows.ts @@ -95,19 +95,25 @@ async function describeWorkflow( store: Pick, workflowId: string, ): Promise { - const ir = await resolveWorkflowIr(store, workflowId); // The display name comes from the persisted definition when available, // otherwise the IR's own name (default workflow). - let name = ir.name; if (isBuiltinWorkflowId(workflowId)) { - name = getBuiltinWorkflow(workflowId)?.name ?? name; - } else { - try { - const def = await store.getWorkflowDefinition(workflowId); - if (def?.name) name = def.name; - } catch { - // fall through to IR name + const ir = await resolveWorkflowIr(store, workflowId); + const name = getBuiltinWorkflow(workflowId)?.name ?? ir.name; + return { id: workflowId, name, columns: describeColumns(ir) }; + } + // Custom workflow: fetch the definition once and derive both IR and name from + // it (previously getWorkflowDefinition was called twice per workflow). + let ir: WorkflowIr = BUILTIN_CODING_WORKFLOW_IR; + let name = ir.name; + try { + const def = await store.getWorkflowDefinition(workflowId); + if (def) { + ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; + name = def.name || ir.name; } + } catch { + // fall through to the default IR/name } return { id: workflowId, name, columns: describeColumns(ir) }; } diff --git a/packages/dashboard/src/routes/register-workflow-routes.ts b/packages/dashboard/src/routes/register-workflow-routes.ts index 29a3dbce67..9c5ece2de8 100644 --- a/packages/dashboard/src/routes/register-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-workflow-routes.ts @@ -1,5 +1,5 @@ import type { WorkflowIr } from "@fusion/core"; -import { ColumnTraitValidationError, OccupiedColumnsError, WorkflowCompileError, WorkflowIrError, compileWorkflowToSteps, listTraits } from "@fusion/core"; +import { ColumnTraitValidationError, OccupiedColumnsError, InvalidRehomeTargetError, WorkflowCompileError, WorkflowIrError, compileWorkflowToSteps, listTraits } from "@fusion/core"; import { ApiError, badRequest, conflict, notFound } from "../api-error.js"; import type { ApiRoutesContext } from "./types.js"; @@ -122,6 +122,11 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { if (err instanceof OccupiedColumnsError) { throw conflict(err.message, { workflowId: err.workflowId, occupancies: err.occupancies }); } + // A supplied rehomeTo naming a non-existent column is a bad request (400), + // not a 409 conflict. + if (err instanceof InvalidRehomeTargetError) { + throw badRequest(err.message, { workflowId: err.workflowId, rehomeTo: err.rehomeTo }); + } if (err instanceof WorkflowIrError) throw badRequest(err.message); if (err instanceof ColumnTraitValidationError) { throw badRequest(err.message, { violations: err.violations }); diff --git a/packages/dashboard/vitest.config.ts b/packages/dashboard/vitest.config.ts index 90d5bba151..e81863ba0c 100644 --- a/packages/dashboard/vitest.config.ts +++ b/packages/dashboard/vitest.config.ts @@ -121,6 +121,7 @@ const qualityAppComponentTests = [ "GitHubBadge", "GroupTaskModal", "InlineCreateCard", + "Lane", "LoginInstructions", "MemoryView", "MergeAdvanceNotice", @@ -174,6 +175,8 @@ const qualityAppComponentTests = [ "TrackingRepoSelect", "WorkflowNodeEditor", "WorkflowResultsTab", + "WorkflowSelector", + "workflow-flow-mapping", "WorktrunkInstallApprovalDetails", ] as const; @@ -187,7 +190,7 @@ const batchedQualityAppComponentTestsB = batchedQualityAppComponentTests.slice(b function buildComponentQualityInclude(testNames: readonly string[]): string[] { return testNames.length > 0 - ? [`app/components/__tests__/{${testNames.join(",")}}.test.tsx`] + ? [`app/components/__tests__/{${testNames.join(",")}}.test.{ts,tsx}`] : []; } diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index 9fddb34227..307eeaebc1 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -41,6 +41,7 @@ import { resolveColumnCapacity, resolveColumnFlags, resolveColumnAdjacency, + DEFAULT_WORKFLOW_POOL_ID, TransitionRejectionError, BUILTIN_CODING_WORKFLOW_IR, getBuiltinWorkflow, @@ -54,8 +55,6 @@ import { } from "@fusion/core"; import { schedulerLog } from "./logger.js"; -const DEFAULT_WORKFLOW_POOL_ID = "__default-workflow__"; - /** A reservation handle returned by {@link HoldReleaseDeps.reserveSlot}. The * sweep calls `release()` if the subsequent move rejects on capacity. */ export interface SlotReservation { @@ -95,7 +94,13 @@ export interface HoldReleaseResult { // ── Workflow IR resolution (read-only, mirrors store + merge-trait) ─────────── -async function resolveTaskWorkflowIr(store: TaskStore, taskId: string): Promise { +async function resolveTaskWorkflowIr( + store: TaskStore, + taskId: string, + // Optional per-sweep cache keyed by workflowId so each distinct workflow's IR + // is resolved (and its definition fetched) at most once per sweep. + irCache?: Map, +): Promise { let workflowId: string | undefined; try { workflowId = store.getTaskWorkflowSelection(taskId)?.workflowId; @@ -103,14 +108,20 @@ async function resolveTaskWorkflowIr(store: TaskStore, taskId: string): Promise< workflowId = undefined; } if (!workflowId) return BUILTIN_CODING_WORKFLOW_IR; + const cached = irCache?.get(workflowId); + if (cached) return cached; if (isBuiltinWorkflowId(workflowId)) { const builtin = getBuiltinWorkflow(workflowId); - return builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR; + const ir = builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR; + irCache?.set(workflowId, ir); + return ir; } try { const def = await store.getWorkflowDefinition(workflowId); if (!def) return BUILTIN_CODING_WORKFLOW_IR; - return typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; + const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; + irCache?.set(workflowId, ir); + return ir; } catch { return BUILTIN_CODING_WORKFLOW_IR; } @@ -287,15 +298,17 @@ function resolveTimerDeadline(holdConfig: Record, task: Task): * arbitration is still the in-txn check, which rejects a losing racer. */ function countCapacitySlot( - store: TaskStore, allTasks: Task[], + // Pre-built taskId → effective workflowId map (one pass per sweep) so this + // counting loop avoids a per-task `effectiveWorkflowId` DB call. + effectiveWorkflowIdByTask: Map, targetColumn: string, workflowId: string, countPending: boolean, ): number { let count = 0; for (const t of allTasks) { - if (effectiveWorkflowId(store, t.id) !== workflowId) continue; + if ((effectiveWorkflowIdByTask.get(t.id) ?? DEFAULT_WORKFLOW_POOL_ID) !== workflowId) continue; if (t.column === targetColumn) { count += 1; continue; @@ -325,6 +338,17 @@ export async function runHoldReleaseSweep( const allTasks = await store.listTasks({ includeArchived: false }); + // Per-sweep caches. `allTasks` is a snapshot-stable read within a sweep, so we + // resolve each workflow's IR at most once (irCache) and pre-build the + // taskId → effective-workflowId map a single time rather than per-task DB + // calls inside the capacity counting loop. The authoritative in-txn capacity + // check is unaffected — this only trims the sweep pre-check cost. + const irCache = new Map(); + const effectiveWorkflowIdByTask = new Map(); + for (const t of allTasks) { + effectiveWorkflowIdByTask.set(t.id, effectiveWorkflowId(store, t.id)); + } + for (const task of allTasks) { // Skip paused / recovery-backoff tasks exactly as the legacy scheduler does. if (task.paused || task.userPaused) { @@ -334,7 +358,7 @@ export async function runHoldReleaseSweep( continue; } - const ir = await resolveTaskWorkflowIr(store, task.id); + const ir = await resolveTaskWorkflowIr(store, task.id, irCache); if (!isHeldTask(ir, task)) continue; const column = findColumn(ir, task.column); @@ -372,8 +396,8 @@ export async function runHoldReleaseSweep( } const capacity = resolveColumnCapacity(ir, target, settings); if (capacity.hasCapacity && Number.isFinite(capacity.limit)) { - const workflowId = effectiveWorkflowId(store, task.id); - const occupants = countCapacitySlot(store, allTasks, target, workflowId, capacity.countPending); + const workflowId = effectiveWorkflowIdByTask.get(task.id) ?? DEFAULT_WORKFLOW_POOL_ID; + const occupants = countCapacitySlot(allTasks, effectiveWorkflowIdByTask, target, workflowId, capacity.countPending); if (occupants >= capacity.limit) { result.held.push({ taskId: task.id, reason: "downstream-full" }); continue; diff --git a/packages/engine/src/merge-trait.ts b/packages/engine/src/merge-trait.ts index 5c320dd319..5b2a8cec5c 100644 --- a/packages/engine/src/merge-trait.ts +++ b/packages/engine/src/merge-trait.ts @@ -51,6 +51,7 @@ import { type WorkflowIr, type WorkflowIrColumn, } from "@fusion/core"; +import { mergerLog } from "./logger.js"; // ── Resolved merge policy ──────────────────────────────────────────────────── @@ -242,7 +243,7 @@ async function mergeOnEnter(store: TaskStore, task: Pick; } +/** + * Await a `saveBranchState` call inside a guard so a Promise-returning impl + * cannot escape as an unhandled rejection, and so a persistence failure never + * kills branch execution (log-and-continue). For a synchronous impl this + * preserves the prior behavior (the write completes before the caller proceeds). + */ +async function persistBranchState( + persistence: WorkflowBranchPersistence | undefined, + state: WorkflowBranchRunState, +): Promise { + try { + await persistence?.saveBranchState?.(state); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + schedulerLog.warn( + `saveBranchState failed for task ${state.taskId} run ${state.runId} branch ${state.branchId}: ${message}`, + ); + } +} + /** Minimal semaphore shape — structurally compatible with AgentSemaphore. */ export interface WorkflowBranchSemaphore { run(fn: () => Promise): Promise; @@ -251,7 +272,7 @@ async function walkBranch( } else { const exec = async (): Promise => env.runBranchNode(node, signal); lastResult = env.semaphore ? await env.semaphore.run(exec) : await exec(); - env.persistence?.saveBranchState?.({ + await persistBranchState(env.persistence, { taskId: env.task.id, runId: env.runId, branchId: startNodeId, @@ -266,7 +287,7 @@ async function walkBranch( } if (lastResult.outcome === "failure") { - env.persistence?.saveBranchState?.({ + await persistBranchState(env.persistence, { taskId: env.task.id, runId: env.runId, branchId: startNodeId, @@ -282,7 +303,7 @@ async function walkBranch( return { outcome: lastResult.outcome, lastNodeId: currentId }; } if (next === joinId) { - env.persistence?.saveBranchState?.({ + await persistBranchState(env.persistence, { taskId: env.task.id, runId: env.runId, branchId: startNodeId,