feat(flows): foundation for interactive WhatsApp messages (PR 1/4)

Foundation for the upcoming WhatsApp chatbot flow builder. No runner,
no UI yet — just the plumbing that lets the codebase SEND interactive
messages (button replies + lists) and RECEIVE the taps as structured
message rows. Plan: ~/.claude/plans/understood-research-such-functionality-binary-lake.md

## What's in this PR

### Migration 010_flows.sql
- New tables: flows, flow_nodes, flow_runs, flow_run_events
- Linchpin partial unique index on flow_runs(user_id, contact_id) WHERE status='active' — at most one active run per contact, makes concurrent webhook deliveries idempotent at the DB layer
- Widens messages.content_type CHECK to allow 'interactive'
- New messages.interactive_reply_id column for the tapped button/row id
- RLS on every table; flow_runs added to supabase_realtime publication
- Idempotent throughout — matches the 009_message_actions.sql template

### Interactive Meta senders (src/lib/whatsapp/meta-api.ts)
- sendInteractiveButtons: type='interactive', interactive.type='button', up to 3 reply buttons
- sendInteractiveList: type='interactive', interactive.type='list', up to 10 rows across sections
- Limits exported as INTERACTIVE_LIMITS so flow validation can share the constants
- All Meta caps validated BEFORE the network call (button title ≤20 chars, list row title ≤24 chars, ≤10 total rows, no duplicate row ids, body ≤1024 chars, etc.) — misconfigured flows fail at save time instead of as a customer-facing 400 mid-conversation

### Webhook (src/app/api/whatsapp/webhook/route.ts)
- Extends WhatsAppMessage interface with `interactive?: { type, button_reply?, list_reply? }`
- parseMessageContent returns a new `interactiveReplyId` field; refactored to use a shared `empty` base so every return site picks up the new field without 9 explicit copies
- New `case 'interactive'`: uses the tapped option's TITLE as content_text (so the inbox bubble reads "Existing customer", not a UUID) and stashes the id separately
- 'interactive' added to ALLOWED_CONTENT_TYPES so the messages insert no longer coerces it to 'text'

### Inbox bubble (src/components/inbox/message-bubble.tsx)
- New `case "interactive"` in MessageContent renders the tap with a small "↩ Button reply" affordance so agents reading the inbox can tell at a glance that this was a tap rather than the customer typing the same words

### Types (src/types/index.ts)
- ContentType union extended with 'interactive'
- Message.interactive_reply_id field

### Engine helpers (src/lib/flows/)
- admin-client.ts: thin service-role client wrapper, mirrors automations/admin-client.ts
- meta-send.ts: engineSendInteractiveButtons / engineSendInteractiveList — adapts the Meta senders for service-role use, persists messages with sender_type='bot' content_type='interactive', uses the same phone-variant retry as automations/meta-send.ts. Unused in PR #1; ready for the runner in PR #2

### Tests
- 13 new vitest cases in meta-api.test.ts covering Meta's limit validation + payload-shape assertions for both interactive senders

## Validation
- npm test → 102/102 (was 89; 13 new)
- npm run lint → 0 errors
- npm run typecheck → clean
- npm run build → clean

