//! Three-tier memory store for Covenant.
//!
//! [`MemoryRecord`] values live in one of three tiers — working,
//! episodic, or long-term — backed by SQLite for persistence with an
//! in-memory implementation suitable for tests. The [`MemoryStore`]
//! trait covers recent-record reads, embedded-vector cosine search,
//! tier-scoped purge ([`MemoryStore::purge_older_than`]), bounded
//! compaction ([`MemoryStore::compact`]), and scoped repair commands
//! ([`MemoryStore::repair`]).

#![deny(unsafe_code)]

pub mod ignore;
pub use ignore::{IgnorePattern, IgnoreSet, IgnoreVerdict};

use async_trait::async_trait;
pub use covenant_types::{
    MemoryCompactionOutcome, MemoryCompactionPolicy, MemoryCompactionRequest, MemoryRepairAction,
    MemoryRepairCommand, MemoryRepairMode, MemoryRepairOutcome, MemoryRepairRequest,
};
use covenant_types::{MemoryRecord, MemoryTier};
use std::collections::BTreeSet;
use std::path::Path;
use std::sync::Mutex;
use tokio::task;
use uuid::Uuid;

#[derive(Debug, thiserror::Error)]
pub enum MemoryError {
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("sqlite: {0}")]
    Sqlite(#[from] rusqlite::Error),
    #[error("serde: {0}")]
    Serde(#[from] serde_json::Error),
    #[error("worker: {0}")]
    Worker(String),
    #[error("memory record {0} not found")]
    RecordNotFound(Uuid),
    #[error("parent mismatch for memory {id}: expected {expected:?}, actual {actual:?}")]
    ParentMismatch {
        id: Uuid,
        expected: Option<Uuid>,
        actual: Option<Uuid>,
    },
    #[error("invalid memory repair request: {0}")]
    InvalidRepair(String),
    #[error("invalid memory compaction request: {0}")]
    InvalidCompaction(String),
}

#[async_trait]
pub trait MemoryStore: Send + Sync {
    async fn put(&self, record: MemoryRecord) -> Result<(), MemoryError>;
    async fn get(&self, id: Uuid) -> Result<Option<MemoryRecord>, MemoryError>;
    async fn all(&self) -> Result<Vec<MemoryRecord>, MemoryError>;
    async fn recent(
        &self,
        tier: Option<MemoryTier>,
        limit: usize,
    ) -> Result<Vec<MemoryRecord>, MemoryError>;
    async fn delete(&self, id: Uuid) -> Result<bool, MemoryError>;
    /// Score every record's `embedding` against `query_embedding` via cosine
    /// similarity and return the top `limit`, optionally filtered by tier.
    /// Records with empty embeddings get score 0 and are returned last (or
    /// dropped, depending on the impl). `min_relevance` is an optional
    /// threshold in `[0.0, 1.0]`: when set, records whose cosine score is
    /// strictly less than the threshold are dropped before the `limit`
    /// truncation, so a high threshold can yield fewer rows than `limit`
    /// even when the unfiltered set is larger. v0 does an in-process
    /// linear scan; LanceDB / sqlite-vec arrive later.
    async fn search_similar(
        &self,
        query_embedding: Vec<f32>,
        tier: Option<MemoryTier>,
        limit: usize,
        min_relevance: Option<f32>,
    ) -> Result<Vec<MemoryRecord>, MemoryError>;
    /// Purge records whose `created_at` is strictly older than `before_ms`,
    /// optionally restricted to a tier. Returns the count deleted. Closes
    /// the spec §11 working-tier-GC pin (clear at task completion is still
    /// the long-term shape; for v0 the operator drives this on a TTL).
    async fn purge_older_than(
        &self,
        tier: Option<MemoryTier>,
        before_ms: u64,
    ) -> Result<u64, MemoryError>;
    /// Operator-controlled repair for verifier drift findings. Dry-run
    /// returns the exact before/after shape without mutating the store;
    /// apply performs the mutation only after the same checks pass.
    async fn repair(
        &self,
        request: MemoryRepairRequest,
    ) -> Result<MemoryRepairOutcome, MemoryError> {
        validate_repair_request(&request)?;
        let id = request.command.id();
        let before = self.get(id).await?.ok_or(MemoryError::RecordNotFound(id))?;
        let action = request.command.action();
        let after = plan_repair(&before, &request.command)?;
        let would_change = after.as_ref() != Some(&before);

        if request.mode == MemoryRepairMode::Apply && would_change {
            match &after {
                Some(record) => self.put(record.clone()).await?,
                None => {
                    let _ = self.delete(id).await?;
                }
            }
        }

        Ok(MemoryRepairOutcome {
            id,
            action,
            mode: request.mode,
            would_change,
            changed: request.mode == MemoryRepairMode::Apply && would_change,
            before: Some(before),
            after,
        })
    }
    /// Deterministic memory compaction. Dry-run and apply compute the
    /// same plan over a single snapshot: expired working/episodic records
    /// are deleted, old long-term records are marked stale instead of
    /// deleted, and parent references are detached when their target is
    /// absent or deleted by the same plan.
    async fn compact(
        &self,
        request: MemoryCompactionRequest,
    ) -> Result<MemoryCompactionOutcome, MemoryError> {
        validate_compaction_request(&request)?;
        let records = self.all().await?;
        let (outcome, updates) = plan_compaction(&records, &request);

        if request.mode == MemoryRepairMode::Apply && outcome.would_change {
            for id in &outcome.deleted {
                let _ = self.delete(*id).await?;
            }
            for record in updates {
                self.put(record).await?;
            }
        }

        Ok(outcome)
    }
}

fn validate_repair_request(request: &MemoryRepairRequest) -> Result<(), MemoryError> {
    if request.reason.trim().is_empty() {
        return Err(MemoryError::InvalidRepair(
            "reason must not be empty".into(),
        ));
    }
    if let MemoryRepairCommand::BackfillProvenance { provenance, .. } = &request.command {
        if provenance.is_null() {
            return Err(MemoryError::InvalidRepair(
                "provenance must not be null".into(),
            ));
        }
    }
    Ok(())
}

fn validate_compaction_request(request: &MemoryCompactionRequest) -> Result<(), MemoryError> {
    if request.reason.trim().is_empty() {
        return Err(MemoryError::InvalidCompaction(
            "reason must not be empty".into(),
        ));
    }
    if request.policy.is_empty() {
        return Err(MemoryError::InvalidCompaction(
            "policy must enable at least one compaction action".into(),
        ));
    }
    Ok(())
}

fn plan_repair(
    record: &MemoryRecord,
    command: &MemoryRepairCommand,
) -> Result<Option<MemoryRecord>, MemoryError> {
    match command {
        MemoryRepairCommand::DetachParent {
            expected_parent, ..
        } => {
            if expected_parent.is_some() && record.parent != *expected_parent {
                return Err(MemoryError::ParentMismatch {
                    id: record.id,
                    expected: *expected_parent,
                    actual: record.parent,
                });
            }
            let mut after = record.clone();
            after.parent = None;
            Ok(Some(after))
        }
        MemoryRepairCommand::DeleteRecord { .. } => Ok(None),
        MemoryRepairCommand::BackfillProvenance { provenance, .. } => {
            let mut after = record.clone();
            let mut metadata = match after.metadata {
                serde_json::Value::Object(map) => map,
                other => {
                    let mut map = serde_json::Map::new();
                    map.insert("previous_metadata".into(), other);
                    map
                }
            };
            metadata.insert("provenance".into(), provenance.clone());
            after.metadata = serde_json::Value::Object(metadata);
            Ok(Some(after))
        }
    }
}

fn plan_compaction(
    records: &[MemoryRecord],
    request: &MemoryCompactionRequest,
) -> (MemoryCompactionOutcome, Vec<MemoryRecord>) {
    let mut deleted: BTreeSet<Uuid> = BTreeSet::new();
    for record in records {
        match record.tier {
            MemoryTier::Working
                if request
                    .policy
                    .delete_working_before_ms
                    .is_some_and(|before| record.created_at < before) =>
            {
                deleted.insert(record.id);
            }
            MemoryTier::Episodic
                if request
                    .policy
                    .delete_episodic_before_ms
                    .is_some_and(|before| record.created_at < before) =>
            {
                deleted.insert(record.id);
            }
            _ => {}
        }
    }

    let retained: BTreeSet<Uuid> = records
        .iter()
        .filter(|record| !deleted.contains(&record.id))
        .map(|record| record.id)
        .collect();

    let mut updates = Vec::new();
    let mut stale_marked = Vec::new();
    let mut parents_detached = Vec::new();
    for record in records {
        if deleted.contains(&record.id) {
            continue;
        }

        let mut after = record.clone();
        let mut changed = false;

        if request.policy.detach_stale_parents {
            if let Some(parent) = after.parent {
                if !retained.contains(&parent) {
                    after.parent = None;
                    parents_detached.push(after.id);
                    changed = true;
                }
            }
        }

        if after.tier == MemoryTier::LongTerm {
            if let Some(before_ms) = request.policy.mark_longterm_stale_before_ms {
                if after.created_at < before_ms {
                    let marked_at_ms = request.policy.marked_at_ms.unwrap_or(before_ms);
                    let stale_context = serde_json::json!({
                        "marked_at_ms": marked_at_ms,
                        "reason": request.reason,
                    });
                    let mut metadata = match after.metadata {
                        serde_json::Value::Object(map) => map,
                        other => {
                            let mut map = serde_json::Map::new();
                            map.insert("previous_metadata".into(), other);
                            map
                        }
                    };
                    if metadata.get("stale_context") != Some(&stale_context) {
                        metadata.insert("stale_context".into(), stale_context);
                        after.metadata = serde_json::Value::Object(metadata);
                        stale_marked.push(after.id);
                        changed = true;
                    } else {
                        after.metadata = serde_json::Value::Object(metadata);
                    }
                }
            }
        }

        if changed {
            updates.push(after);
        }
    }

    let mut deleted: Vec<Uuid> = deleted.into_iter().collect();
    deleted.sort();
    stale_marked.sort();
    parents_detached.sort();
    updates.sort_by_key(|record| record.id);
    let would_change =
        !deleted.is_empty() || !stale_marked.is_empty() || !parents_detached.is_empty();
    (
        MemoryCompactionOutcome {
            mode: request.mode,
            would_change,
            changed: request.mode == MemoryRepairMode::Apply && would_change,
            deleted,
            stale_marked,
            parents_detached,
        },
        updates,
    )
}

/// Cosine similarity over two equal-length vectors. Returns 0.0 for any
/// degenerate input (mismatched length, zero norm, empty).
pub fn cosine(a: &[f32], b: &[f32]) -> f32 {
    if a.is_empty() || a.len() != b.len() {
        return 0.0;
    }
    let mut dot = 0.0_f32;
    let mut na = 0.0_f32;
    let mut nb = 0.0_f32;
    for (x, y) in a.iter().zip(b.iter()) {
        dot += x * y;
        na += x * x;
        nb += y * y;
    }
    if na == 0.0 || nb == 0.0 {
        return 0.0;
    }
    dot / (na.sqrt() * nb.sqrt())
}

/// In-memory backend (tests, defaults when persistence is disabled).
#[derive(Default)]
pub struct InMemoryStore {
    records: Mutex<Vec<MemoryRecord>>,
}

impl InMemoryStore {
    pub fn new() -> Self {
        Self::default()
    }
}

#[async_trait]
impl MemoryStore for InMemoryStore {
    async fn put(&self, record: MemoryRecord) -> Result<(), MemoryError> {
        let mut g = self
            .records
            .lock()
            .map_err(|e| MemoryError::Worker(e.to_string()))?;
        g.retain(|r| r.id != record.id);
        g.push(record);
        Ok(())
    }

    async fn get(&self, id: Uuid) -> Result<Option<MemoryRecord>, MemoryError> {
        let g = self
            .records
            .lock()
            .map_err(|e| MemoryError::Worker(e.to_string()))?;
        Ok(g.iter().find(|r| r.id == id).cloned())
    }

    async fn all(&self) -> Result<Vec<MemoryRecord>, MemoryError> {
        let g = self
            .records
            .lock()
            .map_err(|e| MemoryError::Worker(e.to_string()))?;
        Ok(g.clone())
    }

    async fn recent(
        &self,
        tier: Option<MemoryTier>,
        limit: usize,
    ) -> Result<Vec<MemoryRecord>, MemoryError> {
        let g = self
            .records
            .lock()
            .map_err(|e| MemoryError::Worker(e.to_string()))?;
        let mut out: Vec<MemoryRecord> = g
            .iter()
            .filter(|r| tier.is_none_or(|t| r.tier == t))
            .cloned()
            .collect();
        out.sort_by_key(|r| std::cmp::Reverse(r.created_at));
        out.truncate(limit);
        Ok(out)
    }

    async fn delete(&self, id: Uuid) -> Result<bool, MemoryError> {
        let mut g = self
            .records
            .lock()
            .map_err(|e| MemoryError::Worker(e.to_string()))?;
        let len_before = g.len();
        g.retain(|r| r.id != id);
        Ok(g.len() != len_before)
    }

    async fn search_similar(
        &self,
        query_embedding: Vec<f32>,
        tier: Option<MemoryTier>,
        limit: usize,
        min_relevance: Option<f32>,
    ) -> Result<Vec<MemoryRecord>, MemoryError> {
        let g = self
            .records
            .lock()
            .map_err(|e| MemoryError::Worker(e.to_string()))?;
        let floor = min_relevance.unwrap_or(0.0).max(0.0);
        let mut scored: Vec<(f32, MemoryRecord)> = g
            .iter()
            .filter(|r| tier.is_none_or(|t| r.tier == t))
            .map(|r| (cosine(&query_embedding, &r.embedding), r.clone()))
            .filter(|(s, _)| *s > 0.0 && *s >= floor)
            .collect();
        scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));
        Ok(scored.into_iter().take(limit).map(|(_, r)| r).collect())
    }

    async fn purge_older_than(
        &self,
        tier: Option<MemoryTier>,
        before_ms: u64,
    ) -> Result<u64, MemoryError> {
        let mut g = self
            .records
            .lock()
            .map_err(|e| MemoryError::Worker(e.to_string()))?;
        let len_before = g.len();
        g.retain(|r| !(r.created_at < before_ms && tier.is_none_or(|t| r.tier == t)));
        Ok((len_before - g.len()) as u64)
    }
}

