Address review: terminal-only lazy status gate, terminal-run start fence in local worlds, prologue telemetry, last-completer coverage

- The last completer's lazy runs.get result is now gated on
  isTerminalWorkflowRunStatus (with a debug log): a stale 'pending' read
  — a run with completed steps has necessarily started — no longer
  silently abandons the fan-out's continuation; it falls through to the
  inline replay, whose next entity write is fenced server-side if the
  run truly ended meanwhile.
- world-local / world-postgres now reject step_started on terminal runs
  even when the step row still reads 'running' (a redelivered start a
  previous delivery claimed): starting work on a finished run is never
  valid, and previously the body re-ran with its outcome unconsumable.
  In-flight steps still write their terminal events unchanged. This
  closes the adapter gap behind the fetch-free prologue's reliance on
  the step_started claim as the run-liveness check, and the prologue
  comment now states the contract precisely.
- workflow.step.dispatch_prologue span attribute ('run_context' |
  'runs_get') makes fetch-free adoption and the saved round trip
  observable during version-skew windows.
- Restated why the eager redelivery re-ensure survives on the
  fetch-free path (no run fetch to overlap; still cheaper than the
  in-band recovery's failed-start round trip).
- New two-phase fan-out coverage: a real replay emits the queued step
  message (asserting the stamped runContext), then its redelivery runs
  as the LAST completer — zero reads before the step, exactly one lazy
  runs.get, run completed; plus the stale-'pending' fall-through and
  the genuinely-terminal skip.

Author: TooTallNate

.changeset/terminal-run-start-fence.md

@@ -0,0 +1,6 @@
+---
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+---
+
+Reject `step_started` on terminal runs even when the step row still reads `running`: a redelivered start on a cancelled/completed run previously passed the claim and re-executed the step body whose outcome nothing would consume. In-flight steps can still write their terminal events (`step_completed`/`step_failed`) unchanged.

packages/core/src/runtime.test.ts

@@ -2194,6 +2194,231 @@ describe('workflowEntrypoint resilient step consumption (stepInput re-ensure)',
   });
 });
 
