goosedump 0.10.0

// SPDX-License-Identifier: LGPL-2.1-or-later
// Copyright (C) Jarkko Sakkinen 2026

//! Persistent, content-addressed conversation memory.

use std::collections::{HashMap, HashSet};
use std::env;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::{Context as _, bail};
use rusqlite::{
    Connection, OptionalExtension as _, Transaction, ffi::sqlite3_auto_extension, params,
};
use serde::Serialize;
use sha2::{Digest as _, Sha256};
use sqlite_vec::sqlite3_vec_init;
use zerocopy::IntoBytes as _;

use crate::Client;
use crate::display;
use crate::index::IndexEntry;
use crate::message::{Context, ConversationMessage, MessageView};
use crate::model::{EMBEDDING_MODEL_ID, Embedder, Mutator};

const SCHEMA_VERSION: i64 = 5;
const HASH_VERSION: u8 = 1;
const EMBEDDING_DIMENSIONS: usize = 384;
const ARCHIVE_BINS: i64 = 16;
const RRF_K: f64 = 60.0;
const PLATEAU_WINDOWS: usize = 3;
const PLATEAU_CONTEXTS_PER_WINDOW: u64 = 10;
const PLATEAU_SEARCH_HITS_PER_WINDOW: u64 = 20;
const PLATEAU_QUALITY_DELTA: f64 = 0.01;
const PLATEAU_EXPANSION_RATE_DELTA: f64 = 0.01;
static VEC_REGISTRATION: OnceLock<i32> = OnceLock::new();

/// Optional constraints for a memory recall.
#[derive(Debug, Clone, Default)]
pub struct RecallFilter {
    /// Restrict results to sightings harvested from this provider.
    pub provider: Option<Client>,
    /// Restrict results to this exact session working directory.
    pub path: Option<PathBuf>,
}

/// One content occurrence returned by [`Memory::recall`].
#[derive(Debug, Clone, Serialize)]
pub struct RecallHit {
    /// Row in `search_hits`, used to mark a later expansion.
    pub search_hit_id: i64,
    /// Versioned content hash shared by identical messages.
    pub hash: String,
    /// Collapsed message kind included in the content hash.
    pub kind: String,
    /// Sanitized searchable message text.
    pub text: String,
    /// Positive BM25 or reciprocal-rank-fusion score; larger values are better.
    pub score: f64,
    /// Provider containing this occurrence.
    pub provider: Client,
    /// Provider-native context identifier.
    pub context_id: String,
    /// Entry identifier within the context.
    pub entry_id: String,
    /// Zero-based message position at harvest time.
    pub ordinal: usize,
    /// Working directory associated with the context.
    pub path: PathBuf,
    /// Original context source path from the index.
    pub source_path: PathBuf,
}

/// Counts returned after harvesting one context.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct HarvestReport {
    /// Number of messages visited, including messages with empty text.
    pub messages: usize,
    /// Number of distinct content hashes in this context.
    pub unique_entries: usize,
}

/// Current persistent-memory row counts.
#[derive(Debug, Clone, Copy, PartialEq, Serialize)]
pub struct MemoryStats {
    pub entries: u64,
    pub sightings: u64,
    pub contexts: u64,
    pub searches: u64,
    pub search_hits: u64,
    pub expansions: u64,
    pub embedded: u64,
    pub covered: u64,
    pub archive_entries: u64,
    pub mutations: u64,
    pub stage2: Stage2Status,
}

/// Result of a single Stage-3 archive mutation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct MutationReport {
    pub hash: String,
    pub text: String,
    pub inserted: bool,
}

/// Evidence and result of the Stage-2 consolidation plateau check.
#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize)]
pub struct Stage2Status {
    /// Number of completed, comparable observation windows.
    pub windows: u64,
    /// Harvested contexts across the compared observation windows.
    pub contexts: u64,
    /// Recall hits across the compared observation windows.
    pub search_hits: u64,
    /// Fraction of those recall hits which were explicitly expanded.
    pub expansion_rate: f64,
    /// Current number of occupied MAP-Elites cells.
    pub archive_entries: u64,
    /// Current aggregate archive quality divided by occupied cells.
    pub quality_per_entry: f64,
    /// True only after three sufficiently large windows have stayed stable.
    pub plateaued: bool,
}

/// Rows removed by a forget operation.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
pub struct ForgetReport {
    pub entries: u64,
    pub sightings: u64,
}

/// SQLite-backed persistent memory store.
pub struct Memory {
    conn: Connection,
}

impl Memory {
    /// Open the default state database and initialize it if needed.
    ///
    /// `GOOSEDUMP_STATE_DIR`, when non-empty, takes precedence over the
    /// platform state directory.
    ///
    /// # Errors
    /// Returns an error when no state directory is available or the database
    /// cannot be created, configured, or migrated.
    pub fn open() -> anyhow::Result<Self> {
        Self::open_path(&database_path()?)
    }

    /// Open or create a store at an explicit path.
    ///
    /// This is primarily useful for tests and callers which manage their own
    /// state root.
    ///
    /// # Errors
    /// Returns an error when the parent directory or database cannot be
    /// created, configured, or migrated.
    pub fn open_path(path: &Path) -> anyhow::Result<Self> {
        register_vec()?;
        if let Some(parent) = path.parent()
            && !parent.as_os_str().is_empty()
        {
            fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
        }
        let conn = Connection::open(path).with_context(|| format!("open {}", path.display()))?;
        conn.busy_timeout(Duration::from_secs(5))?;
        conn.pragma_update(None, "foreign_keys", true)?;
        conn.pragma_update(None, "journal_mode", "WAL")?;
        initialize(&conn)?;
        Ok(Self { conn })
    }

    /// Embed and index every entry which does not yet have a vector.
    ///
    /// # Errors
    /// Returns an error if embedding fails or an index row cannot be written.
    pub fn index_missing_embeddings(&mut self, embedder: &Embedder) -> anyhow::Result<usize> {
        let entries = missing_embedding_entries(&self.conn)?;
        if entries.is_empty() {
            return Ok(0);
        }
        let texts = entries
            .iter()
            .map(|(_, text)| text.clone())
            .collect::<Vec<_>>();
        let embeddings = embedder.embed(&texts)?;
        ensure_embedding_batch(&embeddings, entries.len())?;

        let tx = self.conn.transaction()?;
        for ((hash, _), embedding) in entries.iter().zip(&embeddings) {
            insert_embedding(&tx, hash, embedding)?;
        }
        tx.commit()?;
        Ok(entries.len())
    }

    /// Return whether any memory entries have been harvested.
    ///
    /// # Errors
    /// Returns an error if the entry count cannot be read.
    pub fn has_entries(&self) -> anyhow::Result<bool> {
        Ok(self
            .conn
            .query_row("SELECT EXISTS(SELECT 1 FROM entries)", [], |row| {
                row.get::<_, bool>(0)
            })?)
    }

