fix(gateway,tools,agent): close leaked SQLite connections in delivery…

agent/verification_evidence.py

@@ -13,10 +13,11 @@
 import sqlite3
 import tempfile
 import threading
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Iterator, Optional
 
 from hermes_constants import get_hermes_home
 
@@ -63,13 +64,38 @@ def _connect() -> sqlite3.Connection:
     path = _db_path()
     path.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(path)
-    conn.execute("PRAGMA journal_mode=WAL")
-    conn.execute("PRAGMA busy_timeout=5000")
     conn.row_factory = sqlite3.Row
-    _ensure_schema(conn)
+    try:
+        conn.execute("PRAGMA journal_mode=WAL")
+        conn.execute("PRAGMA busy_timeout=5000")
+        _ensure_schema(conn)
+    except Exception:
+        # A PRAGMA/DDL failure after a successful connect() must not leak the
+        # just-opened connection back to the caller.
+        conn.close()
+        raise
     return conn
 
 
+@contextmanager
+def _transaction() -> Iterator[sqlite3.Connection]:
+    """Open a connection, commit/rollback on exit, and ALWAYS close it.
+
+    ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the
+    transaction; they do not close the connection. Using ``with _connect()``
+    alone therefore leaks a connection — and its WAL/SHM file descriptors — on
+    every call, deferring the close to the garbage collector, which over a
+    long-running process can exhaust ``RLIMIT_NOFILE`` (the cron-ledger sibling
+    of this bug was #69567 / PR #69594).
+    """
+    conn = _connect()
+    try:
+        with conn:
+            yield conn
+    finally:
+        conn.close()
+
+
 def _ensure_schema(conn: sqlite3.Connection) -> None:
     conn.execute(
         """
@@ -452,7 +478,7 @@ def record_terminal_result(
 
     created_at = _utc_now()
     with _DB_LOCK:
-        with _connect() as conn:
+        with _transaction() as conn:
             cur = conn.execute(
                 """
                 INSERT INTO verification_events(
@@ -518,7 +544,7 @@ def mark_workspace_edited(
     edited_at = _utc_now()
 
     with _DB_LOCK:
-        with _connect() as conn:
+        with _transaction() as conn:
             row = conn.execute(
                 """
                 SELECT changed_paths_json FROM verification_state
@@ -568,7 +594,7 @@ def verification_status(
     sid = str(session_id or "default")
     root = str(facts.get("root") or Path(cwd or ".").resolve())
     with _DB_LOCK:
-        with _connect() as conn:
+        with _transaction() as conn:
             state = conn.execute(
                 """
                 SELECT last_event_id, last_edit_at, changed_paths_json

gateway/delivery_ledger.py

@@ -46,7 +46,8 @@
 import sqlite3
 import threading
 import time
-from typing import Any, Dict, List, Optional
+from contextlib import contextmanager
+from typing import Any, Dict, Iterator, List, Optional
 
 from hermes_constants import get_hermes_home
 
@@ -78,6 +79,17 @@ def _connect() -> sqlite3.Connection:
     path = _db_path()
     path.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(path, timeout=10)
+    try:
+        _initialize_schema(conn)
+    except Exception:
+        # A PRAGMA/DDL failure after a successful connect() must not leak the
+        # just-opened connection back to the caller.
+        conn.close()
+        raise
+    return conn
+
+
+def _initialize_schema(conn: sqlite3.Connection) -> None:
     conn.execute("PRAGMA journal_mode=WAL")
     conn.execute(
         """CREATE TABLE IF NOT EXISTS delivery_obligations (
@@ -96,7 +108,26 @@ def _connect() -> sqlite3.Connection:
             last_error TEXT
         )"""
     )
-    return conn
+
+
+@contextmanager
+def _transaction() -> Iterator[sqlite3.Connection]:
+    """Open a connection, commit/rollback on exit, and ALWAYS close it.
+
+    ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the
+    transaction; they do not close the connection. Using ``with _connect()``
+    alone therefore leaks a connection — and its WAL/SHM file descriptors — on
+    every call, deferring the close to the garbage collector. On a long-running
+    gateway that exhausts ``RLIMIT_NOFILE`` (the cron-ledger sibling of this
+    bug was #69567 / PR #69594). ``record_obligation`` runs on every outbound
+    final response, so this ledger is the highest-frequency leaker.
+    """
+    conn = _connect()
+    try:
+        with conn:
+            yield conn
+    finally:
+        conn.close()
 
 
 def _owner_stamp() -> tuple[int, Optional[int]]:
@@ -164,7 +195,7 @@ def record_obligation(
     """Record a final response as owed to the platform (state='pending')."""
     now = time.time()
     pid, started = _owner_stamp()
-    with _DB_LOCK, _connect() as conn:
+    with _DB_LOCK, _transaction() as conn:
         conn.execute(
             """INSERT OR REPLACE INTO delivery_obligations
                (obligation_id, session_key, platform, chat_id, thread_id,
@@ -191,7 +222,7 @@ def mark_failed(obligation_id: str, error: str = "") -> None:
 
 
 def _update_state(obligation_id: str, state: str, error: str = "") -> None:
-    with _DB_LOCK, _connect() as conn:
+    with _DB_LOCK, _transaction() as conn:
         conn.execute(
             """UPDATE delivery_obligations
                SET state=?, updated_at=?, last_error=?
@@ -224,7 +255,7 @@ def sweep_recoverable(
     now = now if now is not None else time.time()
     pid, started = _owner_stamp()
     claimed: List[Dict[str, Any]] = []
-    with _DB_LOCK, _connect() as conn:
+    with _DB_LOCK, _transaction() as conn:
         rows = conn.execute(
             """SELECT obligation_id, session_key, platform, chat_id, thread_id,
                       content, state, attempts, created_at,
@@ -277,7 +308,7 @@ def _prune(now: Optional[float] = None) -> None:
     now = now if now is not None else time.time()
     cutoff = now - _RETENTION_SECONDS
     try:
-        with _connect() as conn:
+        with _transaction() as conn:
             conn.execute(
                 """DELETE FROM delivery_obligations
                    WHERE state IN ('delivered', 'abandoned') AND updated_at < ?""",
@@ -321,7 +352,7 @@ def ledger_enabled(config: Optional[Dict[str, Any]] = None) -> bool:
 
 def debug_rows(limit: int = 20) -> str:
     """Human-readable dump for ad-hoc inspection (sqlite3-free path)."""
-    with _DB_LOCK, _connect() as conn:
+    with _DB_LOCK, _transaction() as conn:
         rows = conn.execute(
             """SELECT obligation_id, session_key, state, attempts,
                       created_at, updated_at, last_error

tests/agent/test_verification_evidence_fd_leak.py

@@ -0,0 +1,126 @@
+"""Regression: the verification-evidence ledger must close every connection.
+
+Sibling of the cron execution-ledger leak (#69567 / PR #69594). The evidence
+ledger used ``with _connect() as conn:`` where the connection context manager
+commits/rolls back but never closes, leaking the db/-wal/-shm file descriptors
+on every recorded terminal result, workspace edit, and status read. These tests
+fail if the deterministic ``close()`` is ever removed again.
+"""
+
+import sqlite3
+
+import pytest
+
+from agent import verification_evidence as ve
+
+
+class _TrackingConnection:
+    """Delegates to a real sqlite3.Connection while recording close() calls.
+
+    sqlite3.Connection is a static C type: it has no per-instance __dict__ and
+    its methods can't be monkeypatched, so open/close tracking is done via a
+    delegating wrapper returned in place of the real connection.
+    """
+
+    def __init__(self, real, closed_ids):
+        object.__setattr__(self, "_real", real)
+        object.__setattr__(self, "_closed_ids", closed_ids)
+
+    def close(self):
+        self._closed_ids.append(id(self._real))
+        self._real.close()
+
+    def __enter__(self):
+        self._real.__enter__()
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return self._real.__exit__(exc_type, exc, tb)
+
+    def __getattr__(self, name):
+        return getattr(self._real, name)
+
+    def __setattr__(self, name, value):
+        setattr(self._real, name, value)
+
+
+def _point_ledger(monkeypatch, tmp_path):
+    monkeypatch.setattr(ve, "_db_path", lambda: tmp_path / "verification_evidence.db")
+    return ve
+
+
+def _track_connections(monkeypatch):
+    opened, closed = [], []
+    real_connect = sqlite3.connect
+
+    def tracking_connect(*args, **kwargs):
+        conn = real_connect(*args, **kwargs)
+        opened.append(id(conn))
+        return _TrackingConnection(conn, closed)
+
+    monkeypatch.setattr(ve.sqlite3, "connect", tracking_connect)
+    return opened, closed
+
+
+def _python_project(root):
+    (root / "pyproject.toml").write_text("[tool.pytest.ini_options]\n")
+
+
+def test_ledger_operations_close_every_connection(monkeypatch, tmp_path):
+    """Recording, editing, and status reads must close every connection opened."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _point_ledger(monkeypatch, tmp_path)
+    _python_project(tmp_path)
+    opened, closed = _track_connections(monkeypatch)
+
+    ve.record_terminal_result(
+        command="python -m pytest tests/test_calc.py::test_even -q",
+        cwd=tmp_path, session_id="s1", exit_code=0, output="1 passed",
+    )
+    ve.verification_status(session_id="s1", cwd=tmp_path)
+    ve.mark_workspace_edited(session_id="s1", cwd=tmp_path, paths=["mod.py"])
+
+    assert opened, "expected at least one connection to be opened"
+    assert len(opened) == len(closed)
+    assert set(opened) == set(closed)
+
+
+def test_exception_during_operation_still_closes_connection(monkeypatch, tmp_path):
+    """A failing statement inside the transaction must roll back and close."""
+    _point_ledger(monkeypatch, tmp_path)
+    opened, closed = _track_connections(monkeypatch)
+
+    with pytest.raises(sqlite3.IntegrityError):
+        with ve._transaction() as conn:
+            # Missing NOT NULL columns -> constraint failure inside the block.
+            conn.execute("INSERT INTO verification_events (id) VALUES (1)")
+
+    assert len(opened) == 1
+    assert len(closed) == 1
+
+
+def test_schema_init_failure_still_closes_connection(monkeypatch, tmp_path):
+    """A PRAGMA/DDL failure after connect() must still close the connection."""
+    _point_ledger(monkeypatch, tmp_path)
+    opened, closed = [], []
+    real_connect = sqlite3.connect
+
+    class _FailingSchemaConnection(_TrackingConnection):
+        def execute(self, sql, *args, **kwargs):
+            if "CREATE TABLE" in sql:
+                raise sqlite3.OperationalError("simulated schema init failure")
+            return self._real.execute(sql, *args, **kwargs)
+
+    def tracking_connect(*args, **kwargs):
+        conn = real_connect(*args, **kwargs)
+        opened.append(id(conn))
+        return _FailingSchemaConnection(conn, closed)
+
+    monkeypatch.setattr(ve.sqlite3, "connect", tracking_connect)
+
+    with pytest.raises(sqlite3.OperationalError):
+        with ve._transaction():
+            pass
+
+    assert len(opened) == 1
+    assert len(closed) == 1