[core] Slot event identity + main (race-repro run)

.changeset/inline-claim-reclaim.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Keep a batch of inline steps together when one of its event writes loses a race, instead of discarding the batch

.changeset/late-hook-delivery-divergence.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Stop failing runs with a corrupted-event-log error when a hook delivery arrives after the hook was disposed, or when a step result is still being fetched

.changeset/local-slot-order.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-local': patch
+---
+
+Read a position-numbered event log in position order, so a replay sees the log the order it was written

.changeset/slot-correlation-ids.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'@workflow/world': patch
+---
+
+Scope every queue idempotency key to the run, and number step and wait correlation IDs per kind so that inserting one kind no longer renumbers the others.

.changeset/slot-event-identity-client.md

@@ -0,0 +1,9 @@
+---
+'@workflow/world-vercel': minor
+'@workflow/core': minor
+'@workflow/errors': minor
+'@workflow/world': minor
+'workflow': minor
+---
+
+Event creations on runs that number events by slot now claim their own event id and merge, replay and re-claim when a `SlotConflictError` shows another writer took it first.

.changeset/slot-event-identity-worlds.md

@@ -0,0 +1,8 @@
+---
+'@workflow/world': minor
+'@workflow/world-local': minor
+'@workflow/world-postgres': minor
+'@workflow/core': minor
+---
+
+Number a run's events by position in the Local and Postgres Worlds when `WORKFLOW_SLOT_IDENTITY` is set, so a reader can prove its copy of an event log is complete.

.changeset/slot-identity-default-on.md

@@ -0,0 +1,8 @@
+---
+'@workflow/world': minor
+'@workflow/world-vercel': minor
+'@workflow/world-local': minor
+'@workflow/world-postgres': minor
+---
+
+Number new runs' events by position by default, in every World including the Vercel one. Set `WORKFLOW_SLOT_IDENTITY=0` to keep minting ULID event ids.

.changeset/tidy-moons-observe.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Name the divergent event's pending invocations and the fenced member of an inline step batch in replay-divergence logs

.changeset/turbo-run-started-occurred-at.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Report the same workflow start time on an optimistically started run's first pass and its replays

.gitignore

@@ -42,3 +42,6 @@ packages/swc-plugin-workflow/build-hash.json
 workbench/nextjs-*/public/.well-known/workflow
 workbench/sveltekit/static/.well-known/workflow
 
+
+# Local e2e diagnostics dumps
+e2e-diagnostics-*.json

docs/content/docs/v5/api-reference/workflow-errors/index.mdx