    /// Attribute a summary to current entries and retain their maximum coverage.
    ///
    /// This also fills missing vector-index rows while the current entry texts
    /// are already being embedded.
    ///
    /// # Errors
    /// Returns an error if embedding fails or coverage cannot be persisted.
    pub fn attribute_summary(
        &mut self,
        summary: &str,
        context: &Context,
        embedder: &Embedder,
    ) -> anyhow::Result<usize> {
        let mut seen = HashSet::new();
        let entries = context
            .messages
            .iter()
            .filter_map(|message| {
                let hash = content_hash(message);
                seen.insert(hash.clone())
                    .then(|| (hash, display::searchable_text(message)))
            })
            .collect::<Vec<_>>();
        if entries.is_empty() {
            return Ok(0);
        }
        let mut texts = Vec::with_capacity(entries.len() + 1);
        texts.push(summary.to_string());
        texts.extend(entries.iter().map(|(_, text)| text.clone()));
        let embeddings = embedder.embed(&texts)?;
        ensure_embedding_batch(&embeddings, texts.len())?;
        let summary_embedding = &embeddings[0];

        let tx = self.conn.transaction()?;
        let mut changed = 0;
        for ((hash, _), embedding) in entries.iter().zip(&embeddings[1..]) {
            tx.execute(
                "INSERT OR IGNORE INTO entries_vec(hash, embedding) VALUES(?1, ?2)",
                params![hash, embedding.as_slice().as_bytes()],
            )?;
            let coverage = cosine_similarity(summary_embedding, embedding).clamp(0.0, 1.0);
            changed += tx.execute(
                "UPDATE entries SET coverage = ?2 WHERE hash = ?1 AND coverage < ?2",
                params![hash, coverage],
            )?;
        }
        tx.commit()?;
        Ok(changed)
    }

    /// Harvest every message in `context` as a sighting of immutable content.
    ///
    /// Reharvesting the same range is idempotent. Entry ids, rather than local
    /// range ordinals, identify sightings because each compacted range starts
    /// its ordinal at zero.
    ///
    /// # Errors
    /// Returns an error if the harvest transaction cannot be completed.
    pub fn harvest(
        &mut self,
        index_entry: &IndexEntry,
        context: &Context,
    ) -> anyhow::Result<HarvestReport> {
        let provider = index_entry.provider.as_str();
        let context_id = &index_entry.id;
        let path = index_entry.provider_id.cwd.to_string_lossy();
        let source_path = index_entry.path.to_string_lossy();
        let tx = self.conn.transaction()?;
        let mut hashes = HashSet::new();

        for (ordinal, message) in context.messages.iter().enumerate() {
            let kind = collapsed_kind(message);
            let text = display::searchable_text(message);
            let hash = hash_content(&kind, &text);
            hashes.insert(hash.clone());
            upsert_entry(&tx, &hash, &kind, &text)?;
            upsert_sighting(
                &tx,
                SightingInput {
                    hash: &hash,
                    provider,
                    context_id,
                    entry_id: &message.entry_id,
                    ordinal,
                    path: &path,
                    source_path: &source_path,
                    observed_at: message
                        .timestamp
                        .map_or_else(now_millis, |timestamp| timestamp.timestamp_millis()),
                },
            )?;
        }

        tx.commit()?;

        Ok(HarvestReport {
            messages: context.messages.len(),
            unique_entries: hashes.len(),
        })
    }