/// SQLite backend. Connection is wrapped in a `Mutex` and operations run on
/// a `spawn_blocking` worker since rusqlite is sync.
pub struct SqliteStore {
    conn: std::sync::Arc<Mutex<rusqlite::Connection>>,
}

const SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS memories (
    id            TEXT PRIMARY KEY,
    tier          TEXT NOT NULL,
    owner_display TEXT NOT NULL,
    owner_pubkey  TEXT NOT NULL,
    text          TEXT NOT NULL,
    embedding     BLOB NOT NULL,
    metadata      TEXT NOT NULL,
    created_at    INTEGER NOT NULL,
    parent        TEXT
);
CREATE INDEX IF NOT EXISTS memories_tier_created_idx
    ON memories (tier, created_at DESC);
CREATE INDEX IF NOT EXISTS memories_created_idx
    ON memories (created_at DESC);
"#;

impl SqliteStore {
    pub fn open(path: &Path) -> Result<Self, MemoryError> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let conn = rusqlite::Connection::open(path)?;
        conn.execute_batch(SCHEMA)?;
        Ok(Self {
            conn: std::sync::Arc::new(Mutex::new(conn)),
        })
    }

    pub fn open_in_memory() -> Result<Self, MemoryError> {
        let conn = rusqlite::Connection::open_in_memory()?;
        conn.execute_batch(SCHEMA)?;
        Ok(Self {
            conn: std::sync::Arc::new(Mutex::new(conn)),
        })
    }

    fn tier_str(t: MemoryTier) -> &'static str {
        match t {
            MemoryTier::Working => "working",
            MemoryTier::Episodic => "episodic",
            MemoryTier::LongTerm => "longterm",
        }
    }

    fn parse_tier(s: &str) -> MemoryTier {
        match s {
            "working" => MemoryTier::Working,
            "episodic" => MemoryTier::Episodic,
            _ => MemoryTier::LongTerm,
        }
    }

    fn embedding_to_bytes(v: &[f32]) -> Vec<u8> {
        let mut out = Vec::with_capacity(v.len() * 4);
        for f in v {
            out.extend_from_slice(&f.to_le_bytes());
        }
        out
    }

    fn embedding_from_bytes(b: &[u8]) -> Vec<f32> {
        b.chunks_exact(4)
            .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
            .collect()
    }

    fn row_to_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<MemoryRecord> {
        let id_s: String = row.get(0)?;
        let tier_s: String = row.get(1)?;
        let owner_display: String = row.get(2)?;
        let owner_pubkey_s: String = row.get(3)?;
        let text: String = row.get(4)?;
        let embedding_bytes: Vec<u8> = row.get(5)?;
        let metadata_s: String = row.get(6)?;
        let created_at: i64 = row.get(7)?;
        let parent_s: Option<String> = row.get(8)?;

        let id = Uuid::parse_str(&id_s).map_err(|e| {
            rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e))
        })?;
        let pubkey_vec = bs58::decode(&owner_pubkey_s).into_vec().map_err(|e| {
            rusqlite::Error::FromSqlConversionFailure(3, rusqlite::types::Type::Text, Box::new(e))
        })?;
        if pubkey_vec.len() != 32 {
            return Err(rusqlite::Error::FromSqlConversionFailure(
                3,
                rusqlite::types::Type::Text,
                format!(
                    "owner_pubkey decoded to {} bytes, expected 32",
                    pubkey_vec.len()
                )
                .into(),
            ));
        }
        let mut pubkey = [0u8; 32];
        pubkey.copy_from_slice(&pubkey_vec);
        let metadata: serde_json::Value = serde_json::from_str(&metadata_s).map_err(|e| {
            rusqlite::Error::FromSqlConversionFailure(6, rusqlite::types::Type::Text, Box::new(e))
        })?;
        let parent = parent_s
            .as_deref()
            .map(Uuid::parse_str)
            .transpose()
            .map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    8,
                    rusqlite::types::Type::Text,
                    Box::new(e),
                )
            })?;

        Ok(MemoryRecord {
            id,
            tier: Self::parse_tier(&tier_s),
            owner: covenant_types::AgentId {
                display: owner_display,
                pubkey,
            },
            text,
            embedding: Self::embedding_from_bytes(&embedding_bytes),
            metadata,
            created_at: created_at as u64,
            parent,
        })
    }
}

#[async_trait]
impl MemoryStore for SqliteStore {
    async fn put(&self, record: MemoryRecord) -> Result<(), MemoryError> {
        let conn = self.conn.clone();
        task::spawn_blocking(move || -> Result<(), MemoryError> {
            let g = conn.lock().map_err(|e| MemoryError::Worker(e.to_string()))?;
            g.execute(
                "INSERT OR REPLACE INTO memories
                 (id, tier, owner_display, owner_pubkey, text, embedding, metadata, created_at, parent)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
                rusqlite::params![
                    record.id.to_string(),
                    SqliteStore::tier_str(record.tier),
                    record.owner.display,
                    bs58::encode(record.owner.pubkey).into_string(),
                    record.text,
                    SqliteStore::embedding_to_bytes(&record.embedding),
                    serde_json::to_string(&record.metadata)?,
                    record.created_at as i64,
                    record.parent.as_ref().map(|u| u.to_string()),
                ],
            )?;
            Ok(())
        })
        .await
        .map_err(|e| MemoryError::Worker(e.to_string()))??;
        Ok(())
    }

    async fn get(&self, id: Uuid) -> Result<Option<MemoryRecord>, MemoryError> {
        let conn = self.conn.clone();
        task::spawn_blocking(move || -> Result<Option<MemoryRecord>, MemoryError> {
            let g = conn.lock().map_err(|e| MemoryError::Worker(e.to_string()))?;
            let mut stmt = g.prepare(
                "SELECT id, tier, owner_display, owner_pubkey, text, embedding, metadata, created_at, parent
                 FROM memories WHERE id = ?1",
            )?;
            let mut rows = stmt.query(rusqlite::params![id.to_string()])?;
            if let Some(row) = rows.next()? {
                Ok(Some(SqliteStore::row_to_record(row)?))
            } else {
                Ok(None)
            }
        })
        .await
        .map_err(|e| MemoryError::Worker(e.to_string()))?
    }

    async fn all(&self) -> Result<Vec<MemoryRecord>, MemoryError> {
        let conn = self.conn.clone();
        task::spawn_blocking(move || -> Result<Vec<MemoryRecord>, MemoryError> {
            let g = conn.lock().map_err(|e| MemoryError::Worker(e.to_string()))?;
            let mut stmt = g.prepare(
                "SELECT id, tier, owner_display, owner_pubkey, text, embedding, metadata, created_at, parent
                 FROM memories ORDER BY created_at DESC",
            )?;
            let rows = stmt.query_map([], SqliteStore::row_to_record)?;
            let mut out = Vec::new();
            for r in rows {
                out.push(r?);
            }
            Ok(out)
        })
        .await
        .map_err(|e| MemoryError::Worker(e.to_string()))?
    }

    async fn recent(
        &self,
        tier: Option<MemoryTier>,
        limit: usize,
    ) -> Result<Vec<MemoryRecord>, MemoryError> {
        let conn = self.conn.clone();
        task::spawn_blocking(move || -> Result<Vec<MemoryRecord>, MemoryError> {
            let g = conn.lock().map_err(|e| MemoryError::Worker(e.to_string()))?;
            let (sql, params): (&str, Vec<rusqlite::types::Value>) = match tier {
                Some(t) => (
                    "SELECT id, tier, owner_display, owner_pubkey, text, embedding, metadata, created_at, parent
                     FROM memories WHERE tier = ?1 ORDER BY created_at DESC LIMIT ?2",
                    vec![SqliteStore::tier_str(t).to_string().into(), (limit as i64).into()],
                ),
                None => (
                    "SELECT id, tier, owner_display, owner_pubkey, text, embedding, metadata, created_at, parent
                     FROM memories ORDER BY created_at DESC LIMIT ?1",
                    vec![(limit as i64).into()],
                ),
            };
            let mut stmt = g.prepare(sql)?;
            let rows = stmt.query_map(rusqlite::params_from_iter(params), SqliteStore::row_to_record)?;
            let mut out = Vec::new();
            for r in rows {
                out.push(r?);
            }
            Ok(out)
        })
        .await
        .map_err(|e| MemoryError::Worker(e.to_string()))?
    }

    async fn delete(&self, id: Uuid) -> Result<bool, MemoryError> {
        let conn = self.conn.clone();
        task::spawn_blocking(move || -> Result<bool, MemoryError> {
            let g = conn
                .lock()
                .map_err(|e| MemoryError::Worker(e.to_string()))?;
            let n = g.execute(
                "DELETE FROM memories WHERE id = ?1",
                rusqlite::params![id.to_string()],
            )?;
            Ok(n > 0)
        })
        .await
        .map_err(|e| MemoryError::Worker(e.to_string()))?
    }

