[core] Don't download recorded step inputs on replay (resolveData: 'skip-step-inputs')

.changeset/omit-step-inputs-on-replay.md

@@ -0,0 +1,10 @@
+---
+'@workflow/world': patch
+'@workflow/world-vercel': patch
+'@workflow/core': patch
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+'@workflow/world-testing': patch
+---
+
+Added a `resolveData: 'skip-step-inputs'` option, which directs the World to leave out `input` from `step_created` and `step_started` events. Replay recomputes step arguments by re-running workflow code, and steps take their input from the `step_started` response or from memory, never from the replay log, so replay now reads the event log with this option and no longer downloads recorded step inputs. For workflows that pass growing state into their steps, this removes the part of the replay transfer that grows quadratically. A World that doesn't implement the option must treat it as `'all'`.

packages/core/src/retained-vm-loop.test.ts

@@ -553,8 +553,27 @@ type DriveWorldOptions = {
    * continuing over.
    */
   staleFirstReload?: boolean;
+  /**
+   * Honor `resolveData: 'skip-step-inputs'` the way workflow-server does: every replay event
+   * the World hands back (a list page, the `run_started` preload, an inline
+   * delta) leaves `input` out of `step_created` / `step_started`, while the
+   * created event and the `step` entity a `step_started` returns keep it.
+   * `stripped` counts the events served without their input, so a test can
+   * tell the omission actually happened.
+   */
+  skipStepInputs?: { stripped: number };
 };
 
