fix(memory): connection pooling + circuit breaker for memory_tree SQLite (#2206) (#2212)

src/openhuman/doctor/core.rs

@@ -72,6 +72,7 @@ pub fn run(config: &Config) -> Result<DoctorReport> {
     check_workspace(config, &mut items);
     check_daemon_state(config, &mut items);
     check_environment(&mut items);
+    check_memory_tree_db(config, &mut items);
 
     let errors = items
         .iter()
@@ -771,6 +772,70 @@ fn check_command_available(
     }
 }
 
+// ── Memory-tree DB health ────────────────────────────────────────
+
+/// Probe the memory-tree SQLite database and push a [`DiagnosticItem`].
+///
+/// - If the DB directory / file does not exist yet: `Warn` (not yet created).
+/// - If a stale `.db-shm` file is present alongside the DB: `Warn`.
+/// - If we can open the DB and run a basic probe query: `Ok`.
+/// - If the probe fails: `Error`.
+fn check_memory_tree_db(config: &Config, items: &mut Vec<DiagnosticItem>) {
+    let cat = "memory_tree_db";
+    let db_path = config.workspace_dir.join("memory_tree").join("chunks.db");
+
+    // ── Stale side-files (checked even when chunks.db is absent) ────
+    let base_name = db_path
+        .file_name()
+        .unwrap_or_default()
+        .to_string_lossy()
+        .into_owned();
+    let shm = db_path.with_file_name(format!("{base_name}-shm"));
+    let wal = db_path.with_file_name(format!("{base_name}-wal"));
+    for sidecar in [&shm, &wal] {
+        if sidecar.exists() {
+            items.push(DiagnosticItem::warn(
+                cat,
+                format!(
+                    "stale SQLite side-file present (may indicate unclean shutdown): {}",
+                    sidecar.display()
+                ),
+            ));
+        }
+    }
+
+    // ── File existence ──────────────────────────────────────────────
+    if !db_path.exists() {
+        items.push(DiagnosticItem::warn(
+            cat,
+            format!(
+                "DB not yet created (first ingest will initialise it): {}",
+                db_path.display()
+            ),
+        ));
+        return;
+    }
+
+    // ── Probe connection ─────────────────────────────────────────────
+    match crate::openhuman::memory::tree::store::with_connection(config, |conn| {
+        let n: i64 = conn.query_row("SELECT COUNT(*) FROM mem_tree_chunks", [], |r| r.get(0))?;
+        Ok(n)
+    }) {
+        Ok(count) => {
+            items.push(DiagnosticItem::ok(
+                cat,
+                format!("DB accessible at {} ({count} chunks)", db_path.display()),
+            ));
+        }
+        Err(err) => {
+            items.push(DiagnosticItem::error(
+                cat,
+                format!("DB probe failed at {}: {err:#}", db_path.display()),
+            ));
+        }
+    }
+}
+
 // ── Helpers ──────────────────────────────────────────────────────
 
 fn parse_rfc3339(input: &str) -> Option<DateTime<Utc>> {

src/openhuman/doctor/core_tests.rs

@@ -1,4 +1,11 @@
 use super::*;
+use tempfile::TempDir;
+
+fn test_config_in(tmp: &TempDir) -> Config {
+    let mut cfg = Config::default();
+    cfg.workspace_dir = tmp.path().to_path_buf();
+    cfg
+}
 
 #[test]
 fn config_validation_warns_no_channels() {
@@ -51,3 +58,56 @@ fn embedding_provider_validation_rejects_malformed_url() {
     let err = embedding_provider_validation_error("custom:not a url").expect("should fail");
     assert!(err.contains("invalid custom provider URL"), "{err}");
 }
+
+// ── check_memory_tree_db tests (#2206) ───────────────────────────────────────
+
+/// When the workspace exists but the DB file has never been created,
+/// `check_memory_tree_db` should push exactly one `Warn` item mentioning
+/// "not yet created".
+#[test]
+fn check_memory_tree_db_warns_when_db_missing() {
+    let tmp = TempDir::new().expect("tempdir");
+    let cfg = test_config_in(&tmp);
+
+    let mut items = vec![];
+    check_memory_tree_db(&cfg, &mut items);
+
+    assert_eq!(items.len(), 1, "expected exactly one diagnostic item");
+    assert_eq!(items[0].severity, Severity::Warn);
+    assert!(
+        items[0].message.contains("not yet created"),
+        "unexpected message: {}",
+        items[0].message
+    );
+}
+
+/// After `with_connection` has successfully initialised the DB, the probe
+/// should push an `Ok` item.
+#[test]
+fn check_memory_tree_db_ok_when_accessible() {
+    let tmp = TempDir::new().expect("tempdir");
+    let cfg = test_config_in(&tmp);
+
+    // Trigger DB creation.
+    crate::openhuman::memory::tree::store::with_connection(&cfg, |_conn| Ok(()))
+        .expect("DB init must succeed");
+
+    let mut items = vec![];
+    check_memory_tree_db(&cfg, &mut items);
+
+    // There may be a Warn about the SHM file on some platforms; there must
+    // be at least one Ok item about the DB being accessible.
+    let ok_items: Vec<_> = items
+        .iter()
+        .filter(|i| i.severity == Severity::Ok && i.category == "memory_tree_db")
+        .collect();
+    assert!(
+        !ok_items.is_empty(),
+        "expected at least one Ok memory_tree_db item; got: {items:?}"
+    );
+    assert!(
+        ok_items[0].message.contains("accessible"),
+        "unexpected ok message: {}",
+        ok_items[0].message
+    );
+}

src/openhuman/memory/tree/jobs/worker.rs

@@ -96,8 +96,21 @@ pub fn start(config: Config) {
                             if is_sqlite_busy(&err) {
                                 log::warn!(
                                     "[memory_tree::jobs] worker {idx} hit SQLite busy/locked, \
-                                     backing off: {err:#}"
+                                     backing off 1s: {err:#}"
                                 );
+                                tokio::time::sleep(Duration::from_secs(1)).await;
+                            } else if is_sqlite_io_transient(&err) {
+                                // I/O errors (IOERR_TRUNCATE 1546, IOERR_SHMMAP 4874,
+                                // CANTOPEN 14) or circuit breaker open — transient
+                                // filesystem / WAL condition. Back off 30 s and let the
+                                // connection cache try a fresh open on next poll. These
+                                // are NOT reported to Sentry (they are transient and were
+                                // flooding ~19K events/4 days, see #2206).
+                                log::warn!(
+                                    "[memory_tree::jobs] worker {idx} hit transient I/O error, \
+                                     backing off 30s: {err:#}"
+                                );
+                                tokio::time::sleep(Duration::from_secs(30)).await;
                             } else {
                                 crate::core::observability::report_error(
                                     &err,
@@ -105,8 +118,8 @@ pub fn start(config: Config) {
                                     "tree_jobs_worker",
                                     &[("worker_idx", &idx.to_string())],
                                 );
+                                tokio::time::sleep(Duration::from_secs(1)).await;
                             }
-                            tokio::time::sleep(Duration::from_secs(1)).await;
                         }
                     }
                 }
@@ -226,6 +239,37 @@ pub async fn run_once(config: &Config) -> Result<bool> {
     Ok(true)
 }
 
+/// Classify whether an error is a transient I/O failure that should be
+/// silently backed off without a Sentry report (#2206).
+///
+/// Covers:
+/// - `SQLITE_IOERR_TRUNCATE` (extended code 1546): WAL truncation failed —
+///   usually a transient filesystem hiccup.
+/// - `SQLITE_IOERR_SHMMAP` (extended code 4874): shared-memory mapping
+///   failed — WAL side-file temporarily unavailable.
+/// - `SQLITE_CANTOPEN` / `CannotOpen` (extended code 14): DB file temporarily
+///   inaccessible.
+/// - Text fallback: circuit breaker message, or rusqlite phrases that don't
+///   downcast cleanly after multiple `.context()` layers.
+fn is_sqlite_io_transient(err: &anyhow::Error) -> bool {
+    if let Some(rusqlite::Error::SqliteFailure(f, _)) = err.downcast_ref::<rusqlite::Error>() {
+        if matches!(f.extended_code, 1546 | 4874 | 14) {
+            return true;
+        }
+        if f.code == rusqlite::ErrorCode::CannotOpen {
+            return true;
+        }
+    }
+    // Text fallback for errors wrapped under `.context()` layers or
+    // emitted as plain `anyhow!` strings (e.g. circuit breaker message).
+    let msg = format!("{err:#}").to_ascii_lowercase();
+    msg.contains("circuit breaker open")
+        || msg.contains("disk i/o error")
+        || msg.contains("unable to open database file")
+        || msg.contains("xshmmap")
+        || msg.contains("truncate file")
+}
+
 /// Classify whether an error from `run_once` is a transient SQLite
 /// write-lock contention (`SQLITE_BUSY` or `SQLITE_LOCKED`).
 ///
@@ -335,4 +379,70 @@ mod tests {
         let err = anyhow::anyhow!("upstream returned 500: internal server error");
         assert!(!is_sqlite_busy(&err));
     }
+
+    // ── is_sqlite_io_transient tests (#2206) ─────────────────────────────
+
+    /// SQLITE_IOERR_TRUNCATE (extended code 1546) must be classified as
+    /// transient so the worker backs off without hitting Sentry.
+    #[test]
+    fn is_sqlite_io_transient_matches_ioerr_truncate() {
+        let raw = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::SystemIoFailure,
+                extended_code: 1546, // SQLITE_IOERR_TRUNCATE
+            },
+            Some("disk I/O error".into()),
+        );
+        assert!(is_sqlite_io_transient(&anyhow::Error::from(raw)));
+    }
+
+    /// SQLITE_IOERR_SHMMAP (extended code 4874) must be classified as
+    /// transient — WAL shared-memory mapping is a filesystem hiccup.
+    #[test]
+    fn is_sqlite_io_transient_matches_ioerr_shmmap() {
+        let raw = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::SystemIoFailure,
+                extended_code: 4874, // SQLITE_IOERR_SHMMAP
+            },
+            Some("xshmmap failed".into()),
+        );
+        assert!(is_sqlite_io_transient(&anyhow::Error::from(raw)));
+    }
+
+    /// SQLITE_CANTOPEN (code CannotOpen, extended code 14) must be
+    /// classified as transient — temporary inability to open the file.
+    #[test]
+    fn is_sqlite_io_transient_matches_cantopen() {
+        let raw = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::CannotOpen,
+                extended_code: 14, // SQLITE_CANTOPEN
+            },
+            Some("unable to open database file".into()),
+        );
+        assert!(is_sqlite_io_transient(&anyhow::Error::from(raw)));
+    }
+
+    /// The circuit breaker error message produced by `get_or_init_connection`
+    /// must be classified as transient via the text fallback.
+    #[test]
+    fn is_sqlite_io_transient_text_fallback() {
+        let err = anyhow::anyhow!("memory_tree_db circuit breaker open: too many init failures");
+        assert!(is_sqlite_io_transient(&err));
+    }
+
+    /// UNIQUE constraint violation must NOT be reclassified as a transient
+    /// I/O error — those are genuine bugs.
+    #[test]
+    fn is_sqlite_io_transient_negative_constraint_violation() {
+        let raw = rusqlite::Error::SqliteFailure(
+            rusqlite::ffi::Error {
+                code: rusqlite::ErrorCode::ConstraintViolation,
+                extended_code: 19,
+            },
+            Some("UNIQUE constraint failed: mem_tree_jobs.dedupe_key".into()),
+        );
+        assert!(!is_sqlite_io_transient(&anyhow::Error::from(raw)));
+    }
 }

