feat(core): side-effect-free serialization of workflow VM values (#3257)

* feat(core): side-effect-free serialization of workflow VM values

Serialization runs on the host but inspects values constructed inside the
node:vm sandbox, so ordinary dynamic operations dispatch into the sandbox
realm and execute workflow code: `value.toISOString()`, `Array.from(map)`,
`Object.prototype.toString` (via Symbol.toStringTag), `.source`/`.flags`,
`.href`, view `.buffer`/`.byteOffset`/`.byteLength`, and error
`.message`/`.stack`/`.cause` reads.

That is a determinism hazard. A payload is serialized exactly once and is
never re-serialized on replay, so any workflow-visible side effect it
triggers exists only on the live path — a patched `Date.prototype.toISOString`
that consumes a seeded `Math.random()` draw, for example, shifts every
subsequent draw and diverges from replay.

This makes serialization side-effect free where the data allows it, and
observable where it does not:

- Classification uses engine brand checks (node:util types, internal-slot
  probes) instead of `instanceof global.X` and Object.prototype.toString, so
  it is immune to Symbol.hasInstance, reassigned sandbox globals, and
  Symbol.toStringTag spoofs. An unbranded value claiming a brand-decided tag
  is now classified as a plain object instead of being routed into an
  extractor that requires the real internal slot (unhardened devalue crashes
  on that input).
- Extraction goes through intrinsics captured at module load — host boot,
  before any workflow bundle runs — invoked with explicit receivers.
  Internal slots are realm-agnostic, so host intrinsics read VM-realm
  objects without touching the sandbox's patchable prototypes.
- Property access reads through descriptors, so plain data never invokes
  anything.

Where workflow code must run because the data lives behind it — getters,
proxies, custom [WORKFLOW_SERIALIZE] methods, toString() on
toStringTag-branded objects like Temporal polyfills — the execution is
preserved for compatibility and recorded in a new `CodecOptions.guestCodeStats`
sink, surfaced as workflow.serialization.guest_code_{executions,details} span
attributes. Consumers that retain a VM across steps can treat a non-empty
report as "serialization may have perturbed VM state".

Engine-provided accessors are deliberately not reported: V8 defines `stack`
as an own accessor on every Error instance, so reporting it would flag every
serialized error. Nativeness is decided with the captured host
Function.prototype.toString; the bound-function caveat is documented in
hardened.ts.

Requires devalue 5.9.0 for the pluggable `operations` option.

* chore: shorten changeset

* fix(core): close review gaps in hardened serialization

Five correctness fixes, all with repros:

- Callable proxies were treated as engine accessors. V8 returns
  `function () { [native code] }` from Function.prototype.toString for a
  proxy around a function rather than throwing, so a proxy-wrapped getter
  was cached as engine-provided and invoked unreported. Gate on
  types.isProxy first.

- Host builtins implemented in JavaScript were reported as workflow code.
  Node's DOMException.prototype.message/name are ordinary functions, so
  the nativeness test failed and every serialized DOMException reported
  two getter executions. They belong to the *host* realm, though, and
  workflow code cannot author a host-realm function — so provenance is
  now decided by nativeness OR host-realm `Function.prototype`, which are
  disjoint and together cover both cases (V8 installs `stack` per realm,
  so a VM error's getter is native but VM-realm).

- The extraReducers at the two VM call sites were still unhardened, and
  they run on every value the earlier reducers do not claim — which is
  exactly where the report has to be complete. `instanceof
  global.ReadableStream/WritableStream/Request/Response` consulted
  Symbol.hasInstance on the sandbox class (14 invocations for an ordinary
  payload once the classes are patched), and AbortController's guard did
  a bare `value.signal` read, so a non-enumerable `signal` getter ran
  with an empty report. All five now walk the prototype chain and read
  through descriptors.

- `__closureVarsFn` was invoked unreported on a purity argument that
  nothing checked: the property is reachable from workflow code, which
  can replace the compiler-generated function. step.ts now registers the
  generated function as trusted when it builds the proxy, so provenance
  is verified rather than assumed, and an unrecognized function is
  reported.

- The URL/URLSearchParams test patched prototypes of *host* classes
  injected into the sandbox, mutating them for the rest of the worker
  process. Restored in a finally.

Also, per review:

- `dehydrateStepArguments` / `dehydrateWorkflowReturnValue` take an
  optional GuestCodeStats out-param, so a retained-VM gate can consume
  the report instead of it being spent on span attributes. The
  report-completeness tests use it to exercise the real dehydrate path.

- Every intrinsic capture is now optional. The table is built at module
  scope, so a missing member was an import-time crash of @workflow/core
  rather than a degraded path; only SharedArrayBuffer was guarded, while
  URLSearchParams.prototype.size (Node 19.8+) and the WHATWG classes were
  assumed. Absent captures now make the corresponding reducer decline to
  match.

- Documented that recording is not prevention (a recorded getter calling
  Math.random() still advances the run's seeded PRNG), and that a
  `{ kind: 'proxy' }` report implies a silent shape change (a proxied Map
  serializes as a plain object).

- Parity coverage extended to DataView, boxed primitives, null-prototype
  objects, setter-only properties, DOMException, AggregateError, an
  accessor-valued Symbol.toStringTag, both RetryableError retryAfter
  paths, and a WORKFLOW_SERIALIZE class instance.

* fix(core): keep identifying proxied host classes

Every Next.js e2e job failed on the two webhook tests: the hook POST
returned 404 because `resumeWebhook` could not serialize its step return
value ("Cannot stringify arbitrary non-POJOs"), so no hook was ever
registered.

The value was a `NextRequest`, which Next.js hands over as a **Proxy**.
`isInstanceOfPrototype` rejected proxies outright, so the Request reducer
answered "not a Request" and devalue fell through to the POJO check. The
reasoning behind rejecting them — that proxied built-ins were never
serializable, because internal-slot reads throw on a proxy receiver — is
true for `Map`/`Date`/`URL`, whose reducers read internal slots, but not
for `Request`/`Response`/streams, whose reducers read ordinary
properties. Next's proxy forwards those with the target as receiver, so
they serialized fine before this PR.

Identification now walks through proxies, matching `instanceof`, and
records the traps rather than suppressing the answer. The three reducers
that do read internal slots (URL, URLSearchParams, Headers) fall back to
the dynamic read when the value is a proxy, so their behavior is exactly
what it was before — including throwing for a bare proxy over a built-in,
which threw before too.

Verified against the real thing: the full nextjs-turbopack e2e suite
(135 tests) passes locally, having reproduced the failure first and
confirmed a reverted `serialization.ts` fixed it.

The regression test uses a receiver-correcting proxy, which is what makes
NextRequest work in practice; a comment records that a bare
`new Proxy(request, {})` throws on undici's private slots with or without
this change.

* fix(core): state what the closure-fn mark proves, and correct stale docs

- `isInstanceOfPrototype`'s JSDoc still described the behavior removed in
  8bc462f (proxies rejected without firing traps), which is the opposite
  of what it now does.

- The `__closureVarsFn` provenance check proves the function was passed to
  `useStep`, not that this package generated it: `useStep` is published on
  the sandbox global, so workflow code can call it with a function of its
  own and have it marked. Renamed `registerTrustedFunction` /
  `isTrustedFunction` to `markUseStepClosureFn` / `isUseStepClosureFn` so
  the name states the boundary, and documented the laundering caveat
  alongside the existing ones. Marking still earns its keep — reporting
  every step that captures a variable would bury the signal — and closing
  the gap properly needs a compiler-emitted marker, which is a compiler
  change.

- Added the missing coverage for both sides of that check: an unmarked
  `__closureVarsFn` is invoked and reported, a marked one is invoked and
  not.

- `guestCodeStats` was documented as something a retained-VM gate consumes,
  but no runtime caller passes a sink; the executions reach telemetry from
  every dehydrate path regardless. Reworded both docs to say that, so the
  out-param is not mistaken for wiring that already exists.

Author: TooTallNate

.changeset/hardened-vm-serialization.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Serializing values built inside the `node:vm` workflow VM no longer executes workflow code, using engine brand checks and host intrinsics captured at boot.

packages/core/package.json

@@ -102,7 +102,7 @@
     "@workflow/world-local": "workspace:*",
     "@workflow/world-vercel": "workspace:*",
     "debug": "4.4.3",
-    "devalue": "5.8.1",
+    "devalue": "5.9.0",
     "ms": "2.1.3",
     "nanoid": "5.1.6",
     "seedrandom": "3.0.5",

packages/core/src/serialization.ts

@@ -62,6 +62,11 @@ import {
   isEncrypted,
   peekFormatPrefix,
 } from './serialization/format.js';
+import {
+  type GuestCodeStats,
+  isInstanceOfPrototype,
+  readProperty,
+} from './serialization/hardened.js';
 import {
   getClassReducers,
   getClassRevivers,
@@ -233,6 +238,34 @@ async function recordCompression(
   }
 }
 
+/**
+ * Emits OTel span attributes for workflow (guest) code executions that the
+ * hardened serializer could not avoid (getters, proxies, custom
+ * serializers). No-ops when serialization was fully side-effect free —
+ * the common case. Same never-break-the-data-path contract as
+ * `recordCompression` above.
+ */
+async function recordGuestCodeExecutions(stats: GuestCodeStats): Promise<void> {
+  if (stats.executions.length === 0) return;
+  try {
+    const span = await getActiveSpan();
+    if (!span) return;
+    const details = [
+      ...new Set(
+        stats.executions.map((e) =>
+          e.detail ? `${e.kind} (${e.detail})` : e.kind
+        )
+      ),
+    ];
+    span.setAttributes({
+      ...Attr.SerializationGuestCodeExecutions(stats.executions.length),
+      ...Attr.SerializationGuestCodeDetails(details),
+    });
+  } catch {
+    // ignore telemetry failures
+  }
+}
+
 export function getSerializeStream(
   reducers: Partial<Reducers>,
   cryptoKey: EncryptionKeyParam
@@ -1568,15 +1601,23 @@ function getAllBaseReducers(
     // Request and Response reducers are mode-specific and added by
     // getExternalReducers / getWorkflowReducers / getStepReducers below.
     Request: (value) => {
-      if (!(value instanceof global.Request)) return false;
+      // Chain walk rather than `instanceof global.Request`: see the
+      // ReadableStream reducer in getWorkflowReducers for why. Reads go
+      // through descriptors so a getter cannot run unreported.
+      if (
+        !isInstanceOfPrototype(value, getHostClassPrototype(global, 'Request'))
+      )
+        return false;
       const data: SerializableSpecial['Request'] = {
-        method: value.method,
-        url: value.url,
-        headers: value.headers,
-        body: value.body,
-        duplex: value.duplex,
+        method: readProperty(value, 'method') as string,
+        url: readProperty(value, 'url') as string,
+        headers: readProperty(value, 'headers') as Headers,
+        body: readProperty(value, 'body') as ReadableStream | null,
+        duplex: readProperty(value, 'duplex') as 'half',
       };
-      const responseWritable = value[WEBHOOK_RESPONSE_WRITABLE];
+      const responseWritable = readProperty(value, WEBHOOK_RESPONSE_WRITABLE) as
+        | WritableStream<Response>
+        | undefined;
       if (responseWritable) {
         data.responseWritable = responseWritable;
       }
@@ -1589,25 +1630,30 @@ function getAllBaseReducers(
       // Plain non-aborted native signals are intentionally dropped (would
       // mint stream infra for every Request, including the auto-generated
       // signal on `new Request(url)`).
+      const signal = readProperty(value, 'signal');
       if (
-        value.signal &&
-        (value.signal.aborted ||
-          (value.signal as AbortInternals)[ABORT_STREAM_NAME])
+        signal &&
+        (readProperty(signal, 'aborted') ||
+          readProperty(signal, ABORT_STREAM_NAME))
       ) {
-        data.signal = value.signal;
+        data.signal = signal as AbortSignal;
       }
       return data;
     },
     Response: (value) => {
-      if (!(value instanceof global.Response)) return false;
+      // See the Request reducer above.
+      if (
+        !isInstanceOfPrototype(value, getHostClassPrototype(global, 'Response'))
+      )
+        return false;
       return {
-        type: value.type,
-        url: value.url,
-        status: value.status,
-        statusText: value.statusText,
-        headers: value.headers,
-        body: value.body,
-        redirected: value.redirected,
+        type: readProperty(value, 'type') as Response['type'],
+        url: readProperty(value, 'url') as string,
+        status: readProperty(value, 'status') as number,
+        statusText: readProperty(value, 'statusText') as string,
+        headers: readProperty(value, 'headers') as Headers,
+        body: readProperty(value, 'body') as ReadableStream | null,
+        redirected: readProperty(value, 'redirected') as boolean,
       };
     },
   };
@@ -1937,6 +1983,51 @@ export function getExternalReducers(
  * @param global
  * @returns
  */
+/**
+ * Prototypes used for brand-style identification in the reducers below.
+ *
+ * The stream and abort classes are host classes injected into the sandbox, so
+ * instances carry the host prototype in their chain and a chain walk
+ * identifies them without consulting `Symbol.hasInstance` on the sandbox
+ * class. `undefined` when the runtime lacks the class, in which case
+ * identification falls back to the infrastructure symbols alone.
+ */
+function getStreamPrototype(
+  global: Record<string, any>,
+  kind: 'Readable' | 'Writable'
+): object | undefined {
+  return getHostClassPrototype(global, `${kind}Stream`);
+}
+
+/**
+ * The prototype of a host class that may also be injected into the sandbox.
+ * Prefers the sandbox binding (the same host class object in practice) and
+ * falls back to the host's own, so a chain walk identifies instances from
+ * either realm without consulting `Symbol.hasInstance`.
+ */
+function getHostClassPrototype(
+  global: Record<string, any>,
+  name: string
+): object | undefined {
+  const ctor =
+    global[name] ?? (globalThis as Record<string, any>)[name] ?? undefined;
+  return typeof ctor === 'function' ? ctor.prototype : undefined;
+}
+
+function getAbortControllerPrototype(
+  global: Record<string, any>
+): object | undefined {
+  const ctor = global.AbortController ?? globalThis.AbortController;
+  return typeof ctor === 'function' ? ctor.prototype : undefined;
+}
+
+function getAbortSignalPrototype(
+  global: Record<string, any>
+): object | undefined {
+  const ctor = global.AbortSignal ?? globalThis.AbortSignal;
+  return typeof ctor === 'function' ? ctor.prototype : undefined;
+}
+
 export function getWorkflowReducers(
   global: Record<string, any> = globalThis
 ): Partial<Reducers> {
@@ -1946,45 +2037,68 @@ export function getWorkflowReducers(
     // Readable/Writable streams from within the workflow execution environment
     // are simply "handles" that can be passed around to other steps.
     ReadableStream: (value) => {
-      if (!(value instanceof global.ReadableStream)) return false;
+      // Walk the prototype chain instead of `instanceof global.ReadableStream`:
+      // the class is host-provided (injected into the sandbox), so its
+      // prototype is in the chain of both real streams and the
+      // `Object.create(ReadableStream.prototype)` handles used for request
+      // bodies — but a chain walk never consults `Symbol.hasInstance`, which
+      // the sandbox can define and which ran for every value the earlier
+      // reducers did not claim. Reads below go through descriptors so a
+      // getter on a step argument cannot run unreported.
+      if (!isInstanceOfPrototype(value, getStreamPrototype(global, 'Readable')))
+        return false;
 
       // Check if this is a fake stream storing BodyInit from Request/Response constructor
-      const bodyInit = value[BODY_INIT_SYMBOL];
+      const bodyInit = readProperty(value, BODY_INIT_SYMBOL);
       if (bodyInit !== undefined) {
         // This is a fake stream - serialize the BodyInit directly
         // devalue will handle serializing strings, Uint8Array, etc.
         return { bodyInit };
       }
 
-      const name = value[STREAM_NAME_SYMBOL];
+      const name = readProperty(value, STREAM_NAME_SYMBOL) as string;
       if (!name) {
         throw new WorkflowRuntimeError('ReadableStream `name` is not set');
       }
-      const s: SerializableSpecial['ReadableStream'] = { name };
-      const type = value[STREAM_TYPE_SYMBOL];
+      const s: Extract<
+        SerializableSpecial['ReadableStream'],
+        { name: string }
+      > = { name };
+      const type = readProperty(value, STREAM_TYPE_SYMBOL) as
+        | 'bytes'
+        | undefined;
       if (type) s.type = type;
-      const framing: ByteStreamFraming | undefined =
-        value[STREAM_FRAMING_SYMBOL];
+      const framing = readProperty(value, STREAM_FRAMING_SYMBOL) as
+        | ByteStreamFraming
+        | undefined;
       if (framing) s.framing = framing;
       return s;
     },
     WritableStream: (value) => {
-      if (!(value instanceof global.WritableStream)) return false;
-      const name = value[STREAM_NAME_SYMBOL];
+      // See the ReadableStream reducer above for why this walks the chain.
+      if (!isInstanceOfPrototype(value, getStreamPrototype(global, 'Writable')))
+        return false;
+      const name = readProperty(value, STREAM_NAME_SYMBOL) as string;
       if (!name) {
         throw new WorkflowRuntimeError('WritableStream `name` is not set');
       }
       const s: SerializableSpecial['WritableStream'] = { name };
       // When the handle was forwarded from another run (parent → child
       // via `start()`), preserve the foreign runId so the step-side
       // reviver opens the writable against the original stream.
-      const foreignRunId = value[STREAM_SERVER_RUN_ID_SYMBOL];
+      const foreignRunId = readProperty(value, STREAM_SERVER_RUN_ID_SYMBOL);
       if (typeof foreignRunId === 'string') s.runId = foreignRunId;
-      const foreignDeploymentId = value[STREAM_SERVER_DEPLOYMENT_ID_SYMBOL];
+      const foreignDeploymentId = readProperty(
+        value,
+        STREAM_SERVER_DEPLOYMENT_ID_SYMBOL
+      );
       if (typeof foreignDeploymentId === 'string') {
         s.deploymentId = foreignDeploymentId;
       }
-      const foreignPublicKey = value[STREAM_SERVER_PUBLIC_KEY_SYMBOL];
+      const foreignPublicKey = readProperty(
+        value,
+        STREAM_SERVER_PUBLIC_KEY_SYMBOL
+      );
       if (typeof foreignPublicKey === 'string') {
         s.encryptionPublicKey = foreignPublicKey;
       }
@@ -1996,26 +2110,42 @@ export function getWorkflowReducers(
     // is a plain object (not a class), so instanceof checks won't work for signals.
     // Detect instances by the presence of the ABORT_STREAM_NAME symbol instead.
     AbortController: (value) => {
-      if (!value || !value.signal) return false;
+      // `value.signal` was a bare read, so a `signal` getter on any object
+      // reaching this reducer executed unreported. Read through descriptors
+      // and gate on the infrastructure symbol / prototype first.
+      if (value === null || typeof value !== 'object') return false;
       const holder = value as AbortController & AbortHolder;
-      const hasAbortSymbol =
-        holder[ABORT_STREAM_NAME] ?? holder.signal?.[ABORT_STREAM_NAME];
-      const isNativeAbortController =
-        global.AbortController &&
-        typeof global.AbortController === 'function' &&
-        value instanceof global.AbortController;
-      if (!hasAbortSymbol && !isNativeAbortController) return false;
-      return reduceAbortBySymbol(value.signal, holder);
+      const ownSymbol = readProperty(value, ABORT_STREAM_NAME);
+      const isNativeAbortController = isInstanceOfPrototype(
+        value,
+        getAbortControllerPrototype(global)
+      );
+      if (ownSymbol === undefined && !isNativeAbortController) {
+        // Not ours and not a native controller — but a foreign controller
+        // may still carry the symbol on its signal.
+        const maybeSignal = readProperty(value, 'signal');
+        if (
+          maybeSignal === null ||
+          typeof maybeSignal !== 'object' ||
+          readProperty(maybeSignal, ABORT_STREAM_NAME) === undefined
+        ) {
+          return false;
+        }
+      }
+      const signal = readProperty(value, 'signal');
+      if (!signal) return false;
+      return reduceAbortBySymbol(signal as AbortSignal, holder);
     },
     AbortSignal: (value) => {
-      const signal = value as (AbortSignal & AbortInternals) | undefined;
-      const hasAbortSymbol = signal?.[ABORT_STREAM_NAME];
-      const isNativeAbortSignal =
-        global.AbortSignal &&
-        typeof global.AbortSignal === 'function' &&
-        value instanceof global.AbortSignal;
+      if (value === null || typeof value !== 'object') return false;
+      const hasAbortSymbol =
+        readProperty(value, ABORT_STREAM_NAME) !== undefined;
+      const isNativeAbortSignal = isInstanceOfPrototype(
+        value,
+        getAbortSignalPrototype(global)
+      );
       if (!hasAbortSymbol && !isNativeAbortSignal) return false;
-      return reduceAbortBySymbol(value, value as AbortHolder);
+      return reduceAbortBySymbol(value as AbortSignal, value as AbortHolder);
     },
   };
 }
@@ -3420,21 +3550,34 @@ export async function dehydrateWorkflowReturnValue(
   key: PayloadKey | undefined,
   global: Record<string, any> = globalThis,
   v1Compat = false,
-  compression = false
+  compression = false,
+  /**
+   * Optional sink receiving every workflow-code execution serialization could
+   * not avoid, for callers that need them programmatically (e.g. a
+   * retained-VM gate deciding whether the VM is still reusable). The
+   * executions are emitted as span attributes either way, so omitting this
+   * loses nothing observability-wise. No runtime caller passes one yet.
+   */
+  guestCodeStatsOut?: GuestCodeStats
 ): Promise<Uint8Array | unknown> {
   if (v1Compat) {
     const str = stringify(value, getWorkflowReducers(global));
     return revive(str);
   }
   try {
     const compressionStats: CompressionStats = {};
+    const guestCodeStats: GuestCodeStats = guestCodeStatsOut ?? {
+      executions: [],
+    };
     const result = await stepModule.serialize(value, key, {
       global,
       extraReducers: getStreamAndRequestReducers(getWorkflowReducers(global)),
       compression,
       compressionStats,
+      guestCodeStats,
     });
     await recordCompression(compressionStats, 'serialize');
+    await recordGuestCodeExecutions(guestCodeStats);
     return result;
   } catch (error) {
     const cause = unwrapSerializationCause(error);
@@ -3483,21 +3626,28 @@ export async function dehydrateStepArguments(
   key: PayloadKey | undefined,
   global: Record<string, any> = globalThis,
   v1Compat = false,
-  compression = false
+  compression = false,
+  /** See `dehydrateWorkflowReturnValue`. */
+  guestCodeStatsOut?: GuestCodeStats
 ): Promise<Uint8Array | unknown> {
   if (v1Compat) {
     const str = stringify(value, getWorkflowReducers(global));
     return revive(str);
   }
   try {
     const compressionStats: CompressionStats = {};
+    const guestCodeStats: GuestCodeStats = guestCodeStatsOut ?? {
+      executions: [],
+    };
     const result = await stepModule.serialize(value, key, {
       global,
       extraReducers: getStreamAndRequestReducers(getWorkflowReducers(global)),
       compression,
       compressionStats,
+      guestCodeStats,
     });
     await recordCompression(compressionStats, 'serialize');
+    await recordGuestCodeExecutions(guestCodeStats);
     return result;
   } catch (error) {
     const cause = unwrapSerializationCause(error);