cache key error when user id changes -fixed task 1 31_10_2025 v3

src/context_manager.py

@@ -3,10 +3,49 @@ import sqlite3
 import json
 import logging
 import uuid
+import hashlib
+import threading
+from contextlib import contextmanager
 from datetime import datetime, timedelta
 
 logger = logging.getLogger(__name__)
 
+
+class TransactionManager:
+    """Manage database transactions with proper locking"""
+    
+    def __init__(self, db_path):
+        self.db_path = db_path
+        self._lock = threading.RLock()
+        self._connections = {}
+    
+    @contextmanager
+    def transaction(self, session_id=None):
+        """Context manager for database transactions with automatic rollback"""
+        conn = None
+        cursor = None
+        
+        try:
+            with self._lock:
+                conn = sqlite3.connect(self.db_path, isolation_level='IMMEDIATE')
+                conn.execute('PRAGMA journal_mode=WAL')  # Write-Ahead Logging for better concurrency
+                conn.execute('PRAGMA busy_timeout=5000')  # 5 second timeout for locks
+                cursor = conn.cursor()
+                
+                yield cursor
+                
+                conn.commit()
+                logger.debug(f"Transaction committed for session {session_id}")
+                
+        except Exception as e:
+            if conn:
+                conn.rollback()
+                logger.error(f"Transaction rolled back for session {session_id}: {e}")
+            raise
+        finally:
+            if conn:
+                conn.close()
+
 class EfficientContextManager:
     def __init__(self, llm_router=None):
         self.session_cache = {}  # In-memory for active sessions
@@ -19,7 +58,9 @@ class EfficientContextManager:
         self.db_path = "sessions.db"
         self.llm_router = llm_router  # For generating context summaries
         logger.info(f"Initializing ContextManager with DB path: {self.db_path}")
+        self.transaction_manager = TransactionManager(self.db_path)
         self._init_database()
+        self.optimize_database_indexes()
     
     def _init_database(self):
         """Initialize database and create tables"""
@@ -102,11 +143,95 @@ class EfficientContextManager:
             
             conn.commit()
             conn.close()
+            
+            # Update schema with new columns and tables for user change tracking
+            self._update_database_schema()
+            
             logger.info("Database initialization complete")
             
         except Exception as e:
             logger.error(f"Database initialization error: {e}", exc_info=True)
     
+    def _update_database_schema(self):
+        """Add missing columns and tables for user change tracking"""
+        try:
+            conn = sqlite3.connect(self.db_path)
+            cursor = conn.cursor()
+            
+            # Add needs_refresh column to interaction_contexts
+            try:
+                cursor.execute("""
+                    ALTER TABLE interaction_contexts 
+                    ADD COLUMN needs_refresh INTEGER DEFAULT 0
+                """)
+                logger.info("✓ Added needs_refresh column to interaction_contexts")
+            except sqlite3.OperationalError:
+                pass  # Column already exists
+            
+            # Create user change log table
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS user_change_log (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    session_id TEXT,
+                    old_user_id TEXT,
+                    new_user_id TEXT,
+                    timestamp TIMESTAMP,
+                    FOREIGN KEY(session_id) REFERENCES sessions(session_id)
+                )
+            """)
+            
+            conn.commit()
+            conn.close()
+            logger.info("✓ Database schema updated successfully for user change tracking")
+            
+            # Update interactions table for deduplication
+            self._update_interactions_table()
+            
+        except Exception as e:
+            logger.error(f"Schema update error: {e}", exc_info=True)
+    
+    def _update_interactions_table(self):
+        """Add interaction_hash column for deduplication"""
+        try:
+            conn = sqlite3.connect(self.db_path)
+            cursor = conn.cursor()
+            
+            # Check if column already exists
+            cursor.execute("PRAGMA table_info(interactions)")
+            columns = [row[1] for row in cursor.fetchall()]
+            
+            # Add interaction_hash column if it doesn't exist
+            if 'interaction_hash' not in columns:
+                try:
+                    cursor.execute("""
+                        ALTER TABLE interactions 
+                        ADD COLUMN interaction_hash TEXT
+                    """)
+                    logger.info("✓ Added interaction_hash column to interactions table")
+                except sqlite3.OperationalError:
+                    pass  # Column already exists
+            
+            # Create unique index for deduplication (this enforces uniqueness)
+            try:
+                cursor.execute("""
+                    CREATE UNIQUE INDEX IF NOT EXISTS idx_interaction_hash_unique 
+                    ON interactions(interaction_hash)
+                """)
+                logger.info("✓ Created unique index on interaction_hash")
+            except sqlite3.OperationalError:
+                # Index might already exist, try non-unique index as fallback
+                cursor.execute("""
+                    CREATE INDEX IF NOT EXISTS idx_interaction_hash 
+                    ON interactions(interaction_hash)
+                """)
+            
+            conn.commit()
+            conn.close()
+            logger.info("✓ Interactions table updated for deduplication")
+            
+        except Exception as e:
+            logger.error(f"Error updating interactions table: {e}", exc_info=True)
+    
     async def manage_context(self, session_id: str, user_input: str, user_id: str = "Test_Any") -> dict:
         """
         Efficient context management with separated session/user caching
