fix(enrichment): clear stale manual attempts on edit

Author: TheodoreSpeaks

apps/sim/hooks/queries/tables.test.ts

@@ -19,6 +19,19 @@ const { queryClient, cacheStore } = vi.hoisted(() => {
         cache.set(k, next)
         return next
       }),
+      setQueriesData: vi.fn(
+        (opts: { queryKey: readonly unknown[] }, updater: (value: unknown) => unknown) => {
+          const prefix = JSON.stringify(opts.queryKey).slice(0, -1)
+          const results: Array<[readonly unknown[], unknown]> = []
+          for (const [key, value] of cache.entries()) {
+            if (!key.startsWith(prefix)) continue
+            const next = updater(value)
+            cache.set(key, next)
+            results.push([JSON.parse(key) as readonly unknown[], next])
+          }
+          return results
+        }
+      ),
       getQueriesData: vi.fn((opts: { queryKey: readonly unknown[] }) => {
         const prefix = JSON.stringify(opts.queryKey).slice(0, -1)
         return [...cache.entries()]
@@ -63,6 +76,7 @@ import {
   useDeleteColumn,
   useRestoreTable,
   useUpdateColumn,
+  useUpdateTableRow,
 } from '@/hooks/queries/tables'
 import { tableKeys } from '@/hooks/queries/utils/table-keys'
 
@@ -220,6 +234,67 @@ describe('useUpdateColumn optimistic update', () => {
   })
 })
 
+describe('useUpdateTableRow optimistic update', () => {
+  it('clears a stale manual enrichment attempt when a mapped input changes', async () => {
+    const rowsKey = tableKeys.infiniteRows(TABLE_ID, 'default')
+    setCache(tableKeys.detail(TABLE_ID), {
+      id: TABLE_ID,
+      schema: {
+        columns: [
+          { id: 'domain', name: 'Domain', type: 'string' },
+          { id: 'email', name: 'Email', type: 'string', workflowGroupId: 'group-1' },
+        ],
+        workflowGroups: [
+          {
+            id: 'group-1',
+            workflowId: '',
+            type: 'enrichment',
+            enrichmentId: 'work-email',
+            autoRun: false,
+            inputMappings: [{ inputName: 'companyDomain', columnName: 'domain' }],
+            outputs: [{ blockId: '', path: '', outputId: 'email', columnName: 'email' }],
+          },
+        ],
+      },
+    })
+    setCache(rowsKey, {
+      pages: [
+        {
+          rows: [
+            {
+              id: 'row-1',
+              data: { domain: 'old.example' },
+              executions: {
+                'group-1': {
+                  status: 'completed',
+                  executionId: 'execution-1',
+                  jobId: null,
+                  workflowId: '',
+                  isManualRun: true,
+                  error: null,
+                },
+              },
+              position: 0,
+              createdAt: new Date(),
+              updatedAt: new Date(),
+            },
+          ],
+          totalCount: 1,
+        },
+      ],
+      pageParams: [0],
+    })
+
+    const hook = useUpdateTableRow({ workspaceId: WORKSPACE_ID, tableId: TABLE_ID })
+    await hook.onMutate?.({ rowId: 'row-1', data: { domain: 'new.example' } })
+
+    const rows = getCache<{
+      pages: Array<{ rows: Array<{ executions: Record<string, unknown> }> }>
+    }>(rowsKey)
+    expect(rows?.pages[0]?.rows[0]?.executions).toEqual({})
+  })
+})
+
 describe('useRestoreTable cache invalidation', () => {
   it('primes the table detail cache and clears stale rows for the restored table', () => {
     const hook = useRestoreTable()

apps/sim/lib/table/deps.test.ts

@@ -118,7 +118,26 @@ describe('optimisticallyScheduleNewlyEligibleGroups — enrichment inputs', () =
       domain: 'https://',
     })
 
-    expect(next).toBeNull()
+    expect(next).toEqual({})
+  })
+
+  it('clears a manual enrichment attempt when a mapped input changes', () => {
+    const manualGroup = { ...group, autoRun: false }
+    const before = makeRow(
+      { name: 'Person Name', domain: 'old.example' },
+      {
+        g1: {
+          ...completedExec('wf-g1'),
+          isManualRun: true,
+        },
+      }
+    )
+
+    const next = optimisticallyScheduleNewlyEligibleGroups([manualGroup], columns, before, {
+      domain: 'new.example',
+    })
+
+    expect(next).toEqual({})
   })
 })
 