    /// Log the ranked messages returned by an in-context `search` command.
    ///
    /// # Errors
    /// Returns an error if the search audit transaction cannot be committed.
    pub fn log_context_search(
        &mut self,
        query: &str,
        index_entry: &IndexEntry,
        context: &Context,
        entry_ids: &[String],
    ) -> anyhow::Result<()> {
        let tx = self.conn.transaction()?;
        let provider = index_entry.provider.as_str();
        let path = index_entry.provider_id.cwd.to_string_lossy();
        tx.execute(
            "INSERT INTO searches(query, provider, path, created_at)
             VALUES(?1, ?2, ?3, ?4)",
            params![query, provider, path.as_ref(), now_millis()],
        )?;
        let search_id = tx.last_insert_rowid();
        for (rank, entry_id) in entry_ids.iter().enumerate() {
            let Some(message) = context
                .messages
                .iter()
                .find(|message| message.entry_id == *entry_id)
            else {
                continue;
            };
            tx.execute(
                "INSERT INTO search_hits(
                    search_id, entry_hash, provider, context_id, entry_id,
                    path, rank, score
                 ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, 0.0)",
                params![
                    search_id,
                    content_hash(message),
                    provider,
                    index_entry.id,
                    entry_id,
                    path.as_ref(),
                    i64::try_from(rank).context("search rank")?,
                ],
            )?;
        }
        tx.commit()?;
        Ok(())
    }

    /// Credit the latest unexpanded search hit for each explicitly shown entry.
    ///
    /// # Errors
    /// Returns an error if the search audit rows cannot be updated.
    pub fn mark_context_expanded(
        &mut self,
        provider: Client,
        context_id: &str,
        entry_ids: &[String],
    ) -> anyhow::Result<u64> {
        let tx = self.conn.transaction()?;
        let mut changed = 0_u64;
        for entry_id in entry_ids {
            let updated = tx.execute(
                "UPDATE search_hits SET expanded_at = ?1
                 WHERE id = (
                    SELECT h.id
                    FROM search_hits AS h
                    JOIN searches AS q ON q.id = h.search_id
                    WHERE h.provider = ?2 AND h.context_id = ?3
                      AND h.entry_id = ?4 AND h.expanded_at IS NULL
                    ORDER BY q.created_at DESC, h.id DESC
                    LIMIT 1
                 )",
                params![now_millis(), provider.as_str(), context_id, entry_id],
            )?;
            changed = changed.saturating_add(u64::try_from(updated).context("expanded count")?);
        }
        tx.commit()?;
        Ok(changed)
    }

    /// Recall sightings using deterministic reciprocal-rank fusion of BM25 and
    /// cosine-nearest-neighbor rankings, and log the returned hits.
    ///
    /// Provider and path filters are applied to both rankings. The shadow
    /// MAP-Elites archive is deliberately not used for retrieval.
    ///
    /// # Errors
    /// Returns an error if embedding, querying, or telemetry persistence fails.
    pub fn recall_hybrid(
        &mut self,
        query: &str,
        filter: &RecallFilter,
        limit: usize,
        embedder: &Embedder,
    ) -> anyhow::Result<Vec<RecallHit>> {
        let provider = filter.provider.map(Client::as_str);
        let path = filter
            .path
            .as_ref()
            .map(|value| value.to_string_lossy().into_owned());
        let query_embedding = if query.trim().is_empty() || limit == 0 {
            None
        } else {
            self.index_missing_embeddings(embedder)?;
            let mut embeddings = embedder.embed(&[query.to_string()])?;
            ensure_embedding_batch(&embeddings, 1)?;
            embeddings.pop()
        };

        let tx = self.conn.transaction()?;
        tx.execute(
            "INSERT INTO searches(query, provider, path, created_at)
             VALUES(?1, ?2, ?3, ?4)",
            params![query, provider, path, now_millis()],
        )?;
        let search_id = tx.last_insert_rowid();
        let Some(query_embedding) = query_embedding else {
            tx.commit()?;
            return Ok(Vec::new());
        };

        let candidate_limit = limit.saturating_mul(4).max(limit);
        let sql_candidate_limit =
            i64::try_from(candidate_limit).context("hybrid candidate limit")?;
        let fts_query = fts_query(query);
        let lexical = if fts_query.is_empty() {
            Vec::new()
        } else {
            query_recall(
                &tx,
                &fts_query,
                provider,
                path.as_deref(),
                sql_candidate_limit,
            )?
        };
        let semantic = query_vector_recall(
            &tx,
            &query_embedding,
            provider,
            path.as_deref(),
            sql_candidate_limit,
        )?;
        let rows = reciprocal_rank_fusion(lexical, semantic, limit);

        let mut hits = Vec::with_capacity(rows.len());
        for (rank, row) in rows.into_iter().enumerate() {
            let rank = i64::try_from(rank).context("hybrid recall rank")?;
            tx.execute(
                "INSERT INTO search_hits(
                    search_id, sighting_id, entry_hash, provider, context_id,
                    entry_id, path, rank, score
                 ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
                params![
                    search_id,
                    row.sighting_id,
                    row.hash,
                    row.provider.as_str(),
                    row.context_id,
                    row.entry_id,
                    row.path.to_string_lossy(),
                    rank,
                    row.score,
                ],
            )?;
            let search_hit_id = tx.last_insert_rowid();
            hits.push(RecallHit {
                search_hit_id,
                hash: row.hash,
                kind: row.kind,
                text: row.text,
                score: row.score,
                provider: row.provider,
                context_id: row.context_id,
                entry_id: row.entry_id,
                ordinal: row.ordinal,
                path: row.path,
                source_path: row.source_path,
            });
        }
        tx.commit()?;
        Ok(hits)
    }

    /// Deterministically rebuild the shadow MAP-Elites archive.
    ///
    /// Cells are keyed by message kind and bins of the first two normalized
    /// embedding dimensions. This method does not alter recall behavior.
    ///
    /// # Errors
    /// Returns an error if archive candidates cannot be read or persisted.
    pub fn rebuild_shadow_archive(&mut self) -> anyhow::Result<usize> {
        let rebuilt_at = now_millis();
        let candidates = archive_candidates(&self.conn, rebuilt_at)?;
        let mut elites: HashMap<(String, i64, i64), ArchiveCandidate> = HashMap::new();
        for candidate in candidates {
            let key = (candidate.kind.clone(), candidate.x_bin, candidate.y_bin);
            match elites.get(&key) {
                Some(elite) if !candidate.is_better_than(elite) => {}
                _ => {
                    elites.insert(key, candidate);
                }
            }
        }

        let mut elites = elites.into_values().collect::<Vec<_>>();
        elites.sort_by(|left, right| {
            left.kind
                .cmp(&right.kind)
                .then(left.x_bin.cmp(&right.x_bin))
                .then(left.y_bin.cmp(&right.y_bin))
        });
        let tx = self.conn.transaction()?;
        tx.execute("DELETE FROM memory_archive", [])?;
        for elite in &elites {
            tx.execute(
                "INSERT INTO memory_archive(
                    kind, x_bin, y_bin, entry_hash, quality, recurrence,
                    confirmed_expands, coverage, recency, rebuilt_at
                 ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
                params![
                    elite.kind,
                    elite.x_bin,
                    elite.y_bin,
                    elite.hash,
                    elite.quality,
                    elite.recurrence,
                    elite.confirmed_expands,
                    elite.coverage,
                    elite.recency,
                    rebuilt_at,
                ],
            )?;
        }
        tx.commit()?;
        self.record_stage2_snapshot()?;
        Ok(elites.len())
    }

    /// Merge the two highest-quality archive entries with the local mutator.
    ///
    /// The generated entry inherits source provenance so it can be recalled and
    /// expanded through the original context.
    pub fn mutate_archive(
        &mut self,
        mut mutator: Mutator,
        embedder: &Embedder,
    ) -> anyhow::Result<MutationReport> {
        let parents = mutation_parents(&self.conn)?;
        let [left, right] = parents.as_slice() else {
            bail!("Stage 3 needs at least two archive entries");
        };
        let generated = mutator.merge(&left.text, &right.text)?;
        let text = sanitize_mutation(&generated)?;
        let hash = hash_content("mutation", &text);
        let embedding = embedder.embed(std::slice::from_ref(&text))?;
        ensure_embedding_batch(&embedding, 1)?;

        let tx = self.conn.transaction()?;
        let inserted = upsert_entry(&tx, &hash, "mutation", &text)? != 0;
        if inserted {
            insert_embedding(&tx, &hash, &embedding[0])?;
            tx.execute(
                "INSERT INTO memory_mutations(hash, source_hash, source_sighting_id)
                 VALUES(?1, ?2, ?3)",
                params![hash, left.hash, left.sighting_id],
            )?;
            upsert_sighting(
                &tx,
                SightingInput {
                    hash: &hash,
                    provider: &left.provider,
                    context_id: &left.context_id,
                    entry_id: &format!("mutation:{hash}"),
                    ordinal: left.ordinal,
                    path: &left.path,
                    source_path: &left.source_path,
                    observed_at: now_millis(),
                },
            )?;
        }
        tx.commit()?;
        self.rebuild_shadow_archive()?;
        Ok(MutationReport {
            hash,
            text,
            inserted,
        })
    }

    fn record_stage2_snapshot(&mut self) -> anyhow::Result<()> {
        let current = Stage2Snapshot::from_connection(&self.conn)?;
        let previous = latest_stage2_snapshot(&self.conn)?;
        let Some(previous) = previous else {
            if current.contexts < PLATEAU_CONTEXTS_PER_WINDOW
                || current.search_hits < PLATEAU_SEARCH_HITS_PER_WINDOW
            {
                return Ok(());
            }
            insert_stage2_snapshot(&self.conn, current)?;
            return Ok(());
        };
        if current.contexts.saturating_sub(previous.contexts) < PLATEAU_CONTEXTS_PER_WINDOW
            || current.search_hits.saturating_sub(previous.search_hits)
                < PLATEAU_SEARCH_HITS_PER_WINDOW
        {
            return Ok(());
        }
        insert_stage2_snapshot(&self.conn, current)
    }

    /// Return current content, occurrence, context, and search counts.
    ///
    /// # Errors
    /// Returns an error if any count cannot be read.
    pub fn stats(&self) -> anyhow::Result<MemoryStats> {
        Ok(MemoryStats {
            entries: count(&self.conn, "SELECT count(*) FROM entries")?,
            sightings: count(&self.conn, "SELECT count(*) FROM sightings")?,
            contexts: count(
                &self.conn,
                "SELECT count(*) FROM (
                    SELECT provider, context_id FROM sightings
                    GROUP BY provider, context_id
                 )",
            )?,
            searches: count(&self.conn, "SELECT count(*) FROM searches")?,
            search_hits: count(&self.conn, "SELECT count(*) FROM search_hits")?,
            expansions: count(
                &self.conn,
                "SELECT count(*) FROM search_hits WHERE expanded_at IS NOT NULL",
            )?,
            embedded: count(&self.conn, "SELECT count(*) FROM entries_vec")?,
            covered: count(
                &self.conn,
                "SELECT count(*) FROM entries WHERE coverage > 0.0",
            )?,
            archive_entries: count(&self.conn, "SELECT count(*) FROM memory_archive")?,
            mutations: count(&self.conn, "SELECT count(*) FROM memory_mutations")?,
            stage2: self.stage2_status()?,
        })
    }

    /// Return Stage-2 consolidation evidence from completed observation windows.
    ///
    /// A window is recorded after ten more harvested contexts and twenty more
    /// recall hits. A plateau requires three such windows with stable archive
    /// occupancy, archive quality per entry, and confirmed-expand rate.
    pub fn stage2_status(&self) -> anyhow::Result<Stage2Status> {
        let snapshots = stage2_snapshots(&self.conn, PLATEAU_WINDOWS + 1)?;
        let Some(current) = snapshots.last().copied() else {
            return Ok(Stage2Status::default());
        };
        let windows = snapshots.len().saturating_sub(1);
        let baseline = snapshots.first().copied().expect("non-empty snapshots");
        let plateaued = windows == PLATEAU_WINDOWS
            && snapshots
                .windows(2)
                .all(|pair| pair[0].archive_entries == pair[1].archive_entries)
            && {
                let baseline_quality = baseline.quality_per_entry();
                let current_quality = current.quality_per_entry();
                if baseline_quality.abs() < f64::EPSILON {
                    current_quality.abs()
                } else {
                    ((baseline_quality - current_quality) / baseline_quality).abs()
                }
            } <= PLATEAU_QUALITY_DELTA
            && (baseline.expansion_rate() - current.expansion_rate()).abs()
                <= PLATEAU_EXPANSION_RATE_DELTA;
        Ok(Stage2Status {
            windows: u64::try_from(windows).context("stage-2 window count")?,
            contexts: current.contexts.saturating_sub(baseline.contexts),
            search_hits: current.search_hits.saturating_sub(baseline.search_hits),
            expansion_rate: current.expansion_rate(),
            archive_entries: current.archive_entries,
            quality_per_entry: current.quality_per_entry(),
            plateaued,
        })
    }

    /// Forget content and every sighting of it, while retaining search history.
    ///
    /// # Errors
    /// Returns an error if the delete transaction cannot be completed.
    pub fn forget_hash(&mut self, hash: &str) -> anyhow::Result<ForgetReport> {
        let tx = self.conn.transaction()?;
        let hashes = {
            let mut stmt = tx.prepare(
                "WITH RECURSIVE derived(hash) AS (
                    SELECT ?1
                    UNION
                    SELECT m.hash FROM memory_mutations AS m
                    JOIN derived AS d ON m.source_hash = d.hash
                 )
                 SELECT hash FROM derived",
            )?;
            stmt.query_map(params![hash], |row| row.get::<_, String>(0))?
                .collect::<rusqlite::Result<Vec<_>>>()?
        };
        let mut sightings = 0_u64;
        let mut entries = 0_u64;
        for hash in hashes {
            let count = tx.query_row(
                "SELECT count(*) FROM sightings WHERE entry_hash = ?1",
                params![hash],
                |row| row.get::<_, i64>(0),
            )?;
            let count = u64::try_from(count).context("negative database count")?;
            sightings = sightings.saturating_add(count);
            tx.execute("DELETE FROM entries_vec WHERE hash = ?1", params![hash])?;
            tx.execute("DELETE FROM entries_fts WHERE hash = ?1", params![hash])?;
            entries = entries.saturating_add(u64::try_from(
                tx.execute("DELETE FROM entries WHERE hash = ?1", params![hash])?,
            )?);
        }
        tx.commit()?;
        Ok(ForgetReport { entries, sightings })
    }

    /// Forget one provider context and prune content with no other sightings.
    ///
    /// Shared entries remain available through sightings in other contexts.
    /// Search audit rows retain snapshots of their original context hit.
    ///
    /// # Errors
    /// Returns an error if the delete transaction cannot be completed.
    pub fn forget_context(
        &mut self,
        provider: Client,
        context_id: &str,
    ) -> anyhow::Result<ForgetReport> {
        let tx = self.conn.transaction()?;
        let before = count_tx(&tx, "SELECT count(*) FROM entries")?;
        let sightings = u64::try_from(tx.execute(
            "DELETE FROM sightings WHERE provider = ?1 AND context_id = ?2",
            params![provider.as_str(), context_id],
        )?)
        .context("deleted sighting count")?;
        prune_orphans(&tx)?;
        let after = count_tx(&tx, "SELECT count(*) FROM entries")?;
        tx.commit()?;
        Ok(ForgetReport {
            entries: before.saturating_sub(after),
            sightings,
        })
    }
}