@@ -121,6 +246,14 @@ class EfficientContextManager:
         # Get session context from cache
         session_context = self._get_from_memory_cache(session_cache_key)
         
+        # Check if cached session context matches current user_id
+        if session_context and session_context.get("user_id") != user_id:
+            # User changed, invalidate session cache
+            logger.info(f"User mismatch in cache for session {session_id}, invalidating cache")
+            session_context = None
+            if session_cache_key in self.session_cache:
+                del self.session_cache[session_cache_key]
+        
         # Get user context separately
         user_context = self._get_from_memory_cache(user_cache_key)
         
@@ -128,7 +261,7 @@ class EfficientContextManager:
             # Retrieve from database with user context
             session_context = await self._retrieve_from_db(session_id, user_input, user_id)
             
-            # Cache session context
+            # Cache session context (cache invalidation for user changes is handled in _retrieve_from_db)
             self._warm_memory_cache(session_cache_key, session_context)
         
         # Handle user context separately to prevent loops
@@ -453,13 +586,18 @@ Keep the summary concise (approximately 100 tokens)."""
     
     async def _retrieve_from_db(self, session_id: str, user_input: str, user_id: str = "Test_Any") -> dict:
         """
-        Retrieve context from database with semantic search
+        Retrieve session context with proper user_id synchronization
+        Uses transactions to ensure atomic updates of database and cache
         """
+        conn = None
         try:
             conn = sqlite3.connect(self.db_path)
             cursor = conn.cursor()
             
-            # Get session data
+            # Use transaction to ensure atomic updates
+            cursor.execute("BEGIN TRANSACTION")
+            
+            # Get session data (SQLite doesn't support FOR UPDATE, but transaction ensures consistency)
             cursor.execute("""
                 SELECT context_data, user_metadata, last_activity, user_id
                 FROM sessions 
@@ -474,31 +612,77 @@ Keep the summary concise (approximately 100 tokens)."""
                 last_activity = row[2]
                 session_user_id = row[3] if len(row) > 3 else user_id
                 
-                # Update user_id if it changed
+                # Check for user_id change and update atomically
+                user_changed = False
                 if session_user_id != user_id:
+                    logger.info(f"User change detected: {session_user_id} -> {user_id} for session {session_id}")
+                    user_changed = True
+                    
+                    # Update session with new user_id
                     cursor.execute("""