src/openhuman/memory/tree/store.rs

@@ -6,11 +6,30 @@
 //!
 //! Upsert semantics: writes are idempotent on `chunk.id` so re-ingesting the
 //! same raw source yields no duplicates.
+//!
+//! ## Connection cache (#2206)
+//!
+//! `with_connection()` previously opened a new SQLite connection and re-ran
+//! the full schema init (8 tables, 15+ indexes, 8+ migrations) on **every**
+//! call. With 4 workers polling every 5 s this amounted to ~69K connection
+//! opens/day, and three I/O error codes (1546 IOERR_TRUNCATE, 4874
+//! IOERR_SHMMAP, 14 CANTOPEN) flooded Sentry with ~19K events in 4 days.
+//!
+//! Fix: a process-level `ConnectionCache` keyed by DB path. Each entry holds
+//! one `parking_lot::Mutex<Connection>` that is initialised once (schema +
+//! migrations + legacy-embedding migration) and then reused for all subsequent
+//! calls. A per-entry `CircuitBreaker` stops retrying after 3 consecutive
+//! init failures for 30 s so a broken install does not busy-loop.
 
 use anyhow::{Context, Result};
 use chrono::{DateTime, TimeZone, Utc};
+use parking_lot::Mutex as PMutex;
 use rusqlite::{params, Connection, OptionalExtension, Transaction};
