fix: wrapper layer stale locks and dead imports (fixes #1971) (#1972)

* fix: wrapper layer stale locks and dead imports (fixes #1971)

- Remove dead imports from agents_generator.py (load_dotenv, AutoGenerator, PraisonAIModel)
- Replace BotSessionManager id()-based locks with WeakKeyDictionary to prevent corruption
- Create LockMap utility with LRU eviction and TTL cleanup
- Refactor duplicated lock patterns in _session, _debounce, and _run_control

Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>

* fix: address reviewer feedback on LockMap implementation

- Fix AttributeError: Change _session.py reset() to use .drop() instead of .pop()
- Fix LRU eviction logic: Continue past locked entries instead of breaking
- Update docstring: Clarify asyncio-safe vs thread-safe guarantees

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>

Author: praisonai-triage-agent[bot]

src/praisonai/praisonai/_lockmap.py

@@ -0,0 +1,86 @@
+"""Per-key asyncio.Lock map with bounded LRU eviction and optional TTL.
+
+Use this anywhere you need 'a lock per (user_id|chat_id|key)' instead of
+re-implementing the dict-of-Lock pattern.
+"""
+from __future__ import annotations
+import asyncio
+import time
+from collections import OrderedDict
+from typing import Hashable
+
+
+class LockMap:
+    """Asyncio-safe per-key asyncio.Lock map with automatic cleanup.
+    
+    Provides a lock per key with LRU eviction and TTL-based cleanup to prevent
+    unbounded memory growth in long-running applications.
+    
+    Note: Safe for use within a single asyncio event loop (cooperative
+    multitasking means no interleaving between non-awaited calls), but NOT
+    safe for concurrent access from multiple OS threads.
+    """
+
+    def __init__(self, *, max_entries: int = 10_000, ttl_seconds: float = 3600.0):
+        """Initialize LockMap with bounds.
+        
+        Args:
+            max_entries: Maximum number of locks to cache (LRU eviction)
+            ttl_seconds: Time-to-live for unused locks in seconds
+        """
+        self._locks: "OrderedDict[Hashable, tuple[asyncio.Lock, float]]" = OrderedDict()
+        self._max = max_entries
+        self._ttl = ttl_seconds
+
+    def get(self, key: Hashable) -> asyncio.Lock:
+        """Get or create a lock for the given key.
+        
+        Args:
+            key: The key to get a lock for
+            
+        Returns:
+            asyncio.Lock for the key
+        """
+        now = time.monotonic()
+        entry = self._locks.get(key)
+        if entry is not None:
+            lock, _ = entry
+            # Move to end (most recently used) and update timestamp
+            self._locks.move_to_end(key)
+            self._locks[key] = (lock, now)
+            return lock
+        
+        # Create new lock + insert + LRU/TTL evict
+        lock = asyncio.Lock()
+        self._locks[key] = (lock, now)
+        self._evict_stale(now)
+        return lock
+
+    def _evict_stale(self, now: float) -> None:
+        """Evict expired and excess entries."""
+        # Expire by TTL (only if not currently locked)
+        expired = [
+            k for k, (lock, ts) in self._locks.items()
+            if (now - ts) > self._ttl and not lock.locked()
+        ]
+        for k in expired:
+            self._locks.pop(k, None)
+        
+        # Cap by LRU (don't evict locks currently held)
+        while len(self._locks) > self._max:
+            k, (lock, _) = next(iter(self._locks.items()))
+            if lock.locked():
+                # Don't evict locks currently held; bump them to the end
+                self._locks.move_to_end(k)
+                # Continue trying to evict other unlocked entries
+                continue
+            self._locks.popitem(last=False)
+            break
+
+    def drop(self, key: Hashable) -> None:
+        """Manually remove a lock for the given key."""
+        self._locks.pop(key, None)
+
+    def size(self) -> int:
+        """Return current number of cached locks."""
+        return len(self._locks)

src/praisonai/praisonai/agents_generator.py

@@ -5,9 +5,6 @@
 from .version import __version__
 import yaml, os
 from rich import print
-from dotenv import load_dotenv
-from .auto import AutoGenerator
-from .inc import PraisonAIModel
 import inspect
 from pathlib import Path
 import importlib

src/praisonai/praisonai/bots/_debounce.py

@@ -17,6 +17,7 @@
 import asyncio
 import logging
 from typing import Dict, List
+from .._lockmap import LockMap
 
 logger = logging.getLogger(__name__)
 
@@ -43,12 +44,10 @@ def __init__(
         self._buffers: Dict[str, List[str]] = {}
         self._timers: Dict[str, asyncio.TimerHandle] = {}
         self._futures: Dict[str, List[asyncio.Future]] = {}
-        self._locks: Dict[str, asyncio.Lock] = {}
+        self._locks = LockMap()
 
     def _get_lock(self, user_id: str) -> asyncio.Lock:
-        if user_id not in self._locks:
-            self._locks[user_id] = asyncio.Lock()
-        return self._locks[user_id]
+        return self._locks.get(user_id)
 
     async def debounce(self, user_id: str, text: str) -> str:
         """Buffer *text* for *user_id* and return coalesced result.

src/praisonai/praisonai/bots/_run_control.py

@@ -38,6 +38,7 @@
 import time
 from enum import Enum
 from typing import Dict, Optional, Any, TYPE_CHECKING
+from .._lockmap import LockMap
 
 if TYPE_CHECKING:
     from praisonaiagents.agent.interrupt import InterruptController
@@ -99,13 +100,11 @@ def __init__(
 
         self._busy_ack_template = busy_ack_template
         self._sessions: Dict[str, SessionRunState] = {}
-        self._locks: Dict[str, asyncio.Lock] = {}
+        self._locks = LockMap()
 
     def _get_lock(self, user_id: str) -> asyncio.Lock:
         """Get or create lock for user."""
-        if user_id not in self._locks:
-            self._locks[user_id] = asyncio.Lock()
-        return self._locks[user_id]
+        return self._locks.get(user_id)
 
     def _get_session(self, user_id: str) -> SessionRunState:
         """Get or create session state for user."""
@@ -338,7 +337,7 @@ def cleanup_stale_sessions(self, max_age_seconds: int = 3600) -> int:
 
         for user_id in stale_users:
             del self._sessions[user_id]
-            self._locks.pop(user_id, None)
+            self._locks.drop(user_id)
 
         if stale_users:
             logger.debug(f"SessionRunControl: cleaned up {len(stale_users)} stale sessions")

src/praisonai/praisonai/bots/_session.py

@@ -18,7 +18,9 @@
 import asyncio
 import logging
 import time
+import weakref
 from typing import Any, Dict, List, Optional, TYPE_CHECKING
+from .._lockmap import LockMap
 
 if TYPE_CHECKING:
     from praisonaiagents import Agent
@@ -74,8 +76,8 @@ def __init__(
         run_timeout: float = 300.0,  # 5 minutes default timeout
     ) -> None:
         self._histories: Dict[str, List[Dict[str, Any]]] = {}
-        self._locks: Dict[str, asyncio.Lock] = {}
-        self._agent_locks: Dict[int, asyncio.Lock] = {}
+        self._locks = LockMap()
+        self._agent_locks: "weakref.WeakKeyDictionary[Any, asyncio.Lock]" = weakref.WeakKeyDictionary()
         self._max_history = max_history
         self._store = store
         self._platform = platform
@@ -131,16 +133,15 @@ def _session_key(self, user_id: str) -> str:
     def _get_lock(self, user_id: str) -> asyncio.Lock:
         """Get or create an asyncio.Lock for *user_id* (storage-keyed)."""
         key = self._storage_key(user_id)
-        if key not in self._locks:
-            self._locks[key] = asyncio.Lock()
-        return self._locks[key]
+        return self._locks.get(key)
 
     def _get_agent_lock(self, agent: "Agent") -> asyncio.Lock:
-        """Get or create a lock for the *agent* instance (by id)."""
-        agent_id = id(agent)
-        if agent_id not in self._agent_locks:
-            self._agent_locks[agent_id] = asyncio.Lock()
-        return self._agent_locks[agent_id]
+        """Get or create a lock for the *agent* instance (using WeakKeyDictionary)."""
+        lock = self._agent_locks.get(agent)
+        if lock is None:
+            lock = asyncio.Lock()
+            self._agent_locks[agent] = lock
+        return lock
 
     def _load_history(self, user_id: str) -> List[Dict[str, Any]]:
         """Load user history from store (if available) or in-memory cache."""
@@ -537,7 +538,7 @@ def reap_stale(self, max_age_seconds: int) -> int:
         for storage_key in stale:
             self._histories.pop(storage_key, None)
             self._last_active.pop(storage_key, None)
-            self._locks.pop(storage_key, None)
+            self._locks.drop(storage_key)
             if self._store is not None:
                 key = self._persist_key(storage_key)
                 try:
@@ -571,7 +572,7 @@ def reset(self, user_id: str) -> bool:
         existed = storage_key in self._histories
         self._histories.pop(storage_key, None)
         self._last_active.pop(storage_key, None)
-        self._locks.pop(storage_key, None)
+        self._locks.drop(storage_key)
 
         if self._store is not None:
             persist_key = self._session_key(user_id)