Report RSFS/replay latency telemetry on step terminal events

.changeset/rsfs-replay-telemetry.md

@@ -0,0 +1,7 @@
+---
+'@workflow/core': minor
+'@workflow/world': minor
+'@workflow/world-vercel': minor
+---
+
+Report run-started-to-first-step (rsfs) and final-scheduling-replay (finalSchedulingReplay) latency telemetry on step completion events.

packages/core/src/runtime.test.ts

@@ -1908,6 +1908,14 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => {
       'optimisticStart',
     ]);
 
+    // RSFS/finalSchedulingReplay: under turbo, rsfsAnchorMs is stamped at local run
+    // synthesis (well after the backdated run-id timestamp), so unlike ttfs
+    // it stays within the test's own wall-clock budget.
+    expect(first.eventData.rsfs).toBeGreaterThanOrEqual(0);
+    expect(first.eventData.rsfs).toBeLessThanOrEqual(elapsed);
+    expect(first.eventData.finalSchedulingReplay).toBeGreaterThanOrEqual(0);
+    expect(first.eventData.finalSchedulingReplay).toBeLessThanOrEqual(elapsed);
+
     // Second step ran back-to-back with the first: STSO only, and far
     // smaller than the TTFS anchor distance (it measures the scheduling
     // gap, not run age).
@@ -1921,6 +1929,9 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => {
       'lazyStepStart',
       'optimisticStart',
     ]);
