feat(core): widen retained boundaries to plain data and standard built-ins

.changeset/retained-vm-guest-code-gate.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': minor
+'workflow': minor
+---
+
+Retained workflow VMs now keep the fast path when step arguments are plain data or standard built-ins (`Map`, `Set`, `Date`, typed arrays, `URL`, `Headers`, …), not just primitives. A boundary falls back to a normal replay only when serializing its arguments ran code the workflow controls — a getter, a proxy, a custom serializer — or computed an `Error`'s stack trace.

docs/content/docs/v5/configuration/runtime-tuning.mdx

@@ -101,7 +101,7 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 - Default: enabled
 - Keeps the suspended workflow VM alive across inline steps within one invocation, so each iteration of the inline loop appends only the newly written events instead of replaying the whole event log in a fresh VM.
 - Suspensions involving hooks, waits, or attributes — and any replay divergence — always fall back to a full replay.
-- Step inputs of primitive values remain retainable; anything whose serialization could execute workflow code falls back to replay for that boundary. (Support for plain objects, arrays, and standard built-ins lands in a follow-up.)
+- Step inputs made of plain data (objects, arrays, primitives) and standard built-ins (`Map`, `Set`, `Date`, `RegExp`, typed arrays, `ArrayBuffer`, `URL`, `Headers`) keep the VM retained. Patching or polyfilling built-in prototypes doesn't change that: serialization never calls them. A boundary falls back to a full replay only when serializing its arguments runs code the workflow controls — a getter, a proxy, a custom class serializer — or computes an `Error`'s stack trace.
 - Set `0` or `false` to replay from scratch in a fresh VM on every iteration.
 
 ### `WORKFLOW_INLINE_OWNERSHIP`

packages/core/src/retained-vm-loop.test.ts

@@ -113,6 +113,69 @@ const mixedBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP"
   }
   globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
 
