MCP Console Automation Server

/** * Advanced Workflow Orchestration Engine * Handles complex multi-step automation workflows with conditional execution, * parallel processing, and error handling */ import { EventEmitter } from 'events'; import { v4 as uuidv4 } from 'uuid'; import PQueue from 'p-queue'; import { WorkflowDefinition, WorkflowExecution, WorkflowStatus, TaskExecution, TaskStatus, ExecutionContext, TaskCondition, StateMachine, State, Transition, ApprovalExecution, ExecutionError, ExecutionLog, WorkflowTask, ParallelExecution, LoopConfiguration, RetryPolicy, } from '../types/workflow.js'; import { ConsoleManager } from './ConsoleManager.js'; import { Logger } from '../utils/logger.js'; export class WorkflowEngine extends EventEmitter { private executions: Map<string, WorkflowExecution>; private workflows: Map<string, WorkflowDefinition>; private stateMachines: Map<string, StateMachine>; private taskQueues: Map<string, PQueue>; private consoleManager: ConsoleManager; private logger: Logger; private approvalCallbacks: Map< string, (approved: boolean, comments?: string) => void >; constructor(consoleManager: ConsoleManager) { super(); this.executions = new Map(); this.workflows = new Map(); this.stateMachines = new Map(); this.taskQueues = new Map(); this.consoleManager = consoleManager; this.logger = new Logger('WorkflowEngine'); this.approvalCallbacks = new Map(); } /** * Register a workflow definition */ registerWorkflow(definition: WorkflowDefinition): void { this.workflows.set(definition.id, definition); this.logger.info( `Workflow registered: ${definition.name} (${definition.id})` ); this.emit('workflow-registered', definition); } /** * Execute a workflow with given context */ async executeWorkflow( workflowId: string, context: ExecutionContext, variables?: Record<string, any> ): Promise<string> { const workflow = this.workflows.get(workflowId); if (!workflow) { throw new Error(`Workflow not found: ${workflowId}`); } const executionId = uuidv4(); const execution: WorkflowExecution = { id: executionId, workflowId, status: 'pending', startTime: new Date(), triggeredBy: { type: 'manual', source: 'api', timestamp: new Date(), }, context, tasks: [], approvals: [], variables: { ...workflow.variables.reduce( (acc, v) => ({ ...acc, [v.name]: v.defaultValue }), {} ), ...variables, }, artifacts: [], metrics: { totalTasks: workflow.tasks.length, completedTasks: 0, failedTasks: 0, skippedTasks: 0, avgTaskDuration: 0, resourceUsage: { peakMemory: 0, avgCpu: 0, diskUsed: 0, networkTraffic: 0, }, performance: { queueTime: 0, executionTime: 0, waitTime: 0, overhead: 0, }, }, logs: [], errors: [], }; this.executions.set(executionId, execution); this.createTaskQueue(executionId); try { await this.startExecution(execution, workflow); return executionId; } catch (error: any) { await this.handleExecutionError(execution, error); throw error; } } /** * Start workflow execution */ private async startExecution( execution: WorkflowExecution, workflow: WorkflowDefinition ): Promise<void> { this.updateExecutionStatus(execution, 'running'); this.addExecutionLog( execution, 'info', 'workflow', `Starting workflow: ${workflow.name}` ); try { // Initialize workflow variables await this.initializeVariables(execution, workflow); // Check if workflow uses state machine if (workflow.metadata.dependencies?.includes('state-machine')) { await this.executeStateMachineWorkflow(execution, workflow); } else { await this.executeSequentialWorkflow(execution, workflow); } } catch (error: any) { await this.handleExecutionError(execution, error); throw error; } } /** * Execute workflow using state machine approach */ private async executeStateMachineWorkflow( execution: WorkflowExecution, workflow: WorkflowDefinition ): Promise<void> { const stateMachine = this.createStateMachine(workflow); this.stateMachines.set(execution.id, stateMachine); let currentState = stateMachine.states.find((s) => s.type === 'initial'); if (!currentState) { throw new Error('No initial state found in state machine'); } while (currentState && currentState.type !== 'final') { this.addExecutionLog( execution, 'info', 'state-machine', `Entering state: ${currentState.name}` ); // Execute tasks in current state if (currentState.tasks) { await this.executeStateTasks(execution, workflow, currentState); } // Find next transition const transition = await this.findValidTransition( execution, stateMachine, currentState.id ); if (!transition) { if (currentState.type === 'error') { throw new Error( `Workflow stuck in error state: ${currentState.name}` ); } break; } // Execute transition actions if (transition.actions) { await this.executeTransitionActions(execution, transition.actions); } currentState = stateMachine.states.find((s) => s.id === transition.to); } this.updateExecutionStatus(execution, 'completed'); } /** * Execute workflow sequentially with dependency resolution */ private async executeSequentialWorkflow( execution: WorkflowExecution, workflow: WorkflowDefinition ): Promise<void> { const taskGraph = this.buildTaskDependencyGraph(workflow.tasks); const readyTasks = this.getReadyTasks(taskGraph, new Set()); const completedTasks = new Set<string>(); while (readyTasks.length > 0 || this.hasPendingTasks(execution)) { // Execute ready tasks const taskPromises = readyTasks.map((task) => this.executeTask(execution, workflow, task) ); const results = await Promise.allSettled(taskPromises); // Process results for (let i = 0; i < results.length; i++) { const task = readyTasks[i]; const result = results[i]; if (result.status === 'fulfilled') { completedTasks.add(task.id); execution.metrics.completedTasks++; } else { execution.metrics.failedTasks++; await this.handleTaskError(execution, task, result.reason); if (workflow.errorHandling.global.onError === 'fail') { throw result.reason; } } } // Find new ready tasks readyTasks.length = 0; readyTasks.push(...this.getReadyTasks(taskGraph, completedTasks)); // Check for deadlock if ( readyTasks.length === 0 && !this.allTasksCompleted(taskGraph, completedTasks) ) { throw new Error( 'Workflow deadlock detected - circular dependencies or missing tasks' ); } } this.updateExecutionStatus(execution, 'completed'); } /** * Execute a single task with all its features */ private async executeTask( execution: WorkflowExecution, workflow: WorkflowDefinition, task: WorkflowTask ): Promise<void> { const taskExecution: TaskExecution = { id: uuidv4(), taskId: task.id, status: 'pending', attempts: 0, input: task.input, metrics: { duration: 0, memoryUsage: 0, cpuUsage: 0, diskIO: 0, networkIO: 0, outputSize: 0, }, logs: [], }; execution.tasks.push(taskExecution); this.emit('task-started', execution.id, taskExecution); try { // Check task condition if ( task.condition && !(await this.evaluateCondition(execution, task.condition)) ) { taskExecution.status = 'skipped'; execution.metrics.skippedTasks++; this.addExecutionLog( execution, 'info', 'task', `Task skipped: ${task.name} (condition not met)` ); return; } // Handle approval requirement if (task.approval) { await this.handleTaskApproval(execution, task); } // Execute based on task type and execution mode taskExecution.status = 'running'; taskExecution.startTime = new Date(); if (task.executionMode === 'parallel' && task.parallel) { await this.executeParallelTask( execution, workflow, task, taskExecution ); } else if (task.loop) { await this.executeLoopTask(execution, workflow, task, taskExecution); } else { await this.executeSingleTask(execution, workflow, task, taskExecution); } taskExecution.endTime = new Date(); taskExecution.duration = taskExecution.endTime.getTime() - taskExecution.startTime!.getTime(); taskExecution.status = 'completed'; // Execute success actions if (task.onSuccess) { await this.executeTaskActions(execution, task.onSuccess); } } catch (error: any) { taskExecution.status = 'failed'; taskExecution.error = { timestamp: new Date(), source: task.id, type: error.constructor.name, message: error.message, stack: error.stack, context: { taskId: task.id, input: task.input }, recoverable: this.isRecoverableError(error), }; // Handle retry logic const retryPolicy = task.retryPolicy || workflow.errorHandling.global.retryPolicy; if (retryPolicy && taskExecution.attempts < retryPolicy.maxAttempts) { await this.retryTask( execution, workflow, task, taskExecution, retryPolicy ); return; } // Execute failure actions if (task.onFailure) { await this.executeTaskActions(execution, task.onFailure); } throw error; } finally { this.emit('task-completed', execution.id, taskExecution); } } /** * Execute parallel tasks with concurrency control */ private async executeParallelTask( execution: WorkflowExecution, workflow: WorkflowDefinition, task: WorkflowTask, taskExecution: TaskExecution ): Promise<void> { if (!task.parallel) return; const queue = new PQueue({ concurrency: task.parallel.maxConcurrency, }); const subtasks = this.generateParallelSubtasks(task); const results: any[] = []; try { const promises = subtasks.map((subtask, index) => queue.add(async () => { try { const result = await this.executeSingleTask( execution, workflow, subtask, taskExecution ); results[index] = result; return result; } catch (error) { if (task.parallel!.failFast) { queue.clear(); throw error; } results[index] = { error: error.message }; return null; } }) ); await Promise.all(promises); // Aggregate results based on strategy taskExecution.output = this.aggregateParallelResults( results, task.parallel.aggregationStrategy ); } finally { queue.clear(); } } /** * Execute task with loop configuration */ private async executeLoopTask( execution: WorkflowExecution, workflow: WorkflowDefinition, task: WorkflowTask, taskExecution: TaskExecution ): Promise<void> { if (!task.loop) return; const results: any[] = []; let iteration = 0; while (iteration < task.loop.maxIterations) { // Check loop condition if ( task.loop.condition && !(await this.evaluateCondition(execution, task.loop.condition)) ) { break; } // Check break condition if ( task.loop.breakCondition && (await this.evaluateCondition(execution, task.loop.breakCondition)) ) { break; } // Set loop variables if (task.loop.type === 'foreach' && task.loop.items) { const items = execution.variables[task.loop.items] as any[]; if (!items || iteration >= items.length) break; execution.variables['loop_item'] = items[iteration]; } execution.variables['loop_index'] = iteration; // Execute task iteration try { const result = await this.executeSingleTask( execution, workflow, task, taskExecution ); results.push(result); } catch (error: any) { if (workflow.errorHandling.global.onError === 'fail') { throw error; } results.push({ error: error.message, iteration }); } iteration++; // For 'for' loop, check iterations if ( task.loop.type === 'for' && task.loop.iterations && iteration >= task.loop.iterations ) { break; } } taskExecution.output = { iterations: iteration, results }; } /** * Execute a single task based on its type */ private async executeSingleTask( execution: WorkflowExecution, workflow: WorkflowDefinition, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { const startTime = Date.now(); try { let result: any; switch (task.type) { case 'command': result = await this.executeCommandTask( execution, task, taskExecution ); break; case 'script': result = await this.executeScriptTask(execution, task, taskExecution); break; case 'api_call': result = await this.executeApiCallTask( execution, task, taskExecution ); break; case 'file_operation': result = await this.executeFileOperationTask( execution, task, taskExecution ); break; case 'condition': result = await this.executeConditionTask( execution, task, taskExecution ); break; case 'wait': result = await this.executeWaitTask(execution, task, taskExecution); break; case 'notification': result = await this.executeNotificationTask( execution, task, taskExecution ); break; case 'subworkflow': result = await this.executeSubworkflowTask( execution, task, taskExecution ); break; default: throw new Error(`Unsupported task type: ${task.type}`); } // Update task output if (task.output && result) { if (task.output.variables) { Object.entries(task.output.variables).forEach(([key, expression]) => { execution.variables[key] = this.evaluateExpression( expression, result, execution.variables ); }); } } taskExecution.metrics.duration = Date.now() - startTime; return result; } catch (error: any) { taskExecution.metrics.duration = Date.now() - startTime; this.addExecutionLog( execution, 'error', 'task', `Task failed: ${task.name} - ${error.message}` ); throw error; } } /** * Execute command task using ConsoleManager */ private async executeCommandTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { if (!task.input.command) { throw new Error('Command task requires command in input'); } const command = this.interpolateVariables( task.input.command, execution.variables ); const args = task.input.args?.map((arg) => this.interpolateVariables(arg, execution.variables) ); this.addExecutionLog( execution, 'info', 'task', `Executing command: ${command} ${args?.join(' ') || ''}` ); try { const result = await this.consoleManager.executeCommand(command, args, { cwd: task.input.variables?.cwd as string, env: task.input.variables?.env as Record<string, string>, timeout: task.timeout, detectErrors: true, }); taskExecution.output = result; taskExecution.exitCode = result.exitCode; return result; } catch (error: any) { throw new Error(`Command execution failed: ${error.message}`); } } /** * Execute API call task */ private async executeApiCallTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { if (!task.input.url) { throw new Error('API call task requires URL in input'); } const url = this.interpolateVariables(task.input.url, execution.variables); const method = task.input.method || 'GET'; const headers = task.input.headers || {}; const body = task.input.body; this.addExecutionLog( execution, 'info', 'task', `Making API call: ${method} ${url}` ); try { const response = await fetch(url, { method, headers, body: body ? JSON.stringify(body) : undefined, }); if (!response.ok) { throw new Error(`HTTP ${response.status}: ${response.statusText}`); } const result = await response.json(); taskExecution.output = result; return result; } catch (error: any) { throw new Error(`API call failed: ${error.message}`); } } /** * Execute condition task */ private async executeConditionTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { if (!task.condition) { throw new Error('Condition task requires condition configuration'); } const result = await this.evaluateCondition(execution, task.condition); taskExecution.output = { result }; this.addExecutionLog( execution, 'info', 'task', `Condition evaluated to: ${result}` ); return { result }; } /** * Execute wait task */ private async executeWaitTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { const duration = (task.input.variables?.duration as number) || 1000; this.addExecutionLog( execution, 'info', 'task', `Waiting for ${duration}ms` ); await new Promise((resolve) => setTimeout(resolve, duration)); return { waited: duration }; } /** * Execute notification task */ private async executeNotificationTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { const message = this.interpolateVariables( (task.input.variables?.message as string) || 'Workflow notification', execution.variables ); this.addExecutionLog( execution, 'info', 'task', `Sending notification: ${message}` ); // Emit notification event this.emit('notification', { executionId: execution.id, taskId: task.id, message, timestamp: new Date(), }); return { sent: true, message }; } /** * Execute subworkflow task */ private async executeSubworkflowTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { const subworkflowId = task.input.variables?.workflowId as string; if (!subworkflowId) { throw new Error('Subworkflow task requires workflowId'); } this.addExecutionLog( execution, 'info', 'task', `Executing subworkflow: ${subworkflowId}` ); const subExecutionId = await this.executeWorkflow( subworkflowId, { ...execution.context }, task.input.variables as Record<string, any> ); // Wait for completion const subExecution = await this.waitForExecution(subExecutionId); return { executionId: subExecutionId, status: subExecution.status, output: subExecution.variables, }; } // ... Additional helper methods /** * Evaluate task condition */ private async evaluateCondition( execution: WorkflowExecution, condition: TaskCondition ): Promise<boolean> { switch (condition.type) { case 'expression': return this.evaluateExpression( condition.expression!, null, execution.variables ) as boolean; case 'variable': const variable = execution.variables[condition.variables![0]]; return this.compareValues( variable, condition.operator!, condition.value ); case 'previous_task': const taskResult = execution.tasks.find( (t) => t.taskId === condition.variables![0] ); return taskResult?.status === 'completed'; default: return true; } } private compareValues(actual: any, operator: string, expected: any): boolean { switch (operator) { case 'eq': return actual === expected; case 'ne': return actual !== expected; case 'gt': return actual > expected; case 'lt': return actual < expected; case 'gte': return actual >= expected; case 'lte': return actual <= expected; case 'contains': return String(actual).includes(String(expected)); case 'matches': return new RegExp(expected).test(String(actual)); case 'exists': return actual !== undefined && actual !== null; default: return false; } } /** * Interpolate variables in string templates */ private interpolateVariables( template: string, variables: Record<string, any> ): string { return template.replace(/\{\{(\w+)\}\}/g, (match, key) => { return variables[key] ?? match; }); } /** * Evaluate expression with context */ private evaluateExpression( expression: string, data: any, variables: Record<string, any> ): any { // Simple expression evaluation - in production, use a proper expression engine try { const context = { ...variables, data }; const func = new Function( ...Object.keys(context), `return ${expression}` ); return func(...Object.values(context)); } catch { return false; } } /** * Update execution status and emit events */ private updateExecutionStatus( execution: WorkflowExecution, status: WorkflowStatus ): void { execution.status = status; if ( status === 'completed' || status === 'failed' || status === 'cancelled' ) { execution.endTime = new Date(); execution.duration = execution.endTime.getTime() - execution.startTime.getTime(); } this.emit('execution-status-changed', execution.id, status); } /** * Add execution log entry */ private addExecutionLog( execution: WorkflowExecution, level: 'debug' | 'info' | 'warn' | 'error', source: string, message: string, data?: any ): void { const log: ExecutionLog = { timestamp: new Date(), level, source, message, data, }; execution.logs.push(log); this.logger[level](`[${execution.id}] ${message}`, data); } /** * Execute task actions (onSuccess, onError, etc.) */ private async executeTaskActions( execution: WorkflowExecution, actions: any[] ): Promise<void> { for (const action of actions) { this.addExecutionLog( execution, 'info', 'task-action', `Executing action: ${action.type || 'custom'}` ); // Implementation would depend on action types defined in workflow } } // ... Additional utility methods for task queues, dependency graphs, approvals, etc. /** * Get execution by ID */ getExecution(executionId: string): WorkflowExecution | undefined { return this.executions.get(executionId); } /** * Cancel workflow execution */ async cancelExecution(executionId: string): Promise<void> { const execution = this.executions.get(executionId); if (!execution) { throw new Error(`Execution not found: ${executionId}`); } this.updateExecutionStatus(execution, 'cancelled'); this.addExecutionLog( execution, 'info', 'workflow', 'Workflow cancelled by user' ); // Cancel running tasks const runningTasks = execution.tasks.filter((t) => t.status === 'running'); for (const task of runningTasks) { if (task.sessionId) { await this.consoleManager.stopSession(task.sessionId); } } this.emit('execution-cancelled', executionId); } /** * Get all executions */ getAllExecutions(): WorkflowExecution[] { return Array.from(this.executions.values()); } /** * Clean up completed executions */ cleanup(olderThanHours: number = 24): void { const cutoff = Date.now() - olderThanHours * 60 * 60 * 1000; Array.from(this.executions.entries()).forEach(([id, execution]) => { if (execution.endTime && execution.endTime.getTime() < cutoff) { this.executions.delete(id); this.taskQueues.delete(id); this.stateMachines.delete(id); } }); } // Additional private helper methods would go here... private createTaskQueue(executionId: string): void { this.taskQueues.set(executionId, new PQueue({ concurrency: 5 })); } private async initializeVariables( execution: WorkflowExecution, workflow: WorkflowDefinition ): Promise<void> { // Initialize workflow variables from definition for (const variable of workflow.variables) { if (!(variable.name in execution.variables)) { execution.variables[variable.name] = variable.defaultValue; } } } private buildTaskDependencyGraph( tasks: WorkflowTask[] ): Map<string, WorkflowTask> { return new Map(tasks.map((task) => [task.id, task])); } private getReadyTasks( taskGraph: Map<string, WorkflowTask>, completed: Set<string> ): WorkflowTask[] { const ready: WorkflowTask[] = []; Array.from(taskGraph.values()).forEach((task) => { const allDependenciesCompleted = task.dependsOn.every((dep) => completed.has(dep) ); if (allDependenciesCompleted && !completed.has(task.id)) { ready.push(task); } }); return ready; } private hasPendingTasks(execution: WorkflowExecution): boolean { return execution.tasks.some( (t) => t.status === 'running' || t.status === 'pending' ); } private allTasksCompleted( taskGraph: Map<string, WorkflowTask>, completed: Set<string> ): boolean { return taskGraph.size === completed.size; } private createStateMachine(workflow: WorkflowDefinition): StateMachine { // Create a basic state machine from workflow tasks // In practice, this would be more sophisticated return { id: workflow.id, name: workflow.name, initialState: 'start', states: [ { id: 'start', name: 'Start', type: 'initial', tasks: [] }, { id: 'end', name: 'End', type: 'final', tasks: [] }, ], transitions: [ { id: 'start-to-end', from: 'start', to: 'end', actions: [] }, ], context: {}, }; } private async executeStateTasks( execution: WorkflowExecution, workflow: WorkflowDefinition, state: State ): Promise<void> { if (!state.tasks) return; for (const taskId of state.tasks) { const task = workflow.tasks.find((t) => t.id === taskId); if (task) { await this.executeTask(execution, workflow, task); } } } private async findValidTransition( execution: WorkflowExecution, stateMachine: StateMachine, currentStateId: string ): Promise<Transition | undefined> { const transitions = stateMachine.transitions.filter( (t) => t.from === currentStateId ); for (const transition of transitions) { if ( !transition.condition || (await this.evaluateCondition(execution, transition.condition)) ) { return transition; } } return undefined; } private async executeTransitionActions( execution: WorkflowExecution, actions: any[] ): Promise<void> { // Execute state transition actions for (const action of actions) { // Implementation depends on action types this.addExecutionLog( execution, 'info', 'state-machine', `Executing transition action: ${action.type}` ); } } private async handleTaskError( execution: WorkflowExecution, task: WorkflowTask, error: any ): Promise<void> { const executionError: ExecutionError = { timestamp: new Date(), source: task.id, type: error.constructor.name, message: error.message, stack: error.stack, context: { taskId: task.id }, recoverable: this.isRecoverableError(error), }; execution.errors.push(executionError); this.addExecutionLog( execution, 'error', 'task', `Task error: ${task.name} - ${error.message}` ); } private async handleExecutionError( execution: WorkflowExecution, error: any ): Promise<void> { this.updateExecutionStatus(execution, 'failed'); this.addExecutionLog( execution, 'error', 'workflow', `Workflow failed: ${error.message}` ); } private isRecoverableError(error: any): boolean { // Implement logic to determine if error is recoverable return error.code !== 'FATAL'; } private generateParallelSubtasks(task: WorkflowTask): WorkflowTask[] { // Generate subtasks for parallel execution // This is a simplified implementation return [task]; // In practice, split task into parallel subtasks } private aggregateParallelResults(results: any[], strategy: string): any { switch (strategy) { case 'array': return results; case 'first': return results[0]; case 'last': return results[results.length - 1]; case 'merge': return Object.assign({}, ...results); default: return results; } } private async retryTask( execution: WorkflowExecution, workflow: WorkflowDefinition, task: WorkflowTask, taskExecution: TaskExecution, retryPolicy: RetryPolicy ): Promise<void> { taskExecution.attempts++; const delay = this.calculateRetryDelay(retryPolicy, taskExecution.attempts); this.addExecutionLog( execution, 'info', 'task', `Retrying task ${task.name} (attempt ${taskExecution.attempts}/${retryPolicy.maxAttempts}) after ${delay}ms` ); await new Promise((resolve) => setTimeout(resolve, delay)); // Reset task status and retry taskExecution.status = 'running'; await this.executeSingleTask(execution, workflow, task, taskExecution); } private calculateRetryDelay(policy: RetryPolicy, attempt: number): number { switch (policy.backoff) { case 'exponential': return Math.min( policy.initialDelay * Math.pow(policy.multiplier || 2, attempt - 1), policy.maxDelay || 30000 ); case 'linear': return Math.min( policy.initialDelay * attempt, policy.maxDelay || 30000 ); default: return policy.initialDelay; } } private async handleTaskApproval( execution: WorkflowExecution, task: WorkflowTask ): Promise<void> { if (!task.approval) return; const approvalExecution: ApprovalExecution = { id: uuidv4(), configId: task.approval.configId, status: 'pending', requestTime: new Date(), context: task.approval.context || {}, }; execution.approvals.push(approvalExecution); this.addExecutionLog( execution, 'info', 'approval', `Approval required for task: ${task.name}` ); this.emit('approval-required', execution.id, approvalExecution); // Wait for approval return new Promise((resolve, reject) => { this.approvalCallbacks.set( approvalExecution.id, (approved: boolean, comments?: string) => { approvalExecution.status = approved ? 'approved' : 'rejected'; approvalExecution.responseTime = new Date(); approvalExecution.comments = comments; if (approved) { resolve(); } else { reject( new Error( `Task approval rejected: ${comments || 'No comments provided'}` ) ); } } ); }); } private async executeFileOperationTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { // Implement file operations throw new Error('File operation task not implemented yet'); } private async executeScriptTask( execution: WorkflowExecution, task: WorkflowTask, taskExecution: TaskExecution ): Promise<any> { // Implement script execution throw new Error('Script task not implemented yet'); } private async waitForExecution( executionId: string ): Promise<WorkflowExecution> { return new Promise((resolve, reject) => { const checkStatus = () => { const execution = this.executions.get(executionId); if (!execution) { reject(new Error(`Execution not found: ${executionId}`)); return; } if ( execution.status === 'completed' || execution.status === 'failed' || execution.status === 'cancelled' ) { resolve(execution); } else { setTimeout(checkStatus, 1000); } }; checkStatus(); }); } /** * Approve a pending approval */ approveTask(approvalId: string, approved: boolean, comments?: string): void { const callback = this.approvalCallbacks.get(approvalId); if (callback) { callback(approved, comments); this.approvalCallbacks.delete(approvalId); } } }