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

* fix(core): make step-argument serialization failures catchable in workflow code

A step whose arguments fail to serialize is now finalized by the
suspension handler as step_created + step_failed (mirroring a step-body
failure) instead of rejecting the whole suspension. The next replay —
forced in-process, since no step message is dispatched for the failed
step — rejects the step's promise with the SerializationError, so a
try/catch around the step call observes it. Uncaught, the error
propagates out of the workflow body and fails the run as a fatal
USER_ERROR immediately, instead of redelivering the orchestrator
message until max deliveries (49/48) as reported in production on v4.

* Serialize the step_failed error with the VM global; one-sentence changeset

Addresses review feedback: dehydrateStepError in
finalizeUnserializableStep now receives suspension.globalThis like every
other dehydration in this file. Error detection is realm-independent, so
the host-created SerializationError serializes identically, but VM-realm
values guest code threw into the cause chain are now detected by the
realm-sensitive reducers.

* Address review: QuickJS engine support, deferred-batch join, drain gate, placeholder marker, telemetry, docs

- QuickJS: dumpPendingOps now catches a step input's serialization
  failure per-op, reframes it as a SerializationError with the same
  framed message as dehydrateStepArguments, and surfaces it on the
  pending op instead of failing the whole collection. The entrypoint's
  dispatchPendingOps finalizes such steps as step_created (placeholder
  input) + step_failed, excludes them from inline claims and queue
  publishes, marks them handled, and raises the requeue signal so the
  failure is observed even when the feed lags — mirroring the node:vm
  engine, so both engines agree: catchable in workflow code, USER_ERROR
  with the framed message when uncaught. Both step-argument e2e tests
  now pass on WORKFLOW_VM=quickjs.
- runtime.ts: the failed-step replay path now joins
  suspensionResult.deferredBatchWork before continuing, so a trailing
  chunk commit or step-message publish rejection propagates instead of
  being swallowed after ack; committed inline claims are documented as
  deliberately handed to owned recovery.
- Terminal drain: finalization is gated on a stepDispatch target. The
  drain caller has no replay to observe a finalization, so a completed
  run no longer gains failed-step rows for an unawaited unserializable
  step — the rethrown error is swallowed by the drain's catch,
  preserving its pre-existing behavior.
- The placeholder input now carries a marker string ('[input
  unavailable: step argument serialization failed]', shared via
  runtime/unserializable-step.ts) so inspect/o11y don't render the
  failed step as a genuine zero-argument call.
- New workflow.steps.failed_serialization span attribute on the
  suspension span, so occurrence is measurable without log search.
- Docs: v5 serialization-failed error page documents where each
  boundary's failure surfaces (catchable step failure vs run failure)
  and the no-retry USER_ERROR semantics; foundations/errors-and-retries
  gains a Serialization Failures section with the try/catch shape.

* Guard the finalization crash window; self-contained docs samples

- A crash or transient failure between finalization's two durable
  writes leaves a lone placeholder step_created, and redelivery then
  dispatches the step through normal crash recovery — previously
  running user code with the placeholder arguments. The placeholder
  now carries a structural flag on the input triple's top level (which
  user code never controls, so no false positives), and the step
  executor checks it after hydration: instead of running the body, it
  throws the intended fatal SerializationError, completing the
  interrupted finalization as step_failed. Applies to both engines
  (they share the placeholder and the executor).
- Regression tests: executor fails a placeholder-input step without
  running the body (and doesn't trip on a genuine argument equal to
  the display marker); handleSuspension rejects for redelivery when
  step_failed can't be written after step_created landed, leaving the
  recoverable placeholder behind; mixed bad-step + large fan-out
  returns the failure set alongside still-pending deferredBatchWork
  whose rejection surfaces — the contract the runtime's failed-step
  join (added previously) relies on.
- Docs: the two new code samples are now self-contained so the docs
  code-sample typecheck passes.

