From f58d9f25ccc4c5957de004f4e5f977078c9fe348 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 10 Jul 2026 13:44:29 -0700 Subject: [PATCH 1/6] fix(workflow-edges): enforce edge/block validation server-side, not just client-side MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dragging a connection that creates a cycle correctly refused to render client-side, but the cyclic edge was still queued for realtime persistence and written to the DB, so it reappeared after refresh. Root cause: cycle detection (and several other edge/block rules) only lived in the client Zustand store and was never enforced by the realtime persistence layer, which is the actual source of truth on reload. - Move wouldCreateCycle, edge scope-boundary, annotation-only-block, duplicate-edge, and block-name-conflict checks into @sim/workflow-types so the client store, the collaborative queueing layer, and apps/realtime's DB write path all share one implementation - Wire these into apps/realtime/database/operations.ts's edge-add and block-rename handlers as the authoritative gate - Client-side behavior is unchanged (same call sites, same error messages, same rule ordering) — verified via existing + new test coverage --- apps/realtime/src/database/operations.ts | 186 +++++++++++++- apps/sim/executor/constants.ts | 25 +- apps/sim/hooks/use-collaborative-workflow.ts | 19 +- apps/sim/stores/workflows/utils.ts | 12 +- .../workflow/edge-validation.test.ts | 99 ++++++++ .../workflows/workflow/edge-validation.ts | 46 +--- .../stores/workflows/workflow/store.test.ts | 58 +++++ apps/sim/stores/workflows/workflow/store.ts | 38 +-- apps/sim/stores/workflows/workflow/utils.ts | 54 +--- packages/workflow-types/src/workflow.ts | 230 ++++++++++++++++++ 10 files changed, 639 insertions(+), 128 deletions(-) create mode 100644 apps/sim/stores/workflows/workflow/edge-validation.test.ts diff --git a/apps/realtime/src/database/operations.ts b/apps/realtime/src/database/operations.ts index 7f514bda77f..45e8d3b4756 100644 --- a/apps/realtime/src/database/operations.ts +++ b/apps/realtime/src/database/operations.ts @@ -24,7 +24,16 @@ import { import { randomFloat } from '@sim/utils/random' import { loadWorkflowFromNormalizedTablesRaw } from '@sim/workflow-persistence/load' import { mergeSubBlockValues } from '@sim/workflow-persistence/subblocks' -import { isWorkflowBlockProtected } from '@sim/workflow-types/workflow' +import { + filterAcyclicEdges, + filterUniqueWorkflowEdges, + getWorkflowBlockNameConflict, + getWorkflowEdgeScopeDropReason, + isKnownWorkflowTriggerBlock, + isWorkflowAnnotationOnlyBlockType, + isWorkflowBlockProtected, + wouldCreateCycle, +} from '@sim/workflow-types/workflow' import { and, eq, inArray, isNull, or, sql } from 'drizzle-orm' import { drizzle } from 'drizzle-orm/postgres-js' import postgres from 'postgres' @@ -32,6 +41,22 @@ import { env } from '@/env' const logger = createLogger('SocketDatabase') +interface PersistedEdgeRecord { + sourceBlockId: string + targetBlockId: string + sourceHandle: string | null + targetHandle: string | null +} + +function toEdgeHandles(edge: PersistedEdgeRecord) { + return { + source: edge.sourceBlockId, + target: edge.targetBlockId, + sourceHandle: edge.sourceHandle, + targetHandle: edge.targetHandle, + } +} + // Both realtime pools (this socketDb + the shared @sim/db pool) resolve the // realtime-keyed URL when set, falling back to the shared DATABASE_URL. const connectionString = @@ -406,6 +431,22 @@ async function handleBlockOperationTx( } } + const siblingBlocks = await tx + .select({ id: workflowBlocks.id, name: workflowBlocks.name }) + .from(workflowBlocks) + .where(eq(workflowBlocks.workflowId, workflowId)) + const siblingNamesById: Record = Object.fromEntries( + siblingBlocks.map((b: { id: string; name: string }) => [b.id, b.name]) + ) + + const nameConflict = getWorkflowBlockNameConflict(payload.id, payload.name, siblingNamesById) + if (nameConflict) { + logger.info( + `Skipping rename of block ${payload.id} - name conflict: ${nameConflict.reason}` + ) + break + } + await tx .update(workflowBlocks) .set({ @@ -1285,6 +1326,7 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str const edgeBlocks = await tx .select({ id: workflowBlocks.id, + type: workflowBlocks.type, locked: workflowBlocks.locked, data: workflowBlocks.data, }) @@ -1316,6 +1358,7 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str const parentBlocks = await tx .select({ id: workflowBlocks.id, + type: workflowBlocks.type, locked: workflowBlocks.locked, data: workflowBlocks.data, }) @@ -1331,11 +1374,68 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str } } + if (!blocksById[payload.source] || !blocksById[payload.target]) { + logger.info('Skipping edge add - edge references a missing block') + break + } + if (isWorkflowBlockProtected(payload.target, blocksById)) { logger.info(`Skipping edge add - target block is protected`) break } + if ( + isWorkflowAnnotationOnlyBlockType(blocksById[payload.source].type) || + isWorkflowAnnotationOnlyBlockType(blocksById[payload.target].type) + ) { + logger.info('Skipping edge add - edge references an annotation-only block') + break + } + + if (isKnownWorkflowTriggerBlock(blocksById[payload.target])) { + logger.info('Skipping edge add - trigger blocks cannot be edge targets') + break + } + + const scopeDropReason = getWorkflowEdgeScopeDropReason( + { source: payload.source, target: payload.target }, + blocksById + ) + if (scopeDropReason) { + logger.info(`Skipping edge add - ${scopeDropReason}`) + break + } + + const existingEdgesForCycleCheck: PersistedEdgeRecord[] = await tx + .select({ + sourceBlockId: workflowEdges.sourceBlockId, + targetBlockId: workflowEdges.targetBlockId, + sourceHandle: workflowEdges.sourceHandle, + targetHandle: workflowEdges.targetHandle, + }) + .from(workflowEdges) + .where(eq(workflowEdges.workflowId, workflowId)) + + const candidateEdge = { + source: payload.source, + target: payload.target, + sourceHandle: payload.sourceHandle ?? null, + targetHandle: payload.targetHandle ?? null, + } + const existingEdgeEndpoints = existingEdgesForCycleCheck.map(toEdgeHandles) + + if (filterUniqueWorkflowEdges([candidateEdge], existingEdgeEndpoints).length === 0) { + logger.info('Skipping edge add - duplicate edge already exists') + break + } + + if (wouldCreateCycle(existingEdgeEndpoints, candidateEdge.source, candidateEdge.target)) { + logger.info( + `Skipping edge add - would create a cycle: ${payload.source} -> ${payload.target}` + ) + break + } + await tx.insert(workflowEdges).values({ id: payload.id, workflowId, @@ -1568,6 +1668,7 @@ async function handleEdgesOperationTx( const connectedBlocks = await tx .select({ id: workflowBlocks.id, + type: workflowBlocks.type, locked: workflowBlocks.locked, data: workflowBlocks.data, }) @@ -1600,6 +1701,7 @@ async function handleEdgesOperationTx( const parentBlocks = await tx .select({ id: workflowBlocks.id, + type: workflowBlocks.type, locked: workflowBlocks.locked, data: workflowBlocks.data, }) @@ -1615,13 +1717,87 @@ async function handleEdgesOperationTx( } } - // Filter edges - only add edges where target block is not protected - const safeEdges = (edges as Array>).filter( - (e) => !isWorkflowBlockProtected(e.target as string, blocksById) + const droppedCounts: Record = {} + const countDrop = (reason: string) => { + droppedCounts[reason] = (droppedCounts[reason] ?? 0) + 1 + } + + // Mirror the client's validateEdges rule order: missing block, protected + // target, annotation-only block, trigger-block target, scope boundary. + const structurallyValidEdges = (edges as Array>).filter((e) => { + const source = e.source as string + const target = e.target as string + const sourceBlock = blocksById[source] + const targetBlock = blocksById[target] + + if (!sourceBlock || !targetBlock) { + countDrop('missing block') + return false + } + if (isWorkflowBlockProtected(target, blocksById)) { + countDrop('protected target block') + return false + } + if ( + isWorkflowAnnotationOnlyBlockType(sourceBlock.type) || + isWorkflowAnnotationOnlyBlockType(targetBlock.type) + ) { + countDrop('annotation-only block') + return false + } + if (isKnownWorkflowTriggerBlock(targetBlock)) { + countDrop('trigger block target') + return false + } + const scopeDropReason = getWorkflowEdgeScopeDropReason({ source, target }, blocksById) + if (scopeDropReason) { + countDrop(scopeDropReason) + return false + } + return true + }) + + if (structurallyValidEdges.length === 0) { + logger.info('All edges failed validation, skipping add', { droppedCounts }) + return + } + + const existingEdgesForCycleCheck: PersistedEdgeRecord[] = await tx + .select({ + sourceBlockId: workflowEdges.sourceBlockId, + targetBlockId: workflowEdges.targetBlockId, + sourceHandle: workflowEdges.sourceHandle, + targetHandle: workflowEdges.targetHandle, + }) + .from(workflowEdges) + .where(eq(workflowEdges.workflowId, workflowId)) + const existingEdgeEndpoints = existingEdgesForCycleCheck.map(toEdgeHandles) + + const uniqueEdges = filterUniqueWorkflowEdges( + structurallyValidEdges.map((e) => ({ + source: e.source as string, + target: e.target as string, + sourceHandle: (e.sourceHandle as string | null) ?? null, + targetHandle: (e.targetHandle as string | null) ?? null, + original: e, + })), + existingEdgeEndpoints + ) + + const safeEdges = filterAcyclicEdges(uniqueEdges, existingEdgeEndpoints).map( + (e) => e.original ) + if (safeEdges.length < edges.length) { + logger.info(`Dropped ${edges.length - safeEdges.length} invalid edge(s)`, { + droppedCounts, + droppedDuplicates: structurallyValidEdges.length - uniqueEdges.length, + droppedCyclic: uniqueEdges.length - safeEdges.length, + }) + } + if (safeEdges.length === 0) { - logger.info('All edges target protected blocks, skipping add') + logger.info('All edges were invalid, duplicate, or would create a cycle, skipping add') return } diff --git a/apps/sim/executor/constants.ts b/apps/sim/executor/constants.ts index b5883c61905..affd77355c8 100644 --- a/apps/sim/executor/constants.ts +++ b/apps/sim/executor/constants.ts @@ -1,3 +1,7 @@ +import { + normalizeWorkflowBlockName, + RESERVED_WORKFLOW_BLOCK_NAMES, +} from '@sim/workflow-types/workflow' import { getMaxExecutionTimeout } from '@/lib/core/execution-limits' import type { LoopType, ParallelType } from '@/lib/workflows/types' @@ -151,11 +155,12 @@ export const SPECIAL_REFERENCE_PREFIXES = [ REFERENCE.PREFIX.VARIABLE, ] as const -export const RESERVED_BLOCK_NAMES = [ - REFERENCE.PREFIX.LOOP, - REFERENCE.PREFIX.PARALLEL, - REFERENCE.PREFIX.VARIABLE, -] as const +/** + * Delegates to the shared implementation in `@sim/workflow-types` so the + * client store and the realtime persistence layer agree on the same reserved + * names. Values intentionally mirror REFERENCE.PREFIX.{LOOP,PARALLEL,VARIABLE} above. + */ +export const RESERVED_BLOCK_NAMES = RESERVED_WORKFLOW_BLOCK_NAMES export const LOOP_REFERENCE = { ITERATION: 'iteration', @@ -478,12 +483,10 @@ export function escapeRegExp(value: string): string { * spaces and dots. Used for both block names and variable names to ensure * consistent matching. * - * Dots are stripped because `.` is the reference path delimiter — a name like - * "Trigger.dev 1" must normalize to "triggerdev1" so the reference - * `` parses unambiguously. Dotted names could never be - * referenced before (the first path segment cut the name at the dot), so - * stripping dots cannot break any previously working reference. + * Delegates to the shared implementation in `@sim/workflow-types` so the + * client store and the realtime persistence layer normalize block names + * identically when checking for reserved/duplicate names. */ export function normalizeName(name: string): string { - return name.toLowerCase().replace(/\s+/g, '').replace(/\./g, '') + return normalizeWorkflowBlockName(name) } diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index f24ed617827..46132a31392 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -54,7 +54,11 @@ import type { Position, WorkflowState, } from '@/stores/workflows/workflow/types' -import { findAllDescendantNodes, isBlockProtected } from '@/stores/workflows/workflow/utils' +import { + filterAcyclicEdges, + findAllDescendantNodes, + isBlockProtected, +} from '@/stores/workflows/workflow/utils' const logger = createLogger('CollaborativeWorkflow') @@ -1423,7 +1427,12 @@ export function useCollaborativeWorkflow() { const currentEdges = useWorkflowStore.getState().edges const validEdges = filterValidEdges(edges, blocks) const newEdges = filterNewEdges(validEdges, currentEdges) - if (newEdges.length === 0) return false + // Reject cyclic edges here, before they are queued for realtime/DB + // persistence — the local store also runs this check, but only after + // an unfiltered payload would already be enqueued. Filtering once + // here keeps the queued payload and the local store in agreement. + const acyclicEdges = filterAcyclicEdges(newEdges, currentEdges) + if (acyclicEdges.length === 0) return false const operationId = generateId() @@ -1432,16 +1441,16 @@ export function useCollaborativeWorkflow() { operation: { operation: EDGES_OPERATIONS.BATCH_ADD_EDGES, target: OPERATION_TARGETS.EDGES, - payload: { edges: newEdges }, + payload: { edges: acyclicEdges }, }, workflowId: activeWorkflowId || '', userId: session?.user?.id || 'unknown', }) - useWorkflowStore.getState().batchAddEdges(newEdges, { skipValidation: true }) + useWorkflowStore.getState().batchAddEdges(acyclicEdges, { skipValidation: true }) if (!options?.skipUndoRedo) { - newEdges.forEach((edge) => undoRedo.recordAddEdge(edge.id)) + acyclicEdges.forEach((edge) => undoRedo.recordAddEdge(edge.id)) } return true diff --git a/apps/sim/stores/workflows/utils.ts b/apps/sim/stores/workflows/utils.ts index 5505a7738f1..a57f04b95af 100644 --- a/apps/sim/stores/workflows/utils.ts +++ b/apps/sim/stores/workflows/utils.ts @@ -1,5 +1,6 @@ import { generateId } from '@sim/utils/id' import { mergeSubblockStateWithValues } from '@sim/workflow-persistence/subblocks' +import { filterUniqueWorkflowEdges } from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import { DEFAULT_DUPLICATE_OFFSET } from '@/lib/workflows/autolayout/constants' import { getEffectiveBlockOutputs } from '@/lib/workflows/blocks/block-outputs' @@ -32,16 +33,7 @@ export function filterValidEdges(edges: Edge[], blocks: Record { - if (edge.source === edge.target) return false - return !currentEdges.some( - (e) => - e.source === edge.source && - e.sourceHandle === edge.sourceHandle && - e.target === edge.target && - e.targetHandle === edge.targetHandle - ) - }) + return filterUniqueWorkflowEdges(edgesToAdd, currentEdges) } export interface RegeneratedState { diff --git a/apps/sim/stores/workflows/workflow/edge-validation.test.ts b/apps/sim/stores/workflows/workflow/edge-validation.test.ts new file mode 100644 index 00000000000..d9c4cffda58 --- /dev/null +++ b/apps/sim/stores/workflows/workflow/edge-validation.test.ts @@ -0,0 +1,99 @@ +import type { Edge } from 'reactflow' +import { describe, expect, it } from 'vitest' +import { validateEdges } from '@/stores/workflows/workflow/edge-validation' +import type { BlockState } from '@/stores/workflows/workflow/types' + +function makeBlock(id: string, type: string, overrides?: Partial): BlockState { + return { + id, + type, + name: id, + position: { x: 0, y: 0 }, + subBlocks: {}, + outputs: {}, + enabled: true, + ...overrides, + } +} + +function makeEdge(id: string, source: string, target: string): Edge { + return { id, source, target, type: 'default' } +} + +describe('validateEdges', () => { + it('accepts an edge between two root-scope blocks', () => { + const blocks = { + a: makeBlock('a', 'starter'), + b: makeBlock('b', 'function'), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(1) + expect(result.dropped).toHaveLength(0) + }) + + it('drops an edge referencing a missing block', () => { + const blocks = { a: makeBlock('a', 'starter') } + const result = validateEdges([makeEdge('e1', 'a', 'missing')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toBe('edge references a missing block') + }) + + it('drops an edge touching an annotation-only (note) block', () => { + const blocks = { + a: makeBlock('a', 'note'), + b: makeBlock('b', 'function'), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toBe('edge references an annotation-only block') + }) + + it('drops an edge targeting a trigger block', () => { + const blocks = { + a: makeBlock('a', 'function'), + b: makeBlock('b', 'function', { triggerMode: true }), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toBe('trigger blocks cannot be edge targets') + }) + + it('drops an edge crossing loop scope boundaries', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + inner: makeBlock('inner', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + outer: makeBlock('outer', 'function'), + } + const result = validateEdges([makeEdge('e1', 'inner', 'outer')], blocks) + expect(result.valid).toHaveLength(0) + expect(result.dropped[0].reason).toContain('different scopes') + }) + + it('accepts an edge from a loop container into its own child', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + inner: makeBlock('inner', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + } + const result = validateEdges([makeEdge('e1', 'loop', 'inner')], blocks) + expect(result.valid).toHaveLength(1) + }) + + it('accepts an edge from a loop child back out to its own container', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + inner: makeBlock('inner', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + } + const result = validateEdges([makeEdge('e1', 'inner', 'loop')], blocks) + expect(result.valid).toHaveLength(1) + }) + + it('accepts edges between two siblings inside the same loop', () => { + const blocks = { + loop: makeBlock('loop', 'loop'), + a: makeBlock('a', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + b: makeBlock('b', 'function', { data: { parentId: 'loop', extent: 'parent' } }), + } + const result = validateEdges([makeEdge('e1', 'a', 'b')], blocks) + expect(result.valid).toHaveLength(1) + }) +}) diff --git a/apps/sim/stores/workflows/workflow/edge-validation.ts b/apps/sim/stores/workflows/workflow/edge-validation.ts index 034e87e177d..b6b0539f8cd 100644 --- a/apps/sim/stores/workflows/workflow/edge-validation.ts +++ b/apps/sim/stores/workflows/workflow/edge-validation.ts @@ -1,6 +1,9 @@ +import { + getWorkflowEdgeScopeDropReason, + isWorkflowAnnotationOnlyBlockType, +} from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import { TriggerUtils } from '@/lib/workflows/triggers/triggers' -import { isAnnotationOnlyBlock } from '@/executor/constants' import type { BlockState } from '@/stores/workflows/workflow/types' interface DroppedEdge { @@ -13,40 +16,6 @@ export interface EdgeValidationResult { dropped: DroppedEdge[] } -function isContainerBlock(block: BlockState | undefined): boolean { - return block?.type === 'loop' || block?.type === 'parallel' -} - -function getParentId(block: BlockState | undefined): string | null { - return block?.data?.parentId ?? null -} - -function getScopeDropReason(edge: Edge, blocks: Record): string | null { - const sourceBlock = blocks[edge.source] - const targetBlock = blocks[edge.target] - - if (!sourceBlock || !targetBlock) { - return 'edge references a missing block' - } - - const sourceParent = getParentId(sourceBlock) - const targetParent = getParentId(targetBlock) - - if (sourceParent === targetParent) { - return null - } - - if (targetParent === edge.source && isContainerBlock(sourceBlock)) { - return null - } - - if (sourceParent === edge.target && isContainerBlock(targetBlock)) { - return null - } - - return `blocks are in different scopes (${sourceParent ?? 'root'} -> ${targetParent ?? 'root'})` -} - export function validateEdges( edges: Edge[], blocks: Record @@ -63,7 +32,10 @@ export function validateEdges( continue } - if (isAnnotationOnlyBlock(sourceBlock.type) || isAnnotationOnlyBlock(targetBlock.type)) { + if ( + isWorkflowAnnotationOnlyBlockType(sourceBlock.type) || + isWorkflowAnnotationOnlyBlockType(targetBlock.type) + ) { dropped.push({ edge, reason: 'edge references an annotation-only block' }) continue } @@ -73,7 +45,7 @@ export function validateEdges( continue } - const scopeDropReason = getScopeDropReason(edge, blocks) + const scopeDropReason = getWorkflowEdgeScopeDropReason(edge, blocks) if (scopeDropReason) { dropped.push({ edge, reason: scopeDropReason }) continue diff --git a/apps/sim/stores/workflows/workflow/store.test.ts b/apps/sim/stores/workflows/workflow/store.test.ts index 860418bd7ee..11d9aa73106 100644 --- a/apps/sim/stores/workflows/workflow/store.test.ts +++ b/apps/sim/stores/workflows/workflow/store.test.ts @@ -334,6 +334,52 @@ describe('workflow store', () => { const state = useWorkflowStore.getState() expectEdgeCount(state, 1) }) + + it('should not add a self-loop edge', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + + batchAddEdges([{ id: 'e1', source: 'block-1', target: 'block-1' }]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 0) + }) + + it('should reject an edge that would create a cycle', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'Middle', { x: 200, y: 0 }) + addBlock('block-3', 'function', 'End', { x: 400, y: 0 }) + + batchAddEdges([{ id: 'e1', source: 'block-1', target: 'block-2' }]) + batchAddEdges([{ id: 'e2', source: 'block-2', target: 'block-3' }]) + // block-3 -> block-1 would close the loop back to block-1 + batchAddEdges([{ id: 'e3', source: 'block-3', target: 'block-1' }]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 2) + expect(state.edges.some((e) => e.id === 'e3')).toBe(false) + }) + + it('should reject a cyclic edge within the same batch', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'Middle', { x: 200, y: 0 }) + addBlock('block-3', 'function', 'End', { x: 400, y: 0 }) + + batchAddEdges([ + { id: 'e1', source: 'block-1', target: 'block-2' }, + { id: 'e2', source: 'block-2', target: 'block-3' }, + { id: 'e3', source: 'block-3', target: 'block-1' }, + ]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 2) + expect(state.edges.some((e) => e.id === 'e3')).toBe(false) + }) }) describe('batchRemoveEdges', () => { @@ -1593,6 +1639,18 @@ describe('workflow store', () => { expect(state.blocks.block2.name).toBe('Employee Length') }) + it('should reject reserved names (loop, parallel, variable)', () => { + const { updateBlockName } = useWorkflowStore.getState() + + for (const reserved of ['loop', 'Parallel', 'VARIABLE']) { + const result = updateBlockName('block1', reserved) + expect(result.success).toBe(false) + } + + const state = useWorkflowStore.getState() + expect(state.blocks.block1.name).toBe('Column AD') + }) + it('should return false when trying to rename a non-existent block', () => { const { updateBlockName } = useWorkflowStore.getState() diff --git a/apps/sim/stores/workflows/workflow/store.ts b/apps/sim/stores/workflows/workflow/store.ts index 75b4953a2bc..02d9a58761e 100644 --- a/apps/sim/stores/workflows/workflow/store.ts +++ b/apps/sim/stores/workflows/workflow/store.ts @@ -1,5 +1,6 @@ import { createLogger } from '@sim/logger' import { generateId } from '@sim/utils/id' +import { getWorkflowBlockNameConflict } from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import { create } from 'zustand' import { devtools } from 'zustand/middleware' @@ -8,7 +9,7 @@ import { getDynamicHandleSubblockType, isDynamicHandleSubblock, } from '@/lib/workflows/dynamic-handle-topology' -import { normalizeName, RESERVED_BLOCK_NAMES } from '@/executor/constants' +import { normalizeName } from '@/executor/constants' import { useSubBlockStore } from '@/stores/workflows/subblock/store' import { filterNewEdges, @@ -25,11 +26,11 @@ import type { } from '@/stores/workflows/workflow/types' import { clampParallelBatchSize, + filterAcyclicEdges, findAllDescendantNodes, generateLoopBlocks, generateParallelBlocks, isBlockProtected, - wouldCreateCycle, } from '@/stores/workflows/workflow/utils' import { normalizeWorkflowState } from '@/stores/workflows/workflow/validation' @@ -401,11 +402,10 @@ export const useWorkflowStore = create()( // Skip validation if already validated by caller (e.g., collaborative layer) const validEdges = options?.skipValidation ? edges : filterValidEdges(edges, blocks) const filtered = filterNewEdges(validEdges, currentEdges) - const newEdges = [...currentEdges] - - for (const edge of filtered) { - if (wouldCreateCycle([...newEdges], edge.source, edge.target)) continue - newEdges.push({ + const acyclicEdges = filterAcyclicEdges(filtered, currentEdges) + const newEdges = [ + ...currentEdges, + ...acyclicEdges.map((edge) => ({ id: edge.id || generateId(), source: edge.source, target: edge.target, @@ -413,8 +413,8 @@ export const useWorkflowStore = create()( targetHandle: edge.targetHandle, type: edge.type || 'default', data: edge.data || {}, - }) - } + })), + ] set({ blocks: { ...blocks }, @@ -657,26 +657,26 @@ export const useWorkflowStore = create()( const oldBlock = get().blocks[id] if (!oldBlock) return { success: false, changedSubblocks: [] } - const normalizedNewName = normalizeName(name) + const currentBlocks = get().blocks + const siblingNamesById = Object.fromEntries( + Object.entries(currentBlocks).map(([blockId, block]) => [blockId, block.name]) + ) + const conflict = getWorkflowBlockNameConflict(id, name, siblingNamesById) - if (!normalizedNewName) { + if (conflict?.reason === 'empty') { logger.error(`Cannot rename block to empty name`) return { success: false, changedSubblocks: [] } } - const currentBlocks = get().blocks - const conflictingBlock = Object.entries(currentBlocks).find( - ([blockId, block]) => blockId !== id && normalizeName(block.name) === normalizedNewName - ) - - if (conflictingBlock) { + if (conflict?.reason === 'duplicate') { + const conflictingBlock = currentBlocks[conflict.conflictingBlockId as string] logger.error( - `Cannot rename block to "${name}" - conflicts with "${conflictingBlock[1].name}"` + `Cannot rename block to "${name}" - conflicts with "${conflictingBlock.name}"` ) return { success: false, changedSubblocks: [] } } - if ((RESERVED_BLOCK_NAMES as readonly string[]).includes(normalizedNewName)) { + if (conflict?.reason === 'reserved') { logger.error(`Cannot rename block to reserved name: "${name}"`) return { success: false, changedSubblocks: [] } } diff --git a/apps/sim/stores/workflows/workflow/utils.ts b/apps/sim/stores/workflows/workflow/utils.ts index a7077dc0903..9413beeea99 100644 --- a/apps/sim/stores/workflows/workflow/utils.ts +++ b/apps/sim/stores/workflows/workflow/utils.ts @@ -1,6 +1,8 @@ import { + filterAcyclicEdges as filterAcyclicWorkflowEdges, isWorkflowBlockAncestorLocked, isWorkflowBlockProtected, + wouldCreateCycle as wouldCreateWorkflowEdgeCycle, } from '@sim/workflow-types/workflow' import type { Edge } from 'reactflow' import type { BlockState, Loop, Parallel } from '@/stores/workflows/workflow/types' @@ -19,50 +21,20 @@ export function clampParallelBatchSize(batchSize: unknown): number { /** * Check if adding an edge would create a cycle in the graph. - * Uses depth-first search to detect if the source node is reachable from the target node. - * - * @param edges - Current edges in the graph - * @param sourceId - Source node ID of the proposed edge - * @param targetId - Target node ID of the proposed edge - * @returns true if adding this edge would create a cycle + * Delegates to the shared implementation in `@sim/workflow-types` so the + * client store, the collaborative queueing layer, and the realtime + * persistence layer all agree on the same cyclic edges. */ export function wouldCreateCycle(edges: Edge[], sourceId: string, targetId: string): boolean { - if (sourceId === targetId) { - return true - } - - const adjacencyList = new Map() - for (const edge of edges) { - if (!adjacencyList.has(edge.source)) { - adjacencyList.set(edge.source, []) - } - adjacencyList.get(edge.source)!.push(edge.target) - } - - const visited = new Set() - - function canReachSource(currentNode: string): boolean { - if (currentNode === sourceId) { - return true - } - - if (visited.has(currentNode)) { - return false - } - - visited.add(currentNode) - - const neighbors = adjacencyList.get(currentNode) || [] - for (const neighbor of neighbors) { - if (canReachSource(neighbor)) { - return true - } - } - - return false - } + return wouldCreateWorkflowEdgeCycle(edges, sourceId, targetId) +} - return canReachSource(targetId) +/** + * Filters a batch of candidate edges down to the ones that do not create a + * cycle against `currentEdges`, evaluated incrementally within the batch. + */ +export function filterAcyclicEdges(edgesToAdd: Edge[], currentEdges: Edge[]): Edge[] { + return filterAcyclicWorkflowEdges(edgesToAdd, currentEdges) } /** diff --git a/packages/workflow-types/src/workflow.ts b/packages/workflow-types/src/workflow.ts index 06b9692ddba..24c68f91b8d 100644 --- a/packages/workflow-types/src/workflow.ts +++ b/packages/workflow-types/src/workflow.ts @@ -129,6 +129,236 @@ export function isWorkflowBlockProtected( return Boolean(block.locked || isWorkflowBlockAncestorLocked(blockId, blocks)) } +export interface WorkflowEdgeEndpoints { + source: string + target: string +} + +/** + * Checks whether adding an edge would create a cycle in the graph, via DFS + * reachability from the proposed target back to the proposed source. + * + * Shared between the client store, the collaborative queueing layer, and the + * realtime persistence layer so all three agree on the same cyclic edges — + * an edge rejected by one must never be accepted (and persisted) by another. + */ +export function wouldCreateCycle( + edges: WorkflowEdgeEndpoints[], + sourceId: string, + targetId: string +): boolean { + if (sourceId === targetId) { + return true + } + + const adjacencyList = new Map() + for (const edge of edges) { + if (!adjacencyList.has(edge.source)) { + adjacencyList.set(edge.source, []) + } + adjacencyList.get(edge.source)!.push(edge.target) + } + + const visited = new Set() + + function canReachSource(currentNode: string): boolean { + if (currentNode === sourceId) { + return true + } + + if (visited.has(currentNode)) { + return false + } + + visited.add(currentNode) + + const neighbors = adjacencyList.get(currentNode) || [] + for (const neighbor of neighbors) { + if (canReachSource(neighbor)) { + return true + } + } + + return false + } + + return canReachSource(targetId) +} + +/** + * Filters a batch of candidate edges down to the ones that do not create a + * cycle, evaluated incrementally so edges within the same batch that would + * chain into a cycle are also rejected. + */ +export function filterAcyclicEdges( + edgesToAdd: T[], + currentEdges: WorkflowEdgeEndpoints[] +): T[] { + const workingEdges: WorkflowEdgeEndpoints[] = [...currentEdges] + const acyclicEdges: T[] = [] + + for (const edge of edgesToAdd) { + if (wouldCreateCycle(workingEdges, edge.source, edge.target)) continue + workingEdges.push(edge) + acyclicEdges.push(edge) + } + + return acyclicEdges +} + +/** Edge endpoints that also identify the specific handle used on each side. */ +export interface WorkflowEdgeHandles extends WorkflowEdgeEndpoints { + sourceHandle?: string | null + targetHandle?: string | null +} + +function isDuplicateWorkflowEdge( + edge: WorkflowEdgeHandles, + existing: WorkflowEdgeHandles +): boolean { + return ( + edge.source === existing.source && + (edge.sourceHandle ?? null) === (existing.sourceHandle ?? null) && + edge.target === existing.target && + (edge.targetHandle ?? null) === (existing.targetHandle ?? null) + ) +} + +/** + * Filters a batch of candidate edges down to the ones that are not + * self-loops and do not duplicate an edge already present (same source, + * target, and handles). Shared between the client store, the collaborative + * queueing layer, and the realtime persistence layer. + */ +export function filterUniqueWorkflowEdges( + edgesToAdd: T[], + currentEdges: WorkflowEdgeHandles[] +): T[] { + return edgesToAdd.filter((edge) => { + if (edge.source === edge.target) return false + return !currentEdges.some((existing) => isDuplicateWorkflowEdge(edge, existing)) + }) +} + +const WORKFLOW_CONTAINER_BLOCK_TYPES = new Set(['loop', 'parallel']) +const WORKFLOW_ANNOTATION_ONLY_BLOCK_TYPE = 'note' +/** Legacy trigger block type — see TRIGGER_TYPES.STARTER in apps/sim/lib/workflows/triggers/triggers.ts. */ +const LEGACY_STARTER_BLOCK_TYPE = 'starter' + +export interface WorkflowEdgeScopeBlock extends WorkflowLockBlock { + type?: string +} + +/** + * Checks whether an edge crosses loop/parallel scope boundaries — an edge is + * only valid within the same container, or between a container and its + * direct parent/child. + */ +export function getWorkflowEdgeScopeDropReason( + edge: WorkflowEdgeEndpoints, + blocks: Record +): string | null { + const sourceBlock = blocks[edge.source] + const targetBlock = blocks[edge.target] + + if (!sourceBlock || !targetBlock) { + return 'edge references a missing block' + } + + const sourceParent = getWorkflowBlockParentId(sourceBlock) ?? null + const targetParent = getWorkflowBlockParentId(targetBlock) ?? null + + if (sourceParent === targetParent) { + return null + } + + if (targetParent === edge.source && WORKFLOW_CONTAINER_BLOCK_TYPES.has(sourceBlock.type ?? '')) { + return null + } + + if (sourceParent === edge.target && WORKFLOW_CONTAINER_BLOCK_TYPES.has(targetBlock.type ?? '')) { + return null + } + + return `blocks are in different scopes (${sourceParent ?? 'root'} -> ${targetParent ?? 'root'})` +} + +/** True when a block's type is the annotation-only "note" block, which cannot participate in edges. */ +export function isWorkflowAnnotationOnlyBlockType(blockType: string | undefined): boolean { + return blockType === WORKFLOW_ANNOTATION_ONLY_BLOCK_TYPE +} + +export interface WorkflowTriggerCapableBlock { + type?: string + triggerMode?: boolean +} + +/** + * Portable subset of trigger-block detection: an explicit `triggerMode` + * toggle, or the legacy starter block type. Does NOT cover blocks whose + * trigger status comes from the block registry's `category: 'triggers'` + * field (most modern trigger blocks) — that classification only exists in + * the client's block registry (apps/sim/blocks), which apps/realtime is + * architecturally forbidden from importing (see + * scripts/check-monorepo-boundaries.ts). Persisting a redundant "isTrigger" + * flag on the block row to cover that case would itself be a driftable + * duplicate, so that case remains client-enforced only; see + * TriggerUtils.isTriggerBlock in apps/sim/lib/workflows/triggers/triggers.ts + * for the full (client-only) check. + */ +export function isKnownWorkflowTriggerBlock(block: WorkflowTriggerCapableBlock): boolean { + return block.triggerMode === true || block.type === LEGACY_STARTER_BLOCK_TYPE +} + +/** + * Names reserved for reference-path prefixes (``, ``, + * ``) — a block cannot take one of these as its (normalized) name. + */ +export const RESERVED_WORKFLOW_BLOCK_NAMES = ['loop', 'parallel', 'variable'] as const + +/** + * Normalizes a block name into its reference-safe form: lowercased, whitespace + * stripped, dots stripped. Dots are stripped because `.` is the reference path + * delimiter — a name like "Trigger.dev 1" must normalize to "triggerdev1" so + * the reference `` parses unambiguously. + */ +export function normalizeWorkflowBlockName(name: string): string { + return name.toLowerCase().replace(/\s+/g, '').replace(/\./g, '') +} + +export interface WorkflowBlockNameConflict { + reason: 'empty' | 'reserved' | 'duplicate' + conflictingBlockId?: string +} + +/** + * Checks whether renaming `blockId` to `name` would produce an empty name, a + * reserved name, or collide with another block's normalized name. Block names + * are used as `` reference identifiers, so two blocks landing + * on the same normalized name makes reference resolution ambiguous — shared + * between the client store and the realtime persistence layer so both agree + * on the same conflicts. + */ +export function getWorkflowBlockNameConflict( + blockId: string, + name: string, + siblingNamesById: Record +): WorkflowBlockNameConflict | null { + const normalized = normalizeWorkflowBlockName(name) + if (!normalized) return { reason: 'empty' } + + const conflict = Object.entries(siblingNamesById).find( + ([id, siblingName]) => id !== blockId && normalizeWorkflowBlockName(siblingName) === normalized + ) + if (conflict) return { reason: 'duplicate', conflictingBlockId: conflict[0] } + + if ((RESERVED_WORKFLOW_BLOCK_NAMES as readonly string[]).includes(normalized)) { + return { reason: 'reserved' } + } + + return null +} + export interface SubBlockState { id: string type: SubBlockType From 878b0277f774bbb3f3f2292dce53ed13cde3a91b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 10 Jul 2026 13:52:17 -0700 Subject: [PATCH 2/6] fix(workflow-edges): address review findings on realtime edge validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Select triggerMode when fetching blocks for edge-add validation — isKnownWorkflowTriggerBlock checked block.triggerMode but the column was never fetched from the DB, so trigger-mode blocks could still receive an incoming edge (Cursor Bugbot) - Make filterUniqueWorkflowEdges incremental, so two duplicate edges within the same BATCH_ADD_EDGES payload are also deduped instead of both surviving (Greptile) --- apps/realtime/src/database/operations.ts | 2 ++ .../stores/workflows/workflow/store.test.ts | 17 +++++++++++++++ packages/workflow-types/src/workflow.ts | 21 +++++++++++++------ 3 files changed, 34 insertions(+), 6 deletions(-) diff --git a/apps/realtime/src/database/operations.ts b/apps/realtime/src/database/operations.ts index 45e8d3b4756..33b8e076d1e 100644 --- a/apps/realtime/src/database/operations.ts +++ b/apps/realtime/src/database/operations.ts @@ -1328,6 +1328,7 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str id: workflowBlocks.id, type: workflowBlocks.type, locked: workflowBlocks.locked, + triggerMode: workflowBlocks.triggerMode, data: workflowBlocks.data, }) .from(workflowBlocks) @@ -1670,6 +1671,7 @@ async function handleEdgesOperationTx( id: workflowBlocks.id, type: workflowBlocks.type, locked: workflowBlocks.locked, + triggerMode: workflowBlocks.triggerMode, data: workflowBlocks.data, }) .from(workflowBlocks) diff --git a/apps/sim/stores/workflows/workflow/store.test.ts b/apps/sim/stores/workflows/workflow/store.test.ts index 11d9aa73106..26dd05599cd 100644 --- a/apps/sim/stores/workflows/workflow/store.test.ts +++ b/apps/sim/stores/workflows/workflow/store.test.ts @@ -335,6 +335,23 @@ describe('workflow store', () => { expectEdgeCount(state, 1) }) + it('should not add duplicate connections that appear twice within the same batch', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'End', { x: 200, y: 0 }) + + batchAddEdges([ + { id: 'e1', source: 'block-1', target: 'block-2' }, + { id: 'e2', source: 'block-1', target: 'block-2' }, + ]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 1) + expect(state.edges.some((e) => e.id === 'e1')).toBe(true) + expect(state.edges.some((e) => e.id === 'e2')).toBe(false) + }) + it('should not add a self-loop edge', () => { const { batchAddEdges } = useWorkflowStore.getState() diff --git a/packages/workflow-types/src/workflow.ts b/packages/workflow-types/src/workflow.ts index 24c68f91b8d..fcf1e03cf7b 100644 --- a/packages/workflow-types/src/workflow.ts +++ b/packages/workflow-types/src/workflow.ts @@ -227,17 +227,26 @@ function isDuplicateWorkflowEdge( /** * Filters a batch of candidate edges down to the ones that are not * self-loops and do not duplicate an edge already present (same source, - * target, and handles). Shared between the client store, the collaborative - * queueing layer, and the realtime persistence layer. + * target, and handles), evaluated incrementally so two duplicate edges + * within the same batch are also caught (only the first survives). Shared + * between the client store, the collaborative queueing layer, and the + * realtime persistence layer. */ export function filterUniqueWorkflowEdges( edgesToAdd: T[], currentEdges: WorkflowEdgeHandles[] ): T[] { - return edgesToAdd.filter((edge) => { - if (edge.source === edge.target) return false - return !currentEdges.some((existing) => isDuplicateWorkflowEdge(edge, existing)) - }) + const workingEdges: WorkflowEdgeHandles[] = [...currentEdges] + const uniqueEdges: T[] = [] + + for (const edge of edgesToAdd) { + if (edge.source === edge.target) continue + if (workingEdges.some((existing) => isDuplicateWorkflowEdge(edge, existing))) continue + workingEdges.push(edge) + uniqueEdges.push(edge) + } + + return uniqueEdges } const WORKFLOW_CONTAINER_BLOCK_TYPES = new Set(['loop', 'parallel']) From b959202db50b942d4f88542e1d1a264decd9f2f4 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 10 Jul 2026 14:00:53 -0700 Subject: [PATCH 3/6] fix(workflow-edges): normalize empty-string handles in duplicate-edge check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit filterUniqueWorkflowEdges compared handles with ??, so a `sourceHandle: ''` edge wasn't recognized as a duplicate of an existing null-handle edge — even though both get persisted as the same null value at insert time (edge.sourceHandle || null). Falsy-coalesce in the comparison so '' and null/undefined are treated as the same "no handle" state everywhere. (Greptile) --- .../sim/stores/workflows/workflow/store.test.ts | 17 +++++++++++++++++ packages/workflow-types/src/workflow.ts | 16 ++++++++++++++-- 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/apps/sim/stores/workflows/workflow/store.test.ts b/apps/sim/stores/workflows/workflow/store.test.ts index 26dd05599cd..f5072971faf 100644 --- a/apps/sim/stores/workflows/workflow/store.test.ts +++ b/apps/sim/stores/workflows/workflow/store.test.ts @@ -352,6 +352,23 @@ describe('workflow store', () => { expect(state.edges.some((e) => e.id === 'e2')).toBe(false) }) + it('should treat an empty-string handle as equivalent to no handle when deduping', () => { + const { batchAddEdges } = useWorkflowStore.getState() + + addBlock('block-1', 'starter', 'Start', { x: 0, y: 0 }) + addBlock('block-2', 'function', 'End', { x: 200, y: 0 }) + + batchAddEdges([ + { id: 'e1', source: 'block-1', target: 'block-2', sourceHandle: '' }, + { id: 'e2', source: 'block-1', target: 'block-2' }, + ]) + + const state = useWorkflowStore.getState() + expectEdgeCount(state, 1) + expect(state.edges.some((e) => e.id === 'e1')).toBe(true) + expect(state.edges.some((e) => e.id === 'e2')).toBe(false) + }) + it('should not add a self-loop edge', () => { const { batchAddEdges } = useWorkflowStore.getState() diff --git a/packages/workflow-types/src/workflow.ts b/packages/workflow-types/src/workflow.ts index fcf1e03cf7b..c4aa764aecf 100644 --- a/packages/workflow-types/src/workflow.ts +++ b/packages/workflow-types/src/workflow.ts @@ -212,15 +212,27 @@ export interface WorkflowEdgeHandles extends WorkflowEdgeEndpoints { targetHandle?: string | null } +// Falsy-coalesce (not nullish-coalesce): persistence normalizes a missing +// handle to `null` via `edge.sourceHandle || null` (see +// apps/realtime/src/database/operations.ts), which also maps `''` to +// `null`. Comparing with `??` would treat `''` and `null` as distinct +// handles pre-insert while both are written as the same `null` value, +// letting a `sourceHandle: ''` edge slip past the duplicate check. +function normalizeWorkflowEdgeHandle(handle: string | null | undefined): string | null { + return handle || null +} + function isDuplicateWorkflowEdge( edge: WorkflowEdgeHandles, existing: WorkflowEdgeHandles ): boolean { return ( edge.source === existing.source && - (edge.sourceHandle ?? null) === (existing.sourceHandle ?? null) && + normalizeWorkflowEdgeHandle(edge.sourceHandle) === + normalizeWorkflowEdgeHandle(existing.sourceHandle) && edge.target === existing.target && - (edge.targetHandle ?? null) === (existing.targetHandle ?? null) + normalizeWorkflowEdgeHandle(edge.targetHandle) === + normalizeWorkflowEdgeHandle(existing.targetHandle) ) } From c1ce0a0730d5a0fe39c63e434e1d6071ade84707 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 10 Jul 2026 14:32:13 -0700 Subject: [PATCH 4/6] improvement(workflow-edges): dedup realtime edge-add validation, reuse block-name-conflict helper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /simplify pass on the already-merged-quality PR before sign-off: - Extract filterEdgesForPersist in apps/realtime/src/database/operations.ts: the single-edge ADD and batch BATCH_ADD_EDGES handlers hand-inlined the same six-step validation pipeline (missing block, protected target, annotation-only, trigger-target, scope boundary, duplicate, cycle) and each independently re-fetched blocksById/existingEdgesForCycleCheck. Two copies of one rule in the same file was exactly the drift risk this PR otherwise closes across client/server. One shared helper now backs both. Net -63 lines despite the new shared function. - Fix a real bug this surfaced: droppedCounts was keyed by the free-text, block-id-bearing scope-boundary message, so it could never aggregate across edges/runs. Now keyed by a stable 'scope boundary' reason. - use-collaborative-workflow.ts's collaborativeUpdateBlockName still hand-rolled the empty/reserved/duplicate block-name-conflict checks this PR centralized as getWorkflowBlockNameConflict (already adopted by store.ts). Switched it to the shared helper, which also fixes a latent check-order mismatch between the two (this pre-check ran reserved before duplicate; the store's real gate — after this PR's own store.ts change — runs duplicate before reserved, so they could disagree on which toast a name that was both reserved and duplicate would surface). --- apps/realtime/src/database/operations.ts | 432 ++++++++----------- apps/sim/hooks/use-collaborative-workflow.ts | 21 +- 2 files changed, 195 insertions(+), 258 deletions(-) diff --git a/apps/realtime/src/database/operations.ts b/apps/realtime/src/database/operations.ts index 33b8e076d1e..38ff1f3691b 100644 --- a/apps/realtime/src/database/operations.ts +++ b/apps/realtime/src/database/operations.ts @@ -32,7 +32,6 @@ import { isKnownWorkflowTriggerBlock, isWorkflowAnnotationOnlyBlockType, isWorkflowBlockProtected, - wouldCreateCycle, } from '@sim/workflow-types/workflow' import { and, eq, inArray, isNull, or, sql } from 'drizzle-orm' import { drizzle } from 'drizzle-orm/postgres-js' @@ -57,6 +56,155 @@ function toEdgeHandles(edge: PersistedEdgeRecord) { } } +interface EdgeAddCandidate { + id?: string + source: string + target: string + sourceHandle?: string | null + targetHandle?: string | null +} + +interface FilterEdgesForPersistResult { + safeEdges: T[] + droppedCounts: Record + droppedDuplicates: number + droppedCyclic: number +} + +/** + * Runs the full server-side edge-add validation pipeline — missing block, + * protected target, annotation-only block, trigger-block target, scope + * boundary, duplicate, cycle — against one or more candidate edges. Shared by + * the single-edge `ADD` and `BATCH_ADD_EDGES` handlers so the two paths + * can't drift out of sync with each other, the same way the client and + * realtime layers already share these rules via `@sim/workflow-types`. + */ +async function filterEdgesForPersist( + tx: any, + workflowId: string, + candidates: T[] +): Promise> { + const connectedBlockIds = new Set() + for (const edge of candidates) { + connectedBlockIds.add(edge.source) + connectedBlockIds.add(edge.target) + } + + const connectedBlocks = await tx + .select({ + id: workflowBlocks.id, + type: workflowBlocks.type, + locked: workflowBlocks.locked, + triggerMode: workflowBlocks.triggerMode, + data: workflowBlocks.data, + }) + .from(workflowBlocks) + .where( + and( + eq(workflowBlocks.workflowId, workflowId), + inArray(workflowBlocks.id, Array.from(connectedBlockIds)) + ) + ) + + type EdgeBlockRecord = (typeof connectedBlocks)[number] + const blocksById: Record = Object.fromEntries( + connectedBlocks.map((b: EdgeBlockRecord) => [b.id, b]) + ) + + const parentIds = new Set() + for (const block of connectedBlocks) { + const parentId = (block.data as Record | null)?.parentId as string | undefined + if (parentId && !blocksById[parentId]) { + parentIds.add(parentId) + } + } + + if (parentIds.size > 0) { + const parentBlocks = await tx + .select({ + id: workflowBlocks.id, + type: workflowBlocks.type, + locked: workflowBlocks.locked, + data: workflowBlocks.data, + }) + .from(workflowBlocks) + .where( + and( + eq(workflowBlocks.workflowId, workflowId), + inArray(workflowBlocks.id, Array.from(parentIds)) + ) + ) + for (const b of parentBlocks) { + blocksById[b.id] = b + } + } + + const droppedCounts: Record = {} + const countDrop = (reason: string) => { + droppedCounts[reason] = (droppedCounts[reason] ?? 0) + 1 + } + + // Mirrors the client's validateEdges rule order: missing block, protected + // target, annotation-only block, trigger-block target, scope boundary. + // `scopeDropReason`'s detail (which names the specific block ids/scopes) + // is dropped in favor of a stable 'scope boundary' key so droppedCounts + // aggregates across edges instead of gaining one free-text key per edge. + const structurallyValidEdges = candidates.filter((edge) => { + const sourceBlock = blocksById[edge.source] + const targetBlock = blocksById[edge.target] + + if (!sourceBlock || !targetBlock) { + countDrop('missing block') + return false + } + if (isWorkflowBlockProtected(edge.target, blocksById)) { + countDrop('protected target block') + return false + } + if ( + isWorkflowAnnotationOnlyBlockType(sourceBlock.type) || + isWorkflowAnnotationOnlyBlockType(targetBlock.type) + ) { + countDrop('annotation-only block') + return false + } + if (isKnownWorkflowTriggerBlock(targetBlock)) { + countDrop('trigger block target') + return false + } + if (getWorkflowEdgeScopeDropReason(edge, blocksById)) { + countDrop('scope boundary') + return false + } + return true + }) + + if (structurallyValidEdges.length === 0) { + return { safeEdges: [], droppedCounts, droppedDuplicates: 0, droppedCyclic: 0 } + } + + const existingEdgesForCycleCheck: PersistedEdgeRecord[] = await tx + .select({ + sourceBlockId: workflowEdges.sourceBlockId, + targetBlockId: workflowEdges.targetBlockId, + sourceHandle: workflowEdges.sourceHandle, + targetHandle: workflowEdges.targetHandle, + }) + .from(workflowEdges) + .where(eq(workflowEdges.workflowId, workflowId)) + const existingEdgeEndpoints = existingEdgesForCycleCheck.map(toEdgeHandles) + + const uniqueEdges = filterUniqueWorkflowEdges(structurallyValidEdges, existingEdgeEndpoints) + const safeEdges = filterAcyclicEdges(uniqueEdges, existingEdgeEndpoints) + + return { + safeEdges, + droppedCounts, + droppedDuplicates: structurallyValidEdges.length - uniqueEdges.length, + droppedCyclic: uniqueEdges.length - safeEdges.length, + } +} + // Both realtime pools (this socketDb + the shared @sim/db pool) resolve the // realtime-keyed URL when set, falling back to the shared DATABASE_URL. const connectionString = @@ -1323,117 +1471,28 @@ async function handleEdgeOperationTx(tx: any, workflowId: string, operation: str throw new Error('Missing required fields for add edge operation') } - const edgeBlocks = await tx - .select({ - id: workflowBlocks.id, - type: workflowBlocks.type, - locked: workflowBlocks.locked, - triggerMode: workflowBlocks.triggerMode, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, [payload.source, payload.target]) - ) - ) - - type EdgeBlockRecord = (typeof edgeBlocks)[number] - const blocksById: Record = Object.fromEntries( - edgeBlocks.map((b: EdgeBlockRecord) => [b.id, b]) - ) - - const parentIds = new Set() - for (const block of edgeBlocks) { - const parentId = (block.data as Record | null)?.parentId as - | string - | undefined - if (parentId && !blocksById[parentId]) { - parentIds.add(parentId) - } - } - - // Fetch parent blocks if needed - if (parentIds.size > 0) { - const parentBlocks = await tx - .select({ - id: workflowBlocks.id, - type: workflowBlocks.type, - locked: workflowBlocks.locked, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, Array.from(parentIds)) - ) - ) - for (const b of parentBlocks) { - blocksById[b.id] = b - } - } - - if (!blocksById[payload.source] || !blocksById[payload.target]) { - logger.info('Skipping edge add - edge references a missing block') - break - } - - if (isWorkflowBlockProtected(payload.target, blocksById)) { - logger.info(`Skipping edge add - target block is protected`) - break - } - - if ( - isWorkflowAnnotationOnlyBlockType(blocksById[payload.source].type) || - isWorkflowAnnotationOnlyBlockType(blocksById[payload.target].type) - ) { - logger.info('Skipping edge add - edge references an annotation-only block') - break - } - - if (isKnownWorkflowTriggerBlock(blocksById[payload.target])) { - logger.info('Skipping edge add - trigger blocks cannot be edge targets') - break - } - - const scopeDropReason = getWorkflowEdgeScopeDropReason( - { source: payload.source, target: payload.target }, - blocksById + const { safeEdges, droppedCounts, droppedDuplicates } = await filterEdgesForPersist( + tx, + workflowId, + [ + { + id: payload.id, + source: payload.source, + target: payload.target, + sourceHandle: payload.sourceHandle ?? null, + targetHandle: payload.targetHandle ?? null, + }, + ] ) - if (scopeDropReason) { - logger.info(`Skipping edge add - ${scopeDropReason}`) - break - } - const existingEdgesForCycleCheck: PersistedEdgeRecord[] = await tx - .select({ - sourceBlockId: workflowEdges.sourceBlockId, - targetBlockId: workflowEdges.targetBlockId, - sourceHandle: workflowEdges.sourceHandle, - targetHandle: workflowEdges.targetHandle, - }) - .from(workflowEdges) - .where(eq(workflowEdges.workflowId, workflowId)) - - const candidateEdge = { - source: payload.source, - target: payload.target, - sourceHandle: payload.sourceHandle ?? null, - targetHandle: payload.targetHandle ?? null, - } - const existingEdgeEndpoints = existingEdgesForCycleCheck.map(toEdgeHandles) - - if (filterUniqueWorkflowEdges([candidateEdge], existingEdgeEndpoints).length === 0) { - logger.info('Skipping edge add - duplicate edge already exists') - break - } - - if (wouldCreateCycle(existingEdgeEndpoints, candidateEdge.source, candidateEdge.target)) { - logger.info( - `Skipping edge add - would create a cycle: ${payload.source} -> ${payload.target}` - ) + if (safeEdges.length === 0) { + const [structuralDropReason] = Object.keys(droppedCounts) + const reason = structuralDropReason + ? structuralDropReason + : droppedDuplicates > 0 + ? 'duplicate edge already exists' + : `would create a cycle: ${payload.source} -> ${payload.target}` + logger.info(`Skipping edge add - ${reason}`) break } @@ -1658,143 +1717,22 @@ async function handleEdgesOperationTx( logger.info(`Batch adding ${edges.length} edges to workflow ${workflowId}`) - // Get all connected block IDs to check lock status - const connectedBlockIds = new Set() - edges.forEach((e: Record) => { - connectedBlockIds.add(e.source as string) - connectedBlockIds.add(e.target as string) - }) - - // Fetch blocks to check lock status - const connectedBlocks = await tx - .select({ - id: workflowBlocks.id, - type: workflowBlocks.type, - locked: workflowBlocks.locked, - triggerMode: workflowBlocks.triggerMode, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, Array.from(connectedBlockIds)) - ) - ) - - type AddEdgeBlockRecord = (typeof connectedBlocks)[number] - const blocksById: Record = Object.fromEntries( - connectedBlocks.map((b: AddEdgeBlockRecord) => [b.id, b]) - ) - - // Collect parent IDs that need to be fetched - const parentIds = new Set() - for (const block of connectedBlocks) { - const parentId = (block.data as Record | null)?.parentId as - | string - | undefined - if (parentId && !blocksById[parentId]) { - parentIds.add(parentId) - } - } - - // Fetch parent blocks if needed - if (parentIds.size > 0) { - const parentBlocks = await tx - .select({ - id: workflowBlocks.id, - type: workflowBlocks.type, - locked: workflowBlocks.locked, - data: workflowBlocks.data, - }) - .from(workflowBlocks) - .where( - and( - eq(workflowBlocks.workflowId, workflowId), - inArray(workflowBlocks.id, Array.from(parentIds)) - ) - ) - for (const b of parentBlocks) { - blocksById[b.id] = b - } - } - - const droppedCounts: Record = {} - const countDrop = (reason: string) => { - droppedCounts[reason] = (droppedCounts[reason] ?? 0) + 1 - } - - // Mirror the client's validateEdges rule order: missing block, protected - // target, annotation-only block, trigger-block target, scope boundary. - const structurallyValidEdges = (edges as Array>).filter((e) => { - const source = e.source as string - const target = e.target as string - const sourceBlock = blocksById[source] - const targetBlock = blocksById[target] - - if (!sourceBlock || !targetBlock) { - countDrop('missing block') - return false - } - if (isWorkflowBlockProtected(target, blocksById)) { - countDrop('protected target block') - return false - } - if ( - isWorkflowAnnotationOnlyBlockType(sourceBlock.type) || - isWorkflowAnnotationOnlyBlockType(targetBlock.type) - ) { - countDrop('annotation-only block') - return false - } - if (isKnownWorkflowTriggerBlock(targetBlock)) { - countDrop('trigger block target') - return false - } - const scopeDropReason = getWorkflowEdgeScopeDropReason({ source, target }, blocksById) - if (scopeDropReason) { - countDrop(scopeDropReason) - return false - } - return true - }) - - if (structurallyValidEdges.length === 0) { - logger.info('All edges failed validation, skipping add', { droppedCounts }) - return - } - - const existingEdgesForCycleCheck: PersistedEdgeRecord[] = await tx - .select({ - sourceBlockId: workflowEdges.sourceBlockId, - targetBlockId: workflowEdges.targetBlockId, - sourceHandle: workflowEdges.sourceHandle, - targetHandle: workflowEdges.targetHandle, - }) - .from(workflowEdges) - .where(eq(workflowEdges.workflowId, workflowId)) - const existingEdgeEndpoints = existingEdgesForCycleCheck.map(toEdgeHandles) - - const uniqueEdges = filterUniqueWorkflowEdges( - structurallyValidEdges.map((e) => ({ - source: e.source as string, - target: e.target as string, - sourceHandle: (e.sourceHandle as string | null) ?? null, - targetHandle: (e.targetHandle as string | null) ?? null, - original: e, - })), - existingEdgeEndpoints - ) + const candidates: EdgeAddCandidate[] = (edges as Array>).map((e) => ({ + id: e.id as string, + source: e.source as string, + target: e.target as string, + sourceHandle: (e.sourceHandle as string | null) ?? null, + targetHandle: (e.targetHandle as string | null) ?? null, + })) - const safeEdges = filterAcyclicEdges(uniqueEdges, existingEdgeEndpoints).map( - (e) => e.original - ) + const { safeEdges, droppedCounts, droppedDuplicates, droppedCyclic } = + await filterEdgesForPersist(tx, workflowId, candidates) if (safeEdges.length < edges.length) { logger.info(`Dropped ${edges.length - safeEdges.length} invalid edge(s)`, { droppedCounts, - droppedDuplicates: structurallyValidEdges.length - uniqueEdges.length, - droppedCyclic: uniqueEdges.length - safeEdges.length, + droppedDuplicates, + droppedCyclic, }) } @@ -1803,13 +1741,13 @@ async function handleEdgesOperationTx( return } - const edgeValues = safeEdges.map((edge: Record) => ({ + const edgeValues = safeEdges.map((edge) => ({ id: edge.id as string, workflowId, - sourceBlockId: edge.source as string, - targetBlockId: edge.target as string, - sourceHandle: (edge.sourceHandle as string | null) || null, - targetHandle: (edge.targetHandle as string | null) || null, + sourceBlockId: edge.source, + targetBlockId: edge.target, + sourceHandle: edge.sourceHandle || null, + targetHandle: edge.targetHandle || null, })) await tx diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index 46132a31392..ebab233c7dd 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -12,6 +12,7 @@ import { WORKFLOW_OPERATIONS, } from '@sim/realtime-protocol/constants' import { generateId } from '@sim/utils/id' +import { getWorkflowBlockNameConflict } from '@sim/workflow-types/workflow' import { useQueryClient } from '@tanstack/react-query' import { isEqual } from 'es-toolkit' import type { Edge } from 'reactflow' @@ -27,7 +28,6 @@ import { isSyntheticToolSubBlockId } from '@/lib/workflows/tool-input/synthetic- import { useSocket } from '@/app/workspace/providers/socket-provider' import { getBlock } from '@/blocks' import { getSubBlocksDependingOnChange } from '@/blocks/utils' -import { normalizeName, RESERVED_BLOCK_NAMES } from '@/executor/constants' import { invalidateDeploymentQueries } from '@/hooks/queries/deployments' import { useUndoRedo } from '@/hooks/use-undo-redo' import { @@ -1032,27 +1032,26 @@ export function useCollaborativeWorkflow() { } const trimmedName = name.trim() - const normalizedNewName = normalizeName(trimmedName) + const currentBlocks = useWorkflowStore.getState().blocks + const siblingNamesById = Object.fromEntries( + Object.entries(currentBlocks).map(([blockId, b]) => [blockId, b.name]) + ) + const conflict = getWorkflowBlockNameConflict(id, trimmedName, siblingNamesById) - if (!normalizedNewName) { + if (conflict?.reason === 'empty') { logger.error('Cannot rename block to empty name') toast.error('Block name cannot be empty') return { success: false, error: 'Block name cannot be empty' } } - if ((RESERVED_BLOCK_NAMES as readonly string[]).includes(normalizedNewName)) { + if (conflict?.reason === 'reserved') { logger.error(`Cannot rename block to reserved name: "${trimmedName}"`) toast.error(`"${trimmedName}" is a reserved name and cannot be used`) return { success: false, error: `"${trimmedName}" is a reserved name` } } - const currentBlocks = useWorkflowStore.getState().blocks - const conflictingBlock = Object.entries(currentBlocks).find( - ([blockId, block]) => blockId !== id && normalizeName(block.name) === normalizedNewName - ) - - if (conflictingBlock) { - const conflictName = conflictingBlock[1].name + if (conflict?.reason === 'duplicate') { + const conflictName = currentBlocks[conflict.conflictingBlockId as string].name logger.error(`Cannot rename block to "${trimmedName}" - conflicts with "${conflictName}"`) toast.error(`Block name "${trimmedName}" already exists`) return { success: false, error: `Block name "${trimmedName}" already exists` } From a7fdb6ade53d6e6086a53629b5c63583d985c123 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 10 Jul 2026 15:04:36 -0700 Subject: [PATCH 5/6] docs(workflow-edges): document the per-workflow write-serialization invariant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit No behavior change. Address Greptile P1 on filterEdgesForPersist ('concurrent duplicate writes can persist without a per-workflow write guard') by documenting, at the actual mechanism, why the concern doesn't apply here: persistWorkflowOperation's leading 'UPDATE workflow SET updatedAt ... WHERE id = workflowId' already takes a row lock that serializes every operation (including edge adds) for a given workflowId — a second concurrent call blocks on that UPDATE until the first transaction commits or rolls back, so the validate-then-insert sequence in filterEdgesForPersist can never interleave across two writers on the same workflow. Verified empirically, not just by reading: ran two concurrent transactions against a throwaway local Postgres against the exact statement shape used here (UPDATE the parent row, sleep to simulate the read/validate window, insert, commit). The second transaction's UPDATE blocked for the full duration of the first's transaction and only proceeded once the first committed — confirming the row lock, not any application-level guard, already provides the serialization Greptile flagged as missing. Added a comment at the lock site (not a second, redundant advisory lock) so a future change can't silently break this invariant by making the UPDATE conditional/skippable as a perceived no-op optimization. --- apps/realtime/src/database/operations.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/apps/realtime/src/database/operations.ts b/apps/realtime/src/database/operations.ts index 38ff1f3691b..5bf352a61fa 100644 --- a/apps/realtime/src/database/operations.ts +++ b/apps/realtime/src/database/operations.ts @@ -402,6 +402,15 @@ export async function persistWorkflowOperation(workflowId: string, operation: an } await db.transaction(async (tx) => { + // This UPDATE is also this workflow's write-serialization point, not + // just a timestamp bump: it takes a row lock on `workflow` for the + // rest of the transaction, so two concurrent persistWorkflowOperation + // calls for the same workflowId cannot interleave — the second blocks + // here until the first commits or rolls back. Every handler dispatched + // below (including the edge-add validate-then-insert sequence in + // filterEdgesForPersist) relies on this to read a consistent snapshot; + // do not make this UPDATE conditional/skippable as an optimization + // without replacing the serialization it provides. await tx .update(workflow) .set({ updatedAt: new Date(timestamp) }) From f75cd5e4c8c14dc1b56506ec186258a61107edda Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 10 Jul 2026 15:09:09 -0700 Subject: [PATCH 6/6] fix(workflow-edges): validate edges in BATCH_ADD_BLOCKS before persisting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real gap Cursor's PR summary flagged ('BATCH_ADD_BLOCKS edge inserts... not fully covered by the new server pipeline'), confirmed by reading the code: this handler bulk-inserted the edges from a block-paste/duplicate/import payload directly into workflowEdges with zero validation — no missing-block, protected-target, annotation-only, trigger-target, scope-boundary, duplicate, or cycle check. A client sending edges through this operation instead of BATCH_ADD_EDGES could bypass every rule this PR otherwise enforces server-side, exactly the class of gap the PR exists to close. Routes it through the same filterEdgesForPersist used by the other two edge-add handlers. Runs after the block insert in this same handler, so the shared helper's blocksById lookup also sees the blocks this same batch just inserted (a transaction observes its own prior writes). --- apps/realtime/src/database/operations.ts | 65 +++++++++++++++++------- 1 file changed, 46 insertions(+), 19 deletions(-) diff --git a/apps/realtime/src/database/operations.ts b/apps/realtime/src/database/operations.ts index 5bf352a61fa..697ba4099c0 100644 --- a/apps/realtime/src/database/operations.ts +++ b/apps/realtime/src/database/operations.ts @@ -1018,27 +1018,54 @@ async function handleBlocksOperationTx( } if (edges && edges.length > 0) { - const edgeValues = edges.map((edge: Record) => ({ - id: edge.id as string, - workflowId, - sourceBlockId: edge.source as string, - targetBlockId: edge.target as string, - sourceHandle: (edge.sourceHandle as string | null) || null, - targetHandle: (edge.targetHandle as string | null) || null, - })) + // Runs after the block insert above, so filterEdgesForPersist's + // blocksById lookup (a plain `tx.select` from `workflowBlocks`) also + // sees the blocks this same batch just inserted — reads observe a + // transaction's own prior writes. + const candidates: EdgeAddCandidate[] = (edges as Array>).map( + (e) => ({ + id: e.id as string, + source: e.source as string, + target: e.target as string, + sourceHandle: (e.sourceHandle as string | null) ?? null, + targetHandle: (e.targetHandle as string | null) ?? null, + }) + ) - await tx - .insert(workflowEdges) - .values(edgeValues) - .onConflictDoUpdate({ - target: workflowEdges.id, - set: { - sourceBlockId: sql`excluded.source_block_id`, - targetBlockId: sql`excluded.target_block_id`, - sourceHandle: sql`excluded.source_handle`, - targetHandle: sql`excluded.target_handle`, - }, + const { safeEdges, droppedCounts, droppedDuplicates, droppedCyclic } = + await filterEdgesForPersist(tx, workflowId, candidates) + + if (safeEdges.length < edges.length) { + logger.info(`Dropped ${edges.length - safeEdges.length} invalid edge(s)`, { + droppedCounts, + droppedDuplicates, + droppedCyclic, }) + } + + if (safeEdges.length > 0) { + const edgeValues = safeEdges.map((edge) => ({ + id: edge.id as string, + workflowId, + sourceBlockId: edge.source, + targetBlockId: edge.target, + sourceHandle: edge.sourceHandle || null, + targetHandle: edge.targetHandle || null, + })) + + await tx + .insert(workflowEdges) + .values(edgeValues) + .onConflictDoUpdate({ + target: workflowEdges.id, + set: { + sourceBlockId: sql`excluded.source_block_id`, + targetBlockId: sql`excluded.target_block_id`, + sourceHandle: sql`excluded.source_handle`, + targetHandle: sql`excluded.target_handle`, + }, + }) + } } if (loops && Object.keys(loops).length > 0) {