fix(subconscious): preserve anyhow chain + retry DDL with busy_timeout on SQLITE_BUSY (#2851)

src/openhuman/subconscious/schemas.rs

@@ -289,7 +289,7 @@ fn handle_status(_params: Map<String, Value>) -> ControllerFuture {
                     .unwrap_or((None, 0));
                 Ok((tc, pe, lt, tt))
             })
-            .map_err(|e| e.to_string())?;
+            .map_err(|e| format!("{e:#}"))?;
 
         let provider_unavailable_reason = if hb.enabled && hb.inference_enabled {
             super::engine::subconscious_provider_unavailable_reason(&config)
@@ -355,7 +355,7 @@ fn handle_tasks_list(params: Map<String, Value>) -> ControllerFuture {
         let tasks = store::with_connection(&config.workspace_dir, |conn| {
             store::list_tasks(conn, enabled_only)
         })
-        .map_err(|e| e.to_string())?;
+        .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(tasks, "tasks listed"))
     })
 }
@@ -377,7 +377,7 @@ fn handle_tasks_add(params: Map<String, Value>) -> ControllerFuture {
         let task = engine
             .add_task(&title, source)
             .await
-            .map_err(|e| e.to_string())?;
+            .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(task, "task added"))
     })
 }
@@ -409,7 +409,7 @@ fn handle_tasks_update(params: Map<String, Value>) -> ControllerFuture {
         store::with_connection(&config.workspace_dir, |conn| {
             store::update_task(conn, &task_id, &patch)
         })
-        .map_err(|e| e.to_string())?;
+        .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(
             serde_json::json!({"updated": task_id}),
             "task updated",
@@ -428,7 +428,7 @@ fn handle_tasks_remove(params: Map<String, Value>) -> ControllerFuture {
         store::with_connection(&config.workspace_dir, |conn| {
             store::remove_task(conn, &task_id)
         })
-        .map_err(|e| e.to_string())?;
+        .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(
             serde_json::json!({"removed": task_id}),
             "task removed",
@@ -444,7 +444,7 @@ fn handle_log_list(params: Map<String, Value>) -> ControllerFuture {
         let entries = store::with_connection(&config.workspace_dir, |conn| {
             store::list_log_entries(conn, task_id, limit)
         })
-        .map_err(|e| e.to_string())?;
+        .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(entries, "log entries listed"))
     })
 }
@@ -463,7 +463,7 @@ fn handle_escalations_list(params: Map<String, Value>) -> ControllerFuture {
         let escalations = store::with_connection(&config.workspace_dir, |conn| {
             store::list_escalations(conn, status_filter.as_ref())
         })
-        .map_err(|e| e.to_string())?;
+        .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(escalations, "escalations listed"))
     })
 }
