khive_pack_brain/

persist.rs

//! Brain state persistence — snapshot upsert and namespace-scoped reload.

use std::collections::HashMap;
use std::sync::Mutex;

// Test-only interleaving hook for ensure_loaded.
//
// When set, every cold ensure_loaded call (after the async DB load, before the
// final tracker lock) will:
//   1. Send () on `reached_tx` to signal the test controller "I am at the hook".
//   2. Await `proceed_rx` before continuing.
//
// This lets a test precisely inject the following interleaving:
//   - Loader B reaches hook  →  test controller receives reached signal
//   - Test controller runs Loader A to completion and mutates state
//   - Test controller sends on proceed_tx  →  Loader B continues
//
// With the old code (no re-check), Loader B then clobbered A's mutation.
// With the fix (re-check under final guard), Loader B sees is_active=true
// and returns early, preserving A's mutation.
#[cfg(test)]
pub(crate) struct LoadHook {
    pub reached_tx: tokio::sync::oneshot::Sender<()>,
    pub proceed_rx: tokio::sync::oneshot::Receiver<()>,
}

#[cfg(test)]
pub(crate) static POST_LOAD_HOOK: std::sync::Mutex<Option<LoadHook>> = std::sync::Mutex::new(None);

#[cfg(test)]
pub(crate) fn set_post_load_hook(hook: LoadHook) {
    *POST_LOAD_HOOK.lock().unwrap() = Some(hook);
}

#[cfg(test)]
pub(crate) fn clear_post_load_hook() {
    *POST_LOAD_HOOK.lock().unwrap() = None;
}

use serde_json::Value;

use khive_runtime::{KhiveRuntime, NamespaceToken, RuntimeError};
use khive_storage::types::{SqlStatement, SqlValue};
use khive_storage::{SqlAccess, SqlWriter};

use khive_brain_core::{
    validate_brain_state_snapshot_with_capacity, BrainSignal, BrainState, BrainStateSnapshot,
};

use crate::event::interpret;

const SNAPSHOT_PROFILE_ID: &str = "__brain__";
const DEFAULT_SNAPSHOT_BATCH_SIZE: u64 = 5;

/// Tracks loaded namespaces and dirty event counts; owns the per-namespace state map.
pub struct PersistenceTracker {
    /// Namespace currently reflected in the shared `BrainState` slot, if any.
    pub(crate) active_namespace: Option<String>,
    /// Saved snapshots of in-memory state for namespaces that have been initialised
    /// but are not currently in the active slot.  Used for save-restore on switch.
    saved_states: HashMap<String, BrainState>,
    /// Namespaces for which state has been initialised (from DB or fresh default).
    pub(crate) loaded_namespaces: HashMap<String, ()>,
    dirty_counts: HashMap<String, u64>,
    snapshot_batch_size: u64,
    /// Pre-load accumulator for hook signals that arrive before `ensure_loaded` runs.
    ///
    /// When `on_dispatch` fires for a namespace that has never been loaded from the
    /// DB, the signal is queued here rather than applied to a speculative
    /// `BrainState`.  `ensure_loaded` drains this queue *after* the snapshot +
    /// event-replay path completes, so the ordering guarantee is:
    ///   persisted snapshot → replayed events → queued hook signals
    ///
    /// The namespace is deliberately NOT added to `loaded_namespaces` while
    /// signals are pending; that prevents `ensure_loaded` from skipping the DB
    /// round-trip for a namespace that may have existing persisted history.
    pending_hook_signals: HashMap<String, Vec<BrainSignal>>,
}

impl Default for PersistenceTracker {
    fn default() -> Self {
        Self::new()
    }
}

impl PersistenceTracker {
    /// Create a fresh `PersistenceTracker` with no loaded namespaces.
    pub fn new() -> Self {
        Self {
            active_namespace: None,
            saved_states: HashMap::new(),
            loaded_namespaces: HashMap::new(),
            dirty_counts: HashMap::new(),
            snapshot_batch_size: DEFAULT_SNAPSHOT_BATCH_SIZE,
            pending_hook_signals: HashMap::new(),
        }
    }

    /// Return `true` if `namespace` has been initialised (from DB or fresh default).
    pub fn is_loaded(&self, namespace: &str) -> bool {
        self.loaded_namespaces.contains_key(namespace)
    }

    /// Return `true` if `namespace` is the currently active namespace slot.
    pub fn is_active(&self, namespace: &str) -> bool {
        self.active_namespace.as_deref() == Some(namespace)
    }

    /// Register `namespace` as loaded and set it as the active namespace.
    pub fn mark_loaded(&mut self, namespace: String) {
        self.loaded_namespaces.insert(namespace.clone(), ());
        self.active_namespace = Some(namespace);
    }

    /// Save `from_state` for `from_namespace`, then activate `to_namespace` (returning saved state if any).
    pub fn swap_namespace(
        &mut self,
        from_namespace: &str,
        from_state: BrainState,
        to_namespace: String,
    ) -> Option<BrainState> {
        self.saved_states
            .insert(from_namespace.to_string(), from_state);
        let saved = self.saved_states.remove(&to_namespace);
        self.active_namespace = Some(to_namespace);
        saved
    }

    /// Increment the dirty event count for `namespace`; return `true` when a snapshot is due.
    pub fn increment_dirty(&mut self, namespace: &str) -> bool {
        let count = self.dirty_counts.entry(namespace.to_string()).or_insert(0);
        *count += 1;
        *count >= self.snapshot_batch_size
    }

    pub fn reset_dirty(&mut self, namespace: &str) {
        self.dirty_counts.insert(namespace.to_string(), 0);
    }

    /// Return the total_events counter for any namespace stored in `saved_states`
    /// (saved-off namespaces) or `pending_hook_signals` (cold pre-load queue).
    /// Returns `None` when no state has been initialised for the given namespace.
    ///
    /// Note: does NOT return the counter for the active namespace, which lives in
    /// the shared `BrainState` slot, not in `saved_states`.
    #[cfg(test)]
    pub(crate) fn total_events_for(&self, namespace: &str) -> Option<u64> {
        if let Some(s) = self.saved_states.get(namespace) {
            return Some(s.balanced_recall.total_events);
        }
        self.pending_hook_signals
            .get(namespace)
            .map(|signals| signals.len() as u64)
    }

