Wire encryption into serialization and fix EventsConsumer for async replay

.changeset/e2e-encryption.md

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

packages/core/src/runtime/run.ts

@@ -172,11 +172,11 @@ export class Run<TResult> {
         if (run.status === 'completed') {
           const rawKey = await this.world.getEncryptionKeyForRun?.(run);
           const encryptionKey = rawKey ? await importKey(rawKey) : undefined;
-          return await hydrateWorkflowReturnValue(
+          return (await hydrateWorkflowReturnValue(
             run.output,
             this.runId,
             encryptionKey
-          );
+          )) as TResult;
         }
 
         if (run.status === 'cancelled') {

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

@@ -312,7 +312,10 @@ const stepHandler = getWorldHandlers().createQueueHandler(
             const ops: Promise<void>[] = [];
             const rawKey = await world.getEncryptionKeyForRun?.(workflowRunId);
             const encryptionKey = rawKey ? await importKey(rawKey) : undefined;
-            const hydratedInput = await trace(
+            // The hydrated input is { args, thisVal, closureVars } but
+            // hydrateStepArguments returns `unknown` since the serialization
+            // layer can't know the shape at compile time.
+            const hydratedInput = (await trace(
               'step.hydrate',
               {},
               async (hydrateSpan) => {
@@ -325,12 +328,12 @@ const stepHandler = getWorldHandlers().createQueueHandler(
                 );
                 const durationMs = Date.now() - startTime;
                 hydrateSpan?.setAttributes({
-                  ...Attribute.StepArgumentsCount(result.args.length),
+                  ...Attribute.StepArgumentsCount((result as any).args.length),
                   ...Attribute.QueueDeserializeTimeMs(durationMs),
                 });
                 return result;
               }
-            );
+            )) as { args: unknown[]; thisVal?: unknown; closureVars?: unknown };
 
             const args = hydratedInput.args;
             const thisVal = hydratedInput.thisVal ?? null;

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

@@ -8,13 +8,13 @@ import {
   type WorkflowRun,
   type World,
 } from '@workflow/world';
-import { importKey } from '../encryption.js';
 import type {
   HookInvocationQueueItem,
   StepInvocationQueueItem,
   WaitInvocationQueueItem,
   WorkflowSuspension,
 } from '../global.js';
+import { importKey } from '../encryption.js';
 import { runtimeLogger } from '../logger.js';
 import { dehydrateStepArguments } from '../serialization.js';
 import * as Attribute from '../telemetry/semantic-conventions.js';