fix(tables): reliable stop-all, accurate "X running", and rate/usage gating for cell runs (#4838)

* fix(tables): reliable stop-all, accurate "X running", and rate/usage gating for cell runs Stop-all: - Make the cancellation guard status-based (not executionId-scoped) so a `cancelled` tombstone stamped while a cell is still a dispatcher pre-stamp (null executionId) keeps the cell dead — fixes function-execute cells that resurrected after Stop all. Consolidated into shared isExecCancelled / isExecCancelledAfter predicates in deps.ts, reused by the in-memory guard, the SQL guard, the dispatcher tombstone filter, the worker, and resume. - Add an explicit pre-execution cancellation read so a cell that dequeues after Stop all (e.g. from the trigger.dev queue) never runs. - Resume worker aborts a cancelled paused/awaiting cell before resuming; cancelWorkflowGroupRuns marks paused executions cancelling. "X running": - Emit a dispatch SSE at dispatch start so auto-fired/capped runs surface immediately; show the control whenever a dispatch is active. Checkbox dependency: - Treat boolean `false` as an unmet dependency so unchecking never reruns dependents — only checking does. deriveExecClearsForDataPatch no longer re-arms a downstream group whose deps are unmet after the patch. Rate / usage gating: - Route table cell execution through preprocessExecution (billing actor = workspace billed account, usage limit, per-plan timeout), keeping draft. - Rate limit: pace & retry per cell (async counter) so rows aren't skipped. - Usage limit: halt the dispatch without marking cells and emit a usageLimitReached event; the client shows an Upgrade prompt that routes to subscription settings. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(tables): dedupe usage-limit event + release rate-limited cells on cancel Addresses PR review: - Usage limit: only the cell that transitions the dispatch active→complete (via completeDispatchIfActive) emits usageLimitReached, so concurrent cells don't fire up to 20 identical "upgrade" toasts. - Rate-limit retry: re-check the cancelled tombstone after each sleep so a Stop All mid-wait releases the concurrency slot promptly (signal never fires on the trigger.dev backend). * fix(tables): jitter rate-limit retry backoff to avoid thundering herd Passing the bucket's shared resetAt as retryAfterMs made backoffWithJitter return a fixed clamped value (no jitter, attempt ignored), so all concurrent cells retried in lockstep. Pass null to get jittered exponential backoff. * fix(tables): unstick cells + resync counter on usage-limit halt Addresses PR review: - Clear each blocked cell's pre-stamp on a 402 so it reverts to un-run instead of being stuck "Queued" (no error/cancelled badge); covers auto-fire cells with no owning dispatch. - Client re-syncs run-state counts and refetches rows on usageLimitReached so the stale "X running" / Stop-all control clears and queued cells drop. - Make usageLimitReached.dispatchId optional; client only touches the dispatch overlay when present. * fix(tables): don't emit stale dispatching event after a mid-window halt If a cell halts the dispatch mid-window (usage limit), re-read the dispatch status after the batch and bail instead of emitting a per-window 'dispatching' event that would arrive after the client dropped the dispatch and re-add it (flickering 'X running' back). --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-09-24 15:45:35 +08:00 · 2026-06-01 20:43:32 -04:00
parent ed19b0bcde
commit 3ccb3a392a
14 changed files with 616 additions and 66 deletions
@@ -95,6 +95,10 @@ export interface SelectionSnapshot {
  /** Total running/queued workflow runs across ALL rows. Drives the page-header
   *  RunStatusControl ("N running, Stop all"). */
  totalRunning: number
+  /** Whether any dispatch is active (pending/dispatching). Keeps the RunStatusControl
+   *  + Stop-all visible during a run even when the per-row count momentarily reads 0
+   *  (e.g. the first window of an auto-fired/capped dispatch before cells stamp). */
+  hasActiveDispatch: boolean
  /** Whether the table has any workflow-output columns (drives the Run/Stop visibility). */
  hasWorkflowColumns: boolean
  /** Cells the Play / Refresh / Stop buttons act on. Null when the selection
@@ -333,6 +337,7 @@ export function TableGrid({
  // rows still inside a dispatch's scope — e.g. a cascade where 3 of 4 columns
  // finished would read "4 running" instead of "1".
  const totalRunning = Object.values(runningByRowId).reduce((sum, n) => sum + n, 0)
+  const hasActiveDispatch = (activeDispatches?.length ?? 0) > 0

  const tableRowCountRef = useRef(tableData?.rowCount ?? 0)
  tableRowCountRef.current = tableData?.rowCount ?? 0
@@ -3194,6 +3199,7 @@ export function TableGrid({
      sameStats &&
      prev.runningInActionBarSelection === runningInActionBarSelection &&
      prev.totalRunning === totalRunning &&
+      prev.hasActiveDispatch === hasActiveDispatch &&
      prev.hasWorkflowColumns === hasWorkflowColumns &&
      prev.actionBarRowIds.length === actionBarRowIds.length &&
      prev.actionBarRowIds.every((id, i) => id === actionBarRowIds[i])
@@ -3204,6 +3210,7 @@ export function TableGrid({
      actionBarRowIds,
      runningInActionBarSelection,
      totalRunning,
+      hasActiveDispatch,
      hasWorkflowColumns,
      selectedRunScope,
      selectionStats,
@@ -3215,6 +3222,7 @@ export function TableGrid({
    actionBarRowIds,
    runningInActionBarSelection,
    totalRunning,
+    hasActiveDispatch,
    hasWorkflowColumns,
    selectedRunScope,
    selectionStats,
@@ -1,6 +1,6 @@
 'use client'

-import { useEffect } from 'react'
+import { useEffect, useRef } from 'react'
 import { createLogger } from '@sim/logger'
 import { useQueryClient } from '@tanstack/react-query'
 import type { ActiveDispatch } from '@/lib/api/contracts/tables'
@@ -44,6 +44,9 @@ interface UseTableEventStreamArgs {
  tableId: string | undefined
  workspaceId: string | undefined
  enabled?: boolean
+  /** Fired when the server halts a dispatch because the billed account is over
+   *  its usage limit. The page surfaces an upgrade prompt + redirect. */
+  onUsageLimitReached?: (event: { dispatchId?: string; message: string }) => void
 }

 /**
@@ -59,9 +62,14 @@ export function useTableEventStream({
  tableId,
  workspaceId,
  enabled = true,
+  onUsageLimitReached,
 }: UseTableEventStreamArgs): void {
  const queryClient = useQueryClient()

+  // Ref so a changing callback identity doesn't tear down + reconnect the SSE.
+  const onUsageLimitReachedRef = useRef(onUsageLimitReached)
+  onUsageLimitReachedRef.current = onUsageLimitReached
+
  useEffect(() => {
    if (!enabled || !tableId || !workspaceId) return

@@ -205,6 +213,28 @@ export function useTableEventStream({
      scheduleDispatchInvalidate()
    }

+    const applyUsageLimit = (event: Extract<TableEvent, { kind: 'usageLimitReached' }>): void => {
+      // Drop the halted dispatch from the overlay so the "running" UI clears
+      // immediately (the dispatcher was marked complete server-side). Cascade /
+      // auto-fire events carry no dispatchId — nothing to remove.
+      if (event.dispatchId) {
+        queryClient.setQueryData<TableRunState>(tableKeys.activeDispatches(tableId), (prev) => {
+          if (!prev) return prev
+          const filtered = prev.dispatches.filter((d) => d.id !== event.dispatchId)
+          return filtered.length === prev.dispatches.length
+            ? prev
+            : { ...prev, dispatches: filtered }
+        })
+      }
+      // Blocked cells are left `queued` in the DB with no terminal cell event,
+      // so `runningByRowId` would otherwise stay non-zero (stale "X running").
+      // Re-sync the server counts, and refetch rows so cells whose pre-stamps
+      // the server cleared drop their "Queued" state.
+      scheduleDispatchInvalidate()
+      void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) })
+      onUsageLimitReachedRef.current?.({ dispatchId: event.dispatchId, message: event.message })
+    }
+
    const handlePrune = (payload: PrunedEvent): void => {
      logger.info('Table event buffer pruned — full refetch', { tableId, ...payload })
      void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) })
@@ -253,6 +283,7 @@ export function useTableEventStream({
          savePointer(tableId, lastEventId)
          if (entry.event?.kind === 'cell') applyCell(entry.event)
          else if (entry.event?.kind === 'dispatch') applyDispatch(entry.event)
+          else if (entry.event?.kind === 'usageLimitReached') applyUsageLimit(entry.event)
        } catch (err) {
          logger.warn('Failed to parse table event', { tableId, err })
        }
@@ -36,6 +36,7 @@ import {
  useRunColumn,
 } from '@/hooks/queries/tables'
 import { useInlineRename } from '@/hooks/use-inline-rename'
+import { useSettingsNavigation } from '@/hooks/use-settings-navigation'
 import { useLogDetailsUIStore } from '@/stores/logs/store'
 import type { DeletedRowSnapshot } from '@/stores/table/types'
 import {
@@ -129,7 +130,15 @@ export function Table({
  const posthogRef = useRef(posthog)
  posthogRef.current = posthog

-  useTableEventStream({ tableId, workspaceId })
+  const { navigateToSettings } = useSettingsNavigation()
+  // Plain function: `useTableEventStream` keeps it in a ref (its effect doesn't
+  // depend on the identity), so a stable reference buys nothing here.
+  const onUsageLimitReached = ({ message }: { dispatchId?: string; message: string }) => {
+    toast.error(message, {
+      action: { label: 'Upgrade', onClick: () => navigateToSettings({ section: 'subscription' }) },
+    })
+  }
+  useTableEventStream({ tableId, workspaceId, onUsageLimitReached })

  const [slideout, dispatch] = useReducer(slideoutReducer, { kind: 'none' })
  const [showDeleteTableConfirm, setShowDeleteTableConfirm] = useState(false)
@@ -141,6 +150,7 @@ export function Table({
    actionBarRowIds: [],
    runningInActionBarSelection: 0,
    totalRunning: 0,
+    hasActiveDispatch: false,
    hasWorkflowColumns: false,
    selectedRunScope: null,
    selectionStats: { hasIncompleteOrFailed: false, hasCompleted: false, hasInFlight: false },
@@ -509,7 +519,7 @@ export function Table({
          createTrigger={createTrigger}
          actions={headerActions}
          leadingActions={
-            selection.totalRunning > 0 ? (
+            selection.totalRunning > 0 || selection.hasActiveDispatch ? (
              <RunStatusControl
                running={selection.totalRunning}
                onStopAll={onStopAll}
@@ -527,7 +537,7 @@ export function Table({
        onFilterToggle={() => setFilterOpen((prev) => !prev)}
        filterActive={filterOpen || !!queryOptions.filter}
        trailing={
-          embedded && selection.totalRunning > 0 ? (
+          embedded && (selection.totalRunning > 0 || selection.hasActiveDispatch) ? (
            <RunStatusControl
              running={selection.totalRunning}
              onStopAll={onStopAll}
@@ -3,6 +3,7 @@ import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { task } from '@trigger.dev/sdk'
 import { withCascadeLock } from '@/lib/table/cascade-lock'
+import { isExecCancelled } from '@/lib/table/deps'
 import type { RowData, RowExecutionMetadata } from '@/lib/table/types'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'

@@ -44,6 +45,36 @@ export async function executeResumeJob(payload: ResumeExecutionPayload) {
    const { findCellContextByExecutionId } = await import('@/lib/table/workflow-columns')
    const cellContext = await findCellContextByExecutionId(parentExecutionId)

+    // A paused/awaiting table cell that was cancelled by "Stop all" must not
+    // resume — the cancel write is authoritative (matches the cell-write guard
+    // philosophy). Aborting here also stops the wasted compute the guard alone
+    // can't prevent. Read the cell's current exec and bail if cancelled.
+    if (cellContext) {
+      const { getRowById } = await import('@/lib/table/service')
+      const cellRow = await getRowById(
+        cellContext.tableId,
+        cellContext.rowId,
+        cellContext.workspaceId
+      )
+      if (isExecCancelled(cellRow?.executions?.[cellContext.groupId])) {
+        logger.info('Skipping resume — table cell cancelled', {
+          tableId: cellContext.tableId,
+          rowId: cellContext.rowId,
+          groupId: cellContext.groupId,
+          parentExecutionId,
+        })
+        return {
+          success: false,
+          workflowId,
+          executionId: resumeExecutionId,
+          parentExecutionId,
+          status: 'cancelled' as const,
+          output: undefined,
+          executedAt: new Date().toISOString(),
+        }
+      }
+    }
+
    const writers = cellContext
      ? await buildResumeCellWriters(cellContext, parentExecutionId)
      : null
@@ -2,10 +2,17 @@ import { db } from '@sim/db'
 import { workflow as workflowTable } from '@sim/db/schema'
 import { createLogger, runWithRequestContext } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
+import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
+import { backoffWithJitter } from '@sim/utils/retry'
 import { task } from '@trigger.dev/sdk'
 import { eq } from 'drizzle-orm'
+import { createTimeoutAbortController } from '@/lib/core/execution-limits'
+import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter'
+import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { withCascadeLock } from '@/lib/table/cascade-lock'
+import { isExecCancelled } from '@/lib/table/deps'
+import { appendTableEvent } from '@/lib/table/events'
 import type {
  RowData,
  RowExecutionMetadata,
@@ -18,6 +25,12 @@ export type { WorkflowGroupCellPayload }

 const logger = createLogger('TriggerWorkflowGroupCell')

+/** Max rate-limit retry attempts per cell before giving up and writing a
+ *  re-runnable error. With `backoffWithJitter` (base 500ms, max 30s) this is
+ *  ~1–2 minutes of pacing — enough to ride out a transient burst without
+ *  stalling the dispatcher window indefinitely. */
+const RATE_LIMIT_MAX_ATTEMPTS = 6
+
 /** Cell-task entrypoint. Holds a per-row cascade lock so only one worker
 *  advances a given row at a time; bails on contention. The held lock heart-
 *  beats every 10s so a crashed pod releases within ~30s.
@@ -50,6 +63,9 @@ export async function executeWorkflowGroupCellJob(
      )
      break
    }
+    // Usage limit hit mid-cascade: the dispatch is halted and no cell was
+    // marked, so stop re-driving this row.
+    if (outcome.result === 'blocked') break
    if (signal?.aborted) break
    const freshTable = await getTableById(tableId)
    if (!freshTable) break
@@ -83,7 +99,7 @@ export async function executeWorkflowGroupCellJob(
 export async function runRowCascadeLoop(
  payload: WorkflowGroupCellPayload,
  signal?: AbortSignal
-): Promise<void> {
+): Promise<'blocked' | undefined> {
  const { tableId, rowId, workspaceId } = payload
  const { getTableById, getRowById } = await import('@/lib/table/service')
  const { pickNextEligibleGroupForRow } = await import('@/lib/table/workflow-columns')
@@ -121,6 +137,10 @@ export async function runRowCascadeLoop(
    )

    if (result === 'paused') break
+    // Hard stop (e.g. usage limit): the dispatch was halted and no cell was
+    // marked. Propagate so the outer re-drive loop stops too — otherwise it
+    // would re-pick the still-pending queued marker and spin.
+    if (result === 'blocked') return 'blocked'

    const freshRow = await getRowById(tableId, rowId, workspaceId)
    if (!freshRow) break
@@ -130,17 +150,20 @@ export async function runRowCascadeLoop(
    currentWorkflowId = next.workflowId
    currentExecutionId = generateId()
  }
+  return undefined
 }

 /** Returns `'paused'` to signal the cascade loop must exit (resume worker
- *  takes over). `'completed' | 'error'` keep the loop running. */
+ *  takes over) and `'blocked'` for a hard stop (usage limit — dispatch halted,
+ *  cell left unmarked). `'completed' | 'error'` keep the loop running. */
 async function runWorkflowAndWriteTerminal(
  payload: WorkflowGroupCellPayload,
  signal: AbortSignal | undefined,
  table: TableDefinition,
  group: WorkflowGroup
-): Promise<'completed' | 'error' | 'paused'> {
-  const { tableId, tableName, rowId, groupId, workflowId, workspaceId, executionId } = payload
+): Promise<'completed' | 'error' | 'paused' | 'blocked'> {
+  const { tableId, tableName, rowId, groupId, workflowId, workspaceId, executionId, dispatchId } =
+    payload
  const requestId = `wfgrp-${executionId}`

  return runWithRequestContext({ requestId }, async () => {
@@ -155,6 +178,17 @@ async function runWorkflowAndWriteTerminal(
    const writeState = (executionState: RowExecutionMetadata, dataPatch?: RowData) =>
      writeWorkflowGroupState(cellCtx, { executionState, dataPatch })

+    /** Pre-execution cancellation guard: a cell cancelled while it sat in the
+     *  queue (e.g. trigger.dev concurrency backlog) must not run once it
+     *  dequeues. Reads the already-loaded row's exec — no extra query. */
+    const cancelledBeforeRun = (exec: RowExecutionMetadata | undefined): boolean => {
+      if (!isExecCancelled(exec)) return false
+      logger.info(
+        `Skipping cell — cancelled before execution (table=${tableId} row=${rowId} group=${groupId})`
+      )
+      return true
+    }
+
    // Enrichment groups call a registry function directly instead of running a
    // workflow, reusing the same pickup → run → terminal-write status flow. The
    // `enrichmentId` guard ensures only true registry enrichments take this path
@@ -184,6 +218,8 @@ async function runWorkflowAndWriteTerminal(
        return 'error'
      }

+      if (cancelledBeforeRun(row.executions?.[groupId])) return 'error'
+
      const pickedUp = await markWorkflowGroupPickedUp(cellCtx, {
        workflowId: statusId,
        jobId: null,
@@ -366,8 +402,125 @@ async function runWorkflowAndWriteTerminal(
        return 'error'
      }

-      // SQL guard rejects if a stop click stamped `cancelled` between enqueue
-      // and pickup.
+      if (cancelledBeforeRun(row.executions?.[groupId])) return 'error'
+
+      // Billing / usage / timeout gate — route table cells through the same
+      // preprocessing every other trigger uses. Keep running draft
+      // (checkDeployment: false). Rate limiting is paced separately below so a
+      // retry doesn't re-run the (stable) billing/usage/subscription lookups.
+      // Failures are surfaced via cell state / SSE / dispatch halt, so suppress
+      // preprocessing's own execution-log writes.
+      const preprocess = await preprocessExecution({
+        workflowId,
+        executionId,
+        requestId,
+        workspaceId,
+        workflowRecord,
+        userId: workflowRecord.userId,
+        triggerType: 'workflow',
+        checkDeployment: false,
+        checkRateLimit: false,
+        logPreprocessingErrors: false,
+      })
+      if (!preprocess.success) {
+        // Usage/quota exhausted: retrying won't help. Halt the dispatch without
+        // marking any cell, and signal the client to upgrade.
+        if (preprocess.error?.statusCode === 402) {
+          logger.warn(
+            `Usage limit reached — halting dispatch (table=${tableId} row=${rowId} group=${groupId})`
+          )
+          // Don't leave the cell stuck on its `pending` pre-stamp. Clear this
+          // cell's exec so it reverts to un-run (no error/cancelled badge —
+          // matching "don't mark"; re-runnable after upgrade). Each blocked
+          // cell clears its own.
+          const { updateRow } = await import('@/lib/table/service')
+          await updateRow(
+            { tableId, rowId, data: {}, workspaceId, executionsPatch: { [groupId]: null } },
+            table,
+            requestId
+          ).catch((err) =>
+            logger.warn(`Failed to clear cell pre-stamp on usage limit`, {
+              error: toError(err).message,
+            })
+          )
+          // With up to 20 concurrent cells all hitting the limit at once, only
+          // the cell that transitions the dispatch active→complete emits the
+          // event — otherwise the user sees a toast per in-flight cell. Cells
+          // with no owning dispatch (auto-fire) always emit.
+          let shouldEmit = true
+          if (dispatchId) {
+            const { completeDispatchIfActive } = await import('@/lib/table/dispatcher')
+            shouldEmit = await completeDispatchIfActive(dispatchId)
+          }
+          if (shouldEmit) {
+            await appendTableEvent({
+              kind: 'usageLimitReached',
+              tableId,
+              ...(dispatchId ? { dispatchId } : {}),
+              message:
+                preprocess.error?.message ??
+                'Usage limit exceeded. Please upgrade your plan to continue.',
+            })
+          }
+          return 'blocked'
+        }
+        await writeState({
+          status: 'error',
+          executionId,
+          jobId: null,
+          workflowId,
+          error: preprocess.error?.message ?? 'Workflow could not start',
+        })
+        return 'error'
+      }
+
+      const actorUserId = preprocess.actorUserId ?? workflowRecord.userId
+      const asyncTimeoutMs = preprocess.executionTimeout?.async
+
+      // Rate-limit pacing: tables count against the async counter (background
+      // jobs). On a hit, wait & retry so the row still runs rather than being
+      // skipped — only this cheap check repeats. The waiting cell holds its
+      // concurrency slot, pacing the whole dispatch to the user's rate limit.
+      const rateLimiter = new RateLimiter()
+      for (let attempt = 1; ; attempt++) {
+        if (signal?.aborted) return 'error'
+        const rl = await rateLimiter.checkRateLimitWithSubscription(
+          actorUserId,
+          preprocess.userSubscription ?? null,
+          'workflow',
+          true
+        )
+        if (rl.allowed) break
+        if (attempt >= RATE_LIMIT_MAX_ATTEMPTS) {
+          await writeState({
+            status: 'error',
+            executionId,
+            jobId: null,
+            workflowId,
+            error: 'Rate limit exceeded — please retry later',
+          })
+          return 'error'
+        }
+        // Exponential backoff WITH jitter — pass null, not the bucket's
+        // resetAt. That reset time is shared across all waiters, and
+        // backoffWithJitter clamps a non-null hint to a fixed value with no
+        // jitter, so honoring it would wake all ~20 concurrent cells in
+        // lockstep and stampede the bucket. Jittered backoff spreads retries.
+        const waitMs = backoffWithJitter(attempt, null)
+        logger.info(
+          `Rate limited — waiting ${Math.round(waitMs)}ms before retry ${attempt + 1} (table=${tableId} row=${rowId} group=${groupId})`
+        )
+        await sleep(waitMs)
+        // Stop All can land mid-wait. On the trigger.dev backend `signal` never
+        // fires (cancelByKey is a no-op there), so re-check the DB tombstone and
+        // release this concurrency slot promptly instead of sleeping out the
+        // full retry budget.
+        const refreshed = await getRowById(tableId, rowId, workspaceId)
+        if (!refreshed || cancelledBeforeRun(refreshed.executions?.[groupId])) return 'error'
+      }
+
+      // SQL guard also rejects if a stop click stamped `cancelled` between this
+      // check and pickup.
      const pickedUp = await markWorkflowGroupPickedUp(cellCtx, {
        workflowId,
        jobId: null,
@@ -485,28 +638,42 @@ async function runWorkflowAndWriteTerminal(
        schedulePartialWrite()
      }

-      const result = await executeWorkflow(
-        {
-          id: workflowRecord.id,
-          userId: workflowRecord.userId,
-          workspaceId: workflowRecord.workspaceId,
-          variables: (workflowRecord.variables as Record<string, unknown> | null) ?? {},
-        },
-        requestId,
-        input,
-        workflowRecord.userId,
-        {
-          enabled: true,
-          executionMode: 'sync',
-          workflowTriggerType: 'table',
-          triggerBlockId: startBlock.id,
-          useDraftState: true,
-          abortSignal: signal,
-          onBlockStart,
-          onBlockComplete,
-        },
-        executionId
-      )
+      // Enforce the per-plan execution timeout (from preprocessing), combined
+      // with the existing cancel signal so either a timeout or a Stop aborts.
+      const timeoutController = createTimeoutAbortController(asyncTimeoutMs)
+      const abortSignal = signal
+        ? AbortSignal.any([signal, timeoutController.signal])
+        : timeoutController.signal
+
+      let result: Awaited<ReturnType<typeof executeWorkflow>>
+      try {
+        result = await executeWorkflow(
+          {
+            id: workflowRecord.id,
+            // Workflow owner — drives personal env-var resolution + ownership.
+            userId: workflowRecord.userId,
+            workspaceId: workflowRecord.workspaceId,
+            variables: (workflowRecord.variables as Record<string, unknown> | null) ?? {},
+          },
+          requestId,
+          input,
+          // Billing/usage/rate actor — the workspace billed account.
+          actorUserId,
+          {
+            enabled: true,
+            executionMode: 'sync',
+            workflowTriggerType: 'table',
+            triggerBlockId: startBlock.id,
+            useDraftState: true,
+            abortSignal,
+            onBlockStart,
+            onBlockComplete,
+          },
+          executionId
+        )
+      } finally {
+        timeoutController.cleanup()
+      }

      terminalWritten = true
      await writeChain.catch(() => {})
@@ -3,10 +3,11 @@
 */

 import { loggingSessionMock } from '@sim/testing'
-import { describe, expect, it, vi } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'

-const { mockGetWorkspaceBilledAccountUserId } = vi.hoisted(() => ({
+const { mockGetWorkspaceBilledAccountUserId, mockCheckRateLimit } = vi.hoisted(() => ({
  mockGetWorkspaceBilledAccountUserId: vi.fn(),
+  mockCheckRateLimit: vi.fn(),
 }))

 vi.mock('@sim/db', () => ({ db: {} }))
@@ -21,7 +22,7 @@ vi.mock('@/lib/core/execution-limits', () => ({
  getExecutionTimeout: vi.fn(() => 0),
 }))
 vi.mock('@/lib/core/rate-limiter/rate-limiter', () => ({
-  RateLimiter: vi.fn(),
+  RateLimiter: vi.fn(() => ({ checkRateLimitWithSubscription: mockCheckRateLimit })),
 }))
 vi.mock('@/lib/logs/execution/logging-session', () => loggingSessionMock)
 vi.mock('@/lib/workspaces/utils', () => ({
@@ -36,6 +37,8 @@ vi.mock('@sim/workflow-authz', () => ({
  }),
 }))

+import { checkServerSideUsageLimits } from '@/lib/billing/calculations/usage-monitor'
+import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
 import { preprocessExecution } from './preprocessing'

 describe('preprocessExecution correlation logging', () => {
@@ -88,3 +91,56 @@ describe('preprocessExecution correlation logging', () => {
    })
  })
 })
+
+describe('preprocessExecution logPreprocessingErrors option', () => {
+  const baseOptions = {
+    workflowId: 'workflow-1',
+    userId: 'owner-1',
+    triggerType: 'workflow' as const,
+    executionId: 'execution-1',
+    requestId: 'request-1',
+    checkDeployment: false,
+    checkRateLimit: true,
+    workflowRecord: { id: 'workflow-1', workspaceId: 'workspace-1', isDeployed: false } as any,
+  }
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockGetWorkspaceBilledAccountUserId.mockResolvedValue('billed-account-1')
+    vi.mocked(getHighestPrioritySubscription).mockResolvedValue({ plan: 'free' } as any)
+    vi.mocked(checkServerSideUsageLimits).mockResolvedValue({
+      isExceeded: false,
+      currentUsage: 1,
+      limit: 10,
+    } as any)
+    mockCheckRateLimit.mockResolvedValue({
+      allowed: true,
+      remaining: 100,
+      resetAt: new Date(),
+    })
+  })
+
+  it('suppresses preprocessing-error logging when logPreprocessingErrors is false', async () => {
+    vi.mocked(checkServerSideUsageLimits).mockResolvedValueOnce({
+      isExceeded: true,
+      currentUsage: 20,
+      limit: 10,
+      message: 'Usage limit exceeded. Please upgrade your plan to continue.',
+    } as any)
+
+    const loggingSession = {
+      safeStart: vi.fn().mockResolvedValue(true),
+      safeCompleteWithError: vi.fn().mockResolvedValue(undefined),
+    }
+
+    const result = await preprocessExecution({
+      ...baseOptions,
+      logPreprocessingErrors: false,
+      loggingSession: loggingSession as any,
+    })
+
+    expect(result).toMatchObject({ success: false, error: { statusCode: 402 } })
+    // No execution-log row written — the caller (table cell) surfaces it instead.
+    expect(loggingSession.safeStart).not.toHaveBeenCalled()
+  })
+})
@@ -35,6 +35,7 @@ export interface PreprocessExecutionOptions {
  checkRateLimit?: boolean // Default: false for manual/chat, true for others
  checkDeployment?: boolean // Default: true for non-manual triggers
  skipUsageLimits?: boolean // Default: false (only use for test mode)
+  logPreprocessingErrors?: boolean // Default: true. When false, skip writing workflow_execution_logs error rows (caller surfaces failures itself, e.g. table cells)

  // Context information
  workspaceId?: string // If known, used for billing resolution
@@ -89,6 +90,7 @@ export async function preprocessExecution(
    checkRateLimit = triggerType !== 'manual' && triggerType !== 'chat',
    checkDeployment = triggerType !== 'manual',
    skipUsageLimits = false,
+    logPreprocessingErrors = true,
    workspaceId: providedWorkspaceId,
    loggingSession: providedLoggingSession,
    triggerData,
@@ -97,6 +99,11 @@ export async function preprocessExecution(
    workflowRecord: prefetchedWorkflowRecord,
  } = options

+  // When `logPreprocessingErrors` is false the caller surfaces failures itself
+  // (e.g. table cells use cell state / SSE), so skip the execution-log writes.
+  const recordPreprocessingError: typeof logPreprocessingError = (args) =>
+    logPreprocessingErrors ? logPreprocessingError(args) : Promise.resolve()
+
  logger.info(`[${requestId}] Starting execution preprocessing`, {
    workflowId,
    userId,
@@ -122,7 +129,7 @@ export async function preprocessExecution(
      if (!workflowRecord) {
        logger.warn(`[${requestId}] Workflow not found: ${workflowId}`)

-        await logPreprocessingError({
+        await recordPreprocessingError({
          workflowId,
          executionId,
          triggerType,
@@ -147,7 +154,7 @@ export async function preprocessExecution(
    } catch (error) {
      logger.error(`[${requestId}] Error fetching workflow`, { error, workflowId })

-      await logPreprocessingError({
+      await recordPreprocessingError({
        workflowId,
        executionId,
        triggerType,
@@ -253,7 +260,7 @@ export async function preprocessExecution(
        workspaceId,
      })

-      await logPreprocessingError({
+      await recordPreprocessingError({
        workflowId,
        executionId,
        triggerType,
@@ -277,7 +284,7 @@ export async function preprocessExecution(
  } catch (error) {
    logger.error(`[${requestId}] Error resolving billing actor`, { error, workflowId })
    const fallbackUserId = userId || 'unknown'
-    await logPreprocessingError({
+    await recordPreprocessingError({
      workflowId,
      executionId,
      triggerType,
@@ -319,7 +326,7 @@ export async function preprocessExecution(
          }
        )

-        await logPreprocessingError({
+        await recordPreprocessingError({
          workflowId,
          executionId,
          triggerType,
@@ -349,7 +356,7 @@ export async function preprocessExecution(
        actorUserId,
      })

-      await logPreprocessingError({
+      await recordPreprocessingError({
        workflowId,
        executionId,
        triggerType,
@@ -395,7 +402,7 @@ export async function preprocessExecution(
          resetAt: rateLimitInfo.resetAt,
        })

-        await logPreprocessingError({
+        await recordPreprocessingError({
          workflowId,
          executionId,
          triggerType,
@@ -419,7 +426,7 @@ export async function preprocessExecution(
    } catch (error) {
      logger.error(`[${requestId}] Error checking rate limits`, { error, actorUserId })

-      await logPreprocessingError({
+      await recordPreprocessingError({
        workflowId,
        executionId,
        triggerType,
@@ -10,6 +10,7 @@
 */

 import { createLogger } from '@sim/logger'
+import { isExecCancelled } from '@/lib/table/deps'
 import { appendTableEvent } from '@/lib/table/events'
 import type { RowData, RowExecutionMetadata, RowExecutions, WorkflowGroup } from '@/lib/table/types'

@@ -80,11 +81,13 @@ export async function writeWorkflowGroupState(
    )
    return 'skipped'
  }
-  if (
-    current?.status === 'cancelled' &&
-    current.executionId === executionId &&
-    payload.executionState.status !== 'cancelled'
-  ) {
+  // A `cancelled` cell rejects any worker write regardless of executionId — a
+  // stop click can only stamp the dispatcher pre-stamp's executionId (often
+  // null), so an executionId-matched guard would let the worker that later
+  // claims the cell with its real id resurrect it. `bypassStaleWorker` (a fresh
+  // `queued` claim from a new dispatch, or the authoritative cancel write
+  // itself) still passes; manual re-runs clear the tombstone before stamping.
+  if (!bypassStaleWorker && isExecCancelled(current)) {
    logger.info(
      `Skipping group write — cancelled (table=${tableId} row=${rowId} group=${groupId} executionId=${executionId})`
    )
@@ -0,0 +1,116 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import {
+  areGroupDepsSatisfied,
+  getUnmetGroupDeps,
+  isExecCancelled,
+  isExecCancelledAfter,
+  optimisticallyScheduleNewlyEligibleGroups,
+} from '@/lib/table/deps'
+import type { RowExecutionMetadata, TableRow, WorkflowGroup } from '@/lib/table/types'
+
+function makeGroup(overrides: Partial<WorkflowGroup> & { id: string }): WorkflowGroup {
+  return {
+    workflowId: `wf-${overrides.id}`,
+    outputs: [{ blockId: 'b1', path: 'out', columnName: `${overrides.id}_out` }],
+    ...overrides,
+  }
+}
+
+function makeRow(
+  data: Record<string, unknown> = {},
+  executions: Record<string, RowExecutionMetadata> = {}
+): TableRow {
+  return {
+    id: 'row1',
+    data: data as TableRow['data'],
+    executions,
+    position: 0,
+    createdAt: new Date(),
+    updatedAt: new Date(),
+  }
+}
+
+describe('areGroupDepsSatisfied — checkbox dependency', () => {
+  const group = makeGroup({ id: 'g1', dependencies: { columns: ['flag'] } })
+
+  it('treats a checked box (true) as satisfied', () => {
+    expect(areGroupDepsSatisfied(group, makeRow({ flag: true }))).toBe(true)
+  })
+
+  it('treats an unchecked box (false) as unmet', () => {
+    expect(areGroupDepsSatisfied(group, makeRow({ flag: false }))).toBe(false)
+  })
+
+  it('treats empty / null / undefined as unmet', () => {
+    expect(areGroupDepsSatisfied(group, makeRow({ flag: '' }))).toBe(false)
+    expect(areGroupDepsSatisfied(group, makeRow({ flag: null }))).toBe(false)
+    expect(areGroupDepsSatisfied(group, makeRow({}))).toBe(false)
+  })
+
+  it('reports an unchecked box in unmet deps', () => {
+    expect(getUnmetGroupDeps(group, makeRow({ flag: false })).columns).toEqual(['flag'])
+    expect(getUnmetGroupDeps(group, makeRow({ flag: true })).columns).toEqual([])
+  })
+})
+
+describe('optimisticallyScheduleNewlyEligibleGroups — checkbox toggle', () => {
+  const group = makeGroup({ id: 'g1', autoRun: true, dependencies: { columns: ['flag'] } })
+
+  it('flips the dependent to pending when checking (false → true)', () => {
+    const before = makeRow({ flag: false })
+    const next = optimisticallyScheduleNewlyEligibleGroups([group], before, { flag: true })
+    expect(next?.g1?.status).toBe('pending')
+  })
+
+  it('does NOT schedule anything when unchecking (true → false)', () => {
+    const before = makeRow({ flag: true }, { g1: completedExec('wf-g1') })
+    const next = optimisticallyScheduleNewlyEligibleGroups([group], before, { flag: false })
+    expect(next).toBeNull()
+  })
+})
+
+function completedExec(workflowId: string): RowExecutionMetadata {
+  return { status: 'completed', executionId: 'e1', jobId: null, workflowId, error: null }
+}
+
+describe('isExecCancelled', () => {
+  it('is true only for cancelled status', () => {
+    expect(isExecCancelled({ status: 'cancelled' } as RowExecutionMetadata)).toBe(true)
+    expect(isExecCancelled({ status: 'running' } as RowExecutionMetadata)).toBe(false)
+    expect(isExecCancelled(undefined)).toBe(false)
+  })
+
+  it('is true regardless of executionId — the resurrection-bug guard', () => {
+    // A stop click can only stamp the pre-stamp's (often null) executionId.
+    expect(
+      isExecCancelled({ status: 'cancelled', executionId: null } as RowExecutionMetadata)
+    ).toBe(true)
+  })
+})
+
+describe('isExecCancelledAfter — dispatcher tombstone', () => {
+  const since = new Date('2026-01-01T00:00:00Z')
+
+  it('is true when cancelled after the dispatch was requested', () => {
+    const exec = {
+      status: 'cancelled',
+      cancelledAt: '2026-01-01T00:00:05Z',
+    } as RowExecutionMetadata
+    expect(isExecCancelledAfter(exec, since)).toBe(true)
+  })
+
+  it('is false for a cancel that predates the dispatch (a prior, cleared run)', () => {
+    const exec = {
+      status: 'cancelled',
+      cancelledAt: '2025-12-31T23:59:59Z',
+    } as RowExecutionMetadata
+    expect(isExecCancelledAfter(exec, since)).toBe(false)
+  })
+
+  it('is false without a cancelledAt timestamp', () => {
+    expect(isExecCancelledAfter({ status: 'cancelled' } as RowExecutionMetadata, since)).toBe(false)
+  })
+})
@@ -24,6 +24,37 @@ export function isExecInFlight(exec: RowExecutionMetadata | undefined): boolean
  return s === 'queued' || s === 'running' || s === 'pending'
 }

+/**
+ * A cell run the user/stop killed. The single source of truth for "do not run /
+ * do not write this cell" — used by the in-memory write guard, the worker's
+ * pre-execution check, and the resume worker. The SQL guard in
+ * `writeExecutionsPatch` mirrors this status test in its `WHERE`.
+ */
+export function isExecCancelled(exec: RowExecutionMetadata | undefined): boolean {
+  return exec?.status === 'cancelled'
+}
+
+/**
+ * Cancelled AND killed after `since`. The dispatcher's tombstone test: a cell
+ * cancelled after a dispatch was requested must be skipped by that dispatch's
+ * later windows, even though the dispatcher pre-stamped it before the stop.
+ */
+export function isExecCancelledAfter(exec: RowExecutionMetadata | undefined, since: Date): boolean {
+  if (!isExecCancelled(exec) || !exec?.cancelledAt) return false
+  const at = Date.parse(exec.cancelledAt)
+  return Number.isFinite(at) && at > since.getTime()
+}
+
+/**
+ * A dependency column counts as unmet when its value is empty OR explicitly
+ * `false`. An unchecked checkbox is treated as "dependency not satisfied", so
+ * only checking a box (false→true) makes dependents eligible — unchecking
+ * (true→false) never triggers a rerun.
+ */
+function isDepValueUnmet(value: unknown): boolean {
+  return value === null || value === undefined || value === '' || value === false
+}
+
 /**
 * True when every output column the group writes still has a non-empty value
 * on this row. The "completed" exec status is metadata, but the cells are the
@@ -47,8 +78,7 @@ export function areOutputsFilled(group: WorkflowGroup, row: TableRow): boolean {
 export function areGroupDepsSatisfied(group: WorkflowGroup, row: TableRow): boolean {
  const cols = group.dependencies?.columns ?? []
  for (const colName of cols) {
-    const value = row.data[colName]
-    if (value === null || value === undefined || value === '') return false
+    if (isDepValueUnmet(row.data[colName])) return false
  }
  return true
 }
@@ -66,8 +96,7 @@ export function getUnmetGroupDeps(group: WorkflowGroup, row: TableRow): UnmetDep
  const cols = group.dependencies?.columns ?? []
  const columns: string[] = []
  for (const colName of cols) {
-    const value = row.data[colName]
-    if (value === null || value === undefined || value === '') columns.push(colName)
+    if (isDepValueUnmet(row.data[colName])) columns.push(colName)
  }
  return { columns }
 }
@@ -6,6 +6,7 @@ import { generateId } from '@sim/utils/id'
 import { and, asc, eq, gt, inArray, isNotNull, ne, or, type SQL, sql } from 'drizzle-orm'
 import { getJobQueue } from '@/lib/core/async-jobs/config'
 import { writeWorkflowGroupState } from '@/lib/table/cell-write'
+import { isExecCancelledAfter } from '@/lib/table/deps'
 import { appendTableEvent } from '@/lib/table/events'
 import type { RowExecutionMetadata, RowExecutions, TableRow } from '@/lib/table/types'
 import {
@@ -359,6 +360,22 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
      .update(tableRunDispatches)
      .set({ status: 'dispatching' })
      .where(eq(tableRunDispatches.id, dispatchId))
+    // Announce the dispatch the moment it starts — before the first window's
+    // cells finish. Without this, auto-fired and capped dispatches (no client-
+    // side optimistic seed) emit their first `dispatch` event only after window
+    // 1 completes, so the "X running" / Stop-all control stays hidden while a
+    // long first window runs. The client refetches the run-state count on this.
+    await appendTableEvent({
+      kind: 'dispatch',
+      tableId: dispatch.tableId,
+      dispatchId,
+      status: 'dispatching',
+      scope: dispatch.scope,
+      cursor: dispatch.cursor,
+      mode: dispatch.mode,
+      isManualRun: dispatch.isManualRun,
+      ...(dispatch.limit ? { limit: dispatch.limit } : {}),
+    })
  }

  const filters = [
@@ -444,12 +461,9 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
  const tombstoneFiltered: TableRow[] = []
  for (const r of chunk) {
    const tableRow = toTableRow(r, executionsByRow.get(r.id) ?? {})
-    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()
-    })
+    const tombstoned = dispatch.scope.groupIds.some((gid) =>
+      isExecCancelledAfter(tableRow.executions?.[gid], dispatch.requestedAt)
+    )
    if (!tombstoned) tombstoneFiltered.push(tableRow)
  }

@@ -457,7 +471,7 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
    isManualRun: dispatch.isManualRun,
    groupIds: dispatch.scope.groupIds,
    mode: dispatch.mode,
-  })
+  }).map((p) => ({ ...p, dispatchId }))

  // Cursor advances to the last position in this chunk regardless of
  // eligibility — otherwise a window full of skipped cells loops forever.
@@ -552,6 +566,13 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
    return 'done'
  }

+  // A cell may have halted the dispatch mid-window (e.g. usage limit calls
+  // completeDispatchIfActive). Re-read before emitting the per-window
+  // `dispatching` event — otherwise that stale event arrives after the client
+  // already dropped the dispatch and re-adds it, flickering "X running" back.
+  const current = await readDispatch(dispatchId)
+  if (!current || current.status === 'cancelled' || current.status === 'complete') return 'done'
+
  await Promise.all([
    advanceCursor(dispatchId, lastPosition),
    appendTableEvent({
@@ -625,13 +646,31 @@ async function advanceCursor(dispatchId: string, newCursor: number): Promise<voi
    .where(eq(tableRunDispatches.id, dispatchId))
 }

-async function markDispatchComplete(dispatchId: string): Promise<void> {
+export async function markDispatchComplete(dispatchId: string): Promise<void> {
  await db
    .update(tableRunDispatches)
    .set({ status: 'complete', completedAt: new Date() })
    .where(eq(tableRunDispatches.id, dispatchId))
 }

+/** Complete a dispatch only if it's still active, returning whether THIS call
+ *  performed the transition. Lets concurrent cells that all hit a hard stop
+ *  (e.g. usage limit) elect a single owner — only the winner emits the
+ *  user-facing event, instead of one toast per in-flight cell. */
+export async function completeDispatchIfActive(dispatchId: string): Promise<boolean> {
+  const transitioned = await db
+    .update(tableRunDispatches)
+    .set({ status: 'complete', completedAt: new Date() })
+    .where(
+      and(
+        eq(tableRunDispatches.id, dispatchId),
+        inArray(tableRunDispatches.status, [...ACTIVE_DISPATCH_STATUSES])
+      )
+    )
+    .returning({ id: tableRunDispatches.id })
+  return transitioned.length > 0
+}
+
 export async function markDispatchCancelled(dispatchId: string): Promise<void> {
  await db
    .update(tableRunDispatches)
@@ -113,6 +113,17 @@ export type TableEvent =
       *  skip capped dispatches (see `resolveCellExec`). */
      limit?: { type: 'rows'; max: number }
    }
+  | {
+      /** A dispatch was stopped because the billed account is over its usage
+       *  limit. The client surfaces an upgrade prompt and redirects to billing.
+       *  The dispatch is halted via `markDispatchComplete` and the blocked
+       *  cells' pre-stamps are cleared so they revert to un-run. `dispatchId`
+       *  is absent for cascade/auto-fire payloads with no owning dispatch. */
+      kind: 'usageLimitReached'
+      tableId: string
+      dispatchId?: string
+      message: string
+    }

 export interface TableEventEntry {
  eventId: number
@@ -23,6 +23,7 @@ import { generateRestoreName } from '@/lib/core/utils/restore-name'
 import type { DbOrTx } from '@/lib/db/types'
 import { materializeExecutionData } from '@/lib/logs/execution/trace-store'
 import { COLUMN_TYPES, NAME_PATTERN, TABLE_LIMITS, USER_TABLE_ROWS_SQL_NAME } from './constants'
+import { areGroupDepsSatisfied } from './deps'
 import { buildFilterClause, buildSortClause } from './sql'
 import { fireTableTrigger } from './trigger'
 import type {
@@ -1717,7 +1718,8 @@ function deriveExecClearsForDataPatch(
  dataPatch: RowData,
  schema: TableSchema,
  existingExecutions: RowExecutions,
-  callerPatch: Record<string, RowExecutionMetadata | null> | undefined
+  callerPatch: Record<string, RowExecutionMetadata | null> | undefined,
+  mergedData: RowData
 ): {
  executionsPatch: Record<string, RowExecutionMetadata | null> | undefined
  inFlightDownstreamGroups: string[]
@@ -1739,11 +1741,18 @@ function deriveExecClearsForDataPatch(

  // Left-to-right walk, propagating dirty columns forward.
  const groups = schema.workflowGroups ?? []
+  const afterRow = { data: mergedData } as TableRow
  for (const group of groups) {
    const deps = group.dependencies?.columns ?? []
    const depMatched = deps.some((d) => dirtied.has(d))
    if (!depMatched) continue

+    // A dep column changed, but if the group's deps are no longer satisfied
+    // after the patch — a checkbox was unchecked or a text dep cleared — there's
+    // nothing to recompute. Leave the prior result alone instead of re-arming or
+    // cancelling it; only checking a box / filling a dep drives downstream work.
+    if (!areGroupDepsSatisfied(group, afterRow)) continue
+
    const exec = existingExecutions[group.id]
    if (exec) {
      const status = exec.status
@@ -1912,9 +1921,12 @@ async function writeExecutionsPatch(
            updatedAt: insertValues.updatedAt,
          },
          where: and(
-            // Reject if this group already shows authoritative `cancelled` for
-            // the same executionId — a stop click wrote it first.
-            sql`NOT (${tableRowExecutions.status} = 'cancelled' AND ${tableRowExecutions.executionId} IS NOT DISTINCT FROM ${guardExecutionId})`,
+            // Reject any guarded worker write when the cell is `cancelled` — a
+            // stop click wrote it authoritatively. SQL mirror of `isExecCancelled`
+            // (deps.ts). Status-only (not executionId-scoped): the cancel can
+            // only carry the pre-stamp's executionId (often null), so matching on
+            // id would let the worker's real-id claim resurrect a killed cell.
+            sql`${tableRowExecutions.status} <> 'cancelled'`,
            // Stale-worker: the cell's active run has moved on. Carve-outs
            // permit a fresh worker to take over when the row's executionId
            // is unset (dispatcher's pre-batch `pending` stamp).
@@ -2001,7 +2013,8 @@ export async function updateRow(
      data.data,
      table.schema,
      existingRow.executions,
-      data.executionsPatch
+      data.executionsPatch,
+      mergedData
    )
  const mergedExecutions = applyExecutionsPatch(existingRow.executions, effectiveExecutionsPatch)

@@ -2386,7 +2399,8 @@ export async function batchUpdateRows(
        update.data,
        table.schema,
        existing.executions,
-        update.executionsPatch
+        update.executionsPatch,
+        merged
      )
    const mergedExecutions = applyExecutionsPatch(existing.executions, effectiveExecutionsPatch)

@@ -329,6 +329,10 @@ export interface WorkflowGroupCellPayload {
  enrichmentId?: string
  workspaceId: string
  executionId: string
+  /** Owning dispatch, set by `dispatcherStep`. Lets the cell halt its dispatch
+   *  on a hard stop (e.g. usage limit). Absent for cascade/auto-fire payloads
+   *  that aren't driven by a dispatch. */
+  dispatchId?: string
 }

 /** Per-table concurrency cap. Mirrors trigger.dev's `concurrencyLimit: 20`. */
@@ -454,6 +458,30 @@ export async function cancelWorkflowGroupRuns(

  const mutations: RowMutation[] = Array.from(byRow.values())

+  // Defense-in-depth for paused/awaiting cells: a cell that paused mid-run is
+  // stamped `pending` with a `paused-<executionId>` jobId and keeps a record in
+  // `paused_executions`. Mark those cancelling so a pending waitpoint short-
+  // circuits before it resumes (the resume worker also re-checks the cell's
+  // cancelled tombstone — that's the authoritative stop). No-op for cells with
+  // no paused record.
+  const pausedCancellations = inFlightRows
+    .filter((r) => r.executionId && r.jobId?.startsWith('paused-'))
+    .map((r) => ({ executionId: r.executionId as string, workflowId: r.workflowId }))
+  if (pausedCancellations.length > 0) {
+    const { PauseResumeManager } = await import(
+      '@/lib/workflows/executor/human-in-the-loop-manager'
+    )
+    await Promise.allSettled(
+      pausedCancellations.map((p) =>
+        PauseResumeManager.beginPausedCancellation(p.executionId, p.workflowId).catch((err) => {
+          logger.warn(`beginPausedCancellation failed for ${p.executionId}`, {
+            error: toError(err).message,
+          })
+        })
+      )
+    )
+  }
+
  // Abort in-flight cell runs. The interface method `cancelByKey` is a no-op
  // on the trigger.dev backend (no in-process AbortControllers) and aborts
  // the matching AbortController on the database backend. Trigger.dev's tag