[world-local] Fix CORRUPTED_EVENT_LOG from hook-resume staging slot holes

.changeset/world-local-hook-staging-slots.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-local': patch
+---
+
+Fix `CORRUPTED_EVENT_LOG` after a hook resume that raced another writer or was interrupted mid-write, and deliver a raced resume exactly once instead of duplicating it or reporting a conflict.

AGENTS.md

@@ -137,20 +137,30 @@ control that provides the calibration baseline. Any outcome other than
 `completed` fails the run, except `infra`, which means the harness could not
 reach the deployment.
 
-Run it locally against `@workflow/world-postgres` and a locally started
-workbench app — no Vercel deployment, no credentials:
+Run it locally against a locally started workbench app — no Vercel deployment,
+no credentials:
 
 ```bash
-pnpm run test:e2e:event-log-race-repro:local
+pnpm run test:e2e:event-log-race-repro:local              # world-postgres
+pnpm run test:e2e:event-log-race-repro:local --world local # world-local
 ```
 
-The script (`scripts/event-log-race-repro-local.sh`, `--help` for flags) brings up
-the world-postgres container, applies migrations, builds and starts
-`workbench/nextjs-turbopack` with `WORKFLOW_TARGET_WORLD` and
+The script (`scripts/event-log-race-repro-local.sh`, `--help` for flags) builds
+and starts `workbench/nextjs-turbopack` with `WORKFLOW_TARGET_WORLD` and
 `WORKFLOW_PUBLIC_MANIFEST=1` set **at build time** (both are build-time inputs;
-missing either silently yields a world-local app or a 404 manifest), runs the
-harness, prints the same summary table CI posts, and tears the server down.
-Postgres is left running for the next iteration unless `--teardown` is passed.
+missing either silently yields a default-world app or a 404 manifest), runs the
+harness, prints the same summary table CI posts, and tears the server down. For
+world-postgres it first brings up the container and applies migrations, and
+leaves Postgres running for the next iteration unless `--teardown` is passed;
+the container flags (`--skip-db-setup`, `--no-docker`, `--teardown`) do nothing
+under `--world local`, whose only state is a data directory the script clears
+before each run.
+
+Both worlds are worth running, and neither subsumes the other: world-postgres
+arbitrates event slots inside one SQL statement, while world-local arbitrates
+them with an exclusive `link(2)` against a directory that two processes (the app
+and the harness) both write to. A slot race a transaction closes is not
+automatically closed by a filesystem.
 
 Scale is controlled entirely by `EVENT_LOG_RACE_REPRO_*` environment variables.
 Their defaults live only in `event-log-race-repro.test.ts` — neither the CI
@@ -185,6 +195,17 @@ one heap saturates GC — measured on a 12-core laptop, all 14 attempts came bac
 sets `WORKFLOW_POSTGRES_WORKER_CONCURRENCY=10` (override by exporting it) and
 raises the app's old-space limit (`--heap-mb`). If a local run reports `stuck`
 rather than `CORRUPTED_EVENT_LOG`, suspect the machine before the SDK.
