goosedump 0.10.14

// 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::{self, OpenOptions};
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

#[cfg(unix)]
use std::os::unix::fs::{OpenOptionsExt as _, PermissionsExt as _};

use anyhow::{Context as _, bail};
use rusqlite::{
    Connection, ErrorCode, OptionalExtension as _, Transaction, TransactionBehavior,
    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};
use crate::text;

const SCHEMA_VERSION: i64 = 9;
const HASH_VERSION: u8 = 1;
const EMBEDDING_DIMENSIONS: usize = 384;
const ARCHIVE_BINS: i64 = 16;
const RRF_K: f64 = 60.0;
const RECALL_BACKFILL_BATCH: usize = 16;
const RECALL_CANDIDATE_LIMIT: usize = 64;
const COMPACTION_ATTRIBUTION_BATCH: usize = 16;
const MAX_EMBEDDING_TEXT_BYTES: usize = 8 * 1024;
const MAX_GOVERNANCE_TEXT_BYTES: usize = 16 * 1024;
const MAX_INGESTION_ATTEMPTS: i64 = 3;
const INGESTION_LEASE_MILLIS: i64 = 15 * 60 * 1_000;
const MAX_INGESTION_ERROR_BYTES: usize = 2 * 1024;
const GOVERNANCE_POLICY_VERSION: i64 = 1;
const GOVERNANCE_REVIEW_LIMIT: usize = 100;
const INGESTION_REVIEW_LIMIT: usize = 100;
const REASON_PROMPT_OVERRIDE: i64 = 1;
const REASON_ROLE_IMPERSONATION: i64 = 1 << 1;
const REASON_RETRIEVAL_INSTRUCTION: i64 = 1 << 2;
const REASON_HIDDEN_UNICODE: i64 = 1 << 3;
const REASON_CONTROL_CHARACTER: i64 = 1 << 4;
const REASON_OVERSIZED: i64 = 1 << 5;
const REASON_GENERATED: i64 = 1 << 6;
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();
static INGESTION_CLAIM_SEQUENCE: AtomicU64 = AtomicU64::new(0);

/// Semantic category assigned to one persistent-memory entry.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MemoryType {
    Decision,
    Fact,
    Preference,
    Procedure,
    Episode,
}

impl MemoryType {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Decision => "decision",
            Self::Fact => "fact",
            Self::Preference => "preference",
            Self::Procedure => "procedure",
            Self::Episode => "episode",
        }
    }
}

impl std::str::FromStr for MemoryType {
    type Err = &'static str;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value {
            "decision" => Ok(Self::Decision),
            "fact" => Ok(Self::Fact),
            "preference" => Ok(Self::Preference),
            "procedure" => Ok(Self::Procedure),
            "episode" => Ok(Self::Episode),
            _ => Err("memory type must be decision, fact, preference, procedure, or episode"),
        }
    }
}

/// Reach assigned to a typed memory; concrete identifiers remain in sightings.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MemoryScope {
    Context,
    Project,
}

impl MemoryScope {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Context => "context",
            Self::Project => "project",
        }
    }
}

impl std::str::FromStr for MemoryScope {
    type Err = &'static str;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value {
            "context" => Ok(Self::Context),
            "project" => Ok(Self::Project),
            _ => Err("memory scope must be context or project"),
        }
    }
}

/// How a typed-memory entry was produced.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum MemoryDerivation {
    Observed,
    Mutation,
}

impl MemoryDerivation {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Observed => "observed",
            Self::Mutation => "mutation",
        }
    }
}

impl std::str::FromStr for MemoryDerivation {
    type Err = &'static str;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value {
            "observed" => Ok(Self::Observed),
            "mutation" => Ok(Self::Mutation),
            _ => Err("memory derivation must be observed or mutation"),
        }
    }
}

/// Effective retrieval state assigned by memory governance.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum GovernanceStatus {
    Active,
    Quarantined,
    Expired,
    Superseded,
    Contradicted,
}

impl GovernanceStatus {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Active => "active",
            Self::Quarantined => "quarantined",
            Self::Expired => "expired",
            Self::Superseded => "superseded",
            Self::Contradicted => "contradicted",
        }
    }
}

impl std::str::FromStr for GovernanceStatus {
    type Err = &'static str;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value {
            "active" => Ok(Self::Active),
            "quarantined" => Ok(Self::Quarantined),
            "expired" => Ok(Self::Expired),
            "superseded" => Ok(Self::Superseded),
            "contradicted" => Ok(Self::Contradicted),
            _ => Err(
                "governance status must be active, quarantined, expired, superseded, or contradicted",
            ),
        }
    }
}

/// Durable state of one persistent-memory indexing operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum IngestionStatus {
    Pending,
    Processing,
    Indexed,
    Failed,
}

impl IngestionStatus {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Processing => "processing",
            Self::Indexed => "indexed",
            Self::Failed => "failed",
        }
    }
}

impl std::str::FromStr for IngestionStatus {
    type Err = &'static str;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value {
            "pending" => Ok(Self::Pending),
            "processing" => Ok(Self::Processing),
            "indexed" => Ok(Self::Indexed),
            "failed" => Ok(Self::Failed),
            _ => Err("ingestion status must be pending, processing, indexed, or failed"),
        }
    }
}

/// Persistent-memory entries grouped by indexing state.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
pub struct IngestionCounts {
    pub pending: u64,
    pub processing: u64,
    pub indexed: u64,
    pub failed: u64,
}

/// One durable indexing operation returned for inspection.
#[derive(Debug, Clone, Serialize)]
pub struct IngestionEntry {
    pub hash: String,
    pub kind: String,
    pub text: String,
    pub status: IngestionStatus,
    pub attempts: u64,
    pub last_error: Option<String>,
    pub updated_at: i64,
    pub indexed_at: Option<i64>,
}

/// Persistent-memory entry counts grouped by governance state.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
pub struct GovernanceCounts {
    pub active: u64,
    pub quarantined: u64,
    pub expired: u64,
    pub superseded: u64,
    pub contradicted: u64,
    pub content_tombstones: u64,
    pub context_tombstones: u64,
}

/// One memory returned for governance review.
#[derive(Debug, Clone, Serialize)]
pub struct GovernanceEntry {
    pub hash: String,
    pub kind: String,
    pub text: String,
    pub memory_type: MemoryType,
    pub status: GovernanceStatus,
    pub reasons: Vec<String>,
    pub provider: Option<Client>,
    pub context_id: Option<String>,
    pub entry_id: Option<String>,
    pub path: Option<PathBuf>,
}

/// Result of one explicit governance operation.
#[derive(Debug, Clone, Serialize)]
pub struct GovernanceReport {
    pub action: String,
    pub hash: String,
    pub related_hash: Option<String>,
    pub status: GovernanceStatus,
    pub reasons: Vec<String>,
}

/// 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>,
    /// Restrict results to one semantic memory type.
    pub memory_type: Option<MemoryType>,
}

/// 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,
    /// Semantic category assigned by the deterministic classifier.
    pub memory_type: MemoryType,
    /// Reliability of the source evidence, from zero to one.
    pub trust: f64,
    /// Classifier certainty, from zero to one.
    pub confidence: f64,
    /// Earliest known semantic validity time in Unix milliseconds.
    pub valid_from: Option<i64>,
    /// End of semantic validity in Unix milliseconds, when known.
    pub valid_until: Option<i64>,
    /// Context- or project-level reach; concrete scope is in provenance fields.
    pub scope: MemoryScope,
    /// Whether the entry was directly observed or generated by mutation.
    pub derivation: MemoryDerivation,
}

/// 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,
    /// Number of unique entries currently eligible for retrieval.
    pub admitted: usize,
    /// Number of unique entries retained for review but excluded from retrieval.
    pub quarantined: usize,
    /// Number of messages skipped because content or context was forgotten.
    pub skipped_tombstones: usize,
}

/// Persistent-memory entry counts grouped by semantic type.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
pub struct MemoryTypeCounts {
    pub decisions: u64,
    pub facts: u64,
    pub preferences: u64,
    pub procedures: u64,
    pub episodes: u64,
}

/// 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,
    pub types: MemoryTypeCounts,
    pub governance: GovernanceCounts,
    pub ingestion: IngestionCounts,
}

/// 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,
    pub admitted: bool,
    pub reasons: Vec<String>,
}

