perf(core): retain workflow VM across inline steps (primitives-gated)

.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.

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>

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

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)', () => {

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:

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.