perf(cache): scheduler-facing get_cached_request_context + concurrent-init DDL fix

.claude/rules/reflexio-patterns.md

@@ -6,8 +6,8 @@ paths:
 # Reflexio Architecture Guardrails
 
 ## Reflexio Instance
-- **NEVER** instantiate `Reflexio()` directly in API endpoints
-- **ALWAYS** use `get_reflexio()` from `server/cache/`
+- **NEVER** instantiate `Reflexio()` directly in API endpoints, or `RequestContext()` directly in the hot playbook-aggregation schedulers
+- **ALWAYS** use `get_reflexio()` from `server/cache/` — or `get_cached_request_context()` when a hot aggregation scheduler only needs the `RequestContext`. Direct construction in those per-tick loops rebuilds config decryption, storage client pools, and LLM clients for every org.
 
 ## Storage
 - **NEVER** import storage implementations directly

reflexio/server/cache/__init__.py

@@ -3,12 +3,14 @@
 from reflexio.server.cache.reflexio_cache import (
     clear_reflexio_cache,
     get_cache_stats,
+    get_cached_request_context,
     get_reflexio,
     invalidate_reflexio_cache,
 )
 
 __all__ = [
     "get_reflexio",
+    "get_cached_request_context",
     "invalidate_reflexio_cache",
     "clear_reflexio_cache",
     "get_cache_stats",

reflexio/server/cache/reflexio_cache.py

@@ -8,13 +8,18 @@
 from cachetools import TTLCache
 
 from reflexio.lib.reflexio_lib import Reflexio
+from reflexio.server.api_endpoints.request_context import RequestContext
 from reflexio.server.llm.llm_utils import positive_int_env
 from reflexio.server.tracing import profile_step
 
 logger = logging.getLogger(__name__)
 
-# Cache configuration
-REFLEXIO_CACHE_MAX_SIZE = positive_int_env("REFLEXIO_CACHE_MAX_SIZE", 100, logger)
+# Cache configuration.
+# The default must exceed the combined working set of the background
+# schedulers that page through the whole fleet (two independent 100-org
+# pages) plus request-path headroom — at 100, scheduler sweeps on a
+# 100+-org fleet would continuously evict warm request-path entries.
+REFLEXIO_CACHE_MAX_SIZE = positive_int_env("REFLEXIO_CACHE_MAX_SIZE", 512, logger)
 REFLEXIO_CACHE_TTL_SECONDS = 3600  # 1 hour safety net
 
 # Type alias for cache key: (org_id, storage_base_dir)
@@ -159,6 +164,11 @@ def get_reflexio(org_id: str, storage_base_dir: str | None = None) -> Reflexio:
         # Stale entry. Evict only if the cached version still matches
         # the one we just compared against — another thread may have
         # already replaced the entry while we were probing.
+        # Do NOT close() the evicted instance here: callers receive
+        # instances after the cache lock is released, so another thread
+        # may still be using the one being evicted. Closing its client
+        # pools would cause spurious mid-request failures; correct
+        # cleanup would need refcounting, for no demonstrated leak.
         with profile_step("reflexio.cache.evict_stale") as span:
             with _reflexio_cache_lock:
                 existing = _reflexio_cache.get(cache_key)
@@ -225,6 +235,31 @@ def get_reflexio(org_id: str, storage_base_dir: str | None = None) -> Reflexio:
             return reflexio
 
 
+def get_cached_request_context(
+    org_id: str, storage_base_dir: str | None = None
+) -> RequestContext:
+    """Get the RequestContext owned by the cached Reflexio instance.
+
+    For background schedulers that fan out per-org work: delegates to
+    :func:`get_reflexio` so they share the same cache entry, per-hit
+    config-version eviction, striped construction locks, and every
+    existing ``invalidate_reflexio_cache`` call site as the request
+    path — instead of rebuilding a ``RequestContext`` (config decrypt,
+    storage client pools, LiteLLM client) on every tick.
+
+    Named distinctly from the FastAPI ``get_request_context`` dependency
+    in ``api_endpoints/request_context.py`` to avoid collisions.
+
+    Args:
+        org_id (str): Organization ID
+        storage_base_dir (Optional[str]): Base directory for storage (self-host mode)
+
+    Returns:
+        RequestContext: The context of the cached (or newly constructed) instance.
+    """
+    return get_reflexio(org_id, storage_base_dir).request_context
+
+
 def invalidate_reflexio_cache(org_id: str, storage_base_dir: str | None = None) -> bool:
     """Invalidate cached Reflexio for specific org.
 

reflexio/server/services/storage/sqlite_storage/_base.py

@@ -68,13 +68,25 @@
 logger = logging.getLogger(__name__)
 
 _MINIMUM_SQLITE_VERSION = (3, 35, 0)
+_SQLITE_INITIALIZATION_LOCK_STRIPES = 64
+_sqlite_initialization_locks = tuple(
+    threading.Lock() for _ in range(_SQLITE_INITIALIZATION_LOCK_STRIPES)
+)
 
 
 # ---------------------------------------------------------------------------
 # Module-level helpers
 # ---------------------------------------------------------------------------
 
 
+def _get_sqlite_initialization_lock(db_path: str) -> threading.Lock:
+    """Return a bounded process-local lock for one SQLite database path."""
+    normalized_path = str(Path(db_path).resolve())
+    return _sqlite_initialization_locks[
+        hash(normalized_path) % _SQLITE_INITIALIZATION_LOCK_STRIPES
+    ]
+
+
 def _json_dumps(obj: Any) -> str | None:
     """Serialize a Python object to a JSON string, or None if the object is None."""
     if obj is None:
@@ -786,13 +798,18 @@ def __init__(
 
         # Ensure parent directory exists
         Path(db_path).parent.mkdir(parents=True, exist_ok=True)
-
-        # Open connection
-        self.conn = sqlite3.connect(db_path, check_same_thread=False)
-        self.conn.row_factory = sqlite3.Row
-        register_unicode_lexical_index_function(self.conn)
-        self.conn.execute("PRAGMA journal_mode=WAL")
-        self.conn.execute("PRAGMA foreign_keys=ON")
+        initialization_lock = _get_sqlite_initialization_lock(db_path)
+
+        # SQLite's journal-mode negotiation can fail immediately when another
+        # connection is cold-starting the same file. Serialize that setup by
+        # database path while allowing unrelated SQLite files to initialize in
+        # parallel.
+        with initialization_lock:
+            self.conn = sqlite3.connect(db_path, check_same_thread=False)
+            self.conn.row_factory = sqlite3.Row
+            register_unicode_lexical_index_function(self.conn)
+            self.conn.execute("PRAGMA journal_mode=WAL")
+            self.conn.execute("PRAGMA foreign_keys=ON")
 
         # LLM client for embeddings
         model_setting = SiteVarManager().get_site_var("llm_model_setting")
@@ -830,8 +847,10 @@ def __init__(
         # Optionally load sqlite-vec for native KNN vector search
         self._has_sqlite_vec = self._try_load_sqlite_vec()
 
-        # Create tables
-        self.migrate()
+        # Migrations use an instance-local lock, so separate storage instances
+        # for the same file also need the shared path lock.
+        with initialization_lock:
+            self.migrate()
 
     # ------------------------------------------------------------------
     # Transaction scope

reflexio/server/services/storage/sqlite_storage/playbook/_aggregation.py

@@ -109,9 +109,12 @@
 CREATE INDEX IF NOT EXISTS idx_playbook_aggregation_invalidation_retention
     ON playbook_aggregation_invalidation(processed_at)
     WHERE processed_at IS NOT NULL;
+"""
 
+AGGREGATION_TRIGGER_DDL = """
+BEGIN IMMEDIATE;
 DROP TRIGGER IF EXISTS capture_playbook_aggregation_hard_delete;
-CREATE TRIGGER capture_playbook_aggregation_hard_delete
+CREATE TRIGGER IF NOT EXISTS capture_playbook_aggregation_hard_delete
 BEFORE DELETE ON user_playbooks
 WHEN OLD.status IS NULL AND trim(OLD.agent_version) <> ''
 BEGIN
@@ -125,7 +128,7 @@
 END;
 
 DROP TRIGGER IF EXISTS retire_playbook_aggregation_cluster_on_agent_update;
-CREATE TRIGGER retire_playbook_aggregation_cluster_on_agent_update
+CREATE TRIGGER IF NOT EXISTS retire_playbook_aggregation_cluster_on_agent_update
 AFTER UPDATE OF status, playbook_status, content, trigger, rationale, embedding
 ON agent_playbooks
 WHEN (
@@ -165,7 +168,7 @@
 END;
 
 DROP TRIGGER IF EXISTS retire_playbook_aggregation_cluster_on_agent_delete;
-CREATE TRIGGER retire_playbook_aggregation_cluster_on_agent_delete
+CREATE TRIGGER IF NOT EXISTS retire_playbook_aggregation_cluster_on_agent_delete
 BEFORE DELETE ON agent_playbooks
 BEGIN
     UPDATE playbook_aggregation_item
@@ -191,11 +194,17 @@
     DELETE FROM playbook_aggregation_cluster
     WHERE agent_playbook_id = OLD.agent_playbook_id;
 END;
+COMMIT;
 """
 
 
 def init_playbook_aggregation_tables(conn: sqlite3.Connection) -> None:
     conn.executescript(AGGREGATION_DDL)
+    try:
+        conn.executescript(AGGREGATION_TRIGGER_DDL)
+    except Exception:
+        conn.rollback()
+        raise
     state_columns = {
         str(row[1])
         for row in conn.execute("PRAGMA table_info(playbook_aggregation_state)")