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

.changeset/retained-serialization-taint.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': minor
+'workflow': minor
+---
+
+Retained-VM boundaries now accept plain data and standard built-ins (`Map`, `Set`, `Date`, `Error`, typed arrays, `URL`, `Headers`, …) as step inputs. Serialization reads through captured intrinsics and reports when it had to execute workflow code (getters, proxies, custom serializers); only those boundaries fall back to ordinary replay.

biome.json

@@ -27,7 +27,8 @@
       "!**/.nitro",
       "!**/.output",
       "!**/.tanstack",
-      "!**/routeTree.gen.ts"
+      "!**/routeTree.gen.ts",
+      "!packages/core/src/vendor"
     ],
     "ignoreUnknown": true
   },

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

@@ -81,7 +81,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`, `Error`, typed arrays, `ArrayBuffer`, `URL`, `Headers`) remain retainable: serialization reads them through captured intrinsics, so patching or polyfilling built-in prototypes does not affect retention. Anything whose serialization executes workflow code — getters, proxies, custom class serializers, step-function closures — falls back to replay for that boundary.
 - Set `0` or `false` to replay from scratch in a fresh VM on every iteration.
 
 ### `WORKFLOW_INLINE_OWNERSHIP`

packages/core/e2e/local-build.test.ts

@@ -79,8 +79,10 @@ async function readFileIfExists(filePath: string): Promise<string | null> {
  * Projects that use the VercelBuildOutputAPIBuilder and produce ESM step bundles.
  */
 const ESM_STEP_BUNDLE_PROJECTS: Record<string, string> = {
+  // Steps are bundled into the combined flow function (there is no separate
+  // step.func in the Build Output API layout).
   example:
-    '.vercel/output/functions/.well-known/workflow/v1/step.func/index.mjs',
+    '.vercel/output/functions/.well-known/workflow/v1/flow.func/index.mjs',
 };
 
 const DIAGNOSTICS_MANIFEST_PATHS: Record<string, string> = {

packages/core/package.json

@@ -85,7 +85,7 @@
     "./_workflow": "./dist/workflow/index.js"
   },
   "scripts": {
-    "build": "genversion --es6 src/version.ts && tsc",
+    "build": "genversion --es6 src/version.ts && tsc && cp src/vendor/devalue/LICENSE src/vendor/devalue/README.md dist/vendor/devalue/",
     "dev": "genversion --es6 src/version.ts && tsc --watch",
     "clean": "tsc --build --clean && rm -rf dist src/version.ts docs ||:",
     "test": "cross-env WORKFLOW_TARGET_WORLD=local vitest run src",
@@ -107,7 +107,6 @@
     "@workflow/world-local": "workspace:*",
     "@workflow/world-vercel": "workspace:*",
     "debug": "4.4.3",
-    "devalue": "5.8.1",
     "ms": "2.1.3",
     "nanoid": "5.1.6",
     "seedrandom": "3.0.5",

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

@@ -100,6 +100,56 @@ const mixedBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP"
   }
   globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
 
+// Map/Date/typed-array arguments serialize through captured host intrinsics
+// (see serialization/operations.ts), so these boundaries stay retainable.
+const builtinArgsWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  async function workflow() {
+    const a = await s1({ index: new Map([["k", 1]]), when: new Date(1234) });
+    const b = await s2(new Uint8Array([1, 2, 3]));
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+// 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 = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  Date.prototype.toTemporalInstant = function () { return "instant"; };
+  Set.prototype.union = function (other) { return new Set([...this, ...other]); };
+  Object.groupBy = function () { return {}; };
+  async function workflow() {
+    const a = await s1(new Date(1234));
+    const b = await s2(new Set([1, 2]));
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+// Replacing a serialization-relevant member (Date.prototype.toISOString)
+// does not affect retention: the Date reducer reads through captured host
+// intrinsics (see serialization/operations.ts), so the patched member never
+// executes and the serialized bytes stay pristine in both modes.
+const patchedDateArgWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  Date.prototype.toISOString = function () { return "patched"; };
+  async function workflow() {
+    const a = await s1(new Date(1234));
+    const b = await s2();
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+const patchedDateStringArgWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  Date.prototype.toISOString = function () { return "patched"; };
+  async function workflow() {
+    const a = await s1(new Date(1234).toISOString());
+    const b = await s2();
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
 // `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");
@@ -319,6 +369,42 @@ 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('retains when the patched type is converted to a string first', async () => {
+    const { vmBuilds, output } = await drive(
+      'wrun_retained_patched_date_string',
+      patchedDateStringArgWorkflow
+    );
+    expect(output).toBeInstanceOf(Uint8Array);
+    expect(vmBuilds).toBe(1);
+  });
+
   it('retains a VM that used the synchronous crypto.subtle.digest', async () => {
     const { vmBuilds, output } = await drive(
       'wrun_retained_digest',

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

@@ -397,3 +397,81 @@ describe('handleSuspension', () => {
     ).toBe(false);
   });
 });
