feat(workflows): crash recovery — resume runs orphaned mid-flow

Closes the crash-safety gap: a run left in running/pending by a crash had
nothing to resume it (the scheduler only wakes 'waiting'). Adds a heartbeat
(workflow_runs.updated_at, stamped on every node advance + start/resume) and a
recoverStaleRuns() sweep that re-enters runs whose heartbeat has gone stale
(>10 min) from their persisted node. Runs on the scheduler tick AND the boot
tick, so a restart catches anything stranded during downtime.

Re-entry is at-least-once (the current node may re-execute) — loop counters +
the late-fee math are idempotent, so the only residual risk is a duplicate
reminder email. An attempts counter (migration 145, cap 5) marks a run failed
instead of recovering a node that reliably crashes the process (crash-loop
backstop). Flag-gated. Tests: orphan-resume + crash-loop cap.

backend/__tests__/integration/workflowEngine.test.js

@@ -279,4 +279,40 @@ describe('workflow engine', () => {
     const after = await db('workflows').where({ id: wf.id }).first();
     expect(after.version).toBe(before.version); // unchanged
   });
+
+  test('recoverStaleRuns resumes a run orphaned mid-flow (crash recovery)', async () => {
+    const wfId = await makeWorkflow({
+      trigger: 'recover.event',
+      nodes: [{ key: 'r1', type: 'trigger' }, { key: 'r2', type: 'action', config: { action: 'noop' } }],
+      edges: [{ from: 'r1', to: 'r2' }],
+    });
+    // Simulate a run left 'running' at r2 with a stale heartbeat (crash mid-flow).
+    await db('workflow_runs').insert({
+      workflow_id: wfId, version: 1, trigger_event: 'recover.event', status: 'running', current_node: 'r2',
+      context: JSON.stringify({ vars: {} }), dedup_key: 'recover-1',
+      updated_at: new Date(Date.now() - 3600000).toISOString(),
+    });
+    const run0 = await db('workflow_runs').where({ dedup_key: 'recover-1' }).first();
+    const n = await engine.recoverStaleRuns({ staleMs: 1000 });
+    expect(n).toBeGreaterThanOrEqual(1);
+    const run = await db('workflow_runs').where({ id: run0.id }).first();
+    expect(run.status).toBe('done');
+  });
+
+  test('recoverStaleRuns abandons a crash-looping run after the attempts cap', async () => {
+    const wfId = await makeWorkflow({
+      trigger: 'crashloop.event',
+      nodes: [{ key: 'c1', type: 'trigger' }, { key: 'c2', type: 'action', config: { action: 'noop' } }],
+      edges: [{ from: 'c1', to: 'c2' }],
+    });
+    await db('workflow_runs').insert({
+      workflow_id: wfId, version: 1, trigger_event: 'crashloop.event', status: 'running', current_node: 'c2',
+      context: JSON.stringify({ vars: {} }), dedup_key: 'crash-1', attempts: 5,
+      updated_at: new Date(Date.now() - 3600000).toISOString(),
+    });
+    const run0 = await db('workflow_runs').where({ dedup_key: 'crash-1' }).first();
+    await engine.recoverStaleRuns({ staleMs: 1000 });
+    const run = await db('workflow_runs').where({ id: run0.id }).first();
+    expect(run.status).toBe('failed');
+  });
 });

backend/migrations/core/145_workflow_run_recovery.js

@@ -0,0 +1,33 @@
+/**
+ * Migration 145: crash-recovery fields for workflow runs.
+ *
+ * A run left in 'running'/'pending' by a crash has nothing to resume it (the
+ * scheduler only wakes 'waiting' runs). Add a heartbeat (`updated_at`, stamped
+ * on every step) so a recovery sweep can detect stale runs, plus an `attempts`
+ * counter so a node that reliably crashes the process can't be recovered
+ * forever (crash-loop backstop → marked failed after a cap).
+ */
+exports.up = async function (knex) {
+  if (!(await knex.schema.hasTable('workflow_runs'))) return;
+  const hasUpdated = await knex.schema.hasColumn('workflow_runs', 'updated_at');
+  const hasAttempts = await knex.schema.hasColumn('workflow_runs', 'attempts');
+  await knex.schema.alterTable('workflow_runs', (t) => {
+    if (!hasUpdated) t.timestamp('updated_at').defaultTo(knex.fn.now());
+    if (!hasAttempts) t.integer('attempts').notNullable().defaultTo(0);
+  });
+  // Recovery sweep queries by (status, updated_at).
+  if (!hasUpdated) {
+    try { await knex.schema.alterTable('workflow_runs', (t) => t.index(['status', 'updated_at'], 'workflow_runs_recovery_index')); } catch (_) {}
+  }
+};
+
+exports.down = async function (knex) {
+  if (!(await knex.schema.hasTable('workflow_runs'))) return;
+  try { await knex.schema.alterTable('workflow_runs', (t) => t.dropIndex(['status', 'updated_at'], 'workflow_runs_recovery_index')); } catch (_) {}
+  if (await knex.schema.hasColumn('workflow_runs', 'updated_at')) {
+    await knex.schema.alterTable('workflow_runs', (t) => t.dropColumn('updated_at'));
+  }
+  if (await knex.schema.hasColumn('workflow_runs', 'attempts')) {
+    await knex.schema.alterTable('workflow_runs', (t) => t.dropColumn('attempts'));
+  }
+};

