refactor(table): unify trigger.dev and inline dispatcher paths

`runWorkflowColumn` now always inserts a `table_run_dispatches` row and
drives the dispatcher state machine. The trigger.dev / in-process branch
narrows to a single line: trigger.dev fires `tableRunDispatcherTask` (which
calls the new `runDispatcherToCompletion`), the inline path calls the same
helper fire-and-forget. Deletes `scheduleRunsForRows` and
`stampQueuedOrCancel` — the inline-fallback no longer duplicates window
walking, SSE emission, or cancel.

The dispatcher's window-execute call goes through `JobQueueBackend`:
- New `batchEnqueueAndWait` interface method.
- Trigger.dev impl wraps `tasks.batchTriggerAndWait` behind a
  `taskContext.isInsideTask` guard (clear error if called from outside a
  task).
- Database impl skips `async_jobs` entirely — `Promise.all` over
  `options.runner(payload, signal)` per item, with per-cell AbortControllers
  tracked by `cancelKey` for cancel.

`cancelInlineRun` moves to the interface as `cancelByKey` so
`cancelWorkflowGroupRuns` no longer reaches into the database backend.

Fix `mode: 'new'` SQL filter:
- `${array}::text[]` interpolated as a tuple-cast which Postgres rejected
  ("cannot cast type record to text[]") and every inline dispatch silently
  failed. Switched to `ARRAY[${sql.join(...)}]::text[]`.
- Predicate was `jsonb_exists_any` ("any one targeted group present"),
  which excluded rows that needed at least one group re-run after a
  downstream output was deleted. Switched to `jsonb_exists_all` — per-group
  JS eligibility handles the rest.

Cascade-loop workflowId bug: `runRowCascadeLoop` was not threading the new
group's `workflowId` when advancing across groups. The cell-task ran the
previous group's workflow against the next group's cell, terminating
`completed` with empty `accumulatedData`. Fixed by tracking
`currentWorkflowId` alongside `currentGroupId` / `currentExecutionId`.

Client optimistic-patch tightening:
- `useRunColumn.onMutate` mirrors server eligibility — skip cells with
  unmet deps so unmet rows don't flash Queued and get stuck (no SSE will
  arrive for cells the server skipped).
- `resolveCellExec` overlay synthesizes a virtual `pending` only when
  `areGroupDepsSatisfied` is true. Rows with unmet deps render Waiting,
  matching the dispatcher's actual behavior.

Cleanup from /simplify pass:
- Use `generateShortId(20)` instead of
  `generateId().replace(/-/g, '').slice(0, 20)`.
- Inline `batchEnqueueAndWait` no longer allocates synthetic ids
  (returned `string[]` is unused).
- Flattened the per-cell `tracked` array — only push entries that
  registered controllers, drop the null placeholders.
- Extracted `runDispatcherToCompletion` to share the loop between the
  trigger.dev wrapper and the in-process path.

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

Author: TheodoreSpeaks

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/utils.ts

@@ -6,7 +6,7 @@ import type {
   TableRow as TableRowType,
   WorkflowGroup,
 } from '@/lib/table'
-import { areOutputsFilled } from '@/lib/table/deps'
+import { areGroupDepsSatisfied, areOutputsFilled } from '@/lib/table/deps'
 import type { DeletedRowSnapshot } from '@/stores/table/types'
 import type { DisplayColumn } from './types'
 
