Stamp request provenance on step starts

.changeset/tame-oranges-lead.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Propagate invocation request IDs to step-start events.

packages/core/src/runtime.test.ts

@@ -1450,50 +1450,58 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
       return created;
     };
 
-    const eventsCreate = vi.fn(async (_runId: string, data: any) => {
-      if (data.eventType === 'run_started') {
-        return { run: workflowRun, events: [] as Event[] };
-      }
-      if (data.eventType === 'step_created') {
-        // Eager step_created for the QUEUED step (the one not run inline).
-        // It must be durably created before its dispatch send — the ordering
-        // assertion below checks step_created precedes queue_dispatch_start.
-        order.push('step_created');
-        return { event: recordEvent(data) };
-      }
-      if (data.eventType === 'step_started') {
-        // The inline step's lazy step_started creates the step on the fly:
-        // record a synthetic step_created so replay observes it, then the
-        // step_started, and return a running step so executeStep can run the
-        // (registered, no-op) body to completion.
-        const lazy = data.eventData as { stepName?: string; input?: unknown };
-        if (lazy?.input !== undefined) {
-          recordEvent({
-            eventType: 'step_created',
-            specVersion: SPEC_VERSION_CURRENT,
-            correlationId: data.correlationId,
-            eventData: { stepName: lazy.stepName, input: lazy.input },
-          });
+    const createdEventParams: any[] = [];
+    const stepStartedParams: any[] = [];
+    const eventsCreate = vi.fn(
+      async (_runId: string, data: any, params?: any) => {
+        createdEventParams.push(params);
+        if (data.eventType === 'step_started') {
+          stepStartedParams.push(params);
         }
-        const created = recordEvent(data);
-        return {
-          event: created,
-          step: {
-            runId: workflowRun.runId,
-            stepId: data.correlationId,
-            stepName: lazy?.stepName,
-            status: 'running' as const,
-            attempt: 1,
-            input: lazy?.input,
-            startedAt: new Date(),
-            createdAt: new Date(),
-            updatedAt: new Date(),
-          },
-          ...(lazy?.input !== undefined ? { stepCreated: true } : {}),
-        };
+        if (data.eventType === 'run_started') {
+          return { run: workflowRun, events: [] as Event[] };
+        }
+        if (data.eventType === 'step_created') {
+          // Eager step_created for the QUEUED step (the one not run inline).
+          // It must be durably created before its dispatch send — the ordering
+          // assertion below checks step_created precedes queue_dispatch_start.
+          order.push('step_created');
+          return { event: recordEvent(data) };
+        }
+        if (data.eventType === 'step_started') {
+          // The inline step's lazy step_started creates the step on the fly:
+          // record a synthetic step_created so replay observes it, then the
+          // step_started, and return a running step so executeStep can run the
+          // (registered, no-op) body to completion.
+          const lazy = data.eventData as { stepName?: string; input?: unknown };
+          if (lazy?.input !== undefined) {
+            recordEvent({
+              eventType: 'step_created',
+              specVersion: SPEC_VERSION_CURRENT,
+              correlationId: data.correlationId,
+              eventData: { stepName: lazy.stepName, input: lazy.input },
+            });
+          }
+          const created = recordEvent(data);
+          return {
+            event: created,
+            step: {
+              runId: workflowRun.runId,
+              stepId: data.correlationId,
+              stepName: lazy?.stepName,
+              status: 'running' as const,
+              attempt: 1,
+              input: lazy?.input,
+              startedAt: new Date(),
+              createdAt: new Date(),
+              updatedAt: new Date(),
+            },
+            ...(lazy?.input !== undefined ? { stepCreated: true } : {}),
+          };
+        }
+        return { event: recordEvent(data) };
       }
-      return { event: recordEvent(data) };
-    });
+    );
 
     const queue = vi.fn(async (queueName: string, message: any) => {
       // Only the step-dispatch send carries a stepId; ignore other sends.
@@ -1560,7 +1568,13 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
       }
     );
 
-    return { handlerPromise, order, queue };
+    return {
+      handlerPromise,
+      order,
+      queue,
+      createdEventParams,
+      stepStartedParams,
+    };
   }
 
   it('completes the step-dispatch send before the orchestrator message is acked', async () => {
@@ -1662,7 +1676,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
     // continuation is queued (it carries no stepId).
     process.env.WORKFLOW_MAX_INLINE_STEPS = '3';
 
-    const { handlerPromise, order } = await driveHandler({
+    const { handlerPromise, order, stepStartedParams } = await driveHandler({
       runId: 'wrun_multi_inline',
       queueImpl: async () => ({ messageId: null }),
     });
@@ -1673,6 +1687,10 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
     // No eager step_created and no step-dispatch send: both steps went inline.
     expect(order).not.toContain('step_created');
     expect(order).not.toContain('queue_dispatch_start');
+    expect(stepStartedParams).toHaveLength(2);
+    for (const params of stepStartedParams) {
+      expect(params).toMatchObject({ requestId: 'req_test' });
+    }
   });
 
   it('does not re-queue a throttled inline step as an input-less background step', async () => {
@@ -2000,6 +2018,13 @@ describe('workflowEntrypoint resilient step consumption (stepInput re-ensure)',
     );
     expect(createdIdx).toBeGreaterThanOrEqual(0);
     expect(createdIdx).toBeLessThan(startedIdx);
+    // The queued step start carries the queue invocation's request provenance.
+    const startIdx = createdEvents.findIndex(
+      (e) => e.eventType === 'step_started'
+    );
+    expect(createdEventParams[startIdx]).toMatchObject({
+      requestId: 'req_test',
+    });
     // The step body ran and its terminal event was written.
     expect(stepBodySpy).toHaveBeenCalledWith(2, 3);
     expect(createdEvents).toContainEqual(
@@ -2090,6 +2115,15 @@ describe('workflowEntrypoint resilient step consumption (stepInput re-ensure)',
     expect(createdEventParams[ensureIdx]).toMatchObject({
       viaStepDispatch: true,
     });
+    // Both the failed bare start and the recovery start retain the current
+    // invocation's provenance.
+    for (const [index, event] of createdEvents.entries()) {
+      if (event.eventType === 'step_started') {
+        expect(createdEventParams[index]).toMatchObject({
+          requestId: 'req_test',
+        });
+      }
+    }
   });
 
   it('recovers in-band on attempt 1 with the local-world error shape (no status)', async () => {

packages/core/src/runtime.ts

@@ -1674,6 +1674,7 @@ export function workflowEntrypoint(
                             workflowDeploymentId: bgRun.deploymentId,
                             workflowName,
                             workflowStartedAt: bgStartedAt,
+                            requestId,
                             rootRunId: rootRunIdFrom(bgRun.attributes, runId),
                             stepId: incomingStepId,
                             stepName: incomingStepName,
@@ -2696,6 +2697,7 @@ export function workflowEntrypoint(
                           // requeueing the step.
                           deliveryAttempt: metadata.attempt,
                           ownerMessageId: metadata.messageId,
+                          requestId,
                         });
                         if (quickjsResult?.timeoutSeconds !== undefined) {
                           // Use `reinvoke` rather than returning
@@ -3879,6 +3881,7 @@ export function workflowEntrypoint(
                                 workflowDeploymentId: workflowRun.deploymentId,
                                 workflowName,
                                 workflowStartedAt,
+                                requestId,
                                 rootRunId: rootRunIdFrom(
                                   workflowRun.attributes,
                                   runId

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

@@ -739,6 +739,8 @@ export async function runWorkflowWithQuickJS(params: {
    * the in-flight body instead of requeueing the step.
    */
   ownerMessageId?: string;
+  /** Request ID of the queue invocation, when the queue provides one. */
+  requestId?: string;
   /**
    * Queue namespace resolved at route registration (runtime.ts). Must be
    * threaded into every message publish: the builders bake the namespace
@@ -767,6 +769,7 @@ export async function runWorkflowWithQuickJS(params: {
     maxEventsLimit,
     deliveryAttempt,
     ownerMessageId,
+    requestId,
     namespace,
   } = params;
   // Standalone-caller fallback (tests): without a runtime.ts carrier
@@ -1436,6 +1439,7 @@ export async function runWorkflowWithQuickJS(params: {
                   workflowDeploymentId: workflowRun.deploymentId,
                   workflowName: workflowRun.workflowName,
                   workflowStartedAt,
+                  requestId,
                   rootRunId,
                   stepId: step.correlationId,
                   stepName: step.stepId,

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

@@ -28,8 +28,10 @@ async function setupRunningStep(opts: {
   world: World;
   stepName: string;
   onBody: () => void;
+  register?: boolean;
+  createStep?: boolean;
 }): Promise<{ runId: string; stepId: string }> {
-  const { world, stepName, onBody } = opts;
+  const { world, stepName, onBody, register = true, createStep = true } = opts;
   const runInput = await dehydrateStepArguments([], 'run', undefined);
   const created = await world.events.create(null, {
     eventType: 'run_created',
@@ -48,13 +50,15 @@ async function setupRunningStep(opts: {
   } as never);
 
   const stepId = 'step_timeout_1';
-  const stepInput = await dehydrateStepArguments([], runId, undefined);
-  await world.events.create(runId, {
-    eventType: 'step_created',
-    specVersion: SPEC_VERSION_CURRENT,
-    correlationId: stepId,
-    eventData: { stepName, input: stepInput },
-  });
+  if (createStep) {
+    const stepInput = await dehydrateStepArguments([], runId, undefined);
+    await world.events.create(runId, {
+      eventType: 'step_created',
+      specVersion: SPEC_VERSION_CURRENT,
+      correlationId: stepId,
+      eventData: { stepName, input: stepInput },
+    });
+  }
 
   const stepFn = Object.assign(
     async () => {
@@ -63,7 +67,9 @@ async function setupRunningStep(opts: {
     },
     { maxRetries: MAX_RETRIES }
   );
-  registerStepFunction(stepName, stepFn);
+  if (register) {
+    registerStepFunction(stepName, stepFn);
+  }
 
   return { runId, stepId };
 }
@@ -171,7 +177,7 @@ describe('executeStep — compute instance stamping', () => {
     counter += 1;
   });
 
-  it('stamps computeInstanceId on step_started without displacing the slot snapshot', async () => {
+  it('stamps request and compute provenance on step_started without displacing the slot snapshot', async () => {
     const world = makeWorld();
     const stepName = uniqueStepName();
     const { runId, stepId } = await setupRunningStep({
@@ -191,6 +197,7 @@ describe('executeStep — compute instance stamping', () => {
       workflowRunId: runId,
       workflowName: 'wf',
       workflowStartedAt: Date.now(),
+      requestId: 'req_step_executor',
       stepId,
       stepName,
       slotSnapshot,
@@ -200,11 +207,102 @@ describe('executeStep — compute instance stamping', () => {
       ([, data]) => data.eventType === 'step_started'
     );
     expect(started).toHaveLength(1);
-    expect(started[0]?.[2]?.computeInstanceId).toBe(COMPUTE_INSTANCE_ID);
-    // Both ride the same params object and neither may clobber the other.
+    expect(started[0]?.[2]).toMatchObject({
+      requestId: 'req_step_executor',
+      computeInstanceId: COMPUTE_INSTANCE_ID,
+    });
+    // All dimensions ride the same params object and neither may clobber another.
     expect(started[0]?.[2]?.eventCount).toBe(slotSnapshot.eventCount);
   });
 
+  it('stamps provenance when a lazy unregistered step is materialized', async () => {
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    const { runId, stepId } = await setupRunningStep({
+      world,
+      stepName,
+      onBody: () => {},
+      register: false,
+      createStep: false,
+    });
+    const input = await dehydrateStepArguments([], runId, undefined);
+    const createSpy = vi.spyOn(world.events, 'create');
+
+    const result = await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      requestId: 'req_unregistered',
+      stepId,
+      stepName,
+      lazyStepInput: input,
+    });
+
+    expect(result.type).toBe('failed');
+    const started = createSpy.mock.calls.find(
+      ([, data]) => data.eventType === 'step_started'
+    );
+    expect(started?.[2]).toMatchObject({
+      requestId: 'req_unregistered',
+      computeInstanceId: COMPUTE_INSTANCE_ID,
+    });
+  });
+
+  it('omits an empty requestId from step_started', async () => {
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    const { runId, stepId } = await setupRunningStep({
+      world,
+      stepName,
+      onBody: () => {},
+    });
+    const createSpy = vi.spyOn(world.events, 'create');
+
+    await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      requestId: '',
+      stepId,
+      stepName,
+    });
+
+    const started = createSpy.mock.calls.find(
+      ([, data]) => data.eventType === 'step_started'
+    );
+    expect(started?.[2]?.requestId).toBeUndefined();
+  });
+
+  it('omits requestId from step_started when unavailable', async () => {
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    const { runId, stepId } = await setupRunningStep({
+      world,
+      stepName,
+      onBody: () => {},
+    });
+    const createSpy = vi.spyOn(world.events, 'create');
+
+    await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId,
+      stepName,
+    });
+
+    const started = createSpy.mock.calls.find(
+      ([, data]) => data.eventType === 'step_started'
+    );
+    expect(started?.[2]).toMatchObject({
+      computeInstanceId: COMPUTE_INSTANCE_ID,
+    });
+    expect(started?.[2]?.requestId).toBeUndefined();
+  });
+
   it('advances the snapshot it sends as its own writes land', async () => {
     // The executor writes twice for one step. If the second write still named
     // the position its caller scheduled against, the World would report the