-                        UPDATE sessions SET user_id = ? WHERE session_id = ?
-                    """, (user_id, session_id))
-                    conn.commit()
+                        UPDATE sessions 
+                        SET user_id = ?, last_activity = ?
+                        WHERE session_id = ?
+                    """, (user_id, datetime.now().isoformat(), session_id))
+                    
+                    # Clear any cached interaction contexts for old user by marking for refresh
+                    try:
+                        cursor.execute("""
+                            UPDATE interaction_contexts
+                            SET needs_refresh = 1
+                            WHERE session_id = ?
+                        """, (session_id,))
+                    except sqlite3.OperationalError:
+                        # Column might not exist yet, will be created by schema update
+                        pass
+                    
+                    # Log user change event
+                    try:
+                        cursor.execute("""
+                            INSERT INTO user_change_log (session_id, old_user_id, new_user_id, timestamp)
+                            VALUES (?, ?, ?, ?)
+                        """, (session_id, session_user_id, user_id, datetime.now().isoformat()))
+                    except sqlite3.OperationalError:
+                        # Table might not exist yet, will be created by schema update
+                        pass
+                    
                     # Clear old cache entries when user changes
                     self._clear_user_cache_on_change(session_id, user_id, session_user_id)
                 
-                # Get previous interaction contexts for this session
-                cursor.execute("""
-                    SELECT interaction_summary, created_at
-                    FROM interaction_contexts
-                    WHERE session_id = ?
-                    ORDER BY created_at DESC
-                    LIMIT 20
-                """, (session_id,))
+                cursor.execute("COMMIT")
+                
+                # Get interaction contexts with refresh flag check
+                try:
+                    cursor.execute("""
+                        SELECT interaction_summary, created_at, needs_refresh
+                        FROM interaction_contexts
+                        WHERE session_id = ? AND (needs_refresh IS NULL OR needs_refresh = 0)
+                        ORDER BY created_at DESC
+                        LIMIT 20
+                    """, (session_id,))
+                except sqlite3.OperationalError:
+                    # Column might not exist yet, fall back to query without needs_refresh
+                    cursor.execute("""
+                        SELECT interaction_summary, created_at
+                        FROM interaction_contexts
+                        WHERE session_id = ?
+                        ORDER BY created_at DESC
+                        LIMIT 20
+                    """, (session_id,))
                 
                 interaction_contexts = []
                 for ic_row in cursor.fetchall():
-                    if ic_row[0]:
-                        interaction_contexts.append({
-                            "summary": ic_row[0],
-                            "timestamp": ic_row[1]
-                        })
+                    # Handle both query formats (with and without needs_refresh)
+                    if len(ic_row) >= 2:
+                        summary = ic_row[0]
+                        timestamp = ic_row[1]
+                        needs_refresh = ic_row[2] if len(ic_row) > 2 else 0
+                        
+                        if summary and not needs_refresh:
+                            interaction_contexts.append({
+                                "summary": summary,
+                                "timestamp": timestamp
+                            })
                 
                 context = {
                     "session_id": session_id,
@@ -507,18 +691,20 @@ Keep the summary concise (approximately 100 tokens)."""
                     "preferences": user_metadata.get("preferences", {}),
                     "active_tasks": user_metadata.get("active_tasks", []),
                     "last_activity": last_activity,
-                    "user_context_loaded": False  # Will be loaded in manage_context
+                    "user_context_loaded": False,
+                    "user_changed": user_changed
                 }
                 
                 conn.close()
                 return context
             else:
-                # Create new session
+                # Create new session with transaction
                 cursor.execute("""
                     INSERT INTO sessions (session_id, user_id, created_at, last_activity, context_data, user_metadata)
                     VALUES (?, ?, ?, ?, ?, ?)
                 """, (session_id, user_id, datetime.now().isoformat(), datetime.now().isoformat(), "{}", "{}"))
-                conn.commit()
+                
+                cursor.execute("COMMIT")
                 conn.close()
                 
                 return {
@@ -527,19 +713,47 @@ Keep the summary concise (approximately 100 tokens)."""
                     "interaction_contexts": [],
                     "preferences": {},
                     "active_tasks": [],
-                    "user_context_loaded": False
+                    "user_context_loaded": False,
+                    "user_changed": False
                 }
                 
