vercel · NathanColosimo · Aug 15, 2026
@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Retain workflow VMs across safe step boundaries that also contain open hooks.
@@ -5,7 +5,7 @@
 import { withResolvers } from '@workflow/utils';
 import type { WorldCapabilities } from '@workflow/world';
 import type { EventsConsumer } from './events-consumer.js';
-import type { QueueItem } from './global.js';
+import { type QueueItem, WorkflowSuspension } from './global.js';
 import type { ReplayPayloadCache } from './replay-payload-cache.js';
 import type { Serializable } from './schemas.js';
 import type { DecryptionKey } from './serialization/encryption.js';
@@ -137,12 +137,7 @@ export interface WorkflowOrchestratorContext {
   globalThis: typeof globalThis;
   /**
    * Increments when a suspension is accepted and on every retained-session
-   * resume. STEP suspension signals capture it when scheduled and no-op if
-   * it moved (see step.ts) — this drops same-boundary sibling signals and
-   * timers queued at boundary N that would fire after the session resumed
-   * into boundary N+1. Sleep/hook/attribute signals are intentionally
-   * unguarded: their presence makes the boundary unretainable, so a late
-   * signal correctly demotes the session (workflow.ts `onWorkflowError`).
+   * resume. Step and hook signals capture it so stale signals no-op.
    */
   suspensionGeneration: number;
   eventsConsumer: EventsConsumer;
@@ -793,3 +788,16 @@ export function scheduleWhenIdle(
   };
   setTimeout(check, 0);
 }
+
+/** Schedule a generation-guarded suspension after deliveries settle. */
+export function scheduleWorkflowSuspension(
+  ctx: WorkflowOrchestratorContext
+): void {
+  const generation = ctx.suspensionGeneration;
+  scheduleWhenIdle(ctx, () => {
+    if (generation !== ctx.suspensionGeneration) return;
+    ctx.onWorkflowError(
+      new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis)
+    );
+  });
+}
@@ -5,7 +5,15 @@ import {
   slotToEventId,
   type WorkflowRun,
 } from '@workflow/world';
-import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+  afterEach,
+  assert,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  vi,
+} from 'vitest';
 
 // Spy on VM-context construction while preserving the real implementation, so
 // we can prove the retained path builds ONE VM for a whole run instead of one
@@ -21,9 +29,11 @@ const { registerSerializationClass } = await import('./class-serialization.js');
 const { registerStepFunction } = await import('./private.js');
 const { setWorld } = await import('./runtime/world.js');
 const { workflowEntrypoint } = await import('./runtime.js');