    /// Apply a signal to the state bucket that owns `namespace`.
    ///
    /// - Active namespace: returns `ApplyTarget::ActiveSlot` — caller must apply
    ///   the signal to the shared `BrainState` lock while holding the dispatch gate.
    /// - Saved (loaded) namespace: applies directly to `saved_states` and returns
    ///   `ApplyTarget::Done`.
    /// - Cold/unknown namespace: enqueues the signal in `pending_hook_signals` and
    ///   returns `ApplyTarget::Done`.  The namespace is NOT marked loaded; the DB
    ///   round-trip is preserved for the first `ensure_loaded` call.  The queue is
    ///   drained by `ensure_loaded` *after* snapshot restore and event replay, so
    ///   the ordering guarantee is: snapshot → replayed events → queued signals.
    ///
    /// No event is silently dropped regardless of which slot is currently active.
    pub(crate) fn route_signal(
        &mut self,
        namespace: &str,
        signal: &khive_brain_core::BrainSignal,
        entity_capacity: usize,
    ) -> ApplyTarget {
        if self.is_active(namespace) {
            return ApplyTarget::ActiveSlot;
        }

        if self.is_loaded(namespace) {
            // Namespace has been loaded from the DB but is currently saved off.
            // Apply directly to its saved BrainState.
            if let Some(state) = self.saved_states.get_mut(namespace) {
                state.balanced_recall.apply_signal(signal);
                crate::sync_balanced_recall_record(state);
            }
            return ApplyTarget::Done;
        }

        // Cold/unknown namespace: queue the signal for deferred application.
        // Do NOT mark the namespace loaded — ensure_loaded must still perform
        // the DB snapshot + event-replay before draining this queue.
        let _ = entity_capacity;
        self.pending_hook_signals
            .entry(namespace.to_string())
            .or_default()
            .push(signal.clone());
        ApplyTarget::Done
    }

    /// Drain and return any pending hook signals for `namespace`.
    ///
    /// Called by `ensure_loaded` after the normal load path (snapshot + replay)
    /// completes, so queued signals land on top of persisted history.
    pub(crate) fn drain_pending_signals(&mut self, namespace: &str) -> Vec<BrainSignal> {
        self.pending_hook_signals
            .remove(namespace)
            .unwrap_or_default()
    }
}

/// Indicates where a dispatched signal should be applied by the caller.
pub(crate) enum ApplyTarget {
    /// The namespace is active — the caller must apply the signal to the shared
    /// `BrainState` slot (which the caller already holds behind the dispatch gate).
    ActiveSlot,
    /// Signal was applied inside `PersistenceTracker`; caller has nothing left to do.
    Done,
}

fn sql_err(context: &str, e: impl std::fmt::Display) -> RuntimeError {
    RuntimeError::Internal(format!("brain persistence {context}: {e}"))
}

/// Append one `brain_event_log` row using an already-acquired writer (plain
/// writer or an open transaction — both implement `SqlWriter`).
async fn append_brain_event_on_writer(
    writer: &mut dyn SqlWriter,
    namespace: &str,
    profile_id: &str,
    event_kind: &str,
    payload: &Value,
    created_at_us: i64,
) -> Result<(), RuntimeError> {
    let payload_str = serde_json::to_string(payload).map_err(|e| sql_err("serialize event", e))?;

    writer
        .execute(SqlStatement {
            sql: "INSERT INTO brain_event_log (profile_id, namespace, event_kind, payload, created_at) VALUES (?1, ?2, ?3, ?4, ?5)".into(),
            params: vec![
                SqlValue::Text(profile_id.to_string()),
                SqlValue::Text(namespace.to_string()),
                SqlValue::Text(event_kind.to_string()),
                SqlValue::Text(payload_str),
                SqlValue::Integer(created_at_us),
            ],
            label: Some("brain_event_log_append".into()),
        })
        .await
        .map_err(|e| sql_err("append event", e))?;

    Ok(())
}

/// Upsert the namespace's `brain_profile_snapshots` row using an
/// already-acquired writer (plain writer or an open transaction).
async fn upsert_snapshot_on_writer(
    writer: &mut dyn SqlWriter,
    namespace: &str,
    snapshot: &BrainStateSnapshot,
    updated_at_us: i64,
) -> Result<(), RuntimeError> {
    let snapshot_json =
        serde_json::to_string(snapshot).map_err(|e| sql_err("serialize snapshot", e))?;

    writer
        .execute(SqlStatement {
            sql: "INSERT INTO brain_profile_snapshots (profile_id, namespace, snapshot_json, updated_at) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(profile_id, namespace) DO UPDATE SET snapshot_json = excluded.snapshot_json, updated_at = excluded.updated_at".into(),
            params: vec![
                SqlValue::Text(SNAPSHOT_PROFILE_ID.to_string()),
                SqlValue::Text(namespace.to_string()),
                SqlValue::Text(snapshot_json),
                SqlValue::Integer(updated_at_us),
            ],
            label: Some("brain_snapshot_upsert".into()),
        })
        .await
        .map_err(|e| sql_err("upsert snapshot", e))?;

    Ok(())
}

pub async fn append_brain_event(
    sql: &dyn SqlAccess,
    namespace: &str,
    profile_id: &str,
    event_kind: &str,
    payload: &Value,
    created_at_us: i64,
) -> Result<(), RuntimeError> {
    let mut writer = sql.writer().await.map_err(|e| sql_err("writer", e))?;
    append_brain_event_on_writer(
        writer.as_mut(),
        namespace,
        profile_id,
        event_kind,
        payload,
        created_at_us,
    )
    .await
}

pub async fn upsert_snapshot(
    sql: &dyn SqlAccess,
    namespace: &str,
    snapshot: &BrainStateSnapshot,
    updated_at_us: i64,
) -> Result<(), RuntimeError> {
    let mut writer = sql.writer().await.map_err(|e| sql_err("writer", e))?;
    upsert_snapshot_on_writer(writer.as_mut(), namespace, snapshot, updated_at_us).await
}

/// Descriptor for a single durable `BrainState` mutation (issues #457/#458):
/// which brain-event-log row to append alongside the snapshot upsert.
pub struct BrainMutationEvent {
    pub profile_id: String,
    pub event_kind: String,
    pub payload: Value,
}