+        except sqlite3.Error as e:
+            logger.error(f"Database transaction error: {e}", exc_info=True)
+            if conn:
+                try:
+                    conn.rollback()
+                except:
+                    pass
+                conn.close()
+            # Return safe fallback
+            return {
+                "session_id": session_id,
+                "user_id": user_id,
+                "interaction_contexts": [],
+                "preferences": {},
+                "active_tasks": [],
+                "user_context_loaded": False,
+                "error": str(e),
+                "user_changed": False
+            }
         except Exception as e:
             logger.error(f"Database retrieval error: {e}", exc_info=True)
-            # Fallback to empty context
+            if conn:
+                try:
+                    conn.rollback()
+                except:
+                    pass
+                conn.close()
+            # Return safe fallback
             return {
                 "session_id": session_id,
                 "user_id": user_id,
                 "interaction_contexts": [],
                 "preferences": {},
                 "active_tasks": [],
-                "user_context_loaded": False
+                "user_context_loaded": False,
+                "error": str(e),
+                "user_changed": False
             }
     
     def _warm_memory_cache(self, cache_key: str, context: dict):
@@ -550,39 +764,322 @@ Keep the summary concise (approximately 100 tokens)."""
     
     def _update_context(self, context: dict, user_input: str, response: str = None, user_id: str = "Test_Any") -> dict:
         """