    async fn search_similar(
        &self,
        query_embedding: Vec<f32>,
        tier: Option<MemoryTier>,
        limit: usize,
        min_relevance: Option<f32>,
    ) -> Result<Vec<MemoryRecord>, MemoryError> {
        let conn = self.conn.clone();
        let floor = min_relevance.unwrap_or(0.0).max(0.0);
        task::spawn_blocking(move || -> Result<Vec<MemoryRecord>, MemoryError> {
            let g = conn
                .lock()
                .map_err(|e| MemoryError::Worker(e.to_string()))?;
            let (sql, params): (&str, Vec<rusqlite::types::Value>) = match tier {
                Some(t) => (
                    "SELECT id, tier, owner_display, owner_pubkey, text, embedding, metadata, created_at, parent
                     FROM memories WHERE tier = ?1",
                    vec![SqliteStore::tier_str(t).to_string().into()],
                ),
                None => (
                    "SELECT id, tier, owner_display, owner_pubkey, text, embedding, metadata, created_at, parent
                     FROM memories",
                    vec![],
                ),
            };
            let mut stmt = g.prepare(sql)?;
            let rows = stmt
                .query_map(rusqlite::params_from_iter(params), SqliteStore::row_to_record)?;
            let mut scored: Vec<(f32, MemoryRecord)> = Vec::new();
            for r in rows {
                let r = r?;
                let s = cosine(&query_embedding, &r.embedding);
                if s > 0.0 && s >= floor {
                    scored.push((s, r));
                }
            }
            scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));
            Ok(scored.into_iter().take(limit).map(|(_, r)| r).collect())
        })
        .await
        .map_err(|e| MemoryError::Worker(e.to_string()))?
    }

    async fn purge_older_than(
        &self,
        tier: Option<MemoryTier>,
        before_ms: u64,
    ) -> Result<u64, MemoryError> {
        let conn = self.conn.clone();
        task::spawn_blocking(move || -> Result<u64, MemoryError> {
            let g = conn
                .lock()
                .map_err(|e| MemoryError::Worker(e.to_string()))?;
            let n = match tier {
                Some(t) => g.execute(
                    "DELETE FROM memories WHERE tier = ?1 AND created_at < ?2",
                    rusqlite::params![SqliteStore::tier_str(t), before_ms as i64],
                )?,
                None => g.execute(
                    "DELETE FROM memories WHERE created_at < ?1",
                    rusqlite::params![before_ms as i64],
                )?,
            };
            Ok(n as u64)
        })
        .await
        .map_err(|e| MemoryError::Worker(e.to_string()))?
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use covenant_types::{AgentId, MemoryTier};

    fn record(id: Uuid, tier: MemoryTier, text: &str, created_at: u64) -> MemoryRecord {
        MemoryRecord {
            id,
            tier,
            owner: AgentId::new("user@local", [0u8; 32]),
            text: text.into(),
            embedding: Vec::new(),
            metadata: serde_json::json!({}),
            created_at,
            parent: None,
        }
    }

    #[tokio::test]
    async fn in_memory_store_put_replaces_record_with_same_id() {
        // covenant_memory::InMemoryStore::put runs
        // g.retain(|r| r.id != record.id) before g.push(record), making
        // put behave as an upsert keyed by record.id — a second put
        // with the same id REPLACES the previous record rather than
        // appending a duplicate. Downstream callers (memory.update,
        // memory.repair, the compaction planner) rely on id being a
        // primary key.
        //
        // in_memory_put_get_recent_delete (next test) puts two
        // DIFFERENT ids and does not exercise the upsert path. A
        // refactor that dropped the retain line during a 'simplify by
        // removing the duplicate-check' pass would silently let two
        // records with the same id coexist; get() would return one
        // non-deterministically; recent() and all() would surface
        // both copies; the memory.repair and compaction planners
        // would see phantom duplicates.
        let s = InMemoryStore::new();
        let id = Uuid::new_v4();

        s.put(record(id, MemoryTier::Working, "first", 1))
            .await
            .unwrap();
        s.put(record(id, MemoryTier::Working, "second", 2))
            .await
            .unwrap();

        let got = s
            .get(id)
            .await
            .unwrap()
            .expect("upserted record must be retrievable by id");
        assert_eq!(
            got.text, "second",
            "InMemoryStore::put must REPLACE the previous record when called \
             with the same id — a refactor that dropped the retain line under \
             the rationale that 'callers handle deduplication explicitly' \
             would silently let the first record survive and get(id) would \
             return one of the two non-deterministically; pinning the \
             second-write-wins contract anchors the v0 upsert semantics",
        );
        assert_eq!(
            got.created_at, 2,
            "the replacement record's fields must overwrite the previous record \
             verbatim — a refactor that merged fields from the previous record \
             would silently surface partial state on every memory.update flow",
        );

        let all = s.all().await.unwrap();
        assert_eq!(
            all.len(),
            1,
            "InMemoryStore::all() must return exactly one record after two puts \
             with the same id — a refactor that dropped the retain line would \
             surface BOTH records here, breaking the primary-key contract that \
             every MemoryStore caller depends on",
        );
        assert_eq!(
            all[0].text, "second",
            "the single surviving record in all() must be the second (most \
             recent) write — pinning second-write-wins identically to get()",
        );

        let recent = s.recent(None, 10).await.unwrap();
        assert_eq!(
            recent.len(),
            1,
            "InMemoryStore::recent(None, 10) must return exactly one record \
             — the upsert contract must propagate to recent() identically to \
             all(); a refactor that diverged the two views would silently \
             surface duplicate rows on operator dashboards while keeping \
             get() consistent",
        );
        assert_eq!(recent[0].text, "second");
    }

    #[tokio::test]
    async fn in_memory_put_get_recent_delete() {
        let s = InMemoryStore::new();
        let id1 = Uuid::new_v4();
        let id2 = Uuid::new_v4();
        s.put(record(id1, MemoryTier::Working, "first", 1))
            .await
            .unwrap();
        s.put(record(id2, MemoryTier::Working, "second", 2))
            .await
            .unwrap();
        assert_eq!(s.get(id1).await.unwrap().unwrap().text, "first");
        let recent = s.recent(None, 10).await.unwrap();
        assert_eq!(recent.len(), 2);
        assert_eq!(recent[0].text, "second"); // most recent first
        assert!(s.delete(id1).await.unwrap());
        assert!(s.get(id1).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn in_memory_recent_filters_by_tier() {
        let s = InMemoryStore::new();
        s.put(record(Uuid::new_v4(), MemoryTier::Working, "w", 1))
            .await
            .unwrap();
        s.put(record(Uuid::new_v4(), MemoryTier::Episodic, "e", 2))
            .await
            .unwrap();
        s.put(record(Uuid::new_v4(), MemoryTier::LongTerm, "l", 3))
            .await
            .unwrap();
        let only_episodic = s.recent(Some(MemoryTier::Episodic), 10).await.unwrap();
        assert_eq!(only_episodic.len(), 1);
        assert_eq!(only_episodic[0].text, "e");
    }

    #[tokio::test]
    async fn sqlite_roundtrip() {
        let s = SqliteStore::open_in_memory().unwrap();
        let id = Uuid::new_v4();
        let mut r = record(id, MemoryTier::Episodic, "hello", 100);
        r.embedding = vec![0.1, -0.2, 0.3];
        r.metadata = serde_json::json!({"source": "test"});
        s.put(r.clone()).await.unwrap();
        let got = s.get(id).await.unwrap().unwrap();
        assert_eq!(got.text, "hello");
        assert_eq!(got.tier, MemoryTier::Episodic);
        assert_eq!(got.embedding, vec![0.1, -0.2, 0.3]);
        assert_eq!(got.metadata, serde_json::json!({"source": "test"}));
    }

    #[tokio::test]
    async fn sqlite_recent_orders_by_created_at_desc() {
        let s = SqliteStore::open_in_memory().unwrap();
        let id_a = Uuid::new_v4();
        let id_b = Uuid::new_v4();
        let id_c = Uuid::new_v4();
        s.put(record(id_a, MemoryTier::Working, "a", 100))
            .await
            .unwrap();
        s.put(record(id_b, MemoryTier::Working, "b", 300))
            .await
            .unwrap();
        s.put(record(id_c, MemoryTier::Working, "c", 200))
            .await
            .unwrap();
        let recent = s.recent(Some(MemoryTier::Working), 2).await.unwrap();
        assert_eq!(recent.len(), 2);
        assert_eq!(recent[0].text, "b");
        assert_eq!(recent[1].text, "c");
    }

    #[test]
    fn parse_tier_pins_canonical_tier_mapping_and_silent_catch_all_to_long_term() {
        // SqliteStore::parse_tier is the reverse of SqliteStore::tier_str
        // and is called from row_to_record on every memory record read.
        // The forward mapping (tier_str) is exercised indirectly through
        // every list_records/get_record round-trip test (sqlite_roundtrip,
        // sqlite_recent_orders_by_created_at_desc), but parse_tier has no
        // direct test pinning the three arms ('working' -> Working,
        // 'episodic' -> Episodic, _ catch-all -> LongTerm). Pin each arm
        // so a slug rename in tier_str-without-parse_tier (or vice versa)
        // or a tightening of the catch-all to a strict match is caught
        // at the parse_tier call site, distinct from the helper-internal
        // round-trip tests.
        assert_eq!(
            SqliteStore::parse_tier("working"),
            MemoryTier::Working,
            "working arm: SqliteStore::tier_str(Working) emits 'working' \
             so parse_tier must invert it; a slug rename in either \
             direction without a matching update would silently route \
             every Working record through the catch-all to LongTerm \
             at SQLite read time"
        );
        assert_eq!(
            SqliteStore::parse_tier("episodic"),
            MemoryTier::Episodic,
            "episodic arm: SqliteStore::tier_str(Episodic) emits \
             'episodic' so parse_tier must invert it; a slug rename in \
             either direction without a matching update would silently \
             route every Episodic record through the catch-all to \
             LongTerm at SQLite read time, breaking episodic-tier \
             scoped reads and dashboards"
        );
        assert_eq!(
            SqliteStore::parse_tier("longterm"),
            MemoryTier::LongTerm,
            "catch-all arm via canonical sibling: SqliteStore::tier_str \
             (LongTerm) emits 'longterm' which is intentionally NOT an \
             explicit match arm in parse_tier and routes through the \
             underscore catch-all to LongTerm; a refactor that promoted \
             'longterm' to an explicit arm would change the contract \
             that documents the catch-all as the LongTerm route — \
             future migrations relying on the catch-all to coerce \
             unknown labels would silently regress"
        );
        assert_eq!(
            SqliteStore::parse_tier("unknown-tier-from-future-migration"),
            MemoryTier::LongTerm,
            "catch-all arm via non-canonical value: any tier_s value not \
             matching 'working' or 'episodic' must coerce to LongTerm \
             so a manually-edited SQLite row, a mid-migration state, or \
             a daemon downgrade reading a future tier label gracefully \
             degrades to LongTerm; a refactor that tightened the \
             catch-all to a panic or error would turn a graceful read \
             path into a hard failure that takes down list_records and \
             get_record for the entire SqliteStore"
        );
    }

    #[test]
    fn validate_repair_request_pins_reason_and_backfill_provenance_arms() {
        let id = Uuid::new_v4();
        let detach = || MemoryRepairCommand::DetachParent {
            id,
            expected_parent: None,
        };
        let delete = || MemoryRepairCommand::DeleteRecord { id };
        let backfill = |provenance: serde_json::Value| MemoryRepairCommand::BackfillProvenance {
            id,
            provenance,
        };

        // DetachParent with non-empty reason validates Ok.
        assert!(validate_repair_request(&MemoryRepairRequest {
            mode: MemoryRepairMode::DryRun,
            command: detach(),
            reason: "operator-initiated".into(),
        })
        .is_ok());

        // Empty/whitespace reason rejects for every command variant so
        // the reason field cannot silently widen across the three arms.
        for reason in ["", "   "] {
            for command in [
                detach(),
                delete(),
                backfill(serde_json::json!({"source": "manual"})),
            ] {
                let err = validate_repair_request(&MemoryRepairRequest {
                    mode: MemoryRepairMode::DryRun,
                    command,
                    reason: reason.into(),
                })
                .unwrap_err();
                match err {
                    MemoryError::InvalidRepair(message) => assert!(
                        message.contains("reason must not be empty"),
                        "unexpected InvalidRepair payload: {message:?}",
                    ),
                    other => panic!("expected InvalidRepair, got {other:?}"),
                }
            }
        }

        // BackfillProvenance with null provenance rejects even when
        // reason is non-empty. The provenance check only fires for
        // BackfillProvenance; DetachParent and DeleteRecord do not
        // carry a provenance field.
        let err = validate_repair_request(&MemoryRepairRequest {
            mode: MemoryRepairMode::DryRun,
            command: backfill(serde_json::Value::Null),
            reason: "operator-initiated".into(),
        })
        .unwrap_err();
        match err {
            MemoryError::InvalidRepair(message) => assert!(
                message.contains("provenance must not be null"),
                "unexpected InvalidRepair payload: {message:?}",
            ),
            other => panic!("expected InvalidRepair, got {other:?}"),
        }

        // BackfillProvenance with a non-null provenance object validates
        // Ok so legitimate backfills still flow through.
        assert!(validate_repair_request(&MemoryRepairRequest {
            mode: MemoryRepairMode::DryRun,
            command: backfill(serde_json::json!({"source": "manual"})),
            reason: "operator-initiated".into(),
        })
        .is_ok());

        // DetachParent and DeleteRecord with non-empty reason validate
        // Ok; the provenance branch must not widen onto these arms.
        for command in [detach(), delete()] {
            assert!(validate_repair_request(&MemoryRepairRequest {
                mode: MemoryRepairMode::DryRun,
                command,
                reason: "operator-initiated".into(),
            })
            .is_ok());
        }
    }

    #[test]
    fn validate_compaction_request_pins_reason_and_empty_policy_arms() {
        let request = |policy: MemoryCompactionPolicy, reason: &str| MemoryCompactionRequest {
            mode: MemoryRepairMode::DryRun,
            policy,
            reason: reason.into(),
        };

        let non_empty_policy = MemoryCompactionPolicy {
            delete_working_before_ms: Some(123),
            ..MemoryCompactionPolicy::default()
        };

        // Non-empty reason and non-empty policy validate Ok.
        assert!(validate_compaction_request(&request(
            non_empty_policy.clone(),
            "operator-initiated"
        ))
        .is_ok());

        // Empty/whitespace reason rejects regardless of policy so a
        // policy-only check cannot mask a blank reason.
        for reason in ["", "   "] {
            let err = validate_compaction_request(&request(non_empty_policy.clone(), reason))
                .unwrap_err();
            match err {
                MemoryError::InvalidCompaction(message) => assert!(
                    message.contains("reason must not be empty"),
                    "unexpected InvalidCompaction payload: {message:?}",
                ),
                other => panic!("expected InvalidCompaction, got {other:?}"),
            }
        }

        // All-default policy rejects with the documented message even
        // when reason is non-empty; this is the no-op compaction guard.
        let err =
            validate_compaction_request(&request(MemoryCompactionPolicy::default(), "operator"))
                .unwrap_err();
        match err {
            MemoryError::InvalidCompaction(message) => assert!(
                message.contains("policy must enable at least one compaction action"),
                "unexpected InvalidCompaction payload: {message:?}",
            ),
            other => panic!("expected InvalidCompaction, got {other:?}"),
        }

        // detach_stale_parents=true alone validates Ok; the boolean
        // must NOT be treated as inert by is_empty(), or the legitimate
        // detach-only compaction shape would be silently rejected.
        assert!(validate_compaction_request(&request(
            MemoryCompactionPolicy {
                detach_stale_parents: true,
                ..MemoryCompactionPolicy::default()
            },
            "operator-initiated",
        ))
        .is_ok());

        // Each individual before_ms cutoff alone validates Ok so the
        // is_empty() short-circuit cannot regress to require multiple
        // policy fields to be set.
        for policy in [
            MemoryCompactionPolicy {
                delete_working_before_ms: Some(1),
                ..MemoryCompactionPolicy::default()
            },
            MemoryCompactionPolicy {
                delete_episodic_before_ms: Some(1),
                ..MemoryCompactionPolicy::default()
            },
            MemoryCompactionPolicy {
                mark_longterm_stale_before_ms: Some(1),
                ..MemoryCompactionPolicy::default()
            },
        ] {
            assert!(validate_compaction_request(&request(policy, "operator")).is_ok());
        }
    }

    #[test]
    fn cosine_basics() {
        let a = vec![1.0, 0.0, 0.0];
        let b = vec![1.0, 0.0, 0.0];
        let c = vec![0.0, 1.0, 0.0];
        assert!((cosine(&a, &b) - 1.0).abs() < 1e-6);
        assert!(cosine(&a, &c).abs() < 1e-6);
        assert_eq!(cosine(&[], &[]), 0.0);
        assert_eq!(cosine(&[1.0], &[1.0, 2.0]), 0.0);
    }

    #[test]
    fn cosine_zero_norm_returns_zero_and_anti_parallel_returns_minus_one() {
        // The docstring promises 0.0 for any degenerate input (mismatched
        // length, zero norm, empty). cosine_basics covers length-mismatch
        // and empty; this pin closes the zero-norm arm in both
        // orientations and asserts the result is a finite 0.0, not NaN —
        // a regression that dropped the `na == 0.0 || nb == 0.0` guard
        // would silently NaN every search rank that compares against an
        // all-zero embedding.
        let both_zero = cosine(&[0.0, 0.0, 0.0], &[0.0, 0.0, 0.0]);
        assert_eq!(both_zero, 0.0);
        assert!(!both_zero.is_nan());

        let lhs_zero = cosine(&[0.0, 0.0, 0.0], &[1.0, 0.0, 0.0]);
        assert_eq!(lhs_zero, 0.0);
        assert!(!lhs_zero.is_nan());

        let rhs_zero = cosine(&[1.0, 0.0, 0.0], &[0.0, 0.0, 0.0]);
        assert_eq!(rhs_zero, 0.0);
        assert!(!rhs_zero.is_nan());

        // Anti-parallel unit vectors must score -1.0, not +1.0. A vectorised
        // rewrite that mishandles negative components (e.g. taking |x*y| in
        // the dot product) would silently invert the similarity ordering of
        // every embedding with a sign-bearing dimension.
        let anti_parallel = cosine(&[1.0, 0.0], &[-1.0, 0.0]);
        assert!(
            (anti_parallel - (-1.0)).abs() < 1e-6,
            "anti-parallel must be ~-1.0, got {anti_parallel}",
        );
    }

    #[tokio::test]
    async fn in_memory_search_returns_closest_first() {
        let s = InMemoryStore::new();
        let mut a = record(Uuid::new_v4(), MemoryTier::Working, "alpha", 1);
        a.embedding = vec![1.0, 0.0, 0.0];
        let mut b = record(Uuid::new_v4(), MemoryTier::Working, "beta", 2);
        b.embedding = vec![0.0, 1.0, 0.0];
        let mut c = record(Uuid::new_v4(), MemoryTier::Working, "gamma", 3);
        c.embedding = vec![0.9, 0.1, 0.0];
        s.put(a).await.unwrap();
        s.put(b).await.unwrap();
        s.put(c).await.unwrap();
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], None, 2, None)
            .await
            .unwrap();
        assert_eq!(hits[0].text, "alpha");
        assert_eq!(hits[1].text, "gamma");
    }

    #[tokio::test]
    async fn in_memory_search_min_relevance_drops_below_threshold() {
        let s = InMemoryStore::new();
        let mut close = record(Uuid::new_v4(), MemoryTier::Working, "close", 1);
        close.embedding = vec![1.0, 0.0, 0.0];
        let mut far = record(Uuid::new_v4(), MemoryTier::Working, "far", 2);
        far.embedding = vec![0.1, 1.0, 0.0];
        s.put(close).await.unwrap();
        s.put(far).await.unwrap();
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], None, 10, Some(0.5))
            .await
            .unwrap();
        assert_eq!(hits.len(), 1, "min_relevance 0.5 must drop far record");
        assert_eq!(hits[0].text, "close");
    }

    #[tokio::test]
    async fn sqlite_search_min_relevance_drops_below_threshold() {
        let s = SqliteStore::open_in_memory().unwrap();
        let mut close = record(Uuid::new_v4(), MemoryTier::Working, "close", 1);
        close.embedding = vec![1.0, 0.0, 0.0];
        let mut far = record(Uuid::new_v4(), MemoryTier::Working, "far", 2);
        far.embedding = vec![0.1, 1.0, 0.0];
        s.put(close).await.unwrap();
        s.put(far).await.unwrap();
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], None, 10, Some(0.5))
            .await
            .unwrap();
        assert_eq!(hits.len(), 1, "min_relevance 0.5 must drop far record");
        assert_eq!(hits[0].text, "close");
    }

    #[tokio::test]
    async fn in_memory_search_similar_pins_strict_positive_filter_and_inclusive_floor_equal_boundary(
    ) {
        // InMemoryStore::search_similar applies the paired
        // predicate
        //
        //     .filter(|(s, _)| *s > 0.0 && *s >= floor)
        //
        // The strict-positive `s > 0.0` arm drops degenerate
        // zero-cosine records (zero-norm or mismatched-length embeddings;
        // see cosine()) and anchors the documented
        // 'Records with empty embeddings get score 0 and are... dropped,
        // depending on the impl' contract from the MemoryStore trait
        // doc. The `s >= floor` arm pins the inclusive
        // min_relevance semantic documented on the trait method ('records
        // whose cosine score is strictly less than the threshold are
        // dropped' — i.e., equality passes).
        //
        // The existing in_memory_search_min_relevance_drops_below_threshold
        // (above) probes min_relevance=Some(0.5) against cosine ~0.099
        // — both `>` and `>=` agree on this case, neither arm of the
        // dual predicate is uniquely exercised. The existing
        // in_memory_search_returns_closest_first uses
        // limit=2 against 3 records so a zero-cosine record is silently
        // dropped by limit-truncation rather than by the strict-positive
        // filter. A refactor that changed `s > 0.0` to `s >= 0.0` under
        // a 'simplify the dual predicate' rationale, or that changed
        // `s >= floor` to `s > floor` under a 'use exclusive thresholds
        // for consistency with the doc phrase strictly less than'
        // rationale, would not surface in either existing test.

        let s = InMemoryStore::new();
        let mut exact = record(Uuid::new_v4(), MemoryTier::Working, "exact", 1);
        exact.embedding = vec![1.0, 0.0, 0.0];
        let mut orthogonal = record(Uuid::new_v4(), MemoryTier::Working, "orthogonal", 2);
        orthogonal.embedding = vec![0.0, 1.0, 0.0];
        let mut anti = record(Uuid::new_v4(), MemoryTier::Working, "anti", 3);
        anti.embedding = vec![-1.0, 0.0, 0.0];
        s.put(exact).await.unwrap();
        s.put(orthogonal).await.unwrap();
        s.put(anti).await.unwrap();

        // (1) Strict-positive arm: limit much larger than the non-zero
        // match count, no min_relevance set. The orthogonal record's
        // cosine is exactly 0.0 — under `s > 0.0` it must be filtered;
        // under a regressed `s >= 0.0` it would surface. The
        // anti-parallel record's cosine is -1.0 — filtered by either
        // operator on the strict-positive arm. Only 'exact' surfaces.
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], None, 10, None)
            .await
            .unwrap();
        assert_eq!(
            hits.len(),
            1,
            "strict-positive filter `s > 0.0` must drop the orthogonal \
             cosine-zero record even with abundant limit and no \
             min_relevance set — a refactor to `s >= 0.0` would let \
             zero-norm or mismatched-length embeddings (which v0's mock \
             embedder and Ollama embedder cannot produce, but a future \
             external embedder or corrupted SQLite blob could) pollute \
             every search result silently",
        );
        assert_eq!(
            hits[0].text, "exact",
            "the only surviving record must be the cosine=1.0 match",
        );

        // (2) Inclusive-floor-equality arm: min_relevance = Some(1.0).
        // The 'exact' record's cosine is exactly 1.0 (f32 arithmetic
        // 1.0/(1.0*1.0) = 1.0) — under `s >= floor` it passes; under a
        // regressed `s > floor` it would be rejected. Orthogonal is
        // filtered by strict-positive; anti by strict-positive.
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], None, 10, Some(1.0))
            .await
            .unwrap();
        assert_eq!(
            hits.len(),
            1,
            "inclusive-floor `s >= floor` must accept a record whose \
             cosine is exactly the min_relevance threshold — a refactor \
             to `s > floor` would silently shift the documented \
             inclusive semantic so operators tuning min_relevance to \
             the cosine of their best-match record (a common \
             calibration step after embedding upgrades) would see zero \
             results returned with no signal that the comparison \
             operator drifted",
        );
        assert_eq!(
            hits[0].text, "exact",
            "the exact-cosine record must surface at the equality boundary",
        );
    }

    #[tokio::test]
    async fn sqlite_search_similar_pins_strict_positive_filter_and_inclusive_floor_equal_boundary()
    {
        // SqliteStore::search_similar applies the same paired
        // predicate as InMemoryStore::search_similar:
        //
        //     if s > 0.0 && s >= floor {
        //
        // The two implementations must agree on the dual predicate so
        // storage-tier choice (in-memory for tests/defaults vs SQLite
        // for persistent daemon runs) doesn't change the search result
        // set semantic. The parallel pin (in_memory_search_similar_pins_*
        // above) anchors the contract on the InMemoryStore side; this
        // pin anchors it on the SQLite side. A refactor that diverged
        // the two implementations — e.g., dropped the strict-positive
        // guard on SQLite under a 'rows controlled by put() can never
        // have empty embeddings' rationale — would silently shift
        // result sets when operators migrate from in-memory to SQLite
        // with no audit signal.

        let s = SqliteStore::open_in_memory().unwrap();
        let mut exact = record(Uuid::new_v4(), MemoryTier::Working, "exact", 1);
        exact.embedding = vec![1.0, 0.0, 0.0];
        let mut orthogonal = record(Uuid::new_v4(), MemoryTier::Working, "orthogonal", 2);
        orthogonal.embedding = vec![0.0, 1.0, 0.0];
        let mut anti = record(Uuid::new_v4(), MemoryTier::Working, "anti", 3);
        anti.embedding = vec![-1.0, 0.0, 0.0];
        s.put(exact).await.unwrap();
        s.put(orthogonal).await.unwrap();
        s.put(anti).await.unwrap();

        // (1) Strict-positive arm — same shape as the in-memory pin.
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], None, 10, None)
            .await
            .unwrap();
        assert_eq!(
            hits.len(),
            1,
            "SqliteStore strict-positive filter must drop the \
             orthogonal cosine-zero record — cross-binds the \
             in_memory_search_similar_pins_strict_positive_filter_and_inclusive_floor_equal_boundary \
             arm so the two implementations cannot diverge silently",
        );
        assert_eq!(hits[0].text, "exact");

        // (2) Inclusive-floor-equality arm — same shape as the in-memory
        // pin.
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], None, 10, Some(1.0))
            .await
            .unwrap();
        assert_eq!(
            hits.len(),
            1,
            "SqliteStore inclusive-floor must accept a record at \
             cosine exactly equal to the min_relevance threshold — \
             cross-binds the in_memory pin's inclusive-equality arm",
        );
        assert_eq!(hits[0].text, "exact");
    }

    #[test]
    fn sqlite_embedding_bytes_pins_round_trip_little_endian_and_trailing_partial_chunk_drop() {
        // SqliteStore::embedding_to_bytes and
        // ::embedding_from_bytes are the SQLite BLOB
        // serializer/deserializer pair the persistent memory store
        // uses to round-trip f32 embedding vectors through the
        // embedding column. The pair encodes f32 as little-endian
        // 4-byte chunks with no length prefix, no version byte, no
        // padding — embedding_to_bytes writes f32::to_le_bytes()
        // back-to-back; embedding_from_bytes reads via
        // chunks_exact(4) which silently drops any trailing partial
        // chunk.
        //
        // sqlite_roundtrip and sqlite_recent_orders_by_created_at_desc
        // round-trip records but never directly assert byte-level
        // encoding shape, byte order, or trailing-chunk drop
        // semantics. A refactor that swapped little-endian for big-
        // endian would silently corrupt every persisted embedding
        // across mixed-endian deployments. A refactor that switched
        // chunks_exact to chunks would silently panic on the first
        // truncated row. A refactor that added a length prefix would
        // silently scramble cosine similarity for every existing row.

        // (1) Empty round-trip: no allocation, no panic.
        assert_eq!(
            SqliteStore::embedding_to_bytes(&[]),
            Vec::<u8>::new(),
            "empty embedding must serialize to empty bytes",
        );
        assert_eq!(
            SqliteStore::embedding_from_bytes(&[]),
            Vec::<f32>::new(),
            "empty bytes must deserialize to empty embedding",
        );

        // (2) Single-element round-trip.
        let v = vec![1.5_f32];
        assert_eq!(
            SqliteStore::embedding_from_bytes(&SqliteStore::embedding_to_bytes(&v)),
            v,
            "single-element vector must round-trip exactly",
        );

        // (3) Byte order: cross-bind to f32::to_le_bytes so any
        // switch to big-endian (e.g., 'match network byte order')
        // surfaces immediately.
        assert_eq!(
            SqliteStore::embedding_to_bytes(&[1.0_f32]),
            1.0_f32.to_le_bytes().to_vec(),
            "embedding_to_bytes must emit little-endian f32 — a \
             refactor that swapped to to_be_bytes under a 'match \
             network byte order' rationale would silently corrupt \
             every persisted embedding across mixed-endian \
             deployments (and an in-process round-trip would still \
             pass because both writer and reader would agree on the \
             new order)",
        );

        // (4) Ordering preservation: first f32 → first 4 bytes.
        let bytes = SqliteStore::embedding_to_bytes(&[1.0_f32, 2.0, 3.0]);
        assert_eq!(bytes.len(), 12, "three f32 must produce 12 bytes");
        assert_eq!(
            f32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]),
            1.0_f32,
            "first f32 must occupy bytes [0..4]",
        );
        assert_eq!(
            f32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]),
            2.0_f32,
            "second f32 must occupy bytes [4..8]",
        );
        assert_eq!(
            f32::from_le_bytes([bytes[8], bytes[9], bytes[10], bytes[11]]),
            3.0_f32,
            "third f32 must occupy bytes [8..12]",
        );

        // (5) Trailing partial chunk drop: chunks_exact(4) silently
        // drops the trailing 1-3 bytes. A refactor that switched to
        // chunks(4) and indexed c[0..4] would panic on the partial
        // slice; the silent-drop contract is what protects the
        // recall path against partial writes.
        assert_eq!(
            SqliteStore::embedding_from_bytes(&[0u8, 0, 0, 0, 1, 2, 3]),
            vec![0.0_f32],
            "embedding_from_bytes must yield exactly 1 f32 from 7 \
             bytes — chunks_exact(4) drops the trailing 3 bytes \
             silently, which is what protects the recall path from \
             panicking on partially-written rows after a write-time \
             crash",
        );

        // (6) Length scaling: 100 f32 → 400 bytes (no prefix, no
        // padding, no overhead). A length prefix or version byte
        // would shift this to 404+.
        assert_eq!(
            SqliteStore::embedding_to_bytes(&[0.0_f32; 100]).len(),
            400,
            "100 f32 must produce exactly 400 bytes (4 bytes per f32, \
             no overhead) — a refactor that added a length prefix \
             ('4-byte u32 length || 4N bytes') under an 'evolve the \
             embedding format' rationale without a coordinated \
             migration would shift this to 404 and silently scramble \
             cosine similarity for every existing SQLite row",
        );
    }

    #[tokio::test]
    async fn sqlite_search_respects_tier_filter() {
        let s = SqliteStore::open_in_memory().unwrap();
        let mut w = record(Uuid::new_v4(), MemoryTier::Working, "w-alpha", 1);
        w.embedding = vec![1.0, 0.0, 0.0];
        let mut e = record(Uuid::new_v4(), MemoryTier::Episodic, "e-alpha", 2);
        e.embedding = vec![1.0, 0.0, 0.0];
        s.put(w).await.unwrap();
        s.put(e).await.unwrap();
        let hits = s
            .search_similar(vec![1.0, 0.0, 0.0], Some(MemoryTier::Episodic), 5, None)
            .await
            .unwrap();
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].text, "e-alpha");
    }

    #[tokio::test]
    async fn in_memory_purge_older_than_drops_old_records() {
        let s = InMemoryStore::new();
        s.put(record(Uuid::new_v4(), MemoryTier::Working, "old", 100))
            .await
            .unwrap();
        s.put(record(Uuid::new_v4(), MemoryTier::Working, "newer", 500))
            .await
            .unwrap();
        s.put(record(Uuid::new_v4(), MemoryTier::Episodic, "old-ep", 100))
            .await
            .unwrap();
        // Purge only the working tier records older than 200.
        let n = s
            .purge_older_than(Some(MemoryTier::Working), 200)
            .await
            .unwrap();
        assert_eq!(n, 1);
        let remaining = s.recent(None, 10).await.unwrap();
        assert_eq!(remaining.len(), 2);
    }

    #[tokio::test]
    async fn sqlite_purge_older_than_clears_only_matching_rows() {
        let s = SqliteStore::open_in_memory().unwrap();
        s.put(record(Uuid::new_v4(), MemoryTier::Working, "old", 100))
            .await
            .unwrap();
        s.put(record(Uuid::new_v4(), MemoryTier::Working, "newer", 500))
            .await
            .unwrap();
        let n = s
            .purge_older_than(Some(MemoryTier::Working), 200)
            .await
            .unwrap();
        assert_eq!(n, 1);
        let recent = s.recent(Some(MemoryTier::Working), 10).await.unwrap();
        assert_eq!(recent.len(), 1);
        assert_eq!(recent[0].text, "newer");
    }

    #[tokio::test]
    async fn sqlite_persistence_across_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("memory.db");
        let id = Uuid::new_v4();
        {
            let s = SqliteStore::open(&path).unwrap();
            s.put(record(id, MemoryTier::LongTerm, "persist me", 42))
                .await
                .unwrap();
        }
        let s2 = SqliteStore::open(&path).unwrap();
        let got = s2.get(id).await.unwrap().unwrap();
        assert_eq!(got.text, "persist me");
    }

    #[tokio::test]
    async fn repair_dry_run_detach_parent_does_not_mutate() {
        let s = InMemoryStore::new();
        let id = Uuid::new_v4();
        let parent = Uuid::new_v4();
        let mut r = record(id, MemoryTier::Episodic, "child", 10);
        r.parent = Some(parent);
        s.put(r).await.unwrap();

        let outcome = s
            .repair(MemoryRepairRequest {
                mode: MemoryRepairMode::DryRun,
                command: MemoryRepairCommand::DetachParent {
                    id,
                    expected_parent: Some(parent),
                },
                reason: "parent was deleted".into(),
            })
            .await
            .unwrap();

        assert!(outcome.would_change);
        assert!(!outcome.changed);
        assert_eq!(outcome.after.unwrap().parent, None);
        assert_eq!(s.get(id).await.unwrap().unwrap().parent, Some(parent));
    }

    #[tokio::test]
    async fn repair_apply_detaches_parent() {
        let s = InMemoryStore::new();
        let id = Uuid::new_v4();
        let parent = Uuid::new_v4();
        let mut r = record(id, MemoryTier::Episodic, "child", 10);
        r.parent = Some(parent);
        s.put(r).await.unwrap();

        let outcome = s
            .repair(MemoryRepairRequest {
                mode: MemoryRepairMode::Apply,
                command: MemoryRepairCommand::DetachParent {
                    id,
                    expected_parent: Some(parent),
                },
                reason: "parent was deleted".into(),
            })
            .await
            .unwrap();

        assert!(outcome.changed);
        assert_eq!(s.get(id).await.unwrap().unwrap().parent, None);
    }

    #[tokio::test]
    async fn repair_rejects_parent_mismatch() {
        let s = InMemoryStore::new();
        let id = Uuid::new_v4();
        let actual_parent = Uuid::new_v4();
        let mut r = record(id, MemoryTier::Episodic, "child", 10);
        r.parent = Some(actual_parent);
        s.put(r).await.unwrap();

        let result = s
            .repair(MemoryRepairRequest {
                mode: MemoryRepairMode::Apply,
                command: MemoryRepairCommand::DetachParent {
                    id,
                    expected_parent: Some(Uuid::new_v4()),
                },
                reason: "stale parent repair".into(),
            })
            .await;
        assert!(matches!(result, Err(MemoryError::ParentMismatch { .. })));
        assert_eq!(
            s.get(id).await.unwrap().unwrap().parent,
            Some(actual_parent)
        );
    }

    #[tokio::test]
    async fn repair_apply_deletes_record() {
        let s = SqliteStore::open_in_memory().unwrap();
        let id = Uuid::new_v4();
        s.put(record(id, MemoryTier::Working, "delete me", 10))
            .await
            .unwrap();

        let outcome = s
            .repair(MemoryRepairRequest {
                mode: MemoryRepairMode::Apply,
                command: MemoryRepairCommand::DeleteRecord { id },
                reason: "operator confirmed unsafe memory".into(),
            })
            .await
            .unwrap();

        assert!(outcome.changed);
        assert!(outcome.after.is_none());
        assert!(s.get(id).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn repair_backfills_provenance_metadata() {
        let s = SqliteStore::open_in_memory().unwrap();
        let id = Uuid::new_v4();
        let mut r = record(id, MemoryTier::LongTerm, "needs provenance", 10);
        r.metadata = serde_json::json!({"source": "import"});
        s.put(r).await.unwrap();

        let provenance = serde_json::json!({
            "kind": "manual_backfill",
            "evidence": "audit window checked"
        });
        let outcome = s
            .repair(MemoryRepairRequest {
                mode: MemoryRepairMode::Apply,
                command: MemoryRepairCommand::BackfillProvenance {
                    id,
                    provenance: provenance.clone(),
                },
                reason: "missing provenance evidence".into(),
            })
            .await
            .unwrap();

        assert!(outcome.changed);
        let got = s.get(id).await.unwrap().unwrap();
        assert_eq!(got.metadata["source"], "import");
        assert_eq!(got.metadata["provenance"], provenance);
    }

    #[test]
    fn plan_repair_backfill_provenance_pins_non_object_metadata_under_previous_metadata() {
        // covenant_memory::plan_repair handles
        // MemoryRepairCommand::BackfillProvenance by reading the
        // record's existing metadata. Line 197-204:
        //
        //   let mut metadata = match after.metadata {
        //       serde_json::Value::Object(map) => map,
        //       other => {
        //           let mut map = serde_json::Map::new();
        //           map.insert("previous_metadata".into(), other);
        //           map
        //       }
        //   };
        //   metadata.insert("provenance".into(), provenance.clone());
        //
        // The 'other' arm preserves any non-object metadata under a
        // 'previous_metadata' key in a new object, so backfilling
        // provenance on a legacy record with null/string/array
        // metadata does not silently lose the prior value. The
        // existing repair_backfills_provenance_metadata test
        // only exercises the object-already arm.
        //
        // A refactor that replaces the match with
        // metadata.as_object_mut().unwrap() under a 'metadata is
        // always an object' rationale would panic on legacy records
        // with non-object metadata. A refactor that drops the
        // previous_metadata wrap and just sets metadata =
        // json!({"provenance": ...}) would silently lose the prior
        // non-object metadata value with no audit signal.

        let provenance = serde_json::json!({"kind": "manual_backfill"});
        let cmd = |id: Uuid| MemoryRepairCommand::BackfillProvenance {
            id,
            provenance: provenance.clone(),
        };

        let id = Uuid::new_v4();

        let mut null_record = record(id, MemoryTier::LongTerm, "n", 1);
        null_record.metadata = serde_json::Value::Null;
        let after = plan_repair(&null_record, &cmd(id))
            .expect("BackfillProvenance must not error on null metadata")
            .expect("BackfillProvenance always returns Some(after)");
        assert_eq!(
            after.metadata["previous_metadata"],
            serde_json::Value::Null,
            "null metadata must be preserved verbatim under \
             previous_metadata — pins the non-object arm. \
             A refactor that dropped the wrap would surface here as a \
             missing previous_metadata key, and a refactor that swapped \
             the arm for as_object_mut().unwrap() would have panicked \
             before reaching this assertion",
        );
        assert_eq!(
            after.metadata["provenance"], provenance,
            "provenance must be inserted into the new map alongside \
             previous_metadata — pins that the provenance write happens \
             AFTER the non-object wrap, not BEFORE (which would let \
             previous_metadata shadow it if the key names ever \
             collided)",
        );

        let mut string_record = record(id, MemoryTier::LongTerm, "s", 1);
        string_record.metadata = serde_json::Value::String("legacy-tag".into());
        let after = plan_repair(&string_record, &cmd(id))
            .expect("BackfillProvenance must not error on string metadata")
            .expect("BackfillProvenance always returns Some(after)");
        assert_eq!(
            after.metadata["previous_metadata"], "legacy-tag",
            "string metadata must be preserved verbatim under \
             previous_metadata — pins that the wrap is value-agnostic \
             across primitive JSON types, not just null",
        );
        assert_eq!(after.metadata["provenance"], provenance);

        let mut array_record = record(id, MemoryTier::LongTerm, "a", 1);
        array_record.metadata = serde_json::json!(["tag-a", "tag-b"]);
        let after = plan_repair(&array_record, &cmd(id))
            .expect("BackfillProvenance must not error on array metadata")
            .expect("BackfillProvenance always returns Some(after)");
        assert_eq!(
            after.metadata["previous_metadata"],
            serde_json::json!(["tag-a", "tag-b"]),
            "array metadata must be preserved verbatim under \
             previous_metadata — pins that the wrap captures container \
             values, not just scalars; a refactor that flattened the \
             array into individual map keys under a 'merge legacy tags \
             into provenance' rationale would surface here",
        );
        assert_eq!(after.metadata["provenance"], provenance);

        let mut object_record = record(id, MemoryTier::LongTerm, "o", 1);
        object_record.metadata = serde_json::json!({"source": "import", "rev": 7});
        let after = plan_repair(&object_record, &cmd(id))
            .expect("BackfillProvenance must not error on object metadata")
            .expect("BackfillProvenance always returns Some(after)");
        assert_eq!(
            after.metadata["source"], "import",
            "object metadata happy path: existing keys must be \
             preserved without re-keying under previous_metadata — \
             pins the first match arm. A refactor that \
             accidentally fell through to the 'other' arm for objects \
             would surface here as a missing 'source' key and an \
             unexpected previous_metadata wrap",
        );
        assert_eq!(after.metadata["rev"], 7);
        assert_eq!(after.metadata["provenance"], provenance);
        assert!(
            after.metadata.get("previous_metadata").is_none(),
            "object-already arm must NOT introduce a previous_metadata \
             key — pins that the wrap is exclusive to the non-object \
             case; a refactor that always wrapped under \
             previous_metadata would double-nest the object metadata \
             and silently break callers reading the legacy keys",
        );
    }

    #[tokio::test]
    async fn compaction_dry_run_plans_without_mutating() {
        let s = InMemoryStore::new();
        let old_working = Uuid::new_v4();
        let child = Uuid::new_v4();
        let missing_parent = Uuid::new_v4();
        s.put(record(old_working, MemoryTier::Working, "old", 10))
            .await
            .unwrap();
        let mut child_record = record(child, MemoryTier::Episodic, "child", 30);
        child_record.parent = Some(missing_parent);
        s.put(child_record).await.unwrap();

        let outcome = s
            .compact(MemoryCompactionRequest {
                mode: MemoryRepairMode::DryRun,
                policy: MemoryCompactionPolicy {
                    delete_working_before_ms: Some(20),
                    detach_stale_parents: true,
                    ..MemoryCompactionPolicy::default()
                },
                reason: "routine dry-run".into(),
            })
            .await
            .unwrap();

        assert!(outcome.would_change);
        assert!(!outcome.changed);
        assert_eq!(outcome.deleted, vec![old_working]);
        assert_eq!(outcome.parents_detached, vec![child]);
        assert!(s.get(old_working).await.unwrap().is_some());
        assert_eq!(
            s.get(child).await.unwrap().unwrap().parent,
            Some(missing_parent)
        );
    }

    #[tokio::test]
    async fn compaction_apply_deletes_short_horizon_marks_longterm_and_detaches_parents() {
        let s = SqliteStore::open_in_memory().unwrap();
        let old_working = Uuid::new_v4();
        let old_episodic = Uuid::new_v4();
        let child = Uuid::new_v4();
        let longterm = Uuid::new_v4();
        s.put(record(old_working, MemoryTier::Working, "old working", 10))
            .await
            .unwrap();
        s.put(record(
            old_episodic,
            MemoryTier::Episodic,
            "old episodic",
            10,
        ))
        .await
        .unwrap();
        let mut child_record = record(child, MemoryTier::Episodic, "child", 50);
        child_record.parent = Some(old_working);
        s.put(child_record).await.unwrap();
        s.put(record(longterm, MemoryTier::LongTerm, "durable fact", 10))
            .await
            .unwrap();

        let outcome = s
            .compact(MemoryCompactionRequest {
                mode: MemoryRepairMode::Apply,
                policy: MemoryCompactionPolicy {
                    delete_working_before_ms: Some(20),
                    delete_episodic_before_ms: Some(20),
                    mark_longterm_stale_before_ms: Some(20),
                    detach_stale_parents: true,
                    marked_at_ms: Some(99),
                },
                reason: "age-based compaction".into(),
            })
            .await
            .unwrap();

        assert!(outcome.changed);
        let mut expected_deleted = vec![old_working, old_episodic];
        expected_deleted.sort();
        assert_eq!(outcome.deleted, expected_deleted);
        assert_eq!(outcome.parents_detached, vec![child]);
        assert_eq!(outcome.stale_marked, vec![longterm]);
        assert!(s.get(old_working).await.unwrap().is_none());
        assert!(s.get(old_episodic).await.unwrap().is_none());
        assert_eq!(s.get(child).await.unwrap().unwrap().parent, None);
        let durable = s.get(longterm).await.unwrap().unwrap();
        assert_eq!(durable.metadata["stale_context"]["marked_at_ms"], 99);
        assert_eq!(
            durable.metadata["stale_context"]["reason"],
            "age-based compaction"
        );
    }

    #[test]
    fn plan_compaction_stale_context_pins_non_object_metadata_under_previous_metadata() {
        // covenant_memory::plan_compaction marks
        // LongTerm records stale when their created_at falls below
        // mark_longterm_stale_before_ms. Lines 274-281 use the same
        // non-object preservation pattern as plan_repair:
        //
        //   let mut metadata = match after.metadata {
        //       serde_json::Value::Object(map) => map,
        //       other => {
        //           let mut map = serde_json::Map::new();
        //           map.insert("previous_metadata".into(), other);
        //           map
        //       }
        //   };
        //
        // Every existing compaction test uses record()
        // which seeds metadata = serde_json::json!({}); so they all
        // hit the object-already arm. The 'other' arm
        // that handles legacy LongTerm records with
        // null/string/array metadata is dead code from the test
        // suite's perspective. A refactor that swapped the match for
        // metadata.as_object_mut().unwrap() under a 'metadata is
        // always an object' rationale would panic mid-compaction the
        // first time a legacy record aged past
        // mark_longterm_stale_before_ms.

        let policy = MemoryCompactionPolicy {
            mark_longterm_stale_before_ms: Some(100),
            marked_at_ms: Some(150),
            ..MemoryCompactionPolicy::default()
        };
        let request = MemoryCompactionRequest {
            mode: MemoryRepairMode::Apply,
            policy,
            reason: "stale-context pin".into(),
        };

        let build = |metadata: serde_json::Value| {
            let mut r = record(Uuid::new_v4(), MemoryTier::LongTerm, "legacy", 10);
            r.metadata = metadata;
            r
        };

        let cases = [
            ("null", serde_json::Value::Null, serde_json::Value::Null),
            (
                "string",
                serde_json::Value::String("legacy-tag".into()),
                serde_json::Value::String("legacy-tag".into()),
            ),
            (
                "array",
                serde_json::json!(["tag-a", "tag-b"]),
                serde_json::json!(["tag-a", "tag-b"]),
            ),
        ];
        for (label, initial, expected_previous) in cases {
            let records = vec![build(initial)];
            let (outcome, updates) = plan_compaction(&records, &request);
            assert!(
                outcome.changed,
                "{label}: outcome.changed must be true in Apply mode \
                 when the LongTerm record qualifies for stale marking",
            );
            assert_eq!(
                updates.len(),
                1,
                "{label}: exactly one updated record must be returned \
                 (the stale-marked LongTerm record); a refactor that \
                 skipped writes for non-object metadata would surface \
                 as an empty updates list",
            );
            let after = &updates[0];
            assert_eq!(
                after.metadata["previous_metadata"], expected_previous,
                "{label}: the non-object metadata must be preserved \
                 verbatim under previous_metadata — pins the 'other' \
                 'other' arm. A refactor that dropped the \
                 wrap would surface as a missing previous_metadata \
                 key; a refactor that swapped the match for \
                 as_object_mut().unwrap() would have panicked before \
                 reaching this assertion",
            );
            assert_eq!(
                after.metadata["stale_context"]["marked_at_ms"], 150,
                "{label}: stale_context.marked_at_ms must reflect the \
                 policy field — pins that the stale_context write \
                 happens AFTER the non-object wrap, not BEFORE",
            );
            assert_eq!(
                after.metadata["stale_context"]["reason"], "stale-context pin",
                "{label}: stale_context.reason must reflect the \
                 request.reason — pins reason propagation through the \
                 non-object preservation path",
            );
        }

        let object_record = build(serde_json::json!({"source": "import"}));
        let (outcome, updates) = plan_compaction(&[object_record], &request);
        assert!(outcome.changed);
        let after = &updates[0];
        assert_eq!(
            after.metadata["source"], "import",
            "object-already arm: existing keys must survive the \
             stale_context insert — pins the first match arm at line \
             275. A refactor that always wrapped under \
             previous_metadata would shadow the 'source' key",
        );
        assert_eq!(after.metadata["stale_context"]["marked_at_ms"], 150);
        assert!(
            after.metadata.get("previous_metadata").is_none(),
            "object-already arm must NOT introduce previous_metadata \
             — pins that the wrap is exclusive to the non-object \
             case",
        );
    }

    #[test]
    fn plan_repair_detach_parent_pins_parent_mismatch_arms_and_field_composition() {
        // covenant_memory::plan_repair, DetachParent arm, guards on
        //
        //   if expected_parent.is_some() && record.parent != *expected_parent
        //
        // The guard implements a four-way input semantics:
        //   (1) expected=None always passes (operator opts out of the
        //       check) — the detach proceeds regardless of what parent
        //       the record actually holds;
        //   (2) expected=Some(X), actual=Some(X) passes (matched);
        //   (3) expected=Some(X), actual=Some(Y) where X!=Y returns
        //       Err(ParentMismatch { id, expected: Some(X), actual:
        //       Some(Y) });
        //   (4) expected=Some(X), actual=None returns
        //       Err(ParentMismatch { id, expected: Some(X), actual:
        //       None }).
        //
        // Of these four arms, only arm (3) is pinned today via
        // repair_rejects_parent_mismatch, and only via
        // matches!(_, Err(MemoryError::ParentMismatch { .. })) — the
        // existing pin does NOT inspect the ParentMismatch field
        // values, so a refactor that swapped expected and actual under
        // a 'sort fields alphabetically' rationale would silently flip
        // operator-facing error messages while the existing pin still
        // passes. Arms (1) and (4) have no direct test.
        //
        // A refactor that flipped the != to == under a 'simplify the
        // negation' rationale would invert the entire mismatch
        // contract — matching parents would error and mismatching
        // parents would succeed; pinning the four-arm matrix means
        // each input combination explicitly anchors which side returns
        // Ok vs Err so a flipped check fails three of four arms.
        let id = Uuid::new_v4();
        let parent_a = Uuid::new_v4();
        let parent_b = Uuid::new_v4();

        let with_parent = |p: Option<Uuid>| {
            let mut r = record(id, MemoryTier::Episodic, "child", 10);
            r.parent = p;
            r
        };
        let detach = |expected: Option<Uuid>| MemoryRepairCommand::DetachParent {
            id,
            expected_parent: expected,
        };

        // Arm 1a: expected=None, actual=None — guard short-circuits;
        // detach proceeds.
        let after = plan_repair(&with_parent(None), &detach(None))
            .expect("expected=None must always pass; the guard short-circuits via expected_parent.is_some()")
            .expect("DetachParent always returns Some(after) on success");
        assert_eq!(
            after.parent, None,
            "arm 1a (expected=None, actual=None): detach proceeds; \
             after.parent must be None — pins that DetachParent \
             always sets parent to None on success regardless of the \
             input combination",
        );

        // Arm 1b: expected=None, actual=Some(X) — guard short-circuits;
        // detach proceeds even though the record HAS a parent. This is
        // the documented 'I do not care what parent the record has'
        // contract; a refactor that dropped the expected_parent.is_some()
        // guard would silently turn this into ParentMismatch because
        // record.parent (Some(X)) != *expected_parent (None) is true.
        let after = plan_repair(&with_parent(Some(parent_a)), &detach(None))
            .expect(
                "expected=None must pass even when actual=Some(X); the \
                 expected_parent.is_some() guard is the documented \
                 'operator opts out of the parent check' contract — a \
                 refactor that dropped the guard under a 'simplify the \
                 boolean' rationale would silently turn this arm into \
                 Err(ParentMismatch) because None != Some(X) is true",
            )
            .expect("DetachParent always returns Some(after) on success");
        assert_eq!(after.parent, None, "arm 1b: parent cleared");

        // Arm 2: expected=Some(X), actual=Some(X) — guard true, equality
        // false, no error; detach proceeds.
        let after = plan_repair(&with_parent(Some(parent_a)), &detach(Some(parent_a)))
            .expect(
                "matched expected/actual must pass — pins the equality semantics on the happy path",
            )
            .expect("DetachParent always returns Some(after) on success");
        assert_eq!(after.parent, None, "arm 2: parent cleared on match");

        // Arm 3: expected=Some(X), actual=Some(Y) where X!=Y — guard
        // true, equality false (records differ), ParentMismatch fires.
        // Inspect the FIELD VALUES so a refactor that swapped expected
        // and actual fails the destructure assertion.
        let err = plan_repair(&with_parent(Some(parent_b)), &detach(Some(parent_a))).expect_err(
            "expected=Some(X), actual=Some(Y) with X!=Y must \
                 return ParentMismatch — the existing \
                 repair_rejects_parent_mismatch pin covers \
                 this arm only at the matches!(_, Err(_)) level; this \
                 destructure pins the field VALUES so a refactor that \
                 swapped expected and actual cannot land silently",
        );
        match err {
            MemoryError::ParentMismatch {
                id: err_id,
                expected,
                actual,
            } => {
                assert_eq!(err_id, id, "ParentMismatch.id must equal record.id");
                assert_eq!(
                    expected,
                    Some(parent_a),
                    "ParentMismatch.expected must equal *expected_parent — \
                     a refactor that swapped expected and actual under a \
                     'sort fields alphabetically' rationale would surface \
                     here as expected==Some(parent_b)",
                );
                assert_eq!(
                    actual,
                    Some(parent_b),
                    "ParentMismatch.actual must equal record.parent — \
                     paired with the expected assertion above, this \
                     destructure makes the field-swap regression \
                     fail BOTH arms instead of silently passing",
                );
            }
            other => panic!("expected ParentMismatch, got {other:?}"),
        }

        // Arm 4: expected=Some(X), actual=None — guard true, equality
        // false (None != Some(X)), ParentMismatch fires. This arm is
        // not exercised by repair_rejects_parent_mismatch (which uses
        // actual=Some(Y)). Field destructure pins the actual=None
        // value explicitly.
        let err = plan_repair(&with_parent(None), &detach(Some(parent_a))).expect_err(
            "expected=Some(X), actual=None must return ParentMismatch — \
             the guard fires because expected_parent.is_some() is true \
             and record.parent (None) != *expected_parent (Some(X)) is \
             true; this arm is not covered by \
             repair_rejects_parent_mismatch which only exercises \
             actual=Some(Y)",
        );
        match err {
            MemoryError::ParentMismatch {
                id: err_id,
                expected,
                actual,
            } => {
                assert_eq!(err_id, id);
                assert_eq!(
                    expected,
                    Some(parent_a),
                    "ParentMismatch.expected must equal *expected_parent on the actual=None arm",
                );
                assert_eq!(
                    actual, None,
                    "ParentMismatch.actual must equal record.parent (None) — \
                     pinning the None arm explicitly catches a refactor \
                     that special-cased actual=None to Some(Uuid::nil()) \
                     for 'consistent typing' which would silently change \
                     operator-facing error payloads",
                );
            }
            other => panic!("expected ParentMismatch, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn compaction_rejects_empty_policy_and_reason() {
        let s = InMemoryStore::new();
        let empty_policy = s
            .compact(MemoryCompactionRequest {
                mode: MemoryRepairMode::DryRun,
                policy: MemoryCompactionPolicy::default(),
                reason: "no-op".into(),
            })
            .await;
        assert!(matches!(
            empty_policy,
            Err(MemoryError::InvalidCompaction(_))
        ));

        let empty_reason = s
            .compact(MemoryCompactionRequest {
                mode: MemoryRepairMode::DryRun,
                policy: MemoryCompactionPolicy {
                    detach_stale_parents: true,
                    ..MemoryCompactionPolicy::default()
                },
                reason: " ".into(),
            })
            .await;
        assert!(matches!(
            empty_reason,
            Err(MemoryError::InvalidCompaction(_))
        ));
    }

    #[tokio::test]
    async fn plan_compaction_marked_at_ms_defaults_to_before_ms_when_policy_field_is_none() {
        // covenant_memory::plan_compaction binds the
        // stale_context.marked_at_ms value via:
        //
        //   let marked_at_ms = request.policy.marked_at_ms.unwrap_or(before_ms);
        //
        // where `before_ms` is the mark_longterm_stale_before_ms cutoff
        // bound on the outer `if let Some(before_ms) = ...`. When
        // MemoryCompactionPolicy::marked_at_ms is None, the cutoff
        // itself is the recorded timestamp — anchoring stale_context to
        // the policy boundary that triggered the mark.
        //
        // The existing compaction_apply_deletes_short_horizon_marks_longterm_and_detaches_parents
        // always sets marked_at_ms = Some(99); the
        // default-to-cutoff arm is exercised by no direct test. A
        // refactor that swapped .unwrap_or(before_ms) for
        // .unwrap_or(0), .unwrap_or_default(), or
        // .unwrap_or_else(now_ms) under a 'use sensible default' pass
        // would silently change every stale_context.marked_at_ms emitted
        // by operators who omit the field — dashboards would see 0 or
        // wall-clock timestamps that no longer bind to the policy
        // cutoff, and audit reconstruction would lose its anchor.
        //
        // Cross-bind: the sibling
        // compaction_apply_deletes_short_horizon_marks_longterm_and_detaches_parents
        // pin covers the Some(M) override; this pin closes the None
        // default arm and re-asserts Some(M) inside the same test so
        // both arms travel together.

        let s = InMemoryStore::new();
        let id_default = Uuid::new_v4();
        let id_explicit = Uuid::new_v4();
        s.put(record(id_default, MemoryTier::LongTerm, "defaulted", 10))
            .await
            .unwrap();
        s.put(record(id_explicit, MemoryTier::LongTerm, "explicit", 10))
            .await
            .unwrap();

        // Default arm: marked_at_ms = None, mark_longterm_stale_before_ms = Some(200).
        // The recorded stale_context.marked_at_ms must equal the cutoff (200).
        let outcome = s
            .compact(MemoryCompactionRequest {
                mode: MemoryRepairMode::Apply,
                policy: MemoryCompactionPolicy {
                    mark_longterm_stale_before_ms: Some(200),
                    marked_at_ms: None,
                    ..MemoryCompactionPolicy::default()
                },
                reason: "default-cutoff stale-mark".into(),
            })
            .await
            .unwrap();

        assert!(
            outcome.changed,
            "apply mode with a record below the cutoff must report changed=true",
        );
        assert!(
            outcome.stale_marked.contains(&id_default),
            "the defaulted record must appear in stale_marked: got {:?}",
            outcome.stale_marked,
        );
        let got_default = s.get(id_default).await.unwrap().unwrap();
        assert_eq!(
            got_default.metadata["stale_context"]["marked_at_ms"], 200,
            "marked_at_ms must default to the before_ms cutoff (200) when \
             MemoryCompactionPolicy::marked_at_ms is None — a refactor that \
             swapped .unwrap_or(before_ms) for .unwrap_or(0) would silently \
             surface 0 here on every operator dashboard, breaking the \
             policy-cutoff anchor that audit reconstruction binds to; a \
             refactor to .unwrap_or_else(now_ms) would surface a non-\
             deterministic wall-clock timestamp instead, breaking the \
             reproducibility property that plan_compaction is currently a \
             pure function of (records, policy, reason); a refactor that \
             omitted the field entirely when policy.marked_at_ms is None \
             would make stale_context.marked_at_ms absent here, breaking \
             JSON consumers that assume a uniform schema across records",
        );
        assert_eq!(
            got_default.metadata["stale_context"]["reason"], "default-cutoff stale-mark",
            "reason must travel verbatim into stale_context alongside the \
             marked_at_ms default arm — pinning both fields anchors the \
             two-field stale_context contract that plan_compaction \
             constructs",
        );

        // Explicit-override arm: marked_at_ms = Some(150), mark_longterm_stale_before_ms = Some(300).
        // The recorded stale_context.marked_at_ms must equal the explicit
        // override (150), NOT the cutoff (300). Pinning the override arm
        // here prevents a 'simplify by always using before_ms' regression
        // that would silently drop the Some(M) path while leaving the
        // default arm intact.
        let outcome = s
            .compact(MemoryCompactionRequest {
                mode: MemoryRepairMode::Apply,
                policy: MemoryCompactionPolicy {
                    mark_longterm_stale_before_ms: Some(300),
                    marked_at_ms: Some(150),
                    ..MemoryCompactionPolicy::default()
                },
                reason: "explicit-override stale-mark".into(),
            })
            .await
            .unwrap();

        assert!(outcome.changed);
        assert!(outcome.stale_marked.contains(&id_explicit));
        let got_explicit = s.get(id_explicit).await.unwrap().unwrap();
        assert_eq!(
            got_explicit.metadata["stale_context"]["marked_at_ms"], 150,
            "explicit marked_at_ms = Some(150) must win over the cutoff \
             (300) — a refactor that always used before_ms (dropping the \
             .unwrap_or arm) would silently surface 300 here while the \
             default-arm assertion above would still pass; pinning the \
             explicit override on the same record family anchors both \
             halves of the .unwrap_or contract",
        );
    }

    #[test]
    fn memory_error_display_messages_pin_five_string_variant_format_strings() {
        let worker = format!("{}", MemoryError::Worker("channel closed".into()));
        assert_eq!(
            worker, "worker: channel closed",
            "MemoryError::Worker Display drifted (typo or dropped 'worker:' prefix regression class)"
        );

        let not_found = format!("{}", MemoryError::RecordNotFound(Uuid::nil()));
        assert_eq!(
            not_found, "memory record 00000000-0000-0000-0000-000000000000 not found",
            "MemoryError::RecordNotFound Display drifted (typo or dropped qualifier regression class)"
        );

        let parent_mismatch = format!(
            "{}",
            MemoryError::ParentMismatch {
                id: Uuid::nil(),
                expected: Some(Uuid::from_u128(1)),
                actual: Some(Uuid::from_u128(2)),
            }
        );
        assert!(
            parent_mismatch.contains("parent mismatch for memory 00000000-0000-0000-0000-000000000000"),
            "MemoryError::ParentMismatch id slot drifted (typo regression class): {parent_mismatch}"
        );
        assert!(
            parent_mismatch.contains("expected Some(") && parent_mismatch.contains("00000000-0000-0000-0000-000000000001"),
            "MemoryError::ParentMismatch expected slot drifted (slot-swap or debug-vs-display regression class): {parent_mismatch}"
        );
        assert!(
            parent_mismatch.contains("actual Some(") && parent_mismatch.contains("00000000-0000-0000-0000-000000000002"),
            "MemoryError::ParentMismatch actual slot drifted (slot-swap or debug-vs-display regression class): {parent_mismatch}"
        );
        assert!(
            !parent_mismatch.contains("expected Some(00000000-0000-0000-0000-000000000002"),
            "MemoryError::ParentMismatch expected slot bound to actual value (slot-swap regression class): {parent_mismatch}"
        );
        assert!(
            !parent_mismatch.contains("actual Some(00000000-0000-0000-0000-000000000001"),
            "MemoryError::ParentMismatch actual slot bound to expected value (slot-swap regression class): {parent_mismatch}"
        );

        let invalid_repair = format!("{}", MemoryError::InvalidRepair("reason empty".into()));
        assert_eq!(
            invalid_repair, "invalid memory repair request: reason empty",
            "MemoryError::InvalidRepair Display drifted (typo or dropped 'memory' qualifier regression class — \
             would collide with A2AError::InvalidRepair)"
        );

        let invalid_compaction =
            format!("{}", MemoryError::InvalidCompaction("window empty".into()));
        assert_eq!(
            invalid_compaction, "invalid memory compaction request: window empty",
            "MemoryError::InvalidCompaction Display drifted (typo or prefix-convergence regression class)"
        );
        assert_ne!(
            invalid_compaction, invalid_repair,
            "MemoryError::InvalidCompaction must not converge with MemoryError::InvalidRepair \
             (prefix-convergence regression class would merge memory-compaction with memory-repair rejection paths)"
        );
    }

    #[test]
    fn memory_error_io_and_sqlite_and_serde_display_messages_pin_prefix_and_external_source_display_delegation(
    ) {
        let io_err = MemoryError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            "memory.sqlite missing",
        ));
        let io_message = format!("{io_err}");
        assert!(
            io_message.starts_with("io: "),
            "MemoryError::Io must surface the literal 'io: ' bootstrap-stage prefix so audit-log filters can distinguish memory-store disk faults from SQLite engine faults and JSON parse faults during read/write/repair/compaction (dropped-prefix regression class): {io_message}"
        );
        assert!(
            io_message.contains("memory.sqlite missing"),
            "MemoryError::Io must surface the inner std::io::Error Display rendering after the colon ({{0}}, not {{0:?}}); a Debug refactor would render 'Custom {{ kind: NotFound, error: ... }}' instead of the message payload (Debug-vs-Display formatting regression class on the {{0}} interpolation): {io_message}"
        );
        assert!(
            !io_message.contains("Custom {") && !io_message.contains("Os {"),
            "MemoryError::Io must NOT surface the std::io::Error Debug rendering; a Debug refactor on {{0}} would expose internal struct fields like 'Custom {{ kind: ..., error: ... }}' or 'Os {{ code: ..., kind: ..., message: ... }}' (Debug-vs-Display formatting regression class on the {{0}} interpolation): {io_message}"
        );

        let sqlite_source = rusqlite::Connection::open_in_memory()
            .expect("open in-memory")
            .execute("INSERT INTO nonexistent_table VALUES(1)", [])
            .expect_err("sqlite must fail with no such table");
        let sqlite_err = MemoryError::Sqlite(sqlite_source);
        let sqlite_message = format!("{sqlite_err}");
        assert!(
            sqlite_message.starts_with("sqlite: "),
            "MemoryError::Sqlite must surface the literal 'sqlite: ' bootstrap-stage prefix so audit-log filters can distinguish SQLite engine faults from disk faults and JSON parse faults during read/write/repair/compaction (dropped-prefix regression class): {sqlite_message}"
        );
        assert!(
            sqlite_message.contains("no such table"),
            "MemoryError::Sqlite must surface the inner rusqlite::Error Display rendering after the colon ({{0}}, not {{0:?}}); rusqlite v0.31 Display renders missing-table failures as 'no such table: ...', a Debug refactor on {{0}} would render 'SqliteFailure(Error {{ code: ..., extended_code: ... }}, Some(\"...\"))' instead (Debug-vs-Display formatting regression class on the {{0}} interpolation): {sqlite_message}"
        );
        assert!(
            !sqlite_message.contains("SqliteFailure(Error"),
            "MemoryError::Sqlite must NOT surface the rusqlite::Error Debug rendering; a Debug refactor on {{0}} would expose 'SqliteFailure(Error {{ code: ..., extended_code: ... }}, Some(\"...\"))' internal struct fields and leak extended_code internals (Debug-vs-Display formatting regression class on the {{0}} interpolation): {sqlite_message}"
        );

        let serde_source =
            serde_json::from_str::<serde_json::Value>("not json").expect_err("parse must fail");
        let serde_err = MemoryError::Serde(serde_source);
        let serde_message = format!("{serde_err}");
        assert!(
            serde_message.starts_with("serde: "),
            "MemoryError::Serde must surface the literal 'serde: ' bootstrap-stage prefix so audit-log filters can distinguish JSON parse faults from disk faults and SQLite engine faults during memory-record marshalling (dropped-prefix regression class): {serde_message}"
        );
        assert!(
            serde_message.contains("expected"),
            "MemoryError::Serde must surface the inner serde_json::Error Display rendering after the colon (serde_json renders parse failures with 'expected ...' Display strings); a Debug refactor on {{0}} would render 'Error(\"...\", line: N, column: M)' instead (Debug-vs-Display formatting regression class on the {{0}} interpolation): {serde_message}"
        );
        assert!(
            !serde_message.contains("Error("),
            "MemoryError::Serde must NOT surface the serde_json::Error Debug rendering; a Debug refactor on {{0}} would expose 'Error(\"...\", line: N, column: M)' buffer-position structs (Debug-vs-Display formatting regression class on the {{0}} interpolation): {serde_message}"
        );

        assert_ne!(
            io_message, sqlite_message,
            "MemoryError::Io and MemoryError::Sqlite Display must not converge (prefix-convergence regression class): io={io_message} sqlite={sqlite_message}"
        );
        assert_ne!(
            io_message, serde_message,
            "MemoryError::Io and MemoryError::Serde Display must not converge (prefix-convergence regression class): io={io_message} serde={serde_message}"
        );
        assert_ne!(
            sqlite_message, serde_message,
            "MemoryError::Sqlite and MemoryError::Serde Display must not converge (prefix-convergence regression class): sqlite={sqlite_message} serde={serde_message}"
        );

        assert!(
            !io_message.starts_with("sqlite:") && !io_message.starts_with("serde:"),
            "MemoryError::Io must not start with 'sqlite:' or 'serde:'; a sibling-prefix swap would silently mis-route incident triage (sibling-prefix-swap regression class): {io_message}"
        );
        assert!(
            !sqlite_message.starts_with("io:") && !sqlite_message.starts_with("serde:"),
            "MemoryError::Sqlite must not start with 'io:' or 'serde:'; a sibling-prefix swap would silently mis-route incident triage (sibling-prefix-swap regression class): {sqlite_message}"
        );
        assert!(
            !serde_message.starts_with("io:") && !serde_message.starts_with("sqlite:"),
            "MemoryError::Serde must not start with 'io:' or 'sqlite:'; a sibling-prefix swap would silently mis-route incident triage (sibling-prefix-swap regression class): {serde_message}"
        );

        let string_variant_prefixes = [
            "worker:",
            "memory record ",
            "parent mismatch for memory",
            "invalid memory repair request:",
            "invalid memory compaction request:",
        ];
        for prefix in string_variant_prefixes {
            assert!(
                !io_message.starts_with(prefix),
                "MemoryError::Io must not converge with the string-variant surface '{prefix}' pinned by memory_error_display_messages_pin_five_string_variant_format_strings; a disk fault must not be mis-routed as a structured store-invariant violation (string-surface-convergence regression class): {io_message}"
            );
            assert!(
                !sqlite_message.starts_with(prefix),
                "MemoryError::Sqlite must not converge with the string-variant surface '{prefix}' pinned by memory_error_display_messages_pin_five_string_variant_format_strings; a SQLite engine fault must not be mis-routed as a structured store-invariant violation (string-surface-convergence regression class): {sqlite_message}"
            );
            assert!(
                !serde_message.starts_with(prefix),
                "MemoryError::Serde must not converge with the string-variant surface '{prefix}' pinned by memory_error_display_messages_pin_five_string_variant_format_strings; a JSON parse fault must not be mis-routed as a structured store-invariant violation (string-surface-convergence regression class): {serde_message}"
            );
        }
    }

    #[test]
    fn memory_error_sqlite_source_delegation_pin_returns_inner_rusqlite_error_via_std_error_source()
    {
        use std::error::Error;

        let inner = rusqlite::Connection::open_in_memory()
            .expect("open in-memory")
            .execute("INSERT INTO nonexistent_table VALUES(1)", [])
            .expect_err("sqlite must fail with no such table");
        let expected_display = format!("{inner}");
        let err = MemoryError::Sqlite(inner);
        let source = err.source().expect(
            "MemoryError::Sqlite must surface the inner rusqlite::Error via std::error::Error::source so daemon-side memory repair/compaction retry-policy classifiers can walk the error chain and downcast source() to rusqlite::Error to extract rusqlite::ErrorCode for distinct retry decisions (SQLITE_BUSY/LOCKED retry with backoff, SQLITE_CORRUPT escalates to operator-attention); a refactor that converted the variant from #[from] to a hand-written Error impl returning None (under a 'simpler error wrapping' rationale) would silently change source() to return None while leaving Display intact (dropped-source-attribute regression class)",
        );
        assert_eq!(
            format!("{source}"),
            expected_display,
            "MemoryError::Sqlite source() Display must match a direct format!() of the same rusqlite::Error verbatim; a refactor that swapped the inner field type to Box<dyn Error + Send + Sync> or any other wrapper would silently break daemon-side downcasts even though the wrapper's Display would continue to flow through {{0}} (concrete-source-type regression class)"
        );
        assert!(
            source.downcast_ref::<rusqlite::Error>().is_some(),
            "MemoryError::Sqlite source() must downcast_ref to rusqlite::Error so daemon-side memory retry-policy classifiers can extract rusqlite::ErrorCode for retry decisions; a refactor that wrapped the inner in a project-local newtype (e.g., MemorySqliteError(rusqlite::Error) under a 'distinguish memory-store SQLite failures from sibling SQL surfaces' rationale) would silently break downcast_ref::<rusqlite::Error>() at every downstream callsite (concrete-source-type downcast regression class)"
        );
    }

    #[test]
    fn memory_error_io_source_delegation_pin_returns_inner_std_io_error_via_std_error_source() {
        use std::error::Error;

        let inner = std::io::Error::new(
            std::io::ErrorKind::PermissionDenied,
            "memory.sqlite: permission denied",
        );
        let expected_display = format!("{inner}");
        let err = MemoryError::Io(inner);
        let source = err.source().expect(
            "covenant_memory::MemoryError::Io must surface the inner std::io::Error via std::error::Error::source so daemon-side memory retry-policy classifiers can walk the error chain and downcast source() to std::io::Error to extract io::ErrorKind for distinct retry decisions on memory-store IO (Interrupted retries immediately, WouldBlock backs off briefly, PermissionDenied escalates as security-sensitive on a non-readable memory store, NoSpace blocks new write/repair/compaction); a refactor that converted the variant from #[from] to a hand-written Error impl returning None (under a 'simpler error wrapping' rationale) would silently change source() to return None while leaving Display intact (dropped-source-attribute regression class)",
        );
        assert_eq!(
            format!("{source}"),
            expected_display,
            "covenant_memory::MemoryError::Io source() Display must match a direct format!() of the same std::io::Error verbatim; a refactor that swapped the inner field type to Box<dyn Error + Send + Sync> or any other wrapper would silently break daemon-side downcasts even though the wrapper's Display would continue to flow through {{0}} (concrete-source-type regression class)"
        );
        let kind = source.downcast_ref::<std::io::Error>().map(|e| e.kind());
        assert_eq!(
            kind,
            Some(std::io::ErrorKind::PermissionDenied),
            "covenant_memory::MemoryError::Io source() must downcast_ref to std::io::Error so daemon-side memory retry-policy classifiers can extract io::ErrorKind for retry decisions on memory-store IO; a refactor that wrapped the inner in a project-local newtype (e.g., MemoryIoError(std::io::Error) under a 'tag memory-store IO failures distinctly from sibling Io variants in other crates' rationale) would silently break downcast_ref::<std::io::Error>() at every downstream callsite that classifies memory-store IO faults (concrete-source-type downcast regression class)"
        );
    }

    #[test]
    fn memory_error_serde_source_delegation_pin_returns_inner_serde_json_error_via_std_error_source(
    ) {
        use std::error::Error;

        let inner =
            serde_json::from_str::<serde_json::Value>("not json").expect_err("parse must fail");
        let expected_display = format!("{inner}");
        let err = MemoryError::Serde(inner);
        let source = err.source().expect(
            "covenant_memory::MemoryError::Serde must surface the inner serde_json::Error via std::error::Error::source so daemon-side memory diagnostics can walk the error chain and downcast source() to serde_json::Error to inspect line/column or classify() for malformed-row identification (line/column points the operator at the offending payload buffer offset, classify() distinguishes Syntax-vs-Data-vs-EOF for incident triage on a corrupted JSON-bearing memory record); a refactor that converted the variant from #[from] to a hand-written Error impl returning None (under a 'simpler error wrapping' rationale) would silently change source() to return None while leaving Display intact (dropped-source-attribute regression class)",
        );
        assert_eq!(
            format!("{source}"),
            expected_display,
            "covenant_memory::MemoryError::Serde source() Display must match a direct format!() of the same serde_json::Error verbatim; a refactor that swapped the inner field type to Box<dyn Error + Send + Sync> or any other wrapper would silently break daemon-side downcasts even though the wrapper's Display would continue to flow through {{0}} (concrete-source-type regression class)"
        );
        assert!(
            source.downcast_ref::<serde_json::Error>().is_some(),
            "covenant_memory::MemoryError::Serde source() must downcast_ref to serde_json::Error so daemon-side memory diagnostics can call serde_json::Error::line/column/classify for malformed-row identification; a refactor that wrapped the inner in a project-local newtype (e.g., MemorySerdeError(serde_json::Error) under a 'consolidate parse errors into one Wire variant' rationale) would silently break downcast_ref::<serde_json::Error>() at every downstream callsite that classifies memory-store JSON parse faults (concrete-source-type downcast regression class)"
        );
    }
}