/// Apply a `BrainState` mutation with durability as part of the success
/// contract (issues #457/#458).
///
/// `mutate` runs against a *proposed* copy of the state (never the live
/// `state` mutex) built via a snapshot round-trip. Its result and the
/// resulting snapshot are then persisted — brain event-log append AND
/// snapshot upsert — as ONE atomic unit via `SqlAccess::atomic_unit`. Only
/// after that unit commits does the proposed state replace the live state
/// and the tracker get marked clean. If `mutate` fails, or the atomic unit
/// fails to append, upsert, or commit, this returns `Err` and the live state
/// is left completely untouched — there is no in-memory mutation to roll
/// back because it was never applied to the shared state in the first
/// place. See `crates/khive-pack-brain/docs/api/persist.md` for why this
/// takes `&dyn SqlAccess` and why it uses `atomic_unit` over a manual
/// transaction.
pub async fn persist_brain_state_mutation<R>(
    sql: &dyn SqlAccess,
    token: &NamespaceToken,
    tracker: &Mutex<PersistenceTracker>,
    state: &Mutex<BrainState>,
    event: BrainMutationEvent,
    entity_capacity: usize,
    mutate: impl FnOnce(&mut BrainState) -> Result<R, RuntimeError>,
) -> Result<R, RuntimeError> {
    let namespace = token.namespace().as_str().to_string();
    let now_us = chrono::Utc::now().timestamp_micros();

    let mut proposed = {
        let current = state.lock().unwrap();
        BrainState::from_snapshot(current.to_snapshot(), entity_capacity)
    };

    let result = mutate(&mut proposed)?;
    let snapshot = proposed.to_snapshot();

    let namespace_for_op = namespace.clone();
    let profile_id = event.profile_id;
    let event_kind = event.event_kind;
    let payload = event.payload;
    let op: khive_storage::AtomicUnitOp = Box::new(move |writer| {
        Box::pin(async move {
            append_brain_event_on_writer(
                writer,
                &namespace_for_op,
                &profile_id,
                &event_kind,
                &payload,
                now_us,
            )
            .await
            .map_err(|e| {
                khive_storage::StorageError::driver(
                    khive_storage::StorageCapability::Sql,
                    "brain_persist_append_event",
                    e,
                )
            })?;
            upsert_snapshot_on_writer(writer, &namespace_for_op, &snapshot, now_us)
                .await
                .map_err(|e| {
                    khive_storage::StorageError::driver(
                        khive_storage::StorageCapability::Sql,
                        "brain_persist_upsert_snapshot",
                        e,
                    )
                })?;
            Ok(Box::new(()) as Box<dyn std::any::Any + Send>)
        })
    });

    sql.atomic_unit(op).await?;

    {
        let mut live = state.lock().unwrap();
        *live = proposed;
    }
    {
        let mut t = tracker.lock().unwrap();
        t.reset_dirty(&namespace);
        t.mark_loaded(namespace);
    }

    Ok(result)
}

pub async fn load_latest_snapshot(
    sql: &dyn SqlAccess,
    namespace: &str,
    entity_capacity: usize,
) -> Result<Option<(BrainStateSnapshot, i64)>, RuntimeError> {
    let mut reader = sql.reader().await.map_err(|e| sql_err("reader", e))?;
    let row = reader
        .query_row(SqlStatement {
            sql: "SELECT snapshot_json, updated_at FROM brain_profile_snapshots WHERE profile_id = ?1 AND namespace = ?2 ORDER BY updated_at DESC LIMIT 1".into(),
            params: vec![
                SqlValue::Text(SNAPSHOT_PROFILE_ID.to_string()),
                SqlValue::Text(namespace.to_string()),
            ],
            label: Some("brain_snapshot_load".into()),
        })
        .await
        .map_err(|e| sql_err("load snapshot", e))?;

    match row {
        None => Ok(None),
        Some(row) => {
            let json_str = match row.get("snapshot_json") {
                Some(SqlValue::Text(s)) => s.clone(),
                _ => return Err(sql_err("load snapshot", "missing snapshot_json column")),
            };
            let updated_at = match row.get("updated_at") {
                Some(SqlValue::Integer(n)) => *n,
                _ => return Err(sql_err("load snapshot", "missing updated_at column")),
            };
            let snapshot: BrainStateSnapshot =
                serde_json::from_str(&json_str).map_err(|e| sql_err("deserialize snapshot", e))?;
            validate_brain_state_snapshot_with_capacity(&snapshot, entity_capacity)
                .map_err(|e| sql_err("snapshot invariant violation", e))?;
            Ok(Some((snapshot, updated_at)))
        }
    }
}

/// A single row that was quarantined during replay, with enough metadata to
/// diagnose and re-examine the bad entry without re-running a replay.
pub struct QuarantinedRow {
    /// Row primary key from `brain_event_log`.
    pub id: i64,
    /// Profile id recorded at write time (may be empty string if the column was null).
    pub profile_id: String,
    /// ISO-8601 / epoch-µs created_at value as recorded in the table.
    pub created_at: i64,
    /// Human-readable description of why the row was quarantined.
    pub reason: String,
    /// Leading ~200 chars of the raw payload for quick inspection (truncated with "…").
    pub payload_snippet: String,
}

/// Result of a replay load: valid events and the full quarantine manifest.
pub struct LoadEventsResult {
    pub events: Vec<khive_storage::event::Event>,
    /// Rows that were skipped due to structural or semantic validation failure.
    /// The physical rows remain in `brain_event_log`; this vec makes them queryable.
    pub quarantined: Vec<QuarantinedRow>,
}

impl LoadEventsResult {
    /// Convenience accessor: number of quarantined rows.
    pub fn quarantine_count(&self) -> usize {
        self.quarantined.len()
    }
}

pub async fn load_events_since(
    sql: &dyn SqlAccess,
    namespace: &str,
    since_us: i64,
) -> Result<LoadEventsResult, RuntimeError> {
    let mut reader = sql.reader().await.map_err(|e| sql_err("reader", e))?;
    let rows = reader
        .query_all(SqlStatement {
            sql: "SELECT id, profile_id, event_kind, payload, created_at \
                  FROM brain_event_log \
                  WHERE namespace = ?1 AND created_at > ?2 \
                  ORDER BY created_at ASC, id ASC"
                .into(),
            params: vec![
                SqlValue::Text(namespace.to_string()),
                SqlValue::Integer(since_us),
            ],
            label: Some("brain_events_replay".into()),
        })
        .await
        .map_err(|e| sql_err("load events", e))?;

    let mut events = Vec::with_capacity(rows.len());
    let mut quarantined: Vec<QuarantinedRow> = Vec::new();

    for row in &rows {
        let row_id = match row.get("id") {
            Some(SqlValue::Integer(i)) => *i,
            _ => 0,
        };
        let profile_id = match row.get("profile_id") {
            Some(SqlValue::Text(s)) => s.clone(),
            _ => String::new(),
        };
        let created_at = match row.get("created_at") {
            Some(SqlValue::Integer(i)) => *i,
            _ => 0,
        };

        let mut push_quarantine = |reason: String, payload_raw: &str| {
            let snippet = if payload_raw.len() > 200 {
                let end = payload_raw.floor_char_boundary(200);
                format!("{}…", &payload_raw[..end])
            } else {
                payload_raw.to_string()
            };
            eprintln!(
                "[brain] event-log replay: quarantined row id={row_id} profile={profile_id:?}: {reason}"
            );
            quarantined.push(QuarantinedRow {
                id: row_id,
                profile_id: profile_id.clone(),
                created_at,
                reason,
                payload_snippet: snippet,
            });
        };

        let payload_str = match row.get("payload") {
            Some(SqlValue::Text(s)) => s,
            _ => {
                push_quarantine("missing or non-text payload column".into(), "");
                continue;
            }
        };
        let event = match serde_json::from_str::<khive_storage::event::Event>(payload_str) {
            Ok(ev) => ev,
            Err(e) => {
                push_quarantine(format!("malformed event JSON: {e}"), payload_str);
                continue;
            }
        };
        // Semantic validation: a brain.feedback row with an invalid section_signals
        // payload must be quarantined whole — before any posterior state mutation.
        // This is the shared contract with the live brain.feedback handler.
        if event.verb == "brain.feedback" {
            if let Some(ss) = event.payload.get("section_signals") {
                if let Err(e) = crate::validate_section_signals(ss) {
                    push_quarantine(
                        format!("semantically invalid section_signals: {e}"),
                        payload_str,
                    );
                    continue;
                }
            }
        }
        events.push(event);
    }
    if !quarantined.is_empty() {
        eprintln!(
            "[brain] event-log replay: {} row(s) quarantined out of {} total; \
             replayed {} clean event(s)",
            quarantined.len(),
            rows.len(),
            events.len()
        );
    }
    Ok(LoadEventsResult {
        events,
        quarantined,
    })
}