@@ -82,6 +82,9 @@ All errors extend [`WorkflowError`](/docs/api-reference/workflow-errors/workflow
     <Card href="/docs/api-reference/workflow-errors/precondition-failed-error" title="PreconditionFailedError">
         Thrown when an event creation is rejected because the client's event-log snapshot is stale.
     </Card>
+    <Card href="/docs/api-reference/workflow-errors/slot-conflict-error" title="SlotConflictError">
+        Thrown when an event creation is rejected because another writer already took the event's slot.
+    </Card>
     <Card href="/docs/api-reference/workflow-errors/too-early-error" title="TooEarlyError">
         Thrown when a request is made before the system is ready to process it.
     </Card>

docs/content/docs/v5/api-reference/workflow-errors/meta.json

@@ -15,6 +15,7 @@
     "throttle-error",
     "entity-conflict-error",
     "precondition-failed-error",
+    "slot-conflict-error",
     "run-expired-error",
     "run-not-supported-error",
     "too-early-error"

docs/content/docs/v5/api-reference/workflow-errors/slot-conflict-error.mdx

@@ -0,0 +1,76 @@
+---
+title: SlotConflictError
+description: Thrown when an event creation is rejected because another writer already took the event's slot.
+type: reference
+summary: Catch SlotConflictError when a world rejects an event creation whose slot in the run's event log was already taken.
+related:
+  - /docs/api-reference/workflow-errors/workflow-world-error
+  - /docs/api-reference/workflow-errors/precondition-failed-error
+---
+
+`SlotConflictError` is thrown by world implementations when an event creation is rejected because the event's slot in the run's event log was already taken by another writer. It corresponds to HTTP 409 Conflict semantics.
+
+On a run that numbers its events by slot, each event's id encodes its position in the log: the first event is slot 1, the second slot 2, and so on. Whoever writes a slot first owns it, so a rejected write proves the client was replaying against an event log that was missing at least one event. Retrying the same write can never succeed — the client has to merge the events it was missing, replay, and propose whatever slot that replay lands on.
+
+The rejection carries the missing events inline so that merge usually costs no extra round-trip:
+
+- `events` — the events recorded after the client's snapshot, in ascending slot order. Empty when the backend could not read them, in which case the client reloads the log itself.
+- `cursor` — cursor to continue the delta from.
+- `hasMore` — whether events beyond `events` remain to be fetched.
+
+This is the slot-numbering counterpart to [`PreconditionFailedError`](/docs/api-reference/workflow-errors/precondition-failed-error), which is how the same staleness is reported for runs guarded by an event-log snapshot watermark instead. A run uses one scheme or the other for its whole life, decided when it is created.
+
+<Callout>
+The Workflow runtime handles this error automatically: it merges the events it was missing, replays, and re-proposes the write at a free slot, ultimately re-enqueueing the run for a fresh replay if it cannot catch up. You will only encounter it when interacting with world storage APIs directly.
+</Callout>
+
+```typescript lineNumbers
+import { SlotConflictError } from "workflow/errors"
+declare const world: { events: { create(...args: any[]): Promise<any> } }; // @setup
+declare const runId: string; // @setup
+declare const event: any; // @setup
+
+try {
+  await world.events.create(runId, event);
+} catch (error) {
+  if (SlotConflictError.is(error)) { // [!code highlight]
+    console.log(`Slot ${error.eventId} taken; ${error.events.length} event(s) missed`);
+  }
+}
+```
+
+## API Signature
+
+### Properties
+
+<TSDoc
+definition={`
+interface SlotConflictError {
+  /** The slot-numbered event id that was already taken. */
+  eventId: string;
+  /** The events recorded after the client's snapshot, in ascending slot order. Empty when the backend could not read them. */
+  events: unknown[];
+  /** Cursor to continue the delta from, or null. */
+  cursor: string | null;
+  /** Whether events beyond \`events\` remain to be fetched. */
+  hasMore: boolean;
+  /** The error message. */
+  message: string;
+}
+export default SlotConflictError;`}
+/>
+
+### Static Methods
+
+#### `SlotConflictError.is(value)`
+
+Type-safe check for `SlotConflictError` instances. Preferred over `instanceof` because it works across module boundaries and VM contexts.
+
+```typescript
+import { SlotConflictError } from "workflow/errors"
+declare const error: unknown; // @setup
+
+if (SlotConflictError.is(error)) {
+  // error is typed as SlotConflictError
+}
+```

docs/content/docs/v5/configuration/runtime-tuning.mdx

@@ -48,6 +48,15 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 - Backends that do not support the guard ignore the snapshot; they must not declare the capability, so guard-dependent optimizations stay off against them even when the flag is set.
 - Set `0` to disable.
 
+### `WORKFLOW_SLOT_IDENTITY`
+
+- Default: enabled
+- Numbers a new run's events by position instead of by ULID: `evnt_…001` is the run's first event, `evnt_…002` its second. Positions are allocated in order, so the log reads in the order it was written regardless of clock skew between writers.
+- Contention becomes explicit rather than silent. Two writers proposing the same position cannot both win: the loser gets a 409 ([`SlotConflictError`](/docs/api-reference/workflow-errors/slot-conflict-error)) carrying the events it was missing, and the runtime merges them, replays, and re-proposes above the events it just learned about.
+- Applies only to runs created while it is enabled. A run keeps the identity scheme it was created with for life, so turning the flag on or off never affects runs already in flight.
+- Requires a World that supports it. A World that does not rejects the run outright rather than mis-numbering its events.
+- Set `0` or `false` to disable, which numbers new runs by ULID as before.
+
 ## Inline execution
 
 ### `WORKFLOW_V2_TIMEOUT_MS`
@@ -189,6 +198,12 @@ These variables are primarily for tests, debugging, or unusual deployments.
 - Delay before the unconsumed-event check fires.
 - Minimum: `10`.
 
+### `WORKFLOW_DEFERRED_CHECK_MAX_GRACE_MS`
+
+- Default: `15000`
+- How long the unconsumed-event check keeps waiting while a step result or hook payload is still on its way to the workflow. An event whose consumer has not been registered yet looks exactly like an orphaned one, so the check waits rather than failing the run.
+- Once this budget is spent the check reports regardless, so a delivery that never lands cannot keep a genuinely orphaned event from being detected.
+
 ### `WORKFLOW_LOCK_POLL_INTERVAL_MS`
 
 - Default: `10`

packages/core/e2e/event-log-race-repro.test.ts

@@ -373,6 +373,30 @@ function validateStormReturn(value: unknown): {
   return { stragglers };
 }
 
+/**
+ * Reads a terminal-failed run's error through `returnValue()`, which hydrates
+ * the stored payload into an Error. Returns undefined when the read itself
+ * fails — the outcome is already known from `errorCode`, so a missing message
+ * degrades the report rather than the classification.
+ */
+async function readFailureMessage(
+  run: Run<unknown>
+): Promise<{ name?: string; message?: string } | undefined> {
+  try {
+    await run.returnValue();
+    return undefined;
+  } catch (err) {
+    if (WorkflowRunFailedError.is(err)) {
+      const cause = err.cause;
+      return {
+        name: cause instanceof Error ? cause.name : err.name,
+        message: cause instanceof Error ? cause.message : err.message,
+      };
+    }
+    return undefined;
+  }
+}
+
 async function pollTerminalRun(
   run: Run<unknown>,
   startedAt: number,
@@ -412,13 +436,20 @@ async function pollTerminalRun(
         errorCode?: string;
         error?: { name?: string; message?: string };
       };
+      // `runs.get` hands back the raw serialized error payload, not an Error, so
+      // reading `.message` off it yields undefined and the report records the
+      // code with no diagnosis. Read the failure through the public
+      // return-value path, which hydrates it. For a corruption that message
+      // carries the divergent event and what the replay was waiting for, which
+      // is the whole reason to keep the report.
+      const hydrated = await readFailureMessage(run);
       return {
         ...base,
         outcome: classifyFailure(failure.errorCode),
         status: runData.status,
         errorCode: failure.errorCode,
-        errorMessage: failure.error?.message,
-        errorName: failure.error?.name,
+        errorMessage: hydrated?.message ?? failure.error?.message,
+        errorName: hydrated?.name ?? failure.error?.name,
         durationMs: Date.now() - startedAt,
       };
     }

packages/core/src/abort-consistency.test.ts

@@ -11,6 +11,7 @@ import type { Event, WorkflowRun } from '@workflow/world';
 import * as nanoid from 'nanoid';
 import { monotonicFactory } from 'ulid';
 import { describe, expect, it, vi } from 'vitest';
+import { createCorrelationIdFactory } from './correlation-ids.js';
 import { EventsConsumer } from './events-consumer.js';
 import type { WorkflowSuspension } from './global.js';
 import type { WorkflowOrchestratorContext } from './private.js';
@@ -45,6 +46,10 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    nextCorrelationId: createCorrelationIdFactory({
+      specVersion: undefined,
+      generateUlid: () => ulid(workflowStartedAt),
+    }),
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),

