[core] Decode replay payloads synchronously

.changeset/decode-replay-payloads-synchronously.md

@@ -0,0 +1,9 @@
+---
+"@workflow/core": patch
+"@workflow/cli": patch
+"@workflow/web-shared": patch
+"@workflow/world": patch
+"@workflow/world-vercel": patch
+---
+
+Simplify payload codecs and decode replay payloads synchronously with Node AES-GCM and zstd.

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

@@ -6,7 +6,12 @@
  */
 
 import { inspect } from 'node:util';
-import { getCommonRevivers, maybeDecrypt } from '@workflow/core/serialization';
+import {
+  type DecryptionKey,
+  decrypt,
+  deriveRunPayloadKeys,
+  getCommonRevivers,
+} from '@workflow/core/serialization';
 import {
   ClassInstanceRef,
   extractClassName,
@@ -22,14 +27,12 @@ import { parseClassName } from '@workflow/utils/parse-name';
 import { getEventDataRefFields } from '@workflow/world';
 import chalk from 'chalk';
 
-/**
- * A function that resolves an encryption key for a run, or null to skip
- * decryption. Accepts a runId — the resolver is responsible for looking
- * up the WorkflowRun internally (with caching) if the World needs it.
- */
-export type EncryptionKeyResolver =
-  | ((runId: string) => Promise<Uint8Array | undefined>)
-  | null;
+async function decryptPayload(
+  value: unknown,
+  key: DecryptionKey | undefined
+): Promise<unknown> {
+  return value instanceof Uint8Array ? decrypt(value, key) : value;
+}
 
 // Re-export types and utilities that consumers need
 export {
@@ -313,52 +316,47 @@ function getRevivers(): Revivers {
  * Pre-process a resource's data fields: if the resolver is provided and
  * the field is encrypted, decrypt it before generic hydration.
  *
- * Uses core's `maybeDecrypt()` which handles the 'encr' prefix stripping
- * and AES-GCM decryption transparently.
+ * Binary envelopes go through the core decryptor; legacy values remain
+ * unchanged for the generic hydrator.
  *
- * When the resolver is null (no --decrypt flag), encrypted fields pass
+ * Without a resolver (no --decrypt flag), encrypted fields pass
  * through as Uint8Array and are replaced with EncryptedDataRef in post-processing.
  */
-async function maybeDecryptFields<
-  T extends {
-    runId?: string;
-    input?: any;
-    output?: any;
-    metadata?: any;
-    eventType?: string;
-    eventData?: any;
-  },
->(resource: T, resolver: EncryptionKeyResolver): Promise<T> {
-  if (!resolver) return resource;
+async function maybeDecryptFields<T>(
+  resource: T,
+  resolveKey?: (runId: string) => Promise<Uint8Array | undefined>
+): Promise<T> {
+  if (!resolveKey || !resource || typeof resource !== 'object') return resource;
 
-  const runId = (resource as any).runId as string | undefined;
-  if (!runId) return resource;
+  const source = resource as Record<string, unknown>;
+  if (typeof source.runId !== 'string') return resource;
 
-  const result = { ...resource };
+  const result = { ...source };
 
   try {
-    const rawKey = await resolver(runId);
+    const rawKey = await resolveKey(source.runId);
     // Resolve the full key capability so `--decrypt` can open sealed
     // ('encp') payloads that other runs wrote to this one, not just the
     // run's own symmetric ('encr') payloads.
-    const { deriveRunPayloadKeys } = await import(
-      '@workflow/core/serialization'
-    );
     const k = rawKey ? await deriveRunPayloadKeys(rawKey) : undefined;
 
     // Decrypt input/output/error fields (WorkflowRun, Step)
-    result.input = await maybeDecrypt(result.input, k);
-    result.output = await maybeDecrypt(result.output, k);
-    (result as any).error = await maybeDecrypt((result as any).error, k);
+    result.input = await decryptPayload(result.input, k);
+    result.output = await decryptPayload(result.output, k);
+    result.error = await decryptPayload(result.error, k);
 
     // Decrypt metadata field (Hook)
-    result.metadata = await maybeDecrypt(result.metadata, k);
+    result.metadata = await decryptPayload(result.metadata, k);
 
     // Decrypt eventData fields (Event)
     if (result.eventData && typeof result.eventData === 'object') {
-      const eventData = { ...result.eventData };
-      for (const field of getEventDataRefFields(result.eventType ?? '')) {
-        eventData[field] = await maybeDecrypt(eventData[field], k);
+      const eventData = {
+        ...(result.eventData as Record<string, unknown>),
+      };
+      const eventType =
+        typeof result.eventType === 'string' ? result.eventType : '';
+      for (const field of getEventDataRefFields(eventType)) {
+        eventData[field] = await decryptPayload(eventData[field], k);
       }
       result.eventData = eventData;
     }
@@ -375,10 +373,10 @@ async function maybeDecryptFields<
     // Decryption failed (bad key, corrupted ciphertext, etc.) — fall back
     // to showing encrypted placeholders instead of crashing the CLI.
     const { logger } = await import('../config/log.js');
-    logger.warn(`Decryption failed for resource ${runId}: ${message}`);
+    logger.warn(`Decryption failed for resource ${source.runId}: ${message}`);
   }
 
-  return result;
+  return result as T;
 }
 
 // ---------------------------------------------------------------------------
@@ -421,21 +419,30 @@ function replaceEncryptedAndExpiredWithRef<T>(resource: T): T {
 /**
  * Hydrate the serialized data fields of a resource for CLI display.
  *
- * When `encryptorResolver` is null (default / no --decrypt flag), encrypted
+ * Without `resolveKey` (the default / no --decrypt flag), encrypted
  * fields are shown as styled "🔒 Encrypted" placeholders via EncryptedDataRef.
  *
- * When `encryptorResolver` is provided (--decrypt flag), encrypted fields
+ * When `resolveKey` is provided (--decrypt flag), encrypted fields
  * are decrypted before hydration so the actual user data is displayed.
  */
-export async function hydrateResourceIO<T>(
+export async function hydrateResourceIO<
+  T extends {
+    stepId?: string;
+    hookId?: string;
+    eventId?: string;
+    eventType?: string;
+    input?: unknown;
+    output?: unknown;
+    metadata?: unknown;
+    eventData?: unknown;
+    executionContext?: unknown;
+  },
+>(
   resource: T,
-  keyResolver?: EncryptionKeyResolver
+  resolveKey?: (runId: string) => Promise<Uint8Array | undefined>
 ): Promise<T> {
   // Pre-process: decrypt any encrypted fields when a resolver is provided
-  const preprocessed = await maybeDecryptFields(
-    resource as any,
-    keyResolver ?? null
-  );
+  const preprocessed = await maybeDecryptFields(resource, resolveKey);
   const hydrated = hydrateResourceIOGeneric(preprocessed, getRevivers()) as T;
   // Post-process: swap encrypted Uint8Arrays and expired stubs for CLI-styled objects
   return replaceEncryptedAndExpiredWithRef(hydrated);

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

@@ -25,23 +25,25 @@ import {
   isObservabilityUpgradeRequiredError,
 } from './errors.js';
 import {
-  type EncryptionKeyResolver,
   hydrateResourceIO,
   isEncryptedRef,
   isExpiredRef,
 } from './hydration.js';
 import { resolveTimeWindow } from './time-window.js';
 
 /**
- * Create an EncryptionKeyResolver from a World instance.
- * Returns null if decrypt is false — encrypted data will show as a placeholder.
+ * Create a run-key lookup from a World instance.
+ * Returns undefined when encrypted data should remain a placeholder.
  *
  * The resolver fetches the full WorkflowRun (cached per runId) so that the
  * World can inspect deployment-specific fields for key resolution.
  */
-function createResolver(world: World, decrypt: boolean): EncryptionKeyResolver {
-  if (!decrypt) return null;
-  if (!world.getEncryptionKeyForRun) return null;
+function createResolver(
+  world: World,
+  decrypt: boolean
+): ((runId: string) => Promise<Uint8Array | undefined>) | undefined {
+  if (!decrypt) return;
+  if (!world.getEncryptionKeyForRun) return;
   const cache = new Map<string, Promise<Uint8Array | undefined>>();
   return (runId: string) => {
     let cached = cache.get(runId);

packages/core/src/encryption.test.ts

@@ -14,18 +14,31 @@ async function getOtherKey(): Promise<CryptoKey> {
   return importKey(OTHER_RAW_KEY);
 }
 
+async function captureError(action: () => unknown): Promise<unknown> {
+  try {
+    await action();
+  } catch (error) {
+    return error;
+  }
+  throw new Error('Expected operation to fail');
+}
+
 describe('encryption', () => {
   describe('round-trip', () => {
-    it('encrypt() + decrypt() returns the original plaintext', async () => {
+    it('decrypts synchronously on Node', async () => {
       const key = await getKey();
       const plaintext = new TextEncoder().encode('hello, workflow');
       const ciphertext = await encrypt(key, plaintext);
 
       // Ciphertext is longer than plaintext: 12-byte nonce + 16-byte GCM tag.
       expect(ciphertext.byteLength).toBe(plaintext.byteLength + 12 + 16);
 
-      const decoded = await decrypt(key, ciphertext);
-      expect(new TextDecoder().decode(decoded)).toBe('hello, workflow');
+      const decoded = decrypt(key, ciphertext);
+      expect(decoded).not.toBeInstanceOf(Promise);
+      expect(decoded).toBeInstanceOf(Uint8Array);
+      expect(new TextDecoder().decode(decoded as Uint8Array)).toBe(
+        'hello, workflow'
+      );
     });
   });
 
@@ -42,20 +55,15 @@ describe('encryption', () => {
       const key = await getKey();
       // 12-byte nonce + 16-byte tag = 28 bytes minimum. 10 bytes is too short.
       const tooShort = new Uint8Array(10).fill(0);
-      const error = await decrypt(key, tooShort).catch((e) => e);
+      const error = await captureError(() => decrypt(key, tooShort));
       expect(RuntimeDecryptionError.is(error)).toBe(true);
-      expect(error.message).toMatch(/Encrypted data too short/);
-      expect(error.context).toMatchObject({
-        operation: 'decrypt',
-        byteLength: 10,
+      expect(error).toMatchObject({
+        message: expect.stringMatching(/Encrypted data too short/),
+        context: { operation: 'decrypt', byteLength: 10 },
       });
     });
 
     it('throws RuntimeDecryptionError (not a bare OperationError) on auth-tag failure', async () => {
-      // GCM auth-tag verification failure surfaces from Node's Web
-      // Crypto API as `OperationError: The operation failed for an
-      // operation-specific reason at AESCipherJob.onDone`. The
-      // encryption module must rewrap this as a RuntimeDecryptionError.
       const key = await getKey();
       const plaintext = new TextEncoder().encode('hello, workflow');
       const ciphertext = await encrypt(key, plaintext);
@@ -64,16 +72,14 @@ describe('encryption', () => {
       const tampered = new Uint8Array(ciphertext);
       tampered[tampered.length - 1] ^= 0xff;
 
-      const error = await decrypt(key, tampered).catch((e) => e);
+      const error = await captureError(() => decrypt(key, tampered));
       expect(RuntimeDecryptionError.is(error)).toBe(true);
-      expect(error.cause).toBeDefined();
-      // The original DOMException carries name OperationError on Node 20+,
-      // which is what the wrapping is meant to capture as cause.
-      const cause = error.cause as { name?: string };
-      expect(cause?.name).toBe('OperationError');
-      expect(error.context).toMatchObject({
-        operation: 'decrypt',
-        byteLength: tampered.byteLength,
+      expect(error).toMatchObject({
+        cause: expect.anything(),
+        context: {
+          operation: 'decrypt',
+          byteLength: tampered.byteLength,
+        },
       });
     });
 
@@ -85,11 +91,22 @@ describe('encryption', () => {
         new TextEncoder().encode('secret')
       );
 
-      const error = await decrypt(readerKey, ciphertext).catch((e) => e);
+      const error = await captureError(() => decrypt(readerKey, ciphertext));
       expect(RuntimeDecryptionError.is(error)).toBe(true);
-      // Wrong key → auth tag mismatch → same OperationError as ciphertext corruption.
-      const cause = error.cause as { name?: string };
-      expect(cause?.name).toBe('OperationError');
+      expect(error).toMatchObject({ cause: expect.anything() });
+    });
+
+    it('keeps RuntimeDecryptionError on synchronous auth failure', async () => {
+      const key = await getKey();
+      const ciphertext = await encrypt(
+        key,
+        new TextEncoder().encode('tamper me')
+      );
+      ciphertext[ciphertext.length - 1] ^= 0xff;
+
+      expect(() => decrypt(key, ciphertext)).toThrowError(
+        RuntimeDecryptionError
+      );
     });
 
     it('does not record a formatPrefix at the low-level layer', async () => {
@@ -100,13 +117,14 @@ describe('encryption', () => {
       // The serialization layer attaches the real envelope prefix.
       const key = await getKey();
       const bogus = new Uint8Array(28).fill(0x41); // 28 bytes, passes length check
-      const error = await decrypt(key, bogus).catch((e) => e);
+      const error = await captureError(() => decrypt(key, bogus));
       expect(RuntimeDecryptionError.is(error)).toBe(true);
-      expect(error.context).toMatchObject({
-        operation: 'decrypt',
-        byteLength: 28,
+      expect(error).toMatchObject({
+        context: { operation: 'decrypt', byteLength: 28 },
+      });
+      expect(error).not.toMatchObject({
+        context: { formatPrefix: expect.anything() },
       });
-      expect(error.context.formatPrefix).toBeUndefined();
     });
   });