pub async fn ensure_loaded(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    tracker: &Mutex<PersistenceTracker>,
    state: &Mutex<BrainState>,
    entity_capacity: usize,
) -> Result<(), RuntimeError> {
    let namespace = token.namespace().as_str().to_string();

    {
        let t = tracker.lock().unwrap();
        if t.is_active(&namespace) {
            return Ok(());
        }
    }

    let already_loaded = {
        let t = tracker.lock().unwrap();
        t.is_loaded(&namespace)
    };

    let brain_state: Option<BrainState> = if already_loaded {
        None
    } else {
        let sql = runtime.sql();
        let snapshot_result =
            load_latest_snapshot(sql.as_ref(), &namespace, entity_capacity).await?;

        let bs = if let Some((snapshot, updated_at)) = snapshot_result {
            let replay_result = load_events_since(sql.as_ref(), &namespace, updated_at).await?;

            let mut bs = BrainState::from_snapshot(snapshot, entity_capacity);

            for event in &replay_result.events {
                let signal = interpret(event);
                bs.balanced_recall.apply_signal(&signal);

                let serving_profile = event
                    .payload
                    .get("served_by_profile_id")
                    .and_then(|v| v.as_str())
                    .unwrap_or("balanced-recall-v1");

                {
                    let section_state =
                        crate::ensure_section_state_seeded(&mut bs.section_states, serving_profile);
                    section_state.apply_signal(&signal);
                }
            }

            crate::sync_balanced_recall_record(&mut bs);
            bs
        } else {
            BrainState::new(entity_capacity)
        };
        Some(bs)
    };

    // In test builds, honour the injected interleaving hook (if any).
    // The hook fires only for cold loads (brain_state.is_some() at this
    // point) so warm-path and already-loaded early-returns are unaffected.
    #[cfg(test)]
    if brain_state.is_some() {
        let hook = POST_LOAD_HOOK.lock().unwrap().take();
        if let Some(h) = hook {
            // Signal the test controller: "I am at the hook."
            let _ = h.reached_tx.send(());
            // Wait for the controller to give the go-ahead.
            let _ = h.proceed_rx.await;
        }
    }

    // Lock order: tracker → state (always in this sequence).
    // Reversing this order anywhere would risk deadlock; do not change.
    //
    // Publication is atomic: active_namespace, *state, and loaded_namespaces
    // are all updated inside a single tracker-held critical section.
    //
    // Re-check under the final guard: a concurrent loader may have completed
    // the same namespace while this task awaited the async DB load.  If so,
    // discard the stale brain_state we loaded and defer to the live slot.
    {
        let mut t = tracker.lock().unwrap();

        // Re-check 1: another loader already published this namespace as active.
        // The shared state slot already contains the live state; return without
        // clobbering it with our stale cold-loaded copy.
        if t.is_active(&namespace) {
            return Ok(());
        }

        // Re-check 2: another loader completed a cold load for this namespace
        // (it is now in saved_states) while this task was awaiting the DB.
        // Our brain_state was derived from a snapshot that is now stale relative
        // to any mutations that occurred after that loader published.  Discard
        // it so the save-restore path below uses the live saved state instead.
        let fresh_brain_state = if t.is_loaded(&namespace) {
            None
        } else {
            brain_state
        };

        let current_ns = t.active_namespace.clone();

        // Clone the parts of the current shared state we need for save-restore,
        // then immediately release the state lock so we can re-acquire it for
        // the final write (Rust Mutexes are not reentrant).
        let new_state = {
            let current_state = state.lock().unwrap();

            if let Some(ref from_ns) = current_ns {
                let saved_current = BrainState {
                    profiles: current_state.profiles.clone(),
                    balanced_recall: khive_brain_core::BalancedRecallState::from_snapshot(
                        current_state.balanced_recall.to_snapshot(),
                        entity_capacity,
                    ),
                    profile_states: current_state
                        .profile_states
                        .iter()
                        .map(|(k, v)| {
                            (
                                k.clone(),
                                khive_brain_core::BalancedRecallState::from_snapshot(
                                    v.to_snapshot(),
                                    entity_capacity,
                                ),
                            )
                        })
                        .collect(),
                    bindings: current_state.bindings.clone(),
                    section_states: current_state
                        .section_states
                        .iter()
                        .map(|(k, v)| {
                            (
                                k.clone(),
                                khive_brain_core::SectionPosteriorState::from_snapshot(
                                    v.to_snapshot(),
                                ),
                            )
                        })
                        .collect(),
                    router_state: current_state.router_state.clone(),
                    adapter_set: current_state.adapter_set.clone(),
                };
                // Release the state guard before mutating the tracker (save-restore
                // path) so the re-acquire below cannot deadlock.
                drop(current_state);

                let restored = t.swap_namespace(from_ns, saved_current, namespace.clone());
                fresh_brain_state
                    .or(restored)
                    .unwrap_or_else(|| BrainState::new(entity_capacity))
            } else {
                drop(current_state);
                // No active namespace yet — first load.  active_namespace is set
                // below together with the state write and loaded_namespaces mark.
                fresh_brain_state.unwrap_or_else(|| BrainState::new(entity_capacity))
            }
        };

        // Drain any hook signals that arrived before this load completed.
        // These are signals queued by `route_signal` while the namespace was
        // still cold (no prior `ensure_loaded`).  Applying them here, after the
        // snapshot + replay path, preserves ordering:
        //   persisted snapshot → replayed events → queued hook signals
        let pending = t.drain_pending_signals(&namespace);
        let mut final_state = new_state;
        for sig in &pending {
            final_state.balanced_recall.apply_signal(sig);
        }
        if !pending.is_empty() {
            crate::sync_balanced_recall_record(&mut final_state);
        }

        // Write the new state while the tracker lock is still held.
        // After this line active_namespace, *state, and loaded_namespaces are
        // all consistent; no concurrent dispatch can observe a partial view.
        *state.lock().unwrap() = final_state;

        // For the no-active-namespace (first-load) path swap_namespace was not
        // called, so we set active_namespace here before marking loaded.
        if current_ns.is_none() {
            t.active_namespace = Some(namespace.clone());
        }

        t.loaded_namespaces.insert(namespace, ());
    }

    Ok(())
}