.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 `SerializationError` (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/v5/errors/serialization-failed.mdx

@@ -29,6 +29,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 and step return values** — the *step* fails with the `SerializationError`, exactly like a step whose body threw a fatal error: 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]` when the arguments were the unserializable part.
+- **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.name === "SerializationError"
+    // "Failed to serialize step arguments at path ..."
+  }
+}
+```
+
+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/v5/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 or return value 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 the `SerializationError`:
+
+```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).name === "SerializationError") {
+      // 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 includes an `errorCode` that classifies the failure, alongside the original thrown value (preserved as `cause`):

packages/core/e2e/e2e.test.ts

@@ -1815,6 +1815,120 @@ describe.concurrent('e2e', () => {
       );
     });
 
+    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 SerializationError.
+          const run = await start(
+            await e2e('serializationErrorStepArgsCaught'),
+            []
+          );
+          const result = await run.returnValue;
+
+          expect(result.caught).toBe(true);
+          expect(result.name).toBe('SerializationError');
+          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 steps = await cliInspectJsonUntil(
+            `steps --runId ${run.runId}`,
+            (json) =>
+              json.some(
+                (s: any) =>
+                  s.stepName.includes('acceptAnyValue') && s.status === 'failed'
+              )
+          );
+          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.errorCode).toBe('USER_ERROR');
+          expect(String(error.message)).toContain(
+            'Failed to serialize step arguments'
+          );
+
+          const { json: runData } = await cliInspectJson(`runs ${run.runId}`);
+          expect(runData.status).toBe('failed');
+          expect(runData.errorCode).toBe('USER_ERROR');
+        }
+      );
+
+      test(
+        'step-return-value serialization failure is catchable in workflow code',
+        { timeout: 60_000 },
+        async () => {
+          // The step executor treats a return-value SerializationError as
+          // fatal (skipping the retry loop) and writes step_failed, so the
+          // workflow's try/catch observes it.
+          const run = await start(
+            await e2e('serializationErrorStepReturnCaught'),
+            []
+          );
+          const result = await run.returnValue;
+
+          expect(result.caught).toBe(true);
+          expect(result.name).toBe('SerializationError');
+          expect(result.messageIncludesReturnValue).toBe(true);
+
+          const { json: runData } = await cliInspectJson(`runs ${run.runId}`);
+          expect(runData.status).toBe('completed');
+        }
+      );
+
+      test(
+        'uncaught step-return-value serialization failure fails the run as USER_ERROR',
+        { timeout: 60_000 },
+        async () => {
+          const run = await start(
+            await e2e('serializationErrorStepReturnUncaught'),
+            []
+          );
+          const error = await run.returnValue.catch((e: unknown) => e);
+
+          expect(WorkflowRunFailedError.is(error)).toBe(true);
+          assert(WorkflowRunFailedError.is(error));
+          expect(error.errorCode).toBe('USER_ERROR');
+          expect(String(error.message)).toContain(
+            'Failed to serialize step return value'
+          );
+
+          const { json: runData } = await cliInspectJson(`runs ${run.runId}`);
+          expect(runData.status).toBe('failed');
+          expect(runData.errorCode).toBe('USER_ERROR');
+        }
+      );
+    });
+
     describe('not registered', () => {
       // JS-only: the workflowId is hand-built in the JS scheme, so on another
       // language it names nothing rather than naming something missing.

packages/core/src/runtime.ts

@@ -3329,6 +3329,54 @@ export function workflowEntrypoint(
                           continue;
                         }
 
+                        // Steps whose arguments failed to serialize were
+                        // finalized by the suspension handler as step_created
+                        // + step_failed (see finalizeUnserializableStep). No
+                        // step-execution message is dispatched for them, so
+                        // when such a step is the only pending work nothing
+                        // would ever re-invoke the run — replay in-process
+                        // over the reloaded log instead. The replay rejects
+                        // the step's promise with the SerializationError,
+                        // which a try/catch around the step call observes;
+                        // uncaught, it propagates out of the workflow body
+                        // and fails the run as a USER_ERROR. Healthy sibling
+                        // steps are not dispatched this pass: the replay
+                        // re-suspends over their already-committed
+                        // step_created events and the next pass dispatches
+                        // them as usual.
+                        if (
+                          suspensionResult.failedStepCorrelationIds.size > 0
+                        ) {
+                          // Join the batched fan-out's trailing chunk commits
+                          // and step-message publishes before continuing,
+                          // exactly like the two joins on the dispatch paths
+                          // below: this invocation must not proceed (and
+                          // eventually ack) before every create and publish
+                          // it launched is durable. A rejection propagates
+                          // like theirs — transient world errors rethrow to
+                          // the queue for redelivery.
+                          await suspensionResult.deferredBatchWork;
+                          // Inline steps whose pair-folded step_started this
+                          // pass already committed (`inlineClaims`) are
+                          // deliberately NOT executed on this pass: the
+                          // forced replay below re-suspends over the same
+                          // pending steps, and owned recovery (the claims
+                          // carry this message's ownerMessageId) re-executes
+                          // them there. Its "previous delivery crashed
+                          // mid-body" log is a misnomer on this path —
+                          // nothing crashed, the bodies were never started.
+                          //
+                          // The failed dehydration may have executed
+                          // workflow-owned code (getters/proxies) before
+                          // throwing; demote to a cold replay rather than
+                          // resume a VM that may have diverged. This is a
+                          // rare terminal-error path, so the replay cost is
+                          // irrelevant next to the divergence risk.
+                          retainedSession = null;
+                          eventLog = nextEventLogLoad(eventLog);
+                          continue;
+                        }
+
                         const pendingSteps = suspensionResult.pendingSteps;
 
                         // Inline execution is gated on ownership. The

packages/core/src/runtime/quickjs-entrypoint.ts

@@ -28,6 +28,7 @@ import {
   type RunInput,
   SPEC_VERSION_CURRENT,
   SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT,
+  SPEC_VERSION_SUPPORTS_COMPRESSION,
   type WorkflowRun,
 } from '@workflow/world';
 import { classifyRunError, isRetryableWorldError } from '../classify-error.js';
@@ -39,6 +40,8 @@ import {
 } from '../serialization/encryption.js';
 import {
   dehydrateRunError,
+  dehydrateStepArguments,
+  dehydrateStepError,
   hydrateRunError,
   maybeEncrypt,
 } from '../serialization.js';
@@ -70,6 +73,7 @@ import {
 import { ReplayBudget } from './replay-budget.js';
 import { executeStep, type StepExecutionResult } from './step-executor.js';
 import { runStepSingleFlight } from './step-single-flight.js';
+import { unserializableStepInputPlaceholder } from './unserializable-step.js';
 import { getWaitContinuationDispatch } from './wait-continuation.js';
 import { getWorld } from './world.js';
 
@@ -244,12 +248,30 @@ async function dispatchPendingOps(params: {
    * queues.
    */
   nextTraceCarrier: () => Promise<Record<string, string>>;
+  /**
+   * When true (the inline loop), a step carrying `serializationError` is
+   * finalized as step_created (placeholder input) + step_failed so the
+   * live-VM feed rejects the step's promise and workflow code can catch
+   * it — mirroring the node:vm engine's finalizeUnserializableStep. When
+   * false (the terminal drain), such steps are skipped entirely: the run
+   * is already completing/failing, no replay follows the drain to observe
+   * the failure, and a completed run carrying a failed step would read as
+   * a bug from the dashboard — matching the node:vm drain's behavior.
+   */
+  finalizeUnserializableSteps?: boolean;
   wfdiag: (checkpoint: string, fields: Record<string, unknown>) => void;
 }): Promise<{
   createdAttributeEvent: boolean;
   createdGetConflictHook: boolean;
   /** Step cids already published via resilient dispatch — see above. */
   queuedStepCids: Set<string>;
+  /**
+   * Step cids finalized as failed because their input refused to
+   * serialize (see `finalizeUnserializableSteps`). No execution message
+   * exists for these; the caller must ensure the run observes the
+   * terminal event (the inline loop's feed, or the requeue signal).
+   */
+  failedSerializationStepCids: Set<string>;
 }> {
   const {
     world,
@@ -267,6 +289,9 @@ async function dispatchPendingOps(params: {
   // parallel, message carrying `stepInput`). Reported to the caller so it
   // skips them in its own queueing pass.
   const queuedStepCids = new Set<string>();
+  // Step cids finalized as step_created + step_failed because their input
+  // refused to serialize — see the `finalizeUnserializableSteps` param.
+  const failedSerializationStepCids = new Set<string>();
   // Resilient step dispatch eligibility, shared by every step op below (the
   // per-step input-size check is applied inside the op): feature enabled and
   // a binary-safe (CBOR) queue transport for the run.
@@ -498,6 +523,86 @@ async function dispatchPendingOps(params: {
       const step = op as PendingStep;
       opsPromises.push(
         (async () => {
+          // The step's input refused to serialize while dumping the VM's
+          // pending ops (see PendingStep.serializationError). Finalize it
+          // as step_created (placeholder input — the world requires the
+          // step entity before a terminal event) + step_failed carrying
+          // the SerializationError, so the live-VM feed rejects the
+          // step's promise and workflow code can catch it. Never queue an
+          // execution message for it. Mirrors the node:vm engine's
+          // finalizeUnserializableStep. In the terminal drain
+          // (finalizeUnserializableSteps unset), skip entirely — see the
+          // param docs.
+          if (step.serializationError) {
+            if (!params.finalizeUnserializableSteps) {
+              return;
+            }
+            runtimeLogger.warn(
+              'Step arguments failed to serialize; failing the step so ' +
+                'the workflow can observe the error',
+              {
+                workflowRunId: runId,
+                correlationId: step.correlationId,
+                stepName: step.stepId,
+                error: step.serializationError.message,
+              }
+            );
+            try {
+              await world.events.create(runId, {
+                eventType: 'step_created',
+                specVersion: SPEC_VERSION_CURRENT,
+                correlationId: step.correlationId,
+                eventData: {
+                  stepName: step.stepId,
+                  input: (await dehydrateStepArguments(
+                    unserializableStepInputPlaceholder(),
+                    runId,
+                    encryptionKey,
+                    globalThis,
+                    false,
+                    (workflowRun.specVersion ?? 0) >=
+                      SPEC_VERSION_SUPPORTS_COMPRESSION
+                  )) as Uint8Array,
+                },
+              });
+            } catch (err) {
+              // Concurrent invocation hit the same deterministic failure
+              // and created it first, or the run already finished.
+              if (RunExpiredError.is(err)) return;
+              if (!EntityConflictError.is(err)) throw err;
+            }
+            try {
+              await world.events.create(runId, {
+                eventType: 'step_failed',
+                specVersion: SPEC_VERSION_CURRENT,
+                correlationId: step.correlationId,
+                eventData: {
+                  stepName: step.stepId,
+                  error: await dehydrateStepError(
+                    step.serializationError,
+                    runId,
+                    encryptionKey,
+                    [],
+                    globalThis,
+                    (workflowRun.specVersion ?? 0) >=
+                      SPEC_VERSION_SUPPORTS_COMPRESSION
+                  ),
+                },
+              });
+            } catch (err) {
+              // Step already terminal or run already finished.
+              if (!EntityConflictError.is(err) && !RunExpiredError.is(err)) {
+                throw err;
+              }
+            }
+            failedSerializationStepCids.add(step.correlationId);
+            wfdiag('step_serialization_failed', {
+              stepId: step.stepId,
+              correlationId: step.correlationId,
+            });
+            return;
+          }
+
           // Create step_created event. `step.input` is the
           // format-prefixed devalue bytes ("devl" + devalue) produced
           // by `globalThis[Symbol.for('workflow-serialize')]({args,
@@ -667,7 +772,12 @@ async function dispatchPendingOps(params: {
   // Per-op dispatch runs in parallel.
   await Promise.all(opsPromises);
 
-  return { createdAttributeEvent, createdGetConflictHook, queuedStepCids };
+  return {
+    createdAttributeEvent,
+    createdGetConflictHook,
+    queuedStepCids,
+    failedSerializationStepCids,
+  };
 }
 
 /**
@@ -1146,8 +1256,17 @@ export async function runWorkflowWithQuickJS(params: {
           !executedStepIds.has(op.correlationId) &&
           !queuedStepIds.has(op.correlationId)
       );
+      // Steps whose input refused to serialize (see
+      // PendingStep.serializationError) never execute: they must not be
+      // inline-claimed (a lazy step_started would need the very input that
+      // failed) nor queued. Dispatch below finalizes them as step_created
+      // + step_failed instead; only healthy steps compete for inline
+      // slots and overflow.
+      const healthySteps = freshSteps.filter(
+        (step) => !step.serializationError
+      );
       const inlineCandidates =
-        maxInlineSteps <= 0 ? [] : freshSteps.slice(0, maxInlineSteps);
+        maxInlineSteps <= 0 ? [] : healthySteps.slice(0, maxInlineSteps);
       const inlineClaimCids = new Set(
         inlineCandidates.map((step) => step.correlationId)
       );
@@ -1178,7 +1297,7 @@ export async function runWorkflowWithQuickJS(params: {
       // hasCreatedEvent and would never be queued at all (the wedge behind
       // promiseRaceStressTestWorkflow hanging in the quickjs CI legs). The
       // step-identity-scoped idempotency key makes repeats harmless.
-      const overflowSteps = freshSteps.slice(inlineCandidates.length);
+      const overflowSteps = healthySteps.slice(inlineCandidates.length);
       const dispatched = await dispatchPendingOps({
         world,
         runId,
@@ -1189,6 +1308,7 @@ export async function runWorkflowWithQuickJS(params: {
         pendingOperations: opsToDispatch,
         skipStepCreation: inlineClaimCids,
         queueStepCids: new Set(overflowSteps.map((s) => s.correlationId)),
+        finalizeUnserializableSteps: true,
         wfdiag,
       });
       if (
@@ -1197,6 +1317,19 @@ export async function runWorkflowWithQuickJS(params: {
       ) {
         pendingRequeueSignal = true;
       }
+      // A finalized unserializable step has terminal events durably
+      // written but no execution message anywhere: if the feed below
+      // doesn't surface them (eventually-consistent listing) and the loop
+      // exits, nothing would ever re-invoke the run to observe the
+      // failure. Raise the requeue signal — same mechanism as inline
+      // terminals — and mark the steps handled so later turns don't
+      // re-finalize or backstop-queue them.
+      if (dispatched.failedSerializationStepCids.size > 0) {
+        pendingRequeueSignal = true;
+        for (const cid of dispatched.failedSerializationStepCids) {
+          executedStepIds.add(cid);
+        }
+      }
 
       for (const cid of dispatched.queuedStepCids) {
         queuedStepIds.add(cid);

packages/core/src/runtime/quickjs-runtime.ts

@@ -29,6 +29,7 @@
  * `node:vm` engine's replay determinism.
  */
 
+import { SerializationError } from '@workflow/errors';
 import type {
   Event,
   RunInput,
@@ -49,6 +50,7 @@ import { runtimeLogger } from '../logger.js';
 import { decompress } from '../serialization/compression.js';
 import type { DecryptionKey } from '../serialization/encryption.js';
 import { decrypt } from '../serialization/encryption.js';
+import { formatSerializationError } from '../serialization/errors.js';
 import {
   getReplayTimeoutMs,
   isQuickJSBaselineSnapshotEnabled,
@@ -90,10 +92,25 @@ export interface PendingStep {
   type: 'step';
   correlationId: string;
   stepId: string;
-  /** Format-prefixed devalue-serialized step input (args + closureVars) */
-  input: Uint8Array;
+  /**
+   * Format-prefixed devalue-serialized step input (args + closureVars).
+   * Absent when {@link serializationError} is set — the input is precisely
+   * what refused to serialize.
+   */
+  input?: Uint8Array;
   /** Whether a step_created event already exists for this step */
   hasCreatedEvent: boolean;
+  /**
+   * Set when host-side serialization of the step's raw input failed while
+   * dumping the VM's pending ops (see `dumpPendingOps`). The failure is
+   * deterministic (replaying re-derives the same unserializable value), so
+   * instead of failing the whole collection the op is surfaced with the
+   * reframed error and no `input`; the entrypoint finalizes the step as
+   * `step_created` (placeholder input) + `step_failed`, mirroring the
+   * node:vm engine's `finalizeUnserializableStep`, so a try/catch around
+   * the step call observes the SerializationError.
+   */
+  serializationError?: SerializationError;
 }
 
 export interface PendingWait {
@@ -2512,7 +2529,32 @@ function dumpPendingOps(
       let bytes = byteCache?.get(cacheKey);
       if (!bytes) {
         using valueHandle = rawFields.getProp(String(index));
-        bytes = serde.serialize(valueHandle);
+        try {
+          bytes = serde.serialize(valueHandle);
+        } catch (err) {
+          // A step input that refuses to serialize is a deterministic user
+          // error: failing the whole collection here would fail the run
+          // from the outside, where no workflow code can observe it (and
+          // with a bare DevalueError instead of the framed message the
+          // node:vm engine produces). Reframe it exactly like
+          // `dehydrateStepArguments` does and surface it on the op — the
+          // entrypoint finalizes the step as step_created + step_failed so
+          // the failure rejects into the workflow, catchable. Other raw
+          // fields (hook metadata, abort payloads) keep the throwing
+          // behavior, matching the node:vm engine's scope.
+          if (op.type === 'step' && field === 'input') {
+            const { message, hint } = formatSerializationError(
+              'step arguments',
+              err
+            );
+            (op as PendingStep).serializationError = new SerializationError(
+              message,
+              { hint, cause: err }
+            );
+            continue;
+          }
+          throw err;
+        }
         byteCache?.set(cacheKey, bytes);
       }
       (op as unknown as Record<string, unknown>)[field] = bytes;

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

@@ -6,9 +6,13 @@ import { SPEC_VERSION_CURRENT } from '@workflow/world';
 import { createWorld } from '@workflow/world-local';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import { registerStepFunction } from '../private.js';
-import { dehydrateStepArguments } from '../serialization.js';
+import { dehydrateStepArguments, hydrateStepError } from '../serialization.js';
 import { COMPUTE_INSTANCE_ID } from './compute-instance.js';
 import { executeStep } from './step-executor.js';
+import {
+  UNSERIALIZABLE_STEP_INPUT_MARKER,
+  unserializableStepInputPlaceholder,
+} from './unserializable-step.js';
 
 // The retry ceiling (`authoritativeAttempt`) is what bounds a step that keeps
 // timing out: a timeout hard-kills the body without writing any error, so the
@@ -478,3 +482,114 @@ describe('executeStep — pre-claimed inline start', () => {
     expect(createSpy).not.toHaveBeenCalled();
   });
 });
+
+describe('executeStep — unserializable-argument placeholder guard', () => {
+  afterEach(() => {
+    counter += 1;
+  });
+
+  it('fails the step without running the body when the stored input is the finalization placeholder', async () => {
+    // Simulates the crash window in finalizeUnserializableStep: the
+    // step_created (placeholder input) landed but the process died before
+    // step_failed. Redelivery dispatches the step through normal crash
+    // recovery — the executor must complete the intended failure, not run
+    // user code with placeholder arguments.
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    let bodyRuns = 0;
+    const { runId, stepId } = await setupRunningStep({
+      world,
+      stepName,
+      onBody: () => {
+        bodyRuns += 1;
+      },
+      createStep: false,
+    });
+    await world.events.create(runId, {
+      eventType: 'step_created',
+      specVersion: SPEC_VERSION_CURRENT,
+      correlationId: stepId,
+      eventData: {
+        stepName,
+        input: (await dehydrateStepArguments(
+          unserializableStepInputPlaceholder(),
+          runId,
+          undefined
+        )) as Uint8Array,
+      },
+    });
+
+    const result = await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId,
+      stepName,
+      authoritativeAttempt: 1,
+    });
+
+    expect(result.type).toBe('failed');
+    expect(bodyRuns).toBe(0);
+
+    // Fatal — one attempt, no step_retrying, straight to step_failed.
+    const retrying = await eventsFor(world, runId, stepId, 'step_retrying');
+    expect(retrying).toHaveLength(0);
+    const failures = await eventsFor(world, runId, stepId, 'step_failed');
+    expect(failures).toHaveLength(1);
+    const hydrated = (await hydrateStepError(
+      (failures[0].eventData as { error: unknown }).error,
+      runId,
+      undefined
+    )) as Error;
+    expect(hydrated.name).toBe('SerializationError');
+    expect(hydrated.message).toContain('Failed to serialize step arguments');
+  });
+
+  it('does not trip on a genuine input that merely contains the marker string', async () => {
+    // The structural flag lives on the triple's top level, which user code
+    // never controls — an argument that happens to equal the display marker
+    // must execute normally.
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    let bodyRuns = 0;
+    const { runId, stepId } = await setupRunningStep({
+      world,
+      stepName,
+      onBody: () => {
+        bodyRuns += 1;
+      },
+      createStep: false,
+    });
+    await world.events.create(runId, {
+      eventType: 'step_created',
+      specVersion: SPEC_VERSION_CURRENT,
+      correlationId: stepId,
+      eventData: {
+        stepName,
+        input: (await dehydrateStepArguments(
+          {
+            args: [UNSERIALIZABLE_STEP_INPUT_MARKER],
+            closureVars: [],
+            thisVal: undefined,
+          },
+          runId,
+          undefined
+        )) as Uint8Array,
+      },
+    });
+
+    const result = await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId,
+      stepName,
+      authoritativeAttempt: 1,
+    });
+
+    expect(result.type).toBe('completed');
+    expect(bodyRuns).toBe(1);
+  });
+});

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

@@ -5,6 +5,7 @@ import {
   FatalError,
   RetryableError,
   RunExpiredError,
+  SerializationError,
   ThrottleError,
   TooEarlyError,
   WorkflowRuntimeError,
@@ -31,6 +32,7 @@ import {
 import { runtimeLogger, stepLogger } from '../logger.js';
 import { getStepFunction } from '../private.js';
 import type { PayloadKey } from '../serialization/encryption.js';
+import { formatSerializationError } from '../serialization/errors.js';
 import {
   cancelAbortReaders,
   dehydrateStepError,
@@ -68,6 +70,7 @@ import {
   type StepLatencyEventData,
   type StepLatencyTracking,
 } from './step-latency.js';
+import { isUnserializableStepInputPlaceholder } from './unserializable-step.js';
 import { safeWaitUntil } from './wait-until.js';
 
 export const DEFAULT_STEP_MAX_RETRIES = 3;
@@ -1000,6 +1003,23 @@ export async function executeStep(
         }
       );
 
+      // Finalization of an unserializable-argument step writes step_created
+      // (placeholder input) and step_failed as two separate durable writes.
+      // 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. The
+      // SerializationError is fatal (`fatal: true`), so the catch below
+      // writes step_failed without retries, exactly what the interrupted
+      // finalization was about to do.
+      if (isUnserializableStepInputPlaceholder(hydratedInput)) {
+        const { message, hint } = formatSerializationError(
+          'step arguments',
+          undefined
+        );
+        throw new SerializationError(message, { hint });
+      }
+
       const args = hydratedInput.args;
       const thisVal = hydratedInput.thisVal ?? null;
       const workflowBaseUrl = createWorkflowBaseUrl(

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

@@ -1,7 +1,9 @@
 import { runInNewContext } from 'node:vm';
 import {
+  EntityConflictError,
   FatalError,
   PreconditionFailedError,
+  RunExpiredError,
   WorkflowWorldError,
 } from '@workflow/errors';
 import type { Event } from '@workflow/world';
@@ -14,10 +16,12 @@ import {
 } from '@workflow/world';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { WorkflowSuspension } from '../global.js';
+import { hydrateStepArguments, hydrateStepError } from '../serialization.js';
 import { COMPUTE_INSTANCE_ID } from './compute-instance.js';
 import { maxEventSlot, stepDispatchIdempotencyKey } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
 import { handleSuspension } from './suspension-handler.js';
+import { isUnserializableStepInputPlaceholder } from './unserializable-step.js';
 
 vi.mock('../version.js', () => ({ version: '0.0.0-test' }));
 
@@ -1719,6 +1723,90 @@ describe('handleSuspension batched fan-out', () => {
       });
     });
 
+    it('mixed bad step + large fan-out: deferred rejection still surfaces through deferredBatchWork', async () => {
+      // A step whose args fail serialization is finalized on the sequential
+      // path while the healthy fan-out still defers trailing chunk commits
+      // and publishes. The caller's failed-step replay path must join
+      // deferredBatchWork before continuing (runtime.ts), so its rejection
+      // is observable — this pins the handler-side contract: the failure
+      // set and the still-pending deferred work coexist on one result.
+      class Unserializable {
+        secret = 'not-a-pojo';
+      }
+      let call = 0;
+      let releaseFailure: (() => void) | undefined;
+      const createBatch = vi.fn().mockImplementation((_runId, events) => {
+        call += 1;
+        if (call === 2) {
+          return new Promise((_resolve, reject) => {
+            releaseFailure = () =>
+              reject(
+                new WorkflowWorldError('trailing publish failed', {
+                  status: 500,
+                })
+              );
+          });
+        }
+        let slot = 10;
+        return Promise.resolve({
+          results: events.map(({ event }: { event: object }) => ({
+            status: 200,
+            event: { ...event, eventId: slotToEventId(slot++) },
+          })),
+        });
+      });
+      const eventsCreate = vi
+        .fn()
+        .mockImplementation(async (_runId, event) => ({ event }));
+      const world = {
+        events: { create: eventsCreate, createBatch },
+        queue: vi.fn().mockResolvedValue({ messageId: 'msg_q' }),
+        getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined),
+      } as unknown as World;
+
+      const pending = stepsAndWait(
+        Array.from({ length: 34 }, (_, i) => `s${i + 1}`)
+      ) as Map<string, { args: unknown[] }>;
+      // biome-ignore lint/style/noNonNullAssertion: seeded above
+      pending.get('s5')!.args = [new Unserializable()];
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(
+          pending as ConstructorParameters<typeof WorkflowSuspension>[0],
+          globalThis
+        ),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+        stepDispatch: stepDispatch(),
+        allowDeferredBatchWork: true,
+      });
+
+      // The bad step was finalized sequentially (step_created placeholder +
+      // step_failed), dropped out of the batch fold…
+      expect([...result.failedStepCorrelationIds]).toEqual(['s5']);
+      expect(
+        eventsCreate.mock.calls.map(([, event]) => [
+          event.eventType,
+          event.correlationId,
+        ])
+      ).toEqual([
+        ['step_created', 's5'],
+        ['step_failed', 's5'],
+      ]);
+      // …while the healthy fan-out still handed back live deferred work.
+      expect(result.deferredBatchWork).toBeDefined();
+      expect(await probe(result.deferredBatchWork)).toBe('pending');
+
+      // A trailing rejection surfaces through the deferred promise — the
+      // caller's failed-step path awaits it before replaying.
+      // biome-ignore lint/style/noNonNullAssertion: set by the second call
+      releaseFailure!();
+      await expect(result.deferredBatchWork).rejects.toMatchObject({
+        message: expect.stringContaining('trailing publish failed'),
+      });
+    });
+
     it('settles the trailing chunk before a pair-chunk failure escapes', async () => {
       // settlePhase's invariant: a phase's write set must be final before a
       // failure escapes, or a sibling create lands during the caller's replay
@@ -1804,3 +1892,299 @@ describe('handleSuspension batched fan-out', () => {
     });
   });
 });
