Fix SQLiteSession threading.Lock() bug and file descriptor leak

This PR addresses two critical bugs in SQLiteSession:

## Bug 1: threading.Lock() creating new instances

**Problem:**
In SQLiteSession (4 places) and AdvancedSQLiteSession (8 places), the code used:
```python
with self._lock if self._is_memory_db else threading.Lock():
```

For file-based databases, this creates a NEW Lock() instance on every operation,
providing NO thread safety whatsoever. Only in-memory databases used self._lock.

**Impact:**
- File-based SQLiteSession had zero thread protection
- Race conditions possible but masked by WAL mode's own concurrency handling

## Bug 2: File descriptor leak

**Problem:**
Thread-local connections in ThreadPoolExecutor are never cleaned up:
- asyncio.to_thread() uses ThreadPoolExecutor internally
- Each worker thread creates a connection on first use
- ThreadPoolExecutor reuses threads indefinitely
- Connections persist until program exit, accumulating file descriptors

**Evidence:**
Testing on main branch (60s, 40 concurrent workers):
- My system (FD limit 1,048,575): +789 FDs leaked, 0 errors (limit not reached)
- @ihower's system (likely limit 1,024): 646,802 errors in 20 seconds

Error: `sqlite3.OperationalError: unable to open database file`

## Solution: Unified shared connection approach

Instead of managing thread-local connections that can't be reliably cleaned up
in ThreadPoolExecutor, use a single shared connection for all database types.

**Changes:**
1. Removed thread-local connection logic (eliminates FD leak root cause)
2. All database types now use shared connection + self._lock
3. SQLite's WAL mode provides sufficient concurrency even with single connection
4. Fixed all 12 instances of threading.Lock() bug (4 in SQLiteSession, 8 in AdvancedSQLiteSession)
5. Kept _is_memory_db attribute for backward compatibility with AdvancedSQLiteSession
6. Added close() and __del__() methods for proper cleanup

**Results (60s stress test, 30 writers + 10 readers):**
```
Main branch:
- FD growth: +789 (leak)
- Throughput: 701 ops/s
- Errors: 0 on high-limit systems, 646k+ on normal systems

After fix:
- FD growth: +44 (stable)
- Throughput: 726 ops/s (+3.6% improvement)
- Errors: 0 on all systems
- All 29 SQLite tests pass
```

## Why shared connection performs better

SQLite's WAL (Write-Ahead Logging) mode already provides:
- Multiple concurrent readers
- One writer coexisting with readers
- Readers don't block writer
- Writer doesn't block readers (except during checkpoint)

The overhead of managing multiple connections outweighs any concurrency benefit.

## Backward compatibility

The _is_memory_db attribute is preserved for AdvancedSQLiteSession compatibility,
even though the implementation no longer differentiates connection strategies.

## Testing

Comprehensive stress test available at:
https://gist.github.com/gn00295120/0b6a65fe6c0ac6b7a1ce23654eed3ffe

Run with: `python sqlite_stress_test_final.py`

Author: gn00295120

src/agents/extensions/memory/advanced_sqlite_session.py

@@ -3,7 +3,6 @@
 import asyncio
 import json
 import logging
-import threading
 from contextlib import closing
 from pathlib import Path
 from typing import Any, Union, cast
@@ -146,7 +145,7 @@ def _get_all_items_sync():
                 """Synchronous helper to get all items for a branch."""
                 conn = self._get_connection()
                 # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-                with self._lock if self._is_memory_db else threading.Lock():