// ── BRAIN-007: event-log replay quarantine diagnostics ────────────────────────

#[cfg(test)]
mod brain_007_replay_quarantine {
    use super::*;
    use khive_brain_core::BrainState;
    use khive_runtime::{KhiveRuntime, Namespace};
    use khive_storage::event::Event;
    use khive_types::{EventKind, SubstrateKind};
    use uuid::Uuid;

    async fn insert_raw_payload_at(
        rt: &KhiveRuntime,
        namespace: &str,
        payload: &str,
        created_at: i64,
    ) {
        let sql = rt.sql();
        let mut writer = sql.writer().await.expect("writer");
        writer
            .execute(SqlStatement {
                sql: "INSERT INTO brain_event_log (profile_id, namespace, event_kind, payload, created_at) VALUES (?1, ?2, ?3, ?4, ?5)".into(),
                params: vec![
                    SqlValue::Text("test-profile".to_string()),
                    SqlValue::Text(namespace.to_string()),
                    SqlValue::Text("brain.feedback".to_string()),
                    SqlValue::Text(payload.to_string()),
                    SqlValue::Integer(created_at),
                ],
                label: None,
            })
            .await
            .expect("insert raw row");
    }

    async fn insert_raw_payload(rt: &KhiveRuntime, namespace: &str, payload: &str) {
        insert_raw_payload_at(rt, namespace, payload, 1_000_000).await;
    }

    fn make_valid_event_json(namespace: &str) -> String {
        let ev = Event::new(
            namespace,
            "recall",
            EventKind::Audit,
            SubstrateKind::Note,
            "brain",
        );
        serde_json::to_string(&ev).expect("serialize event")
    }

    /// Build a brain.feedback event JSON with optional section_signals payload.
    fn make_feedback_event_json(
        namespace: &str,
        section_signals: Option<serde_json::Value>,
    ) -> String {
        let mut ev = Event::new(
            namespace,
            "brain.feedback",
            EventKind::Audit,
            SubstrateKind::Event,
            "brain",
        );
        ev.target_id = Some(Uuid::new_v4());
        let mut payload = serde_json::json!({"signal": "useful"});
        if let Some(ss) = section_signals {
            payload["section_signals"] = ss;
        }
        ev.payload = payload;
        serde_json::to_string(&ev).expect("serialize feedback event")
    }

    // ── structural quarantine (pre-existing) ──────────────────────────────────

    #[tokio::test]
    async fn malformed_json_rows_are_quarantined_not_panicked() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        // Insert one malformed row (not valid JSON) and one valid serialized Event.
        insert_raw_payload(&rt, ns, "this is not valid json {{").await;
        insert_raw_payload(&rt, ns, &make_valid_event_json(ns)).await;

        // load_events_since must return without panicking, quarantining the bad row.
        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("load must not fail on malformed rows");