+
+describe('step-argument serialization failure', () => {
+  // A value the workflow serializer cannot dehydrate: a class instance with
+  // no registered serde model. Mirrors serialization.test.ts's unsupported
+  // type coverage — dehydrateStepArguments throws a SerializationError.
+  class Unserializable {
+    secret = 'not-a-pojo';
+  }
+
+  function stepItem(id: string, args: unknown[] = []) {
+    return {
+      type: 'step' as const,
+      correlationId: id,
+      stepName: id,
+      args,
+    };
+  }
+
+  // Finalization requires a dispatch target: a caller without one (the
+  // terminal drain) has no replay to observe the failure — see the
+  // stepDispatch gate in the per-step op.
+  const stepDispatch = () => ({
+    queueName: '__wkf_workflow_test-workflow' as ValidQueueName,
+    getTraceCarrier: vi.fn().mockResolvedValue({}),
+  });
+
+  it('finalizes the step as step_created + step_failed instead of rejecting the suspension', async () => {
+    const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({
+      event,
+    }));
+    const world = createWorld(eventsCreate);
+    const pending = new Map([
+      ['s_bad', stepItem('s_bad', [new Unserializable()])],
+    ]);
+
+    const result = await handleSuspension({
+      suspension: new WorkflowSuspension(pending, globalThis),
+      world,
+      run,
+      stepDispatch: stepDispatch(),
+    });
+
+    // The suspension itself resolves — the failure is scoped to the step.
+    expect(eventsCreate).toHaveBeenCalledTimes(2);
+    const [createdCall, failedCall] = eventsCreate.mock.calls;
+    expect(createdCall[1]).toMatchObject({
+      eventType: 'step_created',
+      correlationId: 's_bad',
+      eventData: expect.objectContaining({
+        stepName: 's_bad',
+        workflowName: run.workflowName,
+      }),
+    });
+    expect(failedCall[1]).toMatchObject({
+      eventType: 'step_failed',
+      correlationId: 's_bad',
+      eventData: expect.objectContaining({ stepName: 's_bad' }),
+    });
+    expect([...result.failedStepCorrelationIds]).toEqual(['s_bad']);
+    // Not owned for dispatch, not deferred for lazy-inline execution: the
+    // step is terminal.
+    expect(result.createdStepCorrelationIds.size).toBe(0);
+    expect(result.lazyInlineSteps).toEqual([]);
+  });
+
+  it('round-trips the SerializationError through the step_failed payload', async () => {
+    const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({
+      event,
+    }));
+    const world = createWorld(eventsCreate);
+    const pending = new Map([
+      ['s_bad', stepItem('s_bad', [new Unserializable()])],
+    ]);
+
+    await handleSuspension({
+      suspension: new WorkflowSuspension(pending, globalThis),
+      world,
+      run,
+      stepDispatch: stepDispatch(),
+    });
+
+    const failedEvent = eventsCreate.mock.calls.find(
+      ([, event]) => event.eventType === 'step_failed'
+    )?.[1];
+    expect(failedEvent).toBeDefined();
+    const hydrated = (await hydrateStepError(
+      failedEvent.eventData.error,
+      run.runId,
+      undefined
+    )) as Error;
+    expect(hydrated).toBeInstanceOf(Error);
+    expect(hydrated.name).toBe('SerializationError');
+    expect(hydrated.message).toContain('Failed to serialize step arguments');
+  });
+
+  it('finalizes the bad step while healthy siblings proceed', async () => {
+    const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({
+      event,
+    }));
+    const world = createWorld(eventsCreate);
+    // Default inline cap (3): both steps are designated lazy-inline, but the
+    // bad one is finalized before deferral, so only the healthy step defers.
+    const pending = new Map([
+      ['s_bad', stepItem('s_bad', [new Unserializable()])],
+      ['s_good', stepItem('s_good', ['fine'])],
+    ]);
+
+    const result = await handleSuspension({
+      suspension: new WorkflowSuspension(pending, globalThis),
+      world,
+      run,
+      stepDispatch: stepDispatch(),
+    });
+
+    expect([...result.failedStepCorrelationIds]).toEqual(['s_bad']);
+    expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([
+      's_good',
+    ]);
+    const eventTypes = eventsCreate.mock.calls.map(([, event]) => [
+      event.eventType,
+      event.correlationId,
+    ]);
+    expect(eventTypes).toEqual([
+      ['step_created', 's_bad'],
+      ['step_failed', 's_bad'],
+    ]);
+  });
+
+  it('drops the bad step out of the batched fan-out onto the sequential path', async () => {
+    vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '1');
+    try {
+      const slotRun: WorkflowRun = { ...run, specVersion: 6 };
+      let slot = 10;
+      const createBatch = vi
+        .fn()
+        .mockImplementation(async (_runId, events) => ({
+          results: events.map(({ event }: { event: object }) => ({
+            status: 200,
+            event: { ...event, eventId: slotToEventId(slot++) },
+          })),
+        }));
+      const eventsCreate = vi
+        .fn()
+        .mockImplementation(async (_runId, event) => ({ event }));
+      const world = {
+        events: { create: eventsCreate, createBatch },
+        // The batch flush publishes chunk step messages when a dispatch
+        // target is provided.
+        queue: vi.fn().mockResolvedValue({ messageId: 'msg_1' }),
+        getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined),
+      } as unknown as World;
+      // s1 defers (cap 1); s_bad fails serialization; s3 + s4 fold into the
+      // batch. The bad step's two writes go through the single-event path.
+      const pending = new Map([
+        ['s1', stepItem('s1')],
+        ['s_bad', stepItem('s_bad', [new Unserializable()])],
+        ['s3', stepItem('s3')],
+        ['s4', stepItem('s4')],
+      ]);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(pending, globalThis),
+        world,
+        run: slotRun,
+        stepDispatch: stepDispatch(),
+      });
+
+      expect([...result.failedStepCorrelationIds]).toEqual(['s_bad']);
+      expect(createBatch).toHaveBeenCalledTimes(1);
+      expect(
+        createBatch.mock.calls[0][1].map(
+          (e: { event: { correlationId: string } }) => e.event.correlationId
+        )
+      ).toEqual(['s3', 's4']);
+      expect(
+        eventsCreate.mock.calls.map(([, event]) => [
+          event.eventType,
+          event.correlationId,
+        ])
+      ).toEqual([
+        ['step_created', 's_bad'],
+        ['step_failed', 's_bad'],
+      ]);
+      expect([...result.createdStepCorrelationIds].sort()).toEqual([
+        's3',
+        's4',
+      ]);
+    } finally {
+      vi.unstubAllEnvs();
+    }
+  });
+
+  it('tolerates a concurrent handler having already finalized the step', async () => {
+    // Both writes conflict: a concurrent replay hit the same deterministic
+    // serialization failure and wrote step_created + step_failed first.
+    const eventsCreate = vi
+      .fn()
+      .mockRejectedValue(new EntityConflictError('already exists'));
+    const world = createWorld(eventsCreate);
+    const pending = new Map([
+      ['s_bad', stepItem('s_bad', [new Unserializable()])],
+    ]);
+
+    const result = await handleSuspension({
+      suspension: new WorkflowSuspension(pending, globalThis),
+      world,
+      run,
+      stepDispatch: stepDispatch(),
+    });
+
+    expect([...result.failedStepCorrelationIds]).toEqual(['s_bad']);
+  });
+
+  it('skips finalization when the run has already finished', async () => {
+    const eventsCreate = vi
+      .fn()
+      .mockRejectedValue(new RunExpiredError('run is gone'));
+    const world = createWorld(eventsCreate);
+    const pending = new Map([
+      ['s_bad', stepItem('s_bad', [new Unserializable()])],
+    ]);
+
+    const result = await handleSuspension({
+      suspension: new WorkflowSuspension(pending, globalThis),
+      world,
+      run,
+      stepDispatch: stepDispatch(),
+    });
+
+    // Nothing to observe the failure — no replay is forced.
+    expect(result.failedStepCorrelationIds.size).toBe(0);
+  });
+
+  it('rejects the suspension when step_failed cannot be written after step_created landed', async () => {
+    // The two finalization writes are separate durable writes. If the second
+    // fails transiently, the suspension must reject so the message
+    // redelivers — leaving a lone placeholder step_created behind. Recovery
+    // for that window lives in the step executor: the placeholder carries a
+    // structural flag (see unserializable-step.ts) that the executor
+    // completes as the intended step_failed instead of running user code
+    // with placeholder arguments (covered in step-executor.test.ts).
+    const writeError = new Error('storage unavailable');
+    const eventsCreate = vi
+      .fn()
+      .mockImplementationOnce(async (_runId, event) => ({ event }))
+      .mockRejectedValueOnce(writeError);
+    const world = createWorld(eventsCreate);
+    const pending = new Map([
+      ['s_bad', stepItem('s_bad', [new Unserializable()])],
+    ]);
+
+    await expect(
+      handleSuspension({
+        suspension: new WorkflowSuspension(pending, globalThis),
+        world,
+        run,
+        stepDispatch: stepDispatch(),
+      })
+    ).rejects.toBe(writeError);
+
+    // The lone step_created that redelivery will find carries the
+    // recoverable placeholder, not a genuine-looking empty input.
+    expect(eventsCreate).toHaveBeenCalledTimes(2);
+    const createdEvent = eventsCreate.mock.calls[0][1];
+    expect(createdEvent.eventType).toBe('step_created');
+    const hydrated = await hydrateStepArguments(
+      createdEvent.eventData.input,
+      run.runId,
+      undefined,
+      []
+    );
+    expect(isUnserializableStepInputPlaceholder(hydrated)).toBe(true);
+  });
+
+  it('rethrows instead of finalizing when no stepDispatch is provided (terminal drain)', async () => {
+    // The drain caller (drainPendingQueueItems) passes no stepDispatch and
+    // swallows the rejection: a run that is already completing must not
+    // gain step_created + step_failed rows nothing can ever observe.
+    const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({
+      event,
+    }));
+    const world = createWorld(eventsCreate);
+    const pending = new Map([
+      ['s_bad', stepItem('s_bad', [new Unserializable()])],
+    ]);
+
+    await expect(
+      handleSuspension({
+        suspension: new WorkflowSuspension(pending, globalThis),
+        world,
+        run,
+      })
+    ).rejects.toMatchObject({ name: 'SerializationError' });
+    expect(eventsCreate).not.toHaveBeenCalled();
+  });
+});

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

