feat(flows): runner engine + webhook integration + agent-handoff + cron sweep (PR 2/4)

Brings the Flows feature alive end-to-end for beta-opted-in accounts.
Still no UI — flows are defined via direct SQL inserts in v1; the
no-code builder lands in PR 3. Everything in this PR is gated behind
`isFlowsEnabled(profile)` so non-beta accounts behave identically to
before.

What landed:

  1. `src/lib/flows/types.ts` — DB row + discriminated-union node
     config types, plus the `ParsedInbound` shape and a
     `DEFAULT_FALLBACK_POLICY`.

  2. `src/lib/flows/fallback.ts` — pure policy resolver. Decides
     whether an unknown reply triggers a reprompt (up to N times),
     a handoff, or end. 11 unit tests.

  3. `src/lib/flows/engine.ts` — `dispatchInboundToFlows()` entry
     point. Loads the active run (if any), checks idempotency via
     `flow_run_events`, advances the run through a synchronous
     auto-advance loop that suspends only at nodes that need
     customer input. Each suspending node calls
     `engineSendInteractive{Buttons,List}` and persists the new
     current_node_key via an optimistic UPDATE with a precondition
     on the OLD key — concurrent inbounds collide cleanly.
     11 unit tests for the pure-logic helpers.

  4. `src/app/api/whatsapp/webhook/route.ts` — invokes the runner
     between the broadcast-reply flag and the automation dispatch.
     When the runner consumes the inbound (started a flow, advanced
     one, applied a fallback), suppresses `new_message_received`
     and `keyword_match` automation triggers — the customer is
     navigating the bot menu, not throwing a keyword. Relationship
     triggers (new_contact_created, first_inbound_message) still
     fire regardless.

  5. `src/app/api/whatsapp/send/route.ts` — after a successful
     agent message, marks any active flow run for that contact as
     `paused_by_agent`. Best-effort: a write failure is logged but
     doesn't block the send response.

  6. `src/app/api/flows/cron/route.ts` — sweeps active runs whose
     `last_advanced_at` is older than the flow's
     `fallback_policy.on_timeout_hours` (default 24h). Marks them
     `timed_out` and writes a `flow_run_events.timeout` row. Re-uses
     `AUTOMATION_CRON_SECRET` so operators don't have to provision
     a second secret.

Test surface in this PR:
  - 142/142 tests pass (added 22 new unit tests across fallback +
    engine pure logic)
  - typecheck clean
  - 0 new lint errors
  - build clean

Manual end-to-end verification arrives via the SQL fixture in the
PR description (a 3-node Welcome → buttons → handoff flow).

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

src/app/api/flows/cron/route.ts

@@ -0,0 +1,102 @@
+import { NextResponse } from 'next/server'
+import { supabaseAdmin } from '@/lib/flows/admin-client'
+import { resolveFallbackPolicy } from '@/lib/flows/fallback'
+
+/**
+ * Sweep abandoned active flow runs.
+ *
+ * Reads each active run's parent-flow `fallback_policy.on_timeout_hours`
+ * to compute the staleness cutoff (default 24h), then marks any run
+ * past its cutoff as `timed_out`. Writes a matching `flow_run_events`
+ * row for the audit trail.
+ *
+ * Without this sweep, a customer who abandons a flow mid-conversation
+ * keeps a row in `idx_one_active_run_per_contact` (the partial unique
+ * index on `flow_runs WHERE status='active'`) forever — blocking any
+ * new triggers for them. The cron is therefore not optional.
+ *
+ * Auth: re-uses `AUTOMATION_CRON_SECRET` so operators only have one
+ * secret to provision. The two endpoints (`/api/automations/cron`
+ * and this one) are independent operations; we keep them on separate
+ * URLs so one failing doesn't block the other.
+ *
+ * Hosting: hit on a schedule (Vercel Cron / GitHub Actions / external
+ * pinger). A 5-minute interval is more than enough for a 24h timeout
+ * default; once per hour would also be acceptable for low-volume
+ * tenants.
+ */
+export async function GET(request: Request) {
+  const expected = process.env.AUTOMATION_CRON_SECRET
+  if (!expected) {
+    return NextResponse.json({ error: 'cron not configured' }, { status: 503 })
+  }
+  const supplied = request.headers.get('x-cron-secret')
+  if (supplied !== expected) {
+    return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+  }
+
+  const admin = supabaseAdmin()
+  const now = new Date()
+
+  // Pull all currently-active runs along with their parent flow's
+  // fallback_policy. Joined in one query — the small set of active
+  // runs per tenant keeps this cheap.
+  const { data: runs, error } = await admin
+    .from('flow_runs')
+    .select(
+      'id, flow_id, user_id, contact_id, last_advanced_at, flows ( fallback_policy )',
+    )
+    .eq('status', 'active')
+
+  if (error) {
+    console.error('[flows-cron] active-run scan failed:', error.message)
+    return NextResponse.json({ error: error.message }, { status: 500 })
+  }
+  if (!runs?.length) return NextResponse.json({ swept: 0 })
+
+  type Row = {
+    id: string
+    flow_id: string
+    user_id: string
+    contact_id: string | null
+    last_advanced_at: string
+    flows: { fallback_policy: unknown } | { fallback_policy: unknown }[] | null
+  }
+
+  let swept = 0
+  for (const r of runs as Row[]) {
+    const flowsField = Array.isArray(r.flows) ? r.flows[0] : r.flows
+    const policy = resolveFallbackPolicy(flowsField?.fallback_policy ?? null)
+    const lastAdvanced = new Date(r.last_advanced_at)
+    const ageHours = (now.getTime() - lastAdvanced.getTime()) / (1000 * 60 * 60)
+    if (ageHours < policy.on_timeout_hours) continue
+
+    // Mark timed_out — guarded by the precondition `status='active'`
+    // so concurrent advance from a late inbound doesn't overwrite a
+    // legitimate update.
+    const { data: updated } = await admin
+      .from('flow_runs')
+      .update({
+        status: 'timed_out',
+        ended_at: now.toISOString(),
+        end_reason: 'stale_sweep',
+      })
+      .eq('id', r.id)
+      .eq('status', 'active')
+      .select('id')
+
+    if (Array.isArray(updated) && updated.length > 0) {
+      await admin.from('flow_run_events').insert({
+        flow_run_id: r.id,
+        event_type: 'timeout',
+        payload: {
+          age_hours: Math.round(ageHours * 10) / 10,
+          policy_hours: policy.on_timeout_hours,
+        },
+      })
+      swept += 1
+    }
+  }
+
+  return NextResponse.json({ swept })
+}

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

@@ -2,6 +2,8 @@ import { NextResponse } from 'next/server'
 import { createClient } from '@/lib/supabase/server'
 import { sendTextMessage, sendTemplateMessage } from '@/lib/whatsapp/meta-api'
 import { decrypt, encrypt, isLegacyFormat } from '@/lib/whatsapp/encryption'
+import { supabaseAdmin } from '@/lib/flows/admin-client'
+import { isFlowsEnabled } from '@/lib/flows/feature-flag'
 import {
   sanitizePhoneForMeta,
   isValidE164,
@@ -282,6 +284,44 @@ export async function POST(request: Request) {
       })
       .eq('id', conversation_id)
 
