diff --git a/web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx b/web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx index c1af965355c..ab570b3c7d9 100644 --- a/web/app/components/workflow-app/components/__tests__/workflow-main.spec.tsx +++ b/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' }], diff --git a/web/app/components/workflow-app/components/workflow-main.tsx b/web/app/components/workflow-app/components/workflow-main.tsx index 5e1a499bd40..d7799b765cb 100644 --- a/web/app/components/workflow-app/components/workflow-main.tsx +++ b/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 diff --git a/web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx b/web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx index af60c08f0cb..1cf0038673e 100644 --- a/web/app/components/workflow/__tests__/update-dsl-modal.spec.tsx +++ b/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() diff --git a/web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.logs-and-events.spec.ts b/web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.logs-and-events.spec.ts index 7ac65fcc8e3..97ec2ceadf4 100644 --- a/web/app/components/workflow/collaboration/core/__tests__/collaboration-manager.logs-and-events.spec.ts +++ b/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 }] diff --git a/web/app/components/workflow/collaboration/core/collaboration-manager.ts b/web/app/components/workflow/collaboration/core/collaboration-manager.ts index dc8c7f6854d..b9e36b72c11 100644 --- a/web/app/components/workflow/collaboration/core/collaboration-manager.ts +++ b/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 diff --git a/web/app/components/workflow/update-dsl-modal.tsx b/web/app/components/workflow/update-dsl-modal.tsx index 2a28f60c8e2..bc420c793dc 100644 --- a/web/app/components/workflow/update-dsl-modal.tsx +++ b/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)