Function cancelWorkflowGroupRuns

apps/sim/lib/table/workflow-columns.ts:365–614

(
  tableId: string,
  rowId?: string,
  options?: { groupIds?: string[]; filter?: Filter; excludeRowIds?: string[] }
)

 * active dispatches running (their in-flight checks skip cancelled cells).
 */
export async function cancelWorkflowGroupRuns(
  tableId: string,
  rowId?: string,
  options?: { groupIds?: string[]; filter?: Filter; excludeRowIds?: string[] }
): Promise<number> {
  const { getTableById } = await import('@/lib/table/service')
  const { updateRow } = await import('@/lib/table/rows/service')
  const { getJobQueue } = await import('@/lib/core/async-jobs/config')
  const { listActiveDispatches, markActiveDispatchesCancelled } = await import(
    '@/lib/table/dispatcher'
  )

  const table = await getTableById(tableId)
  if (!table) {
    logger.warn(`cancelWorkflowGroupRuns: table ${tableId} not found`)
    return 0
  }

  // Per-row cancel leaves the dispatcher alone — other rows in the same
  // dispatch keep running. Table-wide cancel must stop it, else the cursor
  // marches on and re-enqueues fresh cells past what we just cancelled.
  // Filter-scoped cancel stops only dispatches with that exact filter scope
  // (its own run); whole-table or differently-scoped dispatches keep running —
  // their cells cancelled below are skipped via `cancelledAt > requestedAt`.
  if (!rowId) {
    await markActiveDispatchesCancelled(tableId, options?.filter, options?.excludeRowIds)
  }

  const allGroups = table.schema.workflowGroups ?? []
  if (allGroups.length === 0) return 0
  const groupIds = options?.groupIds
    ? new Set(allGroups.filter((g) => options.groupIds?.includes(g.id)).map((g) => g.id))
    : new Set(allGroups.map((g) => g.id))
  if (groupIds.size === 0) return 0

  // Per-row Stop on a row the dispatcher hasn't reached yet has no sidecar
  // entry to cancel — the dispatcher would later walk to that row, see no
  // exec, classify eligible, and re-fire. Pre-write `cancelled` tombstones
  // for active-dispatch in-scope groups so the existing `cancelledAt >
  // dispatch.requestedAt` filter in `dispatcherStep` catches them. Skip
  // when there's no active dispatch (nothing to outrun).
  let aheadOfCursorTombstones: Array<{ groupId: string; workflowId: string }> = []
  if (rowId) {
    const activeDispatches = await listActiveDispatches(tableId)
    const relevant = activeDispatches.filter((d) => {
      if (d.scope.rowIds && !d.scope.rowIds.includes(rowId)) return false
      return d.scope.groupIds.some((gid) => groupIds.has(gid))
    })
    if (relevant.length > 0) {
      // Intersection of targeted groups with active-dispatch scopes — only
      // these groups are at risk of being re-fired by an in-progress dispatch.
      const atRisk = new Set<string>()
      for (const d of relevant) {
        for (const gid of d.scope.groupIds) {
          if (groupIds.has(gid)) atRisk.add(gid)
        }
      }
      aheadOfCursorTombstones = Array.from(atRisk).map((gid) => ({