perf: implement event-sourced architecture with batch event creation

Part 1: Event-sourced architecture
- Add run lifecycle events (run_created, run_started, run_completed, run_failed)
- Update world implementations to create/update entities from events
- Entities (runs, steps, hooks) are now materializations of the event log

Part 2: Batch event creation
- Add events.createBatch() method to World interface
- Implement in world-vercel, world-local, world-postgres
- Update runtime to use batch event creation for wait_completed events

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: pranaygp

.changeset/brave-dots-bake.md

@@ -0,0 +1,16 @@
+---
+"@workflow/core": patch
+"@workflow/world": patch
+"@workflow/world-local": patch
+"@workflow/world-vercel": patch
+"@workflow/world-postgres": patch
+---
+
+perf: add events.createBatch() for batch event creation
+
+- Add `events.createBatch()` method to World interface for creating multiple events atomically
+- Implement batch event creation in world-vercel (parallel API calls)
+- Implement batch event creation in world-local (sequential for ULID ordering)
+- Implement batch event creation in world-postgres (single batch INSERT)
+- Update runtime to use batch event creation for wait_completed events
+- Optimize wait completion from O(n²) to O(n) with Set-based lookup

.changeset/event-sourced-entities.md

@@ -0,0 +1,13 @@
+---
+"@workflow/core": patch
+"@workflow/world": patch
+"@workflow/world-local": patch
+"@workflow/world-postgres": patch
+---
+
+perf: implement event-sourced architecture for runs, steps, and hooks
+
+- Add run lifecycle events (run_created, run_started, run_completed, run_failed, run_cancelled, run_paused, run_resumed)
+- Update world-local and world-postgres to create/update entities from events via createBatch
+- Entities (runs, steps, hooks) are now materializations of the event log
+- This makes the system faster, easier to reason about, and resilient to data inconsistencies

packages/cli/src/lib/inspect/run.ts

@@ -68,6 +68,6 @@ export const startRun = async (
 };
 
 export const cancelRun = async (world: World, runId: string) => {
-  await world.runs.cancel(runId);
+  await world.events.create(runId, { eventType: 'run_cancelled' });
   logger.log(chalk.green(`Cancel signal sent to run ${runId}`));
 };

packages/core/src/events-consumer.test.ts

@@ -73,6 +73,7 @@ describe('EventsConsumer', () => {
       await waitForNextTick();
 
       expect(callback).toHaveBeenCalledWith(event);
+      // Without auto-advance, callback is only called once
       expect(callback).toHaveBeenCalledTimes(1);
     });
   });
@@ -87,6 +88,7 @@ describe('EventsConsumer', () => {
       await waitForNextTick();
 
       expect(callback).toHaveBeenCalledWith(event);
+      // Without auto-advance, callback is only called once
       expect(callback).toHaveBeenCalledTimes(1);
     });
 
@@ -109,23 +111,27 @@ describe('EventsConsumer', () => {
       consumer.subscribe(callback);
       await waitForNextTick();
 
+      // callback finishes at event1, index advances to 1
+      // Without auto-advance, event2 is NOT processed
       expect(consumer.eventIndex).toBe(1);
       expect(consumer.callbacks).toHaveLength(0);
     });
 