+world-local saturates the same single process from its own in-process queue,
+which defaults to 1000 deliveries in flight, so the script holds it at the same
+number via `WORKFLOW_LOCAL_QUEUE_CONCURRENCY`.
+
+world-local's storms come out clean far more often than world-postgres's, so the
+default scale says even less there: the corruption it does produce needs a
+`hook_received` to be staged and then rejected, which the harness reaches only
+in a run's terminal moments. Reach for a unit test in
+`packages/world-local/src/storage/` when a suspected filesystem race can be
+staged directly — it costs milliseconds and does not depend on the interleaving
+showing up.
 
 In CI the same harness runs from `.github/workflows/event-log-race-repro.yml`,
 triggered by adding the `event-log-race-repro` label to a PR (or by
@@ -198,6 +219,10 @@ To poke at a run afterwards, the CLI reads the same world from the environment:
 WORKFLOW_TARGET_WORLD=@workflow/world-postgres \
 WORKFLOW_POSTGRES_URL=postgres://world:world@localhost:5432/world \
   pnpm wf inspect <run-id>
+
+WORKFLOW_TARGET_WORLD=local \
+WORKFLOW_LOCAL_DATA_DIR=workbench/nextjs-turbopack/.next/workflow-data \
+  pnpm wf inspect <run-id>
 ```
 
 ### Example App Development

packages/world-local/src/storage/events-storage.ts

@@ -2734,25 +2734,32 @@ export function createEventsStorage(
               );
             }
 
+            // Staging is private to this attempt, so the name carries a
+            // nonce rather than only the event id. Sharing a name across
+            // writers would make the staging directory a second, invisible
+            // claim on the slot: the allocator probes `events/` alone, so a
+            // staged file is not evidence the position is taken, and bumping
+            // off it moves this writer past a position no one will ever
+            // publish. A crashed attempt (its cleanup lives in a `finally`
+            // the kill skips) would hole the log permanently that way, and
+            // two live writers drawing the same candidate would hole it
+            // whenever the stager is later rejected. The slot is arbitrated
+            // where it is actually taken: the promote below.
             const stagedPath = pendingHookEventPath(
               basedir,
               effectiveRunId,
-              eventId,
+              `${eventId}.${monotonicUlid()}`,
               tag
             );
             const staged = await writeExclusive(stagedPath, serializedEvent);
             if (!staged) {
-              // The staging path can be occupied by a previous crashed
-              // attempt of this very event (which never promoted), or, under
-              // slot ids, by a concurrent writer holding the same position.
-              // Both are handled the same way: fall through to the bump
-              // below, which moves off the position when it can and surfaces
-              // the conflict when it cannot.
-              if (await bumpEventSlot(attempt)) {
-                continue;
-              }
-              throw new EntityConflictError(
-                `Event "${eventId}" already exists for run "${effectiveRunId}"`
+              // A nonced path cannot already exist, so this is a filesystem
+              // fault rather than a lost race. It is deliberately NOT an
+              // `EntityConflictError`: the runtime reads that as a benign
+              // duplicate publish and carries on, which would absorb infra
+              // trouble as "someone else already wrote it".
+              throw new WorkflowWorldError(
+                `Failed to stage event "${eventId}" for run "${effectiveRunId}": staging path already exists`
               );
             }
             try {
@@ -2785,6 +2792,45 @@ export function createEventsStorage(
             notePublishedSlot(effectiveRunId, eventId);
             break;
           }
+          // Losing this publish to THIS SAME resume is the convergence the
+          // claim exists to force, and it has to be answered before the bump
+          // rather than after the loop, because only one of the two takers of
+          // a claim is pinned. The taker that wrote the claim keeps its own
+          // (unpinned) id, since a slot is a position another instance hands
+          // out for unrelated events too; the taker that adopts the claim is
+          // pinned to the claimed position. So whichever one loses the
+          // promote, the loser may be the unpinned owner, and bumping it
+          // publishes a SECOND hook_received for one resumeId: the log stays
+          // dense, and replay delivers the resume twice.
+          //
+          // `converge` above already answers this case with the committed
+          // event — it just could not see it yet, because the other taker had
+          // not published when this attempt read. Answer it the same way. An
+          // occupant that is NOT this resume is the unrelated-event collision
+          // the bump is for, and still bumps (or conflicts, when pinned).
+          if (data.eventType === 'hook_received' && params?.resumeId) {
+            const occupant = await readJSONWithFallback(
+              basedir,
+              'events',
+              `${effectiveRunId}-${eventId}`,
+              EventSchema,
+              tag
+            );
+            if (
+              occupant &&
+              isResumeEvent(occupant, {
+                runId: effectiveRunId,
+                resumeId: params.resumeId,
+                hookId: data.correlationId,
+                eventId,
+              })
+            ) {
+              // The claim already names this position, so there is no claim
+              // rewrite to do: the resume is committed, exactly once, and
+              // both writers return that one event.
+              return { event: occupant };
+            }
+          }
           if (!(await bumpEventSlot(attempt))) {
             break;
           }
@@ -2811,6 +2857,10 @@ export function createEventsStorage(
               tag
             );
           }
+          // A resume that lost its publish to its own committed event already
+          // returned it from inside the loop above, so reaching here means the
+          // occupant is an unrelated event and this is a duplicate publish the
+          // runtime's concurrent-replay catch path handles.
           throw new EntityConflictError(
             `Event "${eventId}" already exists for run "${effectiveRunId}"`
           );