From ce9f7932c3d5db1cd866cb34c8dac281220c0398 Mon Sep 17 00:00:00 2001 From: Nikhil-Doye Date: Mon, 27 Oct 2025 14:30:04 -0400 Subject: [PATCH] Add execution logging and event handling components - Introduced ExecutionLogger component to display real-time execution logs, including workflow and node events. - Added NodeExecutionIndicator component to visualize the status and progress of individual nodes during execution. - Enhanced WorkflowEditor to integrate execution logging, allowing users to toggle the logger and view execution metrics. - Implemented execution event bus to manage and emit events related to workflow execution, including start, progress, completion, and errors. - Updated ExecutionEngine to emit relevant events during execution phases for improved tracking and user feedback. --- src/components/ExecutionLogger.tsx | 295 ++++++++++++++++++++++ src/components/NodeExecutionIndicator.tsx | 142 +++++++++++ src/components/WorkflowEditor.tsx | 119 ++++++++- src/services/executionEngine.ts | 159 ++++++++++-- src/services/executionEventBus.ts | 236 +++++++++++++++++ 5 files changed, 926 insertions(+), 25 deletions(-) create mode 100644 src/components/ExecutionLogger.tsx create mode 100644 src/components/NodeExecutionIndicator.tsx create mode 100644 src/services/executionEventBus.ts diff --git a/src/components/ExecutionLogger.tsx b/src/components/ExecutionLogger.tsx new file mode 100644 index 0000000..6f9788d --- /dev/null +++ b/src/components/ExecutionLogger.tsx @@ -0,0 +1,295 @@ +import React, { useEffect, useRef, useState } from "react"; +import { + executionEventBus, + ExecutionEvent, + NodeCompleteEvent, + NodeStartEvent, +} from "../services/executionEventBus"; +import { + ChevronDown, + ChevronUp, + X, + AlertCircle, + CheckCircle, + Clock, + Zap, + Download, + Trash2, +} from "lucide-react"; + +interface LogEntry { + timestamp: Date; + level: "info" | "success" | "warning" | "error"; + message: string; + nodeId?: string; + nodeLabel?: string; + event?: ExecutionEvent; +} + +interface ExecutionLoggerProps { + executionId?: string; + isOpen: boolean; + onClose: () => void; +} + +export const ExecutionLogger: React.FC = ({ + executionId, + isOpen, + onClose, +}) => { + const [logs, setLogs] = useState([]); + const [isCollapsed, setIsCollapsed] = useState(false); + const [autoScroll, setAutoScroll] = useState(true); + const logsContainerRef = useRef(null); + const unsubscribeRef = useRef<(() => void) | null>(null); + + useEffect(() => { + if (!isOpen) return; + + const handleEvent = (event: ExecutionEvent) => { + const newLog: LogEntry = { + timestamp: event.timestamp, + event, + level: "info", + message: "", + }; + + switch (event.type) { + case "execution:start": + newLog.message = `Workflow execution started (${event.executionMode} mode, ${event.nodeCount} nodes)`; + newLog.level = "info"; + break; + case "node:start": + newLog.message = `Node started`; + newLog.nodeId = event.nodeId; + newLog.nodeLabel = event.nodeLabel; + newLog.level = "info"; + break; + case "node:progress": + newLog.message = `${event.message} (${event.progress}%)`; + newLog.nodeId = event.nodeId; + newLog.level = "info"; + break; + case "node:complete": + const completeEvent = event as NodeCompleteEvent; + if (completeEvent.status === "success") { + newLog.message = `Node completed successfully (${completeEvent.duration}ms)`; + newLog.level = "success"; + } else if (completeEvent.status === "failed") { + newLog.message = `Node failed: ${completeEvent.error}`; + newLog.level = "error"; + } else { + newLog.message = `Node ${completeEvent.status}`; + newLog.level = "warning"; + } + newLog.nodeId = event.nodeId; + break; + case "execution:complete": + newLog.message = `Workflow execution ${event.status} (${event.totalDuration}ms, ${event.completedNodes} completed, ${event.failedNodes} failed)`; + newLog.level = event.status === "success" ? "success" : "error"; + break; + case "execution:error": + newLog.message = `Error during ${event.stage}: ${event.error}`; + newLog.nodeId = event.nodeId; + newLog.level = "error"; + break; + } + + setLogs((prev) => [...prev, newLog]); + }; + + if (executionId) { + unsubscribeRef.current = executionEventBus.subscribeToExecution( + executionId, + handleEvent + ); + } else { + // Subscribe to all events if no specific execution ID + unsubscribeRef.current = () => {}; + const unsubscribers = [ + executionEventBus.subscribe("execution:start", handleEvent), + executionEventBus.subscribe("node:start", handleEvent), + executionEventBus.subscribe("node:progress", handleEvent), + executionEventBus.subscribe("node:complete", handleEvent), + executionEventBus.subscribe("execution:complete", handleEvent), + executionEventBus.subscribe("execution:error", handleEvent), + ]; + unsubscribeRef.current = () => { + unsubscribers.forEach((unsub) => unsub()); + }; + } + + return () => { + if (unsubscribeRef.current) { + unsubscribeRef.current(); + } + }; + }, [isOpen, executionId]); + + // Auto-scroll to bottom when new logs arrive + useEffect(() => { + if (autoScroll && logsContainerRef.current) { + logsContainerRef.current.scrollTop = + logsContainerRef.current.scrollHeight; + } + }, [logs, autoScroll]); + + const downloadLogs = () => { + const logsText = logs + .map( + (log) => + `[${log.timestamp.toISOString()}] [${log.level.toUpperCase()}] ${ + log.nodeLabel ? `(${log.nodeLabel}) ` : "" + }${log.message}` + ) + .join("\n"); + + const blob = new Blob([logsText], { type: "text/plain" }); + const url = URL.createObjectURL(blob); + const link = document.createElement("a"); + link.href = url; + link.download = `execution-logs-${Date.now()}.txt`; + document.body.appendChild(link); + link.click(); + document.body.removeChild(link); + URL.revokeObjectURL(url); + }; + + const clearLogs = () => { + setLogs([]); + }; + + const getLogIcon = (level: string) => { + switch (level) { + case "success": + return ; + case "error": + return ; + case "warning": + return ; + default: + return ; + } + }; + + const getLogBgColor = (level: string) => { + switch (level) { + case "success": + return "bg-green-50 border-l-4 border-green-500"; + case "error": + return "bg-red-50 border-l-4 border-red-500"; + case "warning": + return "bg-yellow-50 border-l-4 border-yellow-500"; + default: + return "bg-blue-50 border-l-4 border-blue-500"; + } + }; + + if (!isOpen) return null; + + return ( +
+ {/* Header */} +
+
+ +

Execution Logs

+ + {logs.length} + +
+ +
+ + + + + +
+
+ + {/* Logs Container */} + {!isCollapsed && ( +
+ {logs.length === 0 ? ( +
+

No logs yet. Execute a workflow to see logs here.

+
+ ) : ( +
+ {logs.map((log, index) => ( +
+
+ {getLogIcon(log.level)} +
+
+ + {log.timestamp.toLocaleTimeString()} + + {log.nodeLabel && ( + + {log.nodeLabel} + + )} +
+

+ {log.message} +

+
+
+
+ ))} +
+ )} +
+ )} +
+ ); +}; diff --git a/src/components/NodeExecutionIndicator.tsx b/src/components/NodeExecutionIndicator.tsx new file mode 100644 index 0000000..6edec32 --- /dev/null +++ b/src/components/NodeExecutionIndicator.tsx @@ -0,0 +1,142 @@ +import React from "react"; +import { + CheckCircle, + AlertCircle, + Clock, + Zap, + ChevronDown, + ChevronUp, +} from "lucide-react"; + +interface ExecutionData { + status: "pending" | "running" | "completed" | "failed" | "skipped"; + progress?: number; + duration?: number; + output?: any; + error?: string; + retryCount?: number; +} + +interface NodeExecutionIndicatorProps { + nodeId: string; + executionData?: ExecutionData; + isExpanded: boolean; + onToggleExpand: () => void; +} + +export const NodeExecutionIndicator: React.FC = ({ + nodeId, + executionData, + isExpanded, + onToggleExpand, +}) => { + if (!executionData) return null; + + const { status, progress, duration, output, error, retryCount } = + executionData; + + const getStatusColor = () => { + switch (status) { + case "running": + return "border-blue-500 bg-blue-50"; + case "completed": + return "border-green-500 bg-green-50"; + case "failed": + return "border-red-500 bg-red-50"; + case "skipped": + return "border-yellow-500 bg-yellow-50"; + default: + return "border-gray-300 bg-gray-50"; + } + }; + + const getStatusIcon = () => { + switch (status) { + case "running": + return ; + case "completed": + return ; + case "failed": + return ; + case "skipped": + return ; + default: + return null; + } + }; + + return ( +
+ {/* Status Overlay Badge */} +
+ {getStatusIcon()} + + {status} + + {retryCount ? ( + + Retry {retryCount} + + ) : null} +
+ + {/* Progress Bar */} + {status === "running" && progress !== undefined && ( +
+
+
+ )} + + {/* Execution Time */} + {duration && status === "completed" && ( +
+ {duration}ms +
+ )} + + {/* Error Badge */} + {status === "failed" && error && ( +
+ {error} +
+ )} + + {/* Output Preview */} + {status === "completed" && output && isExpanded && ( +
+
+ Output + +
+ 
+            {typeof output === "string"
+              ? output
+              : JSON.stringify(output, null, 2)}
+
+
+ )} + + {/* Output Toggle Button */} + {status === "completed" && output && !isExpanded && ( + + )} +
+ ); +}; diff --git a/src/components/WorkflowEditor.tsx b/src/components/WorkflowEditor.tsx index 6931b4d..d0d2a37 100644 --- a/src/components/WorkflowEditor.tsx +++ b/src/components/WorkflowEditor.tsx @@ -1,4 +1,10 @@ -import React, { useCallback, useRef, useState, useMemo } from "react"; +import React, { + useCallback, + useRef, + useState, + useMemo, + useEffect, +} from "react"; import ReactFlow, { Node, Edge, @@ -22,6 +28,7 @@ import { NodeLibrary } from "./NodeLibrary"; import { ConnectionSuggestions } from "./ConnectionSuggestions"; import { ConnectionValidation } from "./ConnectionValidation"; import { InteractiveTutorial } from "./InteractiveTutorial"; +import { ExecutionLogger } from "./ExecutionLogger"; import { WebScrapingNode, LLMNode, @@ -40,7 +47,19 @@ import { GmailNode, } from "./nodes"; import { NodeData } from "../types"; -import { Grid, Trash2, Sparkles, X, Lightbulb, Play } from "lucide-react"; +import { + Grid, + Trash2, + Sparkles, + X, + Lightbulb, + Play, + Activity, +} from "lucide-react"; +import { + executionEventBus, + ExecutionEvent, +} from "../services/executionEventBus"; const nodeTypes: NodeTypes = { webScraping: WebScrapingNode, @@ -76,7 +95,10 @@ export const WorkflowEditor: React.FC = ({ onClose }) => { panelStates, togglePanel, updateNodePosition, + isExecuting, + currentExecution, } = useWorkflowStore(); + const nodes = useMemo( () => currentWorkflow?.nodes || [], [currentWorkflow?.nodes] @@ -86,13 +108,61 @@ export const WorkflowEditor: React.FC = ({ onClose }) => { const [showCopilot, setShowCopilot] = useState(false); const [showSuggestions, setShowSuggestions] = useState(false); const [showTutorial, setShowTutorial] = useState(false); + const [showExecutionLogger, setShowExecutionLogger] = useState(false); const [connectionSource, setConnectionSource] = useState | null>(null); const [connectionTarget, setConnectionTarget] = useState | null>(null); + const [activeExecutionId, setActiveExecutionId] = useState( + null + ); + const [executionMetrics, setExecutionMetrics] = useState<{ + completedNodes: number; + failedNodes: number; + totalDuration: number; + }>({ completedNodes: 0, failedNodes: 0, totalDuration: 0 }); const reactFlowWrapper = useRef(null); const [reactFlowInstance, setReactFlowInstance] = useState(null); + const executionUnsubscribeRef = useRef<(() => void) | null>(null); + + // Subscribe to execution events for real-time feedback + useEffect(() => { + if (isExecuting && currentExecution?.id) { + setActiveExecutionId(currentExecution.id); + setShowExecutionLogger(true); + + // Subscribe to execution events + executionUnsubscribeRef.current = executionEventBus.subscribeToExecution( + currentExecution.id, + (event: ExecutionEvent) => { + // Update execution metrics + if (event.type === "execution:complete") { + setExecutionMetrics({ + completedNodes: event.completedNodes, + failedNodes: event.failedNodes, + totalDuration: event.totalDuration, + }); + } else if (event.type === "node:complete") { + setExecutionMetrics((prev) => ({ + ...prev, + completedNodes: + prev.completedNodes + (event.status === "success" ? 1 : 0), + failedNodes: + prev.failedNodes + (event.status === "failed" ? 1 : 0), + })); + } + } + ); + } + + return () => { + if (executionUnsubscribeRef.current) { + executionUnsubscribeRef.current(); + executionUnsubscribeRef.current = null; + } + }; + }, [isExecuting, currentExecution?.id]); const onConnect = useCallback( (params: Connection) => { @@ -314,6 +384,25 @@ export const WorkflowEditor: React.FC = ({ onClose }) => { Tutorial + {/* Execution Logger Toggle */} + + {selectedNodeId && (
+ {/* Execution Metrics */} + {isExecuting && ( +
+ + + {executionMetrics.completedNodes}/{nodes.length} completed + + {executionMetrics.failedNodes > 0 && ( + + ({executionMetrics.failedNodes} failed) + + )} +
+ )} +
{nodes.length} nodes @@ -370,7 +474,9 @@ export const WorkflowEditor: React.FC = ({ onClose }) => { maxZoom: 2, }} attributionPosition="bottom-left" - className="bg-gradient-to-br from-gray-50 to-blue-50/30" + className={`bg-gradient-to-br from-gray-50 to-blue-50/30 ${ + isExecuting ? "opacity-75" : "" + }`} minZoom={0.1} maxZoom={2} > @@ -474,6 +580,13 @@ export const WorkflowEditor: React.FC = ({ onClose }) => { workflowNodes={nodes} workflowEdges={edges} /> + + {/* Execution Logger */} + setShowExecutionLogger(false)} + />
); }; diff --git a/src/services/executionEngine.ts b/src/services/executionEngine.ts index dc6d3e5..179614b 100644 --- a/src/services/executionEngine.ts +++ b/src/services/executionEngine.ts @@ -1,5 +1,14 @@ // Import variable substitution utilities import { substituteVariables, NodeOutput } from "../utils/variableSubstitution"; +import { + executionEventBus, + ExecutionStartEvent, + NodeStartEvent, + NodeProgressEvent, + NodeCompleteEvent, + ExecutionCompleteEvent, + ExecutionErrorEvent, +} from "./executionEventBus"; export interface ExecutionContext { nodeId: string; @@ -88,6 +97,20 @@ export class ExecutionEngine { options, }); + // Emit execution start event + const executionStartEvent: ExecutionStartEvent = { + type: "execution:start", + executionId, + workflowId, + timestamp: new Date(), + nodeCount: nodes.length, + executionMode: (options.mode || "sequential") as + | "sequential" + | "parallel" + | "conditional", + }; + executionEventBus.emit(executionStartEvent); + // Validate workflow before execution try { this.validateWorkflowStructure(nodes, edges); @@ -96,6 +119,16 @@ export class ExecutionEngine { error instanceof Error ? error.message : "Unknown validation error"; console.error("Workflow validation failed:", errorMessage); + // Emit validation error event + const validationErrorEvent: ExecutionErrorEvent = { + type: "execution:error", + executionId, + stage: "validation", + error: errorMessage, + timestamp: new Date(), + }; + executionEventBus.emit(validationErrorEvent); + // Create a failed execution plan const failedPlan: ExecutionPlan = { id: executionId, @@ -131,14 +164,55 @@ export class ExecutionEngine { this.activeExecutions.set(executionId, plan); try { - await this.executePlan(plan, onNodeUpdate); + await this.executePlan(plan, onNodeUpdate, executionId); plan.status = "completed"; + + // Emit execution complete event + const executionCompleteEvent: ExecutionCompleteEvent = { + type: "execution:complete", + executionId, + workflowId, + status: "success", + totalDuration: plan.totalDuration || 0, + completedNodes: plan.nodes.filter((n) => n.status === "completed") + .length, + failedNodes: plan.nodes.filter((n) => n.status === "failed").length, + results: plan.results, + errors: plan.errors, + timestamp: new Date(), + }; + executionEventBus.emit(executionCompleteEvent); } catch (error) { plan.status = "failed"; - plan.errors.set( - "root", - error instanceof Error ? error.message : "Unknown error" - ); + const errorMessage = + error instanceof Error ? error.message : "Unknown error"; + plan.errors.set("root", errorMessage); + + // Emit execution error event + const executionErrorEvent: ExecutionErrorEvent = { + type: "execution:error", + executionId, + stage: "execution", + error: errorMessage, + timestamp: new Date(), + }; + executionEventBus.emit(executionErrorEvent); + + // Emit execution complete event with failure status + const executionCompleteEvent: ExecutionCompleteEvent = { + type: "execution:complete", + executionId, + workflowId, + status: "failed", + totalDuration: plan.totalDuration || 0, + completedNodes: plan.nodes.filter((n) => n.status === "completed") + .length, + failedNodes: plan.nodes.filter((n) => n.status === "failed").length, + results: plan.results, + errors: plan.errors, + timestamp: new Date(), + }; + executionEventBus.emit(executionCompleteEvent); } finally { plan.endTime = new Date(); plan.totalDuration = @@ -367,20 +441,21 @@ export class ExecutionEngine { status: string, data?: any, error?: string - ) => void + ) => void, + executionId: string ): Promise { plan.status = "running"; plan.startTime = new Date(); switch (plan.executionMode) { case "sequential": - await this.executeSequential(plan, onNodeUpdate); + await this.executeSequential(plan, onNodeUpdate, executionId); break; case "parallel": - await this.executeParallel(plan, onNodeUpdate); + await this.executeParallel(plan, onNodeUpdate, executionId); break; case "conditional": - await this.executeConditional(plan, onNodeUpdate); + await this.executeConditional(plan, onNodeUpdate, executionId); break; default: throw new Error(`Unsupported execution mode: ${plan.executionMode}`); @@ -395,7 +470,8 @@ export class ExecutionEngine { status: string, data?: any, error?: string - ) => void + ) => void, + executionId: string ): Promise { const executionOrder = this.getExecutionOrder(plan.nodes, plan.edges); @@ -418,7 +494,7 @@ export class ExecutionEngine { if (!context) continue; try { - await this.executeNode(context, plan, onNodeUpdate); + await this.executeNode(context, plan, onNodeUpdate, executionId); } catch (error) { context.status = "failed"; context.error = @@ -446,7 +522,8 @@ export class ExecutionEngine { status: string, data?: any, error?: string - ) => void + ) => void, + executionId: string ): Promise { const parallelGroups = this.createParallelGroups( plan.nodes, @@ -456,7 +533,7 @@ export class ExecutionEngine { // Execute groups in parallel const groupPromises = parallelGroups.map((group) => - this.executeParallelGroup(group, plan, onNodeUpdate) + this.executeParallelGroup(group, plan, onNodeUpdate, executionId) ); await Promise.allSettled(groupPromises); @@ -471,13 +548,14 @@ export class ExecutionEngine { status: string, data?: any, error?: string - ) => void + ) => void, + executionId: string ): 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, onNodeUpdate); + return this.executeNode(context, plan, onNodeUpdate, executionId); }); if (group.waitForAll) { @@ -495,7 +573,8 @@ export class ExecutionEngine { status: string, data?: any, error?: string - ) => void + ) => void, + executionId: string ): Promise { const executionOrder = this.getExecutionOrder(plan.nodes, plan.edges); const visited = new Set(); @@ -507,7 +586,7 @@ export class ExecutionEngine { if (!context) continue; try { - await this.executeNode(context, plan, onNodeUpdate); + await this.executeNode(context, plan, onNodeUpdate, executionId); visited.add(nodeId); // Check for conditional branches @@ -517,7 +596,8 @@ export class ExecutionEngine { branches, plan, visited, - onNodeUpdate + onNodeUpdate, + executionId ); } } catch (error) { @@ -568,7 +648,8 @@ export class ExecutionEngine { status: string, data?: any, error?: string - ) => void + ) => void, + executionId: string ): Promise { for (const branch of branches) { const shouldExecuteTrue = await this.evaluateCondition( @@ -587,7 +668,7 @@ export class ExecutionEngine { if (!context) continue; try { - await this.executeNode(context, plan, onNodeUpdate); + await this.executeNode(context, plan, onNodeUpdate, executionId); visited.add(nodeId); } catch (error) { context.status = "failed"; @@ -729,11 +810,22 @@ export class ExecutionEngine { status: string, data?: any, error?: string - ) => void + ) => void, + executionId: string ): Promise { context.status = "running"; context.startTime = new Date(); + // Emit node start event + const nodeStartEvent: NodeStartEvent = { + type: "node:start", + executionId, + nodeId: context.nodeId, + nodeType: context.nodeType, + timestamp: new Date(), + }; + executionEventBus.emit(nodeStartEvent); + // Notify that node is starting if (onNodeUpdate) { onNodeUpdate(context.nodeId, "running"); @@ -782,6 +874,18 @@ export class ExecutionEngine { // Store in plan results plan.results.set(context.nodeId, result); + // Emit node complete event + const nodeCompleteEvent: NodeCompleteEvent = { + type: "node:complete", + executionId, + nodeId: context.nodeId, + nodeType: context.nodeType, + result: result, + duration: context.duration || 0, + timestamp: new Date(), + }; + executionEventBus.emit(nodeCompleteEvent); + // Notify that node completed successfully if (onNodeUpdate) { onNodeUpdate(context.nodeId, "completed", result); @@ -793,6 +897,17 @@ export class ExecutionEngine { context.duration = context.endTime.getTime() - context.startTime.getTime(); + // Emit node error event + const nodeErrorEvent: ExecutionErrorEvent = { + type: "node:error", + executionId, + nodeId: context.nodeId, + nodeType: context.nodeType, + error: context.error || "Unknown error", + timestamp: new Date(), + }; + executionEventBus.emit(nodeErrorEvent); + // Notify about the error if (onNodeUpdate) { onNodeUpdate(context.nodeId, "failed", undefined, context.error); @@ -809,7 +924,7 @@ export class ExecutionEngine { ); // Retry execution - return this.executeNode(context, plan, onNodeUpdate); + return this.executeNode(context, plan, onNodeUpdate, executionId); } throw error; diff --git a/src/services/executionEventBus.ts b/src/services/executionEventBus.ts new file mode 100644 index 0000000..c9d65ad --- /dev/null +++ b/src/services/executionEventBus.ts @@ -0,0 +1,236 @@ +// Event types for execution lifecycle +export interface ExecutionStartEvent { + type: "execution:start"; + executionId: string; + workflowId: string; + timestamp: Date; + nodeCount: number; + executionMode: "sequential" | "parallel" | "conditional"; +} + +export interface NodeStartEvent { + type: "node:start"; + executionId: string; + nodeId: string; + nodeType: string; + nodeLabel?: string; + timestamp: Date; + config?: Record; +} + +export interface NodeProgressEvent { + type: "node:progress"; + executionId: string; + nodeId: string; + progress: number; // 0-100 + message: string; + timestamp: Date; +} + +export interface NodeCompleteEvent { + type: "node:complete"; + executionId: string; + nodeId: string; + nodeType?: string; + status?: "success" | "failed" | "skipped"; + output?: any; + result?: any; + error?: string; + duration: number; // milliseconds + timestamp: Date; + retryCount?: number; +} + +export interface ExecutionCompleteEvent { + type: "execution:complete"; + executionId: string; + workflowId: string; + status: "success" | "failed" | "cancelled"; + totalDuration: number; // milliseconds + completedNodes: number; + failedNodes: number; + results: Map; + errors: Map; + timestamp: Date; +} + +export interface ExecutionErrorEvent { + type: "execution:error" | "node:error"; + executionId: string; + stage?: "validation" | "planning" | "execution"; + error: string; + nodeId?: string; + nodeType?: string; + timestamp: Date; +} + +export type ExecutionEvent = + | ExecutionStartEvent + | NodeStartEvent + | NodeProgressEvent + | NodeCompleteEvent + | ExecutionCompleteEvent + | ExecutionErrorEvent; + +type EventListener = ( + event: T +) => void; + +/** + * ExecutionEventBus - Centralized event emitter for workflow execution events + * Provides real-time feedback on execution progress to all subscribers + */ +export class ExecutionEventBus { + private listeners: Map> = + new Map(); + private eventHistory: ExecutionEvent[] = []; + private maxHistorySize = 1000; + private activeExecutions: Set = new Set(); + + /** + * Subscribe to a specific execution event type + */ + subscribe( + eventType: T["type"], + listener: EventListener + ): () => void { + if (!this.listeners.has(eventType)) { + this.listeners.set(eventType, new Set()); + } + + const listeners = this.listeners.get(eventType)!; + listeners.add(listener as EventListener); + + // Return unsubscribe function + return () => { + listeners.delete(listener as EventListener); + }; + } + + /** + * Subscribe to all events from a specific execution + */ + subscribeToExecution( + executionId: string, + listener: EventListener + ): () => void { + const eventTypes: ExecutionEvent["type"][] = [ + "execution:start", + "node:start", + "node:progress", + "node:complete", + "execution:complete", + "execution:error", + ]; + + const unsubscribers = eventTypes.map((eventType) => + this.subscribe(eventType, (event: any) => { + if (event.executionId === executionId) { + listener(event); + } + }) + ); + + return () => { + unsubscribers.forEach((unsub) => unsub()); + }; + } + + /** + * Emit an execution event + */ + emit(event: ExecutionEvent): void { + // Track active executions + if (event.type === "execution:start") { + this.activeExecutions.add(event.executionId); + } else if (event.type === "execution:complete") { + this.activeExecutions.delete(event.executionId); + } + + // Store in history + this.eventHistory.push(event); + if (this.eventHistory.length > this.maxHistorySize) { + this.eventHistory.shift(); + } + + // Emit to listeners + const listeners = this.listeners.get(event.type); + if (listeners) { + listeners.forEach((listener) => { + try { + listener(event); + } catch (error) { + console.error("Error in event listener:", error); + } + }); + } + + // Log important events + if ( + event.type === "execution:start" || + event.type === "execution:complete" || + event.type === "execution:error" || + (event.type === "node:complete" && event.status !== "success") + ) { + console.log(`[ExecutionEventBus] ${event.type}`, event); + } + } + + /** + * Get event history filtered by type or execution + */ + getHistory(filter?: { + executionId?: string; + eventType?: ExecutionEvent["type"]; + limit?: number; + }): ExecutionEvent[] { + let history = [...this.eventHistory]; + + if (filter?.executionId) { + history = history.filter((e) => e.executionId === filter.executionId); + } + + if (filter?.eventType) { + history = history.filter((e) => e.type === filter.eventType); + } + + if (filter?.limit) { + history = history.slice(-filter.limit); + } + + return history; + } + + /** + * Get active execution IDs + */ + getActiveExecutions(): string[] { + return Array.from(this.activeExecutions); + } + + /** + * Check if execution is active + */ + isExecutionActive(executionId: string): boolean { + return this.activeExecutions.has(executionId); + } + + /** + * Clear event history (useful for testing or cleanup) + */ + clearHistory(): void { + this.eventHistory = []; + } + + /** + * Unsubscribe from all listeners + */ + clear(): void { + this.listeners.clear(); + this.eventHistory = []; + this.activeExecutions.clear(); + } +} + +// Singleton instance +export const executionEventBus = new ExecutionEventBus();