        // Only the valid event should be returned.
        assert_eq!(
            result.events.len(),
            1,
            "one valid event expected; malformed row must be quarantined, not panic"
        );
        assert_eq!(result.quarantine_count(), 1, "quarantine_count must be 1");
    }

    #[tokio::test]
    async fn all_malformed_rows_quarantined_returns_empty_vec() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        insert_raw_payload(&rt, ns, "bad json").await;
        insert_raw_payload(&rt, ns, "{invalid}").await;

        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("load must succeed even when all rows are malformed");

        assert!(
            result.events.is_empty(),
            "all malformed rows must be quarantined"
        );
        assert_eq!(result.quarantine_count(), 2, "quarantine_count must be 2");
    }

    #[tokio::test]
    async fn clean_rows_replay_without_quarantine() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        for _ in 0..3 {
            insert_raw_payload(&rt, ns, &make_valid_event_json(ns)).await;
        }

        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("clean rows must replay without error");

        assert_eq!(result.events.len(), 3, "all 3 clean rows must be returned");
        assert_eq!(result.quarantine_count(), 0, "quarantine_count must be 0");
    }

    // ── semantic quarantine (BRAIN-007 new coverage) ──────────────────────────

    #[tokio::test]
    async fn empty_section_signals_quarantined() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        // brain.feedback with section_signals: {} must be quarantined whole.
        let poison = make_feedback_event_json(ns, Some(serde_json::json!({})));
        insert_raw_payload(&rt, ns, &poison).await;

        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("load must not fail");

        assert!(
            result.events.is_empty(),
            "empty section_signals must be quarantined"
        );
        assert_eq!(result.quarantine_count(), 1);
    }

    #[tokio::test]
    async fn unknown_section_signals_quarantined() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        // Unknown section key must be quarantined.
        let poison = make_feedback_event_json(
            ns,
            Some(serde_json::json!({"not_a_real_section": "useful"})),
        );
        insert_raw_payload(&rt, ns, &poison).await;

        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("load must not fail");

        assert!(
            result.events.is_empty(),
            "unknown section must be quarantined"
        );
        assert_eq!(result.quarantine_count(), 1);
    }

    #[tokio::test]
    async fn semantic_signal_in_section_signals_quarantined() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        // Section fold only accepts useful|not_useful|wrong; semantic event kinds
        // (explicit_positive, correction, …) must be quarantined.
        let poison = make_feedback_event_json(
            ns,
            Some(serde_json::json!({"overview": "explicit_positive"})),
        );
        insert_raw_payload(&rt, ns, &poison).await;

        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("load must not fail");

        assert!(
            result.events.is_empty(),
            "semantic signal in section_signals must be quarantined"
        );
        assert_eq!(result.quarantine_count(), 1);
    }

    // ── state isolation: bad rows must not advance posterior state ────────────

    /// Seed a snapshot, insert bad rows at FIRST / LAST / interleaved positions,
    /// then call the real ensure_loaded path and assert posterior state is unchanged.
    async fn seed_snapshot(rt: &KhiveRuntime, namespace: &str) -> BrainStateSnapshot {
        let state = BrainState::new(16);
        let snapshot = state.to_snapshot();
        let sql = rt.sql();
        upsert_snapshot(sql.as_ref(), namespace, &snapshot, 500_000)
            .await
            .expect("seed snapshot");
        snapshot
    }

    /// Assert that section posteriors and epoch are at the initial (baseline) values,
    /// meaning no section-fold mutation occurred from any replayed event.
    /// Does NOT assert balanced_recall state — clean recall/search events legitimately
    /// advance that without touching section state.
    fn assert_section_posteriors_at_baseline(state: &BrainState, baseline: &BrainState) {
        for key in state.section_states.keys() {
            let s = &state.section_states[key];
            let b = &baseline.section_states[key];
            assert_eq!(
                s.total_events, b.total_events,
                "section_states[{key}].total_events changed; bad row must not advance section state"
            );
            assert_eq!(
                s.exploration_epoch, b.exploration_epoch,
                "section_states[{key}].exploration_epoch changed; bad row must not advance section state"
            );
            for (st, p) in &s.posteriors {
                let bp = &b.posteriors[st];
                assert!(
                    (p.alpha() - bp.alpha()).abs() < 1e-12
                        && (p.beta() - bp.beta()).abs() < 1e-12,
                    "section posterior for {:?} changed: got alpha={} beta={}, expected alpha={} beta={}; \
                     bad row must not mutate posteriors",
                    st, p.alpha(), p.beta(), bp.alpha(), bp.beta()
                );
            }
        }
    }

    #[tokio::test]
    async fn bad_row_first_does_not_mutate_posterior_state() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();

        seed_snapshot(&rt, ns).await;

        // Row 1 (bad — semantic poison): created_at=600_000 (after snapshot at 500_000)
        let poison =
            make_feedback_event_json(ns, Some(serde_json::json!({"overview": "correction"})));
        insert_raw_payload_at(&rt, ns, &poison, 600_001).await;

        // Row 2 (clean recall event): created_at=600_002
        insert_raw_payload_at(&rt, ns, &make_valid_event_json(ns), 600_002).await;

        // Use load_events_since directly and replay manually to assert isolation.
        let sql = rt.sql();
        let result = load_events_since(sql.as_ref(), ns, 500_000)
            .await
            .expect("load must not fail");

        assert_eq!(
            result.quarantine_count(),
            1,
            "bad first row must be quarantined"
        );
        assert_eq!(result.events.len(), 1, "one clean event must pass through");

        // Apply the clean events to a fresh state and confirm section state is at baseline.
        let baseline = BrainState::new(16);
        let mut state = BrainState::new(16);
        for event in &result.events {
            let signal = crate::event::interpret(event);
            state.balanced_recall.apply_signal(&signal);
            for section_state in state.section_states.values_mut() {
                section_state.apply_signal(&signal);
            }
        }
        // The single clean event is a recall, not a feedback; section state unchanged.
        assert_section_posteriors_at_baseline(&state, &baseline);
    }

    #[tokio::test]
    async fn bad_row_last_does_not_mutate_posterior_state() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();

        seed_snapshot(&rt, ns).await;

        // Row 1 (clean): created_at=600_001
        insert_raw_payload_at(&rt, ns, &make_valid_event_json(ns), 600_001).await;
        // Row 2 (bad — empty section_signals): created_at=600_002
        let poison = make_feedback_event_json(ns, Some(serde_json::json!({})));
        insert_raw_payload_at(&rt, ns, &poison, 600_002).await;

        let sql = rt.sql();
        let result = load_events_since(sql.as_ref(), ns, 500_000)
            .await
            .expect("load must not fail");

        assert_eq!(
            result.quarantine_count(),
            1,
            "bad last row must be quarantined"
        );
        assert_eq!(result.events.len(), 1, "one clean event must pass through");

        let baseline = BrainState::new(16);
        let mut state = BrainState::new(16);
        for event in &result.events {
            let signal = crate::event::interpret(event);
            state.balanced_recall.apply_signal(&signal);
            for section_state in state.section_states.values_mut() {
                section_state.apply_signal(&signal);
            }
        }
        assert_section_posteriors_at_baseline(&state, &baseline);
    }

    #[tokio::test]
    async fn bad_rows_interleaved_do_not_mutate_posterior_state() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();

        seed_snapshot(&rt, ns).await;

        // clean, bad, clean, bad, clean
        insert_raw_payload_at(&rt, ns, &make_valid_event_json(ns), 600_001).await;
        let p1 = make_feedback_event_json(
            ns,
            Some(serde_json::json!({"overview": "explicit_negative"})),
        );
        insert_raw_payload_at(&rt, ns, &p1, 600_002).await;
        insert_raw_payload_at(&rt, ns, &make_valid_event_json(ns), 600_003).await;
        let p2 = make_feedback_event_json(ns, Some(serde_json::json!({})));
        insert_raw_payload_at(&rt, ns, &p2, 600_004).await;
        insert_raw_payload_at(&rt, ns, &make_valid_event_json(ns), 600_005).await;

        let sql = rt.sql();
        let result = load_events_since(sql.as_ref(), ns, 500_000)
            .await
            .expect("load must not fail");

        assert_eq!(
            result.quarantine_count(),
            2,
            "2 bad rows must be quarantined"
        );
        assert_eq!(result.events.len(), 3, "3 clean rows must pass through");

        // Apply only clean events; section posteriors must stay at baseline.
        let baseline = BrainState::new(16);
        let mut state = BrainState::new(16);
        for event in &result.events {
            let signal = crate::event::interpret(event);
            state.balanced_recall.apply_signal(&signal);
            for section_state in state.section_states.values_mut() {
                section_state.apply_signal(&signal);
            }
        }
        assert_section_posteriors_at_baseline(&state, &baseline);
    }

    // ── quarantine metadata: id and reason must be returned, not just counted ──

    #[tokio::test]
    async fn quarantined_rows_return_id_and_reason() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        // Two bad rows: one malformed JSON, one semantic poison.
        insert_raw_payload(&rt, ns, "not json at all").await;
        let poison = make_feedback_event_json(ns, Some(serde_json::json!({})));
        insert_raw_payload(&rt, ns, &poison).await;

        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("load must not fail");

        assert_eq!(
            result.quarantine_count(),
            2,
            "both bad rows must be quarantined"
        );
        assert!(result.events.is_empty(), "no clean events expected");

        // Each quarantined entry must carry a non-zero id, non-empty reason,
        // correct profile_id, non-zero created_at, and non-empty payload_snippet.
        for qr in &result.quarantined {
            assert!(
                qr.id > 0,
                "quarantined row id must be a real autoincrement id; got {}",
                qr.id
            );
            assert!(
                !qr.reason.is_empty(),
                "quarantined row must carry a non-empty reason string"
            );
            assert_eq!(
                qr.profile_id, "test-profile",
                "quarantined row profile_id must match inserted value; got {:?}",
                qr.profile_id
            );
            assert!(
                qr.created_at > 0,
                "quarantined row created_at must be non-zero; got {}",
                qr.created_at
            );
            assert!(
                !qr.payload_snippet.is_empty(),
                "quarantined row payload_snippet must be non-empty"
            );
        }

        // First row: malformed JSON — reason must mention JSON or malformed.
        assert!(
            result.quarantined[0].reason.contains("malformed")
                || result.quarantined[0].reason.contains("JSON")
                || result.quarantined[0].reason.contains("json"),
            "first quarantine reason must describe malformed JSON; got: {:?}",
            result.quarantined[0].reason
        );
        // First row payload_snippet must contain a recognizable fragment of the bad payload.
        assert!(
            result.quarantined[0].payload_snippet.contains("not json"),
            "first row snippet must contain 'not json'; got: {:?}",
            result.quarantined[0].payload_snippet
        );

        // Second row: semantic poison (empty section_signals) — reason must mention section_signals.
        assert!(
            result.quarantined[1].reason.contains("section_signals"),
            "second quarantine reason must mention section_signals; got: {:?}",
            result.quarantined[1].reason
        );
        // Second row payload_snippet must be non-empty (it's valid JSON, so it has content).
        assert!(
            !result.quarantined[1].payload_snippet.is_empty(),
            "second row payload_snippet must be non-empty; got: {:?}",
            result.quarantined[1].payload_snippet
        );
    }

    // ── char-boundary safety: multibyte payload does not panic ───────────────

    #[tokio::test]
    async fn payload_snippet_truncation_safe_on_multibyte_chars() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        // '日' is 3 bytes in UTF-8; 250 repetitions = 750 bytes, well over the 200-byte limit.
        // A naive &s[..200] would land mid-char and panic.
        let long_multibyte: String = "日".repeat(250);
        insert_raw_payload(&rt, ns, &long_multibyte).await;

        // Must not panic.
        let result = load_events_since(sql.as_ref(), ns, 0)
            .await
            .expect("load must not fail");

        assert_eq!(result.quarantine_count(), 1);
        let qr = &result.quarantined[0];

        // Snippet must be valid UTF-8 (Rust strings are always valid UTF-8, but
        // verify it is non-empty and ≤ 200 bytes as a byte-length check).
        assert!(
            !qr.payload_snippet.is_empty(),
            "snippet must be non-empty for a non-empty payload"
        );
        // The trailing ellipsis is a multi-byte char itself; strip it before measuring.
        let snippet_body = qr.payload_snippet.trim_end_matches('…');
        assert!(
            snippet_body.len() <= 200,
            "snippet body (sans ellipsis) must be ≤200 bytes; got {} bytes",
            snippet_body.len()
        );
        // The snippet body must itself be valid UTF-8 (Rust ensures this, but also
        // assert it only contains the expected character).
        assert!(
            snippet_body.chars().all(|c| c == '日'),
            "snippet body must contain only '日' chars; got: {:?}",
            snippet_body
        );
    }
}

