Slot event identity (SDK 1/4): per-kind correlation ids and run-scoped idempotency keys

.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.

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())
     ),

packages/core/src/correlation-ids.test.ts

@@ -0,0 +1,92 @@
+import {
+  FIRST_SLOT,
+  SLOT_ID_WIDTH,
+  SPEC_VERSION_CURRENT,
+  SPEC_VERSION_SLOT_IDENTITY,
+  slotFromId,
+} from '@workflow/world';
+import { describe, expect, it } from 'vitest';
+import { createCorrelationIdFactory } from './correlation-ids.js';
+
+/** Stands in for the invocation's seeded, replay-stable ULID generator. */
+function fakeUlids(): () => string {
+  let issued = 0;
+  return () => `01ULID${String(++issued).padStart(20, '0')}`;
+}
+
+function slotFactory() {
+  return createCorrelationIdFactory({
+    specVersion: SPEC_VERSION_SLOT_IDENTITY,
+    generateUlid: fakeUlids(),
+  });
+}
+
+describe('createCorrelationIdFactory', () => {
+  describe('slot identity', () => {
+    it('numbers each kind densely from the first slot', () => {
+      const next = slotFactory();
+      expect(slotFromId(next('step'))).toBe(FIRST_SLOT);
+      expect(slotFromId(next('step'))).toBe(FIRST_SLOT + 1);
+      expect(slotFromId(next('wait'))).toBe(FIRST_SLOT);
+    });
+
+    it('keeps the prefix of the kind it was asked for', () => {
+      const next = slotFactory();
+      expect(next('step')).toMatch(
+        new RegExp(`^step_[0-9]{${SLOT_ID_WIDTH}}$`)
+      );
+      expect(next('wait')).toMatch(
+        new RegExp(`^wait_[0-9]{${SLOT_ID_WIDTH}}$`)
+      );
+    });
+
+    it('issues the same sequence to two fresh invocations', () => {
+      // Replay stability: the VM is rebuilt per replay and the workflow body
+      // issues its operations in the same order, so nothing needs seeding.
+      const replay = () => {
+        const next = slotFactory();
+        return [next('step'), next('wait'), next('step'), next('step')];
+      };
+      expect(replay()).toEqual(replay());
+    });
+
+    it('does not renumber steps or waits when another kind allocates', () => {
+      // Kinds that stay on ULIDs (hooks, attributes) draw from generateUlid,
+      // and a per-kind counter means interleaving them cannot shift a step's
+      // number — which a single shared sequence would.
+      const ulids = fakeUlids();
+      const next = createCorrelationIdFactory({
+        specVersion: SPEC_VERSION_SLOT_IDENTITY,
+        generateUlid: ulids,
+      });
+      const firstStep = next('step');
+      ulids();
+      ulids();
+      const secondStep = next('step');
+      expect(slotFromId(firstStep)).toBe(FIRST_SLOT);
+      expect(slotFromId(secondStep)).toBe(FIRST_SLOT + 1);
+      expect(slotFromId(next('wait'))).toBe(FIRST_SLOT);
+    });
+  });
+
+  describe('ULID identity', () => {
+    it('draws from the invocation generator for every kind', () => {
+      const next = createCorrelationIdFactory({
+        specVersion: SPEC_VERSION_CURRENT,
+        generateUlid: fakeUlids(),
+      });
+      // One shared sequence, exactly as before slots existed: an id's number
+      // reflects the order of allocation across all kinds.
+      expect(next('step')).toBe('step_01ULID00000000000000000001');
+      expect(next('wait')).toBe('wait_01ULID00000000000000000002');
+    });
+
+    it('treats a run with no spec version as ULID-numbered', () => {
+      const next = createCorrelationIdFactory({
+        specVersion: undefined,
+        generateUlid: fakeUlids(),
+      });
+      expect(slotFromId(next('step'))).toBeUndefined();
+    });
+  });
+});

packages/core/src/correlation-ids.ts

