goosedump 0.12.5

// 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::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, params,
};
use serde::Serialize;
use sha2::{Digest as _, Sha256};

use crate::Client;
use crate::display;
use crate::index::IndexEntry;
use crate::message::{Context, ConversationMessage, MessageView};
use crate::model::TextGen;
use crate::text;

const SCHEMA_VERSION: i64 = 10;
const SCHEMA_SQL: &str = r"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,
    coverage REAL NOT NULL DEFAULT 0.0
        CHECK(coverage >= 0.0 AND coverage <= 1.0)
) 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);

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

CREATE TABLE memory_meta(
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL
) WITHOUT ROWID;

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

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;

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

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;

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

PRAGMA user_version = 10;";
const HASH_VERSION: u8 = 1;
const ARCHIVE_BINS: i64 = 16;
const RECALL_BACKFILL_BATCH: usize = 16;
const RECALL_CANDIDATE_LIMIT: usize = 64;
const COMPACTION_ATTRIBUTION_BATCH: usize = 16;
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 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 schema-validated.
    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 schema-validated.
    pub fn open_path(path: &Path) -> anyhow::Result<Self> {
        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 })
    }

    /// Mark one bounded batch of pending ingestion rows as indexed.
    ///
    /// Retrieval indexes are written synchronously at harvest time, so the
    /// only remaining work is ledger bookkeeping.
    ///
    /// # Errors
    /// Returns an error if the ingestion ledger cannot be updated.
    pub fn index_missing_embeddings(&mut self, 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 indexed = claim.entries.len();
        if let Err(error) = finish_ingestion_claim(&mut self.conn, &claim) {
            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.
    ///
    /// Coverage is the lexical overlap between the summary and each candidate
    /// entry.
    ///
    /// # Errors
    /// Returns an error if coverage cannot be persisted.
    pub fn attribute_summary(&mut self, summary: &str, context: &Context) -> 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 coverages = claim
            .entries
            .iter()
            .map(|(_, text)| lexical_coverage(summary, text))
            .collect::<Vec<_>>();
        match finish_attribution_claim(&mut self.conn, &claim, &coverages) {
            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 querying or telemetry persistence fails.
    pub fn recall_hybrid(
        &mut self,
        query: &str,
        filter: &RecallFilter,
        limit: usize,
    ) -> 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 rows = if query.trim().is_empty() || result_limit == 0 {
            Vec::new()
        } else {
            self.index_missing_embeddings(RECALL_BACKFILL_BATCH)?;
            let fts_query = fts_query(query);
            if fts_query.is_empty() {
                Vec::new()
            } else {
                let candidate_limit = i64::try_from(RECALL_CANDIDATE_LIMIT)?;
                let lexical = query_recall(
                    &self.conn,
                    &fts_query,
                    provider,
                    path.as_deref(),
                    filter.memory_type,
                    candidate_limit,
                )?;
                let lexical = rescreen_recall_rows(&mut self.conn, lexical)?;
                lexical.into_iter().take(result_limit).collect()
            }
        };

        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 textgen: TextGen) -> 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 = textgen.merge(&left.text, &right.text)?;
        let text = sanitize_mutation(&generated)?;
        let hash = hash_content("mutation", &text);
        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");
        }
        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 or the governance
    /// transaction cannot be committed.
    pub fn approve_memory(&mut self, hash: &str) -> anyhow::Result<GovernanceReport> {
        if ingestion_is_blocked(&self.conn, hash)? {
            bail!("memory ingestion {hash} is processing or failed; retry it before approval");
        }
        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()],
        )?;
        sync_governed_indexes(&tx, hash)?;
        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_fts")?,
            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_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_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,
}

#[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 initialize(conn: &mut Connection) -> anyhow::Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let version: i64 = tx.pragma_query_value(None, "user_version", |row| row.get(0))?;
    if version == 0 {
        create_initial_schema(&tx)?;
    } else if version != SCHEMA_VERSION {
        bail!(
            "memory database schema version {version} is unsupported; \
             this build expects version {SCHEMA_VERSION}. Back up and remove the DB file to reinitialize."
        );
    }
    // Recover interrupted ingestion work and reapply the current governance
    // policy on every open.
    recover_stale_ingestions(&tx, now_millis())?;
    rescreen_stale_governance(&tx)?;
    tx.commit()?;
    Ok(())
}

fn create_initial_schema(tx: &Transaction<'_>) -> anyhow::Result<()> {
    tx.execute_batch(SCHEMA_SQL)?;
    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 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],
        )?;
    } else {
        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_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(),
    )
}

fn rescreen_recall_rows(
    conn: &mut Connection,
    mut rows: Vec<RecallRow>,
) -> anyhow::Result<Vec<RecallRow>> {
    let mut eligibility = HashMap::new();
    let mut updates = Vec::new();
    for row in &rows {
        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()?;
    }

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

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) -> anyhow::Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let indexed_at = now_millis();
    for (hash, _) in &claim.entries {
        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");
        }
        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,
    coverages: &[f64],
) -> anyhow::Result<usize> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let indexed_at = now_millis();
    let mut changed = 0;
    for ((hash, _), coverage) in claim.entries.iter().zip(coverages) {
        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");
        }
        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)
}

/// Token containment fraction: how many candidate tokens are covered by the
/// summary, normalized by the smaller token set.
fn lexical_coverage(summary: &str, candidate: &str) -> f64 {
    let summary_tokens: HashSet<&str> = summary.split_whitespace().collect();
    let candidate_tokens: HashSet<&str> = candidate.split_whitespace().collect();
    if summary_tokens.is_empty() || candidate_tokens.is_empty() {
        return 0.0;
    }
    let shared = candidate_tokens.intersection(&summary_tokens).count();
    let smaller = summary_tokens.len().min(candidate_tokens.len());
    f64::from(u32::try_from(shared).unwrap_or(u32::MAX))
        / f64::from(u32::try_from(smaller).unwrap_or(u32::MAX))
}

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,
            (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 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 recurrence = row.get::<_, i64>(4)?;
        let confirmed_expands = row.get::<_, i64>(5)?;
        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;
        let hash_bytes = hash.as_bytes();
        let first = hash_bytes.first().copied().unwrap_or(0);
        let second = hash_bytes.get(1).copied().unwrap_or(0);
        Ok(ArchiveCandidate {
            hash,
            kind,
            x_bin: archive_bin(first),
            y_bin: archive_bin(second),
            quality,
            recurrence,
            confirmed_expands,
            coverage,
            recency,
            last_seen_at,
        })
    })?;
    mapped
        .collect::<rusqlite::Result<Vec<_>>>()
        .map_err(Into::into)
}

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