From a774bb0f829e9a3a76e49cef5373199bbf01369d Mon Sep 17 00:00:00 2001 From: David Snelling Date: Wed, 4 Jun 2025 10:01:46 -0700 Subject: [PATCH] feat: implement `SequentialPipeline` with threading and WebSocket support Added the `SequentialPipeline` class for structured augmentation execution. Integrated threading capabilities via the `THREADED` execution mode in `augmentationPipeline`. Enhanced WebSocket support for real-time data processing through pipelines. --- src/augmentationPipeline.ts | 90 ++++++++- src/sequentialPipeline.ts | 364 ++++++++++++++++++++++++++++++++++++ src/types/augmentations.ts | 3 + 3 files changed, 453 insertions(+), 4 deletions(-) create mode 100644 src/sequentialPipeline.ts diff --git a/src/augmentationPipeline.ts b/src/augmentationPipeline.ts index e266da64..bf46f1d5 100644 --- a/src/augmentationPipeline.ts +++ b/src/augmentationPipeline.ts @@ -13,6 +13,8 @@ import { AugmentationResponse, AugmentationType } from './types/augmentations.js' +import { isThreadingAvailable, isBrowser, isNode } from './utils/environment.js' +import { executeInThread } from './utils/workerUtils.js' /** * Type definitions for the augmentation registry @@ -35,7 +37,8 @@ export enum ExecutionMode { SEQUENTIAL = 'sequential', PARALLEL = 'parallel', FIRST_SUCCESS = 'firstSuccess', - FIRST_RESULT = 'firstResult' + FIRST_RESULT = 'firstResult', + THREADED = 'threaded' // Execute in separate threads when available } /** @@ -45,6 +48,8 @@ export interface PipelineOptions { mode?: ExecutionMode; timeout?: number; stopOnError?: boolean; + forceThreading?: boolean; // Force threading even if not in THREADED mode + disableThreading?: boolean; // Disable threading even if in THREADED mode } /** @@ -53,7 +58,9 @@ export interface PipelineOptions { const DEFAULT_PIPELINE_OPTIONS: PipelineOptions = { mode: ExecutionMode.SEQUENTIAL, timeout: 30000, - stopOnError: false + stopOnError: false, + forceThreading: false, + disableThreading: false } /** @@ -501,6 +508,32 @@ export class AugmentationPipeline { return methods.every(method => typeof (augmentation as any)[method] === 'function') } + /** + * Determines if threading should be used based on options and environment + * + * @param options The pipeline options + * @returns True if threading should be used, false otherwise + */ + private shouldUseThreading(options: PipelineOptions): boolean { + // If threading is explicitly disabled, don't use it + if (options.disableThreading) { + return false; + } + + // If threading is explicitly forced, use it if available + if (options.forceThreading) { + return isThreadingAvailable(); + } + + // If in THREADED mode, use threading if available + if (options.mode === ExecutionMode.THREADED) { + return isThreadingAvailable(); + } + + // Otherwise, don't use threading + return false; + } + /** * Execute a pipeline for a specific augmentation type * @@ -551,8 +584,33 @@ export class AugmentationPipeline { }) : null - // Execute the method on the augmentation - const methodPromise = (augmentation[method] as Function)(...args) as AugmentationResponse + // Check if threading should be used + const useThreading = this.shouldUseThreading(options); + + // Execute the method on the augmentation, using threading if appropriate + let methodPromise: Promise>; + + if (useThreading) { + // Execute in a separate thread + try { + // Create a function that can be serialized and executed in a worker + const workerFn = (...workerArgs: any[]) => { + // This function will be stringified and executed in the worker + // It needs to be self-contained + const augFn = augmentation[method as string] as Function; + return augFn.apply(augmentation, workerArgs); + }; + + methodPromise = executeInThread>(workerFn, ...args); + } catch (threadError) { + console.warn(`Failed to execute in thread, falling back to main thread: ${threadError}`); + // Fall back to executing in the main thread + methodPromise = Promise.resolve((augmentation[method] as Function)(...args) as AugmentationResponse); + } + } else { + // Execute in the main thread + methodPromise = Promise.resolve((augmentation[method] as Function)(...args) as AugmentationResponse); + } // Race the method promise against the timeout promise if a timeout is specified const result = timeoutPromise @@ -576,6 +634,30 @@ export class AugmentationPipeline { // Execute all augmentations in parallel return enabledAugmentations.map(executeMethod) + case ExecutionMode.THREADED: + // Execute all augmentations in parallel with threading enabled + // Force threading for this mode + const threadedOptions = { ...options, forceThreading: true }; + + // Create a new executeMethod function that uses the threaded options + const executeMethodThreaded = async (augmentation: T) => { + // Save the original options + const originalOptions = options; + + // Set the options to the threaded options + options = threadedOptions; + + // Execute the method + const result = await executeMethod(augmentation); + + // Restore the original options + options = originalOptions; + + return result; + }; + + return enabledAugmentations.map(executeMethodThreaded); + case ExecutionMode.FIRST_SUCCESS: // Execute augmentations sequentially until one succeeds for (const augmentation of enabledAugmentations) { diff --git a/src/sequentialPipeline.ts b/src/sequentialPipeline.ts new file mode 100644 index 00000000..1d66e9b8 --- /dev/null +++ b/src/sequentialPipeline.ts @@ -0,0 +1,364 @@ +/** + * Sequential Augmentation Pipeline + * + * This module provides a pipeline for executing augmentations in a specific sequence: + * ISense -> IMemory -> ICognition -> IConduit -> IActivation -> IPerception + * + * It supports high-performance streaming data from WebSockets without blocking. + */ + +import { + AugmentationType, + IAugmentation, + IWebSocketSupport, + ISenseAugmentation, + IMemoryAugmentation, + ICognitionAugmentation, + IConduitAugmentation, + IActivationAugmentation, + IPerceptionAugmentation, + AugmentationResponse, + WebSocketConnection +} from './types/augmentations.js' +import { BrainyData } from './brainyData.js' +import { augmentationPipeline } from './augmentationPipeline.js' + +/** + * Options for sequential pipeline execution + */ +export interface SequentialPipelineOptions { + /** + * Timeout for each augmentation execution in milliseconds + */ + timeout?: number; + + /** + * Whether to stop execution if an error occurs + */ + stopOnError?: boolean; + + /** + * BrainyData instance to use for storage + */ + brainyData?: BrainyData; +} + +/** + * Default pipeline options + */ +const DEFAULT_SEQUENTIAL_PIPELINE_OPTIONS: SequentialPipelineOptions = { + timeout: 30000, + stopOnError: false +} + +/** + * Result of a pipeline execution + */ +export interface PipelineResult { + /** + * Whether the pipeline execution was successful + */ + success: boolean; + + /** + * The data returned by the pipeline + */ + data: T; + + /** + * Error message if the pipeline execution failed + */ + error?: string; + + /** + * Results from each stage of the pipeline + */ + stageResults: { + sense?: AugmentationResponse; + memory?: AugmentationResponse; + cognition?: AugmentationResponse; + conduit?: AugmentationResponse; + activation?: AugmentationResponse; + perception?: AugmentationResponse; + } +} + +/** + * SequentialPipeline class + * + * Executes augmentations in a specific sequence: + * ISense -> IMemory -> ICognition -> IConduit -> IActivation -> IPerception + */ +export class SequentialPipeline { + private brainyData: BrainyData; + + /** + * Create a new sequential pipeline + * + * @param options Options for the pipeline + */ + constructor(options: SequentialPipelineOptions = {}) { + this.brainyData = options.brainyData || new BrainyData(); + } + + /** + * Initialize the pipeline + * + * @returns A promise that resolves when initialization is complete + */ + public async initialize(): Promise { + await this.brainyData.init(); + } + + /** + * Process data through the sequential pipeline + * + * @param rawData The raw data to process + * @param dataType The type of data (e.g., 'text', 'image', 'audio') + * @param options Options for pipeline execution + * @returns A promise that resolves with the pipeline result + */ + public async processData( + rawData: Buffer | string, + dataType: string, + options: SequentialPipelineOptions = {} + ): Promise> { + const opts = { ...DEFAULT_SEQUENTIAL_PIPELINE_OPTIONS, ...options }; + const result: PipelineResult = { + success: true, + data: null, + stageResults: {} + }; + + try { + // Step 1: Process raw data with ISense augmentations + const senseResults = await augmentationPipeline.executeSensePipeline( + 'processRawData', + [rawData, dataType], + { timeout: opts.timeout, stopOnError: opts.stopOnError } + ); + + // Get the first successful result + let senseResult: AugmentationResponse<{ nouns: string[], verbs: string[] }> | null = null; + for (const resultPromise of senseResults) { + const res = await resultPromise; + if (res.success) { + senseResult = res; + break; + } + } + + if (!senseResult || !senseResult.success) { + return { + success: false, + data: null, + error: 'Failed to process raw data with ISense augmentations', + stageResults: { sense: senseResult || { success: false, data: null, error: 'No sense augmentations available' } } + }; + } + + result.stageResults.sense = senseResult; + + // Step 2: Store data in BrainyData using IMemory augmentations + const memoryAugmentations = augmentationPipeline.getAugmentationsByType(AugmentationType.MEMORY) as IMemoryAugmentation[]; + + if (memoryAugmentations.length === 0) { + return { + success: false, + data: null, + error: 'No memory augmentations available', + stageResults: result.stageResults + }; + } + + // Use the first available memory augmentation + const memoryAugmentation = memoryAugmentations[0]; + + // Generate a key for the data + const dataKey = `data_${Date.now()}_${Math.random().toString(36).substring(2, 15)}`; + + // Store the data + const memoryResult = await memoryAugmentation.storeData( + dataKey, + { + rawData, + dataType, + nouns: senseResult.data.nouns, + verbs: senseResult.data.verbs, + timestamp: Date.now() + } + ); + + if (!memoryResult.success) { + return { + success: false, + data: null, + error: `Failed to store data: ${memoryResult.error}`, + stageResults: { ...result.stageResults, memory: memoryResult } + }; + } + + result.stageResults.memory = memoryResult; + + // Step 3: Trigger ICognition augmentations to analyze the data + const cognitionResults = await augmentationPipeline.executeCognitionPipeline( + 'reason', + [`Analyze data with key ${dataKey}`, { dataKey }], + { timeout: opts.timeout, stopOnError: opts.stopOnError } + ); + + // Get the first successful result + let cognitionResult: AugmentationResponse<{ inference: string, confidence: number }> | null = null; + for (const resultPromise of cognitionResults) { + const res = await resultPromise; + if (res.success) { + cognitionResult = res; + break; + } + } + + if (cognitionResult) { + result.stageResults.cognition = cognitionResult; + } + + // Step 4: Send notifications to IConduit augmentations + const conduitResults = await augmentationPipeline.executeConduitPipeline( + 'writeData', + [{ dataKey, nouns: senseResult.data.nouns, verbs: senseResult.data.verbs }], + { timeout: opts.timeout, stopOnError: opts.stopOnError } + ); + + // Get the first successful result + let conduitResult: AugmentationResponse | null = null; + for (const resultPromise of conduitResults) { + const res = await resultPromise; + if (res.success) { + conduitResult = res; + break; + } + } + + if (conduitResult) { + result.stageResults.conduit = conduitResult; + } + + // Step 5: Send notifications to IActivation augmentations + const activationResults = await augmentationPipeline.executeActivationPipeline( + 'triggerAction', + ['dataProcessed', { dataKey }], + { timeout: opts.timeout, stopOnError: opts.stopOnError } + ); + + // Get the first successful result + let activationResult: AugmentationResponse | null = null; + for (const resultPromise of activationResults) { + const res = await resultPromise; + if (res.success) { + activationResult = res; + break; + } + } + + if (activationResult) { + result.stageResults.activation = activationResult; + } + + // Step 6: Send notifications to IPerception augmentations + const perceptionResults = await augmentationPipeline.executePerceptionPipeline( + 'interpret', + [senseResult.data.nouns, senseResult.data.verbs, { dataKey }], + { timeout: opts.timeout, stopOnError: opts.stopOnError } + ); + + // Get the first successful result + let perceptionResult: AugmentationResponse> | null = null; + for (const resultPromise of perceptionResults) { + const res = await resultPromise; + if (res.success) { + perceptionResult = res; + break; + } + } + + if (perceptionResult) { + result.stageResults.perception = perceptionResult; + result.data = perceptionResult.data; + } else { + // If no perception result, use the cognition result as the final data + result.data = cognitionResult ? cognitionResult.data : { dataKey }; + } + + return result; + } catch (error) { + return { + success: false, + data: null, + error: `Pipeline execution failed: ${error}`, + stageResults: result.stageResults + }; + } + } + + /** + * Process WebSocket data through the sequential pipeline + * + * @param connection The WebSocket connection + * @param dataType The type of data (e.g., 'text', 'image', 'audio') + * @param options Options for pipeline execution + * @returns A function to handle incoming WebSocket messages + */ + public createWebSocketHandler( + connection: WebSocketConnection, + dataType: string, + options: SequentialPipelineOptions = {} + ): (data: unknown) => void { + return (data: unknown) => { + // Process the data asynchronously without blocking + this.processData( + typeof data === 'string' ? data : JSON.stringify(data), + dataType, + options + ).catch(error => { + console.error('Error processing WebSocket data:', error); + }); + }; + } + + /** + * Set up a WebSocket connection to process data through the pipeline + * + * @param url The WebSocket URL to connect to + * @param dataType The type of data (e.g., 'text', 'image', 'audio') + * @param options Options for pipeline execution + * @returns A promise that resolves with the WebSocket connection + */ + public async setupWebSocketPipeline( + url: string, + dataType: string, + options: SequentialPipelineOptions = {} + ): Promise { + // Get WebSocket-supporting augmentations + const webSocketAugmentations = augmentationPipeline.getWebSocketAugmentations(); + + if (webSocketAugmentations.length === 0) { + throw new Error('No WebSocket-supporting augmentations available'); + } + + // Use the first available WebSocket augmentation + const webSocketAugmentation = webSocketAugmentations[0]; + + // Connect to the WebSocket + const connection = await webSocketAugmentation.connectWebSocket(url); + + // Create a handler for incoming messages + const handler = this.createWebSocketHandler(connection, dataType, options); + + // Register the handler + await webSocketAugmentation.onWebSocketMessage(connection.connectionId, handler); + + return connection; + } +} + +// Create and export a default instance of the sequential pipeline +export const sequentialPipeline = new SequentialPipeline(); diff --git a/src/types/augmentations.ts b/src/types/augmentations.ts index 0fa0aedc..fae4e5ea 100644 --- a/src/types/augmentations.ts +++ b/src/types/augmentations.ts @@ -48,6 +48,9 @@ export interface IAugmentation { shutDown(): Promise getStatus(): Promise<'active' | 'inactive' | 'error'> + + // Allow string indexing for dynamic method access + [key: string]: any; } /**