Backport #3675: fix(core): make step-argument serialization failures catchable in workflow code

.changeset/catchable-step-arg-serialization-errors.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Step-argument serialization failures now fail the step with a catchable error (via a `step_failed` event, like a step-body failure) instead of failing the run from outside the workflow, and when uncaught they fail the run immediately as a `USER_ERROR` rather than retrying until max queue deliveries.

docs/content/docs/errors/serialization-failed.mdx

@@ -25,6 +25,34 @@ This error can appear when:
 - Serializing step arguments
 - Serializing step return values
 
+## Where the Error Surfaces
+
+Where you observe the failure depends on which boundary it crosses:
+
+- **Workflow arguments** — `start()` throws synchronously in your application code.
+- **Step arguments** — the *step* fails, exactly like a step whose body threw a `FatalError`: no retries (the failure is deterministic), and a `try/catch` around the step call in your workflow code observes it. The step's recorded input shows `[input unavailable: step argument serialization failed]`, since the real arguments are precisely what refused to serialize.
+- **Workflow return values** — the workflow body has already returned, so nothing can catch it; the run fails.
+
+```typescript lineNumbers
+async function stepWithBadArguments(value: unknown) {
+  "use step";
+  return value;
+}
+
+export async function processWorkflow(someValue: unknown) {
+  "use workflow";
+
+  try {
+    await stepWithBadArguments(someValue);
+  } catch (err) {
+    // (err as Error).message starts with
+    // "Failed to serialize step arguments"
+  }
+}
+```
+
+Uncaught, the error propagates out of the workflow body and the run fails immediately with the error code `USER_ERROR` — it does not retry.
+
 ## Why This Happens
 
 Workflows persist their state using an event log. Every value that crosses execution boundaries must be:

docs/content/docs/foundations/errors-and-retries.mdx

@@ -139,6 +139,31 @@ callApi.maxRetries = 5; // Retry up to 5 times on failure (6 total attempts)
   step can run up to 4 times total (1 initial attempt + 3 retries).
 </Callout>
 
+## Serialization Failures
+
+A step whose arguments cannot be [serialized](/docs/foundations/serialization) fails like a step whose body threw a `FatalError`: the failure is deterministic, so it skips the retry loop, and a `try/catch` around the step call observes it:
+
+```typescript lineNumbers
+async function someStep(input: unknown) {
+  "use step";
+  return input;
+}
+
+export async function myWorkflow(input: unknown) {
+  "use workflow";
+
+  try {
+    await someStep(input);
+  } catch (err) {
+    if ((err as Error).message.startsWith("Failed to serialize step arguments")) {
+      // e.g. `Failed to serialize step arguments at path "..."`
+    }
+  }
+}
+```
+
+Uncaught, the run fails immediately with the `USER_ERROR` code — without retrying. See [serialization-failed](/docs/errors/serialization-failed) for common causes and fixes.
+
 ## Error Codes
 
 When a workflow run fails, the error may include a `code` that classifies the failure. You can access it programmatically via the `Run` class:

packages/core/e2e/e2e.test.ts

@@ -1,3 +1,3 @@
 import fs from 'node:fs';
 import path from 'node:path';
 import { setTimeout as sleep } from 'node:timers/promises';
@@ -1472,6 +1472,68 @@
       );
     });
 
