[fix] - Close the memory DB handle on setup failure, stamp the schema once per file, cache compiled memory patterns

memory/policy.py

@@ -4,6 +4,7 @@
 
 import logging
 import re
+from functools import lru_cache
 from typing import Any
 
 from utils.errors import PromptInjectionError
@@ -25,8 +26,20 @@ def require_reason(reason: str) -> None:
         raise ValueError("reason must not be empty")
 
 
-def _compile_patterns(base: list[str], cfg: dict[str, Any]) -> list[tuple[str, re.Pattern[str]]]:
+def _compile_patterns(base: list[str], cfg: dict[str, Any]) -> tuple[tuple[str, re.Pattern[str]], ...]:
     pf = (cfg.get("policy") or {}).get("prompt_filter") or {}
+    # Non-str entries become None (skipped below) so the key stays hashable and
+    # each entry keeps its config-list index for the warning.
+    banned = tuple(p if isinstance(p, str) else None for p in (pf.get("banned_patterns") or []))
+    return _compile_cached(tuple(base), banned)
+
+
+@lru_cache(maxsize=8)
+def _compile_cached(base: tuple[str, ...], banned: tuple[str | None, ...]) -> tuple[tuple[str, re.Pattern[str]], ...]:
+    # Cached per distinct (base, banned_patterns) pair: scan_content runs on
+    # every propose and up to twice per apply, and recompiling each time also
+    # re-logged a broken pattern's warning on every memory write. Same
+    # compile-once shape as utils/sanitizer.py.
     compiled: list[tuple[str, re.Pattern[str]]] = []
     seen: set[str] = set()
     flags = re.IGNORECASE | re.DOTALL