backend/src/services/invoiceSchedulerService.js

@@ -46,8 +46,13 @@ async function runTick() {
     // Resume workflow runs whose wait has elapsed. No-op (fails closed) when
     // the `workflows` feature flag is off. Independent try/catch so a workflow
     // failure never suppresses the invoice/reminder jobs above.
-    const resumed = await require('./workflows').runDueWaits();
+    const wf = require('./workflows');
+    const resumed = await wf.runDueWaits();
     if (resumed) logger.info('Workflow scheduler: resumed waiting runs', { resumed });
+    // Recover runs orphaned by a crash (stuck in running/pending). Runs on the
+    // boot tick too, so a restart catches anything stranded during downtime.
+    const recovered = await wf.recoverStaleRuns();
+    if (recovered) logger.warn('Workflow scheduler: recovered orphaned runs', { recovered });
   } catch (err) {
     logger.error('Workflow resume pass failed', { err: err.message });
   }

backend/src/services/workflows/engine.js

@@ -186,7 +186,7 @@ async function advanceRun(runId) {
     }
 
     currentKey = nextKey;
-    await db('workflow_runs').where({ id: runId }).update({ current_node: currentKey || null, context: JSON.stringify(context) });
+    await db('workflow_runs').where({ id: runId }).update({ current_node: currentKey || null, context: JSON.stringify(context), updated_at: db.fn.now() });
   }
 
   await finishRun(runId);
@@ -200,7 +200,7 @@ async function startRun(runId) {
   let entry = null;
   for (const n of nodeByKey.values()) { if (n.type === 'trigger') { entry = n; break; } }
   if (!entry) { await failRun(runId, 'no trigger node'); return; }
-  await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: entry.node_key });
+  await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: entry.node_key, updated_at: db.fn.now() });
   await advanceRun(runId);
 }
 
@@ -214,7 +214,7 @@ async function resumeRun(runId, { decisionHandle = null } = {}) {
   const { edges } = await loadGraph(run.workflow_id, run.version);
   const e = outEdge(edges, run.current_node, decisionHandle);
   const nextKey = e ? e.to_node : null;
-  await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: nextKey, wake_at: null });
+  await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: nextKey, wake_at: null, updated_at: db.fn.now() });
   if (!nextKey) { await finishRun(runId); return; }
   await advanceRun(runId);
 }
@@ -310,9 +310,64 @@ async function runDueWaits(limit = 100) {
   }
 }
 
+const RECOVERY_STALE_MS = 10 * 60 * 1000; // a 'running' run idle this long = orphaned by a crash
+const MAX_RECOVERY_ATTEMPTS = 5;
+
+/**
+ * Resume runs orphaned by a crash. A run left in 'running'/'pending' has nothing
+ * to resume it (the scheduler only wakes 'waiting'), so this sweep picks up ones
+ * whose heartbeat (updated_at) has gone stale and re-enters them from their
+ * persisted node. Re-entry is at-least-once: the current node may re-execute —
+ * loop counters + the late-fee math are idempotent, so the only residual risk is
+ * a duplicate reminder email. `attempts` caps recovery so a node that reliably
+ * crashes the process is marked failed instead of looping forever. Flag-gated
+ * (fails closed when workflows is off). Called from the scheduler tick + boot.
+ */
+async function recoverStaleRuns({ staleMs = RECOVERY_STALE_MS, limit = 50 } = {}) {
+  try {
+    const { isFeatureEnabled } = require('../../middleware/requireFeatureFlag');
+    let enabled = false;
+    try { enabled = await isFeatureEnabled('workflows'); } catch (e) { return 0; }
+    if (!enabled) return 0;
+    if (!(await db.schema.hasColumn('workflow_runs', 'updated_at'))) return 0;
+
+    const cutoff = new Date(Date.now() - staleMs).toISOString();
+    const stale = await db('workflow_runs')
+      .whereIn('status', ['running', 'pending'])
+      .where('updated_at', '<=', cutoff)
+      .limit(limit);
+
+    let recovered = 0;
+    for (const run of stale) {
+      try {
+        const attempts = Number(run.attempts) || 0;
+        if (attempts >= MAX_RECOVERY_ATTEMPTS) {
+          await failRun(run.id, `abandoned after ${attempts} recovery attempts (suspected crash loop)`);
+          continue;
+        }
+        await db('workflow_runs').where({ id: run.id }).update({ attempts: attempts + 1, updated_at: db.fn.now() });
+        if (!run.current_node) {
+          await startRun(run.id);
+        } else {
+          await db('workflow_runs').where({ id: run.id }).update({ status: 'running', updated_at: db.fn.now() });
+          await advanceRun(run.id);
+        }
+        recovered += 1;
+      } catch (err) {
+        logger.error('[workflow] recovery failed', { runId: run.id, error: err.message });
+      }
+    }
+    return recovered;
+  } catch (e) {
+    logger.error('[workflow] recoverStaleRuns failed', { error: e.message });
+    return 0;
+  }
+}
+
 module.exports = {
   emitWorkflowEvent,
   runDueWaits,
+  recoverStaleRuns,
   startRun,
   advanceRun,
   resumeRun,