fix(table): cancel prior runs, scope batch insert dispatch, recover orphan pre-stamps

Addresses cursor + greptile review feedback on table dispatcher edge cases:

- Manual table-wide Run-all / Run-column now cancels prior active dispatches
  AND in-flight cell workers before bulk-clearing. Without this, mode:'all'
  deleted running sidecar rows out from under their workers (which kept
  writing into the wiped state) and a second Run-all could enqueue overlapping
  cells racing on the same rows. Row-scoped manual calls (dep-edit cascade)
  are excluded — those already cancel their own scope.
- batchInsertRowsWithTx now scopes its auto-dispatch to the newly-inserted
  row ids. Without this, after the sidecar migration the NOT EXISTS filter
  matches every existing row (zero sidecar entries), so a CSV import would
  walk the entire table dispatching workflow runs on every pre-existing row.
- classifyEligibility carve-out: pending + executionId=null is an orphan
  pre-stamp (cascade-lock contention, batchEnqueueAndWait failure, etc.),
  treated as claimable so future dispatchers can re-stamp instead of skipping
  it as 'in-flight' forever. Matches pickNextEligibleGroupForRow's logic.
- On batchEnqueueAndWait failure, dispatcherStep now sweeps the orphan
  pre-stamps it wrote for the failed batch so the cells don't render Queued
  forever; the next user action picks them up cleanly.

Author: TheodoreSpeaks

apps/sim/lib/table/dispatcher.ts

@@ -392,8 +392,26 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
       logger.error(`[${dispatchId}] batch dispatch 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.
+      // Reset the orphan pre-stamps so the cells don't render "Queued" forever.
+      // Without this sweep, `pending + null executionId` rows stay until a user
+      // re-triggers the row. The classifyEligibility carve-out lets future
+      // dispatchers re-claim, but the current dispatch has already advanced
+      // its cursor past these rows. Deleting the pre-stamps lets the next
+      // explicit Run pick them up cleanly.
+      await Promise.allSettled(
+        pendingRuns.map((p) =>
+          db
+            .delete(tableRowExecutions)
+            .where(
+              and(
+                eq(tableRowExecutions.rowId, p.rowId),
+                eq(tableRowExecutions.groupId, p.groupId),
+                eq(tableRowExecutions.status, 'pending'),
+                sql`${tableRowExecutions.executionId} IS NULL`
+              )
+            )
+        )
+      )
     }
   }
 

apps/sim/lib/table/service.ts

@@ -1134,9 +1134,14 @@ export async function batchInsertRowsWithTx(
   }))
 
   void fireTableTrigger(data.tableId, table.name, 'insert', result, null, table.schema, requestId)
+  // Scope to the newly-inserted row ids so the dispatcher doesn't walk every
+  // row in the table. After the sidecar migration, all existing rows have
+  // zero entries → `mode:'new'`'s `NOT EXISTS` filter would otherwise include
+  // them, dispatching workflows on every row in a populated table.
   void runWorkflowColumn({
     tableId: table.id,
     workspaceId: table.workspaceId,
+    rowIds: result.map((r) => r.id),
     mode: 'new',
     isManualRun: false,
     requestId,

apps/sim/lib/table/workflow-columns.ts

@@ -72,14 +72,22 @@ export function classifyEligibility(
   if (group.autoRun === false && !isManualRun) return 'autoRun-off'
 
   const exec = row.executions?.[group.id]
-  if (isExecInFlight(exec)) return 'in-flight'
+  // Dispatcher pre-stamp orphans (`pending` + `executionId: null`) are
+  // placeholders left behind when a previous dispatcher loop wrote the stamp
+  // but no cell-task picked up (cascade-lock contention, trigger.dev queue
+  // failure, etc.). Treat them as claimable so a new dispatcher can re-enqueue
+  // — without this carve-out the row would render "Queued" forever. Matches
+  // the `pickNextEligibleGroupForRow` cascade-loop carve-out.
+  const isOrphanPreStamp = exec?.status === 'pending' && exec.executionId == null
+  if (!isOrphanPreStamp && isExecInFlight(exec)) return 'in-flight'
   const status = exec?.status
 
   // `mode: 'new'` is the auto-fire scope: only rows that have never been
   // attempted on this group run. Any pre-existing exec entry — completed,
   // cancelled, or error — keeps the cell sticky until the user manually
   // re-runs via "Run column" / "Run all rows" / "Run this row".
-  if (mode === 'new' && exec) return 'has-prior-attempt'
+  // Exception: orphan pre-stamps are claimable (handled above).
+  if (mode === 'new' && exec && !isOrphanPreStamp) return 'has-prior-attempt'
 
   const completedAndFilled = status === 'completed' && areOutputsFilled(group, row)
   if (!isManualRun && completedAndFilled) return 'completed-on-auto'
@@ -476,11 +484,27 @@ export async function runWorkflowColumn(opts: {
   if (targetGroups.length === 0) return { dispatchId: null }
   const targetGroupIds = targetGroups.map((g) => g.id)
 
-  // Wipe targeted output cols + executions[gid] before any cells fire so the
-  // user sees the column flip to empty/Pending instantly.
   const { bulkClearWorkflowGroupCells, insertDispatch, runDispatcherToCompletion } = await import(
     './dispatcher'
   )
+
+  // For table-wide manual runs (Run all rows / Run column), cancel any prior
+  // active dispatches AND their in-flight cells before clearing. Without this:
+  //  - Two dispatcher loops would walk overlapping rows and burn duplicate work.
+  //  - mode:'all' bulk-clear deletes in-flight sidecar rows without aborting
+  //    workers — those would keep writing into the wiped state.
+  // Row-scoped callers (dep-edit cascade in `updateRow`) already cancel their
+  // specific scope before calling here, so we skip to avoid cancelling
+  // unrelated dispatches on other rows. Auto-fire (`mode:'new'`) is harmless
+  // overlap-wise since the NOT EXISTS filter excludes already-attempted rows.
+  const cancelPriorRuns =
+    isManualRun && (!rowIds || rowIds.length === 0) && (mode === 'all' || mode === 'incomplete')
+  if (cancelPriorRuns) {
+    await cancelWorkflowGroupRuns(tableId, undefined, { groupIds: targetGroupIds })
+  }
+
+  // Wipe targeted output cols + executions[gid] before any cells fire so the
+  // user sees the column flip to empty/Pending instantly.
   await bulkClearWorkflowGroupCells({
     tableId,
     groups: targetGroups.map((g) => ({ id: g.id, outputs: g.outputs })),