fix(db): close SQLite FDs on context exit and fallback gracefully if FTS5 is missing

hermes_cli/kanban_db.py

@@ -1005,6 +1005,38 @@ def _validate_sqlite_header(path: Path) -> None:
     )
 
 
+class _ConnContext(sqlite3.Connection):
+    """sqlite3.Connection subclass that closes itself on context-manager exit.
+
+    ``sqlite3.Connection.__exit__`` only commits or rolls back the active
+    transaction; it deliberately does NOT close the file descriptor (see
+    CPython ``Modules/_sqlite/connection.c``).  Every ``with connect() as
+    conn:`` call site therefore leaks an open FD to kanban.db + its WAL
+    file.  In long-lived gateway / dashboard processes that route every
+    kanban operation through ``connect()``, these accumulate until the
+    process hits the kernel FD limit (``[Errno 24] Too many open files``).
+    Production incident: #33159.
+
+    This wrapper overrides ``__exit__`` to add ``self.close()`` so that
+    ``with connect() as conn:`` is safe without any call-site changes.
+    The ``connect_closing()`` helper is kept for back-compat but is now
+    redundant — it delegates to this class transparently.
+    """
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        # Let the base class handle commit/rollback first.
+        try:
+            super().__exit__(exc_type, exc_val, exc_tb)
+        finally:
+            # Always close the FD regardless of transaction outcome.
+            try:
+                self.close()
+            except Exception:
+                pass
+        # Return False (do not suppress exceptions) — same as base class.
+        return False
+
+
 def connect(
     db_path: Optional[Path] = None,
     *,
@@ -1035,7 +1067,12 @@ def connect(
     path.parent.mkdir(parents=True, exist_ok=True)
     _validate_sqlite_header(path)
     resolved = str(path.resolve())
-    conn = sqlite3.connect(str(path), isolation_level=None, timeout=30)
+    conn = sqlite3.connect(
+        str(path),
+        factory=_ConnContext,
+        isolation_level=None,
+        timeout=30,
+    )
     try:
         conn.row_factory = sqlite3.Row
         with _INIT_LOCK:

hermes_state.py

@@ -337,6 +337,10 @@ def __init__(self, db_path: Path = None):
 
         self._lock = threading.Lock()
         self._write_count = 0
+        self._fts_enabled = True
+        self._trigram_enabled = True
+        self._fts_unavailable_warned = False
+        self._trigram_unavailable_warned = False
         try:
             self._conn = sqlite3.connect(
                 str(self.db_path),
@@ -357,6 +361,12 @@ def __init__(self, db_path: Path = None):
 
             self._init_schema()
         except Exception as exc:
+            if hasattr(self, '_conn') and self._conn is not None:
+                try:
+                    self._conn.close()
+                except Exception:
+                    pass
+                self._conn = None
             # Capture the cause so /resume and friends can surface WHY the
             # session DB is unavailable instead of a bare "Session database
             # not available."  Callers that catch this exception keep their
@@ -462,6 +472,48 @@ def close(self):
                 self._conn.close()
                 self._conn = None
 
+    @staticmethod
+    def _is_fts5_unavailable_error(exc: sqlite3.OperationalError) -> bool:
+        err = str(exc).lower()
+        if "no such module" in err and "fts5" in err:
+            return True
+        if "no such tokenizer: trigram" in err:
+            return True
+        return False
+
+    @staticmethod
+    def _is_trigram_unavailable_error(exc: sqlite3.OperationalError) -> bool:
+        """True when only the trigram tokenizer is missing (FTS5 itself works)."""
+        return "no such tokenizer: trigram" in str(exc).lower()
+
+    def _warn_trigram_unavailable(self, exc: sqlite3.OperationalError) -> None:
+        """Log once that the trigram tokenizer is missing; base FTS5 stays enabled."""
+        if getattr(self, "_trigram_unavailable_warned", False):
+            return
+        self._trigram_unavailable_warned = True
+        logger.info(
+            "SQLite trigram tokenizer unavailable for %s "
+            "(requires SQLite >= 3.34, this build is %s); "
+            "CJK/substring search will fall back to LIKE: %s",
+            self.db_path,
+            sqlite3.sqlite_version,
+            exc,
+        )
+
+    def _warn_fts5_unavailable(self, exc: sqlite3.OperationalError) -> None:
+        self._fts_enabled = False
+        if getattr(self, "_fts_unavailable_warned", False):
+            return
+        self._fts_unavailable_warned = True
+        logger.warning(
+            "SQLite FTS5 unavailable for %s; full-text session search "
+            "disabled. Run `hermes update` to rebuild the venv with a "
+            "current Python (managed uv guarantees FTS5). "
+            "(underlying error: %s)",
+            self.db_path,
+            exc,
+        )
+
     @staticmethod
     def _parse_schema_columns(schema_sql: str) -> Dict[str, Dict[str, str]]:
         """Extract expected columns per table from SCHEMA_SQL.
@@ -608,61 +660,73 @@ def _init_schema(self):
                 # FTS_TRIGRAM_SQL below, but existing rows need a one-time
                 # backfill into the FTS index.
                 try:
-                    cursor.execute("SELECT * FROM messages_fts_trigram LIMIT 0")
-                    _fts_trigram_exists = True
-                except sqlite3.OperationalError:
-                    _fts_trigram_exists = False
-                if not _fts_trigram_exists:
-                    cursor.executescript(FTS_TRIGRAM_SQL)
-                    cursor.execute(
-                        "INSERT INTO messages_fts_trigram(rowid, content) "
-                        "SELECT id, content FROM messages WHERE content IS NOT NULL"
-                    )
+                    try:
+                        cursor.execute("SELECT * FROM messages_fts_trigram LIMIT 0")
+                        _fts_trigram_exists = True
+                    except sqlite3.OperationalError:
+                        _fts_trigram_exists = False
+                    if not _fts_trigram_exists:
+                        cursor.executescript(FTS_TRIGRAM_SQL)
+                        cursor.execute(
+                            "INSERT INTO messages_fts_trigram(rowid, content) "
+                            "SELECT id, content FROM messages WHERE content IS NOT NULL"
+                        )
+                except sqlite3.OperationalError as exc:
+                    if self._is_fts5_unavailable_error(exc) or self._is_trigram_unavailable_error(exc):
+                        pass
+                    else:
+                        raise
             if current_version < 11:
                 # v11: re-index FTS5 tables to cover tool_name + tool_calls and
                 # switch from external-content to inline mode. Existing DBs have
                 # old-schema FTS tables and triggers that IF NOT EXISTS won't
                 # overwrite, so we drop them explicitly and let the post-migration
                 # existence checks (below) recreate them from FTS_SQL /
                 # FTS_TRIGRAM_SQL, then backfill every message row. Fixes #16751.
-                for _trig in (
-                    "messages_fts_insert",
-                    "messages_fts_delete",
-                    "messages_fts_update",
-                    "messages_fts_trigram_insert",
-                    "messages_fts_trigram_delete",
-                    "messages_fts_trigram_update",
-                ):
-                    try:
-                        cursor.execute(f"DROP TRIGGER IF EXISTS {_trig}")
-                    except sqlite3.OperationalError:
-                        pass
-                for _tbl in ("messages_fts", "messages_fts_trigram"):
-                    try:
-                        cursor.execute(f"DROP TABLE IF EXISTS {_tbl}")
-                    except sqlite3.OperationalError:
+                try:
+                    for _trig in (
+                        "messages_fts_insert",
+                        "messages_fts_delete",
+                        "messages_fts_update",
+                        "messages_fts_trigram_insert",
+                        "messages_fts_trigram_delete",
+                        "messages_fts_trigram_update",
+                    ):
+                        try:
+                            cursor.execute(f"DROP TRIGGER IF EXISTS {_trig}")
+                        except sqlite3.OperationalError:
+                            pass
+                    for _tbl in ("messages_fts", "messages_fts_trigram"):
+                        try:
+                            cursor.execute(f"DROP TABLE IF EXISTS {_tbl}")
+                        except sqlite3.OperationalError:
+                            pass
+                    # Recreate virtual tables + triggers with the new inline-mode
+                    # schema that indexes content || tool_name || tool_calls.
+                    cursor.executescript(FTS_SQL)
+                    cursor.executescript(FTS_TRIGRAM_SQL)
+                    # Backfill both indexes from every existing messages row.
+                    cursor.execute(
+                        "INSERT INTO messages_fts(rowid, content) "
+                        "SELECT id, "
+                        "COALESCE(content, '') || ' ' || "
+                        "COALESCE(tool_name, '') || ' ' || "
+                        "COALESCE(tool_calls, '') "
+                        "FROM messages"
+                    )
+                    cursor.execute(
+                        "INSERT INTO messages_fts_trigram(rowid, content) "
+                        "SELECT id, "
+                        "COALESCE(content, '') || ' ' || "
+                        "COALESCE(tool_name, '') || ' ' || "
+                        "COALESCE(tool_calls, '') "
+                        "FROM messages"
+                    )
+                except sqlite3.OperationalError as exc:
+                    if self._is_fts5_unavailable_error(exc) or self._is_trigram_unavailable_error(exc):
                         pass
-                # Recreate virtual tables + triggers with the new inline-mode
-                # schema that indexes content || tool_name || tool_calls.
-                cursor.executescript(FTS_SQL)
-                cursor.executescript(FTS_TRIGRAM_SQL)
-                # Backfill both indexes from every existing messages row.
-                cursor.execute(
-                    "INSERT INTO messages_fts(rowid, content) "
-                    "SELECT id, "
-                    "COALESCE(content, '') || ' ' || "
-                    "COALESCE(tool_name, '') || ' ' || "
-                    "COALESCE(tool_calls, '') "
-                    "FROM messages"
-                )
-                cursor.execute(
-                    "INSERT INTO messages_fts_trigram(rowid, content) "
-                    "SELECT id, "
-                    "COALESCE(content, '') || ' ' || "
-                    "COALESCE(tool_name, '') || ' ' || "
-                    "COALESCE(tool_calls, '') "
-                    "FROM messages"
-                )
+                    else:
+                        raise
             if current_version < SCHEMA_VERSION:
                 cursor.execute(
                     "UPDATE schema_version SET version = ?",
@@ -681,14 +745,43 @@ def _init_schema(self):
         # FTS5 setup (separate because CREATE VIRTUAL TABLE can't be in executescript with IF NOT EXISTS reliably)
         try:
             cursor.execute("SELECT * FROM messages_fts LIMIT 0")
-        except sqlite3.OperationalError:
-            cursor.executescript(FTS_SQL)
+        except sqlite3.OperationalError as exc:
+            if self._is_fts5_unavailable_error(exc):
+                self._warn_fts5_unavailable(exc)
+            else:
+                try:
+                    cursor.executescript(FTS_SQL)
+                except sqlite3.OperationalError as exc_inner:
+                    if self._is_fts5_unavailable_error(exc_inner):
+                        self._warn_fts5_unavailable(exc_inner)
+                    else:
+                        raise
 
         # Trigram FTS5 for CJK/substring search
-        try:
-            cursor.execute("SELECT * FROM messages_fts_trigram LIMIT 0")
-        except sqlite3.OperationalError:
-            cursor.executescript(FTS_TRIGRAM_SQL)
+        if self._fts_enabled:
+            try:
+                cursor.execute("SELECT * FROM messages_fts_trigram LIMIT 0")
+            except sqlite3.OperationalError as exc:
+                if self._is_trigram_unavailable_error(exc):
+                    self._warn_trigram_unavailable(exc)
+                    self._trigram_enabled = False
+                elif self._is_fts5_unavailable_error(exc):
+                    self._warn_fts5_unavailable(exc)
+                    self._trigram_enabled = False
+                else:
+                    try:
+                        cursor.executescript(FTS_TRIGRAM_SQL)
+                    except sqlite3.OperationalError as exc_inner:
+                        if self._is_trigram_unavailable_error(exc_inner):
+                            self._warn_trigram_unavailable(exc_inner)
+                            self._trigram_enabled = False
+                        elif self._is_fts5_unavailable_error(exc_inner):
+                            self._warn_fts5_unavailable(exc_inner)
+                            self._trigram_enabled = False
+                        else:
+                            raise
+        else:
+            self._trigram_enabled = False
 
         self._conn.commit()
 

tests/hermes_cli/test_kanban_db.py

@@ -1981,13 +1981,13 @@ def test_connect_falls_back_to_delete_on_locking_protocol(kanban_home, caplog):
 
     real_connect = _sqlite3.connect
 
-    class _WalBlockingConnection(_sqlite3.Connection):
-        def execute(self, sql, *args, **kwargs):  # type: ignore[override]
-            if "journal_mode=wal" in sql.lower().replace(" ", ""):
-                raise _sqlite3.OperationalError("locking protocol")
-            return super().execute(sql, *args, **kwargs)
-
     def wal_blocking_connect(*args, **kwargs):
+        factory = kwargs.pop("factory", _sqlite3.Connection)
+        class _WalBlockingConnection(factory):
+            def execute(self, sql, *args, **kwargs):  # type: ignore[override]
+                if "journal_mode=wal" in sql.lower().replace(" ", ""):
+                    raise _sqlite3.OperationalError("locking protocol")
+                return super().execute(sql, *args, **kwargs)
         return real_connect(
             *args, factory=_WalBlockingConnection, **kwargs
         )
@@ -2981,3 +2981,66 @@ def test_detect_stale_does_not_tick_failure_counter(kanban_home, monkeypatch):
         assert "stale" in kinds, (
             f"Expected 'stale' event in task_events; got {kinds!r}"
         )
+
+
+def test_connect_closes_fd_on_context_manager_exit(kanban_home):
+    """Regression #33159: with kb.connect() as conn: must close FD on exit.
+
+    sqlite3.Connection.__exit__ only commits/rolls back; it does NOT close
+    the FD. The _ConnContext wrapper must override __exit__ to call close().
+    """
+    import gc
+    import os
+
+    try:
+        fd_before = set(int(f) for f in os.listdir(f"/proc/{os.getpid()}/fd"))
+    except OSError:
+        pytest.skip("Cannot enumerate /proc FDs on this platform")
+
+    with kb.connect() as conn:
+        _ = kb.list_tasks(conn)
+
+    gc.collect()
+
+    fd_after = set(int(f) for f in os.listdir(f"/proc/{os.getpid()}/fd"))
+    new_fds = fd_after - fd_before
+    assert len(new_fds) <= 2, (
+        f"FD leak: {len(new_fds)} new FDs remain open after "
+        f"'with kb.connect() as conn:' exit. new_fds={new_fds}"
+    )
+
+
+def test_connect_context_manager_closes_on_exception(kanban_home):
+    """FD must be closed even when an exception is raised inside the with block."""
+    import gc
+    import os
+
+    try:
+        fd_before = set(int(f) for f in os.listdir(f"/proc/{os.getpid()}/fd"))
+    except OSError:
+        pytest.skip("Cannot enumerate /proc FDs on this platform")
+
+    with pytest.raises(RuntimeError):
+        with kb.connect() as conn:
+            _ = kb.list_tasks(conn)
+            raise RuntimeError("intentional error")
+
+    gc.collect()
+    fd_after = set(int(f) for f in os.listdir(f"/proc/{os.getpid()}/fd"))
+    new_fds = fd_after - fd_before
+    assert len(new_fds) <= 2, (
+        f"FD leak on exception: {len(new_fds)} new FDs. "
+        f"_ConnContext.__exit__ must close even on error."
+    )
+
+
+def test_connect_returns_conn_context_type(kanban_home):
+    """connect() must return a _ConnContext, not a bare sqlite3.Connection."""
+    from hermes_cli.kanban_db import _ConnContext
+    conn = kb.connect()
+    try:
+        assert isinstance(conn, _ConnContext), (
+            f"connect() returned {type(conn).__name__}, expected _ConnContext"
+        )
+    finally:
+        conn.close()