apps/sim/lib/table/deps.ts

@@ -139,8 +139,9 @@ export function getGroupInvalidationColumns(group: WorkflowGroup): string[] {
  * workflow group whose deps were unmet *before* the patch and are satisfied
  * *after*, OR whose dep column was touched by the patch (the server will
  * cancel+re-run via `deriveExecClearsForDataPatch` + the in-flight cancel
- * orchestration), return a new `executions` map with that group flipped to
- * `pending`. Enrichments are only flipped when a provider can build a request.
+ * orchestration), return a new `executions` map. Eligible automatic groups are
+ * flipped to `pending`; terminal manual groups and automatic enrichments that
+ * cannot build provider params are cleared to mirror the server's re-arm.
  * The cell renderer treats `pending` as "Queued".
  *
  * Returns `null` when nothing changed, so callers can short-circuit.
@@ -160,10 +161,29 @@ export function optimisticallyScheduleNewlyEligibleGroups(
   const patchedColumns = new Set(Object.keys(patch))
 
   let next: RowExecutions | null = null
-  let flipped = 0
+  let changed = 0
   let skipped = 0
   for (const group of groups) {
+    const exec = beforeRow.executions?.[group.id]
+    const depTouched = getGroupInvalidationColumns(group).some((columnId) =>
+      patchedColumns.has(columnId)
+    )
+
+    const clearInvalidatedTerminalExecution = (): boolean => {
+      if (
+        !depTouched ||
+        (exec?.status !== 'completed' && exec?.status !== 'error' && exec?.status !== 'cancelled')
+      ) {
+        return false
+      }
+      if (next === null) next = { ...(beforeRow.executions ?? {}) }
+      delete next[group.id]
+      changed++
+      return true
+    }
+
     if (group.autoRun === false) {
+      if (areGroupDepsSatisfied(group, afterRow)) clearInvalidatedTerminalExecution()
       skipped++
       continue
     }
@@ -177,12 +197,12 @@ export function optimisticallyScheduleNewlyEligibleGroups(
         enrichment &&
         !getEnrichmentRowReadiness(enrichment, group, afterRow.data, columns).ready
       ) {
+        clearInvalidatedTerminalExecution()
         skipped++
         continue
       }
     }
 
-    const exec = beforeRow.executions?.[group.id]
     if (exec?.status === 'pending' && exec.jobId) {
       skipped++
       continue
@@ -196,10 +216,6 @@ export function optimisticallyScheduleNewlyEligibleGroups(
     // flight downstream groups, so optimistically flip to `pending`
     // regardless of current exec status (queued/running included — they're
     // about to be cancelled and re-run).
-    const depTouched = getGroupInvalidationColumns(group).some((columnId) =>
-      patchedColumns.has(columnId)
-    )
-
     if (!depTouched && (exec?.status === 'queued' || exec?.status === 'running')) {
       skipped++
       continue
@@ -209,7 +225,7 @@ export function optimisticallyScheduleNewlyEligibleGroups(
       continue
     }
 
-    flipped++
+    changed++
     if (next === null) next = { ...(beforeRow.executions ?? {}) }
     const pending: RowExecutionMetadata = {
       status: 'pending',
@@ -221,8 +237,8 @@ export function optimisticallyScheduleNewlyEligibleGroups(
     }
     next[group.id] = pending
   }
-  if (flipped > 0) {
-    logger.debug(`[OptimisticCascade] row=${beforeRow.id} flipped=${flipped} skipped=${skipped}`)
+  if (changed > 0) {
+    logger.debug(`[OptimisticCascade] row=${beforeRow.id} changed=${changed} skipped=${skipped}`)
   }
   return next
 }