From cb0caed8f4e4899b5bbf20f78edec1467d8d2835 Mon Sep 17 00:00:00 2001 From: Justin Blumencranz <96924014+j15z@users.noreply.github.com> Date: Fri, 11 Sep 2026 10:44:53 -0700 Subject: [PATCH] fix(workflows): save reordered tools and canonical modes atomically --- apps/realtime/src/database/operations.test.ts | 92 ++++++++++++++++++- apps/realtime/src/database/operations.ts | 25 ++++- .../components/tool-input/tool-input.tsx | 35 ++++--- apps/sim/hooks/use-collaborative-workflow.ts | 27 +++++- apps/sim/stores/workflows/subblock/store.ts | 1 + 5 files changed, 159 insertions(+), 21 deletions(-) diff --git a/apps/realtime/src/database/operations.test.ts b/apps/realtime/src/database/operations.test.ts index 061a37af21d..e2446a3a20c 100644 --- a/apps/realtime/src/database/operations.test.ts +++ b/apps/realtime/src/database/operations.test.ts @@ -1,5 +1,9 @@ /** @vitest-environment node */ -import { OPERATION_TARGETS, SUBBLOCK_OPERATIONS } from '@sim/realtime-protocol/constants' +import { + BLOCK_OPERATIONS, + OPERATION_TARGETS, + SUBBLOCK_OPERATIONS, +} from '@sim/realtime-protocol/constants' import { beforeEach, describe, expect, it, vi } from 'vitest' const { mockTransaction, mockSelectWhere, mockSet } = vi.hoisted(() => ({ @@ -121,3 +125,89 @@ describe('search replacement persistence', () => { expect(mockSet).toHaveBeenCalledTimes(1) }) }) + +describe('atomic tool reordering', () => { + const block = { + id: 'agent-1', + type: 'agent', + name: 'Agent', + position: { x: 0, y: 0 }, + locked: false, + subBlocks: { + tools: { + id: 'tools', + type: 'tool-input', + value: [{ type: 'jira', params: { projectId: 'project-1' } }], + }, + }, + data: {}, + } + + beforeEach(() => { + vi.clearAllMocks() + mockTransaction.mockImplementation( + async (callback: (tx: typeof transaction) => Promise) => callback(transaction) + ) + mockSet.mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) }) + mockSelectWhere.mockImplementation(() => + Object.assign( + Promise.resolve([{ ...block, subBlocks: { tools: { value: [{ type: 'function' }] } } }]), + { + limit: async () => [ + { ...block, subBlocks: { tools: { value: [{ type: 'function' }] } } }, + ], + } + ) + ) + }) + + it('persists a reordered tool array and its mode map in one write', async () => { + const first = { + type: 'jira', + params: { projectId: 'project-1', manualProjectId: '' }, + } + const second = { + type: 'jira', + params: { projectId: 'project-2', manualProjectId: '' }, + } + const original = { + ...block, + subBlocks: { tools: { id: 'tools', type: 'tool-input', value: [first, second] } }, + data: { canonicalModes: { '1:projectId': 'advanced' } }, + } + mockSelectWhere.mockResolvedValue([original]) + mockSet.mockReturnValue({ + where: () => + Object.assign(Promise.resolve(undefined), { returning: async () => [{ id: block.id }] }), + }) + const subBlocks = { tools: { id: 'tools', type: 'tool-input', value: [second, first] } } + const canonicalModes = { '0:projectId': 'advanced' } + await expect( + persistWorkflowOperation('workflow-1', { + operation: BLOCK_OPERATIONS.REPLACE_CANONICAL_MODES, + target: OPERATION_TARGETS.BLOCK, + timestamp: Date.now(), + payload: { id: block.id, subBlocks, data: { canonicalModes } }, + }) + ).resolves.toBeUndefined() + expect(mockSet).toHaveBeenLastCalledWith( + expect.objectContaining({ subBlocks, data: { canonicalModes } }) + ) + }) + + it('refuses an atomic tool update inside a locked container', async () => { + mockSelectWhere.mockResolvedValue([ + { ...block, data: { parentId: 'container' } }, + { id: 'container', type: 'loop', locked: true, data: {} }, + ]) + await expect( + persistWorkflowOperation('workflow-1', { + operation: BLOCK_OPERATIONS.REPLACE_CANONICAL_MODES, + target: OPERATION_TARGETS.BLOCK, + timestamp: Date.now(), + payload: { id: block.id, subBlocks: block.subBlocks, data: { canonicalModes: {} } }, + }) + ).rejects.toThrow('locked') + expect(mockSet).toHaveBeenCalledTimes(1) + }) +}) diff --git a/apps/realtime/src/database/operations.ts b/apps/realtime/src/database/operations.ts index 777c6078316..2725b7ac2fd 100644 --- a/apps/realtime/src/database/operations.ts +++ b/apps/realtime/src/database/operations.ts @@ -820,17 +820,34 @@ async function handleBlockOperationTx( throw new Error('Missing required fields for replace canonical modes operation') } - const existingBlock = await tx - .select({ data: workflowBlocks.data }) + const allBlocks = await tx + .select({ + id: workflowBlocks.id, + locked: workflowBlocks.locked, + subBlocks: workflowBlocks.subBlocks, + data: workflowBlocks.data, + }) .from(workflowBlocks) - .where(and(eq(workflowBlocks.id, payload.id), eq(workflowBlocks.workflowId, workflowId))) - .limit(1) + .where(eq(workflowBlocks.workflowId, workflowId)) + const blocksById = Object.fromEntries( + allBlocks.map((block: { id: string; locked: boolean; data: Record }) => [ + block.id, + block, + ]) + ) + if (isWorkflowBlockProtected(payload.id, blocksById)) { + throw new Error(`Block ${payload.id} is locked or inside a locked container`) + } + const existingBlock = allBlocks.filter((block: { id: string }) => block.id === payload.id) const currentData = (existingBlock?.[0]?.data as Record) || {} + const subBlocks = { ...(existingBlock[0]?.subBlocks || {}), ...(payload.subBlocks || {}) } + const updateResult = await tx .update(workflowBlocks) .set({ + ...(payload.subBlocks ? { subBlocks } : {}), data: { ...currentData, canonicalModes: payload.data.canonicalModes, diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/tool-input/tool-input.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/tool-input/tool-input.tsx index abd43aa7c1c..8c8d2ac3875 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/tool-input/tool-input.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/tool-input/tool-input.tsx @@ -395,11 +395,15 @@ export const ToolInput = memo(function ToolInput({ const { collaborativeSetBlockCanonicalMode, collaborativeSetBlockCanonicalModes } = useCollaborativeWorkflow() const reindexCanonicalModesOnMutate = useCallback( - (oldTools: StoredTool[], newTools: StoredTool[]) => { + (oldTools: StoredTool[], newTools: StoredTool[], persistedTools = newTools) => { const next = reindexToolCanonicalModes(oldTools, newTools, canonicalModeOverrides) - if (next) collaborativeSetBlockCanonicalModes(blockId, next) + if (!next) return false + collaborativeSetBlockCanonicalModes(blockId, next, { + [subBlockId]: { id: subBlockId, type: 'tool-input', value: persistedTools }, + }) + return true }, - [canonicalModeOverrides, collaborativeSetBlockCanonicalModes, blockId] + [canonicalModeOverrides, collaborativeSetBlockCanonicalModes, blockId, subBlockId] ) const value = isPreview ? previewValue : storeValue @@ -857,8 +861,9 @@ export const ToolInput = memo(function ToolInput({ (toolIndex: number) => { if (isPreview || disabled) return const updatedTools = selectedTools.filter((_, index) => index !== toolIndex) - reindexCanonicalModesOnMutate(selectedTools, updatedTools) - setStoreValue(updatedTools) + if (!reindexCanonicalModesOnMutate(selectedTools, updatedTools)) { + setStoreValue(updatedTools) + } }, [isPreview, disabled, selectedTools, reindexCanonicalModesOnMutate, setStoreValue] ) @@ -869,8 +874,9 @@ export const ToolInput = memo(function ToolInput({ const updatedTools = selectedTools.filter( (t) => !(t.type === 'mcp' && t.params?.serverId === serverId) ) - reindexCanonicalModesOnMutate(selectedTools, updatedTools) - setStoreValue(updatedTools) + if (!reindexCanonicalModesOnMutate(selectedTools, updatedTools)) { + setStoreValue(updatedTools) + } }, [isPreview, disabled, selectedTools, reindexCanonicalModesOnMutate, setStoreValue] ) @@ -900,8 +906,9 @@ export const ToolInput = memo(function ToolInput({ }) if (updatedTools.length !== selectedTools.length) { - reindexCanonicalModesOnMutate(selectedTools, updatedTools) - setStoreValue(updatedTools) + if (!reindexCanonicalModesOnMutate(selectedTools, updatedTools)) { + setStoreValue(updatedTools) + } } }, [selectedTools, customTools, reindexCanonicalModesOnMutate, setStoreValue] @@ -1077,8 +1084,9 @@ export const ToolInput = memo(function ToolInput({ newTools.splice(adjustedDropIndex, 0, draggedTool) } - reindexCanonicalModesOnMutate(selectedTools, newTools) - setStoreValue(newTools) + if (!reindexCanonicalModesOnMutate(selectedTools, newTools)) { + setStoreValue(newTools) + } setDraggedIndex(null) setDragOverIndex(null) } @@ -1177,8 +1185,9 @@ export const ToolInput = memo(function ToolInput({ ...filteredTools.map((tool) => ({ ...tool, isExpanded: false })), serverBinding, ] - reindexCanonicalModesOnMutate(selectedTools, filteredTools) - setStoreValue(nextTools) + if (!reindexCanonicalModesOnMutate(selectedTools, filteredTools, nextTools)) { + setStoreValue(nextTools) + } setMcpServerDrilldown(null) setOpen(false) }, diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index eb70e2f9089..0f251896076 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -12,7 +12,7 @@ import { WORKFLOW_OPERATIONS, } from '@sim/realtime-protocol/constants' import { generateId } from '@sim/utils/id' -import type { BlockRetryConfig } from '@sim/workflow-types/workflow' +import type { BlockRetryConfig, SubBlockState } from '@sim/workflow-types/workflow' import { filterAcyclicEdges, getWorkflowBlockNameConflict } from '@sim/workflow-types/workflow' import { useQueryClient } from '@tanstack/react-query' import type { Edge } from '@xyflow/react' @@ -59,6 +59,10 @@ import { findAllDescendantNodes, isBlockProtected } from '@/stores/workflows/wor const logger = createLogger('CollaborativeWorkflow') +interface CanonicalModeSubBlockState extends Omit { + value: unknown +} + export function useCollaborativeWorkflow() { const queryClient = useQueryClient() const undoRedo = useUndoRedo() @@ -245,6 +249,13 @@ export function useCollaborativeWorkflow() { useWorkflowStore .getState() .setBlockCanonicalModes(payload.id, payload.data?.canonicalModes ?? {}) + if (payload.subBlocks) { + for (const [subBlockId, subBlock] of Object.entries( + payload.subBlocks as Record + )) { + useSubBlockStore.getState().setValue(payload.id, subBlockId, subBlock.value) + } + } break } } else if (target === OPERATION_TARGETS.BLOCKS) { @@ -1367,14 +1378,24 @@ export function useCollaborativeWorkflow() { * {@link collaborativeSetBlockCanonicalMode}. Needed to reindex nested tool-input overrides on * reorder/removal: a merge can't atomically drop a now-stale index key, and sequential * per-key sets can clobber each other when two tools swap positions. + * Paired tool values travel in the same operation so their indexes stay aligned with the modes. */ const collaborativeSetBlockCanonicalModes = useCallback( - (id: string, canonicalModes: Record) => { + ( + id: string, + canonicalModes: Record, + subBlocks?: Record + ) => { if (isBaselineDiffView) { return } useWorkflowStore.getState().setBlockCanonicalModes(id, canonicalModes) + if (subBlocks) { + for (const [subBlockId, subBlock] of Object.entries(subBlocks)) { + useSubBlockStore.getState().setValue(id, subBlockId, subBlock.value) + } + } if (!activeWorkflowId) { return @@ -1386,7 +1407,7 @@ export function useCollaborativeWorkflow() { operation: { operation: BLOCK_OPERATIONS.REPLACE_CANONICAL_MODES, target: OPERATION_TARGETS.BLOCK, - payload: { id, data: { canonicalModes } }, + payload: { id, data: { canonicalModes }, ...(subBlocks ? { subBlocks } : {}) }, }, workflowId: activeWorkflowId, userId: session?.user?.id || 'unknown', diff --git a/apps/sim/stores/workflows/subblock/store.ts b/apps/sim/stores/workflows/subblock/store.ts index 14c609a57fb..6de02fa9a30 100644 --- a/apps/sim/stores/workflows/subblock/store.ts +++ b/apps/sim/stores/workflows/subblock/store.ts @@ -53,6 +53,7 @@ export const EMPTY_BLOCK_SUBBLOCK_VALUES: Record = {} * * - remote-broadcast application — already persisted server-side * - undo/redo — persists via its own queued inverse operations + * - canonical tool reindexing — queues paired tool values and modes in one operation * - synthetic tool subblock ids — excluded from both persistence and comparison * - whole-document replacement — the server's own state, re-seeded * - webhook management's runtime ids (webhookId/triggerPath/triggerConfig/