Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions acp_adapter/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -621,3 +621,12 @@ def _make_agent(
# Route any incidental human-readable agent output to stderr instead.
agent._print_fn = _acp_stderr_print
return agent

def close(self) -> None:
"""Close the lazily-initialised SessionDB connection."""
if self._db_instance is not None:
try:
self._db_instance.close()
except Exception:
pass
self._db_instance = None
9 changes: 9 additions & 0 deletions agent/insights.py
Original file line number Diff line number Diff line change
Expand Up @@ -919,3 +919,12 @@ def format_gateway(self, report: Dict) -> str:
lines.append(f"**Best streak:** {act['max_streak']} consecutive days")

return "\n".join(lines)

def close(self) -> None:
"""Release the borrowed connection reference.

InsightsEngine borrows db._conn from the owning SessionDB — it
does NOT own the connection, so we only clear our reference.
"""
self._conn = None
self.db = None
36 changes: 35 additions & 1 deletion hermes_cli/kanban_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -1230,6 +1230,31 @@ def _cross_process_init_lock(path: Path):
handle.close()


class _WalSafeConnection(sqlite3.Connection):
"""Connection subclass that checkpoints WAL before close to release
WAL file descriptors immediately.

Long-running processes (gateway kanban dispatcher, notifier) open a
new kanban connection every tick. In WAL mode SQLite defers cleanup
of WAL/shm file descriptors, causing a slow FD leak that eventually
hits the process limit and triggers cascading failures (``too many
open files``, ``unable to open database file``).

Calling ``PRAGMA wal_checkpoint(TRUNCATE)`` before each close forces
SQLite to consolidate the WAL and release its file descriptors so
the FD count stays flat. See issue #30799.
"""

def close(self) -> None:
try:
self.execute("PRAGMA wal_checkpoint(TRUNCATE)")
except sqlite3.ProgrammingError:
pass # already closed
except sqlite3.OperationalError:
pass # network FS / detached -- close without checkpoint
super().close()


def _looks_like_tls_record_at(data: bytes, offset: int) -> bool:
"""Return True for a TLS record header at ``data[offset:]``."""
if len(data) < offset + 5:
Expand Down Expand Up @@ -1449,7 +1474,16 @@ def connect(
# via _INITIALIZED_PATHS so it only runs once per process per path.
_guard_existing_db_is_healthy(path)
resolved = str(path.resolve())
conn = _sqlite_connect(path)
# Use _WalSafeConnection to checkpoint WAL before close, preventing the
# slow FD leak from deferred WAL/shm cleanup in long-running processes.
busy_timeout_ms = _resolve_busy_timeout_ms()
conn = sqlite3.connect(
str(path),
isolation_level=None,
timeout=busy_timeout_ms / 1000.0,
factory=_WalSafeConnection,
)
conn.execute(f"PRAGMA busy_timeout={busy_timeout_ms}")
try:
conn.row_factory = sqlite3.Row
with _INIT_LOCK:
Expand Down
16 changes: 16 additions & 0 deletions plugins/memory/retaindb/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -406,6 +406,16 @@ def shutdown(self) -> None:
self._q.put(_ASYNC_SHUTDOWN)
self._thread.join(timeout=10)

def close(self) -> None:
"""Close the thread-local SQLite connection for the calling thread."""
conn = getattr(self._local, "conn", None)
if conn is not None:
try:
conn.close()
except Exception:
pass
self._local.conn = None


# ---------------------------------------------------------------------------
# Overlay formatter
Expand Down Expand Up @@ -760,6 +770,12 @@ def shutdown(self) -> None:
if self._queue:
self._queue.shutdown()

def close(self) -> None:
"""Shutdown writer threads and close SQLite connections."""
self.shutdown()
if self._queue:
self._queue.close()


def register(ctx) -> None:
"""Register RetainDB as a memory provider plugin."""
Expand Down
212 changes: 212 additions & 0 deletions tests/hermes_cli/test_db_close_idempotency.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,212 @@
"""Tests for close() idempotency on SQLite-owning classes.

Verifies that:
- close() can be called twice without raising
- close() actually releases resources (sets references to None)
- close() on already-closed objects is a no-op

Addresses review feedback from mxnstrexgl on PR #36116.
"""
from __future__ import annotations

import sqlite3
from pathlib import Path
from unittest.mock import MagicMock

import pytest


# ── _WalSafeConnection ──────────────────────────────────────────────────────


class TestWalSafeConnectionClose:
"""Tests for hermes_cli.kanban_db._WalSafeConnection.close()."""

def _make_conn(self, path: Path) -> sqlite3.Connection:
"""Create a _WalSafeConnection to a file-backed database."""
from hermes_cli.kanban_db import _WalSafeConnection

conn = sqlite3.connect(
str(path),
isolation_level=None,
factory=_WalSafeConnection,
)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("CREATE TABLE IF NOT EXISTS test (id INTEGER PRIMARY KEY)")
conn.execute("INSERT INTO test VALUES (1)")
return conn

def test_close_is_idempotent(self, tmp_path: Path) -> None:
"""Calling close() twice must not raise."""
db_path = tmp_path / "test.db"
conn = self._make_conn(db_path)
conn.close()
conn.close() # second call — must not raise

def test_close_checkpoints_wal(self, tmp_path: Path) -> None:
"""close() runs WAL checkpoint before closing."""
db_path = tmp_path / "test.db"
conn = self._make_conn(db_path)

# Verify WAL mode is active
mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
assert mode.lower() == "wal"

conn.close()

# After close, the WAL should be checkpointed (merged into main db)
# Verify by opening a new connection and reading the data
conn2 = sqlite3.connect(str(db_path))
row = conn2.execute("SELECT id FROM test").fetchone()
assert row is not None
assert row[0] == 1
conn2.close()

def test_close_on_in_memory_db(self) -> None:
"""close() works on in-memory databases too."""
from hermes_cli.kanban_db import _WalSafeConnection

conn = sqlite3.connect(":memory:", factory=_WalSafeConnection)
conn.execute("CREATE TABLE t (id INTEGER)")
conn.close()
conn.close() # idempotent

def test_close_handles_already_closed(self, tmp_path: Path) -> None:
"""close() on an already-closed connection is graceful."""
db_path = tmp_path / "test.db"
conn = self._make_conn(db_path)
conn.close()
# The second close might raise ProgrammingError on some Python
# versions, but _WalSafeConnection catches it.
try:
conn.close()
except Exception:
# If it raises, that's also acceptable — the important thing
# is it doesn't crash the process.
pass


# ── InsightsEngine.close() ─────────────────────────────────────────────────


class TestInsightsEngineClose:
"""Tests for agent/insights.py InsightsEngine.close()."""

def test_close_nulls_references(self) -> None:
"""close() sets _conn and db to None."""
from agent.insights import InsightsEngine

engine = InsightsEngine.__new__(InsightsEngine)
engine._conn = MagicMock()
engine.db = MagicMock()

engine.close()

assert engine._conn is None
assert engine.db is None

def test_close_is_idempotent(self) -> None:
"""Calling close() twice doesn't raise."""
from agent.insights import InsightsEngine

engine = InsightsEngine.__new__(InsightsEngine)
engine._conn = MagicMock()
engine.db = MagicMock()

engine.close()
engine.close() # second call — must not raise

assert engine._conn is None
assert engine.db is None

def test_close_on_fresh_instance(self) -> None:
"""close() on an instance that was never opened is safe."""
from agent.insights import InsightsEngine

engine = InsightsEngine.__new__(InsightsEngine)
engine._conn = None
engine.db = None

engine.close() # must not raise
assert engine._conn is None
assert engine.db is None


# ── SessionStore.close() ───────────────────────────────────────────────────


class TestSessionStoreClose:
"""Tests for gateway/session.py SessionStore.close()."""

def test_close_nulls_db(self) -> None:
"""close() sets _db to None."""
from gateway.session import SessionStore

store = SessionStore.__new__(SessionStore)
store._db = MagicMock()

store.close()

assert store._db is None

def test_close_is_idempotent(self) -> None:
"""Calling close() twice doesn't raise."""
from gateway.session import SessionStore

store = SessionStore.__new__(SessionStore)
store._db = MagicMock()

store.close()
store.close() # second call

assert store._db is None

def test_close_when_db_is_none(self) -> None:
"""close() when _db is already None is a no-op."""
from gateway.session import SessionStore

store = SessionStore.__new__(SessionStore)
store._db = None

store.close() # must not raise
assert store._db is None


# ── SessionManager.close() (acp_adapter) ───────────────────────────────────


class TestSessionManagerClose:
"""Tests for acp_adapter/session.py SessionManager.close()."""

def test_close_nulls_db_instance(self) -> None:
"""close() sets _db_instance to None."""
from acp_adapter.session import SessionManager

mgr = SessionManager.__new__(SessionManager)
mgr._db_instance = MagicMock()

mgr.close()

assert mgr._db_instance is None

def test_close_is_idempotent(self) -> None:
"""Calling close() twice doesn't raise."""
from acp_adapter.session import SessionManager

mgr = SessionManager.__new__(SessionManager)
mgr._db_instance = MagicMock()

mgr.close()
mgr.close() # second call

assert mgr._db_instance is None

def test_close_when_db_instance_is_none(self) -> None:
"""close() when _db_instance is already None is a no-op."""
from acp_adapter.session import SessionManager

mgr = SessionManager.__new__(SessionManager)
mgr._db_instance = None

mgr.close() # must not raise
assert mgr._db_instance is None