diff --git a/.changeset/stream-run-started-page.md b/.changeset/stream-run-started-page.md new file mode 100644 index 0000000000..e14e9fde97 --- /dev/null +++ b/.changeset/stream-run-started-page.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-vercel': patch +--- + +Load the first replay page from non-Turbo `run_started` responses while preserving Turbo's preload opt-out. diff --git a/packages/world-vercel/src/events-retry.test.ts b/packages/world-vercel/src/events-retry.test.ts index 68be2d0982..8de70fd2b2 100644 --- a/packages/world-vercel/src/events-retry.test.ts +++ b/packages/world-vercel/src/events-retry.test.ts @@ -38,6 +38,7 @@ const stepCompleted = () => ({ }); const v4Success = () => ({ + type: 'event' as const, eventId: 'evnt_1', runId: RUN_ID, createdAt: '2020-01-01T00:00:00.000Z', diff --git a/packages/world-vercel/src/events-v4.test.ts b/packages/world-vercel/src/events-v4.test.ts index 7b21ff8d83..2781fa2e61 100644 --- a/packages/world-vercel/src/events-v4.test.ts +++ b/packages/world-vercel/src/events-v4.test.ts @@ -1,3 +1,4 @@ +import { Buffer } from 'node:buffer'; import { EntityConflictError, RunExpiredError, @@ -398,68 +399,86 @@ describe('createWorkflowRunEventV4 over HTTP', () => { agent.assertNoPendingInterceptors(); }); - it('forwards skipPreload in the run_started frame meta (turbo preload opt-out)', async () => { + it('requests and decodes the first event page for run_started', async () => { const origin = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; const agent = new MockAgent(); agent.disableNetConnect(); - - // Decode the posted frame's CBOR meta block: - // u32_be(meta_len) || cbor_meta || u32_be(body_len) || body - let capturedMeta: Record | undefined; - const captureMeta = (rawBody: unknown) => { - const bytes = - typeof rawBody === 'string' - ? new TextEncoder().encode(rawBody) - : new Uint8Array(rawBody as ArrayBufferLike); - const metaLen = new DataView( - bytes.buffer, - bytes.byteOffset, - bytes.byteLength - ).getUint32(0, false); - capturedMeta = decode(bytes.subarray(4, 4 + metaLen)) as Record< - string, - unknown - >; - }; + const input = new TextEncoder().encode('"workflow input"'); agent .get(origin) .intercept({ path: '/api/v4/runs/wrun_1/events/run_started', method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, }) .reply( 200, - (opts: { body?: unknown }) => { - captureMeta(opts.body); - return encode({ run: { runId: 'wrun_1', status: 'running' } }); - }, + Buffer.concat([ + encodeFrame( + { + _result: 1, + event: { + eventId: 'evnt_2', + runId: 'wrun_1', + eventType: 'run_started', + }, + run: { runId: 'wrun_1', status: 'running' }, + }, + new Uint8Array() + ), + encodeFrame( + { + eventId: 'evnt_1', + runId: 'wrun_1', + eventType: 'run_created', + }, + input + ), + encodeFrame( + { + eventId: 'evnt_2', + runId: 'wrun_1', + eventType: 'run_started', + }, + new Uint8Array() + ), + encodeFrame( + { _end: 1, next: 'eid:evnt_2', hasMore: false }, + new Uint8Array() + ), + ]), { headers: { - 'x-wf-event-id': 'evnt_1', + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-event-id': 'evnt_2', 'x-wf-run-id': 'wrun_1', 'x-wf-created-at': '2026-06-10T00:00:00.000Z', }, } ); - await createWorkflowRunEventV4( + const result = await createWorkflowRunEventV4( { runId: 'wrun_1', eventType: 'run_started', specVersion: 5, - skipPreload: true, }, { token: 'test-token', dispatcher: agent } ); - expect(capturedMeta?.eventType).toBe('run_started'); - expect(capturedMeta?.skipPreload).toBe(true); + expect(result.type).toBe('event-page'); + if (result.type !== 'event-page') throw new Error('expected event page'); + expect(result.body.run).toMatchObject({ status: 'running' }); + expect(result.page.events).toHaveLength(2); + expect(result.page.events[0].body).toEqual(input); + expect(result.page.next).toBe('eid:evnt_2'); + expect(result.page.hasMore).toBe(false); agent.assertNoPendingInterceptors(); }); - it('omits skipPreload from the frame meta when not set (default / old SDK parity)', async () => { + it('keeps the CBOR response when run_started skips the preload', async () => { const origin = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; const agent = new MockAgent(); @@ -471,6 +490,7 @@ describe('createWorkflowRunEventV4 over HTTP', () => { .intercept({ path: '/api/v4/runs/wrun_1/events/run_started', method: 'POST', + headers: (headers) => headers.accept === '*/*', }) .reply( 200, @@ -489,20 +509,26 @@ describe('createWorkflowRunEventV4 over HTTP', () => { }, { headers: { - 'x-wf-event-id': 'evnt_1', + 'x-wf-event-id': 'evnt_2', 'x-wf-run-id': 'wrun_1', 'x-wf-created-at': '2026-06-10T00:00:00.000Z', }, } ); - await createWorkflowRunEventV4( - { runId: 'wrun_1', eventType: 'run_started', specVersion: 5 }, + const result = await createWorkflowRunEventV4( + { + runId: 'wrun_1', + eventType: 'run_started', + specVersion: 5, + skipPreload: true, + }, { token: 'test-token', dispatcher: agent } ); - expect(capturedMeta?.eventType).toBe('run_started'); - expect('skipPreload' in (capturedMeta ?? {})).toBe(false); + expect(capturedMeta?.skipPreload).toBe(true); + expect(result.type).toBe('event'); + expect(result.body.run).toMatchObject({ status: 'running' }); agent.assertNoPendingInterceptors(); }); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index f5422c81d9..34e88a6686 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -204,16 +204,13 @@ export interface CreateEventV4Input { stateUpdatedAt?: number; } -export interface CreateEventV4Result { +interface CreateEventV4ResultBase { eventId: string; runId: string; createdAt: string; /** - * Materialized-entity bag — CBOR-decoded from the response body. The - * server hands back the same shape v2/v3 use for EventResult so the - * adapter layer can drop these fields into its return value unchanged. - * Keys are unset when the event type doesn't materialize that entity - * kind. + * Materialized-entity bag decoded from CBOR or the leading result frame. + * Keys are unset when the event type doesn't materialize that entity kind. */ body: { event?: unknown; @@ -238,6 +235,21 @@ export interface CreateEventV4Result { }; } +type RequiredEventPageV4Result = Omit< + ListEventsV4Result, + 'next' | 'hasMore' +> & { + next: string; + hasMore: boolean; +}; + +export type CreateEventV4Result = + | (CreateEventV4ResultBase & { type: 'event' }) + | (CreateEventV4ResultBase & { + type: 'event-page'; + page: RequiredEventPageV4Result; + }); + /** Build the CBOR meta map for a v4 POST frame. Drops undefined entries * so the wire shape matches what the server expects to see. */ function buildPostFrameMeta( @@ -363,9 +375,9 @@ export function throwForErrorResponse( /** * POST /api/v4/runs/:runId/events/:eventType * - * Sends the full request as a single v4 frame and returns the event ids - * + materialized-entity bag from the CBOR response body. Throws on - * non-2xx. + * Sends the full request as a single v4 frame. A run_started response may + * include the first event page as frames; skipPreload and other events use + * CBOR. * * The trailing `:eventType` path segment is an alias of the canonical * `/events` route: it exists purely so the event type is visible in @@ -382,6 +394,9 @@ export async function createWorkflowRunEventV4( const { baseUrl, headers: baseHeaders } = await getHttpConfig(config); const headers = new Headers(baseHeaders); headers.set('Content-Type', 'application/octet-stream'); + if (input.eventType === 'run_started' && !input.skipPreload) { + headers.set('Accept', V4_FRAME_CONTENT_TYPE); + } const frame = encodeFrame( buildPostFrameMeta(input), @@ -407,6 +422,35 @@ export async function createWorkflowRunEventV4( throw new Error('v4 createEvent: response missing required x-wf-* headers'); } + const contentType = response.headers.get('content-type'); + if (contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) { + if (input.eventType !== 'run_started') { + throw new Error( + 'v4 createEvent: received an event page for a non-run_started event' + ); + } + const stream = await consumeEventFrameStream(response, 'createEvent'); + switch (stream.type) { + case 'events': + throw new Error( + 'v4 createEvent: frame stream ended without the result frame' + ); + case 'event-page': + return { + type: 'event-page', + eventId, + runId, + createdAt, + body: stream.body, + page: stream.page, + }; + default: { + const exhaustive: never = stream; + throw new Error(`v4 createEvent: unknown stream type ${exhaustive}`); + } + } + } + // Decode the materialized-entity bag from the CBOR response body. const bodyBytes = new Uint8Array(await response.arrayBuffer()); const body = @@ -414,7 +458,7 @@ export async function createWorkflowRunEventV4( ? (decode(bodyBytes) as CreateEventV4Result['body']) : {}; - return { eventId, runId, createdAt, body }; + return { type: 'event', eventId, runId, createdAt, body }; } /** @@ -537,6 +581,91 @@ export interface ListEventsV4Result { hasMore?: boolean; } +type EventFrameStreamResult = + | { type: 'events'; page: ListEventsV4Result } + | { + type: 'event-page'; + body: CreateEventV4ResultBase['body']; + page: RequiredEventPageV4Result; + }; + +type EventFrameStreamState = + | { type: 'events'; events: ListedEventV4[] } + | { + type: 'event-page'; + body: CreateEventV4ResultBase['body']; + events: ListedEventV4[]; + }; + +function finishEventFrameStream( + state: EventFrameStreamState, + meta: Record, + opName: string +): EventFrameStreamResult { + const next = typeof meta.next === 'string' ? meta.next : undefined; + const hasMore = typeof meta.hasMore === 'boolean' ? meta.hasMore : undefined; + + switch (state.type) { + case 'events': + return { type: 'events', page: { events: state.events, next, hasMore } }; + case 'event-page': + if (next === undefined || hasMore === undefined) { + throw new Error(`v4 ${opName}: event page missing pagination state`); + } + return { + type: 'event-page', + body: state.body, + page: { events: state.events, next, hasMore }, + }; + default: { + const exhaustive: never = state; + throw new Error(`v4 ${opName}: unknown stream state ${exhaustive}`); + } + } +} + +async function consumeEventFrameStream( + response: Response, + opName: string +): Promise { + const contentType = response.headers.get('content-type'); + if (!contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) { + throw new Error( + `v4 ${opName}: expected ${V4_FRAME_CONTENT_TYPE}, got ${contentType ?? '(none)'}` + ); + } + + const chunks = response.body as unknown as AsyncIterable; + let state: EventFrameStreamState = { type: 'events', events: [] }; + + for await (const frame of decodeFrames(chunks)) { + if (frame.meta._result === 1) { + if ( + state.type !== 'events' || + state.events.length > 0 || + frame.body.byteLength > 0 + ) { + throw new Error(`v4 ${opName}: invalid result frame`); + } + const { _result, ...body } = frame.meta; + state = { type: 'event-page', body, events: state.events }; + continue; + } + if (frame.meta._end === 1) { + return finishEventFrameStream(state, frame.meta, opName); + } + state.events.push({ + event: frame.meta as unknown as DecodedV4Event, + body: frame.body, + }); + } + + throw new Error( + `v4 ${opName}: frame stream ended without the end-of-stream sentinel ` + + `(${state.events.length} events read) — truncated response?` + ); +} + /** * Drive a v4 frame-stream list response into an in-memory page. Used by * both the by-runId and by-correlationId list endpoints — the wire @@ -558,52 +687,17 @@ async function consumeListFrameStream( config, opName ); - const contentType = response.headers.get('content-type'); - if (!contentType?.startsWith(V4_FRAME_CONTENT_TYPE)) { - throw new Error( - `v4 ${opName}: expected ${V4_FRAME_CONTENT_TYPE}, got ${contentType ?? '(none)'}` - ); - } - - // See getEventV4: fetch's web ReadableStream is async-iterable on Node; the - // cast only works around TS's lib type omitting the async iterator. - const chunks = response.body as unknown as AsyncIterable; - - const events: ListedEventV4[] = []; - let next: string | undefined; - let hasMore: boolean | undefined; - let sawEndSentinel = false; - for await (const frame of decodeFrames(chunks)) { - if (frame.meta._end === 1) { - if (typeof frame.meta.next === 'string') next = frame.meta.next; - if (typeof frame.meta.hasMore === 'boolean') hasMore = frame.meta.hasMore; - sawEndSentinel = true; - break; + const stream = await consumeEventFrameStream(response, opName); + switch (stream.type) { + case 'events': + return stream.page; + case 'event-page': + throw new Error(`v4 ${opName}: unexpected result frame`); + default: { + const exhaustive: never = stream; + throw new Error(`v4 ${opName}: unknown stream type ${exhaustive}`); } - events.push({ - event: frame.meta as unknown as DecodedV4Event, - body: frame.body, - }); } - - // A LIST response always ends with the `{_end: 1}` sentinel frame. EOF - // without it means the response was truncated — and if the cut landed - // between two complete frames, decodeFrames alone can't tell. Returning - // the partial page here would surface as `hasMore: false` and silently - // drop events (replay correctness!), so fail loudly instead; the read - // is idempotent and safe for the caller to retry. - if (!sawEndSentinel) { - throw new Error( - `v4 ${opName}: frame stream ended without the end-of-stream sentinel ` + - `(${events.length} events read) — truncated response?` - ); - } - - return { - events, - ...(next ? { next } : {}), - ...(hasMore !== undefined ? { hasMore } : {}), - }; } /** diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index f78845fdaf..c4479756b4 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -1,3 +1,5 @@ +import { Buffer } from 'node:buffer'; +import { gzipSync } from 'node:zlib'; import type { AnyEventRequest } from '@workflow/world'; import { decode, encode } from 'cbor-x'; import { ulid } from 'ulid'; @@ -629,6 +631,99 @@ describe('createWorkflowRunEvent response coercion', () => { agent.assertNoPendingInterceptors(); }); + it('hydrates streamed run input without decompressing replay events', async () => { + const agent = mockAgent(); + const serializedInput = new TextEncoder().encode('"workflow input"'); + const compressedInput = gzipSync(serializedInput); + const input = new Uint8Array(4 + compressedInput.byteLength); + input.set(new TextEncoder().encode('gzip')); + input.set(compressedInput, 4); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/run_started', + method: 'POST', + headers: { accept: V4_FRAME_CONTENT_TYPE }, + }) + .reply( + 200, + Buffer.concat([ + encodeFrame( + { + _result: 1, + run: { + runId: 'wrun_1', + status: 'running', + startedAt: new Date('2026-06-10T00:00:01.000Z'), + }, + event: { + eventId: 'evnt_2', + runId: 'wrun_1', + eventType: 'run_started', + createdAt: new Date('2026-06-10T00:00:01.000Z'), + }, + }, + new Uint8Array() + ), + encodeFrame( + { + eventId: 'evnt_1', + runId: 'wrun_1', + eventType: 'run_created', + createdAt: new Date('2026-06-10T00:00:00.000Z'), + specVersion: 5, + eventData: { + deploymentId: 'dpl_1', + workflowName: 'wf', + input: { _type: 'RemoteRef', value: 'dbrf:unused' }, + }, + }, + input + ), + encodeFrame( + { + eventId: 'evnt_2', + runId: 'wrun_1', + eventType: 'run_started', + createdAt: new Date('2026-06-10T00:00:01.000Z'), + specVersion: 5, + eventData: {}, + }, + new Uint8Array() + ), + encodeFrame( + { _end: 1, next: 'eid:evnt_2', hasMore: false }, + new Uint8Array() + ), + ]), + { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + 'x-wf-event-id': 'evnt_2', + 'x-wf-run-id': 'wrun_1', + 'x-wf-created-at': '2026-06-10T00:00:01.000Z', + }, + } + ); + + const result = await createWorkflowRunEvent( + 'wrun_1', + { eventType: 'run_started', specVersion: 5 }, + undefined, + { token: 'test-token', dispatcher: agent } + ); + + expect(result.events?.map((event) => event.eventType)).toEqual([ + 'run_created', + 'run_started', + ]); + expect(result.events?.[0].eventData?.input).toEqual(input); + expect(result.run?.input).toEqual(input); + expect(result.cursor).toBe('eid:evnt_2'); + expect(result.hasMore).toBe(false); + agent.assertNoPendingInterceptors(); + }); + it('threads the wait entity through to the EventResult', async () => { const agent = mockAgent(); agent diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index a2d223a882..c686fa0e80 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -16,9 +16,8 @@ * * - POST request body is one v4 frame (meta + payload). The response * surfaces eventId/runId/createdAt as `x-wf-*` headers and carries - * the materialized EventResult (event/run/step/hook/wait/events/ - * cursor/hasMore) as a CBOR body — `remoteRefBehavior` in the frame - * meta still controls server-side ref resolution. + * a materialized EventResult. `run_started` negotiates a frame stream + * containing that result and the first event page; other events use CBOR. * - GET single event returns one v4 frame: the event entity in the * frame meta, the user payload bytes in the frame body. * - LIST events returns a stream of v4 frames terminated by a sentinel @@ -476,30 +475,44 @@ function decodeLegacyStructuredError(payload: Uint8Array): unknown { * * Both GET single-event and LIST use the same frame format: meta is the * full event entity with the payload field as a RefDescriptor, body is - * the resolved payload bytes (possibly empty). This helper splices the - * body bytes into `eventData[fieldName]`, normalizing any zstd wrapper - * back to the raw devalue-with-format-prefix Uint8Array the runtime's - * hydrate helpers (hydrateStepIO, hydrateRunError, …) consume. Stable-line - * structured errors are the exception: the backend stored those as CBOR, - * so they are decoded after checking that the payload is not a current - * format-prefixed serialized value. + * the resolved payload bytes (possibly empty). Read mode normalizes + * compression wrappers for display. Replay mode preserves the serialized + * bytes so the runtime's hydrate helpers own decompression and telemetry. + * Stable-line structured errors are decoded after checking that the payload + * is not a current format-prefixed serialized value. */ function buildEventFromV4( decoded: DecodedV4Event, payloadBody: Uint8Array, - resolveData: 'none' | 'all' + mode: 'none' | 'all' | 'replay' ): Event { const eventData = (decoded.eventData ?? {}) as Record; + let normalizePayload: boolean; + switch (mode) { + case 'none': + case 'all': + normalizePayload = true; + break; + case 'replay': + normalizePayload = false; + break; + default: { + const exhaustive: never = mode; + throw new Error(`Unsupported v4 event mode: ${exhaustive}`); + } + } if (payloadBody.byteLength > 0) { const payloadField = getEventDataPayloadField(decoded.eventType); - const normalizedPayload = normalizeSerializedData(payloadBody); - if (payloadField && normalizedPayload instanceof Uint8Array) { + const payload = normalizePayload + ? normalizeSerializedData(payloadBody) + : payloadBody; + if (payloadField && payload instanceof Uint8Array) { eventData[payloadField] = legacyStructuredErrorEventTypes.has( decoded.eventType ) - ? decodeLegacyStructuredError(normalizedPayload) - : normalizedPayload; + ? decodeLegacyStructuredError(payload) + : payload; } } @@ -526,11 +539,13 @@ function buildEventFromV4( : {}), }; - const event = coerceNormalizedEvent(raw); + const event = normalizePayload + ? coerceNormalizedEvent(raw) + : coerceEventDates(raw); // For resolveData='none', strip eventData entirely. Reuse the world- // side helper so behavior stays in sync with other backends. - return resolveData === 'none' ? stripEventDataRefs(event, 'none') : event; + return mode === 'none' ? stripEventDataRefs(event, 'none') : event; } // ============================================================================= @@ -738,26 +753,18 @@ async function createWorkflowRunEventInner( // matching the v3 path's stripEventAndLegacyRefs behavior. const resolveData = params?.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; const body = result.body; - return { + const baseResult = { event: body.event ? stripEventDataRefs( coerceEventDates(body.event as Record), resolveData ) : undefined, - run: body.run - ? deserializeError(body.run as Record) - : undefined, step: body.step ? deserializeStep(body.step as Parameters[0]) : undefined, hook: body.hook as EventResult['hook'], wait: body.wait as EventResult['wait'], - events: body.events - ? (body.events as Record[]).map(coerceEventDates) - : undefined, - cursor: body.cursor ?? undefined, - hasMore: body.hasMore, // Lazy step start: thread the server's "I created the step on this call" // signal through so the owned-inline runtime path can gate body execution // on it. Absent from older servers → undefined → safe default. @@ -767,4 +774,49 @@ async function createWorkflowRunEventInner( ? { maxEvents: body.maxEvents } : {}), }; + + switch (result.type) { + case 'event': + return { + ...baseResult, + run: body.run + ? deserializeError(body.run as Record) + : undefined, + events: body.events + ? (body.events as Record[]).map(coerceEventDates) + : undefined, + cursor: body.cursor, + hasMore: body.hasMore, + }; + case 'event-page': { + const replayEvents = result.page.events.map(({ event, body }) => + buildEventFromV4(event, body, 'replay') + ); + const runCreated = replayEvents[0]; + if (runCreated?.eventType !== 'run_created') { + throw new Error( + 'v4 createEvent: run_started page must begin with run_created' + ); + } + if (!body.run) { + throw new Error('v4 createEvent: run_started result missing run'); + } + const { inputRef: _inputRef, ...run } = deserializeError< + WorkflowRun & { inputRef?: unknown } + >(body.run as Record); + return { + ...baseResult, + run: { ...run, input: runCreated.eventData.input }, + events: replayEvents.map((event) => + stripEventDataRefs(event, resolveData) + ), + cursor: result.page.next, + hasMore: result.page.hasMore, + }; + } + default: { + const exhaustive: never = result; + throw new Error(`v4 createEvent: unknown result type ${exhaustive}`); + } + } }