feat(world-local): support Hook minimum retention

.changeset/local-hook-min-retention.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-local': minor
+---
+
+Keep Hook tokens reserved through their configured minimum retention.

packages/core/e2e/e2e.test.ts

@@ -2047,6 +2047,39 @@ describe('e2e', () => {
     }
   );
 
+  test.skipIf(
+    !isLocalDeployment() ||
+      process.env.WORKFLOW_TARGET_WORLD === '@workflow/world-postgres'
+  )(
+    'hookMinRetentionWorkflow - terminal Hook cannot resume and its token stays unavailable',
+    { timeout: 60_000 },
+    async () => {
+      const token = `retained-${Math.random().toString(36).slice(2)}`;
+      const owner = await start(await e2e('hookMinRetentionWorkflow'), [
+        token,
+        60_000,
+      ]);
+
+      expect(await owner.returnValue).toEqual({ role: 'owner' });
+
+      const hook = await getHookByToken(token);
+      expect(hook.runId).toBe(owner.runId);
+      await expect(resumeHook(hook, { duplicate: true })).rejects.toSatisfy(
+        (error: unknown) => HookNotFoundError.is(error)
+      );
+
+      const duplicate = await start(await e2e('hookMinRetentionWorkflow'), [
+        token,
+        60_000,
+      ]);
+      expect(await duplicate.returnValue).toEqual({
+        role: 'duplicate',
+        conflictRunId: owner.runId,
+        conflictStatus: 'completed',
+      });
+    }
+  );
+
   test(
     'hookAdoptOwnerResultWorkflow - duplicate adopts the owner result via conflict.returnValue',
     { timeout: 120_000 },

packages/world-local/package.json

@@ -35,6 +35,7 @@
     "@workflow/utils": "workspace:*",
     "@workflow/world": "workspace:*",
     "async-sema": "3.1.1",
+    "proper-lockfile": "4.1.2",
     "ulid": "catalog:",
     "undici": "catalog:",
     "zod": "catalog:"
@@ -43,6 +44,7 @@
     "@opentelemetry/api": "1.9.0",
     "@types/ms": "0.7.34",
     "@types/node": "catalog:",
+    "@types/proper-lockfile": "4.1.4",
     "@workflow/tsconfig": "workspace:*",
     "ms": "2.1.3",
     "typescript": "catalog:",

packages/world-local/src/fs.ts

@@ -413,6 +413,20 @@ export async function readJSON<T>(
   }
 }
 
+/** Read JSON, treating malformed persistence data as absent. */
+export async function readJSONLenient<T>(
+  filePath: string,
+  decoder: z.ZodType<T>
+): Promise<T | null> {
+  try {
+    return await readJSON(filePath, decoder);
+  } catch (error) {
+    if (error instanceof SyntaxError || error instanceof z.ZodError)
+      return null;
+    throw error;
+  }
+}
+
 export async function readBuffer(filePath: string): Promise<Buffer> {
   const content = await fs.readFile(filePath);
   return content;
@@ -507,7 +521,7 @@ interface PaginatedFileSystemQueryConfig<T> {
   cachedItems?: ReadonlyMap<string, T>;
   filePrefix?: string;
   fileIdFilter?: (fileId: string) => boolean;
-  filter?: (item: T) => boolean;
+  filter?: (item: T) => boolean | Promise<boolean>;
   sortOrder?: 'asc' | 'desc';
   limit?: number;
   cursor?: string;
@@ -647,7 +661,7 @@ export async function paginatedFileSystemQuery<T extends { createdAt: Date }>(
     for (const item of loadedBatch) {
       if (!item) continue;
       // Apply custom filter early if provided
-      if (filter && !filter(item)) continue;
+      if (filter && !(await filter(item))) continue;
 
       // Double-check cursor filtering with actual createdAt from JSON
       // (in case ULID timestamp differs from stored createdAt)

packages/world-local/src/index.ts

@@ -70,6 +70,7 @@ export function createWorld(args?: Partial<Config>): LocalWorld {
   const recoverActiveRuns = resolveRecoverActiveRuns(mergedConfig);
   return {
     specVersion: SPEC_VERSION_CURRENT,
+    capabilities: { hookRetention: { active: true } },
     ...queue,
     ...storage,
     ...instrumentObject('world.streams', {