import { SerializedWorkflow, SerializedBlock } from '@/serializer/types' import { ExecutionContext, ExecutionResult, Tool } from './types' import { tools } from '@/tools' import { BlockState } from '@/stores/workflow/types' export class Executor { private workflow: SerializedWorkflow constructor(workflow: SerializedWorkflow) { this.workflow = workflow } private async executeBlock( block: SerializedBlock, inputs: Record, context: ExecutionContext ): Promise> { const config = block.config const toolId = config.tool if (!toolId) { throw new Error(`Block ${block.id} does not specify a tool`) } const tool = tools[toolId] if (!tool) { throw new Error(`Tool not found: ${toolId}`) } // Validate interface compatibility this.validateInterface(block, inputs) // Merge block parameters with runtime inputs const params = { ...config.params, ...inputs } // Validate tool parameters this.validateToolParams(tool, params) try { // Make the HTTP request const url = typeof tool.request.url === 'function' ? tool.request.url(params) : tool.request.url const response = await fetch(url, { method: tool.request.method, headers: tool.request.headers(params), body: tool.request.body ? JSON.stringify(tool.request.body(params)) : undefined }) if (!response.ok) { const error = await response.json().catch(() => ({ message: response.statusText })) throw new Error(tool.transformError(error)) } const result = await tool.transformResponse(response) // Validate the output matches the interface this.validateToolOutput(block, result) return result } catch (error) { throw new Error(`Tool ${toolId} execution failed: ${error instanceof Error ? error.message : 'Unknown error'}`) } } private validateToolParams(tool: Tool, params: Record): void { // Check required parameters for (const [paramName, paramConfig] of Object.entries(tool.params)) { if (paramConfig.required && !(paramName in params)) { throw new Error(`Missing required parameter '${paramName}' for tool ${tool.id}`) } } } private validateInterface(block: SerializedBlock, inputs: Record): void { const { interface: blockInterface } = block.config // Check if all required inputs are provided for (const [inputName, inputType] of Object.entries(blockInterface.inputs)) { if (!(inputName in inputs)) { throw new Error(`Missing required input '${inputName}' of type '${inputType}' for block ${block.id}`) } // Basic type validation (can be enhanced for more complex types) if (!this.validateType(inputs[inputName], inputType)) { throw new Error(`Invalid type for input '${inputName}' in block ${block.id}. Expected ${inputType}`) } } } private validateToolOutput(block: SerializedBlock, output: Record): void { const { interface: blockInterface } = block.config // Check if all promised outputs are present for (const [outputName, outputType] of Object.entries(blockInterface.outputs)) { if (!(outputName in output)) { throw new Error(`Tool output missing required field '${outputName}' of type '${outputType}' for block ${block.id}`) } // Basic type validation (can be enhanced for more complex types) if (!this.validateType(output[outputName], outputType)) { throw new Error(`Invalid type for output '${outputName}' in block ${block.id}. Expected ${outputType}`) } } } private validateType(value: any, expectedType: string): boolean { switch (expectedType.toLowerCase()) { case 'string': return typeof value === 'string' case 'number': return typeof value === 'number' case 'boolean': return typeof value === 'boolean' case 'json': try { if (typeof value === 'string') { JSON.parse(value) } return true } catch { return false } default: // For complex types, we just do basic object/array validation return true } } private determineExecutionOrder(): string[] { const { blocks, connections } = this.workflow const order: string[] = [] const visited = new Set() const inDegree = new Map() blocks.forEach(block => inDegree.set(block.id, 0)) connections.forEach(conn => { const target = conn.target inDegree.set(target, (inDegree.get(target) || 0) + 1) }) const queue = blocks .filter(block => (inDegree.get(block.id) || 0) === 0) .map(block => block.id) while (queue.length > 0) { const blockId = queue.shift()! if (visited.has(blockId)) continue visited.add(blockId) order.push(blockId) connections .filter(conn => conn.source === blockId) .forEach(conn => { const targetId = conn.target inDegree.set(targetId, (inDegree.get(targetId) || 0) - 1) if (inDegree.get(targetId) === 0) { queue.push(targetId) } }) } if (order.length !== blocks.length) { throw new Error('Workflow contains cycles') } return order } private resolveInputs( block: SerializedBlock, context: ExecutionContext ): Record { const inputs: Record = {} // Get all incoming connections for this block const incomingConnections = this.workflow.connections.filter( conn => conn.target === block.id ) // Map outputs from previous blocks to inputs for this block incomingConnections.forEach(conn => { const sourceOutput = context.blockStates.get(conn.source) if (sourceOutput && conn.sourceHandle && conn.targetHandle) { inputs[conn.targetHandle] = sourceOutput[conn.sourceHandle] } }) // If this is a start block with no inputs, use the block's params if (Object.keys(inputs).length === 0) { const targetBlock = this.workflow.blocks.find(b => b.id === block.id) if (targetBlock) { return targetBlock.config.params } } return inputs } async execute(workflowId: string): Promise { const startTime = new Date() const context: ExecutionContext = { workflowId, blockStates: new Map(), metadata: { startTime: startTime.toISOString() } } try { const executionOrder = this.determineExecutionOrder() for (const blockId of executionOrder) { const block = this.workflow.blocks.find(b => b.id === blockId) if (!block) { throw new Error(`Block ${blockId} not found in workflow`) } const blockInputs = this.resolveInputs(block, context) const result = await this.executeBlock(block, blockInputs, context) context.blockStates.set(blockId, result) } const lastBlockId = executionOrder[executionOrder.length - 1] const finalOutput = context.blockStates.get(lastBlockId) const endTime = new Date() return { success: true, data: finalOutput, metadata: { duration: endTime.getTime() - startTime.getTime(), startTime: startTime.toISOString(), endTime: endTime.toISOString() } } } catch (error) { const endTime = new Date() return { success: false, data: {}, error: error instanceof Error ? error.message : 'Unknown error occurred', metadata: { duration: endTime.getTime() - startTime.getTime(), startTime: startTime.toISOString(), endTime: endTime.toISOString() } } } } }