yantrikdb 0.23.0

use rusqlite::params;
use std::collections::HashMap;

use crate::error::Result;
use crate::types::Stats;

use super::{now, YantrikDB};

/// Ceiling on the bounded global ingest queue (`oplog` rows with `applied=0`).
///
/// Single source of truth as of v0.10 Item 4a.6a. This was previously declared
/// three times as a function-local `const` (`log_op_pending`,
/// `log_op_pending_for_reembed_queue`, and now the shared admission check) — a
/// value that gates every foreground write should not be able to drift between
/// its copies.
pub(crate) const MAX_PENDING_OPS: i64 = 10_000;

/// Ops drained per pass by [`YantrikDB::drain_materializer_backlog`].
///
/// The conn lock is released between passes, so this is the granularity at
/// which a foreground drain yields to concurrent writers and to the
/// background materializer pool. Matches the pool's own `DRAIN_BATCH_SIZE`
/// order of magnitude; larger would hold the lock longer for no gain.
const THINK_DRAIN_BATCH: usize = 64;

/// Hard ceiling on ops a single `think()` will drain.
///
/// Deliberately below [`MAX_PENDING_OPS`] (10_000): a foreground call must
/// stay bounded even when the queue is at its admission ceiling. See
/// [`YantrikDB::drain_materializer_backlog`] for what is given up at the
/// boundary and why a bounded drain beats an unbounded one.
const THINK_DRAIN_BUDGET: usize = 4096;

/// `extractor_version` written by the bound extractor (occurrence-local
/// binding with evidence spans; `graph::extract_relations_bound`).
pub(crate) const BOUND_EXTRACTOR_VERSION: &str = "2.0";
/// Refusal rows kept per memory.
const REFUSAL_LEDGER_CAP_PER_MEMORY: usize = 8;

/// The claims table stores spans as `INTEGER`; a span past `i32::MAX`
/// (never on a memory this engine accepts) is stored as none.
fn span_columns(span: Option<(usize, usize)>) -> (Option<i32>, Option<i32>) {
    match span {
        Some((s, e)) => (i32::try_from(s).ok(), i32::try_from(e).ok()),
        None => (None, None),
    }
}

impl YantrikDB {
    /// Get engine statistics. Optionally filter memory counts by namespace.
    pub fn stats(&self, namespace: Option<&str>) -> Result<Stats> {
        let conn = self.conn.lock();
        let ns_filter = namespace
            .map(|ns| format!(" AND namespace = '{}'", ns.replace('\'', "''")))
            .unwrap_or_default();
        let active = conn.query_row(
            &format!(
                "SELECT COUNT(*) FROM memories WHERE consolidation_status = 'active'{}",
                ns_filter
            ),
            [],
            |row| row.get(0),
        )?;
        let consolidated = conn.query_row(
            &format!(
                "SELECT COUNT(*) FROM memories WHERE consolidation_status = 'consolidated'{}",
                ns_filter
            ),
            [],
            |row| row.get(0),
        )?;
        let tombstoned = conn.query_row(
            &format!(
                "SELECT COUNT(*) FROM memories WHERE consolidation_status = 'tombstoned'{}",
                ns_filter
            ),
            [],
            |row| row.get(0),
        )?;
        let archived = conn.query_row(
            &format!(
                "SELECT COUNT(*) FROM memories WHERE storage_tier = 'cold'{}",
                ns_filter
            ),
            [],
            |row| row.get(0),
        )?;
        let edges = conn.query_row(
            "SELECT COUNT(*) FROM edges WHERE tombstoned = 0",
            [],
            |row| row.get(0),
        )?;
        let entities = conn.query_row("SELECT COUNT(*) FROM entities", [], |row| row.get(0))?;
        let chunk_vectors: u64 = conn
            .query_row("SELECT COUNT(*) FROM memory_chunks", [], |row| {
                row.get::<_, i64>(0)
            })
            .map(|n| n.max(0) as u64)
            .unwrap_or(0);
        let operations = conn.query_row("SELECT COUNT(*) FROM oplog", [], |row| row.get(0))?;
        let open_conflicts = conn.query_row(
            "SELECT COUNT(*) FROM conflicts WHERE status = 'open'",
            [],
            |row| row.get(0),
        )?;
        let resolved_conflicts = conn.query_row(
            "SELECT COUNT(*) FROM conflicts WHERE status IN ('resolved', 'dismissed')",
            [],
            |row| row.get(0),
        )?;
        let pending_triggers = conn.query_row(
            "SELECT COUNT(*) FROM trigger_log WHERE status = 'pending'",
            [],
            |row| row.get(0),
        )?;
        let active_patterns = conn.query_row(
            "SELECT COUNT(*) FROM patterns WHERE status = 'active'",
            [],
            |row| row.get(0),
        )?;
        // v0.10 Item 1: how many records are currently superseded
        // (selected active inbound supersedes edge).
        let superseded_records = conn.query_row(
            "SELECT COUNT(DISTINCT target_rid) FROM record_links \
             WHERE link_type = 'supersedes' \
             AND status = 'active' AND selection_state = 'selected'",
            [],
            |row| row.get(0),
        )?;
        // C5b pollution census: apostrophe-bearing entities (phantom
        // possessives + contraction entities minted by the pre-C5a
        // tokenizer) and how many the migration has aliased to their
        // canonicals. before/after on these two IS the migration's
        // success metric. Best-effort on old schemas.
        let apostrophe_entities: u64 = conn
            .query_row(
                "SELECT COUNT(*) FROM entities WHERE name LIKE '%''%'",
                [],
                |row| row.get::<_, i64>(0),
            )
            .map(|n| n.max(0) as u64)
            .unwrap_or(0);
        let possessive_aliases: u64 = conn
            .query_row(
                "SELECT COUNT(*) FROM entity_aliases \
                 WHERE source = 'possessive_migration_v1'",
                [],
                |row| row.get::<_, i64>(0),
            )
            .map(|n| n.max(0) as u64)
            .unwrap_or(0);
        let provenance_verified_records = conn.query_row(
            &format!(
                "SELECT COUNT(*) FROM memories \
                 WHERE consolidation_status != 'tombstoned' \
                 AND json_valid(metadata) \
                 AND json_extract(metadata, '$.provenance_verified') = 1{}",
                ns_filter
            ),
            [],
            |row| row.get(0),
        )?;
        let unverified_user_source_records = conn.query_row(
            &format!(
                "SELECT COUNT(*) FROM memories \
                 WHERE consolidation_status != 'tombstoned' \
                 AND source = 'user' \
                 AND COALESCE(\
                   CASE WHEN json_valid(metadata) \
                     THEN json_extract(metadata, '$.provenance_verified') END,\
                   0\
                 ) != 1{}",
                ns_filter
            ),
            [],
            |row| row.get(0),
        )?;
        let grouped_counts =
            |value_sql: &str, extra_filter: &str| -> Result<HashMap<String, i64>> {
                let sql = format!(
                    "SELECT {value_sql}, COUNT(*) FROM memories \
                 WHERE consolidation_status != 'tombstoned' \
                 {extra_filter}{ns_filter} GROUP BY 1"
                );
                let mut stmt = conn.prepare(&sql)?;
                let rows = stmt.query_map([], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
                })?;
                let mut counts = HashMap::new();
                for row in rows {
                    let (value, count) = row?;
                    counts.insert(value, count);
                }
                Ok(counts)
            };
        let provenance_source_counts =
            grouped_counts("COALESCE(NULLIF(source, ''), 'unknown')", "")?;
        let provenance_method_counts = grouped_counts(
            "COALESCE(NULLIF(CASE WHEN json_valid(metadata) \
               THEN CAST(json_extract(metadata, '$.provenance_method') AS TEXT) END, ''), \
             'unmarked')",
            "",
        )?;
        let unverified_source_counts = grouped_counts(
            "COALESCE(NULLIF(source, ''), 'unknown')",
            "AND COALESCE(CASE WHEN json_valid(metadata) \
               THEN json_extract(metadata, '$.provenance_verified') END, 0) != 1 ",
        )?;
        let recall_candidate_cap_bound_by_namespace_since_boot: HashMap<String, u64> = self
            .recall_candidate_cap_bound_since_boot
            .lock()
            .iter()
            .filter(|(key, _)| namespace.map_or(true, |ns| key.as_str() == ns))
            .map(|(key, count)| (key.clone(), *count))
            .collect();
        let recall_candidate_cap_bound_since_boot =
            recall_candidate_cap_bound_by_namespace_since_boot
                .values()
                .copied()
                .sum();
        let synthesis_fanout_cap = Self::synthesis_fanout_cap_from_conn(&conn)?;
        let synthesis_namespace_filter = namespace
            .map(|ns| format!(" AND d.namespace = '{}'", ns.replace('\'', "''")))
            .unwrap_or_default();
        let (
            synthesis_fanout_current_high_water,
            synthesis_fanout_sources_at_cap,
            synthesis_fanout_sources_over_cap,
        ): (i64, i64, i64) = conn.query_row(
            &format!(
                "WITH fanout AS ( \
                    SELECT d.source_rid, COUNT(DISTINCT d.synthesis_rid) AS n \
                    FROM synthesis_dependencies d \
                    JOIN memories s ON s.rid = d.synthesis_rid \
                    WHERE s.synthesis_state = 'verified' \
                      AND s.consolidation_status = 'active'{} \
                    GROUP BY d.source_rid \
                 ) \
                 SELECT COALESCE(MAX(n), 0), \
                        COALESCE(SUM(CASE WHEN n = ?1 THEN 1 ELSE 0 END), 0), \
                        COALESCE(SUM(CASE WHEN n > ?1 THEN 1 ELSE 0 END), 0) \
                 FROM fanout",
                synthesis_namespace_filter
            ),
            rusqlite::params![synthesis_fanout_cap as i64],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
        )?;
        drop(conn);

