fix(table): X-running count from dispatch scope so reload matches live

The "X running" badge read countRunningCells (sidecar in-flight), but the
dispatcher only stamps one ~20-cell window at a time. During a 1000-row
Run-all the client optimistically showed ~1000 while a reload showed ~20 —
the sidecar never holds more than a window.

Derive the count from the active dispatches instead: rows in scope ahead of
the cursor × |groupIds| (exact for Run-all, upper bound for incomplete/new).
Both scope and cursor are persisted, so a reload computes the same number.

- countActiveRunCells (dispatcher.ts): dispatch-scope total, sidecar fallback
  when no dispatch is active. byRowId stays sidecar-based (the client overlay
  renders queued rows ahead of the cursor).
- Live: applyDispatch re-syncs the badge from the server on every dispatch
  event (one per window, after its cells finish + cursor advances), so the
  badge steps down per window and matches reload. applyCell no longer touches
  runningCellCount (still keeps runningByRowId live for the gutter).
- Optimistic on click: useRunColumn seeds the full run scope (totalCount ×
  groups) so the badge is right before the first window lands.

Author: TheodoreSpeaks

apps/sim/app/api/table/[tableId]/dispatches/route.ts

@@ -5,7 +5,7 @@ import { parseRequest } from '@/lib/api/server'
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { countRunningCells, listActiveDispatches } from '@/lib/table/dispatcher'
+import { countActiveRunCells, listActiveDispatches } from '@/lib/table/dispatcher'
 import { accessError, checkAccess } from '@/app/api/table/utils'
 
 const logger = createLogger('TableDispatchesAPI')