/// Return the default database path without opening it.
///
/// # Errors
/// Returns an error when `GOOSEDUMP_STATE_DIR` is unset or empty and the
/// platform has no state directory.
pub fn database_path() -> anyhow::Result<PathBuf> {
    if let Some(root) = env::var_os("GOOSEDUMP_STATE_DIR").filter(|value| !value.is_empty()) {
        return Ok(PathBuf::from(root).join("goosedump.db"));
    }
    let root = dirs::state_dir().context("state directory not found")?;
    Ok(root.join("goosedump").join("goosedump.db"))
}

/// Compute the stable, versioned content hash used by the memory store.
#[must_use]
pub fn content_hash(message: &ConversationMessage) -> String {
    hash_content(&collapsed_kind(message), &display::searchable_text(message))
}

#[derive(Clone, Copy)]
struct SightingInput<'a> {
    hash: &'a str,
    provider: &'a str,
    context_id: &'a str,
    entry_id: &'a str,
    ordinal: usize,
    path: &'a str,
    source_path: &'a str,
    observed_at: i64,
}

struct RecallRow {
    sighting_id: i64,
    hash: String,
    kind: String,
    text: String,
    score: f64,
    provider: Client,
    context_id: String,
    entry_id: String,
    ordinal: usize,
    path: PathBuf,
    source_path: PathBuf,
}

struct ArchiveCandidate {
    hash: String,
    kind: String,
    x_bin: i64,
    y_bin: i64,
    quality: f64,
    recurrence: i64,
    confirmed_expands: i64,
    coverage: f64,
    recency: f64,
    last_seen_at: i64,
}

struct MutationParent {
    hash: String,
    text: String,
    sighting_id: i64,
    provider: String,
    context_id: String,
    ordinal: usize,
    path: String,
    source_path: String,
}

#[derive(Clone, Copy)]
struct Stage2Snapshot {
    contexts: u64,
    search_hits: u64,
    expansions: u64,
    archive_entries: u64,
    archive_quality: f64,
}

impl Stage2Snapshot {
    fn from_connection(conn: &Connection) -> anyhow::Result<Self> {
        Ok(Self {
            contexts: count(
                conn,
                "SELECT count(*) FROM (
                    SELECT provider, context_id FROM sightings
                    GROUP BY provider, context_id
                 )",
            )?,
            search_hits: count(conn, "SELECT count(*) FROM search_hits")?,
            expansions: count(
                conn,
                "SELECT count(*) FROM search_hits WHERE expanded_at IS NOT NULL",
            )?,
            archive_entries: count(conn, "SELECT count(*) FROM memory_archive")?,
            archive_quality: conn.query_row(
                "SELECT coalesce(sum(quality), 0.0) FROM memory_archive",
                [],
                |row| row.get(0),
            )?,
        })
    }

    fn expansion_rate(self) -> f64 {
        if self.search_hits == 0 {
            0.0
        } else {
            f64::from(u32::try_from(self.expansions).unwrap_or(u32::MAX))
                / f64::from(u32::try_from(self.search_hits).unwrap_or(u32::MAX))
        }
    }

    fn quality_per_entry(self) -> f64 {
        if self.archive_entries == 0 {
            0.0
        } else {
            self.archive_quality
                / f64::from(u32::try_from(self.archive_entries).unwrap_or(u32::MAX))
        }
    }
}