        Ok(Stats {
            active_memories: active,
            consolidated_memories: consolidated,
            tombstoned_memories: tombstoned,
            archived_memories: archived,
            edges,
            entities,
            operations,
            open_conflicts,
            resolved_conflicts,
            pending_triggers,
            active_patterns,
            scoring_cache_entries: self.scoring_cache.read().len(),
            vec_index_entries: self.search_state.load().vec_index.len(),
            graph_index_entities: self.graph_index.read().entity_count(),
            graph_index_edges: self.graph_index.read().edge_count(),
            status_read_policy: if self.status_read_policy() {
                "exclude_superseded".to_string()
            } else {
                "legacy".to_string()
            },
            superseded_records,
            superseded_served_since_boot: self
                .superseded_served_since_boot
                .load(std::sync::atomic::Ordering::Relaxed),
            recall_candidate_cap: super::recall::MAX_OVERSAMPLED_RECALL_CANDIDATES,
            recall_candidate_cap_namespace_capacity:
                super::recall::MAX_TRACKED_RECALL_LIMIT_NAMESPACES,
            recall_candidate_cap_bound_since_boot,
            recall_candidate_cap_bound_by_namespace_since_boot,
            recall_candidate_cap_namespace_stats_truncated_since_boot: self
                .recall_candidate_cap_namespace_stats_truncated_since_boot
                .load(std::sync::atomic::Ordering::Relaxed),
            synthesis_fanout_cap,
            synthesis_fanout_refused_since_boot: self
                .synthesis_fanout_refused_since_boot
                .load(std::sync::atomic::Ordering::Relaxed),
            synthesis_fanout_current_high_water: synthesis_fanout_current_high_water.max(0)
                as usize,
            synthesis_fanout_sources_at_cap,
            synthesis_fanout_sources_over_cap,
            provenance_gate_mode: self.provenance_gate_mode().as_str().to_string(),
            provenance_flagged_since_boot: self
                .provenance_flagged_since_boot
                .load(std::sync::atomic::Ordering::Relaxed),
            claim_chain_gate_mode: self.claim_chain_gate_mode().as_str().to_string(),
            claim_chain_gate_suppressed_since_boot: self
                .claim_chain_gate_suppressed_since_boot
                .lock()
                .iter()
                .map(|(k, v)| (k.clone(), *v))
                .collect(),
            foreign_sqlite_mode: self.foreign_sqlite.mode().as_str().to_string(),
            foreign_sqlite_supported: self.foreign_sqlite.supported(),
            foreign_sqlite_active: self.foreign_sqlite.active(),
            foreign_sqlite_tainted: self.foreign_sqlite.tainted(),
            foreign_sqlite_detected_since_boot: self.foreign_sqlite.detected_since_boot(),
            foreign_sqlite_refused_since_boot: self.foreign_sqlite.refused_since_boot(),
            foreign_commits_detected_since_boot: self
                .foreign_sqlite
                .foreign_commits_detected_since_boot(),
            integrity_check_pending: self.foreign_sqlite.integrity_check_pending(),
            integrity_checks_since_boot: self.foreign_sqlite.integrity_checks_since_boot(),
            last_integrity_check: self.foreign_sqlite.last_integrity().unwrap_or_default(),
            provenance_verified_records,
            unverified_user_source_records,
            provenance_source_counts,
            provenance_method_counts,
            unverified_source_counts,
            embedder_window_chars: self.embedder_window(),
            embedder_truncated_writes: self.embedder_truncated_write_count(),
            embedder_chunked_writes: self.embedder_chunked_write_count(),
            chunk_vectors,
            apostrophe_entities,
            possessive_aliases,
        })
    }

    /// Append an operation to the oplog with HLC and optional embedding hash.
    ///
    /// **Issue #41 brainstorm-2 §1 / brainstorm-4 §6.** Stamps the
    /// v27 `applied_generation` column with the active SearchState
    /// generation. The post-swap materializer (Layer 5) uses this
    /// column to discriminate ops already applied under generation G
    /// (skip them — they're durably indexed) from queued-during-
    /// reembed ops (`applied_generation IS NULL` — need re-encode
    /// under the new embedder). Sync writers call this AFTER their
    /// `vec_index.append` so the generation read here is the same
    /// generation the index entry was written against (the
    /// `SyncWriteGuard` held by the caller prevents reembed from
    /// completing its swap while we read).
    pub fn log_op(
        &self,
        op_type: &str,
        target_rid: Option<&str>,
        payload: &serde_json::Value,
        emb_hash: Option<&[u8]>,
    ) -> Result<String> {
        let op_id = crate::id::new_id();
        let hlc_ts = self.tick_hlc();
        let hlc_bytes = hlc_ts.to_bytes().to_vec();
        // 0.13.2: sealed on encrypted databases (see encode_oplog_payload).
        let payload_str = self.encode_oplog_payload(&serde_json::to_string(payload)?)?;
        let applied_generation: i64 = self.search_state.load().generation as i64;

        let conn = self.conn.lock();
        conn.execute(
            "INSERT INTO oplog (op_id, op_type, timestamp, target_rid, payload, \
             actor_id, hlc, embedding_hash, origin_actor, applied, applied_generation) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 1, ?10)",
            params![
                op_id,
                op_type,
                now(),
                target_rid,
                payload_str,
                self.actor_id,
                hlc_bytes,
                emb_hash,
                self.actor_id,
                applied_generation,
            ],
        )?;
        Ok(op_id)
    }
    /// In-transaction sibling of [`Self::log_op`] (v0.10 Item 4a.6a).
    ///
    /// Writes an `applied=1` oplog row using the CALLER'S transaction instead of
    /// re-locking `self.conn`. This is the primitive that lets a write path
    /// commit its row and its oplog provenance ATOMICALLY. `log_op` cannot do
    /// that: it re-acquires `self.conn` (see [`Self::log_op`]), and
    /// `parking_lot::Mutex` is not reentrant, so calling it while holding a conn
    /// guard deadlocks.
    ///
    /// Generalized from `insert_correct_op_in_tx`, which `correct()` has used
    /// since Item 3 — the shape is proven, this just parameterizes `op_type`.
    ///
    /// `applied_generation` is passed IN rather than read from `search_state`
    /// here: the caller holds a `SyncWriteGuard` pinning the generation for its
    /// whole critical section, and must stamp the same generation its vector
    /// entry was written against.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn log_op_in_tx(
        &self,
        // &Connection, not &Transaction: a `Transaction` derefs to it, so tx
        // callers are unchanged, while SAVEPOINT-guarded callers (record_batch,
        // 4a.6d-2b) have no `Transaction` to offer — same generalization
        // `advance_importance_stats_in_tx` made in 4a.6b.
        tx: &rusqlite::Connection,
        op_type: &str,
        target_rid: Option<&str>,
        payload: &serde_json::Value,
        emb_hash: Option<&[u8]>,
        embedding: Option<&[u8]>,
        applied_generation: i64,
        // 4a.6c: an idempotency claim binds to the authoritative op's id as its
        // recovery evidence, and the claim must be the FIRST statement of the
        // transaction (the v37 partial unique index on memories would otherwise
        // fire before the claim resolves a dup). So the keyed path mints the
        // op_id BEFORE the tx and passes it here; None keeps minting inline.
        preminted_op_id: Option<&str>,
    ) -> Result<String> {
        let hlc_bytes = self.tick_hlc().to_bytes().to_vec();
        self.log_op_at_hlc_in_tx(
            tx,
            op_type,
            target_rid,
            payload,
            emb_hash,
            embedding,
            applied_generation,
            preminted_op_id,
            &hlc_bytes,
        )
    }

    /// Variant used when a materialized row must carry the exact HLC of its
    /// authoritative oplog operation in the same transaction.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn log_op_at_hlc_in_tx(
        &self,
        tx: &rusqlite::Connection,
        op_type: &str,
        target_rid: Option<&str>,
        payload: &serde_json::Value,
        emb_hash: Option<&[u8]>,
        embedding: Option<&[u8]>,
        applied_generation: i64,
        preminted_op_id: Option<&str>,
        hlc_bytes: &[u8],
    ) -> Result<String> {
        let op_id = match preminted_op_id {
            Some(id) => id.to_string(),
            None => crate::id::new_id(),
        };
        // 0.13.2: sealed on encrypted databases (see encode_oplog_payload).
        let payload_str = self.encode_oplog_payload(&serde_json::to_string(payload)?)?;
        tx.execute(
            "INSERT INTO oplog \
             (op_id, op_type, timestamp, target_rid, payload, actor_id, hlc, \
              embedding_hash, origin_actor, applied, applied_generation, embedding) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 1, ?10, ?11)",
            params![
                op_id,
                op_type,
                now(),
                target_rid,
                payload_str,
                self.actor_id,
                hlc_bytes,
                emb_hash,
                self.actor_id,
                applied_generation,
                embedding,
            ],
        )?;
        Ok(op_id)
    }

    /// In-transaction sibling of [`Self::log_op_pending`] (v0.10 Item 4a.6a).
    ///
    /// Writes an `applied=0` oplog row in the caller's transaction.
    ///
    /// **Deliberately does NOT touch `pending_op_count`.** That split is
    /// load-bearing, not laziness. The counter may only be incremented AFTER the
    /// enclosing transaction commits: increment it here and a later rollback
    /// leaves the row gone but the counter raised, with nothing to bring it back
    /// down — [`Self::mark_op_applied`] only decrements rows it actually
    /// transitions, and there is no row. The drift is monotonic, and at
    /// `MAX_PENDING_OPS` of net drift the admission check wedges EVERY foreground
    /// write into `Backpressure` forever, with zero pending ops in SQL. That is
    /// the v0.7.1 counter-leak class. The caller owns the increment, and must run
    /// it only on the committed path (in `record()` the `ReservationGuard` owns
    /// it, so a post-commit unwind cannot skip it).
    ///
    /// Uses a plain `INSERT`, NOT `INSERT OR IGNORE` — and that difference is
    /// deliberate (sol, 4a.6a review finding 4).
    ///
    /// This function mints a FRESH `op_id`, so "ignored" could only ever mean an
    /// id collision or some other constraint — and swallowing it would let the
    /// record transaction commit and `record()` return `Ok` while the
    /// post-materialization enqueue silently did not happen. That write's entity
    /// materialization would then be owed to nobody, forever, which is the exact
    /// failure moving the enqueue into the transaction was meant to prevent. A
    /// plain INSERT turns that into the error it is, rolling the whole write back.
    ///
    /// (I originally copied `OR IGNORE` from [`Self::log_op_pending`] without
    /// checking whether its rationale applied here. It did not — and this doc
    /// then spent a release asserting that `log_op_pending` needed `OR IGNORE`
    /// "because it has cluster-replay callers that re-use an existing `op_id`".
    /// That was false: it mints its own id and no caller can supply one, so the
    /// ignore there could never fire for the stated reason and only ever hid real
    /// violations. #83 made it a plain INSERT too. The rationale I copied was
    /// never real anywhere — sol #83 r2 caught this paragraph still teaching it.)
    ///
    /// Returns the op_id. Deliberately does NOT touch `pending_op_count`: the
    /// counter may only move AFTER the enclosing transaction commits — increment
    /// it here and a rollback leaves the row gone but the counter raised, with
    /// nothing to bring it back down ([`Self::mark_op_applied`] only decrements
    /// rows it actually transitions, and there is no row). The drift is
    /// monotonic, and at `MAX_PENDING_OPS` of net drift the admission check wedges
    /// EVERY foreground write into `Backpressure` forever with zero pending ops in
    /// SQL. That is the v0.7.1 counter-leak class.
    pub(crate) fn log_op_pending_in_tx(
        &self,
        // &Connection, not &Transaction — same generalization as
        // `log_op_in_tx` (4a.6d-2b): Transaction derefs, so tx callers are
        // unchanged, and SAVEPOINT-guarded callers (record_with_rid,
        // 4a.6d-3) have no Transaction to offer.
        tx: &rusqlite::Connection,
        op_type: &str,
        target_rid: Option<&str>,
        payload: &serde_json::Value,
        emb_hash: Option<&[u8]>,
        embedding: Option<&[u8]>,
    ) -> Result<String> {
        let op_id = crate::id::new_id();
        let hlc_bytes = self.tick_hlc().to_bytes().to_vec();
        // 0.13.2: sealed on encrypted databases (see encode_oplog_payload).
        let payload_str = self.encode_oplog_payload(&serde_json::to_string(payload)?)?;
        tx.execute(
            "INSERT INTO oplog \
             (op_id, op_type, timestamp, target_rid, payload, actor_id, hlc, \
              embedding_hash, origin_actor, applied, embedding) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 0, ?10)",
            params![
                op_id,
                op_type,
                now(),
                target_rid,
                payload_str,
                self.actor_id,
                hlc_bytes,
                emb_hash,
                self.actor_id,
                embedding,
            ],
        )?;
        Ok(op_id)
    }

    /// The `MAX_PENDING_OPS` admission check (v0.10 Item 4a.6a).
    ///
    /// **Must be called while holding the conn lock** to be authoritative. It was
    /// originally an unlocked pre-lock read, which sol's 4a.6a review finding 1
    /// showed is a TOCTOU: at 9,999 pending, N writers all read "under the limit",
    /// then serialize under `conn` and each commit an enqueue, pushing the queue
    /// past the ceiling. Under the lock the read and the commit that acts on it
    /// are serialized, so the bound actually holds.
    ///
    /// [`Self::check_pending_backpressure_fast`] is the unlocked variant, useful
    /// only as an early reject before doing expensive work.
    pub(crate) fn check_pending_backpressure_locked(&self) -> Result<()> {
        self.check_pending_backpressure_fast()
    }

    /// Unlocked, advisory admission check — an early reject only. NOT
    /// authoritative: see [`Self::check_pending_backpressure_locked`].
    pub(crate) fn check_pending_backpressure_fast(&self) -> Result<()> {
        use std::sync::atomic::Ordering;
        let pending = self.pending_op_count.load(Ordering::Relaxed);
        if pending >= MAX_PENDING_OPS {
            return Err(crate::error::YantrikDbError::Backpressure {
                pending,
                max: MAX_PENDING_OPS,
                retry_after_ms: 100,
            });
        }
        Ok(())
    }

    // NOTE (4a.6d-2b): `log_record_ops_batch` — the post-commit batched
    // "record"-op writer introduced by #79's fix — was deleted here. Its one
    // caller, `record_batch`, now commits each item's op INSIDE its savepoint
    // via `log_op_in_tx` under a preminted op id (the id an idempotency claim
    // binds to), which is what closed #94's Err-after-commit for that path.

    /// **Decoupled write path RFC, Phase 1.**
    ///
    /// Append a *pending* operation to the oplog (applied=0) carrying the
    /// full embedding bytes. Background materializer workers will later drain
    /// these and apply them to the in-memory indexes (memories table,
    /// vec_index, graph_index, scoring_cache), flipping `applied` to 1.
    ///
    /// This is the "WAL append" step from the RFC's freeway diagram. Foreground
    /// `record()` does not call this yet — Phase 4 of the RFC flips that. For
    /// Phase 1 (this version), the API is exposed for tests and for Phase 3
    /// worker scaffolding.
    ///
    /// **NOT idempotent on op_id, and cannot be.** This rustdoc used to claim
    /// "if the same op_id is appended twice (e.g. via crash-restart replay), the
    /// second INSERT is silently skipped" — describing a capability the signature
    /// does not offer: there is no `op_id` parameter, so a caller CANNOT supply
    /// one. Every call mints a fresh id (below), so the `OR IGNORE` that claim
    /// justified could never fire for the reason given, and instead silently
    /// swallowed real constraint violations — returning `Ok(op_id)` for a row
    /// that was never written. This now uses a plain INSERT so a failure to
    /// enqueue is an error, not a fiction. (The genuine replay case is
    /// replication apply, which really does receive a remote op_id and inserts it
    /// itself.)
    pub fn log_op_pending(
        &self,
        op_type: &str,
        target_rid: Option<&str>,
        payload: &serde_json::Value,
        emb_hash: Option<&[u8]>,
        embedding: Option<&[u8]>,
    ) -> Result<String> {
        use std::sync::atomic::Ordering;
        let op_id = crate::id::new_id();
        let hlc_ts = self.tick_hlc();
        let hlc_bytes = hlc_ts.to_bytes().to_vec();
        // 0.13.2: sealed on encrypted databases (see encode_oplog_payload).
        let payload_str = self.encode_oplog_payload(&serde_json::to_string(payload)?)?;

        // **v0.7.1 perf hotfix.** Backpressure check is now an atomic
        // load against `pending_op_count` instead of `SELECT COUNT(*) FROM
        // oplog WHERE applied = 0`. The previous SQL pattern dominated
        // foreground latency when v0.7.0 wired log_op_pending into the
        // record() hot path — every write paid a Mutex<Connection> acquire
        // + index scan + drop just to check the bound. Cached counter
        // is maintained by `log_op_pending` (fetch_add on insert) and
        // `mark_op_applied` (fetch_sub on apply-win).
        // Advisory fast reject (unlocked). NOT authoritative on its own.
        self.check_pending_backpressure_fast()?;

        let conn = self.conn.lock();
        // THE authoritative check (sol 4a.6a r2 finding 1). The unlocked load
        // above is a TOCTOU: at MAX_PENDING_OPS-1, N producers all read "under the
        // limit", then serialize here and each insert, overshooting the ceiling.
        // Re-reading under the SAME lock that guards the INSERT serializes the
        // read with the write it authorizes, so the bound actually holds. 4a.6a
        // first fixed only record()'s copy of this race and left this one — the
        // instance, not the class.
        self.check_pending_backpressure_locked()?;
        conn.execute(
            "INSERT INTO oplog \
             (op_id, op_type, timestamp, target_rid, payload, \
              actor_id, hlc, embedding_hash, origin_actor, applied, embedding) \
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 0, ?10)",
            params![
                op_id,
                op_type,
                now(),
                target_rid,
                payload_str,
                self.actor_id,
                hlc_bytes,
                emb_hash,
                self.actor_id,
                embedding,
            ],
        )?;
        // **v0.7.1**: maintain the cached counter.
        // Unconditional: a plain INSERT that returned Ok inserted exactly one
        // row — anything else is an Err above. The old `changes() > 0` guard
        // existed to cope with `OR IGNORE` no-ops that could not happen for the
        // documented reason (no caller can supply an op_id), and it turned a
        // swallowed constraint violation into a silently-correct-looking count.
        //
        // Scope of that claim (sol #83 finding 3): it is about THIS statement,
        // not about durability. `conn()` is public, so a caller that opens its
        // own SAVEPOINT around this and rolls back leaves the row gone and this
        // counter high — pre-existing, and not something `changes()` ever
        // caught either. The counter is a cache over `applied = 0` and re-seeds
        // on restart; the in-transaction sibling (`log_op_pending_in_tx`)
        // deliberately leaves it alone precisely because only its committer
        // knows the outcome.
        self.pending_op_count.fetch_add(1, Ordering::Relaxed);
        Ok(op_id)
    }

    /// **Decoupled write path RFC, Phase 1.**
    ///
    /// Count of pending oplog entries (applied=0). Used by tests and by the
    /// background materializer to decide whether to wake up.
    ///
    /// **v0.7.1 hotfix:** returns the cached `pending_op_count` atomic
    /// instead of running `SELECT COUNT(*)`. The atomic is maintained by
    /// `log_op_pending` (`fetch_add` on insert) and `mark_op_applied`
    /// (`fetch_sub` on apply-win) so it's always coherent with the SQL
    /// state. Tests that mutate oplog by hand (rare, only in test
    /// helpers) can still use the SQL form via `count_pending_ops_sql`.
    pub fn count_pending_ops(&self) -> Result<i64> {
        use std::sync::atomic::Ordering;
        Ok(self.pending_op_count.load(Ordering::Relaxed))
    }

    /// SQL-backed count for tests / debug. Same shape as v0.7.0's
    /// `count_pending_ops` but routed through `read_conn`. Kept as a
    /// reconciliation oracle for the cached counter; in production paths
    /// use `count_pending_ops`.
    #[doc(hidden)]
    pub fn count_pending_ops_sql(&self) -> Result<i64> {
        let conn = self.read_conn();
        let count: i64 =
            conn.query_row("SELECT COUNT(*) FROM oplog WHERE applied = 0", [], |row| {
                row.get(0)
            })?;
        Ok(count)
    }

    /// **Decoupled write path RFC, Phase 1.**
    ///
    /// Mark a pending oplog entry as materialized. Called by the background
    /// worker after it has applied the op to the in-memory indexes.
    ///
    /// **Returns** `Ok(true)` iff this caller transitioned the row from
    /// `applied=0` to `applied=1`. `Ok(false)` means another worker
    /// already applied it (race on shared oplog, normal under N workers).
    ///
    /// **This is a completion ACKNOWLEDGEMENT, not a pre-work claim.** The
    /// distinction is load-bearing and was previously documented backwards
    /// ("the work is idempotent so double-execution is safe; this filter just
    /// decides which worker gets to claim the apply count"). Callers do the work
    /// FIRST and only then race on `WHERE applied = 0`, so the filter does not
    /// prevent duplicate execution — N workers draining the same pending op all
    /// perform the work, and it merely picks which one gets to record it.
    /// Retries after an error re-execute it too.
    ///
    /// So `apply_pending_ops_once` is exactly-once in its BOOKKEEPING and
    /// at-least-once in its EFFECTS. Every op handler must therefore be
    /// genuinely idempotent on its own; that is a real obligation, not a
    /// property this function confers. It was violated by the mention-count
    /// bump in `apply_materialize_record_post` (see the note there).
    /// Making this a true pre-work lease is v0.10 Item 4a.6.
    pub fn mark_op_applied(&self, op_id: &str) -> Result<bool> {
        use std::sync::atomic::Ordering;
        let conn = self.conn.lock();
        let changed = conn.execute(
            "UPDATE oplog SET applied = 1 WHERE op_id = ?1 AND applied = 0",
            params![op_id],
        )?;
        let won = changed > 0;
        // **v0.7.1**: decrement the cached counter only when this caller
        // actually transitioned the row. Mirrors log_op_pending's
        // increment-only-on-real-insert pattern; keeps the atomic
        // coherent with SQL applied-state under N concurrent workers.
        if won {
            self.pending_op_count.fetch_sub(1, Ordering::Relaxed);
        }
        Ok(won)
    }

    /// **Decoupled write path RFC, Phase 3 scaffolding.**
    ///
    /// Drain up to `limit` pending oplog entries (applied=0) and apply each
    /// to the engine's in-memory indexes. Returns the number of ops actually
    /// applied this pass. Idempotent on re-entry — already-applied ops are
    /// skipped via the `applied = 0` filter.
    ///
    /// This is the worker's main-loop body as a sync function. Phase 3.5
    /// will wrap it in a thread spawn + condvar wake + Drop-based shutdown.
    /// Phase 4 will switch foreground `record()` to call `log_op_pending()`
    /// instead of materializing inline, at which point this drain becomes
    /// the production write-completion path.
    ///
    /// Op-type dispatch in Phase 3 is intentionally a stub: each op type
    /// has a placeholder materializer that just marks the op applied. Phase 4
    /// fills in the actual application logic (memories INSERT, vec_index
    /// update, graph_index update, scoring_cache insert) — and at that point
    /// foreground record() can stop doing it inline.
    /// THE materializer drain query — one definition, shared by the drain
    /// itself and by the test that pins its query plan (#113).
    ///
    /// It is a const rather than an inline literal specifically so the
    /// plan-assertion test cannot pin a private copy: a test that asserted
    /// against its own duplicated SQL would keep passing while this query
    /// drifted back into a full scan. The index it depends on is
    /// `idx_oplog_pending_ordered` — partial on `(hlc, op_id)`, i.e. on the
    /// SORT KEYS, because a partial index on the filter column alone cannot
    /// serve the ORDER BY and SQLite will silently prefer a full history
    /// scan via `idx_oplog_hlc`. Changing the ORDER BY here without changing
    /// that index re-introduces the idle-CPU defect.
    pub(crate) const PENDING_OPS_QUERY: &'static str =
        "SELECT op_id, op_type, payload, embedding_model FROM oplog \
         WHERE applied = 0 \
         ORDER BY hlc, op_id \
         LIMIT ?1";

    pub fn apply_pending_ops_once(&self, limit: usize) -> Result<usize> {
        // **Issue #41 Layer 5.** Pull `embedding_model` alongside the
        // standard tuple so we can dispatch queued-during-reembed
        // writes through the re-encode path. embedding_model IS NOT
        // NULL is the v27 signature for record_queued ops (sync
        // record() leaves it NULL).
        let pending: Vec<(String, String, String, Option<String>)> = {
            let conn = self.read_conn();
            let mut stmt = conn.prepare(Self::PENDING_OPS_QUERY)?;
            let rows = stmt.query_map(params![limit as i64], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, Option<String>>(3)?,
                ))
            })?;
            rows.collect::<std::result::Result<Vec<_>, _>>()?
        };
        // 0.13.2: payloads are sealed on encrypted databases. Decode HERE,
        // at the single point every apply_* dispatch flows through, so the
        // materializers keep taking plaintext JSON and no future op type
        // can be added that forgets to unseal (the copy-a-pattern class).
        let pending: Vec<(String, String, String, Option<String>)> = pending
            .into_iter()
            .map(|(op_id, op_type, payload, model)| {
                self.decode_oplog_payload(&payload)
                    .map(|p| (op_id, op_type, p, model))
            })
            .collect::<Result<Vec<_>>>()?;

        let mut applied = 0usize;
        for (op_id, op_type, payload, embedding_model) in &pending {
            match op_type.as_str() {
                // **Phase 4.3 — saga task 3.** This is the only op_type
                // whose dispatch is *real* materialization work (the
                // unbounded entity/relation loops that used to be on the
                // foreground request path). Foreground enqueues; worker
                // applies. See docs/phase_4_3_design.md for the contract.
                crate::engine::op_types::OP_MATERIALIZE_RECORD_POST => {
                    match self.apply_materialize_record_post(payload) {
                        Ok(()) => {
                            // Only count this apply if THIS worker won
                            // the race to flip applied=0 -> applied=1.
                            // Other workers may have done duplicate work
                            // (idempotent), but exactly one gets the count.
                            if self.mark_op_applied(op_id)? {
                                applied += 1;
                            }
                        }
                        Err(e) => {
                            // Don't mark applied — leave pending for retry
                            // on next tick. Workers race-safe via the
                            // applied=0 filter, so a transient failure
                            // doesn't lose the op.
                            tracing::warn!(
                                target: "yantrikdb::ingest::materialize",
                                op_id = %op_id,
                                op_type = %op_type,
                                error = %e,
                                "post-record materialization failed; leaving pending for retry"
                            );
                        }
                    }
                }
                // **Phase 4.3 Commit C — saga task 19.** Cluster-mode
                // sibling of OP_MATERIALIZE_RECORD_POST. Same race-safety
                // semantics; the difference is the dispatch logic uses
                // caller-supplied entity list with no extraction.
                crate::engine::op_types::OP_MATERIALIZE_RECORD_WITH_RID_POST => {
                    match self.apply_materialize_record_with_rid_post(payload) {
                        Ok(()) => {
                            if self.mark_op_applied(op_id)? {
                                applied += 1;
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                target: "yantrikdb::ingest::materialize",
                                op_id = %op_id,
                                op_type = %op_type,
                                error = %e,
                                "post-record-with-rid materialization failed; leaving pending for retry"
                            );
                        }
                    }
                }
                // **Issue #41 Layer 5 — Queue-mode record drain.** The
                // signature embedding_model IS NOT NULL identifies an
                // op that was queued by `record_queued` during a
                // reembed cutover. It carries TEXT (not an embedding).
                // After the swap completes, the materializer drains
                // these ops by re-encoding the text under the active
                // generation's embedder + applying directly into the
                // memories table + new vec_index at the new gen.
                //
                // While reembed is still in flight (meta.reembed_state
                // set), defer: leave applied=0 so the next tick after
                // completion picks it up. This matches brainstorm-2 §5
                // "materializer fully PAUSED during reembed" (we pause
                // only this op class — other op types still drain).
                "record" if embedding_model.is_some() => {
                    if self.reembed_status().is_some() {
                        tracing::trace!(
                            target: "yantrikdb::ingest::materialize",
                            op_id = %op_id,
                            "Layer 5: reembed in flight; deferring queued record"
                        );
                        // No mark_op_applied — leave for next tick.
                        continue;
                    }
                    match self.apply_queued_reembed_record(payload) {
                        Ok(()) => {
                            if self.mark_op_applied(op_id)? {
                                applied += 1;
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                target: "yantrikdb::ingest::materialize",
                                op_id = %op_id,
                                op_type = %op_type,
                                error = %e,
                                "Layer 5 queued reembed record apply failed; leaving pending for retry"
                            );
                        }
                    }
                }
                "record" | "forget" | "relate" | "correct" | "consolidate" => {
                    tracing::trace!(
                        target: "yantrikdb::ingest::materialize",
                        op_id = %op_id,
                        op_type = %op_type,
                        "phase 3 stub: marking pending op as applied without inline materialization"
                    );
                    if self.mark_op_applied(op_id)? {
                        applied += 1;
                    }
                }
                other => {
                    tracing::warn!(
                        target: "yantrikdb::ingest::materialize",
                        op_id = %op_id,
                        op_type = %other,
                        "unknown op_type in pending oplog — skipping"
                    );
                }
            }
        }

        Ok(applied)
    }

    /// Drain the materializer backlog to a fixed point, bounded.
    ///
    /// **Why this exists (#95).** Foreground `record()` no longer extracts
    /// entities or claims inline — it enqueues `OP_MATERIALIZE_RECORD_POST`
    /// and returns (`engine/record.rs`). The extraction that actually
    /// populates `entities`, `memory_entities` and `claims` runs in
    /// [`Self::apply_materialize_record_post`], reached only from
    /// [`Self::apply_pending_ops_once`] — which, before this, was called
    /// only by the background materializer pool on its 100 ms idle timer
    /// ([`crate::engine::materializer`]) and had no binding surface at all.
    ///
    /// So every `record(); record(); think()` sequence RACED that timer and
    /// usually lost: the conflict scans read `claims`, `claims` was still
    /// empty, and `think()` reported `conflicts_found: 0`. Adding a `sleep`
    /// before `think()` — changing nothing else — flipped the same input to
    /// `1`. The same race is why `search_entities` and
    /// `backfill_memory_entities` returned 0 for freshly-written memories:
    /// the ops were still sitting in the oplog at `applied = 0`.
    ///
    /// **Bounded, deliberately.** `think()` is a foreground call with a
    /// latency budget; it must not become "block until an arbitrarily large
    /// backlog clears". Two limits:
    ///
    /// - each pass drains at most [`THINK_DRAIN_BATCH`] ops, so the conn
    ///   lock is released between passes and concurrent writers make
    ///   progress;
    /// - the call applies at most [`THINK_DRAIN_BUDGET`] ops in total.
    ///
    /// The budget is the honest tradeoff and worth stating plainly: the
    /// drain query is `ORDER BY hlc, op_id` (oldest first), so on a store
    /// whose backlog exceeds the budget, `think()` clears the *oldest*
    /// ops and the just-written ones may still be pending. Determinism is
    /// therefore guaranteed for the case that matters — an interactive
    /// caller whose backlog is its own handful of writes — and degrades to
    /// today's behaviour under a backlog larger than 4096 ops, rather than
    /// degrading into an unbounded stall. A hang would be strictly worse
    /// than the bug being fixed.
    ///
    /// **Terminates.** Each pass either applies at least one op or returns
    /// zero and breaks. Ops that a materializer refuses (an apply error, or
    /// a record deferred mid-reembed) stay `applied = 0` and are simply not
    /// counted, so a queue of nothing-but-deferred ops yields `0` on the
    /// first pass and exits — it cannot spin.
    ///
    /// **Cannot deadlock.** `apply_pending_ops_once` and the materializers
    /// it dispatches to acquire [`Self::conn`] themselves, and that mutex
    /// is not reentrant. Every caller must therefore hold NO connection
    /// guard across this call. `think()` satisfies that by calling it as
    /// its first statement, before any phase takes a guard.
    ///
    /// **Safe against the background pool.** Concurrent foreground and
    /// worker drains are the design `apply_pending_ops_once` already
    /// documents: the materializers are idempotent, and `mark_op_applied`
    /// awards the count to exactly one racer. A pass that loses every race
    /// returns `0` while the effects it applied are nonetheless in place,
    /// which is why breaking on zero is correct rather than premature.
    ///
    /// Returns the number of ops this call was credited with applying.
    pub(crate) fn drain_materializer_backlog(&self) -> Result<usize> {
        let mut total = 0usize;
        while total < THINK_DRAIN_BUDGET {
            let batch = THINK_DRAIN_BATCH.min(THINK_DRAIN_BUDGET - total);
            let applied = self.apply_pending_ops_once(batch)?;
            if applied == 0 {
                break;
            }
            total += applied;
        }
        Ok(total)
    }

    /// **Phase 4.3 — apply a queued `materialize_record_post` op.**
    ///
    /// Mirrors the post-INSERT entity/relation extraction loop that used to
    /// live on the foreground `record()` path. Now runs on the materializer
    /// thread so the foreground caller is not blocked on the unbounded
    /// loop count (5-15 entities + 0-3 relations per typical record).
    ///
    /// Idempotent: every SQL operation here is `INSERT OR IGNORE` on a
    /// natural key (entity name, memory_entities pair, edge tuple). A
    /// double-apply across worker restarts produces identical state.
    ///
    /// Payload shape (see `docs/phase_4_3_design.md`):
    ///
    /// ```json
    /// {
    ///   "rid":       "01HX...",
    ///   "text":      "<plaintext OR engine-encrypted>",
    ///   "namespace": "default",
    ///   "ts_secs":   1715184000.0,
    ///   "domain":    "general",
    ///   "source":    "user"
    /// }
    /// ```
    /// **Issue #41 Layer 5 — apply a queued-during-reembed record.**
    ///
    /// `record_queued` writes oplog rows with applied=0,
    /// embedding_model=<old_name>, and a payload carrying the
    /// memory's TEXT (not pre-encoded embedding). After reembed
    /// completes, this materializer drain re-encodes the text under
    /// the ACTIVE SearchState's embedder (the new one) and applies
    /// the row to memories + vec_index at the new generation.
    ///
    /// Idempotent: INSERT OR IGNORE on rid + the
    /// search_state-snapshot-once pattern guarantees a re-apply
    /// (from worker race or restart) produces identical state.
    ///
    /// Caller (apply_pending_ops_once) has already verified the op
    /// type is "record" AND embedding_model IS NOT NULL AND no
    /// reembed is in flight.
    fn apply_queued_reembed_record(&self, payload_json: &str) -> Result<()> {
        use crate::serde_helpers::serialize_f32;

        let payload: serde_json::Value = serde_json::from_str(payload_json).map_err(|e| {
            crate::error::YantrikDbError::InvalidInput(format!(
                "Layer 5 record drain: payload parse failed: {e}"
            ))
        })?;

        let rid = payload.get("rid").and_then(|v| v.as_str()).ok_or_else(|| {
            crate::error::YantrikDbError::InvalidInput("Layer 5 record drain: missing rid".into())
        })?;
        let memory_type = payload
            .get("type")
            .and_then(|v| v.as_str())
            .unwrap_or("episodic");
        let text = payload
            .get("text")
            .and_then(|v| v.as_str())
            .ok_or_else(|| {
                crate::error::YantrikDbError::InvalidInput(
                    "Layer 5 record drain: missing text".into(),
                )
            })?;
        let importance = payload
            .get("importance")
            .and_then(|v| v.as_f64())
            .unwrap_or(0.5);
        let valence = payload
            .get("valence")
            .and_then(|v| v.as_f64())
            .unwrap_or(0.0);
        let half_life = payload
            .get("half_life")
            .and_then(|v| v.as_f64())
            .unwrap_or(604800.0);
        let metadata = payload
            .get("metadata")
            .cloned()
            .unwrap_or_else(|| serde_json::json!({}));
        let ts = payload
            .get("created_at")
            .and_then(|v| v.as_f64())
            .unwrap_or_else(super::now);
        let namespace = payload
            .get("namespace")
            .and_then(|v| v.as_str())
            .unwrap_or("default");
        let certainty = payload
            .get("certainty")
            .and_then(|v| v.as_f64())
            .unwrap_or(0.8);
        let domain = payload
            .get("domain")
            .and_then(|v| v.as_str())
            .unwrap_or("general");
        let source = payload
            .get("source")
            .and_then(|v| v.as_str())
            .unwrap_or("user");
        let emotional_state = payload
            .get("emotional_state")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        // 4a.6c: the v37 idempotency columns, carried by keyed queued writes.
        // Absent/null (every pre-4a.6c op, every keyless write) stores NULL —
        // identical to the pre-4a.6c row shape.
        let idempotency_key = payload
            .get("idempotency_key")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());
        let origin_actor = payload
            .get("origin_actor")
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());

        // Snapshot the active SearchState. The embedder + generation
        // here are the post-swap ones (caller verified no reembed in
        // flight; the in-memory state is durable).
        let state = self.search_state.load_full();
        let embedder = state
            .embedder
            .as_ref()
            .ok_or_else(|| {
                crate::error::YantrikDbError::Inference(
                    "Layer 5 record drain: active SearchState has no embedder (cannot re-encode)"
                        .into(),
                )
            })?
            .clone();

        let new_emb = embedder.embed(text).map_err(|e| {
            crate::error::YantrikDbError::Inference(format!(
                "Layer 5 record drain: embedder failed on rid {rid:?}: {e}"
            ))
        })?;
        if new_emb.len() != state.dim() {
            return Err(crate::error::YantrikDbError::Inference(format!(
                "Layer 5 record drain: embedder returned len {} but SearchState dim {}",
                new_emb.len(),
                state.dim(),
            )));
        }
        let emb_blob = serialize_f32(&new_emb);
        let stored_emb = self.encrypt_embedding(&emb_blob)?;

        // The payload's text/metadata are already in engine-stored
        // form (record_queued passed them through). Don't double-
        // encrypt; just hand back as stored.
        let stored_text = text.to_string();
        let stored_meta = serde_json::to_string(&metadata)?;
        // v48 (#149): event-time columns from the SAME value `stored_meta`
        // is serialized from. On an encrypted store the queued payload's
        // metadata is the stored (ciphertext-string) form; bounds are then
        // None, matching a blob that exposes no extractable event time.
        let (event_time_min, event_time_max) = crate::base::datetext::event_time_bounds(&metadata);
        // v50: source_turn from the same value `stored_meta` is serialized
        // from. On an encrypted store that value is ciphertext (a JSON
        // string), the extractor sees no keys, and the column stays NULL —
        // which is exactly why the completeness marker must NOT be restored
        // in that case: this write may have persisted an (encrypted) turn
        // the column does not carry, so the trigger's '0' legitimately
        // stands and `maintain_source_turn_backfill` owes the row a
        // decrypt-and-stamp. Only a plaintext-visible metadata object was
        // stamped faithfully and may preserve the marker.
        let source_turn = crate::engine::thread::extract_source_turn(&metadata);
        let metadata_is_plaintext = metadata.is_object();

        let embedding_generation: i64 = state.generation as i64;

        // INSERT OR IGNORE on rid for idempotency. If a prior worker
        // already inserted this row (race + retry), the OR IGNORE
        // makes this a no-op AND mark_op_applied lets one worker win
        // the count.
        {
            let conn = self.conn();
            let marker_prior = if metadata_is_plaintext {
                Some(crate::engine::thread::marker_snapshot(&conn)?)
            } else {
                None
            };
            conn.execute(
                "INSERT OR IGNORE INTO memories \
                 (rid, type, text, embedding, created_at, updated_at, importance, \
                  half_life, last_access, valence, metadata, namespace, \
                  certainty, domain, source, emotional_state, embedding_generation, \
                  idempotency_key, origin_actor, event_time_min, event_time_max, \
                  source_turn) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, \
                         ?18, ?19, ?20, ?21, ?22)",
                params![
                    rid,
                    memory_type,
                    stored_text,
                    stored_emb,
                    ts,
                    ts,
                    importance,
                    half_life,
                    ts,
                    valence,
                    stored_meta,
                    namespace,
                    certainty,
                    domain,
                    source,
                    emotional_state,
                    embedding_generation,
                    idempotency_key,
                    origin_actor,
                    // v48 (#149) event time.
                    event_time_min,
                    event_time_max,
                    // v50 source turn.
                    source_turn,
                ],
            )?;
            if let Some(prior) = marker_prior {
                crate::engine::thread::marker_restore(&conn, &prior)?;
            }
        }

        // Chunked embeddings: the drain is a re-encode of TEXT under the
        // post-swap embedder, so it chunks exactly like record_text — a
        // queued long record must not silently lose its tail just
        // because it arrived during a reembed cutover. Embedded from the
        // same string the head embed used, under the same snapshot.
        let chunk_vecs: Vec<(usize, Vec<f32>)> = match self.chunk_plan(text) {
            Some(ranges) => {
                let mut cv = Vec::with_capacity(ranges.len());
                for (i, (a, b)) in ranges.iter().enumerate() {
                    let v = embedder.embed(&text[*a..*b]).map_err(|e| {
                        crate::error::YantrikDbError::Inference(format!(
                            "Layer 5 record drain: embedder failed on rid {rid:?} chunk {}: {e}",
                            i + 1
                        ))
                    })?;
                    crate::validate::validate_embedding("record_drain#chunk", &v, state.dim())?;
                    cv.push((i + 1, v));
                }
                cv
            }
            None => Vec::new(),
        };
        if !chunk_vecs.is_empty() {
            let conn = self.conn();
            for (idx, v) in &chunk_vecs {
                let blob = self.encrypt_embedding(&serialize_f32(v))?;
                // OR REPLACE: a retried drain overwrites its own prior
                // partial work rather than erroring on the PK.
                conn.execute(
                    "INSERT OR REPLACE INTO memory_chunks (rid, chunk_idx, embedding) \
                     VALUES (?1, ?2, ?3)",
                    params![rid, *idx as i64, blob],
                )?;
            }
        }

        // Append into the active vec_index. Idempotent: DeltaIndex
        // de-dupes on rid+seq.
        let seq = self
            .vec_seq
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
            + 1;
        state.vec_index.append(rid.to_string(), new_emb, seq)?;
        // Windows after the parent, same seq (a chunk hit collapses to
        // the parent, which must be findable first).
        for (idx, v) in &chunk_vecs {
            let key = crate::vector::chunk::chunk_key(rid, *idx);
            state.vec_index.append(key, v.clone(), seq)?;
        }
        if !chunk_vecs.is_empty() {
            self.note_chunked_write();
        }

        // Bump visible_seq for RYW. Layer 6 will refine the
        // generation-aware semantics; for now bump under the
        // current generation's covers_through_seq logic.
        self.bump_visible_seq(namespace, seq);

        // The two obligations every other write surface discharges and this
        // drain silently didn't (found in the 2026-08-15 surface-family
        // audit):
        //
        // 1. SCORING CACHE. Every recall lane begins `let Some(row) =
        //    cache.get(rid)` and the cache loads from SQL only at open — so
        //    a record written during a reembed cutover was durable, oplogged
        //    and vector-indexed, yet INVISIBLE to every recall until the
        //    engine restarted.
        self.cache_insert(
            rid.to_string(),
            crate::types::ScoringRow {
                created_at: ts,
                importance,
                half_life,
                last_access: ts,
                access_count: 0,
                valence,
                consolidation_status: "active".to_string(),
                synthesis_state: None,
                synthesis_axis: None,
                synthesis_granularity: None,
                memory_type: memory_type.to_string(),
                namespace: namespace.to_string(),
                certainty,
                domain: domain.to_string(),
                source: source.to_string(),
                emotional_state: emotional_state.map(|s| s.to_string()),
            },
        );

        // 2. MATERIALIZE POST-OP. record() enqueues entity extraction /
        //    claims ingestion for every write; the drain never did, so
        //    cutover-written records permanently got no entities, no graph
        //    links, no claims. Not the record_with_rid determinism contract
        //    — this is an ORIGIN write.
        let post_payload = serde_json::json!({
            "rid": rid,
            "text": stored_text,
            "namespace": namespace,
            "ts_secs": ts,
            "domain": domain,
            "source": source,
        });
        self.log_op_pending(
            crate::engine::op_types::OP_MATERIALIZE_RECORD_POST,
            Some(rid),
            &post_payload,
            None,
            None,
        )?;

        Ok(())
    }

    fn apply_materialize_record_post(&self, payload_json: &str) -> Result<()> {
        let payload: serde_json::Value = serde_json::from_str(payload_json).map_err(|e| {
            crate::error::YantrikDbError::InvalidInput(format!(
                "materialize_record_post: payload parse failed: {e}"
            ))
        })?;

        let rid = payload.get("rid").and_then(|v| v.as_str()).ok_or_else(|| {
            crate::error::YantrikDbError::InvalidInput(
                "materialize_record_post: missing rid".into(),
            )
        })?;
        let text_stored = payload
            .get("text")
            .and_then(|v| v.as_str())
            .ok_or_else(|| {
                crate::error::YantrikDbError::InvalidInput(
                    "materialize_record_post: missing text".into(),
                )
            })?;
        let namespace = payload
            .get("namespace")
            .and_then(|v| v.as_str())
            .unwrap_or("default");
        let ts_secs = payload
            .get("ts_secs")
            .and_then(|v| v.as_f64())
            .unwrap_or_else(super::now);
        let domain = payload
            .get("domain")
            .and_then(|v| v.as_str())
            .unwrap_or("general");
        let source = payload
            .get("source")
            .and_then(|v| v.as_str())
            .unwrap_or("user");

        // Decrypt the text field if engine encrypted at rest. The payload
        // stored the engine-encrypted form, so the worker decrypts before
        // running the heuristic extractor on plaintext.
        let text_owned: String = self.decrypt_text(text_stored)?;
        let text = text_owned.as_str();

        let text_tokens = crate::graph::tokenize(text);
        // The store learns how it writes each token BEFORE judging this
        // memory's candidates, so a word the store has been writing in
        // lowercase is already a word when it shows up capitalized at a
        // sentence start here.
        {
            let conn = self.conn();
            self.record_token_case_observations(&conn, text)?;
        }
        let heuristic_entities = self.extract_entities_for(text);

        // Loop A: seed heuristic entities.
        //
        // The mention bump is gated on `memory_entities` so replaying this op is
        // a no-op for the counter. It previously ran unconditionally under a
        // comment calling the statement "idempotent" — the INSERT does not FAIL
        // on conflict, but `mention_count = mention_count + 1` is not idempotent
        // at all, and this op is genuinely executed more than once:
        //
        //   - `mark_op_applied` is a completion ACKNOWLEDGEMENT, not a pre-work
        //     claim. Workers do the work and only THEN race on
        //     `UPDATE ... WHERE applied = 0`, so N workers draining the same
        //     pending op all run this loop; the filter merely picks who counts it.
        //   - on error the op stays pending and is retried, re-running the loop.
        //
        // Each duplicate inflated every heuristic entity's mention_count, which
        // feeds the IDF term in `patterns.rs` (`total_entities / (1 + mention_count)`),
        // so the drift silently skewed salience scoring.
        //
        // `memory_entities` is PK'd on (memory_rid, entity_name) and carries no
        // FK, so `INSERT OR IGNORE` + `changes()` is an exact, durable answer to
        // "is this the first time this memory's mention of this entity has been
        // recorded?" — which is precisely what the counter is supposed to count.
        // It is per-entity, so a run that died midway through the loop still
        // replays correctly: entities already counted are skipped, the rest are
        // counted once.
        if !heuristic_entities.is_empty() {
            let conn = self.conn();
            for entity in &heuristic_entities {
                let entity_type = crate::graph::classify_entity_type(entity);
                // Claim the mention first; Loop B's INSERT OR IGNORE below is a
                // no-op for anything already claimed here.
                let first_mention = conn.execute(
                    "INSERT OR IGNORE INTO memory_entities \
                     (memory_rid, entity_name, entity_name_norm) VALUES (?1, ?2, ?3)",
                    params![
                        rid,
                        entity,
                        crate::engine::thread::normalize_entity_name(entity)
                    ],
                )? > 0;
                // Self-heal AFTER the first-mention read: the repair is a
                // separate UPDATE precisely so it can never count as a
                // mention (see repair_entity_norm).
                crate::engine::thread::repair_entity_norm(&conn, rid, entity)?;
                let inc: i64 = if first_mention { 1 } else { 0 };
                conn.execute(
                    "INSERT INTO entities (name, entity_type, first_seen, last_seen, mention_count) \
                     VALUES (?1, ?2, ?3, ?3, ?4) \
                     ON CONFLICT(name) DO UPDATE SET \
                        last_seen = ?3, \
                        mention_count = mention_count + ?4, \
                        entity_type = CASE \
                            WHEN entity_type = 'unknown' AND ?2 != 'unknown' THEN ?2 \
                            ELSE entity_type END",
                    params![entity, entity_type, ts_secs, inc],
                )?;
            }
        }

        // Compose candidate set: heuristic + already-known entities.
        let mut candidates: std::collections::HashSet<String> =
            heuristic_entities.iter().cloned().collect();
        for known in self.graph_index.read().all_entity_names() {
            if crate::graph::entity_matches_text(&known, &text_tokens) {
                candidates.insert(known);
            }
        }

        if !candidates.is_empty() {
            // Loop B: memory_entities INSERT OR IGNORE.
            {
                let conn = self.conn();
                for entity in &candidates {
                    conn.execute(
                        "INSERT OR IGNORE INTO memory_entities \
                         (memory_rid, entity_name, entity_name_norm) VALUES (?1, ?2, ?3)",
                        params![
                            rid,
                            entity,
                            crate::engine::thread::normalize_entity_name(entity)
                        ],
                    )?;
                    crate::engine::thread::repair_entity_norm(&conn, rid, entity)?;
                }
            }
            // graph_index in-memory update (idempotent — add_entity/link dedupe).
            let mut gi = self.graph_index.write();
            for entity in &candidates {
                let entity_type = crate::graph::classify_entity_type(entity);
                gi.add_entity(entity, entity_type);
                gi.link_memory(rid, entity);
            }
        }

        // Loops C+D+E live in `ingest_extracted_claims` so the one-time
        // re-extraction heal (`reextract_claims`) runs the SAME code the
        // materializer runs — one definition, two callers.
        let heuristic_vec: Vec<String> = heuristic_entities.iter().cloned().collect();
        self.ingest_extracted_claims(rid, text, namespace, &heuristic_vec);

        // Audit telemetry — same shape as the foreground path emitted before.
        let features = crate::graph::analyze_text_features(text, &heuristic_vec);
        tracing::info!(
            target: "yantrikdb::audit::extraction",
            namespace = %namespace,
            memory_rid = %rid,
            domain = %domain,
            source = %source,
            extractor_version = "heuristic_v1",
            char_length = features.char_length,
            sentence_count = features.sentence_count,
            entity_count = features.entity_count,
            entities_matched_in_graph = candidates.len().saturating_sub(heuristic_entities.len()),
            negation_cue_count = features.negation_cue_count,
            temporal_cue_count = features.temporal_cue_count,
            modality_cue_count = features.modality_cue_count,
            has_compound_markers = features.has_compound_markers,
            likely_assertion = features.likely_assertion,
            "extraction audit (materialized post-record)"
        );

        Ok(())
    }

    /// Rewrite the refusal ledger rows for one memory: what the bound
    /// extractor saw and would not bind, capped so a log-shaped memory
    /// cannot flood the table. Best-effort: a ledger failure never fails
    /// extraction.
    pub(crate) fn record_extraction_refusals(
        &self,
        rid: &str,
        namespace: &str,
        refusals: &[crate::graph::ExtractionRefusal],
    ) {
        let conn = self.conn();
        if conn
            .execute(
                "DELETE FROM extraction_refusals WHERE memory_rid = ?1",
                params![rid],
            )
            .is_err()
        {
            return; // pre-v54 store (an old pack): no ledger
        }
        let ts = now();
        for r in refusals.iter().take(REFUSAL_LEDGER_CAP_PER_MEMORY) {
            let _ = conn.execute(
                "INSERT OR REPLACE INTO extraction_refusals (memory_rid, namespace, rel_type, \
                 trigger, reason, left_token, right_token, at, extractor_version, created_at) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
                params![
                    rid,
                    namespace,
                    r.rel_type,
                    r.trigger,
                    r.reason,
                    r.left,
                    r.right,
                    r.at as i64,
                    BOUND_EXTRACTOR_VERSION,
                    ts
                ],
            );
        }
    }

    /// Relation extraction + claim ingestion for one memory (the
    /// materializer's Loops C+D+E): built-in patterns as `heuristic_v1`,
    /// active learned templates as `learned_v1`, never minting a fact any
    /// extractor already recorded. Returns the number of claims written.
    /// Shared by `apply_materialize_record_post` and `reextract_claims`.
    pub(crate) fn ingest_extracted_claims(
        &self,
        rid: &str,
        text: &str,
        namespace: &str,
        heuristic_vec: &[String],
    ) -> usize {
        let mut written = 0usize;
        // Extracted claims inherit the record's temporal tag too: a memory
        // about 2024 yields claims valid from 2024, not from today.
        let event_min = self.memory_event_time_min(rid);
        // Value objects (`0.19.0`, `1985`) are never entities but must stay
        // available as relation OBJECTS, or `runs`/`born_in` claims vanish
        // with the entity junk they used to ride on (issue #213).
        let mut candidates: Vec<String> = heuristic_vec.to_vec();
        for v in crate::graph::extract_value_candidates(text) {
            if !candidates.contains(&v) {
                candidates.push(v);
            }
        }
        let extraction = crate::graph::extract_relations_bound(text, &candidates);
        self.record_extraction_refusals(rid, namespace, &extraction.refusals);
        for rel in &extraction.relations {
            // A value can be an object, never a subject: `2026 -leads-> X`
            // anchors nothing at read time and is refused there anyway.
            // And only a few relations can take a value as an object.
            if crate::graph::is_rejected_entity_name(&rel.src)
                || !crate::graph::relation_admits_value_object(&rel.rel_type, &rel.dst)
            {
                continue;
            }
            let already_exists = {
                let conn = self.conn();
                conn.query_row(
                    "SELECT COUNT(*) FROM edges WHERE src = ?1 AND rel_type = ?2 AND dst = ?3 \
                     AND namespace = ?4 AND extractor = 'heuristic_v1' AND tombstoned = 0",
                    params![rel.src, rel.rel_type, rel.dst, namespace],
                    |row| row.get::<_, i64>(0),
                )
                .unwrap_or(0)
                    > 0
            };
            if already_exists {
                continue;
            }
            let (span_start, span_end) = span_columns(rel.span);
            written += usize::from(
                self.ingest_claim_grounded(
                    &rel.src,
                    &rel.rel_type,
                    &rel.dst,
                    namespace,
                    rel.polarity,
                    &rel.modality,
                    event_min,
                    None,
                    "heuristic_v1",
                    Some(BOUND_EXTRACTOR_VERSION),
                    &rel.confidence_band,
                    Some(rid),
                    span_start,
                    span_end,
                    1.0,
                    crate::engine::claims_lane::GROUNDING_EXTRACTOR_BOUND,
                )
                .is_ok(),
            );
        }

        // Loop E: self-mined templates (see engine::graph_ops, "Self-mined
        // relation templates"). Active templates for this namespace are
        // applied to the same entity windows; a fact any extractor already
        // recorded is not minted twice.
        let templates = {
            let conn = self.conn();
            Self::active_relation_templates(&conn, namespace)
        };
        if !templates.is_empty() {
            let learned = crate::graph::extract_learned_relations(text, &candidates, &templates);
            for rel in &learned {
                if crate::graph::is_rejected_entity_name(&rel.src)
                    || !crate::graph::relation_admits_value_object(&rel.rel_type, &rel.dst)
                {
                    continue;
                }
                let already_exists = {
                    let conn = self.conn();
                    conn.query_row(
                        "SELECT COUNT(*) FROM edges WHERE src = ?1 AND rel_type = ?2 AND dst = ?3 \
                         AND namespace = ?4 AND tombstoned = 0",
                        params![rel.src, rel.rel_type, rel.dst, namespace],
                        |row| row.get::<_, i64>(0),
                    )
                    .unwrap_or(0)
                        > 0
                };
                if already_exists {
                    continue;
                }
                let (span_start, span_end) = span_columns(rel.span);
                written += usize::from(
                    self.ingest_claim_grounded(
                        &rel.src,
                        &rel.rel_type,
                        &rel.dst,
                        namespace,
                        rel.polarity,
                        &rel.modality,
                        event_min,
                        None,
                        crate::engine::graph_ops::LEARNED_CLAIM_EXTRACTOR,
                        Some(BOUND_EXTRACTOR_VERSION),
                        &rel.confidence_band,
                        Some(rid),
                        span_start,
                        span_end,
                        1.0,
                        crate::engine::claims_lane::GROUNDING_EXTRACTOR_BOUND,
                    )
                    .is_ok(),
                );
            }
        }

        written
    }

    /// **Phase 4.3 Commit C — apply a queued `materialize_record_with_rid_post` op.**
    ///
    /// Cluster-mode sibling of [`Self::apply_materialize_record_post`]. Runs the
    /// post-INSERT entity / memory_entities / graph_index updates that used
    /// to live inside the foreground SAVEPOINT block of `record_with_rid()`.
    ///
    /// **Why it differs from the heuristic path.** `record_with_rid` is the
    /// cluster determinism primitive — the leader's apply emits a payload
    /// containing the explicit `extracted_entities` slice; followers must
    /// converge to byte-identical SQL state by replaying that same slice.
    /// Running heuristic extraction on the materializer would risk
    /// divergence (extractor versions or text edge cases differing across
    /// nodes). So this dispatch arm uses the caller's entity list verbatim,
    /// no `extract_heuristic_entities` / `extract_heuristic_relations`
    /// calls.
    ///
    /// `was_new_row` payload field controls whether `entities.mention_count`
    /// gets bumped on insert. Mirrors the original SQL conditional:
    ///
    ///   `mention_count = CASE WHEN ?was_new THEN mention_count + 1 ELSE mention_count END`
    ///
    /// This preserves the contract that replayed-but-not-newly-inserted
    /// memories don't double-count entity mentions.
    ///
    /// Idempotent on every loop step (INSERT OR IGNORE, ON CONFLICT DO UPDATE
    /// with mention_count guarded by `was_new_row`).
    fn apply_materialize_record_with_rid_post(&self, payload_json: &str) -> Result<()> {
        let payload: serde_json::Value = serde_json::from_str(payload_json).map_err(|e| {
            crate::error::YantrikDbError::InvalidInput(format!(
                "materialize_record_with_rid_post: payload parse failed: {e}"
            ))
        })?;

        let rid = payload.get("rid").and_then(|v| v.as_str()).ok_or_else(|| {
            crate::error::YantrikDbError::InvalidInput(
                "materialize_record_with_rid_post: missing rid".into(),
            )
        })?;
        let _namespace = payload
            .get("namespace")
            .and_then(|v| v.as_str())
            .unwrap_or("default");
        let ts_secs = payload
            .get("ts_secs")
            .and_then(|v| v.as_f64())
            .unwrap_or_else(super::now);
        let was_new_row = payload
            .get("was_new_row")
            .and_then(|v| v.as_bool())
            .unwrap_or(false);
        let entities: Vec<String> = payload
            .get("extracted_entities")
            .and_then(|v| v.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|x| x.as_str().map(|s| s.to_string()))
                    .collect()
            })
            .unwrap_or_default();

        if entities.is_empty() {
            // Nothing to materialize — early return is the cheap idempotent path.
            return Ok(());
        }

        // Loop A: entities INSERT (mirrors the inline savepoint block exactly).
        {
            let conn = self.conn();
            for entity in &entities {
                let entity_type = crate::graph::classify_entity_type(entity);
                conn.execute(
                    "INSERT INTO entities (name, entity_type, first_seen, last_seen, mention_count) \
                     VALUES (?1, ?2, ?3, ?3, 1) \
                     ON CONFLICT(name) DO UPDATE SET \
                        last_seen = ?3, \
                        mention_count = CASE WHEN ?4 THEN mention_count + 1 ELSE mention_count END, \
                        entity_type = CASE \
                            WHEN entity_type = 'unknown' AND ?2 != 'unknown' THEN ?2 \
                            ELSE entity_type END",
                    params![entity, entity_type, ts_secs, was_new_row],
                )?;
                conn.execute(
                    "INSERT OR IGNORE INTO memory_entities \
                     (memory_rid, entity_name, entity_name_norm) VALUES (?1, ?2, ?3)",
                    params![
                        rid,
                        entity,
                        crate::engine::thread::normalize_entity_name(entity)
                    ],
                )?;
                crate::engine::thread::repair_entity_norm(&conn, rid, entity)?;
            }
        }

        // graph_index in-memory update (idempotent — add_entity/link_memory dedupe).
        {
            let mut gi = self.graph_index.write();
            for entity in &entities {
                let entity_type = crate::graph::classify_entity_type(entity);
                gi.add_entity(entity, entity_type);
                gi.link_memory(rid, entity);
            }
        }

        Ok(())
    }

    /// **Phase 6 RYW** — allocate or accept a seq for a write primitive.
    ///
    /// Single-node mode: callers pass `None` and the engine allocates a
    /// fresh seq via `vec_seq.fetch_add` (1-indexed via `+ 1`).
    ///
    /// Cluster mode (RFC 010, design lock 2026-05-07): the applier passes
    /// `Some(commit_log_index)` so the seq IS the openraft commit-log
    /// index — leader and followers thereby agree on a single global
    /// monotonic stream. The engine ratchets `vec_seq` up to at least the
    /// supplied value (via `fetch_max`) so any future single-node writes
    /// against the same engine never produce seqs that collide with the
    /// cluster-supplied stream.
    ///
    /// Returns the seq the caller should use to tag the delta entry, the
    /// oplog row, and the visible_seq bump.
    pub(crate) fn assign_seq(&self, requested: Option<u64>) -> u64 {
        use std::sync::atomic::Ordering;
        match requested {
            Some(n) => {
                self.vec_seq.fetch_max(n, Ordering::Relaxed);
                n
            }
            None => self.vec_seq.fetch_add(1, Ordering::Relaxed) + 1,
        }
    }

    /// **Phase 6 RYW** — bump the visible_seq high-water mark for a
    /// namespace. Called by record/record_with_rid and siblings after the
    /// write has been materialized into the in-memory delta. Idempotent:
    /// only advances the watermark via `fetch_max`; same-or-lower seqs
    /// are no-ops.
    ///
    /// Wakes any threads in ``recall_with_seq`` waiting on this namespace
    /// via the paired condvar.
    pub(crate) fn bump_visible_seq(&self, namespace: &str, seq: u64) {
        use std::sync::atomic::Ordering;
        // Fast path: namespace already present — single fetch_max, no
        // hashmap mutation.
        if let Some(entry) = self.visible_seq.get(namespace) {
            entry.fetch_max(seq, Ordering::Release);
        } else {
            // First write for this namespace: insert. The DashMap entry
            // API gives us insert-or-existing semantics atomically per
            // shard. If two threads race to insert the same namespace
            // for the first time, one wins and the other's fetch_max
            // converges anyway.
            self.visible_seq
                .entry(namespace.to_string())
                .or_insert_with(|| std::sync::atomic::AtomicU64::new(0))
                .fetch_max(seq, Ordering::Release);
        }
        self.visible_seq_cv.notify_all();
    }

    /// **Phase 6 RYW** — current visible_seq high-water mark for a namespace.
    /// Returns 0 for namespaces that have never been bumped.
    ///
    /// Lock-free in steady state: a DashMap shard read + an atomic load.
    pub fn visible_seq_for(&self, namespace: &str) -> u64 {
        use std::sync::atomic::Ordering;
        self.visible_seq
            .get(namespace)
            .map(|e| e.load(Ordering::Acquire))
            .unwrap_or(0)
    }

    /// **Phase 6 RYW** — wait until visible_seq[namespace] >= min_seq or
    /// the timeout expires. Returns ``Ok(())`` on watermark reached;
    /// ``Err(Error::RyWaitTimeout)`` on timeout.
    ///
    /// Callers requesting strict read-your-writes pass a seq from a prior
    /// write to gate a subsequent recall. Default ``recall()`` does not
    /// call this — the delta is always visible by virtue of being scanned
    /// during search; this primitive is only needed when the caller wants
    /// to wait through a compaction-in-progress window or a cluster
    /// follower-apply-lag window.
    pub fn wait_for_visible_seq(
        &self,
        namespace: &str,
        min_seq: u64,
        timeout: std::time::Duration,
    ) -> Result<()> {
        let deadline = std::time::Instant::now() + timeout;
        loop {
            let current = self.visible_seq_for(namespace);
            if current >= min_seq {
                return Ok(());
            }
            let now = std::time::Instant::now();
            if now >= deadline {
                return Err(crate::error::YantrikDbError::RyWaitTimeout {
                    namespace: namespace.to_string(),
                    requested_seq: min_seq,
                    observed_seq: current,
                    waited_ms: timeout.as_millis() as u64,
                });
            }
            let remaining = deadline - now;
            // The sentinel mutex is a no-data lock pair for the Condvar.
            // Critical race-avoidance pattern: re-check the watermark AFTER
            // acquiring the mutex but BEFORE waiting, because the writer
            // may have bumped + notified between our outer check and here.
            let mut guard = self.visible_seq_wait_mu.lock();
            let recheck = self.visible_seq_for(namespace);
            if recheck >= min_seq {
                return Ok(());
            }
            let result = self.visible_seq_cv.wait_for(&mut guard, remaining);
            drop(guard);
            if result.timed_out() {
                let final_current = self.visible_seq_for(namespace);
                if final_current >= min_seq {
                    return Ok(());
                }
                return Err(crate::error::YantrikDbError::RyWaitTimeout {
                    namespace: namespace.to_string(),
                    requested_seq: min_seq,
                    observed_seq: final_current,
                    waited_ms: timeout.as_millis() as u64,
                });
            }
            // Spurious wakeup or notify_all — re-check the watermark.
        }
    }
}

