From e856045970043a8c84865cfec991adcacf63bfef Mon Sep 17 00:00:00 2001 From: Nikhil-Doye Date: Mon, 20 Oct 2025 01:02:40 -0400 Subject: [PATCH] Enhance TestingPanel and ExecutionEngine with improved error handling and real-time node status updates - Refactored TestingPanel to include detailed logging during workflow execution and enhanced error handling for better user feedback. - Updated ExecutionEngine to support real-time node status updates through a callback mechanism, allowing for more dynamic execution monitoring. - Introduced new node statuses ('completed' and 'failed') to provide clearer execution state representation. - Improved BaseNode component to handle invalid data gracefully and added debug logging for better traceability. - Enhanced workflow store to manage and log node status updates effectively, improving overall workflow execution visibility. --- src/components/TestingPanel.tsx | 48 ++++++-- src/components/nodes/BaseNode.tsx | 23 +++- src/services/executionEngine.ts | 180 ++++++++++++++++++++++++++---- src/store/workflowStore.ts | 11 +- src/types/index.ts | 8 +- 5 files changed, 233 insertions(+), 37 deletions(-) diff --git a/src/components/TestingPanel.tsx b/src/components/TestingPanel.tsx index ddea157..025ed93 100644 --- a/src/components/TestingPanel.tsx +++ b/src/components/TestingPanel.tsx @@ -36,18 +36,46 @@ export const TestingPanel: React.FC = () => { if (!currentWorkflow) return; setTestResults(null); - await executeWorkflow(testInput); - // Collect all results - const results = currentWorkflow.nodes.map((node) => ({ - nodeId: node.id, - nodeLabel: node.data.label, - status: executionResults[node.id]?.status || "idle", - data: executionResults[node.id]?.data, - error: executionResults[node.id]?.error, - })); + try { + console.log("Starting test execution:", { + currentWorkflow: currentWorkflow.id, + testInput, + nodeCount: currentWorkflow.nodes.length, + }); - setTestResults(results); + await executeWorkflow(testInput); + + // Wait a bit for the execution to complete and results to be processed + await new Promise((resolve) => setTimeout(resolve, 100)); + + console.log("Execution completed, collecting results:", { + executionResults, + }); + + // Collect all results + const results = currentWorkflow.nodes.map((node) => ({ + nodeId: node.id, + nodeLabel: node.data.label, + status: executionResults[node.id]?.status || "idle", + data: executionResults[node.id]?.data, + error: executionResults[node.id]?.error, + })); + + console.log("Collected results:", results); + setTestResults(results); + } catch (error) { + console.error("Test execution failed:", error); + // Set error results + const errorResults = currentWorkflow.nodes.map((node) => ({ + nodeId: node.id, + nodeLabel: node.data.label, + status: "error", + data: null, + error: error instanceof Error ? error.message : "Unknown error", + })); + setTestResults(errorResults); + } }; const handleLoadSample = (sample: any) => { diff --git a/src/components/nodes/BaseNode.tsx b/src/components/nodes/BaseNode.tsx index f8c0bb1..b3b90e6 100644 --- a/src/components/nodes/BaseNode.tsx +++ b/src/components/nodes/BaseNode.tsx @@ -124,12 +124,24 @@ const statusConfig = { bg: "bg-green-100", pulse: false, }, + completed: { + icon: CheckCircle, + color: "text-green-500", + bg: "bg-green-100", + pulse: false, + }, error: { icon: XCircle, color: "text-red-500", bg: "bg-red-100", pulse: false, }, + failed: { + icon: XCircle, + color: "text-red-500", + bg: "bg-red-100", + pulse: false, + }, }; interface BaseNodeProps extends NodeProps { @@ -141,9 +153,14 @@ export const BaseNode: React.FC = ({ selected, ...props }) => { - const Icon = nodeIcons[data.type]; - const colors = nodeColors[data.type]; - const status = statusConfig[data.status]; + // Debug logging + if (!data.type || !data.status) { + console.warn("BaseNode received invalid data:", data); + } + + const Icon = nodeIcons[data.type] || FileText; + const colors = nodeColors[data.type] || nodeColors.dataInput; + const status = statusConfig[data.status] || statusConfig.idle; const StatusIcon = status.icon; return ( diff --git a/src/services/executionEngine.ts b/src/services/executionEngine.ts index 5bdc6dc..a067bb6 100644 --- a/src/services/executionEngine.ts +++ b/src/services/executionEngine.ts @@ -65,19 +65,33 @@ export class ExecutionEngine { retryDelay: number; backoffMultiplier: number; }; - } = {} + } = {}, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void ): Promise { const executionId = `exec_${Date.now()}_${Math.random() .toString(36) .substr(2, 9)}`; + console.log(`Starting workflow execution:`, { + executionId, + workflowId, + nodeCount: nodes.length, + edgeCount: edges.length, + options, + }); + const plan = this.createExecutionPlan(workflowId, nodes, edges, options); plan.id = executionId; this.activeExecutions.set(executionId, plan); try { - await this.executePlan(plan); + await this.executePlan(plan, onNodeUpdate); plan.status = "completed"; } catch (error) { plan.status = "failed"; @@ -114,6 +128,16 @@ export class ExecutionEngine { maxRetries: options.retryPolicy?.maxRetries || 3, })); + console.log("Creating execution plan:", { + workflowId, + nodeCount: nodes.length, + executionNodes: executionNodes.map((n) => ({ + id: n.nodeId, + type: n.nodeType, + })), + edges: edges.map((e) => ({ from: e.source, to: e.target })), + }); + const executionEdges = edges.map((edge) => ({ from: edge.source, to: edge.target, @@ -286,19 +310,27 @@ export class ExecutionEngine { } // Execute the execution plan - private async executePlan(plan: ExecutionPlan): Promise { + private async executePlan( + plan: ExecutionPlan, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void + ): Promise { plan.status = "running"; plan.startTime = new Date(); switch (plan.executionMode) { case "sequential": - await this.executeSequential(plan); + await this.executeSequential(plan, onNodeUpdate); break; case "parallel": - await this.executeParallel(plan); + await this.executeParallel(plan, onNodeUpdate); break; case "conditional": - await this.executeConditional(plan); + await this.executeConditional(plan, onNodeUpdate); break; default: throw new Error(`Unsupported execution mode: ${plan.executionMode}`); @@ -306,21 +338,48 @@ export class ExecutionEngine { } // Sequential execution - private async executeSequential(plan: ExecutionPlan): Promise { + private async executeSequential( + plan: ExecutionPlan, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void + ): Promise { const executionOrder = this.getExecutionOrder(plan.nodes, plan.edges); - for (const nodeId of executionOrder) { + console.log("Execution order:", executionOrder); + console.log( + "Plan nodes:", + plan.nodes.map((n) => ({ id: n.nodeId, type: n.nodeType })) + ); + console.log("Plan edges:", plan.edges); + + // If no execution order is determined (e.g., no edges), execute all nodes in order + const nodesToExecute = + executionOrder.length > 0 + ? executionOrder + : plan.nodes.map((n) => n.nodeId); + console.log("Nodes to execute:", nodesToExecute); + + for (const nodeId of nodesToExecute) { const context = plan.nodes.find((n) => n.nodeId === nodeId); if (!context) continue; try { - await this.executeNode(context, plan); + await this.executeNode(context, plan, onNodeUpdate); } catch (error) { context.status = "failed"; context.error = error instanceof Error ? error.message : "Unknown error"; plan.errors.set(nodeId, context.error); + // Notify about the error + if (onNodeUpdate) { + onNodeUpdate(nodeId, "failed", undefined, context.error); + } + // Decide whether to continue or stop if (!this.shouldContinueOnError(plan, nodeId)) { throw error; @@ -330,7 +389,15 @@ export class ExecutionEngine { } // Parallel execution - private async executeParallel(plan: ExecutionPlan): Promise { + private async executeParallel( + plan: ExecutionPlan, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void + ): Promise { const parallelGroups = this.createParallelGroups( plan.nodes, plan.edges, @@ -339,7 +406,7 @@ export class ExecutionEngine { // Execute groups in parallel const groupPromises = parallelGroups.map((group) => - this.executeParallelGroup(group, plan) + this.executeParallelGroup(group, plan, onNodeUpdate) ); await Promise.allSettled(groupPromises); @@ -348,13 +415,19 @@ export class ExecutionEngine { // Execute a parallel group private async executeParallelGroup( group: ParallelGroup, - plan: ExecutionPlan + plan: ExecutionPlan, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void ): Promise { const nodePromises = group.nodes.map((nodeId) => { const context = plan.nodes.find((n) => n.nodeId === nodeId); if (!context) return Promise.resolve(); - return this.executeNode(context, plan); + return this.executeNode(context, plan, onNodeUpdate); }); if (group.waitForAll) { @@ -365,7 +438,15 @@ export class ExecutionEngine { } // Conditional execution - private async executeConditional(plan: ExecutionPlan): Promise { + private async executeConditional( + plan: ExecutionPlan, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void + ): Promise { const executionOrder = this.getExecutionOrder(plan.nodes, plan.edges); const visited = new Set(); @@ -376,19 +457,29 @@ export class ExecutionEngine { if (!context) continue; try { - await this.executeNode(context, plan); + await this.executeNode(context, plan, onNodeUpdate); visited.add(nodeId); // Check for conditional branches const branches = this.getConditionalBranches(nodeId, plan); if (branches.length > 0) { - await this.executeConditionalBranches(branches, plan, visited); + await this.executeConditionalBranches( + branches, + plan, + visited, + onNodeUpdate + ); } } catch (error) { context.status = "failed"; context.error = error instanceof Error ? error.message : "Unknown error"; plan.errors.set(nodeId, context.error); + + // Notify about the error + if (onNodeUpdate) { + onNodeUpdate(nodeId, "failed", undefined, context.error); + } } } } @@ -421,7 +512,13 @@ export class ExecutionEngine { private async executeConditionalBranches( branches: ConditionalBranch[], plan: ExecutionPlan, - visited: Set + visited: Set, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void ): Promise { for (const branch of branches) { const shouldExecuteTrue = await this.evaluateCondition( @@ -440,13 +537,18 @@ export class ExecutionEngine { if (!context) continue; try { - await this.executeNode(context, plan); + await this.executeNode(context, plan, onNodeUpdate); visited.add(nodeId); } catch (error) { context.status = "failed"; context.error = error instanceof Error ? error.message : "Unknown error"; plan.errors.set(nodeId, context.error); + + // Notify about the error + if (onNodeUpdate) { + onNodeUpdate(nodeId, "failed", undefined, context.error); + } } } } @@ -518,18 +620,39 @@ export class ExecutionEngine { // Execute a single node private async executeNode( context: ExecutionContext, - plan: ExecutionPlan + plan: ExecutionPlan, + onNodeUpdate?: ( + nodeId: string, + status: string, + data?: any, + error?: string + ) => void ): Promise { context.status = "running"; context.startTime = new Date(); + // Notify that node is starting + if (onNodeUpdate) { + onNodeUpdate(context.nodeId, "running"); + } + try { + console.log(`Executing node ${context.nodeId} (${context.nodeType}):`, { + config: context.config, + inputs: Array.from(context.inputs.entries()), + }); + // Import the appropriate node processor const processor = await this.getNodeProcessor(context.nodeType); // Execute the node const result = await processor(context, plan); + console.log(`Node ${context.nodeId} completed successfully:`, { + result: result, + duration: context.duration, + }); + // Store the result context.outputs.set("output", result); context.status = "completed"; @@ -539,6 +662,11 @@ export class ExecutionEngine { // Store in plan results plan.results.set(context.nodeId, result); + + // Notify that node completed successfully + if (onNodeUpdate) { + onNodeUpdate(context.nodeId, "completed", result); + } } catch (error) { context.status = "failed"; context.error = error instanceof Error ? error.message : "Unknown error"; @@ -546,6 +674,11 @@ export class ExecutionEngine { context.duration = context.endTime.getTime() - context.startTime.getTime(); + // Notify about the error + if (onNodeUpdate) { + onNodeUpdate(context.nodeId, "failed", undefined, context.error); + } + // Handle retries if (context.retryCount < context.maxRetries) { context.retryCount++; @@ -557,7 +690,7 @@ export class ExecutionEngine { ); // Retry execution - return this.executeNode(context, plan); + return this.executeNode(context, plan, onNodeUpdate); } throw error; @@ -566,6 +699,8 @@ export class ExecutionEngine { // Get node processor based on type private async getNodeProcessor(nodeType: string): Promise { + console.log(`Loading processor for node type: ${nodeType}`); + // Import the appropriate processor switch (nodeType) { case "dataInput": @@ -585,6 +720,7 @@ export class ExecutionEngine { case "similaritySearch": return (await import("./processors/similarityProcessor")).default; default: + console.log(`Using default processor for unknown type: ${nodeType}`); return (await import("./processors/defaultProcessor")).default; } } @@ -603,8 +739,8 @@ export class ExecutionEngine { // Build dependency graph edges.forEach((edge) => { - dependencies.get(edge.source)!.add(edge.target); - inDegree.set(edge.target, (inDegree.get(edge.target) || 0) + 1); + dependencies.get(edge.from)!.add(edge.to); + inDegree.set(edge.to, (inDegree.get(edge.to) || 0) + 1); }); // Topological sort diff --git a/src/store/workflowStore.ts b/src/store/workflowStore.ts index 94bf0f7..9829a2e 100644 --- a/src/store/workflowStore.ts +++ b/src/store/workflowStore.ts @@ -479,7 +479,16 @@ export const useWorkflowStore = create((set, get) => ({ currentWorkflow.id, currentWorkflow.nodes, currentWorkflow.edges, - execOptions + execOptions, + (nodeId, status, data, error) => { + // Update node status in real-time + console.log(`Node ${nodeId} status updated:`, { + status, + data, + error, + }); + get().updateNodeStatus(nodeId, status as NodeStatus, data, error); + } ); // Store the execution plan diff --git a/src/types/index.ts b/src/types/index.ts index bc85fe5..a5186f8 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -23,7 +23,13 @@ export type NodeType = | "databaseUpdate" | "databaseDelete"; -export type NodeStatus = "idle" | "running" | "success" | "error"; +export type NodeStatus = + | "idle" + | "running" + | "success" + | "completed" + | "error" + | "failed"; export interface WorkflowNode { id: string;