use super::embeddings::EmbeddingProvider;
use super::traits::{
    Memory, MemoryCategory, MemoryEntry, MemoryStats, ResumableSessionEntry, SessionEntry,
    SessionListEntry, SessionSnapshotKind, SessionSnapshotRecord, SessionStateMutation,
    SessionStatePatch, SessionStateRecord, SessionStatus, SlashSessionLifecycle, TaskCreateInput,
    TaskListPage, TaskListQuery, TaskPatch, TaskPriority, TaskRecord, TaskStatus,
};
use super::vector;
use anyhow::Context;
use async_trait::async_trait;
use chrono::Local;
use parking_lot::Mutex;
use rusqlite::{params, Connection, OptionalExtension, Transaction};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::sync::mpsc;
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use uuid::Uuid;

/// Maximum allowed open timeout (seconds) to avoid unreasonable waits.
const SQLITE_OPEN_TIMEOUT_CAP_SECS: u64 = 300;
const MAX_LIST_LIMIT: u32 = 1_000;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum DreamSessionDbStatus {
    Pending,
    Running,
    Completed,
    Failed,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum DreamTriggerDbReason {
    TimeElapsed,
    SessionCount,
    Manual,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct DreamSessionRow {
    session_id: String,
    status: DreamSessionDbStatus,
    trigger_reason: DreamTriggerDbReason,
    completion_recorded_at: String,
    last_attempt_at: Option<String>,
    completed_at: Option<String>,
    artifact_refs: Vec<String>,
    failure_reason: Option<String>,
}

/// SQLite-backed persistent memory — the brain
///
/// Full-stack search engine:
/// - **Vector DB**: embeddings stored as BLOB, cosine similarity search
/// - **Keyword Search**: FTS5 virtual table with BM25 scoring
/// - **Hybrid Merge**: weighted fusion of vector + keyword results
/// - **Embedding Cache**: LRU-evicted cache to avoid redundant API calls
/// - **Safe Reindex**: temp DB → seed → sync → atomic swap → rollback
pub struct SqliteMemory {
    conn: Arc<Mutex<Connection>>,
    db_path: PathBuf,
    embedder: Arc<dyn EmbeddingProvider>,
    vector_weight: f32,
    keyword_weight: f32,
    cache_max: usize,
}

impl SqliteMemory {
    pub fn new(workspace_dir: &Path) -> anyhow::Result<Self> {
        Self::with_embedder(
            workspace_dir,
            Arc::new(super::embeddings::NoopEmbedding),
            0.7,
            0.3,
            10_000,
            None,
        )
    }

    /// Build SQLite memory with optional open timeout.
    ///
    /// If `open_timeout_secs` is `Some(n)`, opening the database is limited to `n` seconds
    /// (capped at 300). Useful when the DB file may be locked or on slow storage.
    /// `None` = wait indefinitely (default).
    pub fn with_embedder(
        workspace_dir: &Path,
        embedder: Arc<dyn EmbeddingProvider>,
        vector_weight: f32,
        keyword_weight: f32,
        cache_max: usize,
        open_timeout_secs: Option<u64>,
    ) -> anyhow::Result<Self> {
        let db_path = workspace_dir.join("memory").join("brain.db");

        if let Some(parent) = db_path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        let conn = Self::open_connection(&db_path, open_timeout_secs)?;

        // ── Production-grade PRAGMA tuning ──────────────────────
        // WAL mode: concurrent reads during writes, crash-safe
        // normal sync: 2× write speed, still durable on WAL
        // mmap 8 MB: let the OS page-cache serve hot reads
        // cache 2 MB: keep ~500 hot pages in-process
        // temp_store memory: temp tables never hit disk
        conn.execute_batch(
            "PRAGMA journal_mode = WAL;
             PRAGMA synchronous  = NORMAL;
             PRAGMA mmap_size    = 8388608;
             PRAGMA cache_size   = -2000;
             PRAGMA temp_store   = MEMORY;
             PRAGMA foreign_keys = ON;",
        )?;

        Self::init_schema(&conn)?;

        Ok(Self {
            conn: Arc::new(Mutex::new(conn)),
            db_path,
            embedder,
            vector_weight,
            keyword_weight,
            cache_max,
        })
    }

    /// Open SQLite connection, optionally with a timeout (for locked/slow storage).
    fn open_connection(
        db_path: &Path,
        open_timeout_secs: Option<u64>,
    ) -> anyhow::Result<Connection> {
        let path_buf = db_path.to_path_buf();

        let conn = if let Some(secs) = open_timeout_secs {
            let capped = secs.min(SQLITE_OPEN_TIMEOUT_CAP_SECS);
            let (tx, rx) = mpsc::channel();
            thread::spawn(move || {
                let result = Connection::open(&path_buf);
                let _ = tx.send(result);
            });
            match rx.recv_timeout(Duration::from_secs(capped)) {
                Ok(Ok(c)) => c,
                Ok(Err(e)) => return Err(e).context("SQLite failed to open database"),
                Err(mpsc::RecvTimeoutError::Timeout) => {
                    anyhow::bail!("SQLite connection open timed out after {} seconds", capped);
                }
                Err(mpsc::RecvTimeoutError::Disconnected) => {
                    anyhow::bail!("SQLite open thread exited unexpectedly");
                }
            }
        } else {
            Connection::open(&path_buf).context("SQLite failed to open database")?
        };

        Ok(conn)
    }

    /// Initialize all tables: memories, FTS5, `embedding_cache`
    fn init_schema(conn: &Connection) -> anyhow::Result<()> {
        conn.execute_batch(
            "-- Core memories table
            CREATE TABLE IF NOT EXISTS memories (
                id          TEXT PRIMARY KEY,
                key         TEXT NOT NULL UNIQUE,
                content     TEXT NOT NULL,
                category    TEXT NOT NULL DEFAULT 'core',
                embedding   BLOB,
                created_at  TEXT NOT NULL,
                updated_at  TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_memories_category ON memories(category);
            CREATE INDEX IF NOT EXISTS idx_memories_key ON memories(key);

            -- FTS5 full-text search (BM25 scoring)
            CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5(
                key, content, content=memories, content_rowid=rowid
            );

            -- FTS5 triggers: keep in sync with memories table
            CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN
                INSERT INTO memories_fts(rowid, key, content)
                VALUES (new.rowid, new.key, new.content);
            END;
            CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN
                INSERT INTO memories_fts(memories_fts, rowid, key, content)
                VALUES ('delete', old.rowid, old.key, old.content);
            END;
            CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN
                INSERT INTO memories_fts(memories_fts, rowid, key, content)
                VALUES ('delete', old.rowid, old.key, old.content);
                INSERT INTO memories_fts(rowid, key, content)
                VALUES (new.rowid, new.key, new.content);
            END;

            -- Embedding cache with LRU eviction
            CREATE TABLE IF NOT EXISTS embedding_cache (
                content_hash TEXT PRIMARY KEY,
                embedding    BLOB NOT NULL,
                created_at   TEXT NOT NULL,
                accessed_at  TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_cache_accessed ON embedding_cache(accessed_at);",
        )?;

        // Migration: add session_id column if not present (safe to run repeatedly)
        let has_session_id: bool = conn
            .prepare("SELECT sql FROM sqlite_master WHERE type='table' AND name='memories'")?
            .query_row([], |row| row.get::<_, String>(0))?
            .contains("session_id");
        if !has_session_id {
            conn.execute_batch(
                "ALTER TABLE memories ADD COLUMN session_id TEXT;
                 CREATE INDEX IF NOT EXISTS idx_memories_session ON memories(session_id);",
            )?;
        }

        // Migration: sessions table for session lifecycle tracking
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS sessions (
                id            TEXT PRIMARY KEY,
                started_at    TEXT NOT NULL,
                ended_at      TEXT,
                status        TEXT NOT NULL DEFAULT 'active',
                message_count INTEGER NOT NULL DEFAULT 0,
                last_activity TEXT NOT NULL,
                token_hash    TEXT,
                metadata      TEXT
            );
             CREATE INDEX IF NOT EXISTS idx_sessions_status ON sessions(status);
             CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at);
             CREATE INDEX IF NOT EXISTS idx_sessions_last_activity ON sessions(last_activity);
             CREATE INDEX IF NOT EXISTS idx_sessions_token ON sessions(token_hash);

             CREATE TABLE IF NOT EXISTS session_snapshots (
                 id                TEXT PRIMARY KEY,
                 session_id        TEXT NOT NULL,
                 snapshot_kind     TEXT NOT NULL,
                 created_at        TEXT NOT NULL,
                 payload           TEXT NOT NULL,
                 is_resume_capable INTEGER NOT NULL DEFAULT 0,
                 FOREIGN KEY(session_id) REFERENCES sessions(id)
             );
             CREATE INDEX IF NOT EXISTS idx_session_snapshots_session_created
                 ON session_snapshots(session_id, created_at DESC);
             CREATE INDEX IF NOT EXISTS idx_session_snapshots_session_kind_created
                 ON session_snapshots(session_id, snapshot_kind, created_at DESC);
             CREATE INDEX IF NOT EXISTS idx_session_snapshots_resume_capable
                 ON session_snapshots(session_id, is_resume_capable, created_at DESC);

             CREATE TABLE IF NOT EXISTS session_state (
                 session_id                    TEXT PRIMARY KEY,
                 lifecycle_state               TEXT NOT NULL DEFAULT 'active',
                 latest_tldr_snapshot_id       TEXT,
                 latest_compact_snapshot_id    TEXT,
                 pending_hydration_snapshot_id TEXT,
                 suspended_at                  TEXT,
                 updated_at                    TEXT NOT NULL,
                 FOREIGN KEY(session_id) REFERENCES sessions(id),
                 FOREIGN KEY(latest_tldr_snapshot_id) REFERENCES session_snapshots(id),
                 FOREIGN KEY(latest_compact_snapshot_id) REFERENCES session_snapshots(id),
                 FOREIGN KEY(pending_hydration_snapshot_id) REFERENCES session_snapshots(id)
             );
             CREATE INDEX IF NOT EXISTS idx_session_state_lifecycle
                  ON session_state(lifecycle_state, updated_at DESC);",
        )?;

        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS tasks (
                 id          TEXT PRIMARY KEY,
                 title       TEXT NOT NULL,
                 description TEXT NOT NULL,
                 status      TEXT NOT NULL CHECK (status IN ('pending', 'in_progress', 'completed', 'cancelled')),
                 priority    TEXT NOT NULL CHECK (priority IN ('low', 'medium', 'high')),
                 session_id  TEXT,
                 created_at  TEXT NOT NULL,
                 updated_at  TEXT NOT NULL,
                 FOREIGN KEY(session_id) REFERENCES sessions(id)
             );
             CREATE INDEX IF NOT EXISTS idx_tasks_created_id
                 ON tasks(created_at DESC, id ASC);
             CREATE INDEX IF NOT EXISTS idx_tasks_session_created_id
                 ON tasks(session_id, created_at DESC, id ASC);
             CREATE INDEX IF NOT EXISTS idx_tasks_status_created_id
                 ON tasks(status, created_at DESC, id ASC);
             CREATE INDEX IF NOT EXISTS idx_tasks_priority_created_id
                 ON tasks(priority, created_at DESC, id ASC);",
        )?;

        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS dream_sessions (
                 session_id              TEXT PRIMARY KEY,
                 status                  TEXT NOT NULL CHECK (status IN ('pending', 'running', 'completed', 'failed')),
                 trigger_reason          TEXT NOT NULL CHECK (trigger_reason IN ('time_elapsed', 'session_count', 'manual')),
                 completion_recorded_at  TEXT NOT NULL,
                 last_attempt_at         TEXT,
                 completed_at            TEXT,
                 artifact_refs           TEXT NOT NULL DEFAULT '[]',
                 failure_reason          TEXT,
                 FOREIGN KEY(session_id) REFERENCES sessions(id)
             );
             CREATE INDEX IF NOT EXISTS idx_dream_sessions_status
                 ON dream_sessions(status);
             CREATE INDEX IF NOT EXISTS idx_dream_sessions_completed_at
                 ON dream_sessions(completed_at DESC);",
        )?;

        Ok(())
    }

    fn category_to_str(cat: &MemoryCategory) -> String {
        match cat {
            MemoryCategory::Core => "core".into(),
            MemoryCategory::Daily => "daily".into(),
            MemoryCategory::Conversation => "conversation".into(),
            MemoryCategory::Custom(name) => name.clone(),
        }
    }

    fn str_to_category(s: &str) -> MemoryCategory {
        match s {
            "core" => MemoryCategory::Core,
            "daily" => MemoryCategory::Daily,
            "conversation" => MemoryCategory::Conversation,
            other => MemoryCategory::Custom(other.to_string()),
        }
    }

    fn lifecycle_from_row(value: String) -> rusqlite::Result<SlashSessionLifecycle> {
        match value.as_str() {
            "active" => Ok(SlashSessionLifecycle::Active),
            "suspended" => Ok(SlashSessionLifecycle::Suspended),
            _ => Err(rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("unknown lifecycle_state value: {value}"),
                )),
            )),
        }
    }

    fn snapshot_kind_from_row(value: String) -> rusqlite::Result<SessionSnapshotKind> {
        match value.as_str() {
            "tldr" => Ok(SessionSnapshotKind::Tldr),
            "compact" => Ok(SessionSnapshotKind::Compact),
            _ => Err(rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("unknown snapshot_kind value: {value}"),
                )),
            )),
        }
    }

    fn task_status_from_row(value: String) -> rusqlite::Result<TaskStatus> {
        match value.as_str() {
            "pending" => Ok(TaskStatus::Pending),
            "in_progress" => Ok(TaskStatus::InProgress),
            "completed" => Ok(TaskStatus::Completed),
            "cancelled" => Ok(TaskStatus::Cancelled),
            _ => Err(rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("unknown task status value: {value}"),
                )),
            )),
        }
    }

    fn task_priority_from_row(value: String) -> rusqlite::Result<TaskPriority> {
        match value.as_str() {
            "low" => Ok(TaskPriority::Low),
            "medium" => Ok(TaskPriority::Medium),
            "high" => Ok(TaskPriority::High),
            _ => Err(rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("unknown task priority value: {value}"),
                )),
            )),
        }
    }

    fn task_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TaskRecord> {
        Ok(TaskRecord {
            id: row.get(0)?,
            title: row.get(1)?,
            description: row.get(2)?,
            status: Self::task_status_from_row(row.get::<_, String>(3)?)?,
            priority: Self::task_priority_from_row(row.get::<_, String>(4)?)?,
            session_id: row.get(5)?,
            created_at: row.get(6)?,
            updated_at: row.get(7)?,
        })
    }

    fn ensure_session_exists(conn: &Connection, session_id: &str) -> anyhow::Result<()> {
        let exists: i64 = conn.query_row(
            "SELECT COUNT(*) FROM sessions WHERE id = ?1",
            params![session_id],
            |row| row.get(0),
        )?;
        if exists == 0 {
            anyhow::bail!("unknown session: {session_id}");
        }
        Ok(())
    }

    fn dream_status_to_str(status: &DreamSessionDbStatus) -> &'static str {
        match status {
            DreamSessionDbStatus::Pending => "pending",
            DreamSessionDbStatus::Running => "running",
            DreamSessionDbStatus::Completed => "completed",
            DreamSessionDbStatus::Failed => "failed",
        }
    }

    fn dream_status_from_row(value: String) -> rusqlite::Result<DreamSessionDbStatus> {
        match value.as_str() {
            "pending" => Ok(DreamSessionDbStatus::Pending),
            "running" => Ok(DreamSessionDbStatus::Running),
            "completed" => Ok(DreamSessionDbStatus::Completed),
            "failed" => Ok(DreamSessionDbStatus::Failed),
            _ => Err(rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("unknown dream status value: {value}"),
                )),
            )),
        }
    }

    fn dream_trigger_reason_to_str(reason: &DreamTriggerDbReason) -> &'static str {
        match reason {
            DreamTriggerDbReason::TimeElapsed => "time_elapsed",
            DreamTriggerDbReason::SessionCount => "session_count",
            DreamTriggerDbReason::Manual => "manual",
        }
    }

    fn dream_trigger_reason_from_row(value: String) -> rusqlite::Result<DreamTriggerDbReason> {
        match value.as_str() {
            "time_elapsed" => Ok(DreamTriggerDbReason::TimeElapsed),
            "session_count" => Ok(DreamTriggerDbReason::SessionCount),
            "manual" => Ok(DreamTriggerDbReason::Manual),
            _ => Err(rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("unknown dream trigger reason value: {value}"),
                )),
            )),
        }
    }

    fn load_dream_session_row(
        conn: &Connection,
        session_id: &str,
    ) -> anyhow::Result<Option<DreamSessionRow>> {
        conn.query_row(
            "SELECT session_id, status, trigger_reason, completion_recorded_at,
                    last_attempt_at, completed_at, artifact_refs, failure_reason
             FROM dream_sessions WHERE session_id = ?1",
            params![session_id],
            |row| {
                let artifact_refs: String = row.get(6)?;
                let refs =
                    serde_json::from_str::<Vec<String>>(&artifact_refs).map_err(|error| {
                        rusqlite::Error::FromSqlConversionFailure(
                            6,
                            rusqlite::types::Type::Text,
                            Box::new(error),
                        )
                    })?;
                Ok(DreamSessionRow {
                    session_id: row.get(0)?,
                    status: Self::dream_status_from_row(row.get::<_, String>(1)?)?,
                    trigger_reason: Self::dream_trigger_reason_from_row(row.get::<_, String>(2)?)?,
                    completion_recorded_at: row.get(3)?,
                    last_attempt_at: row.get(4)?,
                    completed_at: row.get(5)?,
                    artifact_refs: refs,
                    failure_reason: row.get(7)?,
                })
            },
        )
        .optional()
        .map_err(Into::into)
    }

    fn upsert_dream_session_row(tx: &Transaction<'_>, row: &DreamSessionRow) -> anyhow::Result<()> {
        tx.execute(
            "INSERT INTO dream_sessions (
                session_id, status, trigger_reason, completion_recorded_at,
                last_attempt_at, completed_at, artifact_refs, failure_reason
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
             ON CONFLICT(session_id) DO UPDATE SET
                status = excluded.status,
                trigger_reason = excluded.trigger_reason,
                completion_recorded_at = excluded.completion_recorded_at,
                last_attempt_at = excluded.last_attempt_at,
                completed_at = excluded.completed_at,
                artifact_refs = excluded.artifact_refs,
                failure_reason = excluded.failure_reason",
            params![
                row.session_id,
                Self::dream_status_to_str(&row.status),
                Self::dream_trigger_reason_to_str(&row.trigger_reason),
                row.completion_recorded_at,
                row.last_attempt_at,
                row.completed_at,
                serde_json::to_string(&row.artifact_refs)?,
                row.failure_reason,
            ],
        )?;
        Ok(())
    }

    fn load_snapshot_by_id(
        conn: &Connection,
        snapshot_id: &str,
    ) -> anyhow::Result<Option<SessionSnapshotRecord>> {
        let mut stmt = conn.prepare(
            "SELECT id, session_id, snapshot_kind, created_at, payload, is_resume_capable
             FROM session_snapshots WHERE id = ?1",
        )?;
        let mut rows = stmt.query_map(params![snapshot_id], |row| {
            let payload_str: String = row.get(4)?;
            match serde_json::from_str::<serde_json::Value>(&payload_str) {
                Ok(payload) => Ok(SessionSnapshotRecord {
                    id: row.get(0)?,
                    session_id: row.get(1)?,
                    kind: Self::snapshot_kind_from_row(row.get::<_, String>(2)?)?,
                    created_at: row.get(3)?,
                    payload,
                    resume_capable: row.get::<_, i64>(5)? != 0,
                }),
                Err(e) => Err(rusqlite::Error::FromSqlConversionFailure(
                    4, // column index for payload (TEXT)
                    rusqlite::types::Type::Text,
                    Box::new(e),
                )),
            }
        })?;
        match rows.next() {
            Some(Ok(record)) => Ok(Some(record)),
            Some(Err(error)) => Err(error.into()),
            None => Ok(None),
        }
    }

    fn read_session_state(
        conn: &Connection,
        session_id: &str,
    ) -> anyhow::Result<Option<SessionStateRecord>> {
        let mut stmt = conn.prepare(
            "SELECT session_id, lifecycle_state, latest_tldr_snapshot_id, latest_compact_snapshot_id,
                    pending_hydration_snapshot_id, suspended_at, updated_at
             FROM session_state WHERE session_id = ?1",
        )?;
        let mut rows = stmt.query_map(params![session_id], |row| {
            Ok(SessionStateRecord {
                session_id: row.get(0)?,
                lifecycle: Self::lifecycle_from_row(row.get::<_, String>(1)?)?,
                latest_tldr_snapshot_id: row.get(2)?,
                latest_compact_snapshot_id: row.get(3)?,
                pending_hydration_snapshot_id: row.get(4)?,
                suspended_at: row.get(5)?,
                updated_at: row.get(6)?,
            })
        })?;
        match rows.next() {
            Some(Ok(record)) => Ok(Some(record)),
            Some(Err(error)) => Err(error.into()),
            None => Ok(None),
        }
    }

    /// Deterministic content hash for embedding cache.
    /// Uses SHA-256 (truncated) instead of DefaultHasher, which is
    /// explicitly documented as unstable across Rust versions.
    fn content_hash(text: &str) -> String {
        use sha2::{Digest, Sha256};
        let hash = Sha256::digest(text.as_bytes());
        // First 8 bytes → 16 hex chars, matching previous format length
        format!(
            "{:016x}",
            u64::from_be_bytes(
                hash[..8]
                    .try_into()
                    .expect("SHA-256 always produces >= 8 bytes")
            )
        )
    }

    /// Get embedding from cache, or compute + cache it
    async fn get_or_compute_embedding(&self, text: &str) -> anyhow::Result<Option<Vec<f32>>> {
        if self.embedder.dimensions() == 0 {
            return Ok(None); // Noop embedder
        }

        let hash = Self::content_hash(text);
        let now = Local::now().to_rfc3339();

        // Check cache (offloaded to blocking thread)
        let conn = self.conn.clone();
        let hash_c = hash.clone();
        let now_c = now.clone();
        let cached = tokio::task::spawn_blocking(move || -> anyhow::Result<Option<Vec<f32>>> {
            let conn = conn.lock();
            let mut stmt =
                conn.prepare("SELECT embedding FROM embedding_cache WHERE content_hash = ?1")?;
            let blob: Option<Vec<u8>> = stmt.query_row(params![hash_c], |row| row.get(0)).ok();
            if let Some(bytes) = blob {
                conn.execute(
                    "UPDATE embedding_cache SET accessed_at = ?1 WHERE content_hash = ?2",
                    params![now_c, hash_c],
                )?;
                return Ok(Some(vector::bytes_to_vec(&bytes)));
            }
            Ok(None)
        })
        .await??;

        if cached.is_some() {
            return Ok(cached);
        }

        // Compute embedding (async I/O)
        let embedding = self.embedder.embed_one(text).await?;
        let bytes = vector::vec_to_bytes(&embedding);

        // Store in cache + LRU eviction (offloaded to blocking thread)
        let conn = self.conn.clone();
        #[allow(clippy::cast_possible_wrap)]
        let cache_max = self.cache_max as i64;
        tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
            let conn = conn.lock();
            conn.execute(
                "INSERT OR REPLACE INTO embedding_cache (content_hash, embedding, created_at, accessed_at)
                 VALUES (?1, ?2, ?3, ?4)",
                params![hash, bytes, now, now],
            )?;
            conn.execute(
                "DELETE FROM embedding_cache WHERE content_hash IN (
                    SELECT content_hash FROM embedding_cache
                    ORDER BY accessed_at ASC
                    LIMIT MAX(0, (SELECT COUNT(*) FROM embedding_cache) - ?1)
                )",
                params![cache_max],
            )?;
            Ok(())
        })
        .await??;

        Ok(Some(embedding))
    }

    /// FTS5 BM25 keyword search
    fn fts5_search(
        conn: &Connection,
        query: &str,
        limit: usize,
    ) -> anyhow::Result<Vec<(String, f32)>> {
        // Escape FTS5 special chars and build query
        let fts_query: String = query
            .split_whitespace()
            .map(|w| format!("\"{w}\""))
            .collect::<Vec<_>>()
            .join(" OR ");

        if fts_query.is_empty() {
            return Ok(Vec::new());
        }

        let sql = "SELECT m.id, bm25(memories_fts) as score
                   FROM memories_fts f
                   JOIN memories m ON m.rowid = f.rowid
                   WHERE memories_fts MATCH ?1
                   ORDER BY score
                   LIMIT ?2";

        let mut stmt = conn.prepare(sql)?;
        #[allow(clippy::cast_possible_wrap)]
        let limit_i64 = limit as i64;

        let rows = stmt.query_map(params![fts_query, limit_i64], |row| {
            let id: String = row.get(0)?;
            let score: f64 = row.get(1)?;
            // BM25 returns negative scores (lower = better), negate for ranking
            #[allow(clippy::cast_possible_truncation)]
            Ok((id, (-score) as f32))
        })?;

        let mut results = Vec::new();
        for row in rows {
            results.push(row?);
        }
        Ok(results)
    }

    /// Vector similarity search: scan embeddings and compute cosine similarity.
    ///
    /// Optional `category` and `session_id` filters reduce full-table scans
    /// when the caller already knows the scope of relevant memories.
    fn vector_search(
        conn: &Connection,
        query_embedding: &[f32],
        limit: usize,
        category: Option<&str>,
        session_id: Option<&str>,
    ) -> anyhow::Result<Vec<(String, f32)>> {
        use std::fmt::Write;

        let mut sql = "SELECT id, embedding FROM memories WHERE embedding IS NOT NULL".to_string();
        let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
        let mut idx = 1;

        if let Some(cat) = category {
            let _ = write!(sql, " AND category = ?{idx}");
            param_values.push(Box::new(cat.to_string()));
            idx += 1;
        }
        if let Some(sid) = session_id {
            let _ = write!(sql, " AND session_id = ?{idx}");
            param_values.push(Box::new(sid.to_string()));
        }

        let mut stmt = conn.prepare(&sql)?;
        let params_ref: Vec<&dyn rusqlite::types::ToSql> =
            param_values.iter().map(AsRef::as_ref).collect();
        let rows = stmt.query_map(params_ref.as_slice(), |row| {
            let id: String = row.get(0)?;
            let blob: Vec<u8> = row.get(1)?;
            Ok((id, blob))
        })?;

        let mut scored: Vec<(String, f32)> = Vec::new();
        for row in rows {
            let (id, blob) = row?;
            let emb = vector::bytes_to_vec(&blob);
            let sim = vector::cosine_similarity(query_embedding, &emb);
            if sim > 0.0 {
                scored.push((id, sim));
            }
        }

        scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
        scored.truncate(limit);
        Ok(scored)
    }

    /// Safe reindex: rebuild FTS5 + embeddings with rollback on failure
    #[allow(dead_code)]
    pub async fn reindex(&self) -> anyhow::Result<usize> {
        // Step 1: Rebuild FTS5
        {
            let conn = self.conn.clone();
            tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
                let conn = conn.lock();
                conn.execute_batch("INSERT INTO memories_fts(memories_fts) VALUES('rebuild');")?;
                Ok(())
            })
            .await??;
        }

        // Step 2: Re-embed all memories that lack embeddings
        if self.embedder.dimensions() == 0 {
            return Ok(0);
        }

        let conn = self.conn.clone();
        let entries: Vec<(String, String)> = tokio::task::spawn_blocking(move || {
            let conn = conn.lock();
            let mut stmt =
                conn.prepare("SELECT id, content FROM memories WHERE embedding IS NULL")?;
            let rows = stmt.query_map([], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
            })?;
            Ok::<_, anyhow::Error>(rows.filter_map(std::result::Result::ok).collect())
        })
        .await??;

        let mut count = 0;
        for (id, content) in &entries {
            if let Ok(Some(emb)) = self.get_or_compute_embedding(content).await {
                let bytes = vector::vec_to_bytes(&emb);
                let conn = self.conn.clone();
                let id = id.clone();
                tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
                    let conn = conn.lock();
                    conn.execute(
                        "UPDATE memories SET embedding = ?1 WHERE id = ?2",
                        params![bytes, id],
                    )?;
                    Ok(())
                })
                .await??;
                count += 1;
            }
        }

        Ok(count)
    }

    fn hybrid_search(
        conn: &Connection,
        query: &str,
        limit: usize,
        query_embedding: Option<&Vec<f32>>,
        vector_weight: f32,
        keyword_weight: f32,
    ) -> anyhow::Result<Vec<vector::ScoredResult>> {
        let keyword_results = Self::fts5_search(conn, query, limit).unwrap_or_default();

        let vector_results = query_embedding
            .map(|qe| Self::vector_search(conn, qe, limit, None, None).unwrap_or_default())
            .unwrap_or_default();

        if vector_results.is_empty() {
            return Ok(keyword_results
                .iter()
                .map(|(id, score)| vector::ScoredResult {
                    id: id.clone(),
                    vector_score: None,
                    keyword_score: Some(*score),
                    final_score: *score,
                })
                .collect());
        }

        Ok(vector::hybrid_merge(
            &vector_results,
            &keyword_results,
            vector_weight,
            keyword_weight,
            limit,
        ))
    }

    fn fetch_and_filter_entries(
        conn: &Connection,
        merged: Vec<vector::ScoredResult>,
        limit: usize,
        session_filter: Option<&str>,
    ) -> Vec<MemoryEntry> {
        if merged.is_empty() {
            return Vec::new();
        }

        let entries = Self::fetch_entries_by_ids(conn, &merged).unwrap_or_default();
        let filtered = Self::filter_by_session(entries, session_filter);
        filtered.into_iter().take(limit).collect()
    }

    fn fetch_entries_by_ids(
        conn: &Connection,
        merged: &[vector::ScoredResult],
    ) -> anyhow::Result<Vec<MemoryEntry>> {
        let placeholders: String = (1..=merged.len())
            .map(|i| format!("?{i}"))
            .collect::<Vec<_>>()
            .join(", ");
        let sql = format!(
            "SELECT id, key, content, category, created_at, session_id \
             FROM memories WHERE id IN ({placeholders})"
        );
        let mut stmt = conn.prepare(&sql)?;
        let id_params: Vec<Box<dyn rusqlite::types::ToSql>> = merged
            .iter()
            .map(|s| Box::new(s.id.clone()) as Box<dyn rusqlite::types::ToSql>)
            .collect();
        let params_ref: Vec<&dyn rusqlite::types::ToSql> =
            id_params.iter().map(AsRef::as_ref).collect();
        let rows = stmt.query_map(params_ref.as_slice(), |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, String>(3)?,
                row.get::<_, String>(4)?,
                row.get::<_, Option<String>>(5)?,
            ))
        })?;

        let mut entry_map = std::collections::HashMap::new();
        for row in rows {
            let (id, key, content, cat, ts, sid) = row?;
            entry_map.insert(id, (key, content, cat, ts, sid));
        }

        let mut results = Vec::new();
        for scored in merged {
            if let Some((key, content, cat, ts, sid)) = entry_map.remove(&scored.id) {
                results.push(MemoryEntry {
                    id: scored.id.clone(),
                    key,
                    content,
                    category: Self::str_to_category(&cat),
                    timestamp: ts,
                    session_id: sid,
                    score: Some(f64::from(scored.final_score)),
                });
            }
        }
        Ok(results)
    }

    fn filter_by_session(
        mut entries: Vec<MemoryEntry>,
        session_filter: Option<&str>,
    ) -> Vec<MemoryEntry> {
        if let Some(sid) = session_filter {
            entries.retain(|e| e.session_id.as_deref() == Some(sid));
        }
        entries
    }

    fn like_search(
        conn: &Connection,
        query: &str,
        limit: usize,
        session_filter: Option<&str>,
    ) -> anyhow::Result<Vec<MemoryEntry>> {
        const MAX_LIKE_KEYWORDS: usize = 8;
        let keywords: Vec<String> = query
            .split_whitespace()
            .take(MAX_LIKE_KEYWORDS)
            .map(|w| format!("%{w}%"))
            .collect();

        if keywords.is_empty() {
            return Ok(Vec::new());
        }

        let conditions: Vec<String> = keywords
            .iter()
            .enumerate()
            .map(|(i, _)| format!("(content LIKE ?{} OR key LIKE ?{})", i * 2 + 1, i * 2 + 2))
            .collect();
        let where_clause = conditions.join(" OR ");
        let sql = format!(
            "SELECT id, key, content, category, created_at, session_id FROM memories
             WHERE {where_clause}
             ORDER BY updated_at DESC
             LIMIT ?{}",
            keywords.len() * 2 + 1
        );
        let mut stmt = conn.prepare(&sql)?;
        let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
        for kw in &keywords {
            param_values.push(Box::new(kw.clone()));
            param_values.push(Box::new(kw.clone()));
        }
        #[allow(clippy::cast_possible_wrap)]
        param_values.push(Box::new(limit as i64));
        let params_ref: Vec<&dyn rusqlite::types::ToSql> =
            param_values.iter().map(AsRef::as_ref).collect();
        let rows = stmt.query_map(params_ref.as_slice(), |row| {
            Ok(MemoryEntry {
                id: row.get(0)?,
                key: row.get(1)?,
                content: row.get(2)?,
                category: Self::str_to_category(&row.get::<_, String>(3)?),
                timestamp: row.get(4)?,
                session_id: row.get(5)?,
                score: Some(1.0),
            })
        })?;

        let entries: Vec<MemoryEntry> = rows.filter_map(|r| r.ok()).collect();
        Ok(Self::filter_by_session(entries, session_filter))
    }

    fn capped_list_limit(limit: u32) -> i64 {
        i64::from(limit.min(MAX_LIST_LIMIT))
    }
}

