[core] Prepare replay payloads as event frames arrive

.changeset/prepare-streamed-replay-payloads.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Prepare replay payloads as event frames arrive and reuse immutable primitive values across fresh workflow VMs.

packages/core/src/abort-consistency.test.ts

@@ -37,7 +37,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
   return {
     runId: 'wrun_test',
     encryptionKey: undefined,
-    replayPayloadCache: new ReplayPayloadCache(undefined),
+    replayPayloadCache: new ReplayPayloadCache(),
     globalThis: context.globalThis,
     eventsConsumer: new EventsConsumer(events, {
       // Fake context: no deliveries are modeled, so the gate is a no-op here.

packages/core/src/abort-controller.test.ts

@@ -35,7 +35,7 @@ function setupWorkflowContext(
   return {
     runId: 'wrun_test',
     encryptionKey: undefined,
-    replayPayloadCache: new ReplayPayloadCache(undefined),
+    replayPayloadCache: new ReplayPayloadCache(),
     globalThis: context.globalThis,
     eventsConsumer: new EventsConsumer(events, {
       // Fake context: no deliveries are modeled, so the gate is a no-op here.

packages/core/src/abort-replay-ordering.test.ts

@@ -70,7 +70,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
   return {
     runId: 'wrun_test',
     encryptionKey: undefined,
-    replayPayloadCache: new ReplayPayloadCache(undefined),
+    replayPayloadCache: new ReplayPayloadCache(),
     globalThis: context.globalThis,
     eventsConsumer: new EventsConsumer(events, {
       // Fake context: no deliveries are modeled, so the gate is a no-op here.

packages/core/src/async-deserialization-ordering.test.ts

@@ -50,7 +50,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
   return {
     runId: 'wrun_test',
     encryptionKey: undefined,
-    replayPayloadCache: new ReplayPayloadCache(undefined),
+    replayPayloadCache: new ReplayPayloadCache(),
     globalThis: context.globalThis,
     eventsConsumer: new EventsConsumer(events, {
       // Fake context: no deliveries are modeled, so the gate is a no-op here.

packages/core/src/delivery-barrier-coverage.test.ts

@@ -86,7 +86,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     suspensionGeneration: 0,
     runId: 'wrun_test',
     encryptionKey: undefined,
-    replayPayloadCache: new ReplayPayloadCache(undefined),
+    replayPayloadCache: new ReplayPayloadCache(),
     globalThis: context.globalThis,
     eventsConsumer: new EventsConsumer(events, {
       // Fake context: no deliveries are modeled, so the gate is a no-op here.

packages/core/src/replay-payload-cache.test.ts

@@ -54,17 +54,29 @@ function makeEvents(payloads: unknown[]): Event[] {
 }
 
 describe('ReplayPayloadCache', () => {
-  it('deduplicates preparation and accepts a synchronous preparer', async () => {
+  it('deduplicates synchronous preparation without creating a promise', () => {
     const payload = new Uint8Array([1]);
     const preparer = vi.fn<typeof prepareReplayPayload>((value) => value);
+    const hydrate = vi.fn((prepared: unknown) => prepared);
     const cache = new ReplayPayloadCache(undefined, preparer);
 
-    const first = cache.prepareEventPayload('evnt_one', 'result', payload);
-    const second = cache.prepareEventPayload('evnt_one', 'result', payload);
+    const first = cache.getEventValue('evnt_one', payload, hydrate);
+    const second = cache.getEventValue('evnt_one', payload, hydrate);
 
     expect(first).toBe(second);
-    await expect(first).resolves.toBe(payload);
+    expect(first).toEqual(payload);
     expect(preparer).toHaveBeenCalledOnce();
+    expect(hydrate).toHaveBeenCalledTimes(2);
+  });
+
+  it('hydrates and memoizes a primitive without creating a promise', () => {
+    const payload = new Uint8Array([1]);
+    const hydrate = vi.fn(() => 42);
+    const cache = new ReplayPayloadCache(undefined, (value) => value);
+
+    expect(cache.getEventValue('evnt_one', payload, hydrate)).toBe(42);
+    expect(cache.getEventValue('evnt_one', payload, hydrate)).toBe(42);
+    expect(hydrate).toHaveBeenCalledOnce();
   });
 
   it('keeps a failed prewarm until its consumer observes it, then retries', async () => {
@@ -76,13 +88,12 @@ describe('ReplayPayloadCache', () => {
       .mockReturnValueOnce(payload);
     const cache = new ReplayPayloadCache(undefined, preparer);
 
-    await cache.prewarm(run, []);
-    await expect(cache.prepareWorkflowInput(run)).rejects.toThrow(
-      'decrypt failed'
-    );
+    cache.prepareAll(run, []);
+    await Promise.resolve();
+    await expect(cache.getWorkflowInput(run)).rejects.toThrow('decrypt failed');
     expect(preparer).toHaveBeenCalledOnce();
 
-    await expect(cache.prepareWorkflowInput(run)).resolves.toBe(payload);
+    expect(cache.getWorkflowInput(run)).toEqual(payload);
     expect(preparer).toHaveBeenCalledTimes(2);
   });
 
@@ -99,16 +110,61 @@ describe('ReplayPayloadCache', () => {
     const run = makeRun(payloads[0]);
     const events = makeEvents(payloads.slice(1));
 
-    const warming = cache.prewarm(run, events);
+    cache.prepareAll(run, events);
     expect(preparer).toHaveBeenCalledTimes(4);
     for (const resolve of resolvers.reverse()) resolve();
-    await warming;
+    await Promise.all([
+      cache.getWorkflowInput(run),
+      ...events.map((event) => {
+        switch (event.eventType) {
+          case 'step_completed':
+            return cache.getEventValue(
+              event.eventId,
+              event.eventData?.result,
+              (prepared) => prepared
+            );
+          case 'step_failed':
+            return cache.getEventValue(
+              event.eventId,
+              event.eventData?.error,
+              (prepared) => prepared
+            );
+          case 'hook_received':
+            return cache.getEventValue(
+              event.eventId,
+              event.eventData?.payload,
+              (prepared) => prepared
+            );
+          default:
+            throw new Error(`Unexpected event: ${event.eventType}`);
+        }
+      }),
+    ]);
 
-    const allSettled = vi.spyOn(Promise, 'allSettled');
-    await cache.prewarm(run, events);
+    cache.prepareAll(run, events);
     expect(preparer).toHaveBeenCalledTimes(4);
-    expect(allSettled).toHaveBeenLastCalledWith([]);
-    allSettled.mockRestore();
+  });
+
+  it('prepares streamed events synchronously inside the decoder callback', async () => {
+    const payload = new Uint8Array([1]);
+    const order: string[] = [];
+    const preparer = vi.fn<typeof prepareReplayPayload>((value) => {
+      order.push('prepare');
+      return value;
+    });
+    const cache = new ReplayPayloadCache(undefined, preparer);
+    const [event] = makeEvents([payload]);
+
+    cache.prepareEvent(event);
+    expect(preparer).toHaveBeenCalledOnce();
+    expect(order).toEqual(['prepare']);
+
+    cache.prepareEvent(event);
+    expect(order).toEqual(['prepare']);
+
+    expect(
+      cache.getEventValue(event.eventId, payload, (prepared) => prepared)
+    ).toEqual(payload);
   });
 
   it('caches real decrypt/decompress output but revives fresh objects', async () => {
@@ -126,15 +182,19 @@ describe('ReplayPayloadCache', () => {
     const preparer = vi.fn<typeof prepareReplayPayload>(prepareReplayPayload);
     const cache = new ReplayPayloadCache(key, preparer);
 
-    const prepared = await cache.prepareEventPayload(
+    const directPreparation = prepareReplayPayload(serialized, key);
+    expect(directPreparation).not.toBeInstanceOf(Promise);
+    await directPreparation;
+
+    const prepared = await cache.getEventValue(
       'evnt_encrypted',
-      'result',
-      serialized
+      serialized,
+      (value) => value
     );
-    const samePrepared = await cache.prepareEventPayload(
+    const samePrepared = await cache.getEventValue(
       'evnt_encrypted',
-      'result',
-      serialized
+      serialized,
+      (value) => value
     );
     const first = deserializePreparedReplayPayload(prepared) as {
       count: number;
@@ -149,89 +209,115 @@ describe('ReplayPayloadCache', () => {
     expect(second.count).toBe(0);
   });
 
-  it('rescans a log whose missing events were filled in below the scanned prefix', async () => {
-    // A stale-snapshot (412) restart replaces the log with a corrected one, so
-    // the events it was missing appear BELOW the length already scanned and
-    // shift every later position. Resuming from that length skips exactly the
-    // events the reload was for, which is what `resetScan` exists to prevent.
+  it('finds events inserted below a previously prepared prefix', () => {
+    // A stale-snapshot restart can replace the log with a corrected one whose
+    // missing events appear below the old tail. Full scans are cheap because
+    // event-id cache hits do no payload work.
     const payloads = [0, 1, 2].map((value) => new Uint8Array([value]));
     const preparer = vi.fn<typeof prepareReplayPayload>((value) => value);
     const cache = new ReplayPayloadCache(undefined, preparer);
     const run = makeRun(undefined);
     const [first, missing, second] = makeEvents(payloads);
 
-    await cache.prewarm(run, [first, second]);
+    cache.prepareAll(run, [first, second]);
     expect(preparer).toHaveBeenCalledTimes(2);
 
-    // Positional resume: `missing` sits inside the scanned prefix, so it is
-    // skipped and its payload is only prepared on demand.
-    await cache.prewarm(run, [first, missing, second]);
-    expect(preparer).toHaveBeenCalledTimes(2);
-
-    cache.resetScan();
-    await cache.prewarm(run, [first, missing, second]);
-    // Only the inserted event is new: the other two are keyed by event id and
-    // stay prepared across the rescan.
+    cache.prepareAll(run, [first, missing, second]);
     expect(preparer).toHaveBeenCalledTimes(3);
     expect(preparer).toHaveBeenLastCalledWith(payloads[1], undefined);
   });
 
-  it('bypasses legacy values and ignores missing event data during prewarm', async () => {
+  it('bypasses legacy values and ignores missing event data during preparation', async () => {
     const legacy = [0, { value: 1 }];
     const preparer = vi.fn<typeof prepareReplayPayload>((value) => value);
     const cache = new ReplayPayloadCache(undefined, preparer);
 
-    await cache.prepareEventPayload('evnt_legacy', 'result', legacy);
-    await cache.prepareEventPayload('evnt_legacy', 'result', legacy);
+    await cache.getEventValue('evnt_legacy', legacy, (prepared) => prepared);
+    await cache.getEventValue('evnt_legacy', legacy, (prepared) => prepared);
     expect(preparer).not.toHaveBeenCalled();
 
     const events = makeEvents([legacy, legacy, legacy]);
     events[2] = { ...events[2], eventData: undefined } as unknown as Event;
-    await cache.prewarm(makeRun(legacy), events);
+    cache.prepareAll(makeRun(legacy), events);
     expect(preparer).not.toHaveBeenCalled();
   });
 
   it('memoizes primitive step results, including undefined', async () => {
     for (const value of [0, false, '', null, undefined]) {
-      const cache = new ReplayPayloadCache(undefined);
+      const cache = new ReplayPayloadCache();
       const hydrate = vi.fn().mockResolvedValue(value);
 
-      expect(await cache.getStepResult('evnt_result', hydrate)).toBe(value);
-      expect(await cache.getStepResult('evnt_result', hydrate)).toBe(value);
+      expect(await cache.getEventValue('evnt_result', undefined, hydrate)).toBe(
+        value
+      );
+      expect(await cache.getEventValue('evnt_result', undefined, hydrate)).toBe(
+        value
+      );
       expect(hydrate).toHaveBeenCalledOnce();
     }
   });
 
-  it('rehydrates mutable and oversized step results', async () => {
+  it('isolates primitive values by event id', async () => {
+    const cache = new ReplayPayloadCache();
+    const result = vi.fn().mockResolvedValue('result');
+    const error = vi.fn().mockResolvedValue('error');
+
+    await expect(
+      cache.getEventValue('evnt_result', undefined, result)
+    ).resolves.toBe('result');
+    await expect(
+      cache.getEventValue('evnt_error', undefined, error)
+    ).resolves.toBe('error');
+    expect(cache.getEventValue('evnt_result', undefined, result)).toBe(
+      'result'
+    );
+    expect(result).toHaveBeenCalledOnce();
+    expect(error).toHaveBeenCalledOnce();
+  });
+
+  it('rehydrates mutable results and memoizes primitives of any size', async () => {
     const oversized = 'x'.repeat(4097);
     for (const value of [{ count: 0 }, oversized]) {
-      const cache = new ReplayPayloadCache(undefined);
+      const cache = new ReplayPayloadCache();
       const hydrate = vi
         .fn()
         .mockImplementation(async () =>
           typeof value === 'object' ? { ...value } : value
         );
 
-      const first = await cache.getStepResult('evnt_result', hydrate);
-      const second = await cache.getStepResult('evnt_result', hydrate);
-      expect(hydrate).toHaveBeenCalledTimes(2);
-      if (typeof value === 'object') expect(second).not.toBe(first);
+      const first = await cache.getEventValue(
+        'evnt_result',
+        undefined,
+        hydrate
+      );
+      const second = await cache.getEventValue(
+        'evnt_result',
+        undefined,
+        hydrate
+      );
+      if (typeof value === 'object') {
+        expect(hydrate).toHaveBeenCalledTimes(2);
+        expect(second).not.toBe(first);
+      } else {
+        expect(hydrate).toHaveBeenCalledOnce();
+        expect(second).toBe(first);
+      }
     }
   });
 
   it('does not memoize failed step hydration', async () => {
-    const cache = new ReplayPayloadCache(undefined);
+    const cache = new ReplayPayloadCache();
     const hydrate = vi
       .fn()
       .mockRejectedValueOnce(new Error('boom'))
       .mockResolvedValueOnce('ok');
 
-    await expect(cache.getStepResult('evnt_result', hydrate)).rejects.toThrow(
-      'boom'
-    );
-    await expect(cache.getStepResult('evnt_result', hydrate)).resolves.toBe(
-      'ok'
-    );
+    await expect(
+      cache.getEventValue('evnt_result', undefined, hydrate)
+    ).rejects.toThrow('boom');
+    await expect(
+      cache.getEventValue('evnt_result', undefined, hydrate)
+    ).resolves.toBe('ok');
     expect(hydrate).toHaveBeenCalledTimes(2);
   });
 });