From 65c7e2138615f94aaa93df2bc64b4f51382765c3 Mon Sep 17 00:00:00 2001
From: Waleed Latif <walif6@gmail.com>
Date: Thu, 16 Jan 2025 12:30:23 -0800
Subject: [PATCH] Added executor that is block-agnostic and executes serialized
 workflow

---
 executor/__tests__/executor.test.ts     | 272 ++++++++++++++++++++++++
 executor/index.ts                       | 225 ++++++++++++++++++++
 executor/types.ts                       |  27 +++
 serializer/__tests__/serializer.test.ts |  12 +-
 serializer/types.ts                     |   2 -
 5 files changed, 530 insertions(+), 8 deletions(-)
 create mode 100644 executor/__tests__/executor.test.ts
 create mode 100644 executor/index.ts
 create mode 100644 executor/types.ts

diff --git a/executor/__tests__/executor.test.ts b/executor/__tests__/executor.test.ts
new file mode 100644
index 0000000000..45a3c83013
--- /dev/null
+++ b/executor/__tests__/executor.test.ts
@@ -0,0 +1,272 @@
+import { Executor } from '../index';
+import { SerializedWorkflow } from '@/serializer/types';
+import { Tool } from '../types';
+import { toolRegistry } from '@/tools/registry';
+
+// Mock tools
+class MockTool implements Tool {
+  constructor(
+    public name: string,
+    private mockExecute: (params: Record<string, any>) => Promise<Record<string, any>>,
+    private mockValidate: (params: Record<string, any>) => boolean | string = () => true
+  ) {}
+
+  async execute(params: Record<string, any>): Promise<Record<string, any>> {
+    return this.mockExecute(params);
+  }
+
+  validateParams(params: Record<string, any>): boolean | string {
+    return this.mockValidate(params);
+  }
+}
+
+describe('Executor', () => {
+  beforeEach(() => {
+    // Reset toolRegistry mock
+    (toolRegistry as any) = {};
+  });
+
+  describe('Tool Execution', () => {
+    it('should execute a simple workflow with one tool', async () => {
+      const mockTool = new MockTool(
+        'test-tool',
+        async (params) => ({ result: params.input + ' processed' })
+      );
+      (toolRegistry as any)['test-tool'] = mockTool;
+
+      const workflow: SerializedWorkflow = {
+        version: '1.0',
+        blocks: [{
+          id: 'block-1',
+          position: { x: 0, y: 0 },
+          config: {
+            tool: 'test-tool',
+            params: {},
+            interface: {
+              inputs: { input: 'string' },
+              outputs: { result: 'string' }
+            }
+          }
+        }],
+        connections: []
+      };
+
+      const executor = new Executor(workflow);
+      const result = await executor.execute('workflow-1', { input: 'test' });
+
+      expect(result.success).toBe(true);
+      expect(result.data).toEqual({ result: 'test processed' });
+    });
+
+    it('should validate tool parameters', async () => {
+      const mockTool = new MockTool(
+        'test-tool',
+        async () => ({}),
+        (params) => params.required ? true : 'Missing required parameter'
+      );
+      (toolRegistry as any)['test-tool'] = mockTool;
+
+      const workflow: SerializedWorkflow = {
+        version: '1.0',
+        blocks: [{
+          id: 'block-1',
+          position: { x: 0, y: 0 },
+          config: {
+            tool: 'test-tool',
+            params: {},
+            interface: {
+              inputs: {},
+              outputs: {}
+            }
+          }
+        }],
+        connections: []
+      };
+
+      const executor = new Executor(workflow);
+      const result = await executor.execute('workflow-1', {});
+
+      expect(result.success).toBe(false);
+      expect(result.error).toContain('Missing required parameter');
+    });
+  });
+
+  describe('Interface Validation', () => {
+    it('should validate input types', async () => {
+      const mockTool = new MockTool(
+        'test-tool',
+        async (params) => ({ result: params.input })
+      );
+      (toolRegistry as any)['test-tool'] = mockTool;
+
+      const workflow: SerializedWorkflow = {
+        version: '1.0',
+        blocks: [{
+          id: 'block-1',
+          position: { x: 0, y: 0 },
+          config: {
+            tool: 'test-tool',
+            params: {},
+            interface: {
+              inputs: { input: 'number' },
+              outputs: { result: 'number' }
+            }
+          }
+        }],
+        connections: []
+      };
+
+      const executor = new Executor(workflow);
+      const result = await executor.execute('workflow-1', { input: 'not a number' });
+
+      expect(result.success).toBe(false);
+      expect(result.error).toContain('Invalid type for input');
+    });
+
+    it('should validate tool output against interface', async () => {
+      const mockTool = new MockTool(
+        'test-tool',
+        async () => ({ wrongField: 'wrong type' })
+      );
+      (toolRegistry as any)['test-tool'] = mockTool;
+
+      const workflow: SerializedWorkflow = {
+        version: '1.0',
+        blocks: [{
+          id: 'block-1',
+          position: { x: 0, y: 0 },
+          config: {
+            tool: 'test-tool',
+            params: {},
+            interface: {
+              inputs: {},
+              outputs: { result: 'string' }
+            }
+          }
+        }],
+        connections: []
+      };
+
+      const executor = new Executor(workflow);
+      const result = await executor.execute('workflow-1', {});
+
+      expect(result.success).toBe(false);
+      expect(result.error).toContain('Tool output missing required field');
+    });
+  });
+
+  describe('Complex Workflows', () => {
+    it('should execute a workflow with multiple connected blocks', async () => {
+      const processorTool = new MockTool(
+        'processor',
+        async (params) => ({ processed: params.input.toUpperCase() })
+      );
+      const formatterTool = new MockTool(
+        'formatter',
+        async (params) => ({ result: `<${params.processed}>` })
+      );
+      (toolRegistry as any)['processor'] = processorTool;
+      (toolRegistry as any)['formatter'] = formatterTool;
+
+      const workflow: SerializedWorkflow = {
+        version: '1.0',
+        blocks: [
+          {
+            id: 'process',
+            position: { x: 0, y: 0 },
+            config: {
+              tool: 'processor',
+              params: {},
+              interface: {
+                inputs: { input: 'string' },
+                outputs: { processed: 'string' }
+              }
+            }
+          },
+          {
+            id: 'format',
+            position: { x: 100, y: 0 },
+            config: {
+              tool: 'formatter',
+              params: {},
+              interface: {
+                inputs: { processed: 'string' },
+                outputs: { result: 'string' }
+              }
+            }
+          }
+        ],
+        connections: [{
+          source: 'process',
+          target: 'format',
+          sourceHandle: 'processed',
+          targetHandle: 'processed'
+        }]
+      };
+
+      const executor = new Executor(workflow);
+      const result = await executor.execute('workflow-1', { input: 'test' });
+
+      expect(result.success).toBe(true);
+      expect(result.data).toEqual({ result: '<TEST>' });
+    });
+
+    it('should handle circular dependencies', async () => {
+      const mockTool = new MockTool(
+        'test-tool',
+        async () => ({ output: 'test' })
+      );
+      (toolRegistry as any)['test-tool'] = mockTool;
+
+      const workflow: SerializedWorkflow = {
+        version: '1.0',
+        blocks: [
+          {
+            id: 'block-1',
+            position: { x: 0, y: 0 },
+            config: {
+              tool: 'test-tool',
+              params: {},
+              interface: {
+                inputs: { input: 'string' },
+                outputs: { output: 'string' }
+              }
+            }
+          },
+          {
+            id: 'block-2',
+            position: { x: 100, y: 0 },
+            config: {
+              tool: 'test-tool',
+              params: {},
+              interface: {
+                inputs: { input: 'string' },
+                outputs: { output: 'string' }
+              }
+            }
+          }
+        ],
+        connections: [
+          {
+            source: 'block-1',
+            target: 'block-2',
+            sourceHandle: 'output',
+            targetHandle: 'input'
+          },
+          {
+            source: 'block-2',
+            target: 'block-1',
+            sourceHandle: 'output',
+            targetHandle: 'input'
+          }
+        ]
+      };
+
+      const executor = new Executor(workflow);
+      const result = await executor.execute('workflow-1', {});
+
+      expect(result.success).toBe(false);
+      expect(result.error).toContain('Workflow contains cycles');
+    });
+  });
+});
diff --git a/executor/index.ts b/executor/index.ts
new file mode 100644
index 0000000000..74b15848fa
--- /dev/null
+++ b/executor/index.ts
@@ -0,0 +1,225 @@
+import { SerializedWorkflow, SerializedBlock } from '@/serializer/types';
+import { ExecutionContext, ExecutionResult, Tool } from './types';
+import { toolRegistry } from '@/tools/registry';
+
+export class Executor {
+  private workflow: SerializedWorkflow;
+  
+  constructor(workflow: SerializedWorkflow) {
+    this.workflow = workflow;
+  }
+
+  private async executeBlock(
+    block: SerializedBlock,
+    inputs: Record<string, any>,
+    context: ExecutionContext
+  ): Promise<Record<string, any>> {
+    // Get the tool specified by the block's tool property
+    const toolName = block.config.tool;
+    if (!toolName) {
+      throw new Error(`Block ${block.id} does not specify a tool`);
+    }
+
+    const tool = toolRegistry[toolName];
+    if (!tool) {
+      throw new Error(`Tool not found: ${toolName}`);
+    }
+
+    // Validate interface compatibility
+    this.validateInterface(block, inputs);
+
+    // Merge tool parameters with runtime inputs
+    const params = {
+      ...block.config.params,
+      ...inputs
+    };
+
+    // Validate the parameters against tool requirements
+    const validationResult = tool.validateParams(params);
+    if (typeof validationResult === 'string') {
+      throw new Error(`Invalid parameters for tool ${toolName}: ${validationResult}`);
+    }
+
+    try {
+      // Execute the tool and validate its output matches the interface
+      const result = await tool.execute(params);
+      this.validateToolOutput(block, result);
+      return result;
+    } catch (error) {
+      throw new Error(`Tool ${toolName} execution failed: ${error instanceof Error ? error.message : 'Unknown error'}`);
+    }
+  }
+
+  private validateInterface(block: SerializedBlock, inputs: Record<string, any>): 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<string, any>): 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 'object':
+        return typeof value === 'object' && value !== null;
+      case 'array':
+        return Array.isArray(value);
+      case 'function':
+        return typeof value === 'function';
+      default:
+        // For complex types like 'Record<string, any>', 'string[]', etc.
+        // We just do basic object/array validation
+        return true;
+    }
+  }
+
+  private determineExecutionOrder(): string[] {
+    const { blocks, connections } = this.workflow;
+    const order: string[] = [];
+    const visited = new Set<string>();
+    const inDegree = new Map<string, number>();
+    
+    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<string, any> {
+    const inputs: Record<string, any> = {};
+    
+    // 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, pass through workflow inputs
+    if (Object.keys(inputs).length === 0 && context.input) {
+      return context.input;
+    }
+    
+    return inputs;
+  }
+
+  async execute(workflowId: string, input: Record<string, any>): Promise<ExecutionResult> {
+    const startTime = new Date();
+    const context: ExecutionContext = {
+      workflowId,
+      blockStates: new Map(),
+      input,
+      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()
+        }
+      };
+    }
+  }
+}
diff --git a/executor/types.ts b/executor/types.ts
new file mode 100644
index 0000000000..c140018d56
--- /dev/null
+++ b/executor/types.ts
@@ -0,0 +1,27 @@
+export interface Tool {
+  name: string;
+  execute(params: Record<string, any>): Promise<Record<string, any>>;
+  validateParams(params: Record<string, any>): boolean | string;
+}
+
+export interface ToolRegistry {
+  [key: string]: Tool;
+}
+
+export interface ExecutionContext {
+  workflowId: string;
+  blockStates: Map<string, any>;
+  input: Record<string, any>;
+  metadata?: Record<string, any>;
+}
+
+export interface ExecutionResult {
+  success: boolean;
+  data: Record<string, any>;
+  error?: string;
+  metadata?: {
+    duration: number;
+    startTime: string;
+    endTime: string;
+  };
+} 
\ No newline at end of file
diff --git a/serializer/__tests__/serializer.test.ts b/serializer/__tests__/serializer.test.ts
index f867beb754..685dd57ae5 100644
--- a/serializer/__tests__/serializer.test.ts
+++ b/serializer/__tests__/serializer.test.ts
@@ -128,7 +128,7 @@ describe('Serializer', () => {
         data: {
           tool: 'model',
           params: {
-            model: 'gpt-4'
+            model: 'gpt-4o'
           },
           interface: {
             inputs: {},
@@ -142,7 +142,7 @@ describe('Serializer', () => {
       
       expect(block.id).toBe('minimal-1');
       expect(block.config.tool).toBe('model');
-      expect(block.config.params).toEqual({ model: 'gpt-4' });
+      expect(block.config.params).toEqual({ model: 'gpt-4o' });
       expect(block.metadata).toBeUndefined();
     });
 
@@ -173,7 +173,7 @@ describe('Serializer', () => {
           data: {
             tool: 'model',
             params: {
-              model: 'gpt-4',
+              model: 'gpt-4o',
               systemPrompt: 'Process this data'
             },
             interface: {
@@ -254,7 +254,7 @@ describe('Serializer', () => {
         data: {
           tool: 'model',
           params: {
-            model: 'gpt-4',
+            model: 'gpt-4o',
             temperature: 0.7,
             maxTokens: 1000,
             topP: 0.9,
@@ -273,7 +273,7 @@ describe('Serializer', () => {
       const block = serialized.blocks[0];
       
       expect(block.config.params).toEqual({
-        model: 'gpt-4',
+        model: 'gpt-4o',
         temperature: 0.7,
         maxTokens: 1000,
         topP: 0.9,
@@ -374,7 +374,7 @@ describe('Serializer', () => {
           config: {
             tool: 'model',
             params: {
-              model: 'gpt-4'
+              model: 'gpt-4o'
             },
             interface: {
               inputs: {
diff --git a/serializer/types.ts b/serializer/types.ts
index 09f2de4e24..6f34b06d0f 100644
--- a/serializer/types.ts
+++ b/serializer/types.ts
@@ -1,5 +1,3 @@
-import { BlockConfig } from "@/blocks/types/block";
-
 export interface SerializedWorkflow {
   version: string;
   blocks: SerializedBlock[];