diff --git a/strix/core/sessions.py b/strix/core/sessions.py index 8879d359..9286b662 100644 --- a/strix/core/sessions.py +++ b/strix/core/sessions.py @@ -4,6 +4,8 @@ from __future__ import annotations import asyncio import logging +import sqlite3 +from contextlib import contextmanager from typing import TYPE_CHECKING, Any, cast from weakref import WeakKeyDictionary @@ -12,7 +14,7 @@ from agents.memory import SQLiteSession if TYPE_CHECKING: - from collections.abc import Callable + from collections.abc import Callable, Iterator from pathlib import Path from agents.items import TResponseInputItem @@ -22,9 +24,25 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +class _PooledConnectionSession(SQLiteSession): + @contextmanager + def _locked_connection(self) -> Iterator[sqlite3.Connection]: + with self._lock: + if self._closed: + raise RuntimeError("SQLiteSession is closed") + if self._is_memory_db: + yield self._shared_connection + return + connection = sqlite3.connect(str(self.db_path), check_same_thread=False) + try: + yield connection + finally: + connection.close() + + def open_agent_session(agent_id: str, path: Path) -> SQLiteSession: path.parent.mkdir(parents=True, exist_ok=True) - return SQLiteSession(session_id=agent_id, db_path=path) + return _PooledConnectionSession(session_id=agent_id, db_path=path) async def seed_initial_input(session: Session, initial_input: Any) -> bool: diff --git a/tests/test_session_fd.py b/tests/test_session_fd.py new file mode 100644 index 00000000..976758be --- /dev/null +++ b/tests/test_session_fd.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any, cast + +import pytest + +from strix.core.sessions import open_agent_session + + +def _count_open_fds() -> int | None: + for path in (Path("/proc/self/fd"), Path("/dev/fd")): + if path.is_dir(): + return len(list(path.iterdir())) + return None + + +@pytest.mark.asyncio +async def test_sessions_hold_no_descriptors_while_parked(tmp_path: Path) -> None: + """Descriptor use must track live operations, not the number of sessions. + + The SDK keeps a connection per (session, pool thread) open for the session's + whole life. An agent parks rather than exits, so its session lives for the + scan, and fan-out multiplies those handles until the process runs out of file + descriptors (#1018). A session that is not mid-operation should hold none. + """ + baseline = _count_open_fds() + if baseline is None: + pytest.skip("no /proc/self/fd or /dev/fd on this platform") + + sessions = [open_agent_session(f"a{i}", tmp_path / f"s{i}.db") for i in range(60)] + try: + for _ in range(4): + await asyncio.gather( + *(s.add_items([{"role": "user", "content": "x"}]) for s in sessions) + ) + await asyncio.gather(*(s.get_items() for s in sessions)) + parked = _count_open_fds() + assert parked is not None + # 60 parked sessions, yet descriptors are back at the baseline. + assert parked - baseline <= 5, f"parked fds grew by {parked - baseline}" + finally: + for s in sessions: + s.close() + + +@pytest.mark.asyncio +async def test_in_flight_descriptors_track_concurrency_not_session_count( + tmp_path: Path, +) -> None: + baseline = _count_open_fds() + if baseline is None: + pytest.skip("no /proc/self/fd or /dev/fd on this platform") + + sessions = [open_agent_session(f"a{i}", tmp_path / f"s{i}.db") for i in range(200)] + peak = baseline + try: + + async def sample() -> None: + nonlocal peak + for _ in range(500): + current = _count_open_fds() + if current is not None: + peak = max(peak, current) + await asyncio.sleep(0) + + async def load() -> None: + for _ in range(4): + await asyncio.gather( + *(s.add_items([{"role": "user", "content": "x"}]) for s in sessions) + ) + + await asyncio.gather(load(), sample()) + # 200 sessions, but peak is bounded by the thread pool, well under 200. + assert peak - baseline < 100, f"in-flight fds peaked at +{peak - baseline}" + finally: + for s in sessions: + s.close() + + +@pytest.mark.asyncio +async def test_history_survives_the_per_operation_connection(tmp_path: Path) -> None: + session = open_agent_session("agent-1", tmp_path / "agents.db") + try: + for i in range(30): + await session.add_items([{"role": "user", "content": f"m{i}"}]) + items = [cast("dict[str, Any]", i) for i in await session.get_items()] + assert [i["content"] for i in items] == [f"m{i}" for i in range(30)] + finally: + session.close() + + +@pytest.mark.asyncio +async def test_concurrent_sessions_sharing_one_file_stay_consistent(tmp_path: Path) -> None: + db = tmp_path / "shared.db" + sessions = [open_agent_session(f"a{i}", db) for i in range(10)] + try: + await asyncio.gather( + *(s.add_items([{"role": "user", "content": s.session_id}]) for s in sessions) + ) + # Each session sees only its own row despite sharing the file. + for s in sessions: + items = [cast("dict[str, Any]", i) for i in await s.get_items()] + assert [i["content"] for i in items] == [s.session_id] + finally: + for s in sessions: + s.close() + + +@pytest.mark.asyncio +async def test_a_closed_session_refuses_operations(tmp_path: Path) -> None: + session = open_agent_session("agent-1", tmp_path / "agents.db") + await session.add_items([{"role": "user", "content": "x"}]) + session.close() + with pytest.raises(RuntimeError, match="closed"): + await session.add_items([{"role": "user", "content": "y"}])