Centralize workflow event semantics

.changeset/centralize-event-semantics.md

@@ -0,0 +1,12 @@
+---
+'@workflow/world': minor
+'@workflow/core': patch
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+'@workflow/world-vercel': patch
+'@workflow/web-shared': patch
+'@workflow/web': patch
+'@workflow/cli': patch
+---
+
+Centralize workflow event type classifiers and event-data payload field helpers.

packages/cli/src/commands/cancel.ts

@@ -2,7 +2,7 @@ import readline from 'node:readline';
 import { Args, Flags } from '@oclif/core';
 import { cancelRun } from '@workflow/core/runtime';
 import { parseWorkflowName } from '@workflow/utils/parse-name';
-import type { WorkflowRun } from '@workflow/world';
+import { WorkflowRunStatusSchema } from '@workflow/world';
 import chalk from 'chalk';
 import Table from 'easy-table';
 import { BaseCommand } from '../base.js';
@@ -50,7 +50,7 @@ export default class Cancel extends BaseCommand {
     status: Flags.string({
       description: 'Filter runs by status for bulk cancel',
       required: false,
-      options: ['running', 'completed', 'failed', 'cancelled', 'pending'],
+      options: [...WorkflowRunStatusSchema.options],
       helpGroup: 'Bulk Cancel',
       helpLabel: '--status',
     }),
@@ -110,7 +110,9 @@ export default class Cancel extends BaseCommand {
 
     // Fetch matching runs. Only metadata is needed to display and cancel, so
     // prefer the analytics read path when the backend provides one.
-    const status = flags.status as WorkflowRun['status'] | undefined;
+    const status = flags.status
+      ? WorkflowRunStatusSchema.parse(flags.status)
+      : undefined;
     const limit = flags.limit || 50;
     const analytics = world.analytics;
     const fetchMatches = async () => {

packages/cli/src/commands/inspect.ts

@@ -1,5 +1,6 @@
 import { Args, Flags } from '@oclif/core';
 import { VERCEL_403_ERROR_MESSAGE } from '@workflow/errors';
+import { WorkflowRunStatusSchema } from '@workflow/world';
 import { BaseCommand } from '../base.js';
 import { LOGGING_CONFIG, logger } from '../lib/config/log.js';
 import type { InspectCLIOptions } from '../lib/config/types.js';
@@ -125,7 +126,7 @@ export default class Inspect extends BaseCommand {
     status: Flags.string({
       description: 'filter runs by status (only for runs)',
       required: false,
-      options: ['running', 'completed', 'failed', 'cancelled', 'pending'],
+      options: [...WorkflowRunStatusSchema.options],
       helpGroup: 'Filtering',
       helpLabel: '--status',
     }),
@@ -297,7 +298,9 @@ function toInspectOptions(flags: any): InspectCLIOptions {
     sort: flags.sort as 'asc' | 'desc' | undefined,
     limit: flags.limit,
     workflowName: flags.workflowName,
-    status: flags.status,
+    status: flags.status
+      ? WorkflowRunStatusSchema.parse(flags.status)
+      : undefined,
     since: flags.since,
     until: flags.until,
     withData: flags.withData,

packages/cli/src/lib/config/types.ts

@@ -1,4 +1,6 @@
 // Re-export builder types for backwards compatibility
+import type { WorkflowRunStatus } from '@workflow/world';
+
 export type {
   BuildTarget,
   WorkflowConfig,
@@ -18,7 +20,7 @@ export type InspectCLIOptions = {
   sort?: 'asc' | 'desc';
   limit?: number;
   workflowName?: string;
-  status?: string;
+  status?: WorkflowRunStatus;
   /** Listing window start: relative duration (12h, 7d) or timestamp (runs only) */
   since?: string;
   /** Listing window end: relative duration or timestamp; requires --since */

packages/cli/src/lib/inspect/hydration.test.ts

@@ -0,0 +1,32 @@
+import { importKey } from '@workflow/core/encryption';
+import { dehydrateStepError } from '@workflow/core/serialization';
+import { describe, expect, it } from 'vitest';
+import { hydrateResourceIO, isEncryptedRef } from './hydration.js';
+
+describe('hydrateResourceIO', () => {
+  it('displays and decrypts encrypted event errors', async () => {
+    const rawKey = new Uint8Array(32).fill(7);
+    const error = new Error('step failed');
+    const encryptedError = await dehydrateStepError(
+      error,
+      'run_1',
+      await importKey(rawKey),
+      [],
+      globalThis,
+      true
+    );
+    const event = {
+      runId: 'run_1',
+      eventId: 'event_1',
+      eventType: 'step_failed',
+      eventData: { error: encryptedError },
+    };
+
+    const encrypted = await hydrateResourceIO(event);
+    expect(isEncryptedRef(encrypted.eventData.error)).toBe(true);
+
+    const decrypted = await hydrateResourceIO(event, async () => rawKey);
+    expect(decrypted.eventData.error).toBeInstanceOf(Error);
+    expect(decrypted.eventData.error).toHaveProperty('message', error.message);
+  });
+});

packages/cli/src/lib/inspect/hydration.ts

@@ -19,6 +19,7 @@ import {
   serializedInstanceToRef,
 } from '@workflow/core/serialization-format';
 import { parseClassName } from '@workflow/utils/parse-name';
+import { getEventDataRefFields } from '@workflow/world';
 import chalk from 'chalk';
 
 /**
@@ -324,6 +325,7 @@ async function maybeDecryptFields<
     input?: any;
     output?: any;
     metadata?: any;
+    eventType?: string;
     eventData?: any;
   },
 >(resource: T, resolver: EncryptionKeyResolver): Promise<T> {
@@ -350,13 +352,7 @@ async function maybeDecryptFields<
     // Decrypt eventData fields (Event)
     if (result.eventData && typeof result.eventData === 'object') {
       const eventData = { ...result.eventData };
-      for (const field of [
-        'result',
-        'input',
-        'output',
-        'metadata',
-        'payload',
-      ]) {
+      for (const field of getEventDataRefFields(result.eventType ?? '')) {
         eventData[field] = await maybeDecrypt(eventData[field], k);
       }
       result.eventData = eventData;
@@ -406,7 +402,9 @@ function replaceEncryptedAndExpiredWithRef<T>(resource: T): T {
 
   if (result.eventData && typeof result.eventData === 'object') {
     const ed = { ...(result.eventData as Record<string, unknown>) };
-    for (const key of ['result', 'input', 'output', 'metadata', 'payload']) {
+    const eventType =
+      typeof result.eventType === 'string' ? result.eventType : '';
+    for (const key of getEventDataRefFields(eventType)) {
       ed[key] = toDisplayRef(ed[key]);
     }
     result.eventData = ed;

packages/cli/src/lib/inspect/output.ts

@@ -6,7 +6,14 @@ import {
 } from '@workflow/core/serialization';
 import { VERCEL_403_ERROR_MESSAGE } from '@workflow/errors';
 import { parseStepName, parseWorkflowName } from '@workflow/utils/parse-name';
-import type { Event, Hook, Step, WorkflowRun, World } from '@workflow/world';
+import {
+  type Event,
+  type Hook,
+  isWaitEventType,
+  type Step,
+  type WorkflowRun,
+  type World,
+} from '@workflow/world';
 import chalk from 'chalk';
 
 import { formatDistance } from 'date-fns';
@@ -619,7 +626,7 @@ export const listRuns = async (world: World, opts: InspectCLIOptions = {}) => {
 
   const useAnalytics = !opts.withData && Boolean(world.analytics);
   const resolveData = opts.withData ? 'all' : 'none';
-  const status = opts.status as WorkflowRun['status'] | undefined;
+  const status = opts.status;
 
   // Resolve --since/--until into an explicit listing window. Only the
   // analytics read path supports one; the runtime storage APIs have no time
@@ -1371,7 +1378,7 @@ export const listSleeps = async (
     // Group wait events by correlationId
     const waitEventsByCorrelation = new Map<string, Event[]>();
     for (const event of events.data) {
-      if (event.correlationId?.startsWith('wait_')) {
+      if (isWaitEventType(event.eventType) && event.correlationId) {
         const existing = waitEventsByCorrelation.get(event.correlationId) ?? [];
         existing.push(event);
         waitEventsByCorrelation.set(event.correlationId, existing);

packages/core/src/serialization-format.test.ts

@@ -204,42 +204,54 @@ describe('hydrateResourceIO', () => {
 
     const event = {
       eventId: 'evt_123',
+      eventType: 'step_completed',
       eventData: {
-        type: 'step_completed',
         result: resultPayload,
       },
     };
 
     const hydrated = hydrateResourceIO(event, testRevivers);
     expect(hydrated.eventId).toBe('evt_123');
     expect(hydrated.eventData.result).toEqual({ key: 'value' });
-    expect(hydrated.eventData.type).toBe('step_completed');
   });
 
   it('should hydrate event eventData.output', () => {
     const outputPayload = makeDevlPayload({ message: 'done' });
 
     const event = {
       eventId: 'evt_456',
+      eventType: 'run_completed',
       eventData: {
-        type: 'run_completed',
         output: outputPayload,
       },
     };
 
     const hydrated = hydrateResourceIO(event, testRevivers);
     expect(hydrated.eventId).toBe('evt_456');
     expect(hydrated.eventData.output).toEqual({ message: 'done' });
-    expect(hydrated.eventData.type).toBe('run_completed');
+  });
+
+  it.each([
+    'run_started',
+    'step_started',
+  ] as const)('should hydrate eventData.input for %s events', (eventType) => {
+    const event = {
+      eventId: `evt_${eventType}`,
+      eventType,
+      eventData: { input: makeDevlPayload({ value: eventType }) },
+    };
+
+    const hydrated = hydrateResourceIO(event, testRevivers);
+    expect(hydrated.eventData.input).toEqual({ value: eventType });
   });
 
   it('should hydrate event eventData.metadata for hook_created events', () => {
     const metadataPayload = makeDevlPayload({ source: 'webhook', retries: 2 });
 
     const event = {
       eventId: 'evt_hook_created',
+      eventType: 'hook_created',
       eventData: {
-        type: 'hook_created',
         token: 'hook_tok_123',
         metadata: metadataPayload,
       },
@@ -259,8 +271,8 @@ describe('hydrateResourceIO', () => {
 
     const event = {
       eventId: 'evt_hook_received',
+      eventType: 'hook_received',
       eventData: {
-        type: 'hook_received',
         payload,
       },
     };
@@ -278,8 +290,8 @@ describe('hydrateResourceIO', () => {
 
     const event = {
       eventId: 'evt_step_failed',
+      eventType: 'step_failed',
       eventData: {
-        type: 'step_failed',
         error: errorPayload,
       },
     };
@@ -290,7 +302,6 @@ describe('hydrateResourceIO', () => {
       message: 'something blew up',
       stack: 'Error: something blew up\n    at foo:1:1',
     });
-    expect(hydrated.eventData.type).toBe('step_failed');
   });
 
   it('should hydrate hook metadata', () => {

packages/core/src/serialization-format.ts

@@ -6,6 +6,7 @@
  * o11y, CLI o11y). It has NO Node.js dependencies.
  */
 
+import { getEventDataRefFields } from '@workflow/world';
 import { parse, unflatten } from 'devalue';
 
 // ---------------------------------------------------------------------------
@@ -690,66 +691,17 @@ function hydrateWorkflowIO<
 /**
  * Hydrate the eventData fields of an event resource.
  */
-function hydrateEventData<T extends { eventId?: string; eventData?: any }>(
-  resource: T,
-  revivers: Revivers
-): T {
+function hydrateEventData<
+  T extends { eventId?: string; eventType?: string; eventData?: any },
+>(resource: T, revivers: Revivers): T {
   if (!resource.eventData) return resource;
 
   const eventData = { ...resource.eventData };
 
-  // step_completed events have eventData.result (serialized return value)
-  if ('result' in eventData && eventData.result != null) {
-    try {
-      eventData.result = hydrateData(eventData.result, revivers);
-    } catch {
-      // Leave un-hydrated
-    }
-  }
-
-  // step_created events have eventData.input (serialized step arguments)
-  if ('input' in eventData && eventData.input != null) {
-    try {
-      eventData.input = hydrateData(eventData.input, revivers);
-    } catch {
-      // Leave un-hydrated
-    }
-  }
-
-  // run_completed events have eventData.output (serialized return value)
-  if ('output' in eventData && eventData.output != null) {
-    try {
-      eventData.output = hydrateData(eventData.output, revivers);
-    } catch {
-      // Leave un-hydrated
-    }
-  }
-
-  // hook_created events may have serialized metadata
-  if ('metadata' in eventData && eventData.metadata != null) {
-    try {
-      eventData.metadata = hydrateData(eventData.metadata, revivers);
-    } catch {
-      // Leave un-hydrated
-    }
-  }
-
-  // hook_received events have eventData.payload (serialized hook payload)
-  if ('payload' in eventData && eventData.payload != null) {
-    try {
-      eventData.payload = hydrateData(eventData.payload, revivers);
-    } catch {
-      // Leave un-hydrated
-    }
-  }
-
-  // step_failed / step_retrying / run_failed events have eventData.error
-  // (the thrown value, serialized via the error pipeline). Without this,
-  // event listings in o11y tooling would surface the raw `Uint8Array`
-  // payload instead of a hydrated `{ name, message, stack, … }` object.
-  if ('error' in eventData && eventData.error != null) {
+  for (const field of getEventDataRefFields(resource.eventType ?? '')) {
+    if (eventData[field] == null) continue;
     try {
-      eventData.error = hydrateData(eventData.error, revivers);
+      eventData[field] = hydrateData(eventData[field], revivers);
     } catch {
       // Leave un-hydrated
     }
@@ -791,6 +743,7 @@ export function hydrateResourceIO<
     stepId?: string;
     hookId?: string;
     eventId?: string;
+    eventType?: string;
     input?: any;
     output?: any;
     metadata?: any;