VaguelySerious · Jul 27, 2026 · Jul 27, 2026 · Jul 28, 2026 · Jul 27, 2026 · Jul 27, 2026
@@ -0,0 +1,6 @@
+---
+'workflow': minor
+'@workflow/core': minor
+---
+
+Derive correlation IDs from the call site that created the entity instead of from a per-run sequence, so two concurrent replays of the same run agree on IDs even when they disagree about an event. Set `WORKFLOW_CALLSITE_CORRELATION_IDS=0` to opt back into the positional sequence.
@@ -0,0 +1,2 @@
+---
+---
@@ -280,6 +280,9 @@ function renderConfigScale(config) {
   const scenarios = [
     config.stepStormAttempts ? `step-storm ${config.stepStormAttempts}` : '',
     config.hookStormAttempts ? `hook-storm ${config.hookStormAttempts}` : '',
+    config.blockedBranchAttempts
+      ? `blocked-branch ${config.blockedBranchAttempts}`
+      : '',
     config.hookSleepAttempts ? `hook-sleep ${config.hookSleepAttempts}` : '',
     // Historical entries from the pre-storm harness, kept so an old sticky
     // comment still renders its own configuration rather than a blank line.
@@ -309,6 +312,9 @@ function renderConfigTiming(config) {
       ? `step ${config.stepDelayMs}±${config.stepDelayJitterMs ?? 0}ms`
       : '',
     config.hookResumeStaggerMs ? `stagger ${config.hookResumeStaggerMs}ms` : '',
+    config.resumeBurstOffsetMs && config.blockedBranchAttempts
+      ? `burst ${config.resumeBurstOffsetMs}+${config.resumeBurstJitterMs ?? 0}ms`
+      : '',
     config.pokeIntervalMs ? `poke ${config.pokeIntervalMs}ms` : '',
     // The cap belongs next to the cadence it bounds: it BINDS on the local
     // lanes (every step-storm run there reaches it), so a config line that
@@ -331,6 +337,7 @@ function compactConfig(config = {}) {
     attempts: config.attempts,
     stepStormAttempts: config.stepStormAttempts,
     hookStormAttempts: config.hookStormAttempts,
+    blockedBranchAttempts: config.blockedBranchAttempts,
     hookSleepAttempts: config.hookSleepAttempts,
     concurrency: config.concurrency,
     rounds: config.rounds,
@@ -343,6 +350,10 @@ function compactConfig(config = {}) {
     pokeIntervalMs: config.pokeIntervalMs,
     pokeMax: config.pokeMax,
     hookResumeStaggerMs: config.hookResumeStaggerMs,
+    launchStaggerMs: config.launchStaggerMs,
+    resumeBurstOffsetMs: config.resumeBurstOffsetMs,
+    resumeBurstJitterMs: config.resumeBurstJitterMs,
+    blockedBranchWatchdogMs: config.blockedBranchWatchdogMs,
     runTimeoutMs: config.runTimeoutMs,
     // Pre-storm harness keys, retained so history rows recorded by an older
     // revision of this script keep rendering their own configuration.

@@ -43,6 +43,21 @@ on:
       hook_resume_stagger_ms:
         description: 'hook-storm: per-index delay inside a round resume burst. width * this should exceed watchdog_ms'
         required: false
+      blocked_branch_attempts:
+        description: 'Number of blocked-branch runs (launch step before the hook/watchdog race; resume burst aimed at the launch-completion tail)'
+        required: false
+      launch_stagger_ms:
+        description: 'blocked-branch: per-index spacing of launch step durations, so a round completions commit spread out'
+        required: false
+      resume_burst_offset_ms:
+        description: 'blocked-branch: delay from a round hooks existing to the resume burst. Aim at step_delay_ms + (width-1) * launch_stagger_ms'
+        required: false
+      resume_burst_jitter_ms:
+        description: 'blocked-branch: uniform jitter on the burst offset, so rounds sweep the window'
+        required: false
+      blocked_branch_watchdog_ms:
+        description: 'blocked-branch: hook-race watchdog. Size it to normally lose to the burst; it exists to enter a wait into the race'
+        required: false
       poke_interval_ms:
         description: 'step-storm: cadence of out-of-band resumes to the never-read poke hook'
         required: false
@@ -133,6 +148,11 @@ jobs:
           EVENT_LOG_RACE_REPRO_WATCHDOG_MS: ${{ inputs.watchdog_ms }}
           EVENT_LOG_RACE_REPRO_STEP_DELAY_MS: ${{ inputs.step_delay_ms }}
           EVENT_LOG_RACE_REPRO_HOOK_RESUME_STAGGER_MS: ${{ inputs.hook_resume_stagger_ms }}
+          EVENT_LOG_RACE_REPRO_BLOCKED_BRANCH_ATTEMPTS: ${{ inputs.blocked_branch_attempts }}
+          EVENT_LOG_RACE_REPRO_LAUNCH_STAGGER_MS: ${{ inputs.launch_stagger_ms }}
+          EVENT_LOG_RACE_REPRO_RESUME_BURST_OFFSET_MS: ${{ inputs.resume_burst_offset_ms }}
+          EVENT_LOG_RACE_REPRO_RESUME_BURST_JITTER_MS: ${{ inputs.resume_burst_jitter_ms }}
+          EVENT_LOG_RACE_REPRO_BLOCKED_BRANCH_WATCHDOG_MS: ${{ inputs.blocked_branch_watchdog_ms }}
           EVENT_LOG_RACE_REPRO_POKE_INTERVAL_MS: ${{ inputs.poke_interval_ms }}
           EVENT_LOG_RACE_REPRO_POKE_MAX: ${{ inputs.poke_max }}
           EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS: ${{ inputs.run_timeout_ms }}

@@ -115,6 +115,14 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 - Delay before a re-invocation caused by a rejected event creation.
 - Unlike an in-process restart, which re-reads immediately, a re-invocation only happens once the in-process budget failed to catch up — so the delay gives the other writers a moment to quiesce.
 
+### `WORKFLOW_CALLSITE_CORRELATION_IDS`
+
+- Default: enabled
+- Experimental. Derives the correlation ID of each step, hook, wait, and attribute write from its call site — the step name plus a fingerprint of its arguments, or a hook's token — combined with a per-call-site counter.
+- By default every correlation ID is instead the next draw of one per-run sequence, which makes an ID an ordinal: two concurrent replays of the same run that disagree about a single event assign different IDs to every entity after it, so one replay writes events the other can neither match nor consume, and the run fails with `CORRUPTED_EVENT_LOG`. Call-site addressing makes the same entity get the same ID in both replays, so the late write collides with the entity it duplicates instead of renaming everything downstream.
+- Correlation IDs of runs created before the setting changed are not affected: a run keeps replaying on the deployment it started on. On platforms without that guarantee, only enable it while no runs are in flight — a replay that switches schemes mid-run cannot consume its own earlier events.
+- Set `0` to opt back into the positional per-run sequence.
+
 ## Inline execution
 
 ### `WORKFLOW_V2_TIMEOUT_MS`

@@ -57,7 +57,7 @@ const RESULT_PATH = path.resolve(
 const STORM_WORKFLOW_FILE = 'workflows/103_event_log_corruption_repro.ts';
 const CONTROL_WORKFLOW_FILE = 'workflows/101_hook_sleep_repro.ts';
 
-type Scenario = 'step-storm' | 'hook-storm' | 'hook-sleep';
+type Scenario = 'step-storm' | 'hook-storm' | 'blocked-branch' | 'hook-sleep';
 
 type Outcome =
   | 'completed'
@@ -76,6 +76,7 @@ type Outcome =
 interface ReproConfig {
   stepStormAttempts: number;
   hookStormAttempts: number;
+  blockedBranchAttempts: number;
   hookSleepAttempts: number;
   concurrency: number;
   /** Wall-clock budget for *launching* attempts. Once it is spent no new
@@ -118,6 +119,21 @@ interface ReproConfig {
    *  so the burst straddles `watchdogMs` and the straggler count varies. */
   hookResumeStaggerMs: number;
   hookResumeOffsetMs: number;
+  /** `blocked-branch`: per-index spacing of the launch step's duration, so a
+   *  round's launch completions commit spread out instead of in one batch. */
+  launchStaggerMs: number;
+  /** `blocked-branch`: delay from a round's hooks existing to the resume
+   *  burst. Aim it at the tail of the launch-completion spread — the burst has
+   *  to land while a sibling's launch `step_completed` is committing, so a
+   *  hook-woken replay can hold a log that ends just before it. */
+  resumeBurstOffsetMs: number;
+  /** `blocked-branch`: uniform jitter added to the burst offset per round, so
+   *  attempts sweep the window instead of betting on one alignment. */
+  resumeBurstJitterMs: number;
+  /** `blocked-branch`: watchdog for the hook race. Sized to normally LOSE to
+   *  the resume burst — the watchdog's role here is to enter a wait entity
+   *  into the race (the entity whose ordinal gets stolen), not to fire. */
+  blockedBranchWatchdogMs: number;
   /** `hook-sleep` control knobs. */
   iterations: number;
   sleepMs: number;
@@ -185,6 +201,10 @@ function envBoolean(name: string, fallback: boolean) {
 const config: ReproConfig = {
   stepStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_STEP_STORM_ATTEMPTS', 6),
   hookStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_HOOK_STORM_ATTEMPTS', 6),
+  blockedBranchAttempts: envNumber(
+    'EVENT_LOG_RACE_REPRO_BLOCKED_BRANCH_ATTEMPTS',
+    6
+  ),
   hookSleepAttempts: envNumber('EVENT_LOG_RACE_REPRO_ATTEMPTS', 2),
   // Cross-run concurrency is throughput only — the race being reproduced is
   // between concurrent replays *within* one run, driven by `rounds`/`width` and
@@ -225,6 +245,23 @@ const config: ReproConfig = {
     'EVENT_LOG_RACE_REPRO_HOOK_RESUME_OFFSET_MS',
     0
   ),
+  launchStaggerMs: envNumber('EVENT_LOG_RACE_REPRO_LAUNCH_STAGGER_MS', 300),
+  // Default aim: the last launch completes at roughly
+  // `stepDelayMs + (width - 1) * launchStaggerMs` after the round's suspension
+  // (2200 + 7*300 = 4300ms at the defaults) plus dispatch latency; the jitter
+  // sweeps the burst across that tail.
+  resumeBurstOffsetMs: envNumber(
+    'EVENT_LOG_RACE_REPRO_RESUME_BURST_OFFSET_MS',
+    4000
+  ),
+  resumeBurstJitterMs: envNumber(
+    'EVENT_LOG_RACE_REPRO_RESUME_BURST_JITTER_MS',
+    1200
+  ),
+  blockedBranchWatchdogMs: envNumber(
+    'EVENT_LOG_RACE_REPRO_BLOCKED_BRANCH_WATCHDOG_MS',
+    8000
+  ),
   iterations: envNumber('EVENT_LOG_RACE_REPRO_ITERATIONS', 8),
   sleepMs: envNumber('EVENT_LOG_RACE_REPRO_SLEEP_MS', 5000),
   resumeDelayMs: envNumber('EVENT_LOG_RACE_REPRO_RESUME_DELAY_MS', 15_000),
@@ -766,6 +803,106 @@ async function runHookStormAttempt(attempt: number): Promise<ReproRunResult> {
   }
 }
 
+/**
+ * `blocked-branch`: each branch parks on a launch step BEFORE racing its hook
+ * against a watchdog, so a hook-woken replay can hold a log that ends just
+ * before a sibling's launch completion — that sibling then contributes zero
+ * correlation-id draws (not even its watchdog wait), and the woken branch's
+ * finalize takes the ordinal a fresher writer gives the sibling's wait. The
+ * driver's job is to land the whole resume burst (no stagger — the production
+ * shape was near-simultaneous callbacks) at the tail of the round's
+ * launch-completion spread, jittered per round so attempts sweep the window.
+ */
+async function runBlockedBranchAttempt(
+  attempt: number
+): Promise<ReproRunResult> {
+  const scenario: Scenario = 'blocked-branch';
+  const startedAt = Date.now();
+  const token = makeToken(scenario, attempt);
+
+  try {
+    const workflow = await getWorkflowMetadata(
+      deploymentUrl,
+      STORM_WORKFLOW_FILE,
+      'blockedBranchReproWorkflow'
+    );
+    const run = await start(
+      scenario,
+      STORM_WORKFLOW_FILE,
+      'blockedBranchReproWorkflow',
+      workflow,
+      [
+        {
+          ...stormInput(token),
+          launchStaggerMs: config.launchStaggerMs,
+          watchdogMs: config.blockedBranchWatchdogMs,
+        },
+      ]
+    );
+
+    const { runResult, state } = await drive(
+      run,
+      startedAt,
+      scenario,
+      async (driverState) => {
+        for (let round = 0; round < config.rounds; round += 1) {
+          if (driverState.done) return;
+
+          // Round `n`'s hooks only exist once round `n - 1` finished, so this
+          // wait doubles as the round barrier. The hooks are created by the
+          // round's first suspension, which is also when the launch steps
+          // dispatch — so the burst offset below is measured from (roughly)
+          // launch dispatch.
+          await waitForHook(`${token}:${round}:0`, run.runId, driverState);
+          const jitter =
+            config.resumeBurstJitterMs > 0
+              ? Math.floor(Math.random() * config.resumeBurstJitterMs)
+              : 0;
+          await sleep(config.resumeBurstOffsetMs + jitter);
+          if (driverState.done) return;
+
+          // One burst, no stagger: every resume both settles a branch and
+          // wakes its own invocation, and the invocations' event loads race
+          // the last launch completions' commits.
+          await Promise.all(
+            Array.from({ length: config.width }, async (_, index) => {
+              let hook: Awaited<ReturnType<typeof getHookByToken>>;
+              try {
+                hook = await waitForHook(
+                  `${token}:${round}:${index}`,
+                  run.runId,
+                  driverState,
+                  5000
+                );
+              } catch {
+                // The branch's watchdog already fired and disposed the hook.
+                driverState.resumesFailed += 1;
+                return;
+              }
+              await tryResume(driverState, hook, {
+                index,
+                round,
+                sentAt: Date.now(),
+              });
+            })
+          );
+        }
+      }
+    );
+
+    return await finishStormAttempt(
+      run,
+      runResult,
+      state,
+      attempt,
+      scenario,
+      token
+    );
+  } catch (err) {
+    return harnessFailure(scenario, attempt, token, startedAt, err);
+  }
+}
+
 async function finishStormAttempt(
   run: Run<unknown>,
   runResult: ReproRunResult,
@@ -963,6 +1100,7 @@ function summarizeByScenario(results: ReproRunResult[]) {
     {
       'step-storm': emptyOutcomeCounts(),
       'hook-storm': emptyOutcomeCounts(),
+      'blocked-branch': emptyOutcomeCounts(),
       'hook-sleep': emptyOutcomeCounts(),
     }
   );
@@ -979,6 +1117,7 @@ const collected: ReproRunResult[] = [];
 const plannedAttempts =
   config.stepStormAttempts +
   config.hookStormAttempts +
+  config.blockedBranchAttempts +
   config.hookSleepAttempts;
 let overallDeadline = Number.POSITIVE_INFINITY;
 let launchDeadline = Number.POSITIVE_INFINITY;
@@ -1097,6 +1236,34 @@ describe('event log race repro', { retry: 0 }, () => {
       );
     }
 
+    // blocked-branch: the burst must land inside the launch-completion spread
+    // (after the first completion, not after the watchdogs), or no replay can
+    // be missing a sibling's completion while holding a woken branch.
+    const lastLaunchMs =
+      config.stepDelayMs + (config.width - 1) * config.launchStaggerMs;
+    if (
+      config.resumeBurstOffsetMs + config.resumeBurstJitterMs <=
+      config.stepDelayMs
+    ) {
+      console.warn(
+        `[event-log-race-repro] blocked-branch resume burst (offset ${config.resumeBurstOffsetMs}ms ` +
+          `+ jitter ${config.resumeBurstJitterMs}ms) always lands before the first launch ` +
+          `completion (~${config.stepDelayMs}ms). No branch can be parked on a completion ` +
+          `the woken replays are missing.`
+      );
+    }
+    if (
+      config.resumeBurstOffsetMs >=
+      lastLaunchMs + config.blockedBranchWatchdogMs
+    ) {
+      console.warn(
+        `[event-log-race-repro] blocked-branch resume burst offset (${config.resumeBurstOffsetMs}ms) ` +
+          `lands after every watchdog deadline (last launch ~${lastLaunchMs}ms + ` +
+          `watchdog ${config.blockedBranchWatchdogMs}ms). Every branch will take the ` +
+          `recovery path and the resumes race nothing.`
+      );
+    }
+
     const sleepBudgetMs = config.iterations * config.sleepMs;
     const resumeCeilingMs = config.resumeDelayMs + config.resumeJitterMs;
     if (resumeCeilingMs >= sleepBudgetMs) {
@@ -1127,6 +1294,11 @@ describe('event log race repro', { retry: 0 }, () => {
         config.concurrency,
         runHookStormAttempt
       );
+      await runScenario(
+        config.blockedBranchAttempts,
+        config.concurrency,
+        runBlockedBranchAttempt
+      );
       await runScenario(
         config.hookSleepAttempts,
         config.concurrency,

@@ -47,6 +47,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    generateHookToken: (correlationId: string) => `token_${correlationId}`,
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),

@@ -45,6 +45,7 @@ function setupWorkflowContext(
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    generateHookToken: (correlationId: string) => `token_${correlationId}`,
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),

@@ -80,6 +80,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    generateHookToken: (correlationId: string) => `token_${correlationId}`,
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),

@@ -60,6 +60,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     }),
     invocationsQueue: new Map(),
     generateUlid: () => ulid(workflowStartedAt),
+    generateHookToken: (correlationId: string) => `token_${correlationId}`,
     generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) =>
       new Uint8Array(size).map(() => 256 * context.globalThis.Math.random())
     ),