-use std::time::Duration;
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::sync::{Arc, OnceLock};
+use std::time::{Duration, Instant};
 
 use crate::openhuman::config::Config;
 use crate::openhuman::memory::tree::content_store::StagedChunk;
@@ -694,20 +713,136 @@ fn ms_to_utc(ms: i64) -> rusqlite::Result<DateTime<Utc>> {
     })
 }
 
-/// Open the memory_tree SQLite DB and run a closure against it.
+// ── Connection cache (#2206) ─────────────────────────────────────────────────
+
+/// How many consecutive init failures before the circuit breaker trips.
+const CB_THRESHOLD: u32 = 3;
+/// How long the circuit breaker holds the DB closed after tripping.
+const CB_COOLDOWN: Duration = Duration::from_secs(30);
+
+/// Per-path circuit breaker: after [`CB_THRESHOLD`] consecutive init failures
+/// the breaker trips and `get_or_init_connection` returns an error immediately
+/// until [`CB_COOLDOWN`] elapses. On the first success it resets to zero.
+struct CircuitBreaker {
+    consecutive_failures: AtomicU32,
+    tripped: AtomicBool,
+    last_trip: PMutex<Option<Instant>>,
+}
+
+impl CircuitBreaker {
+    fn new() -> Self {
+        Self {
+            consecutive_failures: AtomicU32::new(0),
+            tripped: AtomicBool::new(false),
+            last_trip: PMutex::new(None),
+        }
+    }
+
+    fn record_success(&self) {
+        self.consecutive_failures.store(0, Ordering::Relaxed);
+        self.tripped.store(false, Ordering::Relaxed);
+        *self.last_trip.lock() = None;
+    }
+
+    /// Records one more failure. Returns `true` if this call just tripped the
+    /// breaker (i.e. the threshold was crossed right now).
+    fn record_failure(&self) -> bool {
+        let prev = self.consecutive_failures.fetch_add(1, Ordering::Relaxed);
+        let count = prev + 1;
+        if count >= CB_THRESHOLD && !self.tripped.swap(true, Ordering::Relaxed) {
+            *self.last_trip.lock() = Some(Instant::now());
+            return true;
+        }
+        // Re-arm the cooldown on each subsequent failure while already tripped
+        // so that a failed post-cooldown retry restarts the 30 s window instead
+        // of leaving the stale timestamp in place (which would let `is_open`
+        // return false immediately and allow unlimited retries).
+        if self.tripped.load(Ordering::Relaxed) {
+            *self.last_trip.lock() = Some(Instant::now());
+        }
+        false
+    }
+
+    /// Returns `true` when the breaker is open AND the cooldown has not yet
+    /// elapsed. Returns `false` (allowing a retry) once the cooldown passes.
+    fn is_open(&self) -> bool {
+        if !self.tripped.load(Ordering::Relaxed) {
+            return false;
+        }
+        let guard = self.last_trip.lock();
+        match *guard {
+            Some(t) if t.elapsed() < CB_COOLDOWN => true,
+            _ => false,
+        }
+    }
+}
+
+/// Process-level cache — two separate maps so that a failing init cannot
+/// accidentally serve a dummy connection on the fast path.
 ///
