V48 Gate 3 (implementation-only): fix executions-row race causing FK violations on every synthesis run

Every deposit synthesis run was logging repeated "Failed to persist stream
event" (postgres 23503: execution_events.run_id FK not present in executions)
and the client's post-completion GET /api/executions/history/<runId> could
404. Root cause: enablePipelineStreaming's "ensure an executions row exists"
insert was fire-and-forget (not awaited), while createStreamingExecution emits
its first status event synchronously right after — so the very first (and any
early) execution_events insert could race ahead of the executions row actually
landing.

Fix (pipeline-stream-integration.ts): track the ensure-row insert as a promise
and have the event-persistence subscriber await it before persisting the
first event (no extra round trip — the insert was already in flight), plus a
self-healing backstop that, on any 23503, ensures the row and retries the
event once. This is the shared streaming integration used by every pipeline
(deposit, read, terminal), so the fix applies universally, not just to
deposit synthesis.

Added two regression tests: one proves an event given a slow executions
insert does not persist until the insert resolves (previously it would race
ahead), the other proves a genuine 23503 self-heals via ensure+retry. Package
suite green (23/23); typecheck clean.

Author: geraldarthurdavis

packages/pipelines-generics/src/streaming/__tests__/pipeline-stream-integration.test.ts

@@ -4,10 +4,19 @@ import { enablePipelineStreaming, sourceSafeStreamEvent } from '../../streaming/
 
 // Mock ORM model so we can assert persistence without a real DB
 const createdEvents: any[] = [];