-const { dehydrateWorkflowArguments, hydrateWorkflowReturnValue } = await import(
-  './serialization.js'
-);
+const {
+  dehydrateStepReturnValue,
+  dehydrateWorkflowArguments,
+  hydrateWorkflowReturnValue,
+} = await import('./serialization.js');
 
 vi.mock('@vercel/functions', () => ({
   waitUntil: vi.fn((p: Promise<unknown>) => {
@@ -101,6 +111,39 @@ const parallelBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_ST
   }
   globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
 
+// An open hook and a step both signal the first suspension. The step wins the
+// Promise.race, then a second step advances the same inline replay loop. A
+// retained session must absorb the losing hook's same-generation suspension
+// signal and keep the one VM alive across both step completions.
+const openHookRaceWorkflow = `const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")];
+  const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  async function workflow() {
+    const hook = createHook({ token: "retained-hook" });
+    const a = await Promise.race([hook.then(() => 999), s1()]);
+    const b = await s2();
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+// Hook metadata is serialized at the same suspension boundary as step input.
+// A getter mutating workflow state must demote the retained VM just like an
+// unsafe step argument, because a cold replay skips serialization after the
+// hook_created event exists.
+const impureHookMetadataWorkflow = `const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")];
+  const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const echo = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_echo");
+  async function workflow() {
+    let counter = 0;
+    createHook({
+      token: "unsafe-retained-hook",
+      metadata: { get value() { counter++; return 1; } },
+    });
+    await s1();
+    return await echo(counter);
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
 // A parallel batch where one sibling's input is unsafe must serialize the
 // WHOLE batch through the ordinary VM path (all-or-nothing) and demote.
 const mixedBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
@@ -196,12 +239,17 @@ registerStepFunction('r_echo', async (value) => value);
 // Drive the full workflow handler over a stateful (dynamic) event log so the
 // inline loop makes real progress across its own writes, exactly like a World.
 // Non-turbo (no runInput, attempt 2) to keep the path simple and deterministic.
+type DriveMode =
+  | { type: 'normal' }
+  | { type: 'fail-event'; eventType: Event['eventType'] }
+  | { type: 'inject-hook' };
+
 async function drive(
   runId: string,
   workflowCode = twoStepWorkflow,
-  options: { failEventTypeOnce?: string } = {}
+  initialMode: DriveMode = { type: 'normal' }
 ) {
-  let { failEventTypeOnce } = options;
+  let mode = initialMode;
   const run: WorkflowRun = {
     runId,
     workflowName: 'workflow',
@@ -217,8 +265,8 @@ async function drive(
   let seq = 0;
 
   const eventsCreate = vi.fn(async (_runId: string, data: any) => {
-    if (data.eventType === failEventTypeOnce) {
-      failEventTypeOnce = undefined;
+    if (mode.type === 'fail-event' && data.eventType === mode.eventType) {
+      mode = { type: 'normal' };
       throw new PreconditionFailedError('stale snapshot (test-injected)');
     }
     createdEvents.push(data);
@@ -232,6 +280,29 @@ async function drive(
       ...data,
     } as Event;
     events.push(event);
+    if (data.eventType === 'step_started' && mode.type === 'inject-hook') {
+      mode = { type: 'normal' };
+      const hookCreated = events.find(
+        (candidate) => candidate.eventType === 'hook_created'
+      );
+      assert(hookCreated, 'expected hook_created before step');
+      events.push({
+        eventId: slotToEventId(++seq),
+        runId,
+        eventType: 'hook_received',
+        specVersion: SPEC_VERSION_CURRENT,
+        correlationId: hookCreated.correlationId,
+        eventData: {
+          token: hookCreated.eventData.token,
+          payload: await dehydrateStepReturnValue(
+            { source: 'external-hook' },
+            runId,
+            undefined
+          ),
+        },
+        createdAt: new Date(),
+      });
+    }
     // step_started returns a running step entity so executeStep proceeds to
     // run the body and write step_completed.
     if (data.eventType === 'step_started') {
@@ -358,6 +429,36 @@ describe('retained VM through the inline replay loop', () => {
     expect(vmBuilds).toBe(1);
   });
 
+  it('retains when hook_received extends the log between step_started and step_completed', async () => {
+    const { vmBuilds, result } = await drive(
+      'wrun_retained_hook_between_step_events',
+      openHookRaceWorkflow,
+      { type: 'inject-hook' }
+    );
+    // The hook event precedes step_completed in the durable log, so it wins
+    // the race on resume while the already-started step still completes.
+    expect(result).toBe(1019);
+    expect(vmBuilds).toBe(1);
+  });
+
+  it('matches cold replay when hook metadata serialization mutates workflow state', async () => {
+    process.env.WORKFLOW_RETAINED_VM = '0';
+    const off = await drive(
+      'wrun_impure_hook_metadata_off',
+      impureHookMetadataWorkflow
+    );
+    createContextSpy.mockClear();
+    delete process.env.WORKFLOW_RETAINED_VM;
+
+    const on = await drive(
+      'wrun_impure_hook_metadata_on',
+      impureHookMetadataWorkflow
+    );
+    expect(off.result).toBe(0);
+    expect(on.result).toBe(0);
+    expect(on.vmBuilds).toBeGreaterThan(1);
+  });
+
   it('demotes retention when any input in a parallel batch is unsafe', async () => {
     const { vmBuilds, result } = await drive(
       'wrun_retained_mixed_batch',
@@ -376,7 +477,7 @@ describe('retained VM through the inline replay loop', () => {
     const { vmBuilds, result } = await drive(
       'wrun_retained_412_restart',
       twoStepWorkflow,
-      { failEventTypeOnce: 'run_completed' }
+      { type: 'fail-event', eventType: 'run_completed' }
     );
     expect(result).toBe(30);
     expect(vmBuilds).toBeGreaterThan(1);

@@ -486,11 +486,8 @@ function rootRunIdFrom(
  * `wait_completed`, which the wait timer can resolve with
  * `wait_completed`).
  *
- * This gates VM retention, the inline-delta fast path, and turbo's forced
- * optimistic start. A terminal-step delta can omit an event appended
- * concurrently after that write. With no open hook or wait, only cancellation
- * can do so, and observing it one replay late is safe because the next entity
- * write is rejected.
+ * Open waits block VM retention and inline deltas. Open hooks and waits block
+ * turbo's optimistic start; hooks also require a `step_started` claim.
  *
  * Step-body `attr_set` writes are NOT a concern: they land before the
  * step's terminal write and are therefore already inside the returned
@@ -567,47 +564,6 @@ function replayEventDeploymentId(event: Event): string | undefined {
   return undefined;
 }
 
-/**
- * The whole retention predicate: keep the session only for a pure step
- * boundary (every suspension item is a step — any other item type, present
- * or future, is unretainable by default) whose new step inputs serialized
- * without executing workflow code, with no out-of-band continuation source:
- * attributes require replay; hooks and waits can wake another invocation.
- * `WORKFLOW_RETAINED_VM=0` disables retention entirely.
- *
- * The open hook/wait scan is O(events), so it is taken through a lazy getter
- * and consulted last, after every cheap check has passed.
- *
- * INVARIANT this predicate leans on: every suspension signaler that does NOT
- * carry the step-consumer generation guard (sleep, hook, attribute — see
- * `suspensionGeneration` in private.ts) must be unretainable here, either via
- * a non-step queue item or the open hook/wait scan. A new signaler that
- * satisfies neither would let a stale signal be accepted as a fresh
- * suspension on a resumed session.
- *
- * Quiescence assumes workflow code stays inside the sandbox's determinism
- * contract. Escaping to the host realm (e.g. recovering the host `Function`
- * constructor from an exposed host class to schedule real timers) makes a
- * workflow nondeterministic under ordinary replay too, and is not defended
- * here.
- */
-function canRetainWorkflowSession(
-  suspension: WorkflowSuspension,
-  stepInputsSafe: boolean,
-  openHookWait: { value: ReturnType<typeof openHookAndWaitState> }
-): boolean {
-  if (
-    !isVmRetentionEnabled() ||
-    !stepInputsSafe ||
-    suspension.steps.length === 0 ||
-    !suspension.steps.every((item) => item.type === 'step')
-  ) {
-    return false;
-  }
-  const { openHook, openWait } = openHookWait.value;
-  return !openHook && !openWait;
-}
-
 /**
  * Maximum inline-execution duration for a single handler invocation.
  *
@@ -3341,32 +3297,37 @@ export function workflowEntrypoint(
                         }
 
                         // Open hooks/waits in the log as loaded for this
-                        // replay. This suspension's own hook/wait writes are
-                        // NOT in it — they never reach retention anyway,
-                        // because a suspension containing a non-step item
-                        // fails canRetainWorkflowSession's type check before
-                        // the scan is consulted. Computed
-                        // lazily, at most once, and shared between the
-                        // retention decision here and the delta/turbo gates
-                        // below — the attr-detour and hook-conflict paths
-                        // return/continue before the gates and usually
-                        // short-circuit before ever scanning the log.
+                        // replay. Computed lazily, at most once, and shared
+                        // between the retention decision here and the
+                        // delta/turbo gates below — the attr-detour and
+                        // hook-conflict paths return/continue before the
+                        // gates and usually short-circuit before scanning.
                         const openHookWait = once(() => {
                           assert(eventLog.type === 'ready');
                           return openHookAndWaitState(eventLog.events);
                         });
 
-                        // The single retention decision: keep the parked
-                        // session only across a pure step boundary with no
-                        // out-of-band continuation source and provably
-                        // passive step inputs.
                         if (
                           retainedSession &&
-                          !canRetainWorkflowSession(
-                            err,
-                            suspensionResult.retainedStepInputsSafe,
-                            openHookWait
-                          )
+                          (!isVmRetentionEnabled() ||
+                            !suspensionResult.serializationWasPassive ||
+                            err.stepCount === 0 ||
+                            !err.steps.every((item) => {
+                              switch (item.type) {
+                                case 'step':
+                                case 'hook':
+                                  return true;
+                                case 'wait':
+                                case 'attribute':
+                                  return false;
+                                default:
+                                  item satisfies never;
+                                  throw new Error(
+                                    'Unknown workflow suspension item'
+                                  );
+                              }
+                            }) ||
+                            openHookWait.value.openWait)
                         ) {
                           retainedSession = null;
                         }