fix(workflow): sync imported graph with collaboration state (#42817)

web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx

@@ -12,6 +12,7 @@ const mockSetFeatures = vi.fn()
 const mockSetConversationVariables = vi.fn()
 const mockSetEnvironmentVariables = vi.fn()
 const mockSetEnvSecrets = vi.fn()
+const mockSetSyncWorkflowDraftHash = vi.fn()
 const mockHandleUpdateWorkflowCanvas = vi.hoisted(() => vi.fn())
 const mockFetchWorkflowDraft = vi.hoisted(() => vi.fn())
 const mockOnVarsAndFeaturesUpdate = vi.hoisted(() => vi.fn())
@@ -123,6 +124,7 @@ vi.mock('@/app/components/workflow/store', () => ({
       setConversationVariables: mockSetConversationVariables,
       setEnvironmentVariables: mockSetEnvironmentVariables,
       setEnvSecrets: mockSetEnvSecrets,
+      setSyncWorkflowDraftHash: mockSetSyncWorkflowDraftHash,
     }),
   }),
 }))
@@ -601,6 +603,7 @@ describe('WorkflowMain', () => {
   it('subscribes collaboration listeners and handles sync/workflow update callbacks', async () => {
     collaborationRuntime.isEnabled = true
     mockFetchWorkflowDraft.mockResolvedValue({
+      hash: 'imported-hash',
       features: {
         file_upload: { enabled: true },
         opening_statement: 'hello',
@@ -640,6 +643,7 @@ describe('WorkflowMain', () => {
     await waitFor(() => {
       expect(mockFetchWorkflowDraft).toHaveBeenCalledWith('/apps/app-1/workflows/draft')
       expect(mockSetFeatures).toHaveBeenCalled()
+      expect(mockSetSyncWorkflowDraftHash).toHaveBeenCalledWith('imported-hash')
       expect(mockHandleUpdateWorkflowCanvas).toHaveBeenCalledWith({
         nodes: [{ id: 'n-1' }],
         edges: [{ id: 'e-1' }],

web/app/components/workflow-app/components/workflow-main.tsx

@@ -319,6 +319,8 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => {
       try {
         const response = await fetchWorkflowDraft(`/apps/${appId}/workflows/draft`)
 
+        if (response.hash) workflowStore.getState().setSyncWorkflowDraftHash(response.hash)
+
         // Handle features, variables etc.
         handleWorkflowDataUpdate(response)
 
@@ -337,6 +339,7 @@ const WorkflowMain = ({ nodes, edges, viewport }: WorkflowMainProps) => {
     handleWorkflowDataUpdate,
     handleUpdateWorkflowCanvas,
     isCollaborationEnabled,
+    workflowStore,
   ])
 
   // The server directs this request to the selected saver. Do not gate it on the

web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx

@@ -11,6 +11,8 @@ import UpdateDSLModal from '../update-dsl-modal'
 
 const mockEmit = vi.fn()
 const mockEmitWorkflowUpdate = vi.hoisted(() => vi.fn())
+const mockIsCollaborationConnected = vi.hoisted(() => vi.fn(() => true))
+const mockReplaceGraphFromCommittedDraft = vi.hoisted(() => vi.fn((..._args: unknown[]) => true))
 
 vi.mock('@/app/notifications', () => ({
   toast: {
@@ -56,6 +58,8 @@ vi.mock('@/service/workflow', () => ({
 vi.mock('../collaboration/core/collaboration-manager', () => ({
   collaborationManager: {
     emitWorkflowUpdate: mockEmitWorkflowUpdate,
+    isConnected: mockIsCollaborationConnected,
+    replaceGraphFromCommittedDraft: mockReplaceGraphFromCommittedDraft,
   },
 }))
 
@@ -126,6 +130,8 @@ describe('UpdateDSLModal', () => {
       status: DSLImportStatus.COMPLETED,
       app_id: 'app-1',
     })
+    mockIsCollaborationConnected.mockReturnValue(true)
+    mockReplaceGraphFromCommittedDraft.mockReturnValue(true)
     mockHandleCheckPluginDependencies.mockResolvedValue(undefined)
   })
 
@@ -211,6 +217,76 @@ describe('UpdateDSLModal', () => {
     expect(defaultProps.onCancel).toHaveBeenCalledTimes(1)
   })
 
+  it('commits the imported graph to collaboration before updating the canvas or other clients', async () => {
+    const importedNode = {
+      id: 'imported-start',
+      type: 'custom',
+      position: { x: 0, y: 0 },
+      data: { type: 'start', title: 'Start', desc: '' },
+    }
+    mockFetchWorkflowDraft.mockResolvedValueOnce({
+      graph: { nodes: [importedNode], edges: [], viewport: { x: 0, y: 0, zoom: 1 } },
+      features: {},
+      hash: 'imported-hash',
+      conversation_variables: [],
+      environment_variables: [],
+    })
+    renderModal()
+
+    fireEvent.change(screen.getByTestId('dsl-file-input'), {
+      target: { files: [new File(['workflow'], 'workflow.yml', { type: 'text/yaml' })] },
+    })
+    fireEvent.click(screen.getByRole('button', { name: 'workflow.common.overwriteAndImport' }))
+
+    await waitFor(() => expect(defaultProps.onCancel).toHaveBeenCalledTimes(1))
+    const [appId, nodes, edges] = mockReplaceGraphFromCommittedDraft.mock.calls[0]!
+    expect(appId).toBe('app-1')
+    expect(nodes).toEqual([expect.objectContaining({ id: 'imported-start' })])
+    expect(edges).toEqual([])
+    expect(mockEmit).toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: 'WORKFLOW_DATA_UPDATE',
+        payload: expect.objectContaining({ nodes, edges, hash: 'imported-hash' }),
+      }),
+    )
+    expect(mockReplaceGraphFromCommittedDraft.mock.invocationCallOrder[0]).toBeLessThan(
+      mockEmit.mock.invocationCallOrder[0]!,
+    )
+    expect(mockEmit.mock.invocationCallOrder[0]).toBeLessThan(
+      mockEmitWorkflowUpdate.mock.invocationCallOrder[0]!,
+    )
+  })
+
+  it('updates the imported canvas when collaboration is disconnected', async () => {
+    mockIsCollaborationConnected.mockReturnValue(false)
+    renderModal()
+
+    fireEvent.change(screen.getByTestId('dsl-file-input'), {
+      target: { files: [new File(['workflow'], 'workflow.yml', { type: 'text/yaml' })] },
+    })
+    fireEvent.click(screen.getByRole('button', { name: 'workflow.common.overwriteAndImport' }))
+
+    await waitFor(() => expect(defaultProps.onCancel).toHaveBeenCalledTimes(1))
+    expect(mockReplaceGraphFromCommittedDraft).not.toHaveBeenCalled()
+    expect(mockEmit).toHaveBeenCalledWith(expect.objectContaining({ type: 'WORKFLOW_DATA_UPDATE' }))
+  })
+
+  it('reloads instead of showing a graph that collaboration cannot accept', async () => {
+    const reload = vi.spyOn(window.location, 'reload').mockImplementation(() => {})
+    mockReplaceGraphFromCommittedDraft.mockReturnValue(false)
+    renderModal()
+
+    fireEvent.change(screen.getByTestId('dsl-file-input'), {
+      target: { files: [new File(['workflow'], 'workflow.yml', { type: 'text/yaml' })] },
+    })
+    fireEvent.click(screen.getByRole('button', { name: 'workflow.common.overwriteAndImport' }))
+
+    await waitFor(() => expect(reload).toHaveBeenCalledTimes(1))
+    expect(mockEmit).not.toHaveBeenCalled()
+    expect(defaultProps.onCancel).not.toHaveBeenCalled()
+    reload.mockRestore()
+  })
+
   it('should show Agent package warnings returned by a completed import', async () => {
     mockImportDSL.mockResolvedValue({
       id: 'import-with-agent-warnings',
@@ -409,6 +485,7 @@ describe('UpdateDSLModal', () => {
     expect(toast.success).toHaveBeenCalledWith('workflow.common.importSuccess', undefined)
     expect(defaultProps.onCancel).not.toHaveBeenCalled()
     expect(mockEmitWorkflowUpdate).toHaveBeenCalledWith('app-1')
+    expect(mockReplaceGraphFromCommittedDraft).not.toHaveBeenCalled()
     expect(mockHandleCheckPluginDependencies).not.toHaveBeenCalled()
     expect(reload).toHaveBeenCalledTimes(1)
     reload.mockRestore()

web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.logs-and-events.spec.ts

@@ -177,6 +177,80 @@ describe('CollaborationManager logs and event helpers', () => {
     expect(manager.getEdges().map((currentEdge) => currentEdge.id)).toEqual(['e-start-note'])
   })
 
+  it('replaces a stale CRDT graph with the committed draft before a refresh', () => {
+    const { manager, internals } = setupManagerWithDoc()
+    internals.currentAppId = 'app-1'
+    vi.spyOn(manager, 'canApplyLocalGraphMutation').mockReturnValue(true)
+    const oldNode = createNode('old-node')
+    const oldTarget = createNode('old-target')
+    const oldEdge = createEdge('old-edge', oldNode.id, oldTarget.id)
+    const importedNode = createNode('imported-node')
+    const importedTarget = createNode('imported-target')
+    const importedEdge = createEdge('imported-edge', importedNode.id, importedTarget.id)
+    manager.setNodes([], [oldNode, oldTarget])
+    manager.setEdges([], [oldEdge])
+    const clearUndo = vi.fn()
+    internals.undoManager = {
+      canUndo: vi.fn(),
+      canRedo: vi.fn(),
+      undo: vi.fn(),
+      redo: vi.fn(),
+      clear: clearUndo,
+    }
+
+    expect(
+      manager.replaceGraphFromCommittedDraft(
+        'another-app',
+        [importedNode, importedTarget],
+        [importedEdge],
+      ),
+    ).toBe(false)
+    expect(
+      manager
+        .getNodes()
+        .map((node) => node.id)
+        .sort(),
+    ).toEqual(['old-node', 'old-target'])
+    expect(manager.getEdges().map((edge) => edge.id)).toEqual(['old-edge'])
+
+    expect(
+      manager.replaceGraphFromCommittedDraft(
+        'app-1',
+        [importedNode, importedTarget],
+        [importedEdge],
+      ),
+    ).toBe(true)
+    const refreshedGraphs: Array<{ nodes: Node[]; edges: Edge[] }> = []
+    manager.onGraphImport((graph) => refreshedGraphs.push(graph))
+    manager.refreshGraphSynchronously()
+
+    expect(refreshedGraphs[0]?.nodes.map((node) => node.id).sort()).toEqual([
+      'imported-node',
+      'imported-target',
+    ])
+    expect(refreshedGraphs[0]?.edges.map((edge) => edge.id)).toEqual(['imported-edge'])
+    expect(clearUndo).toHaveBeenCalledTimes(1)
+  })
+
+  it('does not seed the old canvas when importing into an empty CRDT graph', () => {
+    const { manager, internals } = setupManagerWithDoc()
+    internals.currentAppId = 'app-1'
+    vi.spyOn(manager, 'canApplyLocalGraphMutation').mockReturnValue(true)
+    internals.reactFlowStore = {
+      getState: () => ({
+        getNodes: () => [createNode('old-canvas-node')],
+        setNodes: vi.fn(),
+        getEdges: () => [],
+        setEdges: vi.fn(),
+      }),
+    }
+
+    expect(manager.replaceGraphFromCommittedDraft('app-1', [createNode('imported-node')], [])).toBe(
+      true,
+    )
+    expect(manager.getNodes().map((node) => node.id)).toEqual(['imported-node'])
+  })
+
   it('clearGraphImportLog clears logs and pending import snapshot', () => {
     const { manager, internals } = setupManagerWithDoc()
     internals.graphImportLogs = [{ id: 1 }]

web/app/components/workflow/collaboration/core/collaboration-manager.ts

@@ -899,6 +899,18 @@ export class CollaborationManager {
     return true
   }
 
+  replaceGraphFromCommittedDraft(appId: string, nodes: Node[], edges: Edge[]): boolean {
+    if (this.currentAppId !== appId || !this.doc || !this.canApplyLocalGraphMutation()) return false
+
+    // A server-side import or restore replaces the whole draft. Its graph must also replace
+    // the CRDT snapshot before a visibility refresh or page close can persist the old graph.
+    this.syncNodes(this.getNodes(), nodes)
+    this.syncEdges(this.getEdges(), edges)
+    this.doc.commit()
+    this.clearUndoStack()
+    return true
+  }
+
   retryGraphReload(request: GraphReloadRequest): void {
     if (!this.isGraphReloadCurrent(request)) return
 

web/app/components/workflow/update-dsl-modal.tsx

@@ -47,11 +47,19 @@ const UpdateDSLModal = ({ onCancel, onBackup, onImport }: UpdateDSLModalProps) =
         await fetchWorkflowDraft(`/apps/${app_id}/workflows/draft`)
 
       const { nodes, edges, viewport } = graph
+      const importedNodes = initialNodes(nodes, edges)
+      const importedEdges = initialEdges(edges, nodes)
+      if (
+        collaborationManager.isConnected() &&
+        !collaborationManager.replaceGraphFromCommittedDraft(app_id, importedNodes, importedEdges)
+      )
+        throw new Error('Collaborative graph is not ready to apply the imported draft.')
+
       eventEmitter?.emit({
         type: WORKFLOW_DATA_UPDATE,
         payload: {
-          nodes: initialNodes(nodes, edges),
-          edges: initialEdges(edges, nodes),
+          nodes: importedNodes,
+          edges: importedEdges,
           viewport,
           features: normalizeWorkflowFeatures(features),
           hash,
@@ -81,10 +89,10 @@ const UpdateDSLModal = ({ onCancel, onBackup, onImport }: UpdateDSLModalProps) =
             }
           : undefined,
       )
-      collaborationManager.emitWorkflowUpdate(appId)
       try {
         await handleWorkflowUpdate(appId)
       } catch (error) {
+        collaborationManager.emitWorkflowUpdate(appId)
         toast.error(
           t(($) => $.error, { ns: 'common' }),
           {
@@ -95,6 +103,7 @@ const UpdateDSLModal = ({ onCancel, onBackup, onImport }: UpdateDSLModalProps) =
         window.location.reload()
         return
       }
+      collaborationManager.emitWorkflowUpdate(appId)
       onImport?.()
       // Dependency checks own their feedback after the import has already succeeded.
       await handleCheckPluginDependencies(appId)