Wire AES-GCM encryption into serialization layer (#1251)

* fix(core): chain unconsumed event check onto promiseQueue to prevent false positives

The EventsConsumer's unconsumed event check (setTimeout(0)) was racing
against the promiseQueue's async deserialization. When parallel steps
completed and their hydrateStepReturnValue did real async work (e.g.,
decryption), the setTimeout(0) fired before the promise chain resolved
the step results and triggered the next subscribe() call. This caused
step_created events for sequential steps to be falsely flagged as
unconsumed/orphaned.

Fix: chain the unconsumed check onto the promiseQueue via getPromiseQueue()
so it only fires after all pending async work completes. Use
process.nextTick (not setTimeout) after the queue drains to give
synchronous subscribe() calls from resolved user code a chance to cancel.
Version-based cancellation replaces clearTimeout since the check is now
promise-based.

Adds getPromiseQueue option to EventsConsumerOptions. The workflow.ts
context uses a getter/setter to keep the promiseQueue holder in sync.

Reproduction test: parallel steps A+B with 10ms mock deserialization
delay, followed by sequential step C. Previously failed with
'Unconsumed event: step_created(C)'. Now passes.

* fix: chain hydrateWorkflowArguments onto promiseQueue to prevent false unconsumed events

The unconsumed event check was firing during the async gap between
run_started consumption and the workflow function subscribing its
first step callbacks. This happened because hydrateWorkflowArguments
is async, and during its await, the EventsConsumer advanced to
step_created events that had no subscriber yet.

Fix: chain hydrateWorkflowArguments onto the promiseQueue so the
unconsumed check (which waits for the queue to drain) doesn't fire
until after the workflow arguments are hydrated and the workflow
function has been invoked.

* fix: use setTimeout(0) macrotask for unconsumed check to ensure VM promise propagation completes

The process.nextTick-based unconsumed check was still racing against
VM promise propagation. After promiseQueue resolves and the user code's
resolve() fires, there are multiple microtask hops through the VM
boundary before the workflow code actually calls subscribe() for the
next steps. process.nextTick fires before those VM microtasks complete.

setTimeout(0) is a macrotask that is guaranteed to fire only after ALL
microtasks (including VM promise chain propagation) have drained. The
pendingUnconsumedTimeout handle is stored and cleared in subscribe()
to prevent keeping the event loop alive unnecessarily.

* fix: increase unconsumed event check delay to 100ms for cross-VM promise propagation

setTimeout(0) is insufficient because Node.js does not guarantee that
macrotasks fire after all cross-context (VM boundary) microtasks settle.
After promiseQueue resolves and resolve() fires in the host context,
there are multiple microtask hops through the VM boundary before the
workflow code actually calls subscribe(). A 100ms delay provides
sufficient time for this propagation while still detecting truly
orphaned events promptly.

Also update sleep.test.ts to wait 200ms for the unconsumed check.

* Add browser-compatible AES-GCM to core and HKDF key derivation to world-vercel

* update changeset

* Move HKDF key derivation server-side: API returns per-run derived key

* Refactor encrypt/decrypt to accept CryptoKey, export importKey for callers to import once per run

* Overload getEncryptionKeyForRun: accept context for start(), fetch WorkflowRun in resume-hook

* Split changeset into per-package descriptions for world, world-vercel, and core

* Remove unnecessary Uint8Array.from() wrapper around Buffer.from()

* Use zod to parse Vercel API response

* Wire encryption into serialization layer

* Wire AES-GCM encryption into serialization layer

* update changeset

* Add encryption unit tests: primitives, maybeEncrypt/maybeDecrypt, isEncrypted, complex type round-trips

* Accept CryptoKey in encrypt/decrypt, export importKey for callers to import once per run

* Fix review comments: cache stream encryption key, remove redundant casts, fix stale comments

* Trying to clean up some type non-sense

* fix: restore world-vercel files to main versions

The rebase incorrectly picked up older versions of these files from
early encryption branch commits. The main versions are correct and
up-to-date.

* fix: add type cast for hydrateStepReturnValue return in hook.ts

* fix: address review feedback on encryption PR

- Remove Vercel-specific error message from maybeDecrypt (core should
  not reference VERCEL_DEPLOYMENT_KEY)
- Move stream encryption/decryption from transport layer
  (WorkflowServerReadableStream/WritableStream) to framing layer
  (getSerializeStream/getDeserializeStream). Frame length headers stay
  in the clear so frame boundaries are always parseable regardless of
  transport chunking; encryption wraps the frame payload.
- Remove explicit Promise<unknown> return types from all 4 hydrate
  functions. On main these had inferred types (any from devalue),
  so callers didn't need casts. The encryption branch added explicit
  annotations that broke this.
- Revert unnecessary type casts in run.ts, step-handler.ts, hook.ts
  that were only needed due to the explicit Promise<unknown> annotations
- Revert closureVars type from unknown back to Record<string, any>
  in context-storage.ts to match the contract with getClosureVars
- Fix hydrateWorkflowArguments JSDoc for unused _runId parameter

* Revert more unnecessary changes

* cleanup: remove unused runId param, deduplicate processFrames, add legacy comments

- Remove unused _runId parameter from WorkflowServerReadableStream
  constructor and all 4 call sites
- Deduplicate processFrames decryption: decrypt first and reassign
  format/payload, then fall through to single deserialization path
- Add comments on all legacy non-Uint8Array branches explaining when
  this happens (specVersion 1 runs stored data as plain JSON arrays)
- Fix duplicate code block in hydrateStepReturnValue

* feat: wire cryptoKey through stream serialize/deserialize pipeline

Thread the encryption key through the entire stream serialization chain
so that ReadableStream and WritableStream values are encrypted/decrypted
at the framing level.

- Add optional cryptoKey param to getExternalReducers, getStepReducers,
  getExternalRevivers, getStepRevivers
- Pass cryptoKey to getSerializeStream/getDeserializeStream at all 8
  internal call sites within reducers/revivers
- Thread key from dehydrate/hydrate functions into their reducers/revivers
- Cache encryption key in Run class (resolved once via getEncryptionKey(),
  reused for returnValue, getReadable(), etc.)
- Make Run#getReadable() async to resolve the cached key before creating
  the deserialize stream
- Add encryptionKey to step context storage so getWritable() can access
  it during step execution

* fix: make cryptoKey required-but-nullable to prevent silent omission, add stream encryption tests

Change cryptoKey parameter from optional (cryptoKey?) to required-but-
nullable (cryptoKey: CryptoKey | undefined) on all 6 functions:
- getSerializeStream, getDeserializeStream
- getExternalReducers, getStepReducers
- getExternalRevivers, getStepRevivers

This ensures every call site must explicitly pass the key or undefined,
making it impossible to accidentally omit it and silently skip encryption.

Add 7 stream encryption round-trip tests:
- Encrypted frames have 'encr' prefix inside length header
- Full round-trip: encrypt serialize -> decrypt deserialize
- Concatenated encrypted frames (transport coalescing)
- Split encrypted frames (transport splitting)
- Error when encrypted data encountered without key
- No encryption when key is undefined
- Large payload round-trip

Full audit confirms all encryption key threading is complete:
- All 8 dehydrate/hydrate functions pass key to reducers/revivers
- All stream serialize/deserialize call sites pass key
- Run class caches key for reuse across returnValue and getReadable()
- Step context storage carries key for getWritable()

* fix: keep Run#getReadable() sync, resolve encryption key lazily in streams

- Revert Run#getReadable() to synchronous (non-breaking API).
  The encryption key is passed as a Promise through the chain and
  resolved lazily inside the first async transform() call.
- Add EncryptionKeyParam type alias that accepts CryptoKey, undefined,
  or Promise<CryptoKey | undefined>. Used by getSerializeStream,
  getDeserializeStream, and all reducer/reviver functions.
- Key promises are resolved once on first use via a keyState cache
  object inside each stream's transform closure.
- Fix CLI showStream to resolve encryption key from world when runId
  is provided via --run flag, instead of passing undefined.
- Remove incorrect CLI warning that --run is not supported for streams
  (it is now needed for encrypted stream decryption).

* .

* fix: address review feedback from PR #1251

- Fix 4 broken dehydrateWorkflowArguments calls in workflow.test.ts
  that were passing ops as runId (missing runId and key params)
- Use WorkflowRuntimeError instead of plain Error in decodeFormatPrefix
  for unknown serialization formats, for consistency and programmatic
  error handling
- Document maybeDecrypt throw behavior: callers should be aware this
  surfaces as a rejected promise during key rotation/misconfiguration
- Document key-fetch rejection timing in streams: promise rejection
  won't surface until the first chunk is processed

Author: TooTallNate

.changeset/e2e-encryption.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Wire AES-GCM encryption into serialization layer with stream support

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

@@ -1,4 +1,6 @@
+import { importKey } from '@workflow/core/encryption';
 import {
+  type EncryptionKeyParam,
   getDeserializeStream,
   getExternalRevivers,
 } from '@workflow/core/serialization';
@@ -780,16 +782,32 @@ export const showStream = async (
   streamId: string,
   opts: InspectCLIOptions = {}
 ) => {
-  if (opts.runId || opts.stepId) {
+  if (opts.stepId) {
     logger.warn(
-      'Filtering by run-id or step-id is not supported when showing a stream, ignoring filter.'
+      'Filtering by step-id is not supported when showing a stream, ignoring filter.'
     );
   }
   const rawStream = await world.readFromStream(streamId);
 
+  // Resolve the encryption key if a runId is provided (needed for encrypted streams).
+  // The key is passed as a promise so stream construction is synchronous —
+  // it will be resolved lazily on the first encrypted frame.
+  let encryptionKey: EncryptionKeyParam;
+  if (opts.runId) {
+    encryptionKey = (async () => {
+      const rawKey = await world.getEncryptionKeyForRun?.(opts.runId!);
+      return rawKey ? await importKey(rawKey) : undefined;
+    })();
+  }
+
   // Deserialize the stream to get JavaScript objects
-  const revivers = getExternalRevivers(globalThis, [], '');
-  const transform = getDeserializeStream(revivers);
+  const revivers = getExternalRevivers(
+    globalThis,
+    [],
+    opts.runId ?? '',
+    encryptionKey
+  );
+  const transform = getDeserializeStream(revivers, encryptionKey);
   const stream = rawStream.pipeThrough(transform);
 
   logger.info('Streaming to stdout, press CTRL+C to abort.');

packages/core/src/runtime/run.ts

@@ -8,7 +8,7 @@ import {
   type WorkflowRunStatus,
   type World,
 } from '@workflow/world';
-import { importKey } from '../encryption.js';
+import { type CryptoKey, importKey } from '../encryption.js';
 import {
   getExternalRevivers,
   hydrateWorkflowReturnValue,
@@ -63,11 +63,35 @@ export class Run<TResult> {
    */
   private world: World;
 
+  /**
+   * Cached encryption key resolution. Resolved once on first use and
+   * reused for returnValue, getReadable(), etc.
+   * @internal
+   */
+  private encryptionKeyPromise: Promise<CryptoKey | undefined> | null = null;
+
   constructor(runId: string) {
     this.runId = runId;
     this.world = getWorld();
   }
 
+  /**
+   * Resolves and caches the encryption key for this run.
+   * The key is the same for the lifetime of a run, so it only needs
+   * to be resolved once.
+   * @internal
+   */
+  private getEncryptionKey(): Promise<CryptoKey | undefined> {
+    if (!this.encryptionKeyPromise) {
+      this.encryptionKeyPromise = (async () => {
+        const run = await this.world.runs.get(this.runId);
+        const rawKey = await this.world.getEncryptionKeyForRun?.(run);
+        return rawKey ? await importKey(rawKey) : undefined;
+      })();
+    }
+    return this.encryptionKeyPromise;
+  }
+
   /**
    * Interrupts pending `sleep()` calls, resuming the workflow early.
    *
@@ -153,7 +177,15 @@ export class Run<TResult> {
   ): ReadableStream<R> {
     const { ops = [], global = globalThis, startIndex, namespace } = options;
     const name = getWorkflowRunStreamId(this.runId, namespace);
-    return getExternalRevivers(global, ops, this.runId).ReadableStream({
+    // Pass the key as a promise — it will be resolved lazily inside
+    // the first async transform() call of the deserialize stream.
+    const encryptionKey = this.getEncryptionKey();
+    return getExternalRevivers(
+      global,
+      ops,
+      this.runId,
+      encryptionKey
+    ).ReadableStream({
       name,
       startIndex,
     }) as ReadableStream<R>;
@@ -170,8 +202,7 @@ export class Run<TResult> {
         const run = await this.world.runs.get(this.runId);
 
         if (run.status === 'completed') {
-          const rawKey = await this.world.getEncryptionKeyForRun?.(run);
-          const encryptionKey = rawKey ? await importKey(rawKey) : undefined;
+          const encryptionKey = await this.getEncryptionKey();
           return await hydrateWorkflowReturnValue(
             run.output,
             this.runId,

packages/core/src/runtime/step-handler.ts

@@ -356,6 +356,7 @@ const stepHandler = getWorldHandlers().createQueueHandler(
                   },
                   ops,
                   closureVars: hydratedInput.closureVars,
+                  encryptionKey,
                 },
                 () => stepFn.apply(thisVal, args)
               );