+    describe('serialization failures', () => {
+      test(
+        'step-argument serialization failure is catchable in workflow code',
+        { timeout: 60_000 },
+        async () => {
+          // Passing an unserializable value (a class instance with no serde
+          // model) to a step must fail THAT STEP — step_created +
+          // step_failed — not the whole run, so a try/catch around the step
+          // call observes the serialization error.
+          const run = await start(
+            await e2e('serializationErrorStepArgsCaught'),
+            []
+          );
+          const result = await run.returnValue;
+
+          expect(result.caught).toBe(true);
+          expect(result.messageIncludesStepArguments).toBe(true);
+
+          // The workflow completed (the error was caught) …
+          const { json: runData } = await cliInspectJson(`runs ${run.runId}`);
+          expect(runData.status).toBe('completed');
+
+          // … and the step itself is recorded as failed.
+          const { json: steps } = await cliInspectJson(
+            `steps --runId ${run.runId}`
+          );
+          const step = steps.find((s: any) =>
+            s.stepName.includes('acceptAnyValue')
+          );
+          expect(step.status).toBe('failed');
+        }
+      );
+
+      test(
+        'uncaught step-argument serialization failure fails the run as USER_ERROR without redelivery retries',
+        { timeout: 60_000 },
+        async () => {
+          // Regression coverage for the production failure mode where a
+          // step-argument serialization error caused the run to redeliver
+          // until "exceeded max deliveries (49/48)". The run must fail
+          // promptly (well within this test's timeout — 48 redeliveries
+          // with backoff would take many minutes) and classify as
+          // USER_ERROR, not MAX_DELIVERIES_EXCEEDED.
+          const run = await start(
+            await e2e('serializationErrorStepArgsUncaught'),
+            []
+          );
+          const error = await run.returnValue.catch((e: unknown) => e);
+
+          expect(WorkflowRunFailedError.is(error)).toBe(true);
+          assert(WorkflowRunFailedError.is(error));
+          expect(error.cause.code).toBe('USER_ERROR');
+          expect(error.cause.message).toContain(
+            'Failed to serialize step arguments'
+          );
+
+          const { json: runData } = await cliInspectJson(`runs ${run.runId}`);
+          expect(runData.status).toBe('failed');
+        }
+      );
+    });
+
     describe('not registered', () => {
       test(
         'WorkflowNotRegisteredError fails the run when workflow does not exist',

packages/core/src/runtime/step-handler.test.ts

@@ -109,6 +109,9 @@ vi.mock('../serialization.js', () => ({
   dehydrateStepReturnValue: vi
     .fn()
     .mockResolvedValue(new Uint8Array([1, 2, 3])),
+  formatSerializationError: vi.fn(
+    (context: string) => `Failed to serialize ${context}.`
+  ),
 }));
 
 // Mock context storage
@@ -138,6 +141,7 @@ vi.mock('../private.js', () => ({
 // which populates capturedHandlerRef
 import './step-handler.js';
 import { getStepFunction } from '../private.js';
+import { hydrateStepArguments } from '../serialization.js';
 import {
   getErrorName,
   getErrorStack,
@@ -148,6 +152,10 @@ import {
   resetPortCacheForTesting,
   setPortResolverForTesting,
 } from './get-port-lazy.js';
+import {
+  UNSERIALIZABLE_STEP_INPUT_MARKER,
+  unserializableStepInputPlaceholder,
+} from './unserializable-step.js';
 import { getWorld } from './world.js';
 
 const mockPortResolver = vi.fn(async () => 3000);
@@ -729,3 +737,94 @@ describe('step-handler step not found', () => {
     expect(mockQueueMessage).not.toHaveBeenCalled();
   });
 });
+
+describe('step-handler unserializable-argument placeholder recovery', () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.mocked(getStepFunction).mockReturnValue(mockStepFn);
+    vi.mocked(normalizeUnknownError).mockImplementation(
+      async (err: unknown) => ({
+        message: err instanceof Error ? err.message : String(err),
+        name: err instanceof Error ? err.name : 'Error',
+        stack: err instanceof Error ? err.stack : undefined,
+      })
+    );
+    vi.mocked(getErrorName).mockReturnValue('FatalError');
+    vi.mocked(getErrorStack).mockReturnValue('');
+    mockStepFn.mockReset().mockResolvedValue('step-result');
+    mockStepFn.maxRetries = 3;
+    mockQueueMessage.mockResolvedValue(undefined);
+    vi.mocked(getWorld).mockReturnValue({
+      events: { create: mockEventsCreate },
+      queue: mockQueue,
+      getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined),
+    } as any);
+    mockEventsCreate.mockReset().mockResolvedValue({
+      step: {
+        stepId: 'step_abc',
+        status: 'running',
+        attempt: 1,
+        startedAt: new Date(),
+        input: [],
+      },
+      event: {},
+    });
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.mocked(hydrateStepArguments).mockResolvedValue({
+      args: [],
+      thisVal: null,
+      closureVars: undefined,
+    } as any);
+  });
+
+  it('fails a placeholder-input step without running the step body', async () => {
+    // A crash between finalization's two durable writes leaves a lone
+    // step_created carrying the placeholder; crash recovery dispatches it
+    // here. The body must not run — the intended failure completes instead.
+    vi.mocked(hydrateStepArguments).mockResolvedValue(
+      unserializableStepInputPlaceholder() as any
+    );
+
+    const result = await capturedHandler(
+      createMessage(),
+      createMetadata('acceptAnyValue')
+    );
+
+    expect(result).toBeUndefined();
+    expect(mockStepFn).not.toHaveBeenCalled();
+    expect(mockEventsCreate).toHaveBeenCalledWith(
+      'wrun_test123',
+      expect.objectContaining({
+        eventType: 'step_failed',
+        eventData: expect.objectContaining({
+          error: expect.stringContaining('Failed to serialize step arguments'),
+        }),
+      }),
+      expect.anything()
+    );
+    // Fatal: no step_retrying, so no queue attempts are burned.
+    expect(mockEventsCreate).not.toHaveBeenCalledWith(
+      'wrun_test123',
+      expect.objectContaining({ eventType: 'step_retrying' }),
+      expect.anything()
+    );
+  });
+
+  it('does not trip on a genuine argument equal to the display marker', async () => {
+    // The structural flag — not the human-readable marker string — is what
+    // identifies the placeholder, so a step legitimately called with the
+    // marker text still runs.
+    vi.mocked(hydrateStepArguments).mockResolvedValue({
+      args: [UNSERIALIZABLE_STEP_INPUT_MARKER],
+      thisVal: null,
+      closureVars: undefined,
+    } as any);
+
+    await capturedHandler(createMessage(), createMetadata('acceptAnyValue'));
+
+    expect(mockStepFn).toHaveBeenCalledWith(UNSERIALIZABLE_STEP_INPUT_MARKER);
+  });
+});

packages/core/src/runtime/step-handler.ts

@@ -22,6 +22,7 @@ import { runtimeLogger, stepLogger } from '../logger.js';
 import { getStepFunction } from '../private.js';
 import {
   dehydrateStepReturnValue,
+  formatSerializationError,
   hydrateStepArguments,
 } from '../serialization.js';
 import { contextStorage } from '../step/context-storage.js';
@@ -49,6 +50,7 @@ import {
   queueMessage,
   withHealthCheck,
 } from './helpers.js';
+import { isUnserializableStepInputPlaceholder } from './unserializable-step.js';
 import { safeWaitUntil } from './wait-until.js';
 import { getWorld, getWorldHandlers } from './world.js';
 
@@ -533,6 +535,21 @@ const stepHandler = createQueueHandler(
 
           const executionStartTime = Date.now();
           try {
+            // Finalization of an unserializable-argument step writes
+            // step_created (placeholder input) and step_failed as two
+            // separate durable writes (see finalizeUnserializableStep in
+            // suspension-handler.ts). A crash or transient failure between
+            // them leaves this step pending with the placeholder stored as
+            // its input, and normal crash recovery then dispatches it here.
+            // NEVER run user code with placeholder arguments — complete the
+            // intended failure instead. FatalError skips the retry loop
+            // below and writes step_failed, exactly what the interrupted
+            // finalization was about to do.
+            if (isUnserializableStepInputPlaceholder(hydratedInput)) {
+              throw new FatalError(
+                formatSerializationError('step arguments', undefined)
+              );
+            }
             result = await trace('step.execute', {}, async () => {
               return await contextStorage.run(
                 {