From 91d04bbb7335b85144e513f006b42089f70d603a Mon Sep 17 00:00:00 2001 From: Nathan Colosimo <110621881+NathanColosimo@users.noreply.github.com> Date: Mon, 10 Aug 2026 17:20:33 -0700 Subject: [PATCH] feat(core): add atomic start hook admission --- .changeset/atomic-start-hook.md | 11 + packages/cli/src/lib/inspect/hydration.ts | 1 + packages/core/src/runtime.test.ts | 88 ++- packages/core/src/runtime.ts | 49 +- packages/core/src/runtime/quickjs-serde.ts | 40 ++ packages/core/src/runtime/start.test.ts | 247 ++++++++- packages/core/src/runtime/start.ts | 508 +++++++++++------- packages/core/src/serialization.test.ts | 18 +- .../src/serialization/reducers/common-vm.ts | 32 ++ .../core/src/serialization/reducers/common.ts | 22 + packages/core/src/serialization/types.ts | 12 +- packages/errors/src/index.ts | 78 +-- packages/web-shared/src/lib/hydration.ts | 58 +- packages/web-shared/test/hydration.test.ts | 16 +- .../test/serializable-revivers.test.ts | 7 + packages/workflow/src/api.ts | 1 + packages/workflow/src/internal/errors.ts | 1 + packages/world-vercel/src/events-v4.ts | 4 + packages/world-vercel/src/events.test.ts | 21 +- packages/world-vercel/src/events.ts | 7 + packages/world/src/capabilities.ts | 6 + packages/world/src/events.test.ts | 22 + packages/world/src/events.ts | 55 +- packages/world/src/index.ts | 2 + packages/world/src/interfaces.ts | 9 + packages/world/src/queue.ts | 15 +- 26 files changed, 1001 insertions(+), 329 deletions(-) create mode 100644 .changeset/atomic-start-hook.md diff --git a/.changeset/atomic-start-hook.md b/.changeset/atomic-start-hook.md new file mode 100644 index 0000000000..77a1e23543 --- /dev/null +++ b/.changeset/atomic-start-hook.md @@ -0,0 +1,11 @@ +--- +"workflow": minor +"@workflow/core": minor +"@workflow/errors": minor +"@workflow/world": minor +"@workflow/cli": patch +"@workflow/web-shared": patch +"@workflow/world-vercel": patch +--- + +Add atomic workflow admission with `start({ hook })` and a typed error for uncertain start outcomes. diff --git a/packages/cli/src/lib/inspect/hydration.ts b/packages/cli/src/lib/inspect/hydration.ts index eb043aa7d2..9c7bafae93 100644 --- a/packages/cli/src/lib/inspect/hydration.ts +++ b/packages/cli/src/lib/inspect/hydration.ts @@ -153,6 +153,7 @@ const ERROR_REVIVER_KEYS = [ 'EvalError', 'FatalError', 'HookConflictError', + 'WorkflowStartError', 'RangeError', 'ReferenceError', 'RetryableError', diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 8a2e153e95..e53a183699 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -1,7 +1,9 @@ import { EntityConflictError, + HookConflictError, PreconditionFailedError, RUN_ERROR_CODES, + START_HOOK_ADMISSION_REJECTED, ThrottleError, WorkflowWorldError, } from '@workflow/errors'; @@ -2180,6 +2182,7 @@ describe('workflowEntrypoint turbo mode', () => { workflowName: 'workflow', specVersion: SPEC_VERSION_CURRENT, executionContext: {}, + encryptionPublicKey: 'test-public-key', }; } @@ -2194,6 +2197,8 @@ describe('workflowEntrypoint turbo mode', () => { attempt: number; source: string; runStartedGate?: Promise; + startHook?: { token: string }; + runStartedError?: Error; }) { const { runId, attempt, source } = opts; const order = turboOrder; @@ -2223,6 +2228,7 @@ describe('workflowEntrypoint turbo mode', () => { const eventsCreate = vi.fn(async (_runId: string, data: any) => { if (data.eventType === 'run_started') { + if (opts.runStartedError) throw opts.runStartedError; if (opts.runStartedGate) await opts.runStartedGate; order.push('run_started_resolved'); return { run: runEntity, events: [] as Event[] }; @@ -2268,7 +2274,10 @@ describe('workflowEntrypoint turbo mode', () => { { runId, requestedAt: new Date('2024-01-01T00:00:00.000Z'), - runInput: await makeRunInput(runId), + runInput: { + ...(await makeRunInput(runId)), + ...(opts.startHook ? { startHook: opts.startHook } : {}), + }, }, { requestId: 'req_turbo', @@ -2337,6 +2346,9 @@ describe('workflowEntrypoint turbo mode', () => { (c) => (c[1] as any).eventType === 'run_started' ); expect(runStartedCreates).toHaveLength(1); + expect(runStartedCreates[0]?.[1].eventData.encryptionPublicKey).toBe( + 'test-public-key' + ); }); it('does not turbo on a redelivery (attempt > 1): run_started is awaited first', async () => { @@ -2354,6 +2366,80 @@ describe('workflowEntrypoint turbo mode', () => { ); }); + it('awaits atomic start Hook admission before running user code', async () => { + const startHook = { token: 'order:123' }; + const { handlerPromise, order, eventsCreate } = await driveTurbo({ + runId: 'wrun_atomic_start_hook', + attempt: 1, + source: oneStepWorkflow, + startHook, + }); + + expect((await handlerPromise).status).toBe(204); + expect(order.indexOf('run_started_resolved')).toBeLessThan( + order.indexOf('body') + ); + const runStarted = eventsCreate.mock.calls.find( + (call) => call[1].eventType === 'run_started' + ); + expect(runStarted?.[1].eventData.startHook).toEqual(startHook); + }); + + it('acknowledges a queued atomic-start loser without running user code', async () => { + const { handlerPromise, order, eventsCreate } = await driveTurbo({ + runId: 'wrun_atomic_start_loser', + attempt: 1, + source: oneStepWorkflow, + startHook: { token: 'order:123' }, + runStartedError: new HookConflictError('order:123', 'wrun_winner'), + }); + + expect((await handlerPromise).status).toBe(204); + expect(order).not.toContain('body'); + expect( + eventsCreate.mock.calls.some((call) => call[1].eventType === 'run_failed') + ).toBe(false); + }); + + it('acknowledges a permanent atomic-start admission rejection', async () => { + const { handlerPromise, order, eventsCreate } = await driveTurbo({ + runId: 'wrun_atomic_start_rejected', + attempt: 1, + source: oneStepWorkflow, + startHook: { token: 'order:123' }, + runStartedError: new WorkflowWorldError('retention exceeds limit', { + status: 400, + code: START_HOOK_ADMISSION_REJECTED, + }), + }); + + expect((await handlerPromise).status).toBe(204); + expect(order).not.toContain('body'); + expect( + eventsCreate.mock.calls.some((call) => call[1].eventType === 'run_failed') + ).toBe(false); + }); + + it('records unrelated World contract errors on atomic starts', async () => { + const { handlerPromise, order, eventsCreate } = await driveTurbo({ + runId: 'wrun_atomic_start_world_error', + attempt: 1, + source: oneStepWorkflow, + startHook: { token: 'order:123' }, + runStartedError: new WorkflowWorldError('invalid response', { + code: RUN_ERROR_CODES.WORLD_CONTRACT_ERROR, + }), + }); + + expect((await handlerPromise).status).toBe(204); + expect(order).not.toContain('body'); + expect( + eventsCreate.mock.calls.filter( + (call) => call[1].eventType === 'run_failed' + ) + ).toHaveLength(1); + }); + it('does not turbo when WORKFLOW_TURBO=0 (parity with the awaited path)', async () => { process.env.WORKFLOW_TURBO = '0'; const { handlerPromise, order } = await driveTurbo({ diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 4d19bb2a8b..0051fca0db 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -4,6 +4,7 @@ import { CorruptedEventLogError, EntityConflictError, FatalError, + HookConflictError, HookNotFoundError, MaxEventsExceededError, PreconditionFailedError, @@ -11,6 +12,7 @@ import { RUN_ERROR_CODES, type RunErrorCode, RunExpiredError, + START_HOOK_ADMISSION_REJECTED, WorkflowRuntimeError, WorkflowWorldError, } from '@workflow/errors'; @@ -28,6 +30,7 @@ import { isLegacySpecVersion, isTerminalRunEventType, ROOT_RUN_ID_ATTRIBUTE, + type RunCreationData, type RunInput, resolveQueueNamespace, SPEC_VERSION_CURRENT, @@ -170,6 +173,7 @@ export { wakeUpRun, } from './runtime/runs.js'; export { + type StartHookOptions, type StartOptions, type StartOptionsBase, type StartOptionsWithDeploymentId, @@ -1011,6 +1015,7 @@ export function workflowEntrypoint( const turbo = isTurboEnabled() && runInput !== undefined && + runInput.startHook === undefined && metadata.attempt === 1 && incomingStepId === undefined && !replayDivergence; @@ -2049,6 +2054,15 @@ export function workflowEntrypoint( // Contract: events.create('run_started') must be idempotent // for runs already in 'running' status (return the run // without error), not just for pending → running transitions. + let runCreationData: RunCreationData | undefined; + if (runInput) { + const { + environment: _environment, + specVersion: _specVersion, + ...data + } = runInput; + runCreationData = data; + } const runStartedEvent = { eventType: 'run_started' as const, // Use the spec version from the original start() call @@ -2060,18 +2074,8 @@ export function workflowEntrypoint( // create the run if run_created was missed. // Uint8Array values survive the queue natively // (CBOR on world-vercel, JSON reviver on world-local). - ...(runInput - ? { - eventData: { - input: runInput.input, - deploymentId: runInput.deploymentId, - workflowName: runInput.workflowName, - executionContext: runInput.executionContext, - attributes: runInput.attributes, - allowReservedAttributes: - runInput.allowReservedAttributes, - }, - } + ...(runCreationData + ? { eventData: runCreationData } : {}), }; if (turbo && runInput) { @@ -2204,6 +2208,27 @@ export function workflowEntrypoint( return; } } catch (err) { + if (runInput?.startHook !== undefined) { + if (HookConflictError.is(err)) { + return; + } + if ( + WorkflowWorldError.is(err) && + err.code === START_HOOK_ADMISSION_REJECTED + ) { + runtimeLogger.error( + 'Atomic start Hook admission rejected queued candidate', + { + workflowRunId: runId, + error: + err instanceof Error + ? err.message + : String(err), + } + ); + return; + } + } // Run was concurrently completed/failed/cancelled if ( EntityConflictError.is(err) || diff --git a/packages/core/src/runtime/quickjs-serde.ts b/packages/core/src/runtime/quickjs-serde.ts index 159871a4b7..5fc779e7c7 100644 --- a/packages/core/src/runtime/quickjs-serde.ts +++ b/packages/core/src/runtime/quickjs-serde.ts @@ -254,6 +254,7 @@ const SYMBOL_NAMES = [ 'workflow-class-registry', '@workflow/errors//FatalError', '@workflow/errors//HookConflictError', + '@workflow/errors//WorkflowStartError', '@workflow/errors//RetryableError', '@workflow/errors//RuntimeDecryptionError', ] as const; @@ -1116,6 +1117,19 @@ export function createQuickJSSerde( if (Object.hasOwn(shape, 'cause')) reduced.cause = shape.cause; return reduced; }, + WorkflowStartError: (value) => { + if (!isHandle(value) || !value.isError) return false; + if (chainedString(value, 'name') !== 'WorkflowStartError') return false; + const shape = reduceErrorShape(value) as Record; + const reduced: Record = { + message: shape.message, + stack: shape.stack, + runId: own(value, 'runId'), + stage: own(value, 'stage'), + }; + if (Object.hasOwn(shape, 'cause')) reduced.cause = shape.cause; + return reduced; + }, RangeError: namedErrorSubclassReducer('RangeError'), ReferenceError: namedErrorSubclassReducer('ReferenceError'), RetryableError: (value) => { @@ -1777,6 +1791,32 @@ export function createQuickJSSerde( } return error; }, + WorkflowStartError: (value: JSValueHandle) => { + const cls = registeredErrorClass('@workflow/errors//WorkflowStartError'); + const runId = own(value, 'runId') ?? vm.undefined; + const stage = own(value, 'stage') ?? vm.undefined; + const cause = own(value, 'cause') ?? vm.undefined; + const error = cls + ? vm.construct(cls, runId, stage, cause) + : buildError(i.Error, value, { name: 'WorkflowStartError' }); + if (!cls) { + define(error, 'runId', runId); + define(error, 'stage', stage); + } + const message = own(value, 'message'); + if (message) { + define(error, 'message', message); + message.dispose(); + } + const stack = own(value, 'stack'); + if (stack && !stack.isUndefined) define(error, 'stack', stack); + stack?.dispose(); + runId !== vm.undefined && runId.dispose(); + stage !== vm.undefined && stage.dispose(); + cause !== vm.undefined && cause.dispose(); + cls?.dispose(); + return error; + }, RangeError: namedErrorSubclassReviver('RangeError'), ReferenceError: namedErrorSubclassReviver('ReferenceError'), RetryableError: (value: JSValueHandle) => { diff --git a/packages/core/src/runtime/start.test.ts b/packages/core/src/runtime/start.test.ts index 28a62755ba..a0cb126f0f 100644 --- a/packages/core/src/runtime/start.test.ts +++ b/packages/core/src/runtime/start.test.ts @@ -1,4 +1,10 @@ -import { WorkflowRuntimeError, WorkflowWorldError } from '@workflow/errors'; +import { + HookConflictError, + START_HOOK_ADMISSION_REJECTED, + WorkflowRuntimeError, + WorkflowStartError, + WorkflowWorldError, +} from '@workflow/errors'; import { SPEC_VERSION_CURRENT, SPEC_VERSION_LEGACY, @@ -890,6 +896,245 @@ describe('start', () => { }); }); + describe('atomic start Hook', () => { + const now = new Date('2026-08-10T12:00:00.000Z'); + const validWorkflow = Object.assign(() => Promise.resolve('result'), { + workflowId: 'test-workflow', + }); + let eventsCreate: ReturnType; + let queue: ReturnType; + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(now); + eventsCreate = vi.fn(async (runId) => ({ + run: { runId, status: 'pending' }, + })); + queue = vi.fn().mockResolvedValue(undefined); + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: { + atomicStartHook: { active: true }, + hookRetention: { active: true }, + }, + getDeploymentId: vi.fn().mockResolvedValue('deploy_123'), + events: { create: eventsCreate }, + queue, + }); + }); + + afterEach(() => { + vi.useRealTimers(); + setWorld(undefined); + vi.clearAllMocks(); + }); + + it.each([ + ['duration string', '1m', 60_000], + ['milliseconds', 5_000, 5_000], + ['absolute Date', new Date(+now + 10_000), 10_000], + ] as const)('normalizes a %s once and queues it before admission', async (_label, retention, offset) => { + await start(validWorkflow, [], { + hook: { + token: 'order:123', + experimental_minRetention: retention, + }, + }); + + const queuedHook = queue.mock.calls[0][1].runInput.startHook; + const admittedHook = eventsCreate.mock.calls[0][1].eventData.startHook; + expect(queuedHook).toEqual({ + token: 'order:123', + tokenRetentionUntil: new Date(+now + offset), + }); + expect(admittedHook).toEqual(queuedHook); + expect(queue.mock.invocationCallOrder[0]).toBeLessThan( + eventsCreate.mock.invocationCallOrder[0] + ); + }); + + it('fails before side effects when the World lacks support', async () => { + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + getDeploymentId: vi.fn(), + events: { create: eventsCreate }, + queue, + } as any); + + await expect( + start(validWorkflow, [], { hook: { token: 'order:123' } }) + ).rejects.toThrow('does not support atomic start Hooks'); + expect(queue).not.toHaveBeenCalled(); + expect(eventsCreate).not.toHaveBeenCalled(); + }); + + it('requires Hook retention support only when retention is requested', async () => { + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: { atomicStartHook: { active: true } }, + getDeploymentId: vi.fn().mockResolvedValue('deploy_123'), + events: { create: eventsCreate }, + queue, + } as any); + + await start(validWorkflow, [], { hook: { token: 'order:123' } }); + await expect( + start(validWorkflow, [], { + hook: { + token: 'order:456', + experimental_minRetention: '1m', + }, + }) + ).rejects.toThrow('does not support Hook retention'); + expect(queue).toHaveBeenCalledOnce(); + expect(eventsCreate).toHaveBeenCalledOnce(); + }); + + it.each([ + 0, + new Date(+now - 1), + ])('rejects a retention time that is not in the future', async (retention) => { + await expect( + start(validWorkflow, [], { + hook: { + token: 'order:123', + experimental_minRetention: retention, + }, + }) + ).rejects.toThrow('must resolve to a future time'); + expect(queue).not.toHaveBeenCalled(); + }); + + it('surfaces the winning run when admission rejects a duplicate', async () => { + eventsCreate.mockRejectedValue( + new HookConflictError('order:123', 'wrun_winner') + ); + + await expect( + start(validWorkflow, [], { hook: { token: 'order:123' } }) + ).rejects.toMatchObject({ + name: 'HookConflictError', + conflictingRunId: 'wrun_winner', + }); + expect(queue).toHaveBeenCalledOnce(); + }); + + it.each([ + ['queue', () => queue.mockRejectedValue(new Error('offline'))], + [ + 'admission', + () => + eventsCreate.mockRejectedValue( + new WorkflowWorldError('unavailable', { status: 503 }) + ), + ], + ] as const)('reports uncertain %s outcomes', async (stage, fail) => { + fail(); + + const error = await start(validWorkflow, [], { + hook: { token: 'order:123' }, + }).catch((cause) => cause); + + expect(error).toBeInstanceOf(WorkflowStartError); + expect(error).toMatchObject({ stage }); + expect(error.runId).toMatch(/^wrun_/); + }); + + it.each([ + 'queue', + 'admission', + ] as const)('surfaces deterministic %s rejections without wrapping them', async (stage) => { + const rejection = new WorkflowWorldError('invalid token', { + status: 400, + ...(stage === 'admission' + ? { code: START_HOOK_ADMISSION_REJECTED } + : {}), + }); + (stage === 'queue' ? queue : eventsCreate).mockRejectedValue(rejection); + + await expect( + start(validWorkflow, [], { hook: { token: 'order:123' } }) + ).rejects.toBe(rejection); + }); + + it('requires the target deployment to report support', async () => { + const healthResponse = JSON.stringify({ + healthy: true, + specVersion: SPEC_VERSION_CURRENT, + }); + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: { atomicStartHook: { active: true } }, + getDeploymentId: vi.fn().mockResolvedValue('deploy_123'), + events: { create: eventsCreate }, + queue, + streams: { + get: vi.fn( + async () => + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(healthResponse)); + controller.close(); + }, + }) + ), + }, + } as any); + + await expect( + start(validWorkflow, [], { + deploymentId: 'deploy_other', + hook: { token: 'order:123' }, + }) + ).rejects.toThrow('target deployment does not support'); + expect(eventsCreate).not.toHaveBeenCalled(); + expect(queue).toHaveBeenCalledOnce(); + expect(queue.mock.calls[0][0]).toBe('__wkf_workflow_health_check'); + }); + + it('requires the target deployment to support requested retention', async () => { + const healthResponse = JSON.stringify({ + healthy: true, + specVersion: SPEC_VERSION_CURRENT, + capabilities: { atomicStartHook: { active: true } }, + }); + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: { + atomicStartHook: { active: true }, + hookRetention: { active: true }, + }, + getDeploymentId: vi.fn().mockResolvedValue('deploy_123'), + events: { create: eventsCreate }, + queue, + streams: { + get: vi.fn( + async () => + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(healthResponse)); + controller.close(); + }, + }) + ), + }, + } as any); + + await expect( + start(validWorkflow, [], { + deploymentId: 'deploy_other', + hook: { + token: 'order:123', + experimental_minRetention: '1m', + }, + }) + ).rejects.toThrow('target deployment does not support Hook retention'); + expect(eventsCreate).not.toHaveBeenCalled(); + expect(queue).toHaveBeenCalledOnce(); + expect(queue.mock.calls[0][0]).toBe('__wkf_workflow_health_check'); + }); + }); + describe('resilient start (run_created failure)', () => { const validWorkflow = Object.assign(() => Promise.resolve('result'), { workflowId: 'test-workflow', diff --git a/packages/core/src/runtime/start.ts b/packages/core/src/runtime/start.ts index 9ee1dac6e6..d2041b4474 100644 --- a/packages/core/src/runtime/start.ts +++ b/packages/core/src/runtime/start.ts @@ -1,6 +1,20 @@ -import { EntityConflictError, WorkflowRuntimeError } from '@workflow/errors'; +import { + EntityConflictError, + HookConflictError, + RUN_ERROR_CODES, + WorkflowRuntimeError, + WorkflowStartError, + WorkflowWorldError, +} from '@workflow/errors'; +import { parseDurationToDate } from '@workflow/utils'; import { workflowDisplayName } from '@workflow/utils/parse-name'; -import type { WorkflowInvokePayload, World } from '@workflow/world'; +import type { + RunCreationData, + StartHook, + WorkflowInvokePayload, + WorkflowRunStatus, + World, +} from '@workflow/world'; import { HOOK_RESUME_INPUT_VERSION, isLegacySpecVersion, @@ -11,6 +25,7 @@ import { SPEC_VERSION_SUPPORTS_COMPRESSION, workflowRunIdSchema, } from '@workflow/world'; +import type { StringValue } from 'ms'; import { monotonicFactory } from 'ulid'; import { normalizeAttributeChanges } from '../attribute-changes.js'; import { getRunCapabilities } from '../capabilities.js'; @@ -92,7 +107,57 @@ export function _resetLatestNoOpWarnForTests(): void { hasWarnedLatestNoOp = false; } +export interface StartHookOptions { + /** Non-empty token reserved atomically while the workflow is admitted. */ + token: string; + + /** + * **Experimental.** Keeps the token unavailable for at least this long. + * Accepts the same duration string, millisecond number, or absolute `Date` + * as `sleep()` and `createHook({ experimental_minRetention })`. + * + * The workflow remains the owner until it ends even if this time passes + * first. A matching `createHook()` can extend, but cannot shorten, it. + */ + experimental_minRetention?: StringValue | Date | number; +} + +function normalizeStartHook(options: StartHookOptions): StartHook { + if (options.token.length === 0) { + throw new WorkflowRuntimeError('hook.token must be a non-empty string.'); + } + if (options.experimental_minRetention === undefined) { + return { token: options.token }; + } + + const tokenRetentionUntil = parseDurationToDate( + options.experimental_minRetention + ); + if ( + !Number.isFinite(tokenRetentionUntil.getTime()) || + tokenRetentionUntil.getTime() <= Date.now() + ) { + throw new WorkflowRuntimeError( + 'hook.experimental_minRetention must resolve to a future time.' + ); + } + return { token: options.token, tokenRetentionUntil }; +} + +function atomicStartContractError(message: string): WorkflowWorldError { + return new WorkflowWorldError(message, { + code: RUN_ERROR_CODES.WORLD_CONTRACT_ERROR, + }); +} + export interface StartOptionsBase { + /** + * Atomically reserves a Hook token while admitting the workflow run. If + * another run owns the token, `start()` throws `HookConflictError` and the + * duplicate candidate never becomes a run. + */ + hook?: StartHookOptions; + /** * The world to use for the workflow run creation, * by default the world is inferred from the environment variables. @@ -251,38 +316,119 @@ export async function start( options?: StartOptions ) { 'use step'; - return await waitedUntil(() => { - // @ts-expect-error this field is added by our client transform - const workflowName = workflow?.workflowId; - - if (!workflowName) { - throw new WorkflowRuntimeError( - `'start' received an invalid workflow function. Ensure the Workflow SDK is configured correctly and the function includes a 'use workflow' directive.`, - { slug: 'start-invalid-workflow-function' } - ); - } + // @ts-expect-error this field is added by our client transform + const workflowName = workflow?.workflowId; + if (!workflowName) { + throw new WorkflowRuntimeError( + `'start' received an invalid workflow function. Ensure the Workflow SDK is configured correctly and the function includes a 'use workflow' directive.`, + { slug: 'start-invalid-workflow-function' } + ); + } + + let args: Serializable[] = []; + let opts: StartOptions = options ?? {}; + if (Array.isArray(argsOrOptions)) { + args = argsOrOptions as Serializable[]; + } else if (typeof argsOrOptions === 'object') { + opts = argsOrOptions; + } + const startHook = opts.hook && normalizeStartHook(opts.hook); + const spanName = `workflow.start ${workflowDisplayName(workflowName)}`; - const spanName = `workflow.start ${workflowDisplayName(workflowName)}`; + return await waitedUntil(() => { return trace(spanName, async (span) => { span?.setAttributes({ ...Attribute.WorkflowName(workflowName), ...Attribute.WorkflowOperation('start'), - }); - - let args: Serializable[] = []; - let opts: StartOptions = options ?? {}; - if (Array.isArray(argsOrOptions)) { - args = argsOrOptions as Serializable[]; - } else if (typeof argsOrOptions === 'object') { - opts = argsOrOptions; - } - - span?.setAttributes({ ...Attribute.WorkflowArgumentsCount(args.length), }); const world = opts.world ?? (await getWorldLazy()); assertWorldSupportsRuntimeProtocol(world); + if (startHook !== undefined) { + if (world.capabilities?.atomicStartHook?.active !== true) { + throw atomicStartContractError( + 'The configured World does not support atomic start Hooks.' + ); + } + if ( + startHook.tokenRetentionUntil !== undefined && + world.capabilities?.hookRetention?.active !== true + ) { + throw atomicStartContractError( + 'The configured World does not support Hook retention.' + ); + } + } + + // Reject invalid input before deployment lookup or health-check I/O. + const specVersion = opts.specVersion ?? world.specVersion; + const v1Compat = isLegacySpecVersion(specVersion); + if ( + startHook !== undefined && + specVersion < SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT + ) { + throw atomicStartContractError( + 'Atomic start Hooks require a spec version with resilient run input.' + ); + } + + const allowReservedAttributes = opts.allowReservedAttributes === true; + let attributes: Record | undefined; + if (opts.attributes && Object.keys(opts.attributes).length > 0) { + if (specVersion < SPEC_VERSION_SUPPORTS_ATTRIBUTES) { + throw new WorkflowRuntimeError( + 'Initial workflow attributes require a World that supports spec version 4 or later.' + ); + } + // Creation cannot remove attributes, so reject non-string values. + for (const [key, value] of Object.entries(opts.attributes)) { + if (typeof value !== 'string') { + throw new WorkflowRuntimeError( + `Initial workflow attribute ${JSON.stringify(key)} must be a string value.` + ); + } + } + const changes = normalizeAttributeChanges(opts.attributes, { + allowReservedAttributes, + }); + attributes = Object.fromEntries( + changes.map(({ key, value }) => [key, value as string]) + ); + } + + // Child runs inherit the root and record their direct parent. + const lineage = + specVersion >= SPEC_VERSION_SUPPORTS_ATTRIBUTES + ? resolveLineageAttributes() + : undefined; + const runAttributes = lineage + ? { ...lineage, ...attributes } + : attributes; + const attributeSeed = runAttributes + ? { + attributes: runAttributes, + ...(allowReservedAttributes || lineage + ? { allowReservedAttributes: true as const } + : {}), + } + : {}; + const startHookSeed = startHook ? { startHook } : {}; + + // This is persisted as a foreign key to the source run. + if ( + opts.replayedFromRunId !== undefined && + !workflowRunIdSchema.safeParse(opts.replayedFromRunId).success + ) { + throw new WorkflowRuntimeError( + `replayedFromRunId must be a run ID (wrun_); received ${JSON.stringify( + String(opts.replayedFromRunId).slice(0, 64) + )}.` + ); + } + // Pin the run to the VM engine selected when it starts. + const workflowVm = getWorkflowVmFromEnv(); + const currentDeploymentId = await world.getDeploymentId(); let deploymentId = opts.deploymentId ?? currentDeploymentId; @@ -323,8 +469,6 @@ export async function start( // probe has a tight timeout — on miss/failure we fall back to the // legacy raw byte format, which is universally readable. // - // Worlds that don't expose the `streams` API (e.g. minimal test - // mocks) can't service health checks, so we skip the probe for them. // Generate runId client-side so we have it before serialization // (required for future E2E encryption where runId is part of the // encryption context). When the World provides a `createRunId()` @@ -354,12 +498,6 @@ export async function start( // Same deployment: this process is the consumer, so its own constant // is authoritative. targetHookResumeInputVersion = HOOK_RESUME_INPUT_VERSION; - } else if (typeof world.streams?.get !== 'function') { - framedByteStreams = false; - targetSupportsCompression = false; - // No probe channel to the target — cannot attest the consumer honors - // `hookInput`, so leave the marker off (fail closed to sequential). - targetHookResumeInputVersion = undefined; } else { // Ask for this run's public key while we're here. The probe already // blocks `start()` on every cross-deployment call, and the responder @@ -374,6 +512,22 @@ export async function start( timeout: CROSS_DEPLOYMENT_CAPABILITY_PROBE_TIMEOUT_MS, namespace: opts.namespace, }).catch(() => undefined); + if ( + startHook && + probe?.capabilities?.atomicStartHook?.active !== true + ) { + throw atomicStartContractError( + 'The target deployment does not support atomic start Hooks.' + ); + } + if ( + startHook?.tokenRetentionUntil !== undefined && + probe?.capabilities?.hookRetention?.active !== true + ) { + throw atomicStartContractError( + 'The target deployment does not support Hook retention.' + ); + } probedRunPublicKey = probe?.encryptionPublicKey; const capabilities = getRunCapabilities(probe?.workflowCoreVersion); framedByteStreams = capabilities.framedByteStreams; @@ -391,72 +545,6 @@ export async function start( // Serialize current trace context to propagate across queue boundary const traceCarrier = await serializeTraceCarrier(); - // Default new runs to the configured world's spec version. The world - // itself has already been checked against this runtime's spec version. - const specVersion = opts.specVersion ?? world.specVersion; - const v1Compat = isLegacySpecVersion(specVersion); - const allowReservedAttributes = opts.allowReservedAttributes === true; - let attributes: Record | undefined; - if (opts.attributes && Object.keys(opts.attributes).length > 0) { - if (specVersion < SPEC_VERSION_SUPPORTS_ATTRIBUTES) { - throw new WorkflowRuntimeError( - 'Initial workflow attributes require a World that supports spec version 4 or later.' - ); - } - // `normalizeAttributeChanges` treats `undefined` as "remove this - // key", which is meaningless at creation time — reject it up front - // so JS callers get a clear error instead of a downstream schema - // failure (the types already forbid non-string values). - for (const [key, value] of Object.entries(opts.attributes)) { - if (typeof value !== 'string') { - throw new WorkflowRuntimeError( - `Initial workflow attribute ${JSON.stringify(key)} must be a string value.` - ); - } - } - const changes = normalizeAttributeChanges(opts.attributes, { - allowReservedAttributes, - }); - attributes = Object.fromEntries( - changes.map(({ key, value }) => [key, value as string]) - ); - } - - // Cross-run lineage: the reserved keys ride on the run's existing - // attributes, so they add no extra write. Caller attributes are spread - // last, so a caller with allowReservedAttributes can deliberately - // re-parent. - const lineage = - specVersion >= SPEC_VERSION_SUPPORTS_ATTRIBUTES - ? resolveLineageAttributes() - : undefined; - const runAttributes = lineage - ? { ...lineage, ...attributes } - : attributes; - - // Shared by the run_created event and the resilient-start queue input. - const attributeSeed = runAttributes - ? { - attributes: runAttributes, - ...(allowReservedAttributes || lineage != null - ? { allowReservedAttributes: true as const } - : {}), - } - : {}; - - // `replayedFromRunId` is a foreign key to the source run; reject anything - // that isn't a real run ID so the lineage link can't point at garbage. - if ( - opts.replayedFromRunId !== undefined && - !workflowRunIdSchema.safeParse(opts.replayedFromRunId).success - ) { - throw new WorkflowRuntimeError( - `replayedFromRunId must be a run ID (wrun_); received ${JSON.stringify( - String(opts.replayedFromRunId).slice(0, 64) - )}.` - ); - } - // Resolve encryption key for the new run. The runId has already been // generated above (client-generated ULID) and will be used for both // key derivation and the run_created event. The World implementation @@ -523,6 +611,16 @@ export async function start( framedByteStreams, compression ); + // Admission can deliver the queue message before start() returns. + safeWaitUntil(Promise.all(ops), (err) => { + runtimeLogger.warn( + 'Background flush of workflow argument streams failed', + { + workflowRunId: runId, + error: err instanceof Error ? err.message : String(err), + } + ); + }); // The environment this caller's own `run_created` write is attributed // to. Stamped into the queue message's `runInput` (NOT into @@ -542,13 +640,6 @@ export async function start( // is simply absent. const creatorEnvironment = world.getEnvironment?.(); - // If WORKFLOW_VM is set on the client starting the run, stamp the - // engine choice into the run's executionContext so the run keeps - // executing on the engine it started on (the same deployment can - // serve both VM engines). Unknown values throw — see - // getWorkflowVmFromEnv(). - const workflowVm = getWorkflowVmFromEnv(); - const executionContext = { traceCarrier, workflowCoreVersion, @@ -571,117 +662,136 @@ export async function start( : {}), }; - // Call events.create (run_created) and queue in parallel. - // If events.create fails with 429/5xx, the run was still accepted - // via the queue and creation will be re-tried async by the runtime. - const [runCreatedResult, queueResult] = await Promise.allSettled([ - world.events.create( - runId, - { - eventType: 'run_created', - specVersion, - eventData: { - deploymentId: deploymentId, - workflowName: workflowName, - input: workflowArguments, - executionContext, - ...(encryptionPublicKey ? { encryptionPublicKey } : {}), - ...attributeSeed, - }, - }, - { v1Compat } - ), + const runCreationData = { + deploymentId, + workflowName, + input: workflowArguments, + executionContext, + ...(encryptionPublicKey ? { encryptionPublicKey } : {}), + ...attributeSeed, + ...startHookSeed, + } satisfies RunCreationData; + + const runCreatedEvent = { + eventType: 'run_created' as const, + specVersion, + eventData: runCreationData, + }; + const queuePayload = { + runId, + traceCarrier, + ...(specVersion >= SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT + ? { + runInput: { + ...runCreationData, + specVersion, + ...(creatorEnvironment !== undefined + ? { environment: creatorEnvironment } + : {}), + }, + } + : {}), + } satisfies WorkflowInvokePayload; + const createRun = () => + world.events.create(runId, runCreatedEvent, { v1Compat }); + const enqueueRun = () => world.queue( getWorkflowQueueName(workflowName, opts.namespace), - { - runId, - traceCarrier, - ...(specVersion >= SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT - ? { - runInput: { - input: workflowArguments, - deploymentId, - workflowName, - specVersion, - executionContext, - ...(encryptionPublicKey ? { encryptionPublicKey } : {}), - ...(creatorEnvironment !== undefined - ? { environment: creatorEnvironment } - : {}), - ...attributeSeed, - }, - } - : {}), - } satisfies WorkflowInvokePayload, + queuePayload, { deploymentId, specVersion, - // Forward any caller-supplied region hint so worlds with - // per-region queue routing (e.g. world-vercel) can target the - // matching queue. Worlds without a regional dimension ignore - // this field. ...(opts.region !== undefined ? { region: opts.region } : {}), } - ), - ]); - - // Queue failure is always fatal — the run was not enqueued - if (queueResult.status === 'rejected') { - throw queueResult.reason; - } + ); - // Handle events.create result let resilientStart = false; - if (runCreatedResult.status === 'rejected') { - const err = runCreatedResult.reason; - if (EntityConflictError.is(err)) { - // 409: The run already exists. This can happen in extreme cases where - // the run creation call gets a cold start or other slowdown, and the queue - // + run_started call completes faster. We expect this to be <=1% of cases. - // In this case, we can safely return. - } else if (isRetryableWorldError(err)) { - // 429 (ThrottleError), 5xx, and transient transport failures - // (TRANSPORT/TIMEOUT) are retryable — the run was accepted via the - // queue and creation will be re-tried by the runtime when it calls - // run_started. - resilientStart = true; - runtimeLogger.warn( - 'Run creation event failed, but the run was accepted via the queue. ' + - 'The run_created event will be re-tried async by the runtime.', - { workflowRunId: runId, error: err.message } - ); - } else { - throw err; + let createdRunStatus: WorkflowRunStatus | undefined; + if (startHook) { + try { + await enqueueRun(); + } catch (error) { + if ( + (error instanceof WorkflowWorldError || + WorkflowWorldError.is(error)) && + !isRetryableWorldError(error) + ) { + throw error; + } + throw new WorkflowStartError(runId, 'queue', error); + } + + try { + const result = await createRun(); + if (result.run.runId !== runId) { + throw atomicStartContractError( + `World admitted run "${result.run.runId}" for candidate "${runId}".` + ); + } + createdRunStatus = result.run.status; + } catch (error) { + if (HookConflictError.is(error)) { + if (error.conflictingRunId === undefined) { + throw atomicStartContractError( + 'Atomic start Hook conflicts must include conflictingRunId.' + ); + } + throw error; + } + if (EntityConflictError.is(error)) { + throw atomicStartContractError( + 'Atomic start admission must replay the candidate decision instead of returning EntityConflictError.' + ); + } + if ( + (error instanceof WorkflowWorldError || + WorkflowWorldError.is(error)) && + !isRetryableWorldError(error) + ) { + throw error; + } + throw new WorkflowStartError(runId, 'admission', error); } } else { - const result = runCreatedResult.value; - // Verify server accepted our runId - if (!v1Compat && result.run.runId !== runId) { - throw new WorkflowRuntimeError( - `Server returned different runId than requested: expected ${runId}, got ${result.run.runId}` - ); + // Ordinary starts keep the existing parallel, resilient admission + // behavior. Only atomic Hook starts require queue-first ordering. + const [runCreatedResult, queueResult] = await Promise.allSettled([ + createRun(), + enqueueRun(), + ]); + if (queueResult.status === 'rejected') { + throw queueResult.reason; } - } - - // These argument-stream ops are flushed in the background; the promise - // handed to waitUntil must never reject (an unconsumed waitUntil - // rejection crashes the process as unhandledRejection), so unexpected - // failures are logged instead. - safeWaitUntil(Promise.all(ops), (err) => { - runtimeLogger.warn( - 'Background flush of workflow argument streams failed', - { - workflowRunId: runId, - error: err instanceof Error ? err.message : String(err), + if (runCreatedResult.status === 'rejected') { + const err = runCreatedResult.reason; + if (EntityConflictError.is(err)) { + // The queued resilient start created this candidate first. + } else if (isRetryableWorldError(err)) { + resilientStart = true; + runtimeLogger.warn( + 'Run creation event failed, but the run was accepted via the queue. ' + + 'The run_created event will be re-tried async by the runtime.', + { workflowRunId: runId, error: err.message } + ); + } else { + throw err; } - ); - }); + } else { + const result = runCreatedResult.value; + if (!v1Compat && result.run.runId !== runId) { + throw new WorkflowRuntimeError( + `Server returned different runId than requested: expected ${runId}, got ${result.run.runId}` + ); + } + createdRunStatus = result.run.status; + } + } span?.setAttributes({ ...Attribute.WorkflowRunId(runId), ...Attribute.DeploymentId(deploymentId), - ...(runCreatedResult.status === 'fulfilled' - ? Attribute.WorkflowRunStatus(runCreatedResult.value.run.status) + ...(createdRunStatus + ? Attribute.WorkflowRunStatus(createdRunStatus) : {}), }); diff --git a/packages/core/src/serialization.test.ts b/packages/core/src/serialization.test.ts index 6a9cbdfc23..cf6b8aaf2e 100644 --- a/packages/core/src/serialization.test.ts +++ b/packages/core/src/serialization.test.ts @@ -5,6 +5,7 @@ import { HookConflictError, RetryableError, RuntimeDecryptionError, + WorkflowStartError, } from '@workflow/errors'; import { WORKFLOW_DESERIALIZE, WORKFLOW_SERIALIZE } from '@workflow/serde'; import { beforeAll, describe, expect, it, vi } from 'vitest'; @@ -4206,7 +4207,7 @@ describe('DOMException serialization', () => { }); describe('Workflow error serialization', () => { - // FatalError, RetryableError, and HookConflictError are first-class serialization targets + // Workflow-specific errors are first-class serialization targets // (handled by dedicated reducers/revivers in the common reducers module), // so unlike user-defined classes they round-trip without any // `registerSerializationClass` setup. This is what makes them usable @@ -4227,6 +4228,21 @@ describe('Workflow error serialization', () => { ); } + it('should round-trip WorkflowStartError recovery details', async () => { + const hydrated = (await roundTrip( + new WorkflowStartError( + 'wrun_candidate', + 'admission', + new Error('gateway timed out') + ) + )) as WorkflowStartError; + + expect(hydrated).toBeInstanceOf(WorkflowStartError); + expect(hydrated.runId).toBe('wrun_candidate'); + expect(hydrated.stage).toBe('admission'); + expect((hydrated.cause as Error).message).toBe('gateway timed out'); + }); + it('should round-trip FatalError preserving type and message', async () => { const error = new FatalError('step failed permanently'); const hydrated = (await roundTrip(error)) as FatalError; diff --git a/packages/core/src/serialization/reducers/common-vm.ts b/packages/core/src/serialization/reducers/common-vm.ts index 66cb458e56..78489590e2 100644 --- a/packages/core/src/serialization/reducers/common-vm.ts +++ b/packages/core/src/serialization/reducers/common-vm.ts @@ -149,6 +149,20 @@ export function getCommonReducers(): Partial { if ('cause' in value) reduced.cause = (value as any).cause; return reduced; }, + WorkflowStartError: (value) => { + if (!(value instanceof Error) || value.name !== 'WorkflowStartError') { + return false; + } + const error = value as any; + const reduced: SerializableSpecial['WorkflowStartError'] = { + message: value.message, + stack: value.stack, + runId: error.runId, + stage: error.stage, + }; + if ('cause' in value) reduced.cause = error.cause; + return reduced; + }, RangeError: makeNamedErrorSubclassReducer('RangeError'), ReferenceError: makeNamedErrorSubclassReducer('ReferenceError'), // RetryableError carries an extra retryAfter; serialize as numeric @@ -417,6 +431,24 @@ export function getCommonRevivers(): Partial { if ('cause' in value) (error as any).cause = (value as any).cause; return error; }, + WorkflowStartError: (value) => { + const Cls = (globalThis as any)[ + Symbol.for('@workflow/errors//WorkflowStartError') + ]; + let error: Error; + if (typeof Cls === 'function') { + error = new Cls(value.runId, value.stage, value.cause); + } else { + error = new Error(value.message); + error.name = 'WorkflowStartError'; + (error as any).runId = value.runId; + (error as any).stage = value.stage; + } + error.message = value.message; + if (value.stack !== undefined) error.stack = value.stack; + if ('cause' in value) (error as any).cause = value.cause; + return error; + }, RangeError: makeNamedErrorSubclassReviver('RangeError'), ReferenceError: makeNamedErrorSubclassReviver('ReferenceError'), RetryableError: (value) => { diff --git a/packages/core/src/serialization/reducers/common.ts b/packages/core/src/serialization/reducers/common.ts index 81276d587d..0d45d28add 100644 --- a/packages/core/src/serialization/reducers/common.ts +++ b/packages/core/src/serialization/reducers/common.ts @@ -15,6 +15,7 @@ import { HookConflictError, RetryableError, RuntimeDecryptionError, + WorkflowStartError, } from '@workflow/errors'; import { arrayBufferByteLength, @@ -288,6 +289,17 @@ export function getCommonReducers( } return reduced; }, + WorkflowStartError: (value) => { + const base = reduceNamedErrorSubclassBase('WorkflowStartError', value); + if (!base) return false; + const error = value as WorkflowStartError; + const reduced: SerializableSpecial['WorkflowStartError'] = { + ...base, + runId: error.runId, + stage: error.stage, + }; + return reduced; + }, RangeError: makeErrorSubclassReducer('RangeError'), ReferenceError: makeErrorSubclassReducer('ReferenceError'), // RetryableError carries an extra `retryAfter` Date that we serialize as @@ -483,6 +495,16 @@ export function getCommonRevivers( } return error; }, + WorkflowStartError: (value) => { + const Ctor = + ((global as Record)[ + Symbol.for('@workflow/errors//WorkflowStartError') + ] as typeof WorkflowStartError | undefined) ?? WorkflowStartError; + const error = new Ctor(value.runId, value.stage, value.cause); + error.message = value.message; + if (value.stack !== undefined) error.stack = value.stack; + return error; + }, RangeError: makeErrorSubclassReviver(global, 'RangeError'), ReferenceError: makeErrorSubclassReviver(global, 'ReferenceError'), RetryableError: (value) => { diff --git a/packages/core/src/serialization/types.ts b/packages/core/src/serialization/types.ts index 617c4cd205..466b3383aa 100644 --- a/packages/core/src/serialization/types.ts +++ b/packages/core/src/serialization/types.ts @@ -2,7 +2,10 @@ * Shared types for the serialization system. */ -import type { RuntimeDecryptionErrorContext } from '@workflow/errors'; +import type { + RuntimeDecryptionErrorContext, + WorkflowStartStage, +} from '@workflow/errors'; // ---- Format Prefix ---- @@ -93,6 +96,13 @@ export interface SerializableSpecial { // TODO: Make this required when HookConflictError.conflictingRunId is required. conflictingRunId?: string; }; + WorkflowStartError: { + message: string; + stack?: string; + cause?: unknown; + runId: string; + stage: WorkflowStartStage; + }; Int8Array: string; // base64 string Int16Array: string; // base64 string Int32Array: string; // base64 string diff --git a/packages/errors/src/index.ts b/packages/errors/src/index.ts index fb53982131..32a8fc5e5c 100644 --- a/packages/errors/src/index.ts +++ b/packages/errors/src/index.ts @@ -188,6 +188,35 @@ export class WorkflowWorldError extends WorkflowError { } } +/** A start Hook request that the World will never admit. */ +export const START_HOOK_ADMISSION_REJECTED = 'START_HOOK_ADMISSION_REJECTED'; + +export type WorkflowStartStage = 'queue' | 'admission'; + +/** + * Thrown when `start({ hook })` cannot confirm whether a candidate was queued + * or admitted. `runId` identifies that candidate for inspection. + */ +export class WorkflowStartError extends WorkflowWorldError { + readonly runId: string; + readonly stage: WorkflowStartStage; + + constructor(runId: string, stage: WorkflowStartStage, cause: unknown) { + const causeMessage = isError(cause) ? cause.message : String(cause); + super( + `Failed to ${stage === 'queue' ? 'queue' : 'admit'} workflow run "${runId}": ${causeMessage}`, + { cause } + ); + this.name = 'WorkflowStartError'; + this.runId = runId; + this.stage = stage; + } + + static is(value: unknown): value is WorkflowStartError { + return isError(value) && value.name === 'WorkflowStartError'; + } +} + /** * Thrown when a workflow run fails during execution. * @@ -1046,7 +1075,7 @@ export { RUN_ERROR_CODES, type RunErrorCode } from './error-codes.js'; // Cross-realm class registration // --------------------------------------------------------------------------- // -// `FatalError`, `RetryableError`, and `HookConflictError` are not built-ins, so different realms +// Workflow-specific errors are not built-ins, so different realms // (e.g. the workflow VM context vs. the host context that runs the queue // handler) bundle and load their own copies of this module — meaning each // realm has its own distinct class identity. Cross-realm `instanceof` fails @@ -1060,43 +1089,20 @@ export { RUN_ERROR_CODES, type RunErrorCode } from './error-codes.js'; // // First registration in a given realm wins. The descriptor is non-writable // and non-configurable to make accidental clobbering loud. -const FATAL_ERROR_KEY = Symbol.for('@workflow/errors//FatalError'); -const RETRYABLE_ERROR_KEY = Symbol.for('@workflow/errors//RetryableError'); -const HOOK_CONFLICT_ERROR_KEY = Symbol.for( - '@workflow/errors//HookConflictError' -); -const RUNTIME_DECRYPTION_ERROR_KEY = Symbol.for( - '@workflow/errors//RuntimeDecryptionError' -); +const CROSS_REALM_ERROR_CLASSES = { + FatalError, + RetryableError, + HookConflictError, + WorkflowStartError, + RuntimeDecryptionError, +} as const; if (typeof globalThis !== 'undefined') { - if (!Object.hasOwn(globalThis, FATAL_ERROR_KEY)) { - Object.defineProperty(globalThis, FATAL_ERROR_KEY, { - value: FatalError, - writable: false, - enumerable: false, - configurable: false, - }); - } - if (!Object.hasOwn(globalThis, RETRYABLE_ERROR_KEY)) { - Object.defineProperty(globalThis, RETRYABLE_ERROR_KEY, { - value: RetryableError, - writable: false, - enumerable: false, - configurable: false, - }); - } - if (!Object.hasOwn(globalThis, HOOK_CONFLICT_ERROR_KEY)) { - Object.defineProperty(globalThis, HOOK_CONFLICT_ERROR_KEY, { - value: HookConflictError, - writable: false, - enumerable: false, - configurable: false, - }); - } - if (!Object.hasOwn(globalThis, RUNTIME_DECRYPTION_ERROR_KEY)) { - Object.defineProperty(globalThis, RUNTIME_DECRYPTION_ERROR_KEY, { - value: RuntimeDecryptionError, + for (const [name, ErrorClass] of Object.entries(CROSS_REALM_ERROR_CLASSES)) { + const key = Symbol.for(`@workflow/errors//${name}`); + if (Object.hasOwn(globalThis, key)) continue; + Object.defineProperty(globalThis, key, { + value: ErrorClass, writable: false, enumerable: false, configurable: false, diff --git a/packages/web-shared/src/lib/hydration.ts b/packages/web-shared/src/lib/hydration.ts index 25297fb15f..1d8f6d9137 100644 --- a/packages/web-shared/src/lib/hydration.ts +++ b/packages/web-shared/src/lib/hydration.ts @@ -58,6 +58,16 @@ function base64ToArrayBuffer(base64: string): ArrayBuffer { // Web revivers (browser-safe, no Buffer dependency) // --------------------------------------------------------------------------- +type ErrorValue = { message: string; stack?: string; cause?: unknown }; + +function makeWebError(name: string, value: ErrorValue): Error { + const opts = 'cause' in value ? { cause: value.cause } : undefined; + const error = new Error(value.message, opts); + error.name = name; + if (value.stack !== undefined) error.stack = value.stack; + return error; +} + /** * Build a reviver for one of the built-in `Error` subclasses (e.g. * `TypeError`, `RangeError`). The constructor for the named subclass is @@ -90,10 +100,8 @@ function makeWebErrorSubclassReviver( if (typeof Ctor === 'function') { error = new Ctor(value.message, opts); } else { - // Fallback path: no built-in subclass available (exotic env). Construct - // a plain Error with the right `name` and copy `cause` manually since - // the base Error constructor is what we actually called. - error = Object.assign(new Error(value.message, opts), { name }); + // Keep rendering in environments without this built-in subclass. + error = makeWebError(name, value); } if (value.stack !== undefined) error.stack = value.stack; return error; @@ -136,7 +144,7 @@ export function getWebRevivers(): Revivers { // `packages/core/src/serialization/reducers/common.ts`) emits a tagged // entry for each built-in Error subclass plus the workflow-specific // `FatalError` / `RetryableError` / `HookConflictError` / - // `RuntimeDecryptionError` and `AggregateError`. Without + // `WorkflowStartError` / `RuntimeDecryptionError` and `AggregateError`. Without // matching revivers here, `devalue.unflatten` throws "Unknown type X" // — which surfaces in the web o11y UI as "Failed to load resource // details: Unknown type FatalError". @@ -177,34 +185,31 @@ export function getWebRevivers(): Revivers { // class label, but we don't have the real class, so we emit a tagged // Error whose `name` field carries the class identity. This matches // how the existing base `Error` reviver presents unknown subclasses. - FatalError: (value) => { - const opts = 'cause' in value ? { cause: value.cause } : undefined; - const error = new Error(value.message, opts); - error.name = 'FatalError'; - if (value.stack !== undefined) error.stack = value.stack; - return error; - }, + FatalError: (value) => makeWebError('FatalError', value), HookConflictError: (value) => { - const opts = 'cause' in value ? { cause: value.cause } : undefined; - const error = new Error(value.message, opts) as Error & { + const error = makeWebError('HookConflictError', value) as Error & { token?: string; conflictingRunId?: string; }; - error.name = 'HookConflictError'; error.token = value.token; if (value.conflictingRunId !== undefined) { error.conflictingRunId = value.conflictingRunId; } - if (value.stack !== undefined) error.stack = value.stack; + return error; + }, + WorkflowStartError: (value) => { + const error = makeWebError('WorkflowStartError', value) as Error & { + runId: string; + stage: string; + }; + error.runId = value.runId; + error.stage = value.stage; return error; }, RetryableError: (value) => { - const opts = 'cause' in value ? { cause: value.cause } : undefined; - const error = new Error(value.message, opts) as Error & { + const error = makeWebError('RetryableError', value) as Error & { retryAfter?: Date; }; - error.name = 'RetryableError'; - if (value.stack !== undefined) error.stack = value.stack; // `retryAfter` is serialized as an epoch ms number (see the runtime // RetryableError reducer for the rationale around realm-safety). // Rehydrate as a Date so o11y consumers can render it directly. @@ -217,15 +222,12 @@ export function getWebRevivers(): Revivers { return error; }, RuntimeDecryptionError: (value) => { - const opts = 'cause' in value ? { cause: value.cause } : undefined; - const error = new Error(value.message, opts) as Error & { + const error = makeWebError('RuntimeDecryptionError', value) as Error & { context?: unknown; }; - error.name = 'RuntimeDecryptionError'; if (value.context !== undefined) { error.context = value.context; } - if (value.stack !== undefined) error.stack = value.stack; return error; }, DOMException: (value) => { @@ -239,13 +241,7 @@ export function getWebRevivers(): Revivers { if ('cause' in value) (e as { cause?: unknown }).cause = value.cause; return e; } - const error = new Error(value.message); - error.name = value.name; - if (value.stack !== undefined) error.stack = value.stack; - if ('cause' in value) { - (error as Error & { cause?: unknown }).cause = value.cause; - } - return error; + return makeWebError(value.name, value); }, Float32Array: (value: string) => new Float32Array(reviveArrayBuffer(value)), Float64Array: (value: string) => new Float64Array(reviveArrayBuffer(value)), diff --git a/packages/web-shared/test/hydration.test.ts b/packages/web-shared/test/hydration.test.ts index 87e0d72534..f3d4529ccc 100644 --- a/packages/web-shared/test/hydration.test.ts +++ b/packages/web-shared/test/hydration.test.ts @@ -4,7 +4,11 @@ import { dehydrateStepReturnValue, } from '@workflow/core/serialization'; import { hydrateData } from '@workflow/core/serialization-format'; -import { FatalError, RetryableError } from '@workflow/errors'; +import { + FatalError, + RetryableError, + WorkflowStartError, +} from '@workflow/errors'; import { describe, expect, it } from 'vitest'; import { getWebRevivers, @@ -123,6 +127,16 @@ describe('getWebRevivers — error family', () => { expect(revived.conflictingRunId).toBe('wrun_conflicting'); }); + it('hydrates WorkflowStartError details', async () => { + const revived = await roundTrip( + new WorkflowStartError('wrun_candidate', 'queue', new Error('offline')) + ); + + expect(revived.name).toBe('WorkflowStartError'); + expect(revived.runId).toBe('wrun_candidate'); + expect(revived.stage).toBe('queue'); + }); + it('hydrates a RetryableError with retryAfter as a Date', async () => { const retryAt = new Date('2025-01-01T00:00:00.000Z'); const revived = await roundTrip( diff --git a/packages/web-shared/test/serializable-revivers.test.ts b/packages/web-shared/test/serializable-revivers.test.ts index 2c8897d859..0dab8f6827 100644 --- a/packages/web-shared/test/serializable-revivers.test.ts +++ b/packages/web-shared/test/serializable-revivers.test.ts @@ -177,6 +177,13 @@ const SERIALIZABLE_PAYLOADS: Record = { { workflowId: 2 }, 'workflow//example', ], + WorkflowStartError: [ + ['WorkflowStartError', 1], + { message: 2, runId: 3, stage: 4 }, + 'Could not confirm admission', + 'wrun_candidate', + 'admission', + ], WritableStream: [['WritableStream', 1], { name: 2 }, 'stream-a'], }; diff --git a/packages/workflow/src/api.ts b/packages/workflow/src/api.ts index 31ca6ed146..053292a6b8 100644 --- a/packages/workflow/src/api.ts +++ b/packages/workflow/src/api.ts @@ -27,6 +27,7 @@ export { type WorkflowReadableStreamOptions, } from '@workflow/core/runtime/run'; export { + type StartHookOptions, type StartOptions, start, } from '@workflow/core/runtime/start'; diff --git a/packages/workflow/src/internal/errors.ts b/packages/workflow/src/internal/errors.ts index 4490e7ed4f..83fa0b0105 100644 --- a/packages/workflow/src/internal/errors.ts +++ b/packages/workflow/src/internal/errors.ts @@ -15,5 +15,6 @@ export { WorkflowRunNotCompletedError, WorkflowRunNotFoundError, WorkflowRuntimeError, + WorkflowStartError, WorkflowWorldError, } from '@workflow/errors'; diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 0b6eb2b867..d96392da45 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -32,6 +32,7 @@ import { getEventDataPayloadField, HookSchema, type PaginationOptions, + type StartHook, StructuredErrorSchema, WaitSchema, WorkflowRunSchema, @@ -176,6 +177,8 @@ interface CreateEventV4InputBase { * resilient-start path). Validated server-side against the attribute * key/value/count caps. */ attributes?: Record; + /** Atomic start Hook admission data. */ + startHook?: StartHook; /** attr_set's attribute change list ({key, value|null} entries). */ changes?: Array>; /** attr_set's writer provenance ({type:'workflow'} or @@ -442,6 +445,7 @@ function buildPostFrameMeta( meta.executionContext = input.executionContext; } if (input.attributes !== undefined) meta.attributes = input.attributes; + if (input.startHook !== undefined) meta.startHook = input.startHook; if (input.changes !== undefined) meta.changes = input.changes; if (input.writer !== undefined) meta.writer = input.writer; if (input.allowReservedAttributes !== undefined) { diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index ab1eb1632a..823ec04583 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -655,7 +655,26 @@ describe('createWorkflowRunEvent replayDivergenceCount wire field', () => { * runtime tests are the complement: they prove the fields that ARE routed * actually reach the frame meta with the right values and renames. */ -describe('splitEventDataForV4 attribute fields', () => { +describe('splitEventDataForV4 structured fields', () => { + it('carries atomic start Hook admission data as metadata', () => { + const startHook = { + token: 'order:123', + tokenRetentionUntil: new Date('2026-09-01T00:00:00.000Z'), + }; + const { meta } = splitEventDataForV4({ + eventType: 'run_created', + specVersion: 5, + eventData: { + deploymentId: 'dpl_1', + workflowName: 'workflow', + input: new Uint8Array(), + startHook, + }, + }); + + expect(meta.startHook).toEqual(startHook); + }); + it('carries a Hook retention deadline in the frame meta', () => { const tokenRetentionUntil = new Date('2026-07-10T12:00:00.000Z'); const { payload, meta } = splitEventDataForV4({ diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 1e8c482844..7fe187f393 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -46,6 +46,7 @@ import { type ListEventsByCorrelationIdParams, type ListEventsParams, type PaginatedResponse, + type StartHook, validateUlidTimestamp, type WorkflowRun, } from '@workflow/world'; @@ -129,6 +130,8 @@ interface SplitEventData { executionContext?: Record; /** Initial run attributes (run_created / resilient-start run_started). */ attributes?: Record; + /** Atomic start Hook admission data. */ + startHook?: StartHook; /** attr_set change list, included verbatim in frame meta. */ changes?: Array>; /** attr_set writer provenance, included verbatim in frame meta. */ @@ -185,6 +188,7 @@ type MetaSourceField = | 'ownerMessageId' | 'executionContext' | 'attributes' + | 'startHook' | 'changes' | 'writer' | 'allowReservedAttributes' @@ -331,6 +335,9 @@ export function splitEventDataForV4(data: AnyEventRequest): SplitEventData { ) { meta.attributes = eventData.attributes as Record; } + if (eventData.startHook !== undefined) { + meta.startHook = eventData.startHook as StartHook; + } if (Array.isArray(eventData.changes)) { meta.changes = eventData.changes as Array>; } diff --git a/packages/world/src/capabilities.ts b/packages/world/src/capabilities.ts index fa7a115cd1..faea437b40 100644 --- a/packages/world/src/capabilities.ts +++ b/packages/world/src/capabilities.ts @@ -11,6 +11,12 @@ export const WorldCapabilitiesSchema = z.object({ */ hookRetention: z.object({ active: z.boolean() }).optional(), + /** + * Atomically admits a workflow run and reserves its start Hook token. + * Missing or inactive means `start({ hook })` must fail before enqueueing. + */ + atomicStartHook: z.object({ active: z.boolean() }).optional(), + /** * Enforces the event count and update-time preconditions on event creation. * Worlds that accept but ignore either field must leave this unset. diff --git a/packages/world/src/events.test.ts b/packages/world/src/events.test.ts index 587d48535b..574255ae85 100644 --- a/packages/world/src/events.test.ts +++ b/packages/world/src/events.test.ts @@ -22,6 +22,28 @@ describe('hook_created token retention', () => { }); }); +describe('atomic start Hook', () => { + it('parses its absolute retention deadline on run admission', () => { + const parsed = CreateEventSchema.parse({ + eventType: 'run_created', + specVersion: 5, + eventData: { + deploymentId: 'dpl_1', + workflowName: 'workflow', + input: new Uint8Array(), + startHook: { + token: 'order:123', + tokenRetentionUntil: '2026-09-01T00:00:00.000Z', + }, + }, + }); + + expect(parsed.eventData.startHook?.tokenRetentionUntil).toEqual( + new Date('2026-09-01T00:00:00.000Z') + ); + }); +}); + describe('step_started ownerMessageId', () => { it('accepts a bare step_started with no eventData (legacy contract)', () => { const parsed = CreateEventSchema.parse({ diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 7eb2d0d603..709f245f94 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -405,6 +405,13 @@ export const HookCreatedEventSchema = BaseEventSchema.extend({ }), }); +/** Hook token reserved atomically while admitting a workflow run. */ +export const StartHookSchema = z.object({ + token: z.string().min(1), + tokenRetentionUntil: z.coerce.date().optional(), +}); +export type StartHook = z.infer; + const HookReceivedEventSchema = BaseEventSchema.extend({ eventType: z.literal('hook_received'), correlationId: z.string(), @@ -490,27 +497,27 @@ const AttrSetEventSchema = BaseEventSchema.extend({ // Run lifecycle events // ============================================================================= +/** Data shared by direct and resilient run creation. */ +export const RunCreationDataSchema = z.object({ + deploymentId: z.string(), + workflowName: z.string(), + input: SerializedDataSchema, + executionContext: z.record(z.string(), z.any()).optional(), + attributes: z.record(z.string(), z.string()).optional(), + allowReservedAttributes: z.literal(true).optional(), + startHook: StartHookSchema.optional(), + /** Public key used by cross-run writers to seal payloads to this run. */ + encryptionPublicKey: z.string().optional(), +}); +export type RunCreationData = z.infer; + /** * Event created when a workflow run is first created. The World implementation * atomically creates both the event and the run entity with status 'pending'. */ const RunCreatedEventSchema = BaseEventSchema.extend({ eventType: z.literal('run_created'), - eventData: z.object({ - deploymentId: z.string(), - workflowName: z.string(), - input: SerializedDataSchema, - executionContext: z.record(z.string(), z.any()).optional(), - attributes: z.record(z.string(), z.string()).optional(), - allowReservedAttributes: z.literal(true).optional(), - /** - * The run's X25519 public key (base64), stamped by SDKs that support - * sealed (`encp`) envelopes. Persisted onto the run entity so that - * cross-run writers can seal payloads to this run without holding its - * symmetric key. Not secret — see `WorkflowRunBaseSchema`. - */ - encryptionPublicKey: z.string().optional(), - }), + eventData: RunCreationDataSchema, }); /** @@ -524,23 +531,7 @@ const RunCreatedEventSchema = BaseEventSchema.extend({ */ const RunStartedEventSchema = BaseEventSchema.extend({ eventType: z.literal('run_started'), - eventData: z - .object({ - input: SerializedDataSchema.optional(), - deploymentId: z.string().optional(), - workflowName: z.string().optional(), - executionContext: z.record(z.string(), z.any()).optional(), - attributes: z.record(z.string(), z.string()).optional(), - allowReservedAttributes: z.literal(true).optional(), - /** - * Mirrors `run_created.eventData.encryptionPublicKey`. Carried here for - * the resilient-start path: when the `run_created` write failed, the - * server creates the run from this event instead, and without the key - * the run would silently lose its ability to receive sealed writes. - */ - encryptionPublicKey: z.string().optional(), - }) - .optional(), + eventData: RunCreationDataSchema.partial().optional(), }); /** diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index bc0234703d..8a3fb788fb 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -55,7 +55,9 @@ export { isTerminalStepEventType, isWaitEventType, RUN_EVENT_TYPES, + RunCreationDataSchema, STEP_EVENT_TYPES, + StartHookSchema, stripEventDataRefs, TERMINAL_RUN_EVENT_TYPES, TERMINAL_STEP_EVENT_TYPES, diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index 0d5936540e..df7de9eb4e 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -258,6 +258,15 @@ export interface Storage { events: { /** + * A World advertising `atomicStartHook` must atomically reserve + * `eventData.startHook.token` while creating `run_created` or resilient + * `run_started`. Each candidate `runId` has one immutable decision: + * repeated accepted attempts return that run, while repeated rejected + * attempts throw `HookConflictError` with the same `conflictingRunId`. + * A permanent request validation failure must throw `WorkflowWorldError` + * with code `START_HOOK_ADMISSION_REJECTED` so a queued copy can stop + * retrying without hiding unrelated World errors. + * * Create a run_created event to start a new workflow run. * The runId may be provided by the client or left as null for the server to generate. * diff --git a/packages/world/src/queue.ts b/packages/world/src/queue.ts index c756a99ed9..42649f10ee 100644 --- a/packages/world/src/queue.ts +++ b/packages/world/src/queue.ts @@ -1,4 +1,5 @@ import { z } from 'zod/v4'; +import { RunCreationDataSchema } from './events.js'; export type QueueKind = 'workflow'; @@ -97,20 +98,8 @@ export type TraceCarrier = z.infer; * When the runtime processes the message, it passes this data to the * run_started event so the server can create the run if it doesn't exist yet. */ -export const RunInputSchema = z.object({ - input: z.unknown(), - deploymentId: z.string(), - workflowName: z.string(), +export const RunInputSchema = RunCreationDataSchema.extend({ specVersion: z.number(), - executionContext: z.record(z.string(), z.any()).optional(), - /** Initial plaintext run attributes, for resilient run creation. */ - attributes: z.record(z.string(), z.string()).optional(), - /** - * Permits reserved `$`-prefixed keys in `attributes`, mirrored from the - * `start()` option so resilient run creation validates the same way as - * the original `run_created` attempt. - */ - allowReservedAttributes: z.literal(true).optional(), /** * The environment the creating client's writes are attributed to, as * reported by {@link World.getEnvironment} at `start()` time (on Vercel: