refactor(table): dispatcher uses batchTriggerAndWait + tag-based cancel

Switch the per-window cell fan-out from fire-and-forget tasks.trigger to
tasks.batchTriggerAndWait. The dispatcher is now a single long-lived
trigger.dev task that loops dispatcherStep until the table is exhausted;
trigger.dev CRIU-checkpoints the parent during each wait so we don't pay
compute while cells execute. Queue depth is bounded at WINDOW_SIZE per
dispatch — no more flooding trigger.dev with a million queued runs.

- dispatcher.ts builds payloads via the new shared buildPendingRuns helper
  and calls tasks.batchTriggerAndWait directly. Pre-stamps each cell to
  `queued` (jobId=null) so the UI flips instantly.
- table-run-dispatcher.ts is now a plain while-true loop. No
  RUN_BUDGET_MS, no self-re-enqueue, no cold-start tax per window.

Cancel:
- New cancelCellRunsByTags(tags) paginates runs.list + runs.cancel(id).
- cancelWorkflowGroupRuns fires the tag-sweep alongside the per-jobId
  queue.cancelJob path (preserved for auto-fire cells that have real
  jobIds from single tasks.trigger calls).
- Trigger.dev acks the cancel → batchTriggerAndWait resumes → dispatcher
  observes the dispatch-row cancel flag → exits.

Side fixes:
- getAsyncBackendType returns 'trigger-dev' whenever taskContext.isInsideTask
  is true, regardless of TRIGGER_DEV_ENABLED env. The preview/dev-sim
  worker silently routing cell jobs to DatabaseJobQueue (no poller) is
  fixed without any env config change.
- runWorkflowColumn skips the dispatcher entirely when trigger.dev is
  disabled, running cells inline via DatabaseJobQueue.runInline. HTTP
  response returns dispatchId: null in that mode.
- runColumnContract response schema updated to dispatchId.nullable().

Author: TheodoreSpeaks

apps/sim/background/table-run-dispatcher.ts

@@ -1,6 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { task, tasks } from '@trigger.dev/sdk'
+import { task } from '@trigger.dev/sdk'
 import { dispatcherStep } from '@/lib/table/dispatcher'
 
 const logger = createLogger('TableRunDispatcherTask')