-        Update context with new user interaction and persist to database
-        Note: Interaction context generation happens separately after response is generated
+        Update context with deduplication and idempotency checks
+        Prevents duplicate context updates using interaction hashes
         """
         try:
-            # Update session activity
+            # Generate unique interaction hash to prevent duplicates
+            interaction_hash = self._generate_interaction_hash(user_input, context["session_id"], user_id)
+            
+            # Check if this interaction was already processed
+            if self._is_duplicate_interaction(interaction_hash):
+                logger.info(f"Duplicate interaction detected, skipping update: {interaction_hash[:8]}")
+                return context
+            
+            # Use transaction for atomic updates
+            current_time = datetime.now().isoformat()
+            with self.transaction_manager.transaction(context["session_id"]) as cursor:
+                # Update session activity (only if last_activity is older to prevent unnecessary updates)
+                cursor.execute("""
+                    UPDATE sessions 
+                    SET last_activity = ?, user_id = ?
+                    WHERE session_id = ? AND (last_activity IS NULL OR last_activity < ?)
+                """, (current_time, user_id, context["session_id"], current_time))
+                
+                # Store interaction with duplicate prevention using INSERT OR IGNORE
+                session_context = {
+                    "preferences": context.get("preferences", {}),
+                    "active_tasks": context.get("active_tasks", [])
+                }
+                
+                cursor.execute("""
+                    INSERT OR IGNORE INTO interactions (
+                        interaction_hash,
+                        session_id, 
+                        user_input, 
+                        context_snapshot, 
+                        created_at
+                    ) VALUES (?, ?, ?, ?, ?)
+                """, (
+                    interaction_hash,
+                    context["session_id"],
+                    user_input,
+                    json.dumps(session_context),
+                    current_time
+                ))
+            
+            # Mark interaction as processed (outside transaction)
+            self._mark_interaction_processed(interaction_hash)
+            
+            # Update in-memory context
+            context["last_interaction"] = user_input
+            context["last_update"] = current_time
+            
+            logger.info(f"Context updated for session {context['session_id']} with hash {interaction_hash[:8]}")
+            
+            return context
+            
+        except Exception as e:
+            logger.error(f"Error updating context: {e}", exc_info=True)
+            return context
+    
+    def _generate_interaction_hash(self, user_input: str, session_id: str, user_id: str) -> str:
+        """Generate unique hash for interaction to prevent duplicates"""
+        # Use session_id, user_id, and user_input for exact duplicate detection
+        # Normalize user input by stripping whitespace
+        normalized_input = user_input.strip()
+        content = f"{session_id}:{user_id}:{normalized_input}"
+        return hashlib.sha256(content.encode()).hexdigest()
+    
+    def _is_duplicate_interaction(self, interaction_hash: str) -> bool:
+        """Check if interaction was already processed"""
+        # Keep a rolling window of recent interaction hashes in memory
+        if not hasattr(self, '_processed_interactions'):
+            self._processed_interactions = set()
+        
+        # Check in-memory cache first
+        if interaction_hash in self._processed_interactions:
+            return True
+        
+        # Also check database for persistent duplicates
+        try:
             conn = sqlite3.connect(self.db_path)
             cursor = conn.cursor()
+            # Check if interaction_hash column exists and query for duplicates
+            cursor.execute("PRAGMA table_info(interactions)")
+            columns = [row[1] for row in cursor.fetchall()]
+            if 'interaction_hash' in columns:
+                cursor.execute("""
+                    SELECT COUNT(*) FROM interactions 
+                    WHERE interaction_hash IS NOT NULL AND interaction_hash = ?
+                """, (interaction_hash,))
+                count = cursor.fetchone()[0]
+                conn.close()
+                return count > 0
+            else:
+                conn.close()
+                return False
+        except sqlite3.OperationalError:
+            # Column might not exist yet, only check in-memory
+            return interaction_hash in self._processed_interactions
+    
+    def _mark_interaction_processed(self, interaction_hash: str):
+        """Mark interaction as processed"""
+        if not hasattr(self, '_processed_interactions'):
+            self._processed_interactions = set()
+        self._processed_interactions.add(interaction_hash)
+        
+        # Limit memory usage by keeping only last 1000 hashes
+        if len(self._processed_interactions) > 1000:
+            # Keep most recent 500 entries (simple truncation)
+            self._processed_interactions = set(list(self._processed_interactions)[-500:])
+    
+    async def manage_context_optimized(self, session_id: str, user_input: str, user_id: str = "Test_Any") -> dict:
+        """
+        Efficient context management with transaction optimization
+        """
+        # Use session-only cache key
+        session_cache_key = f"session_{session_id}"
+        
+        # Try to get from cache first (no DB access)
+        cached_context = self._get_from_memory_cache(session_cache_key)
+        if cached_context and self._is_cache_valid(cached_context):
+            logger.debug(f"Using cached context for session {session_id}")
+            return cached_context
+        
+        # Use transaction for all DB operations
+        with self.transaction_manager.transaction(session_id) as cursor:
+            # Atomic session retrieval and update
+            cursor.execute("""
+                SELECT s.context_data, s.user_metadata, s.last_activity, s.user_id,
+                       COUNT(ic.interaction_id) as interaction_count
+                FROM sessions s
+                LEFT JOIN interaction_contexts ic ON s.session_id = ic.session_id
+                WHERE s.session_id = ?
+                GROUP BY s.session_id
+            """, (session_id,))
+            
+            row = cursor.fetchone()
             
-            # Update session last_activity
+            if row:
+                # Parse existing session data
+                context_data = json.loads(row[0] or '{}')
+                user_metadata = json.loads(row[1] or '{}')
+                last_activity = row[2]
+                stored_user_id = row[3] or user_id
+                interaction_count = row[4] or 0
+                
+                # Handle user change atomically
+                if stored_user_id != user_id:
+                    self._handle_user_change_atomic(cursor, session_id, stored_user_id, user_id)
+                
+                # Get interaction contexts efficiently
+                interaction_contexts = self._get_interaction_contexts_atomic(cursor, session_id)
+                
+            else:
+                # Create new session atomically
+                cursor.execute("""
+                    INSERT INTO sessions (session_id, user_id, created_at, last_activity, context_data, user_metadata)
+                    VALUES (?, ?, datetime('now'), datetime('now'), '{}', '{}')
+                """, (session_id, user_id))
+                
+                context_data = {}
+                user_metadata = {}
+                interaction_contexts = []
+                interaction_count = 0
+        
+        # Load user context asynchronously (outside transaction)
+        user_context = await self._load_user_context_async(user_id)
+        
+        # Build final context
+        final_context = {
+            "session_id": session_id,
+            "user_id": user_id,
+            "interaction_contexts": interaction_contexts,
+            "user_context": user_context,
+            "preferences": user_metadata.get("preferences", {}),
+            "active_tasks": user_metadata.get("active_tasks", []),
+            "interaction_count": interaction_count,
+            "cache_timestamp": datetime.now().isoformat()
+        }
+        
+        # Update cache
+        self._warm_memory_cache(session_cache_key, final_context)
+        
+        return self._optimize_context(final_context)
+    
+    def _handle_user_change_atomic(self, cursor, session_id: str, old_user_id: str, new_user_id: str):
+        """Handle user change within transaction"""
+        logger.info(f"Handling user change in transaction: {old_user_id} -> {new_user_id}")
+        
+        # Update session
+        cursor.execute("""
+            UPDATE sessions 
+            SET user_id = ?, last_activity = datetime('now')
+            WHERE session_id = ?
+        """, (new_user_id, session_id))
+        
+        # Log the change
+        try:
+            cursor.execute("""
+                INSERT INTO user_change_log (session_id, old_user_id, new_user_id, timestamp)
+                VALUES (?, ?, ?, datetime('now'))
+            """, (session_id, old_user_id, new_user_id))
+        except sqlite3.OperationalError:
+            # Table might not exist yet
+            pass
+        
+        # Invalidate related caches
+        try:
             cursor.execute("""