// ── BRAINCORE-AUD-001: entity-posterior cache-capacity enforcement on load ────

#[cfg(test)]
mod braincore_aud_001_capacity {
    use uuid::Uuid;

    use super::*;
    use khive_brain_core::BetaPosterior;
    use khive_runtime::{KhiveRuntime, Namespace};

    /// Oversized persisted snapshot restore is bounded/rejected: a snapshot
    /// with more entity posteriors than `ENTITY_CACHE_CAPACITY` must fail
    /// capacity-aware validation at the load boundary rather than silently
    /// restoring past the configured bound.
    #[tokio::test]
    async fn oversized_brain_snapshot_load_returns_runtime_error() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        let capacity = 2;
        let mut state = BrainState::new(capacity);
        for _ in 0..capacity {
            state
                .balanced_recall
                .entity_posteriors
                .get_or_insert(Uuid::new_v4(), BetaPosterior::default);
        }
        let mut snapshot = state.to_snapshot();
        // Force one entry past the configured capacity — simulates a crafted
        // or legacy oversized snapshot that live inserts could never produce.
        snapshot
            .balanced_recall
            .entity_posteriors
            .insert(Uuid::new_v4(), BetaPosterior::default());

        upsert_snapshot(sql.as_ref(), ns, &snapshot, 500_000)
            .await
            .expect("seed oversized snapshot");