+
+describe('retainedStepInputsSafe (serialization passivity gate)', () => {
+  function stepPending(args: unknown[]) {
+    return new Map([
+      [
+        'step_1',
+        {
+          type: 'step' as const,
+          correlationId: 'step_1',
+          stepName: 'someStep',
+          args,
+        },
+      ],
+    ]);
+  }
+
+  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/'),
+      new Error('recorded failure'),
+    ]);
+    expect(result.retainedStepInputsSafe).toBe(true);
+  });
+
+  it('reports unsafe when serializing an argument executes a getter', async () => {
+    const value: Record<string, unknown> = {};
+    Object.defineProperty(value, 'lazy', {
+      enumerable: true,
+      get: () => 'computed',
+    });
+    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 tainted inputs successfully (bytes are unaffected)', async () => {
+    const value: Record<string, unknown> = {};
+    Object.defineProperty(value, 'lazy', {
+      enumerable: true,
+      get: () => 'computed',
+    });
+    const eventsCreate = vi
+      .fn()
+      .mockImplementation(async (_runId, event) => ({ event }));
+    const world = createWorld(eventsCreate);
+    const result = await handleSuspension({
+      suspension: new WorkflowSuspension(stepPending([value]), globalThis),
+      world,
+      run,
+    });
+    expect(result.retainedStepInputsSafe).toBe(false);
+    // The step is still prepared for execution as usual.
+    expect(
+      result.lazyInlineSteps.length + result.createdStepCorrelationIds.size
+    ).toBeGreaterThan(0);
+  });
+});

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

@@ -26,25 +26,13 @@ import type {
   WorkflowSuspension,
 } from '../global.js';
 import { runtimeLogger } from '../logger.js';
+import type { SerializationPassivityReport } from '../serialization/operations.js';
 import { dehydrateStepArguments } from '../serialization.js';
 import * as Attribute from '../telemetry/semantic-conventions.js';
 import { getAbortStreamIdFromToken } from '../util.js';
 import { getMaxInlineSteps } from './constants.js';
 import { type MutableEventLog, withPreconditionRetry } from './helpers.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.
-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;
@@ -134,7 +122,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;
 }
 
@@ -506,20 +499,18 @@ 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 stringify operations record exactly that into this report
+  // (see ../serialization/operations.ts); when any input in the batch
+  // taints it, 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 passivityReport: SerializationPassivityReport = {
+    tainted: false,
+    reasons: [],
+  };
 
   // Lazy inline start: defer the step_created write for up to
   // `getMaxInlineSteps()` steps the caller will run inline (in parallel). Each
@@ -564,7 +555,8 @@ export async function handleSuspension({
             encryptionKey,
             suspension.globalThis,
             false,
-            compression
+            compression,
+            passivityReport
           );
           // Deferred (lazy) inline step: skip the step_created write — the
           // caller's inline executeStep will send a lazy step_started carrying
@@ -706,6 +698,15 @@ export async function handleSuspension({
   // step_created and re-dispatches, and recovers the run instead of orphaning it.
   await Promise.all(ops);
 
+  // The step-input dehydrations above have settled, so the report is final.
+  const retainedStepInputsSafe = !passivityReport.tainted;
+  if (passivityReport.tainted) {
+    runtimeLogger.debug(
+      'Serializing step inputs executed workflow code; falling back to replay instead of retaining the VM',
+      { workflowRunId: runId, reasons: passivityReport.reasons }
+    );
+  }
+
   // 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.

packages/core/src/serialization-format.test.ts

@@ -1,4 +1,4 @@
-import { stringify } from 'devalue';
+import { stringify } from './vendor/devalue/index.js';
 import { describe, expect, it } from 'vitest';
 import {
   ClassInstanceRef,

packages/core/src/serialization-format.ts

@@ -7,7 +7,7 @@
  */
 
 import { getEventDataRefFields } from '@workflow/world';
-import { parse, unflatten } from 'devalue';
+import { parse, unflatten } from './vendor/devalue/index.js';
 
 // ---------------------------------------------------------------------------
 // Format prefix constants and encoding/decoding

packages/core/src/serialization.test.ts

@@ -4767,7 +4767,7 @@ describe('getDeserializeStream legacy fallback', () => {
   }
 
   it('should parse legacy newline-delimited devalue text', async () => {
-    const { stringify } = await import('devalue');
+    const { stringify } = await import('./vendor/devalue/index.js');
     const encoder = new TextEncoder();
 
     const line1 = `${stringify({ hello: 'world' })}\n`;
@@ -4781,7 +4781,7 @@ describe('getDeserializeStream legacy fallback', () => {
   });
 
   it('should parse legacy single-line chunks', async () => {
-    const { stringify } = await import('devalue');
+    const { stringify } = await import('./vendor/devalue/index.js');
     const encoder = new TextEncoder();
 
     const chunk1 = encoder.encode(`${stringify('hello')}\n`);