fix(tables): preserve run validation and signals

Author: TheodoreSpeaks

apps/sim/lib/table/application/rows.test.ts

@@ -84,6 +84,7 @@ import {
   queryTableRows,
   replaceTableRows,
   TableRowsValidationError,
+  tablePredicateNamesToFilter,
   upsertTableRow,
 } from '@/lib/table/application/rows'
 
@@ -104,6 +105,19 @@ const TABLE: TableDefinition = {
 
 const PRINCIPAL = { kind: 'session' as const, userId: 'user-1', sessionId: 'session-1' }
 
+describe('table predicate translation', () => {
+  it('maps invalid run filters to the shared row validation error', () => {
+    expect(() =>
+      tablePredicateNamesToFilter({ all: [{ field: 'missing', op: 'eq', value: 'ready' }] }, TABLE)
+    ).toThrowError(
+      expect.objectContaining({
+        name: 'TableRowsValidationError',
+        details: { code: 'INVALID_FILTER' },
+      })
+    )
+  })
+})
+
 describe('replaceTableRows application use case', () => {
   beforeEach(() => {
     vi.clearAllMocks()

apps/sim/lib/table/application/rows.ts

@@ -99,10 +99,14 @@ export function tablePredicateNamesToFilter(
   predicate: TablePredicate,
   table: TableDefinition
 ): Filter {
-  validatePredicateShape(predicate)
-  const translated = predicateToStorage(predicate, table.schema)
-  validateStoragePredicate(translated, table.schema.columns)
-  return predicateToFilter(translated)
+  try {
+    validatePredicateShape(predicate)
+    const translated = predicateToStorage(predicate, table.schema)
+    validateStoragePredicate(translated, table.schema.columns)
+    return predicateToFilter(translated)
+  } catch (error) {
+    rethrowQueryValidation(error)
+  }
 }
 
 async function throwValidationResponse(

apps/sim/lib/table/application/runs.test.ts

@@ -104,7 +104,10 @@ describe('table run application use cases', () => {
       billedAccountUserId: 'billing-owner-1',
     })
     mockGetRowById.mockResolvedValue({ id: 'row-1' })
-    mockRunWorkflowColumn.mockResolvedValue({ dispatchId: 'dispatch-1' })
+    mockRunWorkflowColumn.mockResolvedValue({
+      dispatchId: 'dispatch-1',
+      shouldSignalRowsChanged: true,
+    })
     mockRequireTableRowIds.mockResolvedValue(undefined)
     mockCancelRuns.mockResolvedValue(1)
     mockTranslatePredicate.mockReturnValue({ all: [] })
@@ -200,7 +203,10 @@ describe('table run application use cases', () => {
   })
 
   it('does not signal when the dispatcher reports a no-op', async () => {
-    mockRunWorkflowColumn.mockResolvedValue({ dispatchId: null })
+    mockRunWorkflowColumn.mockResolvedValue({
+      dispatchId: null,
+      shouldSignalRowsChanged: false,
+    })
 
     await startTableRun.execute({
       principal: PRINCIPAL,
@@ -215,6 +221,25 @@ describe('table run application use cases', () => {
     expect(mockSignalRowsChanged).not.toHaveBeenCalled()
   })
 
+  it('signals a cleared row state when cancellation wins before dispatch', async () => {
+    mockRunWorkflowColumn.mockResolvedValue({
+      dispatchId: null,
+      shouldSignalRowsChanged: true,
+    })
+
+    await startTableRun.execute({
+      principal: PRINCIPAL,
+      input: {
+        kind: 'selection',
+        tableId: TABLE.id,
+        groupIds: ['group-1'],
+        mode: 'all',
+      },
+    })
+
+    expect(mockSignalRowsChanged).toHaveBeenCalledWith(TABLE.id)
+  })
+
   it('requires a canonical row for row cancellation', async () => {
     mockGetRowById.mockResolvedValue(null)
 

apps/sim/lib/table/application/runs.ts

@@ -61,6 +61,7 @@ export type StartTableRunInput = StartSelectionRunInput | StartRowEnrichmentInpu
 
 export interface StartTableRunResult extends TableRunResult {
   dispatchId: string | null
+  shouldSignalRowsChanged: boolean
 }
 
 function requireCanonicalGroups(table: TableDefinition, groupIds: string[]): void {
@@ -96,7 +97,11 @@ export const startTableRun = defineAuthorizedTableUseCase({
         requestId: requestId(input),
         triggeredByUserId,
       })
-      return { table: context.table, dispatchId: result.dispatchId }
+      return {
+        table: context.table,
+        dispatchId: result.dispatchId,
+        shouldSignalRowsChanged: result.shouldSignalRowsChanged,
+      }
     }
 
     if (input.rowIds && input.predicate) {
@@ -149,10 +154,14 @@ export const startTableRun = defineAuthorizedTableUseCase({
       requestId: requestId(input),
       triggeredByUserId,
     })
-    return { table: context.table, dispatchId: result.dispatchId }
+    return {
+      table: context.table,
+      dispatchId: result.dispatchId,
+      shouldSignalRowsChanged: result.shouldSignalRowsChanged,
+    }
   },
   afterSuccess: ({ context, result }) => {
-    if (result.dispatchId !== null) signalTableRowsChanged(context.tableId)
+    if (result.shouldSignalRowsChanged) signalTableRowsChanged(context.tableId)
   },
 })
 

apps/sim/lib/table/dispatcher.ts

@@ -21,7 +21,7 @@ import { writeWorkflowGroupState } from '@/lib/table/cell-write'
 import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
 import { isExecCancelledAfter } from '@/lib/table/deps'
 import { appendTableEvent } from '@/lib/table/events'
-import { type DbExecutor, withSeqscanOff } from '@/lib/table/planner'
+import { type DbExecutor, type DbTransaction, withSeqscanOff } from '@/lib/table/planner'
 import { updateTableRowsWithDerivedSecretProvenance } from '@/lib/table/rows/secret-provenance'
 import { buildFilterClause } from '@/lib/table/sql'
 import type {
@@ -87,6 +87,24 @@ export interface DispatchRow {
   requestedAt: Date
 }
 
+async function deleteExecutionRows(trx: DbTransaction, filters: SQL[]): Promise<number> {
+  const countRows = await trx.execute<{ count: number | string }>(sql`
+    WITH deleted AS (
+      DELETE FROM ${tableRowExecutions}
+      WHERE ${and(...filters)}
+      RETURNING 1
+    )
+    SELECT count(*)::integer AS count FROM deleted
+  `)
+  const [countRow] = Array.isArray(countRows) ? countRows : []
+  if (!countRow) throw new Error('Workflow cell clearing did not return a deleted count')
+  const count = Number(countRow.count)
+  if (!Number.isSafeInteger(count) || count < 0) {
+    throw new Error('Workflow cell clearing returned an invalid deleted count')
+  }
+  return count
+}
+
 export type DispatcherStepResult = 'continue' | 'done'
 
 /** Eager bulk clear at click time so the user sees every targeted cell go
@@ -96,17 +114,18 @@ export type DispatcherStepResult = 'continue' | 'done'
  *  already filled, mirroring the eligibility predicate. */
 export async function bulkClearWorkflowGroupCells(input: {
   tableId: string
+  workspaceId: string
   groups: Array<{ id: string; outputs: Array<{ columnName: string }> }>
   rowIds?: string[]
   /** Select-all scope: deselected rows whose outputs must NOT be wiped. */
   excludeRowIds?: string[]
   mode: DispatchMode
-}): Promise<void> {
-  const { tableId, groups, rowIds, excludeRowIds, mode } = input
-  if (groups.length === 0) return
+}): Promise<boolean> {
+  const { tableId, workspaceId, groups, rowIds, excludeRowIds, mode } = input
+  if (groups.length === 0) return false
   // `'new'` mode targets only rows with no prior attempt — nothing to clear.
   // Pre-existing outputs on any other row must not be wiped by an auto-fire.
-  if (mode === 'new') return
+  if (mode === 'new') return false
 
   const groupIds = groups.map((g) => g.id)
   const rowScope = rowIds && rowIds.length > 0 ? rowIds : null
@@ -119,25 +138,34 @@ export async function bulkClearWorkflowGroupCells(input: {
     const outputCols = Array.from(
       new Set(groups.flatMap((g) => g.outputs.map((o) => o.columnName)))
     )
-    const filters: SQL[] = [eq(userTableRows.tableId, tableId)]
+    const filters: SQL[] = [
+      eq(userTableRows.tableId, tableId),
+      eq(userTableRows.workspaceId, workspaceId),
+    ]
     if (rowScope) filters.push(inArray(userTableRows.id, rowScope))
     if (excluded) filters.push(notInArray(userTableRows.id, excluded))
 
-    await db.transaction(async (trx) => {
+    return db.transaction(async (trx) => {
       const rowWhere = and(...filters)!
-      await updateTableRowsWithDerivedSecretProvenance(trx, {
+      const clearedRows = await updateTableRowsWithDerivedSecretProvenance(trx, {
         rowWhere,
         transformation: { mode: 'remove-columns', columnIds: outputCols },
       })
       const execFilters: SQL[] = [
         eq(tableRowExecutions.tableId, tableId),
         inArray(tableRowExecutions.groupId, groupIds),
+        sql`${tableRowExecutions.rowId} IN (
+          SELECT ${userTableRows.id}
+          FROM ${userTableRows}
+          WHERE ${userTableRows.tableId} = ${tableId}
+            AND ${userTableRows.workspaceId} = ${workspaceId}
+        )`,
       ]
       if (rowScope) execFilters.push(inArray(tableRowExecutions.rowId, rowScope))
       if (excluded) execFilters.push(notInArray(tableRowExecutions.rowId, excluded))
-      await trx.delete(tableRowExecutions).where(and(...execFilters))
+      const deletedExecutions = await deleteExecutionRows(trx, execFilters)
+      return clearedRows > 0 || deletedExecutions > 0
     })
-    return
   }
 
   // `incomplete`: clear per-group, not per-row. Only groups that are
@@ -147,20 +175,25 @@ export async function bulkClearWorkflowGroupCells(input: {
   // because a *sibling* group on the same row is incomplete, re-running the
   // completed one. (`never-run` groups have no exec/output to clear — the
   // dispatcher runs them via eligibility.)
-  await db.transaction(async (trx) => {
+  return db.transaction(async (trx) => {
+    let rowsChanged = false
     for (const group of groups) {
       const reRunnable = sql`EXISTS (
         SELECT 1 FROM ${tableRowExecutions} re
         WHERE re.row_id = ${userTableRows.id}
           AND re.group_id = ${group.id}
           AND re.status IN ('error', 'cancelled')
       )`
-      const filters: SQL[] = [eq(userTableRows.tableId, tableId), reRunnable]
+      const filters: SQL[] = [
+        eq(userTableRows.tableId, tableId),
+        eq(userTableRows.workspaceId, workspaceId),
+        reRunnable,
+      ]
       if (rowScope) filters.push(inArray(userTableRows.id, rowScope))
       if (excluded) filters.push(notInArray(userTableRows.id, excluded))
 
       const rowWhere = and(...filters)!
-      await updateTableRowsWithDerivedSecretProvenance(trx, {
+      const clearedRows = await updateTableRowsWithDerivedSecretProvenance(trx, {
         rowWhere,
         transformation: {
           mode: 'remove-columns',
@@ -172,11 +205,19 @@ export async function bulkClearWorkflowGroupCells(input: {
         eq(tableRowExecutions.tableId, tableId),
         eq(tableRowExecutions.groupId, group.id),
         sql`${tableRowExecutions.status} IN ('error', 'cancelled')`,
+        sql`${tableRowExecutions.rowId} IN (
+          SELECT ${userTableRows.id}
+          FROM ${userTableRows}
+          WHERE ${userTableRows.tableId} = ${tableId}
+            AND ${userTableRows.workspaceId} = ${workspaceId}
+        )`,
       ]
       if (rowScope) execFilters.push(inArray(tableRowExecutions.rowId, rowScope))
       if (excluded) execFilters.push(notInArray(tableRowExecutions.rowId, excluded))
-      await trx.delete(tableRowExecutions).where(and(...execFilters))
+      const deletedExecutions = await deleteExecutionRows(trx, execFilters)
+      rowsChanged ||= clearedRows > 0 || deletedExecutions > 0
     }
+    return rowsChanged
   })
 }
 

apps/sim/lib/table/rows/secret-provenance.test.ts

@@ -464,14 +464,18 @@ describe('table row secret provenance', () => {
   it('binds derived rows only after their matching sidecars are written', async () => {
     queueTableRows(userTableRows, [{ id: 'legacy-row' }])
 
-    await updateTableRowsWithDerivedSecretProvenance(dbChainMock.db as unknown as DbTransaction, {
-      rowWhere: eq(userTableRows.id, 'legacy-row'),
-      transformation: {
-        mode: 'remove-columns',
-        columnIds: ['deleted-column', 'deleted-column'],
-      },
-    })
+    const updatedCount = await updateTableRowsWithDerivedSecretProvenance(
+      dbChainMock.db as unknown as DbTransaction,
+      {
+        rowWhere: eq(userTableRows.id, 'legacy-row'),
+        transformation: {
+          mode: 'remove-columns',
+          columnIds: ['deleted-column', 'deleted-column'],
+        },
+      }
+    )
 
+    expect(updatedCount).toBe(1)
     expect(dbChainMockFns.execute).toHaveBeenCalledTimes(2)
     expect(boundArrayValues(dbChainMockFns.execute.mock.calls[0][0])).toEqual([])
     expect(sqlText(dbChainMockFns.execute.mock.calls[0][0])).not.toMatch(

apps/sim/lib/table/rows/secret-provenance.ts

@@ -434,13 +434,13 @@ export async function updateTableRowsWithDerivedSecretProvenance(
     rowWhere: SQL
     transformation: DerivedTableRowTransformation
   }
-): Promise<void> {
+): Promise<number> {
   const removedColumnIds =
     options.transformation.mode === 'remove-columns'
       ? [...new Set(options.transformation.columnIds)]
       : []
   if (options.transformation.mode === 'remove-columns') {
-    if (removedColumnIds.length === 0) return
+    if (removedColumnIds.length === 0) return 0
     if (
       removedColumnIds.length > MAX_PROVENANCE_COLUMNS_PER_ROW ||
       removedColumnIds.some((columnId) => columnId.length === 0)
@@ -481,6 +481,7 @@ export async function updateTableRowsWithDerivedSecretProvenance(
     : sql`source.provenance_entries`
 
   let afterId: string | undefined
+  let updatedCount = 0
   for (;;) {
     const page = await trx
       .select({ id: userTableRows.id })
@@ -491,6 +492,7 @@ export async function updateTableRowsWithDerivedSecretProvenance(
       .for('update')
     if (page.length === 0) break
     const rowIds = page.map((row) => row.id)
+    updatedCount += rowIds.length
 
     await trx.execute(sql`
       WITH source AS MATERIALIZED (
@@ -628,6 +630,7 @@ export async function updateTableRowsWithDerivedSecretProvenance(
     afterId = rowIds[rowIds.length - 1]
     if (page.length < QUERY_CHUNK_SIZE) break
   }
+  return updatedCount
 }
 
 async function readTableRowsVersion(tableId: string, workspaceId: string): Promise<number | null> {