feat(core): experimental in-process VM continuation for inline replay

.changeset/vm-continuation.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': minor
+---
+
+Add in-process VM continuation (on by default; kill switch `WORKFLOW_VM_CONTINUATION=0`). The inline replay loop now keeps a suspended workflow VM alive and feeds newly-appended events into it for step-only suspensions, instead of rebuilding the `vm.Context` and replaying the event log from scratch each pass. Falls back to a full replay on any prefix divergence or non-step suspension, so the kill switch restores the previous behavior exactly.

packages/core/src/events-consumer.ts

@@ -1,3 +1,4 @@
+import { ReplayDivergenceError } from '@workflow/errors';
 import { type Event, envNumber } from '@workflow/world';
 import { eventsLogger } from './logger.js';
 
@@ -69,7 +70,7 @@ export interface EventsConsumerOptions {
 
 export class EventsConsumer {
   eventIndex: number;
-  readonly events: Event[] = [];
+  events: Event[] = [];
   readonly callbacks: EventConsumerCallback[] = [];
   private onConsumedEvent?: (event: Event) => void;
   private onUnconsumedEvent: (event: Event) => void;
@@ -111,6 +112,39 @@ export class EventsConsumer {
     process.nextTick(this.consume);
   }
 
+  /**
+   * Resume consuming against an updated event log without rebuilding the VM.
+   *
+   * Used by the in-process VM-continuation path (see
+   * `isVmContinuationEnabled`): after new events are appended to the run's log
+   * (e.g. an inline step's `step_completed`), the loop adopts the authoritative
+   * array here and re-drives {@link consume} so the still-registered step
+   * consumers advance the SAME live workflow VM to its next suspension point.
+   *
+   * Safety: the events already consumed by this instance (indices
+   * `0..eventIndex`) MUST be a byte-stable prefix of `events` — the durable log
+   * is the source of truth, and a live VM that consumed a now-diverged prefix
+   * can no longer be trusted. Any mismatch throws {@link ReplayDivergenceError}
+   * so the caller falls back to a full replay in a fresh VM. Matching is by
+   * `eventId`, which is immutable per event across reads.
+   */
+  resume(events: Event[]): void {
+    for (let i = 0; i < this.eventIndex; i++) {
+      const consumed = this.events[i];
+      const incoming = events[i];
+      if (!incoming || consumed?.eventId !== incoming.eventId) {
+        throw new ReplayDivergenceError(
+          `VM continuation prefix diverged at index ${i}: consumed ` +
+            `${consumed?.eventId ?? '<none>'} but the authoritative log has ` +
+            `${incoming?.eventId ?? '<none>'}.`,
+          { eventId: incoming?.eventId ?? consumed?.eventId }
+        );
+      }
+    }
+    this.events = events;
+    process.nextTick(this.consume);
+  }
+
   private notifyConsumedEvent(event: Event) {
     if (!this.onConsumedEvent) {
       return;

packages/core/src/global.ts

@@ -68,6 +68,23 @@ export type QueueItem =
   | WaitInvocationQueueItem
   | AttributeInvocationQueueItem;
 
+/**
+ * Handle attached to a {@link WorkflowSuspension} when in-process VM
+ * continuation is armed (see `isVmContinuationEnabled`). Calling `resume`
+ * feeds newly-appended events into the SAME live workflow VM that produced the
+ * suspension and drives it to its next checkpoint, returning the workflow's
+ * dehydrated result on completion or throwing the next `WorkflowSuspension`
+ * (which itself may carry a fresh `continuation`). It throws
+ * `ReplayDivergenceError` if the live VM's consumed prefix no longer matches
+ * the authoritative log, signalling the caller to fall back to a full replay.
+ *
+ * `events` is the authoritative event array (`@workflow/world` `Event[]`),
+ * typed loosely here to keep `global.ts` free of the world dependency.
+ */
+export interface WorkflowContinuation {
+  resume(events: unknown[]): Promise<Uint8Array | unknown>;
+}
+
 /**
  * An error that is thrown when one or more operations (steps/hooks/etc.) are called but do
  * not yet have corresponding entries in the event log. The workflow
@@ -83,6 +100,14 @@ export class WorkflowSuspension extends Error {
   attributeCount: number;
   hookDisposedCount: number;
   abortCount: number;
+  /**
+   * Set by `runWorkflow` when in-process VM continuation is armed for this
+   * (step-only) suspension. When present, the inline replay loop can resume
+   * the SAME live VM with newly-appended events instead of rebuilding the
+   * context and replaying from scratch. Absent when continuation is disabled
+   * or the suspension is not continuation-eligible.
+   */
+  continuation?: WorkflowContinuation;
 
   constructor(stepsInput: Map<string, QueueItem>, global: typeof globalThis) {
     // Convert Map to array for iteration and storage

packages/core/src/runtime.ts

@@ -32,14 +32,19 @@ import {
   isWorldContractError,
 } from './classify-error.js';
 import { describeError } from './describe-error.js';
-import { type StepInvocationQueueItem, WorkflowSuspension } from './global.js';
+import {
+  type StepInvocationQueueItem,
+  type WorkflowContinuation,
+  WorkflowSuspension,
+} from './global.js';
 import { runtimeLogger } from './logger.js';
 import { ReplayPayloadCache } from './replay-payload-cache.js';
 import {
   getMaxQueueDeliveries,
   getReplayDivergenceMaxRetries,
   isInlineOwnershipEnabled,
   isTurboEnabled,
+  isVmContinuationEnabled,
 } from './runtime/constants.js';
 import {
   getQueueOverhead,
@@ -547,6 +552,17 @@ export function workflowEntrypoint(
                     cursor: string | null;
                   } | null = null;
 
+                  // In-process VM continuation (experimental, off by default —
+                  // see isVmContinuationEnabled). When a step-only suspension
+                  // arms a continuation, it is stashed here so the NEXT loop
+                  // iteration resumes the SAME live workflow VM with the freshly
+                  // appended events instead of rebuilding the context and
+                  // replaying the whole event log. Cleared whenever a resume
+                  // diverges (falls back to a fresh replay) or the suspension is
+                  // not continuation-eligible.
+                  const vmContinuationEnabled = isVmContinuationEnabled();
+                  let pendingContinuation: WorkflowContinuation | null = null;
+
                   // Shared state: set by either the background step path
                   // or the run_started setup below.
                   let workflowRun: WorkflowRun | undefined;
@@ -1429,22 +1445,56 @@ export function workflowEntrypoint(
                       // Start every missing decrypt/decompress operation before
                       // VM setup. Web Crypto work can overlap bundle evaluation;
                       // consumers still deserialize and resolve in event order.
+                      // Also warms the cache for a continuation resume, whose
+                      // step consumers hydrate the newly-appended results.
                       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
-                      );
+                      // Fresh replay in a new VM, unless a prior step-only
+                      // suspension armed a continuation — in which case resume
+                      // the SAME live VM with the freshly appended events. Both
+                      // paths obey the same contract: return the dehydrated
+                      // result, or throw a WorkflowSuspension (which may carry a
+                      // fresh continuation). A resume whose consumed prefix has
+                      // diverged from the authoritative log throws
+                      // ReplayDivergenceError; we swallow it here and fall back
+                      // to a full replay in a fresh VM so behavior degrades to
+                      // exactly today's.
+                      const runReplay = () =>
+                        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,
+                          vmContinuationEnabled
+                        );
+                      let result: Uint8Array | unknown;
+                      if (pendingContinuation) {
+                        const continuation = pendingContinuation;
+                        pendingContinuation = null;
+                        try {
+                          result = await continuation.resume(events);
+                        } catch (resumeErr) {
+                          if (ReplayDivergenceError.is(resumeErr)) {
+                            runtimeLogger.debug(
+                              'VM continuation resume diverged; falling back to full replay',
+                              { workflowRunId: runId, loopIteration }
+                            );
+                            result = await runReplay();
+                          } else {
+                            throw resumeErr;
+                          }
+                        }
+                      } else {
+                        result = await runReplay();
+                      }
                       await payloadPrewarm;
                       runtimeLogger.debug('Workflow replay completed', {
                         workflowRunId: runId,
@@ -1496,6 +1546,14 @@ export function workflowEntrypoint(
                       return;
                     } catch (err) {
                       if (WorkflowSuspension.is(err)) {
+                        // VM continuation: a step-only suspension carries a
+                        // handle that resumes the same live VM next iteration.
+                        // Non-eligible suspensions (hooks/waits/attrs) leave it
+                        // undefined, so we fall back to a fresh replay. The
+                        // handle is process-local and dropped if this invocation
+                        // returns instead of looping.
+                        pendingContinuation = err.continuation ?? null;
+
                         // Synchronous `runWorkflow` duration for THIS
                         // suspension only — anchors the `finalSchedulingReplay`
                         // telemetry field below (see

packages/core/src/runtime/constants.ts

@@ -293,6 +293,34 @@ export function isInlineOwnershipEnabled(): boolean {
   return !(raw === '0' || raw.toLowerCase() === 'false');
 }
 
+/**
+ * Whether in-process VM continuation is enabled (default OFF — experimental).
+ *
+ * When on, the inline replay loop keeps a suspended workflow VM alive across
+ * iterations instead of rebuilding the `vm.Context` and replaying the whole
+ * event log from event 1 on every pass. After an inline step's terminal event
+ * is appended, the loop feeds the newly-appended events into the SAME live VM
+ * (resolving the pending step promise) so the workflow body advances to its
+ * next suspension point — skipping the from-scratch replay entirely.
+ *
+ * This is sound ONLY within a single invocation's inline loop: the same process
+ * holds the VM and the same handler owns the appended writes, so the durable
+ * event log stays authoritative. It is scoped further to *simple, step-only*
+ * suspensions (no hooks/waits/attribute writes); any other suspension, or any
+ * divergence between the live VM's consumed prefix and the authoritative log,
+ * makes the continuation bail out and the loop falls back to a full replay in
+ * a fresh VM. Worst case is therefore identical to the from-scratch replay path.
+ *
+ * Default **ON** so CI and benchmarks exercise (and measure) the continuation
+ * path. `WORKFLOW_VM_CONTINUATION=0` (or `false`) is the kill switch that
+ * reverts to rebuilding the VM and replaying from scratch on every pass.
+ */
+export function isVmContinuationEnabled(): boolean {
+  const raw = process.env.WORKFLOW_VM_CONTINUATION;
+  if (raw === undefined || raw === '') return true;
+  return !(raw === '0' || raw.toLowerCase() === 'false');
+}
+
 /**
  * Default inline-ownership lease, in seconds: how long after a step's latest
  * (stamped) `step_started` a non-owner invocation assumes the owning