perf(core): retain workflow VM across inline steps

.changeset/retain-workflow-vm.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Retain workflow execution across inline steps within one invocation.

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/runtime.test.ts

@@ -10,6 +10,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';
@@ -701,6 +702,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',
@@ -759,6 +763,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 () => {

packages/core/src/runtime.ts

@@ -90,7 +90,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';
@@ -306,6 +310,25 @@ function hasOpenHookOrWait(events: Event[]): boolean {
   return false;
 }
 
+/**
+ * Retain only pure step boundaries with no out-of-band continuation source.
+ * Attributes require replay; hooks and waits can wake another invocation.
+ */
+function canRetainWorkflowSession(
+  suspension: WorkflowSuspension,
+  events: Event[]
+): boolean {
+  return (
+    suspension.stepCount > 0 &&
+    suspension.hookCount === 0 &&
+    suspension.waitCount === 0 &&
+    suspension.attributeCount === 0 &&
+    suspension.hookDisposedCount === 0 &&
+    suspension.abortCount === 0 &&
+    !hasOpenHookOrWait(events)
+  );
+}
+
 /**
  * Creates a single route which handles workflow execution requests,
  * executing steps inline when possible to reduce function invocations
@@ -1100,6 +1123,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) {
@@ -1414,37 +1444,76 @@ export function workflowEntrypoint(
                       // point and the inline executeStep mutates eventsCursor.
                       preInlineWriteCursor = eventsCursor;
 
-                      // Replay workflow
-                      runtimeLogger.debug('Starting workflow replay', {
+                      let executionMode = workflowExecution.type;
+                      runtimeLogger.debug('Starting workflow execution', {
                         workflowRunId: runId,
                         loopIteration,
                         eventCount: events.length,
+                        executionMode,
                       });
                       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
-                      );
-                      await payloadPrewarm;
-                      runtimeLogger.debug('Workflow replay completed', {
+                      let workflowResult: WorkflowExecutionResult = {
+                        type: 'replay',
+                      };
+                      if (workflowExecution.type === 'retained') {
+                        workflowResult = await executeWorkflow({
+                          type: 'resume',
+                          session: workflowExecution.session,
+                          events,
+                        });
+                      }
+
+                      if (workflowResult.type === 'replay') {
+                        executionMode = '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,
+                        });
+                        await payloadPrewarm;
+                      }
+
+                      if (workflowResult.type === 'suspended') {
+                        workflowExecution = canRetainWorkflowSession(
+                          workflowResult.suspension,
+                          events
+                        )
+                          ? {
+                              type: 'retained',
+                              session: workflowResult.session,
+                            }
+                          : { type: 'replay' };
+                        throw workflowResult.suspension;
+                      }
+
+                      if (workflowResult.type === 'replay') {
+                        throw new Error(
+                          'Invariant violation: fresh workflow execution requested another replay'
+                        );
+                      }
+
+                      const result = workflowResult.output;
+                      runtimeLogger.debug('Workflow execution completed', {
                         workflowRunId: runId,
                         loopIteration,
                         replayMs: Date.now() - replayStart,
+                        executionMode,
                       });
 
                       // Workflow completed. Send the snapshot but do NOT

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),

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'

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,
+    });
+  });
+});