        let result = load_latest_snapshot(sql.as_ref(), ns, capacity).await;
        assert!(
            result.is_err(),
            "snapshot with entity_posteriors.len() > capacity must be rejected at load"
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("snapshot invariant violation"),
            "error must name the invariant-violation load boundary, got: {err}"
        );
    }

    /// A snapshot within the configured capacity restores without error.
    #[tokio::test]
    async fn in_capacity_brain_snapshot_load_succeeds() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        let capacity = 2;
        let state = BrainState::new(capacity);
        let snapshot = state.to_snapshot();
        upsert_snapshot(sql.as_ref(), ns, &snapshot, 500_000)
            .await
            .expect("seed snapshot");

        let result = load_latest_snapshot(sql.as_ref(), ns, capacity).await;
        assert!(result.is_ok(), "in-capacity snapshot must load cleanly");
    }

    /// PR #535: a persisted snapshot with `{A, B}`
    /// entity_posteriors, `entity_posterior_order = [A]`, and an EXPLICIT
    /// `entity_posteriors_version = 0` must be rejected at the
    /// `load_latest_snapshot` boundary rather than silently normalized by
    /// `restore` to `[A, B]`. Version 0 with a non-empty order is not the
    /// legacy empty-order compatibility case.
    #[tokio::test]
    async fn version_zero_snapshot_with_explicit_partial_order_rejected_at_load() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        let capacity = 10;
        let mut state = BrainState::new(capacity);
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        state
            .balanced_recall
            .entity_posteriors
            .get_or_insert(a, BetaPosterior::default);
        state
            .balanced_recall
            .entity_posteriors
            .get_or_insert(b, BetaPosterior::default);

        let mut snapshot = state.to_snapshot();
        snapshot.balanced_recall.entity_posteriors_version = 0;
        snapshot.balanced_recall.entity_posterior_order = vec![a];

        upsert_snapshot(sql.as_ref(), ns, &snapshot, 500_000)
            .await
            .expect("seed corrupt version-0 snapshot");

        let result = load_latest_snapshot(sql.as_ref(), ns, capacity).await;
        assert!(
            result.is_err(),
            "version-0 snapshot with non-empty partial order must be rejected at load, not normalized"
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("snapshot invariant violation"),
            "error must name the load-boundary invariant violation, got: {err}"
        );
    }

    /// Same corruption as above, but with `entity_posteriors_version` OMITTED
    /// from the stored JSON entirely (rather than set to an explicit `0`) —
    /// the `#[serde(default)]` path that made this reachable in the first
    /// place. Inserted as a raw row (bypassing the typed `upsert_snapshot`,
    /// which always serializes the full struct) to reproduce exactly what a
    /// pre-versioning writer, or a hand-crafted snapshot, would have stored.
    #[tokio::test]
    async fn version_omitted_snapshot_with_partial_order_rejected_at_load() {
        let rt = KhiveRuntime::memory().expect("memory runtime");
        let token = rt.authorize(Namespace::local()).expect("token");
        let ns = token.namespace().as_str();
        let sql = rt.sql();

        let capacity = 10;
        let mut state = BrainState::new(capacity);
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        state
            .balanced_recall
            .entity_posteriors
            .get_or_insert(a, BetaPosterior::default);
        state
            .balanced_recall
            .entity_posteriors
            .get_or_insert(b, BetaPosterior::default);

        let mut snapshot = state.to_snapshot();
        snapshot.balanced_recall.entity_posterior_order = vec![a];

        let mut json = serde_json::to_value(&snapshot).expect("serialize snapshot");
        json["balanced_recall"]
            .as_object_mut()
            .expect("balanced_recall is an object")
            .remove("entity_posteriors_version");
        let json_str = serde_json::to_string(&json).expect("re-serialize snapshot");

        // Round-trip through the typed struct to confirm the omitted field
        // really does serde-default to 0 before we bypass the typed insert
        // path — this is the precondition the fix closes.
        let reparsed: BrainStateSnapshot =
            serde_json::from_str(&json_str).expect("deserialize snapshot with omitted version");
        assert_eq!(
            reparsed.balanced_recall.entity_posteriors_version, 0,
            "omitted entity_posteriors_version must serde-default to 0"
        );

        let mut writer = sql.writer().await.expect("writer");
        writer
            .execute(SqlStatement {
                sql: "INSERT INTO brain_profile_snapshots (profile_id, namespace, snapshot_json, updated_at) VALUES (?1, ?2, ?3, ?4)".into(),
                params: vec![
                    SqlValue::Text(SNAPSHOT_PROFILE_ID.to_string()),
                    SqlValue::Text(ns.to_string()),
                    SqlValue::Text(json_str),
                    SqlValue::Integer(500_000),
                ],
                label: None,
            })
            .await
            .expect("insert raw snapshot row with omitted version");
        drop(writer);

        let result = load_latest_snapshot(sql.as_ref(), ns, capacity).await;
        assert!(
            result.is_err(),
            "snapshot with omitted entity_posteriors_version and non-empty partial order must be rejected at load"
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("snapshot invariant violation"),
            "error must name the load-boundary invariant violation, got: {err}"
        );
    }
}

// ── ADR-067 Fork C slice 2: persist_brain_state_mutation
// write-queue routing ──────────────────────────────────────────────────────

#[cfg(test)]
mod persist_write_queue_routing {
    use super::*;
    use khive_brain_core::BrainState;
    use khive_runtime::{KhiveRuntime, Namespace};

    /// Proves `persist_brain_state_mutation` is actually enqueued on the
    /// pool's shared `WriterTaskHandle` channel when the write queue is
    /// enabled, mirroring `fold_gate.rs`'s equivalent routing test (same
    /// `queue_depth` + occupier-parked-on-oneshot technique; see that test's
    /// doc comment for why a wall-clock test can't distinguish this).
    #[tokio::test]
    async fn persist_brain_state_mutation_routes_through_writer_task_when_flag_enabled() {
        // Token minted from an in-memory runtime: never touches the write
        // queue / env var, used only for its `NamespaceToken`.
        let token_rt = KhiveRuntime::memory().expect("memory runtime for token minting");
        let token = token_rt
            .authorize(Namespace::local())
            .expect("authorize local namespace");

        // The actual storage this test exercises: a bare, file-backed,
        // write-queue-enabled pool/bridge — no KhiveRuntime, no env var.
        let dir = tempfile::tempdir().expect("tempdir");
        let db_path = dir.path().join("persist-write-queue-routing.db");
        let pool_cfg = khive_db::PoolConfig {
            path: Some(db_path),
            write_queue_enabled: true,
            ..khive_db::PoolConfig::default()
        };
        let pool = std::sync::Arc::new(khive_db::ConnectionPool::new(pool_cfg).expect("pool"));
        {
            let mut writer = pool.writer().expect("writer");
            khive_db::run_migrations(writer.conn_mut()).expect("migrations");
        }
        let sql: std::sync::Arc<dyn SqlAccess> =
            std::sync::Arc::new(khive_db::SqlBridge::new(std::sync::Arc::clone(&pool), true));

        let writer_task = pool
            .writer_task_handle()
            .unwrap()
            .expect("writer task must be spawned with the flag on for a file-backed pool");

        let (started_tx, started_rx) = tokio::sync::oneshot::channel::<()>();
        let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
        let occupier = {
            let writer_task = writer_task.clone();
            tokio::spawn(async move {
                writer_task
                    .send(move |_conn| {
                        let _ = started_tx.send(());
                        let _ = release_rx.blocking_recv();
                        Ok::<(), khive_storage::StorageError>(())
                    })
                    .await
            })
        };

        started_rx
            .await
            .expect("occupier must signal it has started running inside the writer task");
        assert_eq!(
            writer_task.queue_depth(),
            0,
            "channel must start empty once the occupier has been dequeued and is running"
        );

        let tracker: Mutex<PersistenceTracker> = Mutex::new(PersistenceTracker::new());
        let state: Mutex<BrainState> = Mutex::new(BrainState::new(16));

        let persist_task = tokio::spawn(async move {
            persist_brain_state_mutation(
                sql.as_ref(),
                &token,
                &tracker,
                &state,
                BrainMutationEvent {
                    profile_id: "write-queue-routing-profile".to_string(),
                    event_kind: "brain.test_mutation".to_string(),
                    payload: serde_json::json!({"probe": true}),
                },
                16,
                |_state: &mut BrainState| -> Result<(), RuntimeError> { Ok(()) },
            )
            .await
        });

        let mut saw_enqueued = false;
        for _ in 0..100 {
            if writer_task.queue_depth() >= 1 {
                saw_enqueued = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        assert!(
            saw_enqueued,
            "persist_brain_state_mutation's atomic_unit request never appeared in the \
             writer task's channel while the occupier held the single drain slot — \
             atomic_unit is not routing this call through the shared writer task"
        );

        release_tx
            .send(())
            .expect("occupier must still be waiting on the release signal");
        occupier
            .await
            .expect("occupier task must not panic")
            .expect("occupier write must succeed");

        persist_task
            .await
            .expect("persist_task must not panic")
            .expect("persist_brain_state_mutation must succeed once unblocked");
    }
}