-                UPDATE sessions 
-                SET last_activity = ?, user_id = ?
+                UPDATE interaction_contexts
+                SET needs_refresh = 1
                 WHERE session_id = ?
-            """, (datetime.now().isoformat(), user_id, context["session_id"]))
+            """, (session_id,))
+        except sqlite3.OperationalError:
+            # Column might not exist yet
+            pass
+    
+    def _get_interaction_contexts_atomic(self, cursor, session_id: str, limit: int = 20):
+        """Get interaction contexts within transaction"""
+        try:
+            cursor.execute("""
+                SELECT interaction_summary, created_at, interaction_id
+                FROM interaction_contexts
+                WHERE session_id = ? AND (needs_refresh IS NULL OR needs_refresh = 0)
+                ORDER BY created_at DESC
+                LIMIT ?
+            """, (session_id, limit))
+        except sqlite3.OperationalError:
+            # Fallback if needs_refresh column doesn't exist
+            cursor.execute("""
+                SELECT interaction_summary, created_at, interaction_id
+                FROM interaction_contexts
+                WHERE session_id = ?
+                ORDER BY created_at DESC
+                LIMIT ?
+            """, (session_id, limit))
+        
+        contexts = []
+        for row in cursor.fetchall():
+            if row[0]:
+                contexts.append({
+                    "summary": row[0],
+                    "timestamp": row[1],
+                    "id": row[2] if len(row) > 2 else None
+                })
+        
+        return contexts
+    
+    async def _load_user_context_async(self, user_id: str):
+        """Load user context asynchronously to avoid blocking"""
+        try:
+            # Check memory cache first
+            user_cache_key = f"user_{user_id}"
+            cached = self._get_from_memory_cache(user_cache_key)
+            if cached:
+                return cached.get("user_context", "")
             
-            # Insert basic interaction record (for backward compatibility)
-            session_context = {
-                "preferences": context.get("preferences", {}),
-                "active_tasks": context.get("active_tasks", [])
-            }
+            # Load from database
+            return await self.get_user_context(user_id)
+        except Exception as e:
+            logger.error(f"Error loading user context: {e}")
+            return ""
+    
+    def _is_cache_valid(self, cached_context: dict, max_age_seconds: int = 60) -> bool:
+        """Check if cached context is still valid"""
+        if not cached_context:
+            return False
+        
+        cache_timestamp = cached_context.get("cache_timestamp")
+        if not cache_timestamp:
+            return False
+        
+        try:
+            cache_time = datetime.fromisoformat(cache_timestamp)
+            age = (datetime.now() - cache_time).total_seconds()
+            return age < max_age_seconds
+        except:
+            return False
+    
+    def optimize_database_indexes(self):
+        """Create database indexes for better query performance"""
+        try:
+            conn = sqlite3.connect(self.db_path)
+            cursor = conn.cursor()
             