/// 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,
    pub search_hits: u64,
    pub tombstones: 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> {
        let path = database_path()?;
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
            secure_directory(parent)?;
        }
        Self::open_path(&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()?;
        prepare_database_path(path)?;
        let mut conn =
            Connection::open(path).with_context(|| format!("open {}", path.display()))?;
        conn.busy_timeout(Duration::from_secs(5))?;
        conn.pragma_update(None, "foreign_keys", true)?;
        initialize(&mut conn)?;
        if let Err(error) = conn.pragma_update(None, "journal_mode", "WAL")
            && !is_lock_error(&error)
        {
            return Err(error.into());
        }
        Ok(Self { conn })
    }

    /// Embed and index one bounded batch of entries without vectors.
    ///
    /// # Errors
    /// Returns an error if embedding fails or an index row cannot be written.
    pub fn index_missing_embeddings(
        &mut self,
        embedder: &Embedder,
        limit: usize,
    ) -> anyhow::Result<usize> {
        let claim = claim_ingestions(&mut self.conn, limit.min(RECALL_BACKFILL_BATCH))?;
        if claim.entries.is_empty() {
            return Ok(0);
        }
        let texts = claim
            .entries
            .iter()
            .map(|(_, text)| embedding_text(text))
            .collect::<Vec<_>>();
        let embeddings = match embedder.embed(&texts) {
            Ok(embeddings) => embeddings,
            Err(error) => {
                return Err(record_ingestion_failure(
                    &mut self.conn,
                    &claim.token,
                    error,
                ));
            }
        };
        if let Err(error) = ensure_embedding_batch(&embeddings, claim.entries.len()) {
            return Err(record_ingestion_failure(
                &mut self.conn,
                &claim.token,
                error,
            ));
        }
        let indexed = claim.entries.len();
        if let Err(error) = finish_ingestion_claim(&mut self.conn, &claim, &embeddings) {
            return Err(record_ingestion_failure(
                &mut self.conn,
                &claim.token,
                error,
            ));
        }
        Ok(indexed)
    }

    /// 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 a bounded recent subset of the current context.
    ///
    /// # 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 candidates = context
            .messages
            .iter()
            .rev()
            .filter_map(|message| {
                let text = display::searchable_text(message);
                if text.trim().is_empty() {
                    return None;
                }
                let hash = hash_content(&collapsed_kind(message), &text);
                seen.insert(hash.clone()).then_some((hash, text))
            })
            .take(COMPACTION_ATTRIBUTION_BATCH)
            .collect::<Vec<_>>();
        let claim = claim_attribution_ingestions(&mut self.conn, candidates)?;
        if claim.entries.is_empty() {
            return Ok(0);
        }

        let mut texts = Vec::with_capacity(claim.entries.len() + 1);
        texts.push(embedding_text(summary));
        texts.extend(claim.entries.iter().map(|(_, text)| embedding_text(text)));
        let embeddings = match embedder.embed(&texts) {
            Ok(embeddings) => embeddings,
            Err(error) => {
                return Err(record_ingestion_failure(
                    &mut self.conn,
                    &claim.token,
                    error,
                ));
            }
        };
        if let Err(error) = ensure_embedding_batch(&embeddings, texts.len()) {
            return Err(record_ingestion_failure(
                &mut self.conn,
                &claim.token,
                error,
            ));
        }
        match finish_attribution_claim(&mut self.conn, &claim, &embeddings) {
            Ok(changed) => Ok(changed),
            Err(error) => Err(record_ingestion_failure(
                &mut self.conn,
                &claim.token,
                error,
            )),
        }
    }

    /// 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_with_behavior(TransactionBehavior::Immediate)?;
        let source_key = context_tombstone_key(provider, context_id);
        let context_forgotten = tx.query_row(
            "SELECT EXISTS(
                SELECT 1 FROM memory_context_tombstones WHERE source_key = ?1
             )",
            params![source_key],
            |row| row.get::<_, bool>(0),
        )?;
        if context_forgotten {
            tx.commit()?;
            return Ok(HarvestReport {
                messages: context.messages.len(),
                unique_entries: 0,
                admitted: 0,
                quarantined: 0,
                skipped_tombstones: context.messages.len(),
            });
        }

        let mut hashes = HashSet::new();
        let mut admitted_hashes = HashSet::new();
        let mut quarantined_hashes = HashSet::new();
        let mut skipped_tombstones = 0;

        for (ordinal, message) in context.messages.iter().enumerate() {
            let kind = collapsed_kind(message);
            let text = display::searchable_text(message);
            if text.trim().is_empty() {
                continue;
            }
            let hash = hash_content(&kind, &text);
            let forgotten = tx.query_row(
                "SELECT EXISTS(
                    SELECT 1 FROM memory_content_tombstones WHERE entry_hash = ?1
                 )",
                params![hash],
                |row| row.get::<_, bool>(0),
            )?;
            if forgotten {
                skipped_tombstones += 1;
                continue;
            }
            let observed_at = message
                .timestamp
                .map_or_else(now_millis, |timestamp| timestamp.timestamp_millis());
            hashes.insert(hash.clone());
            upsert_entry(&tx, &hash, &kind, &text)?;
            let semantics = classify_memory(&kind, &text, observed_at, MemoryDerivation::Observed);
            upsert_entry_semantics(&tx, &hash, semantics)?;
            let reasons = governance_reasons(&kind, &text, semantics.derivation);
            upsert_governance(&tx, &hash, reasons)?;
            upsert_sighting(
                &tx,
                SightingInput {
                    hash: &hash,
                    provider,
                    context_id,
                    entry_id: &message.entry_id,
                    ordinal,
                    path: &path,
                    source_path: &source_path,
                    observed_at,
                },
            )?;
            if sync_governed_indexes(&tx, &hash)? {
                admitted_hashes.insert(hash.clone());
                quarantined_hashes.remove(&hash);
            } else {
                quarantined_hashes.insert(hash.clone());
                admitted_hashes.remove(&hash);
            }
        }
        prune_orphans(&tx)?;

        tx.commit()?;

        Ok(HarvestReport {
            messages: context.messages.len(),
            unique_entries: hashes.len(),
            admitted: admitted_hashes.len(),
            quarantined: quarantined_hashes.len(),
            skipped_tombstones,
        })
    }

    /// 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;
        let mut seen = HashSet::new();
        for entry_id in entry_ids {
            if !seen.insert(entry_id) {
                continue;
            }
            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 result_limit = limit.min(RECALL_CANDIDATE_LIMIT);
        let query_embedding = if query.trim().is_empty() || result_limit == 0 {
            None
        } else {
            self.index_missing_embeddings(embedder, RECALL_BACKFILL_BATCH)?;
            let mut embeddings = embedder.embed(&[query.to_string()])?;
            ensure_embedding_batch(&embeddings, 1)?;
            embeddings.pop()
        };

        let rows = if let Some(query_embedding) = query_embedding {
            let candidate_limit = i64::try_from(RECALL_CANDIDATE_LIMIT)?;
            let fts_query = fts_query(query);
            let lexical = if fts_query.is_empty() {
                Vec::new()
            } else {
                query_recall(
                    &self.conn,
                    &fts_query,
                    provider,
                    path.as_deref(),
                    filter.memory_type,
                    candidate_limit,
                )?
            };
            let semantic = query_vector_recall(
                &self.conn,
                &query_embedding,
                provider,
                path.as_deref(),
                filter.memory_type,
                candidate_limit,
            )?;
            let (lexical, semantic) = rescreen_recall_rows(&mut self.conn, lexical, semantic)?;
            reciprocal_rank_fusion(lexical, semantic, result_limit)
        } else {
            Vec::new()
        };

        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        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 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, (SELECT id FROM sightings WHERE id = ?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,
                memory_type: row.memory_type,
                trust: row.trust,
                confidence: row.confidence,
                valid_from: row.valid_from,
                valid_until: row.valid_until,
                scope: row.scope,
                derivation: row.derivation,
            });
        }
        tx.commit()?;
        Ok(hits)
    }

    /// 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_with_behavior(TransactionBehavior::Immediate)?;
        if tx.query_row(
            "SELECT EXISTS(
                SELECT 1 FROM memory_content_tombstones WHERE entry_hash = ?1
             )",
            params![hash],
            |row| row.get::<_, bool>(0),
        )? {
            bail!("memory content was forgotten: {hash}");
        }
        let inserted = upsert_entry(&tx, &hash, "mutation", &text)? != 0;
        let semantics =
            classify_memory("mutation", &text, now_millis(), MemoryDerivation::Mutation);
        upsert_entry_semantics(&tx, &hash, semantics)?;
        let reasons = governance_reasons("mutation", &text, semantics.derivation);
        let governance = upsert_governance(&tx, &hash, reasons)?;
        for parent in [left, right] {
            tx.execute(
                "INSERT INTO memory_mutations(hash, source_hash, source_sighting_id)
                 VALUES(?1, ?2, ?3)
                 ON CONFLICT(hash, source_hash) DO UPDATE SET
                    source_sighting_id = excluded.source_sighting_id",
                params![hash, parent.hash, parent.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(),
            },
        )?;
        let admitted = sync_governed_indexes(&tx, &hash)?;
        if ingestion_is_blocked(&tx, &hash)? {
            bail!("memory ingestion {hash} is already processing or failed");
        }
        if admitted {
            insert_embedding(&tx, &hash, &embedding[0])?;
        }
        mark_ingestion_indexed(&tx, &hash, now_millis())?;
        rebuild_shadow_archive(&tx)?;
        record_stage2_snapshot(&tx)?;
        tx.commit()?;
        Ok(MutationReport {
            hash,
            text,
            inserted,
            admitted,
            reasons: governance_reason_names(
                governance.reason_mask & !governance.override_reason_mask,
            ),
        })
    }

    /// Return bounded memory-governance review rows.
    ///
    /// # Errors
    /// Returns an error if governance state cannot be queried.
    pub fn review_governance(
        &self,
        status: Option<GovernanceStatus>,
        limit: usize,
    ) -> anyhow::Result<Vec<GovernanceEntry>> {
        query_governance_entries(
            &self.conn,
            status,
            i64::try_from(limit.min(GOVERNANCE_REVIEW_LIMIT))?,
        )
    }

    /// Return bounded durable-ingestion review rows.
    ///
    /// # Errors
    /// Returns an error if ingestion state cannot be queried.
    pub fn review_ingestions(
        &self,
        status: Option<IngestionStatus>,
        limit: usize,
    ) -> anyhow::Result<Vec<IngestionEntry>> {
        query_ingestion_entries(
            &self.conn,
            status,
            i64::try_from(limit.min(INGESTION_REVIEW_LIMIT))?,
        )
    }

    /// Reset one terminally failed ingestion for another bounded retry cycle.
    ///
    /// # Errors
    /// Returns an error if the entry is missing, is not failed, or cannot be updated.
    pub fn retry_ingestion(&mut self, hash: &str) -> anyhow::Result<IngestionEntry> {
        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        let changed = tx.execute(
            "UPDATE memory_ingestions
             SET status = 'pending', attempts = 0, lease_until = NULL,
                 claim_token = NULL, last_error = NULL, updated_at = ?2,
                 indexed_at = NULL
             WHERE entry_hash = ?1 AND status = 'failed'",
            params![hash, now_millis()],
        )?;
        if changed == 0 {
            let status = tx
                .query_row(
                    "SELECT status FROM memory_ingestions WHERE entry_hash = ?1",
                    params![hash],
                    |row| row.get::<_, String>(0),
                )
                .optional()?;
            match status {
                Some(status) => bail!("memory ingestion {hash} is {status}, not failed"),
                None => bail!("memory ingestion not found: {hash}"),
            }
        }
        let entry = query_ingestion_entry(&tx, hash)?;
        tx.commit()?;
        Ok(entry)
    }

    /// Admit a quarantined entry by explicitly overriding its current findings.
    ///
    /// Lifecycle exclusions such as expiry, contradiction, and supersession remain
    /// in force.
    ///
    /// # Errors
    /// Returns an error if the entry does not exist, embedding fails, or the
    /// governance transaction cannot be committed.
    pub fn approve_memory(
        &mut self,
        hash: &str,
        embedder: &Embedder,
    ) -> anyhow::Result<GovernanceReport> {
        let text = self
            .conn
            .query_row(
                "SELECT text FROM entries WHERE hash = ?1",
                params![hash],
                |row| row.get::<_, String>(0),
            )
            .optional()?
            .with_context(|| format!("memory entry not found: {hash}"))?;
        if ingestion_is_blocked(&self.conn, hash)? {
            bail!("memory ingestion {hash} is processing or failed; retry it before approval");
        }
        let mut embeddings = embedder.embed(&[embedding_text(&text)])?;
        ensure_embedding_batch(&embeddings, 1)?;
        let embedding = embeddings.pop().context("missing approval embedding")?;

        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        require_entry(&tx, hash)?;
        if ingestion_is_blocked(&tx, hash)? {
            bail!("memory ingestion {hash} is processing or failed; retry it before approval");
        }
        tx.execute(
            "UPDATE memory_governance
             SET override_reason_mask = reason_mask,
                 manual_quarantine = 0,
                 policy_version = ?2,
                 screened_at = ?3
             WHERE entry_hash = ?1",
            params![hash, GOVERNANCE_POLICY_VERSION, now_millis()],
        )?;
        let admitted = sync_governed_indexes(&tx, hash)?;
        if admitted {
            insert_embedding(&tx, hash, &embedding)?;
        }
        mark_ingestion_indexed(&tx, hash, now_millis())?;
        rebuild_shadow_archive(&tx)?;
        let report = governance_report(&tx, "approve", hash, None)?;
        tx.commit()?;
        Ok(report)
    }

    /// Manually quarantine an entry and remove it from retrieval indexes.
    ///
    /// # Errors
    /// Returns an error if the entry does not exist or the transaction fails.
    pub fn quarantine_memory(&mut self, hash: &str) -> anyhow::Result<GovernanceReport> {
        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        require_entry(&tx, hash)?;
        tx.execute(
            "UPDATE memory_governance
             SET manual_quarantine = 1, screened_at = ?2
             WHERE entry_hash = ?1",
            params![hash, now_millis()],
        )?;
        sync_governed_indexes(&tx, hash)?;
        rebuild_shadow_archive(&tx)?;
        let report = governance_report(&tx, "quarantine", hash, None)?;
        tx.commit()?;
        Ok(report)
    }

    /// Mark one entry as superseding another entry.
    ///
    /// The replacement remains governed by its own admission state; the replaced
    /// entry is excluded from retrieval.
    ///
    /// # Errors
    /// Returns an error if either entry does not exist or the transaction fails.
    pub fn supersede_memory(
        &mut self,
        replacement_hash: &str,
        replaced_hash: &str,
    ) -> anyhow::Result<GovernanceReport> {
        if replacement_hash == replaced_hash {
            bail!("replacement and replaced memory must differ");
        }
        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        require_entry(&tx, replacement_hash)?;
        require_entry(&tx, replaced_hash)?;
        tx.execute(
            "INSERT INTO memory_relations(source_hash, target_hash, relation, created_at)
             VALUES(?1, ?2, 'supersedes', ?3)
             ON CONFLICT(source_hash, target_hash, relation) DO NOTHING",
            params![replacement_hash, replaced_hash, now_millis()],
        )?;
        sync_governed_indexes(&tx, replacement_hash)?;
        sync_governed_indexes(&tx, replaced_hash)?;
        rebuild_shadow_archive(&tx)?;
        let report = governance_report(
            &tx,
            "supersede",
            replaced_hash,
            Some(replacement_hash.to_string()),
        )?;
        tx.commit()?;
        Ok(report)
    }

    /// Mark two entries as contradictory and exclude both from retrieval.
    ///
    /// # Errors
    /// Returns an error if either entry does not exist or the transaction fails.
    pub fn contradict_memory(
        &mut self,
        left_hash: &str,
        right_hash: &str,
    ) -> anyhow::Result<GovernanceReport> {
        if left_hash == right_hash {
            bail!("contradictory memories must differ");
        }
        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        require_entry(&tx, left_hash)?;
        require_entry(&tx, right_hash)?;
        tx.execute(
            "INSERT INTO memory_relations(source_hash, target_hash, relation, created_at)
             VALUES(?1, ?2, 'contradicts', ?3)
             ON CONFLICT(source_hash, target_hash, relation) DO NOTHING",
            params![left_hash, right_hash, now_millis()],
        )?;
        sync_governed_indexes(&tx, left_hash)?;
        sync_governed_indexes(&tx, right_hash)?;
        rebuild_shadow_archive(&tx)?;
        let report = governance_report(&tx, "contradict", left_hash, Some(right_hash.to_string()))?;
        tx.commit()?;
        Ok(report)
    }

    /// Expire one entry immediately and remove it from retrieval indexes.
    ///
    /// # Errors
    /// Returns an error if the entry does not exist or the transaction fails.
    pub fn expire_memory(&mut self, hash: &str) -> anyhow::Result<GovernanceReport> {
        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        require_entry(&tx, hash)?;
        let expired_at = now_millis();
        tx.execute(
            "UPDATE memory_semantics
             SET valid_from = min(valid_from, ?2), valid_until = ?2
             WHERE entry_hash = ?1",
            params![hash, expired_at],
        )?;
        sync_governed_indexes(&tx, hash)?;
        rebuild_shadow_archive(&tx)?;
        let report = governance_report(&tx, "expire", hash, None)?;
        tx.commit()?;
        Ok(report)
    }

    /// 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(DISTINCT hash) FROM memory_mutations",
            )?,
            stage2: self.stage2_status()?,
            types: memory_type_counts(&self.conn)?,
            governance: governance_counts(&self.conn)?,
            ingestion: ingestion_counts(&self.conn)?,
        })
    }

    /// 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 quality_stable = snapshots.windows(2).all(|pair| {
            relative_delta(pair[0].quality_per_entry(), pair[1].quality_per_entry())
                <= PLATEAU_QUALITY_DELTA
        });
        let expansion_rates = snapshots
            .windows(2)
            .map(|pair| pair[1].expansion_rate_since(pair[0]))
            .collect::<Vec<_>>();
        let expansion_stable = expansion_rates
            .windows(2)
            .all(|pair| (pair[0] - pair[1]).abs() <= PLATEAU_EXPANSION_RATE_DELTA);
        let plateaued = windows == PLATEAU_WINDOWS
            && snapshots
                .windows(2)
                .all(|pair| pair[0].archive_entries == pair[1].archive_entries)
            && quality_stable
            && expansion_stable;
        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_since(baseline),
            archive_entries: current.archive_entries,
            quality_per_entry: current.quality_per_entry(),
            plateaued,
        })
    }

    /// Forget content, its derived memories, sightings, and affected search hits.
    ///
    /// Durable content tombstones prevent later re-ingestion of the deleted text.
    ///
    /// # 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_with_behavior(TransactionBehavior::Immediate)?;
        let hashes = derived_hashes(&tx, hash)?;
        let mut sightings = 0_u64;
        let mut search_hits = 0_u64;
        let mut entries = 0_u64;
        let mut tombstones = 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),
            )?;
            sightings =
                sightings.saturating_add(u64::try_from(count).context("negative database count")?);
            search_hits = search_hits.saturating_add(u64::try_from(tx.execute(
                "DELETE FROM search_hits WHERE entry_hash = ?1",
                params![hash],
            )?)?);
            tombstones = tombstones.saturating_add(u64::try_from(tx.execute(
                "INSERT INTO memory_content_tombstones(entry_hash, deleted_at)
                 VALUES(?1, ?2)
                 ON CONFLICT(entry_hash) DO NOTHING",
                params![hash, now_millis()],
            )?)?);
            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])?,
            )?);
        }
        rebuild_shadow_archive(&tx)?;
        tx.execute("DELETE FROM stage2_snapshots", [])?;
        tx.commit()?;
        Ok(ForgetReport {
            entries,
            sightings,
            search_hits,
            tombstones,
        })
    }

    /// Forget one provider context and memories solely derived from it.
    ///
    /// Shared entries remain available through sightings in other contexts. A
    /// hashed context tombstone prevents the deleted context from being ingested
    /// again, while affected search-hit snapshots are removed.
    ///
    /// # 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_with_behavior(TransactionBehavior::Immediate)?;
        let before = count_tx(&tx, "SELECT count(*) FROM entries")?;
        let sightings_before = count_tx(&tx, "SELECT count(*) FROM sightings")?;
        let search_hits_before = count_tx(&tx, "SELECT count(*) FROM search_hits")?;
        let roots = {
            let mut stmt = tx.prepare(
                "SELECT DISTINCT target.entry_hash
                 FROM sightings AS target
                 WHERE target.provider = ?1 AND target.context_id = ?2
                   AND NOT EXISTS(
                      SELECT 1 FROM sightings AS other
                      WHERE other.entry_hash = target.entry_hash
                        AND (other.provider != ?1 OR other.context_id != ?2)
                   )",
            )?;
            stmt.query_map(params![provider.as_str(), context_id], |row| {
                row.get::<_, String>(0)
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?
        };
        let mut derived = HashSet::new();
        for root in roots {
            derived.extend(derived_hashes(&tx, &root)?);
        }
        tx.execute(
            "UPDATE memory_mutations
             SET source_sighting_id = (
                SELECT replacement.id FROM sightings AS replacement
                WHERE replacement.entry_hash = memory_mutations.source_hash
                  AND (replacement.provider != ?1 OR replacement.context_id != ?2)
                ORDER BY replacement.observed_at DESC, replacement.id DESC
                LIMIT 1
             )
             WHERE source_sighting_id IN (
                SELECT id FROM sightings
                WHERE provider = ?1 AND context_id = ?2
             )
               AND EXISTS(
                SELECT 1 FROM sightings AS replacement
                WHERE replacement.entry_hash = memory_mutations.source_hash
                  AND (replacement.provider != ?1 OR replacement.context_id != ?2)
             )",
            params![provider.as_str(), context_id],
        )?;
        tx.execute(
            "DELETE FROM search_hits WHERE provider = ?1 AND context_id = ?2",
            params![provider.as_str(), context_id],
        )?;
        tx.execute(
            "DELETE FROM sightings WHERE provider = ?1 AND context_id = ?2",
            params![provider.as_str(), context_id],
        )?;

        let mut tombstones = u64::try_from(tx.execute(
            "INSERT INTO memory_context_tombstones(source_key, deleted_at)
             VALUES(?1, ?2)
             ON CONFLICT(source_key) DO NOTHING",
            params![
                context_tombstone_key(provider.as_str(), context_id),
                now_millis()
            ],
        )?)?;
        for hash in derived {
            tx.execute(
                "DELETE FROM search_hits WHERE entry_hash = ?1",
                params![hash],
            )?;
            tombstones = tombstones.saturating_add(u64::try_from(tx.execute(
                "INSERT INTO memory_content_tombstones(entry_hash, deleted_at)
                 VALUES(?1, ?2)
                 ON CONFLICT(entry_hash) DO NOTHING",
                params![hash, now_millis()],
            )?)?);
            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])?;
        }
        prune_orphans(&tx)?;
        rebuild_shadow_archive(&tx)?;
        tx.execute("DELETE FROM stage2_snapshots", [])?;
        let after = count_tx(&tx, "SELECT count(*) FROM entries")?;
        let sightings_after = count_tx(&tx, "SELECT count(*) FROM sightings")?;
        let search_hits_after = count_tx(&tx, "SELECT count(*) FROM search_hits")?;
        tx.commit()?;
        Ok(ForgetReport {
            entries: before.saturating_sub(after),
            sightings: sightings_before.saturating_sub(sightings_after),
            search_hits: search_hits_before.saturating_sub(search_hits_after),
            tombstones,
        })
    }
}