packages/core/src/abort-controller.test.ts

@@ -12,6 +12,7 @@ import type { Event } from '@workflow/world';
 import * as nanoid from 'nanoid';
 import { monotonicFactory } from 'ulid';
 import { describe, expect, it, vi } from 'vitest';
+import { createCorrelationIdFactory } from './correlation-ids.js';
 import { DEFERRED_CHECK_DELAY_MS, EventsConsumer } from './events-consumer.js';
 import type { WorkflowOrchestratorContext } from './private.js';
 import { ReplayPayloadCache } from './replay-payload-cache.js';
@@ -43,6 +44,10 @@ function setupWorkflowContext(
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    nextCorrelationId: createCorrelationIdFactory({
+      specVersion: undefined,
+      generateUlid: () => ulid(workflowStartedAt),
+    }),
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),

packages/core/src/abort-replay-ordering.test.ts

@@ -27,6 +27,7 @@ import type { Event } from '@workflow/world';
 import * as nanoid from 'nanoid';
 import { monotonicFactory } from 'ulid';
 import { describe, expect, it, vi } from 'vitest';
+import { createCorrelationIdFactory } from './correlation-ids.js';
 import { EventsConsumer } from './events-consumer.js';
 import {
   scheduleWhenIdle,
@@ -78,6 +79,10 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    nextCorrelationId: createCorrelationIdFactory({
+      specVersion: undefined,
+      generateUlid: () => ulid(workflowStartedAt),
+    }),
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),

packages/core/src/async-deserialization-ordering.test.ts

@@ -4,6 +4,7 @@ import * as nanoid from 'nanoid';
 import { monotonicFactory } from 'ulid';
 import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest';
 import { registerSerializationClass } from './class-serialization.js';
+import { createCorrelationIdFactory } from './correlation-ids.js';
 import { EventsConsumer } from './events-consumer.js';
 import type { WorkflowOrchestratorContext } from './private.js';
 import { ReplayPayloadCache } from './replay-payload-cache.js';
@@ -58,6 +59,10 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    nextCorrelationId: createCorrelationIdFactory({
+      specVersion: undefined,
+      generateUlid: () => ulid(workflowStartedAt),
+    }),
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),