impl ArchiveCandidate {
    fn is_better_than(&self, other: &Self) -> bool {
        self.quality
            .total_cmp(&other.quality)
            .then(self.last_seen_at.cmp(&other.last_seen_at))
            .then_with(|| other.hash.cmp(&self.hash))
            .is_gt()
    }
}

fn register_vec() -> anyhow::Result<()> {
    let result = *VEC_REGISTRATION.get_or_init(|| {
        // sqlite3_auto_extension requires SQLite's extension entry-point shape.
        unsafe {
            sqlite3_auto_extension(Some(std::mem::transmute::<
                *const (),
                unsafe extern "C" fn(
                    *mut rusqlite::ffi::sqlite3,
                    *mut *const std::ffi::c_char,
                    *const rusqlite::ffi::sqlite3_api_routines,
                ) -> std::ffi::c_int,
            >(sqlite3_vec_init as *const ())))
        }
    });
    if result != rusqlite::ffi::SQLITE_OK {
        bail!("register sqlite-vec extension: SQLite error {result}");
    }
    Ok(())
}

fn initialize(conn: &Connection) -> anyhow::Result<()> {
    let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
    if version > SCHEMA_VERSION {
        bail!("memory database schema {version} is newer than supported {SCHEMA_VERSION}");
    }
    if version == 0 {
        create_initial_schema(conn)?;
    }
    if version <= 1 {
        let tx = conn.unchecked_transaction()?;
        tx.execute_batch(
            "ALTER TABLE entries
                ADD COLUMN coverage REAL NOT NULL DEFAULT 0.0
                CHECK(coverage >= 0.0 AND coverage <= 1.0);

             CREATE VIRTUAL TABLE entries_vec USING vec0(
                hash TEXT PRIMARY KEY,
                embedding FLOAT[384] DISTANCE_METRIC=cosine
             );

             CREATE TABLE memory_archive(
                kind TEXT NOT NULL,
                x_bin INTEGER NOT NULL CHECK(x_bin >= 0 AND x_bin < 16),
                y_bin INTEGER NOT NULL CHECK(y_bin >= 0 AND y_bin < 16),
                entry_hash TEXT NOT NULL REFERENCES entries(hash) ON DELETE CASCADE,
                quality REAL NOT NULL,
                recurrence INTEGER NOT NULL,
                confirmed_expands INTEGER NOT NULL,
                coverage REAL NOT NULL,
                recency REAL NOT NULL,
                rebuilt_at INTEGER NOT NULL,
                PRIMARY KEY(kind, x_bin, y_bin)
             ) WITHOUT ROWID;
             CREATE INDEX memory_archive_entry_hash ON memory_archive(entry_hash);

             PRAGMA user_version = 2;",
        )?;
        tx.commit()?;
    }
    if version <= 2 {
        migrate_embedding_metadata(conn)?;
    }
    if version <= 3 {
        migrate_stage2_snapshots(conn)?;
    }
    if version <= 4 {
        migrate_mutations(conn)?;
    }
    reset_stale_embeddings(conn)?;
    Ok(())
}

fn create_initial_schema(conn: &Connection) -> anyhow::Result<()> {
    let tx = conn.unchecked_transaction()?;
    tx.execute_batch(
        "CREATE TABLE entries(
        hash TEXT PRIMARY KEY,
        hash_version INTEGER NOT NULL,
        kind TEXT NOT NULL,
        text TEXT NOT NULL,
        created_at INTEGER NOT NULL,
        last_seen_at INTEGER NOT NULL
     ) WITHOUT ROWID;

     CREATE VIRTUAL TABLE entries_fts USING fts5(hash UNINDEXED, kind, text);

     CREATE TABLE sightings(
        id INTEGER PRIMARY KEY,
        entry_hash TEXT NOT NULL REFERENCES entries(hash) ON DELETE CASCADE,
        provider TEXT NOT NULL,
        context_id TEXT NOT NULL,
        entry_id TEXT NOT NULL,
        ordinal INTEGER NOT NULL CHECK(ordinal >= 0),
        path TEXT NOT NULL,
        source_path TEXT NOT NULL,
        observed_at INTEGER NOT NULL,
        harvested_at INTEGER NOT NULL,
        UNIQUE(provider, context_id, entry_id)
     );
     CREATE INDEX sightings_entry_hash ON sightings(entry_hash);
     CREATE INDEX sightings_context ON sightings(provider, context_id);
     CREATE INDEX sightings_path ON sightings(path);

     CREATE TABLE searches(
        id INTEGER PRIMARY KEY,
        query TEXT NOT NULL,
        provider TEXT,
        path TEXT,
        created_at INTEGER NOT NULL
     );

     CREATE TABLE search_hits(
        id INTEGER PRIMARY KEY,
        search_id INTEGER NOT NULL REFERENCES searches(id) ON DELETE CASCADE,
        sighting_id INTEGER REFERENCES sightings(id) ON DELETE SET NULL,
        entry_hash TEXT NOT NULL,
        provider TEXT NOT NULL,
        context_id TEXT NOT NULL,
        entry_id TEXT NOT NULL,
        path TEXT NOT NULL,
        rank INTEGER NOT NULL,
        score REAL NOT NULL,
        expanded_at INTEGER
     );
     CREATE INDEX search_hits_search ON search_hits(search_id, rank);
     CREATE INDEX search_hits_context ON search_hits(provider, context_id);

     PRAGMA user_version = 1;",
    )?;
    tx.commit()?;
    Ok(())
}

fn migrate_embedding_metadata(conn: &Connection) -> anyhow::Result<()> {
    let tx = conn.unchecked_transaction()?;
    tx.execute_batch(
        "CREATE TABLE memory_meta(
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
         ) WITHOUT ROWID;
         PRAGMA user_version = 3;",
    )?;
    tx.commit()?;
    Ok(())
}

fn migrate_stage2_snapshots(conn: &Connection) -> anyhow::Result<()> {
    let tx = conn.unchecked_transaction()?;
    tx.execute_batch(
        "CREATE TABLE stage2_snapshots(
            id INTEGER PRIMARY KEY,
            contexts INTEGER NOT NULL,
            search_hits INTEGER NOT NULL,
            expansions INTEGER NOT NULL,
            archive_entries INTEGER NOT NULL,
            archive_quality REAL NOT NULL,
            created_at INTEGER NOT NULL
         );
         PRAGMA user_version = 4;",
    )?;
    tx.commit()?;
    Ok(())
}

fn migrate_mutations(conn: &Connection) -> anyhow::Result<()> {
    let tx = conn.unchecked_transaction()?;
    tx.execute_batch(
        "CREATE TABLE memory_mutations(
            hash TEXT PRIMARY KEY REFERENCES entries(hash) ON DELETE CASCADE,
            source_hash TEXT NOT NULL REFERENCES entries(hash) ON DELETE CASCADE,
            source_sighting_id INTEGER NOT NULL REFERENCES sightings(id) ON DELETE CASCADE
         ) WITHOUT ROWID;
         PRAGMA user_version = 5;",
    )?;
    tx.commit()?;
    Ok(())
}

