diff --git a/docs/content/docs/v5/changelog/batched-event-writes.mdx b/docs/content/docs/v5/changelog/batched-event-writes.mdx
index 83ba431cc1..b45dacf4f9 100644
--- a/docs/content/docs/v5/changelog/batched-event-writes.mdx
+++ b/docs/content/docs/v5/changelog/batched-event-writes.mdx
@@ -38,6 +38,8 @@ interface BatchEventRequest {
   event: CreateEventRequest;
   /** Client event time; under slot identity, the source of the durable createdAt. */
   occurredAt?: Date;
+  /** Per-event compute attribution, same as the single create's CreateEventParams. */
+  computeInstanceId?: string;
 }
 
 type BatchEventItemResult =
@@ -60,7 +62,11 @@ The contract:
 
 ## The runtime integration (suspension fan-out fold)
 
-**On by default.** The suspension handler folds a **clean fan-out** — the suspension's eager `step_created` and `wait_created` writes — into `createBatch` calls of at most 32 events (mirroring the server's transaction budgets). The fold only engages when the World implements `createBatch`, the run is on slot identity, and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch; everything else keeps the single-event path byte-for-byte. Lazy-inline steps keep deferring their `step_created` to the lazy start exactly as before.
+**On by default.** The suspension handler folds a **clean fan-out** — the suspension's eager `step_created` and `wait_created` writes — into `createBatch` calls of at most 32 events (mirroring the server's transaction budgets). Chunks of a larger fan-out commit **concurrently**: slot assignment is the World's, so parallel chunks race for slot ranges exactly like the pre-fold path's parallel single writes did, and per-entity conditions — not commit order — carry correctness. The fold only engages when the World implements `createBatch`, the run is on slot identity, and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch; everything else keeps the single-event path byte-for-byte.
+
+**Per-chunk continuation.** Each chunk's follow-on work starts the moment **that chunk** commits, not when the whole fold does: a chunk's step-execution queue messages publish right off its own commit (publish-after-create holds per step), and only the chunk carrying the inline pairs gates the replay's continuation — trailing chunks' commits and publishes are joined before the invocation can acknowledge its message, so the durability contract ("every create durable before ack") is unchanged.
+
+**Pre-claimed inline pairs.** When the fold engages and has company for them (at least two inline steps, or one plus other batchable events), the steps the runtime is about to execute inline join the batch as adjacent `[step_created, step_started]` pairs — the created row carrying the input, the started row a bare ownership-stamped claim the World folds into a born-running create. The inline bodies start straight off the pair chunk's commit (in parallel with the queue publishes and any trailing chunks) with no per-step claim POST at all, and a pair that loses its atomic create-claim to a concurrent delivery skips its body exactly as a lost lazy claim does. A lone inline step with nothing else to batch keeps the optimistic lazy-start path, whose claim overlaps the body.
 
 Per-event `409`s are tolerated the same way the single path tolerates `EntityConflictError` (a concurrent delivery already created the entity); any other per-event failure fails the suspension write the way a single-path rejection would.
 
diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts
index 0c9bc472d2..69604e62d5 100644
--- a/packages/core/src/runtime.ts
+++ b/packages/core/src/runtime.ts
@@ -100,6 +100,7 @@ import { runIdCreatedAt } from './runtime/run-id-time.js';
 import {
   DEFAULT_STEP_MAX_RETRIES,
   executeStep,
+  type PreclaimedInlineStart,
 } from './runtime/step-executor.js';
 import { computeStepLatencyTracking } from './runtime/step-latency.js';
 import {
@@ -3126,6 +3127,21 @@ export function workflowEntrypoint(
                               ),
                               getTraceCarrier: nextTraceCarrier,
                             },
+                            // Inline pre-claims: lets the batched fan-out
+                            // fold the lazy-inline steps' step_created +
+                            // step_started pairs — stamped with this
+                            // message's ownership — into the one commit, so
+                            // the bodies below start straight off it. See
+                            // SuspensionHandlerResult.inlineClaims.
+                            ownerMessageId: metadata.messageId,
+                            // Let the fold return once the pair chunk has
+                            // committed: trailing chunks and the per-chunk
+                            // step-message publishes ride
+                            // `deferredBatchWork`, which this invocation
+                            // joins before it can ack (below, next to the
+                            // dispatch join) — so the durability contract
+                            // is unchanged while the bodies start earlier.
+                            allowDeferredBatchWork: true,
                           });
                         } catch (suspensionError) {
                           // A suspension create was rejected as stale: re-derive
@@ -3586,7 +3602,21 @@ export function workflowEntrypoint(
                             )
                           );
                         }
-                        await Promise.all(dispatches);
+                        // The dispatch publishes and the inline bodies below
+                        // run CONCURRENTLY: the suspension commit already made
+                        // every dispatched step durable (and, when the fold
+                        // engaged, settled the inline pairs' claims), which is
+                        // the only ordering both sides need — so neither waits
+                        // for the other. The joins below (before step results
+                        // are read, and on the no-inline early returns) keep
+                        // the failure contract: a dispatch rejection still
+                        // fails this delivery, after in-flight bodies settle.
+                        const dispatchesSettled = Promise.all(dispatches);
+                        // A rejection must not surface as an unhandled
+                        // rejection while the bodies run (or if setup between
+                        // here and the join throws first); awaiting the
+                        // original promise below still observes it.
+                        dispatchesSettled.catch(() => {});
 
                         // The set of steps THIS invocation executes: the
                         // deferred lazy-inline batch plus any owned-recovery