#[async_trait]
impl Memory for SqliteMemory {
    fn name(&self) -> &str {
        "sqlite"
    }

    async fn store(
        &self,
        key: &str,
        content: &str,
        category: MemoryCategory,
        session_id: Option<&str>,
    ) -> anyhow::Result<()> {
        // Compute embedding (async, before blocking work)
        let embedding_bytes = self
            .get_or_compute_embedding(content)
            .await?
            .map(|emb| vector::vec_to_bytes(&emb));

        let conn = self.conn.clone();
        let key = key.to_string();
        let content = content.to_string();
        let session_id = session_id.map(String::from);

        tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
            let conn = conn.lock();
            let now = Local::now().to_rfc3339();
            let cat = Self::category_to_str(&category);
            let id = Uuid::new_v4().to_string();

            conn.execute(
                "INSERT INTO memories (id, key, content, category, embedding, created_at, updated_at, session_id)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
                 ON CONFLICT(key) DO UPDATE SET
                    content = excluded.content,
                    category = excluded.category,
                    embedding = excluded.embedding,
                    updated_at = excluded.updated_at,
                    session_id = excluded.session_id",
                params![id, key, content, cat, embedding_bytes, now, now, session_id],
            )?;
            Ok(())
        })
        .await?
    }

    async fn recall(
        &self,
        query: &str,
        limit: usize,
        session_id: Option<&str>,
    ) -> anyhow::Result<Vec<MemoryEntry>> {
        if query.trim().is_empty() {
            return Ok(Vec::new());
        }

        let query_embedding = self.get_or_compute_embedding(query).await?;

        let conn = self.conn.clone();
        let query = query.to_string();
        let session_id = session_id.map(String::from);
        let vector_weight = self.vector_weight;
        let keyword_weight = self.keyword_weight;

        tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<MemoryEntry>> {
            let conn = conn.lock();
            let session_ref = session_id.as_deref();

            let merged = Self::hybrid_search(
                &conn,
                &query,
                limit * 2,
                query_embedding.as_ref(),
                vector_weight,
                keyword_weight,
            )?;

            let mut results = Self::fetch_and_filter_entries(&conn, merged, limit, session_ref);
            if results.is_empty() {
                results = Self::like_search(&conn, &query, limit, session_ref).unwrap_or_default();
            }

            Ok(results)
        })
        .await?
    }

    async fn get(&self, key: &str) -> anyhow::Result<Option<MemoryEntry>> {
        let conn = self.conn.clone();
        let key = key.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<MemoryEntry>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT id, key, content, category, created_at, session_id FROM memories WHERE key = ?1",
            )?;

            let mut rows = stmt.query_map(params![key], |row| {
                Ok(MemoryEntry {
                    id: row.get(0)?,
                    key: row.get(1)?,
                    content: row.get(2)?,
                    category: Self::str_to_category(&row.get::<_, String>(3)?),
                    timestamp: row.get(4)?,
                    session_id: row.get(5)?,
                    score: None,
                })
            })?;

            match rows.next() {
                Some(Ok(entry)) => Ok(Some(entry)),
                _ => Ok(None),
            }
        })
        .await?
    }

    async fn list(
        &self,
        category: Option<&MemoryCategory>,
        session_id: Option<&str>,
    ) -> anyhow::Result<Vec<MemoryEntry>> {
        const DEFAULT_LIST_LIMIT: i64 = 1000;

        let conn = self.conn.clone();
        let category = category.cloned();
        let session_id = session_id.map(String::from);

        tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<MemoryEntry>> {
            let conn = conn.lock();
            let session_ref = session_id.as_deref();

            let row_mapper = |row: &rusqlite::Row| -> rusqlite::Result<MemoryEntry> {
                Ok(MemoryEntry {
                    id: row.get(0)?,
                    key: row.get(1)?,
                    content: row.get(2)?,
                    category: Self::str_to_category(&row.get::<_, String>(3)?),
                    timestamp: row.get(4)?,
                    session_id: row.get(5)?,
                    score: None,
                })
            };

            let query = match (&category, &session_ref) {
                (Some(_), Some(_)) => {
                    "SELECT id, key, content, category, created_at, session_id FROM memories
                     WHERE category = ?1 AND session_id = ?2 ORDER BY updated_at DESC LIMIT ?3"
                }
                (Some(_), None) => {
                    "SELECT id, key, content, category, created_at, session_id FROM memories
                     WHERE category = ?1 ORDER BY updated_at DESC LIMIT ?2"
                }
                (None, Some(_)) => {
                    "SELECT id, key, content, category, created_at, session_id FROM memories
                     WHERE session_id = ?1 ORDER BY updated_at DESC LIMIT ?2"
                }
                (None, None) => {
                    "SELECT id, key, content, category, created_at, session_id FROM memories
                     ORDER BY updated_at DESC LIMIT ?1"
                }
            };

            let mut stmt = conn.prepare(query)?;
            let mut results = Vec::new();

            let rows: Vec<MemoryEntry> = match (&category, &session_ref) {
                (Some(cat), Some(sid)) => {
                    let cat_str = Self::category_to_str(cat);
                    stmt.query_map(params![cat_str, sid, DEFAULT_LIST_LIMIT], row_mapper)?
                        .collect::<rusqlite::Result<Vec<_>>>()?
                }
                (Some(cat), None) => {
                    let cat_str = Self::category_to_str(cat);
                    stmt.query_map(params![cat_str, DEFAULT_LIST_LIMIT], row_mapper)?
                        .collect::<rusqlite::Result<Vec<_>>>()?
                }
                (None, Some(sid)) => stmt
                    .query_map(params![sid, DEFAULT_LIST_LIMIT], row_mapper)?
                    .collect::<rusqlite::Result<Vec<_>>>()?,
                (None, None) => stmt
                    .query_map(params![DEFAULT_LIST_LIMIT], row_mapper)?
                    .collect::<rusqlite::Result<Vec<_>>>()?,
            };
            results.extend(rows);

            Ok(results)
        })
        .await?
    }

    async fn forget(&self, key: &str) -> anyhow::Result<bool> {
        let conn = self.conn.clone();
        let key = key.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<bool> {
            let conn = conn.lock();
            let affected = conn.execute("DELETE FROM memories WHERE key = ?1", params![key])?;
            Ok(affected > 0)
        })
        .await?
    }

    async fn count(&self) -> anyhow::Result<usize> {
        let conn = self.conn.clone();

        tokio::task::spawn_blocking(move || -> anyhow::Result<usize> {
            let conn = conn.lock();
            let count: i64 =
                conn.query_row("SELECT COUNT(*) FROM memories", [], |row| row.get(0))?;
            #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
            Ok(count as usize)
        })
        .await?
    }

    async fn health_check(&self) -> bool {
        let conn = self.conn.clone();
        tokio::task::spawn_blocking(move || conn.lock().execute_batch("SELECT 1").is_ok())
            .await
            .unwrap_or(false)
    }

    async fn upsert_session(
        &self,
        session_id: &str,
        token_hash: Option<&str>,
    ) -> anyhow::Result<()> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();
        let token_hash = token_hash.map(String::from);

        tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
            let conn = conn.lock();
            let now = chrono::Utc::now().to_rfc3339();
            let affected = conn.execute(
                "INSERT INTO sessions (id, started_at, status, message_count, last_activity, token_hash)
                 VALUES (?1, ?2, 'active', 0, ?3, ?4)
                 ON CONFLICT(id) DO UPDATE SET
                    last_activity = excluded.last_activity,
                    status = 'active',
                    ended_at = NULL
                 WHERE sessions.token_hash IS excluded.token_hash",
                params![session_id, now, now, token_hash],
            )?;
            if affected == 0 {
                anyhow::bail!("session ownership mismatch for {session_id}");
            }
            Ok(())
        })
        .await?
    }

    async fn end_session(&self, session_id: &str) -> anyhow::Result<()> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
            let conn = conn.lock();
            let now = chrono::Utc::now().to_rfc3339();
            conn.execute(
                "UPDATE sessions SET status = 'ended', ended_at = ?1
                 WHERE id = ?2 AND ended_at IS NULL",
                params![now, session_id],
            )?;
            Ok(())
        })
        .await?
    }

    async fn update_session_activity(
        &self,
        session_id: &str,
        token_hash: Option<&str>,
    ) -> anyhow::Result<()> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();
        let token_hash = token_hash.map(String::from);

        tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
            let conn = conn.lock();
            let now = chrono::Utc::now().to_rfc3339();
            conn.execute(
                "UPDATE sessions SET message_count = message_count + 1, last_activity = ?1
                 WHERE id = ?2 AND ended_at IS NULL AND token_hash IS ?3",
                params![now, session_id, token_hash],
            )?;
            Ok(())
        })
        .await?
    }

    async fn list_sessions(
        &self,
        status: Option<SessionStatus>,
        limit: u32,
        offset: u32,
        sort: &str,
        order: &str,
    ) -> anyhow::Result<(Vec<SessionEntry>, u64)> {
        let conn = self.conn.clone();
        let status = status.map(|s| s.as_str().to_string());
        let limit = Self::capped_list_limit(limit);
        let offset = i64::from(offset);
        let sort_col = match sort {
            "started_at" => "started_at",
            _ => "last_activity",
        };
        let order_dir = if order == "asc" { "ASC" } else { "DESC" };
        let sql_query = format!(
            "SELECT id, started_at, ended_at, status, message_count, last_activity, metadata
             FROM sessions {} ORDER BY {sort_col} {order_dir}, id {order_dir} LIMIT ?1 OFFSET ?2",
            if status.is_some() {
                "WHERE status = ?3"
            } else {
                ""
            }
        );
        let count_query = format!(
            "SELECT COUNT(*) FROM sessions {}",
            if status.is_some() {
                "WHERE status = ?1"
            } else {
                ""
            }
        );

        tokio::task::spawn_blocking(move || -> anyhow::Result<(Vec<SessionEntry>, u64)> {
            let conn = conn.lock();

            #[allow(clippy::cast_sign_loss)]
            let total: u64 = if let Some(ref s) = status {
                conn.query_row(&count_query, params![s], |row| row.get::<_, i64>(0))? as u64
            } else {
                conn.query_row(&count_query, [], |row| row.get::<_, i64>(0))? as u64
            };

            let mut stmt = conn.prepare(&sql_query)?;
            let row_mapper = |row: &rusqlite::Row| -> rusqlite::Result<SessionEntry> {
                #[allow(clippy::cast_sign_loss)]
                Ok(SessionEntry {
                    id: row.get(0)?,
                    started_at: row.get(1)?,
                    ended_at: row.get(2)?,
                    status: row
                        .get::<_, String>(3)?
                        .parse()
                        .unwrap_or(SessionStatus::Active),
                    message_count: row.get::<_, i32>(4)? as u32,
                    last_activity: row.get(5)?,
                    metadata: row
                        .get::<_, Option<String>>(6)?
                        .and_then(|s| serde_json::from_str(&s).ok()),
                })
            };

            let rows: Vec<SessionEntry> = if let Some(ref s) = status {
                stmt.query_map(params![limit, offset, s], row_mapper)?
                    .collect::<rusqlite::Result<Vec<_>>>()?
            } else {
                stmt.query_map(params![limit, offset], row_mapper)?
                    .collect::<rusqlite::Result<Vec<_>>>()?
            };

            Ok((rows, total))
        })
        .await?
    }

    async fn get_session(&self, session_id: &str) -> anyhow::Result<Option<SessionEntry>> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<SessionEntry>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT id, started_at, ended_at, status, message_count, last_activity, metadata
                 FROM sessions WHERE id = ?1",
            )?;

            let mut rows = stmt.query_map(params![session_id], |row| {
                #[allow(clippy::cast_sign_loss)]
                Ok(SessionEntry {
                    id: row.get(0)?,
                    started_at: row.get(1)?,
                    ended_at: row.get(2)?,
                    status: row
                        .get::<_, String>(3)?
                        .parse()
                        .unwrap_or(SessionStatus::Active),
                    message_count: row.get::<_, i32>(4)? as u32,
                    last_activity: row.get(5)?,
                    metadata: row
                        .get::<_, Option<String>>(6)?
                        .and_then(|s| serde_json::from_str(&s).ok()),
                })
            })?;

            match rows.next() {
                Some(Ok(entry)) => Ok(Some(entry)),
                _ => Ok(None),
            }
        })
        .await?
    }

    async fn get_session_for_scope(
        &self,
        session_id: &str,
        caller_scope_key: &str,
    ) -> anyhow::Result<Option<SessionEntry>> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();
        let caller_scope_key = caller_scope_key.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<SessionEntry>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT id, started_at, ended_at, status, message_count, last_activity, metadata
                 FROM sessions WHERE id = ?1 AND token_hash IS ?2",
            )?;

            let mut rows = stmt.query_map(params![session_id, caller_scope_key], |row| {
                #[allow(clippy::cast_sign_loss)]
                Ok(SessionEntry {
                    id: row.get(0)?,
                    started_at: row.get(1)?,
                    ended_at: row.get(2)?,
                    status: row
                        .get::<_, String>(3)?
                        .parse()
                        .unwrap_or(SessionStatus::Active),
                    message_count: row.get::<_, i32>(4)? as u32,
                    last_activity: row.get(5)?,
                    metadata: row
                        .get::<_, Option<String>>(6)?
                        .and_then(|s| serde_json::from_str(&s).ok()),
                })
            })?;

            match rows.next() {
                Some(Ok(record)) => Ok(Some(record)),
                Some(Err(error)) => Err(error.into()),
                None => Ok(None),
            }
        })
        .await?
    }

    async fn list_sessions_for_token(
        &self,
        token_hash: &str,
        limit: u32,
        offset: u32,
    ) -> anyhow::Result<(Vec<SessionEntry>, u64)> {
        let conn = self.conn.clone();
        let token_hash = token_hash.to_string();
        let limit = Self::capped_list_limit(limit);
        let offset = i64::from(offset);

        tokio::task::spawn_blocking(move || -> anyhow::Result<(Vec<SessionEntry>, u64)> {
            let conn = conn.lock();

            #[allow(clippy::cast_sign_loss)]
            let total: u64 = conn.query_row(
                "SELECT COUNT(*) FROM sessions WHERE token_hash = ?1",
                params![token_hash],
                |row| row.get::<_, i64>(0),
            )? as u64;

            let mut stmt = conn.prepare(
                "SELECT id, started_at, ended_at, status, message_count, last_activity, metadata
                 FROM sessions WHERE token_hash = ?1
                 ORDER BY last_activity DESC, id DESC LIMIT ?2 OFFSET ?3",
            )?;

            let rows: Vec<SessionEntry> = stmt
                .query_map(params![token_hash, limit, offset], |row| {
                    #[allow(clippy::cast_sign_loss)]
                    Ok(SessionEntry {
                        id: row.get(0)?,
                        started_at: row.get(1)?,
                        ended_at: row.get(2)?,
                        status: row
                            .get::<_, String>(3)?
                            .parse()
                            .unwrap_or(SessionStatus::Active),
                        message_count: row.get::<_, i32>(4)? as u32,
                        last_activity: row.get(5)?,
                        metadata: row
                            .get::<_, Option<String>>(6)?
                            .and_then(|s| serde_json::from_str(&s).ok()),
                    })
                })?
                .collect::<rusqlite::Result<Vec<_>>>()?;

            Ok((rows, total))
        })
        .await?
    }

    async fn load_session_transcript_excerpt(
        &self,
        session_id: &str,
        limit: usize,
    ) -> anyhow::Result<Vec<MemoryEntry>> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();
        let limit = limit.max(1);

        tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<MemoryEntry>> {
            let conn = conn.lock();
            Self::ensure_session_exists(&conn, &session_id)?;
            let mut stmt = conn.prepare(
                "SELECT id, key, content, category, created_at, session_id
                 FROM memories
                 WHERE session_id = ?1 AND category = 'conversation'
                 ORDER BY updated_at DESC, id DESC
                 LIMIT ?2",
            )?;
            #[allow(clippy::cast_possible_wrap)]
            let limit_i64 = limit as i64;
            let entries = stmt
                .query_map(params![session_id, limit_i64], |row| {
                    Ok(MemoryEntry {
                        id: row.get(0)?,
                        key: row.get(1)?,
                        content: row.get(2)?,
                        category: Self::str_to_category(&row.get::<_, String>(3)?),
                        timestamp: row.get(4)?,
                        session_id: row.get(5)?,
                        score: None,
                    })
                })?
                .collect::<rusqlite::Result<Vec<_>>>()?;
            Ok(entries)
        })
        .await?
    }

    async fn create_session_snapshot(
        &self,
        session_id: &str,
        kind: SessionSnapshotKind,
        payload: serde_json::Value,
        resume_capable: bool,
    ) -> anyhow::Result<SessionSnapshotRecord> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<SessionSnapshotRecord> {
            let conn = conn.lock();
            Self::ensure_session_exists(&conn, &session_id)?;
            let id = Uuid::new_v4().to_string();
            let created_at = chrono::Utc::now().to_rfc3339();
            let payload_text = serde_json::to_string(&payload)?;
            conn.execute(
                "INSERT INTO session_snapshots (
                    id, session_id, snapshot_kind, created_at, payload, is_resume_capable
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    id,
                    session_id,
                    kind.as_str(),
                    created_at,
                    payload_text,
                    if resume_capable { 1_i64 } else { 0_i64 }
                ],
            )?;
            Ok(SessionSnapshotRecord {
                id,
                session_id,
                kind,
                created_at,
                payload,
                resume_capable,
            })
        })
        .await?
    }

    async fn get_session_snapshot(
        &self,
        snapshot_id: &str,
    ) -> anyhow::Result<Option<SessionSnapshotRecord>> {
        let conn = self.conn.clone();
        let snapshot_id = snapshot_id.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<SessionSnapshotRecord>> {
            let conn = conn.lock();
            Self::load_snapshot_by_id(&conn, &snapshot_id)
        })
        .await?
    }

    async fn get_session_state_record(
        &self,
        session_id: &str,
    ) -> anyhow::Result<Option<SessionStateRecord>> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<SessionStateRecord>> {
            let conn = conn.lock();
            Self::ensure_session_exists(&conn, &session_id)?;
            Self::read_session_state(&conn, &session_id)
        })
        .await?
    }

    async fn update_session_state_record(
        &self,
        state: SessionStateMutation,
    ) -> anyhow::Result<SessionStateRecord> {
        let conn = self.conn.clone();

        tokio::task::spawn_blocking(move || -> anyhow::Result<SessionStateRecord> {
            let conn = conn.lock();
            Self::ensure_session_exists(&conn, &state.session_id)?;
            let updated_at = chrono::Utc::now().to_rfc3339();
            conn.execute(
                "INSERT INTO session_state (
                    session_id, lifecycle_state, latest_tldr_snapshot_id, latest_compact_snapshot_id,
                    pending_hydration_snapshot_id, suspended_at, updated_at
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
                 ON CONFLICT(session_id) DO UPDATE SET
                    lifecycle_state = excluded.lifecycle_state,
                    latest_tldr_snapshot_id = excluded.latest_tldr_snapshot_id,
                    latest_compact_snapshot_id = excluded.latest_compact_snapshot_id,
                    pending_hydration_snapshot_id = excluded.pending_hydration_snapshot_id,
                    suspended_at = excluded.suspended_at,
                    updated_at = excluded.updated_at",
                params![
                    state.session_id,
                    state.lifecycle.as_str(),
                    state.latest_tldr_snapshot_id,
                    state.latest_compact_snapshot_id,
                    state.pending_hydration_snapshot_id,
                    state.suspended_at,
                    updated_at,
                ],
            )?;
            Self::read_session_state(&conn, &state.session_id)?.ok_or_else(|| {
                anyhow::anyhow!("session state missing after update for {}", state.session_id)
            })
        })
        .await?
    }

    async fn apply_session_state_patch(
        &self,
        patch: SessionStatePatch,
    ) -> anyhow::Result<SessionStateRecord> {
        let conn = self.conn.clone();

        tokio::task::spawn_blocking(move || -> anyhow::Result<SessionStateRecord> {
            let conn = conn.lock();
            Self::ensure_session_exists(&conn, &patch.session_id)?;

            let updated_at = chrono::Utc::now().to_rfc3339();

            // Convert SessionFieldPatch to params with explicit clear flags.
            // Clear must set NULL via CASE WHEN; Set uses the provided value; Keep preserves existing via COALESCE.
            let (latest_tldr, clear_latest_tldr) = match patch.latest_tldr_snapshot_id {
                crate::memory::SessionFieldPatch::Set(v) => (Some(v), false),
                crate::memory::SessionFieldPatch::Keep => (None, false),
                crate::memory::SessionFieldPatch::Clear => (None, true),
            };
            let (latest_compact, clear_latest_compact) = match patch.latest_compact_snapshot_id {
                crate::memory::SessionFieldPatch::Set(v) => (Some(v), false),
                crate::memory::SessionFieldPatch::Keep => (None, false),
                crate::memory::SessionFieldPatch::Clear => (None, true),
            };
            let (pending_hydration, clear_pending_hydration) = match patch.pending_hydration_snapshot_id {
                crate::memory::SessionFieldPatch::Set(v) => (Some(v), false),
                crate::memory::SessionFieldPatch::Keep => (None, false),
                crate::memory::SessionFieldPatch::Clear => (None, true),
            };
            let (suspended_at_val, clear_suspended_at) = match patch.suspended_at {
                crate::memory::SessionFieldPatch::Set(v) => (Some(v), false),
                crate::memory::SessionFieldPatch::Keep => (None, false),
                crate::memory::SessionFieldPatch::Clear => (None, true),
            };
            // lifecycle uses Option<SlashSessionLifecycle>: None means Keep (preserve existing),
            // Some(v) means Set. It is never nullable (NOT NULL column), so clear is never valid.
            let clear_lifecycle = false;
            let lifecycle_value = patch.lifecycle.as_ref().map(|l| l.as_str().to_string());

            conn.execute(
                "INSERT INTO session_state (
                    session_id,
                    lifecycle_state,
                    latest_tldr_snapshot_id,
                    latest_compact_snapshot_id,
                    pending_hydration_snapshot_id,
                    suspended_at,
                    updated_at
                ) VALUES (?1, COALESCE(?2, 'active'), ?3, ?4, ?5, ?6, ?7)
                ON CONFLICT(session_id) DO UPDATE SET
                    lifecycle_state = CASE WHEN ?8 THEN NULL ELSE COALESCE(excluded.lifecycle_state, session_state.lifecycle_state) END,
                    latest_tldr_snapshot_id = CASE WHEN ?9 THEN NULL ELSE COALESCE(excluded.latest_tldr_snapshot_id, session_state.latest_tldr_snapshot_id) END,
                    latest_compact_snapshot_id = CASE WHEN ?10 THEN NULL ELSE COALESCE(excluded.latest_compact_snapshot_id, session_state.latest_compact_snapshot_id) END,
                    pending_hydration_snapshot_id = CASE WHEN ?11 THEN NULL ELSE COALESCE(excluded.pending_hydration_snapshot_id, session_state.pending_hydration_snapshot_id) END,
                    suspended_at = CASE WHEN ?12 THEN NULL ELSE COALESCE(excluded.suspended_at, session_state.suspended_at) END,
                    updated_at = excluded.updated_at",
                params![
                    patch.session_id,
                    lifecycle_value,
                    latest_tldr,
                    latest_compact,
                    pending_hydration,
                    suspended_at_val,
                    updated_at,
                    clear_lifecycle,
                    clear_latest_tldr,
                    clear_latest_compact,
                    clear_pending_hydration,
                    clear_suspended_at,
                ],
            )?;

            Self::read_session_state(&conn, &patch.session_id)?.ok_or_else(|| {
                anyhow::anyhow!("session state missing after patch for {}", patch.session_id)
            })
        })
        .await?
    }

    async fn list_resumable_sessions(
        &self,
        caller_token_hash: Option<&str>,
        limit: u32,
        offset: u32,
    ) -> anyhow::Result<Vec<ResumableSessionEntry>> {
        let conn = self.conn.clone();
        let caller_token_hash = caller_token_hash.map(str::to_string);
        let limit = Self::capped_list_limit(limit.max(1));
        let offset = i64::from(offset);

        tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<ResumableSessionEntry>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT s.id, s.started_at, s.last_activity, snap.id, snap.created_at, snap.payload
                 FROM sessions s
                 JOIN session_state state ON state.session_id = s.id
                 JOIN session_snapshots snap ON snap.id = state.latest_compact_snapshot_id
                  WHERE s.status != 'ended'
                    AND state.lifecycle_state = 'suspended'
                    AND snap.is_resume_capable = 1
                    AND s.token_hash IS ?3
                  ORDER BY s.last_activity DESC, s.id DESC
                  LIMIT ?1 OFFSET ?2",
            )?;
            let rows = stmt
                .query_map(params![limit, offset, caller_token_hash], |row| {
                    let payload_str = row.get::<_, String>(5)?;
                    let payload =
                        serde_json::from_str::<serde_json::Value>(&payload_str).map_err(|e| {
                            rusqlite::Error::FromSqlConversionFailure(
                                5, // column index for payload (TEXT)
                                rusqlite::types::Type::Text,
                                Box::new(e),
                            )
                        })?;
                    let preview = payload
                        .get("preview")
                        .and_then(serde_json::Value::as_str)
                        .map(str::to_string)
                        .unwrap_or_default();
                    Ok(ResumableSessionEntry {
                        session_id: row.get(0)?,
                        started_at: row.get(1)?,
                        last_activity: row.get(2)?,
                        snapshot_id: row.get(3)?,
                        snapshot_created_at: row.get(4)?,
                        preview,
                    })
                })?
                .collect::<rusqlite::Result<Vec<_>>>()?;
            Ok(rows)
        })
        .await?
    }

    async fn list_session_rows_for_scope(
        &self,
        caller_scope_key: &str,
        limit: u32,
        offset: u32,
    ) -> anyhow::Result<Vec<SessionListEntry>> {
        let conn = self.conn.clone();
        let caller_scope_key = caller_scope_key.to_string();
        let limit = Self::capped_list_limit(limit.max(1));
        let offset = i64::from(offset);

        tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<SessionListEntry>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT s.id,
                        s.last_activity,
                        CASE
                            WHEN state.lifecycle_state = 'suspended' THEN 'suspended'
                            ELSE 'active'
                        END AS lifecycle,
                        CASE
                            WHEN state.lifecycle_state = 'suspended'
                                 AND state.latest_compact_snapshot_id IS NOT NULL
                                 AND snap.id IS NOT NULL
                                 AND snap.is_resume_capable = 1
                            THEN 1
                            ELSE 0
                        END AS resumable
                 FROM sessions s
                 LEFT JOIN session_state state ON state.session_id = s.id
                  LEFT JOIN session_snapshots snap
                      ON snap.id = state.latest_compact_snapshot_id
                      AND snap.session_id = s.id
                      AND snap.snapshot_kind = 'compact'
                 WHERE s.status != 'ended'
                   AND s.token_hash IS ?3
                 ORDER BY s.last_activity DESC, s.id DESC
                 LIMIT ?1 OFFSET ?2",
            )?;
            let rows = stmt
                .query_map(params![limit, offset, caller_scope_key], |row| {
                    Ok(SessionListEntry {
                        id: row.get(0)?,
                        last_activity: row.get(1)?,
                        lifecycle: Self::lifecycle_from_row(row.get::<_, String>(2)?)?,
                        resumable: row.get::<_, i64>(3)? != 0,
                    })
                })?
                .collect::<rusqlite::Result<Vec<_>>>()?;
            Ok(rows)
        })
        .await?
    }

    async fn get_resumable_session_for_scope(
        &self,
        session_id: &str,
        caller_scope_key: &str,
    ) -> anyhow::Result<Option<ResumableSessionEntry>> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();
        let caller_scope_key = caller_scope_key.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<ResumableSessionEntry>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT s.id, s.started_at, s.last_activity, snap.id, snap.created_at, snap.payload
                 FROM sessions s
                 JOIN session_state state ON state.session_id = s.id
                 JOIN session_snapshots snap ON snap.id = state.latest_compact_snapshot_id
                  WHERE s.id = ?1
                    AND s.status != 'ended'
                    AND state.lifecycle_state = 'suspended'
                    AND snap.is_resume_capable = 1
                    AND s.token_hash IS ?2",
            )?;

            stmt.query_row(params![session_id, caller_scope_key], |row| {
                let payload_str = row.get::<_, String>(5)?;
                let payload =
                    serde_json::from_str::<serde_json::Value>(&payload_str).map_err(|e| {
                        rusqlite::Error::FromSqlConversionFailure(
                            5,
                            rusqlite::types::Type::Text,
                            Box::new(e),
                        )
                    })?;
                let preview = payload
                    .get("preview")
                    .and_then(serde_json::Value::as_str)
                    .map(str::to_string)
                    .unwrap_or_default();
                Ok(ResumableSessionEntry {
                    session_id: row.get(0)?,
                    started_at: row.get(1)?,
                    last_activity: row.get(2)?,
                    snapshot_id: row.get(3)?,
                    snapshot_created_at: row.get(4)?,
                    preview,
                })
            })
            .optional()
            .map_err(Into::into)
        })
        .await?
    }

    async fn take_pending_resume_hydration(
        &self,
        session_id: &str,
    ) -> anyhow::Result<Option<SessionSnapshotRecord>> {
        let conn = self.conn.clone();
        let session_id = session_id.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<SessionSnapshotRecord>> {
            let mut conn = conn.lock();
            Self::ensure_session_exists(&conn, &session_id)?;
            let tx = conn.transaction()?;
            let pending_snapshot_id: Option<String> = tx
                .query_row(
                    "SELECT pending_hydration_snapshot_id FROM session_state WHERE session_id = ?1",
                    params![session_id],
                    |row| row.get(0),
                )
                .optional()?
                .flatten();
            let Some(snapshot_id) = pending_snapshot_id else {
                tx.commit()?;
                return Ok(None);
            };
            let snapshot = Self::load_snapshot_by_id(&tx, &snapshot_id)?;
            let Some(snapshot) = snapshot else {
                tx.commit()?;
                anyhow::bail!("pending resume snapshot missing: {snapshot_id}");
            };
            tx.execute(
                "UPDATE session_state SET pending_hydration_snapshot_id = NULL, updated_at = ?2
                 WHERE session_id = ?1",
                params![session_id, chrono::Utc::now().to_rfc3339()],
            )?;
            tx.commit()?;
            Ok(Some(snapshot))
        })
        .await?
    }

    async fn create_task(&self, input: TaskCreateInput) -> anyhow::Result<TaskRecord> {
        let conn = self.conn.clone();

        tokio::task::spawn_blocking(move || -> anyhow::Result<TaskRecord> {
            let conn = conn.lock();
            if let Some(session_id) = input.session_id.as_deref() {
                Self::ensure_session_exists(&conn, session_id)?;
            }

            conn.execute(
                "INSERT INTO tasks (id, title, description, status, priority, session_id, created_at, updated_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
                params![
                    input.id,
                    input.title,
                    input.description,
                    input.status.as_str(),
                    input.priority.as_str(),
                    input.session_id,
                    input.created_at,
                    input.updated_at,
                ],
            )?;

            Ok(TaskRecord {
                id: input.id,
                title: input.title,
                description: input.description,
                status: input.status,
                priority: input.priority,
                session_id: input.session_id,
                created_at: input.created_at,
                updated_at: input.updated_at,
            })
        })
        .await?
    }

    async fn get_task(&self, id: &str) -> anyhow::Result<Option<TaskRecord>> {
        let conn = self.conn.clone();
        let id = id.to_string();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<TaskRecord>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT id, title, description, status, priority, session_id, created_at, updated_at
                 FROM tasks WHERE id = ?1",
            )?;
            stmt.query_row(params![id], Self::task_from_row)
                .optional()
                .map_err(Into::into)
        })
        .await?
    }

    async fn list_tasks(&self, query: TaskListQuery) -> anyhow::Result<TaskListPage> {
        let conn = self.conn.clone();

        tokio::task::spawn_blocking(move || -> anyhow::Result<TaskListPage> {
            let conn = conn.lock();
            let effective_limit = query.limit.clamp(1, MAX_LIST_LIMIT);
            let fetch_limit = i64::from(effective_limit.saturating_add(1));
            let offset = i64::from(query.offset);

            let mut sql = String::from(
                "SELECT id, title, description, status, priority, session_id, created_at, updated_at FROM tasks",
            );
            let mut conditions = Vec::new();
            let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();

            if let Some(status) = query.status {
                conditions.push("status = ?".to_string());
                param_values.push(Box::new(status.as_str().to_string()));
            }
            if let Some(priority) = query.priority {
                conditions.push("priority = ?".to_string());
                param_values.push(Box::new(priority.as_str().to_string()));
            }
            if let Some(session_id) = query.session_id {
                conditions.push("session_id = ?".to_string());
                param_values.push(Box::new(session_id));
            }

            if !conditions.is_empty() {
                sql.push_str(" WHERE ");
                sql.push_str(&conditions.join(" AND "));
            }

            sql.push_str(" ORDER BY created_at DESC, id ASC LIMIT ? OFFSET ?");
            param_values.push(Box::new(fetch_limit));
            param_values.push(Box::new(offset));
            let params_ref: Vec<&dyn rusqlite::types::ToSql> =
                param_values.iter().map(AsRef::as_ref).collect();

            let mut stmt = conn.prepare(&sql)?;
            let mut tasks = stmt
                .query_map(params_ref.as_slice(), Self::task_from_row)?
                .collect::<rusqlite::Result<Vec<_>>>()?;

            let has_more = tasks.len() > effective_limit as usize;
            tasks.truncate(effective_limit as usize);

            Ok(TaskListPage { tasks, has_more })
        })
        .await?
    }

    async fn update_task(&self, patch: TaskPatch) -> anyhow::Result<Option<TaskRecord>> {
        let conn = self.conn.clone();

        tokio::task::spawn_blocking(move || -> anyhow::Result<Option<TaskRecord>> {
            let conn = conn.lock();
            let mut stmt = conn.prepare(
                "SELECT id, title, description, status, priority, session_id, created_at, updated_at
                 FROM tasks WHERE id = ?1",
            )?;
            let current = stmt
                .query_row(params![patch.id.clone()], Self::task_from_row)
                .optional()?;
            let Some(current) = current else {
                return Ok(None);
            };

            let updated_at = chrono::Utc::now().to_rfc3339();
            let next = TaskRecord {
                id: current.id,
                title: patch.title.unwrap_or(current.title),
                description: patch.description.unwrap_or(current.description),
                status: patch.status.unwrap_or(current.status),
                priority: patch.priority.unwrap_or(current.priority),
                session_id: current.session_id,
                created_at: current.created_at,
                updated_at: updated_at.clone(),
            };

            conn.execute(
                "UPDATE tasks
                 SET title = ?2, description = ?3, status = ?4, priority = ?5, updated_at = ?6
                 WHERE id = ?1",
                params![
                    patch.id,
                    next.title,
                    next.description,
                    next.status.as_str(),
                    next.priority.as_str(),
                    updated_at,
                ],
            )?;

            Ok(Some(next))
        })
        .await?
    }

    async fn memory_stats(&self) -> anyhow::Result<MemoryStats> {
        let conn = self.conn.clone();

        tokio::task::spawn_blocking(move || -> anyhow::Result<MemoryStats> {
            let conn = conn.lock();

            #[allow(clippy::cast_sign_loss)]
            let total_entries: u64 = conn.query_row("SELECT COUNT(*) FROM memories", [], |row| {
                row.get::<_, i64>(0)
            })? as u64;

            let mut by_category = std::collections::HashMap::new();
            {
                let mut stmt =
                    conn.prepare("SELECT category, COUNT(*) FROM memories GROUP BY category")?;
                let rows = stmt.query_map([], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
                })?;
                for row in rows {
                    let (cat, count) = row?;
                    #[allow(clippy::cast_sign_loss)]
                    by_category.insert(cat, count as u64);
                }
            }

            #[allow(clippy::cast_sign_loss)]
            let total_sessions: u64 =
                conn.query_row("SELECT COUNT(*) FROM sessions", [], |row| {
                    row.get::<_, i64>(0)
                })? as u64;

            #[allow(clippy::cast_sign_loss)]
            let active_sessions: u64 = conn.query_row(
                "SELECT COUNT(*) FROM sessions WHERE status = 'active'",
                [],
                |row| row.get::<_, i64>(0),
            )? as u64;

            Ok(MemoryStats {
                total_entries,
                by_category,
                total_sessions,
                active_sessions,
                backend: "sqlite".to_string(),
                cerebro_configured: false,
            })
        })
        .await?
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::memory::{
        SessionFieldPatch, TaskCreateInput, TaskListQuery, TaskPatch, TaskPriority, TaskStatus,
    };
    use tempfile::TempDir;

    fn temp_sqlite() -> (TempDir, SqliteMemory) {
        let tmp = TempDir::new().unwrap();
        let mem = SqliteMemory::new(tmp.path()).unwrap();
        (tmp, mem)
    }

    #[test]
    fn sqlite_dream_state_persists_per_session_and_survives_reopen() {
        let tmp = TempDir::new().unwrap();
        let workspace = tmp.path();
        let mem = SqliteMemory::new(workspace).unwrap();

        {
            let conn = mem.conn.lock();
            conn.execute(
                "INSERT INTO sessions (id, started_at, status, message_count, last_activity)
                 VALUES (?1, ?2, 'ended', 1, ?2)",
                params!["sess-123", "2026-04-26T10:00:00Z"],
            )
            .unwrap();
        }

        let core_key = "dream/session/sess-123";
        let core_content = "Dream summary for completed session sess-123";

        {
            let conn = mem.conn.lock();
            let tx = conn.unchecked_transaction().unwrap();
            tx.execute(
                "INSERT INTO memories (id, key, content, category, session_id, created_at, updated_at)
                 VALUES (?1, ?2, ?3, 'core', ?4, ?5, ?5)",
                params![
                    "artifact-1",
                    core_key,
                    core_content,
                    "sess-123",
                    "2026-04-26T10:05:00Z"
                ],
            )
            .unwrap();
            tx.execute(
                "INSERT INTO dream_sessions (
                    session_id, status, trigger_reason, completion_recorded_at,
                    last_attempt_at, completed_at, artifact_refs, failure_reason
                 ) VALUES (?1, 'completed', 'manual', ?2, ?3, ?3, ?4, NULL)
                 ON CONFLICT(session_id) DO UPDATE SET
                    status = excluded.status,
                    trigger_reason = excluded.trigger_reason,
                    completion_recorded_at = excluded.completion_recorded_at,
                    last_attempt_at = excluded.last_attempt_at,
                    completed_at = excluded.completed_at,
                    artifact_refs = excluded.artifact_refs,
                    failure_reason = excluded.failure_reason",
                params![
                    "sess-123",
                    "2026-04-26T10:00:00Z",
                    "2026-04-26T10:05:00Z",
                    "[\"dream/session/sess-123\"]"
                ],
            )
            .unwrap();
            tx.commit().unwrap();
        }

        drop(mem);
        let reopened = SqliteMemory::new(workspace).unwrap();
        let conn = reopened.conn.lock();
        let row: (String, String, String) = conn
            .query_row(
                "SELECT status, trigger_reason, artifact_refs FROM dream_sessions WHERE session_id = ?1",
                params!["sess-123"],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
        assert_eq!(row.0, "completed");
        assert_eq!(row.1, "manual");
        assert_eq!(row.2, "[\"dream/session/sess-123\"]");

        let memory: (String, String) = conn
            .query_row(
                "SELECT key, content FROM memories WHERE session_id = ?1",
                params!["sess-123"],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(memory.0, core_key);
        assert_eq!(memory.1, core_content);
    }

    #[test]
    fn sqlite_dream_metadata_and_artifact_write_is_atomic() {
        let (_tmp, mem) = temp_sqlite();
        {
            let conn = mem.conn.lock();
            conn.execute(
                "INSERT INTO sessions (id, started_at, status, message_count, last_activity)
                 VALUES (?1, ?2, 'ended', 1, ?2)",
                params!["sess-atomic", "2026-04-26T10:00:00Z"],
            )
            .unwrap();
        }

        let conn = mem.conn.lock();
        let tx = conn.unchecked_transaction().unwrap();
        tx.execute(
            "INSERT INTO memories (id, key, content, category, session_id, created_at, updated_at)
             VALUES (?1, ?2, ?3, 'core', ?4, ?5, ?5)",
            params![
                "artifact-atomic",
                "dream/session/sess-atomic",
                "atomic write",
                "sess-atomic",
                "2026-04-26T10:05:00Z"
            ],
        )
        .unwrap();
        tx.execute(
            "INSERT INTO dream_sessions (session_id, status, trigger_reason, completion_recorded_at, artifact_refs)
             VALUES (?1, 'running', 'manual', ?2, ?3)",
            params![
                "sess-atomic",
                "2026-04-26T10:00:00Z",
                "[\"dream/session/sess-atomic\"]"
            ],
        )
        .unwrap();
        tx.rollback().unwrap();

        let artifact_count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM memories WHERE session_id = ?1",
                params!["sess-atomic"],
                |row| row.get(0),
            )
            .unwrap();
        let metadata_count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM dream_sessions WHERE session_id = ?1",
                params!["sess-atomic"],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(artifact_count, 0);
        assert_eq!(metadata_count, 0);
    }

    #[test]
    fn sqlite_dream_duplicate_session_metadata_is_suppressed() {
        let (_tmp, mem) = temp_sqlite();
        {
            let conn = mem.conn.lock();
            conn.execute(
                "INSERT INTO sessions (id, started_at, status, message_count, last_activity)
                 VALUES (?1, ?2, 'ended', 1, ?2)",
                params!["sess-dup", "2026-04-26T10:00:00Z"],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO dream_sessions (session_id, status, trigger_reason, completion_recorded_at, artifact_refs)
                 VALUES (?1, 'completed', 'manual', ?2, ?3)",
                params![
                    "sess-dup",
                    "2026-04-26T10:00:00Z",
                    "[\"dream/session/sess-dup\"]"
                ],
            )
            .unwrap();
        }

        let conn = mem.conn.lock();
        let duplicate = conn.execute(
            "INSERT INTO dream_sessions (session_id, status, trigger_reason, completion_recorded_at, artifact_refs)
             VALUES (?1, 'completed', 'manual', ?2, ?3)",
            params![
                "sess-dup",
                "2026-04-26T10:00:00Z",
                "[\"dream/session/sess-dup\"]"
            ],
        );
        assert!(duplicate.is_err());

        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM dream_sessions WHERE session_id = ?1",
                params!["sess-dup"],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
    }

    #[test]
    fn sqlite_dream_retry_candidates_include_failed_and_pending_but_not_completed() {
        let (_tmp, mem) = temp_sqlite();
        let conn = mem.conn.lock();
        for (session_id, status) in [
            ("sess-pending", "pending"),
            ("sess-failed", "failed"),
            ("sess-completed", "completed"),
        ] {
            conn.execute(
                "INSERT INTO sessions (id, started_at, status, message_count, last_activity)
                 VALUES (?1, ?2, 'ended', 1, ?2)",
                params![session_id, "2026-04-26T10:00:00Z"],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO dream_sessions (session_id, status, trigger_reason, completion_recorded_at, artifact_refs, failure_reason)
                 VALUES (?1, ?2, 'manual', ?3, '[]', CASE WHEN ?2 = 'failed' THEN 'transient' ELSE NULL END)",
                params![session_id, status, "2026-04-26T10:00:00Z"],
            )
            .unwrap();
        }

        let mut stmt = conn
            .prepare(
                "SELECT session_id FROM dream_sessions
                 WHERE status IN ('pending', 'failed')
                 ORDER BY session_id ASC",
            )
            .unwrap();
        let ids: Vec<String> = stmt
            .query_map([], |row| row.get(0))
            .unwrap()
            .map(|row| row.unwrap())
            .collect();

        assert_eq!(
            ids,
            vec!["sess-failed".to_string(), "sess-pending".to_string()]
        );
    }

    #[tokio::test]
    async fn sqlite_name() {
        let (_tmp, mem) = temp_sqlite();
        assert_eq!(mem.name(), "sqlite");
    }

    #[tokio::test]
    async fn sqlite_health() {
        let (_tmp, mem) = temp_sqlite();
        assert!(mem.health_check().await);
    }

    #[tokio::test]
    async fn sqlite_task_roundtrip_create_get_list_and_update() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("session-123", None).await.unwrap();

        let created = mem
            .create_task(TaskCreateInput {
                id: "11111111-1111-4111-8111-111111111111".into(),
                title: "Review parity slice".into(),
                description: "Capture follow-up work".into(),
                status: TaskStatus::Pending,
                priority: TaskPriority::High,
                session_id: Some("session-123".into()),
                created_at: "2026-04-18T00:00:00Z".into(),
                updated_at: "2026-04-18T00:00:00Z".into(),
            })
            .await
            .unwrap();

        assert_eq!(created.id, "11111111-1111-4111-8111-111111111111");
        assert_eq!(created.status, TaskStatus::Pending);
        assert_eq!(created.session_id.as_deref(), Some("session-123"));

        let fetched = mem
            .get_task("11111111-1111-4111-8111-111111111111")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(fetched.title, "Review parity slice");

        let updated = mem
            .update_task(TaskPatch {
                id: "11111111-1111-4111-8111-111111111111".into(),
                title: Some("Review persistent parity slice".into()),
                description: None,
                status: Some(TaskStatus::InProgress),
                priority: Some(TaskPriority::Medium),
            })
            .await
            .unwrap()
            .unwrap();
        assert_eq!(updated.title, "Review persistent parity slice");
        assert_eq!(updated.status, TaskStatus::InProgress);
        assert_eq!(updated.priority, TaskPriority::Medium);
        assert_ne!(updated.updated_at, created.updated_at);

        let page = mem
            .list_tasks(TaskListQuery {
                session_id: Some("session-123".into()),
                status: Some(TaskStatus::InProgress),
                priority: None,
                limit: 10,
                offset: 0,
            })
            .await
            .unwrap();
        assert_eq!(page.tasks.len(), 1);
        assert!(!page.has_more);
        assert_eq!(page.tasks[0].id, "11111111-1111-4111-8111-111111111111");
    }

    #[tokio::test]
    async fn sqlite_task_list_uses_deterministic_order_and_page_metadata() {
        let (_tmp, mem) = temp_sqlite();

        for (id, title, created_at) in [
            (
                "11111111-1111-4111-8111-111111111111",
                "oldest",
                "2026-04-18T00:00:00Z",
            ),
            (
                "22222222-2222-4222-8222-222222222222",
                "middle",
                "2026-04-18T01:00:00Z",
            ),
            (
                "33333333-3333-4333-8333-333333333333",
                "newest",
                "2026-04-18T02:00:00Z",
            ),
        ] {
            mem.create_task(TaskCreateInput {
                id: id.into(),
                title: title.into(),
                description: String::new(),
                status: TaskStatus::Pending,
                priority: TaskPriority::Medium,
                session_id: None,
                created_at: created_at.into(),
                updated_at: created_at.into(),
            })
            .await
            .unwrap();
        }

        let first_page = mem
            .list_tasks(TaskListQuery {
                session_id: None,
                status: None,
                priority: None,
                limit: 2,
                offset: 0,
            })
            .await
            .unwrap();

        assert_eq!(
            first_page
                .tasks
                .iter()
                .map(|task| task.id.as_str())
                .collect::<Vec<_>>(),
            vec![
                "33333333-3333-4333-8333-333333333333",
                "22222222-2222-4222-8222-222222222222",
            ]
        );
        assert!(first_page.has_more);

        let second_page = mem
            .list_tasks(TaskListQuery {
                session_id: None,
                status: None,
                priority: None,
                limit: 2,
                offset: 2,
            })
            .await
            .unwrap();
        assert_eq!(second_page.tasks.len(), 1);
        assert_eq!(
            second_page.tasks[0].id,
            "11111111-1111-4111-8111-111111111111"
        );
        assert!(!second_page.has_more);
    }

    #[tokio::test]
    async fn sqlite_task_create_rejects_unknown_session_association() {
        let (_tmp, mem) = temp_sqlite();

        let error = mem
            .create_task(TaskCreateInput {
                id: "11111111-1111-4111-8111-111111111111".into(),
                title: "Review parity slice".into(),
                description: String::new(),
                status: TaskStatus::Pending,
                priority: TaskPriority::Medium,
                session_id: Some("missing-session".into()),
                created_at: "2026-04-18T00:00:00Z".into(),
                updated_at: "2026-04-18T00:00:00Z".into(),
            })
            .await
            .unwrap_err();

        assert!(error.to_string().contains("unknown session"));
    }

    #[tokio::test]
    async fn sqlite_store_and_get() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("user_lang", "Prefers Rust", MemoryCategory::Core, None)
            .await
            .unwrap();

        let entry = mem.get("user_lang").await.unwrap();
        assert!(entry.is_some());
        let entry = entry.unwrap();
        assert_eq!(entry.key, "user_lang");
        assert_eq!(entry.content, "Prefers Rust");
        assert_eq!(entry.category, MemoryCategory::Core);
    }

    #[tokio::test]
    async fn sqlite_store_upsert() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("pref", "likes Rust", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("pref", "loves Rust", MemoryCategory::Core, None)
            .await
            .unwrap();

        let entry = mem.get("pref").await.unwrap().unwrap();
        assert_eq!(entry.content, "loves Rust");
        assert_eq!(mem.count().await.unwrap(), 1);
    }

    #[tokio::test]
    async fn sqlite_recall_keyword() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "Rust is fast and safe", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("b", "Python is interpreted", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store(
            "c",
            "Rust has zero-cost abstractions",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();

        let results = mem.recall("Rust", 10, None).await.unwrap();
        assert_eq!(results.len(), 2);
        assert!(results
            .iter()
            .all(|r| r.content.to_lowercase().contains("rust")));
    }

    #[tokio::test]
    async fn sqlite_recall_multi_keyword() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "Rust is fast", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("b", "Rust is safe and fast", MemoryCategory::Core, None)
            .await
            .unwrap();

        let results = mem.recall("fast safe", 10, None).await.unwrap();
        assert!(!results.is_empty());
        // Entry with both keywords should score higher
        assert!(results[0].content.contains("safe") && results[0].content.contains("fast"));
    }

    #[tokio::test]
    async fn sqlite_recall_no_match() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "Rust rocks", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("javascript", 10, None).await.unwrap();
        assert!(results.is_empty());
    }

    #[tokio::test]
    async fn sqlite_forget() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("temp", "temporary data", MemoryCategory::Conversation, None)
            .await
            .unwrap();
        assert_eq!(mem.count().await.unwrap(), 1);

        let removed = mem.forget("temp").await.unwrap();
        assert!(removed);
        assert_eq!(mem.count().await.unwrap(), 0);
    }

    #[tokio::test]
    async fn sqlite_forget_nonexistent() {
        let (_tmp, mem) = temp_sqlite();
        let removed = mem.forget("nope").await.unwrap();
        assert!(!removed);
    }

    #[tokio::test]
    async fn sqlite_list_all() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "one", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("b", "two", MemoryCategory::Daily, None)
            .await
            .unwrap();
        mem.store("c", "three", MemoryCategory::Conversation, None)
            .await
            .unwrap();

        let all = mem.list(None, None).await.unwrap();
        assert_eq!(all.len(), 3);
    }

    #[tokio::test]
    async fn sqlite_list_by_category() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "core1", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("b", "core2", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("c", "daily1", MemoryCategory::Daily, None)
            .await
            .unwrap();

        let core = mem.list(Some(&MemoryCategory::Core), None).await.unwrap();
        assert_eq!(core.len(), 2);

        let daily = mem.list(Some(&MemoryCategory::Daily), None).await.unwrap();
        assert_eq!(daily.len(), 1);
    }

    #[tokio::test]
    async fn sqlite_count_empty() {
        let (_tmp, mem) = temp_sqlite();
        assert_eq!(mem.count().await.unwrap(), 0);
    }

    #[tokio::test]
    async fn sqlite_get_nonexistent() {
        let (_tmp, mem) = temp_sqlite();
        assert!(mem.get("nope").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn sqlite_db_persists() {
        let tmp = TempDir::new().unwrap();

        {
            let mem = SqliteMemory::new(tmp.path()).unwrap();
            mem.store("persist", "I survive restarts", MemoryCategory::Core, None)
                .await
                .unwrap();
        }

        // Reopen
        let mem2 = SqliteMemory::new(tmp.path()).unwrap();
        let entry = mem2.get("persist").await.unwrap();
        assert!(entry.is_some());
        assert_eq!(entry.unwrap().content, "I survive restarts");
    }

    #[tokio::test]
    async fn sqlite_category_roundtrip() {
        let (_tmp, mem) = temp_sqlite();
        let categories = [
            MemoryCategory::Core,
            MemoryCategory::Daily,
            MemoryCategory::Conversation,
            MemoryCategory::Custom("project".into()),
        ];

        for (i, cat) in categories.iter().enumerate() {
            mem.store(&format!("k{i}"), &format!("v{i}"), cat.clone(), None)
                .await
                .unwrap();
        }

        for (i, cat) in categories.iter().enumerate() {
            let entry = mem.get(&format!("k{i}")).await.unwrap().unwrap();
            assert_eq!(&entry.category, cat);
        }
    }

    // ── FTS5 search tests ────────────────────────────────────────

    #[tokio::test]
    async fn fts5_bm25_ranking() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "a",
            "Rust is a systems programming language",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        mem.store(
            "b",
            "Python is great for scripting",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        mem.store(
            "c",
            "Rust and Rust and Rust everywhere",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();

        let results = mem.recall("Rust", 10, None).await.unwrap();
        assert!(results.len() >= 2);
        // All results should contain "Rust"
        for r in &results {
            assert!(
                r.content.to_lowercase().contains("rust"),
                "Expected 'rust' in: {}",
                r.content
            );
        }
    }

    #[tokio::test]
    async fn fts5_multi_word_query() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "The quick brown fox jumps", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("b", "A lazy dog sleeps", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("c", "The quick dog runs fast", MemoryCategory::Core, None)
            .await
            .unwrap();

        let results = mem.recall("quick dog", 10, None).await.unwrap();
        assert!(!results.is_empty());
        // "The quick dog runs fast" matches both terms
        assert!(results[0].content.contains("quick"));
    }

    #[tokio::test]
    async fn recall_empty_query_returns_empty() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "data", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("", 10, None).await.unwrap();
        assert!(results.is_empty());
    }

    #[tokio::test]
    async fn recall_whitespace_query_returns_empty() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "data", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("   ", 10, None).await.unwrap();
        assert!(results.is_empty());
    }

    // ── Embedding cache tests ────────────────────────────────────

    #[test]
    fn content_hash_deterministic() {
        let h1 = SqliteMemory::content_hash("hello world");
        let h2 = SqliteMemory::content_hash("hello world");
        assert_eq!(h1, h2);
    }

    #[test]
    fn content_hash_different_inputs() {
        let h1 = SqliteMemory::content_hash("hello");
        let h2 = SqliteMemory::content_hash("world");
        assert_ne!(h1, h2);
    }

    // ── Schema tests ─────────────────────────────────────────────

    #[tokio::test]
    async fn schema_has_fts5_table() {
        let (_tmp, mem) = temp_sqlite();
        let conn = mem.conn.lock();
        // FTS5 table should exist
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='memories_fts'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn schema_has_embedding_cache() {
        let (_tmp, mem) = temp_sqlite();
        let conn = mem.conn.lock();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='embedding_cache'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn schema_memories_has_embedding_column() {
        let (_tmp, mem) = temp_sqlite();
        let conn = mem.conn.lock();
        // Check that embedding column exists by querying it
        let result = conn.execute_batch("SELECT embedding FROM memories LIMIT 0");
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn slash_session_schema_is_additive_and_idempotent() {
        let tmp = TempDir::new().unwrap();
        let mem = SqliteMemory::new(tmp.path()).unwrap();
        {
            let conn = mem.conn.lock();
            let snapshot_table_count: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='session_snapshots'",
                    [],
                    |row| row.get(0),
                )
                .unwrap();
            let state_table_count: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='session_state'",
                    [],
                    |row| row.get(0),
                )
                .unwrap();
            assert_eq!(snapshot_table_count, 1);
            assert_eq!(state_table_count, 1);
        }

        let reopened = SqliteMemory::new(tmp.path()).unwrap();
        let conn = reopened.conn.lock();
        let snapshot_index_count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name='idx_session_snapshots_resume_capable'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(snapshot_index_count, 1);
    }

    #[tokio::test]
    async fn slash_session_persistence_roundtrips_and_pending_hydration_is_single_use() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("session-1", None).await.unwrap();
        mem.store(
            "turn-1",
            "Discuss release checklist",
            MemoryCategory::Conversation,
            Some("session-1"),
        )
        .await
        .unwrap();

        let excerpt = mem
            .load_session_transcript_excerpt("session-1", 5)
            .await
            .unwrap();
        assert_eq!(excerpt.len(), 1);
        assert_eq!(excerpt[0].session_id.as_deref(), Some("session-1"));

        let snapshot = mem
            .create_session_snapshot(
                "session-1",
                SessionSnapshotKind::Compact,
                serde_json::json!({
                    "preview": "Discuss release checklist",
                    "summary": "Discuss release checklist",
                    "resume_context": "Resume from checklist",
                }),
                true,
            )
            .await
            .unwrap();
        let state = mem
            .apply_session_state_patch(SessionStatePatch {
                session_id: "session-1".into(),
                lifecycle: Some(SlashSessionLifecycle::Suspended),
                latest_tldr_snapshot_id: SessionFieldPatch::Clear,
                latest_compact_snapshot_id: SessionFieldPatch::Set(snapshot.id.clone()),
                pending_hydration_snapshot_id: SessionFieldPatch::Set(snapshot.id.clone()),
                suspended_at: SessionFieldPatch::Set("now".into()),
            })
            .await
            .unwrap();

        assert_eq!(state.lifecycle, SlashSessionLifecycle::Suspended);
        assert_eq!(
            state.pending_hydration_snapshot_id.as_deref(),
            Some(snapshot.id.as_str())
        );

        let resumable = mem.list_resumable_sessions(None, 10, 0).await.unwrap();
        assert_eq!(resumable.len(), 1);
        assert_eq!(resumable[0].session_id, "session-1");

        let first_hydration = mem
            .take_pending_resume_hydration("session-1")
            .await
            .unwrap();
        let second_hydration = mem
            .take_pending_resume_hydration("session-1")
            .await
            .unwrap();
        assert_eq!(
            first_hydration
                .as_ref()
                .map(|snapshot| snapshot.id.as_str()),
            Some(snapshot.id.as_str())
        );
        assert!(second_hydration.is_none());
    }

    #[tokio::test]
    async fn pending_hydration_missing_snapshot_keeps_pointer_for_retry() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("session-1", None).await.unwrap();

        let snapshot = mem
            .create_session_snapshot(
                "session-1",
                SessionSnapshotKind::Compact,
                serde_json::json!({
                    "preview": "Preview A",
                    "summary": "Preview A",
                    "resume_context": "Preview A",
                }),
                true,
            )
            .await
            .unwrap();

        mem.apply_session_state_patch(SessionStatePatch {
            session_id: "session-1".into(),
            lifecycle: Some(SlashSessionLifecycle::Active),
            latest_tldr_snapshot_id: SessionFieldPatch::Keep,
            latest_compact_snapshot_id: SessionFieldPatch::Keep,
            pending_hydration_snapshot_id: SessionFieldPatch::Set(snapshot.id.clone()),
            suspended_at: SessionFieldPatch::Clear,
        })
        .await
        .unwrap();

        {
            let conn = mem.conn.lock();
            conn.execute("PRAGMA foreign_keys = OFF", []).unwrap();
            conn.execute(
                "DELETE FROM session_snapshots WHERE id = ?1",
                params![snapshot.id],
            )
            .unwrap();
            conn.execute("PRAGMA foreign_keys = ON", []).unwrap();
        }

        let error = mem
            .take_pending_resume_hydration("session-1")
            .await
            .unwrap_err();
        assert!(error
            .to_string()
            .contains("pending resume snapshot missing"));

        let state = mem
            .get_session_state_record("session-1")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            state.pending_hydration_snapshot_id.as_deref(),
            Some(snapshot.id.as_str())
        );
    }

    #[tokio::test]
    async fn list_resumable_sessions_filters_by_token_scope() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("session-a", Some("token-a"))
            .await
            .unwrap();
        mem.upsert_session("session-b", Some("token-b"))
            .await
            .unwrap();

        for (session_id, preview) in [("session-a", "Preview A"), ("session-b", "Preview B")] {
            let snapshot = mem
                .create_session_snapshot(
                    session_id,
                    SessionSnapshotKind::Compact,
                    serde_json::json!({
                        "preview": preview,
                        "summary": preview,
                        "resume_context": preview,
                    }),
                    true,
                )
                .await
                .unwrap();
            mem.apply_session_state_patch(SessionStatePatch {
                session_id: session_id.into(),
                lifecycle: Some(SlashSessionLifecycle::Suspended),
                latest_tldr_snapshot_id: SessionFieldPatch::Keep,
                latest_compact_snapshot_id: SessionFieldPatch::Set(snapshot.id),
                pending_hydration_snapshot_id: SessionFieldPatch::Clear,
                suspended_at: SessionFieldPatch::Set("now".into()),
            })
            .await
            .unwrap();
        }

        let token_a_results = mem
            .list_resumable_sessions(Some("token-a"), 10, 0)
            .await
            .unwrap();
        assert_eq!(token_a_results.len(), 1);
        assert_eq!(token_a_results[0].session_id, "session-a");
        assert_eq!(token_a_results[0].preview, "Preview A");

        let token_b_results = mem
            .list_resumable_sessions(Some("token-b"), 10, 0)
            .await
            .unwrap();
        assert_eq!(token_b_results.len(), 1);
        assert_eq!(token_b_results[0].session_id, "session-b");
        assert_eq!(token_b_results[0].preview, "Preview B");

        let unscoped_results = mem.list_resumable_sessions(None, 10, 0).await.unwrap();
        assert!(unscoped_results.is_empty());
    }

    #[tokio::test]
    async fn list_session_rows_for_scope_filters_by_scope_and_excludes_ended_sessions() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("session-a", Some("token-a"))
            .await
            .unwrap();
        mem.upsert_session("session-b", Some("token-b"))
            .await
            .unwrap();
        mem.upsert_session("session-ended", Some("token-a"))
            .await
            .unwrap();
        mem.end_session("session-ended").await.unwrap();

        let rows = mem
            .list_session_rows_for_scope("token-a", 10, 0)
            .await
            .unwrap();

        let ids: Vec<_> = rows.into_iter().map(|row| row.id).collect();
        assert_eq!(ids, vec!["session-a"]);
    }

    #[tokio::test]
    async fn list_session_rows_for_scope_derives_lifecycle_and_resumable_authoritatively() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("session-active", Some("token-a"))
            .await
            .unwrap();
        mem.upsert_session("session-suspended", Some("token-a"))
            .await
            .unwrap();

        let snapshot = mem
            .create_session_snapshot(
                "session-suspended",
                SessionSnapshotKind::Compact,
                serde_json::json!({
                    "preview": "Preview A",
                    "summary": "Preview A",
                    "resume_context": "Preview A",
                }),
                true,
            )
            .await
            .unwrap();
        mem.apply_session_state_patch(SessionStatePatch {
            session_id: "session-suspended".into(),
            lifecycle: Some(SlashSessionLifecycle::Suspended),
            latest_tldr_snapshot_id: SessionFieldPatch::Keep,
            latest_compact_snapshot_id: SessionFieldPatch::Set(snapshot.id),
            pending_hydration_snapshot_id: SessionFieldPatch::Clear,
            suspended_at: SessionFieldPatch::Set("now".into()),
        })
        .await
        .unwrap();

        let rows = mem
            .list_session_rows_for_scope("token-a", 10, 0)
            .await
            .unwrap();

        assert!(rows.iter().any(|row| {
            row.id == "session-active"
                && row.lifecycle == SlashSessionLifecycle::Active
                && !row.resumable
        }));
        assert!(rows.iter().any(|row| {
            row.id == "session-suspended"
                && row.lifecycle == SlashSessionLifecycle::Suspended
                && row.resumable
        }));
    }

    #[tokio::test]
    async fn list_session_rows_for_scope_uses_last_activity_then_id_desc_ordering() {
        let (_tmp, mem) = temp_sqlite();
        for session_id in ["b", "a", "c"] {
            mem.upsert_session(session_id, Some("token-a"))
                .await
                .unwrap();
        }

        {
            let conn = mem.conn.lock();
            conn.execute(
                "UPDATE sessions SET last_activity = ?1 WHERE token_hash = ?2",
                params!["2026-03-29T00:00:00Z", "token-a"],
            )
            .unwrap();
        }

        let rows = mem
            .list_session_rows_for_scope("token-a", MAX_LIST_LIMIT + 500, 0)
            .await
            .unwrap();

        let ids: Vec<_> = rows.into_iter().map(|row| row.id).collect();
        assert_eq!(ids, vec!["c", "b", "a"]);
    }

    #[tokio::test]
    async fn get_resumable_session_for_scope_enforces_target_visibility() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("session-a", Some("token-a"))
            .await
            .unwrap();
        mem.upsert_session("session-b", Some("token-b"))
            .await
            .unwrap();

        for (session_id, preview) in [("session-a", "Preview A"), ("session-b", "Preview B")] {
            let snapshot = mem
                .create_session_snapshot(
                    session_id,
                    SessionSnapshotKind::Compact,
                    serde_json::json!({
                        "preview": preview,
                        "summary": preview,
                        "resume_context": preview,
                    }),
                    true,
                )
                .await
                .unwrap();
            mem.apply_session_state_patch(SessionStatePatch {
                session_id: session_id.into(),
                lifecycle: Some(SlashSessionLifecycle::Suspended),
                latest_tldr_snapshot_id: SessionFieldPatch::Keep,
                latest_compact_snapshot_id: SessionFieldPatch::Set(snapshot.id),
                pending_hydration_snapshot_id: SessionFieldPatch::Clear,
                suspended_at: SessionFieldPatch::Set("now".into()),
            })
            .await
            .unwrap();
        }

        let owned = mem
            .get_resumable_session_for_scope("session-a", "token-a")
            .await
            .unwrap();
        assert_eq!(
            owned.as_ref().map(|entry| entry.session_id.as_str()),
            Some("session-a")
        );

        let denied = mem
            .get_resumable_session_for_scope("session-b", "token-a")
            .await
            .unwrap();
        assert!(denied.is_none());
    }

    // ── FTS5 sync trigger tests ──────────────────────────────────

    #[tokio::test]
    async fn fts5_syncs_on_insert() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "test_key",
            "unique_searchterm_xyz",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();

        let conn = mem.conn.lock();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM memories_fts WHERE memories_fts MATCH '\"unique_searchterm_xyz\"'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn fts5_syncs_on_delete() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "del_key",
            "deletable_content_abc",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        mem.forget("del_key").await.unwrap();

        let conn = mem.conn.lock();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM memories_fts WHERE memories_fts MATCH '\"deletable_content_abc\"'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 0);
    }

    #[tokio::test]
    async fn fts5_syncs_on_update() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "upd_key",
            "original_content_111",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        mem.store("upd_key", "updated_content_222", MemoryCategory::Core, None)
            .await
            .unwrap();

        let conn = mem.conn.lock();
        // Old content should not be findable
        let old: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM memories_fts WHERE memories_fts MATCH '\"original_content_111\"'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(old, 0);

        // New content should be findable
        let new: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM memories_fts WHERE memories_fts MATCH '\"updated_content_222\"'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(new, 1);
    }

    // ── Open timeout tests ────────────────────────────────────────

    #[test]
    fn open_with_timeout_succeeds_when_fast() {
        let tmp = TempDir::new().unwrap();
        let embedder = Arc::new(super::super::embeddings::NoopEmbedding);
        let mem = SqliteMemory::with_embedder(tmp.path(), embedder, 0.7, 0.3, 1000, Some(5));
        assert!(
            mem.is_ok(),
            "open with 5s timeout should succeed on fast path"
        );
        assert_eq!(mem.unwrap().name(), "sqlite");
    }

    #[tokio::test]
    async fn open_with_timeout_store_recall_unchanged() {
        let tmp = TempDir::new().unwrap();
        let mem = SqliteMemory::with_embedder(
            tmp.path(),
            Arc::new(super::super::embeddings::NoopEmbedding),
            0.7,
            0.3,
            1000,
            Some(2),
        )
        .unwrap();
        mem.store(
            "timeout_key",
            "value with timeout",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        let entry = mem.get("timeout_key").await.unwrap().unwrap();
        assert_eq!(entry.content, "value with timeout");
    }

    // ── With-embedder constructor test ───────────────────────────

    #[test]
    fn with_embedder_noop() {
        let tmp = TempDir::new().unwrap();
        let embedder = Arc::new(super::super::embeddings::NoopEmbedding);
        let mem = SqliteMemory::with_embedder(tmp.path(), embedder, 0.7, 0.3, 1000, None);
        assert!(mem.is_ok());
        assert_eq!(mem.unwrap().name(), "sqlite");
    }

    // ── Reindex test ─────────────────────────────────────────────

    #[tokio::test]
    async fn reindex_rebuilds_fts() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("r1", "reindex test alpha", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("r2", "reindex test beta", MemoryCategory::Core, None)
            .await
            .unwrap();

        // Reindex should succeed (noop embedder → 0 re-embedded)
        let count = mem.reindex().await.unwrap();
        assert_eq!(count, 0);

        // FTS should still work after rebuild
        let results = mem.recall("reindex", 10, None).await.unwrap();
        assert_eq!(results.len(), 2);
    }

    // ── Recall limit test ────────────────────────────────────────

    #[tokio::test]
    async fn recall_respects_limit() {
        let (_tmp, mem) = temp_sqlite();
        for i in 0..20 {
            mem.store(
                &format!("k{i}"),
                &format!("common keyword item {i}"),
                MemoryCategory::Core,
                None,
            )
            .await
            .unwrap();
        }

        let results = mem.recall("common keyword", 5, None).await.unwrap();
        assert!(results.len() <= 5);
    }

    // ── Score presence test ──────────────────────────────────────

    #[tokio::test]
    async fn recall_results_have_scores() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("s1", "scored result test", MemoryCategory::Core, None)
            .await
            .unwrap();

        let results = mem.recall("scored", 10, None).await.unwrap();
        assert!(!results.is_empty());
        for r in &results {
            assert!(r.score.is_some(), "Expected score on result: {:?}", r.key);
        }
    }

    // ── Edge cases: FTS5 special characters ──────────────────────

    #[tokio::test]
    async fn recall_with_quotes_in_query() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("q1", "He said hello world", MemoryCategory::Core, None)
            .await
            .unwrap();
        // Quotes in query should not crash FTS5
        let results = mem.recall("\"hello\"", 10, None).await.unwrap();
        // May or may not match depending on FTS5 escaping, but must not error
        assert!(results.len() <= 10);
    }

    #[tokio::test]
    async fn recall_with_asterisk_in_query() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a1", "wildcard test content", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("wild*", 10, None).await.unwrap();
        assert!(results.len() <= 10);
    }

    #[tokio::test]
    async fn recall_with_parentheses_in_query() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("p1", "function call test", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("function()", 10, None).await.unwrap();
        assert!(results.len() <= 10);
    }

    #[tokio::test]
    async fn recall_with_sql_injection_attempt() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("safe", "normal content", MemoryCategory::Core, None)
            .await
            .unwrap();
        // Should not crash or leak data
        let results = mem
            .recall("'; DROP TABLE memories; --", 10, None)
            .await
            .unwrap();
        assert!(results.len() <= 10);
        // Table should still exist
        assert_eq!(mem.count().await.unwrap(), 1);
    }

    // ── Edge cases: store ────────────────────────────────────────

    #[tokio::test]
    async fn store_empty_content() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("empty", "", MemoryCategory::Core, None)
            .await
            .unwrap();
        let entry = mem.get("empty").await.unwrap().unwrap();
        assert_eq!(entry.content, "");
    }

    #[tokio::test]
    async fn store_empty_key() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("", "content for empty key", MemoryCategory::Core, None)
            .await
            .unwrap();
        let entry = mem.get("").await.unwrap().unwrap();
        assert_eq!(entry.content, "content for empty key");
    }

    #[tokio::test]
    async fn store_very_long_content() {
        let (_tmp, mem) = temp_sqlite();
        let long_content = "x".repeat(100_000);
        mem.store("long", &long_content, MemoryCategory::Core, None)
            .await
            .unwrap();
        let entry = mem.get("long").await.unwrap().unwrap();
        assert_eq!(entry.content.len(), 100_000);
    }

    #[tokio::test]
    async fn store_unicode_and_emoji() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "emoji_key_🦀",
            "こんにちは 🚀 Ñoño",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        let entry = mem.get("emoji_key_🦀").await.unwrap().unwrap();
        assert_eq!(entry.content, "こんにちは 🚀 Ñoño");
    }

    #[tokio::test]
    async fn store_content_with_newlines_and_tabs() {
        let (_tmp, mem) = temp_sqlite();
        let content = "line1\nline2\ttab\rcarriage\n\nnewparagraph";
        mem.store("whitespace", content, MemoryCategory::Core, None)
            .await
            .unwrap();
        let entry = mem.get("whitespace").await.unwrap().unwrap();
        assert_eq!(entry.content, content);
    }

    // ── Edge cases: recall ───────────────────────────────────────

    #[tokio::test]
    async fn recall_single_character_query() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "x marks the spot", MemoryCategory::Core, None)
            .await
            .unwrap();
        // Single char may not match FTS5 but LIKE fallback should work
        let results = mem.recall("x", 10, None).await.unwrap();
        // Should not crash; may or may not find results
        assert!(results.len() <= 10);
    }

    #[tokio::test]
    async fn recall_limit_zero() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "some content", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("some", 0, None).await.unwrap();
        assert!(results.is_empty());
    }

    #[tokio::test]
    async fn recall_limit_one() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "matching content alpha", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("b", "matching content beta", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("matching content", 1, None).await.unwrap();
        assert_eq!(results.len(), 1);
    }

    #[tokio::test]
    async fn recall_matches_by_key_not_just_content() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "rust_preferences",
            "User likes systems programming",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        // "rust" appears in key but not content — LIKE fallback checks key too
        let results = mem.recall("rust", 10, None).await.unwrap();
        assert!(!results.is_empty(), "Should match by key");
    }

    #[tokio::test]
    async fn recall_unicode_query() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("jp", "日本語のテスト", MemoryCategory::Core, None)
            .await
            .unwrap();
        let results = mem.recall("日本語", 10, None).await.unwrap();
        assert!(!results.is_empty());
    }

    // ── Edge cases: schema idempotency ───────────────────────────

    #[tokio::test]
    async fn schema_idempotent_reopen() {
        let tmp = TempDir::new().unwrap();
        {
            let mem = SqliteMemory::new(tmp.path()).unwrap();
            mem.store("k1", "v1", MemoryCategory::Core, None)
                .await
                .unwrap();
        }
        // Open again — init_schema runs again on existing DB
        let mem2 = SqliteMemory::new(tmp.path()).unwrap();
        let entry = mem2.get("k1").await.unwrap();
        assert!(entry.is_some());
        assert_eq!(entry.unwrap().content, "v1");
        // Store more data — should work fine
        mem2.store("k2", "v2", MemoryCategory::Daily, None)
            .await
            .unwrap();
        assert_eq!(mem2.count().await.unwrap(), 2);
    }

    #[tokio::test]
    async fn schema_triple_open() {
        let tmp = TempDir::new().unwrap();
        let _m1 = SqliteMemory::new(tmp.path()).unwrap();
        let _m2 = SqliteMemory::new(tmp.path()).unwrap();
        let m3 = SqliteMemory::new(tmp.path()).unwrap();
        assert!(m3.health_check().await);
    }

    // ── Edge cases: forget + FTS5 consistency ────────────────────

    #[tokio::test]
    async fn forget_then_recall_no_ghost_results() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "ghost",
            "phantom memory content",
            MemoryCategory::Core,
            None,
        )
        .await
        .unwrap();
        mem.forget("ghost").await.unwrap();
        let results = mem.recall("phantom memory", 10, None).await.unwrap();
        assert!(
            results.is_empty(),
            "Deleted memory should not appear in recall"
        );
    }

    #[tokio::test]
    async fn forget_and_re_store_same_key() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("cycle", "version 1", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.forget("cycle").await.unwrap();
        mem.store("cycle", "version 2", MemoryCategory::Core, None)
            .await
            .unwrap();
        let entry = mem.get("cycle").await.unwrap().unwrap();
        assert_eq!(entry.content, "version 2");
        assert_eq!(mem.count().await.unwrap(), 1);
    }

    // ── Edge cases: reindex ──────────────────────────────────────

    #[tokio::test]
    async fn reindex_empty_db() {
        let (_tmp, mem) = temp_sqlite();
        let count = mem.reindex().await.unwrap();
        assert_eq!(count, 0);
    }

    #[tokio::test]
    async fn reindex_twice_is_safe() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("r1", "reindex data", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.reindex().await.unwrap();
        let count = mem.reindex().await.unwrap();
        assert_eq!(count, 0); // Noop embedder → nothing to re-embed
                              // Data should still be intact
        let results = mem.recall("reindex", 10, None).await.unwrap();
        assert_eq!(results.len(), 1);
    }

    // ── Edge cases: content_hash ─────────────────────────────────

    #[test]
    fn content_hash_empty_string() {
        let h = SqliteMemory::content_hash("");
        assert!(!h.is_empty());
        assert_eq!(h.len(), 16); // 16 hex chars
    }

    #[test]
    fn content_hash_unicode() {
        let h1 = SqliteMemory::content_hash("🦀");
        let h2 = SqliteMemory::content_hash("🦀");
        assert_eq!(h1, h2);
        let h3 = SqliteMemory::content_hash("🚀");
        assert_ne!(h1, h3);
    }

    #[test]
    fn content_hash_long_input() {
        let long = "a".repeat(1_000_000);
        let h = SqliteMemory::content_hash(&long);
        assert_eq!(h.len(), 16);
    }

    // ── Edge cases: category helpers ─────────────────────────────

    #[test]
    fn category_roundtrip_custom_with_spaces() {
        let cat = MemoryCategory::Custom("my custom category".into());
        let s = SqliteMemory::category_to_str(&cat);
        assert_eq!(s, "my custom category");
        let back = SqliteMemory::str_to_category(&s);
        assert_eq!(back, cat);
    }

    #[test]
    fn category_roundtrip_empty_custom() {
        let cat = MemoryCategory::Custom(String::new());
        let s = SqliteMemory::category_to_str(&cat);
        assert_eq!(s, "");
        let back = SqliteMemory::str_to_category(&s);
        assert_eq!(back, MemoryCategory::Custom(String::new()));
    }

    // ── Edge cases: list ─────────────────────────────────────────

    #[tokio::test]
    async fn list_custom_category() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "c1",
            "custom1",
            MemoryCategory::Custom("project".into()),
            None,
        )
        .await
        .unwrap();
        mem.store(
            "c2",
            "custom2",
            MemoryCategory::Custom("project".into()),
            None,
        )
        .await
        .unwrap();
        mem.store("c3", "other", MemoryCategory::Core, None)
            .await
            .unwrap();

        let project = mem
            .list(Some(&MemoryCategory::Custom("project".into())), None)
            .await
            .unwrap();
        assert_eq!(project.len(), 2);
    }

    #[tokio::test]
    async fn list_empty_db() {
        let (_tmp, mem) = temp_sqlite();
        let all = mem.list(None, None).await.unwrap();
        assert!(all.is_empty());
    }

    // ── Session isolation ─────────────────────────────────────────

    #[tokio::test]
    async fn store_and_recall_with_session_id() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("k1", "session A fact", MemoryCategory::Core, Some("sess-a"))
            .await
            .unwrap();
        mem.store("k2", "session B fact", MemoryCategory::Core, Some("sess-b"))
            .await
            .unwrap();
        mem.store("k3", "no session fact", MemoryCategory::Core, None)
            .await
            .unwrap();

        // Recall with session-a filter returns only session-a entry
        let results = mem.recall("fact", 10, Some("sess-a")).await.unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].key, "k1");
        assert_eq!(results[0].session_id.as_deref(), Some("sess-a"));
    }

    #[tokio::test]
    async fn recall_no_session_filter_returns_all() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("k1", "alpha fact", MemoryCategory::Core, Some("sess-a"))
            .await
            .unwrap();
        mem.store("k2", "beta fact", MemoryCategory::Core, Some("sess-b"))
            .await
            .unwrap();
        mem.store("k3", "gamma fact", MemoryCategory::Core, None)
            .await
            .unwrap();

        // Recall without session filter returns all matching entries
        let results = mem.recall("fact", 10, None).await.unwrap();
        assert_eq!(results.len(), 3);
    }

    #[tokio::test]
    async fn cross_session_recall_isolation() {
        let (_tmp, mem) = temp_sqlite();
        mem.store(
            "secret",
            "session A secret data",
            MemoryCategory::Core,
            Some("sess-a"),
        )
        .await
        .unwrap();

        // Session B cannot see session A data
        let results = mem.recall("secret", 10, Some("sess-b")).await.unwrap();
        assert!(results.is_empty());

        // Session A can see its own data
        let results = mem.recall("secret", 10, Some("sess-a")).await.unwrap();
        assert_eq!(results.len(), 1);
    }

    #[tokio::test]
    async fn list_with_session_filter() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("k1", "a1", MemoryCategory::Core, Some("sess-a"))
            .await
            .unwrap();
        mem.store("k2", "a2", MemoryCategory::Conversation, Some("sess-a"))
            .await
            .unwrap();
        mem.store("k3", "b1", MemoryCategory::Core, Some("sess-b"))
            .await
            .unwrap();
        mem.store("k4", "none1", MemoryCategory::Core, None)
            .await
            .unwrap();

        // List with session-a filter
        let results = mem.list(None, Some("sess-a")).await.unwrap();
        assert_eq!(results.len(), 2);
        assert!(results
            .iter()
            .all(|e| e.session_id.as_deref() == Some("sess-a")));

        // List with session-a + category filter
        let results = mem
            .list(Some(&MemoryCategory::Core), Some("sess-a"))
            .await
            .unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].key, "k1");
    }

    #[tokio::test]
    async fn schema_migration_idempotent_on_reopen() {
        let tmp = TempDir::new().unwrap();

        // First open: creates schema + migration
        {
            let mem = SqliteMemory::new(tmp.path()).unwrap();
            mem.store("k1", "before reopen", MemoryCategory::Core, Some("sess-x"))
                .await
                .unwrap();
        }

        // Second open: migration runs again but is idempotent
        {
            let mem = SqliteMemory::new(tmp.path()).unwrap();
            let results = mem.recall("reopen", 10, Some("sess-x")).await.unwrap();
            assert_eq!(results.len(), 1);
            assert_eq!(results[0].key, "k1");
            assert_eq!(results[0].session_id.as_deref(), Some("sess-x"));
        }
    }

    // ── §4.1 Concurrent write contention tests ──────────────

    #[tokio::test]
    async fn sqlite_concurrent_writes_no_data_loss() {
        let (_tmp, mem) = temp_sqlite();
        let mem = std::sync::Arc::new(mem);

        let mut handles = Vec::new();
        for i in 0..10 {
            let mem = std::sync::Arc::clone(&mem);
            handles.push(tokio::spawn(async move {
                mem.store(
                    &format!("concurrent_key_{i}"),
                    &format!("value_{i}"),
                    MemoryCategory::Core,
                    None,
                )
                .await
                .unwrap();
            }));
        }

        for handle in handles {
            handle.await.unwrap();
        }

        let count = mem.count().await.unwrap();
        assert_eq!(
            count, 10,
            "all 10 concurrent writes must succeed without data loss"
        );
    }

    #[tokio::test]
    async fn sqlite_concurrent_read_write_no_panic() {
        let (_tmp, mem) = temp_sqlite();
        let mem = std::sync::Arc::new(mem);

        // Pre-populate
        mem.store("shared_key", "initial", MemoryCategory::Core, None)
            .await
            .unwrap();

        let mut handles = Vec::new();

        // Concurrent reads
        for _ in 0..5 {
            let mem = std::sync::Arc::clone(&mem);
            handles.push(tokio::spawn(async move {
                let _ = mem.get("shared_key").await.unwrap();
            }));
        }

        // Concurrent writes
        for i in 0..5 {
            let mem = std::sync::Arc::clone(&mem);
            handles.push(tokio::spawn(async move {
                mem.store(
                    &format!("key_{i}"),
                    &format!("val_{i}"),
                    MemoryCategory::Core,
                    None,
                )
                .await
                .unwrap();
            }));
        }

        for handle in handles {
            handle.await.unwrap();
        }

        // Should have 6 total entries (1 pre-existing + 5 new)
        assert_eq!(mem.count().await.unwrap(), 6);
    }

    // ── §4.2 Reindex / corruption recovery tests ────────────

    #[tokio::test]
    async fn sqlite_reindex_preserves_data() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("a", "Rust is fast", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("b", "Python is interpreted", MemoryCategory::Core, None)
            .await
            .unwrap();

        mem.reindex().await.unwrap();

        let count = mem.count().await.unwrap();
        assert_eq!(count, 2, "reindex must preserve all entries");

        let entry = mem.get("a").await.unwrap();
        assert!(entry.is_some());
        assert_eq!(entry.unwrap().content, "Rust is fast");
    }

    #[tokio::test]
    async fn sqlite_reindex_idempotent() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("x", "test data", MemoryCategory::Core, None)
            .await
            .unwrap();

        // Multiple reindex calls should be safe
        mem.reindex().await.unwrap();
        mem.reindex().await.unwrap();
        mem.reindex().await.unwrap();

        assert_eq!(mem.count().await.unwrap(), 1);
    }

    // ── Session CRUD tests ───────────────────────────────────────

    #[tokio::test]
    async fn test_sessions_table_migration() {
        let tmp = TempDir::new().unwrap();
        let _mem = SqliteMemory::new(tmp.path()).unwrap();
        // Reopen — migration must be idempotent
        let mem2 = SqliteMemory::new(tmp.path()).unwrap();
        let conn = mem2.conn.lock();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='sessions'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn test_upsert_session_new() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-new", None).await.unwrap();

        let session = mem.get_session("sess-new").await.unwrap();
        assert!(session.is_some());
        let s = session.unwrap();
        assert_eq!(s.id, "sess-new");
        assert_eq!(s.status, SessionStatus::Active);
        assert_eq!(s.message_count, 0);
        assert!(s.ended_at.is_none());
    }

    #[tokio::test]
    async fn test_upsert_session_existing() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-dup", None).await.unwrap();
        let first = mem.get_session("sess-dup").await.unwrap().unwrap();
        let first_started = first.started_at.clone();

        // Small delay then upsert again
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        mem.upsert_session("sess-dup", None).await.unwrap();

        let second = mem.get_session("sess-dup").await.unwrap().unwrap();
        // started_at must not change
        assert_eq!(second.started_at, first_started);
        // message_count must not change
        assert_eq!(second.message_count, 0);
    }

    #[tokio::test]
    async fn test_update_session_activity() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-act", None).await.unwrap();

        mem.update_session_activity("sess-act", None).await.unwrap();
        mem.update_session_activity("sess-act", None).await.unwrap();

        let s = mem.get_session("sess-act").await.unwrap().unwrap();
        assert_eq!(s.message_count, 2);
    }

    #[tokio::test]
    async fn test_end_session() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-end", None).await.unwrap();

        mem.end_session("sess-end").await.unwrap();
        let s = mem.get_session("sess-end").await.unwrap().unwrap();
        assert_eq!(s.status, SessionStatus::Ended);
        assert!(s.ended_at.is_some());
    }

    #[tokio::test]
    async fn test_end_session_idempotent() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-idem", None).await.unwrap();

        mem.end_session("sess-idem").await.unwrap();
        let first_ended = mem
            .get_session("sess-idem")
            .await
            .unwrap()
            .unwrap()
            .ended_at
            .clone();

        // End again — should not change ended_at
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        mem.end_session("sess-idem").await.unwrap();
        let second_ended = mem
            .get_session("sess-idem")
            .await
            .unwrap()
            .unwrap()
            .ended_at;

        assert_eq!(first_ended, second_ended);
    }

    #[tokio::test]
    async fn test_list_sessions_pagination() {
        let (_tmp, mem) = temp_sqlite();
        for i in 0..10 {
            mem.upsert_session(&format!("sess-page-{i}"), None)
                .await
                .unwrap();
        }

        let (sessions, total) = mem
            .list_sessions(None, 3, 0, "last_activity", "desc")
            .await
            .unwrap();
        assert_eq!(sessions.len(), 3);
        assert_eq!(total, 10);

        let (sessions, _) = mem
            .list_sessions(None, 3, 9, "last_activity", "desc")
            .await
            .unwrap();
        assert_eq!(sessions.len(), 1);
    }

    #[tokio::test]
    async fn test_list_sessions_status_filter() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("active-1", None).await.unwrap();
        mem.upsert_session("active-2", None).await.unwrap();
        mem.upsert_session("ended-1", None).await.unwrap();
        mem.end_session("ended-1").await.unwrap();

        let (active, _) = mem
            .list_sessions(Some(SessionStatus::Active), 50, 0, "last_activity", "desc")
            .await
            .unwrap();
        assert_eq!(active.len(), 2);

        let (ended, _) = mem
            .list_sessions(Some(SessionStatus::Ended), 50, 0, "last_activity", "desc")
            .await
            .unwrap();
        assert_eq!(ended.len(), 1);
    }

    #[tokio::test]
    async fn test_get_session_found() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("found-me", None).await.unwrap();

        let s = mem.get_session("found-me").await.unwrap();
        assert!(s.is_some());
        assert_eq!(s.unwrap().id, "found-me");
    }

    #[tokio::test]
    async fn test_get_session_not_found() {
        let (_tmp, mem) = temp_sqlite();
        let s = mem.get_session("nonexistent").await.unwrap();
        assert!(s.is_none());
    }

    #[tokio::test]
    async fn test_list_sessions_for_token() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("tok-a-1", Some("hash_a")).await.unwrap();
        mem.upsert_session("tok-a-2", Some("hash_a")).await.unwrap();
        mem.upsert_session("tok-b-1", Some("hash_b")).await.unwrap();

        let (a_sessions, a_total) = mem.list_sessions_for_token("hash_a", 50, 0).await.unwrap();
        assert_eq!(a_sessions.len(), 2);
        assert_eq!(a_total, 2);

        let (b_sessions, b_total) = mem.list_sessions_for_token("hash_b", 50, 0).await.unwrap();
        assert_eq!(b_sessions.len(), 1);
        assert_eq!(b_total, 1);

        // No cross-token leakage
        let (c_sessions, c_total) = mem.list_sessions_for_token("hash_c", 50, 0).await.unwrap();
        assert!(c_sessions.is_empty());
        assert_eq!(c_total, 0);
    }

    #[tokio::test]
    async fn test_memory_stats() {
        let (_tmp, mem) = temp_sqlite();
        mem.store("k1", "core1", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("k2", "core2", MemoryCategory::Core, None)
            .await
            .unwrap();
        mem.store("k3", "daily1", MemoryCategory::Daily, None)
            .await
            .unwrap();
        mem.upsert_session("s1", None).await.unwrap();
        mem.upsert_session("s2", None).await.unwrap();
        mem.end_session("s2").await.unwrap();

        let stats = mem.memory_stats().await.unwrap();
        assert_eq!(stats.total_entries, 3);
        assert_eq!(stats.by_category.get("core"), Some(&2));
        assert_eq!(stats.by_category.get("daily"), Some(&1));
        assert_eq!(stats.total_sessions, 2);
        assert_eq!(stats.active_sessions, 1);
        assert_eq!(stats.backend, "sqlite");
    }

    #[tokio::test]
    async fn test_activity_update_on_ended_session_is_noop() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-ended-act", None).await.unwrap();
        mem.update_session_activity("sess-ended-act", None)
            .await
            .unwrap();
        mem.end_session("sess-ended-act").await.unwrap();

        // Activity update after end should be a no-op
        mem.update_session_activity("sess-ended-act", None)
            .await
            .unwrap();
        let s = mem.get_session("sess-ended-act").await.unwrap().unwrap();
        assert_eq!(s.message_count, 1); // Only the one before end
    }

    #[tokio::test]
    async fn test_upsert_session_rejects_token_ownership_mismatch() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-owned", Some("hash-a"))
            .await
            .unwrap();

        let err = mem
            .upsert_session("sess-owned", Some("hash-b"))
            .await
            .unwrap_err();
        assert!(err.to_string().contains("ownership mismatch"));
    }

    #[tokio::test]
    async fn test_update_session_activity_respects_token_ownership() {
        let (_tmp, mem) = temp_sqlite();
        mem.upsert_session("sess-owned", Some("hash-a"))
            .await
            .unwrap();

        mem.update_session_activity("sess-owned", Some("hash-b"))
            .await
            .unwrap();

        let s = mem.get_session("sess-owned").await.unwrap().unwrap();
        assert_eq!(s.message_count, 0);
    }

    #[tokio::test]
    async fn test_list_sessions_uses_stable_secondary_sort_and_capped_limit() {
        let (_tmp, mem) = temp_sqlite();
        for session_id in ["b", "a", "c"] {
            mem.upsert_session(session_id, None).await.unwrap();
        }

        {
            let conn = mem.conn.lock();
            conn.execute(
                "UPDATE sessions SET last_activity = ?1",
                params!["2026-03-29T00:00:00Z"],
            )
            .unwrap();
        }

        let (sessions, total) = mem
            .list_sessions(None, MAX_LIST_LIMIT + 500, 0, "last_activity", "asc")
            .await
            .unwrap();

        assert_eq!(total, 3);
        assert_eq!(sessions.len(), 3);
        let ids: Vec<_> = sessions.into_iter().map(|session| session.id).collect();
        assert_eq!(ids, vec!["a", "b", "c"]);
    }

    #[tokio::test]
    async fn test_list_sessions_for_token_uses_stable_secondary_sort_and_capped_limit() {
        let (_tmp, mem) = temp_sqlite();
        for session_id in ["b", "a", "c"] {
            mem.upsert_session(session_id, Some("hash-a"))
                .await
                .unwrap();
        }

        {
            let conn = mem.conn.lock();
            conn.execute(
                "UPDATE sessions SET last_activity = ?1 WHERE token_hash = ?2",
                params!["2026-03-29T00:00:00Z", "hash-a"],
            )
            .unwrap();
        }

        let (sessions, total) = mem
            .list_sessions_for_token("hash-a", MAX_LIST_LIMIT + 500, 0)
            .await
            .unwrap();

        assert_eq!(total, 3);
        assert_eq!(sessions.len(), 3);
        let ids: Vec<_> = sessions.into_iter().map(|session| session.id).collect();
        assert_eq!(ids, vec!["c", "b", "a"]);
    }

    #[tokio::test]
    async fn clear_patch_actually_clears_previously_set_field() {
        let (_temp, mem) = temp_sqlite();

        mem.upsert_session("session-1", None).await.unwrap();

        let snapshot = mem
            .create_session_snapshot(
                "session-1",
                SessionSnapshotKind::Compact,
                serde_json::json!({"summary": "test"}),
                true,
            )
            .await
            .unwrap();

        mem.apply_session_state_patch(SessionStatePatch {
            session_id: "session-1".to_string(),
            lifecycle: None,
            latest_tldr_snapshot_id: SessionFieldPatch::Set(snapshot.id.clone()),
            latest_compact_snapshot_id: SessionFieldPatch::Keep,
            pending_hydration_snapshot_id: SessionFieldPatch::Keep,
            suspended_at: SessionFieldPatch::Clear,
        })
        .await
        .unwrap();

        let state = mem
            .get_session_state_record("session-1")
            .await
            .unwrap()
            .unwrap();
        assert!(state.latest_tldr_snapshot_id.is_some());

        mem.apply_session_state_patch(SessionStatePatch {
            session_id: "session-1".to_string(),
            lifecycle: None,
            latest_tldr_snapshot_id: SessionFieldPatch::Clear,
            latest_compact_snapshot_id: SessionFieldPatch::Keep,
            pending_hydration_snapshot_id: SessionFieldPatch::Keep,
            suspended_at: SessionFieldPatch::Clear,
        })
        .await
        .unwrap();

        let state = mem
            .get_session_state_record("session-1")
            .await
            .unwrap()
            .unwrap();
        assert!(
            state.latest_tldr_snapshot_id.is_none(),
            "Clear should have set latest_tldr_snapshot_id to NULL"
        );
    }
}
