[core] Stop sending a slot snapshot on step executor writes

.changeset/drop-executor-slot-snapshot.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'@workflow/world': patch
+---
+
+Stop sending a slot snapshot (`eventCount`) on step executor writes, so a World no longer reads and returns a skipped-slot event page that the executor only discards.

packages/core/src/runtime.test.ts

@@ -2802,17 +2802,18 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => {
     // The handler responds normally: the rejection restarts the replay inside
     // this delivery, never a run_failed.
     expect(res.status).toBe(204);
-    // Step B's claim was issued from a loaded (non-empty) log, so it named the
-    // position it was decided against. (The very first batch of a run loads an
-    // empty log and has no position to name; reporting is best-effort there,
-    // matching the suspension creates.)
+    // Step B's claim names no log position: the executor has no loaded log
+    // to merge a skipped-slot report into, so it sends no `eventCount` and
+    // the World reads no page for it. The fence this test exercises does not
+    // depend on one either; a World that rejects a claim does so from its own
+    // view of the log, as the scripted rejection above does.
     const rejectedClaim = eventsCreate.mock.calls.find(
       (c) =>
         (c[1] as any).eventType === 'step_started' &&
         ((c[1] as any).eventData as { stepName?: string })?.stepName ===
           'deltaGateStepB'
     );
-    expect(typeof (rejectedClaim?.[2] as any)?.eventCount).toBe('number');
+    expect((rejectedClaim?.[2] as any)?.eventCount).toBeUndefined();
     // The fenced claim's body never ran: step B executes exactly once, on the
     // restarted replay whose claim the backend accepted.
     expect(deltaGateBodyRuns).toEqual(['B']);

packages/core/src/runtime.ts

@@ -4334,36 +4334,6 @@ export function workflowEntrypoint(
                           retained: servedByRetainedSession,
                         });
 
-                        // Slot snapshot for the inline step_started claims: the
-                        // lazy claim is the first durable write of a hot-path
-                        // step (its step_created is deferred), so without a
-                        // snapshot it would name no position at all and a stale
-                        // replay could claim (and commit) a step scheduled off
-                        // a view that misses an event it never loaded.
-                        //
-                        // Taken here rather than inside the executor because
-                        // 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 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, so 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
                         // `reported` latch picks the single step that
