Fix race condition allowing duplicate hook_disposed events (#1523)

* Fix race condition allowing duplicate hook_disposed events

Concurrent workflow invocations could both post hook_disposed for the
same hook, corrupting the event log with duplicate events. This mirrors
the wait_completed race condition fixed in #1057/#1434.

- world-local: Add writeExclusive lock file for hook_disposed (same
  pattern as wait_completed and step terminal states)
- world-postgres: Use DELETE ... RETURNING to atomically detect if
  another caller already deleted the hook entity
- suspension-handler: Improve log messages to distinguish hook-already-
  disposed (EntityConflictError) from run-already-completed (RunExpiredError)

Fixes #1266

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Accept either EntityConflictError or HookNotFoundError in race test

The concurrent hook_disposed race has two possible orderings: the loser
may hit the lock file (EntityConflictError) or find the hook entity
already deleted by the winner (HookNotFoundError). Both are correct.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Assert only one hook_disposed event in event log after race

Verifies the losing concurrent caller didn't sneak an event in before
the guard threw.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: pranaygp

.changeset/fix-duplicate-hook-disposed-race.md

@@ -0,0 +1,7 @@
+---
+'@workflow/core': patch
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+---
+
+Fix race condition allowing duplicate `hook_disposed` events for the same hook

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

@@ -169,7 +169,17 @@ export async function handleSuspension({
         try {
           await world.events.create(runId, hookDisposedEvent, { requestId });
         } catch (err) {
-          if (EntityConflictError.is(err) || RunExpiredError.is(err)) {
+          if (EntityConflictError.is(err)) {
+            // Hook was already disposed by a concurrent invocation — safe to skip
+            runtimeLogger.info(
+              'Hook already disposed, skipping duplicate disposal',
+              {
+                workflowRunId: runId,
+                correlationId: queueItem.correlationId,
+                message: err.message,
+              }
+            );
+          } else if (RunExpiredError.is(err)) {
             runtimeLogger.info(
               'Workflow run already completed, skipping hook disposal',
               {

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

@@ -1952,6 +1952,42 @@ describe('Storage', () => {
         name: 'EntityConflictError',
       });
     });
+
+    it('should reject concurrent hook_disposed for the same hook', async () => {
+      await createHook(storage, testRunId, {
+        hookId: 'hook_race_1',
+        token: 'hook-race-token-1',
+      });
+
+      const results = await Promise.allSettled([
+        disposeHook(storage, testRunId, 'hook_race_1'),
+        disposeHook(storage, testRunId, 'hook_race_1'),
+      ]);
+
+      const fulfilled = results.filter((r) => r.status === 'fulfilled');
+      const rejected = results.filter((r) => r.status === 'rejected');
+
+      expect(fulfilled).toHaveLength(1);
+      expect(rejected).toHaveLength(1);
+      // Depending on timing, the loser may hit the lock file (EntityConflictError)
+      // or find the hook entity already deleted (HookNotFoundError).
+      const reason = (rejected[0] as PromiseRejectedResult).reason as {
+        name?: string;
+      };
+      expect(['EntityConflictError', 'HookNotFoundError']).toContain(
+        reason.name
+      );
+
+      // Verify only one hook_disposed event was written to the log
+      const events = await storage.events.list({
+        runId: testRunId,
+        pagination: {},
+      });
+      const hookDisposedEvents = events.data.filter(
+        (e) => e.eventType === 'hook_disposed'
+      );
+      expect(hookDisposedEvents).toHaveLength(1);
+    });
   });
 
   describe('run terminal state validation', () => {

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

@@ -714,6 +714,19 @@ export function createEventsStorage(
           hook
         );
       } else if (data.eventType === 'hook_disposed') {
+        // hook_disposed: Deletes hook entity, rejects duplicates.
+        // Uses writeExclusive on a lock file to atomically prevent concurrent
+        // invocations from both disposing the same hook (TOCTOU race).
+        const hookLockName = tag
+          ? `${data.correlationId}.disposed.${tag}`
+          : `${data.correlationId}.disposed`;
+        const lockPath = path.join(basedir, '.locks', 'hooks', hookLockName);
+        const claimed = await writeExclusive(lockPath, '');
+        if (!claimed) {
+          throw new EntityConflictError(
+            `Hook "${data.correlationId}" already disposed`
+          );
+        }
         // Read the hook to get its token before deleting
         const hookPath = taggedPath(basedir, 'hooks', data.correlationId, tag);
         const existingHook = await readJSONWithFallback(

packages/world-postgres/src/storage.ts

@@ -1039,11 +1039,19 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
         }
       }
 
-      // Handle hook_disposed event: delete hook entity
+      // Handle hook_disposed event: delete hook entity atomically.
+      // Uses DELETE ... RETURNING to ensure only one concurrent caller
+      // succeeds — if no rows are returned, the hook was already disposed.
       if (data.eventType === 'hook_disposed' && data.correlationId) {
-        await drizzle
+        const [deleted] = await drizzle
           .delete(Schema.hooks)
-          .where(eq(Schema.hooks.hookId, data.correlationId));
+          .where(eq(Schema.hooks.hookId, data.correlationId))
+          .returning({ hookId: Schema.hooks.hookId });
+        if (!deleted) {
+          throw new EntityConflictError(
+            `Hook "${data.correlationId}" already disposed`
+          );
+        }
       }
 
       // Handle wait_created event: create wait entity