@@ -3602,12 +3632,29 @@ export function workflowEntrypoint(
                           correlationId: string;
                           stepName: string;
                           lazyStepInput?: (typeof lazyInlineSteps)[number]['dehydratedInput'];
+                          preclaimedStart?: PreclaimedInlineStart;
                         }> = [
-                          ...lazyInlineSteps.map((s) => ({
-                            correlationId: s.correlationId,
-                            stepName: s.stepName,
-                            lazyStepInput: s.dehydratedInput,
-                          })),
+                          ...lazyInlineSteps.map((s) => {
+                            // Pre-claimed by the suspension batch: the pair
+                            // already settled this step's create + claim, so
+                            // the executor runs (or skips) the body off that
+                            // verdict instead of sending a lazy start with
+                            // the input.
+                            const claim = suspensionResult.inlineClaims.get(
+                              s.correlationId
+                            );
+                            return claim
+                              ? {
+                                  correlationId: s.correlationId,
+                                  stepName: s.stepName,
+                                  preclaimedStart: claim,
+                                }
+                              : {
+                                  correlationId: s.correlationId,
+                                  stepName: s.stepName,
+                                  lazyStepInput: s.dehydratedInput,
+                                };
+                          }),
                           ...ownedRecoverySteps.map((s) => ({
                             correlationId: s.correlationId,
                             stepName: s.stepName,
@@ -3652,6 +3699,14 @@ export function workflowEntrypoint(
                         // queued (or no work needs scheduling). Exit and let
                         // the queue drive subsequent replays.
                         if (inlineExecutions.length === 0) {
+                          // Nothing runs concurrently with the dispatches on
+                          // this path — join them (and the fold's deferred
+                          // chunk commits/publishes) here so a failure fails
+                          // the delivery exactly as it always has.
+                          await Promise.all([
+                            dispatchesSettled,
+                            suspensionResult.deferredBatchWork,
+                          ]);
                           // A `hook.getConflict()` awaiter needs an immediate
                           // re-invocation: the replay consumes the
                           // just-committed hook_created and resolves the
@@ -3848,9 +3903,24 @@ export function workflowEntrypoint(
                         // this is the view the scheduling decision was made
                         // against. The executor advances from it as its own
                         // writes land; see `slotSnapshot` in step-executor.
-                        const inlineClaimSnapshot = slotSnapshotParams(
+                        const loadedSlotSnapshot = slotSnapshotParams(
                           eventLog.events
                         );
+                        // The batched fan-out's own events are not in the
+                        // loaded log yet (the next iteration reloads), but
+                        // this invocation wrote them — fold the batch's
+                        // ceiling in, or every inline terminal write would
+                        // name a pre-batch position and be answered with a
+                        // skipped-slot report echoing the events this
+                        // suspension just committed.
+                        const batchSlotCeiling =
+                          suspensionResult.batchCommittedSlotCeiling;
+                        const inlineClaimSnapshot =
+                          batchSlotCeiling !== undefined &&
+                          batchSlotCeiling >
+                            (loadedSlotSnapshot.eventCount ?? 0)
+                            ? { eventCount: batchSlotCeiling }
+                            : loadedSlotSnapshot;
 
                         // TTR: consumed by this batch. Every step is handed
                         // the SAME tracking object and its one-shot
@@ -3909,7 +3979,8 @@ export function workflowEntrypoint(
                                 // every inline step (which would be O(n²)
                                 // across a long sequential workflow).
                                 authoritativeAttempt:
-                                  s.lazyStepInput !== undefined
+                                  s.lazyStepInput !== undefined ||
+                                  s.preclaimedStart !== undefined
                                     ? 1
                                     : countStepStartedEvents(
                                         eventLog.events,
@@ -3923,8 +3994,15 @@ export function workflowEntrypoint(
                                 // input on step_started so the world creates
                                 // the step on the fly. Absent for
                                 // owned-recovery steps, whose input hydrates
-                                // from the existing step entity.
+                                // from the existing step entity, and for
+                                // pre-claimed steps, whose pair already
+                                // carried it.
                                 lazyStepInput: s.lazyStepInput,
+                                // Pre-claimed inline start: the suspension
+                                // batch settled this step's create + claim;
+                                // the executor runs (or skips) the body off
+                                // that verdict with no start write of its own.
+                                preclaimedStart: s.preclaimedStart,
                                 // Inline ownership: stamp (or re-stamp) this
                                 // invocation's queue message ID on the
                                 // step_started, so wake replays see the body
@@ -3945,7 +4023,8 @@ export function workflowEntrypoint(
                                 runReadyBarrier,
                                 slotSnapshot: inlineClaimSnapshot,
                                 ...(stepIndex === 0 &&
-                                s.lazyStepInput !== undefined &&
+                                (s.lazyStepInput !== undefined ||
+                                  s.preclaimedStart !== undefined) &&
                                 latencyTracking
                                   ? { latencyTracking }
                                   : {}),
@@ -3964,14 +4043,15 @@ export function workflowEntrypoint(
                             // these bodies until they settle — see
                             // assertNoInFlightOwnedSteps.
                             inFlightOwnedSteps.add(s.correlationId);
-                            // Lazy steps are brand-new (their create-claim
-                            // is the exactly-once gate), but an
-                            // owned-recovery step already exists and its
-                            // delayed backstop message may fire mid-body
+                            // Lazy and pre-claimed steps are brand-new
+                            // (their create-claim is the exactly-once gate),
+                            // but an owned-recovery step already exists and
+                            // its delayed backstop message may fire mid-body
                             // in this same process — route those through
                             // the in-process single-flight.
                             const executed =
-                              s.lazyStepInput === undefined
+                              s.lazyStepInput === undefined &&
+                              s.preclaimedStart === undefined
                                 ? runStepSingleFlight(
                                     runId,
                                     s.correlationId,
@@ -3984,6 +4064,26 @@ export function workflowEntrypoint(
                           }
                         );
                         try {
+                          // Join the dispatch publishes launched above and
+                          // the fold's deferred batch work (trailing chunk
+                          // commits + per-chunk step-message publishes) —
+                          // the bodies are already running in parallel with
+                          // both, and this invocation must not ack before
+                          // every create and publish is durable. A failure
+                          // keeps its old contract (fail this delivery so
+                          // the message redelivers), but the in-flight
+                          // bodies must settle first: an owned body left
+                          // running past this handler would race its own
+                          // redelivery.
+                          try {
+                            await Promise.all([
+                              dispatchesSettled,
+                              suspensionResult.deferredBatchWork,
+                            ]);
+                          } catch (dispatchErr) {
+                            await Promise.allSettled(stepExecutionPromises);
+                            throw dispatchErr;
+                          }
                           stepResults = await Promise.all(
                             stepExecutionPromises
                           );
diff --git a/packages/core/src/runtime/step-executor.test.ts b/packages/core/src/runtime/step-executor.test.ts
index ad80d8e1b7..623a299dcd 100644
--- a/packages/core/src/runtime/step-executor.test.ts
+++ b/packages/core/src/runtime/step-executor.test.ts
@@ -247,3 +247,136 @@ describe('executeStep — compute instance stamping', () => {
     }
   });
 });
+
+// Pre-claimed inline starts: the suspension handler's batched fan-out already
+// committed (or lost) the step's step_created + step_started pair, so the
+// executor must run the body straight off that verdict — no start write of
+// its own on the owned path, no write AT ALL on the lost path.
+describe('executeStep — pre-claimed inline start', () => {
+  afterEach(() => {
+    counter += 1;
+  });
+
+  it('runs the body without sending a step_started of its own when owned', async () => {
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    let bodyRuns = 0;
+
+    // Commit the pair the suspension batch would have committed.
+    const runInput = await dehydrateStepArguments([], 'run', undefined);
+    const created = await world.events.create(null, {
+      eventType: 'run_created',
+      specVersion: SPEC_VERSION_CURRENT,
+      eventData: {
+        deploymentId: 'dpl_test',
+        workflowName: 'wf',
+        input: runInput,
+      },
+    });
+    const runId = created.run!.runId;
+    await world.events.create(runId, {
+      eventType: 'run_started',
+      specVersion: SPEC_VERSION_CURRENT,
+      eventData: {},
+    } as never);
+    const stepId = 'step_preclaimed_1';
+    // The shape the suspension handler dehydrates for a pair's created row —
+    // the body's hydration reads `.args` off it.
+    const stepInput = await dehydrateStepArguments(
+      { args: [], closureVars: undefined, thisVal: undefined },
+      runId,
+      undefined
+    );
+    await world.events.create(runId, {
+      eventType: 'step_created',
+      specVersion: SPEC_VERSION_CURRENT,
+      correlationId: stepId,
+      eventData: { stepName, input: stepInput },
+    });
+    const startResult = await world.events.create(runId, {
+      eventType: 'step_started',
+      specVersion: SPEC_VERSION_CURRENT,
+      correlationId: stepId,
+      eventData: { stepName },
+    });
+    registerStepFunction(stepName, async () => {
+      bodyRuns += 1;
+      return 'ok';
+    });
+
+    const createSpy = vi.spyOn(world.events, 'create');
+    const result = await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId,
+      stepName,
+      authoritativeAttempt: 1,
+      preclaimedStart: {
+        owned: true,
+        step: { ...startResult.step!, input: stepInput },
+        batchPostSentAtMs: Date.now() - 5,
+        claimCompletedAtMs: Date.now(),
+      },
+    });
+
+    expect(result.type).toBe('completed');
+    expect(bodyRuns).toBe(1);
+    // The executor wrote ONLY the terminal event — the claim was the batch's.
+    const eventTypesWritten = createSpy.mock.calls.map(
+      (call) => (call[1] as { eventType: string }).eventType
+    );
+    expect(eventTypesWritten).not.toContain('step_started');
+    expect(eventTypesWritten).toContain('step_completed');
+    expect(await eventsFor(world, runId, stepId, 'step_started')).toHaveLength(
+      1
+    );
+  });
+
+  it('skips without any write when the pair lost its claim', async () => {
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    let bodyRuns = 0;
+    registerStepFunction(stepName, async () => {
+      bodyRuns += 1;
+      return 'ok';
+    });
+
+    const createSpy = vi.spyOn(world.events, 'create');
+    const result = await executeStep({
+      world,
+      workflowRunId: 'wrun_never_used',
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId: 'step_lost_claim',
+      stepName,
+      authoritativeAttempt: 1,
+      preclaimedStart: { owned: false },
+    });
+
+    expect(result).toEqual({ type: 'skipped' });
+    expect(bodyRuns).toBe(0);
+    expect(createSpy).not.toHaveBeenCalled();
+  });
+
+  it('skips before the unregistered-step fallback when the claim was lost', async () => {
+    const world = makeWorld();
+    const createSpy = vi.spyOn(world.events, 'create');
+
+    const result = await executeStep({
+      world,
+      workflowRunId: 'wrun_never_used',
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId: 'step_lost_unregistered',
+      // Never registered: the owned path would write step_failed here, but a
+      // lost claim is not this handler's to fail.
+      stepName: 'step//./step-executor-test//neverRegistered',
+      preclaimedStart: { owned: false },
+    });
+
+    expect(result).toEqual({ type: 'skipped' });
+    expect(createSpy).not.toHaveBeenCalled();
+  });
+});
diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts
index 63295800ce..63cacb054d 100644
--- a/packages/core/src/runtime/step-executor.ts
+++ b/packages/core/src/runtime/step-executor.ts
@@ -122,6 +122,17 @@ export interface StepExecutorParams {
    * carries no payload (the legacy contract).
    */
   lazyStepInput?: SerializedData;
+  /**
+   * Pre-claimed inline start: the suspension handler committed (or lost) this
+   * step's `step_created` + `step_started` pair inside its batched fan-out
+   * write, so the start this executor would otherwise send has already been
+   * decided. `owned: false` returns `{ type: 'skipped' }` before any write —
+   * the same outcome as losing the lazy claim's atomic create. `owned: true`
+   * skips both start paths (no start write at all) and runs the body against
+   * the claimed step. Mutually exclusive with `lazyStepInput`: the input
+   * already rode the pair's `step_created`, and the claimed step carries it.
+   */
+  preclaimedStart?: PreclaimedInlineStart;
   /**
    * Inline step ownership: the queue message ID of the invocation this
    * executeStep call runs in (from the queue handler's meta). When set, the
@@ -228,6 +239,44 @@ export interface StepExecutorParams {
   replayRecoveryReporter?: ReplayRecoveryReporter;
 }
 
+/**
+ * The settled outcome of a `step_created` + `step_started` pair the
+ * suspension handler folded into its batched fan-out write (see
+ * `SuspensionHandlerResult.inlineClaims`). Handed to executeStep as
+ * {@link StepExecutorParams.preclaimedStart} so the executor runs (or skips)
+ * the body off the batch's verdict instead of sending a start of its own.
+ */
+export type PreclaimedInlineStart =
+  | {
+      /** The pair committed: this execution owns the step and runs the body. */
+      owned: true;
+      /**
+       * The started step entity from the batch result, with the locally
+       * dehydrated input attached by the suspension handler (batch responses
+       * return refs lazily; the body's hydration wants the same bytes the
+       * pair's `step_created` carried).
+       */
+      step: StartedStep;
+      /**
+       * `Date.now()` taken right before the batch POST that carried the pair
+       * — the claim's "start POST sent" instant, anchoring RSFS exactly like
+       * the lazy claim's own POST would.
+       */
+      batchPostSentAtMs?: number;
+      /**
+       * `Date.now()` taken right after that batch POST returned — the
+       * claim's completion instant (T6 of the hook-resume TTR window).
+       */
+      claimCompletedAtMs?: number;
+    }
+  | {
+      /**
+       * The pair lost its atomic create-claim (per-event 409): a concurrent
+       * writer owns the step, so the body must not run here.
+       */
+      owned: false;
+    };
+
 /**
  * Inline-delta returned by a step-terminal write when the caller passed
  * {@link StepExecutorParams.inlineDeltaSinceCursor} and the World supports
@@ -356,6 +405,24 @@ export async function executeStep(
       ...Attribute.StepId(stepId),
     });
 
+    // A pre-claimed start that LOST the batched pair's atomic create-claim: a
+    // concurrent writer owns this step. Same outcome as losing the lazy
+    // claim (EntityConflictError → skipped), decided before ANY write — the
+    // unregistered-step fallback below included, since a step this handler
+    // does not own is not its to fail.
+    if (params.preclaimedStart && params.preclaimedStart.owned === false) {
+      runtimeLogger.debug('Pre-claimed step start lost, skipping', {
+        stepName,
+        stepId,
+        workflowRunId,
+      });
+      span?.setAttributes({
+        ...Attribute.StepSkipped(true),
+        ...Attribute.StepSkipReason('completed'),
+      });
+      return { type: 'skipped' };
+    }
+
     // Memoized accessor for the per-run encryption key. The first caller
     // (input hydration on the success path, or one of the early-return
     // dehydrateStepError paths if step_started fails) triggers the actual
@@ -587,6 +654,10 @@ export async function executeStep(
     // confirmed, which is exactly the property an operator opts out of with that
     // flag, so an explicit opt-out wins over turbo's force.
     const optimisticStart =
+      // A pre-claimed start already settled its claim in the suspension
+      // batch; there is nothing to fire optimistically (lazyStepInput is
+      // also absent on that path — this term is documentation).
+      params.preclaimedStart === undefined &&
       params.lazyStepInput !== undefined &&
       // Stale-sensitive guarded batches await the claim so the 412 fence
       // covers the body, not just durable writes — see
@@ -637,7 +708,18 @@ export async function executeStep(
       return mapped;
     };
 
-    if (optimisticStart) {
+    if (params.preclaimedStart?.owned) {
+      // Pre-claimed inline start: the suspension handler's batched fan-out
+      // already committed this step's `step_created` + `step_started` pair,
+      // so this execution owns a started attempt-1 step without sending a
+      // start of its own — the body begins straight off the batch commit and
+      // the terminal write below has no in-flight claim to reconcile. The
+      // batch timestamps stand in for the claim's: the POST instant anchors
+      // RSFS, the response instant is TTR's claim completion (T6).
+      step = params.preclaimedStart.step;
+      stepStartPostSentAtMs = params.preclaimedStart.batchPostSentAtMs;
+      stepClaimCompletedAtMs = params.preclaimedStart.claimCompletedAtMs;
+    } else if (optimisticStart) {
       // Chain the lazy `step_started` on the run-ready barrier (turbo mode):
       // the step can't be created before its run exists, but the body below
       // runs immediately against synthesized state, so the `run_started`
@@ -918,6 +1000,7 @@ export async function executeStep(
         attempt,
         lazyStepStart: params.lazyStepInput !== undefined,
         optimisticStart,
+        preclaimedStart: params.preclaimedStart !== undefined,
         stepStartPostSentAtMs,
       });
       if (latencyEventData) {
diff --git a/packages/core/src/runtime/step-latency.ts b/packages/core/src/runtime/step-latency.ts
index 0a4d9b08e3..c30cc2fe4f 100644
--- a/packages/core/src/runtime/step-latency.ts
+++ b/packages/core/src/runtime/step-latency.ts
@@ -326,6 +326,12 @@ export function computeStepLatencyEventData(params: {
   lazyStepStart: boolean;
   /** Whether the body ran optimistically, without awaiting step_started. */
   optimisticStart: boolean;
+  /**
+   * Whether the step's `step_created` + `step_started` pair was pre-claimed
+   * inside the suspension handler's batched fan-out write (no start POST of
+   * its own at all).
+   */
+  preclaimedStart?: boolean;
 }): StepLatencyEventData | undefined {
   const { tracking } = params;
   if (!tracking || params.attempt !== 1) {
@@ -396,6 +402,7 @@ export function computeStepLatencyEventData(params: {
   if (tracking.turbo) optimizations.push('turbo');
   if (params.lazyStepStart) optimizations.push('lazyStepStart');
   if (params.optimisticStart) optimizations.push('optimisticStart');
+  if (params.preclaimedStart) optimizations.push('preclaimedStart');
 
   return {
     ...(ttfs !== undefined ? { ttfs } : {}),
diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts
index aee668d278..b2c440810b 100644
--- a/packages/core/src/runtime/suspension-handler.test.ts
+++ b/packages/core/src/runtime/suspension-handler.test.ts
@@ -14,6 +14,7 @@ import {
 } from '@workflow/world';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { WorkflowSuspension } from '../global.js';
+import { COMPUTE_INSTANCE_ID } from './compute-instance.js';
 import { maxEventSlot, stepDispatchIdempotencyKey } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
 import { handleSuspension } from './suspension-handler.js';
@@ -1222,4 +1223,488 @@ describe('handleSuspension batched fan-out', () => {
     ).toEqual(['s4', 's5']);
     expect([...result.createdStepCorrelationIds].sort()).toEqual(['s4', 's5']);
   });
+
+  describe('pre-claimed inline pairs', () => {
+    it('folds each inline step as a created+started pair, stamped and claimed', async () => {
+      vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2');
+      const eventsCreate = vi.fn();
+      const createBatch = successfulCreateBatch();
+      const world = createBatchWorld(eventsCreate, createBatch);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(
+          stepsAndWait(['s1', 's2', 's3'], 'wait_1'),
+          globalThis
+        ),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+      });
+
+      expect(createBatch).toHaveBeenCalledTimes(1);
+      const events = createBatch.mock.calls[0][1];
+      // s1/s2 are inline: their pairs lead, adjacent; then s3's eager create
+      // and the wait — scheduling order preserved.
+      expect(
+        events.map((e: { event: { eventType: string } }) => e.event.eventType)
+      ).toEqual([
+        'step_created',
+        'step_started',
+        'step_created',
+        'step_started',
+        'step_created',
+        'wait_created',
+      ]);
+      expect(
+        events.map(
+          (e: { event: { correlationId: string } }) => e.event.correlationId
+        )
+      ).toEqual(['s1', 's1', 's2', 's2', 's3', 'wait_1']);
+      // The created rows carry the input; the started rows are bare claims
+      // stamped with this invocation's ownership and compute instance.
+      const s1Created = events[0].event;
+      const s1Started = events[1].event;
+      expect(s1Created.eventData.input).toBeDefined();
+      expect(s1Started.eventData.input).toBeUndefined();
+      expect(s1Started.eventData.ownerMessageId).toBe('msg_owner_1');
+      expect(events[1].computeInstanceId).toBe(COMPUTE_INSTANCE_ID);
+      expect(events[0].computeInstanceId).toBeUndefined();
+      // Claims: both inline steps owned, running attempt 1, input attached
+      // (batch responses return refs lazily — the body hydrates local bytes).
+      expect(result.inlineClaims.size).toBe(2);
+      for (const id of ['s1', 's2']) {
+        const claim = result.inlineClaims.get(id);
+        expect(claim?.owned).toBe(true);
+        if (claim?.owned) {
+          expect(claim.step.status).toBe('running');
+          expect(claim.step.attempt).toBe(1);
+          expect(claim.step.input).toBeDefined();
+          expect(claim.batchPostSentAtMs).toBeTypeOf('number');
+          expect(claim.claimCompletedAtMs).toBeTypeOf('number');
+        }
+      }
+      // Inline steps stay OUT of createdStepCorrelationIds — the started
+      // row's verdict (the claim) is their ownership, and the caller's
+      // dispatch pass skips inline ids regardless.
+      expect([...result.createdStepCorrelationIds]).toEqual(['s3']);
+      // The deferral list is unchanged; the caller keys claims off it.
+      expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([
+        's1',
+        's2',
+      ]);
+      // 6 events at slots 10..15.
+      expect(result.batchCommittedSlotCeiling).toBe(15);
+      expect(eventsCreate).not.toHaveBeenCalled();
+    });
+
+    it('does not fold pairs without the caller ownership stamp', async () => {
+      vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2');
+      const createBatch = successfulCreateBatch();
+      const world = createBatchWorld(vi.fn(), createBatch);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(
+          stepsAndWait(['s1', 's2', 's3', 's4']),
+          globalThis
+        ),
+        world,
+        run: slotRun,
+      });
+
+      // s1/s2 defer to the lazy path; only the eager creates batch.
+      expect(
+        createBatch.mock.calls[0][1].map(
+          (e: { event: { correlationId: string } }) => e.event.correlationId
+        )
+      ).toEqual(['s3', 's4']);
+      expect(result.inlineClaims.size).toBe(0);
+      expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([
+        's1',
+        's2',
+      ]);
+    });
+
+    it('records a lost pair as owned:false and keeps the batch alive', async () => {
+      vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2');
+      let slot = 20;
+      const createBatch = vi
+        .fn()
+        .mockImplementation(async (_runId, events) => ({
+          results: events.map(
+            ({ event }: { event: { correlationId: string } }, index: number) =>
+              event.correlationId === 's1'
+                ? {
+                    status: 409,
+                    error: 'conflict',
+                    message: `row ${index}: already claimed`,
+                  }
+                : {
+                    status: 200,
+                    event: { ...event, eventId: slotToEventId(slot++) },
+                  }
+          ),
+        }));
+      const world = createBatchWorld(vi.fn(), createBatch);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(
+          stepsAndWait(['s1', 's2', 's3']),
+          globalThis
+        ),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+      });
+
+      expect(result.inlineClaims.get('s1')).toEqual({ owned: false });
+      expect(result.inlineClaims.get('s2')?.owned).toBe(true);
+      expect([...result.createdStepCorrelationIds]).toEqual(['s3']);
+    });
+
+    it('keeps the lone inline step on the lazy path (nothing to batch with)', async () => {
+      const eventsCreate = vi.fn();
+      const createBatch = successfulCreateBatch();
+      const world = createBatchWorld(eventsCreate, createBatch);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(stepsAndWait(['s1']), globalThis),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+      });
+
+      // A pair-only batch is the same round trip as the single lazy claim
+      // but gives up the optimistic claim/body overlap — so nothing is
+      // written at all here; the deferral stands.
+      expect(createBatch).not.toHaveBeenCalled();
+      expect(eventsCreate).not.toHaveBeenCalled();
+      expect(result.inlineClaims.size).toBe(0);
+      expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([
+        's1',
+      ]);
+    });
+
+    it('folds a lone inline pair when an eager sibling already batches', async () => {
+      const createBatch = successfulCreateBatch();
+      const world = createBatchWorld(vi.fn(), createBatch);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(
+          stepsAndWait(['s1', 's2']),
+          globalThis
+        ),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+      });
+
+      expect(
+        createBatch.mock.calls[0][1].map(
+          (e: { event: { eventType: string; correlationId: string } }) =>
+            `${e.event.eventType}:${e.event.correlationId}`
+        )
+      ).toEqual(['step_created:s1', 'step_started:s1', 'step_created:s2']);
+      expect(result.inlineClaims.get('s1')?.owned).toBe(true);
+      expect([...result.createdStepCorrelationIds]).toEqual(['s2']);
+    });
+
+    it('keeps pairs whole at the chunk boundary (max inline cap)', async () => {
+      // The inline cap clamps at 16, so 16 pairs = exactly 32 rows — one full
+      // chunk, pairs adjacent throughout — and the eager overflow spills into
+      // the next call. (Pairs always occupy the head rows, so with cap*2 ==
+      // MAX_BATCH_FANOUT_EVENTS a straddle is structurally unreachable; the
+      // chunker still refuses to split one should those constants diverge.)
+      vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '16');
+      const createBatch = successfulCreateBatch();
+      const world = createBatchWorld(vi.fn(), createBatch);
+      const stepIds = Array.from({ length: 17 }, (_, i) => `s${i + 1}`);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+      });
+
+      expect(createBatch).toHaveBeenCalledTimes(2);
+      const head = createBatch.mock.calls[0][1];
+      expect(head).toHaveLength(32);
+      // 16 adjacent created+started pairs, in step order.
+      for (let pair = 0; pair < 16; pair++) {
+        expect(head[2 * pair].event.eventType).toBe('step_created');
+        expect(head[2 * pair + 1].event.eventType).toBe('step_started');
+        expect(head[2 * pair + 1].event.correlationId).toBe(
+          head[2 * pair].event.correlationId
+        );
+      }
+      const tail = createBatch.mock.calls[1][1];
+      expect(
+        tail.map(
+          (e: { event: { eventType: string; correlationId: string } }) =>
+            `${e.event.eventType}:${e.event.correlationId}`
+        )
+      ).toEqual(['step_created:s17']);
+      expect(result.inlineClaims.size).toBe(16);
+      for (const claim of result.inlineClaims.values()) {
+        expect(claim.owned).toBe(true);
+      }
+      expect([...result.createdStepCorrelationIds]).toEqual(['s17']);
+    });
+
+    it('prefers the readback step entity when the World returns one', async () => {
+      vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '1');
+      const serverStartedAt = new Date('2026-08-14T01:02:03.000Z');
+      let slot = 30;
+      const createBatch = vi
+        .fn()
+        .mockImplementation(async (_runId, events) => ({
+          results: events.map(
+            ({
+              event,
+            }: {
+              event: { eventType: string; correlationId: string };
+            }) => ({
+              status: 200,
+              event: { ...event, eventId: slotToEventId(slot++) },
+              ...(event.eventType === 'step_started'
+                ? {
+                    step: {
+                      runId: slotRun.runId,
+                      stepId: event.correlationId,
+                      stepName: event.correlationId,
+                      status: 'running',
+                      attempt: 1,
+                      createdAt: serverStartedAt,
+                      updatedAt: serverStartedAt,
+                      startedAt: serverStartedAt,
+                    },
+                  }
+                : {}),
+            })
+          ),
+        }));
+      const world = createBatchWorld(vi.fn(), createBatch);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(
+          stepsAndWait(['s1', 's2']),
+          globalThis
+        ),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+      });
+
+      const claim = result.inlineClaims.get('s1');
+      expect(claim?.owned).toBe(true);
+      if (claim?.owned) {
+        expect(claim.step.startedAt).toEqual(serverStartedAt);
+        // Input is still re-attached locally over the readback entity.
+        expect(claim.step.input).toBeDefined();
+      }
+    });
+  });
+
+  describe('parallel chunks, per-chunk publishes, deferred work', () => {
+    const queueName = '__wkf_workflow_test-workflow' as ValidQueueName;
+    const stepDispatch = () => ({
+      queueName,
+      getTraceCarrier: vi.fn().mockResolvedValue({ traceparent: '00-abc' }),
+    });
+
+    /**
+     * A createBatch mock whose calls block until released, so tests control
+     * per-chunk commit timing. Results mirror successfulCreateBatch.
+     */
+    function gatedCreateBatch(firstSlot = 10) {
+      let slot = firstSlot;
+      const releases: (() => void)[] = [];
+      const createBatch = vi.fn().mockImplementation(
+        (_runId, events) =>
+          new Promise((resolve) => {
+            releases.push(() =>
+              resolve({
+                results: events.map(
+                  ({ event }: { event: { eventType: string } }) => ({
+                    status: 200,
+                    event: { ...event, eventId: slotToEventId(slot++) },
+                  })
+                ),
+              })
+            );
+          })
+      );
+      return { createBatch, releases };
+    }
+
+    function queueWorld(
+      createBatch: ReturnType<typeof vi.fn>,
+      queue = vi.fn().mockResolvedValue({ messageId: 'msg_q' })
+    ): { world: World; queue: ReturnType<typeof vi.fn> } {
+      const world = {
+        events: { create: vi.fn(), createBatch },
+        queue,
+        getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined),
+      } as unknown as World;
+      return { world, queue };
+    }
+
+    const tick = () => new Promise((resolve) => setImmediate(resolve));
+    /** 'pending' | 'settled' without awaiting the probed promise. */
+    const probe = async (p: Promise<unknown> | undefined) => {
+      let state = 'pending';
+      p?.then(
+        () => {
+          state = 'settled';
+        },
+        () => {
+          state = 'settled';
+        }
+      );
+      await tick();
+      return state;
+    };
+
+    it('POSTs every chunk concurrently instead of serially', async () => {
+      // 34 steps, no pairs (no ownerMessageId): s1 defers lazily, 33 eager
+      // creates chunk as 32 + 1 — and BOTH POSTs must be in flight before
+      // either commits.
+      const { createBatch, releases } = gatedCreateBatch();
+      const { world } = queueWorld(createBatch);
+      const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`);
+
+      const pending = handleSuspension({
+        suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis),
+        world,
+        run: slotRun,
+      });
+      await vi.waitFor(() => {
+        expect(createBatch).toHaveBeenCalledTimes(2);
+      });
+      for (const release of releases) release();
+      const result = await pending;
+      expect(result.createdStepCorrelationIds.size).toBe(33);
+    });
+
+    it('returns off the pair chunk; trailing chunks ride deferredBatchWork', async () => {
+      // 34 steps with a pair: chunk 1 = pair + 30 eager (32 rows), chunk 2 =
+      // 3 eager. Releasing only chunk 1 must resolve the handler with the
+      // claims; chunk 2 settles deferredBatchWork later.
+      const { createBatch, releases } = gatedCreateBatch();
+      const { world, queue } = queueWorld(createBatch);
+      const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`);
+
+      const pending = handleSuspension({
+        suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+        stepDispatch: stepDispatch(),
+        allowDeferredBatchWork: true,
+      });
+      await vi.waitFor(() => {
+        expect(createBatch).toHaveBeenCalledTimes(2);
+      });
+      releases[0]();
+      const result = await pending;
+
+      // The handler returned with chunk 2 still uncommitted.
+      expect(result.inlineClaims.get('s1')?.owned).toBe(true);
+      expect(result.deferredBatchWork).toBeDefined();
+      expect(await probe(result.deferredBatchWork)).toBe('pending');
+      // Every eager step is claimed for in-flush publishing up front, so
+      // the caller's dispatch pass skips them all.
+      expect(result.queuedStepCorrelationIds.size).toBe(33);
+
+      // Chunk 1's publishes fire off its own commit — 30 messages — while
+      // chunk 2's three wait for theirs.
+      await vi.waitFor(() => {
+        expect(queue).toHaveBeenCalledTimes(30);
+      });
+      const publishedNow = queue.mock.calls.map((call) => call[1].stepId);
+      expect(publishedNow).not.toContain('s33');
+
+      releases[1]();
+      // biome-ignore lint/style/noNonNullAssertion: asserted defined above
+      await result.deferredBatchWork!;
+      expect(queue).toHaveBeenCalledTimes(33);
+      // Message shape and idempotency key match the caller's dispatch pass.
+      const [calledQueueName, payload, opts] = queue.mock.calls[0];
+      expect(calledQueueName).toBe(queueName);
+      expect(payload).toMatchObject({
+        runId: slotRun.runId,
+        stepName: payload.stepId,
+        traceCarrier: { traceparent: '00-abc' },
+      });
+      expect(opts.idempotencyKey).toBe(
+        stepDispatchIdempotencyKey(payload.stepId, payload.stepName)
+      );
+    });
+
+    it('surfaces a trailing-chunk failure through deferredBatchWork, not the return', async () => {
+      let call = 0;
+      let releaseFailure: (() => void) | undefined;
+      const createBatch = vi.fn().mockImplementation((_runId, events) => {
+        call += 1;
+        if (call === 2) {
+          return new Promise((_resolve, reject) => {
+            releaseFailure = () =>
+              reject(
+                new WorkflowWorldError('trailing chunk exploded', {
+                  status: 500,
+                })
+              );
+          });
+        }
+        let slot = 10;
+        return Promise.resolve({
+          results: events.map(({ event }: { event: object }) => ({
+            status: 200,
+            event: { ...event, eventId: slotToEventId(slot++) },
+          })),
+        });
+      });
+      const { world } = queueWorld(createBatch);
+      const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`);
+
+      const result = await handleSuspension({
+        suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+        stepDispatch: stepDispatch(),
+        allowDeferredBatchWork: true,
+      });
+      expect(result.inlineClaims.get('s1')?.owned).toBe(true);
+      // biome-ignore lint/style/noNonNullAssertion: set by the second call
+      releaseFailure!();
+      await expect(result.deferredBatchWork).rejects.toMatchObject({
+        message: expect.stringContaining('trailing chunk exploded'),
+      });
+    });
+
+    it('awaits everything at return without the opt-in', async () => {
+      const { createBatch, releases } = gatedCreateBatch();
+      const { world } = queueWorld(createBatch);
+      const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`);
+
+      const pending = handleSuspension({
+        suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis),
+        world,
+        run: slotRun,
+        ownerMessageId: 'msg_owner_1',
+        stepDispatch: stepDispatch(),
+      });
+      await vi.waitFor(() => {
+        expect(createBatch).toHaveBeenCalledTimes(2);
+      });
+      releases[0]();
+      // Chunk 2 unreleased: the handler must still be pending.
+      expect(await probe(pending)).toBe('pending');
+      releases[1]();
+      const result = await pending;
+      expect(result.deferredBatchWork).toBeUndefined();
+      expect(result.inlineClaims.get('s1')?.owned).toBe(true);
+    });
+  });
 });
diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts
index 0e50ac4e09..dfb0336693 100644
--- a/packages/core/src/runtime/suspension-handler.ts
+++ b/packages/core/src/runtime/suspension-handler.ts
@@ -17,6 +17,7 @@ import {
   SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT,
   SPEC_VERSION_SUPPORTS_COMPRESSION,
   SPEC_VERSION_SUPPORTS_SLOT_IDENTITY,
+  type StartedStep,
   type TraceCarrier,
   type ValidQueueName,
   type WorkflowRun,
@@ -36,6 +37,7 @@ import type { GuestCodeStats } from '../serialization/hardened.js';
 import { dehydrateStepArguments } from '../serialization.js';
 import * as Attribute from '../telemetry/semantic-conventions.js';
 import { getAbortStreamIdFromToken } from '../util.js';
+import { COMPUTE_INSTANCE_ID } from './compute-instance.js';
 import {
   getMaxInlineSteps,
   isBatchTransitionsEnabled,
@@ -53,6 +55,7 @@ import {
   stepDispatchIdempotencyKey,
 } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
+import type { PreclaimedInlineStart } from './step-executor.js';
 
 export interface SuspensionHandlerParams {
   suspension: WorkflowSuspension;
@@ -103,6 +106,31 @@ export interface SuspensionHandlerParams {
      */
     getTraceCarrier: () => Promise<TraceCarrier>;
   };
+  /**
+   * Inline step ownership: the queue message ID of the invocation this
+   * suspension runs in (the queue handler's meta). When present AND the
+   * batched fan-out engages, the lazy-inline steps' deferred writes are
+   * folded into the batch as `step_created` + `step_started` pairs — the
+   * started row stamped with this ID, exactly like the lazy claim it
+   * replaces — pre-claiming the steps the caller is about to run inline. See
+   * {@link SuspensionHandlerResult.inlineClaims}. Callers that never
+   * inline-execute (terminal drain) omit it, keeping their lazy steps on the
+   * plain deferred path.
+   */
+  ownerMessageId?: string;
+  /**
+   * Lets the batched fan-out return before every chunk has committed: only
+   * the chunk carrying the pre-claimed inline pairs gates the handler's
+   * return (its claims are what the caller starts bodies from), while the
+   * trailing chunks' commits — and every chunk's in-flush step-message
+   * publishes — ride {@link SuspensionHandlerResult.deferredBatchWork}. A
+   * caller that opts in MUST await that promise before acking its delivery:
+   * the durability contract ("every create durable before ack") moves from
+   * the handler's return to that join, and nothing else re-drives a lost
+   * trailing chunk. Callers that don't opt in (terminal drain, default)
+   * keep the everything-durable-at-return behavior.
+   */
+  allowDeferredBatchWork?: boolean;
 }
 
 /**
@@ -156,6 +184,52 @@ export interface SuspensionHandlerResult {
     stepName: string;
     dehydratedInput: SerializedData;
   }>;
+  /**
+   * Pre-claimed inline starts, by correlation id: the per-step verdicts of
+   * the `step_created` + `step_started` pairs the batched fan-out committed
+   * for the lazy-inline steps. A step with an entry here is passed to
+   * `executeStep` as `preclaimedStart` INSTEAD of `lazyStepInput` — its
+   * input already rode the pair, and the claim is settled: `owned: true`
+   * carries the started attempt-1 entity (input re-attached) so the body
+   * runs straight off the batch commit with no start write of its own;
+   * `owned: false` lost the pair's atomic create-claim to a concurrent
+   * writer, and executeStep returns `skipped` without running the body —
+   * the same outcome as losing the lazy claim. Empty whenever the fold did
+   * not engage (batching off, no `ownerMessageId`, or the lone-inline case,
+   * which keeps the optimistic lazy path and its claim/body overlap).
+   *
+   * Crash window: the pair commits before the caller runs the body, so a
+   * crash between them leaves a started step stamped with this message's
+   * ID. Redelivery of the same message re-executes it via the owned-recovery
+   * path — the exact machinery the lazy claim's crash window already uses.
+   */
+  inlineClaims: Map<string, PreclaimedInlineStart>;
+  /**
+   * The highest slot the batched fan-out committed, when it ran. The batch's
+   * own events are not in the caller's loaded log (the next reload picks
+   * them up), so the caller folds this ceiling into the slot snapshot it
+   * hands the inline executions — otherwise every inline terminal write
+   * would name a pre-batch position and be answered with a skipped-slot
+   * report echoing the events this suspension just wrote. Under
+   * {@link SuspensionHandlerParams.allowDeferredBatchWork} this covers the
+   * chunks that had committed by the handler's return (always the pair
+   * chunk); a trailing chunk that commits later is echoed back on the
+   * terminal writes like any foreign event — reports the executor reads for
+   * position and discards.
+   */
+  batchCommittedSlotCeiling?: number;
+  /**
+   * The batched fan-out's deferred work, present only when the caller opted
+   * in via {@link SuspensionHandlerParams.allowDeferredBatchWork} and
+   * trailing work exists: the commits of every chunk except the pair chunk,
+   * plus every chunk's step-message publishes (each chained on ITS OWN
+   * chunk's commit, so publish-after-create holds per step). The caller
+   * MUST await it before acking — a rejection here is a failed suspension
+   * write and fails the delivery exactly as it would have at the handler's
+   * return. Steps whose messages this work publishes are already in
+   * {@link queuedStepCorrelationIds} at return time.
+   */
+  deferredBatchWork?: Promise<void>;
   /**
    * The soonest pending wait, if any: seconds until it elapses and the
    * correlationId of the wait that produced that timeout. The
@@ -288,6 +362,8 @@ export async function handleSuspension({
   runReadyBarrier,
   replayRecoveryReporter,
   stepDispatch,
+  ownerMessageId,
+  allowDeferredBatchWork,
 }: SuspensionHandlerParams): Promise<SuspensionHandlerResult> {
   const runId = run.runId;
 
@@ -724,12 +800,42 @@ export async function handleSuspension({
    */
   const batchQueue: {
     order: number;
-    kind: 'step' | 'wait';
+    kind: 'step' | 'wait' | 'inline-created' | 'inline-started';
     correlationId: string;
+    /** The step's name, set on step-carrying kinds (the in-flush publishes
+     *  and the dispatch idempotency key need it). */
+    stepName?: string;
     event: CreateEventRequest;
   }[] = [];
   const batchPreps: Promise<void>[] = [];
 
+  // Pre-claimed inline pairs: fold each lazy-inline step's deferred
+  // `step_created` (carrying its input) AND its `step_started` claim (bare,
+  // ownership-stamped) into the batch, so the whole fan-out — the inline
+  // steps' claims included — commits in the one durable write and the caller
+  // starts the bodies straight off that commit instead of posting one claim
+  // per inline step. The lone-inline case (nothing else to batch with) is
+  // excluded: a pair-only batch costs the same round trip as the single lazy
+  // claim while giving up the optimistic claim/body overlap and the
+  // bump-and-report that `createGuarded` provides, so it stays on the lazy
+  // path. Requires the caller's `ownerMessageId` — the started row must
+  // stamp ownership exactly like the lazy claim it replaces (and a caller
+  // that does not inline-execute never provides one).
+  const uncreatedWaitCount = waitItems.filter(
+    (item) => !item.hasCreatedEvent
+  ).length;
+  const inlinePairFoldEligible =
+    batchFanoutEligible &&
+    ownerMessageId !== undefined &&
+    lazyInlineCorrelationIds.size > 0 &&
+    (lazyInlineCorrelationIds.size >= 2 ||
+      stepsNeedingCreation.size -
+        lazyInlineCorrelationIds.size +
+        uncreatedWaitCount >=
+        1);
+  const inlineClaims: SuspensionHandlerResult['inlineClaims'] = new Map();
+  let batchCommittedSlotCeiling: number | undefined;
+
   // The trace carrier for resilient step dispatches, resolved at most once per
   // suspension (the per-step ops run concurrently and share it).
   let stepDispatchTraceCarrier: Promise<TraceCarrier> | undefined;
@@ -750,8 +856,15 @@ export async function handleSuspension({
   for (const queueItem of stepItems) {
     if (stepsNeedingCreation.has(queueItem.correlationId)) {
       // Deterministic position in the batched fold (assigned in stepItems
-      // order, before the concurrent dehydration runs).
-      const stepOrder = batchOrderCounter++;
+      // order, before the concurrent dehydration runs). A pair-folded inline
+      // step occupies two consecutive positions — created row then started
+      // row — which the flush keeps adjacent and never splits across chunks,
+      // so a World can fold them into one born-running create.
+      const pairFolded =
+        inlinePairFoldEligible &&
+        lazyInlineCorrelationIds.has(queueItem.correlationId);
+      const stepOrder = batchOrderCounter;
+      batchOrderCounter += pairFolded ? 2 : 1;
       const stepOp = (async () => {
         // Per-step sink, merged below: the dehydrate wrapper emits span
         // attributes from the sink it is handed, so sharing one across
@@ -783,6 +896,44 @@ export async function handleSuspension({
             stepName: queueItem.stepName,
             dehydratedInput: dehydratedInput as SerializedData,
           });
+          if (pairFolded) {
+            // Enqueue the pair the deferral would otherwise leave to the
+            // caller's lazy `step_started`: the created row carries the
+            // input (payloads ride creates in a batch), the started row is
+            // bare and stamps this invocation's ownership — the same claim
+            // shape the lazy start would have sent, settled by the batch.
+            batchQueue.push({
+              order: stepOrder,
+              kind: 'inline-created',
+              correlationId: queueItem.correlationId,
+              event: {
+                eventType: 'step_created',
+                specVersion: SPEC_VERSION_CURRENT,
+                correlationId: queueItem.correlationId,
+                eventData: {
+                  stepName: queueItem.stepName,
+                  workflowName: run.workflowName,
+                  input: dehydratedInput as SerializedData,
+                },
+              },
+            });
+            batchQueue.push({
+              order: stepOrder + 1,
+              kind: 'inline-started',
+              correlationId: queueItem.correlationId,
+              event: {
+                eventType: 'step_started',
+                specVersion: SPEC_VERSION_CURRENT,
+                correlationId: queueItem.correlationId,
+                eventData: {
+                  stepName: queueItem.stepName,
+                  // Checked by inlinePairFoldEligible; spread keeps the
+                  // narrow-through-closure problem away from the type.
+                  ...(ownerMessageId !== undefined ? { ownerMessageId } : {}),
+                },
+              },
+            });
+          }
           return;
         }
         const stepEvent: CreateEventRequest = {
@@ -892,6 +1043,7 @@ export async function handleSuspension({
             order: stepOrder,
             kind: 'step',
             correlationId: queueItem.correlationId,
+            stepName: queueItem.stepName,
             event: stepEvent,
           });
           return;
@@ -968,12 +1120,27 @@ export async function handleSuspension({
     }
   }
 
-  // The batched fold's flush: ONE durable write for the whole clean fan-out
-  // (chunked at MAX_BATCH_FANOUT_EVENTS), joining `ops` like the per-event
-  // writes it replaces so settlePhase semantics are unchanged. Each event
-  // reports the outcome its own single create would have had: a 409 is the
-  // same already-exists tolerance as the single path, anything else fails
-  // the op the way a single-path rejection would.
+  // The batched fold's flush: the clean fan-out commits through
+  // `createBatch` in chunks of MAX_BATCH_FANOUT_EVENTS — all chunks IN
+  // FLIGHT CONCURRENTLY. Slot assignment is the server's, so parallel
+  // chunks race for slot ranges exactly like the pre-fold path's parallel
+  // single writes did; entity conditions, not commit order, carry
+  // correctness (sibling fan-out events have no cross-order the replay
+  // depends on — it matches by correlation id). Each event reports the
+  // outcome its own single create would have had: a 409 is the same
+  // already-exists tolerance as the single path, anything else fails the
+  // delivery the way a single-path rejection would.
+  //
+  // Latency shape: only the chunk carrying the pre-claimed inline pairs
+  // gates the handler's return (the caller starts bodies off its claims).
+  // Every other chunk's commit — and every chunk's step-message publishes,
+  // which fire the moment ITS creates are durable — rides
+  // `deferredBatchWork` when the caller opted in, joined before ack. A slow
+  // sibling chunk therefore delays neither the inline bodies nor another
+  // chunk's queue messages, while publish-after-create still holds per
+  // step: a step's message is only ever sent after the chunk carrying its
+  // create has committed.
+  let deferredBatchWork: Promise<void> | undefined;
   if (batchFanoutEligible) {
     ops.push(
       (async () => {
@@ -1016,36 +1183,167 @@ export async function handleSuspension({
           }
           return;
         }
-        // Expected next slot for the bump diagnostic below: seeded once from
-        // the caller's view and advanced past each chunk's own committed
-        // events, so chunk 2+ of a multi-chunk fan-out does not misread this
-        // fold's earlier chunks as foreign skips.
-        let expectedFirstSlot = eventLog
+        // Seed for the foreign-interleaving diagnostic below. With chunks
+        // committing in parallel there is no per-chunk "expected next slot"
+        // — the whole fold's committed span is compared against the seed
+        // once every chunk has settled: committed slots are dense per the
+        // World's invariant, so any excess of (max committed slot − seed +
+        // 1) over the fold's own committed count is events OTHER writers
+        // landed in between.
+        const expectedFirstSlot = eventLog
           ? (maxEventSlot(eventLog.events) ?? 0) + 1
           : undefined;
-        for (
-          let start = 0;
-          start < entries.length;
-          start += MAX_BATCH_FANOUT_EVENTS
-        ) {
-          const chunk = entries.slice(start, start + MAX_BATCH_FANOUT_EVENTS);
+        // Pair-aware chunking: a pre-claimed pair's two rows must land in
+        // the same createBatch call — adjacent, so a World can fold them
+        // into one born-running create — and never straddle a chunk
+        // boundary, which would turn the started row into a standalone
+        // claim racing its own create's commit.
+        const chunks: (typeof entries)[] = [];
+        {
+          let current: typeof entries = [];
+          for (let index = 0; index < entries.length; index++) {
+            const entry = entries[index];
+            const next = entries[index + 1];
+            const pairLead =
+              entry.kind === 'inline-created' &&
+              next?.kind === 'inline-started' &&
+              next.correlationId === entry.correlationId;
+            const take = pairLead ? 2 : 1;
+            if (
+              current.length > 0 &&
+              current.length + take > MAX_BATCH_FANOUT_EVENTS
+            ) {
+              chunks.push(current);
+              current = [];
+            }
+            current.push(entry);
+            if (pairLead) {
+              current.push(next);
+              index++;
+            }
+          }
+          if (current.length > 0) chunks.push(current);
+        }
+        // Steps whose queue messages THIS FLUSH will publish (the eager
+        // creates), recorded before any chunk settles so the caller's
+        // dispatch pass — which runs off the handler's return — skips them.
+        // The sends are guaranteed-or-failed by the trailing work the
+        // caller joins before acking, so "will be published by this flush"
+        // and "already published" are equivalent from the caller's side.
+        const publishEagerSteps = stepDispatch !== undefined;
+        if (publishEagerSteps) {
+          for (const entry of entries) {
+            if (entry.kind === 'step') {
+              queuedStepCorrelationIds.add(entry.correlationId);
+            }
+          }
+        }
+
+        // Tallies for the foreign-interleaving diagnostic, folded across
+        // the concurrent chunks and read once all of them settled.
+        let committedCount = 0;
+        let maxCommittedSlot: number | undefined;
+
+        const commitChunk = async (chunk: typeof entries): Promise<void> => {
+          // Anchors for the pre-claimed steps' telemetry: the POST instant
+          // is the claim's "start POST sent" (RSFS's end), the return is the
+          // claim's completion (TTR's T6) — the same two instants the lazy
+          // claim's own POST would have produced.
+          const batchPostSentAtMs = Date.now();
           // biome-ignore lint/style/noNonNullAssertion: batchFanoutEligible implies presence
           const { results } = await world.events.createBatch!(
             runId,
-            chunk.map((entry) => ({ event: entry.event })),
+            chunk.map((entry) => ({
+              event: entry.event,
+              // The started row is the step's executing claim, so it carries
+              // the compute-instance attribution the single claim sends via
+              // CreateEventParams.
+              ...(entry.kind === 'inline-started'
+                ? { computeInstanceId: COMPUTE_INSTANCE_ID }
+                : {}),
+            })),
             // Per-write request attribution, same as the single path's
             // createGuarded(…, { requestId }).
             { requestId }
           );
+          const claimCompletedAtMs = Date.now();
           for (const [index, item] of results.entries()) {
             const entry = chunk[index];
             if (item.error === undefined) {
               if (entry.kind === 'step') {
                 createdStepCorrelationIds.add(entry.correlationId);
+              } else if (entry.kind === 'inline-started') {
+                // The pair committed: this invocation owns the step and the
+                // caller runs the body with no claim of its own. The
+                // readback entity is authoritative where present; a World
+                // that omitted it gets the same locally synthesized running
+                // attempt-1 the optimistic path executes against. Either
+                // way the input is re-attached locally — batch responses
+                // return refs lazily, and the body's hydration wants the
+                // exact bytes the pair's created row carried. (The created
+                // row's success is deliberately NOT membership in
+                // createdStepCorrelationIds: for inline steps ownership is
+                // the started row's verdict, and the caller's dispatch pass
+                // skips inline correlation ids regardless.)
+                const dehydrated = lazyInlineByCorrelationId.get(
+                  entry.correlationId
+                );
+                if (dehydrated === undefined) {
+                  // Unreachable: the same prep op that enqueued the pair set
+                  // this entry, and the flush awaited every prep above.
+                  throw new WorkflowWorldError(
+                    `no dehydrated input for pre-claimed step ${entry.correlationId}`,
+                    { status: 500 }
+                  );
+                }
+                const now = new Date();
+                const startedStep: StartedStep = item.step?.startedAt
+                  ? { ...item.step, startedAt: item.step.startedAt }
+                  : {
+                      runId,
+                      stepId: entry.correlationId,
+                      stepName: dehydrated.stepName,
+                      status: 'running',
+                      attempt: 1,
+                      createdAt: now,
+                      updatedAt: now,
+                      startedAt: now,
+                    };
+                inlineClaims.set(entry.correlationId, {
+                  owned: true,
+                  step: {
+                    ...startedStep,
+                    input: dehydrated.dehydratedInput,
+                  },
+                  batchPostSentAtMs,
+                  claimCompletedAtMs,
+                });
               }
               continue;
             }
             if (item.status === 409) {
+              if (
+                entry.kind === 'inline-created' ||
+                entry.kind === 'inline-started'
+              ) {
+                // The pair lost its atomic create-claim: a concurrent writer
+                // already owns this step (an earlier delivery's create, or a
+                // racing handler's claim). Recorded as a lost claim — the
+                // caller's executeStep returns `skipped` without running the
+                // body, the same outcome as losing the lazy claim. A World
+                // that folds the pair reports the 409 on both rows (set
+                // twice, harmless); one that evaluates rows independently
+                // has the started row — processed second — decide, which is
+                // exactly the single path's semantics (create lost + claim
+                // won still runs the body; create won + claim lost skips).
+                inlineClaims.set(entry.correlationId, { owned: false });
+                runtimeLogger.info('Inline step pre-claim lost, continuing', {
+                  workflowRunId: runId,
+                  correlationId: entry.correlationId,
+                  message: item.message,
+                });
+                continue;
+              }
               // Same tolerance as the single path's EntityConflictError: a
               // concurrent or earlier delivery already created it.
               runtimeLogger.info(
@@ -1066,42 +1364,135 @@ export async function handleSuspension({
               { status: item.status }
             );
           }
-          // Slot-bump visibility: the batch endpoint has no bump-and-report,
-          // so a foreign event landing between our snapshot and the commit
-          // pushes the whole batch to higher slots WITHOUT handing us the
-          // skipped events. That is the same accepted exposure as a dropped
-          // truncated report on the single path (absorbSkippedSlotReport
-          // drops those whole): the local log continues without the foreign
-          // events and the next reload sees them. Logged so a bump is
-          // diagnosable rather than silent.
-          const firstCommitted = results.find(
-            (item) => item.error === undefined
-          )?.event;
-          if (expectedFirstSlot !== undefined && firstCommitted) {
-            const firstSlot = maxEventSlot([firstCommitted]);
-            if (firstSlot !== undefined && firstSlot > expectedFirstSlot) {
-              runtimeLogger.debug('Batched fan-out committed above snapshot', {
-                workflowRunId: runId,
-                expectedFirstSlot,
-                firstSlot,
-                skipped: firstSlot - expectedFirstSlot,
-              });
-            }
-          }
-          // Advance the expectation past this chunk's committed events so the
-          // next chunk's diagnostic measures only foreign interleaving.
+          // Highest slot this chunk committed: the ceiling the caller folds
+          // into the inline executions' slot snapshot (see
+          // SuspensionHandlerResult.batchCommittedSlotCeiling) and one input
+          // of the interleaving diagnostic.
           const chunkMaxSlot = maxEventSlot(
             results.flatMap((item) =>
               item.error === undefined && item.event ? [item.event] : []
             )
           );
           if (
-            expectedFirstSlot !== undefined &&
             chunkMaxSlot !== undefined &&
-            chunkMaxSlot >= expectedFirstSlot
+            (batchCommittedSlotCeiling === undefined ||
+              chunkMaxSlot > batchCommittedSlotCeiling)
+          ) {
+            batchCommittedSlotCeiling = chunkMaxSlot;
+          }
+          committedCount += results.filter(
+            (item) => item.error === undefined
+          ).length;
+          if (
+            chunkMaxSlot !== undefined &&
+            (maxCommittedSlot === undefined || chunkMaxSlot > maxCommittedSlot)
+          ) {
+            maxCommittedSlot = chunkMaxSlot;
+          }
+        };
+
+        // Publish the chunk's eager steps' queue messages the moment ITS
+        // creates are durable — the per-chunk half of publish-after-create.
+        // Same message shape and step-identity-scoped idempotency key as the
+        // caller's dispatch pass, so anything double-published dedupes.
+        const publishChunkSteps = async (
+          chunk: typeof entries
+        ): Promise<void> => {
+          if (!publishEagerSteps) return;
+          const stepEntries = chunk.filter((entry) => entry.kind === 'step');
+          if (stepEntries.length === 0) return;
+          const traceCarrier = await getStepDispatchTraceCarrier();
+          await Promise.all(
+            stepEntries.map((entry) =>
+              queueMessage(
+                world,
+                // biome-ignore lint/style/noNonNullAssertion: publishEagerSteps implies presence
+                stepDispatch!.queueName,
+                {
+                  runId,
+                  stepId: entry.correlationId,
+                  // biome-ignore lint/style/noNonNullAssertion: set on every 'step' entry at enqueue
+                  stepName: entry.stepName!,
+                  traceCarrier,
+                  requestedAt: new Date(),
+                },
+                {
+                  idempotencyKey: stepDispatchIdempotencyKey(
+                    entry.correlationId,
+                    // biome-ignore lint/style/noNonNullAssertion: set on every 'step' entry at enqueue
+                    entry.stepName!
+                  ),
+                }
+              )
+            )
+          );
+        };
+
+        // Launch every chunk's POST now; chain each chunk's publishes on its
+        // OWN commit. A chunk whose commit rejected keeps its messages
+        // unsent (the rejection fails the delivery; redelivery re-creates
+        // and re-dispatches, deduped by the idempotency keys).
+        const commits = chunks.map((chunk) => commitChunk(chunk));
+        const publishes = chunks.map(async (chunk, index) => {
+          await commits[index];
+          await publishChunkSteps(chunk);
+        });
+
+        const trailing = (async () => {
+          // Let every sibling settle before surfacing the first failure: a
+          // chunk that committed must still get its publishes out even when
+          // another chunk failed, and the caller acks only after this
+          // resolves.
+          const settled = await Promise.allSettled([...commits, ...publishes]);
+          // Foreign-interleaving visibility: the batch endpoint has no
+          // bump-and-report, so events other writers landed between the
+          // snapshot and these commits pushed the fold to higher slots
+          // WITHOUT handing us the skipped events. Same accepted exposure
+          // as a dropped truncated report on the single path: the local
+          // log stays a strict prefix and the next reload observes them.
+          // Logged so a bump is diagnosable rather than silent.
+          if (
+            expectedFirstSlot !== undefined &&
+            maxCommittedSlot !== undefined
           ) {
-            expectedFirstSlot = chunkMaxSlot + 1;
+            const interleaved =
+              maxCommittedSlot - expectedFirstSlot + 1 - committedCount;
+            if (interleaved > 0) {
+              runtimeLogger.debug('Batched fan-out committed above snapshot', {
+                workflowRunId: runId,
+                expectedFirstSlot,
+                maxCommittedSlot,
+                committedCount,
+                interleaved,
+              });
+            }
+          }
+          const failure = settled.find(
+            (outcome): outcome is PromiseRejectedResult =>
+              outcome.status === 'rejected'
+          );
+          if (failure) throw failure.reason;
+        })();
+
+        const pairChunkIndex = chunks.findIndex((chunk) =>
+          chunk.some((entry) => entry.kind === 'inline-started')
+        );
+        if (allowDeferredBatchWork) {
+          // The trailing work is the caller's to join before ack. Attach a
+          // handler now so a rejection that races that join (or a foreground
+          // failure that prevents the caller from ever reaching it) is never
+          // an unhandledRejection — awaiting the promise still observes it.
+          trailing.catch(() => {});
+          deferredBatchWork = trailing;
+          // Only the pair chunk gates the return: its claims are what the
+          // caller starts the inline bodies from. With no pairs there is
+          // nothing the caller's post-return work reads from the commits,
+          // so nothing gates.
+          if (pairChunkIndex >= 0) {
+            await commits[pairChunkIndex];
           }
+        } else {
+          await trailing;
         }
       })()
     );
@@ -1229,6 +1620,9 @@ export async function handleSuspension({
     createdStepCorrelationIds,
     queuedStepCorrelationIds,
     lazyInlineSteps,
+    inlineClaims,
+    batchCommittedSlotCeiling,
+    deferredBatchWork,
     // On hook conflict the caller re-invokes immediately and never reads
     // the wait timeout, so don't report one.
     waitTimeout: hasHookConflict ? undefined : soonestWait,
diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts
index a3586f95de..2b9f8f0fe2 100644
--- a/packages/world-vercel/src/events-v4.ts
+++ b/packages/world-vercel/src/events-v4.ts
@@ -881,14 +881,12 @@ export async function createWorkflowRunEventsBatchV4(
     offset += frame.byteLength;
   }
 
-  // Per-type shape for the span, so mixed batches classify as what they
-  // carry rather than as their first event's type alone.
-  const typeCounts = new Map<string, number>();
-  for (const event of input.events) {
-    typeCounts.set(event.eventType, (typeCounts.get(event.eventType) ?? 0) + 1);
-  }
-
   const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/batch`;
+  // Batch identity attributes (size, per-type shape) live on the
+  // world.events.createBatch span (see instrumentObject); this transport
+  // span carries only wire-level facts. workflow.event.type is deliberately
+  // absent — it names a single event write, and tagging a batch with its
+  // first event's type misclassifies the traffic.
   const response = await fetchV4(
     url,
     { method: 'POST', headers, body },
@@ -896,11 +894,6 @@ export async function createWorkflowRunEventsBatchV4(
     'createEventBatch',
     {
       ...WorkflowEventsTransport('http'),
-      ...WorkflowEventType(input.events[0].eventType),
-      'workflow.batch.size': input.events.length,
-      'workflow.batch.shape': [...typeCounts]
-        .map(([type, count]) => `${type}:${count}`)
-        .join(','),
       'workflow.batch.bytes': body.byteLength,
     }
   );
diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts
index 1d0279229d..a89f096a6e 100644
--- a/packages/world-vercel/src/events.ts
+++ b/packages/world-vercel/src/events.ts
@@ -511,7 +511,7 @@ export async function createWorkflowRunEventBatch(
       { status: 400 }
     );
   }
-  const inputs = events.map(({ event, occurredAt }) => {
+  const inputs = events.map(({ event, occurredAt, computeInstanceId }) => {
     const { payload, meta } = splitEventDataForV4(event);
     return {
       runId,
@@ -521,6 +521,9 @@ export async function createWorkflowRunEventBatch(
       // Under slot identity this is the source of the durable createdAt, so
       // the caller's logical time is what every replay observes.
       occurredAt: occurredAt ?? new Date(),
+      // Per-event compute attribution (pre-claimed inline starts) — rides the
+      // frame meta exactly like the single POST's CreateEventParams field.
+      ...(computeInstanceId !== undefined ? { computeInstanceId } : {}),
       // Batch responses carry entities for bookkeeping, not payload reads —
       // default to lazy refs unless the caller explicitly asks for resolved
       // data (the same `resolveData` mapping the read paths use).
diff --git a/packages/world-vercel/src/instrumentObject.ts b/packages/world-vercel/src/instrumentObject.ts
index cfec02211c..649f776a26 100644
--- a/packages/world-vercel/src/instrumentObject.ts
+++ b/packages/world-vercel/src/instrumentObject.ts
@@ -135,6 +135,32 @@ export function instrumentObject<T extends object>(prefix: string, o: T): T {
           }
         }
 
+        // Batch writes describe themselves by size and per-type shape —
+        // deliberately NOT workflow.event.type, which names a single event
+        // write and would misleadingly tag the whole batch with its first
+        // event. These live here (the world.events.createBatch span), not on
+        // the transport span underneath.
+        let batchAttributes: Record<string, string | number> | undefined;
+        if (prefix === 'world.events' && methodName === 'createBatch') {
+          const events = args[1];
+          if (Array.isArray(events)) {
+            const counts = new Map<string, number>();
+            for (const item of events) {
+              const eventType = (item as { event?: { eventType?: unknown } })
+                ?.event?.eventType;
+              if (typeof eventType === 'string') {
+                counts.set(eventType, (counts.get(eventType) ?? 0) + 1);
+              }
+            }
+            batchAttributes = {
+              'workflow.batch.size': events.length,
+              'workflow.batch.shape': [...counts]
+                .map(([type, count]) => `${type}:${count}`)
+                .join(','),
+            };
+          }
+        }
+
         return trace(
           spanName,
           { kind: await getSpanKind('CLIENT') },
@@ -147,6 +173,7 @@ export function instrumentObject<T extends object>(prefix: string, o: T): T {
               ...RpcService(WORKFLOW_SERVER_SERVICE.rpcService),
               ...RpcMethod(spanName),
               ...extractWorkflowAttributes(prefix, methodName, args),
+              ...batchAttributes,
             });
             const result = await f(...args);
             const resultRunId = getRunIdFromResult(result);
diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts
index 6f0c55ba6a..4784becf55 100644
--- a/packages/world/src/events.ts
+++ b/packages/world/src/events.ts
@@ -1031,6 +1031,13 @@ export interface BatchEventRequest {
    * instant the event logically occurred.
    */
   occurredAt?: Date;
+  /**
+   * Compute-instance attribution for this event, same as the single create's
+   * {@link CreateEventParams.computeInstanceId}. Set on the `step_started`
+   * half of a pre-claimed inline pair so a batched claim attributes the
+   * executing instance exactly like the lazy claim it replaces.
+   */
+  computeInstanceId?: string;
 }
 
 /** Per-batch parameters for {@link Storage.events.createBatch}. */