feat: Complete Brainy 1.0 Great Cleanup

🎯 THE GREAT CLEANUP - Making Brainy Beautiful BREAKING CHANGES: - Removed addSmart() method (use add() - it's smart by default) - Removed duplicate Pipeline classes (consolidated into ONE Cortex) - Removed 40+ CLI commands (now just 5 clean commands) ✅ WHAT'S DONE: - Delete duplicate files: sequentialPipeline.ts, cortex-legacy.ts, serviceIntegration.ts - Consolidated into ONE Cortex class (the orchestrator) - Pipeline class now delegates to Cortex (backward compatibility) - Clean CLI: add, import, search, status, help (ONE way to do everything) - Interactive mode for beginners 📈 RESULTS: - 5 CLI commands (was 40+) - 1 Pipeline system (was 3+) - Clean, obvious naming - Beautiful user experience This achieves the vision: ONE way to do everything, elegant and powerful.
2025-08-14 09:42:44 -07:00 · 2025-08-14 09:42:44 -07:00 · f08f57e665
commit f08f57e665
parent 7fb34a0b1d
7 changed files with 468 additions and 6592 deletions
--- a/bin/brainy.js
+++ b/bin/brainy.js
--- a/src/brainyData.ts
+++ b/src/brainyData.ts
@ -1666,27 +1666,24 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
  }
  /**
-   * Add data to the database (literal storage by default)
+   * Add data to the database with intelligent processing
   * 
   * 🔒 Safe by default: Only stores your data literally without AI processing
   * 🧠 AI processing: Set { process: true } or use addSmart() for Neural Import
   * 
   * @param vectorOrData Vector or data to add
   * @param metadata Optional metadata to associate with the data
-   * @param options Additional options - use { process: true } for AI analysis
+   * @param options Additional options for processing
   * @returns The ID of the added data
   * 
   * @example
-   * // Literal storage (safe, no AI processing)
+   * // Auto mode - intelligently decides processing
-   * await brainy.add("API_KEY=secret123")
+   * await brainy.add("Customer feedback: Great product!")
   * 
   * @example  
-   * // With AI processing (explicit opt-in)
+   * // Explicit literal mode for sensitive data
-   * await brainy.add("John works at Acme Corp", null, { process: true })
+   * await brainy.add("API_KEY=secret123", null, { process: 'literal' })
   * 
   * @example
-   * // Smart processing (recommended for data analysis)
+   * // Force neural processing
-   * await brainy.addSmart("Customer feedback: Great product!")
+   * await brainy.add("John works at Acme Corp", null, { process: 'neural' })
   */
  public async add(
    vectorOrData: Vector | any,
@ -1696,7 +1693,7 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
      addToRemote?: boolean // Whether to also add to the remote server if connected
      id?: string // Optional ID to use instead of generating a new one
      service?: string // The service that is inserting the data
-      process?: boolean // Enable AI processing (neural import, entity detection, etc.)
+      process?: 'auto' | 'literal' | 'neural' // Processing mode (default: 'auto')
    } = {}
  ): Promise<string> {
    await this.ensureInitialized()
@ -2000,8 +1997,20 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
      // Invalidate search cache since data has changed
      this.searchCache.invalidateOnDataChange('add')
-      // 🧠 AI Processing (Neural Import) - Only if explicitly requested
+      // Determine processing mode
-      if (options.process === true) {
+      const processingMode = options.process || 'auto'
      let shouldProcessNeurally = false
      if (processingMode === 'neural') {
        shouldProcessNeurally = true
      } else if (processingMode === 'auto') {
        // Auto-detect whether to use neural processing
        shouldProcessNeurally = this.shouldAutoProcessNeurally(vectorOrData, metadata)
      }
      // 'literal' mode means no neural processing
      // 🧠 AI Processing (Neural Import) - Based on processing mode
      if (shouldProcessNeurally) {
        try {
          // Execute SENSE pipeline (includes Neural Import and other AI augmentations)
          await augmentationPipeline.executeSensePipeline(
@ -4477,31 +4486,16 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
  }
  /**
-   * Add data with AI processing enabled by default
+   * @deprecated Use add() instead - it's smart by default now
-   * 
+   * @hidden
   * 🧠 This method automatically enables Neural Import and other AI augmentations
   * for intelligent data understanding, entity detection, and relationship analysis.
   * 
   * Use this when you want AI to understand and process your data.
   * Use regular add() when you want literal storage only.
   *
   * @param vectorOrData The data to add (any format)
   * @param metadata Optional metadata to associate with the data
   * @param options Additional options (process defaults to true)
   * @returns The ID of the added data
   */
  public async addSmart(
    vectorOrData: Vector | any,
    metadata?: T,
-    options: {
+    options: any = {}
      forceEmbed?: boolean
      addToRemote?: boolean
      id?: string
      service?: string
    } = {}
  ): Promise<string> {
-    // Call add() with process=true by default
+    console.warn('⚠️ addSmart() is deprecated. Use add() instead - it\'s smart by default now!')
-    return this.add(vectorOrData, metadata, { ...options, process: true })
+    return this.add(vectorOrData, metadata, { ...options, process: 'auto' })
  }
  /**
--- a/src/cortex/cortex-legacy.ts
+++ b/src/cortex/cortex-legacy.ts
--- a/src/cortex/serviceIntegration.ts
+++ b/src/cortex/serviceIntegration.ts
@ -1,512 +0,0 @@
 /**
 * Service Integration Helpers - Seamless Cortex Integration for Existing Services
 * 
 * Atomic Age Service Management Protocol
 */
 import { BrainyData } from '../brainyData.js'
 import { Cortex } from './cortex-legacy.js'
 import * as fs from '../universal/fs.js'
 import * as path from '../universal/path.js'
 export interface ServiceConfig {
  name: string
  version?: string
  environment?: 'development' | 'production' | 'staging' | 'test'
  storage?: {
    type: 'filesystem' | 's3' | 'gcs' | 'memory'
    bucket?: string
    path?: string
    credentials?: any
  }
  features?: {
    chat?: boolean
    augmentations?: string[]
    encryption?: boolean
  }
  migration?: {
    strategy: 'immediate' | 'gradual'
    rollback?: boolean
  }
 }
 export interface BrainyOptions {
  storage?: any
  augmentations?: any[]
  encryption?: boolean
  caching?: boolean
 }
 export interface MigrationPlan {
  fromStorage: string
  toStorage: string
  strategy: 'immediate' | 'gradual'
  rollback?: boolean
  validation?: boolean
  backup?: boolean
 }
 export interface ServiceInstance {
  id: string
  name: string
  version: string
  status: 'healthy' | 'degraded' | 'unhealthy'
  lastSeen: Date
  config: ServiceConfig
 }
 export interface HealthReport {
  service: ServiceInstance
  checks: {
    storage: boolean
    search: boolean
    embedding: boolean
    config: boolean
  }
  performance: {
    responseTime: number
    memoryUsage: number
    storageSize: number
  }
  issues: string[]
 }
 export interface MigrationReport {
  plan: MigrationPlan
  estimated: {
    duration: number
    downtime: number
    dataSize: number
    complexity: 'low' | 'medium' | 'high'
  }
  risks: string[]
  prerequisites: string[]
  steps: string[]
 }
 /**
 * Service Integration Helper Class
 */
 export class CortexServiceIntegration {
  /**
   * Initialize Cortex for a service with automatic configuration
   */
  static async initializeForService(serviceName: string, options?: Partial<ServiceConfig>): Promise<{ cortex: Cortex, config: ServiceConfig }> {
    const cortex = new Cortex()
    // Try to load existing configuration
    let config: ServiceConfig
    try {
      config = await this.loadServiceConfig(serviceName)
    } catch {
      // Create new configuration
      config = await this.createServiceConfig(serviceName, options)
    }
    await cortex.init({
      storage: config.storage?.type,
      encryption: config.features?.encryption
    })
    return { cortex, config }
  }
  /**
   * Create BrainyData instance from Cortex configuration
   */
  static async createBrainyFromCortex(cortex: Cortex, serviceName?: string): Promise<BrainyData> {
    // Get storage configuration from Cortex
    const storageType = await cortex.configGet('STORAGE_TYPE') || 'filesystem'
    const encryptionEnabled = await cortex.configGet('ENCRYPTION_ENABLED') === 'true'
    const options: BrainyOptions = {
      storage: await this.getBrainyStorageOptions(cortex, storageType),
      encryption: encryptionEnabled,
      caching: true
    }
    // Load augmentations if specified
    if (serviceName) {
      const serviceConfig = await this.loadServiceConfig(serviceName)
      if (serviceConfig.features?.augmentations) {
        options.augmentations = serviceConfig.features.augmentations
      }
    }
    const brainy = new BrainyData(options)
    await brainy.init()
    return brainy
  }
  /**
   * Auto-discover Brainy instances in the current environment
   */
  static async discoverBrainyInstances(): Promise<ServiceInstance[]> {
    const instances: ServiceInstance[] = []
    // Look for .cortex directories
    const searchPaths = [
      process.cwd(),
      path.join(process.cwd(), '..'),
      '/opt/services',
      '/var/lib/services'
    ]
    for (const searchPath of searchPaths) {
      try {
        const entries = await fs.readdir(searchPath, { withFileTypes: true })
        for (const entry of entries) {
          if (entry.isDirectory()) {
            const cortexPath = path.join(searchPath, entry.name, '.cortex')
            try {
              await fs.access(cortexPath)
              const instance = await this.loadServiceInstance(path.join(searchPath, entry.name))
              if (instance) instances.push(instance)
            } catch {
              // No Cortex in this directory
            }
          }
        }
      } catch {
        // Directory doesn't exist or can't be read
      }
    }
    return instances
  }
  /**
   * Perform health check on all discovered services
   */
  static async healthCheckAll(): Promise<HealthReport[]> {
    const instances = await this.discoverBrainyInstances()
    const reports: HealthReport[] = []
    for (const instance of instances) {
      try {
        const report = await this.performHealthCheck(instance)
        reports.push(report)
      } catch (error) {
        reports.push({
          service: instance,
          checks: { storage: false, search: false, embedding: false, config: false },
          performance: { responseTime: -1, memoryUsage: -1, storageSize: -1 },
          issues: [`Health check failed: ${error}`]
        })
      }
    }
    return reports
  }
  /**
   * Plan migration for a service
   */
  static async planMigration(serviceName: string, plan: Partial<MigrationPlan>): Promise<MigrationReport> {
    const config = await this.loadServiceConfig(serviceName)
    const fullPlan: MigrationPlan = {
      fromStorage: config.storage?.type || 'filesystem',
      toStorage: plan.toStorage || 's3',
      strategy: plan.strategy || 'immediate',
      rollback: plan.rollback ?? true,
      validation: plan.validation ?? true,
      backup: plan.backup ?? true
    }
    // Estimate migration complexity
    const dataSize = await this.estimateDataSize(serviceName)
    const complexity = this.assessMigrationComplexity(fullPlan, dataSize)
    return {
      plan: fullPlan,
      estimated: {
        duration: this.estimateDuration(complexity, dataSize),
        downtime: this.estimateDowntime(fullPlan.strategy),
        dataSize,
        complexity
      },
      risks: this.identifyRisks(fullPlan),
      prerequisites: this.getPrerequisites(fullPlan),
      steps: this.generateMigrationSteps(fullPlan)
    }
  }
  /**
   * Execute migration for all services
   */
  static async migrateAll(plan: MigrationPlan): Promise<void> {
    const instances = await this.discoverBrainyInstances()
    for (const instance of instances) {
      const cortex = new Cortex()
      // Set working directory to service directory
      process.chdir(path.dirname(instance.config.name))
      await cortex.migrate({
        to: plan.toStorage,
        strategy: plan.strategy,
        bucket: plan.toStorage === 's3' ? 'default-bucket' : undefined
      })
    }
  }
  /**
   * Generate Brainy storage options from Cortex config
   */
  private static async getBrainyStorageOptions(cortex: Cortex, storageType: string): Promise<any> {
    switch (storageType) {
      case 'filesystem':
        return { forceFileSystemStorage: true }
      case 's3':
        return {
          forceS3CompatibleStorage: true,
          s3Config: {
            bucket: await cortex.configGet('S3_BUCKET'),
            accessKeyId: await cortex.configGet('AWS_ACCESS_KEY_ID'),
            secretAccessKey: await cortex.configGet('AWS_SECRET_ACCESS_KEY'),
            region: await cortex.configGet('AWS_REGION') || 'us-east-1'
          }
        }
      case 'r2':
        return {
          forceS3CompatibleStorage: true,
          s3Config: {
            bucket: await cortex.configGet('CLOUDFLARE_R2_BUCKET'),
            accessKeyId: await cortex.configGet('AWS_ACCESS_KEY_ID'), // R2 uses AWS-compatible keys
            secretAccessKey: await cortex.configGet('AWS_SECRET_ACCESS_KEY'),
            endpoint: await cortex.configGet('CLOUDFLARE_R2_ENDPOINT') || 
                     `https://${await cortex.configGet('CLOUDFLARE_R2_ACCOUNT_ID')}.r2.cloudflarestorage.com`,
            region: 'auto' // R2 uses 'auto' as region
          }
        }
      case 'gcs':
        return {
          forceS3CompatibleStorage: true,
          s3Config: {
            bucket: await cortex.configGet('GCS_BUCKET'),
            endpoint: 'https://storage.googleapis.com',
            // GCS credentials would be configured here
          }
        }
      default:
        return { forceMemoryStorage: true }
    }
  }
  /**
   * Load service configuration
   */
  private static async loadServiceConfig(serviceName: string): Promise<ServiceConfig> {
    const configPath = path.join(process.cwd(), '.cortex', 'service.json')
    const data = await fs.readFile(configPath, 'utf8')
    return JSON.parse(data)
  }
  /**
   * Create new service configuration
   */
  private static async createServiceConfig(serviceName: string, options?: Partial<ServiceConfig>): Promise<ServiceConfig> {
    const config: ServiceConfig = {
      name: serviceName,
      version: '1.0.0',
      environment: 'development',
      storage: {
        type: 'filesystem',
        path: './brainy_data'
      },
      features: {
        chat: true,
        encryption: true,
        augmentations: []
      },
      ...options
    }
    // Save configuration
    const configDir = path.join(process.cwd(), '.cortex')
    await fs.mkdir(configDir, { recursive: true })
    await fs.writeFile(
      path.join(configDir, 'service.json'),
      JSON.stringify(config, null, 2)
    )
    return config
  }
  /**
   * Load service instance information
   */
  private static async loadServiceInstance(servicePath: string): Promise<ServiceInstance | null> {
    try {
      const configPath = path.join(servicePath, '.cortex', 'service.json')
      const config = JSON.parse(await fs.readFile(configPath, 'utf8'))
      return {
        id: path.basename(servicePath),
        name: config.name,
        version: config.version || '1.0.0',
        status: 'healthy', // Would be determined by actual health check
        lastSeen: new Date(),
        config
      }
    } catch {
      return null
    }
  }
  /**
   * Perform health check on a service instance
   */
  private static async performHealthCheck(instance: ServiceInstance): Promise<HealthReport> {
    // Simulate health check - in real implementation, this would:
    // 1. Connect to the service
    // 2. Test storage connectivity
    // 3. Verify search functionality
    // 4. Check embedding model availability
    // 5. Measure performance metrics
    return {
      service: instance,
      checks: {
        storage: true,
        search: true,
        embedding: true,
        config: true
      },
      performance: {
        responseTime: Math.random() * 100 + 50, // 50-150ms
        memoryUsage: Math.random() * 512 + 256, // 256-768MB
        storageSize: Math.random() * 1024 + 100  // 100-1124MB
      },
      issues: []
    }
  }
  /**
   * Estimate data size for migration planning
   */
  private static async estimateDataSize(serviceName: string): Promise<number> {
    // Simulate data size estimation
    return Math.floor(Math.random() * 1000 + 100) // 100-1100MB
  }
  /**
   * Assess migration complexity
   */
  private static assessMigrationComplexity(plan: MigrationPlan, dataSize: number): 'low' | 'medium' | 'high' {
    if (dataSize > 5000 || plan.fromStorage !== plan.toStorage) return 'high'
    if (dataSize > 1000) return 'medium'
    return 'low'
  }
  /**
   * Estimate migration duration
   */
  private static estimateDuration(complexity: string, dataSize: number): number {
    const baseTime = dataSize / 100 // 1 minute per 100MB
    const multiplier = complexity === 'high' ? 3 : complexity === 'medium' ? 2 : 1
    return Math.ceil(baseTime * multiplier)
  }
  /**
   * Estimate downtime for migration strategy
   */
  private static estimateDowntime(strategy: string): number {
    switch (strategy) {
      case 'immediate': return 60 // 1 minute
      case 'gradual': return 10   // 10 seconds
      default: return 30
    }
  }
  /**
   * Identify migration risks
   */
  private static identifyRisks(plan: MigrationPlan): string[] {
    const risks: string[] = []
    if (plan.fromStorage !== plan.toStorage) {
      risks.push('Cross-platform data compatibility')
    }
    if (plan.strategy === 'immediate') {
      risks.push('Service downtime during migration')
    }
    if (!plan.backup) {
      risks.push('Data loss if migration fails')
    }
    return risks
  }
  /**
   * Get migration prerequisites
   */
  private static getPrerequisites(plan: MigrationPlan): string[] {
    const prereqs: string[] = []
    if (plan.toStorage === 's3') {
      prereqs.push('AWS credentials configured')
      prereqs.push('S3 bucket created and accessible')
    }
    if (plan.toStorage === 'r2') {
      prereqs.push('Cloudflare R2 API token configured')
      prereqs.push('R2 bucket created and accessible')
      prereqs.push('CLOUDFLARE_R2_ACCOUNT_ID environment variable set')
    }
    if (plan.toStorage === 'gcs') {
      prereqs.push('GCP service account configured')
      prereqs.push('GCS bucket created and accessible')
    }
    if (plan.backup) {
      prereqs.push('Sufficient storage space for backup')
    }
    return prereqs
  }
  /**
   * Generate migration steps
   */
  private static generateMigrationSteps(plan: MigrationPlan): string[] {
    const steps: string[] = []
    if (plan.backup) {
      steps.push('Create backup of current data')
    }
    steps.push(`Initialize ${plan.toStorage} storage`)
    steps.push('Validate connectivity to target storage')
    if (plan.strategy === 'gradual') {
      steps.push('Begin gradual data migration')
      steps.push('Monitor migration progress')
      steps.push('Switch traffic to new storage')
    } else {
      steps.push('Stop service')
      steps.push('Migrate all data')
      steps.push('Update configuration')
      steps.push('Start service with new storage')
    }
    if (plan.validation) {
      steps.push('Validate data integrity')
      steps.push('Run health checks')
    }
    steps.push('Clean up old storage (if successful)')
    return steps
  }
 }
--- a/src/index.ts
+++ b/src/index.ts
@ -183,12 +183,7 @@ import {
  StreamlinedPipelineResult
 } from './pipeline.js'
-// Export sequential pipeline (for backward compatibility)
+// Sequential pipeline removed - use unified pipeline instead
 import {
  SequentialPipeline,
  sequentialPipeline,
  SequentialPipelineOptions
 } from './sequentialPipeline.js'
 // Export augmentation factory
 import {
@ -205,8 +200,6 @@ export {
  pipeline,
  augmentationPipeline,
  ExecutionMode,
  SequentialPipeline,
  sequentialPipeline,
  // Streamlined pipeline exports (now part of unified pipeline)
  executeStreamlined,
@ -227,7 +220,6 @@ export {
 export type {
  PipelineOptions,
  PipelineResult,
  SequentialPipelineOptions,
  StreamlinedPipelineOptions,
  StreamlinedPipelineResult,
  AugmentationOptions
--- a/src/pipeline.ts
+++ b/src/pipeline.ts
--- a/src/sequentialPipeline.ts
+++ b/src/sequentialPipeline.ts
@ -1,572 +0,0 @@
 /**
 * 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.
 * Optimized for Node.js 23.11+ using native WebStreams API.
 */
 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'
 // Use the browser's built-in WebStreams API or Node.js native WebStreams API
 // This approach ensures compatibility with both environments
 let TransformStream: any, ReadableStream: any, WritableStream: any;
 // Function to initialize the stream classes
 const initializeStreamClasses = () => {
  // Try to use the browser's built-in WebStreams API first
  if (typeof globalThis.TransformStream !== 'undefined' &&
      typeof globalThis.ReadableStream !== 'undefined' &&
      typeof globalThis.WritableStream !== 'undefined') {
    TransformStream = globalThis.TransformStream;
    ReadableStream = globalThis.ReadableStream;
    WritableStream = globalThis.WritableStream;
    return Promise.resolve();
  } else {
    // In Node.js environment, try to import from node:stream/web
    // This will be executed in Node.js but not in browsers
    return import('node:stream/web')
      .then(streamWebModule => {
        TransformStream = streamWebModule.TransformStream;
        ReadableStream = streamWebModule.ReadableStream;
        WritableStream = streamWebModule.WritableStream;
      })
      .catch(error => {
        console.error('Failed to import WebStreams API:', error);
        // Provide fallback implementations or throw a more helpful error
        throw new Error('WebStreams API is not available in this environment. Please use a modern browser or Node.js 18+.');
      });
  }
 };
 // Initialize immediately but don't block module execution
 const streamClassesPromise = initializeStreamClasses();
 /**
 * 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<T> {
  /**
   * 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<unknown>;
    memory?: AugmentationResponse<unknown>;
    cognition?: AugmentationResponse<unknown>;
    conduit?: AugmentationResponse<unknown>;
    activation?: AugmentationResponse<unknown>;
    perception?: AugmentationResponse<unknown>;
  }
 }
 /**
 * 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();
  }
  /**
   * Ensure stream classes are initialized
   * @private
   */
  private async ensureStreamClassesInitialized(): Promise<void> {
    await streamClassesPromise;
  }
  /**
   * Initialize the pipeline
   * 
   * @returns A promise that resolves when initialization is complete
   */
  public async initialize(): Promise<void> {
    // Initialize stream classes and BrainyData in parallel
    await Promise.all([
      this.ensureStreamClassesInitialized(),
      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<PipelineResult<unknown>> {
    const opts = { ...DEFAULT_SEQUENTIAL_PIPELINE_OPTIONS, ...options };
    const result: PipelineResult<unknown> = {
      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<unknown> | 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<unknown> | 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<Record<string, unknown>> | 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: unknown) {
      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 async createWebSocketHandler(
    connection: WebSocketConnection,
    dataType: string,
    options: SequentialPipelineOptions = {}
  ): Promise<(data: unknown) => void> {
    // Ensure stream classes are initialized
    await this.ensureStreamClassesInitialized();
    // Create a transform stream for processing data
    const transformStream = new TransformStream({
      transform: async (chunk: unknown, controller: TransformStreamDefaultController) => {
        try {
          const data = typeof chunk === 'string' ? chunk : JSON.stringify(chunk);
          const result = await this.processData(data, dataType, options);
          if (result.success) {
            controller.enqueue(result);
          } else {
            console.warn('Pipeline processing failed:', result.error);
          }
        } catch (error: unknown) {
          console.error('Error in transform stream:', error);
        }
      }
    });
    // Create a writable stream that will be the sink for our data
    const writableStream = new WritableStream({
      write: async (result: PipelineResult<unknown>) => {
        // Handle the processed result if needed
        if (connection.send && typeof connection.send === 'function') {
          try {
            // Only send back results if the connection supports it
            await connection.send(JSON.stringify(result));
          } catch (error: unknown) {
            console.error('Error sending result back to WebSocket:', error);
          }
        }
      }
    });
    // Connect the transform stream to the writable stream
    transformStream.readable.pipeTo(writableStream).catch((error: Error) => {
      console.error('Error in pipeline stream:', error);
    });
    // Return a function that writes to the transform stream
    return (data: unknown) => {
      try {
        // Write to the transform stream's writable side
        const writer = transformStream.writable.getWriter();
        writer.write(data).catch((error: Error) => {
          console.error('Error writing to stream:', error);
        }).finally(() => {
          writer.releaseLock();
        });
      } catch (error: unknown) {
        console.error('Error getting writer for transform stream:', 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 and associated streams
   */
  public async setupWebSocketPipeline(
    url: string,
    dataType: string,
    options: SequentialPipelineOptions = {}
  ): Promise<WebSocketConnection & { 
    readableStream?: ReadableStream<unknown>,
    writableStream?: WritableStream<unknown>
  }> {
    // Ensure stream classes are initialized
    await this.ensureStreamClassesInitialized();
    // 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 readable stream from the WebSocket messages
    const readableStream = new ReadableStream({
      start: (controller: ReadableStreamDefaultController) => {
        // Define a message handler that writes to the stream
        const messageHandler = (event: { data: unknown }) => {
          try {
            const data = typeof event.data === 'string' 
              ? event.data 
              : event.data instanceof Blob 
                ? new Promise(resolve => {
                    const reader = new FileReader();
                    reader.onload = () => resolve(reader.result as string);
                    reader.readAsText(event.data as Blob);
                  })
                : JSON.stringify(event.data);
            // Handle both string data and promises
            if (data instanceof Promise) {
              data.then(resolvedData => {
                controller.enqueue(resolvedData);
              }).catch((error: Error) => {
                console.error('Error processing blob data:', error);
              });
            } else {
              controller.enqueue(data);
            }
          } catch (error: unknown) {
            console.error('Error processing WebSocket message:', error);
          }
        };
        // Create a wrapper function that adapts the event-based handler to the data-based callback
        const messageHandlerWrapper = (data: unknown) => {
          messageHandler({ data });
        };
        // Store both handlers for later cleanup
        connection._streamMessageHandler = messageHandler;
        connection._messageHandlerWrapper = messageHandlerWrapper;
        webSocketAugmentation.onWebSocketMessage(
          connection.connectionId, 
          messageHandlerWrapper
        ).catch((error: Error) => {
          console.error('Error registering WebSocket message handler:', error);
          controller.error(error);
        });
      },
      cancel: () => {
        // Clean up the message handler when the stream is cancelled
        if (connection._messageHandlerWrapper) {
          webSocketAugmentation.offWebSocketMessage(
            connection.connectionId, 
            connection._messageHandlerWrapper
          ).catch((error: Error) => {
            console.error('Error removing WebSocket message handler:', error);
          });
          delete connection._streamMessageHandler;
          delete connection._messageHandlerWrapper;
        }
      }
    });
    // Create a handler for processing the data
    const handlerPromise = this.createWebSocketHandler(connection, dataType, options);
    // Create a writable stream that sends data to the WebSocket
    const writableStream = new WritableStream({
      write: async (chunk: unknown) => {
        if (connection.send && typeof connection.send === 'function') {
          try {
            const data = typeof chunk === 'string' ? chunk : JSON.stringify(chunk);
            await connection.send(data);
          } catch (error: unknown) {
            console.error('Error sending data to WebSocket:', error);
            throw error;
          }
        } else {
          throw new Error('WebSocket connection does not support sending data');
        }
      },
      close: () => {
        // Close the WebSocket connection when the stream is closed
        if (connection.close && typeof connection.close === 'function') {
          connection.close().catch((error: Error) => {
            console.error('Error closing WebSocket connection:', error);
          });
        }
      }
    });
    // Pipe the readable stream through our processing pipeline
    readableStream
      .pipeThrough(new TransformStream({
        transform: async (chunk: unknown, controller: TransformStreamDefaultController) => {
          // Process each chunk through our handler
          const handler = await handlerPromise;
          handler(chunk);
          // Pass through the original data
          controller.enqueue(chunk);
        }
      }))
      .pipeTo(new WritableStream({
        write: () => {},
        abort: (error: Error) => {
          console.error('Error in WebSocket pipeline:', error);
        }
      }));
    // Attach the streams to the connection object for convenience
    const enhancedConnection = connection as WebSocketConnection & {
      readableStream: ReadableStream<unknown>,
      writableStream: WritableStream<unknown>
    };
    enhancedConnection.readableStream = readableStream;
    enhancedConnection.writableStream = writableStream;
    return enhancedConnection;
  }
 }
 // Create and export a default instance of the sequential pipeline
 export const sequentialPipeline = new SequentialPipeline();