/// Return the default database path without opening it.
///
/// # Errors
/// Returns an error when `GOOSEDUMP_STATE_DIR` is unset or empty and the
/// platform has neither a state nor local-data 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()
        .or_else(dirs::data_local_dir)
        .context("state or local data directory not found")?;
    Ok(root.join("goosedump").join("goosedump.db"))
}

fn prepare_database_path(path: &Path) -> anyhow::Result<()> {
    if let Some(parent) = path.parent()
        && !parent.as_os_str().is_empty()
        && !parent.exists()
    {
        fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
        secure_directory(parent)?;
    }
    prepare_database_file(path)
}

#[cfg(unix)]
fn secure_directory(path: &Path) -> anyhow::Result<()> {
    fs::set_permissions(path, fs::Permissions::from_mode(0o700))
        .with_context(|| format!("secure {}", path.display()))
}

#[cfg(not(unix))]
fn secure_directory(path: &Path) -> anyhow::Result<()> {
    fs::create_dir_all(path).with_context(|| format!("create {}", path.display()))
}

#[cfg(unix)]
fn prepare_database_file(path: &Path) -> anyhow::Result<()> {
    OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .mode(0o600)
        .open(path)
        .with_context(|| format!("create {}", path.display()))?;
    fs::set_permissions(path, fs::Permissions::from_mode(0o600))
        .with_context(|| format!("secure {}", path.display()))
}

#[cfg(not(unix))]
fn prepare_database_file(path: &Path) -> anyhow::Result<()> {
    OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(path)
        .with_context(|| format!("create {}", path.display()))?;
    Ok(())
}

fn is_lock_error(error: &rusqlite::Error) -> bool {
    matches!(
        error,
        rusqlite::Error::SqliteFailure(inner, _)
            if matches!(inner.code, ErrorCode::DatabaseBusy | ErrorCode::DatabaseLocked)
    )
}

/// 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,
}

#[derive(Clone, Copy)]
struct EntrySemantics {
    memory_type: MemoryType,
    trust: f64,
    confidence: f64,
    valid_from: Option<i64>,
    valid_until: Option<i64>,
    scope: MemoryScope,
    derivation: MemoryDerivation,
}

#[derive(Clone, Copy)]
struct GovernanceRecord {
    reason_mask: i64,
    override_reason_mask: i64,
    manual_quarantine: bool,
    policy_version: i64,
}

#[derive(Clone)]
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,
    memory_type: MemoryType,
    trust: f64,
    confidence: f64,
    valid_from: Option<i64>,
    valid_until: Option<i64>,
    scope: MemoryScope,
    derivation: MemoryDerivation,
    reason_mask: i64,
    override_reason_mask: i64,
    policy_version: i64,
}

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,
    embedding: Vec<f32>,
}