+let failNextCreateWithFkViolation = false;
 jest.mock('@bitcode/orm', () => ({
   ExecutionEventsModel: class {
     constructor() {}
     async create(row: any) {
+      if (failNextCreateWithFkViolation) {
+        failNextCreateWithFkViolation = false;
+        const err: any = new Error(
+          'insert or update on table "execution_events" violates foreign key constraint "execution_events_run_id_fkey"',
+        );
+        err.code = '23503';
+        throw err;
+      }
       createdEvents.push(row);
       return { id: String(createdEvents.length), ...row };
     }
@@ -41,6 +50,89 @@ describe('pipeline-stream-integration', () => {
   });
 });
 
+describe('pipeline-stream-integration — executions-row FK race (QA: "Failed to persist stream event" 23503)', () => {
+  it('gates the first persisted event on the executions-row insert completing, instead of racing it', async () => {
+    let resolveInsert: () => void;
+    const insertPromise = new Promise<void>((resolve) => {
+      resolveInsert = resolve;
+    });
+    const fakeSupabase: any = {
+      from: (table: string) => {
+        expect(table).toBe('executions');
+        return {
+          insert: async () => {
+            await insertPromise; // simulates a slow executions-row insert round trip
+            return { error: null };
+          },
+        };
+      },
+    };
+
+    const exec = new Execution('pipeline:race-test');
+    const runId = '11111111-1111-4111-8111-111111111111';
+    const streamer = enablePipelineStreaming(exec, {
+      runId,
+      userId: 'user-race',
+      supabase: fakeSupabase,
+      streamToDatabase: true,
+      streamToSSE: false,
+    });
+
+    const before = createdEvents.length;
+    const writeDone = streamer.writeData(
+      JSON.stringify({ type: 'status', message: 'started', timestamp: Date.now() }),
+    );
+
+    // While the executions-row insert is still in flight, the event must NOT
+    // have persisted yet (this is the FK-violation race: previously the event
+    // insert fired immediately, unordered relative to the executions insert).
+    await new Promise((r) => setTimeout(r, 10));
+    expect(createdEvents.length).toBe(before);
+
+    resolveInsert!();
+    await writeDone;
+    await new Promise((r) => setTimeout(r, 10));
+
+    expect(createdEvents.length).toBe(before + 1);
+    expect(createdEvents[createdEvents.length - 1].run_id).toBe(runId);
+  });
+
+  it('self-heals a 23503 FK violation by ensuring the executions row and retrying once', async () => {
+    let insertCallCount = 0;
+    const fakeSupabase: any = {
+      from: (table: string) => {
+        expect(table).toBe('executions');
+        return {
+          insert: async () => {
+            insertCallCount += 1;
+            return { error: null };
+          },
+        };
+      },
+    };
+    failNextCreateWithFkViolation = true;
+
+    const exec = new Execution('pipeline:retry-test');
+    const runId = '22222222-2222-4222-8222-222222222222';
+    const streamer = enablePipelineStreaming(exec, {
+      runId,
+      userId: 'user-retry',
+      supabase: fakeSupabase,
+      streamToDatabase: true,
+      streamToSSE: false,
+    });
+
+    const before = createdEvents.length;
+    await streamer.writeData(JSON.stringify({ type: 'status', message: 'hello', timestamp: Date.now() }));
+    await new Promise((r) => setTimeout(r, 10));
+
+    expect(createdEvents.length).toBe(before + 1);
+    expect(createdEvents[createdEvents.length - 1].run_id).toBe(runId);
+    // The upfront best-effort insert, plus the retry-triggered ensure-row insert.
+    expect(insertCallCount).toBeGreaterThanOrEqual(2);
+  });
+});
+
 describe('sourceSafeStreamEvent (telemetry source-safety law, V48)', () => {
   const RAW_RESPONSE = '```json\n{\n  "analysis": "secret plan prose",\n  "steps": ["step one"]\n}\n```';
 

packages/pipelines-generics/src/streaming/pipeline-stream-integration.ts

@@ -121,45 +121,71 @@ export function enablePipelineStreaming(
     (config.structuredToDatabase !== true || process?.env?.BITCODE_PIPELINE_LEGACY_EVENTS_DB === '1');
 
   if (legacyExecutionEventsEnabled && config.supabase) {
-    // Best-effort: ensure an executions row exists if runId looks like a UUID
-    try {
-      const uuidRe = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
-      const isUuid = uuidRe.test(String(config.runId || ''));
-      if (isUuid) {
-        const now = new Date().toISOString();
-        // Try insert; ignore duplicate errors
-        config.supabase
-          .from('executions')
-          .insert({
-            id: config.runId,
-            user_id: config.userId,
-            status: 'running',
-            type: 'agentic-execution:asset-pack',
-            started_at: now,
-            created_at: now,
-            updated_at: now
-          } as any)
-          .then(() => {})
-          .catch((e: any) => {
-            if (!String(e?.message || '').toLowerCase().includes('duplicate')) {
-              // swallow non-duplicate as best-effort
-            }
-          });
+    const uuidRe = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
+    const isUuid = uuidRe.test(String(config.runId || ''));
+
+    // Ensure an `executions` row exists before any `execution_events` row can
+    // reference it (execution_events.run_id has a FK to executions.id). Idempotent
+    // (duplicate insert = row already exists, either way).
+    const ensureExecutionsRowExists = async (): Promise<void> => {
+      if (!isUuid) return;
+      const now = new Date().toISOString();
+      try {
+        await config.supabase.from('executions').insert({
+          id: config.runId,
+          user_id: config.userId,
+          status: 'running',
+          type: 'agentic-execution:asset-pack',
+          started_at: now,
+          created_at: now,
+          updated_at: now,
+        } as any);
+      } catch (e: any) {
+        if (!String(e?.message || '').toLowerCase().includes('duplicate')) {
+          // swallow non-duplicate as best-effort; the retry-on-FK-violation below
+          // is the guaranteed backstop.
+        }
       }
-    } catch {}
+    };
+
+    // Fire immediately (don't block registerStreamer/emitEvent below), but track
+    // the promise so the FIRST event to persist can await it — otherwise the
+    // initial "Pipeline execution started" status event (emitted synchronously
+    // right after this function returns, e.g. by createStreamingExecution) races
+    // this insert and can hit a 23503 FK violation before the row lands.
+    let pendingEnsureRow: Promise<void> | null = ensureExecutionsRowExists();
 
     const eventsModel = new ExecutionEventsModel(config.supabase);
-    
+
     // Subscribe to stream events and persist them
     streamer.subscribe(async (event) => {
-      try {
-        await eventsModel.create({
+      if (pendingEnsureRow) {
+        await pendingEnsureRow;
+        pendingEnsureRow = null;
+      }
+      const persist = () =>
+        eventsModel.create({
           run_id: config.runId,
           event_type: event.type,
           event_data: sourceSafeStreamEvent(event),
           created_at: new Date().toISOString(),
         });
-      } catch (error) {
+      try {
+        await persist();
+      } catch (error: any) {
+        // Self-healing backstop: if the executions row still doesn't exist (any
+        // residual race — slow network, replica lag, concurrent dispatch), ensure
+        // it and retry once before giving up.
+        if (error?.code === '23503') {
+          try {
+            await ensureExecutionsRowExists();
+            await persist();
+            return;
+          } catch (retryError) {
+            console.error('Failed to persist stream event after ensuring executions row:', retryError);
+            return;
+          }
+        }
         console.error('Failed to persist stream event:', error);
       }
     });