From bb1b692596ef2cb84699c08b7cb888b93dc3d9b0 Mon Sep 17 00:00:00 2001
From: gsxdsm <gsxdsm@users.noreply.github.com>
Date: Thu, 11 Jun 2026 08:11:18 -0700
Subject: [PATCH] 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.
---
 .../workflow-work-engine-dispatch.test.ts     | 42 +++++++++++++++++++
 .../engine/src/workflow-work-processor.ts     | 32 ++++++++++++--
 2 files changed, 71 insertions(+), 3 deletions(-)

diff --git a/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts b/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts
index 3f190d28f1..efe5c005f3 100644
--- a/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts
+++ b/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts
@@ -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",
+      }),
+    ]);
+  });
 });
diff --git a/packages/engine/src/workflow-work-processor.ts b/packages/engine/src/workflow-work-processor.ts
index 4103e99bc6..2ad2f29b1b 100644
--- a/packages/engine/src/workflow-work-processor.ts
+++ b/packages/engine/src/workflow-work-processor.ts
@@ -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,