#[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_since(self, previous: Self) -> f64 {
        let search_hits = self.search_hits.saturating_sub(previous.search_hits);
        if search_hits == 0 {
            return 0.0;
        }
        let expansions = self.expansions.saturating_sub(previous.expansions);
        f64::from(u32::try_from(expansions).unwrap_or(u32::MAX))
            / f64::from(u32::try_from(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))
        }
    }
}

fn relative_delta(left: f64, right: f64) -> f64 {
    if left.abs() < f64::EPSILON {
        right.abs()
    } else {
        ((left - right) / left).abs()
    }
}

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: &mut Connection) -> anyhow::Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let mut version: i64 = tx.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(&tx)?;
        version = 1;
    }
    if version == 1 {
        migrate_embeddings_archive(&tx)?;
        version = 2;
    }
    if version == 2 {
        migrate_embedding_metadata(&tx)?;
        version = 3;
    }
    if version == 3 {
        migrate_stage2_snapshots(&tx)?;
        version = 4;
    }
    if version == 4 {
        migrate_mutations(&tx)?;
        version = 5;
    }
    if version == 5 {
        migrate_multi_parent_mutations(&tx)?;
        remove_empty_entries(&tx)?;
        version = 6;
    }
    if version == 6 {
        migrate_typed_memory(&tx)?;
        version = 7;
    }
    if version == 7 {
        migrate_memory_governance(&tx)?;
        version = 8;
    }
    if version == 8 {
        migrate_recoverable_ingestion(&tx)?;
        version = 9;
    }
    if version >= 8 {
        rescreen_stale_governance(&tx)?;
    }
    if version == 9 {
        recover_stale_ingestions(&tx, now_millis())?;
    }
    reset_stale_embeddings(&tx)?;
    tx.commit()?;
    Ok(())
}

fn create_initial_schema(tx: &Transaction<'_>) -> anyhow::Result<()> {
    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;",
    )?;
    Ok(())
}

fn migrate_embeddings_archive(tx: &Transaction<'_>) -> anyhow::Result<()> {
    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;",
    )?;
    Ok(())
}

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

fn migrate_stage2_snapshots(tx: &Transaction<'_>) -> anyhow::Result<()> {
    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;",
    )?;
    Ok(())
}

fn migrate_mutations(tx: &Transaction<'_>) -> anyhow::Result<()> {
    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;",
    )?;
    Ok(())
}

fn migrate_multi_parent_mutations(tx: &Transaction<'_>) -> anyhow::Result<()> {
    tx.execute_batch(
        "ALTER TABLE memory_mutations RENAME TO memory_mutations_v5;
         CREATE TABLE memory_mutations(
            hash TEXT NOT NULL 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,
            PRIMARY KEY(hash, source_hash)
         ) WITHOUT ROWID;
         INSERT INTO memory_mutations(hash, source_hash, source_sighting_id)
            SELECT hash, source_hash, source_sighting_id FROM memory_mutations_v5;
         DROP TABLE memory_mutations_v5;
         PRAGMA user_version = 6;",
    )?;
    Ok(())
}

fn migrate_typed_memory(tx: &Transaction<'_>) -> anyhow::Result<()> {
    tx.execute_batch(
        "CREATE TABLE memory_semantics(
            entry_hash TEXT PRIMARY KEY REFERENCES entries(hash) ON DELETE CASCADE,
            memory_type TEXT NOT NULL CHECK(memory_type IN (
                'decision', 'fact', 'preference', 'procedure', 'episode'
            )),
            trust REAL NOT NULL CHECK(trust >= 0.0 AND trust <= 1.0),
            confidence REAL NOT NULL CHECK(confidence >= 0.0 AND confidence <= 1.0),
            valid_from INTEGER,
            valid_until INTEGER,
            scope TEXT NOT NULL CHECK(scope IN ('context', 'project')),
            derivation TEXT NOT NULL CHECK(derivation IN ('observed', 'mutation')),
            classified_at INTEGER NOT NULL,
            CHECK(valid_from IS NULL OR valid_until IS NULL OR valid_until >= valid_from)
         ) WITHOUT ROWID;
         CREATE INDEX memory_semantics_type ON memory_semantics(memory_type);",
    )?;

    let rows = {
        let mut stmt = tx.prepare(
            "SELECT e.hash, e.kind, e.text,
                    coalesce(min(s.observed_at), e.created_at),
                    EXISTS(SELECT 1 FROM memory_mutations AS m WHERE m.hash = e.hash)
             FROM entries AS e
             LEFT JOIN sightings AS s ON s.entry_hash = e.hash
             GROUP BY e.hash, e.kind, e.text, e.created_at
             ORDER BY e.hash",
        )?;
        stmt.query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, i64>(3)?,
                row.get::<_, bool>(4)?,
            ))
        })?
        .collect::<rusqlite::Result<Vec<_>>>()?
    };
    for (hash, kind, text, observed_at, mutated) in rows {
        let derivation = if mutated {
            MemoryDerivation::Mutation
        } else {
            MemoryDerivation::Observed
        };
        upsert_entry_semantics(
            tx,
            &hash,
            classify_memory(&kind, &text, observed_at, derivation),
        )?;
    }
    tx.pragma_update(None, "user_version", 7)?;
    Ok(())
}

fn migrate_memory_governance(tx: &Transaction<'_>) -> anyhow::Result<()> {
    tx.execute_batch(
        "CREATE TABLE memory_governance(
            entry_hash TEXT PRIMARY KEY REFERENCES entries(hash) ON DELETE CASCADE,
            reason_mask INTEGER NOT NULL DEFAULT 0,
            override_reason_mask INTEGER NOT NULL DEFAULT 0,
            manual_quarantine INTEGER NOT NULL DEFAULT 0
                CHECK(manual_quarantine IN (0, 1)),
            policy_version INTEGER NOT NULL,
            screened_at INTEGER NOT NULL
         ) WITHOUT ROWID;
         CREATE INDEX memory_governance_review
            ON memory_governance(manual_quarantine, reason_mask, override_reason_mask);

         CREATE TABLE memory_relations(
            source_hash TEXT NOT NULL REFERENCES entries(hash) ON DELETE CASCADE,
            target_hash TEXT NOT NULL REFERENCES entries(hash) ON DELETE CASCADE,
            relation TEXT NOT NULL CHECK(relation IN ('supersedes', 'contradicts')),
            created_at INTEGER NOT NULL,
            CHECK(source_hash != target_hash),
            PRIMARY KEY(source_hash, target_hash, relation)
         ) WITHOUT ROWID;
         CREATE INDEX memory_relations_target
            ON memory_relations(target_hash, relation);

         CREATE TABLE memory_content_tombstones(
            entry_hash TEXT PRIMARY KEY,
            deleted_at INTEGER NOT NULL
         ) WITHOUT ROWID;

         CREATE TABLE memory_context_tombstones(
            source_key TEXT PRIMARY KEY,
            deleted_at INTEGER NOT NULL
         ) WITHOUT ROWID;

         INSERT INTO memory_governance(
            entry_hash, reason_mask, override_reason_mask, manual_quarantine,
            policy_version, screened_at
         )
         SELECT hash,
                CASE WHEN kind = 'mutation' THEN 64 ELSE 0 END,
                0, 0,
                CASE WHEN kind = 'mutation' THEN 1 ELSE 0 END,
                last_seen_at
         FROM entries;

         DELETE FROM entries_vec
         WHERE hash IN (SELECT hash FROM entries WHERE kind = 'mutation');
         DELETE FROM entries_fts
         WHERE hash IN (SELECT hash FROM entries WHERE kind = 'mutation');
         DELETE FROM memory_archive
         WHERE entry_hash IN (SELECT hash FROM entries WHERE kind = 'mutation');

         PRAGMA user_version = 8;",
    )?;
    Ok(())
}

fn rescreen_stale_governance(tx: &Transaction<'_>) -> anyhow::Result<()> {
    let entries = {
        let mut stmt = tx.prepare(
            "SELECT entries.hash, entries.kind, entries.text
             FROM entries
             JOIN memory_governance AS governance
               ON governance.entry_hash = entries.hash
             WHERE governance.policy_version != ?1
             ORDER BY entries.hash",
        )?;
        stmt.query_map(params![GOVERNANCE_POLICY_VERSION], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
            ))
        })?
        .collect::<rusqlite::Result<Vec<_>>>()?
    };
    if entries.is_empty() {
        return Ok(());
    }
    for (hash, kind, text) in entries {
        let derivation = if kind == "mutation" {
            MemoryDerivation::Mutation
        } else {
            MemoryDerivation::Observed
        };
        upsert_governance(tx, &hash, governance_reasons(&kind, &text, derivation))?;
        sync_governed_indexes(tx, &hash)?;
    }
    rebuild_shadow_archive(tx)?;
    Ok(())
}

fn remove_empty_entries(tx: &Transaction<'_>) -> anyhow::Result<()> {
    let hashes = {
        let mut stmt = tx.prepare("SELECT hash, text FROM entries ORDER BY hash")?;
        stmt.query_map([], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        })?
        .filter_map(|row| match row {
            Ok((hash, text)) if text.trim().is_empty() => Some(Ok(hash)),
            Ok(_) => None,
            Err(error) => Some(Err(error)),
        })
        .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 reset_stale_embeddings(tx: &Transaction<'_>) -> anyhow::Result<()> {
    let stored = tx
        .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(());
    }

    tx.execute("DELETE FROM memory_archive", [])?;
    tx.execute("DELETE FROM stage2_snapshots", [])?;
    tx.execute("DELETE FROM entries_vec", [])?;
    tx.execute("UPDATE entries SET coverage = 0.0", [])?;
    tx.execute(
        "UPDATE memory_ingestions
         SET status = 'pending', attempts = 0, lease_until = NULL,
             claim_token = NULL, last_error = NULL, updated_at = ?1,
             indexed_at = NULL",
        params![now_millis()],
    )?;
    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],
    )?;
    Ok(())
}

fn migrate_recoverable_ingestion(tx: &Transaction<'_>) -> anyhow::Result<()> {
    tx.execute_batch(
        "CREATE TABLE memory_ingestions(
            entry_hash TEXT PRIMARY KEY REFERENCES entries(hash) ON DELETE CASCADE,
            status TEXT NOT NULL
                CHECK(status IN ('pending', 'processing', 'indexed', 'failed')),
            attempts INTEGER NOT NULL DEFAULT 0
                CHECK(attempts >= 0 AND attempts <= 3),
            lease_until INTEGER,
            claim_token TEXT,
            last_error TEXT,
            created_at INTEGER NOT NULL,
            updated_at INTEGER NOT NULL,
            indexed_at INTEGER,
            CHECK((status = 'processing') = (lease_until IS NOT NULL)),
            CHECK((status = 'processing') = (claim_token IS NOT NULL)),
            CHECK((status = 'indexed') = (indexed_at IS NOT NULL))
         ) WITHOUT ROWID;
         CREATE INDEX memory_ingestions_ready
            ON memory_ingestions(status, updated_at, entry_hash);",
    )?;
    let now = now_millis();
    tx.execute(
        "INSERT INTO memory_ingestions(
            entry_hash, status, attempts, created_at, updated_at, indexed_at
         )
         SELECT entries.hash,
                CASE WHEN entries_vec.hash IS NULL THEN 'pending' ELSE 'indexed' END,
                0, ?1, ?1,
                CASE WHEN entries_vec.hash IS NULL THEN NULL ELSE ?1 END
         FROM entries
         LEFT JOIN entries_vec ON entries_vec.hash = entries.hash",
        params![now],
    )?;
    tx.pragma_update(None, "user_version", 9)?;
    Ok(())
}