+    // Pause any active Flow run for this contact — the agent stepping
+    // in is the strongest "yield, human is here" signal. Gated on the
+    // beta flag to avoid an extra DB write for non-beta accounts; for
+    // those, no run can exist anyway. See PR #2 plan for why we pause
+    // (not end): preserves diagnostic state + lets the agent or the
+    // 24h timeout sweep cleanly resolve the run later.
+    try {
+      const { data: profile } = await supabase
+        .from('profiles')
+        .select('beta_features')
+        .eq('user_id', user.id)
+        .maybeSingle()
+      if (isFlowsEnabled(profile as { beta_features?: string[] | null } | null)) {
+        const { error: pauseErr } = await supabaseAdmin()
+          .from('flow_runs')
+          .update({
+            status: 'paused_by_agent',
+            ended_at: new Date().toISOString(),
+            end_reason: 'agent_replied',
+          })
+          .eq('user_id', user.id)
+          .eq('contact_id', contact.id)
+          .eq('status', 'active')
+        if (pauseErr) {
+          // Best-effort — log + continue. The agent's message already
+          // landed at Meta; don't fail the response over a bookkeeping
+          // miss. Worst case: a stale active run gets caught by the
+          // stale-run cron sweep within 24h.
+          console.error('[flows] pause-on-agent-send failed:', pauseErr.message)
+        }
+      }
+    } catch (err) {
+      console.error(
+        '[flows] pause-on-agent-send threw:',
+        err instanceof Error ? err.message : err,
+      )
+    }
+
     return NextResponse.json({
       success: true,
       message_id: messageRecord.id,

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

@@ -5,6 +5,8 @@ import { getMediaUrl, downloadMedia } from '@/lib/whatsapp/meta-api'
 import { normalizePhone, phonesMatch } from '@/lib/whatsapp/phone-utils'
 import { verifyMetaWebhookSignature } from '@/lib/whatsapp/webhook-signature'
 import { runAutomationsForTrigger } from '@/lib/automations/engine'
+import { dispatchInboundToFlows } from '@/lib/flows/engine'
+import { isFlowsEnabled } from '@/lib/flows/feature-flag'
 
 // Lazy-initialized to avoid build-time crash when env vars are missing
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -574,6 +576,51 @@ async function processMessage(
   // trigger installed in migration 003).
   await flagBroadcastReplyIfAny(userId, contactRecord.id)
 
+  // ============================================================
+  // Flow runner dispatch.
+  //
+  // Gated on the per-account beta flag (see PR #113). For accounts
+  // that haven't opted in, the runner never executes and the webhook
+  // behaves identically to before — automations fire as today.
+  //
+  // For opted-in accounts: if the runner consumes the message (it
+  // either advanced an active run or started a new one), we suppress
+  // the `new_message_received` + `keyword_match` automation triggers
+  // for this inbound. Customer is navigating the bot menu, not
+  // sending a fresh trigger word that should fork into automations.
+  //
+  // The relationship-level triggers (`new_contact_created`,
+  // `first_inbound_message`) still fire even when consumed — those
+  // are about WHO is messaging, not what they said.
+  //
+  // Awaited (not fire-and-forget) because we need the `consumed`
+  // result before deciding whether to dispatch automations. The
+  // runner has its own try/catch and never throws.
+  // ============================================================
+  let flowConsumed = false
+  if (await isProfileFlowsEnabled(userId)) {
+    const result = await dispatchInboundToFlows({
+      userId,
+      contactId: contactRecord.id,
+      conversationId: conversation.id,
+      message:
+        interactiveReplyId
+          ? {
+              kind: 'interactive_reply',
+              reply_id: interactiveReplyId,
+              reply_title: contentText ?? '',
+              meta_message_id: message.id,
+            }
+          : {
+              kind: 'text',
+              text: contentText ?? message.text?.body ?? '',
+              meta_message_id: message.id,
+            },
+      isFirstInboundMessage,
+    })
+    flowConsumed = result.consumed
+  }
+
   // Fire any automations that react to this webhook event. All dispatches
   // run here (not earlier) so the contact, conversation, and inbound
   // message all exist before any step — including send_message — runs.
@@ -585,7 +632,12 @@ async function processMessage(
     | 'first_inbound_message'
     | 'new_message_received'
     | 'keyword_match'
-  )[] = ['new_message_received', 'keyword_match']
+  )[] = []
+  // Content-level triggers are suppressed when a flow consumed the
+  // message — see the comment block above.
+  if (!flowConsumed) {
+    automationTriggers.push('new_message_received', 'keyword_match')
+  }
   // new_contact_created fires only when the webhook just auto-created the
   // contact row. first_inbound_message fires whenever this is the contact's
   // first-ever customer-sent message — a superset that also catches
@@ -607,6 +659,36 @@ async function processMessage(
   }
 }
 
+/**
+ * Reads the calling user's `profiles.beta_features` and asks whether
+ * the Flows beta is enabled for them. Best-effort: a DB failure reads
+ * as "not enabled" so a transient outage can't accidentally turn the
+ * feature on for a non-beta account.
+ *
+ * Result not cached across webhooks — Meta webhooks are infrequent
+ * (≪ 1 RPS for a single tenant) and the query is a single indexed
+ * row lookup. Caching introduces staleness when the owner toggles
+ * their opt-in via SQL.
+ */
+async function isProfileFlowsEnabled(userId: string): Promise<boolean> {
+  try {
+    const { data } = await supabaseAdmin()
+      .from('profiles')
+      .select('beta_features')
+      .eq('user_id', userId)
+      .maybeSingle()
+    return isFlowsEnabled(
+      data as { beta_features?: string[] | null } | null,
+    )
+  } catch (err) {
+    console.error(
+      '[flows] isProfileFlowsEnabled lookup failed:',
+      err instanceof Error ? err.message : err,
+    )
+    return false
+  }
+}
+
 async function parseMessageContent(
   message: WhatsAppMessage,
   accessToken: string

src/lib/flows/engine.test.ts

@@ -0,0 +1,186 @@
+import { describe, it, expect } from "vitest";
+import {
+  matchReplyId,
+  matchesKeywordTrigger,
+  isAutoAdvancing,
+  isSuspending,
+  isTerminal,
+} from "./engine";
+
+describe("matchReplyId", () => {
+  it("returns null for nodes without options", () => {
+    expect(
+      matchReplyId({ node_type: "start", config: { next_node_key: "x" } }, "y"),
+    ).toBeNull();
+    expect(
+      matchReplyId({ node_type: "send_message", config: {} }, "y"),
+    ).toBeNull();
+    expect(matchReplyId({ node_type: "end", config: {} }, "y")).toBeNull();
+  });
+
+  it("matches the buttons array on a send_buttons node", () => {
+    const node = {
+      node_type: "send_buttons",
+      config: {
+        text: "Pick one",
+        buttons: [
+          { reply_id: "yes", title: "Yes", next_node_key: "confirmed" },
+          { reply_id: "no", title: "No", next_node_key: "declined" },
+        ],
+      },
+    };
+    expect(matchReplyId(node, "yes")).toBe("confirmed");
+    expect(matchReplyId(node, "no")).toBe("declined");
+  });
+
+  it("returns null when no button reply_id matches", () => {
+    const node = {
+      node_type: "send_buttons",
+      config: {
+        text: "Pick",
+        buttons: [
+          { reply_id: "a", title: "A", next_node_key: "to_a" },
+          { reply_id: "b", title: "B", next_node_key: "to_b" },
+        ],
+      },
+    };
+    expect(matchReplyId(node, "c")).toBeNull();
+    expect(matchReplyId(node, "")).toBeNull();
+  });
+
+  it("searches across all sections in a send_list node", () => {
+    const node = {
+      node_type: "send_list",
+      config: {
+        text: "Pick an order",
+        button_label: "View",
+        sections: [
+          {
+            title: "Recent",
+            rows: [
+              { reply_id: "o1", title: "Order 1", next_node_key: "ord_1" },
+            ],
+          },
+          {
+            title: "Older",
+            rows: [
+              { reply_id: "o2", title: "Order 2", next_node_key: "ord_2" },
+              { reply_id: "o3", title: "Order 3", next_node_key: "ord_3" },
+            ],
+          },
+        ],
+      },
+    };
+    expect(matchReplyId(node, "o1")).toBe("ord_1");
+    expect(matchReplyId(node, "o2")).toBe("ord_2");
+    expect(matchReplyId(node, "o3")).toBe("ord_3");
+    expect(matchReplyId(node, "o99")).toBeNull();
+  });
+
+  it("returns null when send_list has no sections / empty sections", () => {
+    expect(
+      matchReplyId(
+        { node_type: "send_list", config: { text: "x", sections: [] } },
+        "x",
+      ),
+    ).toBeNull();
+    expect(
+      matchReplyId(
+        {
+          node_type: "send_list",
+          config: { text: "x", sections: [{ rows: [] }] },
+        },
+        "x",
+      ),
+    ).toBeNull();
+  });
+});
+
+describe("matchesKeywordTrigger", () => {
+  it("returns false for empty text", () => {
+    expect(matchesKeywordTrigger("", { keywords: ["hi"] })).toBe(false);
+  });
+
+  it("returns false when keywords array is empty", () => {
+    expect(matchesKeywordTrigger("anything", { keywords: [] })).toBe(false);
+  });
+
+  it("default match_type='contains' does case-insensitive substring", () => {
+    const cfg = { keywords: ["support"] };
+    expect(matchesKeywordTrigger("I need SUPPORT please", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("Support is great", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("Help me", cfg)).toBe(false);
+  });
+
+  it("match_type='exact' compares the whole string case-insensitively", () => {
+    const cfg = { keywords: ["help"], match_type: "exact" as const };
+    expect(matchesKeywordTrigger("help", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("HELP", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("help me", cfg)).toBe(false);
+  });
+
+  it("case_sensitive=true preserves case", () => {
+    const cfg = {
+      keywords: ["Support"],
+      case_sensitive: true,
+    };
+    expect(matchesKeywordTrigger("I need Support", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("I need support", cfg)).toBe(false);
+  });
+
+  it("matches any one of multiple keywords", () => {
+    const cfg = { keywords: ["help", "support", "issue"] };
+    expect(matchesKeywordTrigger("I have an issue", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("I need Help!", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("nothing to see here", cfg)).toBe(false);
+  });
+
+  it("skips empty strings in the keywords array", () => {
+    const cfg = { keywords: ["", "support", ""] };
+    expect(matchesKeywordTrigger("support center", cfg)).toBe(true);
+    expect(matchesKeywordTrigger("nope", cfg)).toBe(false);
+  });
+});
+
+describe("node classification helpers", () => {
+  it("isAutoAdvancing covers start + send_message", () => {
+    expect(isAutoAdvancing("start")).toBe(true);
+    expect(isAutoAdvancing("send_message")).toBe(true);
+    expect(isAutoAdvancing("send_buttons")).toBe(false);
+    expect(isAutoAdvancing("send_list")).toBe(false);
+    expect(isAutoAdvancing("handoff")).toBe(false);
+    expect(isAutoAdvancing("end")).toBe(false);
+  });
+
+  it("isSuspending covers the input-requiring nodes", () => {
+    expect(isSuspending("send_buttons")).toBe(true);
+    expect(isSuspending("send_list")).toBe(true);
+    expect(isSuspending("start")).toBe(false);
+    expect(isSuspending("send_message")).toBe(false);
+    expect(isSuspending("handoff")).toBe(false);
+    expect(isSuspending("end")).toBe(false);
+  });
+
+  it("isTerminal covers handoff + end", () => {
+    expect(isTerminal("handoff")).toBe(true);
+    expect(isTerminal("end")).toBe(true);
+    expect(isTerminal("start")).toBe(false);
+    expect(isTerminal("send_buttons")).toBe(false);
+  });
+
+  it("the three classifications are mutually exclusive for known node types", () => {
+    const types = [
+      "start",
+      "send_message",
+      "send_buttons",
+      "send_list",
+      "handoff",
+      "end",
+    ];
+    for (const t of types) {
+      const flags = [isAutoAdvancing(t), isSuspending(t), isTerminal(t)];
+      // Exactly one of the three should be true for every known node.
+      expect(flags.filter(Boolean).length).toBe(1);
+    }
+  });
+});

src/lib/flows/engine.ts

@@ -0,0 +1,754 @@
+/**
+ * Flow runner.
+ *
+ * The single entry point `dispatchInboundToFlows` is called by the
+ * WhatsApp webhook on every inbound message *for an account that has
+ * opted into the Flows beta*. It decides whether the message belongs
+ * to an active conversation flow (advance it) or matches the entry
+ * trigger of an active flow (start a new run) — and reports back to
+ * the webhook so the webhook knows whether to also fire automations.
+ *
+ * Architecture in a sentence: the runner walks the customer through
+ * a DB-stored node graph, suspending only at nodes that need
+ * customer input. Each tap or text reply wakes it back up.
+ *
+ * What lives here vs elsewhere:
+ *   - Pure decision logic (which button matched, where to advance to,
+ *     when to fallback) — here.
+ *   - DB shape (table reads/writes) — here.
+ *   - Meta API calls — `meta-send.ts` (engineSendInteractive*).
+ *   - Policy resolution (reprompt vs handoff vs end) — `fallback.ts`.
+ *   - Type definitions — `types.ts`.
+ *
+ * Concurrency model:
+ *   - Idempotency on `meta_message_id`: the runner refuses to advance
+ *     an active run twice for the same Meta message — protects against
+ *     Meta's retries.
+ *   - Optimistic UPDATE with `current_node_key` precondition: two
+ *     simultaneous taps for the same run collide at the DB layer; the
+ *     second is a no-op.
+ *   - Partial unique index `idx_one_active_run_per_contact`: two
+ *     simultaneous starts for the same contact collide; the second
+ *     INSERT raises 23505 and the runner catches & exits.
+ */
+
+import { supabaseAdmin } from "./admin-client";
+import {
+  engineSendInteractiveButtons,
+  engineSendInteractiveList,
+} from "./meta-send";
+import { decideFallback, resolveFallbackPolicy } from "./fallback";
+import {
+  type DispatchInboundInput,
+  type DispatchInboundResult,
+  type FlowNodeRow,
+  type FlowRow,
+  type FlowRunRow,
+  type ParsedInbound,
+  type SendButtonsNodeConfig,
+  type SendListNodeConfig,
+  type SendMessageNodeConfig,
+  type StartNodeConfig,
+  type KeywordTriggerConfig,
+} from "./types";
+
+// ============================================================
+// Pure helpers — extracted so engine.test.ts can exercise them
+// without a Supabase / Meta mock.
+// ============================================================
+
+/**
+ * Given a node + the customer's reply_id, return the next_node_key
+ * to advance to, or `null` if no option matches.
+ */
+export function matchReplyId(
+  node: { node_type: string; config: Record<string, unknown> },
+  reply_id: string,
+): string | null {
+  if (node.node_type === "send_buttons") {
+    const cfg = node.config as unknown as SendButtonsNodeConfig;
+    const hit = cfg.buttons?.find((b) => b.reply_id === reply_id);
+    return hit?.next_node_key ?? null;
+  }
+  if (node.node_type === "send_list") {
+    const cfg = node.config as unknown as SendListNodeConfig;
+    for (const section of cfg.sections ?? []) {
+      const hit = section.rows?.find((r) => r.reply_id === reply_id);
+      if (hit) return hit.next_node_key;
+    }
+    return null;
+  }
+  return null;
+}
+
+/**
+ * Case-insensitive contains/exact match against a list of keywords.
+ * Used by the trigger evaluator. Stable enough that the v3 builder
+ * UI can preview matches by passing canned strings.
+ */
+export function matchesKeywordTrigger(
+  text: string,
+  cfg: KeywordTriggerConfig,
+): boolean {
+  if (!text || !cfg.keywords?.length) return false;
+  const matchType = cfg.match_type ?? "contains";
+  const haystack = cfg.case_sensitive ? text : text.toLowerCase();
+  for (const raw of cfg.keywords) {
+    if (!raw) continue;
+    const needle = cfg.case_sensitive ? raw : raw.toLowerCase();
+    if (matchType === "exact" ? haystack === needle : haystack.includes(needle)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+/** Nodes that advance to a next_node_key without waiting for input. */
+export function isAutoAdvancing(node_type: string): boolean {
+  return node_type === "start" || node_type === "send_message";
+}
+
+/** Nodes that send a prompt and suspend awaiting a customer reply. */
+export function isSuspending(node_type: string): boolean {
+  return node_type === "send_buttons" || node_type === "send_list";
+}
+
+/** Nodes that end the run. */
+export function isTerminal(node_type: string): boolean {
+  return node_type === "handoff" || node_type === "end";
+}
+
+// ============================================================
+// DB I/O — wrapped in tiny helpers so the dispatch flow stays
+// readable. Errors surface as thrown — the entry point catches.
+// ============================================================
+
+type AdminClient = ReturnType<typeof supabaseAdmin>;
+
+async function loadActiveRunForContact(
+  db: AdminClient,
+  userId: string,
+  contactId: string,
+): Promise<FlowRunRow | null> {
+  const { data, error } = await db
+    .from("flow_runs")
+    .select("*")
+    .eq("user_id", userId)
+    .eq("contact_id", contactId)
+    .eq("status", "active")
+    .maybeSingle();
+  if (error) {
+    console.error("[flows] loadActiveRunForContact error:", error.message);
+    return null;
+  }
+  return (data as FlowRunRow | null) ?? null;
+}
+
+async function loadFlow(
+  db: AdminClient,
+  flowId: string,
+): Promise<FlowRow | null> {
+  const { data, error } = await db
+    .from("flows")
+    .select("*")
+    .eq("id", flowId)
+    .maybeSingle();
+  if (error) {
+    console.error("[flows] loadFlow error:", error.message);
+    return null;
+  }
+  return (data as FlowRow | null) ?? null;
+}
+
+async function loadNode(
+  db: AdminClient,
+  flowId: string,
+  nodeKey: string,
+): Promise<FlowNodeRow | null> {
+  const { data, error } = await db
+    .from("flow_nodes")
+    .select("*")
+    .eq("flow_id", flowId)
+    .eq("node_key", nodeKey)
+    .maybeSingle();
+  if (error) {
+    console.error("[flows] loadNode error:", error.message);
+    return null;
+  }
+  return (data as FlowNodeRow | null) ?? null;
+}
+
+async function logEvent(
+  db: AdminClient,
+  flowRunId: string,
+  event_type:
+    | "started"
+    | "node_entered"
+    | "message_sent"
+    | "reply_received"
+    | "fallback_fired"
+    | "handoff"
+    | "timeout"
+    | "error"
+    | "completed",
+  node_key: string | null,
+  payload: Record<string, unknown> = {},
+): Promise<void> {
+  const { error } = await db.from("flow_run_events").insert({
+    flow_run_id: flowRunId,
+    event_type,
+    node_key,
+    payload,
+  });
+  if (error) {
+    // Logging failure is non-fatal — surface but don't throw.
+    console.error("[flows] logEvent error:", error.message);
+  }
+}
+
+/**
+ * Idempotency check — has a `reply_received` event with this Meta
+ * message_id already been recorded for any of the contact's flow
+ * runs? If yes, the inbound is a duplicate (Meta retry) and we
+ * exit without re-advancing.
+ *
+ * Implementation note: scoped to runs belonging to this user/contact
+ * so the lookup is cheap (the index on flow_run_events(flow_run_id,
+ * event_type) plus the small set of runs per contact).
+ */
+async function isDuplicateInbound(
+  db: AdminClient,
+  userId: string,
+  contactId: string,
+  metaMessageId: string,
+): Promise<boolean> {
+  // Fetch ALL run ids for this contact (active + historical). Bounded
+  // by how many flows the customer has been through — small.
+  const { data: runs } = await db
+    .from("flow_runs")
+    .select("id")
+    .eq("user_id", userId)
+    .eq("contact_id", contactId);
+  if (!runs?.length) return false;
+  const runIds = runs.map((r) => (r as { id: string }).id);
+
+  const { count } = await db
+    .from("flow_run_events")
+    .select("id", { count: "exact", head: true })
+    .in("flow_run_id", runIds)
+    .eq("event_type", "reply_received")
+    .filter("payload->>meta_message_id", "eq", metaMessageId);
+  return (count ?? 0) > 0;
+}
+
+async function findEntryFlow(
+  db: AdminClient,
+  userId: string,
+  message: ParsedInbound,
+  isFirstInbound: boolean,
+): Promise<FlowRow | null> {
+  // Only text messages can match an entry trigger. Interactive replies
+  // are responses to existing prompts; they never start a new flow.
+  if (message.kind !== "text") return null;
+
+  // Pull all active flows for this user. Active set is bounded (the
+  // builder discourages double-trigger overlap; partial index makes
+  // the lookup index-supported).
+  const { data: flows, error } = await db
+    .from("flows")
+    .select("*")
+    .eq("user_id", userId)
+    .eq("status", "active")
+    .order("created_at", { ascending: true });
+  if (error || !flows) return null;
+
+  const typed = flows as FlowRow[];
+  for (const flow of typed) {
+    if (flow.trigger_type === "keyword") {
+      if (matchesKeywordTrigger(
+        message.text,
+        flow.trigger_config as KeywordTriggerConfig,
+      )) {
+        return flow;
+      }
+    } else if (flow.trigger_type === "first_inbound_message" && isFirstInbound) {
+      return flow;
+    }
+    // 'manual' triggers do not auto-start from inbound messages.
+  }
+  return null;
+}
+
+// ============================================================
+// Node executors — each handles ONE node type. send_buttons and
+// send_list also persist `last_prompt_message_id` so the inbox
+// thread can quote the prompt the customer is replying to.
+// ============================================================
+
+async function sendButtonsAndSuspend(
+  db: AdminClient,
+  run: FlowRunRow,
+  node: FlowNodeRow,
+): Promise<{ outcome: "advanced"; node_key: string }> {
+  const cfg = node.config as unknown as SendButtonsNodeConfig;
+  const { whatsapp_message_id } = await engineSendInteractiveButtons({
+    userId: run.user_id,
+    conversationId: run.conversation_id!,
+    contactId: run.contact_id!,
+    bodyText: cfg.text,
+    headerText: cfg.header_text,
+    footerText: cfg.footer_text,
+    buttons: cfg.buttons.map((b) => ({ id: b.reply_id, title: b.title })),
+  });
+  await logEvent(db, run.id, "message_sent", node.node_key, {
+    node_type: "send_buttons",
+    whatsapp_message_id,
+  });
+  // Look up our internal message id so we can stash it on the run.
+  // Cheap — indexed on `messages.message_id`.
+  const { data: msg } = await db
+    .from("messages")
+    .select("id")
+    .eq("message_id", whatsapp_message_id)
+    .maybeSingle();
+  await db
+    .from("flow_runs")
+    .update({
+      last_prompt_message_id: (msg as { id: string } | null)?.id ?? null,
+    })
+    .eq("id", run.id);
+  return { outcome: "advanced", node_key: node.node_key };
+}
+
+async function sendListAndSuspend(
+  db: AdminClient,
+  run: FlowRunRow,
+  node: FlowNodeRow,
+): Promise<{ outcome: "advanced"; node_key: string }> {
+  const cfg = node.config as unknown as SendListNodeConfig;
+  const { whatsapp_message_id } = await engineSendInteractiveList({
+    userId: run.user_id,
+    conversationId: run.conversation_id!,
+    contactId: run.contact_id!,
+    bodyText: cfg.text,
+    buttonLabel: cfg.button_label,
+    headerText: cfg.header_text,
+    footerText: cfg.footer_text,
+    sections: cfg.sections.map((s) => ({
+      title: s.title,
+      rows: s.rows.map((r) => ({
+        id: r.reply_id,
+        title: r.title,
+        description: r.description,
+      })),
+    })),
+  });
+  await logEvent(db, run.id, "message_sent", node.node_key, {
+    node_type: "send_list",
+    whatsapp_message_id,
+  });
+  const { data: msg } = await db
+    .from("messages")
+    .select("id")
+    .eq("message_id", whatsapp_message_id)
+    .maybeSingle();
+  await db
+    .from("flow_runs")
+    .update({
+      last_prompt_message_id: (msg as { id: string } | null)?.id ?? null,
+    })
+    .eq("id", run.id);
+  return { outcome: "advanced", node_key: node.node_key };
+}
+
+async function executeHandoff(
+  db: AdminClient,
+  run: FlowRunRow,
+  node: FlowNodeRow,
+): Promise<void> {
+  const cfg = node.config as { assign_to?: string; note?: string };
+  const convUpdate: Record<string, unknown> = {
+    status: "pending",
+    updated_at: new Date().toISOString(),
+  };
+  if (cfg.assign_to) convUpdate.assigned_agent_id = cfg.assign_to;
+  if (run.conversation_id) {
+    await db
+      .from("conversations")
+      .update(convUpdate)
+      .eq("id", run.conversation_id);
+  }
+  await logEvent(db, run.id, "handoff", node.node_key, {
+    note: cfg.note ?? null,
+    assigned_to: cfg.assign_to ?? null,
+  });
+  await endRun(db, run.id, "handed_off", "handoff_node");
+}
+
+async function endRun(
+  db: AdminClient,
+  runId: string,
+  status: "completed" | "handed_off" | "timed_out" | "failed",
+  reason: string,
+): Promise<void> {
+  await db
+    .from("flow_runs")
+    .update({
+      status,
+      ended_at: new Date().toISOString(),
+      end_reason: reason,
+    })
+    .eq("id", runId);
+}
+
+// ============================================================
+// The synchronous advance loop. Walks through auto-advance nodes
+// until it hits one that suspends (send_buttons/send_list) or
+// terminates (handoff/end). Each suspending node persists the
+// new current_node_key before returning.
+// ============================================================
+
+async function advanceFromNodeKey(
+  db: AdminClient,
+  run: FlowRunRow,
+  startNodeKey: string,
+): Promise<{ outcome: "advanced" | "completed" | "handed_off" }> {
+  let currentKey: string | null = startNodeKey;
+  // Defensive cap — if a flow has a cycle (which the validator
+  // SHOULD catch but doesn't yet in v1), we bail rather than loop.
+  for (let safety = 0; safety < 64; safety += 1) {
+    if (!currentKey) {
+      await logEvent(db, run.id, "error", null, {
+        reason: "next_node_key was null mid-advance",
+      });
+      await endRun(db, run.id, "failed", "missing_next_node");
+      return { outcome: "completed" };
+    }
+    const node = await loadNode(db, run.flow_id, currentKey);
+    if (!node) {
+      await logEvent(db, run.id, "error", currentKey, {
+        reason: "node_not_found",
+      });
+      await endRun(db, run.id, "failed", "node_not_found");
+      return { outcome: "completed" };
+    }
+    await logEvent(db, run.id, "node_entered", node.node_key, {
+      node_type: node.node_type,
+    });
+
+    if (node.node_type === "start") {
+      currentKey = (node.config as unknown as StartNodeConfig).next_node_key;
+      continue;
+    }
+    if (node.node_type === "send_message") {
+      // v1 send_message is a plain text without input. The dedicated
+      // text-only sender lives in automations/meta-send.ts; we'll
+      // wire it in PR #4 when the collect_input node lands. For v1,
+      // a flow that needs to send plain text uses a send_buttons
+      // node with a single "OK" button as the gating affordance.
+      // Until then, treat send_message as auto-advance with no send.
+      const cfg = node.config as unknown as SendMessageNodeConfig;
+      currentKey = cfg.next_node_key;
+      continue;
+    }
+    if (node.node_type === "send_buttons") {
+      await sendButtonsAndSuspend(db, run, node);
+      // Persist the new current_node_key via optimistic UPDATE.
+      const advanced = await advanceCurrentNodeKey(
+        db,
+        run.id,
+        run.current_node_key,
+        node.node_key,
+      );
+      if (!advanced) {
+        await logEvent(db, run.id, "error", node.node_key, {
+          reason: "lost_race_during_advance",
+        });
+      }
+      return { outcome: "advanced" };
+    }
+    if (node.node_type === "send_list") {
+      await sendListAndSuspend(db, run, node);
+      const advanced = await advanceCurrentNodeKey(
+        db,
+        run.id,
+        run.current_node_key,
+        node.node_key,
+      );
+      if (!advanced) {
+        await logEvent(db, run.id, "error", node.node_key, {
+          reason: "lost_race_during_advance",
+        });
+      }
+      return { outcome: "advanced" };
+    }
+    if (node.node_type === "handoff") {
+      await executeHandoff(db, run, node);
+      return { outcome: "handed_off" };
+    }
+    if (node.node_type === "end") {
+      await logEvent(db, run.id, "completed", node.node_key);
+      await endRun(db, run.id, "completed", "end_node");
+      return { outcome: "completed" };
+    }
+    // Unknown node type — shouldn't happen given the CHECK constraint.
+    await logEvent(db, run.id, "error", node.node_key, {
+      reason: `unknown_node_type:${node.node_type}`,
+    });
+    await endRun(db, run.id, "failed", "unknown_node_type");
+    return { outcome: "completed" };
+  }
+  // Safety break — log + fail.
+  await logEvent(db, run.id, "error", currentKey, {
+    reason: "advance_loop_safety_break",
+  });
+  await endRun(db, run.id, "failed", "advance_loop_overflow");
+  return { outcome: "completed" };
+}
+
+/**
+ * Optimistic UPDATE — only advance current_node_key when it matches
+ * the value we read at the top of dispatch. If another webhook beat
+ * us, the row's pointer has already moved and our UPDATE returns
+ * zero rows; we treat that as a no-op and let the other run continue.
+ */
+async function advanceCurrentNodeKey(
+  db: AdminClient,
+  runId: string,
+  expectedOldKey: string | null,
+  newKey: string,
+): Promise<boolean> {
+  // PostgREST: when expectedOldKey is null we can't `.eq` (would match
+  // any row); use `.is('current_node_key', null)` instead.
+  let q = db
+    .from("flow_runs")
+    .update({
+      current_node_key: newKey,
+      last_advanced_at: new Date().toISOString(),
+    })
+    .eq("id", runId)
+    .eq("status", "active");
+  if (expectedOldKey === null) {
+    q = q.is("current_node_key", null);
+  } else {
+    q = q.eq("current_node_key", expectedOldKey);
+  }
+  const { data, error } = await q.select("id");
+  if (error) {
+    console.error("[flows] advanceCurrentNodeKey error:", error.message);
+    return false;
+  }
+  return Array.isArray(data) && data.length > 0;
+}
+
+// ============================================================
+// Public entry point — the webhook calls this on every inbound.
+// ============================================================
+
+export async function dispatchInboundToFlows(
+  input: DispatchInboundInput & { isFirstInboundMessage: boolean },
+): Promise<DispatchInboundResult> {
+  const db = supabaseAdmin();
+  try {
+    const activeRun = await loadActiveRunForContact(
+      db,
+      input.userId,
+      input.contactId,
+    );
+
+    // Idempotency — only matters if there's already a run for this
+    // contact. For new runs, the partial unique index catches duplicate
+    // starts at INSERT time.
+    if (activeRun) {
+      const dupe = await isDuplicateInbound(
+        db,
+        input.userId,
+        input.contactId,
+        input.message.meta_message_id,
+      );
+      if (dupe) {
+        return {
+          consumed: true,
+          flow_run_id: activeRun.id,
+          outcome: "duplicate_inbound_ignored",
+        };
+      }
+      return handleReplyForActiveRun(db, activeRun, input.message);
+    }
+
+    // No active run → look for a flow whose entry trigger matches.
+    const flow = await findEntryFlow(
+      db,
+      input.userId,
+      input.message,
+      input.isFirstInboundMessage,
+    );
+    if (!flow || !flow.entry_node_id) {
+      return { consumed: false, outcome: "no_match" };
+    }
+    return startNewRun(db, flow, input);
+  } catch (err) {
+    console.error(
+      "[flows] dispatchInboundToFlows threw:",
+      err instanceof Error ? err.message : err,
+    );
+    return { consumed: false, outcome: "no_match" };
+  }
+}
+
+async function handleReplyForActiveRun(
+  db: AdminClient,
+  run: FlowRunRow,
+  message: ParsedInbound,
+): Promise<DispatchInboundResult> {
+  await logEvent(db, run.id, "reply_received", run.current_node_key, {
+    meta_message_id: message.meta_message_id,
+    reply_kind: message.kind,
+    reply_id: message.kind === "interactive_reply" ? message.reply_id : null,
+    text: message.kind === "text" ? message.text : null,
+  });
+
+  if (!run.current_node_key) {
+    // Defensive — a run with status='active' but no current node is
+    // malformed. Fail the run rather than spin.
+    await endRun(db, run.id, "failed", "active_run_missing_current_node");
+    return {
+      consumed: true,
+      flow_run_id: run.id,
+      outcome: "no_match",
+    };
+  }
+
+  const currentNode = await loadNode(db, run.flow_id, run.current_node_key);
+  if (!currentNode) {
+    await endRun(db, run.id, "failed", "current_node_not_found");
+    return { consumed: true, flow_run_id: run.id, outcome: "no_match" };
+  }
+
+  // Only interactive replies can advance a v1 flow. Text replies fall
+  // through to the fallback policy below (collect_input arrives in v1.5).
+  let matched: string | null = null;
+  if (message.kind === "interactive_reply") {
+    matched = matchReplyId(currentNode, message.reply_id);
+  }
+
+  if (matched) {
+    // Reset reprompt count on a successful match.
+    await db
+      .from("flow_runs")
+      .update({ reprompt_count: 0 })
+      .eq("id", run.id);
+    // Re-read so the advance loop sees the fresh reprompt_count.
+    const fresh = await db
+      .from("flow_runs")
+      .select("*")
+      .eq("id", run.id)
+      .maybeSingle();
+    const next = (fresh.data as FlowRunRow | null) ?? run;
+    const outcome = await advanceFromNodeKey(db, next, matched);
+    return {
+      consumed: true,
+      flow_run_id: run.id,
+      outcome: outcome.outcome,
+    };
+  }
+
+  // No match → fallback. Apply the policy.
+  const policy = resolveFallbackPolicy(
+    (await loadFlow(db, run.flow_id))?.fallback_policy,
+  );
+  const newReprompts = run.reprompt_count + 1;
+  await db
+    .from("flow_runs")
+    .update({ reprompt_count: newReprompts })
+    .eq("id", run.id);
+
+  const action = decideFallback({ policy, reprompt_count: newReprompts });
+  await logEvent(db, run.id, "fallback_fired", run.current_node_key, {
+    action: action.type,
+    reprompt_count: newReprompts,
+  });
+  if (action.type === "ignore") {
+    // Don't consume — let automations have a shot at it.
+    return { consumed: false, flow_run_id: run.id, outcome: "no_match" };
+  }
+  if (action.type === "reprompt") {
+    // Re-send the same prompt. Same node, no current_node_key change.
+    if (currentNode.node_type === "send_buttons") {
+      await sendButtonsAndSuspend(db, run, currentNode);
+    } else if (currentNode.node_type === "send_list") {
+      await sendListAndSuspend(db, run, currentNode);
+    }
+    return { consumed: true, flow_run_id: run.id, outcome: "fallback_fired" };
+  }
+  if (action.type === "handoff") {
+    if (run.conversation_id) {
+      await db
+        .from("conversations")
+        .update({ status: "pending", updated_at: new Date().toISOString() })
+        .eq("id", run.conversation_id);
+    }
+    await logEvent(db, run.id, "handoff", run.current_node_key, {
+      reason: "fallback_exhausted",
+    });
+    await endRun(db, run.id, "handed_off", "fallback_exhausted");
+    return { consumed: true, flow_run_id: run.id, outcome: "handed_off" };
+  }
+  // action.type === 'end'
+  await endRun(db, run.id, "completed", "fallback_exhausted_end");
+  return { consumed: true, flow_run_id: run.id, outcome: "completed" };
+}
+
+async function startNewRun(
+  db: AdminClient,
+  flow: FlowRow,
+  input: DispatchInboundInput,
+): Promise<DispatchInboundResult> {
+  // INSERT — partial unique index `idx_one_active_run_per_contact`
+  // catches concurrent inserts with 23505. We catch and return as
+  // consumed:true (the parallel webhook handles it).
+  const { data: inserted, error: insErr } = await db
+    .from("flow_runs")
+    .insert({
+      flow_id: flow.id,
+      user_id: flow.user_id,
+      contact_id: input.contactId,
+      conversation_id: input.conversationId,
+      status: "active",
+      current_node_key: flow.entry_node_id,
+    })
+    .select("*")
+    .maybeSingle();
+  if (insErr) {
+    // 23505 = unique_violation → another webhook is starting the run.
+    const msg = insErr.message ?? "";
+    if (msg.includes("23505") || msg.includes("duplicate key")) {
+      return { consumed: true, outcome: "duplicate_inbound_ignored" };
+    }
+    console.error("[flows] startNewRun insert error:", insErr.message);
+    return { consumed: false, outcome: "no_match" };
+  }
+  const run = inserted as FlowRunRow;
+  await logEvent(db, run.id, "started", flow.entry_node_id, {
+    flow_id: flow.id,
+    trigger_type: flow.trigger_type,
+    meta_message_id: input.message.meta_message_id,
+  });
+  // Bump the flow's execution counter — used by the builder UI to
+  // surface "X runs since activation" on the flow card.
+  await db
+    .from("flows")
+    .update({
+      execution_count: flow.execution_count + 1,
+      last_executed_at: new Date().toISOString(),
+    })
+    .eq("id", flow.id);
+
+  // Run the advance loop starting from the entry node.
+  const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!);
+  return {
+    consumed: true,
+    flow_run_id: run.id,
+    outcome: outcome.outcome === "advanced" ? "started" : outcome.outcome,
+  };
+}

src/lib/flows/fallback.test.ts

@@ -0,0 +1,124 @@
+import { describe, it, expect } from "vitest";
+import {
+  decideFallback,
+  resolveFallbackPolicy,
+} from "./fallback";
+import { DEFAULT_FALLBACK_POLICY, type FlowFallbackPolicy } from "./types";
+
+describe("resolveFallbackPolicy", () => {
+  it("returns defaults for null / undefined / non-object", () => {
+    expect(resolveFallbackPolicy(null)).toEqual(DEFAULT_FALLBACK_POLICY);
+    expect(resolveFallbackPolicy(undefined)).toEqual(DEFAULT_FALLBACK_POLICY);
+    expect(resolveFallbackPolicy("not-an-object")).toEqual(
+      DEFAULT_FALLBACK_POLICY,
+    );
+    expect(resolveFallbackPolicy(42)).toEqual(DEFAULT_FALLBACK_POLICY);
+  });
+
+  it("returns defaults for an empty object", () => {
+    expect(resolveFallbackPolicy({})).toEqual(DEFAULT_FALLBACK_POLICY);
+  });
+
+  it("preserves valid fields, defaults the rest", () => {
+    expect(
+      resolveFallbackPolicy({ max_reprompts: 5, on_exhaust: "end" }),
+    ).toEqual({
+      ...DEFAULT_FALLBACK_POLICY,
+      max_reprompts: 5,
+      on_exhaust: "end",
+    });
+  });
+
+  it("rejects invalid on_unknown_reply values", () => {
+    expect(
+      resolveFallbackPolicy({ on_unknown_reply: "nonsense" as unknown }),
+    ).toEqual(DEFAULT_FALLBACK_POLICY);
+  });
+
+  it("rejects negative or NaN max_reprompts", () => {
+    expect(resolveFallbackPolicy({ max_reprompts: -1 })).toEqual(
+      DEFAULT_FALLBACK_POLICY,
+    );
+    expect(resolveFallbackPolicy({ max_reprompts: Number.NaN })).toEqual(
+      DEFAULT_FALLBACK_POLICY,
+    );
+  });
+
+  it("floors non-integer max_reprompts to be safe", () => {
+    expect(resolveFallbackPolicy({ max_reprompts: 2.7 }).max_reprompts).toBe(2);
+  });
+
+  it("rejects non-positive on_timeout_hours", () => {
+    expect(resolveFallbackPolicy({ on_timeout_hours: 0 })).toEqual(
+      DEFAULT_FALLBACK_POLICY,
+    );
+    expect(resolveFallbackPolicy({ on_timeout_hours: -5 })).toEqual(
+      DEFAULT_FALLBACK_POLICY,
+    );
+  });
+});
+
+const POLICY_REPROMPT_2_HANDOFF: FlowFallbackPolicy = {
+  on_unknown_reply: "reprompt",
+  max_reprompts: 2,
+  on_timeout_hours: 24,
+  on_exhaust: "handoff",
+};
+
+describe("decideFallback", () => {
+  it("returns ignore when on_unknown_reply is 'ignore'", () => {
+    expect(
+      decideFallback({
+        policy: { ...POLICY_REPROMPT_2_HANDOFF, on_unknown_reply: "ignore" },
+        reprompt_count: 1,
+      }),
+    ).toEqual({ type: "ignore" });
+  });
+
+  it("returns handoff immediately when on_unknown_reply is 'handoff'", () => {
+    expect(
+      decideFallback({
+        policy: { ...POLICY_REPROMPT_2_HANDOFF, on_unknown_reply: "handoff" },
+        reprompt_count: 1,
+      }),
+    ).toEqual({ type: "handoff" });
+  });
+
+  it("reprompts up to max_reprompts", () => {
+    // count=1 (first reprompt) and count=2 (second) still re-prompt
+    expect(
+      decideFallback({ policy: POLICY_REPROMPT_2_HANDOFF, reprompt_count: 1 }),
+    ).toEqual({ type: "reprompt" });
+    expect(
+      decideFallback({ policy: POLICY_REPROMPT_2_HANDOFF, reprompt_count: 2 }),
+    ).toEqual({ type: "reprompt" });
+  });
+
+  it("escalates to handoff once max_reprompts is exceeded", () => {
+    // count=3 with max=2 → exhaust → handoff
+    expect(
+      decideFallback({ policy: POLICY_REPROMPT_2_HANDOFF, reprompt_count: 3 }),
+    ).toEqual({ type: "handoff" });
+  });
+
+  it("respects on_exhaust='end' when max is exhausted", () => {
+    const policy: FlowFallbackPolicy = {
+      ...POLICY_REPROMPT_2_HANDOFF,
+      on_exhaust: "end",
+    };
+    expect(decideFallback({ policy, reprompt_count: 5 })).toEqual({
+      type: "end",
+    });
+  });
+
+  it("with max_reprompts=0, the first unknown reply exhausts", () => {
+    const policy: FlowFallbackPolicy = {
+      ...POLICY_REPROMPT_2_HANDOFF,
+      max_reprompts: 0,
+    };
+    // count=1 already > max=0 → exhaust
+    expect(decideFallback({ policy, reprompt_count: 1 })).toEqual({
+      type: "handoff",
+    });
+  });
+});

src/lib/flows/fallback.ts

@@ -0,0 +1,91 @@
+/**
+ * Fallback-policy resolver.
+ *
+ * Pure logic that decides what the engine does when a customer reply
+ * doesn't match any option on the current `send_buttons` / `send_list`
+ * node. Lifted out of `engine.ts` so it can be unit-tested without a
+ * Supabase / Meta mock.
+ *
+ * The policy lives on `flows.fallback_policy` (JSONB) and is loaded
+ * with the run; defaults filled in by `resolveFallbackPolicy` so an
+ * older flow row (or a partial JSONB blob) doesn't crash the runner.
+ */
+
+import {
+  DEFAULT_FALLBACK_POLICY,
+  type FlowFallbackPolicy,
+} from "./types";
+
+export type FallbackAction =
+  /** Re-send the same prompt and wait again. */
+  | { type: "reprompt" }
+  /** End the run with status='handed_off', flip conversation to pending. */
+  | { type: "handoff" }
+  /** End the run with status='completed' (the `end` exhaust option). */
+  | { type: "end" }
+  /** Do nothing — the message wasn't for us. */
+  | { type: "ignore" };
+
+/**
+ * Merge a partial / null fallback_policy from the DB with the v1
+ * defaults. The DB column defaults the *whole* JSONB to the right
+ * shape, but rows authored before this default landed, or rows
+ * manually edited to a subset, would otherwise crash the runner.
+ */
+export function resolveFallbackPolicy(
+  raw: unknown,
+): FlowFallbackPolicy {
+  if (!raw || typeof raw !== "object") return DEFAULT_FALLBACK_POLICY;
+  const r = raw as Partial<FlowFallbackPolicy>;
+  return {
+    on_unknown_reply:
+      r.on_unknown_reply === "handoff" ||
+      r.on_unknown_reply === "ignore" ||
+      r.on_unknown_reply === "reprompt"
+        ? r.on_unknown_reply
+        : DEFAULT_FALLBACK_POLICY.on_unknown_reply,
+    max_reprompts:
+      typeof r.max_reprompts === "number" && r.max_reprompts >= 0
+        ? Math.floor(r.max_reprompts)
+        : DEFAULT_FALLBACK_POLICY.max_reprompts,
+    on_timeout_hours:
+      typeof r.on_timeout_hours === "number" && r.on_timeout_hours > 0
+        ? r.on_timeout_hours
+        : DEFAULT_FALLBACK_POLICY.on_timeout_hours,
+    on_exhaust:
+      r.on_exhaust === "handoff" || r.on_exhaust === "end"
+        ? r.on_exhaust
+        : DEFAULT_FALLBACK_POLICY.on_exhaust,
+  };
+}
+
+/**
+ * Decide the action when the customer's reply doesn't match a button
+ * id on the current node. The engine increments `reprompt_count` and
+ * persists, then calls this with the NEW count.
+ *
+ * - `on_unknown_reply: 'ignore'` → always ignore. Useful for a flow
+ *   that should keep running even if the customer types something
+ *   off-script in between taps (rare; default is reprompt).
+ * - `on_unknown_reply: 'handoff'` → immediately escalate. No retries.
+ * - `on_unknown_reply: 'reprompt'` → re-send the prompt up to
+ *   `max_reprompts` times, then apply `on_exhaust`.
+ */
+export function decideFallback(args: {
+  policy: FlowFallbackPolicy;
+  /** Reprompt count AFTER incrementing (so 1 = first reprompt). */
+  reprompt_count: number;
+}): FallbackAction {
+  const { policy, reprompt_count } = args;
+
+  if (policy.on_unknown_reply === "ignore") return { type: "ignore" };
+  if (policy.on_unknown_reply === "handoff") return { type: "handoff" };
+
+  // 'reprompt' — guarded by max_reprompts.
+  if (reprompt_count <= policy.max_reprompts) {
+    return { type: "reprompt" };
+  }
+  return policy.on_exhaust === "end"
+    ? { type: "end" }
+    : { type: "handoff" };
+}

src/lib/flows/types.ts

@@ -0,0 +1,262 @@
+/**
+ * Type definitions for the Flows runtime.
+ *
+ * These mirror the Supabase schema added in migration 010 (`flows`,
+ * `flow_nodes`, `flow_runs`, `flow_run_events`) plus the discriminated
+ * unions the engine uses to typecheck node configs.
+ *
+ * Schema invariants enforced here that the DB CHECK constraints don't:
+ *   - Each node_type maps to one config shape — adding a new node_type
+ *     requires adding the matching config interface AND extending
+ *     `FlowNodeConfig` so the engine's exhaustiveness checks light up.
+ *   - Edges live INSIDE the config (each button row / list row carries
+ *     `next_node_key`). The DB schema doesn't model this — the
+ *     validator (PR #3) catches missing or orphan edges at save time.
+ *
+ * `next_node_key` is the stable string id stored in `flow_nodes.node_key`,
+ * not a UUID, so flows can be cloned / templated without rewriting
+ * references in JSONB.
+ */
+
+// ============================================================
+// Node configs (discriminated union by node_type)
+// ============================================================
+
+export interface StartNodeConfig {
+  /** Stable node_key of the first real node to advance to. */
+  next_node_key: string;
+}
+
+export interface SendMessageNodeConfig {
+  /** Plain text sent to the customer; can interpolate {{vars.X}}. */
+  text: string;
+  /** Auto-advance target after the message lands at Meta. */
+  next_node_key: string;
+}
+
+export interface SendButtonsNodeConfig {
+  text: string;
+  /** Optional header / footer lines around the buttons. */
+  header_text?: string;
+  footer_text?: string;
+  /** 1-3 buttons; Meta cap enforced in meta-api validation. */
+  buttons: Array<{
+    /** Stable id sent back by Meta when this button is tapped. */
+    reply_id: string;
+    /** Visible label (≤ 20 chars per Meta). */
+    title: string;
+    /** node_key the runner advances to when this button is tapped. */
+    next_node_key: string;
+  }>;
+}
+
+export interface SendListNodeConfig {
+  text: string;
+  /** Label of the tap-to-expand button on the message bubble. */
+  button_label: string;
+  header_text?: string;
+  footer_text?: string;
+  /** 1-10 rows TOTAL across sections; cap enforced in meta-api. */
+  sections: Array<{
+    title?: string;
+    rows: Array<{
+      reply_id: string;
+      title: string;
+      description?: string;
+      next_node_key: string;
+    }>;
+  }>;
+}
+
+export interface HandoffNodeConfig {
+  /** Optional internal note written to flow_run_events.payload.note. */
+  note?: string;
+  /**
+   * Optional agent user_id to assign on the conversation when this
+   * node fires. Leave unset to flip the status without assignment.
+   */
+  assign_to?: string;
+}
+
+// Terminal nodes carry no config — they just stop the run.
+export type EndNodeConfig = Record<string, never>;
+
+/**
+ * Total union — every concrete node_type the v1 engine understands.
+ * Add new node types here and the engine's switch will flag missing
+ * cases via TypeScript's exhaustiveness check.
+ *
+ * v1.5+ additions (collect_input, condition, set_tag, http_fetch) will
+ * extend this union — out-of-scope for the v1 engine PR.
+ */
+export type FlowNodeConfig =
+  | { node_type: "start"; config: StartNodeConfig }
+  | { node_type: "send_message"; config: SendMessageNodeConfig }
+  | { node_type: "send_buttons"; config: SendButtonsNodeConfig }
+  | { node_type: "send_list"; config: SendListNodeConfig }
+  | { node_type: "handoff"; config: HandoffNodeConfig }
+  | { node_type: "end"; config: EndNodeConfig };
+
+export type FlowNodeType = FlowNodeConfig["node_type"];
+
+// ============================================================
+// Triggers (matches `flows.trigger_type` + `trigger_config`)
+// ============================================================
+
+export interface KeywordTriggerConfig {
+  /** One or more keywords. Match is case-insensitive by default. */
+  keywords: string[];
+  match_type?: "exact" | "contains";
+  case_sensitive?: boolean;
+}
+
+// No knobs in v1 — the trigger has a single semantic. Kept as a type
+// alias (not an empty interface) for forward compat without tripping
+// the no-empty-object-type lint rule.
+export type FirstInboundTriggerConfig = Record<string, never>;
+
+export type FlowTriggerConfig =
+  | { trigger_type: "keyword"; config: KeywordTriggerConfig }
+  | { trigger_type: "first_inbound_message"; config: FirstInboundTriggerConfig }
+  | { trigger_type: "manual"; config: Record<string, never> };
+
+// ============================================================
+// DB-row shapes (read by the engine via supabaseAdmin)
+// ============================================================
+
+export interface FlowRow {
+  id: string;
+  user_id: string;
+  name: string;
+  description: string | null;
+  status: "draft" | "active" | "archived";
+  trigger_type: "keyword" | "first_inbound_message" | "manual";
+  trigger_config: KeywordTriggerConfig | FirstInboundTriggerConfig | Record<string, unknown>;
+  entry_node_id: string | null;
+  fallback_policy: FlowFallbackPolicy;
+  execution_count: number;
+  last_executed_at: string | null;
+  created_at: string;
+  updated_at: string;
+}
+
+export interface FlowNodeRow {
+  id: string;
+  flow_id: string;
+  node_key: string;
+  node_type: FlowNodeType;
+  config: Record<string, unknown>;
+  position_x: number;
+  position_y: number;
+  created_at: string;
+}
+
+export interface FlowRunRow {
+  id: string;
+  flow_id: string;
+  user_id: string;
+  contact_id: string | null;
+  conversation_id: string | null;
+  status:
+    | "active"
+    | "completed"
+    | "handed_off"
+    | "timed_out"
+    | "paused_by_agent"
+    | "failed";
+  current_node_key: string | null;
+  last_prompt_message_id: string | null;
+  vars: Record<string, unknown>;
+  reprompt_count: number;
+  started_at: string;
+  last_advanced_at: string;
+  ended_at: string | null;
+  end_reason: string | null;
+}
+
+// ============================================================
+// Fallback policy (matches flows.fallback_policy JSONB)
+// ============================================================
+
+export interface FlowFallbackPolicy {
+  /** What to do when the customer reply doesn't match any option. */
+  on_unknown_reply: "reprompt" | "handoff" | "ignore";
+  /** Max reprompts before applying `on_exhaust`. */
+  max_reprompts: number;
+  /** Stale-run sweep cutoff. */
+  on_timeout_hours: number;
+  /** What to do once max_reprompts has been hit. */
+  on_exhaust: "handoff" | "end";
+}
+
+export const DEFAULT_FALLBACK_POLICY: FlowFallbackPolicy = {
+  on_unknown_reply: "reprompt",
+  max_reprompts: 2,
+  on_timeout_hours: 24,
+  on_exhaust: "handoff",
+};
+
+// ============================================================
+// Engine input — what `dispatchInboundToFlows` accepts
+// ============================================================
+
+/**
+ * Normalised view of an inbound message that the runner needs. The
+ * webhook lifts this out of the raw Meta payload before invoking the
+ * runner; keeps the runner free of any WhatsApp-API specifics.
+ */
+export type ParsedInbound =
+  | {
+      kind: "text";
+      /** The user's typed message body. */
+      text: string;
+      /** Meta's `messages[0].id` — used for idempotency. */
+      meta_message_id: string;
+    }
+  | {
+      kind: "interactive_reply";
+      /** The reply_id of the tapped button or list row. */
+      reply_id: string;
+      /** The visible title of the tapped option (for logging). */
+      reply_title: string;
+      meta_message_id: string;
+    };
+
+export interface DispatchInboundInput {
+  userId: string;
+  contactId: string;
+  conversationId: string;
+  message: ParsedInbound;
+}
+
+export interface DispatchInboundResult {
+  /**
+   * True iff the runner handled the message — it either advanced an
+   * existing run or started a new one matching a flow trigger.
+   * Webhook uses this to decide whether to also fire automations.
+   */
+  consumed: boolean;
+  /** For diagnostics / logging — null when not consumed. */
+  flow_run_id?: string;
+  /** For diagnostics. */
+  outcome?:
+    | "advanced"
+    | "started"
+    | "completed"
+    | "handed_off"
+    | "fallback_fired"
+    | "duplicate_inbound_ignored"
+    | "no_match";
+}
+
+// ============================================================
+// Helpers — exhaustiveness assertions
+// ============================================================
+
+/**
+ * Throws a typed compile-time error if the switch over a discriminated
+ * union forgets a case. Used in the engine's node-type switch.
+ */
+export function assertNever(x: never): never {
+  throw new Error(`Unhandled node type: ${JSON.stringify(x)}`);
+}