+                with self._lock:
                     with closing(conn.cursor()) as cursor:
                         if limit is None:
                             cursor.execute(
@@ -191,7 +190,7 @@ def _get_items_sync():
             """Synchronous helper to get items for a specific branch."""
             conn = self._get_connection()
             # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 with closing(conn.cursor()) as cursor:
                     # Get message IDs in correct order for this branch
                     if limit is None:
@@ -333,7 +332,7 @@ def _add_structure_sync():
             """Synchronous helper to add structure metadata to database."""
             conn = self._get_connection()
             # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 # Get the IDs of messages we just inserted, in order
                 with closing(conn.cursor()) as cursor:
                     cursor.execute(
@@ -439,7 +438,7 @@ def _cleanup_sync():
             """Synchronous helper to cleanup orphaned messages."""
             conn = self._get_connection()
             # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 with closing(conn.cursor()) as cursor:
                     # Find messages without structure metadata
                     cursor.execute(
@@ -694,7 +693,7 @@ def _delete_sync():
             """Synchronous helper to delete branch and associated data."""
             conn = self._get_connection()
             # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 with closing(conn.cursor()) as cursor:
                     # First verify the branch exists
                     cursor.execute(
@@ -801,7 +800,7 @@ def _copy_sync():
             """Synchronous helper to copy messages to new branch."""
             conn = self._get_connection()
             # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 with closing(conn.cursor()) as cursor:
                     # Get all messages before the branch point
                     cursor.execute(
@@ -1072,7 +1071,7 @@ def _get_usage_sync():
             """Synchronous helper to get session usage data."""
             conn = self._get_connection()
             # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 if branch_id:
                     # Branch-specific usage
                     query = """
@@ -1236,7 +1235,7 @@ def _update_sync():
             """Synchronous helper to update turn usage data."""
             conn = self._get_connection()
             # TODO: Refactor SQLiteSession to use asyncio.Lock instead of threading.Lock and update this code  # noqa: E501
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 # Serialize token details as JSON
                 input_details_json = None
                 output_details_json = None

src/agents/memory/sqlite_session.py

@@ -38,40 +38,21 @@ def __init__(
         self.db_path = db_path
         self.sessions_table = sessions_table
         self.messages_table = messages_table
-        self._local = threading.local()
         self._lock = threading.Lock()
 
-        # For in-memory databases, we need a shared connection to avoid thread isolation
-        # For file databases, we use thread-local connections for better concurrency
+        # Keep _is_memory_db for backward compatibility with AdvancedSQLiteSession
         self._is_memory_db = str(db_path) == ":memory:"
-        if self._is_memory_db:
-            self._shared_connection = sqlite3.connect(":memory:", check_same_thread=False)
-            self._shared_connection.execute("PRAGMA journal_mode=WAL")
-            self._init_db_for_connection(self._shared_connection)
-        else:
-            # For file databases, initialize the schema once since it persists
-            init_conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
-            init_conn.execute("PRAGMA journal_mode=WAL")
-            self._init_db_for_connection(init_conn)
-            init_conn.close()
+
+        # Use a shared connection for all database types
+        # This avoids file descriptor leaks from thread-local connections
+        # WAL mode enables concurrent readers/writers even with a shared connection
+        self._shared_connection = sqlite3.connect(str(db_path), check_same_thread=False)
+        self._shared_connection.execute("PRAGMA journal_mode=WAL")
+        self._init_db_for_connection(self._shared_connection)
 
     def _get_connection(self) -> sqlite3.Connection:
         """Get a database connection."""
-        if self._is_memory_db:
-            # Use shared connection for in-memory database to avoid thread isolation
-            return self._shared_connection
-        else:
-            # Use thread-local connections for file databases
-            if not hasattr(self._local, "connection"):
-                self._local.connection = sqlite3.connect(
-                    str(self.db_path),
-                    check_same_thread=False,
-                )
-                self._local.connection.execute("PRAGMA journal_mode=WAL")
-            assert isinstance(self._local.connection, sqlite3.Connection), (
-                f"Expected sqlite3.Connection, got {type(self._local.connection)}"
-            )
-            return self._local.connection
+        return self._shared_connection
 
     def _init_db_for_connection(self, conn: sqlite3.Connection) -> None:
         """Initialize the database schema for a specific connection."""
@@ -120,7 +101,7 @@ async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
 
         def _get_items_sync():
             conn = self._get_connection()
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 if limit is None:
                     # Fetch all items in chronological order
                     cursor = conn.execute(
@@ -174,7 +155,7 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
         def _add_items_sync():
             conn = self._get_connection()
 
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 # Ensure session exists
                 conn.execute(
                     f"""
@@ -215,7 +196,7 @@ async def pop_item(self) -> TResponseInputItem | None:
 
         def _pop_item_sync():
             conn = self._get_connection()
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 # Use DELETE with RETURNING to atomically delete and return the most recent item
                 cursor = conn.execute(
                     f"""
@@ -252,7 +233,7 @@ async def clear_session(self) -> None:
 
         def _clear_session_sync():
             conn = self._get_connection()
-            with self._lock if self._is_memory_db else threading.Lock():
+            with self._lock:
                 conn.execute(
                     f"DELETE FROM {self.messages_table} WHERE session_id = ?",
                     (self.session_id,),
@@ -267,9 +248,12 @@ def _clear_session_sync():
 
     def close(self) -> None:
         """Close the database connection."""
-        if self._is_memory_db:
-            if hasattr(self, "_shared_connection"):
-                self._shared_connection.close()
-        else:
-            if hasattr(self._local, "connection"):
-                self._local.connection.close()
+        if hasattr(self, "_shared_connection"):
+            self._shared_connection.close()
+
+    def __del__(self) -> None:
+        """Ensure connection is closed when the session is garbage collected."""
+        try:
+            self.close()
+        except Exception:
+            pass  # Ignore errors during finalization