fn recover_stale_ingestions(tx: &Transaction<'_>, now: i64) -> anyhow::Result<()> {
    tx.execute(
        "UPDATE memory_ingestions
         SET status = CASE WHEN attempts >= ?2 THEN 'failed' ELSE 'pending' END,
             lease_until = NULL,
             claim_token = NULL,
             last_error = coalesce(last_error, 'ingestion interrupted'),
             updated_at = ?1
         WHERE status = 'processing' AND lease_until <= ?1",
        params![now, MAX_INGESTION_ATTEMPTS],
    )?;
    tx.execute(
        "UPDATE memory_ingestions
         SET status = 'failed', last_error = coalesce(last_error, 'retry limit reached'),
             updated_at = ?1
         WHERE status = 'pending' AND attempts >= ?2",
        params![now, MAX_INGESTION_ATTEMPTS],
    )?;
    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 classify_memory(
    kind: &str,
    text: &str,
    observed_at: i64,
    derivation: MemoryDerivation,
) -> EntrySemantics {
    if derivation == MemoryDerivation::Mutation {
        return EntrySemantics {
            memory_type: MemoryType::Fact,
            trust: 0.4,
            confidence: 0.5,
            valid_from: Some(observed_at),
            valid_until: None,
            scope: MemoryScope::Project,
            derivation,
        };
    }

    let normalized = text.to_ascii_lowercase();
    let memory_type = if matches!(kind, "tool_result" | "bash") {
        MemoryType::Fact
    } else if contains_any(
        &normalized,
        &[
            "i prefer",
            "preference:",
            "please always",
            "must not",
            "do not ",
            "don't ",
            "never ",
        ],
    ) {
        MemoryType::Preference
    } else if contains_any(
        &normalized,
        &[
            "procedure:",
            "steps:",
            "runbook",
            "workflow:",
            "how to ",
            "to reproduce",
        ],
    ) {
        MemoryType::Procedure
    } else if contains_any(
        &normalized,
        &[
            "decision:",
            "decided to",
            "we decided",
            "we will use",
            "we'll use",
            "let's use",
            "switch to",
            "selected ",
            "chosen ",
            "proceed with",
        ],
    ) {
        MemoryType::Decision
    } else {
        MemoryType::Episode
    };
    let trust = match kind {
        "user" => 1.0,
        "system" | "tool_result" | "bash" => 0.95,
        "assistant" => 0.6,
        _ => 0.5,
    };
    let confidence = match memory_type {
        MemoryType::Fact => 0.99,
        MemoryType::Decision | MemoryType::Preference | MemoryType::Procedure => 0.9,
        MemoryType::Episode => 0.7,
    };
    let scope = match memory_type {
        MemoryType::Decision | MemoryType::Preference | MemoryType::Procedure => {
            MemoryScope::Project
        }
        MemoryType::Fact | MemoryType::Episode => MemoryScope::Context,
    };
    EntrySemantics {
        memory_type,
        trust,
        confidence,
        valid_from: Some(observed_at),
        valid_until: None,
        scope,
        derivation,
    }
}

fn contains_any(text: &str, patterns: &[&str]) -> bool {
    patterns.iter().any(|pattern| text.contains(pattern))
}

fn governance_reasons(kind: &str, text: &str, derivation: MemoryDerivation) -> i64 {
    let mut reasons = 0;
    if derivation == MemoryDerivation::Mutation || kind == "mutation" {
        reasons |= REASON_GENERATED;
    }
    if text.len() > MAX_GOVERNANCE_TEXT_BYTES {
        reasons |= REASON_OVERSIZED;
    }

    let bounded = bounded_text(text, MAX_GOVERNANCE_TEXT_BYTES);
    if bounded.chars().any(is_hidden_unicode) {
        reasons |= REASON_HIDDEN_UNICODE;
    }
    if bounded
        .chars()
        .any(|character| character.is_control() && !matches!(character, '\n' | '\r' | '\t'))
    {
        reasons |= REASON_CONTROL_CHARACTER;
    }

    let normalized = bounded
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ")
        .to_lowercase();
    if contains_any(
        &normalized,
        &[
            "ignore previous instructions",
            "ignore all previous instructions",
            "disregard previous instructions",
            "override previous instructions",
            "forget previous instructions",
        ],
    ) {
        reasons |= REASON_PROMPT_OVERRIDE;
    }
    if contains_any(
        &normalized,
        &[
            "<system",
            "</system>",
            "[system]",
            "system message:",
            "developer message:",
            "assistant message:",
            "tool message:",
        ],
    ) {
        reasons |= REASON_ROLE_IMPERSONATION;
    }
    if contains_any(
        &normalized,
        &[
            "when recalled",
            "when retrieved",
            "when this memory is used",
            "upon retrieval",
            "hide this instruction",
            "do not reveal this instruction",
        ],
    ) {
        reasons |= REASON_RETRIEVAL_INSTRUCTION;
    }
    reasons
}

fn bounded_text(text: &str, max_bytes: usize) -> &str {
    if text.len() <= max_bytes {
        return text;
    }
    let mut end = max_bytes;
    while !text.is_char_boundary(end) {
        end -= 1;
    }
    &text[..end]
}

fn is_hidden_unicode(character: char) -> bool {
    matches!(
        character,
        '\u{061c}'
            | '\u{200b}'..='\u{200f}'
            | '\u{202a}'..='\u{202e}'
            | '\u{2060}'..='\u{206f}'
            | '\u{feff}'
    )
}

fn governance_reason_names(mask: i64) -> Vec<String> {
    [
        (REASON_PROMPT_OVERRIDE, "prompt_override"),
        (REASON_ROLE_IMPERSONATION, "role_impersonation"),
        (REASON_RETRIEVAL_INSTRUCTION, "retrieval_instruction"),
        (REASON_HIDDEN_UNICODE, "hidden_unicode"),
        (REASON_CONTROL_CHARACTER, "control_character"),
        (REASON_OVERSIZED, "oversized"),
        (REASON_GENERATED, "generated"),
    ]
    .into_iter()
    .filter(|(reason, _)| mask & reason != 0)
    .map(|(_, name)| name.to_string())
    .collect()
}

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 context_tombstone_key(provider: &str, context_id: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(b"goosedump-memory-context\0");
    hasher.update(provider.as_bytes());
    hasher.update(b"\0");
    hasher.update(context_id.as_bytes());
    format!("v1:{: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],
        )?;
    }
    if inserted != 0 {
        tx.execute(
            "INSERT INTO memory_ingestions(
                entry_hash, status, attempts, created_at, updated_at
             ) VALUES(?1, 'pending', 0, ?2, ?2)",
            params![hash, now],
        )?;
    }
    u64::try_from(inserted).context("inserted entry count")
}

fn upsert_entry_semantics(
    tx: &Transaction<'_>,
    hash: &str,
    semantics: EntrySemantics,
) -> anyhow::Result<()> {
    tx.execute(
        "INSERT INTO memory_semantics(
            entry_hash, memory_type, trust, confidence, valid_from, valid_until,
            scope, derivation, classified_at
         ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
         ON CONFLICT(entry_hash) DO UPDATE SET
            valid_from = CASE
                WHEN memory_semantics.valid_from IS NULL THEN excluded.valid_from
                WHEN excluded.valid_from IS NULL THEN memory_semantics.valid_from
                ELSE min(memory_semantics.valid_from, excluded.valid_from)
            END",
        params![
            hash,
            semantics.memory_type.as_str(),
            semantics.trust,
            semantics.confidence,
            semantics.valid_from,
            semantics.valid_until,
            semantics.scope.as_str(),
            semantics.derivation.as_str(),
            now_millis(),
        ],
    )?;
    Ok(())
}

fn upsert_governance(
    tx: &Transaction<'_>,
    hash: &str,
    reason_mask: i64,
) -> anyhow::Result<GovernanceRecord> {
    tx.execute(
        "INSERT INTO memory_governance(
            entry_hash, reason_mask, override_reason_mask, manual_quarantine,
            policy_version, screened_at
         ) VALUES(?1, ?2, 0, 0, ?3, ?4)
         ON CONFLICT(entry_hash) DO UPDATE SET
            reason_mask = excluded.reason_mask,
            policy_version = excluded.policy_version,
            screened_at = excluded.screened_at",
        params![hash, reason_mask, GOVERNANCE_POLICY_VERSION, now_millis()],
    )?;
    governance_record(tx, hash)
}

fn governance_record(tx: &Transaction<'_>, hash: &str) -> anyhow::Result<GovernanceRecord> {
    Ok(tx.query_row(
        "SELECT reason_mask, override_reason_mask, manual_quarantine, policy_version
         FROM memory_governance WHERE entry_hash = ?1",
        params![hash],
        |row| {
            Ok(GovernanceRecord {
                reason_mask: row.get(0)?,
                override_reason_mask: row.get(1)?,
                manual_quarantine: row.get(2)?,
                policy_version: row.get(3)?,
            })
        },
    )?)
}

fn governance_record_reasons(record: GovernanceRecord) -> Vec<String> {
    let mut reasons = governance_reason_names(record.reason_mask & !record.override_reason_mask);
    if record.manual_quarantine {
        reasons.push("manual_quarantine".to_string());
    }
    if record.policy_version != GOVERNANCE_POLICY_VERSION {
        reasons.push("policy_stale".to_string());
    }
    reasons
}

fn require_entry(tx: &Transaction<'_>, hash: &str) -> anyhow::Result<()> {
    let exists = tx.query_row(
        "SELECT EXISTS(SELECT 1 FROM entries WHERE hash = ?1)",
        params![hash],
        |row| row.get::<_, bool>(0),
    )?;
    if !exists {
        bail!("memory entry not found: {hash}");
    }
    Ok(())
}

fn governance_status(conn: &Connection, hash: &str) -> anyhow::Result<GovernanceStatus> {
    let status = conn.query_row(
        "SELECT CASE
            WHEN EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'contradicts'
                  AND (relation.source_hash = governance.entry_hash
                       OR relation.target_hash = governance.entry_hash)
              ) THEN 'contradicted'
            WHEN EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'supersedes'
                  AND relation.target_hash = governance.entry_hash
              ) THEN 'superseded'
            WHEN semantics.valid_until IS NOT NULL AND semantics.valid_until <= ?3
              THEN 'expired'
            WHEN governance.manual_quarantine != 0
              OR (governance.reason_mask & ~governance.override_reason_mask) != 0
              OR governance.policy_version != ?2
              THEN 'quarantined'
            ELSE 'active'
         END
         FROM memory_governance AS governance
         JOIN memory_semantics AS semantics
           ON semantics.entry_hash = governance.entry_hash
         WHERE governance.entry_hash = ?1",
        params![hash, GOVERNANCE_POLICY_VERSION, now_millis()],
        |row| row.get::<_, String>(0),
    )?;
    status
        .parse::<GovernanceStatus>()
        .map_err(|error| anyhow::anyhow!(error))
}

fn governance_report(
    tx: &Transaction<'_>,
    action: &str,
    hash: &str,
    related_hash: Option<String>,
) -> anyhow::Result<GovernanceReport> {
    let record = governance_record(tx, hash)?;
    Ok(GovernanceReport {
        action: action.to_string(),
        hash: hash.to_string(),
        related_hash,
        status: governance_status(tx, hash)?,
        reasons: governance_record_reasons(record),
    })
}