@@ -37,10 +37,8 @@ export const GET = withRouteHandler(async (request: NextRequest, { params }: Rou
     const result = await checkAccess(tableId, authResult.userId, 'read')
     if (!result.ok) return accessError(result, requestId, tableId)
 
-    const [rows, running] = await Promise.all([
-      listActiveDispatches(tableId),
-      countRunningCells(tableId),
-    ])
+    const rows = await listActiveDispatches(tableId)
+    const running = await countActiveRunCells(tableId, rows)
     const dispatches: ActiveDispatch[] = rows.map((r) => ({
       id: r.id,
       status: r.status as 'pending' | 'dispatching',

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

@@ -73,11 +73,11 @@ export function useTableEventStream({
     let lastEventId = loadPointer(tableId)
     let reconnectAttempt = 0
 
-    const updateRunStateCounters = (
-      rowId: string,
-      wasInFlight: boolean,
-      isInFlight: boolean
-    ): void => {
+    // Keeps the per-row gutter (`runningByRowId`) live between dispatch events.
+    // `runningCellCount` (the "X running" badge) is NOT touched here — it's the
+    // server's dispatch-scope count, seeded optimistically on click and
+    // re-synced by `applyDispatch` on every window, so live matches reload.
+    const updateRunningByRow = (rowId: string, wasInFlight: boolean, isInFlight: boolean): void => {
       if (wasInFlight === isInFlight) return
       const delta = isInFlight ? 1 : -1
       queryClient.setQueryData<TableRunState>(tableKeys.activeDispatches(tableId), (prev) => {
@@ -87,11 +87,7 @@ export function useTableEventStream({
         const nextByRow = { ...prev.runningByRowId }
         if (nextForRow === 0) delete nextByRow[rowId]
         else nextByRow[rowId] = nextForRow
-        return {
-          ...prev,
-          runningCellCount: Math.max(0, prev.runningCellCount + delta),
-          runningByRowId: nextByRow,
-        }
+        return { ...prev, runningByRowId: nextByRow }
       })
     }
 
@@ -145,11 +141,7 @@ export function useTableEventStream({
           queryKey: tableKeys.activeDispatches(tableId),
         })
       } else {
-        updateRunStateCounters(
-          rowId,
-          wasInFlight,
-          isExecInFlight({ status } as RowExecutionMetadata)
-        )
+        updateRunningByRow(rowId, wasInFlight, isExecInFlight({ status } as RowExecutionMetadata))
       }
     }
 
@@ -195,6 +187,11 @@ export function useTableEventStream({
         merged[idx] = next
         return { ...base, dispatches: merged }
       })
+      // The dispatcher emits this once per window (after the window's cells
+      // finish + the cursor advances) and on completion. Re-sync the
+      // dispatch-scope `runningCellCount` from the server so the badge steps
+      // down per window and matches a reload exactly.
+      void queryClient.invalidateQueries({ queryKey: tableKeys.activeDispatches(tableId) })
     }
 
     const handlePrune = (payload: PrunedEvent): void => {

apps/sim/hooks/queries/tables.ts

@@ -234,17 +234,40 @@ function countNewlyInFlight(before: RowExecutions, after: RowExecutions): number
   return n
 }
 
-/** Add optimistically-stamped cells to the run-state counter so the "X running"
- *  badge + per-row gutter Stop reflect them instantly (the optimistic stamp
- *  eats the dispatcher's `pending` SSE, so `applyCell` never bumps the count).
- *  Returns the prior snapshot for rollback, or `null` when nothing was bumped. */
+/** First-page `totalCount` from the rows infinite-query cache (or `null` when
+ *  rows haven't loaded yet). Lets a Run-all estimate the full-table cell count
+ *  even though only a page of rows is in memory. */
+function readTotalRowCount(
+  queryClient: ReturnType<typeof useQueryClient>,
+  tableId: string
+): number | null {
+  const entries = queryClient.getQueriesData<InfiniteData<TableRowsResponse, number>>({
+    queryKey: tableKeys.rowsRoot(tableId),
+    exact: false,
+  })
+  for (const [, data] of entries) {
+    const tc = data?.pages?.[0]?.totalCount
+    if (typeof tc === 'number') return tc
+  }
+  return null
+}
+
+/** Optimistically reflect a run on the "X running" badge + per-row gutter Stop
+ *  instantly (the optimistic stamp eats the dispatcher's `pending` SSE, so
+ *  `applyCell` never bumps the count, and the server's dispatch-scope count
+ *  isn't live until the first window). `stampedByRow` drives the per-row gutter
+ *  (loaded rows only); `cellCountDelta` is the badge delta — pass the full run
+ *  scope (rows × groups) for Run-all so it matches the server, or omit to use
+ *  the stamped total. Returns the prior snapshot for rollback. */
 function bumpRunState(
   queryClient: ReturnType<typeof useQueryClient>,
   tableId: string,
-  stampedByRow: Record<string, number>
+  stampedByRow: Record<string, number>,
+  cellCountDelta?: number
 ): { snapshot: TableRunState | undefined } | null {
-  const total = Object.values(stampedByRow).reduce((s, n) => s + n, 0)
-  if (total === 0) return null
+  const stampedTotal = Object.values(stampedByRow).reduce((s, n) => s + n, 0)
+  const countDelta = cellCountDelta ?? stampedTotal
+  if (countDelta === 0 && stampedTotal === 0) return null
   const snapshot = queryClient.getQueryData<TableRunState>(tableKeys.activeDispatches(tableId))
   queryClient.setQueryData<TableRunState>(tableKeys.activeDispatches(tableId), (prev) => {
     const base = prev ?? { dispatches: [], runningCellCount: 0, runningByRowId: {} }
@@ -254,7 +277,7 @@ function bumpRunState(
     }
     return {
       ...base,
-      runningCellCount: base.runningCellCount + total,
+      runningCellCount: base.runningCellCount + countDelta,
       runningByRowId: nextByRow,
     }
   })
@@ -1439,7 +1462,14 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) {
         return { ...r, data: nextData, executions: next }
       })
 
-      const bumped = bumpRunState(queryClient, tableId, stampedByRow)
+      // Badge counts the whole run scope (rows × groups), matching the server's
+      // dispatch-scope count — not just the loaded rows we could stamp. For
+      // Run-all that's the table's totalCount; for a scoped run, the rowIds.
+      const scopeRowCount = targetRowIds
+        ? targetRowIds.size
+        : (readTotalRowCount(queryClient, tableId) ?? Object.keys(stampedByRow).length)
+      const cellCountDelta = scopeRowCount * targetGroupIds.size
+      const bumped = bumpRunState(queryClient, tableId, stampedByRow, cellCountDelta)
       return { snapshots, runStateSnapshot: bumped?.snapshot, didBumpRunState: bumped !== null }
     },
     onError: (_err, _variables, context) => {

apps/sim/lib/table/dispatcher.ts

@@ -195,6 +195,39 @@ export async function countRunningCells(
   return { total, byRowId }
 }
 
+/** Authoritative "cells queued or running" count for the table, derived from
+ *  active dispatches so it survives reload and matches the live count. For each
+ *  active dispatch every row in scope ahead of the cursor still has to run each
+ *  targeted group, so remaining work = (rows ahead of cursor) × |groupIds|.
+ *  Exact for Run-all; an upper bound for incomplete/new (rows the eligibility
+ *  filter later skips are still counted). Falls back to the sidecar in-flight
+ *  count when no dispatch is active (orphan stragglers). `byRowId` stays
+ *  sidecar-based — the client overlay renders queued rows ahead of the cursor. */
+export async function countActiveRunCells(
+  tableId: string,
+  dispatches?: DispatchRow[]
+): Promise<{ total: number; byRowId: Record<string, number> }> {
+  const active = dispatches ?? (await listActiveDispatches(tableId))
+  const sidecar = await countRunningCells(tableId)
+  if (active.length === 0) return sidecar
+
+  let total = 0
+  for (const d of active) {
+    const groupCount = d.scope.groupIds.length
+    if (groupCount === 0) continue
+    const filters = [eq(userTableRows.tableId, tableId), gt(userTableRows.position, d.cursor)]
+    if (d.scope.rowIds && d.scope.rowIds.length > 0) {
+      filters.push(inArray(userTableRows.id, d.scope.rowIds))
+    }
+    const [row] = await db
+      .select({ rowsAhead: sql<number>`count(*)::int` })
+      .from(userTableRows)
+      .where(and(...filters))
+    total += (row?.rowsAhead ?? 0) * groupCount
+  }
+  return { total, byRowId: sidecar.byRowId }
+}
+
 export async function listActiveDispatches(tableId: string): Promise<DispatchRow[]> {
   const rows = await db
     .select()