+/** A replay event as a World honoring `skip-step-inputs` serves it. */
+function withoutStepInput(event: Event): Event {
+  if (event.eventType !== 'step_created' && event.eventType !== 'step_started')
+    return event;
+  const eventData = event.eventData as Record<string, unknown> | undefined;
+  if (!eventData || !('input' in eventData)) return event;
+  const { input: _input, ...rest } = eventData;
+  return { ...event, eventData: rest } as Event;
+}
+
 async function drive(
   runId: string,
   workflowCode = twoStepWorkflow,
@@ -573,6 +592,17 @@ async function drive(
     deploymentId: 'test-deployment',
   };
   const events: Event[] = [];
+  // What a replay read hands back: the log, minus step inputs when the caller
+  // asked for that and the World honors it.
+  const served = (log: Event[], params?: { resolveData?: string }) => {
+    const omission = options.skipStepInputs;
+    if (!omission || params?.resolveData !== 'skip-step-inputs') return log;
+    return log.map((event) => {
+      const stripped = withoutStepInput(event);
+      if (stripped !== event) omission.stripped++;
+      return stripped;
+    });
+  };
   const createdEvents: any[] = [];
   const createParams: any[] = [];
   let seq = 0;
@@ -596,7 +626,7 @@ async function drive(
       createdEvents.push(data);
       createParams.push({ eventType: data.eventType, ...params });
       if (data.eventType === 'run_started') {
-        return { run, events };
+        return { run, events: served(events, params) };
       }
       // A create whose token is already claimed commits `hook_conflict` on the
       // slot the `hook_created` asked for — not an error for the caller, an
@@ -669,7 +699,10 @@ async function drive(
       const delta =
         typeof params?.sinceCursor === 'string' && !options.withholdDelta
           ? {
-              events: events.slice(cursorPosition.get(params.sinceCursor) ?? 0),
+              events: served(
+                events.slice(cursorPosition.get(params.sinceCursor) ?? 0),
+                params
+              ),
               cursor: nextCursor(),
               hasMore: false,
             }
@@ -705,7 +738,7 @@ async function drive(
   );
 
   let listCallCount = 0;
-  const eventsList = vi.fn(async () => {
+  const eventsList = vi.fn(async (params?: { resolveData?: string }) => {
     listCallCount++;
     // The cursor is positioned at what the read actually showed, so a stale
     // read hands back a short log AND a cursor that still covers the events it
@@ -715,7 +748,7 @@ async function drive(
         ? events.slice(0, -1)
         : [...events];
     return {
-      data: visible,
+      data: served(visible, params),
       hasMore: false,
       cursor: nextCursor(visible.length),
     };
@@ -756,6 +789,7 @@ async function drive(
     vmBuilds: createContextSpy.mock.calls.length,
     durableLog: normalizeDurableLog(events),
     listCalls: eventsList.mock.calls.length,
+    listParams: eventsList.mock.calls.map(([params]) => params),
     queueSends: queueSend.mock.calls.length,
     createParams,
     createdHook: createdEvents.some((e) => e.eventType === 'hook_created'),
@@ -1602,3 +1636,78 @@ describe('retained VM through the inline replay loop', () => {
     });
   });
 });
+
+/**
+ * Replay recomputes every step's arguments by re-running workflow code, so a
+ * World may leave the recorded inputs out of the replay events it returns
+ * (`resolveData: 'skip-step-inputs'`). This loop passes growing state rebuilt from earlier step
+ * results — exactly the inputs that are omitted — so any replay path that
+ * still depended on a recorded input would hand a step the wrong state.
+ */
+describe('replay without recorded step inputs', () => {
+  const seenStates: unknown[] = [];
+  registerStepFunction(
+    'r_omit_append',
+    async (state: { items: number[] }, i: number) => {
+      seenStates.push(structuredClone(state));
+      return state.items.length * 10 + i;
+    }
+  );
+  const growingStateWorkflow = `const append = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("r_omit_append");
+  async function workflow() {
+    let state = { items: [] };
+    for (let i = 0; i < 3; i++) {
+      const next = await append(state, i);
+      state = { items: [...state.items, next] };
+    }
+    return state.items;
+  }
+  globalThis.__private_workflows = new Map([["workflow", workflow]]);`;
+
+  const expectedStates = [{ items: [] }, { items: [0] }, { items: [0, 11] }];
+
+  beforeEach(() => {
+    seenStates.length = 0;
+  });
+
+  it.each([
+    ['node:vm, inline deltas', {}, { type: 'normal' }],
+    ['node:vm, list reads', { withholdDelta: true }, { type: 'normal' }],
+    [
+      'node:vm, fresh replay after a 412',
+      {},
+      { type: 'fail-event', eventType: 'run_completed' },
+    ],
+  ] as Array<
+    [string, DriveWorldOptions, DriveMode]
+  >)('completes with the same step inputs (%s)', async (_label, worldOptions, mode) => {
+    const skipStepInputs = { stripped: 0 };
+    const { result, listParams, createParams, durableLog } = await drive(
+      `wrun_omit_${_label.replace(/\W+/g, '_')}`,
+      growingStateWorkflow,
+      mode,
+      { ...worldOptions, skipStepInputs }
+    );
+
+    expect(result).toEqual([0, 11, 22]);
+    expect(seenStates).toEqual(expectedStates);
+    // The World really served replay events without step inputs...
+    expect(skipStepInputs.stripped).toBeGreaterThan(0);
+    expect(listParams.every((p) => p?.resolveData === 'skip-step-inputs')).toBe(
+      true
+    );
+    // ...because every replay read, and every write that asks for replay
+    // events back (an inline delta or the run_started preload), asked for it.
+    expect(
+      createParams
+        .filter(
+          (p) => p.sinceCursor !== undefined || p.eventType === 'run_started'
+        )
+        .every((p) => p.resolveData === 'skip-step-inputs')
+    ).toBe(true);
+    // The recorded log itself still holds each step exactly once.
+    expect(
+      durableLog.filter((e) => e.eventType === 'step_completed')
+    ).toHaveLength(3);
+  });
+});

packages/core/src/runtime.ts

@@ -83,6 +83,7 @@ import {
   parseHealthCheckPayload,
   preconditionEventDelta,
   queueMessage,
+  REPLAY_RESOLVE_DATA,
   resolveRunEncryptionKey,
   rootRunIdFrom,
   runDispatchContext,
@@ -1030,7 +1031,14 @@ export function workflowEntrypoint(
                     params?: CreateEventParams
                   ) => {
                     const sinceCursor = deltaRequestCursor(data, params);
-                    const withSnapshot = { ...slotSnapshot(), ...params };
+                    // Replay events a write hands back (an inline delta or a
+                    // preload) only feed this log; read them the way replay
+                    // reads the log.
+                    const withSnapshot: CreateEventParams = {
+                      ...slotSnapshot(),
+                      resolveData: REPLAY_RESOLVE_DATA,
+                      ...params,
+                    };
                     const result = await replayRecoveryReporter.withEventCreate(
                       sinceCursor === undefined
                         ? withSnapshot

packages/core/src/runtime/helpers.test.ts

@@ -449,6 +449,7 @@ describe('loadWorkflowRunEvents', () => {
     expect(eventsListMock).toHaveBeenCalledWith({
       runId: 'wrun_test',
       pagination: { sortOrder: 'asc', cursor: undefined },
+      resolveData: 'skip-step-inputs',
     });
   });
 
@@ -576,10 +577,12 @@ describe('loadWorkflowRunEvents', () => {
     expect(eventsListMock).toHaveBeenNthCalledWith(1, {
       runId: 'wrun_test',
       pagination: { sortOrder: 'asc', cursor: 'opaque-cursor' },
+      resolveData: 'skip-step-inputs',
     });
     expect(eventsListMock).toHaveBeenNthCalledWith(2, {
       runId: 'wrun_test',
       pagination: { sortOrder: 'asc', cursor: undefined },
+      resolveData: 'skip-step-inputs',
     });
   });
 

packages/core/src/runtime/helpers.ts

@@ -585,6 +585,29 @@ function shouldRetryWithoutEventCursor(
   );
 }
 
+/**
+ * How replay reads the event log. Replay recomputes every step's arguments by
+ * re-running workflow code and never reads the recorded ones (a step reads its
+ * input from the step entity), so replay reads let the World leave them out.
+ * For a workflow passing growing state into its steps, they are the part of
+ * the log that grows quadratically.
+ */
+export const REPLAY_RESOLVE_DATA = 'skip-step-inputs' as const;
+
+/**
+ * Params for a write whose inline delta (`sinceCursor`) is folded into a
+ * replay log: that delta is read the way replay reads the log. The created
+ * event and `step` entity are unaffected, so a write that asks for no delta
+ * is left exactly as it was.
+ */
+export function withReplayDelta(
+  params: CreateEventParams | undefined
+): CreateEventParams | undefined {
+  return params?.sinceCursor === undefined
+    ? params
+    : { resolveData: REPLAY_RESOLVE_DATA, ...params };
+}
+
 /**
  * Loads workflow run events by iterating through all pages of paginated
  * results. Events are returned in chronological (ascending) order for
@@ -634,6 +657,7 @@ export async function loadWorkflowRunEvents(
             sortOrder: 'asc',
             cursor: requestedCursor ?? undefined,
           },
+          resolveData: REPLAY_RESOLVE_DATA,
         });
       } catch (error) {
         if (

packages/core/src/runtime/quickjs-entrypoint.preload.test.ts

@@ -184,6 +184,13 @@ describe('QuickJS lazy hook preload sourcing', () => {
     expect(listEvents).toHaveBeenCalled();
     // ...and replayed what the fetch returned.
     expect(vmEvents.map((e) => e.eventId)).toEqual(log.map((e) => e.eventId));
+    // Replay never reads recorded step inputs, so every fetch lets the World
+    // leave them out.
+    for (const [params] of listEvents.mock.calls as unknown as Array<
+      [{ resolveData?: string }]
+    >) {
+      expect(params.resolveData).toBe('skip-step-inputs');
+    }
   });
 
   it('does not trust an attested but empty preload: fetches via events.list', async () => {

packages/core/src/runtime/quickjs-entrypoint.ts

@@ -61,6 +61,7 @@ import { getPortLazy } from './get-port-lazy.js';
 import {
   getWorkflowQueueName,
   queueMessage,
+  REPLAY_RESOLVE_DATA,
   runDispatchContext,
   stepDispatchIdempotencyKey,
 } from './helpers.js';
@@ -1121,6 +1122,7 @@ export async function runWorkflowWithQuickJS(params: {
           cursor: cursor ?? undefined,
           limit: 1000,
         },
+        resolveData: REPLAY_RESOLVE_DATA,
       });
       eventsFetchedPages++;
       allEvents.push(...response.data);
@@ -1151,6 +1153,9 @@ export async function runWorkflowWithQuickJS(params: {
   const logView = new QuickJSLogView(events, loadedCursor);
   const createEvent: EventCreator = async (data, eventParams) => {
     const result = await world.events.create(runId, data, {
+      // Returned replay events only feed the log; read them the way replay
+      // reads the log.
+      resolveData: REPLAY_RESOLVE_DATA,
       ...eventParams,
       ...logView.snapshotParams(),
     });
@@ -1446,6 +1451,7 @@ export async function runWorkflowWithQuickJS(params: {
           cursor: cursor ?? undefined,
           limit: 1000,
         },
+        resolveData: REPLAY_RESOLVE_DATA,
       });
       for (const e of response.data) {
         if (e.eventId && seenEventIds.has(e.eventId)) continue;

packages/core/src/runtime/quickjs-partial-preload.test.ts

@@ -116,6 +116,7 @@ describe('QuickJS partial run_started preload', () => {
     expect(listEvents).toHaveBeenCalledWith({
       runId,
       pagination: { sortOrder: 'asc', cursor: preloadCursor },
+      resolveData: 'skip-step-inputs',
     });
     expect(runWorkflowWithQuickJS).toHaveBeenCalledWith(
       expect.objectContaining({

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

@@ -56,7 +56,7 @@ import {
   isOptimisticInlineStartExplicitlyDisabled,
 } from './constants.js';
 import { getPortLazy } from './get-port-lazy.js';
-import { memoizeEncryptionKey } from './helpers.js';
+import { memoizeEncryptionKey, withReplayDelta } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
 import {
   computeResumeTtrAttributes,
@@ -393,7 +393,7 @@ export async function executeStep(
     data: T,
     eventParams?: CreateEventParams
   ) =>
-    replayRecoveryReporter.withEventCreate(eventParams, (p) =>
+    replayRecoveryReporter.withEventCreate(withReplayDelta(eventParams), (p) =>
       world.events.create(workflowRunId, data, p)
     );
 

packages/world-local/src/storage/events-storage.ts

@@ -17,11 +17,11 @@ import type {
   CreateEventRequest,
   Event,
   EventResult,
+  EventsResolveData,
   Hook,
   HookCreatedEventRequest,
   PaginatedResponse,
   PaginationOptions,
-  ResolveData,
   SerializedData,
   Step,
   Storage,
@@ -30,6 +30,7 @@ import type {
 } from '@workflow/world';
 import {
   applyAttributeChanges,
+  entityResolveData,
   eventIdToSlot,
   FIRST_EVENT_SLOT,
   getMaxEventsPerRun,
@@ -1252,7 +1253,7 @@ export function createEventsStorage(
               effectiveRunId,
               data,
               currentRun,
-              params
+              params && { resolveData: entityResolveData(params.resolveData) }
             );
           }
         }
@@ -3478,7 +3479,7 @@ export function createEventsStorage(
   async function reportSkippedSlots(
     result: EventResult,
     askedFor: number,
-    resolveData: ResolveData
+    resolveData: EventsResolveData
   ): Promise<EventResult> {
     if (!result.event) {
       return result;