diff --git a/.changeset/retain-workflow-vm.md b/.changeset/retain-workflow-vm.md
new file mode 100644
index 0000000000..718282d550
--- /dev/null
+++ b/.changeset/retain-workflow-vm.md
@@ -0,0 +1,6 @@
+---
+'@workflow/core': minor
+'workflow': minor
+---
+
+Retain workflow execution across inline steps within one invocation; `WORKFLOW_RETAINED_VM=0` disables retention. Boundaries whose step arguments are not primitive values fall back to ordinary replay.
diff --git a/docs/content/docs/v5/api-reference/workflow-serde/workflow-serialize.mdx b/docs/content/docs/v5/api-reference/workflow-serde/workflow-serialize.mdx
index 85c5eb8acf..a5380c3188 100644
--- a/docs/content/docs/v5/api-reference/workflow-serde/workflow-serialize.mdx
+++ b/docs/content/docs/v5/api-reference/workflow-serde/workflow-serialize.mdx
@@ -69,6 +69,7 @@ This method runs inside the workflow context and is subject to the same constrai
 - No Node.js-specific APIs (like `fs`, `path`, `crypto`, etc.)
 - No non-deterministic operations (like `Math.random()` or `Date.now()`)
 - No external network calls
+- No side effects on workflow state — the method may run outside deterministic replay, so mutations would not be reconstructed
 
 Keep this method simple and focused on extracting data from the instance.
 </Callout>
diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx
index 471e1da3e4..9007fb7c2a 100644
--- a/docs/content/docs/v5/configuration/runtime-tuning.mdx
+++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx
@@ -76,6 +76,14 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 - Use only when step side effects are idempotent.
 - Set `0` or `false` to force it off, including the first-delivery fast path used by `WORKFLOW_TURBO`.
 
+### `WORKFLOW_RETAINED_VM`
+
+- 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.)
+- Set `0` or `false` to replay from scratch in a fresh VM on every iteration.
+
 ### `WORKFLOW_INLINE_OWNERSHIP`
 
 - Default: enabled
diff --git a/packages/core/src/events-consumer.test.ts b/packages/core/src/events-consumer.test.ts
index ecf828f728..3e320b16c7 100644
--- a/packages/core/src/events-consumer.test.ts
+++ b/packages/core/src/events-consumer.test.ts
@@ -45,6 +45,15 @@ describe('EventsConsumer', () => {
       expect(consumer.eventIndex).toBe(0);
       expect(consumer.callbacks).toEqual([]);
     });