-            cursor.execute("""
-                INSERT INTO interactions (session_id, user_input, context_snapshot, created_at)
-                VALUES (?, ?, ?, ?)
-            """, (context["session_id"], user_input, json.dumps(session_context), datetime.now().isoformat()))
+            # Create indexes for frequently queried columns
+            indexes = [
+                "CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions(user_id)",
+                "CREATE INDEX IF NOT EXISTS idx_sessions_last_activity ON sessions(last_activity)",
+                "CREATE INDEX IF NOT EXISTS idx_interactions_session_id ON interactions(session_id)",
+                "CREATE INDEX IF NOT EXISTS idx_interaction_contexts_session_id ON interaction_contexts(session_id)",
+                "CREATE INDEX IF NOT EXISTS idx_interaction_contexts_created_at ON interaction_contexts(created_at)",
+                "CREATE INDEX IF NOT EXISTS idx_user_change_log_session_id ON user_change_log(session_id)",
+                "CREATE INDEX IF NOT EXISTS idx_user_contexts_updated_at ON user_contexts(updated_at)"
+            ]
+            
+            for index in indexes:
+                try:
+                    cursor.execute(index)
+                except sqlite3.OperationalError as e:
+                    # Table might not exist yet, skip this index
+                    logger.debug(f"Skipping index creation (table may not exist): {e}")
+            
+            # Analyze database for query optimization
+            try:
+                cursor.execute("ANALYZE")
+            except sqlite3.OperationalError:
+                # ANALYZE might not be available in all SQLite versions
+                pass
             
             conn.commit()
             conn.close()
             
+            logger.info("✓ Database indexes optimized successfully")
+            
         except Exception as e:
-            logger.error(f"Context update error: {e}", exc_info=True)
-        
-        return context
+            logger.error(f"Error optimizing database indexes: {e}", exc_info=True)
     
     def _extract_entities(self, context: dict) -> list:
         """

src/orchestrator_engine.py

@@ -49,35 +49,99 @@ class MVPOrchestrator:
         # User ID tracking for context system
         self._current_user_id = {}  # session_id -> user_id
         
+        # Context cache to prevent loops
+        self._context_cache = {}  # cache_key -> {context, timestamp}
+        
         logger.info("MVPOrchestrator initialized with safety revision thresholds")
     
     def set_user_id(self, session_id: str, user_id: str):
-        """Set user_id for a session"""
-        self._current_user_id[session_id] = user_id
-        logger.info(f"Set user_id={user_id} for session {session_id}")
+        """Set user_id with loop prevention"""
+        # Check if user_id actually changed
+        old_user_id = self._current_user_id.get(session_id)
+        
+        if old_user_id != user_id:
+            self._current_user_id[session_id] = user_id
+            logger.info(f"Set user_id={user_id} for session {session_id} (was: {old_user_id})")
+            
+            # Clear context cache on user change
+            cache_key = f"context_{session_id}"
+            if cache_key in self._context_cache:
+                del self._context_cache[cache_key]
+                logger.info(f"Cleared context cache for session {session_id} due to user change")
+        else:
+            self._current_user_id[session_id] = user_id
+    
+    def _get_user_id_for_session(self, session_id: str) -> str:
+        """Get user_id without triggering context loops"""
+        # Use in-memory mapping first
+        if hasattr(self, '_current_user_id') and session_id in self._current_user_id:
+            return self._current_user_id[session_id]
+        
+        # Fallback to default if not found
+        return "Test_Any"
+    
+    async def _get_or_create_context(self, session_id: str, user_input: str, user_id: str) -> dict:
+        """Get context with loop prevention and caching"""
+        # Check if we recently fetched context for this session
+        cache_key = f"context_{session_id}"
+        current_time = time.time()
+        
+        if hasattr(self, '_context_cache'):
+            cached = self._context_cache.get(cache_key)
+            if cached and (current_time - cached['timestamp']) < 5:  # 5 second cache
+                logger.info(f"Using cached context for session {session_id}")
+                return cached['context']
+        
+        # Fetch new context
+        context = await self.context_manager.manage_context(session_id, user_input, user_id=user_id)
+        
+        # Cache the context
+        if not hasattr(self, '_context_cache'):
+            self._context_cache = {}
+        
+        self._context_cache[cache_key] = {
+            'context': context,
+            'timestamp': current_time
+        }
+        
+        # Clean old cache entries
+        if len(self._context_cache) > 100:
+            # Remove oldest entries
+            sorted_items = sorted(self._context_cache.items(), key=lambda x: x[1]['timestamp'])
+            self._context_cache = dict(sorted_items[-50:])
+        
+        return context
         
     async def process_request(self, session_id: str, user_input: str) -> dict:
         """
