Fix missing realtime broadcasts for step:started and step:completed events (#305)

## Summary

Fixes critical bug where `step:started` and `step:completed` realtime events were not being broadcast to clients due to PostgreSQL query planner optimizing away unused CTEs.

## Bugs Fixed

### 1. Missing step:started broadcasts (Critical)
**Problem:** Clients never received `step:started` events when steps transitioned from Created to Started status.

**Root Cause:** The `broadcast_events` CTE in `start_ready_steps()` was not referenced by any subsequent CTEs or the final RETURN statement. PostgreSQL's query optimizer eliminated it as "dead code", preventing `realtime.send()` from executing.

**Solution:** Moved `realtime.send()` call directly into the RETURNING clause of the UPDATE statement that marks steps as Started. This ensures the broadcast executes atomically with the state change and cannot be optimized away.

### 2. Missing step:completed broadcasts for empty map steps (Critical)
**Problem:** Clients never received `step:completed` events for map steps that received empty arrays and completed immediately.

**Root Cause:** The `broadcast_empty_completed` CTE in `start_ready_steps()` suffered the same optimization issue - not referenced by subsequent operations.

**Solution:** Moved `realtime.send()` call into the RETURNING clause of the UPDATE that completes empty map steps.

### 3. Missing step:completed broadcasts in cascade completion (Critical)
**Problem:** Clients didn't receive events when taskless steps were cascade-completed in `cascade_complete_taskless_steps()`.

**Root Cause:** Similar CTE optimization issue in the cascade completion logic.

**Solution:** Added `realtime.send()` directly in the RETURNING clause of the cascade completion UPDATE.

## Technical Changes

### SQL Schema Updates

**`start_ready_steps.sql`:**
- Removed `broadcast_events` CTE (step:started)
- Removed `broadcast_empty_completed` CTE (step:completed for empty maps)
- Added `realtime.send()` in RETURNING clause of `started_step_states` CTE
- Added `realtime.send()` in RETURNING clause of `completed_empty_steps` CTE

**`cascade_complete_taskless_steps.sql`:**
- Added `realtime.send()` in RETURNING clause of cascade completion UPDATE
- Broadcasts step:completed events atomically with state transition

**`complete_task.sql`:**
- Added PERFORM statements for run:failed and step:failed event broadcasts
- Ensures error events are reliably broadcast

### Client-Side Improvements

**New `applySnapshot()` methods:**
- Added to `FlowRun` and `FlowStep` classes
- Applies database state directly without emitting events
- Used for initial state hydration from `start_flow_with_states()` and `get_run_with_states()`

**Why this matters:** Previously, we used `updateState()` for initial state, which internally tried to emit events. This was wrong - the initial state from the database is a snapshot, not a series of events. The new `applySnapshot()` method makes this distinction clear.

### Test Updates

Completely rewrote broadcast tests to understand the actual behavior:

**Root steps vs Dependent steps:**
- Root steps start in the same transaction as `start_flow()` - already Started when the function returns
- Can't observe their step:started broadcasts (would require listening before the transaction commits)
- Tests now verify state directly instead of trying to catch impossible broadcasts

**Dependent steps:**
- Start AFTER their dependencies complete (separate transaction)
- CAN observe step:started broadcasts (the actual test case for the bug fix)
- New test specifically verifies these broadcasts work correctly

**Empty map steps:**
- Root empty maps complete in `start_flow()` transaction - verify state directly
- Dependent empty maps complete after dependencies - verify broadcasts

## Testing

All tests passing:
- ✅ Root steps verified via state (already Started when startFlow() returns)
- ✅ Dependent steps receive step:started broadcasts
- ✅ Dependent empty maps receive step:completed broadcasts (skip step:started)
- ✅ Cascade completion broadcasts step:completed events
- ✅ Event sequence validation (started → completed order)

## Migration Notes

No breaking changes. This is a pure bug fix that makes the system work as originally intended.

Users who implemented workarounds for missing events can now remove them.

Author: jumski

.changeset/clear-signal-beam.md

@@ -0,0 +1,18 @@
+---
+'@pgflow/client': patch
+'@pgflow/core': patch
+---
+
+Fix missing realtime broadcasts for step:started and step:completed events
+
+**Critical bug fix:** Clients were not receiving `step:started` events when steps transitioned to Started status, and `step:completed` events for empty map steps and cascade completions were also missing.
+
+**Root cause:** PostgreSQL query optimizer was eliminating CTEs containing `realtime.send()` calls because they were not referenced by subsequent operations or the final RETURN statement.
+
+**Solution:** Moved `realtime.send()` calls directly into RETURNING clauses of UPDATE statements, ensuring they execute atomically with state changes and cannot be optimized away.
+
+**Changes:**
+- `start_ready_steps()`: Broadcasts step:started and step:completed events in RETURNING clauses
+- `cascade_complete_taskless_steps()`: Broadcasts step:completed events atomically with cascade completion
+- `complete_task()`: Added PERFORM statements for run:failed and step:failed broadcasts
+- Client: Added `applySnapshot()` methods to FlowRun and FlowStep for proper initial state hydration without event emission

pkgs/client/__tests__/integration/real-flow-execution.test.ts

@@ -36,11 +36,11 @@ describe('Real Flow Execution', () => {
       expect(run.flow_slug).toBe(testFlow.slug);
 
       // Give realtime subscription time to establish properly
-      await new Promise(resolve => setTimeout(resolve, 2000));
+      await new Promise((resolve) => setTimeout(resolve, 2000));
 
       // Poll for task
       const tasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
-      
+
       expect(tasks).toHaveLength(1);
       expect(tasks[0].input.run).toEqual(input);
 
@@ -112,7 +112,7 @@ describe('Real Flow Execution', () => {
       step.on('*', stepTracker.callback);
 
       // Give realtime subscription time to establish
-      await new Promise(resolve => setTimeout(resolve, 100));
+      await new Promise((resolve) => setTimeout(resolve, 100));
 
       // Poll and complete task
       const tasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
@@ -147,54 +147,120 @@ describe('Real Flow Execution', () => {
   );
 
   it(
-    'CRITICAL: broadcasts step:started events (CTE optimization bug check)',
+    'root steps: started immediately (verify via waitForStatus, not broadcasts)',
     withPgNoTransaction(async (sql) => {
-      // This test specifically verifies that step:started events ARE broadcast
-      // It SHOULD FAIL until the CTE optimization bug is fixed in start_ready_steps()
+      // Root steps are started in the same transaction as start_flow()
+      // By the time startFlow() returns, they're already Started
+      // We can't observe these broadcasts - they happen before we can listen
+      // Instead, verify the state directly
 
-      const testFlow = createTestFlow('started_event_flow');
+      const testFlow = createTestFlow('root_started_flow');
       await cleanupFlow(sql, testFlow.slug);
       await grantMinimalPgflowPermissions(sql);
       await sql`SELECT pgflow.create_flow(${testFlow.slug})`;
-      await sql`SELECT pgflow.add_step(${testFlow.slug}, 'test_step')`;
+      await sql`SELECT pgflow.add_step(${testFlow.slug}, 'root_step')`;
 
       const sqlClient = new PgflowSqlClient(sql);
       const supabaseClient = createTestSupabaseClient();
       const pgflowClient = new PgflowClient(supabaseClient);
 
+      // Start flow - root step starts in this transaction
       const run = await pgflowClient.startFlow(testFlow.slug, { test: 'data' });
-      const step = run.step('test_step');
+      const step = run.step('root_step');
 
-      // Track ALL step events with event matchers
-      const tracker = createEventTracker();
-      step.on('*', tracker.callback);
+      // VERIFY: Step is already Started when startFlow() returns
+      expect(step.status).toBe(FlowStepStatus.Started);
+      expect(step.started_at).toBeDefined();
 
-      // Give realtime subscription time to establish
-      await new Promise(resolve => setTimeout(resolve, 100));
+      // waitForStatus should return immediately (already Started)
+      await step.waitForStatus(FlowStepStatus.Started, { timeoutMs: 1000 });
 
-      // Execute the step - this calls start_ready_steps() which should broadcast step:started
+      // Complete for cleanup
       const tasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
-
-      // Wait a moment for broadcast to propagate
-      await new Promise(resolve => setTimeout(resolve, 500));
-
-      // Complete the task
       await sqlClient.completeTask(tasks[0], { result: 'done' });
       await step.waitForStatus(FlowStepStatus.Completed, { timeoutMs: 5000 });
 
-      // CRITICAL ASSERTIONS: Verify step:started WAS broadcast
-      // These will FAIL with the current CTE optimization bug!
+      await supabaseClient.removeAllChannels();
+    }),
+    15000
+  );
+
+  it(
+    'dependent steps: broadcasts step:started when they become ready',
+    withPgNoTransaction(async (sql) => {
+      // Dependent steps start AFTER their dependencies complete
+      // This happens AFTER startFlow() returns, so we CAN observe broadcasts
+      // This is the real test for step:started broadcasts!
+
+      const testFlow = createTestFlow('dependent_started_flow');
+      await cleanupFlow(sql, testFlow.slug);
+      await grantMinimalPgflowPermissions(sql);
+
+      await sql`SELECT pgflow.create_flow(${testFlow.slug})`;
+      await sql`SELECT pgflow.add_step(${testFlow.slug}, 'root_step')`;
+      await sql`SELECT pgflow.add_step(${testFlow.slug}, 'dependent_step', ARRAY['root_step'])`;
+
+      const sqlClient = new PgflowSqlClient(sql);
+      const supabaseClient = createTestSupabaseClient();
+      const pgflowClient = new PgflowClient(supabaseClient);
+
+      // Start flow - only root_step starts
+      const run = await pgflowClient.startFlow(testFlow.slug, { test: 'data' });
+      const rootStep = run.step('root_step');
+      const dependentStep = run.step('dependent_step');
+
+      // Root is started, dependent is still created (waiting for deps)
+      expect(rootStep.status).toBe(FlowStepStatus.Started);
+      expect(dependentStep.status).toBe(FlowStepStatus.Created);
+
+      // NOW set up event tracker (before completing root)
+      const tracker = createEventTracker();
+      dependentStep.on('*', tracker.callback);
+
+      // Give realtime subscription time to establish
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      // Complete root step - this will trigger dependent_step to start
+      const rootTasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
+      expect(rootTasks[0].step_slug).toBe('root_step');
+      await sqlClient.completeTask(rootTasks[0], { result: 'root done' });
+
+      // Wait for dependent to start
+      await dependentStep.waitForStatus(FlowStepStatus.Started, {
+        timeoutMs: 5000,
+      });
+
+      // VERIFY: We received step:started broadcast for dependent step
       expect(tracker).toHaveReceivedEvent('step:started', {
         run_id: run.run_id,
-        step_slug: 'test_step',
+        step_slug: 'dependent_step',
         status: FlowStepStatus.Started,
       });
 
-      // Verify proper event sequence
-      expect(tracker).toHaveReceivedEventSequence(['step:started', 'step:completed']);
-      expect(tracker).toHaveReceivedInOrder('step:started', 'step:completed');
+      // Complete dependent step
+      const dependentTasks = await readAndStart(
+        sql,
+        sqlClient,
+        testFlow.slug,
+        1,
+        5
+      );
+      expect(dependentTasks[0].step_slug).toBe('dependent_step');
+      await sqlClient.completeTask(dependentTasks[0], {
+        result: 'dependent done',
+      });
+
+      // Wait for completion
+      await dependentStep.waitForStatus(FlowStepStatus.Completed, {
+        timeoutMs: 5000,
+      });
 
-      // Verify both events were received
+      // VERIFY: Proper event sequence
+      expect(tracker).toHaveReceivedEventSequence([
+        'step:started',
+        'step:completed',
+      ]);
+      expect(tracker).toHaveReceivedInOrder('step:started', 'step:completed');
       expect(tracker).toHaveReceivedEventCount('step:started', 1);
       expect(tracker).toHaveReceivedEventCount('step:completed', 1);
 
@@ -204,12 +270,13 @@ describe('Real Flow Execution', () => {
   );
 
   it(
-    'empty map steps: skip step:started and go straight to step:completed',
+    'empty map steps (root): completed immediately (verify via state)',
     withPgNoTransaction(async (sql) => {
-      // This test verifies the EXPECTED behavior for empty map steps
-      // They should NOT send step:started, only step:completed
+      // Empty map steps with no tasks complete immediately
+      // Root empty maps complete in the start_flow transaction
+      // Can't observe broadcasts - verify state instead
 
-      const testFlow = createTestFlow('empty_map_flow');
+      const testFlow = createTestFlow('root_empty_map_flow');
       await cleanupFlow(sql, testFlow.slug);
       await grantMinimalPgflowPermissions(sql);
 
@@ -229,31 +296,81 @@ describe('Real Flow Execution', () => {
       const supabaseClient = createTestSupabaseClient();
       const pgflowClient = new PgflowClient(supabaseClient);
 
-      // Start flow with empty array directly (root map steps expect array input)
+      // Start flow with empty array (root map steps expect array input)
       const run = await pgflowClient.startFlow(testFlow.slug, []);
       const step = run.step('empty_map_step');
 
-      // Track events
+      // VERIFY: Step is already Completed when startFlow() returns
+      expect(step.status).toBe(FlowStepStatus.Completed);
+      expect(step.completed_at).toBeDefined();
+
+      // Empty maps DO get started_at set (they transition through started briefly)
+      expect(step.started_at).toBeDefined();
+
+      await supabaseClient.removeAllChannels();
+    }),
+    15000
+  );
+
+  it(
+    'empty map steps (dependent): broadcasts step:completed when triggered',
+    withPgNoTransaction(async (sql) => {
+      // Dependent empty map steps complete AFTER their dependencies
+      // This happens AFTER startFlow() returns, so we CAN observe broadcasts
+      // They skip step:started and go directly to step:completed
+
+      const testFlow = createTestFlow('dependent_empty_map_flow');
+      await cleanupFlow(sql, testFlow.slug);
+      await grantMinimalPgflowPermissions(sql);
+
+      await sql`SELECT pgflow.create_flow(${testFlow.slug})`;
+      await sql`SELECT pgflow.add_step(${testFlow.slug}, 'root_step')`;
+      // Dependent map step that will receive empty array from root
+      await sql`SELECT pgflow.add_step(
+        ${testFlow.slug},
+        'dependent_empty_map',
+        ARRAY['root_step'],
+        NULL,
+        NULL,
+        NULL,
+        NULL,
+        'map'
+      )`;
+
+      const sqlClient = new PgflowSqlClient(sql);
+      const supabaseClient = createTestSupabaseClient();
+      const pgflowClient = new PgflowClient(supabaseClient);
+
+      const run = await pgflowClient.startFlow(testFlow.slug, { test: 'data' });
+      const rootStep = run.step('root_step');
+      const emptyMapStep = run.step('dependent_empty_map');
+
+      // Set up tracker before completing root
       const tracker = createEventTracker();
-      step.on('*', tracker.callback);
+      emptyMapStep.on('*', tracker.callback);
 
-      // Give realtime time to establish
-      await new Promise(resolve => setTimeout(resolve, 100));
+      await new Promise((resolve) => setTimeout(resolve, 100));
 
-      // Wait for step to complete (should happen immediately)
-      await step.waitForStatus(FlowStepStatus.Completed, { timeoutMs: 5000 });
+      // Complete root with empty array (single steps feeding map steps output arrays directly)
+      const rootTasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
+      await sqlClient.completeTask(rootTasks[0], []);
+
+      // Wait for dependent to complete (should happen immediately)
+      await emptyMapStep.waitForStatus(FlowStepStatus.Completed, {
+        timeoutMs: 5000,
+      });
 
-      // Verify NO step:started event (expected for empty maps)
+      // VERIFY: NO step:started (empty maps skip this)
       expect(tracker).toNotHaveReceivedEvent('step:started');
 
-      // Verify step:completed was sent
+      // VERIFY: Received step:completed directly
       expect(tracker).toHaveReceivedEvent('step:completed', {
         run_id: run.run_id,
-        step_slug: 'empty_map_step',
+        step_slug: 'dependent_empty_map',
         status: FlowStepStatus.Completed,
       });
 
-      // Verify only 1 event total
+      // VERIFY: Only 1 event total (completed, no started)
       expect(tracker).toHaveReceivedTotalEvents(1);
 
       await supabaseClient.removeAllChannels();
@@ -285,10 +402,12 @@ describe('Real Flow Execution', () => {
       expect(step.started_at).toBeDefined();
 
       // Give realtime subscription time to establish
-      await new Promise(resolve => setTimeout(resolve, 100));
+      await new Promise((resolve) => setTimeout(resolve, 100));
 
       // waitForStatus should resolve immediately since step is already Started
-      const waitPromise = step.waitForStatus(FlowStepStatus.Started, { timeoutMs: 5000 });
+      const waitPromise = step.waitForStatus(FlowStepStatus.Started, {
+        timeoutMs: 5000,
+      });
       const result = await waitPromise;
       expect(result).toBe(step);
       expect(step.status).toBe(FlowStepStatus.Started);

pkgs/client/src/lib/FlowRun.ts

@@ -264,13 +264,33 @@ export class FlowRun<TFlow extends AnyFlow>
     });
   }
 
+  /**
+   * Apply state from database snapshot (no events emitted)
+   * Used when initializing state from start_flow_with_states() or get_run_with_states()
+   *
+   * @internal This method is only intended for use by PgflowClient.
+   * Applications should not call this directly.
+   */
+  applySnapshot(row: import('@pgflow/core').RunRow): void {
+    // Direct state assignment from database row (no event conversion)
+    this.#state.status = row.status as FlowRunStatus;
+    this.#state.input = row.input as ExtractFlowInput<TFlow>;
+    this.#state.output = row.output as ExtractFlowOutput<TFlow> | null;
+    this.#state.started_at = row.started_at ? new Date(row.started_at) : null;
+    this.#state.completed_at = row.completed_at ? new Date(row.completed_at) : null;
+    this.#state.failed_at = row.failed_at ? new Date(row.failed_at) : null;
+    this.#state.remaining_steps = row.remaining_steps;
+    this.#state.error_message = null; // Database doesn't have error_message for runs
+    this.#state.error = null;
+  }
+
   /**
    * Updates the run state based on an event
-   * 
+   *
    * @internal This method is only intended for use by PgflowClient and tests.
    * Applications should not call this directly - state updates should come from
    * database events through the PgflowClient.
-   * 
+   *
    * TODO: After v1.0, make this method private and refactor tests to use PgflowClient
    * with event emission instead of direct state manipulation.
    */

pkgs/client/src/lib/FlowStep.ts

@@ -176,13 +176,31 @@ export class FlowStep<
     });
   }
 
+  /**
+   * Apply state from database snapshot (no events emitted)
+   * Used when initializing state from start_flow_with_states() or get_run_with_states()
+   *
+   * @internal This method is only intended for use by PgflowClient.
+   * Applications should not call this directly.
+   */
+  applySnapshot(row: import('@pgflow/core').StepStateRow): void {
+    // Direct state assignment from database row (no event conversion)
+    this.#state.status = row.status as FlowStepStatus;
+    this.#state.started_at = row.started_at ? new Date(row.started_at) : null;
+    this.#state.completed_at = row.completed_at ? new Date(row.completed_at) : null;
+    this.#state.failed_at = row.failed_at ? new Date(row.failed_at) : null;
+    this.#state.error_message = row.error_message;
+    this.#state.error = row.error_message ? new Error(row.error_message) : null;
+    // Note: output is not stored in step_states table, remains null
+  }
+
   /**
    * Updates the step state based on an event
-   * 
+   *
    * @internal This method is only intended for use by FlowRun and tests.
    * Applications should not call this directly - state updates should come from
    * database events through the PgflowClient.
-   * 
+   *
    * TODO: After v1.0, make this method private and refactor tests to use PgflowClient
    * with event emission instead of direct state manipulation.
    */