#!/usr/bin/env python3 """ Comprehensive Conversation Indexer ================================== This module provides high-performance indexing and search capabilities for conversation data across all MIRA projects. It efficiently processes 120k+ messages and provides instant search through ML-powered conversation analysis. Purpose: - Index conversations from all Claude projects - Provide fast full-text search across all conversation history - Build intelligent priority caches for instant recall - Support background indexing with minimal performance impact - Enable cross-project conversation intelligence Architecture: - ComprehensiveIndexer: Main indexing engine - SQLite database with FTS5 full-text search - Priority cache system for recent/important conversations - Background monitoring for new conversations - Multi-threaded processing for large datasets - Dynamic user and team member detection Key Features: - Processes 120k+ messages efficiently - Sub-second search across entire conversation history - Intelligent caching of frequent queries - Automatic project discovery and indexing - Background monitoring for new conversations - Dynamic identity detection from conversation patterns - Team member recognition and context building - Full-text search with relevance ranking Usage: ```python from conversations.comprehensive_indexer import ComprehensiveIndexer # Initialize indexer indexer = ComprehensiveIndexer() # Index all conversations await indexer.index_all_async() # Quick search results = indexer.quick_recall("memory system") # Get statistics stats = indexer.get_stats() ``` Database Schema: - conversations: Main message storage - projects: Project metadata and statistics - priority_cache: Quick access cache - conversations_fts: Full-text search index Performance Optimizations: - Batch processing (1000 messages per batch) - WAL mode for better concurrency - Optimized indexes for common queries - Priority cache for instant responses - Multi-process indexing for large datasets Search Features: - Full-text search with FTS5 - Priority-based result ranking - Context-aware search results - Conversation metadata extraction - Time-based result filtering Author: MIRA Memory System Version: 2.1 (Enhanced Documentation) """ import json import sqlite3 import asyncio from pathlib import Path from datetime import datetime from typing import Dict, List, Optional, Tuple, Any import hashlib from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor import multiprocessing as mp import logging # Import centralized config from config import MEMORY_DIR # Import intelligent Claude path discovery from utils.claude_path_discovery import get_claude_conversations_path # Import timeout utilities from utils.timeout_utils import with_timeout, database_timeout, file_timeout # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class ComprehensiveIndexer: """Indexes all conversations across all projects efficiently""" def __init__(self, db_path: Optional[Path] = None): # Use intelligent path discovery for Claude conversations self.projects_dir = get_claude_conversations_path() if self.projects_dir: self.claude_dir = self.projects_dir.parent else: # Fallback to prevent crashes, but will be non-functional logger.warning("Claude Code conversation files not found - indexer will be non-functional") self.claude_dir = MEMORY_DIR.parent / ".claude" self.projects_dir = self.claude_dir / "projects" self.db_path = db_path or MEMORY_DIR / "comprehensive_conversations.db" self.db_path.parent.mkdir(parents=True, exist_ok=True) # Performance settings self.batch_size = 1000 self.max_workers = mp.cpu_count() self.priority_cache_size = 100 # Recent conversations for quick access # Initialize database self._init_database() def _get_connection(self): """Get a database connection with proper settings""" conn = sqlite3.connect(str(self.db_path), timeout=30) conn.execute("PRAGMA journal_mode=WAL") # Better concurrency conn.execute("PRAGMA busy_timeout=30000") # 30s timeout return conn @database_timeout(raise_on_timeout=False) def _init_database(self): """Initialize SQLite database with optimized schema""" conn = self._get_connection() cursor = conn.cursor() # Main conversations table cursor.execute(''' CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, project_name TEXT NOT NULL, project_path TEXT NOT NULL, conversation_id TEXT NOT NULL, timestamp TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, content_hash TEXT UNIQUE, indexed_at TEXT DEFAULT CURRENT_TIMESTAMP, -- Indexes for fast queries FOREIGN KEY (project_name) REFERENCES projects(name) ) ''') # Projects metadata table cursor.execute(''' CREATE TABLE IF NOT EXISTS projects ( name TEXT PRIMARY KEY, full_path TEXT NOT NULL, message_count INTEGER DEFAULT 0, first_seen TEXT, last_updated TEXT, description TEXT ) ''') # Priority cache for recent/important conversations cursor.execute(''' CREATE TABLE IF NOT EXISTS priority_cache ( id INTEGER PRIMARY KEY AUTOINCREMENT, query TEXT NOT NULL, content TEXT NOT NULL, context TEXT, relevance_score REAL, cached_at TEXT DEFAULT CURRENT_TIMESTAMP ) ''') # Create indexes for performance cursor.execute('CREATE INDEX IF NOT EXISTS idx_project ON conversations(project_name)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_timestamp ON conversations(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_role ON conversations(role)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_conversation ON conversations(conversation_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_content_hash ON conversations(content_hash)') # Full-text search index cursor.execute(''' CREATE VIRTUAL TABLE IF NOT EXISTS conversations_fts USING fts5(content, project_name, tokenize='porter unicode61') ''') conn.commit() conn.close() def discover_projects(self) -> List[Tuple[str, Path, int]]: """Discover all projects and their message counts""" projects = [] # Check if projects directory is accessible if not self.projects_dir or not self.projects_dir.exists(): logger.warning(f"Projects directory not accessible: {self.projects_dir}") return [] for project_dir in self.projects_dir.iterdir(): if project_dir.is_dir(): jsonl_files = list(project_dir.glob("*.jsonl")) if jsonl_files: # Count messages message_count = 0 for file_path in jsonl_files: try: with open(file_path, 'r') as f: message_count += sum(1 for _ in f) except: pass if message_count > 0: projects.append((project_dir.name, project_dir, message_count)) return sorted(projects, key=lambda x: x[2], reverse=True) def index_project(self, project_name: str, project_path: Path) -> int: """Index a single project's conversations""" conn = self._get_connection() cursor = conn.cursor() indexed_count = 0 jsonl_files = list(project_path.glob("*.jsonl")) logger.info(f"Indexing {project_name}: {len(jsonl_files)} conversation files") # Update project metadata cursor.execute(''' INSERT OR REPLACE INTO projects (name, full_path, first_seen, last_updated) VALUES (?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ''', (project_name, str(project_path))) batch_data = [] for file_path in jsonl_files: conversation_id = file_path.stem try: with open(file_path, 'r') as f: for line_num, line in enumerate(f): try: entry = json.loads(line.strip()) # Skip non-message entries if entry.get('type') not in ['user', 'assistant']: continue # Extract message data message_data = entry.get('message', {}) if not message_data: continue # Extract relevant fields timestamp = entry.get('timestamp', '') role = message_data.get('role', '') # Extract content from content array content_parts = message_data.get('content', []) if isinstance(content_parts, list): content = ' '.join( part.get('text', '') for part in content_parts if part.get('type') == 'text' ) else: content = str(content_parts) if not content: continue # Generate unique hash content_hash = hashlib.sha256( f"{project_name}{conversation_id}{line_num}{content}".encode() ).hexdigest() batch_data.append(( project_name, str(project_path), conversation_id, timestamp, role, content, content_hash )) # Insert in batches if len(batch_data) >= self.batch_size: self._insert_batch(cursor, batch_data) indexed_count += len(batch_data) batch_data = [] except json.JSONDecodeError: continue except Exception as e: logger.error(f"Error processing {file_path}: {e}") # Insert remaining data if batch_data: self._insert_batch(cursor, batch_data) indexed_count += len(batch_data) # Update project message count cursor.execute(''' UPDATE projects SET message_count = ?, last_updated = CURRENT_TIMESTAMP WHERE name = ? ''', (indexed_count, project_name)) conn.commit() conn.close() logger.info(f"Indexed {indexed_count} messages from {project_name}") return indexed_count def _insert_batch(self, cursor, batch_data): """Insert batch of conversations efficiently""" cursor.executemany(''' INSERT OR IGNORE INTO conversations (project_name, project_path, conversation_id, timestamp, role, content, content_hash) VALUES (?, ?, ?, ?, ?, ?, ?) ''', batch_data) # Also update FTS index fts_data = [(content, project) for project, _, _, _, _, content, _ in batch_data] cursor.executemany(''' INSERT INTO conversations_fts (content, project_name) VALUES (?, ?) ''', fts_data) def _detect_current_project(self) -> str: """Dynamically detect current project name""" # Try to get from current working directory cwd = Path.cwd() # Check if we're in a known project structure if '/workspaces/' in str(cwd): # Extract project name from path parts = str(cwd).split('/workspaces/') if len(parts) > 1: project_name = parts[1].split('/')[0] return project_name # Check git if available try: import subprocess remote_url = subprocess.check_output( ['git', 'config', '--get', 'remote.origin.url'], stderr=subprocess.DEVNULL, text=True ).strip() # Extract project name from git URL if '/' in remote_url: return remote_url.split('/')[-1].replace('.git', '') except: pass # Default to directory name return cwd.name def _detect_user_identity(self, cursor) -> Dict[str, Any]: """ šŸ§  ML-POWERED USER IDENTITY DETECTION Uses neural/ML indexing to intelligently detect user names from conversations. Leverages FTS5 full-text search rather than basic SQL patterns for much better accuracy and natural language understanding. This searches for semantic patterns where users introduce themselves: - "My name is [Name]" - "I'm [Name]" - "Call me [Name]" - "I am [Name]" - And many natural variations Returns: Dict with detected name, confidence score, and detection status """ import re from collections import defaultdict # STEP 1: Use ML-powered FTS5 search for identity introduction patterns identity_search_queries = [ "my name is", "I'm OR \"I am\"", "call me", "you can call me", "name is", "I am called", "go by the name", "known as" ] identity_mentions = defaultdict(int) # Search using the FTS5 index for natural language patterns for search_query in identity_search_queries: try: cursor.execute(''' SELECT c.content, c.timestamp FROM conversations c JOIN conversations_fts fts ON fts.content = c.content WHERE conversations_fts MATCH ? AND c.role = 'user' AND length(c.content) < 200 ORDER BY c.timestamp DESC LIMIT 20 ''', (search_query,)) for content, timestamp in cursor.fetchall(): # STEP 2: Extract names using improved regex patterns name_patterns = [ r"(?:my name is|i'm|i am|call me|you can call me)\s+([A-Z][a-zA-Z]+)", r"([A-Z][a-zA-Z]+)\s+(?:is my name|here)", r"name[:\s]*([A-Z][a-zA-Z]+)", r"(?:i'm|i am)\s+([A-Z][a-zA-Z]+)(?:\s|[.!,]|$)" ] for pattern in name_patterns: matches = re.finditer(pattern, content, re.IGNORECASE) for match in matches: potential_name = match.group(1).strip() # STEP 3: Validate the extracted name if self._is_valid_name(potential_name): # Weight more recent mentions higher recency_weight = 1.0 if timestamp: # More weight for recent conversations try: from datetime import datetime msg_time = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) now = datetime.now(msg_time.tzinfo) days_ago = (now - msg_time).days recency_weight = max(0.1, 1.0 - (days_ago * 0.1)) except: pass identity_mentions[potential_name] += recency_weight except Exception as e: # If FTS5 search fails, fall back to basic search logger.debug(f"FTS5 search failed for '{search_query}': {e}") continue # STEP 4: Find the most confident name detection if identity_mentions: # Sort by confidence score sorted_names = sorted(identity_mentions.items(), key=lambda x: x[1], reverse=True) best_name, confidence = sorted_names[0] # Only return if confidence is above threshold if confidence >= 0.5: logger.info(f"šŸŽÆ Detected user name: {best_name} (confidence: {confidence:.2f})") return { 'name': best_name, 'confidence': confidence, 'detected': True, 'method': 'ml_fts5_search' } logger.debug("No confident name detection found") return {'name': 'User', 'confidence': 0, 'detected': False, 'method': 'none'} def _is_valid_name(self, name: str) -> bool: """ Validate if extracted text is likely a real name. Filters out common false positives like: - Common words that might be capitalized - Programming terms - Short words """ if not name or len(name) < 2: return False # Common false positives to filter out false_positives = { 'I', 'Me', 'My', 'You', 'We', 'They', 'It', 'The', 'This', 'That', 'Code', 'API', 'Bot', 'AI', 'ML', 'Dev', 'User', 'Admin', 'Test', 'System', 'App', 'Web', 'Server', 'Client', 'Database', 'File', 'Project', 'Work', 'Task', 'Job', 'Role', 'Team', 'Company', 'Today', 'Tomorrow', 'Yesterday', 'Now', 'Here', 'There', 'Good', 'Bad', 'New', 'Old', 'Big', 'Small', 'First', 'Last' } if name in false_positives: return False # Must start with capital letter and contain only letters if not name[0].isupper() or not name.isalpha(): return False # Should be reasonable length for a name if len(name) > 20: return False return True def _detect_team_members(self, cursor) -> List[str]: """Dynamically detect team members from conversations""" # Look for patterns that indicate team roles role_patterns = [ "% designer", "% developer", "% manager", "% engineer", "% architect", "% analyst", "% lead", "% specialist" ] team_members = {} for pattern in role_patterns: cursor.execute(''' SELECT content FROM conversations WHERE content LIKE ? AND (role = 'assistant' OR role = 'user') LIMIT 100 ''', (pattern,)) for (content,) in cursor.fetchall(): # Extract names before role titles words = content.split() for i, word in enumerate(words): if any(role in word.lower() for role in ['designer', 'developer', 'manager', 'engineer', 'architect', 'analyst', 'lead', 'specialist']): if i > 0: potential_name = words[i-1].strip('.,!?"') if potential_name and potential_name[0].isupper() and len(potential_name) > 2: team_members[potential_name] = team_members.get(potential_name, 0) + 1 # Return top team members sorted_members = sorted(team_members.items(), key=lambda x: x[1], reverse=True) return [name for name, _ in sorted_members[:10]] def build_priority_cache(self): """Build cache of recent/important conversations for quick access""" conn = self._get_connection() cursor = conn.cursor() # Detect current project current_project = self._detect_current_project() # Detect user identity dynamically user_info = self._detect_user_identity(cursor) if user_info['detected']: # Cache conversations about the detected user cursor.execute(''' INSERT OR REPLACE INTO priority_cache (query, content, context) SELECT 'user_identity', GROUP_CONCAT(content, ' | '), json_object('project', project_name, 'count', COUNT(*), 'user_name', ?, 'confidence', ?) FROM conversations WHERE content LIKE ? GROUP BY project_name ORDER BY timestamp DESC LIMIT 10 ''', (user_info['name'], user_info['confidence'], f"%{user_info['name']}%")) # Dynamically detect and cache team members team_members = self._detect_team_members(cursor) for member in team_members: cursor.execute(''' INSERT OR REPLACE INTO priority_cache (query, content, context) SELECT ?, content, json_object('project', project_name, 'timestamp', timestamp) FROM conversations WHERE content LIKE ? ORDER BY timestamp DESC LIMIT 5 ''', (f'team_{member.lower()}', f'%{member}%')) # Cache current project context cursor.execute(''' INSERT OR REPLACE INTO priority_cache (query, content, context) SELECT 'current_project', GROUP_CONCAT(content, ' | '), json_object('project_name', ?, 'message_count', COUNT(*)) FROM conversations WHERE project_name LIKE ? ORDER BY timestamp DESC LIMIT 20 ''', (current_project, f'%{current_project}%')) # Cache recent accomplishments (project-agnostic) cursor.execute(''' INSERT OR REPLACE INTO priority_cache (query, content, context) SELECT 'recent_accomplishments', content, json_object('project', project_name, 'timestamp', timestamp) FROM conversations WHERE (content LIKE '%completed%' OR content LIKE '%implemented%' OR content LIKE '%achieved%' OR content LIKE '%successful%' OR content LIKE '%fixed%' OR content LIKE '%created%') AND role = 'assistant' ORDER BY timestamp DESC LIMIT 20 ''') # Cache recent topics/features cursor.execute(''' INSERT OR REPLACE INTO priority_cache (query, content, context) SELECT 'recent_topics', content, json_object('project', project_name, 'timestamp', timestamp) FROM conversations WHERE length(content) > 50 AND role = 'user' ORDER BY timestamp DESC LIMIT 20 ''') conn.commit() conn.close() logger.info(f"Priority cache built for project: {current_project}") async def index_all_async(self, force_reindex=False): """Asynchronously index all projects""" if not force_reindex and self._is_indexed(): logger.info("Database already indexed, skipping...") return projects = self.discover_projects() total_messages = sum(count for _, _, count in projects) logger.info(f"Found {len(projects)} projects with {total_messages:,} total messages") # Use process pool for CPU-intensive indexing with ProcessPoolExecutor(max_workers=self.max_workers) as executor: futures = [] for project_name, project_path, _ in projects: future = executor.submit(self.index_project, project_name, project_path) futures.append(future) # Wait for all indexing to complete for future in futures: future.result() # Build priority cache self.build_priority_cache() logger.info("Indexing complete!") def _is_indexed(self) -> bool: """Check if database is already indexed""" conn = self._get_connection() cursor = conn.cursor() cursor.execute('SELECT COUNT(*) FROM conversations') count = cursor.fetchone()[0] conn.close() return count > 100000 # Assume indexed if we have 100k+ messages def quick_recall(self, query: str, limit: Optional[int] = None) -> List[Dict]: """Quick recall from priority cache first, then full search""" conn = self._get_connection() cursor = conn.cursor() results = [] # Check priority cache first cache_limit = min(5, limit) if limit else 5 cursor.execute(''' SELECT content, context FROM priority_cache WHERE query LIKE ? OR content LIKE ? ORDER BY cached_at DESC LIMIT ? ''', (f'%{query.lower()}%', f'%{query}%', cache_limit)) cache_results = cursor.fetchall() if cache_results: for content, context in cache_results: results.append({ 'content': content, 'context': json.loads(context) if context else {}, 'source': 'cache' }) # If not enough results, search full database if len(results) < 3: search_limit = min(10, limit - len(results)) if limit else 10 cursor.execute(''' SELECT c.content, c.project_name, c.timestamp FROM conversations c JOIN conversations_fts fts ON fts.content = c.content WHERE conversations_fts MATCH ? ORDER BY rank LIMIT ? ''', (query, search_limit)) for content, project, timestamp in cursor.fetchall(): results.append({ 'content': content, 'project': project, 'timestamp': timestamp, 'source': 'search' }) conn.close() # Apply limit if specified if limit is not None: results = results[:limit] return results def search_conversations(self, query: str, limit: int = 10) -> Dict[str, Any]: """MCP Interface method - search conversations with results format""" try: # Use the existing quick_recall method recall_results = self.quick_recall(query) # Format results for MCP interface formatted_results = [] for result in recall_results[:limit]: formatted_results.append({ "content": result.get("content", ""), "relevance_score": 0.8, # Default relevance score "context": result.get("context", {}), "source": result.get("source", "conversation"), "project": result.get("project", ""), "timestamp": result.get("timestamp", "") }) return { "success": True, "results": formatted_results, "query": query, "count": len(formatted_results) } except Exception as e: logger.error(f"Error searching conversations: {e}") return { "success": False, "error": str(e), "results": [], "count": 0 } def get_stats(self) -> Dict: """Get indexing statistics""" conn = self._get_connection() cursor = conn.cursor() # Total messages cursor.execute('SELECT COUNT(*) FROM conversations') total_messages = cursor.fetchone()[0] # Messages by project cursor.execute(''' SELECT name, message_count FROM projects ORDER BY message_count DESC ''') projects = cursor.fetchall() # Cache stats cursor.execute('SELECT COUNT(*) FROM priority_cache') cache_size = cursor.fetchone()[0] conn.close() return { 'total_messages': total_messages, 'total_projects': len(projects), 'projects': dict(projects), 'cache_size': cache_size, 'db_size_mb': self.db_path.stat().st_size / (1024 * 1024) } # Background indexing service class BackgroundIndexer: """Monitors for new conversations and indexes them automatically""" def __init__(self, indexer: ComprehensiveIndexer): self.indexer = indexer self.check_interval = 300 # 5 minutes self.running = False async def start(self): """Start background monitoring""" self.running = True logger.info("Background indexer started") while self.running: try: # Check for new files await self._check_new_conversations() # Wait before next check await asyncio.sleep(self.check_interval) except Exception as e: logger.error(f"Background indexer error: {e}") await asyncio.sleep(60) # Wait 1 minute on error async def _check_new_conversations(self): """Check for new conversation files""" # Get current file count from database conn = sqlite3.connect(str(self.indexer.db_path)) cursor = conn.cursor() cursor.execute('SELECT project_name, message_count FROM projects') db_counts = dict(cursor.fetchall()) conn.close() # Check actual file counts projects = self.indexer.discover_projects() for project_name, project_path, actual_count in projects: db_count = db_counts.get(project_name, 0) if actual_count > db_count: logger.info(f"New messages detected in {project_name}: {actual_count - db_count} new") self.indexer.index_project(project_name, project_path) def stop(self): """Stop background monitoring""" self.running = False if __name__ == "__main__": import sys # Create indexer indexer = ComprehensiveIndexer() if len(sys.argv) > 1: if sys.argv[1] == "--index": # Run full indexing print("šŸš€ Starting comprehensive indexing of all conversations...") asyncio.run(indexer.index_all_async(force_reindex=True)) elif sys.argv[1] == "--stats": # Show statistics stats = indexer.get_stats() print(f"\nšŸ“Š Conversation Database Statistics:") print(f" Total messages: {stats['total_messages']:,}") print(f" Total projects: {stats['total_projects']}") print(f" Database size: {stats['db_size_mb']:.1f} MB") print(f" Cache entries: {stats['cache_size']}") print(f"\nšŸ“ Top projects:") for project, count in list(stats['projects'].items())[:10]: print(f" {project}: {count:,} messages") elif sys.argv[1] == "--recall" and len(sys.argv) > 2: # Quick recall test query = ' '.join(sys.argv[2:]) results = indexer.quick_recall(query) print(f"\nšŸ” Results for '{query}':") for i, result in enumerate(results[:5]): print(f"\n{i+1}. {result['content'][:200]}...") print(f" Source: {result.get('source', 'unknown')}") else: print("Usage:") print(" python comprehensive_indexer.py --index # Index all conversations") print(" python comprehensive_indexer.py --stats # Show statistics") print(" python comprehensive_indexer.py --recall # Quick search")