@@ -10,10 +10,12 @@ export interface TableRunDispatcherPayload {
 }
 
 /**
- * Trigger.dev wrapper around `dispatcherStep`. Each task run processes one
- * window of rows and re-enqueues itself with `concurrencyKey: dispatchId` so
- * a single dispatch can't fork. Self-re-enqueue caps each task run's
- * duration; the persisted cursor handles crash recovery.
+ * Trigger.dev wrapper around `dispatcherStep`. One task run holds the
+ * dispatcher loop for the dispatch's entire lifetime — each iteration
+ * processes a window of cells via `batchTriggerAndWait`, which checkpoints
+ * the parent via CRIU during the wait so we don't pay compute while cells
+ * execute. The cursor is persisted in DB; if this run crashes, trigger.dev
+ * retries and the next attempt resumes from the persisted cursor.
  */
 export const tableRunDispatcherTask = task({
   id: 'table-run-dispatcher',
@@ -26,16 +28,12 @@ export const tableRunDispatcherTask = task({
   run: async (payload: TableRunDispatcherPayload) => {
     const { dispatchId } = payload
     try {
-      const result = await dispatcherStep(dispatchId)
-      if (result === 'continue') {
-        await tasks.trigger<typeof tableRunDispatcherTask>(
-          'table-run-dispatcher',
-          { dispatchId },
-          { concurrencyKey: dispatchId }
-        )
+      while (true) {
+        const result = await dispatcherStep(dispatchId)
+        if (result === 'done') return
       }
     } catch (err) {
-      logger.error(`[${dispatchId}] dispatcher step failed`, { error: toError(err).message })
+      logger.error(`[${dispatchId}] dispatcher loop failed`, { error: toError(err).message })
       throw err
     }
   },

apps/sim/lib/api/contracts/tables.ts

@@ -900,9 +900,11 @@ export const runColumnContract = defineRouteContract({
     /**
      * `dispatchId` is the id of the `table_run_dispatches` row created for
      * this run. The dispatcher task picks it up and crawls the table row by
-     * row; clients receive cell + dispatch events via SSE.
+     * row; clients receive cell + dispatch events via SSE. Null when
+     * trigger.dev is disabled — in that mode cells run inline in-process and
+     * no dispatch row is created.
      */
-    schema: successResponseSchema(z.object({ dispatchId: z.string().min(1) })),
+    schema: successResponseSchema(z.object({ dispatchId: z.string().min(1).nullable() })),
   },
 })
 

apps/sim/lib/core/async-jobs/config.ts

@@ -1,4 +1,5 @@
 import { createLogger } from '@sim/logger'
+import { taskContext } from '@trigger.dev/core/v3'
 import type { AsyncBackendType, JobQueueBackend } from '@/lib/core/async-jobs/types'
 import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 
@@ -10,10 +11,15 @@ let cachedInlineBackend: JobQueueBackend | null = null
 
 /**
  * Determines which async backend to use based on environment configuration.
- * Follows the fallback chain: trigger.dev → database
+ * Falls back to the database backend when trigger.dev isn't enabled — except
+ * when this process IS a trigger.dev worker (`taskContext.isInsideTask`), in
+ * which case the SDK runtime is available regardless of env vars and we
+ * always want to enqueue back through trigger.dev. Without this carve-out, a
+ * worker pod missing `TRIGGER_DEV_ENABLED=true` silently routes cell jobs to
+ * the database backend that nothing's draining.
  */
 export function getAsyncBackendType(): AsyncBackendType {
-  if (isTriggerDevEnabled) {
+  if (isTriggerDevEnabled || taskContext.isInsideTask) {
     return 'trigger-dev'
   }
 

apps/sim/lib/table/dispatcher.ts

@@ -1,14 +1,17 @@
 import { db } from '@sim/db'
 import { tableRunDispatches, userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
+import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
+import { tasks } from '@trigger.dev/sdk'
 import { and, asc, eq, gt, inArray, type SQL, sql } from 'drizzle-orm'
+import { writeWorkflowGroupState } from '@/lib/table/cell-write'
 import { appendTableEvent } from '@/lib/table/events'
 import type { TableRow } from '@/lib/table/types'
 import {
-  isGroupEligible,
-  scheduleRunsForRows,
-  type ScheduleOpts,
+  buildPendingRuns,
+  cellTagsFor,
+  type RunGroupCellOptions,
   TABLE_CONCURRENCY_LIMIT,
   toTableRow,
 } from './workflow-columns'
@@ -207,36 +210,54 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
     return 'done'
   }
 
-  // Rows were bulk-cleared at click time, so the chunk is ready to enqueue
-  // as-is. We only filter out cells the user cancelled mid-cascade (the
-  // tombstone) and cells whose deps still aren't satisfied.
-  const eligibleRows: TableRow[] = []
+  // Strip rows the user cancelled mid-cascade (post-dispatch tombstones)
+  // before running the shared eligibility filter — `buildPendingRuns`
+  // doesn't know about the per-dispatch cancel tombstone.
+  const tombstoneFiltered: TableRow[] = []
   for (const r of chunk) {
     const tableRow = toTableRow(r)
-    const anyEligible = targetGroups.some((g) => {
-      const exec = tableRow.executions?.[g.id]
-      if (exec?.cancelledAt) {
-        const cancelledAtMs = Date.parse(exec.cancelledAt)
-        if (Number.isFinite(cancelledAtMs) && cancelledAtMs > dispatch.requestedAt.getTime()) {
-          return false
-        }
-      }
-      return isGroupEligible(g, tableRow, { isManualRun: true, mode: dispatch.mode })
+    const tombstoned = dispatch.scope.groupIds.some((gid) => {
+      const exec = tableRow.executions?.[gid]
+      if (!exec?.cancelledAt) return false
+      const cancelledAtMs = Date.parse(exec.cancelledAt)
+      return Number.isFinite(cancelledAtMs) && cancelledAtMs > dispatch.requestedAt.getTime()
     })
-    if (anyEligible) eligibleRows.push(tableRow)
+    if (!tombstoned) tombstoneFiltered.push(tableRow)
   }
 
+  const pendingRuns = buildPendingRuns(table, tombstoneFiltered, {
+    isManualRun: true,
+    groupIds: dispatch.scope.groupIds,
+    mode: dispatch.mode,
+  })
+
   // Cursor advances to the last position in this chunk regardless of
   // eligibility — otherwise a window full of skipped cells loops forever.
   const lastPosition = chunk[chunk.length - 1].position
 
-  if (eligibleRows.length > 0) {
-    const scheduleOpts: ScheduleOpts = {
-      isManualRun: true,
-      groupIds: dispatch.scope.groupIds,
-      mode: dispatch.mode,
+  if (pendingRuns.length > 0) {
+    await stampQueuedForBatch(pendingRuns)
+
+    // `batchTriggerAndWait` blocks the parent dispatcher until every cell
+    // terminates (success / fail / cancel). Trigger.dev checkpoints the
+    // parent during the wait via CRIU, so we don't pay compute. Bounds the
+    // queue depth at WINDOW_SIZE per dispatch — no flooding trigger.dev.
+    const items = pendingRuns.map((runOpts) => ({
+      payload: runOpts,
+      options: {
+        concurrencyKey: runOpts.tableId,
+        tags: cellTagsFor(runOpts),
+      },
+    }))
+    try {
+      await tasks.batchTriggerAndWait('workflow-group-cell', items)
+    } catch (err) {
+      logger.error(`[${dispatchId}] batchTriggerAndWait failed`, {
+        error: toError(err).message,
+      })
+      // Don't bail the dispatch — terminal states are already in the DB
+      // (workers wrote them) or will be reconciled on the next user click.
     }
-    await scheduleRunsForRows(table, eligibleRows, scheduleOpts)
   }
 
   await Promise.all([
@@ -252,6 +273,26 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
   return 'continue'
 }
 
+/** Pre-batch stamp: write each cell's exec to `queued` before firing the
+ *  batch. We don't know per-cell jobIds (batchTriggerAndWait returns only a
+ *  batch handle), so `jobId` stays null. Workers update to `running` and
+ *  terminal states from inside their own run. */
+async function stampQueuedForBatch(pendingRuns: RunGroupCellOptions[]): Promise<void> {
+  await Promise.allSettled(
+    pendingRuns.map((runOpts) =>
+      writeWorkflowGroupState(runOpts, {
+        executionState: {
+          status: 'queued',
+          executionId: runOpts.executionId,
+          jobId: null,
+          workflowId: runOpts.workflowId,
+          error: null,
+        },
+      })
+    )
+  )
+}
+
 async function advanceCursor(dispatchId: string, newCursor: number): Promise<void> {
   await db
     .update(tableRunDispatches)