fn reset_stale_embeddings(conn: &Connection) -> anyhow::Result<()> {
    let stored = conn
        .query_row(
            "SELECT value FROM memory_meta WHERE key = 'embedding_model'",
            [],
            |row| row.get::<_, String>(0),
        )
        .optional()?;
    if stored.as_deref() == Some(EMBEDDING_MODEL_ID) {
        return Ok(());
    }

    let tx = conn.unchecked_transaction()?;
    tx.execute("DELETE FROM memory_archive", [])?;
    tx.execute("DELETE FROM entries_vec", [])?;
    tx.execute("UPDATE entries SET coverage = 0.0", [])?;
    tx.execute(
        "INSERT INTO memory_meta(key, value) VALUES('embedding_model', ?1)
         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
        params![EMBEDDING_MODEL_ID],
    )?;
    tx.commit()?;
    Ok(())
}

fn collapsed_kind(message: &ConversationMessage) -> String {
    match message.view() {
        MessageView::Text { role, .. } => {
            if role.is_empty() {
                "unknown".to_string()
            } else {
                role.to_ascii_lowercase()
            }
        }
        MessageView::Assistant { .. } => "assistant".to_string(),
        MessageView::ToolResult(_) => "tool_result".to_string(),
        MessageView::Bash(_) => "bash".to_string(),
    }
}

fn hash_content(kind: &str, text: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(b"goosedump-memory\0");
    hasher.update([HASH_VERSION]);
    hasher.update(b"\0");
    hasher.update(kind.as_bytes());
    hasher.update(b"\0");
    hasher.update(text.as_bytes());
    format!("v{HASH_VERSION}:{:x}", hasher.finalize())
}

fn upsert_entry(tx: &Transaction<'_>, hash: &str, kind: &str, text: &str) -> anyhow::Result<u64> {
    let now = now_millis();
    let inserted = tx.execute(
        "INSERT INTO entries(hash, hash_version, kind, text, created_at, last_seen_at)
         VALUES(?1, ?2, ?3, ?4, ?5, ?5)
         ON CONFLICT(hash) DO NOTHING",
        params![hash, HASH_VERSION, kind, text, now],
    )?;
    if inserted == 0 {
        tx.execute(
            "UPDATE entries SET last_seen_at = ?2 WHERE hash = ?1",
            params![hash, now],
        )?;
    } else {
        tx.execute(
            "INSERT INTO entries_fts(hash, kind, text) VALUES(?1, ?2, ?3)",
            params![hash, kind, text],
        )?;
    }
    u64::try_from(inserted).context("inserted entry count")
}

fn upsert_sighting(tx: &Transaction<'_>, input: SightingInput<'_>) -> anyhow::Result<()> {
    let ordinal = i64::try_from(input.ordinal).context("message ordinal")?;
    tx.execute(
        "INSERT INTO sightings(
            entry_hash, provider, context_id, entry_id, ordinal, path,
            source_path, observed_at, harvested_at
         ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
         ON CONFLICT(provider, context_id, entry_id) DO UPDATE SET
            entry_hash = excluded.entry_hash,
            ordinal = excluded.ordinal,
            path = excluded.path,
            source_path = excluded.source_path,
            observed_at = excluded.observed_at,
            harvested_at = excluded.harvested_at",
        params![
            input.hash,
            input.provider,
            input.context_id,
            input.entry_id,
            ordinal,
            input.path,
            input.source_path,
            input.observed_at,
            now_millis(),
        ],
    )?;
    Ok(())
}

fn prune_orphans(tx: &Transaction<'_>) -> anyhow::Result<()> {
    let hashes = {
        let mut stmt = tx.prepare(
            "SELECT hash FROM entries
             WHERE NOT EXISTS(
                SELECT 1 FROM sightings WHERE entry_hash = entries.hash
             )",
        )?;
        stmt.query_map([], |row| row.get::<_, String>(0))?
            .collect::<rusqlite::Result<Vec<_>>>()?
    };
    for hash in hashes {
        tx.execute("DELETE FROM entries_vec WHERE hash = ?1", params![hash])?;
        tx.execute("DELETE FROM entries_fts WHERE hash = ?1", params![hash])?;
        tx.execute("DELETE FROM entries WHERE hash = ?1", params![hash])?;
    }
    Ok(())
}

fn query_recall(
    tx: &Transaction<'_>,
    query: &str,
    provider: Option<&str>,
    path: Option<&str>,
    limit: i64,
) -> anyhow::Result<Vec<RecallRow>> {
    let mut stmt = tx.prepare(
        "SELECT
            coalesce(source.id, s.id), f.hash, f.kind, f.text, -bm25(entries_fts, 0.0, 0.2, 1.0),
            coalesce(source.provider, s.provider), coalesce(source.context_id, s.context_id),
            coalesce(source.entry_id, s.entry_id), coalesce(source.ordinal, s.ordinal),
            coalesce(source.path, s.path), coalesce(source.source_path, s.source_path)
         FROM entries_fts AS f
         JOIN sightings AS s ON s.entry_hash = f.hash
         LEFT JOIN memory_mutations AS m ON m.hash = f.hash
         LEFT JOIN sightings AS source ON source.id = m.source_sighting_id
         WHERE entries_fts MATCH ?1
            AND (?2 IS NULL OR s.provider = ?2)
            AND (?3 IS NULL OR s.path = ?3)
            AND s.id = (
                SELECT latest.id FROM sightings AS latest
                WHERE latest.entry_hash = f.hash
                  AND (?2 IS NULL OR latest.provider = ?2)
                  AND (?3 IS NULL OR latest.path = ?3)
                ORDER BY latest.observed_at DESC, latest.id DESC
                LIMIT 1
            )
          ORDER BY bm25(entries_fts, 0.0, 0.2, 1.0), s.observed_at DESC, s.id
         LIMIT ?4",
    )?;
    let mapped = stmt.query_map(params![query, provider, path, limit], map_recall_row)?;
    mapped
        .collect::<rusqlite::Result<Vec<_>>>()
        .map_err(Into::into)
}

fn map_recall_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<RecallRow> {
    let provider_name = row.get::<_, String>(5)?;
    let provider = provider_name.parse::<Client>().map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(
            5,
            rusqlite::types::Type::Text,
            std::io::Error::other(error).into(),
        )
    })?;
    let ordinal = usize::try_from(row.get::<_, i64>(8)?).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(8, rusqlite::types::Type::Integer, error.into())
    })?;
    Ok(RecallRow {
        sighting_id: row.get(0)?,
        hash: row.get(1)?,
        kind: row.get(2)?,
        text: row.get(3)?,
        score: row.get(4)?,
        provider,
        context_id: row.get(6)?,
        entry_id: row.get(7)?,
        ordinal,
        path: PathBuf::from(row.get::<_, String>(9)?),
        source_path: PathBuf::from(row.get::<_, String>(10)?),
    })
}