## What's NOT in this PR
- The runner (dispatchInboundToFlows) — PR #2
- Builder UI / API surface — PR #3
- collect_input / condition / set_tag nodes + templates — PR #4
- react-flow canvas + http_fetch — v2 (PR #5)

## Risks tracked from the plan
- messages.content_type CHECK widening: grep'd every consumer; message-bubble.tsx, reply-quote.tsx, send/route.ts, automations/meta-send.ts all handle the new variant correctly (or accept it as a defaulted fallthrough)
- engineSendText / engineSendInteractive share ~80% of code; deliberately not pre-factored yet — let both stabilize, extract a shared base in PR #2 once the shape settles

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/app/api/whatsapp/webhook/route.ts

@@ -32,6 +32,17 @@ interface WhatsAppMessage {
   sticker?: { id: string; mime_type: string }
   location?: { latitude: number; longitude: number; name?: string; address?: string }
   reaction?: { message_id: string; emoji: string }
+  /**
+   * Set when the customer taps a button or list row on an interactive
+   * message we sent. `button_reply.id` / `list_reply.id` is whatever id
+   * we put on the button/row when sending — the Flows engine uses this
+   * to advance the per-contact run.
+   */
+  interactive?: {
+    type: 'button_reply' | 'list_reply'
+    button_reply?: { id: string; title: string }
+    list_reply?: { id: string; title: string; description?: string }
+  }
   /** Present when the customer swipe-replies to one of our messages. */
   context?: { id: string }
 }
@@ -468,10 +479,8 @@ async function processMessage(
   }
 
   // Parse message content based on type
-  const { contentText, mediaUrl, mediaType } = await parseMessageContent(
+  const { contentText, mediaUrl, mediaType, interactiveReplyId } =
-    message,
+    await parseMessageContent(message, accessToken)
-    accessToken
-  )
 
   // Resolve swipe-reply context if present. A missing parent is fine —
   // we just store NULL and the UI renders the message without a quote.
@@ -498,12 +507,14 @@ async function processMessage(
   // parseMessageContent. Silence the unused-var warning:
   void mediaType
 
-  // The messages.content_type CHECK constraint only allows:
+  // The messages.content_type CHECK constraint (widened in migration 010
-  //   text, image, document, audio, video, location, template
+  // to add 'interactive' for button/list taps) allows:
+  //   text, image, document, audio, video, location, template, interactive
   // Map incoming WhatsApp types that aren't in that list to the closest
   // allowed value so the INSERT doesn't fail with a constraint error.
   const ALLOWED_CONTENT_TYPES = new Set([
-    'text', 'image', 'document', 'audio', 'video', 'location', 'template',
+    'text', 'image', 'document', 'audio', 'video',
+    'location', 'template', 'interactive',
   ])
   const contentType = ALLOWED_CONTENT_TYPES.has(message.type)
     ? message.type
@@ -532,6 +543,10 @@ async function processMessage(
     status: 'delivered',
     created_at: new Date(parseInt(message.timestamp) * 1000).toISOString(),
     reply_to_message_id: replyToInternalId,
+    // Only populated for content_type='interactive'. Migration 010 added
+    // the column; null for every other content_type so existing inserts
+    // behave identically.
+    interactive_reply_id: interactiveReplyId,
   })
 
   if (msgError) {
@@ -599,6 +614,14 @@ async function parseMessageContent(
   contentText: string | null
   mediaUrl: string | null
   mediaType: string | null
+  /**
+   * For interactive button / list replies: the stable id of the tapped
+   * option (whatever we put on the button when sending). Used by the
+   * Flows engine to advance the per-contact run; persisted to
+   * `messages.interactive_reply_id` so the inbox bubble can render the
+   * tap with the right affordance. Null for everything else.
+   */
+  interactiveReplyId: string | null
 }> {
   // getMediaUrl signature is (mediaId, accessToken) — earlier code had
   // the args swapped, so every verification hit an invalid Meta URL and
@@ -619,54 +642,62 @@ async function parseMessageContent(
     }
   }
 
-  switch (message.type) {
+  // Default shape — each case overrides only the fields it cares about.
-    case 'text':
+  // Keeps the new `interactiveReplyId` field DRY across every return site.
-      return {
+  const empty = {
-        contentText: message.text?.body || null,
+    contentText: null,
     mediaUrl: null,
     mediaType: null,
+    interactiveReplyId: null,
   }
 
+  switch (message.type) {
+    case 'text':
+      return { ...empty, contentText: message.text?.body || null }
+
     case 'image':
       if (message.image?.id) {
         return {
+          ...empty,
           contentText: message.image.caption || null,
           mediaUrl: await verifyAndBuildUrl(message.image.id),
           mediaType: message.image.mime_type,
         }
       }
-      return { contentText: null, mediaUrl: null, mediaType: null }
+      return empty
 
     case 'video':
       if (message.video?.id) {
         return {
+          ...empty,
           contentText: message.video.caption || null,
           mediaUrl: await verifyAndBuildUrl(message.video.id),
           mediaType: message.video.mime_type,
         }
       }
-      return { contentText: null, mediaUrl: null, mediaType: null }
+      return empty
 
     case 'document':
       if (message.document?.id) {
         return {
+          ...empty,
           contentText:
             message.document.caption || message.document.filename || null,
           mediaUrl: await verifyAndBuildUrl(message.document.id),
           mediaType: message.document.mime_type,
         }
       }
-      return { contentText: null, mediaUrl: null, mediaType: null }
+      return empty
 
     case 'audio':
       if (message.audio?.id) {
         return {
-          contentText: null,
+          ...empty,
           mediaUrl: await verifyAndBuildUrl(message.audio.id),
           mediaType: message.audio.mime_type,
         }
       }
-      return { contentText: null, mediaUrl: null, mediaType: null }
+      return empty
 
     case 'sticker':
       // Stickers are images under the hood. Treat them as such so the
@@ -674,12 +705,12 @@ async function parseMessageContent(
       // content_type to 'image' for the CHECK constraint.
       if (message.sticker?.id) {
         return {
-          contentText: null,
+          ...empty,
           mediaUrl: await verifyAndBuildUrl(message.sticker.id),
           mediaType: message.sticker.mime_type,
         }
       }
-      return { contentText: null, mediaUrl: null, mediaType: null }
+      return empty
 
     case 'location':
       if (message.location) {
@@ -687,26 +718,36 @@ async function parseMessageContent(
         const locationText = [loc.name, loc.address, `${loc.latitude},${loc.longitude}`]
           .filter(Boolean)
           .join(' - ')
-        return {
+        return { ...empty, contentText: locationText }
-          contentText: locationText,
-          mediaUrl: null,
-          mediaType: null,
       }
-      }
+      return empty
-      return { contentText: null, mediaUrl: null, mediaType: null }
 
     case 'reaction':
+      return { ...empty, contentText: message.reaction?.emoji || null }
+
+    case 'interactive': {
+      // The customer tapped a reply button or a list row on a message
+      // we previously sent. Meta delivers `interactive.button_reply` for
+      // 3-button messages and `interactive.list_reply` for list messages.
+      // Use the human-readable title as contentText so the inbox bubble
+      // renders the tap legibly ("Existing customer"), and stash the
+      // stable id separately so the Flows engine can route on it.
+      const reply =
+        message.interactive?.button_reply ?? message.interactive?.list_reply
+      if (reply?.id) {
         return {
-        contentText: message.reaction?.emoji || null,
+          ...empty,
-        mediaUrl: null,
+          contentText: reply.title || reply.id,
-        mediaType: null,
+          interactiveReplyId: reply.id,
+        }
+      }
+      return { ...empty, contentText: '[Interactive reply]' }
     }
 
     default:
       return {
+        ...empty,
         contentText: `[Unsupported message type: ${message.type}]`,
-        mediaUrl: null,
-        mediaType: null,
       }
   }
 }

src/components/inbox/message-bubble.tsx

@@ -12,6 +12,7 @@ import {
   MapPin,
   LayoutTemplate,
   ImageOff,
+  CornerDownLeft,
 } from "lucide-react";
 import { format } from "date-fns";
 import { ReplyQuote } from "./reply-quote";
@@ -212,6 +213,24 @@ function MessageContent({ message }: { message: Message }) {
         </div>
       );
 
+    case "interactive":
+      // Customer tapped a reply button or list row on a message the bot
+      // sent. We show the tapped option's title (already in content_text,
+      // set by parseMessageContent in the webhook) with a small affordance
+      // so agents reading the inbox can tell at a glance that this is a
+      // tap rather than the customer typing the same words.
+      return (
+        <div className="flex flex-col gap-0.5">
+          <span className="inline-flex items-center gap-1 text-[10px] font-medium uppercase tracking-wide text-slate-400">
+            <CornerDownLeft className="h-3 w-3" />
+            Button reply
+          </span>
+          <p className="whitespace-pre-wrap break-words text-sm">
+            {message.content_text || "[Interactive reply]"}
+          </p>
+        </div>
+      );
+
     default:
       return (
         <p className="whitespace-pre-wrap break-words text-sm">

src/lib/flows/admin-client.ts

@@ -0,0 +1,16 @@
+import { createClient, type SupabaseClient } from '@supabase/supabase-js'
+
+// Lazy, shared service-role client for the Flows engine.
+// Mirrors src/lib/automations/admin-client.ts — same shape so anyone
+// reading either file picks up the convention immediately.
+let _adminClient: SupabaseClient | null = null
+
+export function supabaseAdmin(): SupabaseClient {
+  if (!_adminClient) {
+    _adminClient = createClient(
+      process.env.NEXT_PUBLIC_SUPABASE_URL!,
+      process.env.SUPABASE_SERVICE_ROLE_KEY!,
+    )
+  }
+  return _adminClient
+}

src/lib/flows/meta-send.ts

@@ -0,0 +1,201 @@
+import {
+  sendInteractiveButtons,
+  sendInteractiveList,
+  type InteractiveButton,
+  type InteractiveListSection,
+} from '@/lib/whatsapp/meta-api'
+import { decrypt } from '@/lib/whatsapp/encryption'
+import {
+  sanitizePhoneForMeta,
+  isValidE164,
+  phoneVariants,
+  isRecipientNotAllowedError,
+} from '@/lib/whatsapp/phone-utils'
+import { supabaseAdmin } from './admin-client'
+
+// ------------------------------------------------------------
+// Flows-side Meta sender (interactive variants).
+//
+// Mirrors src/lib/automations/meta-send.ts (engineSendText /
+// engineSendTemplate) but emits interactive button + list messages.
+// Kept separate from the automations file so the two engines don't
+// fight over each other's shape — once both stabilize, the
+// phone-variant retry + DB persistence are obvious extraction
+// candidates into a shared base.
+//
+// PR #1 ships this in isolation: callers don't exist yet. PR #2
+// brings the flow runner online and wires it up. Shipping it now
+// keeps the foundation PR self-contained and unit-testable.
+// ------------------------------------------------------------
+
+interface SendInteractiveButtonsEngineArgs {
+  userId: string
+  conversationId: string
+  contactId: string
+  bodyText: string
+  buttons: InteractiveButton[]
+  headerText?: string
+  footerText?: string
+}
+
+interface SendInteractiveListEngineArgs {
+  userId: string
+  conversationId: string
+  contactId: string
+  bodyText: string
+  buttonLabel: string
+  sections: InteractiveListSection[]
+  headerText?: string
+  footerText?: string
+}
+
+/**
+ * Send an interactive-button WhatsApp message from the Flows engine.
+ *
+ * Persists the outgoing message to `messages` with
+ * `content_type='interactive'` and `sender_type='bot'` so the inbox
+ * surfaces it with the "Button reply" affordance and the conversation
+ * thread reflects the bot's prompt.
+ *
+ * Returns the Meta message id so the caller (engine) can stash it on
+ * the `flow_runs.last_prompt_message_id` field for later reference.
+ */
+export async function engineSendInteractiveButtons(
+  args: SendInteractiveButtonsEngineArgs,
+): Promise<{ whatsapp_message_id: string }> {
+  return sendInteractiveViaMeta({ ...args, kind: 'buttons' })
+}
+
+/**
+ * Send an interactive-list WhatsApp message from the Flows engine.
+ * Used when the flow needs more than 3 options (Meta's button cap).
+ */
+export async function engineSendInteractiveList(
+  args: SendInteractiveListEngineArgs,
+): Promise<{ whatsapp_message_id: string }> {
+  return sendInteractiveViaMeta({ ...args, kind: 'list' })
+}
+
+type SendInput =
+  | (SendInteractiveButtonsEngineArgs & { kind: 'buttons' })
+  | (SendInteractiveListEngineArgs & { kind: 'list' })
+
+async function sendInteractiveViaMeta(
+  input: SendInput,
+): Promise<{ whatsapp_message_id: string }> {
+  const db = supabaseAdmin()
+
+  // Scope the contact lookup by user_id — same defense-in-depth
+  // rationale as automations/meta-send.ts. Service-role client
+  // bypasses RLS, so an attacker who could call into the engine
+  // with a contact_id from another tenant would otherwise send
+  // through their own WhatsApp config to a stranger's number.
+  const { data: contact, error: contactErr } = await db
+    .from('contacts')
+    .select('id, phone')
+    .eq('id', input.contactId)
+    .eq('user_id', input.userId)
+    .maybeSingle()
+  if (contactErr || !contact?.phone) {
+    throw new Error('contact not found for this user')
+  }
+
+  const sanitized = sanitizePhoneForMeta(contact.phone)
+  if (!isValidE164(sanitized)) {
+    throw new Error(`contact phone invalid: ${contact.phone}`)
+  }
+
+  const { data: config, error: configErr } = await db
+    .from('whatsapp_config')
+    .select('*')
+    .eq('user_id', input.userId)
+    .single()
+  if (configErr || !config) {
+    throw new Error('WhatsApp not configured for this account')
+  }
+
+  const accessToken = decrypt(config.access_token)
+
+  const attempt = async (phone: string): Promise<string> => {
+    if (input.kind === 'buttons') {
+      const r = await sendInteractiveButtons({
+        phoneNumberId: config.phone_number_id,
+        accessToken,
+        to: phone,
+        bodyText: input.bodyText,
+        buttons: input.buttons,
+        headerText: input.headerText,
+        footerText: input.footerText,
+      })
+      return r.messageId
+    }
+    const r = await sendInteractiveList({
+      phoneNumberId: config.phone_number_id,
+      accessToken,
+      to: phone,
+      bodyText: input.bodyText,
+      buttonLabel: input.buttonLabel,
+      sections: input.sections,
+      headerText: input.headerText,
+      footerText: input.footerText,
+    })
+    return r.messageId
+  }
+
+  // Same phone-variant retry as automations/meta-send.ts. Numbers
+  // registered with/without a trunk 0 + Meta's sandbox quirks all
+  // need this to reliably land a message.
+  const variants = phoneVariants(sanitized)
+  let workingPhone = sanitized
+  let waMessageId = ''
+  let lastError: unknown = null
+  for (const v of variants) {
+    try {
+      waMessageId = await attempt(v)
+      workingPhone = v
+      lastError = null
+      break
+    } catch (err) {
+      const msg = err instanceof Error ? err.message : String(err)
+      if (!isRecipientNotAllowedError(msg)) throw err
+      lastError = err
+    }
+  }
+  if (lastError) throw lastError
+
+  if (workingPhone !== sanitized) {
+    await db.from('contacts').update({ phone: workingPhone }).eq('id', contact.id)
+  }
+
+  // Persist the bot's prompt to the messages table so it appears in
+  // the inbox. content_type='interactive' is supported as of
+  // migration 010; sender_type='bot' distinguishes flow sends from
+  // manual agent sends (the conversation list preview will pick up
+  // last_message_text as a sensible summary).
+  //
+  // We do NOT set interactive_reply_id here — that column is reserved
+  // for the customer's tap on this message, populated by the webhook
+  // when their reply arrives.
+  const { error: msgErr } = await db.from('messages').insert({
+    conversation_id: input.conversationId,
+    sender_type: 'bot',
+    content_type: 'interactive',
+    content_text: input.bodyText,
+    message_id: waMessageId,
+    status: 'sent',
+  })
+  if (msgErr) {
+    throw new Error(`sent to Meta but DB insert failed: ${msgErr.message}`)
+  }
+
+  await db
+    .from('conversations')
+    .update({
+      last_message_text: input.bodyText,
+      last_message_at: new Date().toISOString(),
+      updated_at: new Date().toISOString(),
+    })
+    .eq('id', input.conversationId)
+
+  return { whatsapp_message_id: waMessageId }
+}

src/lib/whatsapp/meta-api.test.ts

@@ -0,0 +1,269 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  INTERACTIVE_LIMITS,
+  sendInteractiveButtons,
+  sendInteractiveList,
+} from "./meta-api";
+
+// All assertions in this file run BEFORE the network call. We stub fetch
+// to a never-resolving mock so a test that accidentally falls through to
+// the request body would hang (and fail) rather than silently hit
+// graph.facebook.com.
+const neverFetch = () =>
+  new Promise<Response>(() => {
+    /* intentionally never resolves */
+  });
+
+const BASE_ARGS = {
+  phoneNumberId: "test-phone",
+  accessToken: "test-token",
+  to: "1234567890",
+  bodyText: "Body text",
+} as const;
+
+describe("sendInteractiveButtons — validation", () => {
+  beforeEach(() => {
+    vi.stubGlobal("fetch", vi.fn(neverFetch));
+  });
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  it("rejects an empty buttons array", async () => {
+    await expect(
+      sendInteractiveButtons({ ...BASE_ARGS, buttons: [] }),
+    ).rejects.toThrow(/1-3 buttons/);
+  });
+
+  it(`rejects more than ${INTERACTIVE_LIMITS.maxButtons} buttons (Meta cap)`, async () => {
+    await expect(
+      sendInteractiveButtons({
+        ...BASE_ARGS,
+        buttons: [
+          { id: "a", title: "A" },
+          { id: "b", title: "B" },
+          { id: "c", title: "C" },
+          { id: "d", title: "D" },
+        ],
+      }),
+    ).rejects.toThrow(/1-3 buttons/);
+  });
+
+  it("rejects a button title longer than 20 chars (Meta cap)", async () => {
+    await expect(
+      sendInteractiveButtons({
+        ...BASE_ARGS,
+        buttons: [
+          { id: "a", title: "x".repeat(INTERACTIVE_LIMITS.buttonTitleMaxLength + 1) },
+        ],
+      }),
+    ).rejects.toThrow(/exceeds 20 chars/);
+  });
+
+  it("rejects a button missing its id", async () => {
+    await expect(
+      sendInteractiveButtons({
+        ...BASE_ARGS,
+        buttons: [{ id: "", title: "Choose me" }],
+      }),
+    ).rejects.toThrow(/missing id/);
+  });
+
+  it("rejects an empty body text", async () => {
+    await expect(
+      sendInteractiveButtons({
+        ...BASE_ARGS,
+        bodyText: "",
+        buttons: [{ id: "a", title: "A" }],
+      }),
+    ).rejects.toThrow(/requires bodyText/);
+  });
+
+  it("rejects a header text over the limit", async () => {
+    await expect(
+      sendInteractiveButtons({
+        ...BASE_ARGS,
+        headerText: "x".repeat(INTERACTIVE_LIMITS.headerTextMaxLength + 1),
+        buttons: [{ id: "a", title: "A" }],
+      }),
+    ).rejects.toThrow(/headerText exceeds/);
+  });
+
+  it("sends the right payload shape when all inputs are valid", async () => {
+    let captured: { url: string; body: unknown; method: string } | null = null;
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async (url: string, init: RequestInit) => {
+        captured = {
+          url,
+          method: init.method ?? "GET",
+          body: JSON.parse(String(init.body)),
+        };
+        return new Response(
+          JSON.stringify({ messages: [{ id: "wamid.PASS" }] }),
+          { status: 200 },
+        );
+      }),
+    );
+
+    const result = await sendInteractiveButtons({
+      ...BASE_ARGS,
+      headerText: "Hello",
+      footerText: "Tap one",
+      buttons: [
+        { id: "yes", title: "Yes" },
+        { id: "no", title: "No" },
+      ],
+    });
+
+    expect(result).toEqual({ messageId: "wamid.PASS" });
+    expect(captured).not.toBeNull();
+    expect(captured!.method).toBe("POST");
+    expect(captured!.url).toContain("test-phone/messages");
+    expect(captured!.body).toMatchObject({
+      messaging_product: "whatsapp",
+      recipient_type: "individual",
+      to: "1234567890",
+      type: "interactive",
+      interactive: {
+        type: "button",
+        body: { text: "Body text" },
+        header: { type: "text", text: "Hello" },
+        footer: { text: "Tap one" },
+        action: {
+          buttons: [
+            { type: "reply", reply: { id: "yes", title: "Yes" } },
+            { type: "reply", reply: { id: "no", title: "No" } },
+          ],
+        },
+      },
+    });
+  });
+});
+
+describe("sendInteractiveList — validation", () => {
+  beforeEach(() => {
+    vi.stubGlobal("fetch", vi.fn(neverFetch));
+  });
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  const ROW = { id: "r1", title: "Row 1" };
+
+  it("rejects zero sections", async () => {
+    await expect(
+      sendInteractiveList({
+        ...BASE_ARGS,
+        buttonLabel: "Open",
+        sections: [],
+      }),
+    ).rejects.toThrow(/1-10 sections/);
+  });
+
+  it(`rejects more than ${INTERACTIVE_LIMITS.maxListRowsTotal} rows total across sections (Meta cap)`, async () => {
+    const rows = Array.from({ length: 11 }, (_, i) => ({
+      id: `r${i}`,
+      title: `Row ${i}`,
+    }));
+    await expect(
+      sendInteractiveList({
+        ...BASE_ARGS,
+        buttonLabel: "Open",
+        sections: [{ rows }],
+      }),
+    ).rejects.toThrow(/1-10 rows total/);
+  });
+
+  it("rejects a row title longer than 24 chars (Meta cap)", async () => {
+    await expect(
+      sendInteractiveList({
+        ...BASE_ARGS,
+        buttonLabel: "Open",
+        sections: [
+          {
+            rows: [
+              {
+                id: "r1",
+                title: "x".repeat(INTERACTIVE_LIMITS.listRowTitleMaxLength + 1),
+              },
+            ],
+          },
+        ],
+      }),
+    ).rejects.toThrow(/exceeds 24 chars/);
+  });
+
+  it("rejects duplicate row ids across sections", async () => {
+    await expect(
+      sendInteractiveList({
+        ...BASE_ARGS,
+        buttonLabel: "Open",
+        sections: [
+          { rows: [{ id: "dupe", title: "First" }] },
+          { rows: [{ id: "dupe", title: "Second" }] },
+        ],
+      }),
+    ).rejects.toThrow(/duplicate row id/);
+  });
+
+  it("rejects an empty buttonLabel", async () => {
+    await expect(
+      sendInteractiveList({
+        ...BASE_ARGS,
+        buttonLabel: "",
+        sections: [{ rows: [ROW] }],
+      }),
+    ).rejects.toThrow(/requires a buttonLabel/);
+  });
+
+  it("sends the right payload shape when valid", async () => {
+    let captured: { body: unknown } | null = null;
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async (_url: string, init: RequestInit) => {
+        captured = { body: JSON.parse(String(init.body)) };
+        return new Response(
+          JSON.stringify({ messages: [{ id: "wamid.LIST" }] }),
+          { status: 200 },
+        );
+      }),
+    );
+
+    const result = await sendInteractiveList({
+      ...BASE_ARGS,
+      buttonLabel: "Open menu",
+      sections: [
+        {
+          title: "Orders",
+          rows: [
+            { id: "order_1", title: "Order #1", description: "€12" },
+            { id: "order_2", title: "Order #2" },
+          ],
+        },
+      ],
+    });
+
+    expect(result).toEqual({ messageId: "wamid.LIST" });
+    expect(captured).not.toBeNull();
+    expect(captured!.body).toMatchObject({
+      type: "interactive",
+      interactive: {
+        type: "list",
+        body: { text: "Body text" },
+        action: {
+          button: "Open menu",
+          sections: [
+            {
+              title: "Orders",
+              rows: [
+                { id: "order_1", title: "Order #1", description: "€12" },
+                { id: "order_2", title: "Order #2" },
+              ],
+            },
+          ],
+        },
+      },
+    });
+  });
+});

src/lib/whatsapp/meta-api.ts

@@ -226,6 +226,285 @@ export async function sendReactionMessage(
   return { messageId: data.messages[0].id }
 }
 
+// ============================================================
+// Interactive (button replies + list messages)
+// ============================================================
+//
+// Meta's two flavours of interactive message — used by the Flows
+// engine to drive scripted chatbot menus. Caller passes plain
+// JS values; helpers shape the Meta payload and enforce Meta's
+// limits BEFORE the network call so the failure mode is a
+// developer-facing error rather than a customer-facing one.
+
+/**
+ * Meta limits for interactive messages, hard-coded so violations
+ * fail at build/save time rather than as a 400 from the Meta API
+ * mid-conversation. See:
+ *   https://developers.facebook.com/docs/whatsapp/cloud-api/messages/interactive-reply-buttons-messages
+ *   https://developers.facebook.com/docs/whatsapp/cloud-api/messages/interactive-list-messages
+ */
+export const INTERACTIVE_LIMITS = {
+  maxButtons: 3,
+  buttonTitleMaxLength: 20,
+  maxListSections: 10,
+  maxListRowsTotal: 10,
+  listRowTitleMaxLength: 24,
+  listRowDescriptionMaxLength: 72,
+  bodyMaxLength: 1024,
+  footerMaxLength: 60,
+  headerTextMaxLength: 60,
+} as const
+
+export interface InteractiveButton {
+  /** Stable id sent back in the webhook when tapped (≤ 256 chars). */
+  id: string
+  /** Visible label (≤ 20 chars per Meta). */
+  title: string
+}
+
+export interface SendInteractiveButtonsArgs {
+  phoneNumberId: string
+  accessToken: string
+  to: string
+  /** The body text — what the customer reads above the buttons. */
+  bodyText: string
+  /** Optional plain-text header (≤ 60 chars). */
+  headerText?: string
+  /** Optional grey footer line under the buttons (≤ 60 chars). */
+  footerText?: string
+  /** 1–3 buttons. Validated against Meta's limits before sending. */
+  buttons: InteractiveButton[]
+  /** Meta's message_id of the message being replied to (quote preview). */
+  contextMessageId?: string
+}
+
+/**
+ * Send an interactive message with up to 3 inline reply buttons. The
+ * customer taps one and Meta delivers a webhook with
+ * `messages[0].interactive.button_reply.id` set to the matching button.id.
+ *
+ * Validation throws BEFORE the network call so misconfigured flows
+ * fail at save time, not during a live conversation.
+ */
+export async function sendInteractiveButtons(
+  args: SendInteractiveButtonsArgs
+): Promise<MetaSendResult> {
+  const {
+    phoneNumberId, accessToken, to,
+    bodyText, headerText, footerText, buttons, contextMessageId,
+  } = args
+  validateInteractiveBody(bodyText)
+  validateInteractiveHeaderFooter(headerText, footerText)
+  if (buttons.length < 1 || buttons.length > INTERACTIVE_LIMITS.maxButtons) {
+    throw new Error(
+      `Interactive button message requires 1-${INTERACTIVE_LIMITS.maxButtons} buttons (got ${buttons.length}).`
+    )
+  }
+  for (const btn of buttons) {
+    if (!btn.id) throw new Error('Interactive button missing id.')
+    if (!btn.title) throw new Error(`Interactive button "${btn.id}" missing title.`)
+    if (btn.title.length > INTERACTIVE_LIMITS.buttonTitleMaxLength) {
+      throw new Error(
+        `Interactive button title "${btn.title}" exceeds ${INTERACTIVE_LIMITS.buttonTitleMaxLength} chars.`
+      )
+    }
+  }
+
+  const interactive: Record<string, unknown> = {
+    type: 'button',
+    body: { text: bodyText },
+    action: {
+      buttons: buttons.map((b) => ({
+        type: 'reply',
+        reply: { id: b.id, title: b.title },
+      })),
+    },
+  }
+  if (headerText) interactive.header = { type: 'text', text: headerText }
+  if (footerText) interactive.footer = { text: footerText }
+
+  const body: Record<string, unknown> = {
+    messaging_product: 'whatsapp',
+    recipient_type: 'individual',
+    to,
+    type: 'interactive',
+    interactive,
+  }
+  if (contextMessageId) body.context = { message_id: contextMessageId }
+
+  const url = `${META_API_BASE}/${phoneNumberId}/messages`
+  const response = await fetch(url, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      Authorization: `Bearer ${accessToken}`,
+    },
+    body: JSON.stringify(body),
+  })
+  if (!response.ok) {
+    await throwMetaError(response, `Meta API error: ${response.status}`)
+  }
+  const data = await response.json()
+  return { messageId: data.messages[0].id }
+}
+
+export interface InteractiveListRow {
+  /** Stable id sent back in the webhook when tapped (≤ 200 chars). */
+  id: string
+  /** Visible row title (≤ 24 chars per Meta). */
+  title: string
+  /** Optional secondary line shown under the title (≤ 72 chars). */
+  description?: string
+}
+
+export interface InteractiveListSection {
+  /** Optional section header shown above its rows. */
+  title?: string
+  rows: InteractiveListRow[]
+}
+
+export interface SendInteractiveListArgs {
+  phoneNumberId: string
+  accessToken: string
+  to: string
+  bodyText: string
+  /** Label of the tap-to-expand button on the message bubble. */
+  buttonLabel: string
+  headerText?: string
+  footerText?: string
+  /**
+   * 1–10 rows TOTAL across all sections. Meta caps the *total*, not
+   * per-section. Validation enforces this before send.
+   */
+  sections: InteractiveListSection[]
+  contextMessageId?: string
+}
+
+/**
+ * Send an interactive message with a tap-to-expand list of selectable
+ * rows. Use when there are more options than the 3-button limit allows.
+ * Webhook arrives with `messages[0].interactive.list_reply.id` set to
+ * the matching row.id.
+ */
+export async function sendInteractiveList(
+  args: SendInteractiveListArgs
+): Promise<MetaSendResult> {
+  const {
+    phoneNumberId, accessToken, to,
+    bodyText, buttonLabel, headerText, footerText, sections, contextMessageId,
+  } = args
+  validateInteractiveBody(bodyText)
+  validateInteractiveHeaderFooter(headerText, footerText)
+  if (!buttonLabel) throw new Error('Interactive list requires a buttonLabel.')
+  if (buttonLabel.length > INTERACTIVE_LIMITS.buttonTitleMaxLength) {
+    throw new Error(
+      `Interactive list buttonLabel "${buttonLabel}" exceeds ${INTERACTIVE_LIMITS.buttonTitleMaxLength} chars.`
+    )
+  }
+  if (sections.length < 1 || sections.length > INTERACTIVE_LIMITS.maxListSections) {
+    throw new Error(
+      `Interactive list requires 1-${INTERACTIVE_LIMITS.maxListSections} sections (got ${sections.length}).`
+    )
+  }
+  const totalRows = sections.reduce((sum, s) => sum + s.rows.length, 0)
+  if (totalRows < 1 || totalRows > INTERACTIVE_LIMITS.maxListRowsTotal) {
+    throw new Error(
+      `Interactive list requires 1-${INTERACTIVE_LIMITS.maxListRowsTotal} rows total across all sections (got ${totalRows}).`
+    )
+  }
+  const seenIds = new Set<string>()
+  for (const section of sections) {
+    for (const row of section.rows) {
+      if (!row.id) throw new Error('Interactive list row missing id.')
+      if (seenIds.has(row.id)) {
+        throw new Error(`Interactive list has duplicate row id "${row.id}".`)
+      }
+      seenIds.add(row.id)
+      if (!row.title) throw new Error(`Interactive list row "${row.id}" missing title.`)
+      if (row.title.length > INTERACTIVE_LIMITS.listRowTitleMaxLength) {
+        throw new Error(
+          `Interactive list row title "${row.title}" exceeds ${INTERACTIVE_LIMITS.listRowTitleMaxLength} chars.`
+        )
+      }
+      if (
+        row.description &&
+        row.description.length > INTERACTIVE_LIMITS.listRowDescriptionMaxLength
+      ) {
+        throw new Error(
+          `Interactive list row description for "${row.id}" exceeds ${INTERACTIVE_LIMITS.listRowDescriptionMaxLength} chars.`
+        )
+      }
+    }
+  }
+
+  const interactive: Record<string, unknown> = {
+    type: 'list',
+    body: { text: bodyText },
+    action: {
+      button: buttonLabel,
+      sections: sections.map((s) => ({
+        ...(s.title ? { title: s.title } : {}),
+        rows: s.rows.map((r) => ({
+          id: r.id,
+          title: r.title,
+          ...(r.description ? { description: r.description } : {}),
+        })),
+      })),
+    },
+  }
+  if (headerText) interactive.header = { type: 'text', text: headerText }
+  if (footerText) interactive.footer = { text: footerText }
+
+  const body: Record<string, unknown> = {
+    messaging_product: 'whatsapp',
+    recipient_type: 'individual',
+    to,
+    type: 'interactive',
+    interactive,
+  }
+  if (contextMessageId) body.context = { message_id: contextMessageId }
+
+  const url = `${META_API_BASE}/${phoneNumberId}/messages`
+  const response = await fetch(url, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      Authorization: `Bearer ${accessToken}`,
+    },
+    body: JSON.stringify(body),
+  })
+  if (!response.ok) {
+    await throwMetaError(response, `Meta API error: ${response.status}`)
+  }
+  const data = await response.json()
+  return { messageId: data.messages[0].id }
+}
+
+function validateInteractiveBody(bodyText: string): void {
+  if (!bodyText) throw new Error('Interactive message requires bodyText.')
+  if (bodyText.length > INTERACTIVE_LIMITS.bodyMaxLength) {
+    throw new Error(
+      `Interactive bodyText exceeds ${INTERACTIVE_LIMITS.bodyMaxLength} chars.`
+    )
+  }
+}
+
+function validateInteractiveHeaderFooter(
+  headerText: string | undefined,
+  footerText: string | undefined,
+): void {
+  if (headerText && headerText.length > INTERACTIVE_LIMITS.headerTextMaxLength) {
+    throw new Error(
+      `Interactive headerText exceeds ${INTERACTIVE_LIMITS.headerTextMaxLength} chars.`
+    )
+  }
+  if (footerText && footerText.length > INTERACTIVE_LIMITS.footerMaxLength) {
+    throw new Error(
+      `Interactive footerText exceeds ${INTERACTIVE_LIMITS.footerMaxLength} chars.`
+    )
+  }
+}
+
 // ============================================================
 // Media
 // ============================================================

