fix(tables): follow latest workflow deployment

Author: TheodoreSpeaks

apps/sim/background/workflow-column-execution.test.ts

@@ -7,16 +7,23 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { createTimeoutAbortController, getExecutionDeadlineAt } from '@/lib/core/execution-limits'
 import { abortManualExecution } from '@/lib/execution/manual-cancellation'
 import {
+  assertWorkflowGroupMatchesLatestDeployment,
   buildTableAbortState,
   buildTableUsageLimitClear,
   createWorkflowGroupAttemptTimeoutController,
   createWorkflowGroupCarrierTimeoutController,
   terminalizeAbortedQueuedCarrierMarker,
 } from '@/background/workflow-column-execution'
 
-const { appendTableEventMock } = vi.hoisted(() => ({ appendTableEventMock: vi.fn() }))
+const { appendTableEventMock, flattenWorkflowOutputsMock } = vi.hoisted(() => ({
+  appendTableEventMock: vi.fn(),
+  flattenWorkflowOutputsMock: vi.fn(),
+}))
 
 vi.mock('@/lib/table/events', () => ({ appendTableEvent: appendTableEventMock }))
+vi.mock('@/lib/workflows/blocks/flatten-outputs', () => ({
+  flattenWorkflowOutputs: flattenWorkflowOutputsMock,
+}))
 
 beforeEach(() => {
   vi.clearAllMocks()
@@ -50,6 +57,90 @@ const QUEUED_PAYLOAD = {
   },
 }
 
+function latestDeployment(): Parameters<typeof assertWorkflowGroupMatchesLatestDeployment>[1] {
+  return {
+    blocks: {
+      start: {
+        id: 'start',
+        type: 'start_trigger',
+        subBlocks: {
+          inputFormat: { value: [{ name: 'company', type: 'string' }] },
+        },
+      },
+      agent: {
+        id: 'agent',
+        type: 'agent',
+        subBlocks: {},
+      },
+    },
+    edges: [{ id: 'start-agent', source: 'start', target: 'agent' }],
+    loops: {},
+    parallels: {},
+    variables: {},
+    isFromNormalizedTables: false,
+    deploymentVersionId: 'deployment-version-latest',
+  } as Parameters<typeof assertWorkflowGroupMatchesLatestDeployment>[1]
+}
+
+describe('latest table workflow deployment mappings', () => {
+  beforeEach(() => {
+    flattenWorkflowOutputsMock.mockReturnValue([
+      {
+        blockId: 'agent',
+        blockName: 'Agent',
+        blockType: 'agent',
+        path: 'content',
+        leafType: 'string',
+      },
+    ])
+  })
+
+  it('accepts saved mappings that the latest active deployment still supports', () => {
+    expect(() =>
+      assertWorkflowGroupMatchesLatestDeployment(
+        {
+          id: 'group-1',
+          workflowId: 'workflow-1',
+          outputs: [{ blockId: 'agent', path: 'content', columnName: 'column-output' }],
+          inputMappings: [{ inputName: 'company', columnName: 'column-company' }],
+        },
+        latestDeployment()
+      )
+    ).not.toThrow()
+  })
+
+  it('rejects an output mapping removed by the latest active deployment', () => {
+    expect(() =>
+      assertWorkflowGroupMatchesLatestDeployment(
+        {
+          id: 'group-1',
+          workflowId: 'workflow-1',
+          outputs: [{ blockId: 'agent', path: 'score', columnName: 'column-output' }],
+        },
+        latestDeployment()
+      )
+    ).toThrow(
+      'Workflow group group-1 output agent::score is not available in the latest active deployment'
+    )
+  })
+
+  it('rejects an input mapping removed by the latest active deployment', () => {
+    expect(() =>
+      assertWorkflowGroupMatchesLatestDeployment(
+        {
+          id: 'group-1',
+          workflowId: 'workflow-1',
+          outputs: [{ blockId: 'agent', path: 'content', columnName: 'column-output' }],
+          inputMappings: [{ inputName: 'website', columnName: 'column-website' }],
+        },
+        latestDeployment()
+      )
+    ).toThrow(
+      'Workflow group group-1 input website is not available in the latest active deployment'
+    )
+  })
+})
+
 describe('table workflow carrier deadline', () => {
   it('preserves one absolute deadline when a later cascade group creates its controller', () => {
     vi.useFakeTimers()

apps/sim/background/workflow-column-execution.ts

@@ -54,12 +54,57 @@ import {
   type QueuedWorkflowGroupCellPayload,
   type WorkflowGroupCellPayload,
 } from '@/lib/table/workflow-columns'
+import { flattenWorkflowOutputs } from '@/lib/workflows/blocks/flatten-outputs'
+import { normalizeInputFormatValue } from '@/lib/workflows/input-format'
+import type { DeployedWorkflowData } from '@/lib/workflows/persistence/utils'
 import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
 
 export type { WorkflowGroupCellPayload }
 
 const logger = createLogger('TriggerWorkflowGroupCell')
 
+/**
+ * Fails before workflow execution when saved table mappings are incompatible
+ * with the latest active deployment loaded for this cell run.
+ */
+export function assertWorkflowGroupMatchesLatestDeployment(
+  group: WorkflowGroup,
+  deployment: DeployedWorkflowData
+): void {
+  const validOutputs = new Set(
+    flattenWorkflowOutputs(Object.values(deployment.blocks), deployment.edges).map(
+      (output) => `${output.blockId}::${output.path}`
+    )
+  )
+  const invalidOutput = group.outputs.find(
+    (output) => !validOutputs.has(`${output.blockId}::${output.path}`)
+  )
+  if (invalidOutput) {
+    throw new Error(
+      `Workflow group ${group.id} output ${invalidOutput.blockId}::${invalidOutput.path} is not available in the latest active deployment`
+    )
+  }
+
+  const startBlock = Object.values(deployment.blocks).find(
+    (block) => block.type === 'start_trigger'
+  )
+  if (!startBlock) {
+    throw new Error('Workflow is missing a Start trigger')
+  }
+
+  const validInputNames = new Set(
+    normalizeInputFormatValue(startBlock.subBlocks?.inputFormat?.value).map((input) => input.name)
+  )
+  const invalidInput = (group.inputMappings ?? []).find(
+    (mapping) => !validInputNames.has(mapping.inputName)
+  )
+  if (invalidInput) {
+    throw new Error(
+      `Workflow group ${group.id} input ${invalidInput.inputName} is not available in the latest active deployment`
+    )
+  }
+}
+
 function requirePayloadBillingAttribution(
   payload: WorkflowGroupCellPayload
 ): BillingAttributionSnapshot {
@@ -395,9 +440,7 @@ async function runWorkflowAndWriteTerminal(
     return await runWithRequestContext({ requestId }, async () => {
       const { getRowById } = await import('@/lib/table/rows/service')
       const { executeWorkflow } = await import('@/lib/workflows/executor/execute-workflow')
-      const { loadWorkflowDeploymentVersionState } = await import(
-        '@/lib/workflows/persistence/utils'
-      )
+      const { loadDeployedWorkflowState } = await import('@/lib/workflows/persistence/utils')
       const {
         buildCancelledExecution,
         createWorkflowCellProgressWriter,
@@ -689,18 +732,10 @@ async function runWorkflowAndWriteTerminal(
           return 'error'
         }
 
-        let normalizedData: Awaited<ReturnType<typeof loadWorkflowDeploymentVersionState>>
+        let normalizedData: Awaited<ReturnType<typeof loadDeployedWorkflowState>>
         try {
-          if (!group.deploymentVersionId) {
-            throw new Error(
-              `Workflow group ${group.id} has no pinned deployment version; run the table workflow deployment backfill`
-            )
-          }
-          normalizedData = await loadWorkflowDeploymentVersionState(
-            workflowId,
-            group.deploymentVersionId,
-            workspaceId
-          )
+          normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId)
+          assertWorkflowGroupMatchesLatestDeployment(group, normalizedData)
         } catch (err) {
           await writeState({
             status: 'error',

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

@@ -89,7 +89,6 @@ import {
 const group: WorkflowGroup = {
   id: 'group-1',
   workflowId: 'workflow-1',
-  deploymentVersionId: 'deployment-version-1',
   outputs: [{ blockId: 'block-1', path: 'content', columnName: 'column-result' }],
 }
 const table: TableDefinition = {
@@ -148,7 +147,6 @@ const principal = {
 }
 const resolvedWorkflow = {
   workflowId: 'workflow-1',
-  deploymentVersionId: 'deployment-version-1',
   outputs: [
     {
       blockId: 'block-1',
@@ -239,9 +237,6 @@ describe('workflow and enrichment Table application commands', () => {
         ...(input.workflowId ? { workflowId: input.workflowId } : {}),
         ...(input.name ? { name: input.name } : {}),
         ...(input.outputs ? { outputs: input.outputs } : {}),
-        ...(input.resolvedDeployment
-          ? { deploymentVersionId: input.resolvedDeployment.deploymentVersionId }
-          : {}),
         ...(input.autoRun !== undefined ? { autoRun: input.autoRun } : {}),
       })
     )
@@ -276,7 +271,6 @@ describe('workflow and enrichment Table application commands', () => {
         group: expect.objectContaining({
           id: 'generated-id',
           workflowId: 'workflow-1',
-          deploymentVersionId: 'deployment-version-1',
           name: 'Scoring',
           autoRun: false,
           outputs: [{ blockId: 'block-2', path: 'score', columnName: 'score' }],
@@ -538,7 +532,6 @@ describe('workflow and enrichment Table application commands', () => {
       'request-1'
     )
     expect(result.group.workflowId).toBe('workflow-1')
-    expect(result.group.deploymentVersionId).toBe('deployment-version-1')
   })
 
   it('still creates an enrichment-template group that carries a backing workflow', async () => {
@@ -582,64 +575,6 @@ describe('workflow and enrichment Table application commands', () => {
     expect(mocks.audit).not.toHaveBeenCalled()
   })
 
-  it('refuses to repin mappings that the new active deployment cannot produce', async () => {
-    mocks.loadWorkflowOutputs.mockResolvedValueOnce({
-      ...resolvedWorkflow,
-      deploymentVersionId: 'deployment-version-2',
-      outputs: resolvedWorkflow.outputs.filter((output) => output.blockId !== 'block-1'),
-    })
-
-    await expect(
-      updateTableGroupUseCase.execute({
-        principal,
-        input: {
-          tableId: table.id,
-          workspaceId: table.workspaceId,
-          groupId: group.id,
-          workflowId: group.workflowId,
-          outputs: group.outputs,
-        },
-      })
-    ).rejects.toMatchObject({
-      code: 'validation',
-      message: expect.stringContaining('Invalid output(s) for workflow workflow-1'),
-    })
-
-    expect(mocks.updateGroup).not.toHaveBeenCalled()
-  })
-
-  it('pins a compatible active deployment when workflow mappings are saved', async () => {
-    mocks.loadWorkflowOutputs.mockResolvedValueOnce({
-      ...resolvedWorkflow,
-      deploymentVersionId: 'deployment-version-2',
-    })
-
-    await updateTableGroupUseCase.execute({
-      principal,
-      input: {
-        tableId: table.id,
-        workspaceId: table.workspaceId,
-        groupId: group.id,
-        workflowId: group.workflowId,
-        outputs: group.outputs,
-      },
-    })
-
-    expect(mocks.updateGroup).toHaveBeenCalledWith(
-      expect.objectContaining({
-        resolvedDeployment: {
-          workflowId: 'workflow-1',
-          deploymentVersionId: 'deployment-version-2',
-          validOutputCoordinates: [
-            { blockId: 'block-1', path: 'content' },
-            { blockId: 'block-2', path: 'score' },
-          ],
-        },
-      }),
-      'request-1'
-    )
-  })
-
   it('preserves an existing output coordinate that is no longer pickable', async () => {
     mocks.loadWorkflowOutputs.mockResolvedValueOnce({
       ...resolvedWorkflow,
@@ -1130,7 +1065,6 @@ describe('workflow and enrichment Table application commands', () => {
       expect.objectContaining({
         resolvedOutput: expect.objectContaining({
           workflowId: 'workflow-1',
-          deploymentVersionId: 'deployment-version-1',
           columnType: 'number',
           order: expect.arrayContaining([
             expect.objectContaining({ blockId: 'block-2', executionDistance: 2 }),