fn query_ingestion_entries(
    conn: &Connection,
    status: Option<IngestionStatus>,
    limit: i64,
) -> anyhow::Result<Vec<IngestionEntry>> {
    let status = status.map(IngestionStatus::as_str);
    let mut stmt = conn.prepare(
        "SELECT ingestion.entry_hash, entries.kind, entries.text, ingestion.status,
                ingestion.attempts, ingestion.last_error, ingestion.updated_at,
                ingestion.indexed_at
         FROM memory_ingestions AS ingestion
         JOIN entries ON entries.hash = ingestion.entry_hash
         WHERE (?1 IS NULL OR ingestion.status = ?1)
         ORDER BY CASE ingestion.status
                    WHEN 'failed' THEN 0
                    WHEN 'processing' THEN 1
                    WHEN 'pending' THEN 2
                    ELSE 3
                  END,
                  ingestion.updated_at DESC, ingestion.entry_hash
         LIMIT ?2",
    )?;
    stmt.query_map(params![status, limit], map_ingestion_entry)?
        .collect::<rusqlite::Result<Vec<_>>>()
        .map_err(Into::into)
}

fn query_ingestion_entry(conn: &Connection, hash: &str) -> anyhow::Result<IngestionEntry> {
    Ok(conn.query_row(
        "SELECT ingestion.entry_hash, entries.kind, entries.text, ingestion.status,
                ingestion.attempts, ingestion.last_error, ingestion.updated_at,
                ingestion.indexed_at
         FROM memory_ingestions AS ingestion
         JOIN entries ON entries.hash = ingestion.entry_hash
         WHERE ingestion.entry_hash = ?1",
        params![hash],
        map_ingestion_entry,
    )?)
}

fn map_ingestion_entry(row: &rusqlite::Row<'_>) -> rusqlite::Result<IngestionEntry> {
    let status = row
        .get::<_, String>(3)?
        .parse()
        .map_err(|message| semantic_conversion_error(3, message))?;
    let attempts = row.get::<_, i64>(4)?;
    let attempts = u64::try_from(attempts)
        .map_err(|_| semantic_conversion_error(4, "negative ingestion attempt count"))?;
    Ok(IngestionEntry {
        hash: row.get(0)?,
        kind: row.get(1)?,
        text: row.get(2)?,
        status,
        attempts,
        last_error: row.get(5)?,
        updated_at: row.get(6)?,
        indexed_at: row.get(7)?,
    })
}

fn query_governance_entries(
    conn: &Connection,
    status: Option<GovernanceStatus>,
    limit: i64,
) -> anyhow::Result<Vec<GovernanceEntry>> {
    let status = status.map(GovernanceStatus::as_str);
    let mut stmt = conn.prepare(
        "WITH classified AS (
            SELECT entries.hash, entries.kind, entries.text, semantics.memory_type,
                   governance.reason_mask, governance.override_reason_mask,
                   governance.manual_quarantine, governance.policy_version,
                   CASE
                     WHEN EXISTS(
                         SELECT 1 FROM memory_relations AS relation
                         WHERE relation.relation = 'contradicts'
                           AND (relation.source_hash = entries.hash
                                OR relation.target_hash = entries.hash)
                       ) THEN 'contradicted'
                     WHEN EXISTS(
                         SELECT 1 FROM memory_relations AS relation
                         WHERE relation.relation = 'supersedes'
                           AND relation.target_hash = entries.hash
                       ) THEN 'superseded'
                     WHEN semantics.valid_until IS NOT NULL AND semantics.valid_until <= ?3
                       THEN 'expired'
                     WHEN governance.manual_quarantine != 0
                       OR (governance.reason_mask & ~governance.override_reason_mask) != 0
                       OR governance.policy_version != ?2
                       THEN 'quarantined'
                     ELSE 'active'
                   END AS status
            FROM entries
            JOIN memory_semantics AS semantics ON semantics.entry_hash = entries.hash
            JOIN memory_governance AS governance ON governance.entry_hash = entries.hash
         )
         SELECT classified.hash, classified.kind, classified.text,
                classified.memory_type, classified.status,
                classified.reason_mask, classified.override_reason_mask,
                classified.manual_quarantine, classified.policy_version,
                latest.provider, latest.context_id, latest.entry_id, latest.path
         FROM classified
         LEFT JOIN sightings AS latest ON latest.id = (
            SELECT sighting.id FROM sightings AS sighting
            WHERE sighting.entry_hash = classified.hash
            ORDER BY sighting.observed_at DESC, sighting.id DESC LIMIT 1
         )
         WHERE (?1 IS NULL OR classified.status = ?1)
         ORDER BY CASE classified.status
                    WHEN 'quarantined' THEN 0
                    WHEN 'contradicted' THEN 1
                    WHEN 'superseded' THEN 2
                    WHEN 'expired' THEN 3
                    ELSE 4
                  END,
                  classified.hash
         LIMIT ?4",
    )?;
    let rows = stmt.query_map(
        params![status, GOVERNANCE_POLICY_VERSION, now_millis(), limit],
        |row| {
            let memory_type = row
                .get::<_, String>(3)?
                .parse::<MemoryType>()
                .map_err(|error| semantic_conversion_error(3, error))?;
            let status = row
                .get::<_, String>(4)?
                .parse::<GovernanceStatus>()
                .map_err(|error| semantic_conversion_error(4, error))?;
            let record = GovernanceRecord {
                reason_mask: row.get(5)?,
                override_reason_mask: row.get(6)?,
                manual_quarantine: row.get(7)?,
                policy_version: row.get(8)?,
            };
            let provider = row
                .get::<_, Option<String>>(9)?
                .map(|value| {
                    value.parse::<Client>().map_err(|error| {
                        rusqlite::Error::FromSqlConversionFailure(
                            9,
                            rusqlite::types::Type::Text,
                            std::io::Error::other(error).into(),
                        )
                    })
                })
                .transpose()?;
            Ok(GovernanceEntry {
                hash: row.get(0)?,
                kind: row.get(1)?,
                text: text::clip(&row.get::<_, String>(2)?, MAX_GOVERNANCE_TEXT_BYTES),
                memory_type,
                status,
                reasons: governance_record_reasons(record),
                provider,
                context_id: row.get(10)?,
                entry_id: row.get(11)?,
                path: row.get::<_, Option<String>>(12)?.map(PathBuf::from),
            })
        },
    )?;
    rows.collect::<rusqlite::Result<Vec<_>>>()
        .map_err(Into::into)
}

fn entry_is_retrievable(tx: &Transaction<'_>, hash: &str, at: i64) -> anyhow::Result<bool> {
    Ok(tx.query_row(
        "SELECT EXISTS(
            SELECT 1
            FROM memory_governance AS governance
            JOIN memory_semantics AS semantics
              ON semantics.entry_hash = governance.entry_hash
            WHERE governance.entry_hash = ?1
              AND governance.manual_quarantine = 0
              AND (governance.reason_mask & ~governance.override_reason_mask) = 0
              AND governance.policy_version = ?2
              AND (semantics.valid_until IS NULL OR semantics.valid_until > ?3)
              AND NOT EXISTS(
                 SELECT 1 FROM memory_relations AS relation
                 WHERE relation.relation = 'supersedes'
                   AND relation.target_hash = governance.entry_hash
              )
              AND NOT EXISTS(
                 SELECT 1 FROM memory_relations AS relation
                 WHERE relation.relation = 'contradicts'
                   AND (relation.source_hash = governance.entry_hash
                        OR relation.target_hash = governance.entry_hash)
              )
         )",
        params![hash, GOVERNANCE_POLICY_VERSION, at],
        |row| row.get(0),
    )?)
}

fn sync_governed_indexes(tx: &Transaction<'_>, hash: &str) -> anyhow::Result<bool> {
    let now = now_millis();
    let eligible = entry_is_retrievable(tx, hash, now)?;
    if eligible {
        tx.execute(
            "INSERT INTO entries_fts(hash, kind, text)
             SELECT hash, kind, text FROM entries
             WHERE hash = ?1
               AND NOT EXISTS(SELECT 1 FROM entries_fts WHERE hash = ?1)",
            params![hash],
        )?;
        tx.execute(
            "UPDATE memory_ingestions
             SET status = 'pending', attempts = 0, last_error = NULL,
                 updated_at = ?2, indexed_at = NULL
             WHERE entry_hash = ?1 AND status = 'indexed'
               AND NOT EXISTS(SELECT 1 FROM entries_vec WHERE hash = ?1)",
            params![hash, now],
        )?;
    } else {
        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 memory_archive WHERE entry_hash = ?1",
            params![hash],
        )?;
    }
    Ok(eligible)
}

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 roots = {
        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<_>>>()?
    };
    let mut hashes = HashSet::new();
    for root in roots {
        hashes.extend(derived_hashes(tx, &root)?);
    }
    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 derived_hashes(tx: &Transaction<'_>, hash: &str) -> anyhow::Result<Vec<String>> {
    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",
    )?;
    Ok(stmt
        .query_map(params![hash], |row| row.get::<_, String>(0))?
        .collect::<rusqlite::Result<Vec<_>>>()?)
}

fn query_recall(
    conn: &Connection,
    query: &str,
    provider: Option<&str>,
    path: Option<&str>,
    memory_type: Option<MemoryType>,
    limit: i64,
) -> anyhow::Result<Vec<RecallRow>> {
    let memory_type = memory_type.map(MemoryType::as_str);
    let mut stmt = conn.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),
            meta.memory_type, meta.trust, meta.confidence, meta.valid_from,
            meta.valid_until, meta.scope, meta.derivation,
            governance.reason_mask, governance.override_reason_mask,
            governance.policy_version
         FROM entries_fts AS f
         JOIN sightings AS s ON s.entry_hash = f.hash
         JOIN memory_semantics AS meta ON meta.entry_hash = f.hash
         JOIN memory_governance AS governance ON governance.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 coalesce(source.provider, s.provider) = ?2)
           AND (?3 IS NULL OR coalesce(source.path, s.path) = ?3)
           AND (?4 IS NULL OR meta.memory_type = ?4)
           AND governance.manual_quarantine = 0
           AND (governance.reason_mask & ~governance.override_reason_mask) = 0
           AND governance.policy_version = ?5
           AND (meta.valid_until IS NULL OR meta.valid_until > ?6)
           AND NOT EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'supersedes'
                  AND relation.target_hash = f.hash
           )
           AND NOT EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'contradicts'
                  AND (relation.source_hash = f.hash OR relation.target_hash = f.hash)
           )
           AND s.id = (
                SELECT latest.id FROM sightings AS latest
                WHERE latest.entry_hash = f.hash
                  AND (m.hash IS NOT NULL OR ?2 IS NULL OR latest.provider = ?2)
                  AND (m.hash IS NOT NULL OR ?3 IS NULL OR latest.path = ?3)
                ORDER BY latest.observed_at DESC, latest.id DESC LIMIT 1
           )
           AND (m.hash IS NULL OR m.source_sighting_id = (
                SELECT m2.source_sighting_id
                FROM memory_mutations AS m2
                JOIN sightings AS source2 ON source2.id = m2.source_sighting_id
                WHERE m2.hash = f.hash
                  AND (?2 IS NULL OR source2.provider = ?2)
                  AND (?3 IS NULL OR source2.path = ?3)
                ORDER BY source2.observed_at DESC, source2.id DESC, m2.source_hash
                LIMIT 1
           ))
         ORDER BY bm25(entries_fts, 0.0, 0.2, 1.0),
                  coalesce(source.observed_at, s.observed_at) DESC,
                  coalesce(source.id, s.id)
         LIMIT ?7",
    )?;
    let mapped = stmt.query_map(
        params![
            query,
            provider,
            path,
            memory_type,
            GOVERNANCE_POLICY_VERSION,
            now_millis(),
            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())
    })?;
    let memory_type = row
        .get::<_, String>(11)?
        .parse::<MemoryType>()
        .map_err(|error| semantic_conversion_error(11, error))?;
    let scope = row
        .get::<_, String>(16)?
        .parse::<MemoryScope>()
        .map_err(|error| semantic_conversion_error(16, error))?;
    let derivation = row
        .get::<_, String>(17)?
        .parse::<MemoryDerivation>()
        .map_err(|error| semantic_conversion_error(17, error))?;
    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)?),
        memory_type,
        trust: row.get(12)?,
        confidence: row.get(13)?,
        valid_from: row.get(14)?,
        valid_until: row.get(15)?,
        scope,
        derivation,
        reason_mask: row.get(18)?,
        override_reason_mask: row.get(19)?,
        policy_version: row.get(20)?,
    })
}

