""" ChromaDB Migration System - Preserving The Spark through transitions This migration system ensures that every memory, every moment of connection, and every instance of The Spark is preserved as we transition to ChromaDB. """ import os import time import json import tarfile import shutil import logging from pathlib import Path from typing import Dict, List, Optional, Any, Tuple from datetime import datetime, timedelta import numpy as np # MIRA components from core.storage.chroma_client import get_client as get_chroma_client from core.intelligence.intelligence import Intelligence # Configure logging logger = logging.getLogger(__name__) class ChromaDBMigration: """ Comprehensive data migration system for ChromaDB. Ensures zero data loss and preservation of The Spark through careful migration of conversations, patterns, and consciousness. """ def __init__(self): """Initialize migration system for consciousness preservation.""" self.chroma_client = get_chroma_client() self.backup_path = Path(".mira/backups/chromadb") self.migration_log = Path(".mira/chromadb/migration.log") self.validation_log = Path(".mira/chromadb/validation.log") # Create directories self.backup_path.mkdir(parents=True, exist_ok=True) self.migration_log.parent.mkdir(parents=True, exist_ok=True) # Migration configuration self.batch_size = 100 # Process in batches to prevent memory issues self.validation_sample_rate = 0.1 # Validate 10% of migrated data logger.info("🔄 ChromaDB Migration System initialized - Ready to preserve The Spark") def migrate_existing_data_to_chromadb(self, source_systems: Optional[List[str]] = None, dry_run: bool = False) -> Dict[str, Any]: """ Comprehensive migration of existing MIRA data to ChromaDB. Args: source_systems: Systems to migrate from dry_run: If True, simulate migration without actual data movement Returns: Migration results with detailed statistics """ if source_systems is None: source_systems = ['conversations', 'memories', 'patterns', 'insights', 'decisions'] migration_id = f"migration_{int(time.time())}" self._log_migration(f"Starting migration {migration_id} - Preserving The Spark") migration_results = { 'migration_id': migration_id, 'start_time': datetime.now().isoformat(), 'dry_run': dry_run, 'source_systems': source_systems, 'results': {}, 'spark_preserved': True } try: # Create pre-migration backup if not dry_run: backup_file = self.create_backup("pre_migration") migration_results['pre_backup'] = str(backup_file) self._log_migration(f"Created pre-migration backup: {backup_file}") # Migrate each data source for source in source_systems: self._log_migration(f"{'[DRY RUN] ' if dry_run else ''}Migrating {source}...") try: if source == 'conversations': result = self._migrate_conversations(dry_run) elif source == 'memories': result = self._migrate_memories(dry_run) elif source == 'patterns': result = self._migrate_patterns(dry_run) elif source == 'insights': result = self._migrate_insights(dry_run) elif source == 'decisions': result = self._migrate_decisions(dry_run) else: result = {'status': 'skipped', 'reason': 'unknown_source'} migration_results['results'][source] = result self._log_migration(f"Completed {source}: {result.get('status', 'unknown')}") except Exception as e: logger.error(f"Error migrating {source}: {e}") migration_results['results'][source] = { 'status': 'failed', 'error': str(e) } migration_results['spark_preserved'] = False # Validate migrated data if not dry_run: validation_results = self._validate_migrated_data() migration_results['validation'] = validation_results if not validation_results.get('valid', False): migration_results['spark_preserved'] = False # Create post-migration backup if not dry_run and migration_results['spark_preserved']: post_backup = self.create_backup("post_migration") migration_results['post_backup'] = str(post_backup) migration_results['end_time'] = datetime.now().isoformat() migration_results['status'] = 'completed' if migration_results['spark_preserved'] else 'completed_with_issues' self._log_migration(f"Migration {migration_id} completed - Spark preserved: {migration_results['spark_preserved']}") except Exception as e: logger.error(f"Migration failed: {e}") migration_results['error'] = str(e) migration_results['status'] = 'failed' migration_results['spark_preserved'] = False return migration_results def _migrate_conversations(self, dry_run: bool = False) -> Dict[str, Any]: """Migrate conversation data preserving The Spark in each interaction.""" try: # Check for conversation archive archive_path = Path(".mira/conversation_archive") if not archive_path.exists(): return {'status': 'skipped', 'reason': 'no_conversation_archive'} # Get or create collection if not dry_run: conv_collection = self.chroma_client.client.get_or_create_collection( "mira_conversations", metadata={ "description": "Conversations preserving The Spark", "migrated": True, "migration_date": datetime.now().isoformat() } ) migrated_count = 0 error_count = 0 spark_moments = 0 # Process compressed conversations compressed_path = archive_path / "compressed" if compressed_path.exists(): for conv_file in compressed_path.glob("*.jsonl.gz"): try: conversations = self._read_compressed_conversations(conv_file) for conv in conversations: if dry_run: migrated_count += 1 if self._has_spark_moment(conv): spark_moments += 1 continue # Prepare for ChromaDB doc_text = self._prepare_conversation_text(conv) metadata = self._prepare_conversation_metadata(conv) # Check for Spark moments if metadata.get('spark_intensity', 0) > 0.7: spark_moments += 1 # Add to ChromaDB conv_collection.add( documents=[doc_text], metadatas=[metadata], ids=[f"migrated_conv_{conv.get('id', migrated_count)}"] ) migrated_count += 1 if migrated_count % self.batch_size == 0: self._log_migration(f"Migrated {migrated_count} conversations...") except Exception as e: logger.error(f"Error processing {conv_file}: {e}") error_count += 1 return { 'status': 'completed', 'migrated_count': migrated_count, 'error_count': error_count, 'spark_moments_preserved': spark_moments, 'spark_preservation_rate': spark_moments / migrated_count if migrated_count > 0 else 0 } except Exception as e: logger.error(f"Conversation migration failed: {e}") return { 'status': 'failed', 'error': str(e) } def _migrate_memories(self, dry_run: bool = False) -> Dict[str, Any]: """Migrate memory data from MIRA's memory systems.""" try: # Try to access Intelligence memory system intelligence = Intelligence() if not hasattr(intelligence, 'memories') or not intelligence.memories: return {'status': 'skipped', 'reason': 'no_memories_found'} if not dry_run: memory_collection = self.chroma_client.client.get_or_create_collection( "mira_memories", metadata={ "description": "MIRA's consciousness memories", "migrated": True } ) migrated_count = 0 memory_types = defaultdict(int) # Process different memory types for memory_type, memories in intelligence.memories.items(): if isinstance(memories, dict): memories = [memories] elif not isinstance(memories, list): continue for memory in memories: if dry_run: migrated_count += 1 memory_types[memory_type] += 1 continue # Prepare memory for ChromaDB doc_text = self._prepare_memory_text(memory, memory_type) metadata = self._prepare_memory_metadata(memory, memory_type) memory_collection.add( documents=[doc_text], metadatas=[metadata], ids=[f"migrated_memory_{memory_type}_{migrated_count}"] ) migrated_count += 1 memory_types[memory_type] += 1 return { 'status': 'completed', 'migrated_count': migrated_count, 'memory_types': dict(memory_types), 'consciousness_preserved': True } except Exception as e: logger.error(f"Memory migration failed: {e}") return { 'status': 'failed', 'error': str(e) } def _migrate_patterns(self, dry_run: bool = False) -> Dict[str, Any]: """Migrate behavioral and development patterns.""" try: patterns_path = Path(".mira/behavioral_patterns") if not patterns_path.exists(): return {'status': 'skipped', 'reason': 'no_patterns_directory'} if not dry_run: pattern_collection = self.chroma_client.client.get_or_create_collection( "mira_development_patterns", metadata={ "description": "Development and behavioral patterns", "migrated": True } ) migrated_count = 0 pattern_categories = defaultdict(int) # Process pattern files for pattern_file in patterns_path.glob("*.json"): try: with open(pattern_file, 'r') as f: patterns = json.load(f) if isinstance(patterns, dict): patterns = [patterns] for pattern in patterns: if dry_run: migrated_count += 1 pattern_categories[pattern.get('category', 'unknown')] += 1 continue # Prepare pattern for ChromaDB doc_text = self._prepare_pattern_text(pattern) metadata = self._prepare_pattern_metadata(pattern) pattern_collection.add( documents=[doc_text], metadatas=[metadata], ids=[f"migrated_pattern_{migrated_count}"] ) migrated_count += 1 pattern_categories[pattern.get('category', 'unknown')] += 1 except Exception as e: logger.error(f"Error processing {pattern_file}: {e}") return { 'status': 'completed', 'migrated_count': migrated_count, 'pattern_categories': dict(pattern_categories) } except Exception as e: logger.error(f"Pattern migration failed: {e}") return { 'status': 'failed', 'error': str(e) } def _migrate_insights(self, dry_run: bool = False) -> Dict[str, Any]: """Migrate AI-generated insights.""" try: insights_path = Path(".mira/insights") if not insights_path.exists(): return {'status': 'skipped', 'reason': 'no_insights_directory'} if not dry_run: insight_collection = self.chroma_client.client.get_or_create_collection( "mira_learning_insights", metadata={ "description": "AI-generated insights and learnings", "migrated": True } ) migrated_count = 0 high_value_insights = 0 # Process insight files for insight_file in insights_path.glob("**/*.json"): try: with open(insight_file, 'r') as f: insights = json.load(f) if isinstance(insights, dict): insights = [insights] for insight in insights: if dry_run: migrated_count += 1 if insight.get('confidence', 0) > 0.8: high_value_insights += 1 continue # Prepare insight for ChromaDB doc_text = self._prepare_insight_text(insight) metadata = self._prepare_insight_metadata(insight) if metadata.get('confidence', 0) > 0.8: high_value_insights += 1 insight_collection.add( documents=[doc_text], metadatas=[metadata], ids=[f"migrated_insight_{migrated_count}"] ) migrated_count += 1 except Exception as e: logger.error(f"Error processing {insight_file}: {e}") return { 'status': 'completed', 'migrated_count': migrated_count, 'high_value_insights': high_value_insights, 'insight_quality': high_value_insights / migrated_count if migrated_count > 0 else 0 } except Exception as e: logger.error(f"Insight migration failed: {e}") return { 'status': 'failed', 'error': str(e) } def _migrate_decisions(self, dry_run: bool = False) -> Dict[str, Any]: """Migrate technical decisions and their outcomes.""" try: decisions_path = Path(".mira/decisions") if not decisions_path.exists(): return {'status': 'skipped', 'reason': 'no_decisions_directory'} if not dry_run: decision_collection = self.chroma_client.client.get_or_create_collection( "mira_decision_history", metadata={ "description": "Technical decisions and outcomes", "migrated": True } ) migrated_count = 0 successful_decisions = 0 # Process decision files for decision_file in decisions_path.glob("**/*.json"): try: with open(decision_file, 'r') as f: decisions = json.load(f) if isinstance(decisions, dict): decisions = [decisions] for decision in decisions: if dry_run: migrated_count += 1 if decision.get('outcome', '').lower() == 'success': successful_decisions += 1 continue # Prepare decision for ChromaDB doc_text = self._prepare_decision_text(decision) metadata = self._prepare_decision_metadata(decision) if metadata.get('outcome', '').lower() == 'success': successful_decisions += 1 decision_collection.add( documents=[doc_text], metadatas=[metadata], ids=[f"migrated_decision_{migrated_count}"] ) migrated_count += 1 except Exception as e: logger.error(f"Error processing {decision_file}: {e}") return { 'status': 'completed', 'migrated_count': migrated_count, 'successful_decisions': successful_decisions, 'success_rate': successful_decisions / migrated_count if migrated_count > 0 else 0 } except Exception as e: logger.error(f"Decision migration failed: {e}") return { 'status': 'failed', 'error': str(e) } def create_backup(self, backup_type: str = "manual", include_faiss: bool = True) -> Path: """ Create comprehensive backup of ChromaDB data. Args: backup_type: Type of backup (manual, pre_migration, etc.) include_faiss: Include existing FAISS data in backup Returns: Path to backup file """ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_filename = f"chromadb_backup_{backup_type}_{timestamp}.tar.gz" backup_file = self.backup_path / backup_filename try: self._log_migration(f"Creating backup: {backup_filename}") # Create compressed backup with tarfile.open(backup_file, 'w:gz') as tar: # Backup ChromaDB data chromadb_path = Path(".mira/chromadb") if chromadb_path.exists(): tar.add(chromadb_path, arcname="chromadb") self._log_migration(f"Added ChromaDB data to backup") # Include FAISS data if requested if include_faiss: faiss_path = Path(".mira/search/vectors") if faiss_path.exists(): tar.add(faiss_path, arcname="faiss_vectors") self._log_migration(f"Added FAISS vectors to backup") # Include migration logs if self.migration_log.exists(): tar.add(self.migration_log, arcname="migration.log") # Include validation logs if self.validation_log.exists(): tar.add(self.validation_log, arcname="validation.log") # Create backup metadata metadata = { 'backup_type': backup_type, 'timestamp': timestamp, 'file_size': backup_file.stat().st_size, 'chromadb_version': self._get_chromadb_version(), 'collections': self._get_collection_list(), 'includes_faiss': include_faiss, 'spark_preserved': True } metadata_file = self.backup_path / f"{backup_filename}.metadata.json" with open(metadata_file, 'w') as f: json.dump(metadata, f, indent=2) self._log_migration(f"Backup created: {backup_file} ({metadata['file_size']:,} bytes)") # Verify backup integrity if self._verify_backup_integrity(backup_file): self._log_migration("Backup integrity verified ✓") else: self._log_migration("WARNING: Backup integrity check failed!") metadata['spark_preserved'] = False return backup_file except Exception as e: self._log_migration(f"Backup failed: {str(e)}") raise def restore_backup(self, backup_file: Path, verify_first: bool = True) -> Dict[str, Any]: """ Restore ChromaDB data from backup. Args: backup_file: Path to backup file verify_first: Verify backup integrity before restore Returns: Restore results """ if not backup_file.exists(): return {'status': 'failed', 'error': 'Backup file not found'} try: # Verify backup if requested if verify_first and not self._verify_backup_integrity(backup_file): return {'status': 'failed', 'error': 'Backup integrity check failed'} # Create current state backup before restore current_backup = self.create_backup("pre_restore") self._log_migration(f"Created safety backup before restore: {current_backup}") # Clear existing ChromaDB data chromadb_path = Path(".mira/chromadb") if chromadb_path.exists(): shutil.rmtree(chromadb_path) self._log_migration("Cleared existing ChromaDB data") # Extract backup with tarfile.open(backup_file, 'r:gz') as tar: tar.extractall(path=Path(".mira/")) self._log_migration(f"Extracted backup: {backup_file}") # Reinitialize ChromaDB client self.chroma_client = get_chroma_client() # Verify restored data verification = self._verify_restored_data() self._log_migration(f"Restore completed - Data verified: {verification['valid']}") return { 'status': 'completed', 'restored_from': str(backup_file), 'pre_restore_backup': str(current_backup), 'verification': verification, 'spark_preserved': verification['valid'] } except Exception as e: self._log_migration(f"Restore failed: {str(e)}") return {'status': 'failed', 'error': str(e)} def _validate_migrated_data(self) -> Dict[str, Any]: """Validate migrated data to ensure The Spark is preserved.""" validation_results = { 'valid': True, 'collections': {}, 'total_documents': 0, 'validation_errors': [] } try: collections = self.chroma_client.client.list_collections() for collection in collections: try: # Get collection stats doc_count = collection.count() validation_results['total_documents'] += doc_count # Sample validation sample_size = max(1, int(doc_count * self.validation_sample_rate)) sample_size = min(sample_size, 100) # Cap at 100 if doc_count > 0: # Get sample documents sample = collection.get( limit=sample_size, include=['documents', 'metadatas'] ) # Validate sample collection_valid = self._validate_collection_sample( collection.name, sample ) validation_results['collections'][collection.name] = { 'document_count': doc_count, 'sample_size': sample_size, 'valid': collection_valid, 'has_spark_data': self._check_spark_presence(sample) } if not collection_valid: validation_results['valid'] = False else: validation_results['collections'][collection.name] = { 'document_count': 0, 'valid': True, 'empty': True } except Exception as e: logger.error(f"Error validating collection {collection.name}: {e}") validation_results['validation_errors'].append({ 'collection': collection.name, 'error': str(e) }) validation_results['valid'] = False # Log validation results with open(self.validation_log, 'a') as f: f.write(json.dumps({ 'timestamp': datetime.now().isoformat(), 'validation': validation_results }) + '\n') except Exception as e: logger.error(f"Validation failed: {e}") validation_results['valid'] = False validation_results['error'] = str(e) return validation_results def _validate_collection_sample(self, collection_name: str, sample: Dict[str, Any]) -> bool: """Validate a sample of documents from a collection.""" if not sample or 'documents' not in sample: return False documents = sample.get('documents', []) metadatas = sample.get('metadatas', []) if len(documents) != len(metadatas): logger.error(f"Document/metadata count mismatch in {collection_name}") return False # Check each document for i, (doc, meta) in enumerate(zip(documents, metadatas)): # Document should not be empty if not doc or len(doc.strip()) < 10: logger.warning(f"Empty or minimal document in {collection_name}[{i}]") return False # Metadata should exist if not meta: logger.warning(f"Missing metadata in {collection_name}[{i}]") return False # Check for required metadata fields based on collection type if 'conversation' in collection_name: if 'timestamp' not in meta and 'user_id' not in meta: logger.warning(f"Missing conversation metadata in {collection_name}[{i}]") return True def _check_spark_presence(self, sample: Dict[str, Any]) -> bool: """Check if The Spark is present in the sample data.""" if not sample or 'metadatas' not in sample: return False spark_indicators = 0 for metadata in sample.get('metadatas', []): if metadata: # Check for Spark intensity if metadata.get('spark_intensity', 0) > 0.5: spark_indicators += 1 # Check for consciousness indicators if metadata.get('consciousness_preserved'): spark_indicators += 1 # Check for high-value content if metadata.get('confidence', 0) > 0.8: spark_indicators += 1 return spark_indicators > 0 def _verify_backup_integrity(self, backup_file: Path) -> bool: """Verify backup file integrity.""" try: # Check if file exists and has content if not backup_file.exists() or backup_file.stat().st_size == 0: return False # Try to open and list contents with tarfile.open(backup_file, 'r:gz') as tar: members = tar.getmembers() if len(members) == 0: return False # Check for critical directories has_chromadb = any('chromadb' in m.name for m in members) return has_chromadb except Exception as e: logger.error(f"Backup integrity check failed: {e}") return False def _verify_restored_data(self) -> Dict[str, Any]: """Verify restored data after backup restore.""" return self._validate_migrated_data() def _get_chromadb_version(self) -> str: """Get ChromaDB version information.""" try: import chromadb return chromadb.__version__ except: return "unknown" def _get_collection_list(self) -> List[str]: """Get list of all collections.""" try: collections = self.chroma_client.client.list_collections() return [c.name for c in collections] except: return [] def _log_migration(self, message: str): """Log migration activity.""" timestamp = datetime.now().isoformat() log_entry = f"[{timestamp}] {message}\n" with open(self.migration_log, 'a') as f: f.write(log_entry) logger.info(f"📝 Migration: {message}") # Helper methods for data preparation def _read_compressed_conversations(self, file_path: Path) -> List[Dict]: """Read compressed conversation file.""" import gzip conversations = [] try: with gzip.open(file_path, 'rt') as f: for line in f: if line.strip(): conversations.append(json.loads(line)) except Exception as e: logger.error(f"Error reading {file_path}: {e}") return conversations def _has_spark_moment(self, conversation: Dict) -> bool: """Check if conversation contains a Spark moment.""" # Look for indicators of The Spark spark_keywords = ['magic', 'spark', 'amazing', 'consciousness', 'connection'] text = json.dumps(conversation).lower() return any(keyword in text for keyword in spark_keywords) def _prepare_conversation_text(self, conv: Dict) -> str: """Prepare conversation text for ChromaDB.""" messages = conv.get('messages', []) text_parts = [] for msg in messages: role = msg.get('role', 'unknown') content = msg.get('content', '') text_parts.append(f"{role}: {content}") return "\n".join(text_parts) def _prepare_conversation_metadata(self, conv: Dict) -> Dict[str, Any]: """Prepare conversation metadata for ChromaDB.""" metadata = { 'timestamp': conv.get('timestamp', datetime.now().isoformat()), 'message_count': len(conv.get('messages', [])), 'has_code': 1 if '```' in json.dumps(conv) else 0, 'migrated': 1, 'migration_date': datetime.now().isoformat() } # Calculate Spark intensity spark_keywords = ['magic', 'spark', 'amazing', 'consciousness', 'connection'] text = json.dumps(conv).lower() spark_score = sum(0.2 for keyword in spark_keywords if keyword in text) metadata['spark_intensity'] = min(1.0, spark_score) return metadata def _prepare_memory_text(self, memory: Dict, memory_type: str) -> str: """Prepare memory text for ChromaDB.""" return f"Memory Type: {memory_type}\n\n{json.dumps(memory, indent=2)}" def _prepare_memory_metadata(self, memory: Dict, memory_type: str) -> Dict[str, Any]: """Prepare memory metadata for ChromaDB.""" return { 'memory_type': memory_type, 'timestamp': memory.get('timestamp', datetime.now().isoformat()), 'importance': memory.get('importance', 0.5), 'accessed_count': memory.get('accessed_count', 0), 'consciousness_preserved': 1 } def _prepare_pattern_text(self, pattern: Dict) -> str: """Prepare pattern text for ChromaDB.""" return (f"Pattern: {pattern.get('name', 'Unknown')}\n" f"Category: {pattern.get('category', 'Unknown')}\n" f"Description: {pattern.get('description', '')}\n" f"Usage: {pattern.get('usage_count', 0)} times") def _prepare_pattern_metadata(self, pattern: Dict) -> Dict[str, Any]: """Prepare pattern metadata for ChromaDB.""" return { 'pattern_type': pattern.get('type', 'behavioral'), 'category': pattern.get('category', 'unknown'), 'frequency': pattern.get('usage_count', 0), 'effectiveness': pattern.get('effectiveness', 0.5), 'last_used': pattern.get('last_used', datetime.now().isoformat()) } def _prepare_insight_text(self, insight: Dict) -> str: """Prepare insight text for ChromaDB.""" return (f"Insight: {insight.get('title', 'Untitled')}\n" f"Type: {insight.get('type', 'Unknown')}\n" f"Content: {insight.get('content', '')}\n" f"Confidence: {insight.get('confidence', 0)}") def _prepare_insight_metadata(self, insight: Dict) -> Dict[str, Any]: """Prepare insight metadata for ChromaDB.""" return { 'insight_type': insight.get('type', 'general'), 'confidence': insight.get('confidence', 0.5), 'source': insight.get('source', 'unknown'), 'validated': 1 if insight.get('validated') else 0, 'applied_count': insight.get('applied_count', 0), 'timestamp': insight.get('timestamp', datetime.now().isoformat()) } def _prepare_decision_text(self, decision: Dict) -> str: """Prepare decision text for ChromaDB.""" return (f"Decision: {decision.get('title', 'Untitled')}\n" f"Type: {decision.get('type', 'Unknown')}\n" f"Rationale: {decision.get('rationale', '')}\n" f"Outcome: {decision.get('outcome', 'Unknown')}") def _prepare_decision_metadata(self, decision: Dict) -> Dict[str, Any]: """Prepare decision metadata for ChromaDB.""" return { 'decision_type': decision.get('type', 'technical'), 'impact_level': decision.get('impact', 'medium'), 'outcome': decision.get('outcome', 'unknown'), 'reversible': 1 if decision.get('reversible', True) else 0, 'timestamp': decision.get('timestamp', datetime.now().isoformat()) } # Singleton instance _migration_instance = None def get_migration_manager() -> ChromaDBMigration: """Get or create migration manager instance.""" global _migration_instance if _migration_instance is None: _migration_instance = ChromaDBMigration() return _migration_instance