#[cfg(test)]
mod pending_ops_tests {
    use super::*;
    use crate::YantrikDB;

    fn open_test_db() -> YantrikDB {
        // Use :memory: so tests don't touch disk and migrations are fresh.
        YantrikDB::new(":memory:", 64).expect("open test db")
    }

    /// **#113 — the idle-CPU defect, pinned at the query plan.**
    ///
    /// A field user burned ~55% of a 32-core machine while IDLE because every
    /// 100ms materializer poll scanned the ENTIRE oplog: the drain query's
    /// `ORDER BY hlc, op_id` could not be served by a partial index on the
    /// filter column, so SQLite chose `idx_oplog_hlc`, walked all history
    /// filtering row-by-row, and added a temp B-tree — and with nothing
    /// pending, no LIMIT short-circuit ever fired. Cost grew with history
    /// depth, times 16 workers, times N engines.
    ///
    /// Asserting the PLAN, not a latency budget: the plan is deterministic
    /// and the regression is structural (a re-ordered ORDER BY, a dropped
    /// index, a "harmless" index rename all reintroduce it), whereas a timing
    /// assertion would be flaky on CI and would only fail once a corpus grew
    /// large enough to hurt — i.e. in a user's process, not here.
    ///
    /// It runs against [`YantrikDB::PENDING_OPS_QUERY`], the same const the
    /// drain executes. That is deliberate: pinning a copy of the SQL would
    /// pass forever while the real query drifted.
    #[test]
    fn pending_ops_query_uses_the_partial_index_and_never_scans_history() {
        let db = open_test_db();
        let conn = db.conn();

        // The index must exist on a FRESH database — the original
        // `idx_oplog_pending` was added in a migration and never in
        // SCHEMA_SQL, so every new install since v24 silently had no pending
        // index at all.
        let idx_exists: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' \
                 AND name = 'idx_oplog_pending_ordered'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(
            idx_exists, 1,
            "fresh databases must carry idx_oplog_pending_ordered (it must live in \
             SCHEMA_SQL, not only in a migration)"
        );