-    it('should not increment event index when callback returns false', async () => {
+    it('should NOT auto-advance when all callbacks return NotConsumed', async () => {
       const event = createMockEvent();
       const consumer = new EventsConsumer([event]);
       const callback = vi.fn().mockReturnValue(EventConsumerResult.NotConsumed);
 
       consumer.subscribe(callback);
       await waitForNextTick();
+      await waitForNextTick(); // Extra tick to confirm no auto-advance
 
+      // Without auto-advance, eventIndex stays at 0
       expect(consumer.eventIndex).toBe(0);
       expect(consumer.callbacks).toContain(callback);
     });
 
-    it('should process multiple callbacks until one returns true', async () => {
+    it('should process multiple callbacks until one returns Consumed or Finished', async () => {
       const event = createMockEvent();
       const consumer = new EventsConsumer([event]);
       const callback1 = vi
@@ -140,15 +146,17 @@ describe('EventsConsumer', () => {
       consumer.subscribe(callback2);
       consumer.subscribe(callback3);
       await waitForNextTick();
+      await waitForNextTick(); // For next event processing
 
       expect(callback1).toHaveBeenCalledWith(event);
       expect(callback2).toHaveBeenCalledWith(event);
+      // callback3 sees the next event (null since we only have one event)
       expect(callback3).toHaveBeenCalledWith(null);
       expect(consumer.eventIndex).toBe(1);
       expect(consumer.callbacks).toEqual([callback1, callback3]);
     });
 
-    it('should process all callbacks when none return true', async () => {
+    it('should NOT advance when all callbacks return NotConsumed', async () => {
       const event = createMockEvent();
       const consumer = new EventsConsumer([event]);
       const callback1 = vi
@@ -169,6 +177,7 @@ describe('EventsConsumer', () => {
       expect(callback1).toHaveBeenCalledWith(event);
       expect(callback2).toHaveBeenCalledWith(event);
       expect(callback3).toHaveBeenCalledWith(event);
+      // Without auto-advance, eventIndex stays at 0
       expect(consumer.eventIndex).toBe(0);
       expect(consumer.callbacks).toEqual([callback1, callback2, callback3]);
     });
@@ -211,7 +220,7 @@ describe('EventsConsumer', () => {
       expect(callback2).toHaveBeenCalledWith(null);
     });
 
-    it('should handle complex event processing scenario', async () => {
+    it('should handle complex event processing with multiple consumers', async () => {
       const events = [
         createMockEvent({ id: 'event-1', event_type: 'type-a' }),
         createMockEvent({ id: 'event-2', event_type: 'type-b' }),
@@ -241,13 +250,14 @@ describe('EventsConsumer', () => {
       consumer.subscribe(typeBCallback);
       await waitForNextTick();
       await waitForNextTick(); // Wait for recursive processing
-      await waitForNextTick(); // Wait for final processing
 
-      // typeACallback processes event-1 and gets removed, so it won't process event-3
+      // typeACallback processes event-1 and gets removed
       expect(typeACallback).toHaveBeenCalledTimes(1); // Called for event-1 only
+      // typeBCallback processes event-2 and gets removed
       expect(typeBCallback).toHaveBeenCalledTimes(1); // Called for event-2
-      expect(consumer.eventIndex).toBe(2); // Only 2 events processed (event-3 remains)
-      expect(consumer.callbacks).toHaveLength(0); // Both callbacks removed after consuming their events
+      // eventIndex is at 2 (after event-1 and event-2 were consumed)
+      expect(consumer.eventIndex).toBe(2);
+      expect(consumer.callbacks).toHaveLength(0);
     });
   });
 
@@ -297,8 +307,9 @@ describe('EventsConsumer', () => {
       consumer.subscribe(callback3);
       await waitForNextTick();
 
-      // callback2 should be removed when it returns true
+      // callback2 should be removed when it returns Finished
       expect(consumer.callbacks).toEqual([callback1, callback3]);
+      // callback3 is called with the next event (null after event-1)
       expect(callback3).toHaveBeenCalledWith(null);
     });
 
@@ -314,25 +325,6 @@ describe('EventsConsumer', () => {
       expect(consumer.eventIndex).toBe(1);
     });
 
-    it('should handle multiple subscriptions happening in sequence', async () => {
-      const event1 = createMockEvent({ id: 'event-1' });
-      const event2 = createMockEvent({ id: 'event-2' });
-      const consumer = new EventsConsumer([event1, event2]);
-
-      const callback1 = vi.fn().mockReturnValue(EventConsumerResult.Finished);
-      const callback2 = vi.fn().mockReturnValue(EventConsumerResult.Finished);
-
-      consumer.subscribe(callback1);
-      await waitForNextTick();
-
-      consumer.subscribe(callback2);
-      await waitForNextTick();
-
-      expect(callback1).toHaveBeenCalledWith(event1);
-      expect(callback2).toHaveBeenCalledWith(event2);
-      expect(consumer.eventIndex).toBe(2);
-    });
-
     it('should handle empty events array gracefully', async () => {
       const consumer = new EventsConsumer([]);
       const callback = vi.fn().mockReturnValue(EventConsumerResult.NotConsumed);
@@ -343,5 +335,49 @@ describe('EventsConsumer', () => {
       expect(callback).toHaveBeenCalledWith(null);
       expect(consumer.eventIndex).toBe(0);
     });
+
+    it('should process events in order with proper consumers', async () => {
+      // This test simulates the workflow scenario:
+      // - run_created consumer consumes it
+      // - step consumer gets step_created, step_completed
+      const events = [
+        createMockEvent({ id: 'run-created', event_type: 'run_created' }),
+        createMockEvent({ id: 'step-created', event_type: 'step_created' }),
+        createMockEvent({ id: 'step-completed', event_type: 'step_completed' }),
+      ];
+      const consumer = new EventsConsumer(events);
+
+      // Run lifecycle consumer - consumes run_created
+      const runConsumer = vi.fn().mockImplementation((event: Event | null) => {
+        if (event?.event_type === 'run_created') {
+          return EventConsumerResult.Consumed;
+        }
+        return EventConsumerResult.NotConsumed;
+      });
+
+      // Step consumer - consumes step_created, finishes on step_completed
+      const stepConsumer = vi.fn().mockImplementation((event: Event | null) => {
+        if (event?.event_type === 'step_created') {
+          return EventConsumerResult.Consumed;
+        }
+        if (event?.event_type === 'step_completed') {
+          return EventConsumerResult.Finished;
+        }
+        return EventConsumerResult.NotConsumed;
+      });
+
+      consumer.subscribe(runConsumer);
+      consumer.subscribe(stepConsumer);
+      await waitForNextTick();
+      await waitForNextTick();
+      await waitForNextTick();
+
+      // runConsumer consumes run_created
+      expect(runConsumer).toHaveBeenCalledWith(events[0]);
+      // stepConsumer consumes step_created, then finishes on step_completed
+      expect(stepConsumer).toHaveBeenCalledWith(events[1]);
+      expect(stepConsumer).toHaveBeenCalledWith(events[2]);
+      expect(consumer.eventIndex).toBe(3);
+    });
   });
 });

packages/core/src/events-consumer.ts

@@ -78,5 +78,10 @@ export class EventsConsumer {
         return;
       }
     }
+
+    // If we reach here, all callbacks returned NotConsumed.
+    // We do NOT auto-advance - every event must have a consumer.
+    // With proper consumers for run_created/run_started/step_created,
+    // this should not cause events to get stuck.
   };
 }

packages/core/src/global.ts

@@ -6,6 +6,7 @@ export interface StepInvocationQueueItem {
   stepName: string;
   args: Serializable[];
   closureVars?: Record<string, Serializable>;
+  hasCreatedEvent?: boolean;
 }
 
 export interface HookInvocationQueueItem {

packages/core/src/runtime.ts

@@ -104,7 +104,9 @@ export class Run<TResult> {
    * Cancels the workflow run.
    */
   async cancel(): Promise<void> {
-    await this.world.runs.cancel(this.runId);
+    await this.world.events.create(this.runId, {
+      eventType: 'run_cancelled',
+    });
   }
 
   /**
@@ -272,10 +274,14 @@ export function workflowEntrypoint(
               let workflowRun = await world.runs.get(runId);
 
               if (workflowRun.status === 'pending') {
-                workflowRun = await world.runs.update(runId, {
-                  // This sets the `startedAt` timestamp at the database level
-                  status: 'running',
+                // Transition run to 'running' via event (event-sourced architecture)
+                const result = await world.events.create(runId, {
+                  eventType: 'run_started',
                 });
+                // Use the run entity from the event response (no extra get call needed)
+                if (result.run) {
+                  workflowRun = result.run;
+                }
               }
 
               // At this point, the workflow is "running" and `startedAt` should
@@ -310,27 +316,38 @@ export function workflowEntrypoint(
               // Load all events into memory before running
               const events = await getAllWorkflowRunEvents(workflowRun.runId);
 
-              // Check for any elapsed waits and create wait_completed events
+              // Check for any elapsed waits and batch create wait_completed events
               const now = Date.now();
-              for (const event of events) {
-                if (event.eventType === 'wait_created') {
-                  const resumeAt = event.eventData.resumeAt as Date;
-                  const hasCompleted = events.some(
-                    (e) =>
-                      e.eventType === 'wait_completed' &&
-                      e.correlationId === event.correlationId
-                  );
 
-                  // If wait has elapsed and hasn't been completed yet
-                  if (!hasCompleted && now >= resumeAt.getTime()) {
-                    const completedEvent = await world.events.create(runId, {
-                      eventType: 'wait_completed',
-                      correlationId: event.correlationId,
-                    });
-                    // Add the event to the events array so the workflow can see it
-                    events.push(completedEvent);
-                  }
-                }
+              // Pre-compute completed correlation IDs for O(n) lookup instead of O(n²)
+              const completedWaitIds = new Set(
+                events
+                  .filter((e) => e.eventType === 'wait_completed')
+                  .map((e) => e.correlationId)
+              );
+
+              // Collect all waits that need completion
+              const waitsToComplete = events
+                .filter(
+                  (e): e is typeof e & { correlationId: string } =>
+                    e.eventType === 'wait_created' &&
+                    e.correlationId !== undefined &&
+                    !completedWaitIds.has(e.correlationId) &&
+                    now >= (e.eventData.resumeAt as Date).getTime()
+                )
+                .map((e) => ({
+                  eventType: 'wait_completed' as const,
+                  correlationId: e.correlationId,
+                }));
+
+              // Batch create all wait_completed events
+              if (waitsToComplete.length > 0) {
+                const completedResults = await world.events.createBatch(
+                  runId,
+                  waitsToComplete
+                );
+                // Add the events to the events array so the workflow can see them
+                events.push(...completedResults.map((r) => r.event));
               }
 
               const result = await runWorkflow(
@@ -339,10 +356,12 @@ export function workflowEntrypoint(
                 events
               );
 
-              // Update the workflow run with the result
-              await world.runs.update(runId, {
-                status: 'completed',
-                output: result as Serializable,
+              // Complete the workflow run via event (event-sourced architecture)
+              await world.events.create(runId, {
+                eventType: 'run_completed',
+                eventData: {
+                  output: result as Serializable,
+                },
               });
 
               span?.setAttributes({
@@ -393,11 +412,14 @@ export function workflowEntrypoint(
                 console.error(
                   `${errorName} while running "${runId}" workflow:\n\n${errorStack}`
                 );
-                await world.runs.update(runId, {
-                  status: 'failed',
-                  error: {
-                    message: errorMessage,
-                    stack: errorStack,
+                // Fail the workflow run via event (event-sourced architecture)
+                await world.events.create(runId, {
+                  eventType: 'run_failed',
+                  eventData: {
+                    error: {
+                      message: errorMessage,
+                      stack: errorStack,
+                    },
                     // TODO: include error codes when we define them
                   },
                 });