ai-space/workflow-builder
3
0
Fork 0
mirror of https://github.com/Nikhil-Doye/workflow-builder.git synced 2026-07-22 02:01:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

Enhance TestingPanel and ExecutionEngine with improved error handling and real-time node status updates

Browse Source 
- 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.
This commit is contained in:
Nikhil-Doye
2025-10-20 01:02:40 -04:00
parent 6352974028
commit e856045970
5 changed files with 233 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/TestingPanel.tsx
+38 -10
View File
@@ -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) => {
src/components/nodes/BaseNode.tsx
+20 -3
View File
@@ -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<BaseNodeProps> = ({
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 (
src/services/executionEngine.ts
+158 -22
View File
@@ -65,19 +65,33 @@ export class ExecutionEngine {
retryDelay: number;
backoffMultiplier: number;
};
} = {}
} = {},
onNodeUpdate?: (
nodeId: string,
status: string,
data?: any,
error?: string
) => void
): Promise<ExecutionPlan> {
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<void> {
private async executePlan(
plan: ExecutionPlan,
onNodeUpdate?: (
nodeId: string,
status: string,
data?: any,
error?: string
) => void
): Promise<void> {
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<void> {
private async executeSequential(
plan: ExecutionPlan,
onNodeUpdate?: (
nodeId: string,
status: string,
data?: any,
error?: string
) => void
): Promise<void> {
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<void> {
private async executeParallel(
plan: ExecutionPlan,
onNodeUpdate?: (
nodeId: string,
status: string,
data?: any,
error?: string
) => void
): Promise<void> {
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<void> {
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<void> {
private async executeConditional(
plan: ExecutionPlan,
onNodeUpdate?: (
nodeId: string,
status: string,
data?: any,
error?: string
) => void
): Promise<void> {
const executionOrder = this.getExecutionOrder(plan.nodes, plan.edges);
const visited = new Set<string>();
@@ -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<string>
visited: Set<string>,
onNodeUpdate?: (
nodeId: string,
status: string,
data?: any,
error?: string
) => void
): Promise<void> {
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<void> {
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<any> {
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
src/store/workflowStore.ts
+10 -1
View File
@@ -479,7 +479,16 @@ export const useWorkflowStore = create<WorkflowStore>((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
src/types/index.ts
+7 -1
View File
@@ -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;