@@ -170,10 +170,12 @@ export function readExecution(
 
 /**
  * Resolves a cell's execution state with the "about to run" overlay applied:
- * for cells in an active dispatch's scope ahead of its cursor, returns a
- * synthetic `pending` exec so the renderer shows `Queued`. Cells with a real
- * DB exec always win — the overlay only fills the gap between dispatch start
- * and the dispatcher's per-row pending stamp.
+ * for cells in an active dispatch's scope ahead of its cursor whose deps are
+ * already satisfied, returns a synthetic `pending` exec so the renderer
+ * shows `Queued`. Cells with a real DB exec always win — the overlay only
+ * fills the gap between dispatch start and the dispatcher's per-row pending
+ * stamp. Cells with unmet deps still render as `Waiting` (the renderer
+ * computes that from `waitingOnLabels`).
  */
 export function resolveCellExec(
   row: TableRowType,
@@ -185,6 +187,7 @@ export function resolveCellExec(
   if (real) return real
   if (!activeDispatches || activeDispatches.length === 0) return undefined
   if (areOutputsFilled(group, row)) return undefined
+  if (!areGroupDepsSatisfied(group, row)) return undefined
   for (const d of activeDispatches) {
     if (!d.scope.groupIds.includes(group.id)) continue
     if (d.scope.rowIds && !d.scope.rowIds.includes(row.id)) continue

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

@@ -1,7 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { task } from '@trigger.dev/sdk'
-import { dispatcherStep } from '@/lib/table/dispatcher'
+import { runDispatcherToCompletion } from '@/lib/table/dispatcher'
 
 const logger = createLogger('TableRunDispatcherTask')
 
@@ -28,10 +28,7 @@ export const tableRunDispatcherTask = task({
   run: async (payload: TableRunDispatcherPayload) => {
     const { dispatchId } = payload
     try {
-      while (true) {
-        const result = await dispatcherStep(dispatchId)
-        if (result === 'done') return
-      }
+      await runDispatcherToCompletion(dispatchId)
     } catch (err) {
       logger.error(`[${dispatchId}] dispatcher loop failed`, { error: toError(err).message })
       throw err

apps/sim/background/workflow-column-execution.ts

@@ -43,6 +43,7 @@ export async function runRowCascadeLoop(
   const { pickNextEligibleGroupForRow } = await import('@/lib/table/workflow-columns')
 
   let currentGroupId = payload.groupId
+  let currentWorkflowId = payload.workflowId
   // Fresh executionId per iteration: SQL guard rejects writes whose id ≠
   // row.executions[gid].executionId, so we need a new claim per group.
   let currentExecutionId = payload.executionId
@@ -62,7 +63,12 @@ export async function runRowCascadeLoop(
     }
 
     const result = await runWorkflowAndWriteTerminal(
-      { ...payload, groupId: currentGroupId, executionId: currentExecutionId },
+      {
+        ...payload,
+        groupId: currentGroupId,
+        workflowId: currentWorkflowId,
+        executionId: currentExecutionId,
+      },
       signal,
       freshTable,
       currentGroup
@@ -75,6 +81,7 @@ export async function runRowCascadeLoop(
     const next = pickNextEligibleGroupForRow(freshTable, freshRow, currentGroupId)
     if (!next) break
     currentGroupId = next.id
+    currentWorkflowId = next.workflowId
     currentExecutionId = generateId()
   }
 }

apps/sim/hooks/queries/tables.ts

@@ -72,6 +72,7 @@ import type {
   WorkflowGroupOutput,
 } from '@/lib/table'
 import {
+  areGroupDepsSatisfied,
   areOutputsFilled,
   isExecInFlight,
   optimisticallyScheduleNewlyEligibleGroups,
@@ -1331,22 +1332,23 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) {
         for (const groupId of targetGroupIds) {
           const exec = executions[groupId] as RowExecutionMetadata | undefined
           if (isExecInFlight(exec)) continue
+          const group = groupsById.get(groupId)
+          // Mirror server eligibility: rows with unmet deps are skipped by the
+          // dispatcher regardless of mode. Stamping pending here would leave
+          // the cell flashing Queued indefinitely (no SSE event will arrive).
+          if (group && !areGroupDepsSatisfied(group, r)) continue
           // Mirror server eligibility for `mode: 'incomplete'`: skip cells whose
           // outputs are filled, regardless of exec status. A cancelled/error
           // cell with a leftover value from a prior run was rendering as filled
           // but flipping to "queued" optimistically here even though the server
           // would skip it.
-          if (runMode === 'incomplete') {
-            const group = groupsById.get(groupId)
-            if (group && areOutputsFilled(group, r)) continue
-          }
+          if (runMode === 'incomplete' && group && areOutputsFilled(group, r)) continue
           next[groupId] = buildPendingExec(exec)
           // Mirror the server-side bulk clear: wipe output values so the cell
           // doesn't render the stale completed value behind a pending badge.
           // Without this the cell-render path's "value wins" branch keeps
           // showing the previous run's output and the Queued/Running pill
           // never appears.
-          const group = groupsById.get(groupId)
           if (group) {
             for (const o of group.outputs) {
               if (o.columnName in nextData) nextData[o.columnName] = null

apps/sim/lib/core/async-jobs/backends/database.ts

@@ -1,7 +1,7 @@
 import { asyncJobs, db } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { generateId } from '@sim/utils/id'
+import { generateShortId } from '@sim/utils/id'
 import { eq, sql } from 'drizzle-orm'
 import {
   type EnqueueOptions,
@@ -37,6 +37,14 @@ function rowToJob(row: AsyncJobRow): Job {
 
 const inlineAbortControllers = new Map<string, AbortController>()
 
+/**
+ * Per-cancel-key abort controllers for the `batchEnqueueAndWait` direct-call
+ * path. Distinct from `inlineAbortControllers` (which keys by jobId) — this
+ * map keys by the domain `cancelKey` callers pass in, since the await-blocking
+ * path skips `async_jobs` entirely and has no jobId to cancel by.
+ */
+const inlineCancelKeyControllers = new Map<string, AbortController>()
+
 interface Semaphore {
   available: number
   waiters: Array<() => void>
@@ -73,7 +81,7 @@ export class DatabaseJobQueue implements JobQueueBackend {
     payload: TPayload,
     options?: EnqueueOptions
   ): Promise<string> {
-    const jobId = options?.jobId ?? `run_${generateId().replace(/-/g, '').slice(0, 20)}`
+    const jobId = options?.jobId ?? `run_${generateShortId(20)}`
     const now = new Date()
 
     await db
@@ -112,7 +120,7 @@ export class DatabaseJobQueue implements JobQueueBackend {
     if (items.length === 0) return []
     const now = new Date()
     const rows = items.map(({ payload, options }) => ({
-      id: `run_${generateId().replace(/-/g, '').slice(0, 20)}`,
+      id: `run_${generateShortId(20)}`,
       type,
       payload: payload as Record<string, unknown>,
       status: JOB_STATUS.PENDING,
@@ -144,6 +152,44 @@ export class DatabaseJobQueue implements JobQueueBackend {
     return rows.map((r) => r.id)
   }
 
+  /** Skips `async_jobs` entirely — ids are returned empty since callers can't
+   *  look up rows that don't exist. Cancel goes through `cancelByKey`. */
+  async batchEnqueueAndWait<TPayload>(
+    type: JobType,
+    items: Array<{ payload: TPayload; options?: EnqueueOptions }>
+  ): Promise<string[]> {
+    if (items.length === 0) return []
+    const tracked: Array<{ key: string; controller: AbortController }> = []
+    const runs = items.map((item) => {
+      const runner = item.options?.runner
+      if (!runner) return Promise.resolve()
+      const controller = new AbortController()
+      const cancelKey = item.options?.cancelKey
+      if (cancelKey) {
+        inlineCancelKeyControllers.set(cancelKey, controller)
+        tracked.push({ key: cancelKey, controller })
+      }
+      return runner(item.payload, controller.signal).catch((err) => {
+        logger.error(`[${type}] Inline run failed`, {
+          cancelKey,
+          error: toError(err).message,
+        })
+      })
+    })
+    try {
+      await Promise.all(runs)
+    } finally {
+      // Compare-and-delete guards against a re-enqueue under the same key
+      // racing with our cleanup.
+      for (const t of tracked) {
+        if (inlineCancelKeyControllers.get(t.key) === t.controller) {
+          inlineCancelKeyControllers.delete(t.key)
+        }
+      }
+    }
+    return items.map(() => '')
+  }
+
   async getJob(jobId: string): Promise<Job | null> {
     const [row] = await db.select().from(asyncJobs).where(eq(asyncJobs.id, jobId)).limit(1)
 
@@ -224,6 +270,14 @@ export class DatabaseJobQueue implements JobQueueBackend {
     logger.debug('Marked job as cancelled (DB queue)', { jobId, abortedInline: aborted })
   }
 
+  cancelByKey(cancelKey: string): boolean {
+    const controller = inlineCancelKeyControllers.get(cancelKey)
+    if (!controller) return false
+    controller.abort('Cancelled')
+    inlineCancelKeyControllers.delete(cancelKey)
+    return true
+  }
+
   /**
    * Fire-and-forget IIFE that owns the lifecycle for an inline job: registers
    * the abort controller (so `cancelJob` can interrupt mid-flight), acquires

apps/sim/lib/core/async-jobs/backends/trigger-dev.ts

@@ -1,4 +1,5 @@
 import { createLogger } from '@sim/logger'
+import { taskContext } from '@trigger.dev/core/v3'
 import { runs, type TriggerOptions, tasks } from '@trigger.dev/sdk'
 import {
   type EnqueueOptions,
@@ -103,6 +104,50 @@ export class TriggerDevJobQueue implements JobQueueBackend {
     return ids
   }
 
+  async batchEnqueueAndWait<TPayload>(
+    type: JobType,
+    items: Array<{ payload: TPayload; options?: EnqueueOptions }>
+  ): Promise<string[]> {
+    if (items.length === 0) return []
+    // The SDK's checkpoint-and-resume requires task runtime context. The only
+    // caller (`dispatcherStep` invoked by `tableRunDispatcherTask.run`) is
+    // always inside a task; check defensively so misuse fails at the boundary
+    // instead of as a confusing SDK internal error.
+    if (!taskContext.isInsideTask) {
+      throw new Error(
+        'batchEnqueueAndWait requires trigger.dev task runtime context — call from within a registered task'
+      )
+    }
+
+    const taskId = JOB_TYPE_TO_TASK_ID[type]
+    if (!taskId) throw new Error(`Unknown job type: ${type}`)
+
+    const batchItems = items.map(({ payload, options }) => {
+      const enrichedPayload =
+        options?.metadata && typeof payload === 'object' && payload !== null
+          ? { ...payload, ...options.metadata }
+          : payload
+      const tags = buildTags(options)
+      const batchItem: {
+        payload: unknown
+        options?: { concurrencyKey?: string; tags?: string[] }
+      } = { payload: enrichedPayload }
+      const batchOpts: { concurrencyKey?: string; tags?: string[] } = {}
+      if (options?.concurrencyKey) batchOpts.concurrencyKey = options.concurrencyKey
+      if (tags.length > 0) batchOpts.tags = tags
+      if (Object.keys(batchOpts).length > 0) batchItem.options = batchOpts
+      return batchItem
+    })
+
+    const result = await tasks.batchTriggerAndWait(taskId, batchItems)
+    logger.debug('batchTriggerAndWait completed', {
+      type,
+      taskId,
+      runCount: result.runs.length,
+    })
+    return result.runs.map((r) => r.id)
+  }
+
   async getJob(jobId: string): Promise<Job | null> {
     try {
       const run = await runs.retrieve(jobId)
@@ -168,6 +213,13 @@ export class TriggerDevJobQueue implements JobQueueBackend {
       throw error
     }
   }
+
+  cancelByKey(_cancelKey: string): boolean {
+    // No in-process AbortControllers to abort — trigger.dev runs are cancelled
+    // by jobId or via tag sweep (see `cancelCellRunsByTags`). Callers that
+    // need both surfaces should fan out themselves.
+    return false
+  }
 }
 
 /**

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

@@ -96,6 +96,14 @@ export interface EnqueueOptions {
    * payload and an `AbortSignal` driven by `cancelJob`.
    */
   runner?: <TPayload>(payload: TPayload, signal: AbortSignal) => Promise<void>
+  /**
+   * Stable identity for cancellation lookups on the database backend's
+   * `batchEnqueueAndWait` path (which skips `async_jobs` entirely, so there
+   * is no jobId to cancel by). Lets callers map a domain identity (e.g.
+   * `tableId:rowId:groupId`) to the in-flight `AbortController`. Ignored
+   * by trigger.dev — runs there are cancelled by tag or jobId.
+   */
+  cancelKey?: string
 }
 
 /**
@@ -118,6 +126,28 @@ export interface JobQueueBackend {
     items: Array<{ payload: TPayload; options?: EnqueueOptions }>
   ): Promise<string[]>
 
+  /**
+   * Enqueue a batch and block until every job has reached a terminal state
+   * (completed, failed, or cancelled). The caller — typically a dispatcher
+   * walking work in windows — uses this to gate window N+1 on window N's
+   * completion.
+   *
+   * Backend implementations:
+   * - Trigger.dev: wraps `tasks.batchTriggerAndWait`. MUST be called from
+   *   inside a registered trigger.dev task (the SDK's checkpoint-and-resume
+   *   requires task runtime context). Backends guard with
+   *   `taskContext.isInsideTask` and throw a clear error otherwise.
+   * - Database (in-process): bypasses `async_jobs` entirely. Since the
+   *   caller is awaiting in-process, the row would serve no live purpose
+   *   (no cross-process recovery, no by-id lookup, no semaphore needed —
+   *   window size IS the concurrency cap). Calls the runner directly via
+   *   `Promise.all` and resolves on the runner's exit.
+   */
+  batchEnqueueAndWait<TPayload>(
+    type: JobType,
+    items: Array<{ payload: TPayload; options?: EnqueueOptions }>
+  ): Promise<string[]>
+
   /**
    * Get a job by ID
    */
@@ -144,6 +174,15 @@ export interface JobQueueBackend {
    * should resolve quietly so callers can drive cancel from possibly-stale state.
    */
   cancelJob(jobId: string): Promise<void>
+
+  /**
+   * Cancel an in-flight job by its `cancelKey` (the domain identity callers
+   * stamped on enqueue via `EnqueueOptions.cancelKey`). Used by
+   * `batchEnqueueAndWait` paths that skip per-job ids; the trigger.dev
+   * backend has no in-process AbortControllers to abort and returns `false`.
+   * Returns `true` if a matching controller was found and aborted.
+   */
+  cancelByKey(cancelKey: string): boolean
 }
 
 export type AsyncBackendType = 'trigger-dev' | 'database'