        let plan: Vec<String> = {
            let sql = format!("EXPLAIN QUERY PLAN {}", YantrikDB::PENDING_OPS_QUERY);
            let mut stmt = conn.prepare(&sql).unwrap();
            let rows = stmt
                .query_map(params![64_i64], |r| r.get::<_, String>(3))
                .unwrap();
            rows.collect::<std::result::Result<Vec<_>, _>>().unwrap()
        };
        let joined = plan.join(" | ");

        assert!(
            joined.contains("idx_oplog_pending_ordered"),
            "the drain must use the partial pending index; plan was: {joined}"
        );
        // The two signatures of the defect, asserted separately so a failure
        // says WHICH half regressed.
        assert!(
            !joined.contains("idx_oplog_hlc"),
            "the drain fell back to the full-history hlc index — this is the #113 \
             idle-scan defect; plan was: {joined}"
        );
        assert!(
            !joined.to_uppercase().contains("TEMP B-TREE"),
            "the drain is sorting in a temp B-tree, so the index no longer serves \
             the ORDER BY; plan was: {joined}"
        );
    }

    fn fake_embedding(seed: f32, dim: usize) -> Vec<u8> {
        let raw: Vec<f32> = (0..dim).map(|i| (seed + i as f32) * 0.1).collect();
        let norm: f32 = raw.iter().map(|x| x * x).sum::<f32>().sqrt().max(1e-9);
        let normalized: Vec<f32> = raw.iter().map(|x| x / norm).collect();
        crate::serde_helpers::serialize_f32(&normalized)
    }

    #[test]
    fn pending_op_round_trip() {
        let db = open_test_db();
        assert_eq!(
            db.count_pending_ops().unwrap(),
            0,
            "fresh db has no pending"
        );

        let payload = serde_json::json!({
            "rid": "test_rid_1",
            "type": "episodic",
            "text": "first pending op",
        });
        let emb_bytes = fake_embedding(1.0, 64);
        let op_id = db
            .log_op_pending(
                "record",
                Some("test_rid_1"),
                &payload,
                None,
                Some(&emb_bytes),
            )
            .expect("log_op_pending");

        assert_eq!(
            db.count_pending_ops().unwrap(),
            1,
            "one pending op after append"
        );

        db.mark_op_applied(&op_id).expect("mark applied");
        assert_eq!(db.count_pending_ops().unwrap(), 0, "no pending after mark");
    }

    /// NOT a typo: this asserts the OPPOSITE of idempotency, which is the real
    /// contract. It was named `pending_op_idempotent_on_double_append` — a lie
    /// that agreed with the rustdoc `log_op_pending` used to carry, and which
    /// justified the `INSERT OR IGNORE` this test's own body disproves.
    #[test]
    fn pending_op_double_append_writes_two_distinct_rows() {
        let db = open_test_db();
        let payload = serde_json::json!({"rid": "rid_idem"});
        let emb_bytes = fake_embedding(2.0, 64);

        let op_id_a = db
            .log_op_pending("record", Some("rid_idem"), &payload, None, Some(&emb_bytes))
            .unwrap();
        let op_id_b = db
            .log_op_pending("record", Some("rid_idem"), &payload, None, Some(&emb_bytes))
            .unwrap();

        // Two distinct op_ids generated (uuid7), but each is a separate row.
        assert_ne!(op_id_a, op_id_b, "each call generates a distinct op_id");
        assert_eq!(db.count_pending_ops().unwrap(), 2);
    }

    #[test]
    fn pending_op_persists_embedding_blob() {
        let db = open_test_db();
        let emb_bytes = fake_embedding(3.0, 64);
        let op_id = db
            .log_op_pending(
                "record",
                Some("rid_emb"),
                &serde_json::json!({}),
                None,
                Some(&emb_bytes),
            )
            .unwrap();

        let conn = db.read_conn();
        let stored: Option<Vec<u8>> = conn
            .query_row(
                "SELECT embedding FROM oplog WHERE op_id = ?1",
                params![op_id],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            stored.as_deref(),
            Some(emb_bytes.as_slice()),
            "embedding bytes round-trip exactly"
        );
    }

    #[test]
    fn mark_op_applied_idempotent() {
        let db = open_test_db();
        let op_id = db
            .log_op_pending("record", None, &serde_json::json!({}), None, None)
            .unwrap();
        db.mark_op_applied(&op_id).unwrap();
        // Calling again on already-applied op is a no-op.
        db.mark_op_applied(&op_id).unwrap();
        assert_eq!(db.count_pending_ops().unwrap(), 0);
    }

    #[test]
    fn count_pending_ignores_applied_ops() {
        let db = open_test_db();
        // Old log_op writes applied=1 directly.
        db.log_op("record", Some("rid_old"), &serde_json::json!({}), None)
            .unwrap();
        assert_eq!(
            db.count_pending_ops().unwrap(),
            0,
            "log_op (applied=1) is not pending"
        );

        // log_op_pending writes applied=0.
        db.log_op_pending(
            "record",
            Some("rid_new"),
            &serde_json::json!({}),
            None,
            None,
        )
        .unwrap();
        assert_eq!(db.count_pending_ops().unwrap(), 1);
    }

    #[test]
    fn backpressure_engages_at_max_pending() {
        // Saturate the queue with 10_000 pending ops, then verify the
        // 10_001st returns Error::Backpressure with sane fields.
        let db = open_test_db();
        for i in 0..10_000 {
            db.log_op_pending(
                "record",
                Some(&format!("rid_{i}")),
                &serde_json::json!({}),
                None,
                None,
            )
            .expect("first 10k succeed");
        }
        assert_eq!(db.count_pending_ops().unwrap(), 10_000);

        let err = db
            .log_op_pending(
                "record",
                Some("rid_overflow"),
                &serde_json::json!({}),
                None,
                None,
            )
            .expect_err("11k must fail with backpressure");
        match err {
            crate::error::YantrikDbError::Backpressure {
                pending,
                max,
                retry_after_ms,
            } => {
                assert_eq!(max, 10_000);
                assert_eq!(pending, 10_000);
                assert!(retry_after_ms > 0, "retry hint must be non-zero");
            }
            other => panic!("expected Backpressure, got {other:?}"),
        }

        // After draining one, the next push must succeed (proves backpressure
        // is reactive, not sticky). v0.7.1: drain via the public
        // `mark_op_applied` API so the cached `pending_op_count` atomic
        // stays coherent. Bypassing it with raw SQL (the v0.7.0 shape)
        // wouldn't decrement the counter and would falsely keep
        // backpressure engaged — the new test path verifies the
        // atomic counter contract end-to-end.
        let one_op_id: String = {
            let conn = db.read_conn();
            conn.query_row(
                "SELECT op_id FROM oplog WHERE applied = 0 LIMIT 1",
                [],
                |row| row.get(0),
            )
            .unwrap()
        };
        let was_unset = db.mark_op_applied(&one_op_id).unwrap();
        assert!(was_unset, "mark_op_applied should win the transition");
        db.log_op_pending(
            "record",
            Some("rid_after_drain"),
            &serde_json::json!({}),
            None,
            None,
        )
        .expect("succeeds after one drained");
    }

    #[test]
    fn apply_pending_drains_then_marks() {
        let db = open_test_db();
        // Seed 3 pending ops of various types.
        for (op_type, target) in [
            ("record", "rid_1"),
            ("forget", "rid_2"),
            ("relate", "rid_3"),
        ] {
            db.log_op_pending(op_type, Some(target), &serde_json::json!({}), None, None)
                .unwrap();
        }
        assert_eq!(db.count_pending_ops().unwrap(), 3);

        let applied = db.apply_pending_ops_once(10).unwrap();
        assert_eq!(applied, 3, "all 3 pending ops drained in one pass");
        assert_eq!(db.count_pending_ops().unwrap(), 0);
    }

    #[test]
    fn apply_pending_respects_limit() {
        let db = open_test_db();
        for i in 0..5 {
            db.log_op_pending(
                "record",
                Some(&format!("rid_{i}")),
                &serde_json::json!({}),
                None,
                None,
            )
            .unwrap();
        }
        let applied = db.apply_pending_ops_once(2).unwrap();
        assert_eq!(applied, 2, "only 2 of 5 drained when limit=2");
        assert_eq!(db.count_pending_ops().unwrap(), 3);

        // Subsequent drain picks up the rest.
        let applied2 = db.apply_pending_ops_once(10).unwrap();
        assert_eq!(applied2, 3);
        assert_eq!(db.count_pending_ops().unwrap(), 0);
    }

    #[test]
    fn apply_pending_idempotent_when_empty() {
        let db = open_test_db();
        // No pending ops — drain returns 0 cleanly.
        assert_eq!(db.apply_pending_ops_once(100).unwrap(), 0);
        assert_eq!(db.apply_pending_ops_once(100).unwrap(), 0);
    }

    #[test]
    fn apply_pending_skips_unknown_op_type() {
        let db = open_test_db();
        // v0.7.1: enqueue via log_op_pending with a synthetic op_type so
        // the cached `pending_op_count` atomic increments. v0.7.0 used
        // direct SQL INSERT here, but that bypasses the counter and
        // mismatches the public-API count_pending_ops contract — fixed
        // alongside the perf hotfix.
        db.log_op_pending(
            "made_up_op",
            Some("synth_unknown"),
            &serde_json::json!({}),
            None,
            None,
        )
        .unwrap();
        assert_eq!(db.count_pending_ops().unwrap(), 1);

        // Drain doesn't apply unknown op types — they stay pending so a
        // future runtime that knows the op type can drain them.
        let applied = db.apply_pending_ops_once(10).unwrap();
        assert_eq!(applied, 0);
        assert_eq!(
            db.count_pending_ops().unwrap(),
            1,
            "unknown op_type stays pending"
        );
    }

    #[test]
    fn schema_v25_columns_present() {
        // Open a fresh DB so the canonical SCHEMA_SQL runs and creates
        // memories with the v25 columns. Then verify column metadata.
        let db = open_test_db();
        let conn = db.read_conn();
        let mut stmt = conn.prepare("PRAGMA table_info(memories)").unwrap();
        let cols: Vec<String> = stmt
            .query_map([], |row| row.get::<_, String>(1))
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert!(
            cols.contains(&"tombstone_reason".to_string()),
            "tombstone_reason missing — schema v25 not applied"
        );
        assert!(
            cols.contains(&"created_at_unix_micros".to_string()),
            "created_at_unix_micros missing — schema v25 not applied"
        );
        assert!(
            cols.contains(&"embedding_model".to_string()),
            "embedding_model missing — schema v25 not applied"
        );
    }

    #[test]
    fn schema_version_meta_at_current() {
        // Locks the meta-stamp to the SCHEMA_VERSION constant so a future
        // bump (e.g. v26 → v27 from RFC 026's next phase) automatically
        // moves this test forward without a manual literal edit.
        // Previously hard-coded "25"; v26 (issue #29) made the brittleness
        // obvious so the renaming + constant reference go together.
        let db = open_test_db();
        let conn = db.read_conn();
        let v: String = conn
            .query_row(
                "SELECT value FROM meta WHERE key = 'schema_version'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(v, crate::schema::SCHEMA_VERSION.to_string());
    }

    // ── Perf regression tests (added v0.7.1 after the SELECT COUNT incident) ──
    //
    // The v0.7.0 → v0.7.1 hotfix story (yantrikdb-server msg b951a2de):
    // log_op_pending was running `SELECT COUNT(*) FROM oplog WHERE applied=0`
    // on every foreground call. At 8 writers that's 16 conn acquisitions/sec
    // just for the backpressure check. v0.7.1 replaced it with an
    // AtomicI64 cached counter. These tests structurally pin both
    // properties (atomic stays coherent with SQL truth + backpressure check
    // is O(1) under load) so a future regression of the same class is
    // caught BEFORE it ships, not by yantrikdb-server's homelab bench.

    /// Atomic counter must never drift from SQL truth across mixed
    /// log_op_pending / mark_op_applied / apply_pending_ops_once
    /// operations. If this drifts, the backpressure check is wrong AND
    /// `count_pending_ops()` lies to the materializer. Drift = silent
    /// data loss possibility.
    #[test]
    fn pending_op_count_atomic_matches_sql_after_workload() {
        let db = open_test_db();

        // Phase 1: pure inserts.
        for i in 0..50 {
            db.log_op_pending(
                "record",
                Some(&format!("rid_w1_{i}")),
                &serde_json::json!({}),
                None,
                None,
            )
            .unwrap();
            assert_eq!(
                db.count_pending_ops().unwrap(),
                db.count_pending_ops_sql().unwrap(),
                "drift after insert #{i}"
            );
        }

        // Phase 2: mixed inserts + applies via apply_pending_ops_once
        // (which exercises the full mark_op_applied → atomic.fetch_sub path).
        let drained = db.apply_pending_ops_once(20).unwrap();
        assert!(drained > 0, "drained at least one");
        assert_eq!(
            db.count_pending_ops().unwrap(),
            db.count_pending_ops_sql().unwrap(),
            "drift after apply_pending_ops_once"
        );

        // Phase 3: more inserts after partial drain.
        for i in 0..30 {
            db.log_op_pending(
                "record",
                Some(&format!("rid_w3_{i}")),
                &serde_json::json!({}),
                None,
                None,
            )
            .unwrap();
        }
        assert_eq!(
            db.count_pending_ops().unwrap(),
            db.count_pending_ops_sql().unwrap(),
            "drift after second-wave inserts"
        );

        // Phase 4: drain all the rest.
        loop {
            let n = db.apply_pending_ops_once(100).unwrap();
            if n == 0 {
                break;
            }
        }
        assert_eq!(db.count_pending_ops().unwrap(), 0);
        assert_eq!(db.count_pending_ops_sql().unwrap(), 0);

        // Phase 5: idempotent re-mark must not double-decrement.
        let extra_op_id = db
            .log_op_pending(
                "record",
                Some("rid_extra"),
                &serde_json::json!({}),
                None,
                None,
            )
            .unwrap();
        let first_mark = db.mark_op_applied(&extra_op_id).unwrap();
        let second_mark = db.mark_op_applied(&extra_op_id).unwrap();
        assert!(first_mark, "first mark wins the transition");
        assert!(!second_mark, "second mark is a no-op (idempotent)");
        assert_eq!(
            db.count_pending_ops().unwrap(),
            0,
            "double-mark must NOT push counter negative"
        );
    }

    /// **Regression guard for the v0.7.0 → v0.7.1 hotfix.** Before the
    /// fix, `log_op_pending` did a SELECT COUNT(*) WHERE applied=0 per
    /// call. That made the foreground hot path scale O(pending_count)
    /// despite the partial index. The fix made it a single atomic load.
    ///
    /// This test pins the property: insert N pending ops, then time the
    /// next log_op_pending call. Even with 5000 pending ops in the
    /// oplog, the call should complete in <10ms (the actual production
    /// number is ~1µs; the threshold is ~10000× looser to absorb CI
    /// noise without false positives). If a regression re-introduces
    /// SELECT COUNT, the call time grows with the count and the
    /// assertion fires.
    #[test]
    fn log_op_pending_is_o1_under_pending_load() {
        use std::time::Instant;

        let db = open_test_db();
        // Seed 5000 pending ops. Each insert is a real conn acquire +
        // INSERT — this is the setup cost, not the test of interest.
        for i in 0..5000 {
            db.log_op_pending(
                "record",
                Some(&format!("rid_seed_{i}")),
                &serde_json::json!({}),
                None,
                None,
            )
            .expect("seed insert");
        }
        assert_eq!(db.count_pending_ops().unwrap(), 5000);

        // Time the next call. Pre-v0.7.1 this scanned 5000 oplog rows
        // via the partial index AND did a separate read_conn acquire.
        // Post-fix it's an atomic load. We give 10ms headroom for CI
        // noise; the actual cost should be sub-millisecond.
        let t0 = Instant::now();
        let _id = db
            .log_op_pending(
                "record",
                Some("rid_under_load"),
                &serde_json::json!({}),
                None,
                None,
            )
            .expect("call under load");
        let elapsed = t0.elapsed();
        assert!(
            elapsed.as_millis() < 10,
            "log_op_pending should be O(1); 5001st call took {elapsed:?} \
             (regression: probably SELECT COUNT re-introduced)"
        );
    }

    /// Atomic counter must remain non-negative and stable when many
    /// concurrent workers race on mark_op_applied. The applied=0 filter
    /// in the SQL guarantees exactly-once SQL transition; the
    /// `if won { fetch_sub }` guard inside mark_op_applied must
    /// preserve that into the atomic.
    #[test]
    fn pending_op_count_under_concurrent_mark() {
        use std::sync::Arc;
        use std::thread;

        let db = Arc::new(open_test_db());
        for i in 0..100 {
            db.log_op_pending(
                "record",
                Some(&format!("rid_conc_{i}")),
                &serde_json::json!({}),
                None,
                None,
            )
            .unwrap();
        }
        assert_eq!(db.count_pending_ops().unwrap(), 100);

        // Snapshot all op_ids so each worker can race to mark them.
        let op_ids: Vec<String> = {
            let conn = db.read_conn();
            let mut stmt = conn
                .prepare("SELECT op_id FROM oplog WHERE applied = 0")
                .unwrap();
            stmt.query_map([], |row| row.get::<_, String>(0))
                .unwrap()
                .filter_map(|r| r.ok())
                .collect()
        };

        // 4 workers each try to mark every op. The applied=0 filter +
        // bool return + `if won` guard means each op is decremented
        // exactly once across all workers.
        let mut handles = Vec::new();
        for _ in 0..4 {
            let db_c = Arc::clone(&db);
            let ids_c = op_ids.clone();
            handles.push(thread::spawn(move || {
                for id in ids_c {
                    let _ = db_c.mark_op_applied(&id);
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(
            db.count_pending_ops().unwrap(),
            0,
            "concurrent racing on mark_op_applied must converge atomic to 0"
        );
        assert_eq!(
            db.count_pending_ops_sql().unwrap(),
            0,
            "and SQL truth must agree"
        );
    }

    #[test]
    fn schema_v25_indexes_present() {
        let db = open_test_db();
        let conn = db.read_conn();
        let mut stmt = conn
            .prepare("SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='memories'")
            .unwrap();
        let names: Vec<String> = stmt
            .query_map([], |row| row.get::<_, String>(0))
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert!(
            names.iter().any(|n| n == "idx_memories_created_at_micros"),
            "idx_memories_created_at_micros missing"
        );
        assert!(
            names.iter().any(|n| n == "idx_memories_embedding_model"),
            "idx_memories_embedding_model missing"
        );
    }

    // ── Phase 4.3 (saga task 3): materialize_record_post dispatch arm ──
    //
    // These tests exercise the worker-side materialization path WITHOUT
    // changing foreground behavior. They enqueue an op directly via
    // log_op_pending and drain via apply_pending_ops_once, asserting the
    // resulting SQL/graph state matches what the foreground inline loop
    // would have produced. Commit B will flip foreground to enqueue;
    // these tests act as the contract pin so the flip is provably safe.

    fn enqueue_post_record(db: &YantrikDB, rid: &str, text: &str, namespace: &str) -> String {
        // First INSERT a stub memories row so memory_entities + claims
        // FK references are valid. Foreground (Commit B) will INSERT the
        // memories row before enqueuing; tests mirror that ordering.
        let conn = db.conn();
        let stored_text = db.encrypt_text(text).unwrap();
        let ts = super::super::now();
        conn.execute(
            "INSERT INTO memories \
             (rid, type, text, embedding, created_at, updated_at, importance, \
              half_life, last_access, valence, metadata, namespace, \
              certainty, domain, source, emotional_state) \
             VALUES (?1, 'episodic', ?2, NULL, ?3, ?3, 0.5, 604800.0, ?3, 0.0, '{}', ?4, 0.8, 'general', 'user', NULL)",
            params![rid, stored_text, ts, namespace],
        ).unwrap();
        drop(conn);

        let payload = serde_json::json!({
            "rid": rid,
            "text": stored_text,
            "namespace": namespace,
            "ts_secs": ts,
            "domain": "general",
            "source": "user",
        });
        db.log_op_pending(
            crate::engine::op_types::OP_MATERIALIZE_RECORD_POST,
            Some(rid),
            &payload,
            None,
            None,
        )
        .expect("log_op_pending")
    }

    #[test]
    fn materialize_record_post_inserts_entities() {
        let db = open_test_db();
        let _op_id = enqueue_post_record(&db, "r1", "Alice met Acme yesterday", "default");
        assert_eq!(db.count_pending_ops().unwrap(), 1);

        let n = db.apply_pending_ops_once(10).unwrap();
        assert_eq!(n, 1, "one op drained");
        assert_eq!(db.count_pending_ops().unwrap(), 0);

        // Entities table should now contain Alice and Acme.
        let conn = db.read_conn();
        let alice: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM entities WHERE name = 'Alice'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        let acme: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM entities WHERE name = 'Acme'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(alice, 1, "Alice seeded by heuristic");
        assert_eq!(acme, 1, "Acme seeded by heuristic");
    }

    #[test]
    fn materialize_record_post_inserts_memory_entities() {
        let db = open_test_db();
        let _op_id = enqueue_post_record(&db, "r2", "Bob works at Beta Corp", "default");
        let _ = db.apply_pending_ops_once(10).unwrap();

        let conn = db.read_conn();
        let count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM memory_entities WHERE memory_rid = 'r2'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(
            count >= 2,
            "memory_entities has at least 2 rows for Bob+Beta Corp; got {count}"
        );
    }

    #[test]
    fn materialize_record_post_idempotent_on_double_drain() {
        // Drain twice — second drain must be a no-op (op already marked
        // applied by first drain). entities mention_count must stay at 1.
        let db = open_test_db();
        let _op_id = enqueue_post_record(&db, "r3", "Charlie went to Delta", "default");
        let n1 = db.apply_pending_ops_once(10).unwrap();
        let n2 = db.apply_pending_ops_once(10).unwrap();
        assert_eq!(n1, 1, "first drain applies");
        assert_eq!(n2, 0, "second drain finds nothing pending");

        let conn = db.read_conn();
        let mc: i64 = conn
            .query_row(
                "SELECT mention_count FROM entities WHERE name = 'Charlie'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(mc, 1, "mention_count not double-bumped");
    }

    #[test]
    fn materialize_record_post_updates_graph_index() {
        // graph_index in-memory state must reflect the worker's apply
        // — this is what makes recall-by-entity find the new memory.
        let db = open_test_db();
        let _op_id = enqueue_post_record(&db, "r4", "Eve climbed Everest", "default");
        let _ = db.apply_pending_ops_once(10).unwrap();

        let gi = db.graph_index.read();
        let names = gi.all_entity_names();
        assert!(names.iter().any(|n| n == "Eve"), "Eve in graph_index");
        assert!(
            names.iter().any(|n| n == "Everest"),
            "Everest in graph_index"
        );
    }

    #[test]
    fn materialize_record_post_concurrent_workers_no_double_apply() {
        // 4 worker threads + 20 ops → exactly-once semantics on the
        // applied=0 filter. Same race-safety as the existing materializer
        // tests, but exercising the new dispatch path.
        use std::sync::Arc;
        use std::thread;

        let db = Arc::new(open_test_db());
        for i in 0..20 {
            let _ = enqueue_post_record(
                &db,
                &format!("rcc_{i}"),
                &format!("Person{i} met Place{i}"),
                "default",
            );
        }
        assert_eq!(db.count_pending_ops().unwrap(), 20);

        let mut handles = Vec::new();
        for _ in 0..4 {
            let db_c = Arc::clone(&db);
            handles.push(thread::spawn(move || {
                let mut total = 0;
                while db_c.count_pending_ops().unwrap() > 0 {
                    total += db_c.apply_pending_ops_once(50).unwrap();
                    if total >= 20 {
                        break;
                    }
                }
                total
            }));
        }
        let totals: Vec<usize> = handles.into_iter().map(|h| h.join().unwrap()).collect();
        assert_eq!(
            totals.iter().sum::<usize>(),
            20,
            "exactly 20 applies across all workers, no double-counting; got {totals:?}"
        );
        assert_eq!(db.count_pending_ops().unwrap(), 0);

        // entities should have 20 distinct Person* rows, 20 distinct Place* rows.
        let conn = db.read_conn();
        let person_count: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM entities WHERE name LIKE 'Person%'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(person_count, 20);
    }

    #[test]
    fn materialize_record_post_invalid_payload_leaves_op_pending() {
        // Malformed payload → worker logs warning, leaves op pending for
        // retry. Must not advance applied flag (otherwise we'd silently
        // lose data on a transient parse failure).
        let db = open_test_db();
        let bad_payload = serde_json::json!({"not_a_rid": "oops"});
        let _ = db
            .log_op_pending(
                crate::engine::op_types::OP_MATERIALIZE_RECORD_POST,
                Some("r_bad"),
                &bad_payload,
                None,
                None,
            )
            .unwrap();
        let n = db.apply_pending_ops_once(10).unwrap();
        assert_eq!(n, 0, "malformed op not applied");
        assert_eq!(
            db.count_pending_ops().unwrap(),
            1,
            "still pending for retry"
        );
    }
}