@@ -0,0 +1,65 @@
+/**
+ * Correlation-id allocation for a single workflow invocation.
+ *
+ * A correlation id names an operation the workflow body issued — a step call, a
+ * sleep — and must come out identical on every replay of that run, because it
+ * is how a replay recognises the event that already recorded the operation.
+ *
+ * Two schemes exist. A run on ULID identity draws from the invocation's seeded
+ * monotonic ULID generator, which is replay-stable because its seed and initial
+ * clock are derived from the run. A run on slot identity counts: the first step
+ * of the run is `step_…001`, the second `step_…002`, zero-padded to ULID width
+ * (see `slotIdBody`). Which scheme applies is fixed by the run's persisted
+ * `specVersion` and never by the build, so a run started before slot identity
+ * keeps proposing the ids its log already holds.
+ *
+ * Counters are **per kind**, and there is nothing to seed them with. The VM is
+ * rebuilt for every replay and the workflow body issues its operations in the
+ * same order every time, which is the same argument that licenses the seeded
+ * ULID sequence today. Recovering counters from the loaded log would be actively
+ * wrong: the n-th step's id would then depend on how much of the log this
+ * replay happened to load.
+ *
+ * Per-kind is a strict improvement over the shared ULID sequence. Today all
+ * four id kinds draw from one generator, so introducing a hook allocation
+ * renumbers every step and wait issued after it; separate counters mean a step's
+ * number depends only on the steps before it.
+ *
+ * Hook and attribute ids stay on ULIDs and are not allocated here. Both are
+ * written from outside the VM in cases where no counter exists (an attribute set
+ * on a run from the outside), and both already carry their own per-run
+ * idempotency, so slots would buy them nothing.
+ */
+
+import { FIRST_SLOT, slotIdBody, usesSlotIdentity } from '@workflow/world';
+
+/** Operation kinds whose correlation ids are allocated per run and per kind. */
+export type CorrelationKind = 'step' | 'wait';
+
+/**
+ * Allocates the next correlation id for a kind, prefix included. Returning the
+ * finished id — rather than a number or a bare body — keeps the prefix from
+ * ever diverging from the counter it was drawn against.
+ */
+export type CorrelationIdFactory = (kind: CorrelationKind) => string;
+
+export function createCorrelationIdFactory({
+  specVersion,
+  generateUlid,
+}: {
+  /** The run's *persisted* spec version. */
+  specVersion: number | undefined;
+  /** The invocation's replay-stable ULID generator. */
+  generateUlid: () => string;
+}): CorrelationIdFactory {
+  if (!usesSlotIdentity(specVersion)) {
+    return (kind) => `${kind}_${generateUlid()}`;
+  }
+
+  const allocated = new Map<CorrelationKind, number>();
+  return (kind) => {
+    const slot = (allocated.get(kind) ?? FIRST_SLOT - 1) + 1;
+    allocated.set(kind, slot);
+    return `${kind}_${slotIdBody(slot)}`;
+  };
+}

packages/core/src/delivery-barrier-coverage.test.ts

@@ -45,6 +45,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 { WorkflowSuspension } from './global.js';
 import {
@@ -86,6 +87,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/hook-sleep-interaction.test.ts

@@ -4,6 +4,7 @@ import type { Event } from '@workflow/world';
 import * as nanoid from 'nanoid';
 import { monotonicFactory } from 'ulid';
 import { afterEach, describe, expect, it, vi } from 'vitest';
+import { createCorrelationIdFactory } from './correlation-ids.js';
 import { EventsConsumer } from './events-consumer.js';
 import { WorkflowSuspension } from './global.js';
 import type { WorkflowOrchestratorContext } from './private.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())
     ),

packages/core/src/private.ts

@@ -4,6 +4,7 @@
 
 import { withResolvers } from '@workflow/utils';
 import type { WorldCapabilities } from '@workflow/world';
+import type { CorrelationIdFactory } from './correlation-ids.js';
 import type { EventsConsumer } from './events-consumer.js';
 import type { QueueItem } from './global.js';
 import type { ReplayPayloadCache } from './replay-payload-cache.js';
@@ -142,6 +143,13 @@ export interface WorkflowOrchestratorContext {
    */
   invocationsQueue: Map<string, QueueItem>;
   onWorkflowError: (error: Error) => void;
+  /**
+   * Allocates the correlation id for a step or wait the workflow body just
+   * issued. Replay-stable, and the only place those ids are minted — see
+   * `correlation-ids.ts` for why the two kinds count separately and why
+   * nothing seeds them from the loaded log.
+   */
+  nextCorrelationId: CorrelationIdFactory;
   generateUlid: () => string;
   generateNanoid: () => string;
   /**

packages/core/src/runtime.ts

@@ -70,6 +70,7 @@ import {
   queueMessage,
   withHealthCheck,
 } from './runtime/helpers.js';
+import { runScopedKey } from './runtime/idempotency-key.js';
 import {
   handleReplayBudgetExhausted,
   ReplayBudget,
@@ -2546,7 +2547,10 @@ export function workflowEntrypoint(
                                 },
                                 {
                                   delaySeconds: backstopDelaySeconds,
-                                  idempotencyKey: backstopIdempotencyKey(step),
+                                  idempotencyKey: backstopIdempotencyKey(
+                                    runId,
+                                    step
+                                  ),
                                 }
                               )
                             );
@@ -2564,7 +2568,10 @@ export function workflowEntrypoint(
                                 requestedAt: new Date(),
                               },
                               {
-                                idempotencyKey: step.correlationId,
+                                idempotencyKey: runScopedKey(
+                                  runId,
+                                  step.correlationId
+                                ),
                               }
                             )
                           );
@@ -2580,6 +2587,7 @@ export function workflowEntrypoint(
                                 requestedAt: new Date(),
                               },
                               getWaitContinuationDispatch(
+                                runId,
                                 suspensionResult.waitTimeout.seconds,
                                 suspensionResult.waitTimeout.correlationId
                               )
@@ -3151,7 +3159,7 @@ export function workflowEntrypoint(
                                   // correlationId so it dedupes against the
                                   // keyed re-dispatch the suspension handler
                                   // performs on replay (it also uses
-                                  // `idempotencyKey: step.correlationId`).
+                                  // the same run-scoped correlationId key).
                                   //
                                   // Without this, a mixed batch where one step
                                   // `completed` with unflushed background ops
@@ -3170,7 +3178,10 @@ export function workflowEntrypoint(
                                   // retry body could run early/concurrently.
                                   // Sharing the key lets the earlier delayed
                                   // message win, honoring the backoff.
-                                  idempotencyKey: step.correlationId,
+                                  idempotencyKey: runScopedKey(
+                                    runId,
+                                    step.correlationId
+                                  ),
                                 }
                               )
                             )