@@ -43,12 +56,12 @@ def _compile_patterns(base: list[str], cfg: dict[str, Any]) -> list[tuple[str, r
                 "memory banned pattern failed to compile (%s); it is skipped",
                 exc,
             )
-    for idx, p in enumerate(pf.get("banned_patterns") or []):
-        if not isinstance(p, str) or p in seen:
+    for idx, entry in enumerate(banned):
+        if entry is None or entry in seen:
             continue
         try:
-            compiled.append((p, re.compile(p, flags)))
-            seen.add(p)
+            compiled.append((entry, re.compile(entry, flags)))
+            seen.add(entry)
         except re.error as exc:
             # Config-list index, matching utils/sanitizer.py. Combined-list
             # index would shift with ENFORCED vs OWASP base length and dedup.
@@ -57,7 +70,7 @@ def _compile_patterns(base: list[str], cfg: dict[str, Any]) -> list[tuple[str, r
                 idx,
                 exc,
             )
-    return compiled
+    return tuple(compiled)
 
 
 def scan_content(content: str, cfg: dict[str, Any], *, enforced: bool = True) -> list[str]:

memory/store.py

@@ -87,15 +87,32 @@ def connect(cfg: Mapping[str, Any]) -> sqlite3.Connection:
         except OSError:
             logger.warning("Could not harden memory DB permissions to 0600: %s", path)
     conn = sqlite3.connect(str(path), check_same_thread=False)
-    conn.row_factory = sqlite3.Row
-    conn.execute("PRAGMA journal_mode=WAL")
-    conn.execute("PRAGMA foreign_keys=ON")
-    conn.execute("PRAGMA busy_timeout=5000")
-    _ensure_schema(conn)
+    try:
+        conn.row_factory = sqlite3.Row
+        conn.execute("PRAGMA journal_mode=WAL")
+        conn.execute("PRAGMA foreign_keys=ON")
+        conn.execute("PRAGMA busy_timeout=5000")
+        _ensure_schema(conn)
+    except BaseException:
+        # A corrupt file ("file is not a database") or a locked DB fails here;
+        # without the close every such request leaked one SQLite handle until GC.
+        conn.close()
+        raise
     return conn
 
 
+# Stamped into PRAGMA user_version once the DDL below has run. Every read helper
+# opens a fresh connection (search_facts_fts runs per query when fusion is on),
+# and re-running eleven CREATE ... IF NOT EXISTS statements -- plus the implicit
+# COMMIT executescript issues -- on each one is wasted work. The stamp lives in
+# the DB file itself, so a deleted/recreated DB (version 0) gets its schema
+# again with no process-wide cache to go stale. Bump it when the DDL changes.
+_SCHEMA_VERSION = 1
+
+
 def _ensure_schema(conn: sqlite3.Connection) -> None:
+    if conn.execute("PRAGMA user_version").fetchone()[0] == _SCHEMA_VERSION:
+        return
     conn.executescript(
         """
         CREATE TABLE IF NOT EXISTS facts (
@@ -175,6 +192,8 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
         CREATE INDEX IF NOT EXISTS idx_proposals_status ON memory_proposals(status);
         """
     )
+    # PRAGMA takes no bound parameters; _SCHEMA_VERSION is a module int constant.
+    conn.execute(f"PRAGMA user_version = {int(_SCHEMA_VERSION)}")
     conn.commit()
 
 

tests/test_memory_policy.py

@@ -4,10 +4,26 @@
 
 import pytest
 
-from memory.policy import check_content_size, check_tags, enforce_content, require_reason, scan_content
+from memory.policy import (
+    _compile_cached,
+    check_content_size,
+    check_tags,
+    enforce_content,
+    require_reason,
+    scan_content,
+)
 from utils.errors import PromptInjectionError
 
 
+@pytest.fixture(autouse=True)
+def _fresh_pattern_cache():
+    # _compile_cached logs a broken pattern once per distinct pattern list; clear
+    # it so a warning-asserting test never depends on what ran before it.
+    _compile_cached.cache_clear()
+    yield
+    _compile_cached.cache_clear()
+
+
 CFG = {
     "memory": {"facts": {"max_content_chars": 32}},
     "policy": {
@@ -116,3 +132,30 @@ def test_scan_malformed_regex_uses_config_index_not_combined_list(caplog):
     assert compile_msgs
     assert all("banned_patterns entry #1" in msg for msg in compile_msgs)
     assert not any("entry #9" in msg or "entry #13" in msg for msg in compile_msgs)
+
+
+def test_compiled_patterns_are_cached_and_a_broken_pattern_warns_once(caplog):
+    # scan_content runs on every propose and up to twice per apply; it used to
+    # recompile every pattern and re-log a broken one's warning each time.
+    cfg = {"policy": {"prompt_filter": {"banned_patterns": [r"(unclosed", r"cyclaw-only-sentinel"]}}}
+    with caplog.at_level("WARNING", logger="cyclaw.memory.policy"):
+        for _ in range(5):
+            assert scan_content("please cyclaw-only-sentinel now", cfg, enforced=True)
+    warnings = [r for r in caplog.records if "failed to compile" in r.message]
+    assert len(warnings) == 1
+    assert _compile_cached.cache_info().hits == 4
+
+
+def test_pattern_cache_tracks_a_changed_banned_patterns_list():
+    # The cache key is the pattern list itself, so a different config never
+    # reuses another config's compiled set.
+    first = {"policy": {"prompt_filter": {"banned_patterns": [r"alpha-sentinel"]}}}
+    second = {"policy": {"prompt_filter": {"banned_patterns": [r"beta-sentinel"]}}}
+    assert scan_content("alpha-sentinel", first, enforced=True)
+    assert scan_content("alpha-sentinel", second, enforced=True) == []
+    assert scan_content("beta-sentinel", second, enforced=True)
+
+
+def test_non_string_banned_pattern_entries_are_skipped_without_breaking_the_cache():
+    cfg = {"policy": {"prompt_filter": {"banned_patterns": [{"not": "a str"}, 7, r"cyclaw-only-sentinel"]}}}
+    assert scan_content("cyclaw-only-sentinel", cfg, enforced=True)

tests/test_memory_store.py

@@ -2,13 +2,15 @@
 
 from __future__ import annotations
 
+import sqlite3
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
 
 import pytest
 
 from memory.policy import enforce_content, require_reason
 from memory.store import (
+    _SCHEMA_VERSION,
     apply_proposal,
     connect,
     count_active_facts,
@@ -351,3 +353,54 @@ def test_corrupt_proposal_json_columns_do_not_raise(mem_cfg):
     assert loaded.reason == "operator note"
     listed = list_proposals(mem_cfg)
     assert any(p.id == prop.id and p.payload == {} for p in listed)
+
+
+def test_connect_closes_the_handle_when_setup_fails(mem_cfg: dict, monkeypatch: pytest.MonkeyPatch) -> None:
+    # A corrupt file fails at the first PRAGMA; connect() used to leak the open
+    # handle (one per request) because it only returned on success.
+    closed: list[bool] = []
+
+    class _Tracking(sqlite3.Connection):
+        def close(self) -> None:
+            closed.append(True)
+            super().close()
+
+    real_connect = sqlite3.connect
+    monkeypatch.setattr(sqlite3, "connect", lambda *a, **k: real_connect(*a, factory=_Tracking, **k))
+    Path(mem_cfg["memory"]["db_path"]).write_bytes(b"this is not a sqlite database" * 64)
+    with pytest.raises(sqlite3.DatabaseError):
+        connect(mem_cfg)
+    assert closed == [True]
+
+
+def test_schema_ddl_runs_once_per_db_file_and_again_for_a_recreated_one(
+    mem_cfg: dict, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Every read helper opens a fresh connection; the DDL script must run only
+    # when the file is not yet stamped, and again for a deleted/recreated DB.
+    scripts: list[str] = []
+
+    class _Counting(sqlite3.Connection):
+        def executescript(self, sql: str) -> sqlite3.Cursor:
+            scripts.append(sql)
+            return super().executescript(sql)
+
+    real_connect = sqlite3.connect
+    monkeypatch.setattr(sqlite3, "connect", lambda *a, **k: real_connect(*a, factory=_Counting, **k))
+
+    connect(mem_cfg).close()
+    assert len(scripts) == 1
+    connect(mem_cfg).close()
+    assert search_facts_fts(mem_cfg, "anything") == []
+    assert len(scripts) == 1, "an already-stamped DB must not re-run the schema DDL"
+
+    db = Path(mem_cfg["memory"]["db_path"])
+    for suffix in ("", "-wal", "-shm"):
+        Path(f"{db}{suffix}").unlink(missing_ok=True)
+    conn = connect(mem_cfg)
+    try:
+        assert len(scripts) == 2
+        assert conn.execute("PRAGMA user_version").fetchone()[0] == _SCHEMA_VERSION
+        assert conn.execute("SELECT name FROM sqlite_master WHERE name = 'facts'").fetchone() is not None
+    finally:
+        conn.close()