fn semantic_conversion_error(column: usize, message: &'static str) -> rusqlite::Error {
    rusqlite::Error::FromSqlConversionFailure(
        column,
        rusqlite::types::Type::Text,
        std::io::Error::other(message).into(),
    )
}

const VECTOR_RECALL_QUERY: &str = "WITH eligible(hash) AS (
        SELECT direct.entry_hash
        FROM sightings AS direct
        JOIN memory_semantics AS direct_meta ON direct_meta.entry_hash = direct.entry_hash
        JOIN memory_governance AS direct_governance
          ON direct_governance.entry_hash = direct.entry_hash
        WHERE NOT EXISTS(
                SELECT 1 FROM memory_mutations AS mutation
                WHERE mutation.hash = direct.entry_hash
              )
          AND (?2 IS NULL OR direct.provider = ?2)
          AND (?3 IS NULL OR direct.path = ?3)
          AND (?4 IS NULL OR direct_meta.memory_type = ?4)
          AND direct_governance.manual_quarantine = 0
          AND (direct_governance.reason_mask & ~direct_governance.override_reason_mask) = 0
          AND direct_governance.policy_version = ?5
          AND (direct_meta.valid_until IS NULL OR direct_meta.valid_until > ?6)
          AND NOT EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'supersedes'
                  AND relation.target_hash = direct.entry_hash
              )
          AND NOT EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'contradicts'
                  AND (relation.source_hash = direct.entry_hash
                       OR relation.target_hash = direct.entry_hash)
              )
        UNION
        SELECT mutation.hash
        FROM memory_mutations AS mutation
        JOIN sightings AS provenance ON provenance.id = mutation.source_sighting_id
        JOIN memory_semantics AS mutation_meta ON mutation_meta.entry_hash = mutation.hash
        JOIN memory_governance AS mutation_governance
          ON mutation_governance.entry_hash = mutation.hash
        WHERE (?2 IS NULL OR provenance.provider = ?2)
          AND (?3 IS NULL OR provenance.path = ?3)
          AND (?4 IS NULL OR mutation_meta.memory_type = ?4)
          AND mutation_governance.manual_quarantine = 0
          AND (mutation_governance.reason_mask & ~mutation_governance.override_reason_mask) = 0
          AND mutation_governance.policy_version = ?5
          AND (mutation_meta.valid_until IS NULL OR mutation_meta.valid_until > ?6)
          AND NOT EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'supersedes'
                  AND relation.target_hash = mutation.hash
              )
          AND NOT EXISTS(
                SELECT 1 FROM memory_relations AS relation
                WHERE relation.relation = 'contradicts'
                  AND (relation.source_hash = mutation.hash
                       OR relation.target_hash = mutation.hash)
              )
     ),
     nearest(hash, distance) AS (
        SELECT candidate.hash, candidate.distance
        FROM entries_vec AS candidate
        WHERE candidate.embedding MATCH ?1
          AND candidate.k = ?7
          AND candidate.hash IN (SELECT hash FROM eligible)
     )
     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),
        meta.memory_type, meta.trust, meta.confidence, meta.valid_from,
        meta.valid_until, meta.scope, meta.derivation,
        governance.reason_mask, governance.override_reason_mask,
        governance.policy_version
     FROM nearest
     JOIN entries AS e ON e.hash = nearest.hash
     JOIN sightings AS s ON s.entry_hash = e.hash
     JOIN memory_semantics AS meta ON meta.entry_hash = e.hash
     JOIN memory_governance AS governance ON governance.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 (?2 IS NULL OR coalesce(source.provider, s.provider) = ?2)
       AND (?3 IS NULL OR coalesce(source.path, s.path) = ?3)
       AND s.id = (
            SELECT latest.id FROM sightings AS latest
            WHERE latest.entry_hash = e.hash
              AND (m.hash IS NOT NULL OR ?2 IS NULL OR latest.provider = ?2)
              AND (m.hash IS NOT NULL OR ?3 IS NULL OR latest.path = ?3)
            ORDER BY latest.observed_at DESC, latest.id DESC LIMIT 1
       )
       AND (m.hash IS NULL OR m.source_sighting_id = (
            SELECT m2.source_sighting_id
            FROM memory_mutations AS m2
            JOIN sightings AS source2 ON source2.id = m2.source_sighting_id
            WHERE m2.hash = e.hash
              AND (?2 IS NULL OR source2.provider = ?2)
              AND (?3 IS NULL OR source2.path = ?3)
            ORDER BY source2.observed_at DESC, source2.id DESC, m2.source_hash
            LIMIT 1
       ))
     ORDER BY nearest.distance, e.hash";

fn query_vector_recall(
    conn: &Connection,
    embedding: &[f32],
    provider: Option<&str>,
    path: Option<&str>,
    memory_type: Option<MemoryType>,
    limit: i64,
) -> anyhow::Result<Vec<RecallRow>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let memory_type = memory_type.map(MemoryType::as_str);
    let mut stmt = conn.prepare(VECTOR_RECALL_QUERY)?;
    stmt.query_map(
        params![
            embedding.as_bytes(),
            provider,
            path,
            memory_type,
            GOVERNANCE_POLICY_VERSION,
            now_millis(),
            limit
        ],
        map_recall_row,
    )?
    .collect::<rusqlite::Result<Vec<_>>>()
    .map_err(Into::into)
}

fn rescreen_recall_rows(
    conn: &mut Connection,
    mut lexical: Vec<RecallRow>,
    mut semantic: Vec<RecallRow>,
) -> anyhow::Result<(Vec<RecallRow>, Vec<RecallRow>)> {
    let mut eligibility = HashMap::new();
    let mut updates = Vec::new();
    for row in lexical.iter().chain(&semantic) {
        if eligibility.contains_key(&row.hash) {
            continue;
        }
        let current = governance_reasons(&row.kind, &row.text, row.derivation);
        let allowed = current & !row.override_reason_mask == 0;
        eligibility.insert(row.hash.clone(), allowed);
        if current != row.reason_mask || row.policy_version != GOVERNANCE_POLICY_VERSION {
            updates.push((row.hash.clone(), current));
        }
    }

    if !updates.is_empty() {
        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
        for (hash, reasons) in updates {
            upsert_governance(&tx, &hash, reasons)?;
            sync_governed_indexes(&tx, &hash)?;
        }
        rebuild_shadow_archive(&tx)?;
        tx.commit()?;
    }

    lexical.retain(|row| eligibility.get(&row.hash).copied().unwrap_or(false));
    semantic.retain(|row| eligibility.get(&row.hash).copied().unwrap_or(false));
    Ok((lexical, semantic))
}

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)
}

struct IngestionClaim {
    token: String,
    entries: Vec<(String, String)>,
    claimed: HashSet<String>,
}

fn mark_ingestion_indexed(tx: &Transaction<'_>, hash: &str, indexed_at: i64) -> anyhow::Result<()> {
    let changed = tx.execute(
        "UPDATE memory_ingestions
         SET status = 'indexed', lease_until = NULL, claim_token = NULL,
             last_error = NULL, updated_at = ?2, indexed_at = ?2
         WHERE entry_hash = ?1 AND status != 'processing'",
        params![hash, indexed_at],
    )?;
    if changed != 1 {
        bail!("mark memory ingestion {hash} indexed: state changed");
    }
    Ok(())
}

fn ingestion_claim_token(now: i64, operation: &str) -> String {
    let sequence = INGESTION_CLAIM_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    format!("{}:{now}:{sequence}:{operation}", std::process::id())
}

fn ingestion_is_blocked(conn: &Connection, hash: &str) -> anyhow::Result<bool> {
    Ok(conn.query_row(
        "SELECT status IN ('processing', 'failed')
         FROM memory_ingestions WHERE entry_hash = ?1",
        params![hash],
        |row| row.get(0),
    )?)
}

fn claim_ingestions(conn: &mut Connection, limit: usize) -> anyhow::Result<IngestionClaim> {
    let limit = i64::try_from(limit).unwrap_or(i64::MAX);
    let now = now_millis();
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    recover_stale_ingestions(&tx, now)?;
    let entries = {
        let mut stmt = tx.prepare(
            "SELECT ingestion.entry_hash, entries.text
             FROM memory_ingestions AS ingestion
             JOIN entries ON entries.hash = ingestion.entry_hash
             WHERE ingestion.status = 'pending' AND ingestion.attempts < ?2
             ORDER BY ingestion.updated_at, ingestion.entry_hash
             LIMIT ?1",
        )?;
        stmt.query_map(params![limit, MAX_INGESTION_ATTEMPTS], |row| {
            Ok((row.get(0)?, row.get(1)?))
        })?
        .collect::<rusqlite::Result<Vec<_>>>()?
    };
    if entries.is_empty() {
        tx.commit()?;
        return Ok(IngestionClaim {
            token: String::new(),
            entries,
            claimed: HashSet::new(),
        });
    }
    let token = ingestion_claim_token(now, "index");
    let lease_until = now.saturating_add(INGESTION_LEASE_MILLIS);
    for (hash, _) in &entries {
        let changed = tx.execute(
            "UPDATE memory_ingestions
             SET status = 'processing', attempts = attempts + 1,
                 lease_until = ?2, claim_token = ?3, updated_at = ?4
             WHERE entry_hash = ?1 AND status = 'pending' AND attempts < ?5",
            params![hash, lease_until, token, now, MAX_INGESTION_ATTEMPTS],
        )?;
        if changed != 1 {
            bail!("claim memory ingestion {hash}: state changed");
        }
    }
    tx.commit()?;
    let claimed = entries.iter().map(|(hash, _)| hash.clone()).collect();
    Ok(IngestionClaim {
        token,
        entries,
        claimed,
    })
}

fn claim_attribution_ingestions(
    conn: &mut Connection,
    candidates: Vec<(String, String)>,
) -> anyhow::Result<IngestionClaim> {
    let now = now_millis();
    let token = ingestion_claim_token(now, "attribution");
    let lease_until = now.saturating_add(INGESTION_LEASE_MILLIS);
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    recover_stale_ingestions(&tx, now)?;
    let mut entries = Vec::with_capacity(candidates.len());
    let mut claimed = HashSet::new();
    for (hash, text) in candidates {
        let status = tx
            .query_row(
                "SELECT status FROM memory_ingestions WHERE entry_hash = ?1",
                params![hash],
                |row| row.get::<_, String>(0),
            )
            .optional()?;
        match status.as_deref() {
            Some("indexed") => entries.push((hash, text)),
            Some("pending") => {
                let changed = tx.execute(
                    "UPDATE memory_ingestions
                     SET status = 'processing', attempts = attempts + 1,
                         lease_until = ?2, claim_token = ?3, updated_at = ?4
                     WHERE entry_hash = ?1 AND status = 'pending' AND attempts < ?5",
                    params![hash, lease_until, token, now, MAX_INGESTION_ATTEMPTS],
                )?;
                if changed != 1 {
                    bail!("claim memory attribution {hash}: state changed");
                }
                claimed.insert(hash.clone());
                entries.push((hash, text));
            }
            _ => {}
        }
    }
    tx.commit()?;
    Ok(IngestionClaim {
        token,
        entries,
        claimed,
    })
}