fn query_vector_recall(
    tx: &Transaction<'_>,
    embedding: &[f32],
    provider: Option<&str>,
    path: Option<&str>,
    limit: i64,
) -> anyhow::Result<Vec<RecallRow>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let mut stmt = tx.prepare(
        "SELECT
            coalesce(source.id, s.id), e.hash, e.kind, e.text, 1.0 - nearest.distance,
            coalesce(source.provider, s.provider), coalesce(source.context_id, s.context_id),
            coalesce(source.entry_id, s.entry_id), coalesce(source.ordinal, s.ordinal),
            coalesce(source.path, s.path), coalesce(source.source_path, s.source_path)
         FROM (
            SELECT hash, distance FROM entries_vec
            WHERE embedding MATCH ?1 AND k = ?2
         ) AS nearest
         JOIN entries AS e ON e.hash = nearest.hash
         JOIN sightings AS s ON s.entry_hash = e.hash
         LEFT JOIN memory_mutations AS m ON m.hash = e.hash
         LEFT JOIN sightings AS source ON source.id = m.source_sighting_id
         WHERE s.id = (
            SELECT latest.id FROM sightings AS latest
            WHERE latest.entry_hash = e.hash
              AND (?3 IS NULL OR latest.provider = ?3)
              AND (?4 IS NULL OR latest.path = ?4)
            ORDER BY latest.observed_at DESC, latest.id DESC LIMIT 1
         )
         ORDER BY nearest.distance, e.hash",
    )?;
    stmt.query_map(
        params![embedding.as_bytes(), limit, provider, path],
        map_recall_row,
    )?
    .collect::<rusqlite::Result<Vec<_>>>()
    .map_err(Into::into)
}

fn reciprocal_rank_fusion(
    lexical: Vec<RecallRow>,
    semantic: Vec<RecallRow>,
    limit: usize,
) -> Vec<RecallRow> {
    struct Candidate {
        row: RecallRow,
        lexical_rank: Option<usize>,
        semantic_rank: Option<usize>,
    }

    let mut candidates = HashMap::new();
    for (rank, row) in lexical.into_iter().enumerate() {
        candidates.insert(
            row.hash.clone(),
            Candidate {
                row,
                lexical_rank: Some(rank),
                semantic_rank: None,
            },
        );
    }
    for (rank, row) in semantic.into_iter().enumerate() {
        candidates
            .entry(row.hash.clone())
            .and_modify(|candidate| candidate.semantic_rank = Some(rank))
            .or_insert(Candidate {
                row,
                lexical_rank: None,
                semantic_rank: Some(rank),
            });
    }
    let mut candidates = candidates.into_values().collect::<Vec<_>>();
    candidates.sort_by(|left, right| {
        let left_score = rrf_score(left.lexical_rank, left.semantic_rank);
        let right_score = rrf_score(right.lexical_rank, right.semantic_rank);
        right_score
            .total_cmp(&left_score)
            .then_with(|| {
                best_rank(left.lexical_rank, left.semantic_rank)
                    .cmp(&best_rank(right.lexical_rank, right.semantic_rank))
            })
            .then(left.row.hash.cmp(&right.row.hash))
            .then(left.row.sighting_id.cmp(&right.row.sighting_id))
    });
    candidates
        .into_iter()
        .take(limit)
        .map(|mut candidate| {
            candidate.row.score = rrf_score(candidate.lexical_rank, candidate.semantic_rank);
            candidate.row
        })
        .collect()
}

fn rrf_score(lexical_rank: Option<usize>, semantic_rank: Option<usize>) -> f64 {
    [lexical_rank, semantic_rank]
        .into_iter()
        .flatten()
        .map(|rank| {
            let rank = u32::try_from(rank).unwrap_or(u32::MAX);
            1.0 / (RRF_K + f64::from(rank) + 1.0)
        })
        .sum()
}

fn best_rank(lexical_rank: Option<usize>, semantic_rank: Option<usize>) -> usize {
    lexical_rank
        .into_iter()
        .chain(semantic_rank)
        .min()
        .unwrap_or(usize::MAX)
}

fn missing_embedding_entries(conn: &Connection) -> anyhow::Result<Vec<(String, String)>> {
    let mut stmt = conn.prepare(
        "SELECT hash, text FROM entries
         WHERE NOT EXISTS(SELECT 1 FROM entries_vec WHERE entries_vec.hash = entries.hash)
         ORDER BY hash",
    )?;
    let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?;
    rows.collect::<rusqlite::Result<Vec<_>>>()
        .map_err(Into::into)
}

fn ensure_embedding_batch(embeddings: &[Vec<f32>], expected: usize) -> anyhow::Result<()> {
    if embeddings.len() != expected {
        bail!(
            "embedder returned {} rows for {expected} texts",
            embeddings.len()
        );
    }
    for embedding in embeddings {
        if embedding.len() != EMBEDDING_DIMENSIONS {
            bail!(
                "embedder returned {} dimensions, expected {EMBEDDING_DIMENSIONS}",
                embedding.len()
            );
        }
        if embedding.iter().any(|value| !value.is_finite()) {
            bail!("embedder returned a non-finite value");
        }
    }
    Ok(())
}

fn insert_embedding(tx: &Transaction<'_>, hash: &str, embedding: &[f32]) -> anyhow::Result<()> {
    ensure_embedding_batch(&[embedding.to_vec()], 1)?;
    tx.execute(
        "INSERT OR IGNORE INTO entries_vec(hash, embedding) VALUES(?1, ?2)",
        params![hash, embedding.as_bytes()],
    )?;
    Ok(())
}

fn cosine_similarity(left: &[f32], right: &[f32]) -> f64 {
    let dot = left
        .iter()
        .zip(right)
        .map(|(left, right)| f64::from(*left) * f64::from(*right))
        .sum::<f64>();
    let left_norm = left
        .iter()
        .map(|value| f64::from(*value).powi(2))
        .sum::<f64>()
        .sqrt();
    let right_norm = right
        .iter()
        .map(|value| f64::from(*value).powi(2))
        .sum::<f64>()
        .sqrt();
    if left_norm == 0.0 || right_norm == 0.0 {
        0.0
    } else {
        dot / (left_norm * right_norm)
    }
}

