Skip to content

[core] Add atomic start Hook admission #3426

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NathanColosimo wants to merge 1 commit into from
Open

[core] Add atomic start Hook admission #3426

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions .changeset/atomic-start-hook.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
1 change: 1 addition & 0 deletions packages/cli/src/lib/inspect/hydration.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,7 @@ const ERROR_REVIVER_KEYS = [
'EvalError',
'FatalError',
'HookConflictError',
'WorkflowStartError',
'RangeError',
'ReferenceError',
'RetryableError',
Expand Down
88 changes: 87 additions & 1 deletion packages/core/src/runtime.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import {
EntityConflictError,
HookConflictError,
PreconditionFailedError,
RUN_ERROR_CODES,
START_HOOK_ADMISSION_REJECTED,
ThrottleError,
WorkflowWorldError,
} from '@workflow/errors';
Expand Down Expand Up @@ -2180,6 +2182,7 @@ describe('workflowEntrypoint turbo mode', () => {
workflowName: 'workflow',
specVersion: SPEC_VERSION_CURRENT,
executionContext: {},
encryptionPublicKey: 'test-public-key',
};
}

Expand All @@ -2194,6 +2197,8 @@ describe('workflowEntrypoint turbo mode', () => {
attempt: number;
source: string;
runStartedGate?: Promise<void>;
startHook?: { token: string };
runStartedError?: Error;
}) {
const { runId, attempt, source } = opts;
const order = turboOrder;
Expand Down Expand Up @@ -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[] };
Expand Down Expand Up @@ -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',
Expand Down Expand Up @@ -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 () => {
Expand All @@ -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({
Expand Down
49 changes: 37 additions & 12 deletions packages/core/src/runtime.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,15 @@ import {
CorruptedEventLogError,
EntityConflictError,
FatalError,
HookConflictError,
HookNotFoundError,
MaxEventsExceededError,
PreconditionFailedError,
ReplayDivergenceError,
RUN_ERROR_CODES,
type RunErrorCode,
RunExpiredError,
START_HOOK_ADMISSION_REJECTED,
WorkflowRuntimeError,
WorkflowWorldError,
} from '@workflow/errors';
Expand All @@ -28,6 +30,7 @@ import {
isLegacySpecVersion,
isTerminalRunEventType,
ROOT_RUN_ID_ATTRIBUTE,
type RunCreationData,
type RunInput,
resolveQueueNamespace,
SPEC_VERSION_CURRENT,
Expand Down Expand Up @@ -170,6 +173,7 @@ export {
wakeUpRun,
} from './runtime/runs.js';
export {
type StartHookOptions,
type StartOptions,
type StartOptionsBase,
type StartOptionsWithDeploymentId,
Expand Down Expand Up @@ -1011,6 +1015,7 @@ export function workflowEntrypoint(
const turbo =
isTurboEnabled() &&
runInput !== undefined &&
runInput.startHook === undefined &&
metadata.attempt === 1 &&
incomingStepId === undefined &&
!replayDivergence;
Expand Down Expand Up @@ -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;

@VaguelySerious VaguelySerious Aug 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AI Review: Note

Behavior change that applies to every run, not just atomic-start ones.

Replacing the explicit six-field pick with {...runInput} minus environment/specVersion means run_started now also carries encryptionPublicKey on the resilient-start path. Verified empirically by porting the new assertion on line 2348 back to the base branch:

base: expected undefined to be 'test-public-key'
head: passes

This looks like a fix, and it matches what the (now-deleted) RunStartedEventSchema comment described: on the resilient path the run is created from this event, and without the key it silently loses the ability to receive sealed writes. But it ships unannounced — the changeset doesn't mention it, and the only coverage is an added assertion inside a test named for turbo optimistic start. Worth its own changeset line and a test that names the behavior, so a future refactor doesn't drop it again.

The spread also means any field added to RunInput later is auto-forwarded into run_started rather than opted in. That's the mechanism that just quietly changed the payload here.

}
const runStartedEvent = {
eventType: 'run_started' as const,
// Use the spec version from the original start() call
Expand All @@ -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) {
Expand Down Expand Up @@ -2204,6 +2208,27 @@ export function workflowEntrypoint(
return;
}
} catch (err) {
if (runInput?.startHook !== undefined) {

@VaguelySerious VaguelySerious Aug 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AI Review: Blocking

This prelude sits before the EntityConflictError/RunExpiredError check and before the getWorkflowSetupErrorCode -> recordFatalRunError path. EntityConflictError, RunExpiredError, and PreconditionFailedError all extend WorkflowWorldError and all classify as non-retryable, so err instanceof WorkflowWorldError && !isRetryableWorldError(err) shadows every one of them.

For a world-contract error the pre-existing path writes run_failed; the atomic path logs at ERROR and acks the message. The run stays in pending forever with no terminal event and no retry.

I confirmed this against the driveTurbo harness in this PR (scratch test, not committed):

non-atomic run + WorkflowWorldError{code: WORLD_CONTRACT_ERROR}  -> 1 run_failed
atomic-start run + the identical error                            -> 0 run_failed

A second scratch test showed RunExpiredError in the atomic path now logs error: "Atomic start Hook admission rejected queued candidate" and never logs the pre-existing info: "Run already finished during setup, skipping" — same outcome, wrong severity, misleading message.

The second branch is strictly worse than falling through for contract errors, since the fall-through already stops the retry loop and records the failure. Narrowing the swallow to HookConflictError.is(err) alone fixes the case that matters. If you also want to stop queue retries for non-contract, non-retryable world errors, route those through recordFatalRunError rather than a bare return.

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) ||
Expand Down
40 changes: 40 additions & 0 deletions packages/core/src/runtime/quickjs-serde.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<string, unknown>;
const reduced: Record<string, unknown> = {
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) => {
Expand Down Expand Up @@ -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) => {
Expand Down
Loading
Loading