[core] Prepare replay payloads as event frames arrive

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

@@ -0,0 +1,7 @@
+---
+"@workflow/core": patch
+"@workflow/world": patch
+"@workflow/world-vercel": patch
+---
+
+Prepare replay payloads as validated 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

@@ -57,14 +57,16 @@ describe('ReplayPayloadCache', () => {
   it('deduplicates preparation', async () => {
     const payload = new Uint8Array([1]);
     const preparer = vi.fn<typeof prepareReplayPayload>(async (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);
+    await expect(second).resolves.toBe(payload);
     expect(preparer).toHaveBeenCalledOnce();
+    expect(hydrate).toHaveBeenCalledTimes(2);
   });
 
   it('compacts prepared bytes before retaining them', async () => {
@@ -75,17 +77,29 @@ describe('ReplayPayloadCache', () => {
       vi.fn<typeof prepareReplayPayload>(async () => prepared)
     );
 
-    const retained = await cache.prepareEventPayload(
+    const retained = await cache.getEventValue(
       'evnt_compact',
-      'result',
-      new Uint8Array([1])
+      new Uint8Array([1]),
+      (value) => value
     );
 
     assert(retained instanceof Uint8Array);
     expect(retained).toEqual(prepared);
     expect(retained.buffer.byteLength).toBe(retained.byteLength);
   });
 
+  it('hydrates and memoizes a primitive', async () => {
+    const payload = new Uint8Array([1]);
+    const hydrate = vi.fn(() => 42);
+    const cache = new ReplayPayloadCache(undefined, async (value) => value);
+
+    await expect(
+      cache.getEventValue('evnt_one', payload, hydrate)
+    ).resolves.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 () => {
     const payload = new Uint8Array([1]);
     const run = makeRun(payload);
@@ -95,13 +109,12 @@ describe('ReplayPayloadCache', () => {
       .mockResolvedValueOnce(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);
+    await expect(cache.getWorkflowInput(run)).resolves.toBe(payload);
     expect(preparer).toHaveBeenCalledTimes(2);
   });
 
@@ -118,16 +131,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('starts streamed preparation inside the decoder callback', async () => {
+    const payload = new Uint8Array([1]);
+    const order: string[] = [];
+    const preparer = vi.fn<typeof prepareReplayPayload>(async (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']);
+
+    await expect(
+      cache.getEventValue(event.eventId, payload, (prepared) => prepared)
+    ).resolves.toBe(payload);
   });
 
   it('caches real decrypt/decompress output but revives fresh objects', async () => {
@@ -145,15 +203,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).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;
@@ -168,89 +230,136 @@ 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('prepares only events appended after the scanned prefix', () => {
     const payloads = [0, 1, 2].map((value) => new Uint8Array([value]));
     const preparer = vi.fn<typeof prepareReplayPayload>(async (value) => value);
     const cache = new ReplayPayloadCache(undefined, preparer);
     const run = makeRun(undefined);
-    const [first, missing, second] = makeEvents(payloads);
+    const [first, second, third] = makeEvents(payloads);
+    const prepareEvent = vi.spyOn(cache, 'prepareEvent');
 
-    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]);
+    prepareEvent.mockClear();
+    cache.prepareAll(run, [first, second, third]);
+    expect(prepareEvent).toHaveBeenCalledOnce();
+    expect(prepareEvent).toHaveBeenCalledWith(third);
+    expect(preparer).toHaveBeenCalledTimes(3);
+  });
+
+  it('rescans a corrected event log after reset', () => {
+    const payloads = [0, 1, 2].map((value) => new Uint8Array([value]));
+    const preparer = vi.fn<typeof prepareReplayPayload>(async (value) => value);
+    const cache = new ReplayPayloadCache(undefined, preparer);
+    const run = makeRun(undefined);
+    const [first, missing, second] = makeEvents(payloads);
+
+    cache.prepareAll(run, [first, second]);
+    cache.prepareAll(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>(async (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 () => {
-    const oversized = 'x'.repeat(4097);
-    for (const value of [{ count: 0 }, oversized]) {
-      const cache = new ReplayPayloadCache(undefined);
+  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('memoizes primitives of any size and rehydrates object results', async () => {
+    const commonText = 'x'.repeat(256 * 1024);
+    const oversizedText = 'x'.repeat(16 * 1024 * 1024 + 1);
+    for (const [value, expectedHydrations] of [
+      [{ count: 0 }, 2],
+      [commonText, 1],
+      [oversizedText, 1],
+    ] as const) {
+      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
+      );
+      expect(hydrate).toHaveBeenCalledTimes(expectedHydrations);
+      if (typeof value === 'object') {
+        expect(second).not.toBe(first);
+      } else if (expectedHydrations === 1) {
+        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);
   });
 });