+    // STSO-only steps never qualify for RSFS (it shares TTFS eligibility).
+    expect(second.eventData.rsfs).toBeUndefined();
+    expect(second.eventData.finalSchedulingReplay).toBeUndefined();
   });
 
   it('still reports ttfs without turbo (redelivery), minus turbo-only optimization flags', async () => {
@@ -1934,6 +1945,10 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => {
     const [first] = stepCompleted;
     expect(first.eventData.ttfs).toBeGreaterThanOrEqual(backdateMs);
     expect(first.eventData.optimizations).toEqual(['lazyStepStart']);
+    // Non-turbo: rsfsAnchorMs is stamped right after the real (awaited)
+    // run_started response, so rsfs is a small non-negative duration too.
+    expect(first.eventData.rsfs).toBeGreaterThanOrEqual(0);
+    expect(first.eventData.finalSchedulingReplay).toBeGreaterThanOrEqual(0);
   });
 
   it('reports nothing when the first step is scheduled alongside a wait', async () => {
@@ -1944,6 +1959,8 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => {
     expect(stepCompleted).toHaveLength(1);
     expect(stepCompleted[0].eventData.ttfs).toBeUndefined();
     expect(stepCompleted[0].eventData.stso).toBeUndefined();
+    expect(stepCompleted[0].eventData.rsfs).toBeUndefined();
+    expect(stepCompleted[0].eventData.finalSchedulingReplay).toBeUndefined();
     expect(stepCompleted[0].eventData.optimizations).toBeUndefined();
   });
 

packages/core/src/runtime.ts

@@ -554,6 +554,14 @@ export function workflowEntrypoint(
                   // cannot be measured wall-clock, so TTFS is not reported.
                   // Set once, on the first iteration's loaded snapshot.
                   let invocationStartedClean: boolean | undefined;
+                  // Epoch ms the `run_started` response was received/parsed
+                  // by the SDK — anchors RSFS (run_started → first step's
+                  // start POST). Set once, in the run_started setup below.
+                  // Under turbo, run_started is backgrounded rather than
+                  // awaited, so this is stamped at the point the run is
+                  // synthesized locally instead of the real response — see
+                  // StepLatencyTracking.rsfsAnchorMs.
+                  let runStartedReceivedAtMs: number | undefined;
                   // Wall-clock ms spent committing hook_created events before
                   // the first step ran, accumulated across suspension passes
                   // and subtracted from TTFS.
@@ -964,6 +972,11 @@ export function workflowEntrypoint(
                         updatedAt: now,
                       };
                       workflowStartedAt = +now;
+                      // See the `runStartedReceivedAtMs` declaration above:
+                      // turbo synthesizes the run before the real
+                      // `run_started` response lands, so anchor RSFS here
+                      // rather than at an actual response instant.
+                      runStartedReceivedAtMs = +now;
                       span?.setAttributes({
                         ...Attribute.WorkflowRunStatus('running'),
                         ...Attribute.WorkflowStartedAt(workflowStartedAt),
@@ -982,6 +995,8 @@ export function workflowEntrypoint(
                           );
                         }
                         workflowRun = result.run;
+                        // Anchors RSFS — see the declaration above.
+                        runStartedReceivedAtMs = Date.now();
 
                         // If the response includes events, use them to skip
                         // the initial events.list call and reduce TTFB.
@@ -1434,10 +1449,36 @@ export function workflowEntrypoint(
                       return;
                     } catch (err) {
                       if (WorkflowSuspension.is(err)) {
+                        // Synchronous `runWorkflow` duration for THIS
+                        // suspension only — anchors the `finalSchedulingReplay`
+                        // telemetry field below (see
+                        // StepLatencyTracking.replayMs). This is the FINAL
+                        // replay pass, the one that reached and scheduled the
+                        // first step: valid rsfs paths can replay more than
+                        // once before the first step (e.g. a workflow-body
+                        // `setAttributes()` detour replays twice), and a
+                        // redelivery omits earlier invocations' replay work
+                        // entirely. This value is NOT accumulated across
+                        // those earlier passes, so it must not be read as
+                        // "the replay portion of rsfs" — rsfs covers the
+                        // whole run_started-to-first-step window;
+                        // finalSchedulingReplay covers only this last pass.
+                        // Captured here, before `handleSuspension`'s awaited
+                        // I/O, so it excludes that I/O.
+                        //
+                        // This duplicates what OTEL already captures on the
+                        // run/invocation span, but is collected as client
+                        // telemetry so the server can emit it as an
+                        // UNSAMPLED, full-population metric: workflow-server's
+                        // server spans are heavily sampled in production
+                        // (~7%), and client spans can't be filtered by SDK
+                        // version, so neither can serve as the dashboard's
+                        // exact TTFS decomposition.
+                        const replayDurationMs = Date.now() - replayStart;
                         runtimeLogger.debug('Workflow suspended', {
                           workflowRunId: runId,
                           loopIteration,
-                          replayMs: Date.now() - replayStart,
+                          replayMs: replayDurationMs,
                           steps: err.stepCount,
                           hooks: err.hookCount,
                           waits: err.waitCount,
@@ -1957,6 +1998,8 @@ export function workflowEntrypoint(
                           runCreatedAtMs:
                             runIdCreatedAt(runId) ??
                             (turbo ? undefined : +workflowRun.createdAt),
+                          runStartedReceivedAtMs,
+                          replayMs: replayDurationMs,
                           preStepBlockingMs,
                           preStepBlockingBeforeAttrMs,
                           // This suspension's own hook/wait writes are not in

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

@@ -393,6 +393,10 @@ export async function executeStep(
           !isOptimisticInlineStartExplicitlyDisabled()));
 
     let step: Step;
+    // `Date.now()` taken immediately before the `step_started` create is
+    // issued (either path below) — anchors RSFS's end point. See
+    // StepLatencyEventData.rsfs and the call sites below.
+    let stepStartPostSentAtMs: number | undefined;
     // Settled outcome of the in-flight optimistic `step_started`. Handlers are
     // attached synchronously (`.then(ok, err)`) so a fast rejection never
     // surfaces as an unhandledRejection while the body runs.
@@ -423,8 +427,12 @@ export async function executeStep(
       // round-trip overlaps the body rather than blocking it. Outside turbo the
       // barrier is undefined and this is a plain create.
       const startedPromise = (params.runReadyBarrier ?? Promise.resolve()).then(
-        () =>
-          world.events.create(workflowRunId, {
+        () => {
+          // Taken right before the create fires, not before the barrier —
+          // RSFS measures the run_started-to-POST stretch, and the barrier
+          // wait IS part of that stretch under turbo.
+          stepStartPostSentAtMs = Date.now();
+          return world.events.create(workflowRunId, {
             eventType: 'step_started',
             specVersion: SPEC_VERSION_CURRENT,
             correlationId: stepId,
@@ -437,7 +445,8 @@ export async function executeStep(
                 ? { ownerMessageId: params.ownerMessageId }
                 : {}),
             },
-          })
+          });
+        }
       );
       optimisticStartSettled = startedPromise.then(
         () => ({ ok: true as const }),
@@ -474,6 +483,7 @@ export async function executeStep(
           params.ownerMessageId !== undefined
             ? { ownerMessageId: params.ownerMessageId }
             : {};
+        stepStartPostSentAtMs = Date.now();
         const startResult = await world.events.create(workflowRunId, {
           eventType: 'step_started',
           specVersion: SPEC_VERSION_CURRENT,
@@ -596,6 +606,30 @@ export async function executeStep(
     // catch below) can attach it to step_failed too.
     let latencyEventData: StepLatencyEventData | undefined;
 
+    // Backfill RSFS onto the already-computed telemetry once the optimistic
+    // turbo start has settled. On that path the step-start POST fires inside
+    // the run-ready barrier's `.then`, so `stepStartPostSentAtMs` — and
+    // therefore RSFS — is usually still unset when `latencyEventData` is
+    // first computed just before user code (the barrier is still in flight
+    // for any non-trivial `run_started` round-trip, which is exactly the
+    // slow-run_started case RSFS exists to measure). By the time
+    // `reconcileOptimisticStart()` has awaited the barrier the POST timestamp
+    // is known, so we patch RSFS in before the terminal event is written.
+    // Without this, slow-run_started samples are dropped, biasing RSFS
+    // percentiles low (missing-not-at-random). Recomputes only RSFS —
+    // TTFS/STSO stay anchored to `executionStartTime` as computed above.
+    const backfillOptimisticRsfs = (): void => {
+      if (!latencyEventData || latencyEventData.rsfs !== undefined) return;
+      const anchorMs = params.latencyTracking?.rsfsAnchorMs;
+      if (anchorMs === undefined || stepStartPostSentAtMs === undefined) return;
+      latencyEventData.rsfs = Math.max(0, stepStartPostSentAtMs - anchorMs);
+      if (span) {
+        span.setAttributes({
+          ...Attribute.StepRsfsMs(latencyEventData.rsfs),
+        });
+      }
+    };
+
     try {
       const attempt = step.attempt;
 
@@ -656,17 +690,26 @@ export async function executeStep(
         attempt,
         lazyStepStart: params.lazyStepInput !== undefined,
         optimisticStart,
+        stepStartPostSentAtMs,
       });
       if (latencyEventData) {
         // Mirror the latency telemetry onto the step span so traces show
-        // TTFS/STSO alongside the flame graph, not just Datadog metrics.
+        // TTFS/STSO/RSFS alongside the flame graph, not just Datadog metrics.
         span?.setAttributes({
           ...(latencyEventData.ttfs !== undefined
             ? Attribute.StepTtfsMs(latencyEventData.ttfs)
             : {}),
           ...(latencyEventData.stso !== undefined
             ? Attribute.StepStsoMs(latencyEventData.stso)
             : {}),
+          ...(latencyEventData.rsfs !== undefined
+            ? Attribute.StepRsfsMs(latencyEventData.rsfs)
+            : {}),
+          ...(latencyEventData.finalSchedulingReplay !== undefined
+            ? Attribute.StepFinalSchedulingReplayMs(
+                latencyEventData.finalSchedulingReplay
+              )
+            : {}),
           ...Attribute.StepLatencyOptimizations(
             latencyEventData.optimizations ?? []
           ),
@@ -802,6 +845,9 @@ export async function executeStep(
       if (optimisticStart) {
         const reconcile = await reconcileOptimisticStart();
         if (reconcile) return reconcile;
+        // Barrier resolved — the step-start POST timestamp is now known, so
+        // RSFS can be attached to the step_completed event below.
+        backfillOptimisticRsfs();
       }
 
       // Commit must-be-durable ops (e.g. a step-initiated abort's
@@ -830,6 +876,8 @@ export async function executeStep(
       if (optimisticStart) {
         const reconcile = await reconcileOptimisticStart();
         if (reconcile) return reconcile;
+        // Barrier resolved — attach RSFS to the step_failed event(s) below.
+        backfillOptimisticRsfs();
       }
 
       // Order any must-be-durable ops (e.g. a step-initiated abort's

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

@@ -1468,4 +1468,63 @@ describe('executeStep optimistic inline start', () => {
     expect(calls).toContain('step_started');
     expect(calls.indexOf('body')).toBeLessThan(calls.indexOf('step_started'));
   });
+
+  it('still records rsfs on the terminal event when runReadyBarrier resolves AFTER the latency computation', async () => {
+    // Regression: on the optimistic turbo path the step-start POST fires
+    // inside the run-ready barrier's `.then`, so `stepStartPostSentAtMs` (the
+    // RSFS end anchor) is unset when `latencyEventData` is first computed just
+    // before user code. If the barrier is still in flight at that point — the
+    // common slow-`run_started` case RSFS exists to measure — RSFS would be
+    // dropped, biasing percentiles low. The terminal event must carry rsfs
+    // regardless of barrier timing.
+    delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START;
+    let release!: () => void;
+    const barrier = new Promise<void>((r) => {
+      release = r;
+    });
+    mockEventsCreate
+      .mockReset()
+      .mockImplementation((_runId: string, _event: { eventType: string }) =>
+        Promise.resolve({ event: {} })
+      );
+    mockStepFn.mockReset().mockResolvedValue('step-result');
+
+    // Anchor RSFS/TTFS in the past so both are computable and non-negative.
+    const anchorMs = Date.now() - 10;
+
+    const world = await getWorld();
+    const resultPromise = executeStep({
+      world: world as never,
+      ...baseParams,
+      forceOptimisticStart: true,
+      runReadyBarrier: barrier,
+      latencyTracking: {
+        ttfsAnchorMs: anchorMs,
+        rsfsAnchorMs: anchorMs,
+        replayMs: 1,
+        turbo: true,
+      },
+    });
+
+    // Let the body run and reach the latency computation while the barrier —
+    // and therefore the step-start POST — is still pending.
+    await vi.waitFor(() => expect(mockStepFn).toHaveBeenCalledTimes(1), {
+      timeout: 15_000,
+    });
+
+    release();
+    const result = await resultPromise;
+    expect(result.type).toBe('completed');
+
+    const completedWrite = mockEventsCreate.mock.calls.find(
+      ([, event]) =>
+        (event as { eventType: string }).eventType === 'step_completed'
+    );
+    expect(completedWrite).toBeDefined();
+    const eventData = (
+      completedWrite![1] as { eventData: { rsfs?: number } }
+    ).eventData;
+    expect(eventData.rsfs).toBeDefined();
+    expect(eventData.rsfs).toBeGreaterThanOrEqual(0);
+  });
 });