@@ -481,7 +481,7 @@ fn handle_escalations_approve(params: Map<String, Value>) -> ControllerFuture {
         engine
             .approve_escalation(&escalation_id)
             .await
-            .map_err(|e| e.to_string())?;
+            .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(
             serde_json::json!({"approved": escalation_id}),
             "escalation approved and executed",
@@ -502,7 +502,7 @@ fn handle_escalations_dismiss(params: Map<String, Value>) -> ControllerFuture {
         engine
             .dismiss_escalation(&escalation_id)
             .await
-            .map_err(|e| e.to_string())?;
+            .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(
             serde_json::json!({"dismissed": escalation_id}),
             "escalation dismissed",
@@ -520,7 +520,7 @@ fn handle_reflections_list(params: Map<String, Value>) -> ControllerFuture {
         let reflections = store::with_connection(&config.workspace_dir, |conn| {
             reflection_store::list_recent(conn, limit, since_ts)
         })
-        .map_err(|e| e.to_string())?;
+        .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(reflections, "reflections listed"))
     })
 }
@@ -537,7 +537,7 @@ fn handle_reflections_act(params: Map<String, Value>) -> ControllerFuture {
         let reflection = store::with_connection(&config.workspace_dir, |conn| {
             reflection_store::get_reflection(conn, &reflection_id)
         })
-        .map_err(|e| e.to_string())?
+        .map_err(|e| format!("{e:#}"))?
         .ok_or_else(|| format!("reflection not found: {reflection_id}"))?;
 
         // Spawn a fresh conversation thread for this action. Reflections never
@@ -660,7 +660,7 @@ fn handle_reflections_dismiss(params: Map<String, Value>) -> ControllerFuture {
         store::with_connection(&config.workspace_dir, |conn| {
             reflection_store::mark_dismissed(conn, &reflection_id, now)
         })
-        .map_err(|e| e.to_string())?;
+        .map_err(|e| format!("{e:#}"))?;
         to_json(RpcOutcome::single_log(
             serde_json::json!({"dismissed": reflection_id}),
             "reflection dismissed",

src/openhuman/subconscious/schemas_tests.rs

@@ -166,3 +166,90 @@ fn field_opt_helper_is_not_required() {
     let f = field_opt("name", TypeSchema::String, "desc");
     assert!(!f.required);
 }
+
+// ── Error chain preservation ───────────────────────────────────
+//
+// The RPC handlers in this module bridge `anyhow::Result` (from
+// `store::with_connection` and the wrapped rusqlite errors) into the
+// JSON-RPC `Result<Value, String>` boundary via `map_err(|e| ...)`.
+//
+// **Critical for observability**: plain `e.to_string()` on an
+// `anyhow::Error` returns ONLY the outermost context. For a
+// `with_connection` failure the outer wrap is
+// `"failed to run subconscious schema DDL"` — the underlying rusqlite
+// root (the actual SQLite error code + message) is dropped. That
+// stringified message is what `jsonrpc::invoke_method_inner` later
+// passes to `report_error_or_expected`, which in turn captures it in
+// Sentry. Without the chain, Sentry events for TAURI-RUST-A only
+// surface the generic wrapper text and the rusqlite root cause is
+// permanently invisible.
+//
+// All `map_err` sites in `schemas.rs` use `format!("{e:#}")` (anyhow's
+// alternate Display walks the cause chain inline joined by `": "`) so
+// the rusqlite root reaches Sentry. These guard tests pin the format
+// so future contributors don't silently regress to `e.to_string()`.
+#[test]
+fn anyhow_alternate_display_walks_chain() {
+    use anyhow::Context;
+
+    let inner = anyhow::anyhow!("database is locked").context("execute_batch failed");
+    let outer: anyhow::Result<()> = Err(inner).context("failed to run subconscious schema DDL");
+
+    let err = outer.unwrap_err();
+
+    // Plain to_string() — the broken (pre-fix) shape. Only outer
+    // wrapper reaches the caller, root cause lost.
+    let lossy = err.to_string();
+    assert_eq!(lossy, "failed to run subconscious schema DDL");
+    assert!(
+        !lossy.contains("database is locked"),
+        "plain Display must drop the root cause — if this changes the chain-formatter \
+         is no longer load-bearing, revisit observability assumptions"
+    );
+
+    // Alternate Display — what schemas.rs map_err now produces. Every
+    // layer joined by ": " so the rusqlite root reaches Sentry.
+    let full = format!("{err:#}");
+    assert!(
+        full.contains("failed to run subconscious schema DDL"),
+        "chain-formatted message must include outer wrapper, got: {full}"
+    );
+    assert!(
+        full.contains("execute_batch failed"),
+        "chain-formatted message must include middle context, got: {full}"
+    );
+    assert!(
+        full.contains("database is locked"),
+        "chain-formatted message must include the rusqlite root, got: {full}"
+    );
+}
+
+#[test]
+fn anyhow_alternate_display_includes_rusqlite_error_chain() {
+    use anyhow::Context;
+
+    // Simulate the exact shape produced by `with_connection`:
+    // a real rusqlite Error wrapped in `with_context(...)`.
+    let raw = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error {
+            code: rusqlite::ErrorCode::DatabaseBusy,
+            extended_code: 5,
+        },
+        Some("database is locked".into()),
+    );
+    let wrapped: anyhow::Result<()> =
+        Err(anyhow::Error::from(raw)).context("failed to run subconscious schema DDL");
+
+    let err = wrapped.unwrap_err();
+    let chained = format!("{err:#}");
+
+    // Outer wrapper preserved.
+    assert!(chained.contains("failed to run subconscious schema DDL"));
+    // rusqlite-rendered root preserved — this is the signal Sentry
+    // needs to distinguish a DDL lock-race from a corruption / disk-full
+    // / permission failure. Without it, all four fingerprint identically.
+    assert!(
+        chained.contains("database is locked"),
+        "rusqlite root must appear in chain-formatted message, got: {chained}"
+    );
+}

src/openhuman/subconscious/store.rs

@@ -6,6 +6,7 @@
 use anyhow::{Context, Result};
 use rusqlite::{Connection, OptionalExtension};
 use std::path::Path;
+use std::time::Duration;
 use uuid::Uuid;
 
 use super::types::{
@@ -13,7 +14,46 @@ use super::types::{
     TaskPatch, TaskRecurrence, TaskSource,
 };
 
+/// Per-connection busy handler window. Tracks the value used by the cron
+/// module + other domain stores (`memory_store::unified::init`, 15s) and
+/// the higher-throughput whatsapp/memory_queue path (5s). 5 s is enough
+/// for the subconscious tick — RPC handlers are user-driven (status
+/// polling at 3 s, manual triggers) and we'd rather fail fast than block
+/// the UI thread for the full 15 s of contention.
+const BUSY_TIMEOUT: Duration = Duration::from_millis(5000);
+
+/// Maximum number of application-level retries after rusqlite's busy
+/// handler is exhausted. The first attempt is "attempt 0" — total
+/// attempts = `OPEN_RETRY_ATTEMPTS` + 1.
+const OPEN_RETRY_ATTEMPTS: u32 = 3;
+
+/// Base backoff for application-level retries; per-attempt sleep is
+/// `BASE * 3^attempt` so the schedule is `100 ms / 300 ms / 900 ms`
+/// totalling ≤ 1.3 s before the final attempt fails through.
+const OPEN_RETRY_BASE_MS: u64 = 100;
+
 /// Open the subconscious database and run schema migrations.
+///
+/// Three layers of defence against transient `SQLITE_BUSY` / `SQLITE_LOCKED`
+/// at the open / DDL boundary, motivated by Sentry TAURI-RUST-A
+/// (cross-platform, ~1.3k events / 24 h, RPC paths `subconscious_tasks_list`
+/// and `subconscious_status`):
+///
+/// 1. **Per-connection busy timeout** (`BUSY_TIMEOUT`, 5 s): SQLite's
+///    default is `0` — first lock contention returns `SQLITE_BUSY`
+///    immediately. The subconscious domain serialises several RPCs
+///    (status poll every 3 s, tasks-list on Intelligence page, manual
+///    trigger), each opening its own connection; without a timeout the
+///    first concurrent open races and one returns `SQLITE_BUSY` mid-DDL.
+/// 2. **Application-level retry** (3 attempts, exponential backoff
+///    100 / 300 / 900 ms): catches the residual case where the busy
+///    handler is exhausted (long-running external write txn, AV scan
+///    holding the file). Mirrors `whatsapp_data::sqlite_retry` /
+///    `memory_queue::worker::is_sqlite_busy`.
+/// 3. **Retry classifier** (`is_sqlite_busy`): only retries
+///    `DatabaseBusy` / `DatabaseLocked`. Schema / syntax / corruption
+///    errors are real bugs or unrecoverable file-state failures —
+///    retrying just delays the report.
 pub fn with_connection<T>(
     workspace_dir: &Path,
     f: impl FnOnce(&Connection) -> Result<T>,
@@ -24,11 +64,72 @@ pub fn with_connection<T>(
             .with_context(|| format!("failed to create subconscious dir: {}", parent.display()))?;
     }
 
-    let conn = Connection::open(&db_path)
+    let conn = open_and_initialize_with_retry(&db_path)?;
+    f(&conn)
+}
+
+/// Open the SQLite file, set `busy_timeout`, run `SCHEMA_DDL`, and apply
+/// the idempotent reflection-store migrations — retrying the whole
+/// sequence on `SQLITE_BUSY` / `SQLITE_LOCKED`. Split out so the retry
+/// loop has a single failure surface to classify and the happy path
+/// stays linear.
+fn open_and_initialize_with_retry(db_path: &Path) -> Result<Connection> {
+    let mut last_err: Option<anyhow::Error> = None;
+
+    for attempt in 0..=OPEN_RETRY_ATTEMPTS {
+        match open_and_initialize(db_path) {
+            Ok(conn) => {
+                if attempt > 0 {
+                    tracing::debug!(
+                        target: "openhuman::subconscious::store",
+                        attempt = attempt,
+                        db_path = %db_path.display(),
+                        "[subconscious::store] open/DDL succeeded after {attempt} busy retries"
+                    );
+                }
+                return Ok(conn);
+            }
+            Err(e) => {
+                if !is_sqlite_busy(&e) || attempt == OPEN_RETRY_ATTEMPTS {
+                    last_err = Some(e);
+                    break;
+                }
+                let sleep_ms = OPEN_RETRY_BASE_MS
+                    .saturating_mul(3u64.saturating_pow(attempt))
+                    .min(900);
+                tracing::warn!(
+                    target: "openhuman::subconscious::store",
+                    attempt = attempt + 1,
+                    max_attempts = OPEN_RETRY_ATTEMPTS + 1,
+                    sleep_ms = sleep_ms,
+                    error = %format!("{e:#}"),
+                    "[subconscious::store] SQLite busy/locked on open or DDL; retrying"
+                );
+                std::thread::sleep(Duration::from_millis(sleep_ms));
+                last_err = Some(e);
+            }
+        }
+    }
+
+    Err(last_err.expect("OPEN_RETRY_ATTEMPTS >= 0 ensures at least one attempt"))
+}
+
+/// Single-shot open + DDL + migrations. Each invocation returns an
+/// owned `Connection`; on failure the partially-initialised connection
+/// is dropped before the caller retries.
+fn open_and_initialize(db_path: &Path) -> Result<Connection> {
+    let conn = Connection::open(db_path)
         .with_context(|| format!("failed to open subconscious DB: {}", db_path.display()))?;
 
+    // Set busy_timeout BEFORE running DDL — the very first PRAGMA / CREATE
+    // TABLE in SCHEMA_DDL can race with another in-process connection
+    // (subconscious RPCs each call `with_connection` independently), and
+    // SQLite's default busy_timeout is 0.
+    conn.busy_timeout(BUSY_TIMEOUT)
+        .context("configure subconscious busy_timeout")?;
+
     conn.execute_batch(SCHEMA_DDL)
-        .with_context(|| "failed to run subconscious schema DDL")?;
+        .context("failed to run subconscious schema DDL")?;
 
     // Drop the legacy `disposition` / `surfaced_at` columns + their index
     // from previously-migrated DBs. Idempotent — fresh installs and
@@ -39,7 +140,33 @@ pub fn with_connection<T>(
     // Idempotent (duplicate-column errors swallowed).
     super::reflection_store::migrate_add_source_chunks_column(&conn);
 
-    f(&conn)
+    Ok(conn)
+}
+
+/// Returns true when `err` is transient SQLite contention worth retrying
+/// (`SQLITE_BUSY` / `SQLITE_LOCKED`). Schema / syntax / corruption errors
+/// are NOT retried — the retry would just delay the same failure.
+///
+/// Modelled on [`crate::openhuman::memory_queue::worker::is_sqlite_busy`]
+/// and [`crate::openhuman::whatsapp_data::sqlite_retry::is_sqlite_busy`];
+/// kept private to the subconscious store so the retry policy can evolve
+/// independently of those sibling domains.
+fn is_sqlite_busy(err: &anyhow::Error) -> bool {
+    if let Some(rusqlite::Error::SqliteFailure(sqlite_err, _)) =
+        err.downcast_ref::<rusqlite::Error>()
+    {
+        return matches!(
+            sqlite_err.code,
+            rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
+        );
+    }
+    // Fallback for errors wrapped under `.context(...)` layers — the
+    // rusqlite root may sit a few levels deep after `with_context`
+    // wraps the open / DDL failure. anyhow's alternate Display joins
+    // every cause with ": " so the SQLite-rendered phrase remains
+    // searchable.
+    let msg = format!("{err:#}").to_ascii_lowercase();
+    msg.contains("database is locked") || msg.contains("database table is locked")
 }
 
 const SCHEMA_DDL: &str = "

src/openhuman/subconscious/store_tests.rs

@@ -156,3 +156,186 @@ fn recurrence_roundtrip() {
         TaskRecurrence::Cron("0 8 * * *".into())
     );
 }
+
+// ── DDL resilience: classifier + retry happy path ──────────────
+//
+// These guards back Sentry TAURI-RUST-A: the production failure is
+// `Connection::open` + `execute_batch(SCHEMA_DDL)` racing against
+// another in-process connection that holds the write lock. With
+// `BUSY_TIMEOUT` set and the application-level retry loop in place,
+// the race resolves on its own; without them the first attempt
+// returns `SQLITE_BUSY` and the user sees "failed to run subconscious
+// schema DDL" in Sentry with no further context.
+
+#[test]
+fn is_sqlite_busy_matches_database_busy_code() {
+    let raw = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error {
+            code: rusqlite::ErrorCode::DatabaseBusy,
+            extended_code: 5, // SQLITE_BUSY
+        },
+        Some("database is locked".into()),
+    );
+    let err = anyhow::Error::from(raw);
+    assert!(is_sqlite_busy(&err));
+}
+
+#[test]
+fn is_sqlite_busy_matches_database_locked_code() {
+    let raw = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error {
+            code: rusqlite::ErrorCode::DatabaseLocked,
+            extended_code: 6, // SQLITE_LOCKED
+        },
+        Some("database table is locked".into()),
+    );
+    let err = anyhow::Error::from(raw);
+    assert!(is_sqlite_busy(&err));
+}
+
+#[test]
+fn is_sqlite_busy_does_not_match_constraint_violation() {
+    let raw = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error {
+            code: rusqlite::ErrorCode::ConstraintViolation,
+            extended_code: 19,
+        },
+        Some("UNIQUE constraint failed".into()),
+    );
+    let err = anyhow::Error::from(raw);
+    assert!(!is_sqlite_busy(&err));
+}
+
+#[test]
+fn is_sqlite_busy_does_not_match_schema_syntax_error() {
+    // A genuine bug in `SCHEMA_DDL` (e.g. typo in CREATE TABLE) would
+    // surface as a `SqliteFailure(Unknown, ...)` with "syntax error"
+    // in the message — retrying just delays the same failure, so the
+    // classifier must reject it. Use Unknown + non-busy message.
+    let raw = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error {
+            code: rusqlite::ErrorCode::Unknown,
+            extended_code: 1,
+        },
+        Some("near \"FOO\": syntax error".into()),
+    );
+    let err = anyhow::Error::from(raw);
+    assert!(!is_sqlite_busy(&err));
+}
+
+#[test]
+fn is_sqlite_busy_matches_through_context_layers() {
+    // The production failure shape: a rusqlite error wrapped under
+    // `with_context("failed to run subconscious schema DDL")` —
+    // exactly what `open_and_initialize` produces. Downcast must
+    // still find the rusqlite root.
+    let raw = rusqlite::Error::SqliteFailure(
+        rusqlite::ffi::Error {
+            code: rusqlite::ErrorCode::DatabaseBusy,
+            extended_code: 5,
+        },
+        Some("database is locked".into()),
+    );
+    let wrapped: anyhow::Result<()> = Err(anyhow::Error::from(raw))
+        .with_context(|| "failed to run subconscious schema DDL".to_string());
+    let err = wrapped.unwrap_err();
+    assert!(is_sqlite_busy(&err));
+}
+
+#[test]
+fn is_sqlite_busy_text_fallback_when_downcast_misses() {
+    // If a future refactor stringifies the rusqlite error before
+    // wrapping (e.g. via anyhow!("{e}")), the downcast misses but
+    // the chain-formatter text still preserves "database is locked".
+    let err = anyhow::anyhow!("failed to run subconscious schema DDL: database is locked");
+    assert!(is_sqlite_busy(&err));
+}
+
+#[test]
+fn with_connection_resolves_external_write_contention() {
+    use std::sync::mpsc;
+    use tempfile::TempDir;
+
+    let tmp = TempDir::new().unwrap();
+    let workspace = tmp.path().to_path_buf();
+
+    // First call: prime the DB so the file exists and the schema is
+    // initialized. Subsequent calls take the fast path.
+    with_connection(&workspace, |conn| {
+        add_task(conn, "primer", TaskSource::User, TaskRecurrence::Once)?;
+        Ok(())
+    })
+    .expect("prime DB");
+
+    // Hold an EXCLUSIVE write lock for ~250 ms in a side thread.
+    // The DDL loop in `open_and_initialize` re-runs PRAGMA journal_mode
+    // and CREATE TABLE IF NOT EXISTS — both are no-ops on an already
+    // initialized DB but still acquire the write lock briefly, which
+    // races against the held lock. The application-level retry
+    // (100 / 300 / 900 ms) plus the 5 s `busy_timeout` must absorb
+    // this and let the second `with_connection` succeed.
+    let db_path = workspace.join("subconscious").join("subconscious.db");
+    let (lock_ready_tx, lock_ready_rx) = mpsc::channel::<()>();
+    let (release_tx, release_rx) = mpsc::channel::<()>();
+    let blocker = std::thread::spawn(move || {
+        let conn = rusqlite::Connection::open(&db_path).expect("open blocker conn");
+        conn.busy_timeout(std::time::Duration::from_millis(100))
+            .expect("blocker busy_timeout");
+        let tx = conn
+            .unchecked_transaction()
+            .expect("begin blocker transaction");
+        // Force write-lock acquisition immediately.
+        tx.execute(
+            "INSERT INTO subconscious_tasks \
+             (id, title, source, recurrence, created_at) \
+             VALUES ('blocker', 'blocker', 'user', 'pending', 0.0)",
+            [],
+        )
+        .expect("blocker insert");
+        lock_ready_tx.send(()).expect("signal lock acquired");
+        // Wait for the main thread to start contending, then a touch
+        // longer so the first one or two retries collide.
+        release_rx
+            .recv_timeout(std::time::Duration::from_secs(2))
+            .expect("release signal");
+        std::thread::sleep(std::time::Duration::from_millis(50));
+        tx.rollback().expect("rollback blocker txn");
+    });
+
+    // Wait until the blocker actually holds the write lock.
+    lock_ready_rx
+        .recv_timeout(std::time::Duration::from_secs(2))
+        .expect("blocker never acquired lock");
+
+    // Contender: should retry through the busy window and succeed
+    // once the blocker rolls back. We release the blocker after
+    // ~250 ms so the second / third retry attempt lands in the
+    // unlocked window.
+    let release_tx_for_timer = release_tx.clone();
+    let timer = std::thread::spawn(move || {
+        std::thread::sleep(std::time::Duration::from_millis(250));
+        let _ = release_tx_for_timer.send(());
+    });
+
+    let result = with_connection(&workspace, |conn| {
+        // Confirm we can issue a real query through the contended
+        // connection — proves the open + DDL completed cleanly.
+        let count: i64 = conn
+            .query_row("SELECT COUNT(*) FROM subconscious_tasks", [], |row| {
+                row.get(0)
+            })
+            .unwrap_or(-1);
+        Ok(count)
+    });
+
+    timer.join().expect("timer thread panicked");
+    blocker.join().expect("blocker thread panicked");
+
+    let count = result.expect("contended with_connection must succeed via retry");
+    // Primer row is "primer"; blocker's INSERT was rolled back, so
+    // the count should be exactly 1.
+    assert_eq!(
+        count, 1,
+        "expected only the primer row after blocker rollback, got {count}"
+    );
+}