@@ -5,6 +5,7 @@ import {
   HookNotFoundError,
   PreconditionFailedError,
   RunExpiredError,
+  SerializationError,
   WorkflowWorldError,
 } from '@workflow/errors';
 import {
@@ -34,7 +35,10 @@ import type {
 } from '../global.js';
 import { runtimeLogger } from '../logger.js';
 import type { GuestCodeStats } from '../serialization/hardened.js';
-import { dehydrateStepArguments } from '../serialization.js';
+import {
+  dehydrateStepArguments,
+  dehydrateStepError,
+} from '../serialization.js';
 import * as Attribute from '../telemetry/semantic-conventions.js';
 import { getAbortStreamIdFromToken } from '../util.js';
 import { COMPUTE_INSTANCE_ID } from './compute-instance.js';
@@ -56,6 +60,7 @@ import {
 } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
 import type { PreclaimedInlineStart } from './step-executor.js';
+import { unserializableStepInputPlaceholder } from './unserializable-step.js';
 
 export interface SuspensionHandlerParams {
   suspension: WorkflowSuspension;
@@ -149,6 +154,18 @@ export interface SuspensionHandlerResult {
    * into the same batch boundary.
    */
   createdStepCorrelationIds: Set<string>;
+  /**
+   * Correlation IDs of steps whose arguments failed to serialize. Each was
+   * finalized here as `step_created` (with a placeholder input — the real
+   * input is precisely what refused to serialize) followed by `step_failed`
+   * carrying the SerializationError, so the next replay rejects the step's
+   * promise and a try/catch around the step call observes the error —
+   * exactly like a step-body failure. No step-execution message is
+   * dispatched for these, so the caller MUST force an in-process replay:
+   * when the failed step was the only pending work, nothing else will ever
+   * re-invoke the run to observe the terminal event.
+   */
+  failedStepCorrelationIds: Set<string>;
   /**
    * Correlation IDs of steps this suspension call already published
    * step-execution queue messages for, via resilient step dispatch (the
@@ -716,6 +733,135 @@ export async function handleSuspension({
   // racing with concurrent handlers on step execution.
   const createdStepCorrelationIds = new Set<string>();
 
+  // Correlation IDs of steps finalized as failed because their arguments
+  // refused to serialize — see finalizeUnserializableStep below.
+  const failedStepCorrelationIds = new Set<string>();
+
+  /**
+   * A step whose arguments fail to serialize is deterministic: every replay
+   * re-derives the same unserializable value, so redelivering the
+   * orchestrator message can never succeed. Instead of rejecting the whole
+   * suspension (which fails the run from the outside, where no user code can
+   * observe it), treat it exactly like a step-body failure: write
+   * `step_created` with a placeholder input (every World requires the step
+   * entity to exist before a terminal step event, and the real input is
+   * precisely what refused to serialize) followed by `step_failed` carrying
+   * the SerializationError. The next replay rejects the step's promise with
+   * it, so a try/catch around the step call observes the error; uncaught, it
+   * propagates out of the workflow body and fails the run as a USER_ERROR —
+   * without burning queue redeliveries either way.
+   */
+  const finalizeUnserializableStep = async (
+    queueItem: StepInvocationQueueItem,
+    error: SerializationError
+  ): Promise<void> => {
+    runtimeLogger.warn(
+      'Step arguments failed to serialize; failing the step so the ' +
+        'workflow can observe the error',
+      {
+        workflowRunId: runId,
+        correlationId: queueItem.correlationId,
+        stepName: queueItem.stepName,
+        error: error.message,
+      }
+    );
+    await ensureRunReady();
+    // Marker placeholder (not empty args): byte-identical-to-zero-args would
+    // make `workflow inspect steps` show "no arguments" for the one step
+    // whose entire problem was its arguments.
+    const placeholderInput = (await dehydrateStepArguments(
+      unserializableStepInputPlaceholder(),
+      runId,
+      encryptionKey,
+      suspension.globalThis,
+      false,
+      compression
+    )) as SerializedData;
+    try {
+      await createGuarded(
+        {
+          eventType: 'step_created' as const,
+          specVersion: SPEC_VERSION_CURRENT,
+          correlationId: queueItem.correlationId,
+          eventData: {
+            stepName: queueItem.stepName,
+            workflowName: run.workflowName,
+            input: placeholderInput,
+          },
+        },
+        { requestId }
+      );
+    } catch (createErr) {
+      if (EntityConflictError.is(createErr)) {
+        // A concurrent handler already created the step — the failure is
+        // deterministic, so it is racing toward the same step_failed below.
+        runtimeLogger.info('Step already exists, continuing', {
+          workflowRunId: runId,
+          correlationId: queueItem.correlationId,
+          message: createErr.message,
+        });
+      } else if (RunExpiredError.is(createErr)) {
+        // Run already finished — nothing to observe the failure.
+        return;
+      } else {
+        throw createErr;
+      }
+    }
+    try {
+      await createGuarded(
+        {
+          eventType: 'step_failed' as const,
+          specVersion: SPEC_VERSION_CURRENT,
+          correlationId: queueItem.correlationId,
+          eventData: {
+            stepName: queueItem.stepName,
+            // The error itself is a plain WorkflowError (name, message with
+            // framed hint, cause chain) — serializable even though the step
+            // input was not. Error detection is realm-independent
+            // (types.isNativeError), so the host-created error serializes
+            // the same under either global; the VM global is passed for
+            // consistency with every other dehydration in this file and so
+            // any VM-realm values guest code threw into the cause chain
+            // (getters/proxies executed during the failed dehydration) are
+            // detected by the realm-sensitive reducers.
+            error: await dehydrateStepError(
+              error,
+              runId,
+              encryptionKey,
+              [],
+              suspension.globalThis,
+              compression
+            ),
+          },
+        },
+        { requestId }
+      );
+    } catch (failErr) {
+      if (EntityConflictError.is(failErr) || RunExpiredError.is(failErr)) {
+        // Step already terminal (a concurrent handler wrote the same
+        // deterministic failure) or the run already finished.
+        runtimeLogger.info(
+          'Tried failing step, but step or run has already finished.',
+          {
+            workflowRunId: runId,
+            correlationId: queueItem.correlationId,
+            message: failErr.message,
+          }
+        );
+      } else {
+        throw failErr;
+      }
+    }
+    failedStepCorrelationIds.add(queueItem.correlationId);
+    // Release the inline slot bookkeeping: the step never runs, so it must
+    // not appear in the rebuilt `lazyInlineSteps`. (Its slot in the first-N
+    // selection and in `inlinePairFoldEligible`'s arithmetic was consumed
+    // before dehydration could reveal the failure — inherent to selecting
+    // before serializing, and bounded to one wasted slot on a pass that
+    // ends in a forced replay anyway.)
+    lazyInlineCorrelationIds.delete(queueItem.correlationId);
+  };
+
   // Serialization always runs through the one ordinary path below, so the
   // durable bytes cannot depend on retention. What retention needs to know is
   // whether that serialization *executed* workflow code (getters, proxy
@@ -878,19 +1024,45 @@ export async function handleSuspension({
         // attributes from the sink it is handed, so sharing one across
         // steps would re-emit (and misattribute) earlier steps' entries.
         const stepGuestCode: GuestCodeStats = { executions: [] };
-        const dehydratedInput = await dehydrateStepArguments(
-          {
-            args: queueItem.args,
-            closureVars: queueItem.closureVars,
-            thisVal: queueItem.thisVal,
-          },
-          runId,
-          encryptionKey,
-          suspension.globalThis,
-          false,
-          compression,
-          stepGuestCode
-        );
+        let dehydratedInput: Uint8Array | unknown;
+        try {
+          dehydratedInput = await dehydrateStepArguments(
+            {
+              args: queueItem.args,
+              closureVars: queueItem.closureVars,
+              thisVal: queueItem.thisVal,
+            },
+            runId,
+            encryptionKey,
+            suspension.globalThis,
+            false,
+            compression,
+            stepGuestCode
+          );
+        } catch (err) {
+          // The sink records executions as they happen, so guest code that
+          // ran before the failure still counts against retention.
+          guestCodeStats.executions.push(...stepGuestCode.executions);
+          if (!SerializationError.is(err)) {
+            // e.g. RuntimeDecryptionError — an SDK fault, not a user value
+            // problem. Keep its identity (RUNTIME_ERROR) and current
+            // fail-the-suspension behavior.
+            throw err;
+          }
+          if (!stepDispatch) {
+            // No dispatch target means no replay will observe a
+            // finalization: this is the terminal drain (or a create-only
+            // test caller). The run is already completing/failing, so
+            // writing step_created + step_failed here would leave e.g. a
+            // COMPLETED run carrying a failed step nothing can ever
+            // observe — reading as a bug from the dashboard. Rethrow
+            // instead; the drain's own catch swallows it, preserving its
+            // pre-existing behavior (no rows for the unawaited step).
+            throw err;
+          }
+          await finalizeUnserializableStep(queueItem, err);
+          return;
+        }
         guestCodeStats.executions.push(...stepGuestCode.executions);
         // Deferred (lazy) inline step: skip the step_created write — the
         // caller's inline executeStep will send a lazy step_started carrying
@@ -1648,6 +1820,11 @@ export async function handleSuspension({
     ...Attribute.WorkflowStepsCreated(stepItems.length),
     ...Attribute.WorkflowHooksCreated(hooksNeedingCreation.length),
     ...Attribute.WorkflowWaitsCreated(waitItems.length),
+    ...(failedStepCorrelationIds.size > 0
+      ? Attribute.WorkflowStepsFailedSerialization(
+          failedStepCorrelationIds.size
+        )
+      : {}),
     ...(resilientDispatchRecovered > 0
       ? Attribute.StepResilientDispatchRecovered(resilientDispatchRecovered)
       : {}),
@@ -1656,6 +1833,7 @@ export async function handleSuspension({
   return {
     pendingSteps: stepItems,
     createdStepCorrelationIds,
+    failedStepCorrelationIds,
     queuedStepCorrelationIds,
     lazyInlineSteps,
     inlineClaims,

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

@@ -0,0 +1,59 @@
+/**
+ * Shared shape for finalizing a step whose arguments failed to serialize
+ * (used by both the node:vm suspension handler and the QuickJS entrypoint).
+ *
+ * The world requires a `step_created` before any terminal step event, and
+ * the step's real input is precisely what refused to serialize — so the
+ * finalization writes a placeholder input. The marker string makes the
+ * placeholder distinguishable from a genuine zero-argument step in
+ * `workflow inspect steps` and the observability UI: a reader sees
+ * "input unavailable" instead of "no arguments".
+ */
+export const UNSERIALIZABLE_STEP_INPUT_MARKER =
+  '[input unavailable: step argument serialization failed]';
+
+/**
+ * Structural discriminator on the placeholder's top level. The
+ * `{ args, closureVars, thisVal }` triple is built by the SDK — user code
+ * never controls its top-level keys — so this flag cannot false-positive on
+ * a legitimate input, unlike the display marker inside `args`.
+ */
+const UNSERIALIZABLE_FLAG = '__workflowUnserializableStepInput';
+
+/**
+ * The placeholder value serialized into the failed step's `step_created`
+ * input. Matches the `{ args, closureVars, thisVal }` triple
+ * `dehydrateStepArguments` / the QuickJS bootstrap produce for real steps,
+ * so every consumer hydrates it uniformly, plus the structural flag the
+ * step executor checks before running user code (see
+ * {@link isUnserializableStepInputPlaceholder}).
+ */
+export function unserializableStepInputPlaceholder(): Record<string, unknown> {
+  return {
+    args: [UNSERIALIZABLE_STEP_INPUT_MARKER],
+    closureVars: [],
+    thisVal: undefined,
+    [UNSERIALIZABLE_FLAG]: true,
+  };
+}
+
+/**
+ * Whether a hydrated step input is the finalization placeholder.
+ *
+ * Finalization writes `step_created` (placeholder) and `step_failed` as two
+ * separate durable writes; a crash or transient failure between them leaves
+ * a pending step whose stored input is the placeholder. Redelivery then
+ * dispatches that step through normal crash recovery — the executor calls
+ * this before running user code and completes the intended failure (a fatal
+ * SerializationError → `step_failed`) instead of silently invoking the step
+ * body with placeholder arguments.
+ */
+export function isUnserializableStepInputPlaceholder(
+  hydratedInput: unknown
+): boolean {
+  return (
+    typeof hydratedInput === 'object' &&
+    hydratedInput !== null &&
+    (hydratedInput as Record<string, unknown>)[UNSERIALIZABLE_FLAG] === true
+  );
+}

packages/core/src/telemetry/semantic-conventions.ts

@@ -196,6 +196,15 @@ export const WorkflowWaitsCreated = SemanticConvention<number>(
   'workflow.waits.created'
 );
 
+/**
+ * Number of steps this suspension finalized as failed because their
+ * arguments refused to serialize (step_created placeholder + step_failed
+ * carrying the SerializationError — see finalizeUnserializableStep).
+ */
+export const WorkflowStepsFailedSerialization = SemanticConvention<number>(
+  'workflow.steps.failed_serialization'
+);
+
 /**
  * Number of inline-owned steps this invocation re-executed because it is a
  * redelivery of their owning queue message (crash recovery for inline

workbench/example/workflows/99_e2e.ts

@@ -1473,6 +1473,91 @@ export async function errorStepThrowNonErrorValue() {
   }
 }
 
+// ---
+
+/**
+ * A class instance with no registered serde model cannot cross the
+ * workflow/step boundary — the serializer rejects non-POJO instances.
+ * Used by the serialization-error tests below.
+ */
+class UnserializableValue {
+  secret = 'not-serializable';
+}
+
+async function acceptAnyValue(value: unknown) {
+  'use step';
+  return { received: value !== undefined };
+}
+
+/**
+ * Test: step ARGUMENTS that cannot be serialized. The suspension handler
+ * fails the step (step_created + step_failed) instead of failing the run
+ * from the outside, so a try/catch around the step call observes the
+ * SerializationError — same shape as catching a step-body failure.
+ */
+export async function serializationErrorStepArgsCaught() {
+  'use workflow';
+  try {
+    await acceptAnyValue(new UnserializableValue());
+    return { caught: false } as any;
+  } catch (err: any) {
+    return {
+      caught: true,
+      name: err?.name,
+      messageIncludesStepArguments:
+        typeof err?.message === 'string' &&
+        err.message.includes('Failed to serialize step arguments'),
+    };
+  }
+}
+
+/**
+ * Test: uncaught step-argument serialization failure fails the run as a
+ * fatal USER_ERROR immediately — no queue-redelivery retry loop.
+ */
+export async function serializationErrorStepArgsUncaught() {
+  'use workflow';
+  // Don't catch — the SerializationError propagates and fails the run.
+  await acceptAnyValue(new UnserializableValue());
+  return { caught: false };
+}
+
+async function returnUnserializableValue() {
+  'use step';
+  return new UnserializableValue();
+}
+
+/**
+ * Test: step RETURN VALUE that cannot be serialized. The step executor
+ * treats the SerializationError as fatal (skipping the retry loop) and
+ * writes step_failed, so the workflow can catch it.
+ */
+export async function serializationErrorStepReturnCaught() {
+  'use workflow';
+  try {
+    await returnUnserializableValue();
+    return { caught: false } as any;
+  } catch (err: any) {
+    return {
+      caught: true,
+      name: err?.name,
+      messageIncludesReturnValue:
+        typeof err?.message === 'string' &&
+        err.message.includes('Failed to serialize step return value'),
+    };
+  }
+}
+
+/**
+ * Test: uncaught step-return-value serialization failure fails the run as
+ * a fatal USER_ERROR.
+ */
+export async function serializationErrorStepReturnUncaught() {
+  'use workflow';
+  await returnUnserializableValue();
+  return { caught: false };
+}
+
 // ------------------------------------------------------------
 // SECTION 4: NOT REGISTERED ERRORS
 // Tests for step/workflow not registered in the current deployment