-/// Visible to sibling modules (e.g. `score::store`) so Phase 2 can reuse
-/// the same connection setup / schema initialisation without duplication.
-pub(crate) fn with_connection<T>(
-    config: &Config,
-    f: impl FnOnce(&Connection) -> Result<T>,
-) -> Result<T> {
-    let dir = config.workspace_dir.join(DB_DIR);
-    std::fs::create_dir_all(&dir)
-        .with_context(|| format!("Failed to create memory_tree dir: {}", dir.display()))?;
-    let db_path = dir.join(DB_FILE);
-    let conn = Connection::open(&db_path)
-        .with_context(|| format!("Failed to open memory_tree DB: {}", db_path.display()))?;
+/// `connections`: only fully-initialised (schema + migrations run) entries.
+/// `breakers`:    persists across failed init attempts so the circuit breaker
+///                survives even when `connections` has no entry for that path.
+struct ConnectionCache {
+    connections: PMutex<HashMap<PathBuf, Arc<PMutex<Connection>>>>,
+    breakers: PMutex<HashMap<PathBuf, Arc<CircuitBreaker>>>,
+}
+
+static CONN_CACHE: OnceLock<ConnectionCache> = OnceLock::new();
+
+fn conn_cache() -> &'static ConnectionCache {
+    CONN_CACHE.get_or_init(|| ConnectionCache {
+        connections: PMutex::new(HashMap::new()),
+        breakers: PMutex::new(HashMap::new()),
+    })
+}
+
+/// Compute the canonical DB path from `config`.
+fn db_path_for(config: &Config) -> PathBuf {
+    config.workspace_dir.join(DB_DIR).join(DB_FILE)
+}
+
+/// Delete stale WAL/SHM side-files (`<db>-shm`, `<db>-wal`) that can block a
+/// clean DB open after a crash. Logs what was deleted and returns `true` if
+/// anything was removed.
+///
+/// SQLite names these files `<db_path>-shm` and `<db_path>-wal`.
+/// For `chunks.db` that is `chunks.db-shm` / `chunks.db-wal`.
+pub(crate) fn try_cleanup_stale_files(db_path: &std::path::Path) -> bool {
+    let mut cleaned = false;
+    for suffix in &["-shm", "-wal"] {
+        // Build the side-file path: append suffix to the full db filename.
+        let side = {
+            let mut p = db_path.to_path_buf();
+            let name = p
+                .file_name()
+                .unwrap_or_default()
+                .to_string_lossy()
+                .into_owned();
+            p.set_file_name(format!("{name}{suffix}"));
+            p
+        };
+        if side.exists() {
+            match std::fs::remove_file(&side) {
+                Ok(()) => {
+                    log::warn!("[memory_tree] removed stale side-file: {}", side.display());
+                    cleaned = true;
+                }
+                Err(e) => {
+                    log::warn!(
+                        "[memory_tree] failed to remove stale side-file {}: {e}",
+                        side.display()
+                    );
+                }
+            }
+        }
+    }
+    cleaned
+}
+
+/// Run the full one-time DB initialisation (WAL, schema, migrations) against
+/// an already-open `Connection`. Used by `get_or_init_connection`.
+fn init_db(conn: &Connection, config: &Config) -> Result<()> {
     conn.busy_timeout(SQLITE_BUSY_TIMEOUT)
         .context("Failed to configure memory_tree busy timeout")?;
     if let Err(wal_err) = conn.execute_batch("PRAGMA journal_mode=WAL;") {
@@ -718,28 +853,20 @@ pub(crate) fn with_connection<T>(
     conn.execute_batch(SCHEMA)
         .context("Failed to initialize memory_tree schema")?;
     // Phase 2 migrations — additive, idempotent.
-    add_column_if_missing(&conn, "mem_tree_chunks", "embedding", "BLOB")?;
+    add_column_if_missing(conn, "mem_tree_chunks", "embedding", "BLOB")?;
     // Phase 2 LLM-NER follow-up: per-chunk LLM importance signal +
     // human-readable reason. Both nullable; absence is treated as
     // "no LLM signal available" by readers.
-    add_column_if_missing(&conn, "mem_tree_score", "llm_importance", "REAL")?;
-    add_column_if_missing(&conn, "mem_tree_score", "llm_importance_reason", "TEXT")?;
-    // Phase 3a (#709): parent-summary backlink on leaves. Populated when
-    // the L0 buffer seals into an L1 summary so traversal can walk
-    // leaf → parent without scanning `mem_tree_summaries.child_ids_json`.
-    add_column_if_missing(&conn, "mem_tree_chunks", "parent_summary_id", "TEXT")?;
+    add_column_if_missing(conn, "mem_tree_score", "llm_importance", "REAL")?;
+    add_column_if_missing(conn, "mem_tree_score", "llm_importance_reason", "TEXT")?;
+    // Phase 3a (#709): parent-summary backlink on leaves.
+    add_column_if_missing(conn, "mem_tree_chunks", "parent_summary_id", "TEXT")?;
     // Phase 4 (#710): sealed-summary embeddings for semantic rerank.
-    // Blob layout matches `mem_tree_chunks.embedding` — see
-    // `score::embed::{pack_embedding, unpack_embedding}`. Nullable so
-    // legacy summaries from Phases 1-3 read back as None; retrieval
-    // tolerates NULL by dropping the row to the bottom of a rerank.
-    add_column_if_missing(&conn, "mem_tree_summaries", "embedding", "BLOB")?;
+    add_column_if_missing(conn, "mem_tree_summaries", "embedding", "BLOB")?;
     // Async-pipeline lifecycle flag. Default 'admitted' so chunks ingested
-    // before the queue migration stay queryable. New writes start at
-    // 'pending_extraction'; the extract handler advances them to 'admitted'
-    // (then 'buffered' / 'sealed') or 'dropped'.
+    // before the queue migration stay queryable.
     add_column_if_missing(
-        &conn,
+        conn,
         "mem_tree_chunks",
         "lifecycle_status",
         "TEXT NOT NULL DEFAULT 'admitted'",
@@ -749,43 +876,189 @@ pub(crate) fn with_connection<T>(
          ON mem_tree_chunks(lifecycle_status);",
     )
     .context("Failed to create mem_tree_chunks lifecycle index")?;
-    // Phase MD-content (#TBD): pointer + integrity hash. Body lives at
-    // <content_root>/<content_path> as a .md file. Both nullable so chunks
-    // ingested before this migration read back with NULL (body still in
-    // `content`). New writes populate both columns. The `content` column
-    // stores a 500-char plain-text preview instead of the full body.
-    add_column_if_missing(&conn, "mem_tree_chunks", "content_path", "TEXT")?;
-    add_column_if_missing(&conn, "mem_tree_chunks", "content_sha256", "TEXT")?;
-    // Phase MD-content (summaries): same pointer pattern for summary nodes.
-    // `content_path` is the relative path to the .md file under
-    // `<content_root>/summaries/...`. `content_sha256` is the SHA-256 hex
-    // of the body bytes only (front-matter excluded). Both nullable so
-    // legacy rows (from before this migration) read back with NULL — callers
-    // fall back to the `content` column for those rows.
-    add_column_if_missing(&conn, "mem_tree_summaries", "content_path", "TEXT")?;
-    add_column_if_missing(&conn, "mem_tree_summaries", "content_sha256", "TEXT")?;
-    // Raw-archive pointer column. JSON array of {path, start, end} —
-    // used by chunks whose body comes from one or more files under
-    // `<content_root>/raw/...` (today: email). When set, `read_chunk_body`
-    // reads + concatenates those byte ranges instead of fetching from
-    // disk via `content_path` or falling back to the SQL `content`
-    // preview. Nullable so legacy chunks keep working unchanged.
-    add_column_if_missing(&conn, "mem_tree_chunks", "raw_refs_json", "TEXT")?;
-    // #1365: is_user flag on indexed entity rows. Set at write time by
-    // running the canonical id through the Composio identity registry
-    // (`is_self_identity_any_toolkit`). Default 0 so legacy rows read
-    // back as "not user" until the backfill job re-tags them.
+    // Phase MD-content (#TBD): pointer + integrity hash.
+    add_column_if_missing(conn, "mem_tree_chunks", "content_path", "TEXT")?;
+    add_column_if_missing(conn, "mem_tree_chunks", "content_sha256", "TEXT")?;
+    // Phase MD-content (summaries).
+    add_column_if_missing(conn, "mem_tree_summaries", "content_path", "TEXT")?;
+    add_column_if_missing(conn, "mem_tree_summaries", "content_sha256", "TEXT")?;
+    // Raw-archive pointer column.
+    add_column_if_missing(conn, "mem_tree_chunks", "raw_refs_json", "TEXT")?;
+    // #1365: is_user flag on indexed entity rows.
     add_column_if_missing(
-        &conn,
+        conn,
         "mem_tree_entity_index",
         "is_user",
         "INTEGER NOT NULL DEFAULT 0",
     )?;
-    // #1574 §7: one-shot, version-gated copy of legacy embedding columns
-    // into the per-model sidecar at the active signature. Runs once
-    // (PRAGMA user_version gate); cheap no-op on every subsequent open.
-    migrate_legacy_embeddings_to_sidecar(&conn, config)?;
-    f(&conn)
+    // #1574 §7: one-shot, version-gated legacy→sidecar embedding migration.
+    migrate_legacy_embeddings_to_sidecar(conn, config)?;
+    Ok(())
+}
+
+/// Whether `err` looks like one of the I/O error codes that warrant a
+/// stale-file cleanup + single retry before giving up.
+fn is_io_open_error(err: &anyhow::Error) -> bool {
+    if let Some(rusqlite::Error::SqliteFailure(f, _)) = err.downcast_ref::<rusqlite::Error>() {
+        // 1546 = SQLITE_IOERR_TRUNCATE, 4874 = SQLITE_IOERR_SHMMAP, 14 = SQLITE_CANTOPEN
+        return matches!(f.extended_code, 1546 | 4874 | 14)
+            || f.code == rusqlite::ErrorCode::CannotOpen;
+    }
+    let msg = format!("{err:#}").to_ascii_lowercase();
+    msg.contains("disk i/o error")
+        || msg.contains("unable to open database file")
+        || msg.contains("xshmmap")
+        || msg.contains("truncate file")
+}
+
+/// Obtain (or lazily create) a cached connection for the workspace described
+/// by `config`. Returns `Err` immediately when the circuit breaker is open.
+fn get_or_init_connection(config: &Config) -> Result<Arc<PMutex<Connection>>> {
+    let db_path = db_path_for(config);
+
+    // ── Fast path: reject immediately if the breaker is open ─────────────
+    {
+        let breakers = conn_cache().breakers.lock();
+        if let Some(breaker) = breakers.get(&db_path) {
+            if breaker.is_open() {
+                anyhow::bail!(
+                    "[memory_tree] circuit breaker open for {}: too many consecutive init failures",
+                    db_path.display()
+                );
+            }
+        }
+    }
+    // ── Fast path: return cached connection if already initialised ────────
+    {
+        let guard = conn_cache().connections.lock();
+        if let Some(conn) = guard.get(&db_path) {
+            return Ok(Arc::clone(conn));
+        }
+    }
+
+    // ── Slow path: initialise DB, then insert into cache ─────────────────
+    log::debug!(
+        "[memory_tree] opening and initialising DB at {}",
+        db_path.display()
+    );
+
+    // Attempt to open + init the connection (dir creation is inside
+    // `open_and_init` so every failure — including EEXIST on the dir —
+    // reaches the circuit-breaker recording logic below). On certain I/O
+    // errors (#2206) we clean up stale WAL/SHM side-files and retry once.
+    let conn = open_and_init(&db_path, config).or_else(|first_err| {
+        if is_io_open_error(&first_err) {
+            log::warn!(
+                "[memory_tree] I/O error on first open attempt ({}), cleaning stale files and retrying",
+                first_err
+            );
+            try_cleanup_stale_files(&db_path);
+            open_and_init(&db_path, config)
+        } else {
+            Err(first_err)
+        }
+    });
+
+    match conn {
+        Ok(conn) => {
+            let arc_conn = Arc::new(PMutex::new(conn));
+            conn_cache()
+                .connections
+                .lock()
+                .insert(db_path.clone(), Arc::clone(&arc_conn));
+            // Reset any prior failure counter now that init succeeded.
+            if let Some(breaker) = conn_cache().breakers.lock().get(&db_path) {
+                breaker.record_success();
+            }
+            log::debug!("[memory_tree] DB connection cached and ready");
+            Ok(arc_conn)
+        }
+        Err(err) => {
+            // Persist the breaker so the failure count accumulates across
+            // calls even though no connection entry exists yet.
+            let breaker = {
+                let mut guard = conn_cache().breakers.lock();
+                guard
+                    .entry(db_path.clone())
+                    .or_insert_with(|| Arc::new(CircuitBreaker::new()))
+                    .clone()
+            };
+            let just_tripped = breaker.record_failure();
+            if just_tripped {
+                log::error!(
+                    "[memory_tree] circuit breaker tripped for {}: {} consecutive init failures",
+                    db_path.display(),
+                    CB_THRESHOLD
+                );
+                let _ = crate::core::event_bus::publish_global(
+                    crate::core::event_bus::DomainEvent::HealthChanged {
+                        component: "memory_tree_db".to_string(),
+                        healthy: false,
+                        message: Some(format!(
+                            "Schema init failed {CB_THRESHOLD} consecutive times"
+                        )),
+                    },
+                );
+            }
+            Err(err)
+        }
+    }
+}
+
+/// Ensure the DB directory exists, open the SQLite file, and run the full
+/// schema init sequence. All errors (dir creation, file open, schema init)
+/// are returned as `Err` so callers can funnel them through the circuit
+/// breaker logic in a single place.
+fn open_and_init(db_path: &std::path::Path, config: &Config) -> Result<Connection> {
+    let dir = db_path.parent().expect("db_path always has a parent");
+    std::fs::create_dir_all(dir)
+        .with_context(|| format!("Failed to create memory_tree dir: {}", dir.display()))?;
+    let conn = Connection::open(db_path)
+        .with_context(|| format!("Failed to open memory_tree DB: {}", db_path.display()))?;
+    init_db(&conn, config)
+        .with_context(|| format!("Failed to init memory_tree schema: {}", db_path.display()))?;
+    Ok(conn)
+}
+
+/// Remove the cached connection for `config`'s workspace (forces a fresh open
+/// on the next `with_connection` call). Also clears the breaker so the next
+/// open attempt is not immediately rejected. Does nothing if no entry exists.
+#[allow(dead_code)]
+pub(crate) fn invalidate_connection(config: &Config) {
+    let db_path = db_path_for(config);
+    conn_cache().connections.lock().remove(&db_path);
+    conn_cache().breakers.lock().remove(&db_path);
+    log::debug!(
+        "[memory_tree] connection invalidated for {}",
+        db_path.display()
+    );
+}
+
+/// Clear the entire connection cache. For test isolation only.
+#[cfg(test)]
+pub(crate) fn clear_connection_cache() {
+    conn_cache().connections.lock().clear();
+    conn_cache().breakers.lock().clear();
+}
+
+/// Open the memory_tree SQLite DB and run a closure against it.
+///
+/// Visible to sibling modules (e.g. `score::store`) so Phase 2 can reuse
+/// the same connection setup / schema initialisation without duplication.
+///
+/// # Connection caching (#2206)
+///
+/// The underlying connection is initialised once per workspace path and then
+/// reused from a process-level cache. Schema migrations run exactly once on
+/// the first call for a given `config.workspace_dir`. Subsequent calls pay
+/// only the cost of a `parking_lot::Mutex` lock and the closure itself.
+pub(crate) fn with_connection<T>(
+    config: &Config,
+    f: impl FnOnce(&Connection) -> Result<T>,
+) -> Result<T> {
+    let conn_arc = get_or_init_connection(config)?;
+    let guard = conn_arc.lock();
+    f(&guard)
 }
 
 /// One-shot migration (#1574 §7, vN): copy legacy `mem_tree_chunks.embedding`

src/openhuman/memory/tree/store_tests.rs

@@ -1,5 +1,14 @@
 //! Unit tests for [`super`] — chunk upsert / list / lifecycle / embedding /
 //! content-pointer accessors against a tempdir-backed SQLite store.
+//!
+//! ## Test isolation for the connection cache
+//!
+//! Because the connection cache is a process-level singleton, tests that want
+//! to exercise cache behaviour (same Arc, independent workspaces, circuit
+//! breaker, cleanup) must call `clear_connection_cache()` at the start — or
+//! be careful to use unique tempdirs that cannot collide with other tests.
+//! The call is cheap (a mutex lock + HashMap clear) and harmless for tests
+//! that don't need it.
 
 use super::*;
 use crate::openhuman::memory::tree::types::chunk_id;
@@ -296,6 +305,10 @@ fn legacy_embeddings_migrate_to_sidecar_once() {
     })
     .unwrap();
 
+    // Evict the cached connection so the next open sees user_version = 0
+    // and re-runs migrate_legacy_embeddings_to_sidecar.
+    invalidate_connection(&cfg);
+
     // The next store open (this getter's `with_connection`) sees
     // user_version = 0 and runs the migration before returning.
     assert_eq!(
@@ -340,3 +353,105 @@ fn legacy_embeddings_migrate_to_sidecar_once() {
         Some(match_vec)
     );
 }
+
+// ── Connection cache tests (#2206) ───────────────────────────────────────────
+
+/// Two `with_connection` calls for the same workspace must return the same
+/// cached `Arc` (pointer identity proves no re-init happened).
+#[test]
+fn connection_cache_returns_same_arc_for_same_workspace() {
+    clear_connection_cache();
+    let (_tmp, cfg) = test_config();
+
+    let arc1 = get_or_init_connection(&cfg).expect("first get_or_init");
+    let arc2 = get_or_init_connection(&cfg).expect("second get_or_init");
+    assert!(
+        Arc::ptr_eq(&arc1, &arc2),
+        "expected the same Arc from the connection cache on the second call"
+    );
+}
+
+/// Two configs pointing at different tempdirs must produce independent
+/// connections (separate Arc pointers, no cross-contamination).
+#[test]
+fn connection_cache_uses_separate_connections_for_different_workspaces() {
+    clear_connection_cache();
+    let (_tmp1, cfg1) = test_config();
+    let (_tmp2, cfg2) = test_config();
+
+    let arc1 = get_or_init_connection(&cfg1).expect("workspace 1");
+    let arc2 = get_or_init_connection(&cfg2).expect("workspace 2");
+    assert!(
+        !Arc::ptr_eq(&arc1, &arc2),
+        "different workspaces must have independent connections"
+    );
+
+    // Sanity: each DB is usable independently.
+    let c = sample_chunk("s", 0, 1_700_000_000_000);
+    upsert_chunks(&cfg1, &[c.clone()]).unwrap();
+    assert_eq!(count_chunks(&cfg1).unwrap(), 1);
+    assert_eq!(count_chunks(&cfg2).unwrap(), 0);
+}
+
+/// Pointing the DB path at a *file* (not a directory) makes it impossible to
+/// create the DB, so `get_or_init_connection` must fail. After
+/// `CB_THRESHOLD` failures the circuit breaker trips and subsequent calls
+/// return an error immediately without touching the filesystem.
+#[test]
+fn circuit_breaker_trips_after_threshold() {
+    clear_connection_cache();
+    let tmp = TempDir::new().expect("tempdir");
+
+    // Create a regular file where the memory_tree *directory* would be —
+    // this prevents `create_dir_all` from succeeding.
+    let blocker = tmp.path().join(DB_DIR);
+    std::fs::write(&blocker, b"not a dir").expect("write blocker file");
+
+    let mut cfg = Config::default();
+    cfg.workspace_dir = tmp.path().to_path_buf();
+
+    // First CB_THRESHOLD calls should all fail (can't create dir over a file).
+    for i in 0..CB_THRESHOLD {
+        let result = get_or_init_connection(&cfg);
+        assert!(
+            result.is_err(),
+            "call {i}: expected error before breaker trips"
+        );
+    }
+
+    // The CB_THRESHOLD+1'th call should be rejected immediately by the
+    // circuit breaker (error message contains "circuit breaker").
+    let cb_err = get_or_init_connection(&cfg)
+        .expect_err("expected circuit breaker error on call after threshold");
+    let msg = format!("{cb_err:#}").to_ascii_lowercase();
+    assert!(
+        msg.contains("circuit breaker"),
+        "expected circuit breaker message, got: {msg}"
+    );
+}
+
+/// `try_cleanup_stale_files` removes `.db-shm` and `.db-wal` side-files that
+/// exist alongside the main DB file.
+#[test]
+fn stale_shm_cleanup_removes_files() {
+    let tmp = TempDir::new().expect("tempdir");
+    let db_path = tmp.path().join("chunks.db");
+
+    // Create the main DB file and the two stale side-files.
+    std::fs::write(&db_path, b"").expect("create db file");
+    let shm = tmp.path().join("chunks.db-shm");
+    let wal = tmp.path().join("chunks.db-wal");
+    std::fs::write(&shm, b"stale shm").expect("create shm");
+    std::fs::write(&wal, b"stale wal").expect("create wal");
+
+    assert!(shm.exists(), "shm must exist before cleanup");
+    assert!(wal.exists(), "wal must exist before cleanup");
+
+    let cleaned = try_cleanup_stale_files(&db_path);
+    assert!(
+        cleaned,
+        "cleanup should return true when files were removed"
+    );
+    assert!(!shm.exists(), "shm must be removed");
+    assert!(!wal.exists(), "wal must be removed");
+}