fn archive_candidates(conn: &Connection, rebuilt_at: i64) -> anyhow::Result<Vec<ArchiveCandidate>> {
    let mut stmt = conn.prepare(
        "SELECT
            e.hash, e.kind, e.coverage, e.last_seen_at, v.embedding,
            (SELECT count(*) FROM (
                SELECT s.provider, s.context_id FROM sightings AS s
                WHERE s.entry_hash = e.hash
                GROUP BY s.provider, s.context_id
             )),
            (SELECT count(*) FROM search_hits AS h
             WHERE h.entry_hash = e.hash AND h.expanded_at IS NOT NULL)
         FROM entries AS e
         JOIN entries_vec AS v ON v.hash = e.hash
         ORDER BY e.hash",
    )?;
    let mapped = stmt.query_map([], |row| {
        let hash = row.get::<_, String>(0)?;
        let kind = row.get::<_, String>(1)?;
        let coverage = row.get::<_, f64>(2)?;
        let last_seen_at = row.get::<_, i64>(3)?;
        let bytes = row.get::<_, Vec<u8>>(4)?;
        let recurrence = row.get::<_, i64>(5)?;
        let confirmed_expands = row.get::<_, i64>(6)?;
        if bytes.len() != EMBEDDING_DIMENSIONS * size_of::<f32>() {
            return Err(rusqlite::Error::FromSqlConversionFailure(
                4,
                rusqlite::types::Type::Blob,
                std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid embedding size")
                    .into(),
            ));
        }
        let first = f32::from_ne_bytes(bytes[0..4].try_into().expect("four-byte slice"));
        let second = f32::from_ne_bytes(bytes[4..8].try_into().expect("four-byte slice"));
        let age_days = rebuilt_at.saturating_sub(last_seen_at) / 86_400_000;
        let age_days = f64::from(u32::try_from(age_days).unwrap_or(u32::MAX));
        let recency = 1.0 / (1.0 + age_days / 30.0);
        let recurrence_quality = f64::from(u32::try_from(recurrence).unwrap_or(u32::MAX));
        let expansion_quality = f64::from(u32::try_from(confirmed_expands).unwrap_or(u32::MAX));
        let quality =
            recurrence_quality.ln_1p() + 2.0 * expansion_quality.ln_1p() + 2.0 * coverage + recency;
        Ok(ArchiveCandidate {
            hash,
            kind,
            x_bin: archive_bin(first),
            y_bin: archive_bin(second),
            quality,
            recurrence,
            confirmed_expands,
            coverage,
            recency,
            last_seen_at,
        })
    })?;
    mapped
        .collect::<rusqlite::Result<Vec<_>>>()
        .map_err(Into::into)
}

fn mutation_parents(conn: &Connection) -> anyhow::Result<Vec<MutationParent>> {
    let mut stmt = conn.prepare(
        "SELECT e.hash, e.text, s.id, s.provider, s.context_id, s.ordinal, s.path, s.source_path
         FROM memory_archive AS a
         JOIN entries AS e ON e.hash = a.entry_hash
         JOIN sightings AS s ON s.entry_hash = e.hash
         WHERE s.id = (
             SELECT latest.id FROM sightings AS latest
             WHERE latest.entry_hash = e.hash
             ORDER BY latest.observed_at DESC, latest.id DESC LIMIT 1
         )
         ORDER BY a.quality DESC, a.entry_hash
         LIMIT 2",
    )?;
    let rows = stmt.query_map([], |row| {
        let ordinal = usize::try_from(row.get::<_, i64>(5)?).map_err(|error| {
            rusqlite::Error::FromSqlConversionFailure(
                5,
                rusqlite::types::Type::Integer,
                error.into(),
            )
        })?;
        Ok(MutationParent {
            hash: row.get(0)?,
            text: row.get(1)?,
            sighting_id: row.get(2)?,
            provider: row.get(3)?,
            context_id: row.get(4)?,
            ordinal,
            path: row.get(6)?,
            source_path: row.get(7)?,
        })
    })?;
    rows.collect::<rusqlite::Result<Vec<_>>>()
        .map_err(Into::into)
}

fn sanitize_mutation(text: &str) -> anyhow::Result<String> {
    let text = text.trim();
    if text.is_empty() || text.contains("<think>") || text.contains("</think>") {
        bail!("mutation model did not return a final statement");
    }
    if text.chars().count() > 800 {
        bail!("mutation model returned more than 800 characters");
    }
    Ok(text.to_string())
}

fn archive_bin(value: f32) -> i64 {
    const THRESHOLDS: [f32; 15] = [
        -0.875, -0.75, -0.625, -0.5, -0.375, -0.25, -0.125, 0.0, 0.125, 0.25, 0.375, 0.5, 0.625,
        0.75, 0.875,
    ];
    let value = value.clamp(-1.0, 1.0);
    let bin = THRESHOLDS.partition_point(|threshold| value >= *threshold);
    i64::try_from(bin).unwrap_or(ARCHIVE_BINS - 1)
}

fn fts_query(query: &str) -> String {
    query
        .split_whitespace()
        .filter(|term| !term.is_empty())
        .map(|term| format!("\"{}\"", term.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" OR ")
}

fn count(conn: &Connection, sql: &str) -> anyhow::Result<u64> {
    let value = conn.query_row(sql, [], |row| row.get::<_, i64>(0))?;
    u64::try_from(value).context("negative database count")
}

fn count_tx(tx: &Transaction<'_>, sql: &str) -> anyhow::Result<u64> {
    let value = tx.query_row(sql, [], |row| row.get::<_, i64>(0))?;
    u64::try_from(value).context("negative database count")
}

fn latest_stage2_snapshot(conn: &Connection) -> anyhow::Result<Option<Stage2Snapshot>> {
    conn.query_row(
        "SELECT contexts, search_hits, expansions, archive_entries, archive_quality
         FROM stage2_snapshots ORDER BY id DESC LIMIT 1",
        [],
        stage2_snapshot_from_row,
    )
    .optional()
    .map_err(Into::into)
}

fn stage2_snapshots(conn: &Connection, limit: usize) -> anyhow::Result<Vec<Stage2Snapshot>> {
    let limit = i64::try_from(limit).context("stage-2 snapshot limit")?;
    let mut stmt = conn.prepare(
        "SELECT contexts, search_hits, expansions, archive_entries, archive_quality
         FROM stage2_snapshots ORDER BY id DESC LIMIT ?1",
    )?;
    let mut snapshots = stmt
        .query_map(params![limit], stage2_snapshot_from_row)?
        .collect::<rusqlite::Result<Vec<_>>>()?;
    snapshots.reverse();
    Ok(snapshots)
}

fn stage2_snapshot_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Stage2Snapshot> {
    let contexts = u64::try_from(row.get::<_, i64>(0)?).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Integer, error.into())
    })?;
    let search_hits = u64::try_from(row.get::<_, i64>(1)?).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(1, rusqlite::types::Type::Integer, error.into())
    })?;
    let expansions = u64::try_from(row.get::<_, i64>(2)?).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(2, rusqlite::types::Type::Integer, error.into())
    })?;
    let archive_entries = u64::try_from(row.get::<_, i64>(3)?).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(3, rusqlite::types::Type::Integer, error.into())
    })?;
    Ok(Stage2Snapshot {
        contexts,
        search_hits,
        expansions,
        archive_entries,
        archive_quality: row.get(4)?,
    })
}

fn insert_stage2_snapshot(conn: &Connection, snapshot: Stage2Snapshot) -> anyhow::Result<()> {
    conn.execute(
        "INSERT INTO stage2_snapshots(
            contexts, search_hits, expansions, archive_entries, archive_quality, created_at
         ) VALUES(?1, ?2, ?3, ?4, ?5, ?6)",
        params![
            i64::try_from(snapshot.contexts).context("stage-2 context count")?,
            i64::try_from(snapshot.search_hits).context("stage-2 search hit count")?,
            i64::try_from(snapshot.expansions).context("stage-2 expansion count")?,
            i64::try_from(snapshot.archive_entries).context("stage-2 archive entry count")?,
            snapshot.archive_quality,
            now_millis(),
        ],
    )?;
    Ok(())
}

fn now_millis() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .ok()
        .and_then(|duration| i64::try_from(duration.as_millis()).ok())
        .unwrap_or(0)
}