Materialize waits as entities to prevent duplicate wait_completed events (#1057)

* Handle 409 conflict when completing waits that were already completed

When multiple concurrent workflow invocations race to complete the same
wait, the server returns 409 (conflict) for duplicates. This change
handles the 409 gracefully in both runtime.ts (sleep elapsed check) and
runs.ts (wakeUpRun), preventing crashes and treating already-completed
waits as successful.

Also updates event-sourcing docs to reflect that waits are now
materialized as entities in storage with atomic completion guarantees.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Point e2e tests at workflow-server preview for wait materialization branch

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Materialize waits as entities in local and postgres worlds to prevent duplicate wait_completed events

Adds Wait type/schema to the shared world package and implements wait entity
materialization in both local (filesystem) and postgres world implementations,
matching the DynamoDB behavior. wait_created creates a wait entity with status
'waiting', and wait_completed transitions it to 'completed' with guards that
reject duplicates (409).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Fix waitId to use composite key for consistency with postgres world

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Clean up wait entities on terminal run states and register waits migration

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Clear WORKFLOW_SERVER_URL_OVERRIDE for merge

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Update changeset for all affected packages and restore server URL override

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Clear WORKFLOW_SERVER_URL_OVERRIDE now that server PR is merged

The server-side changes (vercel/workflow-server#265) have been merged to main
and deployed to production, so we no longer need to point at the preview URL.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: pranaygp

.changeset/wait-complete-guard.md

@@ -0,0 +1,13 @@
+---
+"@workflow/core": patch
+"@workflow/world": patch
+"@workflow/world-local": patch
+"@workflow/world-postgres": patch
+---
+
+Materialize waits as entities to prevent duplicate wait_completed events
+
+- `@workflow/core`: Handle 409 conflict gracefully when creating wait_completed events, preventing crashes when multiple concurrent invocations race to complete the same wait
+- `@workflow/world`: Add `Wait` type, `WaitSchema`, and `WaitStatusSchema` exports; add optional `wait` field to `EventResult`
+- `@workflow/world-local`: Materialize wait entities on wait_created/wait_completed with duplicate detection; clean up waits on terminal run states
+- `@workflow/world-postgres`: Add `workflow_waits` table with `wait_status` enum; materialize wait entities with conditional writes for duplicate prevention; clean up waits on terminal run states

docs/content/docs/how-it-works/event-sourcing.mdx

@@ -33,7 +33,7 @@ In the Workflow DevKit, the following entity types are managed through events:
 - **Runs**: Workflow execution instances (materialized in storage)
 - **Steps**: Individual atomic operations within a workflow (materialized in storage)
 - **Hooks**: Suspension points that can receive external data (materialized in storage)
-- **Waits**: Sleep or delay operations (tracked via events only, not materialized)
+- **Waits**: Sleep or delay operations (materialized in storage)
 
 ## Entity Lifecycles
 
@@ -151,7 +151,7 @@ flowchart TD
 - `completed`: Delay period has elapsed, workflow can resume
 
 <Callout type="info">
-Unlike Runs, Steps, and Hooks, waits are conceptual entities tracked only through events. There is no separate "Wait" record in storage that can be queried—the wait state is derived entirely from the `wait_created` and `wait_completed` events in the event log.
+Like Runs, Steps, and Hooks, waits are materialized as entities in storage. When a `wait_created` event is processed, a wait entity is created with status `waiting`. When a `wait_completed` event is processed, the wait entity is atomically transitioned to `completed` — this guarantees that a wait can only be completed exactly once, even if multiple concurrent invocations attempt to complete it simultaneously.
 </Callout>
 
 ## Event Types Reference

packages/core/src/runtime.ts

@@ -220,12 +220,26 @@ export function workflowEntrypoint(
 
                     // Create all wait_completed events
                     for (const waitEvent of waitsToComplete) {
-                      const result = await world.events.create(
-                        runId,
-                        waitEvent
-                      );
-                      // Add the event to the events array so the workflow can see it
-                      events.push(result.event!);
+                      try {
+                        const result = await world.events.create(
+                          runId,
+                          waitEvent
+                        );
+                        // Add the event to the events array so the workflow can see it
+                        events.push(result.event!);
+                      } catch (err) {
+                        if (WorkflowAPIError.is(err) && err.status === 409) {
+                          runtimeLogger.info(
+                            'Wait already completed, skipping',
+                            {
+                              workflowRunId: runId,
+                              correlationId: waitEvent.correlationId,
+                            }
+                          );
+                          continue;
+                        }
+                        throw err;
+                      }
                     }
 
                     const result = await trace(

packages/core/src/runtime/runs.test.ts

@@ -0,0 +1,96 @@
+import { WorkflowAPIError } from '@workflow/errors';
+import type { Event, World } from '@workflow/world';
+import { describe, expect, it, vi } from 'vitest';
+
+// Mock version module to avoid missing generated file
+vi.mock('../version.js', () => ({ version: '0.0.0-test' }));
+
+import { wakeUpRun } from './runs.js';
+
+describe('wakeUpRun', () => {
+  function createMockWorld(
+    overrides: {
+      run?: Partial<
+        ReturnType<World['runs']['get']> extends Promise<infer T> ? T : never
+      >;
+      events?: Event[];
+      createError?: Error;
+    } = {}
+  ): World {
+    const run = {
+      runId: 'wrun_123',
+      workflowName: 'test-workflow',
+      status: 'running' as const,
+      specVersion: 2,
+      input: [],
+      createdAt: new Date(),
+      updatedAt: new Date(),
+      startedAt: new Date(),
+      deploymentId: 'test-deployment',
+      ...overrides.run,
+    };
+
+    const events = overrides.events ?? [];
+
+    return {
+      runs: {
+        get: vi.fn().mockResolvedValue(run),
+      },
+      events: {
+        list: vi.fn().mockResolvedValue({
+          data: events,
+          hasMore: false,
+          cursor: null,
+        }),
+        create: overrides.createError
+          ? vi.fn().mockRejectedValue(overrides.createError)
+          : vi.fn().mockResolvedValue({ event: {} }),
+      },
+      queue: vi.fn().mockResolvedValue(undefined),
+    } as unknown as World;
+  }
+
+  it('should count 409 conflict as a successful stop', async () => {
+    const events: Event[] = [
+      {
+        eventId: 'evnt_0',
+        runId: 'wrun_123',
+        eventType: 'wait_created',
+        correlationId: 'wait_abc',
+        eventData: { resumeAt: new Date('2024-01-01T00:00:01.000Z') },
+        createdAt: new Date(),
+      },
+    ];
+
+    const conflict = new WorkflowAPIError('Wait already completed', {
+      status: 409,
+    });
+
+    const world = createMockWorld({ events, createError: conflict });
+    const result = await wakeUpRun(world, 'wrun_123');
+
+    expect(result.stoppedCount).toBe(1);
+    expect(world.queue).toHaveBeenCalled();
+  });
+
+  it('should throw for non-409 errors', async () => {
+    const events: Event[] = [
+      {
+        eventId: 'evnt_0',
+        runId: 'wrun_123',
+        eventType: 'wait_created',
+        correlationId: 'wait_abc',
+        eventData: { resumeAt: new Date('2024-01-01T00:00:01.000Z') },
+        createdAt: new Date(),
+      },
+    ];
+
+    const serverError = new WorkflowAPIError('Internal server error', {
+      status: 500,
+    });
+
+    const world = createMockWorld({ events, createError: serverError });
+
+    await expect(wakeUpRun(world, 'wrun_123')).rejects.toThrow(AggregateError);
+  });
+});

packages/core/src/runtime/runs.ts

@@ -1,3 +1,4 @@
+import { WorkflowAPIError } from '@workflow/errors';
 import { hydrateWorkflowArguments } from '../serialization.js';
 import {
   type Event,
@@ -183,7 +184,11 @@ export async function wakeUpRun(
         await world.events.create(runId, eventData, { v1Compat: compatMode });
         stoppedCount++;
       } catch (err) {
-        errors.push(err instanceof Error ? err : new Error(String(err)));
+        if (WorkflowAPIError.is(err) && err.status === 409) {
+          stoppedCount++;
+        } else {
+          errors.push(err instanceof Error ? err : new Error(String(err)));
+        }
       }
     }
 

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

@@ -7,6 +7,7 @@ import type {
   SerializedData,
   Step,
   Storage,
+  Wait,
   WorkflowRun,
 } from '@workflow/world';
 import {
@@ -16,6 +17,7 @@ import {
   requiresNewerWorld,
   SPEC_VERSION_CURRENT,
   StepSchema,
+  WaitSchema,
   WorkflowRunSchema,
 } from '@workflow/world';
 import { DEFAULT_RESOLVE_DATA_OPTION } from '../config.js';
@@ -31,6 +33,25 @@ import { getObjectCreatedAt, monotonicUlid } from './helpers.js';
 import { deleteAllHooksForRun } from './hooks-storage.js';
 import { handleLegacyEvent } from './legacy.js';
 
+/**
+ * Helper function to delete all waits associated with a workflow run.
+ * Called when a run reaches a terminal state.
+ */
+async function deleteAllWaitsForRun(
+  basedir: string,
+  runId: string
+): Promise<void> {
+  const waitsDir = path.join(basedir, 'waits');
+  const files = await listJSONFiles(waitsDir);
+
+  for (const file of files) {
+    if (file.startsWith(`${runId}-`)) {
+      const waitPath = path.join(waitsDir, `${file}.json`);
+      await deleteJSON(waitPath);
+    }
+  }
+}
+
 /**
  * Creates the events storage implementation using the filesystem.
  * Implements the Storage['events'] interface with create, list, and listByCorrelationId operations.
@@ -156,7 +177,8 @@ export function createEventsStorage(basedir: string): Storage['events'] {
         // Creating new entities on terminal runs is not allowed
         if (
           data.eventType === 'step_created' ||
-          data.eventType === 'hook_created'
+          data.eventType === 'hook_created' ||
+          data.eventType === 'wait_created'
         ) {
           throw new WorkflowAPIError(
             `Cannot create new entities on run in terminal state "${currentRun.status}"`,
@@ -240,6 +262,7 @@ export function createEventsStorage(basedir: string): Storage['events'] {
       let run: WorkflowRun | undefined;
       let step: Step | undefined;
       let hook: Hook | undefined;
+      let wait: Wait | undefined;
 
       // Create/update entity based on event type (event-sourced architecture)
       // Run lifecycle events
@@ -312,7 +335,10 @@ export function createEventsStorage(basedir: string): Storage['events'] {
             updatedAt: now,
           };
           await writeJSON(runPath, run, { overwrite: true });
-          await deleteAllHooksForRun(basedir, effectiveRunId);
+          await Promise.all([
+            deleteAllHooksForRun(basedir, effectiveRunId),
+            deleteAllWaitsForRun(basedir, effectiveRunId),
+          ]);
         }
       } else if (data.eventType === 'run_failed' && 'eventData' in data) {
         const failedData = data.eventData as {
@@ -346,7 +372,10 @@ export function createEventsStorage(basedir: string): Storage['events'] {
             updatedAt: now,
           };
           await writeJSON(runPath, run, { overwrite: true });
-          await deleteAllHooksForRun(basedir, effectiveRunId);
+          await Promise.all([
+            deleteAllHooksForRun(basedir, effectiveRunId),
+            deleteAllWaitsForRun(basedir, effectiveRunId),
+          ]);
         }
       } else if (data.eventType === 'run_cancelled') {
         // Reuse currentRun from validation (already read above)
@@ -369,7 +398,10 @@ export function createEventsStorage(basedir: string): Storage['events'] {
             updatedAt: now,
           };
           await writeJSON(runPath, run, { overwrite: true });
-          await deleteAllHooksForRun(basedir, effectiveRunId);
+          await Promise.all([
+            deleteAllHooksForRun(basedir, effectiveRunId),
+            deleteAllWaitsForRun(basedir, effectiveRunId),
+          ]);
         }
       } else if (
         // Step lifecycle events
@@ -611,6 +643,62 @@ export function createEventsStorage(basedir: string): Storage['events'] {
           `${data.correlationId}.json`
         );
         await deleteJSON(hookPath);
+      } else if (data.eventType === 'wait_created' && 'eventData' in data) {
+        // wait_created: Creates wait entity with status 'waiting'
+        const waitData = data.eventData as {
+          resumeAt?: Date;
+        };
+        const waitCompositeKey = `${effectiveRunId}-${data.correlationId}`;
+        const waitPath = path.join(
+          basedir,
+          'waits',
+          `${waitCompositeKey}.json`
+        );
+        const existingWait = await readJSON(waitPath, WaitSchema);
+        if (existingWait) {
+          throw new WorkflowAPIError(
+            `Wait "${data.correlationId}" already exists`,
+            { status: 409 }
+          );
+        }
+        wait = {
+          waitId: waitCompositeKey,
+          runId: effectiveRunId,
+          status: 'waiting',
+          resumeAt: waitData.resumeAt,
+          completedAt: undefined,
+          createdAt: now,
+          updatedAt: now,
+          specVersion: effectiveSpecVersion,
+        };
+        await writeJSON(waitPath, wait);
+      } else if (data.eventType === 'wait_completed') {
+        // wait_completed: Transitions wait to 'completed', rejects duplicates
+        const waitCompositeKey = `${effectiveRunId}-${data.correlationId}`;
+        const waitPath = path.join(
+          basedir,
+          'waits',
+          `${waitCompositeKey}.json`
+        );
+        const existingWait = await readJSON(waitPath, WaitSchema);
+        if (!existingWait) {
+          throw new WorkflowAPIError(`Wait "${data.correlationId}" not found`, {
+            status: 404,
+          });
+        }
+        if (existingWait.status === 'completed') {
+          throw new WorkflowAPIError(
+            `Wait "${data.correlationId}" already completed`,
+            { status: 409 }
+          );
+        }
+        wait = {
+          ...existingWait,
+          status: 'completed',
+          completedAt: now,
+          updatedAt: now,
+        };
+        await writeJSON(waitPath, wait, { overwrite: true });
       }
       // Note: hook_received events are stored in the event log but don't
       // modify the Hook entity (which doesn't have a payload field)
@@ -629,6 +717,7 @@ export function createEventsStorage(basedir: string): Storage['events'] {
         run,
         step,
         hook,
+        wait,
       };
     },
 

packages/world-postgres/src/drizzle/migrations/0007_add_waits_table.sql

@@ -0,0 +1,13 @@
+CREATE TYPE "public"."wait_status" AS ENUM('waiting', 'completed');--> statement-breakpoint
+CREATE TABLE IF NOT EXISTS "workflow"."workflow_waits" (
+	"wait_id" varchar PRIMARY KEY NOT NULL,
+	"run_id" varchar NOT NULL,
+	"status" "wait_status" NOT NULL,
+	"resume_at" timestamp,
+	"completed_at" timestamp,
+	"created_at" timestamp DEFAULT now() NOT NULL,
+	"updated_at" timestamp DEFAULT now() NOT NULL,
+	"spec_version" integer
+);
+--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_waits_run_id_index" ON "workflow"."workflow_waits" USING btree ("run_id");

packages/world-postgres/src/drizzle/migrations/meta/_journal.json

@@ -50,6 +50,13 @@
       "when": 1768500000000,
       "tag": "0006_add_error_cbor",
       "breakpoints": true
+    },
+    {
+      "idx": 7,
+      "version": "7",
+      "when": 1769500000000,
+      "tag": "0007_add_waits_table",
+      "breakpoints": true
     }
   ]
 }