+
+    it('should own its event array', () => {
+      const events = [createMockEvent()];
+      const consumer = new EventsConsumer(events, defaultOptions);
+
+      events.push(createMockEvent({ id: 'event-2' }));
+
+      expect(consumer.events).toHaveLength(1);
+    });
   });
 
   describe('subscribe', () => {
@@ -82,6 +91,24 @@ describe('EventsConsumer', () => {
       expect(callback).toHaveBeenCalledWith(event);
       expect(callback).toHaveBeenCalledTimes(1);
     });
+
+    it('should consume appended events', async () => {
+      const event = createMockEvent();
+      const consumer = new EventsConsumer([], defaultOptions);
+      const callback = vi
+        .fn()
+        .mockImplementation((value: Event | null) =>
+          value ? EventConsumerResult.Finished : EventConsumerResult.NotConsumed
+        );
+      consumer.subscribe(callback);
+      await waitForNextTick();
+
+      consumer.append([event]);
+      await waitForNextTick();
+
+      expect(callback).toHaveBeenLastCalledWith(event);
+      expect(consumer.eventIndex).toBe(1);
+    });
   });
 
   describe('consume (implicit)', () => {
diff --git a/packages/core/src/events-consumer.ts b/packages/core/src/events-consumer.ts
index 4b7cf0742e..7b4afd6a9a 100644
--- a/packages/core/src/events-consumer.ts
+++ b/packages/core/src/events-consumer.ts
@@ -79,13 +79,18 @@ export class EventsConsumer {
   private unconsumedCheckVersion = 0;
 
   constructor(events: Event[], options: EventsConsumerOptions) {
-    this.events = events;
+    this.events = [...events];
     this.eventIndex = 0;
     this.onConsumedEvent = options.onConsumedEvent;
     this.onUnconsumedEvent = options.onUnconsumedEvent;
     this.getPromiseQueue = options.getPromiseQueue;
   }
 
+  append(events: Event[]): void {
+    this.events.push(...events);
+    process.nextTick(this.consume);
+  }
+
   /**
    * Registers a callback function to be called after an event has been consumed
    * by a different callback. The callback can return:
diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts
index eeef58688b..2f7dfd1e72 100644
--- a/packages/core/src/private.ts
+++ b/packages/core/src/private.ts
@@ -135,6 +135,13 @@ export interface WorkflowOrchestratorContext {
   encryptionKey: CryptoKey | undefined;
   worldCapabilities?: WorldCapabilities;
   globalThis: typeof globalThis;
+  /**
+   * Increments on every retained-session resume. Suspension signals capture
+   * it when scheduled and no-op if it moved — a timer queued at boundary N
+   * must not signal after the session resumed into boundary N+1, so the
+   * state machine never has to reason about cross-boundary timer ordering.
+   */
+  suspensionGeneration: number;
   eventsConsumer: EventsConsumer;
   /**
    * Map of pending invocations keyed by correlationId.
diff --git a/packages/core/src/retained-vm-loop.test.ts b/packages/core/src/retained-vm-loop.test.ts
new file mode 100644
index 0000000000..d24b59bcc2
--- /dev/null
+++ b/packages/core/src/retained-vm-loop.test.ts
@@ -0,0 +1,330 @@
+import {
+  type Event,
+  SPEC_VERSION_CURRENT,
+  type WorkflowRun,
+} from '@workflow/world';
+import { afterEach, 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
+// per replay iteration. workflow.ts imports `createContext` from this same
+// module, so the spy observes its constructions too.
+vi.mock('./vm/index.js', async (importActual) => {
+  const actual = await importActual<typeof import('./vm/index.js')>();
+  return { ...actual, createContext: vi.fn(actual.createContext) };
+});
+
+const { createContext } = await import('./vm/index.js');
+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'
+);
+
+vi.mock('@vercel/functions', () => ({
+  waitUntil: vi.fn((p: Promise<unknown>) => {
+    p.catch(() => {});
+  }),
+}));
+
+const createContextSpy = createContext as unknown as ReturnType<typeof vi.fn>;
+
+// Sequential two-step workflow: two replay-advancing suspensions before it
+// completes — so a from-scratch replay builds the VM three times while the
+// retained path builds it once.
+const twoStepWorkflow = `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();
+    const b = await s2();
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+// Serializing this step's arguments executes the getter after suspension. Its
+// state mutation is not reconstructed by a cold replay, so this boundary must
+// demote even though it does not draw randomness.
+const impureArgsWorkflow = `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;
+    await s1({ get x() { counter++; return 1; } });
+    return await echo(counter);
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+const impureSerializerWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1");
+  const echo = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_echo");
+  class Value {
+    static classId = "test/RetainedSerializerValue";
+    static [Symbol.for("workflow-serialize")](instance) {
+      instance.onSerialize();
+      return { value: instance.value };
+    }
+    constructor(value, onSerialize) {
+      this.value = value;
+      this.onSerialize = onSerialize;
+    }
+  }
+  async function workflow() {
+    let counter = 0;
+    await s1(new Value(1, () => counter++));
+    return await echo(counter);
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+class RetainedSerializerValue {
+  constructor(readonly value: number) {}
+
+  static [Symbol.for('workflow-deserialize')](data: { value: number }) {
+    return new RetainedSerializerValue(data.value);
+  }
+}
+registerSerializationClass(
+  'test/RetainedSerializerValue',
+  RetainedSerializerValue
+);
+
+// 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");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  async function workflow() {
+    const [a, b] = await Promise.all([
+      s1({ get x() { return 1; } }),
+      s2({ plain: true }),
+    ]);
+    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");
+  const s2 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s2");
+  async function workflow() {
+    await crypto.subtle.digest("SHA-256", new Uint8Array(8));
+    const a = await s1();
+    const b = await s2();
+    return a + b;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+registerStepFunction('r_s1', async () => 10);
+registerStepFunction('r_s2', async () => 20);
+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.
+async function drive(runId: string, workflowCode = twoStepWorkflow) {
+  const run: WorkflowRun = {
+    runId,
+    workflowName: 'workflow',
+    status: 'running',
+    input: await dehydrateWorkflowArguments([], runId, undefined, []),
+    createdAt: new Date('2024-01-01T00:00:00.000Z'),
+    updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+    startedAt: new Date('2024-01-01T00:00:00.000Z'),
+    deploymentId: 'test-deployment',
+  };
+  const events: Event[] = [];
+  const createdEvents: any[] = [];
+  let seq = 0;
+
+  const eventsCreate = vi.fn(async (_runId: string, data: any) => {
+    createdEvents.push(data);
+    if (data.eventType === 'run_started') {
+      return { run, events };
+    }
+    const event = {
+      eventId: `e-${++seq}`,
+      runId,
+      createdAt: new Date(),
+      ...data,
+    } as Event;
+    events.push(event);
+    // step_started returns a running step entity so executeStep proceeds to
+    // run the body and write step_completed.
+    if (data.eventType === 'step_started') {
+      const d = data.eventData as { stepName?: string; input?: unknown };
+      return {
+        event,
+        step: {
+          runId,
+          stepId: data.correlationId,
+          stepName: d?.stepName,
+          status: 'running' as const,
+          attempt: 1,
+          input: d?.input,
+          startedAt: new Date(),
+          createdAt: new Date(),
+          updatedAt: new Date(),
+        },
+        ...(d?.input !== undefined ? { stepCreated: true } : {}),
+      };
+    }
+    return { event };
+  });
+
+  setWorld({
+    specVersion: SPEC_VERSION_CURRENT,
+    createQueueHandler: vi.fn(
+      (_p: string, handler: (m: unknown, md: unknown) => Promise<unknown>) =>
+        async () => {
+          await handler(
+            { runId, requestedAt: new Date('2024-01-01T00:00:00.000Z') },
+            {
+              requestId: 'req_retained',
+              attempt: 2,
+              queueName: '__wkf_workflow_workflow',
+              messageId: 'msg_retained',
+            }
+          );
+          return new Response(null, { status: 204 });
+        }
+    ),
+    events: {
+      create: eventsCreate,
+      list: vi.fn(async () => ({
+        data: [...events],
+        hasMore: false,
+        cursor: 'cursor_retained',
+      })),
+    },
+    runs: { get: vi.fn(async () => run) },
+    queue: vi.fn(async () => ({ messageId: null })),
+    getEncryptionKeyForRun: vi.fn(async () => undefined),
+  } as any);
+
+  await workflowEntrypoint(workflowCode)(new Request('https://example.test'));
+
+  const runCompleted = createdEvents.find(
+    (e) => e.eventType === 'run_completed'
+  );
+  return {
+    vmBuilds: createContextSpy.mock.calls.length,
+    output: runCompleted?.eventData?.output as Uint8Array | undefined,
+  };
+}
+
+describe('retained VM through the inline replay loop', () => {
+  beforeEach(() => {
+    createContextSpy.mockClear();
+  });
+  afterEach(() => {
+    delete process.env.WORKFLOW_RETAINED_VM;
+    setWorld(undefined);
+    vi.clearAllMocks();
+  });
+
+  it('rebuilds the VM once per replay under the kill switch (WORKFLOW_RETAINED_VM=0)', async () => {
+    process.env.WORKFLOW_RETAINED_VM = '0';
+    const { vmBuilds, output } = await drive('wrun_retained_off');
+    expect(output).toBeInstanceOf(Uint8Array);
+    // A 2-step sequential workflow suspends twice then completes → >1 replay,
+    // each building a fresh VM.
+    expect(vmBuilds).toBeGreaterThan(1);
+  });
+
+  it('builds the VM once and resumes it when retention is ON (the default)', async () => {
+    // Baseline via the kill switch, to compare the dehydrated result against.
+    process.env.WORKFLOW_RETAINED_VM = '0';
+    const off = await drive('wrun_retained_baseline_off');
+    createContextSpy.mockClear();
+
+    // Unset → retention is ON by default.
+    delete process.env.WORKFLOW_RETAINED_VM;
+    const on = await drive('wrun_retained_on');
+
+    expect(on.output).toBeInstanceOf(Uint8Array);
+    // The whole run is served by a single VM: built once on the first pass,
+    // resumed (not rebuilt) for every subsequent step.
+    expect(on.vmBuilds).toBe(1);
+    // And it produces the identical dehydrated result as the replay path.
+    expect(on.output).toEqual(off.output);
+  });
+
+  it('matches cold replay when argument serialization mutates workflow state', async () => {
+    process.env.WORKFLOW_RETAINED_VM = '0';
+    const off = await drive(
+      'wrun_retained_impure_args_off',
+      impureArgsWorkflow
+    );
+    createContextSpy.mockClear();
+
+    delete process.env.WORKFLOW_RETAINED_VM;
+    const on = await drive('wrun_retained_impure_args_on', impureArgsWorkflow);
+
+    expect(on.vmBuilds).toBeGreaterThan(1);
+    expect(
+      await hydrateWorkflowReturnValue(
+        off.output,
+        'wrun_retained_impure_args_off',
+        undefined,
+        []
+      )
+    ).toBe(0);
+    expect(
+      await hydrateWorkflowReturnValue(
+        on.output,
+        'wrun_retained_impure_args_on',
+        undefined,
+        []
+      )
+    ).toBe(0);
+  });
+
+  it('matches cold replay when a custom serializer mutates workflow state', async () => {
+    process.env.WORKFLOW_RETAINED_VM = '0';
+    const off = await drive(
+      'wrun_retained_impure_serializer_off',
+      impureSerializerWorkflow
+    );
+    createContextSpy.mockClear();
+
+    delete process.env.WORKFLOW_RETAINED_VM;
+    const on = await drive(
+      'wrun_retained_impure_serializer_on',
+      impureSerializerWorkflow
+    );
+
+    expect(on.vmBuilds).toBeGreaterThan(1);
+    expect(
+      await hydrateWorkflowReturnValue(
+        off.output,
+        'wrun_retained_impure_serializer_off',
+        undefined,
+        []
+      )
+    ).toBe(0);
+    expect(
+      await hydrateWorkflowReturnValue(
+        on.output,
+        'wrun_retained_impure_serializer_on',
+        undefined,
+        []
+      )
+    ).toBe(0);
+  });
+
+  it('demotes retention when any input in a parallel batch is unsafe', async () => {
+    const { vmBuilds, output } = await drive(
+      'wrun_retained_mixed_batch',
+      mixedBatchWorkflow
+    );
+    expect(output).toBeInstanceOf(Uint8Array);
+    expect(vmBuilds).toBeGreaterThan(1);
+  });
+
+  it('retains a VM that used the synchronous crypto.subtle.digest', async () => {
+    const { vmBuilds, output } = await drive(
+      'wrun_retained_digest',
+      digestWorkflow
+    );
+    expect(output).toBeInstanceOf(Uint8Array);
+    expect(vmBuilds).toBe(1);
+  });
+});
diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts
index aafec00075..1e229e2fec 100644
--- a/packages/core/src/runtime.test.ts
+++ b/packages/core/src/runtime.test.ts
@@ -11,6 +11,7 @@ import {
 } from '@workflow/world';
 import { ulid } from 'ulid';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { runtimeLogger } from './logger.js';
 import { registerStepFunction } from './private.js';
 import { REPLAY_DIVERGENCE_MAX_RETRIES } from './runtime/constants.js';
 import { setWorld } from './runtime/world.js';
@@ -702,6 +703,9 @@ describe('workflowEntrypoint replay guards', () => {
   });
 
   it('replays attribute events before executing a step that loses the same race', async () => {
+    const debug = vi
+      .spyOn(runtimeLogger, 'debug')
+      .mockImplementation(() => undefined);
     const ops: Promise<any>[] = [];
     const workflowRun: WorkflowRun = {
       runId: 'wrun_attribute_step_race',
@@ -760,6 +764,11 @@ describe('workflowEntrypoint replay guards', () => {
     expect(createdEvents).not.toContainEqual(
       expect.objectContaining({ eventType: 'step_started' })
     );
+    const executionModes = debug.mock.calls
+      .filter(([message]) => message === 'Starting workflow execution')
+      .map(([, context]) => context?.executionMode);
+    expect(executionModes).toEqual(['replay', 'replay']);
+    debug.mockRestore();
   });
 
   it('fails the run when the World rejects an attr_set event as invalid', async () => {
diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts
index 9122fe8e52..b30a9568d0 100644
--- a/packages/core/src/runtime.ts
+++ b/packages/core/src/runtime.ts
@@ -43,6 +43,7 @@ import {
   getReplayDivergenceMaxRetries,
   isInlineOwnershipEnabled,
   isTurboEnabled,
+  isVmRetentionEnabled,
 } from './runtime/constants.js';
 import {
   getQueueOverhead,
@@ -98,7 +99,11 @@ import {
 } from './telemetry.js';
 import { getErrorName, getErrorStack, normalizeUnknownError } from './types.js';
 import { buildWorkflowSuspensionMessage } from './util.js';
-import { runWorkflow } from './workflow.js';
+import {
+  executeWorkflow,
+  type WorkflowExecutionResult,
+  type WorkflowSession,
+} from './workflow.js';
 
 export type { Event, WorkflowRun };
 export { WorkflowSuspension } from './global.js';
@@ -355,6 +360,32 @@ function openHookAndWaitState(events: Event[]): {
   return { openHook, openWait };
 }
 
+/**
+ * Retain only pure step boundaries (every suspension item is a step — any
+ * other item type, present or future, is unretainable by default) with no
+ * out-of-band continuation source: attributes require replay; hooks and
+ * waits can wake another invocation. `WORKFLOW_RETAINED_VM=0` disables
+ * retention entirely.
+ *
+ * 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,
+  openHookWaitState: { openHook: boolean; openWait: boolean }
+): boolean {
+  return (
+    isVmRetentionEnabled() &&
+    suspension.steps.length > 0 &&
+    suspension.steps.every((item) => item.type === 'step') &&
+    !openHookWaitState.openHook &&
+    !openHookWaitState.openWait
+  );
+}
+
 /**
  * Creates a single route which handles workflow execution requests,
  * executing steps inline when possible to reduce function invocations
@@ -1233,6 +1264,13 @@ export function workflowEntrypoint(
                     encryptionKey
                   );
 
+                  let workflowExecution:
+                    | { readonly type: 'replay' }
+                    | {
+                        readonly type: 'retained';
+                        readonly session: WorkflowSession;
+                      } = { type: 'replay' };
+
                   // Main replay loop
                   // biome-ignore lint/correctness/noConstantCondition: intentional loop
                   while (true) {
@@ -1561,38 +1599,65 @@ export function workflowEntrypoint(
                       // point and the inline executeStep mutates eventsCursor.
                       preInlineWriteCursor = eventsCursor;
 
-                      // Replay workflow
-                      runtimeLogger.debug('Starting workflow replay', {
+                      runtimeLogger.debug('Starting workflow execution', {
                         workflowRunId: runId,
                         loopIteration,
                         eventCount: events.length,
+                        executionMode: workflowExecution.type,
                       });
                       replayStart = Date.now();
-                      // Start every missing decrypt/decompress operation before
-                      // VM setup. Web Crypto work can overlap bundle evaluation;
-                      // consumers still deserialize and resolve in event order.
-                      const payloadPrewarm = replayPayloadCache.prewarm(
-                        workflowRun,
-                        events
-                      );
-                      const result = await runWorkflow(
-                        workflowCode,
-                        workflowRun,
-                        events,
-                        encryptionKey,
-                        replayPayloadCache,
-                        // Turbo: the end-of-run drain inside runWorkflow commits
-                        // fire-and-forget `*_created` events before the terminal
-                        // `awaitRunReady()` below, so gate those writes on the
-                        // backgrounded run_started too. Undefined outside turbo.
-                        runReadyBarrier,
-                        world.capabilities
-                      );
-                      await payloadPrewarm;
-                      runtimeLogger.debug('Workflow replay completed', {
+                      let workflowResult: WorkflowExecutionResult =
+                        workflowExecution.type === 'retained'
+                          ? await executeWorkflow({
+                              type: 'resume',
+                              session: workflowExecution.session,
+                              events,
+                            })
+                          : { type: 'replay' };
+
+                      if (workflowResult.type === 'replay') {
+                        workflowExecution = { type: 'replay' };
+                        // Start every missing decrypt/decompress operation
+                        // before VM setup. Web Crypto work can overlap bundle
+                        // evaluation; consumers still deserialize and resolve
+                        // in event order.
+                        const payloadPrewarm = replayPayloadCache.prewarm(
+                          workflowRun,
+                          events
+                        );
+                        workflowResult = await executeWorkflow({
+                          type: 'replay',
+                          workflowCode,
+                          workflowRun,
+                          events,
+                          encryptionKey,
+                          replayPayloadCache,
+                          // Turbo: the end-of-run drain inside workflow
+                          // execution commits fire-and-forget `*_created`
+                          // events before the terminal `awaitRunReady()` below.
+                          runReadyBarrier,
+                          worldCapabilities: world.capabilities,
+                        });
+                        await payloadPrewarm;
+                      }
+
+                      if (workflowResult.type === 'suspended') {
+                        // Park the live session; the suspension catch below
+                        // makes the one retention decision — keep it for the
+                        // next iteration or discard it for a fresh replay.
+                        workflowExecution = {
+                          type: 'retained',
+                          session: workflowResult.session,
+                        };
+                        throw workflowResult.suspension;
+                      }
+
+                      const result = workflowResult.output;
+                      runtimeLogger.debug('Workflow execution completed', {
                         workflowRunId: runId,
                         loopIteration,
                         replayMs: Date.now() - replayStart,
+                        executionMode: workflowExecution.type,
                       });
 
                       // Workflow completed. Send the snapshot but do NOT
@@ -1808,6 +1873,28 @@ export function workflowEntrypoint(
                           return;
                         }
                         eventsCursor = suspensionLog.cursor;
+
+                        // Open hooks/waits in the cumulative log (this
+                        // suspension's writes are already merged in), computed
+                        // once for the retention decision here and the
+                        // delta/turbo gates below.
+                        const openHookWaitState =
+                          openHookAndWaitState(cachedEvents);
+
+                        // 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 (
+                          workflowExecution.type === 'retained' &&
+                          !(
+                            canRetainWorkflowSession(err, openHookWaitState) &&
+                            suspensionResult.retainedStepInputsSafe
+                          )
+                        ) {
+                          workflowExecution = { type: 'replay' };
+                        }
+
                         preStepBlockingMs += suspensionResult.hookCreationMs;
                         if (
                           suspensionResult.hasAttributeEvents &&
@@ -2128,12 +2215,6 @@ export function workflowEntrypoint(
                         // this the replay-budget check at the top of the
                         // next loop iteration would (incorrectly) charge
                         // the step body against the budget.
-                        // Open hooks/waits in the cumulative log, computed
-                        // once for the two gates below.
-                        const openHookWaitState = openHookAndWaitState(
-                          cachedEvents ?? []
-                        );
-
                         // Inline-delta fast path gate. We request the delta —
                         // and on the next iteration consume it in place of the
                         // events.list — only when ALL hold:
diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts
index 619c50f890..2d25f9163f 100644
--- a/packages/core/src/runtime/constants.ts
+++ b/packages/core/src/runtime/constants.ts
@@ -304,6 +304,23 @@ export function isTurboEnabled(): boolean {
   return !(raw === '0' || raw.toLowerCase() === 'false');
 }
 
+/**
+ * Whether the inline loop retains a suspended workflow VM across inline steps
+ * within one invocation (default ON). When on, a step-only suspension keeps
+ * the live VM, event consumer, and hydrated state alive, and the next loop
+ * iteration appends only the newly durable events instead of rebuilding the
+ * `vm.Context` and replaying the whole event log. Non-step suspensions and
+ * replay divergence always fall back to the ordinary durable replay path.
+ *
+ * `WORKFLOW_RETAINED_VM=0` (or `false`) is the kill switch: every iteration
+ * replays from scratch in a fresh VM, matching the pre-retention behavior.
+ */
+export function isVmRetentionEnabled(): boolean {
+  const raw = process.env.WORKFLOW_RETAINED_VM;
+  if (raw === undefined || raw === '') return true;
+  return !(raw === '0' || raw.toLowerCase() === 'false');
+}
+
 /**
  * Whether inline step ownership is enabled (default ON). When on, the lazy
  * `step_started` that creates an inline step records the owning queue
diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts
index a9c102caeb..72f5411f3c 100644
--- a/packages/core/src/runtime/suspension-handler.ts
+++ b/packages/core/src/runtime/suspension-handler.ts
@@ -32,6 +32,19 @@ 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;
@@ -121,6 +134,8 @@ 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. */
+  retainedStepInputsSafe: boolean;
 }
 
 async function createHookEvent({
@@ -489,6 +504,23 @@ export async function handleSuspension({
   // racing with concurrent handlers on step execution.
   const createdStepCorrelationIds = new Set<string>();
 
+  // 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))
+  );
+
   // Lazy inline start: defer the step_created write for up to
   // `getMaxInlineSteps()` steps the caller will run inline (in parallel). Each
   // step is created on the fly by the lazy `step_started` executeStep sends
@@ -717,6 +749,7 @@ export async function handleSuspension({
     hasAttributeEvents: attributeItems.length > 0,
     hasHookEvents: hooksNeedingCreation.length > 0,
     hookCreationMs,
+    retainedStepInputsSafe,
   };
 }
 
diff --git a/packages/core/src/step.ts b/packages/core/src/step.ts
index 1b14165b91..40d20bc914 100644
--- a/packages/core/src/step.ts
+++ b/packages/core/src/step.ts
@@ -56,7 +56,11 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) {
           // Crucially, if we got here, then this step Promise does
           // not resolve so that the user workflow code does not proceed any further.
           // Notify the workflow handler that this step has not been run / has not completed yet.
+          const generation = ctx.suspensionGeneration;
           scheduleWhenIdle(ctx, () => {
+            // A retained session may have resumed past this boundary while
+            // the timer was queued; a stale signal must not fire.
+            if (generation !== ctx.suspensionGeneration) return;
             ctx.onWorkflowError(
               new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis)
             );
diff --git a/packages/core/src/telemetry.ts b/packages/core/src/telemetry.ts
index da69d2756d..0c3a32f9d5 100644
--- a/packages/core/src/telemetry.ts
+++ b/packages/core/src/telemetry.ts
@@ -249,7 +249,10 @@ export async function trace<T>(
         code: otel.SpanStatusCode.ERROR,
         message: (e as Error).message,
       });
-      applyWorkflowSuspensionToSpan(e, otel, span);
+      if (WorkflowSuspension.is(e)) {
+        span.setStatus({ code: otel.SpanStatusCode.OK });
+        applyWorkflowSuspensionToSpan(e, span);
+      }
       throw e;
     } finally {
       span.end();
@@ -275,19 +278,12 @@ export async function recordElapsedSpan(
 }
 
 /**
- * Applies workflow suspension attributes to the given span if the error is a WorkflowSuspension
- * which is technically not an error, but an algebraic effect indicating suspension.
+ * Applies the workflow suspension algebraic effect to an active span.
  */
-function applyWorkflowSuspensionToSpan(
-  error: unknown,
-  otel: typeof api,
+export function applyWorkflowSuspensionToSpan(
+  error: WorkflowSuspension,
   span: api.Span
-) {
-  if (!error || !WorkflowSuspension.is(error)) {
-    return;
-  }
-
-  span.setStatus({ code: otel.SpanStatusCode.OK });
+): void {
   span.setAttributes({
     ...Attr.WorkflowSuspensionState('suspended'),
     ...Attr.WorkflowSuspensionStepCount(error.stepCount),
diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts
index bcc6d1dabf..0793b26d87 100644
--- a/packages/core/src/telemetry/semantic-conventions.ts
+++ b/packages/core/src/telemetry/semantic-conventions.ts
@@ -77,6 +77,11 @@ export const WorkflowEventsCount = SemanticConvention<number>(
   'workflow.events.count'
 );
 
+/** Whether workflow execution starts with replay or resumes a retained VM */
+export const WorkflowExecutionMode = SemanticConvention<'replay' | 'retained'>(
+  'workflow.execution.mode'
+);
+
 /** Number of arguments passed to the workflow */
 export const WorkflowArgumentsCount = SemanticConvention<number>(
   'workflow.arguments.count'
diff --git a/packages/core/src/workflow-session-telemetry.test.ts b/packages/core/src/workflow-session-telemetry.test.ts
new file mode 100644
index 0000000000..49058a4585
--- /dev/null
+++ b/packages/core/src/workflow-session-telemetry.test.ts
@@ -0,0 +1,69 @@
+import { trace as otelTrace } from '@opentelemetry/api';
+import {
+  BasicTracerProvider,
+  InMemorySpanExporter,
+  SimpleSpanProcessor,
+} from '@opentelemetry/sdk-trace-base';
+import type { WorkflowRun } from '@workflow/world';
+import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest';
+import { ReplayPayloadCache } from './replay-payload-cache.js';
+import { dehydrateWorkflowArguments } from './serialization.js';
+import { executeWorkflow } from './workflow.js';
+
+const exporter = new InMemorySpanExporter();
+const provider = new BasicTracerProvider();
+
+beforeAll(() => {
+  provider.addSpanProcessor(new SimpleSpanProcessor(exporter));
+  otelTrace.setGlobalTracerProvider(provider);
+});
+
+afterAll(async () => {
+  await provider.shutdown();
+  otelTrace.disable();
+});
+
+afterEach(() => {
+  exporter.reset();
+});
+
+describe('retained workflow telemetry', () => {
+  it('records suspension attributes on workflow.run spans', async () => {
+    const runId = 'wrun_retained_telemetry';
+    const workflowRun: WorkflowRun = {
+      runId,
+      workflowName: 'workflow',
+      status: 'running',
+      input: await dehydrateWorkflowArguments([], runId, undefined, []),
+      createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+      startedAt: new Date('2024-01-01T00:00:00.000Z'),
+      deploymentId: 'test-deployment',
+    };
+    const workflowCode = `
+      const step = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step");
+      async function workflow() { await step(); }
+      globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+    const result = await executeWorkflow({
+      type: 'replay',
+      workflowCode,
+      workflowRun,
+      events: [],
+      encryptionKey: undefined,
+      replayPayloadCache: new ReplayPayloadCache(undefined),
+    });
+    expect(result.type).toBe('suspended');
+
+    const span = exporter
+      .getFinishedSpans()
+      .find((candidate) => candidate.name === 'workflow.run workflow');
+    expect(span?.attributes).toMatchObject({
+      'workflow.execution.mode': 'replay',
+      'workflow.suspension.state': 'suspended',
+      'workflow.suspension.step_count': 1,
+      'workflow.suspension.hook_count': 0,
+      'workflow.suspension.wait_count': 0,
+    });
+  });
+});
diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts
index 2d903b94a7..6830603819 100644
--- a/packages/core/src/workflow.test.ts
+++ b/packages/core/src/workflow.test.ts
@@ -6,6 +6,7 @@ import { monotonicFactory } from 'ulid';
 import { afterEach, assert, describe, expect, it, vi } from 'vitest';
 import { DEFERRED_CHECK_DELAY_MS } from './events-consumer.js';
 import type { WorkflowSuspension } from './global.js';
+import { ReplayPayloadCache } from './replay-payload-cache.js';
 import { setWorld } from './runtime/world.js';
 import {
   dehydrateStepReturnValue,
@@ -13,7 +14,7 @@ import {
   hydrateWorkflowReturnValue,
 } from './serialization.js';
 import { createContext } from './vm/index.js';
-import { runWorkflow } from './workflow.js';
+import { executeWorkflow, runWorkflow } from './workflow.js';
 
 // No encryption key = encryption disabled
 const noEncryptionKey = undefined;
@@ -221,6 +222,251 @@ describe('runWorkflow', () => {
     ).toEqual(3);
   });
 
+  it('should retain workflow execution across sequential step suspensions', async () => {
+    const ops: Promise<any>[] = [];
+    const workflowRun: WorkflowRun = {
+      runId: 'wrun_retained',
+      workflowName: 'workflow',
+      status: 'running',
+      input: await dehydrateWorkflowArguments(
+        [],
+        'wrun_retained',
+        noEncryptionKey,
+        ops
+      ),
+      createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+      startedAt: new Date('2024-01-01T00:00:00.000Z'),
+      deploymentId: 'test-deployment',
+    };
+    const workflowCode = `
+      const add = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("add");
+      async function workflow() {
+        console.log("retained:entered");
+        const first = await add(1, 2);
+        console.log("retained:continued");
+        const second = await add(first, 3);
+        return second;
+      }
+      ${getWorkflowTransformCode('workflow')}`;
+    const log = vi.spyOn(console, 'log').mockImplementation(() => undefined);
+    const events: Event[] = [];
+    let eventNumber = 0;
+    const appendStepEvents = async (
+      correlationId: string,
+      result: number
+    ): Promise<void> => {
+      const createdAt = new Date(`2024-01-01T00:00:0${eventNumber + 1}.000Z`);
+      events.push(
+        {
+          eventId: `event-${eventNumber++}`,
+          runId: workflowRun.runId,
+          eventType: 'step_created',
+          correlationId,
+          eventData: { stepName: 'add' },
+          createdAt,
+        },
+        {
+          eventId: `event-${eventNumber++}`,
+          runId: workflowRun.runId,
+          eventType: 'step_started',
+          correlationId,
+          eventData: { stepName: 'add' },
+          createdAt,
+        },
+        {
+          eventId: `event-${eventNumber++}`,
+          runId: workflowRun.runId,
+          eventType: 'step_completed',
+          correlationId,
+          eventData: {
+            stepName: 'add',
+            result: await dehydrateStepReturnValue(
+              result,
+              workflowRun.runId,
+              noEncryptionKey,
+              ops
+            ),
+          },
+          createdAt,
+        }
+      );
+    };
+
+    const first = await executeWorkflow({
+      type: 'replay',
+      workflowCode,
+      workflowRun,
+      events,
+      encryptionKey: noEncryptionKey,
+      replayPayloadCache: new ReplayPayloadCache(noEncryptionKey),
+    });
+    assert(first.type === 'suspended');
+    const firstStep = first.suspension.steps[0];
+    assert(firstStep?.type === 'step');
+
+    await appendStepEvents(firstStep.correlationId, 3);
+
+    const second = await executeWorkflow({
+      type: 'resume',
+      session: first.session,
+      events,
+    });
+    assert(second.type === 'suspended');
+    expect(second.session).toBe(first.session);
+    const secondStep = second.suspension.steps[0];
+    assert(secondStep?.type === 'step');
+
+    await appendStepEvents(secondStep.correlationId, 6);
+
+    const completed = await executeWorkflow({
+      type: 'resume',
+      session: second.session,
+      events,
+    });
+    assert(completed.type === 'completed');
+
+    expect(
+      await hydrateWorkflowReturnValue(
+        completed.output as any,
+        workflowRun.runId,
+        noEncryptionKey,
+        ops
+      )
+    ).toBe(6);
+    expect(
+      log.mock.calls
+        .map(([message]) => message)
+        .filter((message) => String(message).startsWith('retained:'))
+    ).toEqual(['retained:entered', 'retained:continued']);
+    log.mockRestore();
+  });
+
+  it('falls back to replay permanently when the event prefix diverges', async () => {
+    const ops: Promise<any>[] = [];
+    const workflowRun: WorkflowRun = {
+      runId: 'wrun_retained_divergence',
+      workflowName: 'workflow',
+      status: 'running',
+      input: await dehydrateWorkflowArguments(
+        [],
+        'wrun_retained_divergence',
+        noEncryptionKey,
+        ops
+      ),
+      createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+      startedAt: new Date('2024-01-01T00:00:00.000Z'),
+      deploymentId: 'test-deployment',
+    };
+    const workflowCode = `
+      const step = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step");
+      async function workflow() { await step(); }
+      ${getWorkflowTransformCode('workflow')}`;
+    const events: Event[] = [
+      {
+        eventId: 'event-run-created',
+        runId: workflowRun.runId,
+        eventType: 'run_created',
+        createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      },
+    ];
+
+    const suspended = await executeWorkflow({
+      type: 'replay',
+      workflowCode,
+      workflowRun,
+      events,
+      encryptionKey: noEncryptionKey,
+      replayPayloadCache: new ReplayPayloadCache(noEncryptionKey),
+    });
+    assert(suspended.type === 'suspended');
+
+    // A log whose consumed prefix was rewritten is not a strict extension.
+    const rewritten = [
+      { ...events[0], eventId: 'rewritten-event' } as Event,
+      { ...events[0], eventId: 'new-event' } as Event,
+    ];
+    expect(
+      await executeWorkflow({
+        type: 'resume',
+        session: suspended.session,
+        events: rewritten,
+      })
+    ).toEqual({ type: 'replay' });
+
+    // The fallback is permanent, even for a well-formed extension.
+    expect(
+      await executeWorkflow({
+        type: 'resume',
+        session: suspended.session,
+        events,
+      })
+    ).toEqual({ type: 'replay' });
+  });
+
+  it('falls back to replay (not failure) after mid-execution divergence', async () => {
+    const ops: Promise<any>[] = [];
+    const workflowRun: WorkflowRun = {
+      runId: 'wrun_retained_mid_divergence',
+      workflowName: 'workflow',
+      status: 'running',
+      input: await dehydrateWorkflowArguments(
+        [],
+        'wrun_retained_mid_divergence',
+        noEncryptionKey,
+        ops
+      ),
+      createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+      startedAt: new Date('2024-01-01T00:00:00.000Z'),
+      deploymentId: 'test-deployment',
+    };
+    const workflowCode = `
+      const step = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step");
+      async function workflow() { await step(); }
+      ${getWorkflowTransformCode('workflow')}`;
+    const events: Event[] = [];
+
+    const suspended = await executeWorkflow({
+      type: 'replay',
+      workflowCode,
+      workflowRun,
+      events,
+      encryptionKey: noEncryptionKey,
+      replayPayloadCache: new ReplayPayloadCache(noEncryptionKey),
+    });
+    assert(suspended.type === 'suspended');
+
+    // A strict extension whose appended suffix the VM cannot consume: the
+    // resume itself starts, then diverges mid-execution.
+    await expect(
+      executeWorkflow({
+        type: 'resume',
+        session: suspended.session,
+        events: [
+          {
+            eventId: 'event-alien',
+            runId: workflowRun.runId,
+            eventType: 'hook_received',
+            correlationId: 'hook_unknown',
+            eventData: {},
+            createdAt: new Date('2024-01-01T00:00:01.000Z'),
+          } as Event,
+        ],
+      })
+    ).rejects.toThrow(/could not consume event/);
+
+    // The session must demote to replay — not throw "cannot resume".
+    expect(
+      await executeWorkflow({
+        type: 'resume',
+        session: suspended.session,
+        events: [],
+      })
+    ).toEqual({ type: 'replay' });
+  });
+
   it('regenerates step correlation IDs independent of startedAt (turbo replay-stability)', async () => {
     // Turbo's first delivery synthesizes `startedAt` from the local clock,
     // while later (non-turbo) deliveries load the server-canonical `startedAt`.
diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts
index 06fe1d79e2..f56580e0d0 100644
--- a/packages/core/src/workflow.ts
+++ b/packages/core/src/workflow.ts
@@ -4,7 +4,11 @@ import {
   WorkflowNotRegisteredError,
   WorkflowRuntimeError,
 } from '@workflow/errors';
-import { createWorkflowBaseUrl, withResolvers } from '@workflow/utils';
+import {
+  createWorkflowBaseUrl,
+  type PromiseWithResolvers,
+  withResolvers,
+} from '@workflow/utils';
 import { parseWorkflowName } from '@workflow/utils/parse-name';
 import type { Event, WorkflowRun, WorldCapabilities } from '@workflow/world';
 import { SPEC_VERSION_SUPPORTS_COMPRESSION } from '@workflow/world';
@@ -36,7 +40,7 @@ import {
   WORKFLOW_USE_STEP,
 } from './symbols.js';
 import * as Attribute from './telemetry/semantic-conventions.js';
-import { trace } from './telemetry.js';
+import { applyWorkflowSuspensionToSpan, trace } from './telemetry.js';
 import { getWorkflowRunStreamId } from './util.js';
 import { createContext } from './vm/index.js';
 import { runCachedWorkflowScript } from './vm/script-cache.js';
@@ -133,6 +137,127 @@ async function drainPendingQueueItems(
   }
 }
 
+interface WorkflowCompletion {
+  readonly output: unknown;
+  readonly resultType: string;
+}
+
+type WorkflowSessionState =
+  | {
+      readonly type: 'running';
+      readonly interruption: PromiseWithResolvers<never>;
+    }
+  | { readonly type: 'suspended'; readonly suspension: WorkflowSuspension }
+  | { readonly type: 'replay' }
+  | { readonly type: 'completed' };
+
+export interface WorkflowSession {
+  readonly workflowRun: WorkflowRun;
+  readonly argumentCount: number;
+  resume(events: Event[]): WorkflowSessionResumeResult;
+}
+
+type WorkflowSessionResumeResult =
+  | {
+      readonly type: 'resumed';
+      readonly execution: Promise<WorkflowCompletion>;
+    }
+  | { readonly type: 'replay' };
+
+interface WorkflowSessionOptions {
+  readonly workflowCode: string;
+  readonly workflowRun: WorkflowRun;
+  readonly events: Event[];
+  readonly encryptionKey: CryptoKey | undefined;
+  readonly replayPayloadCache: ReplayPayloadCache;
+  readonly runReadyBarrier?: Promise<unknown>;
+  readonly worldCapabilities?: WorldCapabilities;
+}
+
+type WorkflowExecutionRequest =
+  | ({ readonly type: 'replay' } & WorkflowSessionOptions)
+  | {
+      readonly type: 'resume';
+      readonly session: WorkflowSession;
+      readonly events: Event[];
+    };
+
+type WorkflowReplayResult =
+  | { readonly type: 'completed'; readonly output: unknown }
+  | {
+      readonly type: 'suspended';
+      readonly suspension: WorkflowSuspension;
+      readonly session: WorkflowSession;
+    };
+
+export type WorkflowExecutionResult =
+  | { readonly type: 'replay' }
+  | WorkflowReplayResult;
+
+// A fresh replay always completes or suspends; only resuming a session can
+// request another replay (stale session or diverged event prefix).
+export async function executeWorkflow(
+  request: { readonly type: 'replay' } & WorkflowSessionOptions
+): Promise<WorkflowReplayResult>;
+export async function executeWorkflow(
+  request: WorkflowExecutionRequest
+): Promise<WorkflowExecutionResult>;
+export async function executeWorkflow(
+  request: WorkflowExecutionRequest
+): Promise<WorkflowExecutionResult> {
+  const workflowRun =
+    request.type === 'replay'
+      ? request.workflowRun
+      : request.session.workflowRun;
+  const mode = request.type === 'replay' ? 'replay' : 'retained';
+
+  return trace(`workflow.run ${workflowRun.workflowName}`, async (span) => {
+    span?.setAttributes({
+      ...Attribute.WorkflowName(workflowRun.workflowName),
+      ...Attribute.WorkflowRunId(workflowRun.runId),
+      ...Attribute.WorkflowRunStatus(workflowRun.status),
+      ...Attribute.WorkflowEventsCount(request.events.length),
+      ...Attribute.WorkflowExecutionMode(mode),
+    });
+
+    let session: WorkflowSession;
+    let execution: Promise<WorkflowCompletion>;
+    switch (request.type) {
+      case 'replay': {
+        const started = await createWorkflowSession(request);
+        session = started.session;
+        execution = started.execution;
+        break;
+      }
+      case 'resume': {
+        session = request.session;
+        const resumed = session.resume(request.events);
+        if (resumed.type === 'replay') return resumed;
+        execution = resumed.execution;
+        break;
+      }
+    }
+
+    span?.setAttributes({
+      ...Attribute.WorkflowArgumentsCount(session.argumentCount),
+    });
+
+    try {
+      const completed = await execution;
+      span?.setAttributes({
+        ...Attribute.WorkflowResultType(completed.resultType),
+      });
+      return { type: 'completed', output: completed.output };
+    } catch (error) {
+      if (WorkflowSuspension.is(error)) {
+        if (span) applyWorkflowSuspensionToSpan(error, span);
+        return { type: 'suspended', suspension: error, session };
+      }
+      throw error;
+    }
+  });
+}
+
 export async function runWorkflow(
   workflowCode: string,
   workflowRun: WorkflowRun,
@@ -140,8 +265,7 @@ export async function runWorkflow(
   encryptionKey: CryptoKey | undefined,
   /**
    * Optional per-run cache for replay payload preparation and immutable final
-   * values. Owned by the inline replay loop so it survives fresh VM contexts
-   * created by successive iterations of this invocation.
+   * values. Owned by the inline execution loop for this invocation.
    */
   replayPayloadCache: ReplayPayloadCache = new ReplayPayloadCache(
     encryptionKey
@@ -159,293 +283,329 @@ export async function runWorkflow(
    */
   worldCapabilities?: WorldCapabilities
 ): Promise<Uint8Array | unknown> {
-  return trace(`workflow.run ${workflowRun.workflowName}`, async (span) => {
-    span?.setAttributes({
-      ...Attribute.WorkflowName(workflowRun.workflowName),
-      ...Attribute.WorkflowRunId(workflowRun.runId),
-      ...Attribute.WorkflowRunStatus(workflowRun.status),
-      ...Attribute.WorkflowEventsCount(events.length),
-    });
-
-    const startedAt = workflowRun.startedAt;
-    if (!startedAt) {
-      throw new WorkflowRuntimeError(
-        `Workflow run "${workflowRun.runId}" has no "startedAt" timestamp (should not happen)`
-      );
-    }
-
-    // The deterministic RNG seed is derived from identifiers that are all
-    // known the instant the queue message arrives — `runId`, `workflowName`,
-    // and `deploymentId` — with no timestamp component. `runId` alone already
-    // makes the seed unique-per-run and replay-stable; `workflowName` and
-    // `deploymentId` are included for extra entropy. Dropping the timestamp
-    // means the seed no longer depends on `startedAt`/`createdAt`, so it (and
-    // the VM context) can be computed before any server round-trip.
-    //
-    // The VM's initial fixed clock is derived from the run's creation time,
-    // recovered from the ULID embedded in `runId` (also available immediately),
-    // falling back to the run snapshot's `createdAt` for non-ULID ids. This
-    // initial `fixedTimestamp` only governs `Date.now()` / `new Date()` in the
-    // window before the first event is consumed; thereafter `updateTimestamp`
-    // advances the VM clock to each consumed event's `createdAt` (see the
-    // EventsConsumer below), starting with `run_created`.
-    const fixedTimestamp =
-      runIdCreatedAt(workflowRun.runId) ?? +workflowRun.createdAt;
-
-    // Get the port before creating VM context to avoid async operations
-    // affecting the deterministic timestamp
-    const isVercel = process.env.VERCEL_URL !== undefined;
-    // Load getPort lazily to prevent Turbopack from tracing get-port's
-    // fs ops (readdir, readFile) into the flow route bundle. The resolved
-    // port is cached per process (see get-port-lazy.ts), so this is cheap
-    // on replays after the first.
-    const workflowBaseUrl = createWorkflowBaseUrl(
-      isVercel
-        ? `https://${process.env.VERCEL_URL}`
-        : `http://localhost:${(await getPortLazy()) ?? 3000}`
-    );
-
-    const {
-      context,
-      globalThis: vmGlobalThis,
-      updateTimestamp,
-    } = createContext({
-      seed: `${workflowRun.runId}:${workflowRun.workflowName}:${workflowRun.deploymentId}`,
-      fixedTimestamp,
-    });
-
-    const workflowDiscontinuation = withResolvers<void>();
+  const result = await executeWorkflow({
+    type: 'replay',
+    workflowCode,
+    workflowRun,
+    events,
+    encryptionKey,
+    replayPayloadCache,
+    runReadyBarrier,
+    worldCapabilities,
+  });
+  if (result.type === 'suspended') throw result.suspension;
+  return result.output;
+}
 
-    const ulid = monotonicFactory(() => vmGlobalThis.Math.random());
-    const generateNanoid = nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
-      new Uint8Array(size).map(() => 256 * vmGlobalThis.Math.random())
+async function createWorkflowSession({
+  workflowCode,
+  workflowRun,
+  events,
+  encryptionKey,
+  replayPayloadCache,
+  runReadyBarrier,
+  worldCapabilities,
+}: WorkflowSessionOptions): Promise<{
+  session: WorkflowSession;
+  execution: Promise<WorkflowCompletion>;
+}> {
+  const startedAt = workflowRun.startedAt;
+  if (!startedAt) {
+    throw new WorkflowRuntimeError(
+      `Workflow run "${workflowRun.runId}" has no "startedAt" timestamp (should not happen)`
     );
+  }
 
-    // Create a mutable holder for the promise queue so the EventsConsumer
-    // can access the current queue state via a getter. The queue is mutated
-    // by step/hook/sleep callbacks as events are processed.
-    const promiseQueueHolder = { current: Promise.resolve() };
-
-    const eventsConsumer = new EventsConsumer(events, {
-      onConsumedEvent: (event) => {
-        updateTimestamp(+event.createdAt);
-      },
-      onUnconsumedEvent: (event) => {
-        workflowDiscontinuation.reject(
-          new ReplayDivergenceError(
-            `Replay could not consume event: eventType=${event.eventType}, correlationId=${event.correlationId}, eventId=${event.eventId}.`,
-            { eventId: event.eventId }
-          )
-        );
-      },
-      getPromiseQueue: () => promiseQueueHolder.current,
-    });
-
-    const workflowContext: WorkflowOrchestratorContext = {
-      runId: workflowRun.runId,
-      encryptionKey,
-      worldCapabilities,
-      globalThis: vmGlobalThis,
-      onWorkflowError: workflowDiscontinuation.reject,
-      eventsConsumer,
-      // Correlation IDs (step_/wait_/hook_) are derived from `generateUlid`, so
-      // the time prefix fed to `ulid()` MUST be replay-stable across every
-      // delivery — otherwise a redelivery regenerates different correlation IDs
-      // and replay throws ReplayDivergenceError. `startedAt` is NOT safe here:
-      // under turbo the first delivery synthesizes `startedAt` from the local
-      // clock, but later (non-turbo) deliveries load the server-canonical
-      // `startedAt`, which differs by >=1ms. Use the same replay-stable value
-      // that already seeds the RNG and the VM clock (`fixedTimestamp`, recovered
-      // from the run ID's ULID and known the instant the message arrives).
-      generateUlid: () => ulid(fixedTimestamp),
-      generateNanoid,
-      invocationsQueue: new Map(),
-      // Use getter/setter so the EventsConsumer's getPromiseQueue() always
-      // sees the latest queue state as it's mutated by step/hook/sleep callbacks.
-      get promiseQueue() {
-        return promiseQueueHolder.current;
-      },
-      set promiseQueue(value: Promise<void>) {
-        promiseQueueHolder.current = value;
-      },
-      pendingDeliveries: 0,
-      pendingDeliveryBarriers: new Map(),
-      replayPayloadCache,
-    };
-
-    // Consume run lifecycle events - these are structural events that don't
-    // need special handling in the workflow, but must be consumed to advance
-    // past them in the event log
-    workflowContext.eventsConsumer.subscribe((event) => {
-      if (!event) {
-        return EventConsumerResult.NotConsumed;
-      }
-
-      // Consume run_created - every run has exactly one
-      if (event.eventType === 'run_created') {
-        return EventConsumerResult.Consumed;
-      }
+  // The deterministic RNG seed is derived from identifiers that are all
+  // known the instant the queue message arrives — `runId`, `workflowName`,
+  // and `deploymentId` — with no timestamp component. `runId` alone already
+  // makes the seed unique-per-run and replay-stable; `workflowName` and
+  // `deploymentId` are included for extra entropy. Dropping the timestamp
+  // means the seed no longer depends on `startedAt`/`createdAt`, so it (and
+  // the VM context) can be computed before any server round-trip.
+  //
+  // The VM's initial fixed clock is derived from the run's creation time,
+  // recovered from the ULID embedded in `runId` (also available immediately),
+  // falling back to the run snapshot's `createdAt` for non-ULID ids. This
+  // initial `fixedTimestamp` only governs `Date.now()` / `new Date()` in the
+  // window before the first event is consumed; thereafter `updateTimestamp`
+  // advances the VM clock to each consumed event's `createdAt` (see the
+  // EventsConsumer below), starting with `run_created`.
+  const fixedTimestamp =
+    runIdCreatedAt(workflowRun.runId) ?? +workflowRun.createdAt;
+
+  // Get the port before creating VM context to avoid async operations
+  // affecting the deterministic timestamp
+  const isVercel = process.env.VERCEL_URL !== undefined;
+  // Load getPort lazily to prevent Turbopack from tracing get-port's
+  // fs ops (readdir, readFile) into the flow route bundle. The resolved
+  // port is cached per process (see get-port-lazy.ts), so this is cheap
+  // on replays after the first.
+  const workflowBaseUrl = createWorkflowBaseUrl(
+    isVercel
+      ? `https://${process.env.VERCEL_URL}`
+      : `http://localhost:${(await getPortLazy()) ?? 3000}`
+  );
+
+  const {
+    context,
+    globalThis: vmGlobalThis,
+    updateTimestamp,
+  } = createContext({
+    seed: `${workflowRun.runId}:${workflowRun.workflowName}:${workflowRun.deploymentId}`,
+    fixedTimestamp,
+  });
 
-      // Consume run_started - every run has exactly one
-      if (event.eventType === 'run_started') {
-        return EventConsumerResult.Consumed;
+  const initialInterruption = withResolvers<never>();
+  let state: WorkflowSessionState = {
+    type: 'running',
+    interruption: initialInterruption,
+  };
+
+  const onWorkflowError = (error: Error): void => {
+    switch (state.type) {
+      case 'running': {
+        const { interruption } = state;
+        state = WorkflowSuspension.is(error)
+          ? { type: 'suspended', suspension: error }
+          : { type: 'replay' };
+        interruption.reject(error);
+        return;
       }
-
-      // Attribute writes performed from a step have no workflow-body call to
-      // consume them during replay; they are already reflected in the run
-      // snapshot and remain structural until a read API is introduced.
-      if (
-        event.eventType === 'attr_set' &&
-        event.eventData.writer.type === 'step'
-      ) {
-        return EventConsumerResult.Consumed;
+      case 'suspended': {
+        // A suspended VM is quiescent, but each parked step consumer
+        // signals its own (identical) suspension via scheduleWhenIdle —
+        // absorb those duplicates, treat anything else as unretainable.
+        // The suspension counts are all derived from `steps`, so identical
+        // items mean an identical boundary.
+        const { steps } = state.suspension;
+        const isDuplicateSignal =
+          WorkflowSuspension.is(error) &&
+          error.steps.length === steps.length &&
+          error.steps.every((item, index) => item === steps[index]);
+        if (!isDuplicateSignal) state = { type: 'replay' };
+        return;
       }
+      case 'replay':
+      case 'completed':
+        return;
+    }
+    state satisfies never;
+  };
+
+  const ulid = monotonicFactory(() => vmGlobalThis.Math.random());
+  const generateNanoid = nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
+    new Uint8Array(size).map(() => 256 * vmGlobalThis.Math.random())
+  );
+
+  // Create a mutable holder for the promise queue so the EventsConsumer
+  // can access the current queue state via a getter. The queue is mutated
+  // by step/hook/sleep callbacks as events are processed.
+  const promiseQueueHolder = { current: Promise.resolve() };
+
+  const eventsConsumer = new EventsConsumer(events, {
+    onConsumedEvent: (event) => {
+      updateTimestamp(+event.createdAt);
+    },
+    onUnconsumedEvent: (event) => {
+      onWorkflowError(
+        new ReplayDivergenceError(
+          `Replay could not consume event: eventType=${event.eventType}, correlationId=${event.correlationId}, eventId=${event.eventId}.`,
+          { eventId: event.eventId }
+        )
+      );
+    },
+    getPromiseQueue: () => promiseQueueHolder.current,
+  });
 
+  const workflowContext: WorkflowOrchestratorContext = {
+    runId: workflowRun.runId,
+    encryptionKey,
+    worldCapabilities,
+    globalThis: vmGlobalThis,
+    onWorkflowError,
+    eventsConsumer,
+    // Correlation IDs (step_/wait_/hook_) are derived from `generateUlid`, so
+    // the time prefix fed to `ulid()` MUST be replay-stable across every
+    // delivery — otherwise a redelivery regenerates different correlation IDs
+    // and replay throws ReplayDivergenceError. `startedAt` is NOT safe here:
+    // under turbo the first delivery synthesizes `startedAt` from the local
+    // clock, but later (non-turbo) deliveries load the server-canonical
+    // `startedAt`, which differs by >=1ms. Use the same replay-stable value
+    // that already seeds the RNG and the VM clock (`fixedTimestamp`, recovered
+    // from the run ID's ULID and known the instant the message arrives).
+    generateUlid: () => ulid(fixedTimestamp),
+    generateNanoid,
+    invocationsQueue: new Map(),
+    // Use getter/setter so the EventsConsumer's getPromiseQueue() always
+    // sees the latest queue state as it's mutated by step/hook/sleep callbacks.
+    get promiseQueue() {
+      return promiseQueueHolder.current;
+    },
+    set promiseQueue(value: Promise<void>) {
+      promiseQueueHolder.current = value;
+    },
+    pendingDeliveries: 0,
+    suspensionGeneration: 0,
+    pendingDeliveryBarriers: new Map(),
+    replayPayloadCache,
+  };
+
+  // Consume run lifecycle events - these are structural events that don't
+  // need special handling in the workflow, but must be consumed to advance
+  // past them in the event log
+  workflowContext.eventsConsumer.subscribe((event) => {
+    if (!event) {
       return EventConsumerResult.NotConsumed;
-    });
+    }
 
-    const useStep = createUseStep(workflowContext);
-    const createHook = createCreateHook(workflowContext);
-    const sleep = createSleep(workflowContext);
-    const setAttributes = createSetAttributes(workflowContext);
-
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[WORKFLOW_USE_STEP] = useStep;
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[WORKFLOW_SET_ATTRIBUTES] = setAttributes;
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[WORKFLOW_CREATE_HOOK] = createHook;
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[WORKFLOW_SLEEP] = sleep;
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[WORKFLOW_GET_STREAM_ID] = (namespace?: string) =>
-      getWorkflowRunStreamId(workflowRun.runId, namespace);
-
-    // For the workflow VM, we store the context in a symbol on the `globalThis` object
-    const ctx: WorkflowMetadata = {
-      workflowName: workflowRun.workflowName,
-      workflowRunId: workflowRun.runId,
-      workflowStartedAt: new vmGlobalThis.Date(+startedAt),
-      url: workflowBaseUrl,
-      features: { encryption: !!encryptionKey },
-    };
+    // Consume run_created - every run has exactly one
+    if (event.eventType === 'run_created') {
+      return EventConsumerResult.Consumed;
+    }
 
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[WORKFLOW_CONTEXT_SYMBOL] = ctx;
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[STABLE_ULID] = ulid;
+    // Consume run_started - every run has exactly one
+    if (event.eventType === 'run_started') {
+      return EventConsumerResult.Consumed;
+    }
 
-    // NOTE: Will have a config override to use the custom fetch step.
-    //       For now `fetch` must be explicitly imported from `workflow`.
-    vmGlobalThis.fetch = () => {
-      throw new vmGlobalThis.Error(
-        `Global "fetch" is unavailable in workflow functions. Use the "fetch" step function from "workflow" to make HTTP requests.\n\nLearn more: https://workflow-sdk.dev/err/${ERROR_SLUGS.FETCH_IN_WORKFLOW_FUNCTION}`
-      );
-    };
+    // Attribute writes performed from a step have no workflow-body call to
+    // consume them during replay; they are already reflected in the run
+    // snapshot and remain structural until a read API is introduced.
+    if (
+      event.eventType === 'attr_set' &&
+      event.eventData.writer.type === 'step'
+    ) {
+      return EventConsumerResult.Consumed;
+    }
 
-    // Override timeout/interval functions to throw helpful errors
-    // These are not supported in workflow functions because they rely on
-    // asynchronous scheduling which breaks deterministic replay
-    const timeoutErrorMessage =
-      'Timeout functions like "setTimeout" and "setInterval" are not supported in workflow functions. Use the "sleep" function from "workflow" for time-based delays.';
+    return EventConsumerResult.NotConsumed;
+  });
 
-    (vmGlobalThis as any).setTimeout = () => {
-      throw new WorkflowRuntimeError(timeoutErrorMessage, {
-        slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
-      });
-    };
-    (vmGlobalThis as any).setInterval = () => {
-      throw new WorkflowRuntimeError(timeoutErrorMessage, {
-        slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
-      });
-    };
-    (vmGlobalThis as any).clearTimeout = () => {
-      throw new WorkflowRuntimeError(timeoutErrorMessage, {
-        slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
-      });
-    };
-    (vmGlobalThis as any).clearInterval = () => {
-      throw new WorkflowRuntimeError(timeoutErrorMessage, {
-        slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
-      });
-    };
-    (vmGlobalThis as any).setImmediate = () => {
-      throw new WorkflowRuntimeError(timeoutErrorMessage, {
-        slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
-      });
-    };
-    (vmGlobalThis as any).clearImmediate = () => {
-      throw new WorkflowRuntimeError(timeoutErrorMessage, {
-        slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
-      });
-    };
+  const useStep = createUseStep(workflowContext);
+  const createHook = createCreateHook(workflowContext);
+  const sleep = createSleep(workflowContext);
+  const setAttributes = createSetAttributes(workflowContext);
+
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[WORKFLOW_USE_STEP] = useStep;
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[WORKFLOW_SET_ATTRIBUTES] = setAttributes;
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[WORKFLOW_CREATE_HOOK] = createHook;
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[WORKFLOW_SLEEP] = sleep;
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[WORKFLOW_GET_STREAM_ID] = (namespace?: string) =>
+    getWorkflowRunStreamId(workflowRun.runId, namespace);
+
+  // For the workflow VM, we store the context in a symbol on the `globalThis` object
+  const ctx: WorkflowMetadata = {
+    workflowName: workflowRun.workflowName,
+    workflowRunId: workflowRun.runId,
+    workflowStartedAt: new vmGlobalThis.Date(+startedAt),
+    url: workflowBaseUrl,
+    features: { encryption: !!encryptionKey },
+  };
+
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[WORKFLOW_CONTEXT_SYMBOL] = ctx;
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[STABLE_ULID] = ulid;
+
+  // NOTE: Will have a config override to use the custom fetch step.
+  //       For now `fetch` must be explicitly imported from `workflow`.
+  vmGlobalThis.fetch = () => {
+    throw new vmGlobalThis.Error(
+      `Global "fetch" is unavailable in workflow functions. Use the "fetch" step function from "workflow" to make HTTP requests.\n\nLearn more: https://workflow-sdk.dev/err/${ERROR_SLUGS.FETCH_IN_WORKFLOW_FUNCTION}`
+    );
+  };
 
-    // `AbortController` and `AbortSignal` in the workflow VM are hook-backed
-    // for deterministic replay. The controller's abort() queues a hook resumption,
-    // and signal.aborted is updated when the hook event is processed during replay.
-    (vmGlobalThis as any).AbortController =
-      createCreateAbortController(workflowContext);
-    const abortSignalStatics = createAbortSignalStatics();
-    (vmGlobalThis as any).AbortSignal = {
-      abort: abortSignalStatics.abort,
-      any: abortSignalStatics.any,
-      timeout: abortSignalStatics.timeout,
-    };
+  // Override timeout/interval functions to throw helpful errors
+  // These are not supported in workflow functions because they rely on
+  // asynchronous scheduling which breaks deterministic replay
+  const timeoutErrorMessage =
+    'Timeout functions like "setTimeout" and "setInterval" are not supported in workflow functions. Use the "sleep" function from "workflow" for time-based delays.';
 
-    // `Request` and `Response` are special built-in classes that invoke steps
-    // for the `json()`, `text()` and `arrayBuffer()` instance methods
-    class Request implements globalThis.Request {
-      cache!: globalThis.Request['cache'];
-      credentials!: globalThis.Request['credentials'];
-      destination!: globalThis.Request['destination'];
-      headers!: Headers;
-      integrity!: string;
-      method!: string;
-      mode!: globalThis.Request['mode'];
-      redirect!: globalThis.Request['redirect'];
-      referrer!: string;
-      referrerPolicy!: globalThis.Request['referrerPolicy'];
-      url!: string;
-      keepalive!: boolean;
-      signal!: AbortSignal;
-      duplex!: 'half';
-      body!: ReadableStream<any> | null;
-
-      constructor(input: any, init?: RequestInit) {
-        // Handle URL input
-        if (typeof input === 'string' || input instanceof vmGlobalThis.URL) {
-          const urlString = String(input);
-          // Validate URL format
-          try {
-            new vmGlobalThis.URL(urlString);
-            this.url = urlString;
-          } catch (cause) {
-            throw new TypeError(`Failed to parse URL from ${urlString}`, {
-              cause,
-            });
-          }
-        } else {
-          // Input is a Request object - clone its properties
-          this.url = input.url;
-          if (!init) {
-            this.method = input.method;
-            this.headers = new vmGlobalThis.Headers(input.headers);
-            this.body = input.body;
-            this.mode = input.mode;
-            this.credentials = input.credentials;
-            this.cache = input.cache;
-            this.redirect = input.redirect;
-            this.referrer = input.referrer;
-            this.referrerPolicy = input.referrerPolicy;
-            this.integrity = input.integrity;
-            this.keepalive = input.keepalive;
-            this.signal = input.signal;
-            this.duplex = input.duplex;
-            this.destination = input.destination;
-            return;
-          }
-          // If init is provided, merge: use source properties, then override with init
-          // Copy all properties from the source Request first
+  (vmGlobalThis as any).setTimeout = () => {
+    throw new WorkflowRuntimeError(timeoutErrorMessage, {
+      slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
+    });
+  };
+  (vmGlobalThis as any).setInterval = () => {
+    throw new WorkflowRuntimeError(timeoutErrorMessage, {
+      slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
+    });
+  };
+  (vmGlobalThis as any).clearTimeout = () => {
+    throw new WorkflowRuntimeError(timeoutErrorMessage, {
+      slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
+    });
+  };
+  (vmGlobalThis as any).clearInterval = () => {
+    throw new WorkflowRuntimeError(timeoutErrorMessage, {
+      slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
+    });
+  };
+  (vmGlobalThis as any).setImmediate = () => {
+    throw new WorkflowRuntimeError(timeoutErrorMessage, {
+      slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
+    });
+  };
+  (vmGlobalThis as any).clearImmediate = () => {
+    throw new WorkflowRuntimeError(timeoutErrorMessage, {
+      slug: ERROR_SLUGS.TIMEOUT_FUNCTIONS_IN_WORKFLOW,
+    });
+  };
+
+  // `AbortController` and `AbortSignal` in the workflow VM are hook-backed
+  // for deterministic replay. The controller's abort() queues a hook resumption,
+  // and signal.aborted is updated when the hook event is processed during replay.
+  (vmGlobalThis as any).AbortController =
+    createCreateAbortController(workflowContext);
+  const abortSignalStatics = createAbortSignalStatics();
+  (vmGlobalThis as any).AbortSignal = {
+    abort: abortSignalStatics.abort,
+    any: abortSignalStatics.any,
+    timeout: abortSignalStatics.timeout,
+  };
+
+  // `Request` and `Response` are special built-in classes that invoke steps
+  // for the `json()`, `text()` and `arrayBuffer()` instance methods
+  class Request implements globalThis.Request {
+    cache!: globalThis.Request['cache'];
+    credentials!: globalThis.Request['credentials'];
+    destination!: globalThis.Request['destination'];
+    headers!: Headers;
+    integrity!: string;
+    method!: string;
+    mode!: globalThis.Request['mode'];
+    redirect!: globalThis.Request['redirect'];
+    referrer!: string;
+    referrerPolicy!: globalThis.Request['referrerPolicy'];
+    url!: string;
+    keepalive!: boolean;
+    signal!: AbortSignal;
+    duplex!: 'half';
+    body!: ReadableStream<any> | null;
+
+    constructor(input: any, init?: RequestInit) {
+      // Handle URL input
+      if (typeof input === 'string' || input instanceof vmGlobalThis.URL) {
+        const urlString = String(input);
+        // Validate URL format
+        try {
+          new vmGlobalThis.URL(urlString);
+          this.url = urlString;
+        } catch (cause) {
+          throw new TypeError(`Failed to parse URL from ${urlString}`, {
+            cause,
+          });
+        }
+      } else {
+        // Input is a Request object - clone its properties
+        this.url = input.url;
+        if (!init) {
           this.method = input.method;
           this.headers = new vmGlobalThis.Headers(input.headers);
           this.body = input.body;
@@ -460,437 +620,485 @@ export async function runWorkflow(
           this.signal = input.signal;
           this.duplex = input.duplex;
           this.destination = input.destination;
+          return;
         }
+        // If init is provided, merge: use source properties, then override with init
+        // Copy all properties from the source Request first
+        this.method = input.method;
+        this.headers = new vmGlobalThis.Headers(input.headers);
+        this.body = input.body;
+        this.mode = input.mode;
+        this.credentials = input.credentials;
+        this.cache = input.cache;
+        this.redirect = input.redirect;
+        this.referrer = input.referrer;
+        this.referrerPolicy = input.referrerPolicy;
+        this.integrity = input.integrity;
+        this.keepalive = input.keepalive;
+        this.signal = input.signal;
+        this.duplex = input.duplex;
+        this.destination = input.destination;
+      }
 
-        // Override with init options if provided
-        // Set method
-        if (init?.method) {
-          this.method = init.method.toUpperCase();
-        } else if (typeof this.method !== 'string') {
-          // Fallback to default for string input case
-          this.method = 'GET';
-        }
-
-        // Set headers
-        if (init?.headers) {
-          this.headers = new vmGlobalThis.Headers(init.headers);
-        } else if (
-          typeof input === 'string' ||
-          input instanceof vmGlobalThis.URL
-        ) {
-          // For string/URL input, create empty headers
-          this.headers = new vmGlobalThis.Headers();
-        }
+      // Override with init options if provided
+      // Set method
+      if (init?.method) {
+        this.method = init.method.toUpperCase();
+      } else if (typeof this.method !== 'string') {
+        // Fallback to default for string input case
+        this.method = 'GET';
+      }
 
-        // Set other properties with init values or defaults
-        if (init?.mode !== undefined) {
-          this.mode = init.mode;
-        } else if (typeof this.mode !== 'string') {
-          this.mode = 'cors';
-        }
+      // Set headers
+      if (init?.headers) {
+        this.headers = new vmGlobalThis.Headers(init.headers);
+      } else if (
+        typeof input === 'string' ||
+        input instanceof vmGlobalThis.URL
+      ) {
+        // For string/URL input, create empty headers
+        this.headers = new vmGlobalThis.Headers();
+      }
 
-        if (init?.credentials !== undefined) {
-          this.credentials = init.credentials;
-        } else if (typeof this.credentials !== 'string') {
-          this.credentials = 'same-origin';
-        }
+      // Set other properties with init values or defaults
+      if (init?.mode !== undefined) {
+        this.mode = init.mode;
+      } else if (typeof this.mode !== 'string') {
+        this.mode = 'cors';
+      }
 
-        // `any` cast here because @types/node v22 does not yet have `cache`
-        if ((init as any)?.cache !== undefined) {
-          this.cache = (init as any).cache;
-        } else if (typeof this.cache !== 'string') {
-          this.cache = 'default';
-        }
+      if (init?.credentials !== undefined) {
+        this.credentials = init.credentials;
+      } else if (typeof this.credentials !== 'string') {
+        this.credentials = 'same-origin';
+      }
 
-        if (init?.redirect !== undefined) {
-          this.redirect = init.redirect;
-        } else if (typeof this.redirect !== 'string') {
-          this.redirect = 'follow';
-        }
+      // `any` cast here because @types/node v22 does not yet have `cache`
+      if ((init as any)?.cache !== undefined) {
+        this.cache = (init as any).cache;
+      } else if (typeof this.cache !== 'string') {
+        this.cache = 'default';
+      }
 
-        if (init?.referrer !== undefined) {
-          this.referrer = init.referrer;
-        } else if (typeof this.referrer !== 'string') {
-          this.referrer = 'about:client';
-        }
+      if (init?.redirect !== undefined) {
+        this.redirect = init.redirect;
+      } else if (typeof this.redirect !== 'string') {
+        this.redirect = 'follow';
+      }
 
-        if (init?.referrerPolicy !== undefined) {
-          this.referrerPolicy = init.referrerPolicy;
-        } else if (typeof this.referrerPolicy !== 'string') {
-          this.referrerPolicy = '';
-        }
+      if (init?.referrer !== undefined) {
+        this.referrer = init.referrer;
+      } else if (typeof this.referrer !== 'string') {
+        this.referrer = 'about:client';
+      }
 
-        if (init?.integrity !== undefined) {
-          this.integrity = init.integrity;
-        } else if (typeof this.integrity !== 'string') {
-          this.integrity = '';
-        }
+      if (init?.referrerPolicy !== undefined) {
+        this.referrerPolicy = init.referrerPolicy;
+      } else if (typeof this.referrerPolicy !== 'string') {
+        this.referrerPolicy = '';
+      }
 
-        if (init?.keepalive !== undefined) {
-          this.keepalive = init.keepalive;
-        } else if (typeof this.keepalive !== 'boolean') {
-          this.keepalive = false;
-        }
+      if (init?.integrity !== undefined) {
+        this.integrity = init.integrity;
+      } else if (typeof this.integrity !== 'string') {
+        this.integrity = '';
+      }
 
-        if (init?.signal !== undefined) {
-          // @ts-expect-error - AbortSignal stub
-          this.signal = init.signal;
-        } else if (!this.signal) {
-          // @ts-expect-error - AbortSignal stub
-          this.signal = { aborted: false };
-        }
+      if (init?.keepalive !== undefined) {
+        this.keepalive = init.keepalive;
+      } else if (typeof this.keepalive !== 'boolean') {
+        this.keepalive = false;
+      }
 
-        if (!this.duplex) {
-          this.duplex = 'half';
-        }
+      if (init?.signal !== undefined) {
+        // @ts-expect-error - AbortSignal stub
+        this.signal = init.signal;
+      } else if (!this.signal) {
+        // @ts-expect-error - AbortSignal stub
+        this.signal = { aborted: false };
+      }
 
-        if (!this.destination) {
-          this.destination = 'document';
-        }
+      if (!this.duplex) {
+        this.duplex = 'half';
+      }
 
-        const body = init?.body;
+      if (!this.destination) {
+        this.destination = 'document';
+      }
 
-        // Validate that GET/HEAD methods don't have a body
-        if (
-          body !== null &&
-          body !== undefined &&
-          (this.method === 'GET' || this.method === 'HEAD')
-        ) {
-          throw new TypeError(`Request with GET/HEAD method cannot have body.`);
-        }
+      const body = init?.body;
 
-        // Store the original BodyInit for serialization
-        if (body !== null && body !== undefined) {
-          // Create a "fake" ReadableStream that stores the original body
-          // This avoids doing async work during workflow replay
-          this.body = Object.create(vmGlobalThis.ReadableStream.prototype, {
-            [BODY_INIT_SYMBOL]: {
-              value: body,
-              writable: false,
-            },
-          });
-        } else {
-          this.body = null;
-        }
+      // Validate that GET/HEAD methods don't have a body
+      if (
+        body !== null &&
+        body !== undefined &&
+        (this.method === 'GET' || this.method === 'HEAD')
+      ) {
+        throw new TypeError(`Request with GET/HEAD method cannot have body.`);
       }
 
-      clone(): Request {
-        ENOTSUP();
+      // Store the original BodyInit for serialization
+      if (body !== null && body !== undefined) {
+        // Create a "fake" ReadableStream that stores the original body
+        // This avoids doing async work during workflow replay
+        this.body = Object.create(vmGlobalThis.ReadableStream.prototype, {
+          [BODY_INIT_SYMBOL]: {
+            value: body,
+            writable: false,
+          },
+        });
+      } else {
+        this.body = null;
       }
+    }
 
-      get bodyUsed() {
-        return false;
-      }
+    clone(): Request {
+      ENOTSUP();
+    }
 
-      // TODO: implement these
-      blob!: () => Promise<Blob>;
-      formData!: () => Promise<FormData>;
+    get bodyUsed() {
+      return false;
+    }
 
-      arrayBuffer!: () => Promise<ArrayBuffer>;
-      json!: () => Promise<any>;
-      text!: () => Promise<string>;
+    // TODO: implement these
+    blob!: () => Promise<Blob>;
+    formData!: () => Promise<FormData>;
 
-      async bytes() {
-        return new Uint8Array(await this.arrayBuffer());
-      }
+    arrayBuffer!: () => Promise<ArrayBuffer>;
+    json!: () => Promise<any>;
+    text!: () => Promise<string>;
+
+    async bytes() {
+      return new Uint8Array(await this.arrayBuffer());
     }
-    vmGlobalThis.Request = Request;
-
-    Object.defineProperties(Request.prototype, {
-      arrayBuffer: {
-        value: useStep<[], ArrayBuffer>('__builtin_response_array_buffer'),
-        writable: true,
-        configurable: true,
-      },
-      json: {
-        value: useStep<[], any>('__builtin_response_json'),
-        writable: true,
-        configurable: true,
-      },
-      text: {
-        value: useStep<[], string>('__builtin_response_text'),
-        writable: true,
-        configurable: true,
-      },
-    });
+  }
+  vmGlobalThis.Request = Request;
+
+  Object.defineProperties(Request.prototype, {
+    arrayBuffer: {
+      value: useStep<[], ArrayBuffer>('__builtin_response_array_buffer'),
+      writable: true,
+      configurable: true,
+    },
+    json: {
+      value: useStep<[], any>('__builtin_response_json'),
+      writable: true,
+      configurable: true,
+    },
+    text: {
+      value: useStep<[], string>('__builtin_response_text'),
+      writable: true,
+      configurable: true,
+    },
+  });
 
-    class Response implements globalThis.Response {
-      type!: globalThis.Response['type'];
-      url!: string;
-      status!: number;
-      statusText!: string;
-      body!: ReadableStream<Uint8Array> | null;
-      headers!: Headers;
-      redirected!: boolean;
-
-      constructor(body?: any, init?: ResponseInit) {
-        this.status = init?.status ?? 200;
-        this.statusText = init?.statusText ?? '';
-        this.headers = new vmGlobalThis.Headers(init?.headers);
-        this.type = 'default';
-        this.url = '';
-        this.redirected = false;
-
-        // Validate that null-body status codes don't have a body
-        // Per HTTP spec: 204 (No Content), 205 (Reset Content), and 304 (Not Modified)
-        if (
-          body !== null &&
-          body !== undefined &&
-          (this.status === 204 || this.status === 205 || this.status === 304)
-        ) {
-          throw new TypeError(
-            `Response constructor: Invalid response status code ${this.status}`
-          );
-        }
+  class Response implements globalThis.Response {
+    type!: globalThis.Response['type'];
+    url!: string;
+    status!: number;
+    statusText!: string;
+    body!: ReadableStream<Uint8Array> | null;
+    headers!: Headers;
+    redirected!: boolean;
+
+    constructor(body?: any, init?: ResponseInit) {
+      this.status = init?.status ?? 200;
+      this.statusText = init?.statusText ?? '';
+      this.headers = new vmGlobalThis.Headers(init?.headers);
+      this.type = 'default';
+      this.url = '';
+      this.redirected = false;
+
+      // Validate that null-body status codes don't have a body
+      // Per HTTP spec: 204 (No Content), 205 (Reset Content), and 304 (Not Modified)
+      if (
+        body !== null &&
+        body !== undefined &&
+        (this.status === 204 || this.status === 205 || this.status === 304)
+      ) {
+        throw new TypeError(
+          `Response constructor: Invalid response status code ${this.status}`
+        );
+      }
 
-        // Store the original BodyInit for serialization
-        if (body !== null && body !== undefined) {
-          // Create a "fake" ReadableStream that stores the original body
-          // This avoids doing async work during workflow replay
-          this.body = Object.create(vmGlobalThis.ReadableStream.prototype, {
-            [BODY_INIT_SYMBOL]: {
-              value: body,
-              writable: false,
-            },
-          });
-        } else {
-          this.body = null;
-        }
+      // Store the original BodyInit for serialization
+      if (body !== null && body !== undefined) {
+        // Create a "fake" ReadableStream that stores the original body
+        // This avoids doing async work during workflow replay
+        this.body = Object.create(vmGlobalThis.ReadableStream.prototype, {
+          [BODY_INIT_SYMBOL]: {
+            value: body,
+            writable: false,
+          },
+        });
+      } else {
+        this.body = null;
       }
+    }
 
-      // TODO: implement these
-      clone!: () => Response;
-      blob!: () => Promise<globalThis.Blob>;
-      formData!: () => Promise<globalThis.FormData>;
+    // TODO: implement these
+    clone!: () => Response;
+    blob!: () => Promise<globalThis.Blob>;
+    formData!: () => Promise<globalThis.FormData>;
 
-      get ok() {
-        return this.status >= 200 && this.status < 300;
-      }
+    get ok() {
+      return this.status >= 200 && this.status < 300;
+    }
 
-      get bodyUsed() {
-        return false;
-      }
+    get bodyUsed() {
+      return false;
+    }
 
-      arrayBuffer!: () => Promise<ArrayBuffer>;
-      json!: () => Promise<any>;
-      text!: () => Promise<string>;
+    arrayBuffer!: () => Promise<ArrayBuffer>;
+    json!: () => Promise<any>;
+    text!: () => Promise<string>;
 
-      async bytes() {
-        return new Uint8Array(await this.arrayBuffer());
-      }
+    async bytes() {
+      return new Uint8Array(await this.arrayBuffer());
+    }
 
-      static json(data: any, init?: ResponseInit): Response {
-        const body = JSON.stringify(data);
-        const headers = new vmGlobalThis.Headers(init?.headers);
-        if (!headers.has('content-type')) {
-          headers.set('content-type', 'application/json');
-        }
-        return new Response(body, { ...init, headers });
+    static json(data: any, init?: ResponseInit): Response {
+      const body = JSON.stringify(data);
+      const headers = new vmGlobalThis.Headers(init?.headers);
+      if (!headers.has('content-type')) {
+        headers.set('content-type', 'application/json');
       }
+      return new Response(body, { ...init, headers });
+    }
 
-      static error(): Response {
-        ENOTSUP();
-      }
+    static error(): Response {
+      ENOTSUP();
+    }
 
-      static redirect(url: string | URL, status: number = 302): Response {
-        // Validate status code - only specific redirect codes are allowed
-        if (![301, 302, 303, 307, 308].includes(status)) {
-          throw new RangeError(
-            `Invalid redirect status code: ${status}. Must be one of: 301, 302, 303, 307, 308`
-          );
-        }
+    static redirect(url: string | URL, status: number = 302): Response {
+      // Validate status code - only specific redirect codes are allowed
+      if (![301, 302, 303, 307, 308].includes(status)) {
+        throw new RangeError(
+          `Invalid redirect status code: ${status}. Must be one of: 301, 302, 303, 307, 308`
+        );
+      }
 
-        // Create response with Location header
-        const headers = new vmGlobalThis.Headers();
-        headers.set('Location', String(url));
+      // Create response with Location header
+      const headers = new vmGlobalThis.Headers();
+      headers.set('Location', String(url));
 
-        const response = Object.create(Response.prototype);
-        response.status = status;
-        response.statusText = '';
-        response.headers = headers;
-        response.body = null;
-        response.type = 'default';
-        response.url = '';
-        response.redirected = false;
+      const response = Object.create(Response.prototype);
+      response.status = status;
+      response.statusText = '';
+      response.headers = headers;
+      response.body = null;
+      response.type = 'default';
+      response.url = '';
+      response.redirected = false;
 
-        return response;
-      }
+      return response;
     }
-    vmGlobalThis.Response = Response;
-
-    Object.defineProperties(Response.prototype, {
-      arrayBuffer: {
-        value: useStep<[], ArrayBuffer>('__builtin_response_array_buffer'),
-        writable: true,
-        configurable: true,
-      },
-      json: {
-        value: useStep<[], any>('__builtin_response_json'),
-        writable: true,
-        configurable: true,
-      },
-      text: {
-        value: useStep<[], string>('__builtin_response_text'),
-        writable: true,
-        configurable: true,
-      },
-    });
+  }
+  vmGlobalThis.Response = Response;
+
+  Object.defineProperties(Response.prototype, {
+    arrayBuffer: {
+      value: useStep<[], ArrayBuffer>('__builtin_response_array_buffer'),
+      writable: true,
+      configurable: true,
+    },
+    json: {
+      value: useStep<[], any>('__builtin_response_json'),
+      writable: true,
+      configurable: true,
+    },
+    text: {
+      value: useStep<[], string>('__builtin_response_text'),
+      writable: true,
+      configurable: true,
+    },
+  });
 
-    class ReadableStream<T> implements globalThis.ReadableStream<T> {
-      constructor() {
-        ENOTSUP();
-      }
+  class ReadableStream<T> implements globalThis.ReadableStream<T> {
+    constructor() {
+      ENOTSUP();
+    }
 
-      get locked() {
-        return false;
-      }
+    get locked() {
+      return false;
+    }
 
-      cancel(): any {
-        ENOTSUP();
-      }
+    cancel(): any {
+      ENOTSUP();
+    }
 
-      getReader(): any {
-        ENOTSUP();
-      }
+    getReader(): any {
+      ENOTSUP();
+    }
 
-      pipeThrough(): any {
-        ENOTSUP();
-      }
+    pipeThrough(): any {
+      ENOTSUP();
+    }
 
-      pipeTo(): any {
-        ENOTSUP();
-      }
+    pipeTo(): any {
+      ENOTSUP();
+    }
 
-      tee(): any {
-        ENOTSUP();
-      }
+    tee(): any {
+      ENOTSUP();
+    }
 
-      values(): any {
-        ENOTSUP();
-      }
+    values(): any {
+      ENOTSUP();
+    }
 
-      static from(): any {
-        ENOTSUP();
-      }
+    static from(): any {
+      ENOTSUP();
+    }
 
-      [Symbol.asyncIterator](): any {
-        ENOTSUP();
-      }
+    [Symbol.asyncIterator](): any {
+      ENOTSUP();
     }
-    vmGlobalThis.ReadableStream = ReadableStream;
+  }
+  vmGlobalThis.ReadableStream = ReadableStream;
 
-    class WritableStream<T> implements globalThis.WritableStream<T> {
-      constructor() {
-        ENOTSUP();
-      }
+  class WritableStream<T> implements globalThis.WritableStream<T> {
+    constructor() {
+      ENOTSUP();
+    }
 
-      get locked() {
-        return false;
-      }
+    get locked() {
+      return false;
+    }
 
-      abort(): any {
-        ENOTSUP();
-      }
+    abort(): any {
+      ENOTSUP();
+    }
 
-      close(): any {
-        ENOTSUP();
-      }
+    close(): any {
+      ENOTSUP();
+    }
 
-      getWriter(): any {
-        ENOTSUP();
-      }
+    getWriter(): any {
+      ENOTSUP();
     }
-    vmGlobalThis.WritableStream = WritableStream;
+  }
+  vmGlobalThis.WritableStream = WritableStream;
 
-    class TransformStream<I, O> implements globalThis.TransformStream<I, O> {
-      readable: globalThis.ReadableStream<O>;
-      writable: globalThis.WritableStream<I>;
+  class TransformStream<I, O> implements globalThis.TransformStream<I, O> {
+    readable: globalThis.ReadableStream<O>;
+    writable: globalThis.WritableStream<I>;
 
-      constructor() {
-        ENOTSUP();
-      }
+    constructor() {
+      ENOTSUP();
     }
-    vmGlobalThis.TransformStream = TransformStream;
-
-    // Eventually we'll probably want to provide our own `console` object,
-    // but for now we'll just expose the global one.
-    vmGlobalThis.console = globalThis.console;
-
-    // HACK: propagate symbol needed for AI gateway usage
-    const SYMBOL_FOR_REQ_CONTEXT = Symbol.for('@vercel/request-context');
-    // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
-    vmGlobalThis[SYMBOL_FOR_REQ_CONTEXT] = (globalThis as any)[
-      SYMBOL_FOR_REQ_CONTEXT
-    ];
-
-    // Get a reference to the user-defined workflow function.
-    // The filename parameter ensures stack traces show a meaningful name
-    // (e.g., "example/workflows/99_e2e.ts") instead of "evalmachine.<anonymous>".
-    const parsedName = parseWorkflowName(workflowRun.workflowName);
-    const filename = parsedName?.moduleSpecifier || workflowRun.workflowName;
-
-    // Evaluate the workflow bundle against the fresh context using a
-    // process-wide cache of the compiled `vm.Script`. The bundle is the same
-    // string for every replay and every invocation in this process, and
-    // compilation is a pure function of `(code, filename)`, so reusing the
-    // compiled Script across replays is determinism-safe: it produces the same
-    // workflow function and the same `filename` source attribution as
-    // re-parsing the bundle every time, but skips the (expensive) re-parse.
-    // Evaluating the bundle registers every workflow on
-    // `globalThis.__private_workflows`; the trailing lookup expression then
-    // retrieves the requested workflow function. The lookup is evaluated as a
-    // separate cached Script under the same `filename`, so error stack frames
-    // still attribute to the workflow's source file (`remapErrorStack` keys on
-    // `filename`). The one behavioural difference from the previous
-    // single-combined-string approach is the *line number* of an error thrown
-    // by the lookup expression itself: it now reports line 1 of the lookup
-    // Script rather than the line just past the end of the bundle. That path
-    // is rare (it requires the lookup `?.get(...)` expression to throw) and
-    // does not affect the workflow function or replay determinism.
-    runCachedWorkflowScript(workflowCode, filename, context);
-    const workflowFn = runCachedWorkflowScript(
-      `globalThis.__private_workflows?.get(${JSON.stringify(workflowRun.workflowName)})`,
-      filename,
-      context
-    );
+  }
+  vmGlobalThis.TransformStream = TransformStream;
+
+  // Eventually we'll probably want to provide our own `console` object,
+  // but for now we'll just expose the global one.
+  vmGlobalThis.console = globalThis.console;
+
+  // HACK: propagate symbol needed for AI gateway usage
+  const SYMBOL_FOR_REQ_CONTEXT = Symbol.for('@vercel/request-context');
+  // @ts-expect-error - `@types/node` says symbol is not valid, but it does work
+  vmGlobalThis[SYMBOL_FOR_REQ_CONTEXT] = (globalThis as any)[
+    SYMBOL_FOR_REQ_CONTEXT
+  ];
+
+  // Get a reference to the user-defined workflow function.
+  // The filename parameter ensures stack traces show a meaningful name
+  // (e.g., "example/workflows/99_e2e.ts") instead of "evalmachine.<anonymous>".
+  const parsedName = parseWorkflowName(workflowRun.workflowName);
+  const filename = parsedName?.moduleSpecifier || workflowRun.workflowName;
+
+  // Evaluate the workflow bundle against the fresh context using a
+  // process-wide cache of the compiled `vm.Script`. The bundle is the same
+  // string for every replay and every invocation in this process, and
+  // compilation is a pure function of `(code, filename)`, so reusing the
+  // compiled Script across replays is determinism-safe: it produces the same
+  // workflow function and the same `filename` source attribution as
+  // re-parsing the bundle every time, but skips the (expensive) re-parse.
+  // Evaluating the bundle registers every workflow on
+  // `globalThis.__private_workflows`; the trailing lookup expression then
+  // retrieves the requested workflow function. The lookup is evaluated as a
+  // separate cached Script under the same `filename`, so error stack frames
+  // still attribute to the workflow's source file (`remapErrorStack` keys on
+  // `filename`). The one behavioural difference from the previous
+  // single-combined-string approach is the *line number* of an error thrown
+  // by the lookup expression itself: it now reports line 1 of the lookup
+  // Script rather than the line just past the end of the bundle. That path
+  // is rare (it requires the lookup `?.get(...)` expression to throw) and
+  // does not affect the workflow function or replay determinism.
+  runCachedWorkflowScript(workflowCode, filename, context);
+  const workflowFn = runCachedWorkflowScript(
+    `globalThis.__private_workflows?.get(${JSON.stringify(workflowRun.workflowName)})`,
+    filename,
+    context
+  );
+
+  if (typeof workflowFn !== 'function') {
+    throw new WorkflowNotRegisteredError(workflowRun.workflowName);
+  }
 
-    if (typeof workflowFn !== 'function') {
-      throw new WorkflowNotRegisteredError(workflowRun.workflowName);
-    }
-
-    // Chain workflow argument hydration onto the promiseQueue so that the
-    // unconsumed event check (which waits for the queue to drain) doesn't
-    // fire during the async gap between run_started consumption and the
-    // workflow function subscribing its first step callbacks.
-    let args: unknown[] = [];
-    workflowContext.promiseQueue = workflowContext.promiseQueue.then(
-      async () => {
-        const prepared =
-          await replayPayloadCache.prepareWorkflowInput(workflowRun);
-        args = await hydrateWorkflowArguments(
-          workflowRun.input,
-          workflowRun.runId,
-          encryptionKey,
-          vmGlobalThis,
-          {},
-          prepared
-        );
-      }
+  // Chain workflow argument hydration onto the promiseQueue so that the
+  // unconsumed event check (which waits for the queue to drain) doesn't
+  // fire during the async gap between run_started consumption and the
+  // workflow function subscribing its first step callbacks.
+  let args: unknown[] = [];
+  workflowContext.promiseQueue = workflowContext.promiseQueue.then(async () => {
+    const prepared = await replayPayloadCache.prepareWorkflowInput(workflowRun);
+    args = await hydrateWorkflowArguments(
+      workflowRun.input,
+      workflowRun.runId,
+      encryptionKey,
+      vmGlobalThis,
+      {},
+      prepared
+    );
+  });
+  await workflowContext.promiseQueue;
+
+  const workflowExecution = (async (): Promise<unknown> => {
+    return await workflowFn(...args);
+  })();
+
+  const failWorkflow = async (error: unknown): Promise<never> => {
+    // Control-flow signals are handled by the runtime and do not mean the
+    // workflow has terminally failed. `onWorkflowError` usually already moved
+    // the state machine, but a divergence can also arrive via a step
+    // promise's direct rejection (bypassing `onWorkflowError`) — demote so
+    // every control-flow path converges on `replay` and a later resume falls
+    // back instead of throwing.
+    if (WorkflowSuspension.is(error) || ReplayDivergenceError.is(error)) {
+      if (state.type === 'running') state = { type: 'replay' };
+      throw error;
+    }
+    state = { type: 'completed' };
+
+    await drainPendingQueueItems(
+      workflowRun.runId,
+      workflowContext.invocationsQueue,
+      vmGlobalThis,
+      workflowRun,
+      'failed',
+      runReadyBarrier
     );
-    await workflowContext.promiseQueue;
 
-    span?.setAttributes({
-      ...Attribute.WorkflowArgumentsCount(args.length),
-    });
+    throw error;
+  };
 
-    // Invoke user workflow
+  const waitForExecution = async (
+    interruption: PromiseWithResolvers<never>
+  ): Promise<WorkflowCompletion> => {
+    let result: unknown;
     try {
-      const result = await Promise.race([
-        workflowFn(...args),
-        workflowDiscontinuation.promise,
-      ]);
+      result = await Promise.race([workflowExecution, interruption.promise]);
+    } catch (error) {
+      if (state.type === 'suspended' && error === state.suspension) {
+        throw error;
+      }
+      return failWorkflow(error);
+    }
 
-      const dehydrated = await dehydrateWorkflowReturnValue(
+    state = { type: 'completed' };
+    try {
+      const output = await dehydrateWorkflowReturnValue(
         result,
         workflowRun.runId,
         encryptionKey,
@@ -902,10 +1110,6 @@ export async function runWorkflow(
         (workflowRun.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION
       );
 
-      span?.setAttributes({
-        ...Attribute.WorkflowResultType(typeof result),
-      });
-
       await drainPendingQueueItems(
         workflowRun.runId,
         workflowContext.invocationsQueue,
@@ -915,24 +1119,51 @@ export async function runWorkflow(
         runReadyBarrier
       );
 
-      return dehydrated;
-    } catch (err) {
-      // Control-flow signals are handled by the runtime and do not mean the
-      // workflow has terminally failed.
-      if (WorkflowSuspension.is(err) || ReplayDivergenceError.is(err)) {
-        throw err;
-      }
-
-      await drainPendingQueueItems(
-        workflowRun.runId,
-        workflowContext.invocationsQueue,
-        vmGlobalThis,
-        workflowRun,
-        'failed',
-        runReadyBarrier
-      );
-
-      throw err;
+      return { output, resultType: typeof result };
+    } catch (error) {
+      return failWorkflow(error);
     }
-  });
+  };
+
+  const session: WorkflowSession = {
+    workflowRun,
+    argumentCount: args.length,
+    resume(nextEvents) {
+      switch (state.type) {
+        case 'suspended': {
+          const consumedEvents = eventsConsumer.events;
+          const isStrictExtension =
+            nextEvents.length > consumedEvents.length &&
+            consumedEvents.every(
+              (event, index) => event.eventId === nextEvents[index]?.eventId
+            );
+          if (!isStrictExtension) {
+            state = { type: 'replay' };
+            return { type: 'replay' };
+          }
+          const interruption = withResolvers<never>();
+          state = { type: 'running', interruption };
+          workflowContext.suspensionGeneration++;
+          eventsConsumer.append(nextEvents.slice(consumedEvents.length));
+          return {
+            type: 'resumed',
+            execution: waitForExecution(interruption),
+          };
+        }
+        case 'replay':
+          return { type: 'replay' };
+        case 'completed':
+        case 'running':
+          throw new WorkflowRuntimeError(
+            `Cannot resume ${state.type} workflow "${workflowRun.runId}"`
+          );
+      }
+      state satisfies never;
+    },
+  };
+
+  return {
+    session,
+    execution: waitForExecution(initialInterruption),
+  };
 }