@@ -4464,7 +4434,6 @@ export function workflowEntrypoint(
                                 // See suppressOptimisticStart above.
                                 suppressOptimisticStart,
                                 runReadyBarrier,
-                                slotSnapshot: inlineClaimSnapshot,
                                 ...(stepIndex === 0 &&
                                 (s.lazyStepInput !== undefined ||
                                   s.preclaimedStart !== undefined) &&

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

@@ -194,8 +194,6 @@ describe('executeStep — compute instance stamping', () => {
     // persist — so observe the call itself rather than the stored event.
     const createSpy = vi.spyOn(world.events, 'create');
 
-    const slotSnapshot = { eventCount: 7 };
-
     await executeStep({
       world,
       workflowRunId: runId,
@@ -204,7 +202,6 @@ describe('executeStep — compute instance stamping', () => {
       requestId: 'req_step_executor',
       stepId,
       stepName,
-      slotSnapshot,
     });
 
     const started = createSpy.mock.calls.filter(
@@ -215,8 +212,9 @@ describe('executeStep — compute instance stamping', () => {
       requestId: 'req_step_executor',
       computeInstanceId: COMPUTE_INSTANCE_ID,
     });
-    // All dimensions ride the same params object and neither may clobber another.
-    expect(started[0]?.[2]?.eventCount).toBe(slotSnapshot.eventCount);
+    // An executor write names no log position: it has no log to merge a
+    // skipped-slot report into, so it must not ask the World to read one.
+    expect(started[0]?.[2]?.eventCount).toBeUndefined();
   });
 
   it('stamps provenance when a lazy unregistered step is materialized', async () => {
@@ -307,10 +305,12 @@ describe('executeStep — compute instance stamping', () => {
     expect(started?.[2]?.requestId).toBeUndefined();
   });
 
-  it('advances the snapshot it sends as its own writes land', async () => {
-    // The executor writes twice for one step. If the second write still named
-    // the position its caller scheduled against, the World would report the
-    // first one back to it on every step, forever.
+  it('sends no slot snapshot on any of its writes', async () => {
+    // The only thing a World does with `eventCount` is bump-and-report: read
+    // the events between the named position and the committed one and hand
+    // them back. The executor has no loaded log to merge that page into, so
+    // naming a position would make the World read a page nobody consumes, on
+    // every contended step_started.
     const world = makeWorld();
     const stepName = uniqueStepName();
     const { runId, stepId } = await setupRunningStep({
@@ -319,13 +319,6 @@ describe('executeStep — compute instance stamping', () => {
       onBody: () => {},
     });
 
-    // The position the caller would have scheduled against, taken from the log
-    // rather than written down, so the seed stays below the slots the executor
-    // is about to commit at. Seeding it above them would leave `observeSlot`
-    // with nothing to raise and the test would pass without exercising it.
-    const { data: seeded } = await world.events.list({ runId });
-    const scheduledAt = seeded.length;
-
     const createSpy = vi.spyOn(world.events, 'create');
 
     await executeStep({
@@ -335,18 +328,11 @@ describe('executeStep — compute instance stamping', () => {
       workflowStartedAt: Date.now(),
       stepId,
       stepName,
-      slotSnapshot: { eventCount: scheduledAt },
     });
 
-    // world-local mints slots for a run created on this scheme, so every write
-    // reads back as a position and each one has to name the position its
-    // predecessor landed on.
     const counts = createSpy.mock.calls.map((call) => call[2]?.eventCount);
     expect(counts.length).toBeGreaterThan(1);
-    expect(counts[0]).toBe(scheduledAt);
-    for (let i = 1; i < counts.length; i++) {
-      expect(counts[i]).toBeGreaterThan(counts[i - 1] as number);
-    }
+    expect(counts.every((count) => count === undefined)).toBe(true);
   });
 });
 

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

@@ -25,7 +25,6 @@ import type {
   World,
 } from '@workflow/world';
 import {
-  requireEventSlot,
   SPEC_VERSION_CURRENT,
   SPEC_VERSION_SUPPORTS_COMPRESSION,
 } from '@workflow/world';
@@ -55,11 +54,7 @@ import {
   isOptimisticInlineStartExplicitlyDisabled,
 } from './constants.js';
 import { getPortLazy } from './get-port-lazy.js';
-import {
-  maxEventSlot,
-  memoizeEncryptionKey,
-  type SlotSnapshotParams,
-} from './helpers.js';
+import { memoizeEncryptionKey } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
 import {
   computeResumeTtrAttributes,
@@ -163,26 +158,6 @@ export interface StepExecutorParams {
    * handler is the sole inline writer for the run on this iteration.
    */
   inlineDeltaSinceCursor?: string;
-  /**
-   * How much of the run's log the caller's replay had loaded when it scheduled
-   * this step, as the highest slot that log occupies. Seeds the snapshot every
-   * write this executor makes carries; each committed event advances it to its
-   * own slot, so a later write never names a position that predates an earlier
-   * one from the same step.
-   *
-   * It matters most on the lazy inline path, where the `step_started` claim is
-   * the step's FIRST durable write (its `step_created` is deferred): without a
-   * seed the claim would name no position at all, and a replay working from a
-   * stale view could claim (and then commit) a step scheduled without
-   * observing an event it never loaded.
-   *
-   * A World that fences rejects a stale claim with `PreconditionFailedError`
-   * (412); executeStep does NOT translate that rejection (re-claiming in place
-   * would still commit the stale schedule), so it propagates for the caller to
-   * abandon the batch and restart its replay. Undefined for a caller with
-   * nothing loaded.
-   */
-  slotSnapshot?: SlotSnapshotParams;
   /**
    * Suppress optimistic inline start for this step regardless of
    * `WORKFLOW_OPTIMISTIC_INLINE_START` / `forceOptimisticStart`: take the
@@ -348,59 +323,22 @@ export async function executeStep(
     (params.runSpecVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION;
   const replayRecoveryReporter =
     params.replayRecoveryReporter ?? ReplayRecoveryReporter.inert();
-  /**
-   * The highest log slot this executor knows about, seeded from the view its
-   * caller scheduled the step against and advanced by every event it commits.
-   *
-   * Advancing is what keeps the snapshot honest across a step's own writes. A
-   * step commits `step_started` and then `step_completed`; if the second still
-   * named the caller's original position, the World would report the first one
-   * back as an event this writer had not seen, on every step, forever.
-   *
-   * This reads a report for its highest position and then discards it, where
-   * the replay loop and the suspension handler merge theirs with
-   * `absorbSkippedSlotReport`. That is the difference between the callers, not
-   * an oversight: an executor holds no loaded log to merge into. It runs from a
-   * queued delivery whose only view of the log is the integer its caller passed
-   * in, so the position is the entire value the report has to it. Whoever
-   * replays next loads the log and gets the events themselves.
-   */
-  let knownSlot = params.slotSnapshot?.eventCount;
-  const observeSlot = (result: { event?: Event; events?: Event[] }): void => {
-    if (knownSlot === undefined) {
-      // The caller scheduled this step without naming a position, so there is
-      // no snapshot to advance and the writes below send none. Not the same as
-      // a run without positions: every run has them, this executor was not
-      // told which one it started from.
-      return;
-    }
-    const observed: number[] = [];
-    if (result.event) {
-      observed.push(requireEventSlot(result.event.eventId));
-    }
-    const reported = maxEventSlot(result.events ?? []);
-    if (reported !== undefined) {
-      observed.push(reported);
-    }
-    for (const slot of observed) {
-      if (slot > knownSlot) {
-        knownSlot = slot;
-      }
-    }
-  };
+  // Executor writes carry no slot snapshot (`CreateEventParams.eventCount`).
+  // The only thing a World does with one is bump-and-report: when the write
+  // lands above the position named, it reads the events in between and hands
+  // them back. The replay loop and the suspension handler merge that page into
+  // their loaded log; this executor has no log to merge into, so the page was
+  // read only to be discarded, and in production that read fell on a third of
+  // all `step_started` writes. Omitting the count is the documented shape for
+  // a caller with no loaded log to be stale against, and the conditional
+  // write on (runId, correlationId) remains the ownership fence.
   const createEvent = async <T extends CreateEventRequest>(
     data: T,
     eventParams?: CreateEventParams
-  ) => {
-    const result = await replayRecoveryReporter.withEventCreate(
-      knownSlot === undefined
-        ? eventParams
-        : { eventCount: knownSlot, ...eventParams },
-      (p) => world.events.create(workflowRunId, data, p)
+  ) =>
+    replayRecoveryReporter.withEventCreate(eventParams, (p) =>
+      world.events.create(workflowRunId, data, p)
     );
-    observeSlot(result);
-    return result;
-  };
 
   // `step_started` identifies the invocation that performed this attempt.
   // Keep request and compute provenance independent: world-vercel serializes
@@ -709,9 +647,7 @@ export async function executeStep(
     }
 
     let step: StartedStep;
-    // Params for the `step_started` create on either path below. The slot
-    // snapshot is not spread here: `createEvent` attaches it to every write,
-    // this one included.
+    // Params for the `step_started` create on either path below.
     const startEventParams = stepStartedEventParams;
     // `Date.now()` taken immediately before the `step_started` create is
     // issued (either path below); anchors RSFS's end point. See
@@ -785,10 +721,9 @@ export async function executeStep(
                   : {}),
               },
             },
-            // Guard the claim; see StepExecutorParams.slotSnapshot. A
-            // stale (412) rejection surfaces via reconcileOptimisticStart as a
-            // non-translatable error: the body result is discarded and the
-            // rejection propagates to the caller.
+            // A 412 rejection from a fencing World surfaces via
+            // reconcileOptimisticStart as a non-translatable error: the body
+            // result is discarded and the rejection propagates to the caller.
             startEventParams
           );
         }
@@ -845,10 +780,9 @@ export async function executeStep(
                   }
                 : { stepName, ...ownershipStamp },
           },
-          // Guard the claim; see StepExecutorParams.slotSnapshot. A
-          // stale (412) rejection is intentionally NOT translated by
-          // startErrorToResult below, so it propagates to the caller for a
-          // fresh replay.
+          // A 412 rejection from a fencing World is intentionally NOT
+          // translated by startErrorToResult below, so it propagates to the
+          // caller for a fresh replay.
           startEventParams
         );
         stepClaimCompletedAtMs = Date.now();

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

@@ -1584,7 +1584,6 @@ describe('handleSuspension batched fan-out', () => {
 
     expect(result.inlineClaims.size).toBe(0);
     expect(result.deferredBatchWork).toBeUndefined();
-    expect(result.batchCommittedSlotCeiling).toBeUndefined();
     // The deferred inline step still carries its input for the lazy start.
     expect(result.lazyInlineSteps).toHaveLength(1);
     expect(result.lazyInlineSteps[0].correlationId).toBe('s1');
@@ -1788,8 +1787,6 @@ describe('handleSuspension batched fan-out', () => {
         's1',
         's2',
       ]);
-      // 6 events at slots 10..15.
-      expect(result.batchCommittedSlotCeiling).toBe(15);
       expect(eventsCreate).not.toHaveBeenCalled();
     });