+/**
+ * A two-step workflow source: optional prelude, then `s1(argA)` and
+ * `s2(argB)` in sequence. The interesting part of each fixture is exactly
+ * (prelude, argA, argB).
+ */
+function twoStepSource(prelude: string, argA: string, argB = ''): string {
+  return `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  ${prelude}
+  async function workflow() {
+    const a = await s1(${argA});
+    const b = await s2(${argB});
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+}
+
+// Map/Date/typed-array arguments serialize through captured host intrinsics
+// (see serialization/hardened.ts), so these boundaries stay retainable.
+const builtinArgsWorkflow = twoStepSource(
+  '',
+  '{ index: new Map([["k", 1]]), when: new Date(1234) }',
+  'new Uint8Array([1, 2, 3])'
+);
+
+// The Temporal / core-js pattern: polyfills add new data-valued methods to
+// built-in prototypes and constructor statics. Serialization never reads
+// them, so retention is unaffected.
+const polyfillArgsWorkflow = twoStepSource(
+  `Date.prototype.toTemporalInstant = function () { return "instant"; };
+  Set.prototype.union = function (other) { return new Set([...this, ...other]); };
+  Object.groupBy = function () { return {}; };`,
+  'new Date(1234)',
+  'new Set([1, 2])'
+);
+
+// Replacing a serialization-relevant member (Date.prototype.toISOString)
+// does not affect retention: the Date reducer reads through captured host
+// intrinsics (see serialization/hardened.ts), so the patched member never
+// executes and the serialized bytes stay pristine in both modes.
+const patchedDateArgWorkflow = twoStepSource(
+  'Date.prototype.toISOString = function () { return "patched"; };',
+  'new Date(1234)'
+);
+
+// Serializing an Error with a lazy stack records the read (it runs the
+// engine's format-and-cache, and any Error.prepareStackTrace) — the
+// boundary demotes and the formatter's side effects land in a doomed VM.
+const prepareStackTraceWorkflow = twoStepSource(
+  'Error.prepareStackTrace = () => "formatted";',
+  'new Error("boom")'
+);
+
+// A formatter that deletes itself during the stack read still demotes: the
+// gate records the stack read itself, not the formatter's presence.
+const selfDeletingFormatterWorkflow = twoStepSource(
+  `Error.prepareStackTrace = () => {
+    delete Error.prepareStackTrace;
+    return "formatted";
+  };`,
+  'new Error("boom")'
+);
+
 // `crypto.subtle.digest` computes synchronously via node:crypto, so a
 // digest-using VM stays quiescent at suspension and remains retainable.
 const digestWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
@@ -318,6 +381,49 @@ describe('retained VM through the inline replay loop', () => {
     expect(vmBuilds).toBeGreaterThan(1);
   });
 
+  it('retains boundaries whose args are supported built-ins', async () => {
+    const { vmBuilds, output } = await drive(
+      'wrun_retained_builtins',
+      builtinArgsWorkflow
+    );
+    expect(output).toBeInstanceOf(Uint8Array);
+    expect(vmBuilds).toBe(1);
+  });
+
+  it('retains boundaries when prototypes carry polyfilled data methods', async () => {
+    const { vmBuilds, output } = await drive(
+      'wrun_retained_polyfill',
+      polyfillArgsWorkflow
+    );
+    expect(output).toBeInstanceOf(Uint8Array);
+    expect(vmBuilds).toBe(1);
+  });
+
+  it('retains a Date arg even when a serialization member is replaced', async () => {
+    const { vmBuilds, output } = await drive(
+      'wrun_retained_patched_date',
+      patchedDateArgWorkflow
+    );
+    expect(output).toBeInstanceOf(Uint8Array);
+    expect(vmBuilds).toBe(1);
+  });
+
+  it('demotes when the workflow replaced Error.prepareStackTrace', async () => {
+    const { vmBuilds } = await drive(
+      'wrun_retained_prepare_stack_trace',
+      prepareStackTraceWorkflow
+    );
+    expect(vmBuilds).toBeGreaterThan(1);
+  });
+
+  it('demotes when the formatter deletes itself during serialization', async () => {
+    const { vmBuilds } = await drive(
+      'wrun_retained_self_deleting_formatter',
+      selfDeletingFormatterWorkflow
+    );
+    expect(vmBuilds).toBeGreaterThan(1);
+  });
+
   it('retains a VM that used the synchronous crypto.subtle.digest', async () => {
     const { vmBuilds, result } = await drive(
       'wrun_retained_digest',

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

@@ -1,3 +1,4 @@
+import { runInNewContext } from 'node:vm';
 import { PreconditionFailedError } from '@workflow/errors';
 import type { WorkflowRun, World } from '@workflow/world';
 import { describe, expect, it, vi } from 'vitest';
@@ -508,3 +509,84 @@ describe('handleSuspension', () => {
     ).rejects.toBeInstanceOf(PreconditionFailedError);
   });
 });
+
+describe('retainedStepInputsSafe (serialization passivity gate)', () => {
+  function stepPending(args: unknown[]) {
+    return new Map([
+      [
+        'step_1',
+        {
+          type: 'step' as const,
+          correlationId: 'step_1',
+          stepName: 'someStep',
+          args,
+        },
+      ],
+    ]);
+  }
+
+  /** An object with a VM-realm getter — exactly what the sink records. */
+  function vmGetterObject() {
+    return runInNewContext(
+      `const o = {};
+       Object.defineProperty(o, 'lazy', {
+         enumerable: true,
+         get: () => 'computed',
+       });
+       o`
+    );
+  }
+
+  async function runSuspension(args: unknown[]) {
+    const eventsCreate = vi
+      .fn()
+      .mockImplementation(async (_runId, event) => ({ event }));
+    const world = createWorld(eventsCreate);
+    return handleSuspension({
+      suspension: new WorkflowSuspension(stepPending(args), globalThis),
+      world,
+      run,
+    });
+  }
+
+  it('reports safe for plain data and supported built-ins', async () => {
+    const result = await runSuspension([
+      { nested: [{ ok: true }, 'text', 42n], flag: false },
+      new Map([['k', new Set([1])]]),
+      new Date(1700000000000),
+      new Uint8Array([1, 2, 3]),
+      /pattern/gi,
+      new URL('https://example.com/'),
+    ]);
+    expect(result.retainedStepInputsSafe).toBe(true);
+  });
+
+  it('reports unsafe for an Error argument (stack materialization)', async () => {
+    // Serializing an error reads `stack`, an own engine accessor whose first
+    // invocation formats-and-caches the trace and runs any
+    // `Error.prepareStackTrace` — neither is repeated by a cold replay, so
+    // the boundary must demote.
+    const result = await runSuspension([new Error('lazy stack')]);
+    expect(result.retainedStepInputsSafe).toBe(false);
+  });
+
+  it('reports unsafe when serializing an argument executes a getter', async () => {
+    const value = vmGetterObject();
+    const result = await runSuspension([{ deep: [value] }]);
+    expect(result.retainedStepInputsSafe).toBe(false);
+  });
+
+  it('reports unsafe when an argument is a proxy', async () => {
+    const result = await runSuspension([new Proxy({ a: 1 }, {})]);
+    expect(result.retainedStepInputsSafe).toBe(false);
+  });
+
+  it('still serializes recorded inputs successfully (bytes are unaffected)', async () => {
+    const value = vmGetterObject();
+    const result = await runSuspension([value]);
+    expect(result.retainedStepInputsSafe).toBe(false);
+    // The step is still prepared for execution as usual (a single uncreated
+    // step always lands in the lazy inline slice).
+    expect(result.lazyInlineSteps).toHaveLength(1);
+  });
+});

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

@@ -25,6 +25,7 @@ import type {
   WorkflowSuspension,
 } from '../global.js';
 import { runtimeLogger } from '../logger.js';
+import type { GuestCodeStats } from '../serialization/hardened.js';
 import { dehydrateStepArguments } from '../serialization.js';
 import * as Attribute from '../telemetry/semantic-conventions.js';
 import { getAbortStreamIdFromToken } from '../util.js';
@@ -36,21 +37,6 @@ import {
 } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
 
-// Serializing a primitive executes no code of any kind. BigInt is excluded:
-// its encoding calls a prototype method. Widened to plain data and standard
-// built-ins by the retained-input walker in a follow-up. (Distinct from
-// replay-payload-cache's `isMemoizablePrimitive`, a size-gated memoization
-// filter — do not merge them.)
-function isPrimitiveStepArgument(value: unknown): boolean {
-  return (
-    value === null ||
-    value === undefined ||
-    typeof value === 'boolean' ||
-    typeof value === 'number' ||
-    typeof value === 'string'
-  );
-}
-
 export interface SuspensionHandlerParams {
   suspension: WorkflowSuspension;
   world: World;
@@ -144,7 +130,12 @@ export interface SuspensionHandlerResult {
    * durably creating the user's hooks doesn't count as runtime overhead.
    */
   hookCreationMs: number;
-  /** Whether every newly serialized step input was passive retained-VM data. */
+  /**
+   * Whether serializing this suspension's new step inputs was passive (did
+   * not execute workflow-owned code such as getters, proxy traps, or custom
+   * serializers). `false` means the retained VM may have diverged from what
+   * a cold replay would compute, so the caller must demote to replay.
+   */
   retainedStepInputsSafe: boolean;
 }
 
@@ -547,20 +538,15 @@ export async function handleSuspension({
 
   // 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 will execute workflow code (getters, hooks,
-  // patched prototype members) — side effects a cold replay would not repeat.
-  // For now only primitive arguments are provably passive (serializing them
-  // executes no code at all); a follow-up widens this to plain data and the
-  // standard built-ins. If any input in the batch is not provably passive,
-  // the caller demotes the session so the side effects land in a VM that is
-  // about to be discarded, exactly like the pre-retention runtime.
-  const retainedStepInputsSafe = stepItems.every(
-    (item) =>
-      !stepsNeedingCreation.has(item.correlationId) ||
-      (item.thisVal === undefined &&
-        item.closureVars === undefined &&
-        item.args.every(isPrimitiveStepArgument))
-  );
+  // whether that serialization *executed* workflow code (getters, proxy
+  // traps, custom serializers) — side effects a cold replay would not
+  // repeat, since a replay skips dehydration for already-recorded steps.
+  // The hardened serializer records exactly that into this sink (see
+  // ../serialization/hardened.ts); when any input in the batch records an
+  // execution, the caller demotes the session so the side effects land in a
+  // VM that is about to be discarded, exactly like the pre-retention
+  // runtime.
+  const guestCodeStats: GuestCodeStats = { executions: [] };
 
   // Lazy inline start: defer the step_created write for up to
   // `getMaxInlineSteps()` steps the caller will run inline (in parallel). Each
@@ -595,6 +581,10 @@ export async function handleSuspension({
     if (stepsNeedingCreation.has(queueItem.correlationId)) {
       ops.push(
         (async () => {
+          // Per-step sink, merged below: the dehydrate wrapper emits span
+          // 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,
@@ -605,8 +595,10 @@ export async function handleSuspension({
             encryptionKey,
             suspension.globalThis,
             false,
-            compression
+            compression,
+            stepGuestCode
           );
+          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
           // this input, and the world creates the step (entity + synthetic
@@ -752,6 +744,20 @@ export async function handleSuspension({
   // step_created and re-dispatches, and recovers the run instead of orphaning it.
   await settlePhase(ops);
 
+  // The step-input dehydrations above have settled, so the sink is final.
+  const retainedStepInputsSafe = guestCodeStats.executions.length === 0;
+  if (!retainedStepInputsSafe) {
+    runtimeLogger.debug(
+      'Serializing step inputs executed workflow code; falling back to replay instead of retaining the VM',
+      {
+        workflowRunId: runId,
+        executions: guestCodeStats.executions
+          .slice(0, 5)
+          .map((e) => (e.detail ? `${e.kind}(${e.detail})` : e.kind)),
+      }
+    );
+  }
+
   // Rebuild the inline batch in deterministic order. `lazyInlineCorrelationIds`
   // is a Set seeded from the ordered first-N slice, so iterating it preserves
   // stepItems order; every id in it was set by the lazy branch above.