+// The fan-out path this PR optimizes end to end: phase 1 drives a REAL replay
+// of a two-step Promise.all fan-out (inline cap 1, so the runtime itself
+// emits the queued step message with its seeded correlation id and stamped
+// runContext); phase 2 redelivers that exact message against the shared event
+// log, making it the LAST completer — exercising the fetch-free prologue, the
+// single lazy runs.get before the inline replay, and the terminal-only status
+// gate on its result.
+describe('workflowEntrypoint fan-out last completer (fetch-free prologue)', () => {
+  beforeEach(() => {
+    process.env.WORKFLOW_MAX_INLINE_STEPS = '1';
+  });
+  afterEach(() => {
+    delete process.env.WORKFLOW_MAX_INLINE_STEPS;
+    setWorld(undefined);
+    vi.clearAllMocks();
+  });
+
+  const getWorkflowTransformCode = (workflowName: string) =>
+    `;globalThis.__private_workflows = new Map();
+    globalThis.__private_workflows.set(${JSON.stringify(workflowName)}, ${workflowName});`;
+
+  const fanOutWorkflow = `const fanA = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("fanA");
+    const fanB = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("fanB");
+    async function workflow() {
+      const [a, b] = await Promise.all([fanA(1, 2), fanB(3, 4)]);
+      return a + b;
+    }${getWorkflowTransformCode('workflow')}`;
+
+  registerStepFunction('fanA', async (a: number, b: number) => a + b);
+  registerStepFunction('fanB', async (a: number, b: number) => a + b);
+
+  async function driveFanOut(opts: {
+    runId: string;
+    /** Status the LAST completer's lazy runs.get returns. */
+    lazyRunStatus: 'running' | 'pending' | 'completed';
+  }) {
+    const workflowRun: WorkflowRun = {
+      runId: opts.runId,
+      workflowName: 'workflow',
+      status: 'running',
+      specVersion: SPEC_VERSION_CURRENT,
+      input: await dehydrateWorkflowArguments([], opts.runId, undefined, []),
+      createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+      startedAt: new Date('2024-01-01T00:00:00.000Z'),
+      deploymentId: 'test-deployment',
+    };
+
+    // Shared durable state across the two deliveries.
+    let eventSeq = 0;
+    const durableEvents: Event[] = [
+      {
+        eventId: 'event-run-created',
+        runId: opts.runId,
+        createdAt: new Date('2024-01-01T00:00:00.000Z'),
+        eventType: 'run_created',
+        specVersion: SPEC_VERSION_CURRENT,
+        eventData: {
+          deploymentId: 'test-deployment',
+          workflowName: 'workflow',
+          input: workflowRun.input,
+        },
+      } as unknown as Event,
+    ];
+    const recordEvent = (data: any): Event => {
+      eventSeq += 1;
+      const created = {
+        eventId: `event-${eventSeq}`,
+        runId: opts.runId,
+        createdAt: new Date(),
+        ...data,
+      } as Event;
+      durableEvents.push(created);
+      return created;
+    };
+    // Step inputs recorded at creation, so a BARE start can return the entity
+    // with the input the body will hydrate (like a real World).
+    const stepInputsByCorrelationId = new Map<string, unknown>();
+
+    const eventsCreate = vi.fn(async (_runId: string, data: any) => {
+      if (data.eventType === 'run_started') {
+        return { run: workflowRun, events: [...durableEvents] };
+      }
+      if (data.eventType === 'step_created') {
+        stepInputsByCorrelationId.set(
+          data.correlationId,
+          data.eventData?.input
+        );
+        return { event: recordEvent(data) };
+      }
+      if (data.eventType === 'step_started') {
+        const lazy = data.eventData as { stepName?: string; input?: unknown };
+        if (lazy?.input !== undefined) {
+          // Lazy inline start: create the step on the fly.
+          stepInputsByCorrelationId.set(data.correlationId, lazy.input);
+          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: opts.runId,
+            stepId: data.correlationId,
+            stepName: lazy?.stepName,
+            status: 'running' as const,
+            attempt: 1,
+            input: stepInputsByCorrelationId.get(data.correlationId),
+            startedAt: new Date(),
+            createdAt: new Date(),
+            updatedAt: new Date(),
+          },
+          ...(lazy?.input !== undefined ? { stepCreated: true } : {}),
+        };
+      }
+      return { event: recordEvent(data) };
+    });
+
+    const queuedMessages: any[] = [];
+    const queue = vi.fn(async (_queueName: string, message: any) => {
+      queuedMessages.push(message);
+      return { messageId: null };
+    });
+    const runsGet = vi.fn(async () => ({
+      ...workflowRun,
+      status: opts.lazyRunStatus,
+    }));
+
+    // Capture the queue handler so both phases can deliver messages directly.
+    let handler!: (message: unknown, metadata: unknown) => Promise<unknown>;
+    setWorld({
+      specVersion: SPEC_VERSION_CURRENT,
+      createQueueHandler: vi.fn((_prefix: string, h: typeof handler) => {
+        handler = h;
+        return async () => new Response(null, { status: 204 });
+      }),
+      events: {
+        create: eventsCreate,
+        list: vi.fn(async () => ({
+          data: [...durableEvents],
+          hasMore: false,
+          cursor: 'cursor_test',
+        })),
+      },
+      runs: { get: runsGet },
+      queue,
+      getEncryptionKeyForRun: vi.fn(async () => undefined),
+    } as any);
+
+    const entry = workflowEntrypoint(fanOutWorkflow);
+    await entry(new Request('https://example.test')); // binds `handler`
+
+    // Phase 1: flow delivery. The real replay inline-executes one step (cap 1)
+    // and queues the other with its seeded correlation id and runContext.
+    await handler(
+      { runId: opts.runId, requestedAt: new Date() },
+      {
+        requestId: 'req_1',
+        attempt: 1,
+        queueName: '__wkf_workflow_workflow',
+        messageId: 'msg_flow_1',
+      }
+    );
+    const stepMessage = queuedMessages.find((m) => m && m.stepId);
+    expect(stepMessage).toBeDefined();
+    expect(runsGet).not.toHaveBeenCalled();
+
+    // Phase 2: redeliver the runtime's own step message — the last completer.
+    await handler(stepMessage, {
+      requestId: 'req_2',
+      attempt: 1,
+      queueName: '__wkf_workflow_workflow',
+      messageId: 'msg_step_1',
+    });
+
+    return { stepMessage, durableEvents, runsGet };
+  }
+
+  it('completes the run via the fetch-free prologue with exactly one lazy runs.get', async () => {
+    const { stepMessage, durableEvents, runsGet } = await driveFanOut({
+      runId: 'wrun_fanout_last_completer',
+      lazyRunStatus: 'running',
+    });
+
+    // The producer stamped the run identity on its own message.
+    expect(stepMessage.runContext).toMatchObject({
+      deploymentId: 'test-deployment',
+      specVersion: SPEC_VERSION_CURRENT,
+      rootRunId: 'wrun_fanout_last_completer',
+    });
+    // Zero reads before the step; exactly one for the inline replay.
+    expect(runsGet).toHaveBeenCalledTimes(1);
+    // The last completer replayed inline and finished the run.
+    expect(durableEvents.map((e) => e.eventType)).toContain('run_completed');
+  });
+
+  it('falls through a stale `pending` read instead of silently abandoning the fan-out', async () => {
+    const { durableEvents, runsGet } = await driveFanOut({
+      runId: 'wrun_fanout_stale_pending',
+      lazyRunStatus: 'pending',
+    });
+
+    // A run with completed steps cannot truly be pending — the read is stale.
+    // The terminal-only gate lets the replay proceed, so the continuation is
+    // not dropped on the floor.
+    expect(runsGet).toHaveBeenCalledTimes(1);
+    expect(durableEvents.map((e) => e.eventType)).toContain('run_completed');
+  });
+
+  it('still skips the inline replay when the run is genuinely terminal', async () => {
+    const { durableEvents } = await driveFanOut({
+      runId: 'wrun_fanout_terminal',
+      lazyRunStatus: 'completed',
+    });
+
+    expect(durableEvents.map((e) => e.eventType)).not.toContain(
+      'run_completed'
+    );
+  });
+});
+
 describe('workflowEntrypoint turbo mode', () => {
   const ORIG_TURBO = process.env.WORKFLOW_TURBO;
   const ORIG_OPT = process.env.WORKFLOW_OPTIMISTIC_INLINE_START;

packages/core/src/runtime.ts

@@ -27,6 +27,7 @@ import {
   getQueueTopicPrefix,
   isLegacySpecVersion,
   isTerminalRunEventType,
+  isTerminalWorkflowRunStatus,
   type RunInput,
   resolveQueueNamespace,
   SPEC_VERSION_CURRENT,
@@ -1415,11 +1416,14 @@ export function workflowEntrypoint(
                       //    ~300s visibility-timeout redelivery — measured
                       //    exactly so in the durabench parallel sweeps before
                       //    this path existed.
-                      //  - EAGERLY on a genuine redelivery (attempt > 1),
-                      //    in parallel with the run fetch below — a
+                      //  - EAGERLY on a genuine redelivery (attempt > 1) — a
                       //    redelivered dispatch already had its create race
-                      //    resolved either way, so this saves the failed
-                      //    start round-trip at no wall-time cost. First
+                      //    resolved either way, so ensuring up front saves the
+                      //    failed-start round trip the in-band recovery would
+                      //    otherwise pay. On the legacy prologue it overlaps
+                      //    the run fetch (no wall-time cost); on the
+                      //    fetch-free (runContext) prologue it is the sole
+                      //    pre-step write and still the cheaper trade. First
                       //    deliveries skip it: the producer's write almost
                       //    always lands, and an eager ensure would burn a
                       //    conditional write per step.
@@ -1492,18 +1496,37 @@ export function workflowEntrypoint(
                       // entirely — one less round trip on the TTLS-critical
                       // path, and N fewer reads on the run's partition per
                       // fan-out (vercel/workflow#3456). The run-status early
-                      // exit is not lost: a terminal run rejects the
-                      // `step_started` claim server-side (RunExpired → gone,
-                      // terminal step → skipped). Older messages without the
-                      // field keep the legacy fetch.
+                      // exit is not lost: every World rejects a `step_started`
+                      // claim on a terminal run (RunExpired → gone, terminal
+                      // step → skipped) — including a redelivered start whose
+                      // step row still reads `running`, which world-local and
+                      // world-postgres reject as of this change (previously
+                      // only world-vercel's run-status fence covered that
+                      // shape, and the body could re-run on a finished run).
+                      // Older messages without the field keep the legacy
+                      // fetch.
                       let bgRun: WorkflowRun | undefined;
                       let runIdentity: {
                         deploymentId: string;
                         specVersion: number;
                         startedAt?: number;
                         rootRunId?: string;
                       };
+                      // Which prologue ran — makes adoption of the fetch-free
+                      // path (and the round trip it saves) directly observable
+                      // during version-skew windows.
+                      span?.setAttributes(
+                        Attribute.StepDispatchPrologue(
+                          runContext ? 'run_context' : 'runs_get'
+                        )
+                      );
                       if (runContext) {
+                        // The eager redelivery re-ensure has no run fetch to
+                        // overlap with on this path — it is kept because a
+                        // redelivered dispatch has already had its create race
+                        // resolved, so one conditional write here is cheaper
+                        // than letting the bare start fail and paying the
+                        // in-band recovery's extra start round trip.
                         const ensureOutcome =
                           stepInput && metadata.attempt > 1
                             ? await ensureStepFromMessage()
@@ -1782,7 +1805,23 @@ export function workflowEntrypoint(
                           (await world.runs.get(runId, {
                             resolveData: 'none',
                           }));
-                        if (replayRunRow.status !== 'running') {
+                        // Terminal statuses only: a `pending` read here is a
+                        // stale row (this run has completed steps, so it has
+                        // started), and under the fetch-free prologue this is
+                        // the last completer's ONLY status read — returning on
+                        // it would silently abandon the fan-out's continuation
+                        // (the final step_completed is written, the inline
+                        // replay never runs). Fall through instead: the replay
+                        // synthesizes `running` and the next entity write is
+                        // fenced server-side if the run truly ended meanwhile.
+                        if (isTerminalWorkflowRunStatus(replayRunRow.status)) {
+                          runtimeLogger.debug(
+                            'Run already finished, skipping inline replay after background step',
+                            {
+                              workflowRunId: runId,
+                              status: replayRunRow.status,
+                            }
+                          );
                           return;
                         }
                         const runCreatedEvent = loaded.events.find(

packages/core/src/telemetry/semantic-conventions.ts

@@ -491,6 +491,17 @@ export const StepResilientDispatchMaterialized = SemanticConvention<boolean>(
   'workflow.step.resilient_dispatch_materialized'
 );
 
+/**
+ * How the queued-step consumer resolved the run identity for this execution:
+ * `run_context` — carried on the dispatch message, no `runs.get` before the
+ * step (the fetch-free prologue); `runs_get` — the legacy blocking fetch
+ * (message from an older producer). Distinguishes the two paths during
+ * version-skew windows and makes the saved round trip directly measurable.
+ */
+export const StepDispatchPrologue = SemanticConvention<
+  'run_context' | 'runs_get'
+>('workflow.step.dispatch_prologue');
+
 // Webhook attributes
 
 /** Number of webhook handlers triggered */

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

@@ -1305,11 +1305,21 @@ export function createEventsStorage(
               );
             }
 
-            // On terminal runs: only allow completing/failing in-progress steps
+            // On terminal runs: only allow completing/failing in-progress
+            // steps. A step_started is never that — it begins work, and no
+            // work should begin on a finished run — so it is rejected even
+            // when the step row still reads `running` (a redelivery of a
+            // start a previous delivery already claimed). Without this, a
+            // redelivered start on a cancelled/completed run passes the
+            // claim and executes the step body whose outcome nothing will
+            // ever consume.
             if (currentRun && isTerminalWorkflowRunStatus(currentRun.status)) {
-              if (validatedStep.status !== 'running') {
+              if (
+                validatedStep.status !== 'running' ||
+                data.eventType === 'step_started'
+              ) {
                 throw new RunExpiredError(
-                  `Cannot modify non-running step on run in terminal state "${currentRun.status}"`
+                  `Cannot ${data.eventType === 'step_started' ? 'start' : 'modify non-running'} step on run in terminal state "${currentRun.status}"`
                 );
               }
             }