-        Main orchestration flow with academic differentiation and enhanced reasoning chain
+        Main orchestration flow with loop prevention
         """
         logger.info(f"Processing request for session {session_id}")
         logger.info(f"User input: {user_input[:100]}")
         
-        # Safety context bypass: Skip safety checks for binary responses to safety prompts
+        # Critical: Prevent safety check loops on binary responses
         user_input_upper = user_input.strip().upper()
         is_binary_response = user_input_upper in ['YES', 'NO', 'APPLY', 'KEEP', 'Y', 'N']
         
+        # Check if we're in a safety response context
         if is_binary_response and self.awaiting_safety_response.get(session_id, False):
-            logger.info(f"Binary safety response detected ({user_input_upper}) - bypassing safety check to prevent loop")
-            # Clear the flag immediately to prevent re-triggering
+            logger.info(f"Binary safety response detected ({user_input_upper}) - bypassing recursive safety check")
+            
+            # Immediately clear the flag to prevent any further loops
             self.awaiting_safety_response[session_id] = False
             
-            # Return a signal that this should be handled by the choice handler, not normal processing
+            # Remove from pending choices if exists
+            if hasattr(self, '_pending_choices'):
+                self._pending_choices.pop(session_id, None)
+            
+            # Return with skip flag to prevent further processing
             return {
                 'is_safety_response': True,
                 'response': user_input_upper,
                 'requires_user_choice': False,
-                'skip_safety_check': True
+                'skip_safety_check': True,
+                'final_response': f"Choice '{user_input_upper}' has been applied.",
+                'bypass_reason': 'binary_safety_response'
             }
         
         # Clear previous trace for new request
@@ -98,11 +162,15 @@ class MVPOrchestrator:
             interaction_id = self._generate_interaction_id(session_id)
             logger.info(f"Generated interaction ID: {interaction_id}")
             
-            # Step 2: Context management with reasoning
-            logger.info("Step 2: Managing context...")
-            # Get user_id from context or default to Test_Any (passed from app.py)
-            user_id = getattr(self, '_current_user_id', {}).get(session_id, "Test_Any")
-            context = await self.context_manager.manage_context(session_id, user_input, user_id=user_id)
+            # Step 2: Context management with loop prevention
+            logger.info("Step 2: Managing context with loop prevention...")
+            
+            # Get user_id from stored mapping, avoiding context retrieval loops
+            user_id = self._get_user_id_for_session(session_id)
+            
+            # Use context with deduplication check
+            context = await self._get_or_create_context(session_id, user_input, user_id)
+            
             logger.info(f"Context retrieved: {len(context.get('interaction_contexts', []))} interaction contexts")
             
             # Add context analysis to reasoning chain