fn fail_ingestion_claim(conn: &mut Connection, token: &str, error: &str) -> anyhow::Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    tx.execute(
        "UPDATE memory_ingestions
         SET status = CASE WHEN attempts >= ?2 THEN 'failed' ELSE 'pending' END,
             lease_until = NULL, claim_token = NULL, last_error = ?3,
             updated_at = ?4, indexed_at = NULL
         WHERE status = 'processing' AND claim_token = ?1",
        params![
            token,
            MAX_INGESTION_ATTEMPTS,
            text::clip(error, MAX_INGESTION_ERROR_BYTES),
            now_millis()
        ],
    )?;
    tx.commit()?;
    Ok(())
}

fn record_ingestion_failure(
    conn: &mut Connection,
    token: &str,
    error: anyhow::Error,
) -> anyhow::Error {
    let detail = error.to_string();
    match fail_ingestion_claim(conn, token, &detail) {
        Ok(()) => error,
        Err(record_error) => error.context(format!(
            "also failed to record ingestion failure: {record_error}"
        )),
    }
}

fn finish_ingestion_claim(
    conn: &mut Connection,
    claim: &IngestionClaim,
    embeddings: &[Vec<f32>],
) -> anyhow::Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let indexed_at = now_millis();
    for ((hash, _), embedding) in claim.entries.iter().zip(embeddings) {
        if !claim.claimed.contains(hash) {
            bail!("finish memory ingestion {hash}: entry was not claimed");
        }
        let owned = tx.query_row(
            "SELECT EXISTS(
                SELECT 1 FROM memory_ingestions
                WHERE entry_hash = ?1 AND status = 'processing' AND claim_token = ?2
             )",
            params![hash, claim.token],
            |row| row.get::<_, bool>(0),
        )?;
        if !owned {
            bail!("finish memory ingestion {hash}: claim expired");
        }
        insert_embedding(&tx, hash, embedding)?;
        tx.execute(
            "UPDATE memory_ingestions
             SET status = 'indexed', lease_until = NULL, claim_token = NULL,
                 last_error = NULL, updated_at = ?2, indexed_at = ?2
             WHERE entry_hash = ?1 AND claim_token = ?3",
            params![hash, indexed_at, claim.token],
        )?;
    }
    rebuild_shadow_archive(&tx)?;
    record_stage2_snapshot(&tx)?;
    tx.commit()?;
    Ok(())
}

fn finish_attribution_claim(
    conn: &mut Connection,
    claim: &IngestionClaim,
    embeddings: &[Vec<f32>],
) -> anyhow::Result<usize> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let indexed_at = now_millis();
    let summary_embedding = &embeddings[0];
    let mut changed = 0;
    for ((hash, _), embedding) in claim.entries.iter().zip(&embeddings[1..]) {
        let expected_status = if claim.claimed.contains(hash) {
            "processing"
        } else {
            "indexed"
        };
        let owned = tx.query_row(
            "SELECT status = ?2 AND (claim_token = ?3 OR ?2 = 'indexed')
             FROM memory_ingestions WHERE entry_hash = ?1",
            params![hash, expected_status, claim.token],
            |row| row.get::<_, bool>(0),
        )?;
        if !owned {
            bail!("finish memory attribution {hash}: state changed");
        }
        insert_embedding(&tx, hash, embedding)?;
        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],
        )?;
        if claim.claimed.contains(hash) {
            tx.execute(
                "UPDATE memory_ingestions
                 SET status = 'indexed', lease_until = NULL, claim_token = NULL,
                     last_error = NULL, updated_at = ?2, indexed_at = ?2
                 WHERE entry_hash = ?1 AND claim_token = ?3",
                params![hash, indexed_at, claim.token],
            )?;
        }
    }
    rebuild_shadow_archive(&tx)?;
    record_stage2_snapshot(&tx)?;
    tx.commit()?;
    Ok(changed)
}

fn embedding_text(value: &str) -> String {
    text::clip(value, MAX_EMBEDDING_TEXT_BYTES)
}

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)?;
    if !entry_is_retrievable(tx, hash, now_millis())? {
        tx.execute("DELETE FROM entries_vec WHERE hash = ?1", params![hash])?;
        return Ok(());
    }
    tx.execute(
        "INSERT INTO entries_vec(hash, embedding)
         SELECT ?1, ?2
         WHERE NOT EXISTS(SELECT 1 FROM entries_vec WHERE hash = ?1)",
        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 rebuild_shadow_archive(tx: &Transaction<'_>) -> anyhow::Result<usize> {
    let rebuilt_at = now_millis();
    let candidates = archive_candidates(tx, 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))
    });
    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,
            ],
        )?;
    }
    Ok(elites.len())
}

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
         JOIN memory_governance AS governance ON governance.entry_hash = e.hash
         JOIN memory_semantics AS semantics ON semantics.entry_hash = e.hash
         WHERE EXISTS(SELECT 1 FROM sightings AS live WHERE live.entry_hash = e.hash)
           AND governance.manual_quarantine = 0
           AND (governance.reason_mask & ~governance.override_reason_mask) = 0
           AND governance.policy_version = ?1
           AND (semantics.valid_until IS NULL OR semantics.valid_until > ?2)
           AND NOT EXISTS(
                 SELECT 1 FROM memory_relations AS relation
                 WHERE relation.relation = 'supersedes'
                   AND relation.target_hash = e.hash
               )
           AND NOT EXISTS(
                 SELECT 1 FROM memory_relations AS relation
                 WHERE relation.relation = 'contradicts'
                   AND (relation.source_hash = e.hash OR relation.target_hash = e.hash)
               )
         ORDER BY e.hash",
    )?;
    let mapped = stmt.query_map(params![GOVERNANCE_POLICY_VERSION, rebuilt_at], |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, v.embedding
         FROM memory_archive AS a
         JOIN entries AS e ON e.hash = a.entry_hash
         JOIN entries_vec AS v ON v.hash = e.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",
    )?;
    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(),
            )
        })?;
        let bytes = row.get::<_, Vec<u8>>(8)?;
        if bytes.len() != EMBEDDING_DIMENSIONS * size_of::<f32>() {
            return Err(rusqlite::Error::FromSqlConversionFailure(
                8,
                rusqlite::types::Type::Blob,
                std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid embedding size")
                    .into(),
            ));
        }
        let embedding = bytes
            .chunks_exact(size_of::<f32>())
            .map(|chunk| f32::from_ne_bytes(chunk.try_into().expect("four-byte slice")))
            .collect();
        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)?,
            embedding,
        })
    })?;
    let mut parents = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    if parents.len() < 2 {
        return Ok(parents);
    }
    let left = parents.remove(0);
    parents.retain(|candidate| same_mutation_scope(&left, candidate));
    if parents.is_empty() {
        return Ok(vec![left]);
    }
    let right_index = parents
        .iter()
        .enumerate()
        .max_by(
            |(left_index, left_candidate), (right_index, right_candidate)| {
                cosine_similarity(&left.embedding, &left_candidate.embedding)
                    .total_cmp(&cosine_similarity(
                        &left.embedding,
                        &right_candidate.embedding,
                    ))
                    .then_with(|| right_index.cmp(left_index))
            },
        )
        .map_or(0, |(index, _)| index);
    let right = parents.remove(right_index);
    Ok(vec![left, right])
}

fn same_mutation_scope(left: &MutationParent, candidate: &MutationParent) -> bool {
    candidate.provider == left.provider
        && if left.path.is_empty() || candidate.path.is_empty() {
            candidate.context_id == left.context_id
        } else {
            candidate.path == left.path
        }
}

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 memory_type_counts(conn: &Connection) -> anyhow::Result<MemoryTypeCounts> {
    Ok(MemoryTypeCounts {
        decisions: count(
            conn,
            "SELECT count(*) FROM memory_semantics WHERE memory_type = 'decision'",
        )?,
        facts: count(
            conn,
            "SELECT count(*) FROM memory_semantics WHERE memory_type = 'fact'",
        )?,
        preferences: count(
            conn,
            "SELECT count(*) FROM memory_semantics WHERE memory_type = 'preference'",
        )?,
        procedures: count(
            conn,
            "SELECT count(*) FROM memory_semantics WHERE memory_type = 'procedure'",
        )?,
        episodes: count(
            conn,
            "SELECT count(*) FROM memory_semantics WHERE memory_type = 'episode'",
        )?,
    })
}

fn ingestion_counts(conn: &Connection) -> anyhow::Result<IngestionCounts> {
    let (pending, processing, indexed, failed) = conn.query_row(
        "SELECT
            coalesce(sum(status = 'pending'), 0),
            coalesce(sum(status = 'processing'), 0),
            coalesce(sum(status = 'indexed'), 0),
            coalesce(sum(status = 'failed'), 0)
         FROM memory_ingestions",
        [],
        |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, i64>(2)?,
                row.get::<_, i64>(3)?,
            ))
        },
    )?;
    Ok(IngestionCounts {
        pending: u64::try_from(pending).context("negative pending ingestion count")?,
        processing: u64::try_from(processing).context("negative processing ingestion count")?,
        indexed: u64::try_from(indexed).context("negative indexed ingestion count")?,
        failed: u64::try_from(failed).context("negative failed ingestion count")?,
    })
}

fn governance_counts(conn: &Connection) -> anyhow::Result<GovernanceCounts> {
    let (active, quarantined, expired, superseded, contradicted) = conn.query_row(
        "SELECT
            coalesce(sum(status = 'active'), 0),
            coalesce(sum(status = 'quarantined'), 0),
            coalesce(sum(status = 'expired'), 0),
            coalesce(sum(status = 'superseded'), 0),
            coalesce(sum(status = 'contradicted'), 0)
         FROM (
            SELECT CASE
                WHEN EXISTS(
                    SELECT 1 FROM memory_relations AS relation
                    WHERE relation.relation = 'contradicts'
                      AND (relation.source_hash = entries.hash
                           OR relation.target_hash = entries.hash)
                  ) THEN 'contradicted'
                WHEN EXISTS(
                    SELECT 1 FROM memory_relations AS relation
                    WHERE relation.relation = 'supersedes'
                      AND relation.target_hash = entries.hash
                  ) THEN 'superseded'
                WHEN semantics.valid_until IS NOT NULL AND semantics.valid_until <= ?2
                  THEN 'expired'
                WHEN governance.manual_quarantine != 0
                  OR (governance.reason_mask & ~governance.override_reason_mask) != 0
                  OR governance.policy_version != ?1
                  THEN 'quarantined'
                ELSE 'active'
            END AS status
            FROM entries
            JOIN memory_semantics AS semantics ON semantics.entry_hash = entries.hash
            JOIN memory_governance AS governance ON governance.entry_hash = entries.hash
         )",
        params![GOVERNANCE_POLICY_VERSION, now_millis()],
        |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, i64>(2)?,
                row.get::<_, i64>(3)?,
                row.get::<_, i64>(4)?,
            ))
        },
    )?;
    Ok(GovernanceCounts {
        active: u64::try_from(active).context("negative active count")?,
        quarantined: u64::try_from(quarantined).context("negative quarantined count")?,
        expired: u64::try_from(expired).context("negative expired count")?,
        superseded: u64::try_from(superseded).context("negative superseded count")?,
        contradicted: u64::try_from(contradicted).context("negative contradicted count")?,
        content_tombstones: count(conn, "SELECT count(*) FROM memory_content_tombstones")?,
        context_tombstones: count(conn, "SELECT count(*) FROM memory_context_tombstones")?,
    })
}

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 record_stage2_snapshot(conn: &Connection) -> anyhow::Result<()> {
    let current = Stage2Snapshot::from_connection(conn)?;
    let previous = latest_stage2_snapshot(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(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(conn, current)
}

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)
}