fix(table): address remaining cursor/greptile review feedback

- Mothership update_row no longer double-dispatches. updateRow already fires
  the auto-cascade internally; the second `mode: 'incomplete'` call here
  raced with it and could bulk-clear sibling-group outputs.
- SSE dispatch events no longer dropped when the activeDispatches cache is
  cold. Seed an empty TableRunState if the initial fetch hasn't landed yet
  so the queued overlay doesn't lose the first dispatch event.
- batchUpdateRows now runs cancel+rerun for per-row in-flight downstream
  groups, mirroring updateRow. Without this, dep edits in a batch left
  running workflows reading stale upstream values.

Author: TheodoreSpeaks

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts

@@ -147,18 +147,25 @@ export function useTableEventStream({
     const applyDispatch = (event: Extract<TableEvent, { kind: 'dispatch' }>): void => {
       const { dispatchId, status, scope, cursor, mode, isManualRun } = event
       queryClient.setQueryData<TableRunState>(tableKeys.activeDispatches(tableId), (prev) => {
-        if (!prev) return prev
-        const list = prev.dispatches
+        // SSE may arrive before the initial fetch lands. Seed an empty
+        // run-state so the dispatch isn't dropped; counters are reconciled
+        // by the subsequent fetch / per-cell SSE events.
+        const base: TableRunState = prev ?? {
+          dispatches: [],
+          runningCellCount: 0,
+          runningByRowId: {},
+        }
+        const list = base.dispatches
         // Terminal states drop the dispatch from the overlay; client renders
         // the row's authoritative DB exec state from here.
         if (status === 'complete' || status === 'cancelled') {
           const filtered = list.filter((d) => d.id !== dispatchId)
-          return filtered.length === list.length ? prev : { ...prev, dispatches: filtered }
+          return filtered.length === list.length ? base : { ...base, dispatches: filtered }
         }
         if (scope === undefined || cursor === undefined || mode === undefined) {
           // Defensive: a legacy emit without the new fields can't drive the
           // overlay. Leave existing cache alone.
-          return prev
+          return base
         }
         const idx = list.findIndex((d) => d.id === dispatchId)
         const existing = idx === -1 ? undefined : list[idx]
@@ -174,10 +181,10 @@ export function useTableEventStream({
           cursor,
           scope,
         }
-        if (idx === -1) return { ...prev, dispatches: [...list, next] }
+        if (idx === -1) return { ...base, dispatches: [...list, next] }
         const merged = list.slice()
         merged[idx] = next
-        return { ...prev, dispatches: merged }
+        return { ...base, dispatches: merged }
       })
     }
 

apps/sim/lib/copilot/tools/server/table/user-table.ts

@@ -525,16 +525,11 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
             // doesn't, so the guard never trips here. Defensive narrowing.
             return { success: false, message: 'Row update was skipped' }
           }
-          void runWorkflowColumn({
-            tableId: args.tableId,
-            workspaceId,
-            rowIds: [updatedRow.id],
-            mode: 'incomplete',
-            isManualRun: false,
-            requestId,
-          }).catch((err) =>
-            logger.error(`[${requestId}] auto-dispatch (Mothership update_row) failed:`, err)
-          )
+          // Auto-dispatch for user edits is handled inside `updateRow`
+          // (mode: 'new' for newly-cleared groups + cancel+rerun for in-flight
+          // downstream groups). Firing a second mode: 'incomplete' dispatch
+          // here would race with the internal one AND bulk-clear sibling-group
+          // outputs (mode: 'incomplete' wipes terminal-state cells in scope).
 
           return {
             success: true,

apps/sim/lib/table/service.ts

@@ -2309,21 +2309,22 @@ export async function batchUpdateRows(
     mergedData: RowData
     mergedExecutions: RowExecutions
     executionsPatch?: Record<string, RowExecutionMetadata | null>
+    inFlightDownstreamGroups: string[]
   }> = []
   for (const update of data.updates) {
     const existing = existingMap.get(update.rowId)!
     const merged = { ...existing.data, ...update.data }
     // Auto-clear exec records for workflow output columns the user just
     // wiped AND downstream dep-changed terminal groups — same rationale as
-    // `updateRow`. In-flight downstream groups are dropped here (batch
-    // updates don't run the cancel+restart orchestration — those go through
-    // single-row `updateRow`).
-    const { executionsPatch: effectiveExecutionsPatch } = deriveExecClearsForDataPatch(
-      update.data,
-      table.schema,
-      existing.executions,
-      update.executionsPatch
-    )
+    // `updateRow`. Per-row in-flight downstream groups are surfaced so we
+    // can run the cancel+rerun orchestration after the batch commits.
+    const { executionsPatch: effectiveExecutionsPatch, inFlightDownstreamGroups } =
+      deriveExecClearsForDataPatch(
+        update.data,
+        table.schema,
+        existing.executions,
+        update.executionsPatch
+      )
     const mergedExecutions = applyExecutionsPatch(existing.executions, effectiveExecutionsPatch)
 
     const sizeValidation = validateRowSize(merged)
@@ -2341,6 +2342,7 @@ export async function batchUpdateRows(
       mergedData: merged,
       mergedExecutions,
       executionsPatch: effectiveExecutionsPatch,
+      inFlightDownstreamGroups,
     })
   }
 
@@ -2404,6 +2406,39 @@ export async function batchUpdateRows(
     table.schema,
     requestId
   )
+  // Per-row cancel+rerun for in-flight downstream groups whose deps just
+  // changed — same orchestration as single-row `updateRow`. Without this,
+  // batch updates would leave running workflows reading stale dep values.
+  // Each row needs its own cancel + manual-incomplete dispatch because
+  // `cancelWorkflowGroupRuns`'s `groupIds` filter is per-row.
+  const rowsWithInFlightDownstream = mergedUpdates.filter(
+    (u) => u.inFlightDownstreamGroups.length > 0
+  )
+  if (rowsWithInFlightDownstream.length > 0) {
+    void (async () => {
+      try {
+        for (const { rowId, inFlightDownstreamGroups } of rowsWithInFlightDownstream) {
+          await cancelWorkflowGroupRuns(data.tableId, rowId, {
+            groupIds: inFlightDownstreamGroups,
+          })
+          await runWorkflowColumn({
+            tableId: data.tableId,
+            workspaceId: data.workspaceId,
+            mode: 'incomplete',
+            isManualRun: true,
+            rowIds: [rowId],
+            groupIds: inFlightDownstreamGroups,
+            requestId,
+          })
+        }
+      } catch (err) {
+        logger.error(
+          `[${requestId}] cancel+rerun for in-flight downstream groups (batch) failed:`,
+          err
+        )
+      }
+    })()
+  }
   void runWorkflowColumn({
     tableId: table.id,
     workspaceId: table.workspaceId,