From 60946880a278d8be9fd89686737d097e1432ef61 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 16:35:47 -0700 Subject: [PATCH 01/13] fix(tables): run workflow groups from deployments --- .../workflow-sidebar/workflow-sidebar.tsx | 27 -- .../background/workflow-column-execution.ts | 46 +-- ...ackfill-table-workflow-deployments.test.ts | 207 ++++++++++ .../backfill-table-workflow-deployments.ts | 370 ++++++++++++++++++ 4 files changed, 591 insertions(+), 59 deletions(-) create mode 100644 apps/sim/scripts/backfill-table-workflow-deployments.test.ts create mode 100644 apps/sim/scripts/backfill-table-workflow-deployments.ts diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx index 762f158a783..33e851bbcf4 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx @@ -3,8 +3,6 @@ import { useMemo, useState } from 'react' import { Button, - ButtonGroup, - ButtonGroupItem, ChipCombobox, ChipInput, type ComboboxOptionGroup, @@ -36,7 +34,6 @@ import type { ColumnDefinition, WorkflowGroup, WorkflowGroupDependencies, - WorkflowGroupDeploymentMode, WorkflowGroupInputMapping, WorkflowGroupOutput, } from '@/lib/table' @@ -312,11 +309,6 @@ export function WorkflowSidebarBody({ const [autoRun, setAutoRun] = useState(() => existingGroup ? existingGroup.autoRun !== false : false ) - // Which workflow state per-cell runs execute against. Defaults to `'live'` - // (the editable draft) for both new and pre-feature groups. - const [deploymentMode, setDeploymentMode] = useState( - () => existingGroup?.deploymentMode ?? 'live' - ) // Deps default to none selected. With auto-run on, at least one is required // (enforced via `depsValid` below); a legacy group with empty deps will // surface the error on first open until the user picks at least one column. @@ -676,7 +668,6 @@ export function WorkflowSidebarBody({ outputs: fullOutputs, ...(newOutputColumns.length > 0 ? { newOutputColumns } : {}), inputMappings: inputMappingsList, - deploymentMode, autoRun, }) toast.success(`Saved "${existingGroup.name ?? 'Workflow'}"`) @@ -708,7 +699,6 @@ export function WorkflowSidebarBody({ dependencies, outputs: groupOutputs, inputMappings: inputMappingsList, - deploymentMode, autoRun, } await addWorkflowGroup.mutateAsync({ group, outputColumns: newOutputColumns }) @@ -993,23 +983,6 @@ export function WorkflowSidebarBody({ {showAdvanced && ( <> - {!isEnrichment && ( - <> -
- - - setDeploymentMode(v === 'deployed' ? 'deployed' : 'live') - } - > - Live - Deployed - -
- - - )} { const { getRowById } = await import('@/lib/table/rows/service') const { executeWorkflow } = await import('@/lib/workflows/executor/execute-workflow') - const { loadWorkflowFromNormalizedTables, loadDeployedWorkflowState } = await import( - '@/lib/workflows/persistence/utils' - ) + const { loadDeployedWorkflowState } = await import('@/lib/workflows/persistence/utils') const { buildCancelledExecution, createWorkflowCellProgressWriter, @@ -693,27 +687,18 @@ async function runWorkflowAndWriteTerminal( return 'error' } - // `deployed` groups run the workflow's latest active deployment; `live` - // (default) runs the editable draft. A `deployed` group whose workflow - // has never been deployed fails the cell — no silent fallback to draft. - let normalizedData: Awaited> - if (deploymentMode === 'deployed') { - try { - normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId) - } catch (err) { - // Surface the real reason (missing deployment vs. transient DB/migration - // failure) rather than always claiming the workflow isn't deployed. - await writeState({ - status: 'error', - executionId, - jobId: null, - workflowId, - error: toError(err).message, - }) - return 'error' - } - } else { - normalizedData = await loadWorkflowFromNormalizedTables(workflowId) + let normalizedData: Awaited> + try { + normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId) + } catch (err) { + await writeState({ + status: 'error', + executionId, + jobId: null, + workflowId, + error: toError(err).message, + }) + return 'error' } const startBlock = normalizedData ? Object.values(normalizedData.blocks).find((b) => b?.type === 'start_trigger') @@ -1007,10 +992,7 @@ async function runWorkflowAndWriteTerminal( executionMode: 'sync', workflowTriggerType: 'table', triggerBlockId: startBlock.id, - // `deployed` groups execute the latest active deployment; everything - // else runs the editable draft (the table default). Matches the - // state loaded above for start-block / output-block resolution. - useDraftState: deploymentMode !== 'deployed', + useDraftState: false, abortSignal: attemptSignal, onBlockStart: progressWriter.onBlockStart, onBlockComplete: progressWriter.onBlockComplete, diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts new file mode 100644 index 00000000000..914c593cb2c --- /dev/null +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -0,0 +1,207 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockPerformFullDeploy } = vi.hoisted(() => ({ + mockPerformFullDeploy: vi.fn(), +})) + +vi.mock('@/lib/workflows/orchestration/deploy', () => ({ + performFullDeploy: mockPerformFullDeploy, +})) + +import { + backfillTableWorkflowDeployments, + deployTableWorkflow, + TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE, + type TableWorkflowDeploymentCandidate, + type TableWorkflowDeploymentStore, +} from '@/scripts/backfill-table-workflow-deployments' + +function candidate(workflowId: string): TableWorkflowDeploymentCandidate { + return { + workflowId, + workspaceId: 'workspace-1', + userId: 'user-1', + } +} + +function store( + overrides: Partial = {} +): TableWorkflowDeploymentStore { + return { + assertIntegrity: vi.fn().mockResolvedValue(undefined), + listCandidates: vi.fn().mockResolvedValue([]), + isDeployed: vi.fn().mockResolvedValue(false), + ...overrides, + } +} + +describe('backfillTableWorkflowDeployments', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('deploys bounded keyset pages and verifies the final desired state', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')]) + .mockResolvedValueOnce([candidate('workflow-c')]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]) + const deploymentState = new Map() + const isDeployed = vi + .fn() + .mockImplementation(async (workflowId) => deploymentState.get(workflowId) ?? false) + const deploy = vi.fn(async (workflow: TableWorkflowDeploymentCandidate) => { + deploymentState.set(workflow.workflowId, true) + return { + success: true, + activeDeployment: { + deploymentVersionId: `version-${workflow.workflowId}`, + version: 1, + deployedAt: new Date().toISOString(), + }, + } + }) + const backfillStore = store({ listCandidates, isDeployed }) + + await expect( + backfillTableWorkflowDeployments(backfillStore, deploy, { batchSize: 2 }) + ).resolves.toEqual({ + scanned: 3, + deployed: 3, + alreadyDeployed: 0, + }) + expect(listCandidates.mock.calls).toEqual([ + ['', 2], + ['workflow-b', 2], + ['workflow-c', 2], + ['', 1], + ]) + expect(backfillStore.assertIntegrity).toHaveBeenCalledTimes(2) + expect(deploy.mock.calls.map(([workflow]) => workflow.workflowId)).toEqual([ + 'workflow-a', + 'workflow-b', + 'workflow-c', + ]) + }) + + it('does not redeploy an already deployed workflow', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a')]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]) + const deploy = vi.fn() + + await expect( + backfillTableWorkflowDeployments( + store({ + listCandidates, + isDeployed: vi.fn().mockResolvedValue(true), + }), + deploy + ) + ).resolves.toEqual({ + scanned: 1, + deployed: 0, + alreadyDeployed: 1, + }) + expect(deploy).not.toHaveBeenCalled() + }) + + it('fails fast when canonical deployment fails', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')]) + const deploy = vi.fn().mockResolvedValue({ + success: false, + error: 'invalid trigger configuration', + }) + + await expect( + backfillTableWorkflowDeployments(store({ listCandidates }), deploy) + ).rejects.toThrow('Failed to deploy table workflow workflow-a: invalid trigger configuration') + expect(deploy).toHaveBeenCalledTimes(1) + }) + + it('fails when deployment does not activate or persist a valid active version', async () => { + const firstList = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a')]) + await expect( + backfillTableWorkflowDeployments(store({ listCandidates: firstList }), async () => ({ + success: true, + activeDeployment: null, + })) + ).rejects.toThrow('did not reach an active deployment state') + + const secondList = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-b')]) + await expect( + backfillTableWorkflowDeployments(store({ listCandidates: secondList }), async () => ({ + success: true, + activeDeployment: { + deploymentVersionId: 'version-b', + version: 1, + deployedAt: new Date().toISOString(), + }, + })) + ).rejects.toThrow('completed without a valid active version') + }) + + it('rejects invalid batch and page behavior before it can loop or skip data', async () => { + const invalidBatchStore = store() + await expect( + backfillTableWorkflowDeployments(invalidBatchStore, vi.fn(), { batchSize: 0 }) + ).rejects.toThrow('positive integer') + expect(invalidBatchStore.assertIntegrity).not.toHaveBeenCalled() + + const oversizedStore = store({ + listCandidates: vi.fn().mockResolvedValue([candidate('workflow-a'), candidate('workflow-b')]), + }) + await expect( + backfillTableWorkflowDeployments(oversizedStore, vi.fn(), { batchSize: 1 }) + ).rejects.toThrow('oversized page') + + const duplicateStore = store({ + listCandidates: vi.fn().mockResolvedValue([candidate('workflow-a'), candidate('workflow-a')]), + }) + await expect( + backfillTableWorkflowDeployments(duplicateStore, vi.fn(), { batchSize: 2 }) + ).rejects.toThrow('duplicate workflow ids') + }) + + it('uses the canonical deployer with backfill attribution and a stable idempotency key', async () => { + mockPerformFullDeploy.mockResolvedValue({ + success: true, + activeDeployment: { + deploymentVersionId: 'version-1', + version: 1, + deployedAt: new Date().toISOString(), + }, + }) + + await deployTableWorkflow(candidate('workflow-1')) + + expect(mockPerformFullDeploy).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + userId: 'user-1', + actorId: 'table-workflow-deployment-backfill', + captureAnalytics: false, + requestId: 'table-workflow-deployment-backfill:v2:workflow-1', + idempotencyKey: 'table-workflow-deployment-backfill:v2:workflow-1', + }) + }) + + it('uses the repository batch-size default', async () => { + const listCandidates = vi.fn().mockResolvedValue([]) + + await backfillTableWorkflowDeployments(store({ listCandidates }), vi.fn()) + + expect(listCandidates).toHaveBeenNthCalledWith(1, '', TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE) + }) +}) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts new file mode 100644 index 00000000000..3b90f50589c --- /dev/null +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -0,0 +1,370 @@ +#!/usr/bin/env bun + +/** + * Deploys every workflow referenced by a table workflow group. + * + * The script is idempotent and resumable: it pages over only workflows that do + * not have the full desired state, and each deployment uses a stable idempotency + * key. It runs the canonical deployment orchestration so webhook, schedule, + * MCP, audit, and notification side effects stay consistent with a + * user-initiated deployment. + * + * Usage: + * DATABASE_URL=... bun run apps/sim/scripts/backfill-table-workflow-deployments.ts + */ + +import { db } from '@sim/db' +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import { sql } from 'drizzle-orm' +import { + type PerformFullDeployResult, + performFullDeploy, +} from '@/lib/workflows/orchestration/deploy' + +const logger = createLogger('BackfillTableWorkflowDeployments') + +export const TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE = 25 +const BACKFILL_ACTOR_ID = 'table-workflow-deployment-backfill' +const BACKFILL_OPERATION_VERSION = 'v2' + +export interface TableWorkflowDeploymentCandidate { + workflowId: string + workspaceId: string + userId: string +} + +export interface TableWorkflowDeploymentStore { + assertIntegrity(): Promise + listCandidates( + afterWorkflowId: string, + limit: number + ): Promise + isDeployed(workflowId: string): Promise +} + +export interface TableWorkflowDeploymentSummary { + scanned: number + deployed: number + alreadyDeployed: number +} + +interface TableWorkflowDeploymentBackfillOptions { + batchSize?: number +} + +export type DeployTableWorkflow = ( + candidate: TableWorkflowDeploymentCandidate +) => Promise + +interface InvalidTableSchemaRow extends Record { + table_id: string +} + +interface InvalidWorkflowGroupRow extends Record { + group_index: string + table_id: string +} + +interface InvalidWorkflowReferenceRow extends Record { + table_id: string + table_workspace_id: string + workflow_id: string + workflow_workspace_id: string | null +} + +interface MultipleActiveVersionsRow extends Record { + active_version_count: number + workflow_id: string +} + +interface CandidateRow extends Record { + user_id: string + workflow_id: string + workspace_id: string +} + +interface DeploymentStateRow extends Record { + active_version_count: number + is_deployed: boolean +} + +/** Ensures the table group references can be traversed without silently dropping corrupt data. */ +async function assertTableWorkflowIntegrity(): Promise { + const [invalidSchema] = await db.execute(sql` + SELECT id AS table_id + FROM user_table_definitions + WHERE schema ? 'workflowGroups' + AND jsonb_typeof(schema->'workflowGroups') IS DISTINCT FROM 'array' + LIMIT 1 + `) + if (invalidSchema) { + throw new Error( + `Table ${invalidSchema.table_id} has a workflowGroups value that is not an array` + ) + } + + const [invalidGroup] = await db.execute(sql` + SELECT + table_definition.id AS table_id, + workflow_group.ordinality::text AS group_index + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) WITH ORDINALITY AS workflow_group(value, ordinality) + WHERE jsonb_typeof(workflow_group.value) IS DISTINCT FROM 'object' + OR NOT (workflow_group.value ? 'workflowId') + OR jsonb_typeof(workflow_group.value->'workflowId') IS DISTINCT FROM 'string' + OR workflow_group.value->>'workflowId' = '' + LIMIT 1 + `) + if (invalidGroup) { + throw new Error( + `Table ${invalidGroup.table_id} workflow group ${invalidGroup.group_index} has an invalid workflowId` + ) + } + + const [invalidReference] = await db.execute(sql` + WITH table_workflow_groups AS ( + SELECT + table_definition.id AS table_id, + table_definition.workspace_id AS table_workspace_id, + workflow_group.value->>'workflowId' AS workflow_id + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + ) + SELECT + table_workflow_groups.table_id, + table_workflow_groups.table_workspace_id, + table_workflow_groups.workflow_id, + workflow.workspace_id AS workflow_workspace_id + FROM table_workflow_groups + LEFT JOIN workflow ON workflow.id = table_workflow_groups.workflow_id + WHERE table_workflow_groups.workflow_id <> '' + AND ( + workflow.id IS NULL + OR ( + workflow.archived_at IS NULL + AND workflow.workspace_id IS DISTINCT FROM table_workflow_groups.table_workspace_id + ) + ) + LIMIT 1 + `) + if (invalidReference) { + const workflowScope = invalidReference.workflow_workspace_id ?? 'missing' + throw new Error( + `Table ${invalidReference.table_id} in workspace ${invalidReference.table_workspace_id} ` + + `references workflow ${invalidReference.workflow_id} in workspace ${workflowScope}` + ) + } + + const [multipleActiveVersions] = await db.execute(sql` + WITH referenced_workflow_ids AS ( + SELECT DISTINCT workflow_group.value->>'workflowId' AS workflow_id + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + INNER JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' + AND workflow.archived_at IS NULL + WHERE workflow_group.value->>'workflowId' <> '' + ) + SELECT + deployment_version.workflow_id, + COUNT(*)::int AS active_version_count + FROM workflow_deployment_version AS deployment_version + INNER JOIN referenced_workflow_ids + ON referenced_workflow_ids.workflow_id = deployment_version.workflow_id + WHERE deployment_version.is_active = true + GROUP BY deployment_version.workflow_id + HAVING COUNT(*) > 1 + LIMIT 1 + `) + if (multipleActiveVersions) { + throw new Error( + `Workflow ${multipleActiveVersions.workflow_id} has ${multipleActiveVersions.active_version_count} active deployment versions` + ) + } +} + +export const postgresTableWorkflowDeploymentStore: TableWorkflowDeploymentStore = { + assertIntegrity: assertTableWorkflowIntegrity, + + async listCandidates(afterWorkflowId, limit) { + const rows = await db.execute(sql` + WITH referenced_workflow_ids AS ( + SELECT DISTINCT workflow_group.value->>'workflowId' AS workflow_id + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + WHERE workflow_group.value->>'workflowId' <> '' + ) + SELECT + workflow.id AS workflow_id, + workflow.workspace_id AS workspace_id, + workflow.user_id + FROM referenced_workflow_ids + INNER JOIN workflow ON workflow.id = referenced_workflow_ids.workflow_id + AND workflow.archived_at IS NULL + WHERE workflow.id COLLATE "C" > ${afterWorkflowId}::text COLLATE "C" + AND ( + workflow.is_deployed = false + OR NOT EXISTS ( + SELECT 1 + FROM workflow_deployment_version AS active_version + WHERE active_version.workflow_id = workflow.id + AND active_version.is_active = true + ) + ) + ORDER BY workflow.id COLLATE "C" + LIMIT ${limit} + `) + + return rows.map((row) => ({ + workflowId: row.workflow_id, + workspaceId: row.workspace_id, + userId: row.user_id, + })) + }, + + async isDeployed(workflowId) { + const [state] = await db.execute(sql` + SELECT + workflow.is_deployed, + (COUNT(deployment_version.id) FILTER (WHERE deployment_version.is_active))::int + AS active_version_count + FROM workflow + LEFT JOIN workflow_deployment_version AS deployment_version + ON deployment_version.workflow_id = workflow.id + WHERE workflow.id = ${workflowId} + GROUP BY workflow.id, workflow.is_deployed + `) + if (!state) { + throw new Error(`Workflow ${workflowId} disappeared during the deployment backfill`) + } + if (state.active_version_count > 1) { + throw new Error( + `Workflow ${workflowId} has ${state.active_version_count} active deployment versions` + ) + } + return state.is_deployed && state.active_version_count === 1 + }, +} + +/** Deploys one table workflow through the same orchestration used by application surfaces. */ +export async function deployTableWorkflow( + candidate: TableWorkflowDeploymentCandidate +): Promise { + return performFullDeploy({ + workflowId: candidate.workflowId, + userId: candidate.userId, + actorId: BACKFILL_ACTOR_ID, + captureAnalytics: false, + requestId: `${BACKFILL_ACTOR_ID}:${BACKFILL_OPERATION_VERSION}:${candidate.workflowId}`, + idempotencyKey: `${BACKFILL_ACTOR_ID}:${BACKFILL_OPERATION_VERSION}:${candidate.workflowId}`, + }) +} + +/** + * Reaches and verifies that every table workflow has an active deployment. + */ +export async function backfillTableWorkflowDeployments( + store: TableWorkflowDeploymentStore, + deploy: DeployTableWorkflow, + options: TableWorkflowDeploymentBackfillOptions = {} +): Promise { + const batchSize = options.batchSize ?? TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE + if (!Number.isInteger(batchSize) || batchSize <= 0) { + throw new Error('Table workflow deployment backfill batch size must be a positive integer') + } + + await store.assertIntegrity() + + const summary: TableWorkflowDeploymentSummary = { + scanned: 0, + deployed: 0, + alreadyDeployed: 0, + } + let afterWorkflowId = '' + + for (;;) { + const candidates = await store.listCandidates(afterWorkflowId, batchSize) + if (candidates.length === 0) break + if (candidates.length > batchSize) { + throw new Error('Table workflow deployment store returned an oversized page') + } + + const pageIds = new Set(candidates.map((candidate) => candidate.workflowId)) + if (pageIds.size !== candidates.length) { + throw new Error('Table workflow deployment store returned duplicate workflow ids') + } + const lastWorkflowId = candidates.at(-1)?.workflowId + if (!lastWorkflowId || lastWorkflowId === afterWorkflowId) { + throw new Error('Table workflow deployment store returned a non-advancing page') + } + + for (const candidate of candidates) { + summary.scanned += 1 + if (await store.isDeployed(candidate.workflowId)) { + summary.alreadyDeployed += 1 + } else { + logger.info('Deploying workflow referenced by a table workflow group', { + workflowId: candidate.workflowId, + workspaceId: candidate.workspaceId, + }) + const result = await deploy(candidate) + if (!result.success) { + throw new Error( + `Failed to deploy table workflow ${candidate.workflowId}: ${result.error ?? 'deployment returned no error'}` + ) + } + if (!result.activeDeployment) { + throw new Error( + `Table workflow ${candidate.workflowId} did not reach an active deployment state` + ) + } + if (!(await store.isDeployed(candidate.workflowId))) { + throw new Error( + `Table workflow ${candidate.workflowId} deployment completed without a valid active version` + ) + } + summary.deployed += 1 + } + } + + afterWorkflowId = lastWorkflowId + } + + await store.assertIntegrity() + const remaining = await store.listCandidates('', 1) + if (remaining.length > 0) { + throw new Error( + `Table workflow deployment backfill left workflow ${remaining[0].workflowId} undeployed` + ) + } + + return summary +} + +export async function runTableWorkflowDeploymentBackfill(): Promise { + logger.info('Starting table workflow deployment backfill') + const summary = await backfillTableWorkflowDeployments( + postgresTableWorkflowDeploymentStore, + deployTableWorkflow + ) + logger.info('Table workflow deployment backfill completed', summary) +} + +if ((import.meta as { main?: boolean }).main) { + runTableWorkflowDeploymentBackfill() + .then(() => process.exit(0)) + .catch((error: unknown) => { + logger.error('Table workflow deployment backfill failed', { + error: getErrorMessage(error), + }) + process.exit(1) + }) +} From f1fa8de8be0b5a5ae40999af57c6a85f50840a87 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 16:52:41 -0700 Subject: [PATCH 02/13] fix(tables): validate deployed workflow mappings --- .../workflow-sidebar/workflow-sidebar.tsx | 162 ++---------------- apps/sim/lib/table/application/groups.test.ts | 2 +- apps/sim/lib/table/application/groups.ts | 4 +- .../resolve-workflow-outputs.test.ts | 51 +++++- .../application/resolve-workflow-outputs.ts | 54 ++++-- 5 files changed, 112 insertions(+), 161 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx index 33e851bbcf4..5536ae391c5 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx @@ -19,17 +19,11 @@ import { import { ArrowLeft, ChevronDown, SquareArrowUpRight, X } from '@sim/emcn/icons' import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' -import { useMutation, useQueryClient } from '@tanstack/react-query' import { findValidationIssue, isValidationError } from '@/lib/api/client/errors' -import { requestJson } from '@/lib/api/client/request' import type { AddWorkflowGroupBodyInput, UpdateWorkflowGroupBodyInput, } from '@/lib/api/contracts/tables' -import { - putWorkflowNormalizedStateContract, - type WorkflowStateContractInput, -} from '@/lib/api/contracts/workflows' import type { ColumnDefinition, WorkflowGroup, @@ -39,7 +33,6 @@ import type { } from '@/lib/table' import { getColumnId } from '@/lib/table/column-keys' import { columnTypeForLeaf, deriveOutputColumnName } from '@/lib/table/column-naming' -import { columnTypeById } from '@/lib/table/column-types' import { type FlattenOutputsBlockInput, type FlattenOutputsEdgeInput, @@ -55,12 +48,12 @@ import { } from '@/app/workspace/[workspaceId]/tables/[tableId]/components/sidebar-fields' import { PreviewWorkflow } from '@/app/workspace/[workspaceId]/w/components/preview' import { BlockTile } from '@/blocks/block-tile' +import { useDeployedWorkflowState } from '@/hooks/queries/deployments' import { useAddWorkflowGroup, useUpdateColumn, useUpdateWorkflowGroup, } from '@/hooks/queries/tables' -import { useWorkflowState, workflowKeys } from '@/hooks/queries/workflows' import type { WorkflowMetadata } from '@/stores/workflows/registry/types' import { InputMappingSection } from './input-mapping-section' import { RunSettingsSection } from './run-settings-section' @@ -139,25 +132,6 @@ interface BlockOutputGroup { paths: string[] } -interface WorkflowStatePayload { - blocks: Record< - string, - { - type: string - subBlocks?: Record - } & Record - > - edges: unknown[] - loops: unknown - parallels: unknown - lastSaved?: number - isDeployed?: boolean -} - -function tableColumnTypeToInputType(colType: ColumnDefinition['type'] | undefined): string { - return columnTypeById(colType).workflowInputType -} - /** * Right-edge sidebar for workflow group configuration. Three flows: * - create a new group (workflow + outputs + deps), @@ -274,11 +248,6 @@ export function WorkflowSidebarBody({ */ const otherColumns = anchorIdx >= allColumns.length ? allColumns : allColumns.slice(0, anchorIdx) - // Used by the "missing workflow input" suggestion below — for edit-output - // we exclude the column being edited (you can't suggest it as its own - // input). - const anchorColumnName = config.mode === 'edit-output' ? config.columnName : null - // Every left-of-current column is a valid dep — workflow output columns // included. Exclude this group's own outputs (you can't depend on yourself). const ownOutputIds = new Set(existingGroup?.outputs.map((o) => o.columnName) ?? []) @@ -324,101 +293,21 @@ export function WorkflowSidebarBody({ const [showValidation, setShowValidation] = useState(false) const [nameError, setNameError] = useState(null) - const workflowState = useWorkflowState(selectedWorkflowId || undefined) + const workflowState = useDeployedWorkflowState(selectedWorkflowId || null) - /** Resolves the unified Start block id and its current `inputFormat` field - * names. The "Add inputs" mutation only adds rows for table columns that - * aren't already represented in the start block. */ - const startBlockInputs = useMemo<{ - blockId: string | null - existingNames: Set - existing: InputFormatField[] - }>(() => { + /** Resolves Start-block inputs from the active deployment used by table runs. */ + const startBlockInputs = useMemo(() => { const blocks = (workflowState.data as { blocks?: Record } | null) ?.blocks - if (!blocks) return { blockId: null, existingNames: new Set(), existing: [] } + if (!blocks) return [] const candidate = TriggerUtils.findStartBlock(blocks, 'manual') - if (!candidate) return { blockId: null, existingNames: new Set(), existing: [] } + if (!candidate) return [] const block = blocks[candidate.blockId] as | { subBlocks?: Record } | undefined - const existing = normalizeInputFormatValue(block?.subBlocks?.inputFormat?.value) - return { - blockId: candidate.blockId, - existingNames: new Set(existing.map((f) => f.name).filter((n): n is string => !!n)), - existing, - } + return normalizeInputFormatValue(block?.subBlocks?.inputFormat?.value) }, [workflowState.data]) - const missingInputColumnNames = useMemo(() => { - if (!startBlockInputs.blockId) return [] - const anchor = anchorColumnName - return allColumns - .filter( - (c) => - getColumnId(c) !== anchor && - !c.workflowGroupId && - !startBlockInputs.existingNames.has(c.name) - ) - .map((c) => c.name) - }, [allColumns, anchorColumnName, startBlockInputs]) - - const queryClient = useQueryClient() - const addInputsMutation = useMutation({ - mutationFn: async () => { - const wfId = selectedWorkflowId - const startBlockId = startBlockInputs.blockId - const state = workflowState.data as WorkflowStatePayload | null | undefined - if (!wfId || !startBlockId || !state || missingInputColumnNames.length === 0) { - throw new Error('Nothing to add') - } - const startBlock = state.blocks[startBlockId] - if (!startBlock) throw new Error('Start block missing from workflow') - - const newFields: InputFormatField[] = missingInputColumnNames.map((name) => { - const col = allColumns.find((c) => c.name === name) - return { - id: generateId(), - name, - type: tableColumnTypeToInputType(col?.type), - value: '', - collapsed: false, - } as InputFormatField & { id: string; collapsed: boolean } - }) - - const updatedSubBlock = { - ...(startBlock.subBlocks?.inputFormat ?? { id: 'inputFormat', type: 'input-format' }), - value: [...startBlockInputs.existing, ...newFields], - } - const updatedBlocks = { - ...state.blocks, - [startBlockId]: { - ...startBlock, - subBlocks: { ...startBlock.subBlocks, inputFormat: updatedSubBlock }, - }, - } - - const rawBody = { - blocks: updatedBlocks, - edges: state.edges, - loops: state.loops, - parallels: state.parallels, - lastSaved: state.lastSaved ?? Date.now(), - isDeployed: state.isDeployed ?? false, - } - // double-cast-allowed: WorkflowStatePayload is the loose local view of - // useWorkflowState; round-trip back to the strict PUT body shape. - const body = rawBody as unknown as WorkflowStateContractInput - await requestJson(putWorkflowNormalizedStateContract, { params: { id: wfId }, body }) - }, - onError: (err) => { - toast.error(toError(err).message) - }, - onSettled: () => { - return queryClient.invalidateQueries({ queryKey: workflowKeys.state(selectedWorkflowId) }) - }, - }) - const blockOutputGroups = useMemo(() => { const state = workflowState.data as | { @@ -509,13 +398,13 @@ export function WorkflowSidebarBody({ // Once the Start block's input fields resolve, auto-fill any field that has no // persisted mapping yet but matches a table column by name. Runs once; never // overrides a persisted or user-picked mapping. - if (!inputMappingsHydrated && startBlockInputs.existing.length > 0) { + if (!inputMappingsHydrated && startBlockInputs.length > 0) { // Map a Start input field to the column sharing its name, storing the // column id (the value the dropdowns and persisted mappings key on). const idByColumnName = new Map(depOptions.map((c) => [c.name, getColumnId(c)])) const next = { ...inputMappings } let changed = false - for (const field of startBlockInputs.existing) { + for (const field of startBlockInputs) { if (!field.name || next[field.name]) continue const colId = idByColumnName.get(field.name) if (colId) { @@ -805,29 +694,6 @@ export function WorkflowSidebarBody({
- {!isEnrichment && - startBlockInputs.blockId && - missingInputColumnNames.length > 0 && ( - - - - - - Adds {missingInputColumnNames.join(', ')} to the workflow's Start block - - - )}
{workflowState.isLoading ? ( @@ -886,12 +752,16 @@ export function WorkflowSidebarBody({
Workflow ({ label: wf.name, value: wf.id })) ?? []} + options={ + workflows + ?.filter((workflow) => workflow.isDeployed) + .map((workflow) => ({ label: workflow.name, value: workflow.id })) ?? [] + } value={selectedWorkflowId} onChange={(v) => setSelectedWorkflowId(v)} placeholder='Select a workflow' disabled={!workflows || workflows.length === 0 || isEditOutputMode || isEnrichment} - emptyMessage='No manual triggers configured' + emptyMessage='No deployed workflows available' maxHeight={260} searchable searchPlaceholder='Search workflows...' @@ -984,7 +854,7 @@ export function WorkflowSidebarBody({ {showAdvanced && ( <> ({ resolveActiveWorkflowApplicationContext: mocks.resolveWorkflowContext, })) vi.mock('@/lib/workflows/application/resolve-workflow-outputs', () => ({ - loadResolvedWorkflowOutputs: mocks.loadWorkflowOutputs, + loadResolvedDeployedWorkflowOutputs: mocks.loadWorkflowOutputs, })) import { v2WorkflowGroupSchema } from '@/lib/api/contracts/v2/tables' diff --git a/apps/sim/lib/table/application/groups.ts b/apps/sim/lib/table/application/groups.ts index f8eef13eaca..5dd93947ef4 100644 --- a/apps/sim/lib/table/application/groups.ts +++ b/apps/sim/lib/table/application/groups.ts @@ -35,7 +35,7 @@ import { } from '@/lib/table/workflow-groups/service' import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context' import type { ResolveWorkflowOutputsResult } from '@/lib/workflows/application/resolve-workflow-outputs' -import { loadResolvedWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' +import { loadResolvedDeployedWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' import { getEnrichment } from '@/enrichments/registry' import type { EnrichmentConfig } from '@/enrichments/types' @@ -64,7 +64,7 @@ async function resolveWorkflowForAuthorizedTableCommand( workflowId, assertedWorkspaceId: workspaceId, }) - return loadResolvedWorkflowOutputs(workflowContext) + return loadResolvedDeployedWorkflowOutputs(workflowContext) } async function resolveRelatedWorkflowForTableRoute( diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts index 6ccf06f983a..f750feea8b8 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts @@ -7,7 +7,9 @@ import { OrchestrationError } from '@/lib/core/orchestration/types' const mocks = vi.hoisted(() => ({ flatten: vi.fn(), + loadDeployed: vi.fn(), load: vi.fn(), + NoActiveDeploymentError: class NoActiveDeploymentError extends Error {}, order: vi.fn(), resolveContext: vi.fn(), resolvePermission: vi.fn(), @@ -33,10 +35,15 @@ vi.mock('@/lib/workflows/blocks/flatten-outputs', () => ({ })) vi.mock('@/lib/workflows/persistence/utils', () => ({ + NoActiveDeploymentError: mocks.NoActiveDeploymentError, + loadDeployedWorkflowState: mocks.loadDeployed, loadWorkflowFromNormalizedTables: mocks.load, })) -import { resolveWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' +import { + loadResolvedDeployedWorkflowOutputs, + resolveWorkflowOutputs, +} from '@/lib/workflows/application/resolve-workflow-outputs' const principal = { kind: 'delegated' as const, @@ -59,12 +66,16 @@ describe('resolveWorkflowOutputs', () => { workspaceOrganizationId: null, allowPersonalApiKeys: true, billedAccountUserId: 'billing-owner-1', - workflow: { id: 'workflow-1' }, + workflow: { id: 'workflow-1', isDeployed: true }, }) mocks.load.mockResolvedValue({ blocks: { block1: { id: 'block-1', type: 'agent', name: 'Agent', subBlocks: {} } }, edges: [], }) + mocks.loadDeployed.mockResolvedValue({ + blocks: { block1: { id: 'block-1', type: 'agent', name: 'Agent', subBlocks: {} } }, + edges: [], + }) mocks.flatten.mockReturnValue([ { blockId: 'block-1', @@ -110,6 +121,42 @@ describe('resolveWorkflowOutputs', () => { expect(mocks.load).not.toHaveBeenCalled() }) + it('resolves table mappings from the active deployment state', async () => { + const context = await mocks.resolveContext() + + await expect(loadResolvedDeployedWorkflowOutputs(context)).resolves.toMatchObject({ + workflowId: 'workflow-1', + outputs: [{ blockId: 'block-1', path: 'content' }], + }) + + expect(mocks.loadDeployed).toHaveBeenCalledWith('workflow-1', 'workspace-1') + expect(mocks.load).not.toHaveBeenCalled() + }) + + it('rejects a workflow without an active deployment before resolving mappings', async () => { + const context = { + ...(await mocks.resolveContext()), + workflow: { id: 'workflow-1', isDeployed: false }, + } + + await expect(loadResolvedDeployedWorkflowOutputs(context)).rejects.toMatchObject({ + code: 'validation', + message: 'Workflow must have an active deployment', + }) + expect(mocks.loadDeployed).not.toHaveBeenCalled() + }) + + it('rejects inconsistent deployment metadata without returning draft mappings', async () => { + const context = await mocks.resolveContext() + mocks.loadDeployed.mockRejectedValueOnce(new mocks.NoActiveDeploymentError()) + + await expect(loadResolvedDeployedWorkflowOutputs(context)).rejects.toMatchObject({ + code: 'validation', + message: 'Workflow must have an active deployment', + }) + expect(mocks.load).not.toHaveBeenCalled() + }) + it('rejects expired delegated scope before loading workflow state', async () => { await expect( resolveWorkflowOutputs.execute({ diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts index 209d26107f6..cb60823c006 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts @@ -1,3 +1,4 @@ +import { OrchestrationError } from '@/lib/core/orchestration/types' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { type ActiveWorkflowApplicationContext, @@ -9,7 +10,11 @@ import { flattenWorkflowOutputs, getBlockExecutionOrder, } from '@/lib/workflows/blocks/flatten-outputs' -import { loadWorkflowFromNormalizedTables } from '@/lib/workflows/persistence/utils' +import { + loadDeployedWorkflowState, + loadWorkflowFromNormalizedTables, + NoActiveDeploymentError, +} from '@/lib/workflows/persistence/utils' export interface ResolveWorkflowOutputsInput { workflowId: string @@ -22,14 +27,14 @@ export interface ResolveWorkflowOutputsResult { executionOrderByBlockId: Record } -/** Loads output metadata after a top-level application command has authorized this workflow context. */ -export async function loadResolvedWorkflowOutputs( - context: ActiveWorkflowApplicationContext -): Promise { - const normalized = await loadWorkflowFromNormalizedTables(context.workflowId) - if (!normalized) { - return { workflowId: context.workflowId, outputs: null, executionOrderByBlockId: {} } - } +type ResolvableWorkflowState = + | NonNullable>> + | Awaited> + +function resolveWorkflowOutputsFromState( + workflowId: string, + normalized: ResolvableWorkflowState +): ResolveWorkflowOutputsResult { const blocks = Object.values(normalized.blocks ?? {}).map((block) => ({ id: block.id, type: block.type, @@ -38,12 +43,41 @@ export async function loadResolvedWorkflowOutputs( subBlocks: block.subBlocks as Record | undefined, })) return { - workflowId: context.workflowId, + workflowId, outputs: flattenWorkflowOutputs(blocks, normalized.edges ?? []), executionOrderByBlockId: getBlockExecutionOrder(blocks, normalized.edges ?? []), } } +/** Loads output metadata after a top-level application command has authorized this workflow context. */ +export async function loadResolvedWorkflowOutputs( + context: ActiveWorkflowApplicationContext +): Promise { + const normalized = await loadWorkflowFromNormalizedTables(context.workflowId) + if (!normalized) { + return { workflowId: context.workflowId, outputs: null, executionOrderByBlockId: {} } + } + return resolveWorkflowOutputsFromState(context.workflowId, normalized) +} + +/** Loads output metadata from the active deployment after workflow authorization. */ +export async function loadResolvedDeployedWorkflowOutputs( + context: ActiveWorkflowApplicationContext +): Promise { + if (!context.workflow.isDeployed) { + throw new OrchestrationError('validation', 'Workflow must have an active deployment') + } + try { + const normalized = await loadDeployedWorkflowState(context.workflowId, context.workspaceId) + return resolveWorkflowOutputsFromState(context.workflowId, normalized) + } catch (error) { + if (error instanceof NoActiveDeploymentError) { + throw new OrchestrationError('validation', 'Workflow must have an active deployment') + } + throw error + } +} + export const resolveWorkflowOutputs = defineAuthorizedWorkflowUseCase({ operation: workflowOperations.read, resolveContext: ({ input }: { input: ResolveWorkflowOutputsInput }) => From 2510780ff5635b22039f7fa2eafcb9895e522d0d Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 16:59:09 -0700 Subject: [PATCH 03/13] fix(tables): pin table workflow deployment state --- .../background/workflow-column-execution.ts | 1 + .../executor/execute-workflow.test.ts | 22 +++++++++++++++++++ .../workflows/executor/execute-workflow.ts | 3 +++ .../backfill-table-workflow-deployments.ts | 9 +++++++- 4 files changed, 34 insertions(+), 1 deletion(-) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index b747e070e88..48015457f55 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -993,6 +993,7 @@ async function runWorkflowAndWriteTerminal( workflowTriggerType: 'table', triggerBlockId: startBlock.id, useDraftState: false, + workflowStateOverride: normalizedData, abortSignal: attemptSignal, onBlockStart: progressWriter.onBlockStart, onBlockComplete: progressWriter.onBlockComplete, diff --git a/apps/sim/lib/workflows/executor/execute-workflow.test.ts b/apps/sim/lib/workflows/executor/execute-workflow.test.ts index 4240058f7d3..984cf79f478 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.test.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.test.ts @@ -234,6 +234,28 @@ describe('executeWorkflow', () => { ) }) + it('forwards a trusted immutable workflow state to the execution snapshot', async () => { + const workflowStateOverride = { + blocks: { 'block-1': { id: 'block-1', type: 'start_trigger' } }, + edges: [], + loops: {}, + parallels: {}, + deploymentVersionId: 'deployment-version-1', + } + + await executeWorkflow(workflow, 'request-1', { prompt: 'hello' }, 'actor-1', { + enabled: true, + principal, + billingAttribution, + workflowStateOverride, + }) + + const coreParams = executeWorkflowCoreMock.mock.calls[0]?.[0] as { + snapshot: ExecutionSnapshot + } + expect(coreParams.snapshot.metadata.workflowStateOverride).toEqual(workflowStateOverride) + }) + it('waits for post-execution persistence before resolving', async () => { let resolvePostExecution!: () => void waitForPostExecutionMock.mockReturnValueOnce( diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts index 8336ea332d2..f4994ca5a6d 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.ts @@ -48,6 +48,8 @@ export interface ExecuteWorkflowOptions { abortSignal?: AbortSignal /** Use the live/draft workflow state instead of the deployed state. Used by copilot. */ useDraftState?: boolean + /** Immutable workflow state selected by a trusted server-side trigger boundary. */ + workflowStateOverride?: NonNullable /** Stop execution after this block completes. Used for "run until block" feature. */ stopAfterBlockId?: string /** Run-from-block configuration using a prior execution snapshot. */ @@ -142,6 +144,7 @@ export async function executeWorkflow( triggerType, triggerBlockId: streamConfig?.triggerBlockId, useDraftState: streamConfig?.useDraftState ?? false, + workflowStateOverride: streamConfig?.workflowStateOverride, startTime: new Date().toISOString(), isClientSession: false, enforceCredentialAccess: streamConfig?.enforceCredentialAccess ?? false, diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index 3b90f50589c..82ed9180ab0 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -115,7 +115,14 @@ async function assertTableWorkflowIntegrity(): Promise { WHERE jsonb_typeof(workflow_group.value) IS DISTINCT FROM 'object' OR NOT (workflow_group.value ? 'workflowId') OR jsonb_typeof(workflow_group.value->'workflowId') IS DISTINCT FROM 'string' - OR workflow_group.value->>'workflowId' = '' + OR ( + workflow_group.value->>'workflowId' = '' + AND ( + NOT (workflow_group.value ? 'enrichmentId') + OR jsonb_typeof(workflow_group.value->'enrichmentId') IS DISTINCT FROM 'string' + OR workflow_group.value->>'enrichmentId' = '' + ) + ) LIMIT 1 `) if (invalidGroup) { From 66665231fc1815346c9da801e9f928896d4737bf Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 17:03:57 -0700 Subject: [PATCH 04/13] fix(tables): pin deployed workflow variables --- apps/sim/executor/execution/types.ts | 1 + apps/sim/lib/workflows/executor/execute-workflow.test.ts | 4 ++++ apps/sim/lib/workflows/executor/execute-workflow.ts | 2 +- 3 files changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/sim/executor/execution/types.ts b/apps/sim/executor/execution/types.ts index 35cf038381f..ba7e7db6a8a 100644 --- a/apps/sim/executor/execution/types.ts +++ b/apps/sim/executor/execution/types.ts @@ -56,6 +56,7 @@ export interface ExecutionMetadata { edges: Edge[] loops?: Record parallels?: Record + variables?: Record deploymentVersionId?: string } largeValueExecutionIds?: string[] diff --git a/apps/sim/lib/workflows/executor/execute-workflow.test.ts b/apps/sim/lib/workflows/executor/execute-workflow.test.ts index 984cf79f478..664ff6f511a 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.test.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.test.ts @@ -240,6 +240,9 @@ describe('executeWorkflow', () => { edges: [], loops: {}, parallels: {}, + variables: { + 'variable-1': { id: 'variable-1', name: 'deployed', value: 'frozen' }, + }, deploymentVersionId: 'deployment-version-1', } @@ -254,6 +257,7 @@ describe('executeWorkflow', () => { snapshot: ExecutionSnapshot } expect(coreParams.snapshot.metadata.workflowStateOverride).toEqual(workflowStateOverride) + expect(coreParams.snapshot.workflowVariables).toEqual(workflowStateOverride.variables) }) it('waits for post-execution persistence before resolving', async () => { diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts index f4994ca5a6d..b2589685fb3 100644 --- a/apps/sim/lib/workflows/executor/execute-workflow.ts +++ b/apps/sim/lib/workflows/executor/execute-workflow.ts @@ -166,7 +166,7 @@ export async function executeWorkflow( metadata, workflow, input, - workflow.variables || {}, + streamConfig?.workflowStateOverride?.variables ?? workflow.variables ?? {}, streamConfig?.selectedOutputs || [] ) From 66b5951bf45b2c42d3820d8449c0fe4ba20f806c Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 17:21:31 -0700 Subject: [PATCH 05/13] fix(tables): pin workflow group deployment versions --- .../background/workflow-column-execution.ts | 17 +- apps/sim/lib/table/application/groups.test.ts | 66 ++++++++ apps/sim/lib/table/application/groups.ts | 80 ++++++++-- apps/sim/lib/table/types.ts | 8 + apps/sim/lib/table/workflow-groups/service.ts | 40 ++++- .../resolve-workflow-outputs.test.ts | 2 + .../application/resolve-workflow-outputs.ts | 11 +- ...ackfill-table-workflow-deployments.test.ts | 7 +- .../backfill-table-workflow-deployments.ts | 151 +++++++++++++++++- 9 files changed, 362 insertions(+), 20 deletions(-) diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 48015457f55..5009492f5c4 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -395,7 +395,9 @@ 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 { loadDeployedWorkflowState } = await import('@/lib/workflows/persistence/utils') + const { loadWorkflowDeploymentVersionState } = await import( + '@/lib/workflows/persistence/utils' + ) const { buildCancelledExecution, createWorkflowCellProgressWriter, @@ -687,9 +689,18 @@ async function runWorkflowAndWriteTerminal( return 'error' } - let normalizedData: Awaited> + let normalizedData: Awaited> try { - normalizedData = await loadDeployedWorkflowState(workflowId, workspaceId) + 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 + ) } catch (err) { await writeState({ status: 'error', diff --git a/apps/sim/lib/table/application/groups.test.ts b/apps/sim/lib/table/application/groups.test.ts index 59f3135caa7..24b566321d3 100644 --- a/apps/sim/lib/table/application/groups.test.ts +++ b/apps/sim/lib/table/application/groups.test.ts @@ -89,6 +89,7 @@ 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 = { @@ -147,6 +148,7 @@ const principal = { } const resolvedWorkflow = { workflowId: 'workflow-1', + deploymentVersionId: 'deployment-version-1', outputs: [ { blockId: 'block-1', @@ -237,6 +239,9 @@ 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 } : {}), }) ) @@ -271,6 +276,7 @@ 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' }], @@ -532,6 +538,7 @@ 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 () => { @@ -575,6 +582,64 @@ 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, @@ -1065,6 +1130,7 @@ 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 }), diff --git a/apps/sim/lib/table/application/groups.ts b/apps/sim/lib/table/application/groups.ts index 5dd93947ef4..50513f5d679 100644 --- a/apps/sim/lib/table/application/groups.ts +++ b/apps/sim/lib/table/application/groups.ts @@ -34,7 +34,10 @@ import { updateWorkflowGroup, } from '@/lib/table/workflow-groups/service' import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context' -import type { ResolveWorkflowOutputsResult } from '@/lib/workflows/application/resolve-workflow-outputs' +import type { + ResolveDeployedWorkflowOutputsResult, + ResolveWorkflowOutputsResult, +} from '@/lib/workflows/application/resolve-workflow-outputs' import { loadResolvedDeployedWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' import { getEnrichment } from '@/enrichments/registry' import type { EnrichmentConfig } from '@/enrichments/types' @@ -59,7 +62,7 @@ function groupFromTable(table: TableDefinition, groupId: string): WorkflowGroup async function resolveWorkflowForAuthorizedTableCommand( workflowId: string, workspaceId: string -): Promise { +): Promise { const workflowContext = await resolveActiveWorkflowApplicationContext({ workflowId, assertedWorkspaceId: workspaceId, @@ -70,7 +73,7 @@ async function resolveWorkflowForAuthorizedTableCommand( async function resolveRelatedWorkflowForTableRoute( workflowId: string, workspaceId: string -): Promise { +): Promise { try { return await resolveWorkflowForAuthorizedTableCommand(workflowId, workspaceId) } catch (error) { @@ -245,6 +248,7 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({ * with a backing workflow and workflow output coordinates, and only a group * with no workflow is filled from the enrichment registry. */ + let deploymentVersionId: string | undefined if (input.group.workflowId) { const resolvedWorkflow = await resolveRelatedWorkflowForTableRoute( input.group.workflowId, @@ -258,6 +262,7 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({ resolvedWorkflow, input.group.workflowId ) + deploymentVersionId = resolvedWorkflow.deploymentVersionId } else if (input.group.enrichmentId) { requireKnownEnrichmentOutputIds( requireEnrichment(input.group.enrichmentId), @@ -284,6 +289,7 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({ ...input.group, id: groupId, workflowId: input.group.workflowId ?? '', + ...(deploymentVersionId ? { deploymentVersionId } : {}), outputs: input.group.outputs.map((output) => ({ ...output, blockId: output.blockId ?? '', @@ -404,6 +410,7 @@ export const createWorkflowTableGroup = defineAuthorizedTableUseCase({ const group: WorkflowGroup = { id: groupId, workflowId: input.workflowId, + deploymentVersionId: resolvedWorkflow.deploymentVersionId, ...(input.name ? { name: input.name } : {}), ...(input.dependencies ? { dependencies: input.dependencies } : {}), ...(input.deploymentMode ? { deploymentMode: input.deploymentMode } : {}), @@ -719,9 +726,10 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({ const workflowMetadataRequired = input.workflowId !== undefined || outputCoordinatesToValidate.length > 0 || - (input.mappingUpdates?.length ?? 0) > 0 + (input.mappingUpdates?.length ?? 0) > 0 || + input.inputMappings !== undefined const targetWorkflowId = input.workflowId ?? previousGroup?.workflowId - let resolvedWorkflow: ResolveWorkflowOutputsResult | undefined + let resolvedWorkflow: ResolveDeployedWorkflowOutputsResult | undefined if (workflowMetadataRequired) { if (!targetWorkflowId) { throw new OrchestrationError('not_found', 'Workflow not found') @@ -730,8 +738,19 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({ targetWorkflowId, context.workspaceId ) - if (outputCoordinatesToValidate.length > 0) { - validateRequestedOutputs(outputCoordinatesToValidate, resolvedWorkflow, targetWorkflowId) + const remappedPreviousOutputs = (previousGroup?.outputs ?? []).map((output) => { + const mapping = input.mappingUpdates?.find( + (candidate) => candidate.columnName === output.columnName + ) + return mapping ? { ...output, blockId: mapping.blockId, path: mapping.path } : output + }) + const resultingOutputs = input.outputs ?? remappedPreviousOutputs + const deploymentChanged = + resolvedWorkflow.deploymentVersionId !== previousGroup?.deploymentVersionId + const coordinatesToValidate = + workflowChanged || deploymentChanged ? resultingOutputs : outputCoordinatesToValidate + if (coordinatesToValidate.length > 0) { + validateRequestedOutputs(coordinatesToValidate, resolvedWorkflow, targetWorkflowId) } } const actorUserId = attributedUserId(principal, context.billedAccountUserId) @@ -779,6 +798,18 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({ : {}), ...(input.mappingUpdates !== undefined ? { mappingUpdates: input.mappingUpdates } : {}), ...(resolvedMappingTypes ? { resolvedMappingTypes } : {}), + ...(resolvedWorkflow + ? { + resolvedDeployment: { + workflowId: resolvedWorkflow.workflowId, + deploymentVersionId: resolvedWorkflow.deploymentVersionId, + validOutputCoordinates: (resolvedWorkflow.outputs ?? []).map((output) => ({ + blockId: output.blockId, + path: output.path, + })), + }, + } + : {}), ...(input.inputMappings !== undefined ? { inputMappings: input.inputMappings } : {}), ...(input.deploymentMode !== undefined ? { deploymentMode: input.deploymentMode } : {}), ...(input.type !== undefined ? { type: input.type } : {}), @@ -870,10 +901,19 @@ export const updateWorkflowTableGroup = defineAuthorizedTableUseCase({ const resolvedWorkflow = workflowMetadataRequired ? await resolveWorkflowForAuthorizedTableCommand(targetWorkflowId, context.workspaceId) : undefined - if (input.outputs && resolvedWorkflow) { - validateRequestedOutputs(input.outputs, resolvedWorkflow, targetWorkflowId) - } else if (input.workflowId && resolvedWorkflow) { - validateRequestedOutputs(previousGroup.outputs, resolvedWorkflow, targetWorkflowId) + if (resolvedWorkflow) { + const remappedPreviousOutputs = previousGroup.outputs.map((output) => { + const mapping = input.mappingUpdates?.find( + (candidate) => candidate.columnName === output.columnName + ) + return mapping ? { ...output, blockId: mapping.blockId, path: mapping.path } : output + }) + const resultingOutputs = input.outputs ?? remappedPreviousOutputs + const deploymentChanged = + resolvedWorkflow.deploymentVersionId !== previousGroup.deploymentVersionId + if (input.outputs || input.workflowId || deploymentChanged) { + validateRequestedOutputs(resultingOutputs, resolvedWorkflow, targetWorkflowId) + } } let outputs: WorkflowGroupOutput[] | undefined @@ -970,6 +1010,18 @@ export const updateWorkflowTableGroup = defineAuthorizedTableUseCase({ ...(newOutputColumns !== undefined ? { newOutputColumns } : {}), ...(input.mappingUpdates !== undefined ? { mappingUpdates: input.mappingUpdates } : {}), ...(resolvedMappingTypes ? { resolvedMappingTypes } : {}), + ...(resolvedWorkflow + ? { + resolvedDeployment: { + workflowId: resolvedWorkflow.workflowId, + deploymentVersionId: resolvedWorkflow.deploymentVersionId, + validOutputCoordinates: (resolvedWorkflow.outputs ?? []).map((output) => ({ + blockId: output.blockId, + path: output.path, + })), + }, + } + : {}), ...(input.deploymentMode !== undefined ? { deploymentMode: input.deploymentMode } : {}), ...(input.autoRun !== undefined ? { autoRun: input.autoRun } : {}), }, @@ -1079,6 +1131,11 @@ export const addWorkflowTableGroupOutput = defineAuthorizedTableUseCase({ context.workspaceId ) const outputs = requireWorkflowOutputs(resolvedWorkflow, group.workflowId) + validateRequestedOutputs( + [...group.outputs, { blockId: input.blockId, path: input.path }], + resolvedWorkflow, + group.workflowId + ) const output = outputs.find( (candidate) => candidate.blockId === input.blockId && candidate.path === input.path ) @@ -1101,6 +1158,7 @@ export const addWorkflowTableGroupOutput = defineAuthorizedTableUseCase({ }).attributedUserId, resolvedOutput: { workflowId: resolvedWorkflow.workflowId, + deploymentVersionId: resolvedWorkflow.deploymentVersionId, columnType: columnTypeForLeaf(output.leafType), order: outputs.map((candidate, discoveryIndex) => { const distance = resolvedWorkflow.executionOrderByBlockId[candidate.blockId] diff --git a/apps/sim/lib/table/types.ts b/apps/sim/lib/table/types.ts index 9319e794c10..92e4d27ae6a 100644 --- a/apps/sim/lib/table/types.ts +++ b/apps/sim/lib/table/types.ts @@ -138,6 +138,8 @@ export interface WorkflowGroup { id: string /** Backing workflow id for `manual` groups. `''` for enrichment groups. */ workflowId: string + /** Immutable deployment version whose input and output coordinates this group stores. */ + deploymentVersionId?: string /** Registry enrichment id for `enrichment` groups. */ enrichmentId?: string /** Display name; defaults to the workflow's / enrichment's name. */ @@ -966,6 +968,12 @@ export interface UpdateWorkflowGroupData { workflowId: string columns: Array<{ columnName: string; type: ColumnDefinition['type'] }> } + /** Workflow-authorized deployment snapshot to pin after this update. */ + resolvedDeployment?: { + workflowId: string + deploymentVersionId: string + validOutputCoordinates: Array<{ blockId: string; path: string }> + } /** Replace the group's input mappings. Omit to leave them unchanged. */ inputMappings?: WorkflowGroupInputMapping[] /** Change which workflow state the group runs against. Omit to leave unchanged. */ diff --git a/apps/sim/lib/table/workflow-groups/service.ts b/apps/sim/lib/table/workflow-groups/service.ts index 7f526593b5a..ccc30d89b1b 100644 --- a/apps/sim/lib/table/workflow-groups/service.ts +++ b/apps/sim/lib/table/workflow-groups/service.ts @@ -296,6 +296,13 @@ export async function updateWorkflowGroup( throw new OrchestrationError('not_found', `Workflow group "${data.groupId}" not found`) } const group = groups[groupIndex] + const finalWorkflowId = data.workflowId ?? group.workflowId + if (data.resolvedDeployment && data.resolvedDeployment.workflowId !== finalWorkflowId) { + throw new OrchestrationError( + 'conflict', + `Workflow group "${data.groupId}" changed concurrently; retry the update.` + ) + } // Normalize every caller-supplied column reference to its stable id, so // the diff/splice/clear logic below operates uniformly in id-space (the @@ -361,7 +368,6 @@ export async function updateWorkflowGroup( // Only apply the out-of-lock leaf-type resolution if the group still // points at the workflow we resolved against. A concurrent workflow // remap invalidates the command snapshot and must be retried. - const finalWorkflowId = data.workflowId ?? group.workflowId if (remapLeafTypeById.size > 0 && resolvedForWorkflowId !== finalWorkflowId) { throw new OrchestrationError( 'conflict', @@ -382,6 +388,22 @@ export async function updateWorkflowGroup( // If the caller passed `outputs`, that's the new full set. If only // `mappingUpdates` was sent, the new set is the remapped old set. const newOutputs = outputsInput ?? oldOutputs + if (data.resolvedDeployment) { + const validCoordinates = new Set( + data.resolvedDeployment.validOutputCoordinates.map( + (output) => `${output.blockId}::${output.path}` + ) + ) + const invalidOutput = newOutputs.find( + (output) => !validCoordinates.has(`${output.blockId}::${output.path}`) + ) + if (invalidOutput) { + throw new OrchestrationError( + 'conflict', + `Workflow group "${data.groupId}" mappings changed concurrently; retry the update.` + ) + } + } // Enrichment outputs all share empty `blockId`/`path`, so keying on those // alone collapses every sibling to one entry (dropping columns on diff). Key // on the registry `outputId` when present; fall back to `blockId::path` for @@ -445,7 +467,10 @@ export async function updateWorkflowGroup( const updatedGroup: WorkflowGroup = { ...group, - workflowId: data.workflowId ?? group.workflowId, + workflowId: finalWorkflowId, + ...(data.resolvedDeployment + ? { deploymentVersionId: data.resolvedDeployment.deploymentVersionId } + : {}), name: data.name ?? group.name, dependencies: dependenciesInput ?? group.dependencies, outputs: newOutputs, @@ -629,6 +654,7 @@ export async function addWorkflowGroupOutput( actorUserId?: string | null resolvedOutput: { workflowId: string + deploymentVersionId: string columnType: ColumnDefinition['type'] order: Array<{ blockId: string @@ -741,9 +767,19 @@ export async function addWorkflowGroupOutput( const [db, ib] = orderKey(b) return da !== db ? da - db : ia - ib }) + const invalidOutput = allGroupOutputs.find( + (output) => !resolvedOrder.has(`${output.blockId}::${output.path}`) + ) + if (invalidOutput) { + throw new OrchestrationError( + 'conflict', + `Workflow group "${data.groupId}" mappings changed concurrently; retry the add.` + ) + } const orderedGroupColIds = allGroupOutputs.map((o) => o.columnName) const updatedGroup: WorkflowGroup = { ...group, + deploymentVersionId: data.resolvedOutput.deploymentVersionId, outputs: allGroupOutputs, } const nextGroups = groups.map((g, i) => (i === groupIndex ? updatedGroup : g)) diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts index f750feea8b8..8aa9152c24a 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts @@ -75,6 +75,7 @@ describe('resolveWorkflowOutputs', () => { mocks.loadDeployed.mockResolvedValue({ blocks: { block1: { id: 'block-1', type: 'agent', name: 'Agent', subBlocks: {} } }, edges: [], + deploymentVersionId: 'deployment-version-1', }) mocks.flatten.mockReturnValue([ { @@ -126,6 +127,7 @@ describe('resolveWorkflowOutputs', () => { await expect(loadResolvedDeployedWorkflowOutputs(context)).resolves.toMatchObject({ workflowId: 'workflow-1', + deploymentVersionId: 'deployment-version-1', outputs: [{ blockId: 'block-1', path: 'content' }], }) diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts index cb60823c006..3334feb8bb7 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts @@ -27,6 +27,10 @@ export interface ResolveWorkflowOutputsResult { executionOrderByBlockId: Record } +export interface ResolveDeployedWorkflowOutputsResult extends ResolveWorkflowOutputsResult { + deploymentVersionId: string +} + type ResolvableWorkflowState = | NonNullable>> | Awaited> @@ -63,13 +67,16 @@ export async function loadResolvedWorkflowOutputs( /** Loads output metadata from the active deployment after workflow authorization. */ export async function loadResolvedDeployedWorkflowOutputs( context: ActiveWorkflowApplicationContext -): Promise { +): Promise { if (!context.workflow.isDeployed) { throw new OrchestrationError('validation', 'Workflow must have an active deployment') } try { const normalized = await loadDeployedWorkflowState(context.workflowId, context.workspaceId) - return resolveWorkflowOutputsFromState(context.workflowId, normalized) + return { + ...resolveWorkflowOutputsFromState(context.workflowId, normalized), + deploymentVersionId: normalized.deploymentVersionId, + } } catch (error) { if (error instanceof NoActiveDeploymentError) { throw new OrchestrationError('validation', 'Workflow must have an active deployment') diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts index 914c593cb2c..d09b8b6f103 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -34,6 +34,7 @@ function store( assertIntegrity: vi.fn().mockResolvedValue(undefined), listCandidates: vi.fn().mockResolvedValue([]), isDeployed: vi.fn().mockResolvedValue(false), + pinActiveDeploymentVersions: vi.fn().mockResolvedValue(0), ...overrides, } } @@ -65,7 +66,8 @@ describe('backfillTableWorkflowDeployments', () => { }, } }) - const backfillStore = store({ listCandidates, isDeployed }) + const pinActiveDeploymentVersions = vi.fn().mockResolvedValue(3) + const backfillStore = store({ listCandidates, isDeployed, pinActiveDeploymentVersions }) await expect( backfillTableWorkflowDeployments(backfillStore, deploy, { batchSize: 2 }) @@ -73,6 +75,7 @@ describe('backfillTableWorkflowDeployments', () => { scanned: 3, deployed: 3, alreadyDeployed: 0, + pinnedGroups: 3, }) expect(listCandidates.mock.calls).toEqual([ ['', 2], @@ -81,6 +84,7 @@ describe('backfillTableWorkflowDeployments', () => { ['', 1], ]) expect(backfillStore.assertIntegrity).toHaveBeenCalledTimes(2) + expect(pinActiveDeploymentVersions).toHaveBeenCalledWith(2) expect(deploy.mock.calls.map(([workflow]) => workflow.workflowId)).toEqual([ 'workflow-a', 'workflow-b', @@ -108,6 +112,7 @@ describe('backfillTableWorkflowDeployments', () => { scanned: 1, deployed: 0, alreadyDeployed: 1, + pinnedGroups: 0, }) expect(deploy).not.toHaveBeenCalled() }) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index 82ed9180ab0..c0f5f1ee1b0 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -1,7 +1,8 @@ #!/usr/bin/env bun /** - * Deploys every workflow referenced by a table workflow group. + * Deploys every workflow referenced by a table workflow group and pins each + * group to the active deployment whose mappings it uses. * * The script is idempotent and resumable: it pages over only workflows that do * not have the full desired state, and each deployment uses a stable idempotency @@ -41,12 +42,14 @@ export interface TableWorkflowDeploymentStore { limit: number ): Promise isDeployed(workflowId: string): Promise + pinActiveDeploymentVersions(batchSize: number): Promise } export interface TableWorkflowDeploymentSummary { scanned: number deployed: number alreadyDeployed: number + pinnedGroups: number } interface TableWorkflowDeploymentBackfillOptions { @@ -89,6 +92,20 @@ interface DeploymentStateRow extends Record { is_deployed: boolean } +interface PinnedGroupCountRow extends Record { + pinned_group_count: number +} + +interface TablePinCandidateRow extends Record { + table_id: string +} + +interface UnpinnedWorkflowGroupRow extends Record { + group_id: string + table_id: string + workflow_id: string +} + /** Ensures the table group references can be traversed without silently dropping corrupt data. */ async function assertTableWorkflowIntegrity(): Promise { const [invalidSchema] = await db.execute(sql` @@ -259,6 +276,135 @@ export const postgresTableWorkflowDeploymentStore: TableWorkflowDeploymentStore } return state.is_deployed && state.active_version_count === 1 }, + + async pinActiveDeploymentVersions(batchSize) { + let afterTableId = '' + let pinnedGroups = 0 + + for (;;) { + const tables = await db.execute(sql` + SELECT table_definition.id AS table_id + FROM user_table_definitions AS table_definition + WHERE table_definition.id COLLATE "C" > ${afterTableId}::text COLLATE "C" + AND EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + INNER JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' + AND workflow.archived_at IS NULL + INNER JOIN workflow_deployment_version AS active_version + ON active_version.workflow_id = workflow.id + AND active_version.is_active = true + WHERE workflow_group.value->>'workflowId' <> '' + AND workflow_group.value->>'deploymentVersionId' + IS DISTINCT FROM active_version.id + ) + ORDER BY table_definition.id COLLATE "C" + LIMIT ${batchSize} + `) + if (tables.length === 0) break + if (tables.length > batchSize) { + throw new Error('Table workflow deployment pin store returned an oversized page') + } + const tableIds = new Set(tables.map((table) => table.table_id)) + if (tableIds.size !== tables.length) { + throw new Error('Table workflow deployment pin store returned duplicate table ids') + } + const lastTableId = tables.at(-1)?.table_id + if (!lastTableId || lastTableId === afterTableId) { + throw new Error('Table workflow deployment pin store returned a non-advancing page') + } + + for (const table of tables) { + const [result] = await db.execute(sql` + WITH rewritten AS ( + SELECT + jsonb_agg( + CASE + WHEN workflow.id IS NOT NULL + AND workflow.archived_at IS NULL + AND active_version.id IS NOT NULL + AND workflow_group.value->>'deploymentVersionId' + IS DISTINCT FROM active_version.id + THEN jsonb_set( + workflow_group.value, + '{deploymentVersionId}', + to_jsonb(active_version.id), + true + ) + ELSE workflow_group.value + END + ORDER BY workflow_group.ordinality + ) AS workflow_groups, + ( + COUNT(*) FILTER ( + WHERE workflow.id IS NOT NULL + AND workflow.archived_at IS NULL + AND active_version.id IS NOT NULL + AND workflow_group.value->>'deploymentVersionId' + IS DISTINCT FROM active_version.id + ) + )::int AS pinned_group_count + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) WITH ORDINALITY AS workflow_group(value, ordinality) + LEFT JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' + LEFT JOIN workflow_deployment_version AS active_version + ON active_version.workflow_id = workflow.id + AND active_version.is_active = true + WHERE table_definition.id = ${table.table_id} + ) + UPDATE user_table_definitions AS table_definition + SET + schema = jsonb_set( + table_definition.schema, + '{workflowGroups}', + rewritten.workflow_groups, + false + ), + updated_at = NOW() + FROM rewritten + WHERE table_definition.id = ${table.table_id} + AND rewritten.pinned_group_count > 0 + RETURNING rewritten.pinned_group_count + `) + pinnedGroups += result?.pinned_group_count ?? 0 + } + + afterTableId = lastTableId + } + + const [unpinned] = await db.execute(sql` + SELECT + table_definition.id AS table_id, + workflow_group.value->>'id' AS group_id, + workflow_group.value->>'workflowId' AS workflow_id + FROM user_table_definitions AS table_definition + CROSS JOIN LATERAL jsonb_array_elements( + COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) + ) AS workflow_group(value) + INNER JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' + AND workflow.archived_at IS NULL + LEFT JOIN workflow_deployment_version AS active_version + ON active_version.workflow_id = workflow.id + AND active_version.is_active = true + WHERE workflow_group.value->>'workflowId' <> '' + AND ( + active_version.id IS NULL + OR workflow_group.value->>'deploymentVersionId' IS DISTINCT FROM active_version.id + ) + LIMIT 1 + `) + if (unpinned) { + throw new Error( + `Table ${unpinned.table_id} workflow group ${unpinned.group_id} did not pin active deployment for workflow ${unpinned.workflow_id}` + ) + } + + return pinnedGroups + }, } /** Deploys one table workflow through the same orchestration used by application surfaces. */ @@ -294,6 +440,7 @@ export async function backfillTableWorkflowDeployments( scanned: 0, deployed: 0, alreadyDeployed: 0, + pinnedGroups: 0, } let afterWorkflowId = '' @@ -353,6 +500,8 @@ export async function backfillTableWorkflowDeployments( ) } + summary.pinnedGroups = await store.pinActiveDeploymentVersions(batchSize) + return summary } From 17760cb8a45d7f79acf3deb0b716722363400497 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 17:39:33 -0700 Subject: [PATCH 06/13] fix(tables): follow latest workflow deployment --- .../workflow-column-execution.test.ts | 93 ++++++++++- .../background/workflow-column-execution.ts | 63 ++++++-- apps/sim/lib/table/application/groups.test.ts | 66 -------- apps/sim/lib/table/application/groups.ts | 75 ++------- apps/sim/lib/table/types.ts | 8 - apps/sim/lib/table/workflow-groups/service.ts | 31 +--- .../resolve-workflow-outputs.test.ts | 2 - .../application/resolve-workflow-outputs.ts | 11 +- ...ackfill-table-workflow-deployments.test.ts | 7 +- .../backfill-table-workflow-deployments.ts | 151 +----------------- 10 files changed, 158 insertions(+), 349 deletions(-) diff --git a/apps/sim/background/workflow-column-execution.test.ts b/apps/sim/background/workflow-column-execution.test.ts index 3cb88624c7d..12d63aff070 100644 --- a/apps/sim/background/workflow-column-execution.test.ts +++ b/apps/sim/background/workflow-column-execution.test.ts @@ -7,6 +7,7 @@ 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, @@ -14,9 +15,15 @@ import { 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[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[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() diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 5009492f5c4..85f750a7034 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/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> + let normalizedData: Awaited> 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', diff --git a/apps/sim/lib/table/application/groups.test.ts b/apps/sim/lib/table/application/groups.test.ts index 24b566321d3..59f3135caa7 100644 --- a/apps/sim/lib/table/application/groups.test.ts +++ b/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 }), diff --git a/apps/sim/lib/table/application/groups.ts b/apps/sim/lib/table/application/groups.ts index 50513f5d679..485c960b78e 100644 --- a/apps/sim/lib/table/application/groups.ts +++ b/apps/sim/lib/table/application/groups.ts @@ -34,10 +34,7 @@ import { updateWorkflowGroup, } from '@/lib/table/workflow-groups/service' import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context' -import type { - ResolveDeployedWorkflowOutputsResult, - ResolveWorkflowOutputsResult, -} from '@/lib/workflows/application/resolve-workflow-outputs' +import type { ResolveWorkflowOutputsResult } from '@/lib/workflows/application/resolve-workflow-outputs' import { loadResolvedDeployedWorkflowOutputs } from '@/lib/workflows/application/resolve-workflow-outputs' import { getEnrichment } from '@/enrichments/registry' import type { EnrichmentConfig } from '@/enrichments/types' @@ -62,7 +59,7 @@ function groupFromTable(table: TableDefinition, groupId: string): WorkflowGroup async function resolveWorkflowForAuthorizedTableCommand( workflowId: string, workspaceId: string -): Promise { +): Promise { const workflowContext = await resolveActiveWorkflowApplicationContext({ workflowId, assertedWorkspaceId: workspaceId, @@ -73,7 +70,7 @@ async function resolveWorkflowForAuthorizedTableCommand( async function resolveRelatedWorkflowForTableRoute( workflowId: string, workspaceId: string -): Promise { +): Promise { try { return await resolveWorkflowForAuthorizedTableCommand(workflowId, workspaceId) } catch (error) { @@ -248,7 +245,6 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({ * with a backing workflow and workflow output coordinates, and only a group * with no workflow is filled from the enrichment registry. */ - let deploymentVersionId: string | undefined if (input.group.workflowId) { const resolvedWorkflow = await resolveRelatedWorkflowForTableRoute( input.group.workflowId, @@ -262,7 +258,6 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({ resolvedWorkflow, input.group.workflowId ) - deploymentVersionId = resolvedWorkflow.deploymentVersionId } else if (input.group.enrichmentId) { requireKnownEnrichmentOutputIds( requireEnrichment(input.group.enrichmentId), @@ -289,7 +284,6 @@ export const createTableGroupUseCase = defineAuthorizedTableUseCase({ ...input.group, id: groupId, workflowId: input.group.workflowId ?? '', - ...(deploymentVersionId ? { deploymentVersionId } : {}), outputs: input.group.outputs.map((output) => ({ ...output, blockId: output.blockId ?? '', @@ -410,7 +404,6 @@ export const createWorkflowTableGroup = defineAuthorizedTableUseCase({ const group: WorkflowGroup = { id: groupId, workflowId: input.workflowId, - deploymentVersionId: resolvedWorkflow.deploymentVersionId, ...(input.name ? { name: input.name } : {}), ...(input.dependencies ? { dependencies: input.dependencies } : {}), ...(input.deploymentMode ? { deploymentMode: input.deploymentMode } : {}), @@ -726,10 +719,9 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({ const workflowMetadataRequired = input.workflowId !== undefined || outputCoordinatesToValidate.length > 0 || - (input.mappingUpdates?.length ?? 0) > 0 || - input.inputMappings !== undefined + (input.mappingUpdates?.length ?? 0) > 0 const targetWorkflowId = input.workflowId ?? previousGroup?.workflowId - let resolvedWorkflow: ResolveDeployedWorkflowOutputsResult | undefined + let resolvedWorkflow: ResolveWorkflowOutputsResult | undefined if (workflowMetadataRequired) { if (!targetWorkflowId) { throw new OrchestrationError('not_found', 'Workflow not found') @@ -738,19 +730,8 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({ targetWorkflowId, context.workspaceId ) - const remappedPreviousOutputs = (previousGroup?.outputs ?? []).map((output) => { - const mapping = input.mappingUpdates?.find( - (candidate) => candidate.columnName === output.columnName - ) - return mapping ? { ...output, blockId: mapping.blockId, path: mapping.path } : output - }) - const resultingOutputs = input.outputs ?? remappedPreviousOutputs - const deploymentChanged = - resolvedWorkflow.deploymentVersionId !== previousGroup?.deploymentVersionId - const coordinatesToValidate = - workflowChanged || deploymentChanged ? resultingOutputs : outputCoordinatesToValidate - if (coordinatesToValidate.length > 0) { - validateRequestedOutputs(coordinatesToValidate, resolvedWorkflow, targetWorkflowId) + if (outputCoordinatesToValidate.length > 0) { + validateRequestedOutputs(outputCoordinatesToValidate, resolvedWorkflow, targetWorkflowId) } } const actorUserId = attributedUserId(principal, context.billedAccountUserId) @@ -798,18 +779,6 @@ export const updateTableGroupUseCase = defineAuthorizedTableUseCase({ : {}), ...(input.mappingUpdates !== undefined ? { mappingUpdates: input.mappingUpdates } : {}), ...(resolvedMappingTypes ? { resolvedMappingTypes } : {}), - ...(resolvedWorkflow - ? { - resolvedDeployment: { - workflowId: resolvedWorkflow.workflowId, - deploymentVersionId: resolvedWorkflow.deploymentVersionId, - validOutputCoordinates: (resolvedWorkflow.outputs ?? []).map((output) => ({ - blockId: output.blockId, - path: output.path, - })), - }, - } - : {}), ...(input.inputMappings !== undefined ? { inputMappings: input.inputMappings } : {}), ...(input.deploymentMode !== undefined ? { deploymentMode: input.deploymentMode } : {}), ...(input.type !== undefined ? { type: input.type } : {}), @@ -901,19 +870,10 @@ export const updateWorkflowTableGroup = defineAuthorizedTableUseCase({ const resolvedWorkflow = workflowMetadataRequired ? await resolveWorkflowForAuthorizedTableCommand(targetWorkflowId, context.workspaceId) : undefined - if (resolvedWorkflow) { - const remappedPreviousOutputs = previousGroup.outputs.map((output) => { - const mapping = input.mappingUpdates?.find( - (candidate) => candidate.columnName === output.columnName - ) - return mapping ? { ...output, blockId: mapping.blockId, path: mapping.path } : output - }) - const resultingOutputs = input.outputs ?? remappedPreviousOutputs - const deploymentChanged = - resolvedWorkflow.deploymentVersionId !== previousGroup.deploymentVersionId - if (input.outputs || input.workflowId || deploymentChanged) { - validateRequestedOutputs(resultingOutputs, resolvedWorkflow, targetWorkflowId) - } + if (input.outputs && resolvedWorkflow) { + validateRequestedOutputs(input.outputs, resolvedWorkflow, targetWorkflowId) + } else if (input.workflowId && resolvedWorkflow) { + validateRequestedOutputs(previousGroup.outputs, resolvedWorkflow, targetWorkflowId) } let outputs: WorkflowGroupOutput[] | undefined @@ -1010,18 +970,6 @@ export const updateWorkflowTableGroup = defineAuthorizedTableUseCase({ ...(newOutputColumns !== undefined ? { newOutputColumns } : {}), ...(input.mappingUpdates !== undefined ? { mappingUpdates: input.mappingUpdates } : {}), ...(resolvedMappingTypes ? { resolvedMappingTypes } : {}), - ...(resolvedWorkflow - ? { - resolvedDeployment: { - workflowId: resolvedWorkflow.workflowId, - deploymentVersionId: resolvedWorkflow.deploymentVersionId, - validOutputCoordinates: (resolvedWorkflow.outputs ?? []).map((output) => ({ - blockId: output.blockId, - path: output.path, - })), - }, - } - : {}), ...(input.deploymentMode !== undefined ? { deploymentMode: input.deploymentMode } : {}), ...(input.autoRun !== undefined ? { autoRun: input.autoRun } : {}), }, @@ -1158,7 +1106,6 @@ export const addWorkflowTableGroupOutput = defineAuthorizedTableUseCase({ }).attributedUserId, resolvedOutput: { workflowId: resolvedWorkflow.workflowId, - deploymentVersionId: resolvedWorkflow.deploymentVersionId, columnType: columnTypeForLeaf(output.leafType), order: outputs.map((candidate, discoveryIndex) => { const distance = resolvedWorkflow.executionOrderByBlockId[candidate.blockId] diff --git a/apps/sim/lib/table/types.ts b/apps/sim/lib/table/types.ts index 92e4d27ae6a..9319e794c10 100644 --- a/apps/sim/lib/table/types.ts +++ b/apps/sim/lib/table/types.ts @@ -138,8 +138,6 @@ export interface WorkflowGroup { id: string /** Backing workflow id for `manual` groups. `''` for enrichment groups. */ workflowId: string - /** Immutable deployment version whose input and output coordinates this group stores. */ - deploymentVersionId?: string /** Registry enrichment id for `enrichment` groups. */ enrichmentId?: string /** Display name; defaults to the workflow's / enrichment's name. */ @@ -968,12 +966,6 @@ export interface UpdateWorkflowGroupData { workflowId: string columns: Array<{ columnName: string; type: ColumnDefinition['type'] }> } - /** Workflow-authorized deployment snapshot to pin after this update. */ - resolvedDeployment?: { - workflowId: string - deploymentVersionId: string - validOutputCoordinates: Array<{ blockId: string; path: string }> - } /** Replace the group's input mappings. Omit to leave them unchanged. */ inputMappings?: WorkflowGroupInputMapping[] /** Change which workflow state the group runs against. Omit to leave unchanged. */ diff --git a/apps/sim/lib/table/workflow-groups/service.ts b/apps/sim/lib/table/workflow-groups/service.ts index ccc30d89b1b..9437427fef3 100644 --- a/apps/sim/lib/table/workflow-groups/service.ts +++ b/apps/sim/lib/table/workflow-groups/service.ts @@ -296,13 +296,6 @@ export async function updateWorkflowGroup( throw new OrchestrationError('not_found', `Workflow group "${data.groupId}" not found`) } const group = groups[groupIndex] - const finalWorkflowId = data.workflowId ?? group.workflowId - if (data.resolvedDeployment && data.resolvedDeployment.workflowId !== finalWorkflowId) { - throw new OrchestrationError( - 'conflict', - `Workflow group "${data.groupId}" changed concurrently; retry the update.` - ) - } // Normalize every caller-supplied column reference to its stable id, so // the diff/splice/clear logic below operates uniformly in id-space (the @@ -368,6 +361,7 @@ export async function updateWorkflowGroup( // Only apply the out-of-lock leaf-type resolution if the group still // points at the workflow we resolved against. A concurrent workflow // remap invalidates the command snapshot and must be retried. + const finalWorkflowId = data.workflowId ?? group.workflowId if (remapLeafTypeById.size > 0 && resolvedForWorkflowId !== finalWorkflowId) { throw new OrchestrationError( 'conflict', @@ -388,22 +382,6 @@ export async function updateWorkflowGroup( // If the caller passed `outputs`, that's the new full set. If only // `mappingUpdates` was sent, the new set is the remapped old set. const newOutputs = outputsInput ?? oldOutputs - if (data.resolvedDeployment) { - const validCoordinates = new Set( - data.resolvedDeployment.validOutputCoordinates.map( - (output) => `${output.blockId}::${output.path}` - ) - ) - const invalidOutput = newOutputs.find( - (output) => !validCoordinates.has(`${output.blockId}::${output.path}`) - ) - if (invalidOutput) { - throw new OrchestrationError( - 'conflict', - `Workflow group "${data.groupId}" mappings changed concurrently; retry the update.` - ) - } - } // Enrichment outputs all share empty `blockId`/`path`, so keying on those // alone collapses every sibling to one entry (dropping columns on diff). Key // on the registry `outputId` when present; fall back to `blockId::path` for @@ -467,10 +445,7 @@ export async function updateWorkflowGroup( const updatedGroup: WorkflowGroup = { ...group, - workflowId: finalWorkflowId, - ...(data.resolvedDeployment - ? { deploymentVersionId: data.resolvedDeployment.deploymentVersionId } - : {}), + workflowId: data.workflowId ?? group.workflowId, name: data.name ?? group.name, dependencies: dependenciesInput ?? group.dependencies, outputs: newOutputs, @@ -654,7 +629,6 @@ export async function addWorkflowGroupOutput( actorUserId?: string | null resolvedOutput: { workflowId: string - deploymentVersionId: string columnType: ColumnDefinition['type'] order: Array<{ blockId: string @@ -779,7 +753,6 @@ export async function addWorkflowGroupOutput( const orderedGroupColIds = allGroupOutputs.map((o) => o.columnName) const updatedGroup: WorkflowGroup = { ...group, - deploymentVersionId: data.resolvedOutput.deploymentVersionId, outputs: allGroupOutputs, } const nextGroups = groups.map((g, i) => (i === groupIndex ? updatedGroup : g)) diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts index 8aa9152c24a..f750feea8b8 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.test.ts @@ -75,7 +75,6 @@ describe('resolveWorkflowOutputs', () => { mocks.loadDeployed.mockResolvedValue({ blocks: { block1: { id: 'block-1', type: 'agent', name: 'Agent', subBlocks: {} } }, edges: [], - deploymentVersionId: 'deployment-version-1', }) mocks.flatten.mockReturnValue([ { @@ -127,7 +126,6 @@ describe('resolveWorkflowOutputs', () => { await expect(loadResolvedDeployedWorkflowOutputs(context)).resolves.toMatchObject({ workflowId: 'workflow-1', - deploymentVersionId: 'deployment-version-1', outputs: [{ blockId: 'block-1', path: 'content' }], }) diff --git a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts index 3334feb8bb7..cb60823c006 100644 --- a/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts +++ b/apps/sim/lib/workflows/application/resolve-workflow-outputs.ts @@ -27,10 +27,6 @@ export interface ResolveWorkflowOutputsResult { executionOrderByBlockId: Record } -export interface ResolveDeployedWorkflowOutputsResult extends ResolveWorkflowOutputsResult { - deploymentVersionId: string -} - type ResolvableWorkflowState = | NonNullable>> | Awaited> @@ -67,16 +63,13 @@ export async function loadResolvedWorkflowOutputs( /** Loads output metadata from the active deployment after workflow authorization. */ export async function loadResolvedDeployedWorkflowOutputs( context: ActiveWorkflowApplicationContext -): Promise { +): Promise { if (!context.workflow.isDeployed) { throw new OrchestrationError('validation', 'Workflow must have an active deployment') } try { const normalized = await loadDeployedWorkflowState(context.workflowId, context.workspaceId) - return { - ...resolveWorkflowOutputsFromState(context.workflowId, normalized), - deploymentVersionId: normalized.deploymentVersionId, - } + return resolveWorkflowOutputsFromState(context.workflowId, normalized) } catch (error) { if (error instanceof NoActiveDeploymentError) { throw new OrchestrationError('validation', 'Workflow must have an active deployment') diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts index d09b8b6f103..914c593cb2c 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -34,7 +34,6 @@ function store( assertIntegrity: vi.fn().mockResolvedValue(undefined), listCandidates: vi.fn().mockResolvedValue([]), isDeployed: vi.fn().mockResolvedValue(false), - pinActiveDeploymentVersions: vi.fn().mockResolvedValue(0), ...overrides, } } @@ -66,8 +65,7 @@ describe('backfillTableWorkflowDeployments', () => { }, } }) - const pinActiveDeploymentVersions = vi.fn().mockResolvedValue(3) - const backfillStore = store({ listCandidates, isDeployed, pinActiveDeploymentVersions }) + const backfillStore = store({ listCandidates, isDeployed }) await expect( backfillTableWorkflowDeployments(backfillStore, deploy, { batchSize: 2 }) @@ -75,7 +73,6 @@ describe('backfillTableWorkflowDeployments', () => { scanned: 3, deployed: 3, alreadyDeployed: 0, - pinnedGroups: 3, }) expect(listCandidates.mock.calls).toEqual([ ['', 2], @@ -84,7 +81,6 @@ describe('backfillTableWorkflowDeployments', () => { ['', 1], ]) expect(backfillStore.assertIntegrity).toHaveBeenCalledTimes(2) - expect(pinActiveDeploymentVersions).toHaveBeenCalledWith(2) expect(deploy.mock.calls.map(([workflow]) => workflow.workflowId)).toEqual([ 'workflow-a', 'workflow-b', @@ -112,7 +108,6 @@ describe('backfillTableWorkflowDeployments', () => { scanned: 1, deployed: 0, alreadyDeployed: 1, - pinnedGroups: 0, }) expect(deploy).not.toHaveBeenCalled() }) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index c0f5f1ee1b0..82ed9180ab0 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -1,8 +1,7 @@ #!/usr/bin/env bun /** - * Deploys every workflow referenced by a table workflow group and pins each - * group to the active deployment whose mappings it uses. + * Deploys every workflow referenced by a table workflow group. * * The script is idempotent and resumable: it pages over only workflows that do * not have the full desired state, and each deployment uses a stable idempotency @@ -42,14 +41,12 @@ export interface TableWorkflowDeploymentStore { limit: number ): Promise isDeployed(workflowId: string): Promise - pinActiveDeploymentVersions(batchSize: number): Promise } export interface TableWorkflowDeploymentSummary { scanned: number deployed: number alreadyDeployed: number - pinnedGroups: number } interface TableWorkflowDeploymentBackfillOptions { @@ -92,20 +89,6 @@ interface DeploymentStateRow extends Record { is_deployed: boolean } -interface PinnedGroupCountRow extends Record { - pinned_group_count: number -} - -interface TablePinCandidateRow extends Record { - table_id: string -} - -interface UnpinnedWorkflowGroupRow extends Record { - group_id: string - table_id: string - workflow_id: string -} - /** Ensures the table group references can be traversed without silently dropping corrupt data. */ async function assertTableWorkflowIntegrity(): Promise { const [invalidSchema] = await db.execute(sql` @@ -276,135 +259,6 @@ export const postgresTableWorkflowDeploymentStore: TableWorkflowDeploymentStore } return state.is_deployed && state.active_version_count === 1 }, - - async pinActiveDeploymentVersions(batchSize) { - let afterTableId = '' - let pinnedGroups = 0 - - for (;;) { - const tables = await db.execute(sql` - SELECT table_definition.id AS table_id - FROM user_table_definitions AS table_definition - WHERE table_definition.id COLLATE "C" > ${afterTableId}::text COLLATE "C" - AND EXISTS ( - SELECT 1 - FROM jsonb_array_elements( - COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) - ) AS workflow_group(value) - INNER JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' - AND workflow.archived_at IS NULL - INNER JOIN workflow_deployment_version AS active_version - ON active_version.workflow_id = workflow.id - AND active_version.is_active = true - WHERE workflow_group.value->>'workflowId' <> '' - AND workflow_group.value->>'deploymentVersionId' - IS DISTINCT FROM active_version.id - ) - ORDER BY table_definition.id COLLATE "C" - LIMIT ${batchSize} - `) - if (tables.length === 0) break - if (tables.length > batchSize) { - throw new Error('Table workflow deployment pin store returned an oversized page') - } - const tableIds = new Set(tables.map((table) => table.table_id)) - if (tableIds.size !== tables.length) { - throw new Error('Table workflow deployment pin store returned duplicate table ids') - } - const lastTableId = tables.at(-1)?.table_id - if (!lastTableId || lastTableId === afterTableId) { - throw new Error('Table workflow deployment pin store returned a non-advancing page') - } - - for (const table of tables) { - const [result] = await db.execute(sql` - WITH rewritten AS ( - SELECT - jsonb_agg( - CASE - WHEN workflow.id IS NOT NULL - AND workflow.archived_at IS NULL - AND active_version.id IS NOT NULL - AND workflow_group.value->>'deploymentVersionId' - IS DISTINCT FROM active_version.id - THEN jsonb_set( - workflow_group.value, - '{deploymentVersionId}', - to_jsonb(active_version.id), - true - ) - ELSE workflow_group.value - END - ORDER BY workflow_group.ordinality - ) AS workflow_groups, - ( - COUNT(*) FILTER ( - WHERE workflow.id IS NOT NULL - AND workflow.archived_at IS NULL - AND active_version.id IS NOT NULL - AND workflow_group.value->>'deploymentVersionId' - IS DISTINCT FROM active_version.id - ) - )::int AS pinned_group_count - FROM user_table_definitions AS table_definition - CROSS JOIN LATERAL jsonb_array_elements( - COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) - ) WITH ORDINALITY AS workflow_group(value, ordinality) - LEFT JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' - LEFT JOIN workflow_deployment_version AS active_version - ON active_version.workflow_id = workflow.id - AND active_version.is_active = true - WHERE table_definition.id = ${table.table_id} - ) - UPDATE user_table_definitions AS table_definition - SET - schema = jsonb_set( - table_definition.schema, - '{workflowGroups}', - rewritten.workflow_groups, - false - ), - updated_at = NOW() - FROM rewritten - WHERE table_definition.id = ${table.table_id} - AND rewritten.pinned_group_count > 0 - RETURNING rewritten.pinned_group_count - `) - pinnedGroups += result?.pinned_group_count ?? 0 - } - - afterTableId = lastTableId - } - - const [unpinned] = await db.execute(sql` - SELECT - table_definition.id AS table_id, - workflow_group.value->>'id' AS group_id, - workflow_group.value->>'workflowId' AS workflow_id - FROM user_table_definitions AS table_definition - CROSS JOIN LATERAL jsonb_array_elements( - COALESCE(table_definition.schema->'workflowGroups', '[]'::jsonb) - ) AS workflow_group(value) - INNER JOIN workflow ON workflow.id = workflow_group.value->>'workflowId' - AND workflow.archived_at IS NULL - LEFT JOIN workflow_deployment_version AS active_version - ON active_version.workflow_id = workflow.id - AND active_version.is_active = true - WHERE workflow_group.value->>'workflowId' <> '' - AND ( - active_version.id IS NULL - OR workflow_group.value->>'deploymentVersionId' IS DISTINCT FROM active_version.id - ) - LIMIT 1 - `) - if (unpinned) { - throw new Error( - `Table ${unpinned.table_id} workflow group ${unpinned.group_id} did not pin active deployment for workflow ${unpinned.workflow_id}` - ) - } - - return pinnedGroups - }, } /** Deploys one table workflow through the same orchestration used by application surfaces. */ @@ -440,7 +294,6 @@ export async function backfillTableWorkflowDeployments( scanned: 0, deployed: 0, alreadyDeployed: 0, - pinnedGroups: 0, } let afterWorkflowId = '' @@ -500,8 +353,6 @@ export async function backfillTableWorkflowDeployments( ) } - summary.pinnedGroups = await store.pinActiveDeploymentVersions(batchSize) - return summary } From 65e8d12a2736dde71330179f6b77bfaa55949b65 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 17:51:37 -0700 Subject: [PATCH 07/13] fix(tables): remove copilot deployment mode --- .../lib/copilot/generated/tool-catalog-v1.ts | 12 ----------- .../lib/copilot/generated/tool-schemas-v1.ts | 12 ----------- .../tools/server/table/user-table.test.ts | 20 +++++++++++++++++++ .../copilot/tools/server/table/user-table.ts | 14 ------------- 4 files changed, 20 insertions(+), 38 deletions(-) diff --git a/apps/sim/lib/copilot/generated/tool-catalog-v1.ts b/apps/sim/lib/copilot/generated/tool-catalog-v1.ts index b50898eb45a..ed93f30ed58 100644 --- a/apps/sim/lib/copilot/generated/tool-catalog-v1.ts +++ b/apps/sim/lib/copilot/generated/tool-catalog-v1.ts @@ -5312,12 +5312,6 @@ export const TableAutomations: ToolCatalogEntry = { }, }, }, - deploymentMode: { - type: 'string', - description: - 'Which workflow version rows execute: "live" (default, editable draft — edits take effect immediately) or "deployed" (latest active deployment; fails if the workflow was never deployed).', - enum: ['live', 'deployed'], - }, groupId: { type: 'string', description: @@ -6092,12 +6086,6 @@ export const UserTable: ToolCatalogEntry = { }, }, }, - deploymentMode: { - type: 'string', - description: - "Which version of the backing workflow this group's per-row runs execute, for add_workflow_group and update_workflow_group. 'live' (default) runs the editable draft, so later edits take effect immediately. 'deployed' runs the workflow's latest active deployment, pinning rows to a published version — if that workflow has never been deployed the cell fails rather than falling back to the draft. Only meaningful for workflow groups; enrichment groups have no backing workflow.", - enum: ['live', 'deployed'], - }, description: { type: 'string', description: "Table description (optional for 'create')" }, enrichmentId: { type: 'string', diff --git a/apps/sim/lib/copilot/generated/tool-schemas-v1.ts b/apps/sim/lib/copilot/generated/tool-schemas-v1.ts index 7625f38c296..1fecc8476a6 100644 --- a/apps/sim/lib/copilot/generated/tool-schemas-v1.ts +++ b/apps/sim/lib/copilot/generated/tool-schemas-v1.ts @@ -5173,12 +5173,6 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { }, }, }, - deploymentMode: { - type: 'string', - description: - 'Which workflow version rows execute: "live" (default, editable draft — edits take effect immediately) or "deployed" (latest active deployment; fails if the workflow was never deployed).', - enum: ['live', 'deployed'], - }, groupId: { type: 'string', description: @@ -6034,12 +6028,6 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { }, }, }, - deploymentMode: { - type: 'string', - description: - "Which version of the backing workflow this group's per-row runs execute, for add_workflow_group and update_workflow_group. 'live' (default) runs the editable draft, so later edits take effect immediately. 'deployed' runs the workflow's latest active deployment, pinning rows to a published version — if that workflow has never been deployed the cell fails rather than falling back to the draft. Only meaningful for workflow groups; enrichment groups have no backing workflow.", - enum: ['live', 'deployed'], - }, description: { type: 'string', description: "Table description (optional for 'create')", diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.test.ts b/apps/sim/lib/copilot/tools/server/table/user-table.test.ts index 85d39c44e4d..1e96164e1b5 100644 --- a/apps/sim/lib/copilot/tools/server/table/user-table.test.ts +++ b/apps/sim/lib/copilot/tools/server/table/user-table.test.ts @@ -258,6 +258,7 @@ vi.mock('@/lib/workflows/application/context', () => ({ })) vi.mock('@/lib/workflows/application/resolve-workflow-outputs', () => ({ + loadResolvedDeployedWorkflowOutputs: async () => mockExecuteCopilotWorkflowUseCase(), loadResolvedWorkflowOutputs: async () => mockExecuteCopilotWorkflowUseCase(), resolveWorkflowOutputs: { operation: { id: 'workflows.read' } }, })) @@ -959,6 +960,25 @@ describe('userTableServerTool workflow scope', () => { expect(mockAddWorkflowGroup).not.toHaveBeenCalled() }) + it('does not pass a legacy deployment mode into workflow group creation', async () => { + const result = await userTableServerTool.execute( + { + operation: 'add_workflow_group', + args: { + tableId: 'tbl_1', + workflowId: 'workflow-1', + outputs: [{ blockId: 'block-1', path: 'content' }], + deploymentMode: 'live', + }, + }, + buildToolContext() + ) + + expect(result.success).toBe(true) + expect(mockAddWorkflowGroup).toHaveBeenCalledTimes(1) + expect(mockAddWorkflowGroup.mock.calls[0][0].group).not.toHaveProperty('deploymentMode') + }) + it('conceals unknown application failures from tool output', async () => { mockQueryRows.mockRejectedValueOnce(new Error('database host unavailable')) diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.ts b/apps/sim/lib/copilot/tools/server/table/user-table.ts index 372d3aa9898..d48ac344d42 100644 --- a/apps/sim/lib/copilot/tools/server/table/user-table.ts +++ b/apps/sim/lib/copilot/tools/server/table/user-table.ts @@ -62,7 +62,6 @@ import type { TablePredicateInput, TableSchema, WorkflowGroupDependencies, - WorkflowGroupDeploymentMode, } from '@/lib/table/types' import { viewConfigIdsToNames } from '@/lib/table/views/service' import type { ResolvedSecretTraceProvenanceV1 } from '@/executor/utils/resolved-secret-trace-registry' @@ -132,16 +131,6 @@ function resolveAuthorizedWorkflowOutputs( }) } -/** - * Narrows a raw `deploymentMode` arg to the `'live' | 'deployed'` union, or - * `undefined` when absent/invalid (leaving the group's existing value — which - * itself defaults to `'live'`). Lets Mothership choose whether a group's - * per-cell runs execute the live draft or the latest active deployment. - */ -function parseDeploymentMode(value: unknown): WorkflowGroupDeploymentMode | undefined { - return value === 'live' || value === 'deployed' ? value : undefined -} - /** Validates an optional row limit against the policy for the requested surface operation. */ function limitError(limit: unknown, max?: number): string | null { if (limit === undefined) return null @@ -1242,7 +1231,6 @@ export const userTableServerTool: BaseServerTool const dependencies = args.dependencies as WorkflowGroupDependencies | undefined const name = args.name as string | undefined - const deploymentMode = parseDeploymentMode(args.deploymentMode) assertNotAborted() const autoRun = args.autoRun === true const { table: updated, group } = await executeCopilotCreateWorkflowTableGroup(context, { @@ -1252,7 +1240,6 @@ export const userTableServerTool: BaseServerTool outputs: rawOutputs, name, dependencies, - deploymentMode, autoRun, }) return { @@ -1294,7 +1281,6 @@ export const userTableServerTool: BaseServerTool dependencies: args.dependencies as WorkflowGroupDependencies | undefined, outputs: updateOutputs, mappingUpdates, - deploymentMode: parseDeploymentMode(args.deploymentMode), autoRun: typeof args.autoRun === 'boolean' ? args.autoRun : undefined, }) return { From 25f8c71d5593aacf863f86e20b737c2caec79e88 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 28 Aug 2026 17:54:46 -0700 Subject: [PATCH 08/13] fix(tables): resolve canonical workflow starts --- .../workflow-column-execution.test.ts | 20 +++++++++++++++++-- .../background/workflow-column-execution.ts | 19 +++++++++--------- 2 files changed, 27 insertions(+), 12 deletions(-) diff --git a/apps/sim/background/workflow-column-execution.test.ts b/apps/sim/background/workflow-column-execution.test.ts index 12d63aff070..306e925133f 100644 --- a/apps/sim/background/workflow-column-execution.test.ts +++ b/apps/sim/background/workflow-column-execution.test.ts @@ -57,12 +57,14 @@ const QUEUED_PAYLOAD = { }, } -function latestDeployment(): Parameters[1] { +function latestDeployment( + startType = 'start_trigger' +): Parameters[1] { return { blocks: { start: { id: 'start', - type: 'start_trigger', + type: startType, subBlocks: { inputFormat: { value: [{ name: 'company', type: 'string' }] }, }, @@ -109,6 +111,20 @@ describe('latest table workflow deployment mappings', () => { ).not.toThrow() }) + it('accepts a canonical split manual start block', () => { + expect(() => + assertWorkflowGroupMatchesLatestDeployment( + { + id: 'group-1', + workflowId: 'workflow-1', + outputs: [{ blockId: 'agent', path: 'content', columnName: 'column-output' }], + inputMappings: [{ inputName: 'company', columnName: 'column-company' }], + }, + latestDeployment('manual_trigger') + ) + ).not.toThrow() + }) + it('rejects an output mapping removed by the latest active deployment', () => { expect(() => assertWorkflowGroupMatchesLatestDeployment( diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 85f750a7034..33c21174d70 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/apps/sim/background/workflow-column-execution.ts @@ -57,6 +57,7 @@ import { import { flattenWorkflowOutputs } from '@/lib/workflows/blocks/flatten-outputs' import { normalizeInputFormatValue } from '@/lib/workflows/input-format' import type { DeployedWorkflowData } from '@/lib/workflows/persistence/utils' +import { TriggerUtils } from '@/lib/workflows/triggers/triggers' import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' export type { WorkflowGroupCellPayload } @@ -85,15 +86,15 @@ export function assertWorkflowGroupMatchesLatestDeployment( ) } - const startBlock = Object.values(deployment.blocks).find( - (block) => block.type === 'start_trigger' - ) - if (!startBlock) { + const startCandidate = TriggerUtils.findStartBlock(deployment.blocks, 'manual') + if (!startCandidate) { throw new Error('Workflow is missing a Start trigger') } const validInputNames = new Set( - normalizeInputFormatValue(startBlock.subBlocks?.inputFormat?.value).map((input) => input.name) + normalizeInputFormatValue(startCandidate.block.subBlocks?.inputFormat?.value).map( + (input) => input.name + ) ) const invalidInput = (group.inputMappings ?? []).find( (mapping) => !validInputNames.has(mapping.inputName) @@ -746,10 +747,8 @@ async function runWorkflowAndWriteTerminal( }) return 'error' } - const startBlock = normalizedData - ? Object.values(normalizedData.blocks).find((b) => b?.type === 'start_trigger') - : undefined - if (!startBlock) { + const startCandidate = TriggerUtils.findStartBlock(normalizedData.blocks, 'manual') + if (!startCandidate) { await writeState({ status: 'error', executionId, @@ -1037,7 +1036,7 @@ async function runWorkflowAndWriteTerminal( }, executionMode: 'sync', workflowTriggerType: 'table', - triggerBlockId: startBlock.id, + triggerBlockId: startCandidate.blockId, useDraftState: false, workflowStateOverride: normalizedData, abortSignal: attemptSignal, From 319c3dcd31c9cc12c6ba447234174d189d333caf Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Sat, 29 Aug 2026 11:30:07 -0700 Subject: [PATCH 09/13] fix(scripts): simplify staging workflow backfill --- ...ackfill-table-workflow-deployments.test.ts | 73 +++++++++++++++- .../backfill-table-workflow-deployments.ts | 84 +++++++++++++++++-- 2 files changed, 149 insertions(+), 8 deletions(-) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts index 914c593cb2c..6f878b99de6 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -1,12 +1,17 @@ /** * @vitest-environment node */ -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockPerformFullDeploy } = vi.hoisted(() => ({ +const { mockLoadRuntimeSecrets, mockPerformFullDeploy } = vi.hoisted(() => ({ + mockLoadRuntimeSecrets: vi.fn(), mockPerformFullDeploy: vi.fn(), })) +vi.mock('@sim/runtime-secrets', () => ({ + loadRuntimeSecrets: mockLoadRuntimeSecrets, +})) + vi.mock('@/lib/workflows/orchestration/deploy', () => ({ performFullDeploy: mockPerformFullDeploy, })) @@ -14,11 +19,28 @@ vi.mock('@/lib/workflows/orchestration/deploy', () => ({ import { backfillTableWorkflowDeployments, deployTableWorkflow, + parseTableWorkflowDeploymentBackfillArgs, + prepareTableWorkflowDeploymentBackfillEnvironment, TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE, type TableWorkflowDeploymentCandidate, type TableWorkflowDeploymentStore, } from '@/scripts/backfill-table-workflow-deployments' +const ORIGINAL_ENV = { + DATABASE_URL: process.env.DATABASE_URL, + DATABASE_URL_WEB: process.env.DATABASE_URL_WEB, + SIM_ENV_SECRET_ID: process.env.SIM_ENV_SECRET_ID, +} + +function restoreEnvironmentVariable(key: keyof typeof ORIGINAL_ENV): void { + const value = ORIGINAL_ENV[key] + if (value === undefined) { + Reflect.deleteProperty(process.env, key) + } else { + process.env[key] = value + } +} + function candidate(workflowId: string): TableWorkflowDeploymentCandidate { return { workflowId, @@ -43,6 +65,53 @@ describe('backfillTableWorkflowDeployments', () => { vi.clearAllMocks() }) + afterEach(() => { + restoreEnvironmentVariable('DATABASE_URL') + restoreEnvironmentVariable('DATABASE_URL_WEB') + restoreEnvironmentVariable('SIM_ENV_SECRET_ID') + }) + + it('loads the staging runtime secret before database modules are needed', async () => { + Reflect.deleteProperty(process.env, 'DATABASE_URL') + Reflect.deleteProperty(process.env, 'DATABASE_URL_WEB') + Reflect.deleteProperty(process.env, 'SIM_ENV_SECRET_ID') + mockLoadRuntimeSecrets.mockImplementation(async () => { + process.env.DATABASE_URL = 'postgres://staging/database' + }) + + await prepareTableWorkflowDeploymentBackfillEnvironment(['--environment=staging']) + + expect(process.env.SIM_ENV_SECRET_ID).toBe('/staging/sim/env-vars') + expect(mockLoadRuntimeSecrets).toHaveBeenCalledTimes(1) + }) + + it('keeps the existing local DATABASE_URL mode when no environment is requested', async () => { + process.env.DATABASE_URL = 'postgres://local/database' + + await prepareTableWorkflowDeploymentBackfillEnvironment([]) + + expect(mockLoadRuntimeSecrets).not.toHaveBeenCalled() + expect(process.env.DATABASE_URL).toBe('postgres://local/database') + }) + + it('rejects unsupported, unknown, duplicate, and locally configured staging arguments', async () => { + expect(() => parseTableWorkflowDeploymentBackfillArgs(['--environment=production'])).toThrow( + 'Unsupported backfill environment: production' + ) + expect(() => parseTableWorkflowDeploymentBackfillArgs(['--dry-run'])).toThrow( + 'Unknown argument: --dry-run' + ) + expect(() => + parseTableWorkflowDeploymentBackfillArgs(['--environment=staging', '--environment=staging']) + ).toThrow('can only be provided once') + + process.env.DATABASE_URL = 'postgres://local/database' + await expect( + prepareTableWorkflowDeploymentBackfillEnvironment(['--environment=staging']) + ).rejects.toThrow('local configuration cannot override staging') + expect(mockLoadRuntimeSecrets).not.toHaveBeenCalled() + }) + it('deploys bounded keyset pages and verifies the final desired state', async () => { const listCandidates = vi .fn() diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index 82ed9180ab0..a486bcc3e98 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -11,22 +11,25 @@ * * Usage: * DATABASE_URL=... bun run apps/sim/scripts/backfill-table-workflow-deployments.ts + * AWS_PROFILE=sim-admin bun --no-env-file apps/sim/scripts/backfill-table-workflow-deployments.ts --environment=staging */ -import { db } from '@sim/db' import { createLogger } from '@sim/logger' +import { loadRuntimeSecrets } from '@sim/runtime-secrets' import { getErrorMessage } from '@sim/utils/errors' import { sql } from 'drizzle-orm' -import { - type PerformFullDeployResult, - performFullDeploy, -} from '@/lib/workflows/orchestration/deploy' +import type { PerformFullDeployResult } from '@/lib/workflows/orchestration/deploy' const logger = createLogger('BackfillTableWorkflowDeployments') export const TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE = 25 const BACKFILL_ACTOR_ID = 'table-workflow-deployment-backfill' const BACKFILL_OPERATION_VERSION = 'v2' +const STAGING_RUNTIME_SECRET_ID = '/staging/sim/env-vars' + +interface TableWorkflowDeploymentBackfillCliOptions { + environment?: 'staging' +} export interface TableWorkflowDeploymentCandidate { workflowId: string @@ -89,8 +92,69 @@ interface DeploymentStateRow extends Record { is_deployed: boolean } +/** Parses the deliberately small CLI surface for the backfill. */ +export function parseTableWorkflowDeploymentBackfillArgs( + args: readonly string[] +): TableWorkflowDeploymentBackfillCliOptions { + let environment: TableWorkflowDeploymentBackfillCliOptions['environment'] + + for (const arg of args) { + if (!arg.startsWith('--environment=')) { + throw new Error(`Unknown argument: ${arg}`) + } + if (environment) { + throw new Error('The --environment argument can only be provided once') + } + + const requestedEnvironment = arg.slice('--environment='.length) + if (requestedEnvironment !== 'staging') { + throw new Error(`Unsupported backfill environment: ${requestedEnvironment || '(empty)'}`) + } + environment = requestedEnvironment + } + + return { environment } +} + +/** Loads staging configuration before modules that read database settings are imported. */ +export async function prepareTableWorkflowDeploymentBackfillEnvironment( + args: readonly string[] +): Promise { + const { environment } = parseTableWorkflowDeploymentBackfillArgs(args) + if (!environment) return + + const configuredSecretId = process.env.SIM_ENV_SECRET_ID + if (configuredSecretId && configuredSecretId !== STAGING_RUNTIME_SECRET_ID) { + throw new Error( + `SIM_ENV_SECRET_ID is already set to ${configuredSecretId}; expected ${STAGING_RUNTIME_SECRET_ID}` + ) + } + + const configuredDatabaseVariables = ['DATABASE_URL', 'DATABASE_URL_WEB'].filter( + (key) => key in process.env + ) + if (configuredDatabaseVariables.length > 0) { + throw new Error( + `Unset ${configuredDatabaseVariables.join(', ')} before using --environment=staging so local configuration cannot override staging` + ) + } + + process.env.SIM_ENV_SECRET_ID = STAGING_RUNTIME_SECRET_ID + await loadRuntimeSecrets() + + if (!process.env.DATABASE_URL && !process.env.DATABASE_URL_WEB) { + throw new Error(`${STAGING_RUNTIME_SECRET_ID} did not provide a database URL`) + } +} + +async function getDatabase() { + const { db } = await import('@sim/db') + return db +} + /** Ensures the table group references can be traversed without silently dropping corrupt data. */ async function assertTableWorkflowIntegrity(): Promise { + const db = await getDatabase() const [invalidSchema] = await db.execute(sql` SELECT id AS table_id FROM user_table_definitions @@ -200,6 +264,7 @@ export const postgresTableWorkflowDeploymentStore: TableWorkflowDeploymentStore assertIntegrity: assertTableWorkflowIntegrity, async listCandidates(afterWorkflowId, limit) { + const db = await getDatabase() const rows = await db.execute(sql` WITH referenced_workflow_ids AS ( SELECT DISTINCT workflow_group.value->>'workflowId' AS workflow_id @@ -238,6 +303,7 @@ export const postgresTableWorkflowDeploymentStore: TableWorkflowDeploymentStore }, async isDeployed(workflowId) { + const db = await getDatabase() const [state] = await db.execute(sql` SELECT workflow.is_deployed, @@ -265,6 +331,7 @@ export const postgresTableWorkflowDeploymentStore: TableWorkflowDeploymentStore export async function deployTableWorkflow( candidate: TableWorkflowDeploymentCandidate ): Promise { + const { performFullDeploy } = await import('@/lib/workflows/orchestration/deploy') return performFullDeploy({ workflowId: candidate.workflowId, userId: candidate.userId, @@ -365,8 +432,13 @@ export async function runTableWorkflowDeploymentBackfill(): Promise { logger.info('Table workflow deployment backfill completed', summary) } +async function main(): Promise { + await prepareTableWorkflowDeploymentBackfillEnvironment(process.argv.slice(2)) + await runTableWorkflowDeploymentBackfill() +} + if ((import.meta as { main?: boolean }).main) { - runTableWorkflowDeploymentBackfill() + main() .then(() => process.exit(0)) .catch((error: unknown) => { logger.error('Table workflow deployment backfill failed', { From 793b8dac3531cc6ff32e791f5e1133cd523c1a69 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Sat, 29 Aug 2026 11:35:03 -0700 Subject: [PATCH 10/13] fix(scripts): skip deleted table workflows --- ...backfill-table-workflow-deployments.test.ts | 18 ++++++++++++++++++ .../backfill-table-workflow-deployments.ts | 16 +++++----------- 2 files changed, 23 insertions(+), 11 deletions(-) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts index 6f878b99de6..f8da9eaf9fa 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -1,6 +1,7 @@ /** * @vitest-environment node */ +import { dbChainMockFns, resetDbChainMock } from '@sim/testing' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const { mockLoadRuntimeSecrets, mockPerformFullDeploy } = vi.hoisted(() => ({ @@ -20,6 +21,7 @@ import { backfillTableWorkflowDeployments, deployTableWorkflow, parseTableWorkflowDeploymentBackfillArgs, + postgresTableWorkflowDeploymentStore, prepareTableWorkflowDeploymentBackfillEnvironment, TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE, type TableWorkflowDeploymentCandidate, @@ -32,6 +34,10 @@ const ORIGINAL_ENV = { SIM_ENV_SECRET_ID: process.env.SIM_ENV_SECRET_ID, } +interface MockSqlQuery { + toSQL(): { sql: string } +} + function restoreEnvironmentVariable(key: keyof typeof ORIGINAL_ENV): void { const value = ORIGINAL_ENV[key] if (value === undefined) { @@ -63,6 +69,7 @@ function store( describe('backfillTableWorkflowDeployments', () => { beforeEach(() => { vi.clearAllMocks() + resetDbChainMock() }) afterEach(() => { @@ -112,6 +119,17 @@ describe('backfillTableWorkflowDeployments', () => { expect(mockLoadRuntimeSecrets).not.toHaveBeenCalled() }) + it('silently excludes missing and archived workflow references from integrity checks', async () => { + await postgresTableWorkflowDeploymentStore.assertIntegrity() + + const referenceQuery = dbChainMockFns.execute.mock.calls[2]?.[0] as MockSqlQuery + const queryText = referenceQuery.toSQL().sql + expect(queryText).toContain('INNER JOIN workflow') + expect(queryText).toContain('workflow.archived_at IS NULL') + expect(queryText).not.toContain('LEFT JOIN workflow') + expect(queryText).not.toContain('workflow.id IS NULL') + }) + it('deploys bounded keyset pages and verifies the final desired state', async () => { const listCandidates = vi .fn() diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index a486bcc3e98..72d4debcc4e 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -73,7 +73,7 @@ interface InvalidWorkflowReferenceRow extends Record { table_id: string table_workspace_id: string workflow_id: string - workflow_workspace_id: string | null + workflow_workspace_id: string } interface MultipleActiveVersionsRow extends Record { @@ -212,22 +212,16 @@ async function assertTableWorkflowIntegrity(): Promise { table_workflow_groups.workflow_id, workflow.workspace_id AS workflow_workspace_id FROM table_workflow_groups - LEFT JOIN workflow ON workflow.id = table_workflow_groups.workflow_id + INNER JOIN workflow ON workflow.id = table_workflow_groups.workflow_id + AND workflow.archived_at IS NULL WHERE table_workflow_groups.workflow_id <> '' - AND ( - workflow.id IS NULL - OR ( - workflow.archived_at IS NULL - AND workflow.workspace_id IS DISTINCT FROM table_workflow_groups.table_workspace_id - ) - ) + AND workflow.workspace_id IS DISTINCT FROM table_workflow_groups.table_workspace_id LIMIT 1 `) if (invalidReference) { - const workflowScope = invalidReference.workflow_workspace_id ?? 'missing' throw new Error( `Table ${invalidReference.table_id} in workspace ${invalidReference.table_workspace_id} ` + - `references workflow ${invalidReference.workflow_id} in workspace ${workflowScope}` + `references workflow ${invalidReference.workflow_id} in workspace ${invalidReference.workflow_workspace_id}` ) } From 4c1a53854bbccc5658aaef61f6b0d389dd4e2448 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Sat, 29 Aug 2026 11:42:06 -0700 Subject: [PATCH 11/13] fix(scripts): avoid private Redis in staging backfill --- .../backfill-table-workflow-deployments.test.ts | 12 ++++++++++++ .../scripts/backfill-table-workflow-deployments.ts | 6 ++++++ 2 files changed, 18 insertions(+) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts index f8da9eaf9fa..8bd7d6c2f9e 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -31,6 +31,8 @@ import { const ORIGINAL_ENV = { DATABASE_URL: process.env.DATABASE_URL, DATABASE_URL_WEB: process.env.DATABASE_URL_WEB, + REDIS_TLS_SERVERNAME: process.env.REDIS_TLS_SERVERNAME, + REDIS_URL: process.env.REDIS_URL, SIM_ENV_SECRET_ID: process.env.SIM_ENV_SECRET_ID, } @@ -75,30 +77,40 @@ describe('backfillTableWorkflowDeployments', () => { afterEach(() => { restoreEnvironmentVariable('DATABASE_URL') restoreEnvironmentVariable('DATABASE_URL_WEB') + restoreEnvironmentVariable('REDIS_TLS_SERVERNAME') + restoreEnvironmentVariable('REDIS_URL') restoreEnvironmentVariable('SIM_ENV_SECRET_ID') }) it('loads the staging runtime secret before database modules are needed', async () => { Reflect.deleteProperty(process.env, 'DATABASE_URL') Reflect.deleteProperty(process.env, 'DATABASE_URL_WEB') + Reflect.deleteProperty(process.env, 'REDIS_TLS_SERVERNAME') + Reflect.deleteProperty(process.env, 'REDIS_URL') Reflect.deleteProperty(process.env, 'SIM_ENV_SECRET_ID') mockLoadRuntimeSecrets.mockImplementation(async () => { process.env.DATABASE_URL = 'postgres://staging/database' + process.env.REDIS_TLS_SERVERNAME = 'cache.staging.internal' + process.env.REDIS_URL = 'rediss://cache.staging.internal:6379' }) await prepareTableWorkflowDeploymentBackfillEnvironment(['--environment=staging']) expect(process.env.SIM_ENV_SECRET_ID).toBe('/staging/sim/env-vars') + expect(process.env.REDIS_TLS_SERVERNAME).toBeUndefined() + expect(process.env.REDIS_URL).toBeUndefined() expect(mockLoadRuntimeSecrets).toHaveBeenCalledTimes(1) }) it('keeps the existing local DATABASE_URL mode when no environment is requested', async () => { process.env.DATABASE_URL = 'postgres://local/database' + process.env.REDIS_URL = 'redis://localhost:6379' await prepareTableWorkflowDeploymentBackfillEnvironment([]) expect(mockLoadRuntimeSecrets).not.toHaveBeenCalled() expect(process.env.DATABASE_URL).toBe('postgres://local/database') + expect(process.env.REDIS_URL).toBe('redis://localhost:6379') }) it('rejects unsupported, unknown, duplicate, and locally configured staging arguments', async () => { diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index 72d4debcc4e..dd8b6b0ef0c 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -26,6 +26,8 @@ export const TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE = 25 const BACKFILL_ACTOR_ID = 'table-workflow-deployment-backfill' const BACKFILL_OPERATION_VERSION = 'v2' const STAGING_RUNTIME_SECRET_ID = '/staging/sim/env-vars' +/** Container-private services that a locally executed staging backfill must not initialize. */ +const LOCAL_STAGING_OMITTED_VARIABLES = ['REDIS_URL', 'REDIS_TLS_SERVERNAME'] as const interface TableWorkflowDeploymentBackfillCliOptions { environment?: 'staging' @@ -145,6 +147,10 @@ export async function prepareTableWorkflowDeploymentBackfillEnvironment( if (!process.env.DATABASE_URL && !process.env.DATABASE_URL_WEB) { throw new Error(`${STAGING_RUNTIME_SECRET_ID} did not provide a database URL`) } + + for (const key of LOCAL_STAGING_OMITTED_VARIABLES) { + Reflect.deleteProperty(process.env, key) + } } async function getDatabase() { From 0e9c9ea35855891e68d2425b67efbb0f2427af13 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Sat, 29 Aug 2026 12:12:30 -0700 Subject: [PATCH 12/13] fix(scripts): skip locked table workflows --- .../workflows/orchestration/deploy.test.ts | 1 + .../sim/lib/workflows/orchestration/deploy.ts | 7 +- ...ackfill-table-workflow-deployments.test.ts | 50 +++++++++++ .../backfill-table-workflow-deployments.ts | 83 ++++++++++++++----- 4 files changed, 117 insertions(+), 24 deletions(-) diff --git a/apps/sim/lib/workflows/orchestration/deploy.test.ts b/apps/sim/lib/workflows/orchestration/deploy.test.ts index 262a45c30d7..08e4f63fce8 100644 --- a/apps/sim/lib/workflows/orchestration/deploy.test.ts +++ b/apps/sim/lib/workflows/orchestration/deploy.test.ts @@ -896,6 +896,7 @@ describe('mutation lock on the orchestration entry points', () => { expect(result.success).toBe(false) expect(result.error).toContain('locked') + expect(result.errorCode).toBe('locked') expect(mockRecordAudit).not.toHaveBeenCalled() }) diff --git a/apps/sim/lib/workflows/orchestration/deploy.ts b/apps/sim/lib/workflows/orchestration/deploy.ts index 5584a52412f..9290143972e 100644 --- a/apps/sim/lib/workflows/orchestration/deploy.ts +++ b/apps/sim/lib/workflows/orchestration/deploy.ts @@ -151,7 +151,7 @@ export async function performFullDeploy( // Backstop for every caller — routes may assert first to render their own 423, // but the copilot deploy tools call this directly. const lockDenial = await workflowLockDenial(workflowId) - if (lockDenial) return { success: false, error: lockDenial, errorCode: 'validation' } + if (lockDenial) return { success: false, error: lockDenial, errorCode: 'locked' } const [workflowRecord] = await db .select() @@ -509,6 +509,7 @@ export interface PerformFullUndeployParams { export interface PerformFullUndeployResult { success: boolean error?: string + errorCode?: OrchestrationErrorCode warnings?: string[] } @@ -526,7 +527,7 @@ export async function performFullUndeploy( const requestId = params.requestId ?? generateRequestId() const lockDenial = await workflowLockDenial(workflowId) - if (lockDenial) return { success: false, error: lockDenial } + if (lockDenial) return { success: false, error: lockDenial, errorCode: 'locked' } const [workflowRecord] = await db .select() @@ -661,7 +662,7 @@ export async function performActivateVersion( const idempotencyKey = params.idempotencyKey ?? generateId() const lockDenial = await workflowLockDenial(workflowId) - if (lockDenial) return { success: false, error: lockDenial, errorCode: 'validation' } + if (lockDenial) return { success: false, error: lockDenial, errorCode: 'locked' } const [versionRow] = await db .select({ diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts index 8bd7d6c2f9e..bb87b5ee287 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -172,6 +172,7 @@ describe('backfillTableWorkflowDeployments', () => { scanned: 3, deployed: 3, alreadyDeployed: 0, + skippedLocked: 0, }) expect(listCandidates.mock.calls).toEqual([ ['', 2], @@ -207,10 +208,59 @@ describe('backfillTableWorkflowDeployments', () => { scanned: 1, deployed: 0, alreadyDeployed: 1, + skippedLocked: 0, }) expect(deploy).not.toHaveBeenCalled() }) + it('reports and skips locked workflows while continuing the backfill', async () => { + const listCandidates = vi + .fn() + .mockResolvedValueOnce([candidate('workflow-a'), candidate('workflow-b')]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([candidate('workflow-a')]) + .mockResolvedValueOnce([]) + const deploymentState = new Map() + const isDeployed = vi + .fn() + .mockImplementation(async (workflowId) => deploymentState.get(workflowId) ?? false) + const deploy = vi.fn(async (workflow: TableWorkflowDeploymentCandidate) => { + if (workflow.workflowId === 'workflow-a') { + return { + success: false, + error: 'Workflow is locked by its containing folder', + errorCode: 'locked' as const, + } + } + deploymentState.set(workflow.workflowId, true) + return { + success: true, + activeDeployment: { + deploymentVersionId: `version-${workflow.workflowId}`, + version: 1, + deployedAt: new Date().toISOString(), + }, + } + }) + + await expect( + backfillTableWorkflowDeployments(store({ listCandidates, isDeployed }), deploy) + ).resolves.toEqual({ + scanned: 2, + deployed: 1, + alreadyDeployed: 0, + skippedLocked: 1, + }) + expect(deploy.mock.calls.map(([workflow]) => workflow.workflowId)).toEqual([ + 'workflow-a', + 'workflow-b', + ]) + expect(listCandidates.mock.calls.slice(-2)).toEqual([ + ['', 1], + ['workflow-a', 1], + ]) + }) + it('fails fast when canonical deployment fails', async () => { const listCandidates = vi .fn() diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index dd8b6b0ef0c..83e57524e75 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -1,7 +1,7 @@ #!/usr/bin/env bun /** - * Deploys every workflow referenced by a table workflow group. + * Deploys every mutable workflow referenced by a table workflow group. * * The script is idempotent and resumable: it pages over only workflows that do * not have the full desired state, and each deployment uses a stable idempotency @@ -52,6 +52,7 @@ export interface TableWorkflowDeploymentSummary { scanned: number deployed: number alreadyDeployed: number + skippedLocked: number } interface TableWorkflowDeploymentBackfillOptions { @@ -158,6 +159,50 @@ async function getDatabase() { return db } +function validateCandidatePage( + candidates: TableWorkflowDeploymentCandidate[], + afterWorkflowId: string, + limit: number +): string | null { + if (candidates.length === 0) return null + if (candidates.length > limit) { + throw new Error('Table workflow deployment store returned an oversized page') + } + + const pageIds = new Set(candidates.map((candidate) => candidate.workflowId)) + if (pageIds.size !== candidates.length) { + throw new Error('Table workflow deployment store returned duplicate workflow ids') + } + + const lastWorkflowId = candidates.at(-1)?.workflowId + if (!lastWorkflowId || lastWorkflowId === afterWorkflowId) { + throw new Error('Table workflow deployment store returned a non-advancing page') + } + return lastWorkflowId +} + +async function assertOnlyLockedCandidatesRemain( + store: TableWorkflowDeploymentStore, + lockedWorkflowIds: ReadonlySet +): Promise { + let afterWorkflowId = '' + for (;;) { + const candidates = await store.listCandidates(afterWorkflowId, 1) + const lastWorkflowId = validateCandidatePage(candidates, afterWorkflowId, 1) + if (!lastWorkflowId) return + + const unexpectedCandidate = candidates.find( + (candidate) => !lockedWorkflowIds.has(candidate.workflowId) + ) + if (unexpectedCandidate) { + throw new Error( + `Table workflow deployment backfill left workflow ${unexpectedCandidate.workflowId} undeployed` + ) + } + afterWorkflowId = lastWorkflowId + } +} + /** Ensures the table group references can be traversed without silently dropping corrupt data. */ async function assertTableWorkflowIntegrity(): Promise { const db = await getDatabase() @@ -343,7 +388,7 @@ export async function deployTableWorkflow( } /** - * Reaches and verifies that every table workflow has an active deployment. + * Reaches and verifies that every mutable table workflow has an active deployment. */ export async function backfillTableWorkflowDeployments( store: TableWorkflowDeploymentStore, @@ -361,24 +406,15 @@ export async function backfillTableWorkflowDeployments( scanned: 0, deployed: 0, alreadyDeployed: 0, + skippedLocked: 0, } + const lockedWorkflowIds = new Set() let afterWorkflowId = '' for (;;) { const candidates = await store.listCandidates(afterWorkflowId, batchSize) - if (candidates.length === 0) break - if (candidates.length > batchSize) { - throw new Error('Table workflow deployment store returned an oversized page') - } - - const pageIds = new Set(candidates.map((candidate) => candidate.workflowId)) - if (pageIds.size !== candidates.length) { - throw new Error('Table workflow deployment store returned duplicate workflow ids') - } - const lastWorkflowId = candidates.at(-1)?.workflowId - if (!lastWorkflowId || lastWorkflowId === afterWorkflowId) { - throw new Error('Table workflow deployment store returned a non-advancing page') - } + const lastWorkflowId = validateCandidatePage(candidates, afterWorkflowId, batchSize) + if (!lastWorkflowId) break for (const candidate of candidates) { summary.scanned += 1 @@ -391,6 +427,16 @@ export async function backfillTableWorkflowDeployments( }) const result = await deploy(candidate) if (!result.success) { + if (result.errorCode === 'locked') { + lockedWorkflowIds.add(candidate.workflowId) + summary.skippedLocked += 1 + logger.warn('Skipping locked workflow referenced by a table workflow group', { + workflowId: candidate.workflowId, + workspaceId: candidate.workspaceId, + reason: result.error ?? 'Workflow is locked', + }) + continue + } throw new Error( `Failed to deploy table workflow ${candidate.workflowId}: ${result.error ?? 'deployment returned no error'}` ) @@ -413,12 +459,7 @@ export async function backfillTableWorkflowDeployments( } await store.assertIntegrity() - const remaining = await store.listCandidates('', 1) - if (remaining.length > 0) { - throw new Error( - `Table workflow deployment backfill left workflow ${remaining[0].workflowId} undeployed` - ) - } + await assertOnlyLockedCandidatesRemain(store, lockedWorkflowIds) return summary } From 2efa0d72cfea7f38faa56b4bd9887377ff433902 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Sat, 29 Aug 2026 14:17:06 -0700 Subject: [PATCH 13/13] fix(scripts): support production workflow backfill --- ...ackfill-table-workflow-deployments.test.ts | 48 +++++++++++-------- .../backfill-table-workflow-deployments.ts | 35 +++++++++----- 2 files changed, 51 insertions(+), 32 deletions(-) diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts index bb87b5ee287..1eae182e775 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.test.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.test.ts @@ -82,25 +82,31 @@ describe('backfillTableWorkflowDeployments', () => { restoreEnvironmentVariable('SIM_ENV_SECRET_ID') }) - it('loads the staging runtime secret before database modules are needed', async () => { - Reflect.deleteProperty(process.env, 'DATABASE_URL') - Reflect.deleteProperty(process.env, 'DATABASE_URL_WEB') - Reflect.deleteProperty(process.env, 'REDIS_TLS_SERVERNAME') - Reflect.deleteProperty(process.env, 'REDIS_URL') - Reflect.deleteProperty(process.env, 'SIM_ENV_SECRET_ID') - mockLoadRuntimeSecrets.mockImplementation(async () => { - process.env.DATABASE_URL = 'postgres://staging/database' - process.env.REDIS_TLS_SERVERNAME = 'cache.staging.internal' - process.env.REDIS_URL = 'rediss://cache.staging.internal:6379' - }) - - await prepareTableWorkflowDeploymentBackfillEnvironment(['--environment=staging']) - - expect(process.env.SIM_ENV_SECRET_ID).toBe('/staging/sim/env-vars') - expect(process.env.REDIS_TLS_SERVERNAME).toBeUndefined() - expect(process.env.REDIS_URL).toBeUndefined() - expect(mockLoadRuntimeSecrets).toHaveBeenCalledTimes(1) - }) + it.each([ + ['production', '/production/sim/env-vars'], + ['staging', '/staging/sim/env-vars'], + ] as const)( + 'loads the %s runtime secret before database modules are needed', + async (environment, runtimeSecretId) => { + Reflect.deleteProperty(process.env, 'DATABASE_URL') + Reflect.deleteProperty(process.env, 'DATABASE_URL_WEB') + Reflect.deleteProperty(process.env, 'REDIS_TLS_SERVERNAME') + Reflect.deleteProperty(process.env, 'REDIS_URL') + Reflect.deleteProperty(process.env, 'SIM_ENV_SECRET_ID') + mockLoadRuntimeSecrets.mockImplementation(async () => { + process.env.DATABASE_URL = `postgres://${environment}/database` + process.env.REDIS_TLS_SERVERNAME = `cache.${environment}.internal` + process.env.REDIS_URL = `rediss://cache.${environment}.internal:6379` + }) + + await prepareTableWorkflowDeploymentBackfillEnvironment([`--environment=${environment}`]) + + expect(process.env.SIM_ENV_SECRET_ID).toBe(runtimeSecretId) + expect(process.env.REDIS_TLS_SERVERNAME).toBeUndefined() + expect(process.env.REDIS_URL).toBeUndefined() + expect(mockLoadRuntimeSecrets).toHaveBeenCalledTimes(1) + } + ) it('keeps the existing local DATABASE_URL mode when no environment is requested', async () => { process.env.DATABASE_URL = 'postgres://local/database' @@ -114,8 +120,8 @@ describe('backfillTableWorkflowDeployments', () => { }) it('rejects unsupported, unknown, duplicate, and locally configured staging arguments', async () => { - expect(() => parseTableWorkflowDeploymentBackfillArgs(['--environment=production'])).toThrow( - 'Unsupported backfill environment: production' + expect(() => parseTableWorkflowDeploymentBackfillArgs(['--environment=prod'])).toThrow( + 'Unsupported backfill environment: prod' ) expect(() => parseTableWorkflowDeploymentBackfillArgs(['--dry-run'])).toThrow( 'Unknown argument: --dry-run' diff --git a/apps/sim/scripts/backfill-table-workflow-deployments.ts b/apps/sim/scripts/backfill-table-workflow-deployments.ts index 83e57524e75..07f4fd0cf57 100644 --- a/apps/sim/scripts/backfill-table-workflow-deployments.ts +++ b/apps/sim/scripts/backfill-table-workflow-deployments.ts @@ -12,6 +12,7 @@ * Usage: * DATABASE_URL=... bun run apps/sim/scripts/backfill-table-workflow-deployments.ts * AWS_PROFILE=sim-admin bun --no-env-file apps/sim/scripts/backfill-table-workflow-deployments.ts --environment=staging + * AWS_PROFILE=sim-admin bun --no-env-file apps/sim/scripts/backfill-table-workflow-deployments.ts --environment=production */ import { createLogger } from '@sim/logger' @@ -25,12 +26,17 @@ const logger = createLogger('BackfillTableWorkflowDeployments') export const TABLE_WORKFLOW_DEPLOYMENT_BATCH_SIZE = 25 const BACKFILL_ACTOR_ID = 'table-workflow-deployment-backfill' const BACKFILL_OPERATION_VERSION = 'v2' -const STAGING_RUNTIME_SECRET_ID = '/staging/sim/env-vars' -/** Container-private services that a locally executed staging backfill must not initialize. */ -const LOCAL_STAGING_OMITTED_VARIABLES = ['REDIS_URL', 'REDIS_TLS_SERVERNAME'] as const +const RUNTIME_SECRET_IDS = { + production: '/production/sim/env-vars', + staging: '/staging/sim/env-vars', +} as const +/** Container-private services that a locally executed hosted backfill must not initialize. */ +const LOCAL_HOSTED_OMITTED_VARIABLES = ['REDIS_URL', 'REDIS_TLS_SERVERNAME'] as const + +type TableWorkflowDeploymentBackfillEnvironment = keyof typeof RUNTIME_SECRET_IDS interface TableWorkflowDeploymentBackfillCliOptions { - environment?: 'staging' + environment?: TableWorkflowDeploymentBackfillEnvironment } export interface TableWorkflowDeploymentCandidate { @@ -95,6 +101,12 @@ interface DeploymentStateRow extends Record { is_deployed: boolean } +function isTableWorkflowDeploymentBackfillEnvironment( + value: string +): value is TableWorkflowDeploymentBackfillEnvironment { + return Object.hasOwn(RUNTIME_SECRET_IDS, value) +} + /** Parses the deliberately small CLI surface for the backfill. */ export function parseTableWorkflowDeploymentBackfillArgs( args: readonly string[] @@ -110,7 +122,7 @@ export function parseTableWorkflowDeploymentBackfillArgs( } const requestedEnvironment = arg.slice('--environment='.length) - if (requestedEnvironment !== 'staging') { + if (!isTableWorkflowDeploymentBackfillEnvironment(requestedEnvironment)) { throw new Error(`Unsupported backfill environment: ${requestedEnvironment || '(empty)'}`) } environment = requestedEnvironment @@ -126,10 +138,11 @@ export async function prepareTableWorkflowDeploymentBackfillEnvironment( const { environment } = parseTableWorkflowDeploymentBackfillArgs(args) if (!environment) return + const runtimeSecretId = RUNTIME_SECRET_IDS[environment] const configuredSecretId = process.env.SIM_ENV_SECRET_ID - if (configuredSecretId && configuredSecretId !== STAGING_RUNTIME_SECRET_ID) { + if (configuredSecretId && configuredSecretId !== runtimeSecretId) { throw new Error( - `SIM_ENV_SECRET_ID is already set to ${configuredSecretId}; expected ${STAGING_RUNTIME_SECRET_ID}` + `SIM_ENV_SECRET_ID is already set to ${configuredSecretId}; expected ${runtimeSecretId}` ) } @@ -138,18 +151,18 @@ export async function prepareTableWorkflowDeploymentBackfillEnvironment( ) if (configuredDatabaseVariables.length > 0) { throw new Error( - `Unset ${configuredDatabaseVariables.join(', ')} before using --environment=staging so local configuration cannot override staging` + `Unset ${configuredDatabaseVariables.join(', ')} before using --environment=${environment} so local configuration cannot override ${environment}` ) } - process.env.SIM_ENV_SECRET_ID = STAGING_RUNTIME_SECRET_ID + process.env.SIM_ENV_SECRET_ID = runtimeSecretId await loadRuntimeSecrets() if (!process.env.DATABASE_URL && !process.env.DATABASE_URL_WEB) { - throw new Error(`${STAGING_RUNTIME_SECRET_ID} did not provide a database URL`) + throw new Error(`${runtimeSecretId} did not provide a database URL`) } - for (const key of LOCAL_STAGING_OMITTED_VARIABLES) { + for (const key of LOCAL_HOSTED_OMITTED_VARIABLES) { Reflect.deleteProperty(process.env, key) } }