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. diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 1236f5e160..098fde3df2 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -96,6 +96,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/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 53afa6de25..4ff997d99a 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -445,6 +445,31 @@ describe('e2e', () => { expect(returnValue).toBe('B'); }); + // Interleaves VM-retention modes: retained boundaries (primitive step + // args), demoted boundaries (object args), sleeps, a step-vs-sleep race, + // and a hook resolved in parallel with a step. Asserts the exact composite + // result so a dropped/duplicated/misordered boundary fails loudly. + test('retainedInterleavingWorkflow', { timeout: 90_000 }, async () => { + const token = Math.random().toString(36).slice(2); + const run = await start(await e2e('retainedInterleavingWorkflow'), [token]); + + const hook = await waitForHook(token, { runId: run.runId }); + await resumeHook(hook, { delta: 5 }); + + const returnValue = await run.returnValue; + expect(returnValue).toEqual({ + a: 3, + b: 3, + c: 13, + d: 23, + e: 13, + f: 24, + winner: 'step', + g: 137, + h: 142, + }); + }); + test.skipIf(!isNext)( 'importedStepOnlyWorkflow', { timeout: 60_000 }, diff --git a/packages/core/src/delivery-barrier-coverage.test.ts b/packages/core/src/delivery-barrier-coverage.test.ts index 836d7d3948..44248742e6 100644 --- a/packages/core/src/delivery-barrier-coverage.test.ts +++ b/packages/core/src/delivery-barrier-coverage.test.ts @@ -72,6 +72,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { const promiseQueueHolder = { current: Promise.resolve() }; const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { + suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, replayPayloadCache: new ReplayPayloadCache(undefined), diff --git a/packages/core/src/events-consumer.ts b/packages/core/src/events-consumer.ts index 4b7cf0742e..cc05c6f975 100644 --- a/packages/core/src/events-consumer.ts +++ b/packages/core/src/events-consumer.ts @@ -69,7 +69,7 @@ export interface EventsConsumerOptions { export class EventsConsumer { eventIndex: number; - readonly events: Event[] = []; + readonly events: Event[]; readonly callbacks: EventConsumerCallback[] = []; private onConsumedEvent?: (event: Event) => void; private onUnconsumedEvent: (event: Event) => void; @@ -79,13 +79,21 @@ export class EventsConsumer { private unconsumedCheckVersion = 0; constructor(events: Event[], options: EventsConsumerOptions) { - this.events = events; + // Own copy: the runtime mutates its event array in place, and a retained + // session must only observe new events through append() so resume()'s + // strict-extension check stays meaningful. + this.events = [...events]; this.eventIndex = 0; this.onConsumedEvent = options.onConsumedEvent; this.onUnconsumedEvent = options.onUnconsumedEvent; this.getPromiseQueue = options.getPromiseQueue; } + append(events: Event[]): void { + for (const event of events) this.events.push(event); + 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/hook-sleep-interaction.test.ts b/packages/core/src/hook-sleep-interaction.test.ts index 29d64fcc58..135cea5056 100644 --- a/packages/core/src/hook-sleep-interaction.test.ts +++ b/packages/core/src/hook-sleep-interaction.test.ts @@ -42,6 +42,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { // silently dropping them. const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { + suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, replayPayloadCache: new ReplayPayloadCache(undefined), diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index 16fb6a3e4e..5d93f46fcd 100644 --- a/packages/core/src/private.ts +++ b/packages/core/src/private.ts @@ -135,6 +135,16 @@ export interface WorkflowOrchestratorContext { encryptionKey: PayloadKey | undefined; worldCapabilities?: WorldCapabilities; globalThis: typeof globalThis; + /** + * Increments when a suspension is accepted and on every retained-session + * resume. STEP suspension signals capture it when scheduled and no-op if + * it moved (see step.ts) — this drops same-boundary sibling signals and + * timers queued at boundary N that would fire after the session resumed + * into boundary N+1. Sleep/hook/attribute signals are intentionally + * unguarded: their presence makes the boundary unretainable, so a late + * signal correctly demotes the session (workflow.ts `onWorkflowError`). + */ + 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..9e8beeef6f --- /dev/null +++ b/packages/core/src/retained-vm-loop.test.ts @@ -0,0 +1,329 @@ +import { PreconditionFailedError } from '@workflow/errors'; +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(); + 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) => { + p.catch(() => {}); + }), +})); + +const createContextSpy = createContext as unknown as ReturnType; + +// 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 all-primitive batch: both parked step consumers schedule their +// own (identical) suspension signal for the same boundary — the first one is +// the suspension, the sibling must be absorbed by the generation guard +// without demoting the session. +const parallelBatchWorkflow = `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(1), s2(2)]); + return a + b; + } + globalThis.__private_workflows = new Map([["workflow", workflow]]);`; + +// A parallel batch where one sibling's input is unsafe must serialize the +// WHOLE batch through the ordinary VM path (all-or-nothing) and demote. +const mixedBatchWorkflow = `const s1 = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_s1"); + 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, + options: { failEventTypeOnce?: string } = {} +) { + let { failEventTypeOnce } = options; + 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) => { + if (data.eventType === failEventTypeOnce) { + failEventTypeOnce = undefined; + throw new PreconditionFailedError('stale snapshot (test-injected)'); + } + 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) => + 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 output = createdEvents.find((e) => e.eventType === 'run_completed') + ?.eventData?.output as Uint8Array | undefined; + return { + vmBuilds: createContextSpy.mock.calls.length, + output, + result: + output === undefined + ? undefined + : await hydrateWorkflowReturnValue(output, runId, 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, result } = await drive('wrun_retained_off'); + expect(result).toBe(30); + // 2 sequential steps → 2 suspensions + completion → 3 replays, each + // building a fresh VM. + expect(vmBuilds).toBeGreaterThan(1); + }); + + it('builds the VM once when retention is ON (the default), with byte-identical output', async () => { + // Baseline via the kill switch, to compare the dehydrated bytes against. + process.env.WORKFLOW_RETAINED_VM = '0'; + const off = await drive('wrun_retained_baseline_off'); + createContextSpy.mockClear(); + delete process.env.WORKFLOW_RETAINED_VM; + + const on = await drive('wrun_retained_on'); + expect(on.result).toBe(30); + // One VM for the whole run: built on the first pass, resumed after. + expect(on.vmBuilds).toBe(1); + expect(on.output).toEqual(off.output); + }); + + it.each([ + ['an argument getter', impureArgsWorkflow, 'impure_args'], + ['a custom serializer', impureSerializerWorkflow, 'impure_serializer'], + ])('matches cold replay when %s mutates workflow state', async (_name, workflowCode, slug) => { + process.env.WORKFLOW_RETAINED_VM = '0'; + const off = await drive(`wrun_${slug}_off`, workflowCode); + createContextSpy.mockClear(); + delete process.env.WORKFLOW_RETAINED_VM; + + const on = await drive(`wrun_${slug}_on`, workflowCode); + // The boundary demoted (multiple VMs), and the result is what a cold + // replay computes: the serialization-time mutation is NOT visible. + expect(on.vmBuilds).toBeGreaterThan(1); + expect(off.result).toBe(0); + expect(on.result).toBe(0); + }); + + it('retains one VM for a parallel batch (sibling suspension signals absorbed)', async () => { + const { vmBuilds, result } = await drive( + 'wrun_retained_parallel_batch', + parallelBatchWorkflow + ); + expect(result).toBe(30); + expect(vmBuilds).toBe(1); + }); + + it('demotes retention when any input in a parallel batch is unsafe', async () => { + const { vmBuilds, result } = await drive( + 'wrun_retained_mixed_batch', + mixedBatchWorkflow + ); + expect(result).toBe(30); + expect(vmBuilds).toBeGreaterThan(1); + }); + + it('discards the retained session when a 412 forces an in-process restart', async () => { + // A stale-snapshot rejection of run_completed restarts the replay in + // process (see restartReplayInProcess). The parked session belongs to the + // discarded log — resuming it would replay a completed session (throw → + // run_failed) or bypass the retention decision entirely. The restart must + // fall back to a fresh replay and still complete the run. + const { vmBuilds, result } = await drive( + 'wrun_retained_412_restart', + twoStepWorkflow, + { failEventTypeOnce: 'run_completed' } + ); + expect(result).toBe(30); + expect(vmBuilds).toBeGreaterThan(1); + }); + + it('retains a VM that used the synchronous crypto.subtle.digest', async () => { + const { vmBuilds, result } = await drive( + 'wrun_retained_digest', + digestWorkflow + ); + expect(result).toBe(30); + expect(vmBuilds).toBe(1); + }); +}); diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 750cef009f..4f6a4e81a7 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'; @@ -765,6 +766,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[] = []; const workflowRun: WorkflowRun = { runId: 'wrun_attribute_step_race', @@ -823,6 +827,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 53f5ea80ab..b34df4e209 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -12,7 +12,7 @@ import { RunExpiredError, WorkflowRuntimeError, } from '@workflow/errors'; -import { setWorkflowBasePath } from '@workflow/utils'; +import { once, setWorkflowBasePath } from '@workflow/utils'; import { parseWorkflowName, workflowDisplayName, @@ -52,6 +52,7 @@ import { getReplayDivergenceMaxRetries, isInlineOwnershipEnabled, isTurboEnabled, + isVmRetentionEnabled, } from './runtime/constants.js'; import { countStepStartedEvents } from './runtime/count-step-started-events.js'; import { @@ -110,7 +111,12 @@ import { } from './telemetry.js'; import { getErrorName, getErrorStack, normalizeUnknownError } from './types.js'; import { buildWorkflowSuspensionMessage } from './util.js'; -import { runWorkflow } from './workflow.js'; +import { + replayWorkflow, + resumeWorkflow, + type WorkflowResumeResult, + type WorkflowSession, +} from './workflow.js'; export type { Event, WorkflowRun }; export { WorkflowSuspension } from './global.js'; @@ -435,10 +441,11 @@ function rootRunIdFrom( * `wait_completed`, which the wait timer can resolve with * `wait_completed`). * - * This gates the inline-delta fast path and turbo's forced optimistic start. - * A terminal-step delta can omit an event appended concurrently after that - * write. With no open hook or wait, only cancellation can do so, and observing - * it one replay late is safe because the next entity write is rejected. + * This gates VM retention, the inline-delta fast path, and turbo's forced + * optimistic start. A terminal-step delta can omit an event appended + * concurrently after that write. With no open hook or wait, only cancellation + * can do so, and observing it one replay late is safe because the next entity + * write is rejected. * * Step-body `attr_set` writes are NOT a concern: they land before the * step's terminal write and are therefore already inside the returned @@ -475,6 +482,47 @@ function openHookAndWaitState(events: Event[]): { return { openHook, openWait }; } +/** + * The whole retention predicate: keep the session only for a pure step + * boundary (every suspension item is a step — any other item type, present + * or future, is unretainable by default) whose new step inputs serialized + * without executing workflow code, with no out-of-band continuation source: + * attributes require replay; hooks and waits can wake another invocation. + * `WORKFLOW_RETAINED_VM=0` disables retention entirely. + * + * The open hook/wait scan is O(events), so it is taken through a lazy getter + * and consulted last, after every cheap check has passed. + * + * INVARIANT this predicate leans on: every suspension signaler that does NOT + * carry the step-consumer generation guard (sleep, hook, attribute — see + * `suspensionGeneration` in private.ts) must be unretainable here, either via + * a non-step queue item or the open hook/wait scan. A new signaler that + * satisfies neither would let a stale signal be accepted as a fresh + * suspension on a resumed session. + * + * Quiescence assumes workflow code stays inside the sandbox's determinism + * contract. Escaping to the host realm (e.g. recovering the host `Function` + * constructor from an exposed host class to schedule real timers) makes a + * workflow nondeterministic under ordinary replay too, and is not defended + * here. + */ +function canRetainWorkflowSession( + suspension: WorkflowSuspension, + stepInputsSafe: boolean, + openHookWait: { value: ReturnType } +): boolean { + if ( + !isVmRetentionEnabled() || + !stepInputsSafe || + suspension.steps.length === 0 || + !suspension.steps.every((item) => item.type === 'step') + ) { + return false; + } + const { openHook, openWait } = openHookWait.value; + return !openHook && !openWait; +} + /** * Creates a single route which handles workflow execution requests, * executing steps inline when possible to reduce function invocations @@ -997,6 +1045,14 @@ export function workflowEntrypoint( return false; } preconditionRestarts++; + // Every stale-snapshot restart invalidates the parked VM: + // the log it consumed was missing events, so only a fresh + // replay over the corrected log is authoritative. This + // also covers the suspension-create 412, which fires + // before the retention decision (kill switch + step-input + // gate) ever ran, and the run_completed 412, where a + // completed session must not be resumed again. + retainedSession = null; // A World MAY return the events we were missing on the 412. // Trust it only on the FIRST restart: its completeness proof // leans on the backend's own bookkeeping, so if that @@ -1606,7 +1662,7 @@ export function workflowEntrypoint( } // end if (!workflowRun) // Resolve the encryption key for this run's deployment. - // Used eagerly here since both runWorkflow (input + // Used eagerly here since both workflow execution (input // hydration / hook payload decryption) and the run_failed // dehydrate path below need it. Memoized accessor: first // call triggers the actual fetch / HKDF derivation, @@ -1625,6 +1681,11 @@ export function workflowEntrypoint( encryptionKey ); + // The live VM parked at the previous boundary, when the + // retention decision kept it. null → this iteration cold- + // replays. Invocation-scoped: dies with this delivery. + let retainedSession: WorkflowSession | null = null; + // Main replay loop // biome-ignore lint/correctness/noConstantCondition: intentional loop while (true) { @@ -2086,38 +2147,57 @@ 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: retainedSession ? 'retained' : 'replay', }); replayStart = Date.now(); - // Start every missing decrypt/decompress operation before - // VM setup. Web Crypto work can overlap bundle evaluation; + // Start every missing decrypt/decompress operation up + // front (already-prepared payloads are skipped). Web + // Crypto work overlaps VM setup on the replay path and + // the appended events' consumption on the resume path; // 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 - ); + let workflowResult: WorkflowResumeResult = retainedSession + ? await resumeWorkflow(retainedSession, events) + : { type: 'replay' }; + + if (workflowResult.type === 'replay') { + retainedSession = null; + workflowResult = await replayWorkflow({ + 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; - runtimeLogger.debug('Workflow replay completed', { + + 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. + retainedSession = workflowResult.session; + throw workflowResult.suspension; + } + + const result = workflowResult.output; + runtimeLogger.debug('Workflow execution completed', { workflowRunId: runId, loopIteration, replayMs: Date.now() - replayStart, + executionMode: retainedSession ? 'retained' : 'replay', }); replayRecoveryReporter.activate(); @@ -2164,7 +2244,7 @@ export function workflowEntrypoint( } catch (err) { if (WorkflowSuspension.is(err)) { replayRecoveryReporter.activate(); - // Synchronous `runWorkflow` duration for THIS + // Synchronous workflow-execution duration for THIS // suspension only — anchors the `finalSchedulingReplay` // telemetry field below (see // StepLatencyTracking.replayMs). This is the FINAL @@ -2188,7 +2268,10 @@ export function workflowEntrypoint( // server spans are heavily sampled in production // (~7%), and client spans can't be filtered by SDK // version, so neither can serve as the dashboard's - // exact TTFS decomposition. + // exact TTFS decomposition. On a retained-VM + // resume this measures the resume (typically ~0ms), + // not a replay, so the field's distribution is + // bimodal once retention is active. const replayDurationMs = Date.now() - replayStart; runtimeLogger.debug('Workflow suspended', { workflowRunId: runId, @@ -2347,6 +2430,41 @@ export function workflowEntrypoint( return; } eventsCursor = suspensionLog.cursor; + + // Open hooks/waits in the log as loaded for this + // replay. This suspension's own hook/wait writes are + // NOT in it — they never reach retention anyway, + // because a suspension containing a non-step item + // fails canRetainWorkflowSession's type check before + // the scan is consulted. Computed + // lazily, at most once, and shared between the + // retention decision here and the delta/turbo gates + // below — the attr-detour and hook-conflict paths + // return/continue before the gates and usually + // short-circuit before ever scanning the log. + // Narrowed alias: the closure below would otherwise + // lose `cachedEvents`'s non-null narrowing. Nothing in + // this catch scope reassigns the array. + const suspensionEvents = cachedEvents; + const openHookWait = once(() => + openHookAndWaitState(suspensionEvents) + ); + + // The single retention decision: keep the parked + // session only across a pure step boundary with no + // out-of-band continuation source and provably + // passive step inputs. + if ( + retainedSession && + !canRetainWorkflowSession( + err, + suspensionResult.retainedStepInputsSafe, + openHookWait + ) + ) { + retainedSession = null; + } + preStepBlockingMs += suspensionResult.hookCreationMs; if ( suspensionResult.hasAttributeEvents && @@ -2664,18 +2782,9 @@ export function workflowEntrypoint( return; } - // Execute inline step. Pause the replay budget - // for the duration of the step body — step - // duration is bounded by the platform's function - // maxDuration, not by the replay timeout. Without - // 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 ?? [] - ); + // Open hooks/waits are consulted by all three gates + // below; resolve the memoized scan once here. + const openHookWaitState = openHookWait.value; // Inline-delta fast path gate. We request the delta — // and on the next iteration consume it in place of the @@ -2842,7 +2951,7 @@ export function workflowEntrypoint( // snapshot has a local-clock createdAt, so under // turbo only the run-id ULID timestamp is trusted. const latencyTracking = computeStepLatencyTracking({ - events: cachedEvents ?? [], + events: cachedEvents, invocationStartedClean: invocationStartedClean === true, runCreatedAtMs: @@ -2874,7 +2983,7 @@ export function workflowEntrypoint( // outside guarded deployments; Worlds that don't // enforce the guard ignore it. const inlineClaimSnapshot = preconditionSnapshotParams( - cachedEvents ?? [], + cachedEvents, preInlineWriteCursor ); diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index d5fea8383b..dadd81ca5d 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 8cc8dbf208..a5fac2353a 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -36,6 +36,21 @@ import { } from './helpers.js'; import { ReplayRecoveryReporter } from './replay-recovery-reporter.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. (Distinct from +// replay-payload-cache's `isMemoizablePrimitive`, a size-gated memoization +// filter — do not merge them.) +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; @@ -129,6 +144,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({ @@ -528,6 +545,23 @@ export async function handleSuspension({ // racing with concurrent handlers on step execution. const createdStepCorrelationIds = new Set(); + // 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 @@ -761,6 +795,7 @@ export async function handleSuspension({ hasAttributeEvents: attributeItems.length > 0, hasHookEvents: hooksNeedingCreation.length > 0, hookCreationMs, + retainedStepInputsSafe, }; } diff --git a/packages/core/src/step-delivery-hop-count.test.ts b/packages/core/src/step-delivery-hop-count.test.ts index f7ea3e09b1..7f9ce88134 100644 --- a/packages/core/src/step-delivery-hop-count.test.ts +++ b/packages/core/src/step-delivery-hop-count.test.ts @@ -55,6 +55,7 @@ function setupWorkflowContext( const promiseQueueHolder = { current: Promise.resolve() }; const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { + suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, replayPayloadCache, diff --git a/packages/core/src/step-delivery-ordering.test.ts b/packages/core/src/step-delivery-ordering.test.ts index 85933b7e7c..5bf7257047 100644 --- a/packages/core/src/step-delivery-ordering.test.ts +++ b/packages/core/src/step-delivery-ordering.test.ts @@ -102,6 +102,7 @@ function setupWorkflowContext( const promiseQueueHolder = { current: Promise.resolve() }; const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { + suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, replayPayloadCache, diff --git a/packages/core/src/step.ts b/packages/core/src/step.ts index c6dbb13494..1fcc4c11df 100644 --- a/packages/core/src/step.ts +++ b/packages/core/src/step.ts @@ -59,7 +59,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..fd126abb3e 100644 --- a/packages/core/src/telemetry.ts +++ b/packages/core/src/telemetry.ts @@ -245,11 +245,15 @@ export async function trace( span.setStatus({ code: otel.SpanStatusCode.OK }); return result; } catch (e) { - span.setStatus({ - 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); + } else { + span.setStatus({ + code: otel.SpanStatusCode.ERROR, + message: (e as Error).message, + }); + } throw e; } finally { span.end(); @@ -275,19 +279,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 f2991767d9..38933257ae 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( '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( 'workflow.arguments.count' diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index 384121c1f0..a089d7a533 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 { replayWorkflow, resumeWorkflow, runWorkflow } from './workflow.js'; // No encryption key = encryption disabled const noEncryptionKey = undefined; @@ -221,6 +222,183 @@ describe('runWorkflow', () => { ).toEqual(3); }); + describe('workflow sessions (retained VM)', () => { + const sessionRun = async (runId: string): Promise => ({ + runId, + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments([], runId, noEncryptionKey, []), + 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 replay = ( + workflowRun: WorkflowRun, + workflowCode: string, + events: Event[] + ) => + replayWorkflow({ + workflowCode, + workflowRun, + events, + encryptionKey: noEncryptionKey, + replayPayloadCache: new ReplayPayloadCache(noEncryptionKey), + }); + + // Append the step_created/started/completed triplet for the suspension's + // single pending step, as the runtime's inline execution would. + let eventCounter = 0; + const completeStep = async ( + run: WorkflowRun, + events: Event[], + suspension: WorkflowSuspension, + result: number + ): Promise => { + const step = suspension.steps[0]; + assert(step?.type === 'step'); + const base = { + runId: run.runId, + correlationId: step.correlationId, + createdAt: new Date(Date.UTC(2024, 0, 1, 0, 0, ++eventCounter)), + }; + events.push( + { + ...base, + eventId: `event-${eventCounter}-created`, + eventType: 'step_created', + eventData: { stepName: 'add' }, + } as Event, + { + ...base, + eventId: `event-${eventCounter}-started`, + eventType: 'step_started', + eventData: { stepName: 'add' }, + } as Event, + { + ...base, + eventId: `event-${eventCounter}-completed`, + eventType: 'step_completed', + eventData: { + stepName: 'add', + result: await dehydrateStepReturnValue( + result, + run.runId, + noEncryptionKey, + [] + ), + }, + } as Event + ); + }; + + it('resumes one VM across sequential step boundaries', async () => { + const run = await sessionRun('wrun_retained'); + 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"); + return await add(first, 3); + } + ${getWorkflowTransformCode('workflow')}`; + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); + const events: Event[] = []; + + const first = await replay(run, workflowCode, events); + assert(first.type === 'suspended'); + await completeStep(run, events, first.suspension, 3); + + const second = await resumeWorkflow(first.session, events); + assert(second.type === 'suspended'); + expect(second.session).toBe(first.session); + await completeStep(run, events, second.suspension, 6); + + const completed = await resumeWorkflow(second.session, events); + assert(completed.type === 'completed'); + expect( + await hydrateWorkflowReturnValue( + completed.output as any, + run.runId, + noEncryptionKey, + [] + ) + ).toBe(6); + // The body ran straight through exactly once — never re-entered. + expect( + log.mock.calls + .map(([message]) => message) + .filter((message) => String(message).startsWith('retained:')) + ).toEqual(['retained:entered', 'retained:continued']); + log.mockRestore(); + }); + + it('declines resume permanently when the event prefix diverges', async () => { + const run = await sessionRun('wrun_retained_divergence'); + 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: run.runId, + eventType: 'run_created', + createdAt: new Date('2024-01-01T00:00:00.000Z'), + } as Event, + ]; + + const suspended = await replay(run, workflowCode, events); + 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 resumeWorkflow(suspended.session, rewritten)).toEqual({ + type: 'replay', + }); + + // The fallback is permanent, even for a well-formed extension. + expect(await resumeWorkflow(suspended.session, events)).toEqual({ + type: 'replay', + }); + }); + + it('demotes to replay (not failure) after mid-execution divergence', async () => { + const run = await sessionRun('wrun_retained_mid_divergence'); + const workflowCode = ` + const step = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step"); + async function workflow() { await step(); } + ${getWorkflowTransformCode('workflow')}`; + + const suspended = await replay(run, workflowCode, []); + assert(suspended.type === 'suspended'); + + // A strict extension whose appended suffix the VM cannot consume: the + // resume starts, then diverges mid-execution. + const alien = { + eventId: 'event-alien', + runId: run.runId, + eventType: 'hook_received', + correlationId: 'hook_unknown', + eventData: {}, + createdAt: new Date('2024-01-01T00:00:01.000Z'), + } as Event; + await expect(resumeWorkflow(suspended.session, [alien])).rejects.toThrow( + /could not consume event/ + ); + + // The session demotes to replay — resume() must not throw. + expect(await resumeWorkflow(suspended.session, [])).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 6c09703327..113836e91d 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -1,10 +1,15 @@ +import type { Span } from '@opentelemetry/api'; import { ERROR_SLUGS, ReplayDivergenceError, 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 +41,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'; @@ -112,6 +117,131 @@ async function drainPendingQueueItems( } } +/** Everything needed to cold-start a workflow VM over an event log. */ +interface WorkflowSessionOptions { + readonly workflowCode: string; + readonly workflowRun: WorkflowRun; + readonly events: Event[]; + readonly encryptionKey: PayloadKey | undefined; + readonly replayPayloadCache: ReplayPayloadCache; + readonly runReadyBarrier?: Promise; + readonly worldCapabilities?: WorldCapabilities; +} + +/** + * A live workflow VM, parked at a suspension boundary. `resume` advances it + * by appending events instead of replaying from scratch. + */ +export interface WorkflowSession { + readonly workflowRun: WorkflowRun; + readonly argumentCount: number; + resume(events: Event[]): Promise; +} + +/** A finished execution attempt: the workflow's output or a live boundary. */ +export type WorkflowResult = + | { + readonly type: 'completed'; + readonly output: unknown; + /** `typeof` the raw return value, for telemetry. */ + readonly resultType: string; + } + | { + readonly type: 'suspended'; + readonly suspension: WorkflowSuspension; + readonly session: WorkflowSession; + }; + +/** + * `resume` can additionally decline — `{ type: 'replay' }` means "this + * session is unusable, cold-replay instead". A fresh replay never declines. + */ +export type WorkflowResumeResult = WorkflowResult | { readonly type: 'replay' }; + +/** The session's private state machine. */ +type WorkflowSessionState = + | { + readonly type: 'running'; + readonly interruption: PromiseWithResolvers; + } + | { readonly type: 'suspended'; readonly suspension: WorkflowSuspension } + | { readonly type: 'replay' } + | { readonly type: 'completed' }; + +/** Cold-start: build a fresh VM session and replay it over `events`. */ +export function replayWorkflow( + options: WorkflowSessionOptions +): Promise { + return traceExecution( + 'replay', + options.workflowRun, + options.events, + async (span) => { + const { session, execution } = await createWorkflowSession(options); + span?.setAttributes({ + ...Attribute.WorkflowArgumentsCount(session.argumentCount), + }); + return recordResult(await execution, span); + } + ); +} + +/** Warm-start: advance a retained session by appending events. */ +export function resumeWorkflow( + session: WorkflowSession, + events: Event[] +): Promise { + return traceExecution( + 'retained', + session.workflowRun, + events, + async (span) => { + span?.setAttributes({ + ...Attribute.WorkflowArgumentsCount(session.argumentCount), + }); + const result = await session.resume(events); + return result.type === 'replay' ? result : recordResult(result, span); + } + ); +} + +function traceExecution( + mode: 'replay' | 'retained', + workflowRun: WorkflowRun, + events: Event[], + fn: (span: Span | undefined) => Promise +): Promise { + return trace(`workflow.run ${workflowRun.workflowName}`, (span) => { + span?.setAttributes({ + ...Attribute.WorkflowName(workflowRun.workflowName), + ...Attribute.WorkflowRunId(workflowRun.runId), + ...Attribute.WorkflowRunStatus(workflowRun.status), + ...Attribute.WorkflowEventsCount(events.length), + ...Attribute.WorkflowExecutionMode(mode), + }); + return fn(span); + }); +} + +function recordResult( + result: WorkflowResult, + span: Span | undefined +): WorkflowResult { + if (result.type === 'completed') { + span?.setAttributes({ + ...Attribute.WorkflowResultType(result.resultType), + }); + } else if (span) { + applyWorkflowSuspensionToSpan(result.suspension, span); + } + return result; +} + +/** + * Single-shot replay: execute the workflow over `events` and either return + * its output or throw its suspension. Kept for the extensive existing test + * suites — production code goes through `replayWorkflow`/`resumeWorkflow`. + */ export async function runWorkflow( workflowCode: string, workflowRun: WorkflowRun, @@ -119,8 +249,7 @@ export async function runWorkflow( encryptionKey: PayloadKey | 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 @@ -138,275 +267,307 @@ export async function runWorkflow( */ worldCapabilities?: WorldCapabilities ): Promise { - 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)` - ); - } - - // Seed and initial clock must be available before I/O and remain stable on - // replay. After the first event, EventsConsumer advances the VM clock from - // each event's `createdAt`. - const fixedTimestamp = - runIdCreatedAt(workflowRun.runId) ?? +workflowRun.createdAt; - - // Truthiness, not presence: `vercel env pull` writes `VERCEL_URL=""` into - // `.env.local`, and a framework that loads that file locally would otherwise - // put us on the Vercel branch with nothing to build a host from, making - // `https://` the base URL of every run. - const isVercel = Boolean(process.env.VERCEL_URL); - // 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(); + const result = await replayWorkflow({ + 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; +}> { + 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 must be replay-stable. `startedAt` differs between a - // turbo delivery and a later server-backed replay, so use fixedTimestamp. - 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) { - 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; - } - - // Consume run_started - every run has exactly one - if (event.eventType === 'run_started') { - return EventConsumerResult.Consumed; - } + // Seed and initial clock must be available before I/O and remain stable on + // replay. After the first event, EventsConsumer advances the VM clock from + // each event's `createdAt`. + const fixedTimestamp = + runIdCreatedAt(workflowRun.runId) ?? +workflowRun.createdAt; + + // Truthiness, not presence: `vercel env pull` writes `VERCEL_URL=""` into + // `.env.local`, and a framework that loads that file locally would otherwise + // put us on the Vercel branch with nothing to build a host from, making + // `https://` the base URL of every run. + const isVercel = Boolean(process.env.VERCEL_URL); + // 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, + }); - // 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; + const initialInterruption = withResolvers(); + 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' }; + // Each parked step consumer schedules its own (identical) suspension + // signal; the first one lands here, and bumping the generation makes + // the step-consumer guard drop the rest at fire time. + workflowContext.suspensionGeneration++; + interruption.reject(error); + return; } + case 'suspended': + // Same-boundary duplicates were staled by the generation bump above, + // so anything landing here is out-of-band — an unguarded sleep/hook/ + // attribute signal or a divergence. Those boundaries are unretainable + // (the runtime demotes them too), so fall back to replay. + 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 must be replay-stable. `startedAt` differs between a + // turbo delivery and a later server-backed replay, so use fixedTimestamp. + 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) { + 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; + } - // Workflow code must import the deterministic `fetch` step 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; + + // Workflow code must import the deterministic `fetch` step 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 | 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 | 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; @@ -421,420 +582,471 @@ 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; - formData!: () => Promise; + get bodyUsed() { + return false; + } - arrayBuffer!: () => Promise; - json!: () => Promise; - text!: () => Promise; + // TODO: implement these + blob!: () => Promise; + formData!: () => Promise; - async bytes() { - return new Uint8Array(await this.arrayBuffer()); - } + arrayBuffer!: () => Promise; + json!: () => Promise; + text!: () => Promise; + + 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 | 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 | 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; - formData!: () => Promise; + // TODO: implement these + clone!: () => Response; + blob!: () => Promise; + formData!: () => Promise; - 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; - json!: () => Promise; - text!: () => Promise; + arrayBuffer!: () => Promise; + json!: () => Promise; + text!: () => Promise; - 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 implements globalThis.ReadableStream { - constructor() { - ENOTSUP(); - } + class ReadableStream implements globalThis.ReadableStream { + 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 implements globalThis.WritableStream { - constructor() { - ENOTSUP(); - } + class WritableStream implements globalThis.WritableStream { + 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 implements globalThis.TransformStream { - readable: globalThis.ReadableStream; - writable: globalThis.WritableStream; + class TransformStream implements globalThis.TransformStream { + readable: globalThis.ReadableStream; + writable: globalThis.WritableStream; - constructor() { - ENOTSUP(); - } + constructor() { + ENOTSUP(); } - vmGlobalThis.TransformStream = TransformStream; - - vmGlobalThis.console = globalThis.console; - - // Expose the request-context symbol required by AI Gateway. - 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.". - const parsedName = parseWorkflowName(workflowRun.workflowName); - const filename = parsedName?.moduleSpecifier || workflowRun.workflowName; - - // Reuse compiled scripts by `(code, filename)`: compilation is deterministic - // and the filename preserves workflow source attribution in stack traces. - // The bundle registers workflows on `globalThis.__private_workflows`. - runCachedWorkflowScript(workflowCode, filename, context); - const workflowFn = runCachedWorkflowScript( - `globalThis.__private_workflows?.get(${JSON.stringify(workflowRun.workflowName)})`, - filename, - context - ); + } + vmGlobalThis.TransformStream = TransformStream; + + vmGlobalThis.console = globalThis.console; + + // Expose the request-context symbol required by AI Gateway. + 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.". + const parsedName = parseWorkflowName(workflowRun.workflowName); + const filename = parsedName?.moduleSpecifier || workflowRun.workflowName; + + // Reuse compiled scripts by `(code, filename)`: compilation is deterministic + // and the filename preserves workflow source attribution in stack traces. + // The bundle registers workflows on `globalThis.__private_workflows`. + 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; + + // The user function's promise. It may stay pending across many resumes + // (each parked step promise holds it up) and is raced against the current + // attempt's interruption in waitForExecution. + const workflowBody = (async (): Promise => { + return await workflowFn(...args); + })(); + + const failWorkflow = async (error: unknown): Promise => { + // 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 + ): Promise => { + let result: unknown; try { - const result = await Promise.race([ - workflowFn(...args), - workflowDiscontinuation.promise, - ]); + result = await Promise.race([workflowBody, interruption.promise]); + } catch (error) { + if (state.type === 'suspended' && error === state.suspension) { + return { type: 'suspended', suspension: state.suspension, session }; + } + return failWorkflow(error); + } - const dehydrated = await dehydrateWorkflowReturnValue( + state = { type: 'completed' }; + try { + const output = await dehydrateWorkflowReturnValue( result, workflowRun.runId, encryptionKey, @@ -845,10 +1057,6 @@ export async function runWorkflow( (workflowRun.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION ); - span?.setAttributes({ - ...Attribute.WorkflowResultType(typeof result), - }); - await drainPendingQueueItems( workflowRun.runId, workflowContext.invocationsQueue, @@ -858,24 +1066,53 @@ 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 { type: 'completed', output, resultType: typeof result }; + } catch (error) { + return failWorkflow(error); } - }); + }; + + const session: WorkflowSession = { + workflowRun, + argumentCount: args.length, + async resume(nextEvents) { + switch (state.type) { + case 'suspended': { + // The full O(known events) prefix compare is required: the runtime + // usually grows one array in place, but its wait-completion and + // stale-reload paths REPLACE the array with a fresh full fetch, so + // a rewritten prefix is a real input, not paranoia. The compares + // are cheap string equality and dwarfed by the replay they avoid. + const knownEvents = eventsConsumer.events; + const isStrictExtension = + nextEvents.length > knownEvents.length && + knownEvents.every( + (event, index) => event.eventId === nextEvents[index].eventId + ); + if (!isStrictExtension) { + state = { type: 'replay' }; + return { type: 'replay' }; + } + const interruption = withResolvers(); + state = { type: 'running', interruption }; + workflowContext.suspensionGeneration++; + eventsConsumer.append(nextEvents.slice(knownEvents.length)); + return 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), + }; } diff --git a/packages/core/src/workflow/sleep.test.ts b/packages/core/src/workflow/sleep.test.ts index a7c4723dd7..79c2202264 100644 --- a/packages/core/src/workflow/sleep.test.ts +++ b/packages/core/src/workflow/sleep.test.ts @@ -20,6 +20,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { const ulid = monotonicFactory(() => context.globalThis.Math.random()); const workflowStartedAt = context.globalThis.Date.now(); const ctx: WorkflowOrchestratorContext = { + suspensionGeneration: 0, runId: 'wrun_test', encryptionKey: undefined, replayPayloadCache: new ReplayPayloadCache(undefined), diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index 2fb89e129d..0ef0bcc661 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -246,6 +246,48 @@ export async function stepWinsRaceWorkflow() { ////////////////////////////////////////////////////////// +// Takes an OBJECT argument: under VM retention (WORKFLOW_RETAINED_VM), a +// boundary whose new step has a non-primitive input falls back to cold +// replay instead of resuming the retained VM. +async function unwrapValue(box: { value: number }) { + 'use step'; + return box.value; +} + +/** + * Interleaves every retention mode the runtime can hit: retained boundaries + * (primitive step args), demoted boundaries (object args), wait boundaries + * (sleep, step-vs-sleep race), and a hook awaited in parallel with a step. + * The chained arithmetic makes any dropped, duplicated, or misordered + * boundary visible in the final output. + */ +export async function retainedInterleavingWorkflow(token: string) { + 'use workflow'; + // Retained: sequential primitive-arg step. + const a = await add(1, 2); // 3 + // Demoted: object argument. + const b = await unwrapValue({ value: a }); // 3 + // Retained: parallel all-primitive batch. + const [c, d] = await Promise.all([add(b, 10), add(b, 20)]); // 13, 23 + // Demoted: mixed parallel batch (one object arg, one primitive). + const [e, f] = await Promise.all([unwrapValue({ value: c }), add(d, 1)]); // 13, 24 + // Wait boundary: step races (and beats) a sleep. + const winner = await Promise.race([ + delayMsStep(100, 'step'), + sleep('30s').then(() => 'sleep'), + ]); // 'step' + // Wait boundary: plain sleep. + await sleep('1s'); + // Hook boundary: hook payload awaited in parallel with a primitive step. + using hook = createHook<{ delta: number }>({ token }); + const [payload, g] = await Promise.all([hook, add(e + f, 100)]); // _, 137 + // Retained again after all the demotions. + const h = await add(g, payload.delta); // 137 + delta + return { a, b, c, d, e, f, winner, g, h }; +} + +////////////////////////////////////////////////////////// + async function nullByteStep() { 'use step'; return 'null byte \0';