perf(core): lazy inline step start (save one world round-trip per step) (#2478)

* perf(core): lazy inline step start to save a world round-trip per step

The owned-inline runtime path used to write step_created (suspension
handler) and then step_started (executeStep) as two separate world
round-trips for a step it already owns and is about to run inline. This
defers the step_created write: executeStep sends a single step_started
carrying the step input, and the world creates the step on the fly
(materializing the step entity plus a synthetic step_created event so
replay still observes it). Mirrors the existing resilient run_started ->
run_created pattern.

Exactly-one ownership is preserved by the world's atomic create-claim:
the loser of a concurrent lazy step_started gets EntityConflictError,
which executeStep maps to `skipped`, so it never runs the body. A lazy
step_started is only ever sent for a brand-new step (the suspension
handler defers only steps with no prior step_created), so crash recovery
still re-runs a `running` step via the normal non-lazy step_started.

Worlds updated: world-local, world-postgres (implicit create + synthetic
step_created event), world-vercel (routes the input as the v4 frame
payload and threads the server's stepCreated flag). @workflow/world adds
optional `input` to step_started and a `stepCreated` EventResult signal.

Rollout: server-first. The matching workflow-server change must deploy
before this ships; the Vercel world targets a single Vercel-operated
backend (server always >= SDK). For local/postgres the world ships in the
same package as the runtime, so there is no version skew.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(core): materialize deferred step before failing unregistered step on lazy inline path

The lazy inline step-start optimization defers a step's step_created write,
expecting executeStep to materialize the step via a lazy step_started carrying
its input. For an UNREGISTERED step, executeStep bails out before sending that
step_started and writes step_failed directly — but the step entity was never
created, so the world's "step must exist" ordering guard rejects the
step_failed and the run wedges (times out).

This regressed the StepNotRegisteredError e2e tests uniformly across every
framework/world (the ghost step never reached `failed`). Fix: on the lazy path,
send the lazy step_started first to materialize the step (entity + synthetic
step_created, keeping replay correct), then write step_failed. The lazy
step_started's atomic create-claim preserves exactly-one-owner: a concurrent
winner makes ours reject with EntityConflictError → skipped, so the failure is
never written twice.

Adds world-level regression tests (world-local, world-postgres) asserting a
lazy step_started followed by step_failed marks the step failed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: pranaygp

.changeset/lazy-inline-step-start.md

@@ -0,0 +1,18 @@
+---
+'workflow': patch
+'@workflow/core': patch
+'@workflow/world': patch
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+'@workflow/world-vercel': patch
+---
+
+Lazy inline step start: the owned-inline runtime path now sends a single `step_started` carrying the step input, letting the world create the step on the fly and saving one round-trip per inline step.
+
+`@workflow/world`: `step_started` event data accepts an optional `input`, and `EventResult` gains a `stepCreated` ownership signal.
+
+`@workflow/world-local`: `step_started` with input atomically creates the step plus a synthetic `step_created` event; a lazy `step_started` for an already-existing step throws `EntityConflictError` so concurrent losers skip (exactly-once).
+
+`@workflow/world-postgres`: same lazy-create + exactly-once create-claim for the Postgres backend.
+
+`@workflow/world-vercel`: sends the step input on `step_started` over the v4 wire and threads the server's `stepCreated` signal into `EventResult`.

packages/core/src/runtime.test.ts

@@ -1,14 +1,11 @@
-import {
-  EntityConflictError,
-  RUN_ERROR_CODES,
-  WorkflowWorldError,
-} from '@workflow/errors';
+import { RUN_ERROR_CODES, WorkflowWorldError } from '@workflow/errors';
 import {
   type Event,
   SPEC_VERSION_CURRENT,
   type WorkflowRun,
 } from '@workflow/world';
 import { afterEach, describe, expect, it, vi } from 'vitest';
+import { registerStepFunction } from './private.js';
 import { REPLAY_DIVERGENCE_MAX_RETRIES } from './runtime/constants.js';
 import { setWorld } from './runtime/world.js';
 import { workflowEntrypoint } from './runtime.js';
@@ -629,7 +626,13 @@ describe('workflowEntrypoint replay guards', () => {
     expect(firstAttemptEvents).toContainEqual(
       expect.objectContaining({ eventType: 'attr_set' })
     );
-    expect(firstAttemptEvents).toContainEqual(
+    // Under lazy inline start the step that loses the attribute race is NOT
+    // eagerly created: its step_created is deferred for a lazy step_started
+    // that never fires, because the attr_set event triggers an immediate
+    // re-invocation before any step executes. So no step_created is written on
+    // this attempt — strictly less event-log garbage than the eager model,
+    // and correct because the step loses the race and is abandoned on replay.
+    expect(firstAttemptEvents).not.toContainEqual(
       expect.objectContaining({ eventType: 'step_created' })
     );
     expect(firstAttemptMessages).toEqual([]);
@@ -774,7 +777,7 @@ describe('workflowEntrypoint replay guards', () => {
     );
   });
 
-  it('propagates transient step_created failures to the queue handler without an unhandled rejection', async () => {
+  it('propagates transient step-creation failures (lazy step_started) to the queue handler without an unhandled rejection', async () => {
     const createdEvents: unknown[] = [];
     const workflowRun: WorkflowRun = {
       runId: 'wrun_step_created_parse',
@@ -792,7 +795,10 @@ describe('workflowEntrypoint replay guards', () => {
       deploymentId: 'test-deployment',
     };
     // Simulates a transient network failure on POST /runs/{id}/events
-    // (e.g. the connection terminated mid-response-body).
+    // (e.g. the connection terminated mid-response-body). Under lazy inline
+    // start the step is created on the fly by its step_started, so the
+    // transient failure surfaces there (the standalone step_created round-trip
+    // no longer exists on this path).
     const parseError = new WorkflowWorldError(
       'Failed to parse response body for POST /v3/runs/wrun_step_created_parse/events (Content-Type: application/cbor):\n\nTypeError: terminated',
       { code: 'PARSE_ERROR' }
@@ -801,7 +807,7 @@ describe('workflowEntrypoint replay guards', () => {
       if (data.eventType === 'run_started') {
         return { run: workflowRun, events: [] };
       }
-      if (data.eventType === 'step_created') {
+      if (data.eventType === 'step_started') {
         throw parseError;
       }
       createdEvents.push(data);
@@ -893,21 +899,28 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
     `;globalThis.__private_workflows = new Map();
     globalThis.__private_workflows.set(${JSON.stringify(workflowName)}, ${workflowName});`;
 
-  // A workflow that suspends on a step AND a sleep. The harness's
-  // events.create answers step_created with EntityConflictError (see
-  // driveHandler), so this handler observes the pending step without owning
-  // it — the crash-recovery shape where a prior handler wrote step_created
-  // but died before queueing. That forces the unified dispatch to QUEUE the
-  // step (inline execution is ownership-gated), exercising the
-  // progress-critical step-dispatch queue() send that must complete before
-  // the orchestrator message is acked.
+  // A workflow that suspends on TWO parallel steps and a sleep. Under the
+  // lazy-inline-start model exactly one pending step is run inline (its
+  // step_created is deferred and folded into a lazy step_started); every other
+  // pending step keeps its eager step_created and is QUEUED via the unified
+  // dispatch. So the second step here is always queued, exercising the
+  // progress-critical step-dispatch queue() send that must complete before the
+  // orchestrator message is acked — independent of which step the runtime
+  // happens to pick for inline execution.
   const stepWithSleepWorkflow = `const add = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("add");
+    const addB = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("addB");
     const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")];
     async function workflow() {
-      const [a] = await Promise.all([add(1, 2), sleep('1h')]);
+      const [a] = await Promise.all([add(1, 2), addB(3, 4), sleep('1h')]);
       return a;
     }${getWorkflowTransformCode('workflow')}`;
 
+  // Register the two steps so the one chosen for inline execution actually
+  // runs (and completes) instead of failing as unregistered; the other is
+  // queued. Both are no-ops — these tests only assert dispatch/ack ordering.
+  registerStepFunction('add', async () => undefined);
+  registerStepFunction('addB', async () => undefined);
+
   async function makeRunningRun(runId: string): Promise<WorkflowRun> {
     return {
       runId,
@@ -946,25 +959,68 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
     // `.only`, not just the afterEach reset between this suite's tests.
     waitUntilPromises.length = 0;
 
+    // Stateful event log so replay converges instead of re-suspending forever:
+    // the inline step's events and the queued step's eager step_created are
+    // recorded here and returned by `list`, so a later loop iteration observes
+    // the inline step as done and the queued step as already-created (and thus
+    // not re-run/re-inlined).
+    let eventSeq = 0;
+    const durableEvents: Event[] = [];
+    const recordEvent = (data: any): Event => {
+      eventSeq += 1;
+      const created = {
+        eventId: `event-${eventSeq}`,
+        runId: workflowRun.runId,
+        createdAt: new Date(),
+        ...data,
+      } as Event;
+      durableEvents.push(created);
+      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');
-        // Concurrent-handler simulation: the step_created event already
-        // exists, so this handler doesn't own the step and must queue it
-        // rather than execute it inline.
-        throw new EntityConflictError('step already exists');
+        return { event: recordEvent(data) };
       }
-      return {
-        event: {
-          eventId: `event-${order.length}`,
-          runId: workflowRun.runId,
-          createdAt: new Date(),
-          ...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) };
     });
 
     const queue = vi.fn(async (queueName: string, message: any) => {
@@ -1004,8 +1060,12 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
       ),
       events: {
         create: eventsCreate,
+        // Return the accumulated event log so replay converges: a later loop
+        // iteration sees the inline step completed and the queued step already
+        // created (so neither is re-run), and the handler returns instead of
+        // re-suspending forever.
         list: vi.fn(async () => ({
-          data: [] as Event[],
+          data: [...durableEvents],
           hasMore: false,
           cursor: 'cursor_test',
         })),

packages/core/src/runtime.ts

@@ -1245,41 +1245,42 @@ export function workflowEntrypoint(
 
                         const pendingSteps = suspensionResult.pendingSteps;
 
-                        // Inline execution is gated on ownership: only the
-                        // handler that actually wrote the step_created event
-                        // may run the step body inline. The world-level
-                        // step_created is atomic per-correlationId, so
-                        // exactly one handler owns each step — concurrent
-                        // handlers can't race on step execution.
-                        const ownedPendingSteps = pendingSteps.filter((s) =>
-                          suspensionResult.createdStepCorrelationIds.has(
-                            s.correlationId
-                          )
-                        );
-
-                        // Pick one owned step to execute inline (if any).
-                        // The rest of the pending steps, plus any wait
-                        // timer, are queued below in a single parallel
-                        // batch.
+                        // Inline execution is gated on ownership. The
+                        // suspension handler deferred the step_created write for
+                        // exactly one step (`lazyInlineStep`) so we can run it
+                        // inline via a lazy `step_started` that creates the step
+                        // on the fly — saving one world round-trip. Ownership is
+                        // still atomic and exactly-one: the world's
+                        // create-claim inside that step_started returns
+                        // `EntityConflictError` (→ executeStep `skipped`) to any
+                        // concurrent loser, so only one handler ever runs the
+                        // body. Every other pending step keeps its eager
+                        // step_created (in `createdStepCorrelationIds`) and is
+                        // queued below.
                         //
-                        // Skip inline execution when a created hook has a
-                        // `hook.getConflict()` awaiter: an inline `await
-                        // executeStep(...)` blocks this handler for the
-                        // full step duration, so the awaiter's continuation
-                        // (which only advances on the next replay) would be
-                        // serialized behind the step — defeating work the
-                        // workflow expressed as parallel (e.g.
-                        // `hook.getConflict().then(() => stepB())` racing
-                        // `await stepA()`). Queue every step and re-invoke
-                        // immediately instead: the re-invocation replays
-                        // over the just-committed hook_created and resolves
-                        // the awaiter while the queued steps execute in
+                        // The suspension handler only designates a
+                        // `lazyInlineStep` when no `hook.getConflict()` awaiter
+                        // is present. That awaiter case must execute nothing
+                        // inline: an inline `await executeStep(...)` blocks this
+                        // handler for the full step duration, so the awaiter's
+                        // continuation (which only advances on the next replay)
+                        // would be serialized behind the step — defeating work
+                        // the workflow expressed as parallel (e.g.
+                        // `hook.getConflict().then(() => stepB())` racing `await
+                        // stepA()`). In that case `lazyInlineStep` is undefined
+                        // and every step is queued for re-invocation, which
+                        // replays over the just-committed hook_created and
+                        // resolves the awaiter while queued steps run in
                         // parallel invocations.
+                        const lazyInline = suspensionResult.lazyInlineStep;
                         const inlineStep:
                           | (typeof pendingSteps)[number]
-                          | undefined = suspensionResult.hasAwaitedHookCreation
-                          ? undefined
-                          : ownedPendingSteps[0];
+                          | undefined = lazyInline
+                          ? pendingSteps.find(
+                              (s) =>
+                                s.correlationId === lazyInline.correlationId
+                            )
+                          : undefined;
 
                         // Unified queue dispatch for everything we are NOT
                         // inline-executing. Steps are queued with stepId so
@@ -1389,9 +1390,11 @@ export function workflowEntrypoint(
                         //    (`preInlineWriteCursor`; a World may return none on
                         //    the initial load).
                         //  - This is the clean single-step sequential case:
-                        //    this suspension created exactly one step and no
-                        //    hooks or waits (`err.{step,hook,wait}Count`), and
-                        //    that one pending step is owned by us (no parallel
+                        //    this suspension produced exactly one step and no
+                        //    hooks or waits (`err.{step,hook,wait}Count`), that
+                        //    one step is the lone pending step
+                        //    (`pendingSteps.length === 1`), and it is the one we
+                        //    are running inline (`inlineStep` — no parallel
                         //    siblings queued to background handlers, which would
                         //    write their own events out of band).
                         //  - No pending wait timer from THIS suspension.
@@ -1413,7 +1416,7 @@ export function workflowEntrypoint(
                           err.hookCount === 0 &&
                           err.waitCount === 0 &&
                           pendingSteps.length === 1 &&
-                          ownedPendingSteps.length === 1 &&
+                          inlineStep !== undefined &&
                           !suspensionResult.waitTimeout &&
                           !hasOpenHookOrWait(cachedEvents ?? []);
 
@@ -1429,6 +1432,11 @@ export function workflowEntrypoint(
                             stepId: inlineStep.correlationId,
                             stepName: inlineStep.stepName,
                             runSpecVersion: workflowRun.specVersion,
+                            // Lazy inline start: send the deferred step's input
+                            // on step_started so the world creates the step on
+                            // the fly. lazyInline is defined whenever inlineStep
+                            // is (both derive from suspensionResult.lazyInlineStep).
+                            lazyStepInput: lazyInline?.dehydratedInput,
                             ...(requestInlineDelta && preInlineWriteCursor
                               ? { inlineDeltaSinceCursor: preInlineWriteCursor }
                               : {}),