diff --git a/.changeset/zstd-step-error-display.md b/.changeset/zstd-step-error-display.md new file mode 100644 index 0000000000..8a104882fe --- /dev/null +++ b/.changeset/zstd-step-error-display.md @@ -0,0 +1,7 @@ +--- +'@workflow/world-vercel': patch +'@workflow/web-shared': patch +'@workflow/web': patch +--- + +Decompress gzip- and zstd-prefixed serialized data returned from Vercel Workflow storage, and route OSS web hydration through the async WASM-capable path for compressed payloads. diff --git a/packages/web-shared/src/components/sidebar/events-list.tsx b/packages/web-shared/src/components/sidebar/events-list.tsx index e91bbde12d..5996049228 100644 --- a/packages/web-shared/src/components/sidebar/events-list.tsx +++ b/packages/web-shared/src/components/sidebar/events-list.tsx @@ -1,8 +1,12 @@ 'use client'; -import { EVENT_DATA_REF_FIELDS, type Event } from '@workflow/world'; +import type { Event } from '@workflow/world'; import { useCallback, useLayoutEffect, useMemo, useRef, useState } from 'react'; -import { hasEncryptedFields, isExpiredMarker } from '../../lib/hydration'; +import { + getEventDataRefFields, + hasEncryptedFields, + isExpiredMarker, +} from '../../lib/hydration'; import { Collapsible, CollapsibleContent, @@ -230,7 +234,7 @@ function EventItem({ /** * Check if an eventData object has only expired marker values in ref/payload - * fields for this event type (see {@link EVENT_DATA_REF_FIELDS}). Other keys + * fields for this event type (see {@link getEventDataRefFields}). Other keys * (e.g. `resumeAt`, `stepName`) are ignored. */ function hasOnlyExpiredFields(data: unknown, eventType: string): boolean { @@ -238,7 +242,7 @@ function hasOnlyExpiredFields(data: unknown, eventType: string): boolean { return false; } const record = data as Record; - const refKeys = EVENT_DATA_REF_FIELDS[eventType] ?? []; + const refKeys = getEventDataRefFields(eventType); const presentKeys = refKeys.filter((k) => k in record); return ( presentKeys.length > 0 && diff --git a/packages/web-shared/src/index.ts b/packages/web-shared/src/index.ts index eeb564d735..0fcb536b70 100644 --- a/packages/web-shared/src/index.ts +++ b/packages/web-shared/src/index.ts @@ -14,13 +14,6 @@ export { waitEventsToWaitEntity, } from './components/workflow-traces/trace-span-construction'; export type { EventAnalysis } from './lib/event-analysis'; -export { - parseExactWorkflowSearchId, - looksLikeWorkflowIdSearchInput, - type ExactWorkflowSearchId, - type ExactWorkflowSearchIdKind, - type ExactIdSearchResult, -} from './lib/exact-event-search-id'; export { analyzeEvents, hasPendingHooksFromEvents, @@ -40,6 +33,13 @@ export { materializeSteps, materializeWaits, } from './lib/event-materialization'; +export { + type ExactIdSearchResult, + type ExactWorkflowSearchId, + type ExactWorkflowSearchIdKind, + looksLikeWorkflowIdSearchInput, + parseExactWorkflowSearchId, +} from './lib/exact-event-search-id'; export type { Revivers, StreamRef } from './lib/hydration'; export { CLASS_INSTANCE_REF_TYPE, @@ -49,6 +49,7 @@ export { getWebRevivers, hasEncryptedFields, hydrateResourceIO, + hydrateResourceIOAsync, hydrateResourceIOWithKey, isClassInstanceRef, isEncryptedMarker, diff --git a/packages/web-shared/src/lib/hydration.ts b/packages/web-shared/src/lib/hydration.ts index b7472021b9..edefa0caea 100644 --- a/packages/web-shared/src/lib/hydration.ts +++ b/packages/web-shared/src/lib/hydration.ts @@ -18,6 +18,20 @@ import { } from '@workflow/core/serialization-format'; import { EVENT_DATA_REF_FIELDS } from '@workflow/world'; +const V4_EXTRA_EVENT_DATA_REF_FIELDS: Record = { + run_started: ['input'], + step_started: ['input'], +}; + +export function getEventDataRefFields(eventType: string): string[] { + return [ + ...new Set([ + ...(EVENT_DATA_REF_FIELDS[eventType] ?? []), + ...(V4_EXTRA_EVENT_DATA_REF_FIELDS[eventType] ?? []), + ]), + ]; +} + // Re-export types and utilities that consumers need export { CLASS_INSTANCE_REF_TYPE, @@ -446,7 +460,7 @@ function replaceEncryptedAndExpiredWithMarkers(resource: T): T { if (result.eventData && typeof result.eventData === 'object') { const eventType = typeof result.eventType === 'string' ? result.eventType : ''; - const refKeys = EVENT_DATA_REF_FIELDS[eventType] ?? []; + const refKeys = getEventDataRefFields(eventType); const ed = { ...(result.eventData as Record) }; for (const key of refKeys) { if (key in ed) { @@ -465,17 +479,31 @@ function replaceEncryptedAndExpiredWithMarkers(resource: T): T { * When a key is provided, encrypted fields are decrypted before hydration. * This is the async version used when the user clicks "Decrypt" in the web UI. * - * Handles both top-level fields (input, output, metadata) and nested - * eventData subfields per `EVENT_DATA_REF_FIELDS` from `@workflow/world` for that event type. + * Handles both top-level fields (input, output, metadata) and nested eventData + * payload fields for that event type. */ export async function hydrateResourceIOWithKey( resource: T, key: Uint8Array +): Promise { + return hydrateResourceIOAsync(resource, key); +} + +/** + * Async hydration for web display. + * + * This follows the same resource-field mapping as {@link hydrateResourceIO}, + * but can also inflate compressed browser payloads through the registered + * zstd WASM decoder. When a key is provided, encrypted fields are decrypted + * first and then inflated/hydrated. + */ +export async function hydrateResourceIOAsync( + resource: T, + key?: Uint8Array ): Promise { const { hydrateDataWithKey } = await import( '@workflow/core/serialization-format' ); - const { importKey } = await import('@workflow/core/encryption'); // Payloads may be zstd-compressed (the Web DecompressionStream has no zstd); // register the WASM-backed browser decoder before hydrating. Idempotent and // lazy — the WASM is only compiled when a zstd payload is actually decoded. @@ -483,25 +511,24 @@ export async function hydrateResourceIOWithKey( './zstd-browser-decoder.js' ); ensureZstdDecoderRegistered(); - const cryptoKey = await importKey(key); + const cryptoKey = key + ? await import('@workflow/core/encryption').then(({ importKey }) => + importKey(key) + ) + : undefined; const revivers = getRevivers(); - /** Extract original encrypted bytes from a marker or raw Uint8Array, then decrypt + hydrate */ - async function decryptField( - value: unknown, - rev: Revivers, - k: Awaited> - ): Promise { + async function hydrateField(value: unknown): Promise { // Already-hydrated: encrypted marker with stored bytes if (isEncryptedMarker(value)) { const raw = (value as any).__encryptedData as Uint8Array; - return hydrateDataWithKey(raw, rev, k); + return cryptoKey ? hydrateDataWithKey(raw, revivers, cryptoKey) : value; } - // Raw encrypted Uint8Array (not yet hydrated) + // Raw Uint8Array: may be encrypted, compressed, or plain devalue. if (value instanceof Uint8Array) { - return hydrateDataWithKey(value, rev, k); + return hydrateDataWithKey(value, revivers, cryptoKey); } - // Not encrypted — return as-is + // Not serialized — return as-is. return value; } @@ -511,29 +538,25 @@ export async function hydrateResourceIOWithKey( // Decrypt + hydrate top-level serialized fields (runs, steps, hooks) for (const field of ['input', 'output', 'metadata', 'error']) { if (field in result) { - result[field] = await decryptField(result[field], revivers, cryptoKey); + result[field] = await hydrateField(result[field]); } } - // Decrypt + hydrate eventData subfields (events) + // Hydrate eventData subfields (events) if (result.eventData && typeof result.eventData === 'object') { const eventType = typeof result.eventType === 'string' ? result.eventType : ''; - const refKeys = EVENT_DATA_REF_FIELDS[eventType] ?? []; + const refKeys = getEventDataRefFields(eventType); const eventData = { ...(result.eventData as Record) }; for (const field of refKeys) { if (field in eventData) { - eventData[field] = await decryptField( - eventData[field], - revivers, - cryptoKey - ); + eventData[field] = await hydrateField(eventData[field]); } } result.eventData = eventData; } - return result as T; + return replaceEncryptedAndExpiredWithMarkers(result as T); } /** @@ -552,7 +575,7 @@ export function hasEncryptedFields(resource: unknown): boolean { if (r.eventData && typeof r.eventData === 'object') { const eventType = typeof r.eventType === 'string' ? r.eventType : ''; - const refKeys = EVENT_DATA_REF_FIELDS[eventType] ?? []; + const refKeys = getEventDataRefFields(eventType); const ed = r.eventData as Record; for (const key of refKeys) { if (key in ed && isEncryptedMarker(ed[key])) return true; diff --git a/packages/web-shared/test/hydration.test.ts b/packages/web-shared/test/hydration.test.ts index 576c170622..87e0d72534 100644 --- a/packages/web-shared/test/hydration.test.ts +++ b/packages/web-shared/test/hydration.test.ts @@ -1,8 +1,19 @@ -import { dehydrateStepError } from '@workflow/core/serialization'; +import { importKey } from '@workflow/core/encryption'; +import { + dehydrateStepError, + dehydrateStepReturnValue, +} from '@workflow/core/serialization'; import { hydrateData } from '@workflow/core/serialization-format'; import { FatalError, RetryableError } from '@workflow/errors'; import { describe, expect, it } from 'vitest'; -import { getWebRevivers } from '../src/lib/hydration.js'; +import { + getWebRevivers, + hasEncryptedFields, + hydrateResourceIO, + hydrateResourceIOAsync, + hydrateResourceIOWithKey, + isEncryptedMarker, +} from '../src/lib/hydration.js'; /** * The web reviver set must mirror every key in `SerializableSpecial` (see @@ -20,6 +31,7 @@ import { getWebRevivers } from '../src/lib/hydration.js'; */ const REVIVERS = getWebRevivers(); +const textDecoder = new TextDecoder(); /** Run a real value through the production wire path with no encryption. */ async function roundTrip(value: unknown): Promise { @@ -141,3 +153,98 @@ describe('getWebRevivers — error family', () => { expect(revived.retryAfter).toBeUndefined(); }); }); + +describe('front hydration — encrypted compressed payloads', () => { + const runId = 'wrun_test'; + const rawKey = new Uint8Array(32).fill(7); + + function formatPrefix(value: unknown): string { + expect(value).toBeInstanceOf(Uint8Array); + return textDecoder.decode((value as Uint8Array).subarray(0, 4)); + } + + it('keeps encrypted compressed step errors as markers until decrypting them with the run key', async () => { + const cryptoKey = await importKey(rawKey); + const original = new Error( + `boom ${'front encrypted payload '.repeat(400)}` + ); + const wire = await dehydrateStepError( + original, + runId, + cryptoKey, + [], + globalThis, + true + ); + + expect(formatPrefix(wire)).toBe('encr'); + + const hydrated = hydrateResourceIO({ + stepId: 'step_test', + error: wire, + }); + + expect(isEncryptedMarker(hydrated.error)).toBe(true); + expect(hasEncryptedFields(hydrated)).toBe(true); + + const decrypted = await hydrateResourceIOWithKey(hydrated, rawKey); + expect(decrypted.error).toBeInstanceOf(Error); + expect((decrypted.error as Error).message).toBe(original.message); + }); + + it('hydrates unencrypted compressed step errors through the async web path', async () => { + const original = new Error( + `boom ${'oss web compressed payload '.repeat(400)}` + ); + const wire = await dehydrateStepError( + original, + runId, + undefined, + [], + globalThis, + true + ); + + expect(['gzip', 'zstd']).toContain(formatPrefix(wire)); + + const hydrated = await hydrateResourceIOAsync({ + stepId: 'step_test', + error: wire, + }); + + expect(hydrated.error).toBeInstanceOf(Error); + expect((hydrated.error as Error).message).toBe(original.message); + }); + + it('decrypts encrypted compressed v4 step_started input payloads', async () => { + const cryptoKey = await importKey(rawKey); + const input = ['probe', { message: 'encrypted front payload' }]; + const wire = await dehydrateStepReturnValue( + input, + runId, + cryptoKey, + [], + globalThis, + false, + false, + true + ); + + expect(formatPrefix(wire)).toBe('encr'); + + const hydrated = hydrateResourceIO({ + eventId: 'evnt_test', + eventType: 'step_started', + eventData: { + stepName: 'probe', + input: wire, + }, + }); + + expect(isEncryptedMarker(hydrated.eventData.input)).toBe(true); + expect(hasEncryptedFields(hydrated)).toBe(true); + + const decrypted = await hydrateResourceIOWithKey(hydrated, rawKey); + expect(decrypted.eventData.input).toEqual(input); + }); +}); diff --git a/packages/web/app/components/run-detail-view.tsx b/packages/web/app/components/run-detail-view.tsx index f8fd77c442..58090ef571 100644 --- a/packages/web/app/components/run-detail-view.tsx +++ b/packages/web/app/components/run-detail-view.tsx @@ -4,8 +4,7 @@ import { DecryptButton, ErrorBoundary, EventListView, - hydrateResourceIO, - hydrateResourceIOWithKey, + hydrateResourceIOAsync, NewTraceViewer, type SidebarDataContextValue, StreamViewer, @@ -301,9 +300,10 @@ export function RunDetailView({ if (error) { throw error; } - const fullEvent = encryptionKeyRef.current - ? await hydrateResourceIOWithKey(result, encryptionKeyRef.current) - : hydrateResourceIO(result); + const fullEvent = await hydrateResourceIOAsync( + result, + encryptionKeyRef.current ?? undefined + ); if ('eventData' in fullEvent) { return fullEvent.eventData; } @@ -321,9 +321,10 @@ export function RunDetailView({ if (error) { throw error; } - const fullEvent = encryptionKeyRef.current - ? await hydrateResourceIOWithKey(result, encryptionKeyRef.current) - : hydrateResourceIO(result); + const fullEvent = await hydrateResourceIOAsync( + result, + encryptionKeyRef.current ?? undefined + ); if ('eventData' in fullEvent) { return fullEvent.eventData; } diff --git a/packages/web/app/lib/client/hooks/use-events-list-data.ts b/packages/web/app/lib/client/hooks/use-events-list-data.ts index 8daf975c15..c6c327fc81 100644 --- a/packages/web/app/lib/client/hooks/use-events-list-data.ts +++ b/packages/web/app/lib/client/hooks/use-events-list-data.ts @@ -1,15 +1,12 @@ 'use client'; -import type { Event } from '@workflow/world'; import type { ExactIdSearchResult, ExactWorkflowSearchIdKind, } from '@workflow/web-shared'; +import { hydrateResourceIOAsync } from '@workflow/web-shared'; +import type { Event } from '@workflow/world'; import { useCallback, useEffect, useRef, useState } from 'react'; -import { - hydrateResourceIO, - hydrateResourceIOWithKey, -} from '@workflow/web-shared'; import { unwrapServerActionResult } from '~/lib/client/workflow-errors'; import { fetchEvent, @@ -52,14 +49,8 @@ export function useEventsListData( encryptionKeyRef.current = encryptionKey; const hydrateEvents = useCallback(async (rawEvents: Event[]) => { - const hydrated = rawEvents.map(hydrateResourceIO); const key = encryptionKeyRef.current; - if (key) { - return Promise.all( - hydrated.map((ev) => hydrateResourceIOWithKey(ev, key)) - ); - } - return hydrated; + return Promise.all(rawEvents.map((ev) => hydrateResourceIOAsync(ev, key))); }, []); const fetchInitial = useCallback(async () => { @@ -102,7 +93,7 @@ export function useEventsListData( useEffect(() => { if (!encryptionKey || events.length === 0) return; let cancelled = false; - Promise.all(events.map((ev) => hydrateResourceIOWithKey(ev, encryptionKey))) + Promise.all(events.map((ev) => hydrateResourceIOAsync(ev, encryptionKey))) .then((decrypted) => { if (!cancelled) setEvents(decrypted); }) diff --git a/packages/web/app/lib/client/hooks/use-resource-data.test.ts b/packages/web/app/lib/client/hooks/use-resource-data.test.ts index 8f7d9d1fd5..4ab2a6ccaf 100644 --- a/packages/web/app/lib/client/hooks/use-resource-data.test.ts +++ b/packages/web/app/lib/client/hooks/use-resource-data.test.ts @@ -1,9 +1,12 @@ import { renderHook, waitFor } from '@testing-library/react'; import { beforeEach, describe, expect, it, vi } from 'vitest'; -import { useWorkflowResourceData } from './use-resource-data'; +import { + fetchSpanDetailResource, + useWorkflowResourceData, +} from './use-resource-data'; vi.mock('@workflow/web-shared', () => ({ - hydrateResourceIO: (x: T): T => x, + hydrateResourceIOAsync: vi.fn(async (x: T): Promise => x), waitEventsToWaitEntity: vi.fn(), })); @@ -14,9 +17,12 @@ vi.mock('~/lib/rpc-client', () => ({ fetchEvents: vi.fn(), })); -import { waitEventsToWaitEntity } from '@workflow/web-shared'; -import type { WorkflowRun } from '@workflow/world'; -import { fetchEvents, fetchHook, fetchRun } from '~/lib/rpc-client'; +import { + hydrateResourceIOAsync, + waitEventsToWaitEntity, +} from '@workflow/web-shared'; +import type { Step, WorkflowRun } from '@workflow/world'; +import { fetchEvents, fetchHook, fetchRun, fetchStep } from '~/lib/rpc-client'; const env = { SOME_VAR: 'test' }; @@ -79,7 +85,7 @@ describe('useWorkflowResourceData', () => { expect(result.current.error).not.toBeNull(); }); - expect(result.current.error!.message).toBe('not found'); + expect(result.current.error?.message).toBe('not found'); }); it('shows hook data after loading', async () => { @@ -110,6 +116,41 @@ describe('useWorkflowResourceData', () => { expect(result.current.error).toBeNull(); }); + it('hydrates step data through the async path without an encryption key', async () => { + const rawStep = { + runId: 'run-1', + stepId: 'step-1', + stepName: 'step-1', + status: 'failed', + input: new Uint8Array([1, 2, 3]), + error: new Uint8Array([122, 115, 116, 100]), + attempt: 1, + createdAt: new Date(), + updatedAt: new Date(), + specVersion: 5, + } as Step; + const hydratedStep = { + ...rawStep, + error: new Error('Fatal step error'), + } as Step; + + vi.mocked(fetchStep).mockResolvedValue({ + success: true, + data: rawStep, + }); + vi.mocked(hydrateResourceIOAsync).mockResolvedValueOnce(hydratedStep); + + await expect( + fetchSpanDetailResource(env, { + resource: 'step', + resourceId: 'step-1', + runId: 'run-1', + }) + ).resolves.toBe(hydratedStep); + + expect(hydrateResourceIOAsync).toHaveBeenCalledWith(rawStep, undefined); + }); + it('shows sleep entity constructed from events', async () => { const events = [ { @@ -158,7 +199,7 @@ describe('useWorkflowResourceData', () => { expect(result.current.error).not.toBeNull(); }); - expect(result.current.error!.message).toContain( + expect(result.current.error?.message).toContain( 'missing required event data' ); }); diff --git a/packages/web/app/lib/client/hooks/use-resource-data.ts b/packages/web/app/lib/client/hooks/use-resource-data.ts index cf74566bd2..8e59cc789b 100644 --- a/packages/web/app/lib/client/hooks/use-resource-data.ts +++ b/packages/web/app/lib/client/hooks/use-resource-data.ts @@ -1,6 +1,5 @@ import { - hydrateResourceIO, - hydrateResourceIOWithKey, + hydrateResourceIOAsync, waitEventsToWaitEntity, } from '@workflow/web-shared'; import type { Event, Hook, Step, WorkflowRun } from '@workflow/world'; @@ -67,9 +66,7 @@ export async function fetchSpanDetailResource( const { resource, resourceId, runId } = selection; const { encryptionKey } = options; const hydrate = async (value: T): Promise => - encryptionKey - ? hydrateResourceIOWithKey(value, encryptionKey) - : hydrateResourceIO(value); + hydrateResourceIOAsync(value, encryptionKey); if (resource === 'hook') { const result = await unwrapOrThrow(fetchHook(env, resourceId, 'all')); @@ -86,11 +83,14 @@ export async function fetchSpanDetailResource( const result = await unwrapOrThrow( fetchEvents(env, runId, { sortOrder: 'asc', limit: 1000, withData: true }) ); - const allEvents = (result.data as unknown as Event[]).map( - hydrateResourceIO - ); + // `correlationId` is a top-level event field, untouched by hydration, so + // filter first and only hydrate the wait's own events. Hydrating the + // whole page would needlessly decrypt every event in the run when a key + // is present. const waitEvents = await Promise.all( - allEvents.filter((e) => e.correlationId === resourceId).map(hydrate) + (result.data as unknown as Event[]) + .filter((e) => e.correlationId === resourceId) + .map(hydrate) ); const data = waitEventsToWaitEntity(waitEvents); if (data === null) { diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 2e02222388..41dcd0d1c3 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -57,6 +57,10 @@ import { getWorkflowRunEventsV4, } from './events-v4.js'; import { cancelWorkflowRunV1, createWorkflowRunV1 } from './runs.js'; +import { + normalizeEventData, + normalizeSerializedData, +} from './serialized-data.js'; import { deserializeStep } from './steps.js'; import { type APIConfig, @@ -402,6 +406,10 @@ function coerceEventDates(raw: Record): Event { return raw as unknown as Event; } +function coerceNormalizedEvent(raw: Record): Event { + return coerceEventDates(normalizeEventData(raw)); +} + /** * Turn a v4 event (frame meta + frame body) into the Event shape the * workflow runtime expects. @@ -409,10 +417,11 @@ function coerceEventDates(raw: Record): Event { * 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]` unchanged — the runtime's - * hydrate helpers (hydrateStepIO, hydrateRunError, …) consume the raw - * devalue-with-format-prefix Uint8Array directly. No CBOR decode here, - * symmetric with the pass-through write in `splitEventDataForV4`. + * 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. No CBOR + * decode here, symmetric with the pass-through write in + * `splitEventDataForV4`. */ function buildEventFromV4( decoded: DecodedV4Event, @@ -423,7 +432,10 @@ function buildEventFromV4( if (payloadBody.byteLength > 0) { const payloadField = payloadFieldFor(decoded.eventType); - if (payloadField) eventData[payloadField] = payloadBody; + const normalizedPayload = normalizeSerializedData(payloadBody); + if (payloadField && normalizedPayload instanceof Uint8Array) { + eventData[payloadField] = normalizedPayload; + } } const raw = { @@ -449,7 +461,7 @@ function buildEventFromV4( : {}), }; - const event = coerceEventDates(raw); + const event = coerceNormalizedEvent(raw); // For resolveData='none', strip eventData entirely. Reuse the world- // side helper so behavior stays in sync with other backends. @@ -639,18 +651,23 @@ async function createWorkflowRunEventInner( ); // The server already CBOR-decoded into result.body — just thread the - // fields through. Step has a wire-format adapter; runs use the - // pass-through deserializeError helper (run/step dates arrive as real - // Dates — the server's entity getters convert before CBOR-encoding). - // The returned `event` and preloaded `events` go through - // coerceEventDates: they can be read back from the backing store - // server-side (e.g. the run_started TTFB preload queries the event - // log), where nested eventData dates are ISO strings — same coercion - // the GET/LIST path applies, and the v3 path applied via its zod wire - // schemas. - // The returned event honors the caller's resolveData: 'none' strips - // payload fields, matching the v3 path's stripEventAndLegacyRefs - // behavior and the Storage contract. + // fields through. This is the runtime's event-append path (world.events + // .create is only ever called from the workflow runtime, never from + // o11y), and the runtime re-hydrates every payload it consumes through + // the decompress-aware helpers (hydrateStepReturnValue, hydrateRunError, + // …). So we deliberately do NOT decompress here: doing so would be + // redundant work on the TTFB-sensitive run_started/inline-delta path and + // would make the runtime's deserialize compression telemetry report + // `codec: none` for payloads that were compressed at rest. gzip/zstd + // normalization for o11y/display lives on the read paths (getEvent, + // getWorkflowRunEvents, getStep, getRun, getHook). + // + // `event`/`events` go through coerceEventDates only: they can be read + // back from the backing store server-side (e.g. the run_started TTFB + // preload queries the event log), where nested eventData dates are ISO + // strings — same coercion the GET/LIST path applies. The returned event + // honors the caller's resolveData: 'none' strips payload fields, + // matching the v3 path's stripEventAndLegacyRefs behavior. const resolveData = params?.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION; const body = result.body; return { diff --git a/packages/world-vercel/src/hooks.ts b/packages/world-vercel/src/hooks.ts index 82ab7043b6..e3add16a2d 100644 --- a/packages/world-vercel/src/hooks.ts +++ b/packages/world-vercel/src/hooks.ts @@ -8,6 +8,7 @@ import type { } from '@workflow/world'; import { HookSchema, PaginatedResponseSchema } from '@workflow/world'; import z from 'zod'; +import { normalizeHookData } from './serialized-data.js'; import type { APIConfig } from './utils.js'; import { DEFAULT_RESOLVE_DATA_OPTION, makeRequest } from './utils.js'; @@ -17,7 +18,7 @@ function filterHookData(hook: any, resolveData: 'none' | 'all'): Hook { const { metadataRef: _metadataRef, ...rest } = hook; return rest; } - return hook; + return normalizeHookData(hook) as Hook; } const HookWithRefsSchema = HookSchema.omit({ metadata: true, diff --git a/packages/world-vercel/src/runs.ts b/packages/world-vercel/src/runs.ts index b2390a148f..5e8273eb96 100644 --- a/packages/world-vercel/src/runs.ts +++ b/packages/world-vercel/src/runs.ts @@ -14,6 +14,7 @@ import { type WorkflowRunWithoutData, } from '@workflow/world'; import { z } from 'zod'; +import { normalizeWorkflowRunData } from './serialized-data.js'; import type { APIConfig } from './utils.js'; import { DEFAULT_RESOLVE_DATA_OPTION, @@ -68,21 +69,29 @@ function filterRunData( resolveData: 'none' | 'all' ): WorkflowRun | WorkflowRunWithoutData; -// Implementation +// Implementation. This is a read/display entry point (getRun/listRuns), +// so it decompresses gzip/zstd payload wrappers via +// `normalizeWorkflowRunData`. The runtime write path (events.create) +// re-hydrates run errors through `hydrateRunError`, which decompresses +// on its own, so it deliberately does not route through here. function filterRunData( run: any, resolveData: 'none' | 'all' ): WorkflowRun | WorkflowRunWithoutData { if (resolveData === 'none') { const { inputRef: _inputRef, outputRef: _outputRef, ...rest } = run; - const deserialized = deserializeError(rest); + const deserialized = normalizeWorkflowRunData( + deserializeError(rest) as unknown as Record + ); return { ...deserialized, input: undefined, output: undefined, } as WorkflowRunWithoutData; } - return deserializeError(run); + return normalizeWorkflowRunData( + deserializeError(run) as unknown as Record + ) as unknown as WorkflowRun; } // Functions diff --git a/packages/world-vercel/src/serialized-data.ts b/packages/world-vercel/src/serialized-data.ts new file mode 100644 index 0000000000..5fc4087477 --- /dev/null +++ b/packages/world-vercel/src/serialized-data.ts @@ -0,0 +1,131 @@ +import { WorkflowWorldError } from '@workflow/errors'; +import { EVENT_DATA_REF_FIELDS } from '@workflow/world'; + +const FORMAT_PREFIX_LENGTH = 4; +const GZIP_FORMAT_PREFIX = 'gzip'; +const ZSTD_FORMAT_PREFIX = 'zstd'; +const formatDecoder = new TextDecoder(); +const V4_EXTRA_EVENT_DATA_REF_FIELDS: Record = { + run_started: ['input'], + step_started: ['input'], +}; + +interface NodeZlibDecode { + gunzipSync?: (data: Uint8Array) => Uint8Array; + zstdDecompressSync?: (data: Uint8Array) => Uint8Array; +} + +function getNodeZlib(): NodeZlibDecode | undefined { + try { + return ( + globalThis as { + process?: { getBuiltinModule?: (id: string) => NodeZlibDecode }; + } + ).process?.getBuiltinModule?.('node:zlib'); + } catch { + return undefined; + } +} + +function peekFormatPrefix(value: unknown): string | null { + if ( + !(value instanceof Uint8Array) || + value.byteLength < FORMAT_PREFIX_LENGTH + ) { + return null; + } + return formatDecoder.decode(value.subarray(0, FORMAT_PREFIX_LENGTH)); +} + +function decompress(format: string, payload: Uint8Array): Uint8Array { + const zlib = getNodeZlib(); + const decompress = + format === ZSTD_FORMAT_PREFIX ? zlib?.zstdDecompressSync : zlib?.gunzipSync; + + if (!decompress) { + throw new WorkflowWorldError( + `Received ${format}-compressed workflow data, but this Node.js runtime does not support ${format} decompression. Use a compatible Node.js runtime or request unresolved data.` + ); + } + + return new Uint8Array(decompress(payload)); +} + +export function normalizeSerializedData(value: unknown): unknown { + const format = peekFormatPrefix(value); + if (format !== ZSTD_FORMAT_PREFIX && format !== GZIP_FORMAT_PREFIX) { + return value; + } + const bytes = value as Uint8Array; + return decompress(format, bytes.subarray(FORMAT_PREFIX_LENGTH)); +} + +export function normalizeWorkflowRunData>( + run: T +): T { + return { + ...run, + input: normalizeSerializedData(run.input), + output: normalizeSerializedData(run.output), + error: normalizeSerializedData(run.error), + }; +} + +export function normalizeStepData>( + step: T +): T { + // Only the resolved payload fields can carry a compression wrapper. + // `*Ref` fields are RefDescriptor objects (lazy mode), never byte + // payloads, so they need no normalization. + return { + ...step, + input: normalizeSerializedData(step.input), + output: normalizeSerializedData(step.output), + error: normalizeSerializedData(step.error), + }; +} + +export function normalizeHookData>( + hook: T +): T { + return { + ...hook, + metadata: normalizeSerializedData(hook.metadata), + }; +} + +export function normalizeEventData>( + event: T +): T { + const eventData = event.eventData; + if (!eventData || typeof eventData !== 'object') { + return event; + } + + const eventType = typeof event.eventType === 'string' ? event.eventType : ''; + const refFields = [ + ...new Set([ + ...(EVENT_DATA_REF_FIELDS[eventType] ?? []), + ...(V4_EXTRA_EVENT_DATA_REF_FIELDS[eventType] ?? []), + ]), + ]; + if (refFields.length === 0) { + return event; + } + + const normalizedEventData = { ...(eventData as Record) }; + let changed = false; + for (const field of refFields) { + if (!(field in normalizedEventData)) { + continue; + } + const before = normalizedEventData[field]; + const after = normalizeSerializedData(before); + if (after !== before) { + normalizedEventData[field] = after; + changed = true; + } + } + + return changed ? { ...event, eventData: normalizedEventData } : event; +} diff --git a/packages/world-vercel/src/steps.test.ts b/packages/world-vercel/src/steps.test.ts new file mode 100644 index 0000000000..1ed8ad4a4a --- /dev/null +++ b/packages/world-vercel/src/steps.test.ts @@ -0,0 +1,140 @@ +import * as zlib from 'node:zlib'; +import { encode } from 'cbor-x'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { deserializeStep, getStep } from './steps.js'; + +vi.mock('@vercel/oidc', () => ({ + getVercelOidcToken: vi.fn().mockRejectedValue(new Error('no OIDC')), +})); + +type ZlibWithZstd = typeof zlib & { + zstdCompressSync?: (buf: NodeJS.ArrayBufferView) => Buffer; +}; + +const zstdCompressSync = (zlib as ZlibWithZstd).zstdCompressSync; +const zstdIt = zstdCompressSync ? it : it.skip; + +function cborResponse(data: unknown): Response { + const bytes = encode(data); + return new Response(new Uint8Array(bytes), { + headers: { 'Content-Type': 'application/cbor' }, + }); +} + +function zstdWrapped(bytes: Uint8Array): Uint8Array { + if (!zstdCompressSync) { + throw new Error('zstdCompressSync unavailable'); + } + const compressed = zstdCompressSync(bytes); + const result = new Uint8Array(4 + compressed.byteLength); + result.set(new TextEncoder().encode('zstd'), 0); + result.set(compressed, 4); + return result; +} + +function gzipWrapped(bytes: Uint8Array): Uint8Array { + const compressed = zlib.gzipSync(bytes); + const result = new Uint8Array(4 + compressed.byteLength); + result.set(new TextEncoder().encode('gzip'), 0); + result.set(compressed, 4); + return result; +} + +describe('getStep', () => { + const originalEnv = process.env; + + beforeEach(() => { + process.env = { ...originalEnv }; + process.env.VERCEL_WORKFLOW_SERVER_URL = 'https://workflow.test'; + }); + + afterEach(() => { + process.env = originalEnv; + vi.unstubAllGlobals(); + }); + + zstdIt('decompresses zstd-prefixed serialized step errors', async () => { + const serializedError = new TextEncoder().encode( + 'devl[{"name":1,"message":2}, "Error", "boom"]' + ); + const fetchMock = vi.fn().mockResolvedValue( + cborResponse({ + runId: 'wrun_test', + stepId: 'step_test', + stepName: 'step//./workflows/test//explode', + status: 'failed', + error: zstdWrapped(serializedError), + attempt: 1, + createdAt: '2026-06-26T00:00:00.000Z', + updatedAt: '2026-06-26T00:00:01.000Z', + completedAt: '2026-06-26T00:00:01.000Z', + specVersion: 5, + }) + ); + vi.stubGlobal('fetch', fetchMock); + + const step = await getStep('wrun_test', 'step_test', { + resolveData: 'all', + }); + + expect(step.error).toEqual(serializedError); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('decompresses gzip-prefixed serialized step errors', async () => { + const serializedError = new TextEncoder().encode( + 'devl[{"name":1,"message":2}, "Error", "boom"]' + ); + const fetchMock = vi.fn().mockResolvedValue( + cborResponse({ + runId: 'wrun_test', + stepId: 'step_test', + stepName: 'step//./workflows/test//explode', + status: 'failed', + error: gzipWrapped(serializedError), + attempt: 1, + createdAt: '2026-06-26T00:00:00.000Z', + updatedAt: '2026-06-26T00:00:01.000Z', + completedAt: '2026-06-26T00:00:01.000Z', + specVersion: 5, + }) + ); + vi.stubGlobal('fetch', fetchMock); + + const step = await getStep('wrun_test', 'step_test', { + resolveData: 'all', + }); + + expect(step.error).toEqual(serializedError); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); +}); + +describe('deserializeStep (runtime write/append path)', () => { + // The runtime consumes events.create/createStep/updateStep results and + // re-hydrates payloads through the decompress-aware hydrate helpers, so the + // wire→shape adapter must NOT decompress. Decompressing here would skew the + // runtime's deserialize compression telemetry to `codec: none`. Compression + // normalization is the read path's (filterStepData) job — covered by the + // getStep tests above. + it('passes a compressed step error through unchanged (no decompression)', () => { + const serializedError = new TextEncoder().encode( + 'devl[{"name":1,"message":2}, "Error", "boom"]' + ); + const wrapped = gzipWrapped(serializedError); + + const step = deserializeStep({ + runId: 'wrun_test', + stepId: 'step_test', + stepName: 'step//./workflows/test//explode', + status: 'failed', + error: wrapped, + attempt: 1, + specVersion: 5, + }); + + // Still the compressed wire bytes — the runtime's hydrateStepError + // inflates them, this layer does not. + expect(step.error).toBe(wrapped); + }); +}); diff --git a/packages/world-vercel/src/steps.ts b/packages/world-vercel/src/steps.ts index 5fc109ab96..c347628010 100644 --- a/packages/world-vercel/src/steps.ts +++ b/packages/world-vercel/src/steps.ts @@ -11,6 +11,7 @@ import { type UpdateStepRequest, } from '@workflow/world'; import { z } from 'zod'; +import { normalizeStepData } from './serialized-data.js'; import type { APIConfig } from './utils.js'; import { DEFAULT_RESOLVE_DATA_OPTION, @@ -50,6 +51,13 @@ const StepWireWithRefsSchema = StepWireSchema.omit({ * The `error` field on Step is SerializedData (Uint8Array) from the * serialization pipeline — we pass through the wire-format `error` (or * the resolved `errorRef`) as-is. Consumers hydrate via `hydrateStepError`. + * + * Wire→shape only: this does NOT decompress. The runtime write paths + * (createStep/updateStep/events.create) re-hydrate step payloads through + * `hydrateStepReturnValue`/`hydrateStepError`, which decompress on their + * own, so decompressing here would be redundant work and would skew the + * runtime's deserialize compression telemetry. Compression normalization + * for o11y/display is applied in {@link filterStepData}, the read path. */ export function deserializeStep(wireStep: any): Step { const { error, errorRef, ...rest } = wireStep; @@ -70,21 +78,25 @@ function filterStepData( ): Step | StepWithoutData; // Implementation - when resolveData='none', returns Step with input/output set to undefined -// to match other World implementations (world-local, world-postgres) +// to match other World implementations (world-local, world-postgres). +// +// This is the read/display entry point, so it decompresses gzip/zstd +// payload wrappers via `normalizeStepData` (the runtime write paths use +// `deserializeStep` directly and skip this — see its doc comment). function filterStepData( step: any, resolveData: 'none' | 'all' ): Step | StepWithoutData { if (resolveData === 'none') { const { inputRef: _inputRef, outputRef: _outputRef, ...rest } = step; - const deserialized = deserializeStep(rest); + const deserialized = normalizeStepData(deserializeStep(rest)); return { ...deserialized, input: undefined, output: undefined, } as StepWithoutData; } - return deserializeStep(step); + return normalizeStepData(deserializeStep(step)); } // Functions