src/types/index.ts

@@ -82,7 +82,16 @@ export interface Conversation {
 }
 
 export type SenderType = 'customer' | 'agent' | 'bot';
-export type ContentType = 'text' | 'image' | 'document' | 'audio' | 'video' | 'location' | 'template';
+export type ContentType =
+  | 'text'
+  | 'image'
+  | 'document'
+  | 'audio'
+  | 'video'
+  | 'location'
+  | 'template'
+  /** Customer tapped a reply button or list row on a message we sent. */
+  | 'interactive';
 export type MessageStatus = 'sending' | 'sent' | 'delivered' | 'read' | 'failed';
 
 export interface Message {
@@ -98,6 +107,13 @@ export interface Message {
   status: MessageStatus;
   created_at: string;
   reply_to_message_id?: string;
+  /**
+   * Only set when `content_type === 'interactive'` — the stable id of
+   * the button or list row the customer tapped. The Flows engine uses
+   * this to route the next node; the inbox bubble uses it as a styling
+   * cue (renders with a "↩ button reply" affordance).
+   */
+  interactive_reply_id?: string;
 }
 
 export type ReactionActor = 'customer' | 'agent';

supabase/migrations/010_flows.sql

@@ -0,0 +1,280 @@
+-- ============================================================
+-- Conversational Flows: stateful, branching WhatsApp chatbot.
+--
+-- What this migration adds:
+--
+--   1. `flows` — the definition envelope (name, trigger config,
+--      entry node, fallback policy, status). One row per authored bot.
+--
+--   2. `flow_nodes` — the graph rows. Edges live INSIDE each node's
+--      `config` JSONB (e.g. each button row carries its own
+--      `next_node_key`). Why edges-in-config rather than a separate
+--      `flow_edges` table:
+--        - The runner only ever asks "given current node X, where does
+--          reply Y go?" — that's a single-row lookup with the JSON
+--          already on the row. Splitting edges out forces a join per
+--          inbound message.
+--        - The builder's natural unit of edit is the node ("change this
+--          button's label and target"); a side table would force
+--          coordinated inserts/deletes on every save.
+--      Cross-node integrity is enforced at save-time by the validator
+--      (mirrors what `automation_steps`/`validate.ts` already does).
+--
+--      `node_key` is a STABLE STRING (e.g. "menu_existing"), not the
+--      UUID. Edge targets reference node_key, which means:
+--        - Cloning a flow doesn't require UUID rewriting in JSON edges.
+--        - Templates ship with human-readable keys.
+--        - Direct DB inspection is debuggable.
+--      The (flow_id, node_key) UNIQUE constraint guarantees lookup
+--      determinism.
+--
+--   3. `flow_runs` — per-contact runtime state machine. The linchpin
+--      is the partial unique index `idx_one_active_run_per_contact`:
+--      at most one ACTIVE run per (user_id, contact_id). Two concurrent
+--      webhook deliveries trying to start a run both attempt INSERT;
+--      the second fails with 23505 and the runner catches & exits.
+--      No locking required.
+--
+--   4. `flow_run_events` — append-only audit. Used by the runner for
+--      idempotency (refuses to advance twice on the same Meta
+--      message_id) and by the future run-history viewer.
+--
+--   5. Widens `messages.content_type` CHECK to allow 'interactive', and
+--      adds `messages.interactive_reply_id`. With this, button/list
+--      taps become first-class message rows with a queryable reply id
+--      instead of getting silently coerced into the "Unsupported
+--      message type" fallback in parseMessageContent.
+--
+-- Idempotent — safe to run multiple times.
+-- ============================================================
+
+-- ============================================================
+-- 1. Messages table — widen content_type, add interactive_reply_id
+-- ============================================================
+
+-- Drop & re-add the CHECK constraint to add 'interactive' as an allowed
+-- value. Migration 001 named it `messages_content_type_check` (Postgres
+-- default for an inline CHECK on a TEXT column).
+ALTER TABLE messages
+  DROP CONSTRAINT IF EXISTS messages_content_type_check;
+
+ALTER TABLE messages
+  ADD CONSTRAINT messages_content_type_check
+  CHECK (content_type IN (
+    'text', 'image', 'document', 'audio', 'video',
+    'location', 'template', 'interactive'
+  ));
+
+-- Reply id of the button / list row the customer tapped. NULL for
+-- everything that isn't an interactive reply. No FK — Meta button ids
+-- are arbitrary user-chosen strings, not row references.
+ALTER TABLE messages
+  ADD COLUMN IF NOT EXISTS interactive_reply_id TEXT;
+
+-- ============================================================
+-- 2. flows
+-- ============================================================
+CREATE TABLE IF NOT EXISTS flows (
+  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+  user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
+  name TEXT NOT NULL,
+  description TEXT,
+  status TEXT NOT NULL DEFAULT 'draft'
+    CHECK (status IN ('draft', 'active', 'archived')),
+  trigger_type TEXT NOT NULL
+    CHECK (trigger_type IN ('keyword', 'first_inbound_message', 'manual')),
+  trigger_config JSONB NOT NULL DEFAULT '{}'::jsonb,
+  -- References `flow_nodes.node_key` (a string, not the UUID). NULL
+  -- while the flow is being authored; required before activation
+  -- (enforced by the validator, not at the DB level so drafts can save).
+  entry_node_id TEXT,
+  fallback_policy JSONB NOT NULL DEFAULT
+    '{"on_unknown_reply":"reprompt","max_reprompts":2,"on_timeout_hours":24,"on_exhaust":"handoff"}'::jsonb,
+  execution_count INTEGER NOT NULL DEFAULT 0,
+  last_executed_at TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Active-only lookups dominate the runner's hot path. Partial index
+-- keeps it small even when archived flows accumulate.
+CREATE INDEX IF NOT EXISTS idx_flows_active_trigger
+  ON flows(user_id, trigger_type)
+  WHERE status = 'active';
+
+ALTER TABLE flows ENABLE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS "Users can manage own flows" ON flows;
+CREATE POLICY "Users can manage own flows" ON flows FOR ALL
+  USING (auth.uid() = user_id);
+
+-- ============================================================
+-- 3. flow_nodes
+-- ============================================================
+CREATE TABLE IF NOT EXISTS flow_nodes (
+  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+  flow_id UUID NOT NULL REFERENCES flows(id) ON DELETE CASCADE,
+  node_key TEXT NOT NULL,
+  node_type TEXT NOT NULL CHECK (node_type IN (
+    'start',
+    'send_buttons',
+    'send_list',
+    'send_message',
+    'collect_input',
+    'condition',
+    'set_tag',
+    'handoff',
+    'http_fetch',
+    'end'
+  )),
+  config JSONB NOT NULL DEFAULT '{}'::jsonb,
+  -- Reserved for the v2 react-flow canvas. v1 list editor leaves both
+  -- at 0; carrying the columns now avoids a follow-up migration when
+  -- the canvas ships.
+  position_x INTEGER NOT NULL DEFAULT 0,
+  position_y INTEGER NOT NULL DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  UNIQUE (flow_id, node_key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_flow_nodes_flow
+  ON flow_nodes(flow_id);
+
+ALTER TABLE flow_nodes ENABLE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS "Users manage nodes on their flows" ON flow_nodes;
+CREATE POLICY "Users manage nodes on their flows" ON flow_nodes FOR ALL
+  USING (EXISTS (
+    SELECT 1 FROM flows f
+    WHERE f.id = flow_nodes.flow_id
+      AND f.user_id = auth.uid()
+  ));
+
+-- ============================================================
+-- 4. flow_runs
+-- ============================================================
+CREATE TABLE IF NOT EXISTS flow_runs (
+  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+  flow_id UUID NOT NULL REFERENCES flows(id) ON DELETE CASCADE,
+  user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
+  -- contact_id intentionally SET NULL on delete (matches the
+  -- automation_logs / broadcast_recipients pattern in migration 004):
+  -- deleting a contact must not erase the historical audit trail.
+  contact_id UUID REFERENCES contacts(id) ON DELETE SET NULL,
+  conversation_id UUID REFERENCES conversations(id) ON DELETE SET NULL,
+  status TEXT NOT NULL DEFAULT 'active' CHECK (status IN (
+    'active',           -- currently awaiting customer input
+    'completed',        -- reached an end node naturally
+    'handed_off',       -- ended via a handoff node
+    'timed_out',        -- swept by the cron after fallback_policy.on_timeout_hours
+    'paused_by_agent',  -- an agent manually replied; flow yielded
+    'failed'            -- runner hit an unrecoverable error
+  )),
+  current_node_key TEXT,
+  last_prompt_message_id UUID REFERENCES messages(id) ON DELETE SET NULL,
+  -- Captured collect_input values + http_fetch responses. Interpolated
+  -- into downstream node configs at advance time.
+  vars JSONB NOT NULL DEFAULT '{}'::jsonb,
+  reprompt_count INTEGER NOT NULL DEFAULT 0,
+  started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  last_advanced_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  ended_at TIMESTAMPTZ,
+  end_reason TEXT
+);
+
+-- Linchpin of idempotency / concurrency safety. At most one active run
+-- per (user_id, contact_id). Two concurrent webhook deliveries each
+-- trying to start a run will collide on this index; the second INSERT
+-- fails with 23505 and the runner catches & returns consumed:true.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_one_active_run_per_contact
+  ON flow_runs(user_id, contact_id)
+  WHERE status = 'active';
+
+-- Cron sweep query: "find active runs older than X hours" needs to be
+-- index-supported so the sweeper stays cheap as flow volume grows.
+CREATE INDEX IF NOT EXISTS idx_flow_runs_active_advanced
+  ON flow_runs(last_advanced_at)
+  WHERE status = 'active';
+
+-- Detail / history page queries: "list runs for this flow, newest first".
+CREATE INDEX IF NOT EXISTS idx_flow_runs_flow_started
+  ON flow_runs(flow_id, started_at DESC);
+
+ALTER TABLE flow_runs ENABLE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS "Users see own flow runs" ON flow_runs;
+CREATE POLICY "Users see own flow runs" ON flow_runs FOR SELECT
+  USING (auth.uid() = user_id);
+
+-- The runner uses service_role for all writes; users never INSERT /
+-- UPDATE / DELETE flow_runs from the client. Omitting those policies
+-- keeps the surface tight (mirrors automation_pending_executions).
+
+-- ============================================================
+-- 5. flow_run_events
+-- ============================================================
+CREATE TABLE IF NOT EXISTS flow_run_events (
+  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+  flow_run_id UUID NOT NULL REFERENCES flow_runs(id) ON DELETE CASCADE,
+  event_type TEXT NOT NULL CHECK (event_type IN (
+    'started',
+    'node_entered',
+    'message_sent',
+    'reply_received',
+    'fallback_fired',
+    'handoff',
+    'timeout',
+    'error',
+    'completed'
+  )),
+  node_key TEXT,
+  payload JSONB NOT NULL DEFAULT '{}'::jsonb,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Idempotency check in the runner needs fast lookup by
+-- (flow_run_id, event_type, payload->>'meta_message_id'). The runner
+-- does the JSONB extraction client-side; index just needs the first
+-- two columns to narrow.
+CREATE INDEX IF NOT EXISTS idx_flow_run_events_run_type
+  ON flow_run_events(flow_run_id, event_type);
+
+-- History viewer: reverse-chronological scan per run.
+CREATE INDEX IF NOT EXISTS idx_flow_run_events_run_time
+  ON flow_run_events(flow_run_id, created_at DESC);
+
+ALTER TABLE flow_run_events ENABLE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS "Users see events on their runs" ON flow_run_events;
+CREATE POLICY "Users see events on their runs" ON flow_run_events FOR SELECT
+  USING (EXISTS (
+    SELECT 1 FROM flow_runs r
+    WHERE r.id = flow_run_events.flow_run_id
+      AND r.user_id = auth.uid()
+  ));
+
+-- ============================================================
+-- 6. updated_at trigger on flows
+-- ============================================================
+-- Reuses update_updated_at_column() from migration 001. Trigger name
+-- matches the convention used on every other table that has one
+-- (see migration 001 lines 361-367).
+DROP TRIGGER IF EXISTS set_updated_at ON flows;
+CREATE TRIGGER set_updated_at BEFORE UPDATE ON flows
+  FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
+
+-- ============================================================
+-- 7. Realtime publication
+-- ============================================================
+-- Add flow_runs so the inbox can render "this contact is in flow X at
+-- node Y" live as the runner advances. Other flow tables don't need
+-- realtime — the builder reads on demand, the runner is server-side.
+DO $$
+BEGIN
+  IF NOT EXISTS (
+    SELECT 1 FROM pg_publication_tables
+    WHERE pubname = 'supabase_realtime' AND tablename = 'flow_runs'
+  ) THEN
+    ALTER PUBLICATION supabase_realtime ADD TABLE flow_runs;
+  END IF;
+END $$;