fix(table): trigger cascade race fixes, polling, workflow column flag

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/new-column-dropdown/new-column-dropdown.tsx

@@ -8,9 +8,14 @@ import {
   DropdownMenuTrigger,
 } from '@/components/emcn'
 import { Plus } from '@/components/emcn/icons'
+import { isWorkflowColumnsEnabledClient } from '@/lib/core/config/feature-flags'
 import type { ColumnDefinition } from '@/lib/table'
 import { COLUMN_TYPE_OPTIONS } from '../column-config-sidebar'
 
+const VISIBLE_COLUMN_TYPE_OPTIONS = isWorkflowColumnsEnabledClient
+  ? COLUMN_TYPE_OPTIONS
+  : COLUMN_TYPE_OPTIONS.filter((o) => o.type !== 'workflow')
+
 const CELL_HEADER =
   'border-[var(--border)] border-r border-b bg-[var(--bg)] px-2 py-[7px] text-left align-middle'
 
@@ -56,7 +61,7 @@ export function NewColumnDropdown({
         )}
       </DropdownMenuTrigger>
       <DropdownMenuContent align='start' side='bottom' sideOffset={4}>
-        {COLUMN_TYPE_OPTIONS.map((option) => {
+        {VISIBLE_COLUMN_TYPE_OPTIONS.map((option) => {
           const Icon = option.icon
           const onSelect =
             option.type === 'workflow'

apps/sim/hooks/queries/tables.ts

@@ -4,6 +4,7 @@
  * React Query hooks for managing user-defined tables.
  */
 
+import { useEffect, useMemo } from 'react'
 import { createLogger } from '@sim/logger'
 import {
   type InfiniteData,
@@ -68,7 +69,7 @@ import type {
   WorkflowGroupDependencies,
   WorkflowGroupOutput,
 } from '@/lib/table'
-import { optimisticallyScheduleNewlyEligibleGroups } from '@/lib/table/deps'
+import { areOutputsFilled, optimisticallyScheduleNewlyEligibleGroups } from '@/lib/table/deps'
 
 /** Short poll to surface running → completed transitions from the server without a dedicated realtime channel. */
 const ROWS_POLL_INTERVAL_WHILE_RUNNING_MS = 1500
@@ -84,14 +85,6 @@ function hasRunningGroupExecution(rows: TableRow[] | undefined): boolean {
   return false
 }
 
-function hasRunningGroupExecutionInPages(pages: TableRowsResponse[] | undefined): boolean {
-  if (!pages) return false
-  for (const page of pages) {
-    if (hasRunningGroupExecution(page.rows)) return true
-  }
-  return false
-}
-
 const logger = createLogger('TableQueries')
 
 type TableQueryScope = 'active' | 'archived' | 'all'
@@ -293,9 +286,10 @@ export function useInfiniteTableRows({
     filter: filter ?? null,
     sort: sort ?? null,
   })
+  const queryKey = useMemo(() => tableKeys.infiniteRows(tableId, paramsKey), [tableId, paramsKey])
 
-  return useInfiniteQuery({
-    queryKey: tableKeys.infiniteRows(tableId, paramsKey),
+  const query = useInfiniteQuery({
+    queryKey,
     queryFn: ({ pageParam, signal }) =>
       fetchTableRows({
         workspaceId,
@@ -314,23 +308,65 @@ export function useInfiniteTableRows({
     },
     enabled: Boolean(workspaceId && tableId) && enabled,
     staleTime: 30 * 1000,
-    /**
-     * Poll while any row has a `pending` or `running` group execution.
-     * Realtime sockets push every cell write, but cross-network paths
-     * (trigger.dev workers → realtime ECS, client through CloudFront/proxy)
-     * occasionally drop events. Polling at the running cadence is the
-     * safety net so cells reach their terminal state without a refresh.
-     * No polling when nothing is running and no polling while a mutation
-     * is in flight (optimistic-update guard).
-     */
-    refetchInterval: (query) => {
-      if (queryClient.isMutating() > 0) return false
-      return hasRunningGroupExecutionInPages(query.state.data?.pages)
-        ? ROWS_POLL_INTERVAL_WHILE_RUNNING_MS
-        : false
-    },
-    refetchIntervalInBackground: false,
   })
+
+  /**
+   * Per-page polling. Built-in `refetchInterval` would refetch every loaded
+   * page on each tick — wasteful when only one page has running cells.
+   * Instead, walk pages each tick and refetch ONLY the dirty ones, splicing
+   * results back into the cache. Polling stops when no page has in-flight
+   * cells, or while a mutation is running (optimistic-update guard).
+   */
+  useEffect(() => {
+    if (!enabled || !workspaceId || !tableId) return
+    let cancelled = false
+    const tick = async () => {
+      if (cancelled) return
+      if (queryClient.isMutating() > 0) return
+      const data = queryClient.getQueryData<InfiniteData<TableRowsResponse, number>>(queryKey)
+      if (!data) return
+      const dirty: number[] = []
+      for (let i = 0; i < data.pages.length; i++) {
+        if (hasRunningGroupExecution(data.pages[i].rows)) {
+          dirty.push(data.pageParams[i] ?? i * pageSize)
+        }
+      }
+      if (dirty.length === 0) return
+      await Promise.all(
+        dirty.map(async (offset) => {
+          try {
+            const fresh = await fetchTableRows({
+              workspaceId,
+              tableId,
+              limit: pageSize,
+              offset,
+              filter,
+              sort,
+              includeTotal: offset === 0,
+            })
+            if (cancelled) return
+            queryClient.setQueryData<InfiniteData<TableRowsResponse, number>>(queryKey, (prev) => {
+              if (!prev) return prev
+              const idx = prev.pageParams.indexOf(offset)
+              if (idx === -1) return prev
+              const nextPages = prev.pages.slice()
+              nextPages[idx] = fresh
+              return { ...prev, pages: nextPages }
+            })
+          } catch {
+            // Transient fetch failure — next tick retries. Don't kill the loop.
+          }
+        })
+      )
+    }
+    const intervalId = setInterval(() => void tick(), ROWS_POLL_INTERVAL_WHILE_RUNNING_MS)
+    return () => {
+      cancelled = true
+      clearInterval(intervalId)
+    }
+  }, [enabled, workspaceId, tableId, pageSize, filter, sort, queryClient, queryKey])
+
+  return query
 }
 
 /**
@@ -1176,6 +1212,10 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) {
     onMutate: async ({ groupIds, runMode = 'all', rowIds }) => {
       const targetRowIds = rowIds && rowIds.length > 0 ? new Set(rowIds) : null
       const targetGroupIds = new Set(groupIds)
+      const groups =
+        queryClient.getQueryData<TableDefinition>(tableKeys.detail(tableId))?.schema
+          .workflowGroups ?? []
+      const groupsById = new Map(groups.map((g) => [g.id, g]))
       const snapshots = await snapshotAndMutateRows(queryClient, tableId, (r) => {
         if (targetRowIds && !targetRowIds.has(r.id)) return null
         const executions = r.executions ?? {}
@@ -1184,7 +1224,15 @@ export function useRunColumn({ workspaceId, tableId }: RowMutationContext) {
         for (const groupId of targetGroupIds) {
           const exec = executions[groupId] as RowExecutionMetadata | undefined
           if (isOptimisticInFlight(exec)) continue
-          if (runMode === 'incomplete' && exec?.status === 'completed') continue
+          // Mirror server eligibility for `mode: 'incomplete'`: skip cells whose
+          // outputs are filled, regardless of exec status. A cancelled/error
+          // cell with a leftover value from a prior run was rendering as filled
+          // but flipping to "queued" optimistically here even though the server
+          // would skip it.
+          if (runMode === 'incomplete') {
+            const group = groupsById.get(groupId)
+            if (group && areOutputsFilled(group, r)) continue
+          }
           next[groupId] = buildPendingExec(exec)
           changed = true
         }

apps/sim/lib/core/config/env.ts

@@ -455,6 +455,7 @@ export const env = createEnv({
     NEXT_PUBLIC_AUDIT_LOGS_ENABLED:        z.boolean().optional(),                   // Enable audit logs on self-hosted (bypasses hosted requirements)
     NEXT_PUBLIC_DATA_RETENTION_ENABLED:   z.boolean().optional(),                   // Enable data retention settings on self-hosted (bypasses hosted requirements)
     NEXT_PUBLIC_DATA_DRAINS_ENABLED:      z.boolean().optional(),                   // Enable data drains on self-hosted (bypasses hosted requirements)
+    NEXT_PUBLIC_WORKFLOW_COLUMNS_ENABLED: z.boolean().optional(),                   // Show the "Workflow" column type in user tables (defaults to false)
     NEXT_PUBLIC_ORGANIZATIONS_ENABLED:     z.boolean().optional(),                   // Enable organizations on self-hosted (bypasses plan requirements)
     NEXT_PUBLIC_DISABLE_INVITATIONS:       z.boolean().optional(),                   // Disable workspace invitations globally (for self-hosted deployments)
     NEXT_PUBLIC_DISABLE_PUBLIC_API:        z.boolean().optional(),                   // Disable public API access UI toggle globally
@@ -493,6 +494,7 @@ export const env = createEnv({
     NEXT_PUBLIC_AUDIT_LOGS_ENABLED: process.env.NEXT_PUBLIC_AUDIT_LOGS_ENABLED,
     NEXT_PUBLIC_DATA_RETENTION_ENABLED: process.env.NEXT_PUBLIC_DATA_RETENTION_ENABLED,
     NEXT_PUBLIC_DATA_DRAINS_ENABLED: process.env.NEXT_PUBLIC_DATA_DRAINS_ENABLED,
+    NEXT_PUBLIC_WORKFLOW_COLUMNS_ENABLED: process.env.NEXT_PUBLIC_WORKFLOW_COLUMNS_ENABLED,
     NEXT_PUBLIC_ORGANIZATIONS_ENABLED: process.env.NEXT_PUBLIC_ORGANIZATIONS_ENABLED,
     NEXT_PUBLIC_DISABLE_INVITATIONS: process.env.NEXT_PUBLIC_DISABLE_INVITATIONS,
     NEXT_PUBLIC_DISABLE_PUBLIC_API: process.env.NEXT_PUBLIC_DISABLE_PUBLIC_API,

apps/sim/lib/core/config/feature-flags.ts

@@ -141,6 +141,15 @@ export const isDataRetentionEnabled = isTruthy(env.DATA_RETENTION_ENABLED)
  */
 export const isDataDrainsEnabled = isTruthy(env.DATA_DRAINS_ENABLED)
 
+/**
+ * Are workflow output columns enabled in user tables.
+ * Defaults to false; set NEXT_PUBLIC_WORKFLOW_COLUMNS_ENABLED=true to show
+ * the "Workflow" column type in the new-column dropdown.
+ */
+export const isWorkflowColumnsEnabledClient = isTruthy(
+  getEnv('NEXT_PUBLIC_WORKFLOW_COLUMNS_ENABLED')
+)
+
 /**
  * Is E2B enabled for remote code execution
  */

apps/sim/lib/table/cell-write.ts

@@ -56,24 +56,29 @@ export async function writeWorkflowGroupState(
   }
   const current = row.executions?.[groupId] as RowExecutionMetadata | undefined
   // Stale-worker guard: only blocks writes FROM an old worker (status =
-  // running / completed / error / pending). A `queued` stamp is the scheduler
-  // claiming the cell for a brand-new run — the new executionId is supposed
-  // to overwrite whatever was there. Same for `cancelled` (authoritative).
-  // Without this carve-out, the new run's stamp gets rejected and the cell
-  // is stuck in its old state forever.
-  const isAuthoritativeNewStamp =
-    payload.executionState.status === 'queued' || payload.executionState.status === 'cancelled'
-  if (
-    !isAuthoritativeNewStamp &&
-    current &&
-    current.executionId &&
-    current.executionId !== executionId
-  ) {
+  // running / completed / error / pending). A `queued` stamp from the
+  // scheduler can claim the cell for a brand-new run — that's the new
+  // authority. Same for `cancelled` (always authoritative, written by stop).
+  const isCancelStamp = payload.executionState.status === 'cancelled'
+  const isQueuedStamp = payload.executionState.status === 'queued'
+  const isNewQueuedStamp = isQueuedStamp && current?.executionId !== executionId
+  const bypassStaleWorker = isNewQueuedStamp || isCancelStamp
+  if (!bypassStaleWorker && current && current.executionId && current.executionId !== executionId) {
     logger.info(
       `Skipping group write — stale worker (table=${tableId} row=${rowId} group=${groupId} mine=${executionId} active=${current.executionId})`
     )
     return 'skipped'
   }
+  // A late `queued` stamp for the SAME run that's already moved past queued
+  // (worker called markWorkflowGroupPickedUp before our parallel stamp landed)
+  // must NOT overwrite the further-along state. Without this, a cell can show
+  // "queued" forever while the worker is actually running.
+  if (isQueuedStamp && current?.executionId === executionId && current.status !== 'pending') {
+    logger.info(
+      `Skipping queued stamp — same run already at status=${current.status} (table=${tableId} row=${rowId} group=${groupId} executionId=${executionId})`
+    )
+    return 'skipped'
+  }
   if (
     current?.status === 'cancelled' &&
     current.executionId === executionId &&
@@ -89,7 +94,7 @@ export async function writeWorkflowGroupState(
   // stamps from the scheduler also bypass — they ARE the new authority. Cell-
   // task writes (running/completed/error) get the SQL guard so an in-flight
   // partial can't clobber a stop click or a newer run that already committed.
-  const cancellationGuard = isAuthoritativeNewStamp ? undefined : { groupId, executionId }
+  const cancellationGuard = bypassStaleWorker ? undefined : { groupId, executionId }
   const result = await updateRow(
     {
       tableId,