feat(world-local): support atomic start hooks

.changeset/atomic-start-hook-local.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-local': minor
+---
+
+Support atomic workflow admission with `start({ hook })` in the Local World.

packages/core/e2e/e2e.test.ts

@@ -2129,6 +2129,37 @@ describe('e2e', () => {
     }
   );
 
+  test.skipIf(
+    !isLocalDeployment() ||
+      process.env.WORKFLOW_TARGET_WORLD === '@workflow/world-postgres'
+  )(
+    'atomic start Hooks admit one concurrent run',
+    { timeout: 60_000 },
+    async () => {
+      const token = `atomic-start-${Math.random().toString(36).slice(2)}`;
+      const workflow = await e2e('sleepingWorkflow');
+      const results = await Promise.allSettled(
+        Array.from({ length: 6 }, () =>
+          start(workflow, [5_000], { hook: { token } })
+        )
+      );
+      const winner = results.find((result) => result.status === 'fulfilled');
+      assert(winner?.status === 'fulfilled');
+
+      expect(
+        results.filter((result) => result.status === 'fulfilled')
+      ).toHaveLength(1);
+      for (const result of results) {
+        if (result.status === 'rejected') {
+          expect(HookConflictError.is(result.reason)).toBe(true);
+          assert(HookConflictError.is(result.reason));
+          expect(result.reason.conflictingRunId).toBe(winner.value.runId);
+        }
+      }
+      await winner.value.returnValue;
+    }
+  );
+
   test(
     'hookAdoptOwnerResultWorkflow - duplicate adopts the owner result via conflict.returnValue',
     { timeout: 120_000 },

packages/world-local/src/index.ts

@@ -18,7 +18,11 @@ import {
 import { initDataDir } from './init.js';
 import { instrumentObject } from './instrumentObject.js';
 import { createQueue, type DirectHandler } from './queue.js';
-import { hashToken, hookRecoveryMarkerPath } from './storage/helpers.js';
+import {
+  hookRecoveryMarkerPath,
+  releaseHookTokenClaimIfOwnedBy,
+  StartHookAdmissionSchema,
+} from './storage/helpers.js';
 import { resetHookIndexEnsureCache } from './storage/hook-index.js';
 import { createStorage } from './storage.js';
 import { createStreamer } from './streamer.js';
@@ -75,6 +79,7 @@ export function createWorld(args?: Partial<Config>): LocalWorld {
     specVersion: SPEC_VERSION_CURRENT,
     capabilities: {
       hookRetention: { active: true },
+      atomicStartHook: { active: true },
       // world-local deduplicates concurrent `hook_received` writes sharing a
       // `(runId, resumeId)` via a filesystem sidecar claim (see
       // events-storage.ts `claimHookResume`), so resumeHook()'s parallel fast
@@ -128,12 +133,8 @@ export function createWorld(args?: Partial<Config>): LocalWorld {
         // Selectively delete only files matching this tag
         const basedir = mergedConfig.dataDir;
 
-        // Delete hook token constraint files (and recovery markers,
-        // for disk hygiene) BEFORE deleting the hooks, since we need
-        // to read each hook to extract its token hash. Constraint
-        // files and markers are untagged (`{sha256}.json` and
-        // `{sha256}.recovery.json`) so listTaggedFiles won't find
-        // them — we must resolve them via the hook data.
+        // Claims and recovery markers are untagged, so release them through
+        // their tagged Hook or admission before deleting tagged entities.
         const hooksDir = path.join(basedir, 'hooks');
         const taggedHookFiles = await listTaggedFiles(hooksDir, tag);
         const { HookSchema } = await import('@workflow/world');
@@ -144,9 +145,11 @@ export function createWorld(args?: Partial<Config>): LocalWorld {
               HookSchema
             );
             if (hook?.token) {
-              await deleteJSON(
-                path.join(hooksDir, 'tokens', `${hashToken(hook.token)}.json`)
-              );
+              await releaseHookTokenClaimIfOwnedBy(basedir, hook.token, {
+                runId: hook.runId,
+                hookId: hook.hookId,
+                tag,
+              });
               await deleteJSON(
                 hookRecoveryMarkerPath(
                   basedir,
@@ -159,6 +162,26 @@ export function createWorld(args?: Partial<Config>): LocalWorld {
           })
         );
 
+        const admissionsDir = path.join(basedir, 'hooks', 'admissions');
+        const taggedAdmissionFiles = await listTaggedFiles(admissionsDir, tag);
+        await Promise.all(
+          taggedAdmissionFiles.map(async (admissionFile) => {
+            const admissionPath = path.join(admissionsDir, admissionFile);
+            const admission = await readJSON(
+              admissionPath,
+              StartHookAdmissionSchema
+            );
+            if (admission && !('redirectRunId' in admission)) {
+              await releaseHookTokenClaimIfOwnedBy(basedir, admission.token, {
+                runId: admission.runId,
+                eventId: admission.eventId,
+                tag,
+              });
+            }
+            await deleteJSON(admissionPath);
+          })
+        );
+
         // Delete tagged entity files across all directories
         const entityDirs = [
           'runs',