fix(FN-000): contain workflow processor runtime errors

Address PR #1581 feedback by converting runWorkItem throws into failed work-item state while returning the claimed work identity to polling callers.
This commit is contained in:
gsxdsm
2026-06-11 08:11:18 -07:00
parent c5b824137e
commit bb1b692596
2 changed files with 71 additions and 3 deletions

View File

@@ -228,4 +228,46 @@ describe("workflow work processor", () => {
expect.objectContaining({ state: "succeeded", leaseOwner: null, leaseExpiresAt: null }),
]);
});
it("marks claimed work failed when runtime dispatch throws", async () => {
const task = await store.createTask({ description: "processor failure task" });
await store.moveTask(task.id, "todo");
await store.moveTask(task.id, "in-progress");
await store.handoffToReview(task.id, {
ownerAgentId: "agent-test",
evidence: { reason: "fn_task_done", runId: "run-processor-failure", agentId: "agent-test" },
now: "2026-06-09T00:00:00.000Z",
});
const runtime = new WorkflowTaskRuntime({
store,
primitives: primitives(),
runCustomNode: async () => ({ outcome: "success" }),
});
vi.spyOn(runtime, "runWorkItem").mockRejectedValue(new Error("sqlite busy"));
const result = await processDueWorkflowWorkItem(store, runtime, { experimentalFeatures: {} } as any, {
now: "2026-06-09T00:00:00.000Z",
leaseOwner: "processor-a",
leaseDurationMs: 60_000,
kinds: workflowMergeWorkKinds(),
});
expect(result).toMatchObject({
claimed: true,
taskId: task.id,
runtime: {
disposition: "failed",
outcome: "failure",
reason: "workflow-work-item-runtime-error:sqlite busy",
},
});
expect(store.listWorkflowWorkItemsForTask(task.id, { kinds: ["merge"] })).toEqual([
expect.objectContaining({
state: "failed",
leaseOwner: null,
leaseExpiresAt: null,
lastError: "workflow-work-item-runtime-error:sqlite busy",
}),
]);
});
});

View File

@@ -1,4 +1,4 @@
import type { Settings, WorkflowWorkItemKind } from "@fusion/core";
import type { Settings, WorkflowWorkItem, WorkflowWorkItemKind, WorkflowWorkItemState } from "@fusion/core";
import { claimDueWorkflowWorkItem, type WorkflowWorkSchedulerStore } from "./workflow-work-scheduler.js";
import { WorkflowTaskRuntime, type WorkflowTaskRuntimeResult } from "./workflow-task-runtime.js";
@@ -16,8 +16,16 @@ export interface WorkflowWorkProcessorResult {
runtime?: WorkflowTaskRuntimeResult;
}
type WorkflowWorkProcessorStore = WorkflowWorkSchedulerStore & {
transitionWorkflowWorkItem?: (
id: string,
state: WorkflowWorkItemState,
patch?: { now?: string; lastError?: string | null; leaseOwner?: string | null; leaseExpiresAt?: string | null },
) => WorkflowWorkItem;
};
export async function processDueWorkflowWorkItem(
store: WorkflowWorkSchedulerStore,
store: WorkflowWorkProcessorStore,
runtime: WorkflowTaskRuntime,
settings: (Pick<Settings, "experimentalFeatures"> & Partial<Settings>) | undefined,
opts: WorkflowWorkProcessorOptions,
@@ -30,7 +38,25 @@ export async function processDueWorkflowWorkItem(
});
if (!dispatch) return { claimed: false };
const runtimeResult = await runtime.runWorkItem(dispatch.workItem, settings);
let runtimeResult: WorkflowTaskRuntimeResult;
try {
runtimeResult = await runtime.runWorkItem(dispatch.workItem, settings);
} catch (err) {
const reason = `workflow-work-item-runtime-error:${err instanceof Error ? err.message : String(err)}`;
store.transitionWorkflowWorkItem?.(dispatch.workItem.id, "failed", {
now: opts.now,
leaseOwner: null,
leaseExpiresAt: null,
lastError: reason,
});
runtimeResult = {
disposition: "failed",
outcome: "failure",
visitedNodeIds: [],
context: {},
reason,
};
}
return {
claimed: true,
workItemId: dispatch.workItem.id,