tidaldb/tidal/src/db/mod.rs

//! The public entry point for tidalDB.
//!
//! This module provides [`TidalDb`] (the database handle) and
//! [`TidalDbBuilder`] (the fluent construction API). All interaction
//! with tidalDB starts here.
//!
//! # Quick Start
//!
//! ```
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! use tidaldb::TidalDb;
//!
//! // In-memory database — no filesystem access, perfect for tests:
//! let db = TidalDb::builder().ephemeral().open()?;
//! db.health_check()?;
//! db.close()?;
//! # Ok(())
//! # }
//! ```

pub mod builder;
pub mod config;
#[cfg(feature = "metrics")]
pub mod http;
pub mod metrics;
pub mod paths;
#[cfg(any(test, feature = "test-utils"))]
pub mod temp;
pub(crate) mod wal_bridge;

pub use builder::TidalDbBuilder;
pub use config::{Config, ConfigError, StorageMode};
pub use metrics::MetricsState;
pub use paths::Paths;
#[cfg(any(test, feature = "test-utils"))]
pub use temp::TempTidalHome;

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Duration;

use roaring::RoaringBitmap;

use crate::entities::{
    CreatorItemsBitmap, HardNegIndex, InteractionLedger, PreferenceVectors, RelationshipType,
    UserStateIndex,
};
use crate::query::executor::RetrieveExecutor;
use crate::query::retrieve::{Results, Retrieve};
use crate::ranking::builtins::register_builtins;
use crate::ranking::registry::ProfileRegistry;
use crate::schema::{DurabilityError, EntityId, EntityKind, Schema, TidalError, Timestamp, Window};
#[allow(unused_imports)]
// Session API types used by upcoming start_session/close_session/signal_with_session methods.
use crate::session::{
    self as session_mod, AgentId, AuditEntry, SessionHandle, SessionId, SessionInfo,
    SessionSnapshot, SessionState, SessionSummary,
};
use crate::signals::{NoopWalWriter, SignalLedger, SignalTypeId};
use crate::storage::indexes::bitmap::BitmapIndex;
use crate::storage::indexes::range::RangeIndex;
use crate::storage::vector::registry::EmbeddingSlotRegistry;
use crate::storage::{InMemoryBackend, StorageEngine, Tag, encode_key};
use crate::wal::{WalConfig, WalHandle};

use self::wal_bridge::WalHandleWriter;

// ── Storage abstraction ───────────────────────────────────────────────────────

/// Wraps either in-memory backends (ephemeral mode) or a fjall storage
/// (persistent mode) behind a uniform interface.
///
/// M3 provides three backends: items, users, creators. In ephemeral mode
/// each is an independent `InMemoryBackend`; in persistent mode they share
/// a single `FjallStorage` with three keyspaces.
pub(crate) enum StorageBox {
    Memory {
        items: InMemoryBackend,
        users: InMemoryBackend,
        creators: InMemoryBackend,
    },
    Fjall(crate::storage::FjallStorage),
}

impl StorageBox {
    /// Reference to the items storage engine.
    fn items_engine(&self) -> &dyn StorageEngine {
        match self {
            Self::Memory { items, .. } => items,
            Self::Fjall(f) => f.backend(EntityKind::Item),
        }
    }

    /// Reference to the users storage engine.
    fn users_engine(&self) -> &dyn StorageEngine {
        match self {
            Self::Memory { users, .. } => users,
            Self::Fjall(f) => f.backend(EntityKind::User),
        }
    }

    /// Reference to the creators storage engine.
    fn creators_engine(&self) -> &dyn StorageEngine {
        match self {
            Self::Memory { creators, .. } => creators,
            Self::Fjall(f) => f.backend(EntityKind::Creator),
        }
    }

    /// Flush all buffered writes to durable storage.
    fn flush(&self) -> crate::Result<()> {
        match self {
            Self::Memory { .. } => Ok(()),
            Self::Fjall(f) => f.flush_all().map_err(TidalError::from),
        }
    }
}

// ── Metadata serialization ────────────────────────────────────────────────────

/// Serialize `HashMap<String, String>` as length-prefixed binary pairs.
///
/// This is the **items-only** metadata format (no embedding header).
/// For user/creator entities (which include an optional embedding prefix),
/// see [`entities::serialize_entity`](crate::entities::serialize_entity).
///
/// The two formats are intentionally different: items store only metadata
/// (embeddings go to the HNSW index), while users/creators store an optional
/// embedding inline with their metadata for cold-start preference vectors.
///
/// Format (all lengths little-endian u32):
/// ```text
/// [num_entries: 4 bytes]
/// for each entry:
///   [key_len: 4 bytes][key bytes]
///   [val_len: 4 bytes][value bytes]
/// ```
fn serialize_metadata(map: &HashMap<String, String>) -> Vec<u8> {
    #[allow(clippy::cast_possible_truncation)]
    let mut buf = Vec::new();
    buf.extend_from_slice(&(map.len() as u32).to_le_bytes());
    for (k, v) in map {
        buf.extend_from_slice(&(k.len() as u32).to_le_bytes());
        buf.extend_from_slice(k.as_bytes());
        buf.extend_from_slice(&(v.len() as u32).to_le_bytes());
        buf.extend_from_slice(v.as_bytes());
    }
    buf
}

/// Deserialize `HashMap<String, String>` from the binary format produced by
/// [`serialize_metadata`].
///
/// This is the **items-only** metadata format (no embedding header).
/// For user/creator entities, see [`entities::deserialize_entity`](crate::entities::deserialize_entity).
///
/// Returns an empty map if the bytes are empty or malformed -- metadata reads
/// must never panic or fail the query.
fn deserialize_metadata(bytes: &[u8]) -> HashMap<String, String> {
    let mut map = HashMap::new();
    if bytes.len() < 4 {
        return map;
    }
    let count = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize;
    let mut pos = 4;
    for _ in 0..count {
        if pos + 4 > bytes.len() {
            break;
        }
        let key_len =
            u32::from_le_bytes([bytes[pos], bytes[pos + 1], bytes[pos + 2], bytes[pos + 3]])
                as usize;
        pos += 4;
        if pos + key_len > bytes.len() {
            break;
        }
        let key = String::from_utf8_lossy(&bytes[pos..pos + key_len]).to_string();
        pos += key_len;
        if pos + 4 > bytes.len() {
            break;
        }
        let val_len =
            u32::from_le_bytes([bytes[pos], bytes[pos + 1], bytes[pos + 2], bytes[pos + 3]])
                as usize;
        pos += 4;
        if pos + val_len > bytes.len() {
            break;
        }
        let val = String::from_utf8_lossy(&bytes[pos..pos + val_len]).to_string();
        pos += val_len;
        map.insert(key, val);
    }
    map
}

// ── Signal classification ────────────────────────────────────────────────

/// Returns `true` for positive engagement signal types that should update
/// the user's preference vector.
///
/// These are high-intent signals: the user explicitly chose to engage with
/// the content. Views are excluded because they are low-signal (the user
/// may have scrolled past without interest).
fn is_positive_engagement_signal(signal_type: &str) -> bool {
    matches!(signal_type, "like" | "share" | "completion")
}

// ── Entity state rebuild ─────────────────────────────────────────────────────

/// Rebuild in-memory entity state from durable storage on restart.
///
/// Scans the users keyspace for relationship edges and the items keyspace for
/// `creator_id` metadata. Populates:
/// 1. `user_state.blocked` from `RelationshipType::Blocks` edges
/// 2. `user_state.seen` (hidden items) from `RelationshipType::Hide` edges
/// 3. `user_state.follows` from `RelationshipType::Follows` edges
/// 4. `creator_items` bitmap from items with `creator_id` metadata
/// 5. `interaction_ledger` from `RelationshipType::InteractionWeight` edges
///
/// For ephemeral mode, all engines are empty, so this is effectively a no-op.
fn rebuild_entity_state(
    storage: &StorageBox,
    user_state: &UserStateIndex,
    creator_items: &CreatorItemsBitmap,
    interaction_ledger: &InteractionLedger,
) -> crate::Result<()> {
    use crate::entities::relationship::{
        RelationshipType, deserialize_relationship_value, parse_relationship_to,
    };
    use crate::storage::keys::parse_key;

    // Scan the users keyspace for all relationship edges.
    // The relationship key format is:
    //   [from_entity_id: 8 BE][0x00][Tag::Rel (0x04)][rel_type: 1][to_entity_id: 8 BE]
    // We scan with an empty prefix to get all keys, then filter for Tag::Rel.
    let mut rel_count = 0u64;
    for entry in storage.users_engine().scan_prefix(&[]) {
        let (key, value) = entry.map_err(TidalError::from)?;

        // Only process relationship keys (Tag::Rel = 0x04).
        if let Some((from_id, Tag::Rel, suffix)) = parse_key(&key) {
            // suffix = [rel_type: 1 byte][to_entity_id: 8 BE]
            if suffix.is_empty() {
                continue;
            }
            let rel_type_byte = suffix[0];
            let Some(rel_type) = RelationshipType::from_byte(rel_type_byte) else {
                continue;
            };
            let Some(to_id) = parse_relationship_to(&key) else {
                continue;
            };
            let from_id_u64 = from_id.as_u64();

            match rel_type {
                RelationshipType::Blocks => {
                    user_state.add_block_creator(from_id_u64, to_id.as_u64());
                    rel_count += 1;
                }
                RelationshipType::Hide => {
                    #[allow(clippy::cast_possible_truncation)]
                    user_state.add_hide(from_id_u64, to_id.as_u64() as u32);
                    rel_count += 1;
                }
                RelationshipType::Follows => {
                    user_state.add_follow(from_id_u64, to_id.as_u64());
                    rel_count += 1;
                }
                RelationshipType::InteractionWeight => {
                    // Reconstruct interaction weight from the stored edge value.
                    if let Some((weight, ts_nanos)) = deserialize_relationship_value(&value) {
                        interaction_ledger.record(from_id_u64, to_id.as_u64(), weight, ts_nanos);
                        rel_count += 1;
                    }
                }
                RelationshipType::Mute => {
                    // Mute edges do not have in-memory state (yet).
                    rel_count += 1;
                }
            }
        }
    }

    // Scan items keyspace for creator_id metadata to rebuild creator_items bitmap.
    let mut item_count = 0u64;
    for entry in storage.items_engine().scan_prefix(&[]) {
        let (key, value) = entry.map_err(TidalError::from)?;

        if let Some((entity_id, Tag::Meta, _suffix)) = parse_key(&key) {
            let meta = deserialize_metadata(&value);
            if let Some(creator_str) = meta.get("creator_id")
                && let Ok(creator_id) = creator_str.parse::<u64>()
            {
                #[allow(clippy::cast_possible_truncation)]
                creator_items.add_item(creator_id, entity_id.as_u64() as u32);
                item_count += 1;
            }
        }
    }

    if rel_count > 0 || item_count > 0 {
        tracing::info!(
            relationships = rel_count,
            creator_items = item_count,
            "entity state rebuilt from durable storage"
        );
    }

    Ok(())
}

// ── Periodic checkpoint ───────────────────────────────────────────────────────

/// Background thread body: checkpoint signal state to storage every 30 seconds.
///
/// Polls the shutdown flag every 500ms so the thread exits promptly when
/// `shutdown_inner()` is called. Only runs in persistent mode (ephemeral opens
/// never spawn this thread).
///
/// The `Arc` arguments are intentionally passed by value: the thread must own
/// them for its entire lifetime (references cannot satisfy the `'static` bound
/// required by `std::thread::spawn`).
#[allow(clippy::needless_pass_by_value)]
fn run_checkpoint_thread(
    shutdown: Arc<AtomicBool>,
    ledger: Arc<SignalLedger>,
    storage: Box<dyn StorageEngine + Send + Sync>,
    last_wal_seq: Arc<AtomicU64>,
) {
    const CHECKPOINT_INTERVAL: Duration = Duration::from_secs(30);
    const POLL_INTERVAL: Duration = Duration::from_millis(500);

    let mut elapsed = Duration::ZERO;
    loop {
        std::thread::sleep(POLL_INTERVAL);
        if shutdown.load(Ordering::Acquire) {
            break;
        }
        elapsed += POLL_INTERVAL;
        if elapsed >= CHECKPOINT_INTERVAL {
            elapsed = Duration::ZERO;
            let meta = crate::signals::checkpoint::CheckpointMeta {
                checkpoint_time_ns: Timestamp::now().as_nanos(),
                wal_sequence: last_wal_seq.load(Ordering::Relaxed),
            };
            if let Err(e) = ledger.checkpoint(storage.as_ref(), meta) {
                tracing::error!(error = %e, "periodic checkpoint failed");
            } else {
                tracing::debug!("periodic checkpoint written");
            }
        }
    }
}

// ── OpenResult ──────────────────────────────────────────────────────────────

/// Bundle returned by [`TidalDb::open_with_schema`] to avoid a fragile tuple.
///
/// Each field is consumed by [`TidalDb::from_parts`] during construction.
pub(crate) struct OpenResult {
    pub storage: StorageBox,
    pub ledger: Arc<SignalLedger>,
    pub wal: Option<WalHandle>,
    pub last_seq: Arc<AtomicU64>,
    pub profile_registry: ProfileRegistry,
    pub category_index: BitmapIndex,
    pub format_index: BitmapIndex,
    pub creator_index: BitmapIndex,
    pub tag_index: BitmapIndex,
    pub duration_index: RangeIndex<u32>,
    pub created_at_index: RangeIndex<u64>,
    pub universe: RoaringBitmap,
    pub embedding_registry: EmbeddingSlotRegistry,
    pub schema_def: Schema,
    // M3 entity state rebuilt from durable storage.
    pub creator_items: CreatorItemsBitmap,
    pub user_state: UserStateIndex,
    pub hard_negatives: HardNegIndex,
    pub interaction_ledger: InteractionLedger,
    pub preference_vectors: PreferenceVectors,
}

// ── TidalDb ───────────────────────────────────────────────────────────────────

/// A tidalDB database instance.
///
/// Created via [`TidalDb::builder()`]. After M1p5, the database wires in the
/// storage engine, signal ledger, and WAL behind this facade. M2 adds bitmap
/// indexes, range indexes, a universe bitmap, embedding registry, profile
/// registry, and the RETRIEVE query executor.
///
/// # Shutdown
///
/// Call [`close`](Self::close) for explicit, checked shutdown. If dropped
/// without calling `close`, the [`Drop`] implementation will run best-effort
/// cleanup and log any errors via `tracing::error!`.
///
/// # Thread Safety
///
/// `TidalDb` is `Send + Sync`. Wrap it in an `Arc` to share across threads.
pub struct TidalDb {
    config: Config,
    /// Whether `close()` has been called. Prevents double-shutdown.
    closed: AtomicBool,
    /// Runtime metrics shared with the optional HTTP server.
    metrics: Arc<MetricsState>,
    /// Handle to the metrics HTTP server thread (metrics feature only).
    #[cfg(feature = "metrics")]
    metrics_handle: Option<http::MetricsHandle>,
    /// Signal ledger: in-memory hot + warm tier state.
    /// `None` if no schema was provided at open time.
    ledger: Option<Arc<SignalLedger>>,
    /// Storage engine: routes to the correct backend by entity kind.
    /// `None` in ephemeral mode without a schema.
    storage: Option<StorageBox>,
    /// The live WAL handle. Wrapped in `Mutex<Option<...>>` so
    /// `shutdown_inner(&self)` can take ownership for graceful shutdown.
    wal: std::sync::Mutex<Option<WalHandle>>,
    /// Highest WAL sequence number committed by `WalHandleWriter`.
    /// Shared with the bridge; read at checkpoint time.
    last_wal_seq: Arc<AtomicU64>,
    /// Shutdown flag for the periodic checkpoint background thread.
    /// Set to `true` in `shutdown_inner()` to stop the thread.
    shutdown_checkpoint: Arc<AtomicBool>,
    /// Periodic checkpoint background thread (persistent mode only).
    /// Wrapped in `Mutex<Option<...>>` so `shutdown_inner(&self)` can join it.
    checkpoint_thread: std::sync::Mutex<Option<std::thread::JoinHandle<()>>>,

    // ── M2 indexes ────────────────────────────────────────────────────
    /// Bitmap index: `entity_id` -> category tag.
    category_index: BitmapIndex,
    /// Bitmap index: `entity_id` -> format tag.
    format_index: BitmapIndex,
    /// Bitmap index: `entity_id` -> creator ID (as string).
    creator_index: BitmapIndex,
    /// Bitmap index: `entity_id` -> tag set.
    tag_index: BitmapIndex,
    /// Range index: `entity_id` -> duration (seconds as u32).
    duration_index: RangeIndex<u32>,
    /// Range index: `entity_id` -> `created_at` timestamp (nanos as u64).
    created_at_index: RangeIndex<u64>,
    /// Universe bitmap: set of all known entity IDs (u32 truncated).
    universe: Arc<RwLock<RoaringBitmap>>,
    /// Embedding slot registry: maps (`EntityKind`, `slot_name`) to HNSW index state.
    embedding_registry: Arc<RwLock<EmbeddingSlotRegistry>>,
    /// Ranking profile registry: named, versioned scoring function definitions.
    profile_registry: ProfileRegistry,
    /// Frozen schema definition. Stored so new instances can inspect signal types,
    /// embedding slots, and session policies at runtime.
    #[allow(dead_code)]
    // Used by upcoming session policy evaluation and schema introspection API.
    schema_def: Option<Schema>,

    // ── M3 entities ──────────────────────────────────────────────────
    /// Maps each creator to the set of item IDs they have produced.
    /// Populated during `write_item_with_metadata` when `creator_id` is present.
    creator_items: Arc<CreatorItemsBitmap>,
    /// Per-user state bitmaps (seen, blocked, saved, liked, completion).
    /// Thread-safe via `DashMap` -- no mutex on the hot path.
    user_state: Arc<UserStateIndex>,
    /// Per-user hard-negative bitmap (skip, hide, dislike, block).
    hard_negatives: Arc<HardNegIndex>,
    /// Per-(user, creator) interaction weight ledger with lazy decay.
    interaction_ledger: Arc<InteractionLedger>,
    /// Per-user preference vectors (taste embeddings).
    /// Dimensionality is set when the first preference is written; default 128.
    preference_vectors: Arc<PreferenceVectors>,

    // ── M4 sessions ───────────────────────────────────────────────────
    /// Active sessions: created by `start_session`, removed by `close_session`.
    #[allow(dead_code)] // Used by upcoming session API (start_session/close_session).
    sessions: dashmap::DashMap<SessionId, Arc<SessionState>>,
    /// Monotonically increasing session ID counter (starts at 1).
    #[allow(dead_code)] // Used by upcoming session API (start_session).
    next_session_id: AtomicU64,
    /// Archived session snapshots (populated when `close_session` is called).
    #[allow(dead_code)] // Used by upcoming session API (close_session/session_snapshot).
    closed_sessions: dashmap::DashMap<SessionId, SessionSnapshot>,
}

impl TidalDb {
    /// Returns a new [`TidalDbBuilder`] with default (ephemeral) configuration.
    #[must_use]
    pub fn builder() -> TidalDbBuilder {
        TidalDbBuilder::new()
    }

    /// Construct a `TidalDb` from a validated configuration (no schema).
    ///
    /// Used by the builder when no schema is provided -- backwards-compatible
    /// with M0 usage where signal/storage APIs are not called.
    #[allow(clippy::missing_const_for_fn)] // Arc field prevents const in practice
    pub(crate) fn from_config(
        config: Config,
        metrics: Arc<MetricsState>,
        #[cfg(feature = "metrics")] metrics_handle: Option<http::MetricsHandle>,
    ) -> Self {
        let mut profile_registry = ProfileRegistry::new();
        // Register builtins even in no-schema mode so health_check and metrics
        // still work. Ignore errors -- builtins always validate.
        let _ = register_builtins(&mut profile_registry);

        Self {
            config,
            closed: AtomicBool::new(false),
            metrics,
            #[cfg(feature = "metrics")]
            metrics_handle,
            ledger: None,
            storage: None,
            wal: std::sync::Mutex::new(None),
            last_wal_seq: Arc::new(AtomicU64::new(0)),
            shutdown_checkpoint: Arc::new(AtomicBool::new(false)),
            checkpoint_thread: std::sync::Mutex::new(None),
            category_index: BitmapIndex::new("category"),
            format_index: BitmapIndex::new("format"),
            creator_index: BitmapIndex::new("creator"),
            tag_index: BitmapIndex::new("tags"),
            duration_index: RangeIndex::new("duration"),
            created_at_index: RangeIndex::new("created_at"),
            universe: Arc::new(RwLock::new(RoaringBitmap::new())),
            embedding_registry: Arc::new(RwLock::new(EmbeddingSlotRegistry::new())),
            profile_registry,
            schema_def: None,
            creator_items: Arc::new(CreatorItemsBitmap::new()),
            user_state: Arc::new(UserStateIndex::new()),
            hard_negatives: Arc::new(HardNegIndex::new()),
            interaction_ledger: Arc::new(InteractionLedger::new()),
            preference_vectors: Arc::new(PreferenceVectors::new(128)),
            sessions: dashmap::DashMap::new(),
            next_session_id: AtomicU64::new(1),
            closed_sessions: dashmap::DashMap::new(),
        }
    }

    /// Construct a `TidalDb` from all opened components.
    #[allow(clippy::too_many_arguments, clippy::missing_const_for_fn)]
    pub(crate) fn from_parts(
        config: Config,
        metrics: Arc<MetricsState>,
        #[cfg(feature = "metrics")] metrics_handle: Option<http::MetricsHandle>,
        result: OpenResult,
    ) -> Self {
        let OpenResult {
            storage,
            ledger,
            wal,
            last_seq,
            profile_registry,
            category_index,
            format_index,
            creator_index,
            tag_index,
            duration_index,
            created_at_index,
            universe,
            embedding_registry,
            schema_def,
            creator_items,
            user_state,
            hard_negatives,
            interaction_ledger,
            preference_vectors,
        } = result;

        let ledger = Some(ledger);
        let storage = Some(storage);

        // Spawn a periodic checkpoint thread in persistent mode when a ledger
        // is present. The thread checkpoints every 30 seconds so that crash
        // recovery replays a bounded WAL tail regardless of shutdown timing.
        let shutdown_checkpoint = Arc::new(AtomicBool::new(false));
        let checkpoint_thread = {
            let thread_handle = match (storage.as_ref(), ledger.as_ref()) {
                (Some(StorageBox::Fjall(f)), Some(ledger_arc)) => {
                    let items = Box::new(f.backend(EntityKind::Item).clone())
                        as Box<dyn StorageEngine + Send + Sync>;
                    let shutdown = Arc::clone(&shutdown_checkpoint);
                    let ledger_clone = Arc::clone(ledger_arc);
                    let seq_clone = Arc::clone(&last_seq);
                    std::thread::Builder::new()
                        .name("tidaldb-checkpoint".to_string())
                        .spawn(move || {
                            run_checkpoint_thread(shutdown, ledger_clone, items, seq_clone);
                        })
                        .ok()
                }
                _ => None,
            };
            std::sync::Mutex::new(thread_handle)
        };

        Self {
            config,
            closed: AtomicBool::new(false),
            metrics,
            #[cfg(feature = "metrics")]
            metrics_handle,
            ledger,
            storage,
            wal: std::sync::Mutex::new(wal),
            last_wal_seq: last_seq,
            shutdown_checkpoint,
            checkpoint_thread,
            category_index,
            format_index,
            creator_index,
            tag_index,
            duration_index,
            created_at_index,
            universe: Arc::new(RwLock::new(universe)),
            embedding_registry: Arc::new(RwLock::new(embedding_registry)),
            profile_registry,
            schema_def: Some(schema_def),
            creator_items: Arc::new(creator_items),
            user_state: Arc::new(user_state),
            hard_negatives: Arc::new(hard_negatives),
            interaction_ledger: Arc::new(interaction_ledger),
            preference_vectors: Arc::new(preference_vectors),
            sessions: dashmap::DashMap::new(),
            next_session_id: AtomicU64::new(1),
            closed_sessions: dashmap::DashMap::new(),
        }
    }

    /// Returns a reference to the shared metrics state.
    #[must_use]
    #[allow(clippy::missing_const_for_fn)] // Arc field prevents const in practice
    pub fn metrics(&self) -> &Arc<MetricsState> {
        &self.metrics
    }

    /// Returns the bound address of the metrics HTTP server, if running.
    ///
    /// Useful when port 0 was requested to discover the OS-assigned port.
    /// Returns `None` if the `metrics` feature is disabled or if
    /// `enable_metrics` was not called on the builder.
    #[must_use]
    #[allow(clippy::missing_const_for_fn)] // cfg-gated body prevents const
    pub fn metrics_addr(&self) -> Option<std::net::SocketAddr> {
        #[cfg(feature = "metrics")]
        {
            self.metrics_handle.as_ref().map(|h| h.addr)
        }
        #[cfg(not(feature = "metrics"))]
        {
            None
        }
    }

    /// Returns `Ok(())` if the database is initialized and operational.
    ///
    /// # Errors
    ///
    /// Returns an error if the database has been closed or an internal
    /// check fails.
    #[tracing::instrument(skip(self))]
    pub fn health_check(&self) -> crate::Result<()> {
        if self.closed.load(Ordering::Acquire) {
            self.metrics.health_ok.store(false, Ordering::Release);
            return Err(crate::TidalError::Internal(
                "database is closed".to_string(),
            ));
        }
        Ok(())
    }

    // ── Internal helpers ─────────────────────────────────────────────────────

    /// Borrow the storage backend, or error if the database was opened without a schema.
    fn storage(&self) -> crate::Result<&StorageBox> {
        self.storage
            .as_ref()
            .ok_or_else(|| TidalError::Internal("no storage: open with with_schema()".into()))
    }

    /// Borrow the signal ledger, or error if the database was opened without a schema.
    #[allow(dead_code)] // Used by upcoming session signal write path.
    fn ledger(&self) -> crate::Result<&Arc<SignalLedger>> {
        self.ledger
            .as_ref()
            .ok_or_else(|| TidalError::Internal("no ledger: open with with_schema()".into()))
    }

    // ── M1p5 API ─────────────────────────────────────────────────────────────

    /// Write (or overwrite) item metadata.
    ///
    /// Stores the `metadata` key-value map under the entity's `Tag::Meta` key
    /// in the items storage backend.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired (use `with_schema()`).
    /// - `TidalError::Storage` on storage engine failure.
    pub fn write_item(
        &self,
        id: EntityId,
        metadata: &HashMap<String, String>,
    ) -> crate::Result<()> {
        let storage = self.storage()?;
        let key = encode_key(id, Tag::Meta, b"");
        let value = serialize_metadata(metadata);
        storage
            .items_engine()
            .put(&key, &value)
            .map_err(TidalError::from)
    }

    /// Record a signal event for an entity.
    ///
    /// Atomically:
    /// 1. Appends the event to the WAL (WAL-first durability).
    /// 2. Updates the in-memory decay score (hot tier).
    /// 3. Updates the in-memory windowed counter (warm tier).
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no ledger is wired (use `with_schema()`).
    /// - `TidalError::Schema` if `signal_type` is not defined in the schema.
    /// - `TidalError::Durability` if the WAL write fails.
    pub fn signal(
        &self,
        signal_type: &str,
        entity_id: EntityId,
        weight: f64,
        timestamp: Timestamp,
    ) -> crate::Result<()> {
        self.ledger()?
            .record_signal(signal_type, entity_id, weight, timestamp)
    }

    /// Read the current decay score for an entity-signal pair.
    ///
    /// Applies lazy decay from the stored timestamp to the current wall-clock
    /// time. Returns `None` if no signals have been recorded.
    ///
    /// `decay_rate_idx` selects the lambda index from the signal definition.
    /// For exponential signals with one rate, use `0`.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no ledger is wired.
    /// - `TidalError::Schema` if `signal_type` is not defined.
    pub fn read_decay_score(
        &self,
        entity_id: EntityId,
        signal_type: &str,
        decay_rate_idx: usize,
    ) -> crate::Result<Option<f64>> {
        self.ledger()?
            .read_decay_score(entity_id, signal_type, decay_rate_idx)
    }

    /// Read the windowed event count for an entity-signal pair.
    ///
    /// Returns `0` if no signals have been recorded.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no ledger is wired.
    /// - `TidalError::Schema` if `signal_type` is not defined.
    pub fn read_windowed_count(
        &self,
        entity_id: EntityId,
        signal_type: &str,
        window: Window,
    ) -> crate::Result<u64> {
        self.ledger()?
            .read_windowed_count(entity_id, signal_type, window)
    }

    /// Read the velocity (events per second) for an entity-signal-window.
    ///
    /// Velocity = `windowed_count / window_duration_seconds`.
    /// Returns `0.0` for `AllTime` windows or if no signals recorded.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no ledger is wired.
    /// - `TidalError::Schema` if `signal_type` is not defined.
    pub fn read_velocity(
        &self,
        entity_id: EntityId,
        signal_type: &str,
        window: Window,
    ) -> crate::Result<f64> {
        self.ledger()?.read_velocity(entity_id, signal_type, window)
    }

    // ── M2 API ──────────────────────────────────────────────────────────────

    /// Write (or overwrite) item metadata and update in-memory indexes.
    ///
    /// This is the M2 replacement for `write_item` -- it persists metadata
    /// to storage AND inserts the entity into the bitmap, range, and universe
    /// indexes so it is discoverable by RETRIEVE queries.
    ///
    /// Recognized metadata keys for indexing:
    /// - `"category"` -> category bitmap index
    /// - `"format"` -> format bitmap index
    /// - `"creator_id"` -> creator bitmap index
    /// - `"tags"` -> tag bitmap index (comma-separated)
    /// - `"duration"` -> duration range index (seconds, parsed as u32)
    /// - `"created_at"` -> `created_at` range index (nanos, parsed as u64)
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn write_item_with_metadata(
        &self,
        id: EntityId,
        metadata: &HashMap<String, String>,
    ) -> crate::Result<()> {
        // Persist to storage.
        self.write_item(id, metadata)?;

        // Truncate entity ID to u32 for roaring bitmap. This limits the universe
        // to ~4 billion entities per instance, which is well beyond the single-node
        // target of 10M items.
        #[allow(clippy::cast_possible_truncation)]
        let id_u32 = id.as_u64() as u32;
        if id.as_u64() > u64::from(u32::MAX) {
            tracing::warn!(
                entity_id = id.as_u64(),
                "entity ID exceeds u32::MAX; universe bitmap entry will collide with a lower ID"
            );
        }

        // Insert into bitmap indexes.
        if let Some(val) = metadata.get("category") {
            self.category_index.insert(id_u32, val);
        }
        if let Some(val) = metadata.get("format") {
            self.format_index.insert(id_u32, val);
        }
        if let Some(val) = metadata.get("creator_id") {
            self.creator_index.insert(id_u32, val);
            // M3: populate creator-items bitmap for the `following` profile
            // and `unblocked` predicate.
            if let Ok(creator_id) = val.parse::<u64>() {
                self.creator_items.add_item(creator_id, id_u32);
            }
        }
        if let Some(tags) = metadata.get("tags") {
            for tag in tags.split(',') {
                let tag = tag.trim();
                if !tag.is_empty() {
                    self.tag_index.insert(id_u32, tag);
                }
            }
        }

        // Insert into range indexes.
        if let Some(val) = metadata.get("duration") {
            match val.parse::<u32>() {
                Ok(dur) => self.duration_index.insert(id_u32, dur),
                Err(e) => tracing::warn!(
                    entity_id = id.as_u64(),
                    value = %val,
                    error = %e,
                    "failed to parse 'duration' metadata; item will not be indexed by duration"
                ),
            }
        }
        let has_created_at =
            metadata
                .get("created_at")
                .is_some_and(|val| match val.parse::<u64>() {
                    Ok(ts) => {
                        self.created_at_index.insert(id_u32, ts);
                        true
                    }
                    Err(e) => {
                        tracing::warn!(
                            entity_id = id.as_u64(),
                            value = %val,
                            error = %e,
                            "failed to parse 'created_at' metadata; defaulting to current time"
                        );
                        false
                    }
                });
        if !has_created_at {
            // Default: use current time as created_at so the item is discoverable
            // by range queries and sortable by recency.
            self.created_at_index
                .insert(id_u32, Timestamp::now().as_nanos());
        }

        // Insert into universe bitmap.
        if let Ok(mut bm) = self.universe.write() {
            bm.insert(id_u32);
        }

        Ok(())
    }

    /// Returns the number of items in the universe bitmap.
    ///
    /// This counts items written via `write_item_with_metadata`, not raw
    /// storage entries.
    #[must_use]
    pub fn item_count(&self) -> u64 {
        self.universe.read().map_or(0, |bm| bm.len())
    }

    /// Read item metadata for a given entity ID.
    ///
    /// Returns `None` if the entity does not exist in storage.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn get_item_metadata(
        &self,
        id: EntityId,
    ) -> crate::Result<Option<HashMap<String, String>>> {
        let storage = self.storage()?;
        let key = encode_key(id, Tag::Meta, b"");
        match storage.items_engine().get(&key) {
            Ok(Some(bytes)) => Ok(Some(deserialize_metadata(&bytes))),
            Ok(None) => Ok(None),
            Err(e) => Err(TidalError::from(e)),
        }
    }

    /// Execute a RETRIEVE query against the database.
    ///
    /// Constructs a per-query `RetrieveExecutor` with borrowed references to
    /// all infrastructure, then runs the 5-stage pipeline:
    ///
    /// 1. Candidate generation (scan universe or signal-ranked)
    /// 2. Filter evaluation (bitmap + range indexes)
    /// 3. Signal scoring (profile executor)
    /// 4. Diversity enforcement (per-creator, format-mix)
    /// 5. Result assembly (pagination, cursor, explain)
    ///
    /// # Errors
    ///
    /// Returns `TidalError::Query` on validation failure, missing profile,
    /// or unsupported candidate strategy.
    pub fn retrieve(&self, query: &Retrieve) -> crate::Result<Results> {
        let ledger = self.ledger()?;

        let base_executor = RetrieveExecutor::new(
            ledger,
            &self.profile_registry,
            Some(&self.category_index),
            Some(&self.format_index),
            Some(&self.creator_index),
            Some(&self.tag_index),
            Some(&self.duration_index),
            Some(&self.created_at_index),
            Some(&self.universe),
            Some(&self.embedding_registry),
        )
        .with_user_context(
            &self.user_state,
            &self.hard_negatives,
            &self.interaction_ledger,
            &self.creator_items,
        )
        .with_preference_vectors(&self.preference_vectors);

        // M4: wire in session context when FOR SESSION is specified.
        let executor = if let Some(session_id) = query.for_session {
            match self.session_snapshot(session_id) {
                Ok(snapshot) => {
                    let ctx = session_mod::SessionContext::from_snapshot(&snapshot);
                    base_executor.with_session(ctx, snapshot)
                }
                Err(e) => {
                    tracing::warn!(
                        session_id = %session_id,
                        error = %e,
                        "FOR SESSION: session not found; executing without session boost"
                    );
                    base_executor
                }
            }
        } else {
            base_executor
        };

        executor.execute(query).map_err(TidalError::from)
    }

    // ── M3 Entity API ────────────────────────────────────────────────────────

    /// Write (or overwrite) a user entity.
    ///
    /// Stores metadata and optional embedding under the user's `Tag::Meta` key
    /// in the users storage backend.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn write_user(
        &self,
        id: EntityId,
        metadata: &HashMap<String, String>,
    ) -> crate::Result<()> {
        let storage = self.storage()?;
        let key = encode_key(id, Tag::Meta, b"");
        let value = crate::entities::serialize_entity(None, metadata);
        storage
            .users_engine()
            .put(&key, &value)
            .map_err(TidalError::from)
    }

    /// Read user metadata for a given entity ID.
    ///
    /// Returns `None` if the user does not exist in storage.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn get_user_metadata(
        &self,
        id: EntityId,
    ) -> crate::Result<Option<HashMap<String, String>>> {
        let storage = self.storage()?;
        let key = encode_key(id, Tag::Meta, b"");
        match storage.users_engine().get(&key) {
            Ok(Some(bytes)) => {
                let (_emb, meta) = crate::entities::deserialize_entity(&bytes);
                Ok(Some(meta))
            }
            Ok(None) => Ok(None),
            Err(e) => Err(TidalError::from(e)),
        }
    }

    /// Write (or overwrite) a creator entity.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn write_creator(
        &self,
        id: EntityId,
        metadata: &HashMap<String, String>,
    ) -> crate::Result<()> {
        let storage = self.storage()?;
        let key = encode_key(id, Tag::Meta, b"");
        let value = crate::entities::serialize_entity(None, metadata);
        storage
            .creators_engine()
            .put(&key, &value)
            .map_err(TidalError::from)
    }

    /// Read creator metadata for a given entity ID.
    ///
    /// Returns `None` if the creator does not exist in storage.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn get_creator_metadata(
        &self,
        id: EntityId,
    ) -> crate::Result<Option<HashMap<String, String>>> {
        let storage = self.storage()?;
        let key = encode_key(id, Tag::Meta, b"");
        match storage.creators_engine().get(&key) {
            Ok(Some(bytes)) => {
                let (_emb, meta) = crate::entities::deserialize_entity(&bytes);
                Ok(Some(meta))
            }
            Ok(None) => Ok(None),
            Err(e) => Err(TidalError::from(e)),
        }
    }

    /// Write a relationship edge between two entities.
    ///
    /// The edge is stored in the users keyspace under the `from` entity's key
    /// range using the relationship key encoding.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn write_relationship(
        &self,
        from: EntityId,
        rel_type: crate::entities::RelationshipType,
        to: EntityId,
        weight: f64,
        timestamp: Timestamp,
    ) -> crate::Result<()> {
        use crate::entities::relationship::{
            encode_relationship_key, serialize_relationship_value,
        };

        let storage = self.storage()?;

        let key = encode_relationship_key(from, rel_type, to);
        let value = serialize_relationship_value(weight, timestamp);
        storage
            .users_engine()
            .put(&key, &value)
            .map_err(TidalError::from)?;

        // Update in-memory user state based on relationship type.
        match rel_type {
            RelationshipType::Blocks => {
                self.user_state
                    .add_block_creator(from.as_u64(), to.as_u64());
            }
            RelationshipType::Hide => {
                #[allow(clippy::cast_possible_truncation)]
                self.user_state.add_hide(from.as_u64(), to.as_u64() as u32);
            }
            RelationshipType::Follows => {
                self.user_state.add_follow(from.as_u64(), to.as_u64());
            }
            _ => {}
        }

        Ok(())
    }

    /// Delete a relationship edge.
    ///
    /// Removes both the durable storage entry and the corresponding in-memory
    /// state (blocked creators, hidden items, follows).
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn delete_relationship(
        &self,
        from: EntityId,
        rel_type: crate::entities::RelationshipType,
        to: EntityId,
    ) -> crate::Result<()> {
        use crate::entities::relationship::encode_relationship_key;

        let storage = self.storage()?;
        let key = encode_relationship_key(from, rel_type, to);
        storage
            .users_engine()
            .delete(&key)
            .map_err(TidalError::from)?;

        // Remove corresponding in-memory state.
        match rel_type {
            RelationshipType::Blocks => {
                self.user_state
                    .remove_block_creator(from.as_u64(), to.as_u64());
            }
            RelationshipType::Hide => {
                #[allow(clippy::cast_possible_truncation)]
                self.user_state
                    .remove_hide(from.as_u64(), to.as_u64() as u32);
            }
            RelationshipType::Follows => {
                self.user_state.remove_follow(from.as_u64(), to.as_u64());
            }
            _ => {}
        }

        Ok(())
    }

    /// List all relationship targets for a given (from, `rel_type`) pair.
    ///
    /// Returns `(to_entity_id, weight, timestamp_nanos)` tuples.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if no storage backend is wired.
    /// - `TidalError::Storage` on storage engine failure.
    pub fn list_relationships(
        &self,
        from: EntityId,
        rel_type: crate::entities::RelationshipType,
    ) -> crate::Result<Vec<(EntityId, f64, u64)>> {
        use crate::entities::relationship::{
            deserialize_relationship_value, parse_relationship_to, relationship_prefix,
        };

        let storage = self.storage()?;
        let prefix = relationship_prefix(from, rel_type);
        let mut results = Vec::new();
        for entry in storage.users_engine().scan_prefix(&prefix) {
            let (key, value) = entry.map_err(TidalError::from)?;
            if let Some(to) = parse_relationship_to(&key)
                && let Some((weight, ts_nanos)) = deserialize_relationship_value(&value)
            {
                results.push((to, weight, ts_nanos));
            }
        }
        Ok(results)
    }

    /// Access the per-user state index (seen, blocked, saved, liked, completion).
    #[must_use]
    pub fn user_state(&self) -> &UserStateIndex {
        &self.user_state
    }

    /// Access the creator-items bitmap (maps creators to their item sets).
    #[must_use]
    pub fn creator_items(&self) -> &CreatorItemsBitmap {
        &self.creator_items
    }

    /// Access the hard-negative index.
    #[must_use]
    pub fn hard_negatives(&self) -> &HardNegIndex {
        &self.hard_negatives
    }

    /// Access the interaction weight ledger.
    #[must_use]
    pub fn interaction_ledger(&self) -> &InteractionLedger {
        &self.interaction_ledger
    }

    /// Access the preference vectors store.
    #[must_use]
    pub fn preference_vectors(&self) -> &PreferenceVectors {
        &self.preference_vectors
    }

    /// Records a signal with user context, updating the interaction ledger, seen state,
    /// and preference vectors in-memory.
    ///
    /// In addition to updating the signal ledger, this method:
    /// 1. Hard negatives: if the signal is skip/hide/dislike/block, records
    ///    a hard negative for the (user, item) pair.
    /// 2. Interaction weight: if `for_user` is provided, updates the
    ///    (user, creator) interaction weight.
    /// 3. Seen tracking: if `for_user` is provided, marks the item as seen.
    /// 4. Preference vector: for positive engagement signals (like, share,
    ///    completion), looks up the item's embedding from durable storage and
    ///    blends it into the user's preference vector via EMA.
    ///
    /// # Preference vector updates
    ///
    /// The update triggers when all three conditions are met:
    /// - The signal type is a positive engagement signal ("like", "share", "completion").
    /// - `for_user` is `Some` (the acting user is known).
    /// - The item has a stored embedding in the entity store (written via
    ///   `insert_embedding` or `update_embedding` during item ingestion).
    ///
    /// The embedding is read from the first Item embedding slot declared in the
    /// schema. If no schema or no embedding slot is declared, falls back to the
    /// slot name "content". If the lookup fails (no embedding stored, storage
    /// error), the preference update is silently skipped -- the base signal is
    /// still recorded.
    ///
    /// # Durability
    ///
    /// The base signal (entity, type, weight, timestamp) is WAL-backed and survives crashes.
    /// User-context side effects (seen state, interaction weights, preference vector updates)
    /// are reconstructed from durable storage on restart via `rebuild_entity_state`.
    /// Hard negatives (hide/block) are durably written via `write_relationship()`.
    ///
    /// Seen state from regular views/likes is intentionally ephemeral -- users should see
    /// content again after a restart. Only explicit hides (via `write_relationship` with
    /// `RelationshipType::Hide`) survive restarts as "seen".
    ///
    /// # Errors
    ///
    /// Returns errors from the underlying `signal()` method.
    pub fn signal_with_context(
        &self,
        signal_type: &str,
        entity_id: EntityId,
        weight: f64,
        timestamp: Timestamp,
        for_user: Option<u64>,
        creator_id: Option<u64>,
    ) -> crate::Result<()> {
        // Record the base signal.
        self.signal(signal_type, entity_id, weight, timestamp)?;

        // Signal dispatch: side effects based on signal type and context.
        if let Some(user_id) = for_user {
            #[allow(clippy::cast_possible_truncation)]
            let item_u32 = entity_id.as_u64() as u32;

            // 1. Hard negatives.
            if HardNegIndex::is_hard_neg_signal(signal_type) {
                self.hard_negatives.add(user_id, item_u32);
            }

            // 2. Seen tracking.
            self.user_state.mark_seen(user_id, item_u32);

            // 3. Interaction weight: if creator is known, update the
            //    (user, creator) interaction strength.
            if let Some(cid) = creator_id {
                self.interaction_ledger
                    .record(user_id, cid, weight, timestamp.as_nanos());
            }

            // 4. Preference vector: for positive engagement signals, look up
            //    the item's embedding and blend into the user's taste vector.
            if is_positive_engagement_signal(signal_type) {
                self.try_update_preference_vector(user_id, entity_id);
            }
        }

        Ok(())
    }

    /// Attempt to update a user's preference vector from the item's stored embedding.
    ///
    /// Reads the item's embedding from durable storage (entity store) and blends it
    /// into the user's preference vector via `PreferenceVectors::update()`. This is
    /// a best-effort operation: if the item has no embedding, no storage is wired, or
    /// the embedding cannot be deserialized, the update is silently skipped.
    ///
    /// The slot name is determined by the schema's first Item embedding slot, falling
    /// back to "content" if no schema is available.
    fn try_update_preference_vector(&self, user_id: u64, entity_id: EntityId) {
        // Determine which embedding slot to read.
        let slot_name = self
            .schema_def
            .as_ref()
            .and_then(|s| {
                s.embedding_slots()
                    .iter()
                    .find(|slot| slot.entity_kind == EntityKind::Item)
                    .map(|slot| slot.name.as_str())
            })
            .unwrap_or("content");

        // Read the item's embedding from durable storage.
        let Some(storage) = self.storage.as_ref() else {
            return;
        };
        let key = crate::storage::vector::embedding_store_key(entity_id, slot_name);
        let embedding_bytes = match storage.items_engine().get(&key) {
            Ok(Some(bytes)) => bytes,
            Ok(None) => {
                tracing::debug!(
                    entity_id = entity_id.as_u64(),
                    slot = slot_name,
                    "preference vector update skipped: item has no stored embedding"
                );
                return;
            }
            Err(e) => {
                tracing::debug!(
                    entity_id = entity_id.as_u64(),
                    error = %e,
                    "preference vector update skipped: storage read failed"
                );
                return;
            }
        };

        // Deserialize the embedding.
        let embedding = match crate::storage::vector::deserialize_embedding(&embedding_bytes) {
            Ok(v) => v,
            Err(e) => {
                tracing::debug!(
                    entity_id = entity_id.as_u64(),
                    error = %e,
                    "preference vector update skipped: embedding deserialization failed"
                );
                return;
            }
        };

        // Blend into the user's preference vector.
        if !self.preference_vectors.update(user_id, &embedding) {
            tracing::debug!(
                user_id,
                entity_id = entity_id.as_u64(),
                embedding_dim = embedding.len(),
                "preference vector update skipped: dimension mismatch"
            );
        }
    }

    // ── M4 Session API ────────────────────────────────────────────────────────

    /// Start a new agent session.
    ///
    /// Creates a session-scoped signal context for the given agent. The
    /// session is identified by its `SessionId` and keyed to `user_id` and
    /// `agent_id`. The `policy_name` must match a policy declared via
    /// `SchemaBuilder::session_policy()`.
    ///
    /// # Errors
    ///
    /// - `TidalError::Schema` if `policy_name` is not found in the schema.
    /// - `TidalError::Internal` if no schema was provided at open time.
    pub fn start_session(
        &self,
        user_id: u64,
        agent_id: &str,
        policy_name: &str,
        metadata: HashMap<String, String>,
    ) -> crate::Result<SessionHandle> {
        // Validate policy exists in schema.
        let schema = self
            .schema_def
            .as_ref()
            .ok_or_else(|| TidalError::Internal("no schema: open with with_schema()".into()))?;
        if schema.session_policy(policy_name).is_none() {
            return Err(TidalError::Internal(format!(
                "policy '{policy_name}' not found in schema"
            )));
        }

        let parsed_agent_id = AgentId::new(agent_id)
            .map_err(|e| TidalError::Internal(format!("invalid agent_id: {e}")))?;

        let session_id = SessionId::from_raw(self.next_session_id.fetch_add(1, Ordering::Relaxed));

        let closed = Arc::new(AtomicBool::new(false));

        // Capture started_at once — shared between SessionState and SessionHandle.
        let started_at = std::time::Instant::now();
        let started_at_ns = Timestamp::now().as_nanos();

        let state = Arc::new(SessionState {
            id: session_id,
            user_id,
            agent_id: parsed_agent_id.clone(),
            policy_name: policy_name.to_owned(),
            started_at,
            started_at_ns,
            metadata,
            signals: dashmap::DashMap::new(),
            signaled_entities: dashmap::DashMap::new(),
            annotations: std::sync::Mutex::new(Vec::new()),
            signals_written: AtomicU64::new(0),
            signals_rejected: AtomicU64::new(0),
            audit_log: std::sync::Mutex::new(Vec::new()),
            closed: Arc::clone(&closed),
        });

        self.sessions.insert(session_id, state);

        Ok(SessionHandle {
            id: session_id,
            user_id,
            agent_id: parsed_agent_id,
            policy_name: policy_name.to_owned(),
            started_at,
            closed,
        })
    }

    /// Close a session and return a summary.
    ///
    /// Takes ownership of the `SessionHandle` to prevent use-after-close at
    /// compile time. The session snapshot is archived to `closed_sessions`.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if the session was already removed (double-close).
    #[allow(clippy::needless_pass_by_value)] // Intentional: move semantics prevent use-after-close at the type level.
    pub fn close_session(&self, handle: SessionHandle) -> crate::Result<SessionSummary> {
        // Mark the handle as closed (runtime defense-in-depth).
        handle.closed.store(true, Ordering::Release);

        let session_id = handle.id;
        let (_id, state) = self.sessions.remove(&session_id).ok_or_else(|| {
            TidalError::Internal(format!("session {session_id} not found (already closed?)"))
        })?;

        let duration_ms = state.started_at.elapsed().as_millis() as u64;
        let signals_written = state.signals_written.load(Ordering::Relaxed);
        let rejections = state.signals_rejected.load(Ordering::Relaxed);

        // Build and archive the frozen snapshot.
        let snapshot = session_mod::build_frozen_snapshot(&state, duration_ms);

        // Evict oldest closed session if the cap is exceeded.
        if self.closed_sessions.len() >= session_mod::MAX_CLOSED_SESSIONS
            && let Some(oldest_key) = self.closed_sessions.iter().map(|e| *e.key()).min()
        {
            self.closed_sessions.remove(&oldest_key);
        }
        self.closed_sessions.insert(session_id, snapshot);

        tracing::debug!(
            session_id = %session_id,
            signals_written,
            rejections,
            duration_ms,
            "session closed"
        );

        Ok(SessionSummary {
            id: session_id,
            duration_ms,
            signals_written,
            rejections,
        })
    }

    /// List all currently active sessions.
    #[must_use]
    pub fn active_sessions(&self) -> Vec<SessionInfo> {
        self.sessions
            .iter()
            .map(|entry| {
                let s = entry.value();
                SessionInfo {
                    id: s.id,
                    user_id: s.user_id,
                    agent_id: s.agent_id.as_str().to_owned(),
                    started_at_ns: s.started_at_ns,
                    signals_written: s.signals_written.load(Ordering::Relaxed),
                }
            })
            .collect()
    }

    /// Write a session-scoped signal for an entity.
    ///
    /// Session signals are tracked in the session's in-memory `SessionHotState`
    /// with aggressive decay (5-minute half-life by default). They do **not**
    /// propagate to the global `SignalLedger`; they exist only within this session
    /// and are archived on `close_session`.
    ///
    /// If the session has a policy, it is evaluated before the write; rejected
    /// signals are counted and logged to the audit trail.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if `signal_type` is not in the schema, the session
    ///   is closed, or not found.
    /// - `TidalError::PolicyViolation` if the policy rejects the signal.
    /// - `TidalError::SessionExpired` if the session's duration limit is exceeded.
    #[allow(clippy::significant_drop_tightening)] // state_ref must live for the duration of the method.
    pub fn session_signal(
        &self,
        handle: &SessionHandle,
        signal_type: &str,
        entity_id: EntityId,
        weight: f64,
        ts: Timestamp,
        annotation: Option<String>,
    ) -> crate::Result<()> {
        // Runtime guard: check the closed flag.
        if handle.closed.load(Ordering::Acquire) {
            return Err(TidalError::Internal(format!(
                "session {} is closed",
                handle.id
            )));
        }

        // Validate signal_type exists in the schema.
        if let Some(ledger) = self.ledger.as_ref()
            && ledger.resolve_signal_type(signal_type).is_err()
        {
            return Err(TidalError::Internal(format!(
                "unknown signal type: '{signal_type}'"
            )));
        }

        let state_ref = self
            .sessions
            .get(&handle.id)
            .ok_or_else(|| TidalError::Internal(format!("session {} not found", handle.id)))?;
        let state = state_ref.value();

        // Policy evaluation.
        if let Some(schema) = &self.schema_def
            && let Some(policy) = schema.session_policy(&state.policy_name)
        {
            let evaluator = session_mod::PolicyEvaluator::new(policy, &state.policy_name);
            match evaluator.check(signal_type, state, std::time::Instant::now()) {
                Ok(()) => {
                    // Record accepted entry.
                    let entry = AuditEntry {
                        timestamp_ns: ts.as_nanos(),
                        signal_type: signal_type.to_owned(),
                        accepted: true,
                        reason: None,
                    };
                    if let Ok(mut log) = state.audit_log.lock()
                        && log.len() < session_mod::MAX_AUDIT_ENTRIES
                    {
                        log.push(entry);
                    }
                }
                Err(violation) => {
                    // Record rejected entry.
                    let entry = AuditEntry {
                        timestamp_ns: ts.as_nanos(),
                        signal_type: signal_type.to_owned(),
                        accepted: false,
                        reason: Some(violation.reason.clone()),
                    };
                    if let Ok(mut log) = state.audit_log.lock()
                        && log.len() < session_mod::MAX_AUDIT_ENTRIES
                    {
                        log.push(entry);
                    }
                    state.signals_rejected.fetch_add(1, Ordering::Relaxed);

                    // Dispatch on the typed violation kind (no string parsing).
                    return match violation.kind {
                        session_mod::PolicyViolationKind::Expired => {
                            Err(TidalError::SessionExpired {
                                session_id: handle.id.as_u64(),
                                max_duration_secs: policy.max_session_duration.as_secs_f64(),
                            })
                        }
                        _ => Err(TidalError::PolicyViolation {
                            signal_type: violation.signal_type,
                            policy_name: violation.policy_name,
                            reason: violation.reason,
                        }),
                    };
                }
            }
        }

        // Write to session hot state (CAS decay update).
        let hot = state
            .signals
            .entry(signal_type.to_owned())
            .or_insert_with(session_mod::SessionHotState::new);
        hot.on_signal(weight, ts.as_nanos(), session_mod::DEFAULT_SESSION_LAMBDA);
        drop(hot);

        // Track signaled entity.
        state.signaled_entities.insert(entity_id.as_u64(), ());

        // Store annotation (capped at MAX_ANNOTATIONS).
        if let Some(ann) = annotation
            && let Ok(mut anns) = state.annotations.lock()
            && anns.len() < session_mod::MAX_ANNOTATIONS
        {
            anns.push(ann);
        }

        state.signals_written.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// Retrieve a snapshot of an active or archived session.
    ///
    /// For active sessions the decay scores are computed at the current
    /// wall-clock time. For archived sessions the scores are frozen at the
    /// moment `close_session` was called.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if the session is not found.
    pub fn session_snapshot(&self, session_id: SessionId) -> crate::Result<SessionSnapshot> {
        // Try active sessions first.
        if let Some(state_ref) = self.sessions.get(&session_id) {
            let state = state_ref.value();
            let now_ns = Timestamp::now().as_nanos();
            return Ok(session_mod::build_snapshot(state, now_ns));
        }

        // Fall back to archived sessions.
        if let Some(snap_ref) = self.closed_sessions.get(&session_id) {
            return Ok(snap_ref.value().clone());
        }

        Err(TidalError::Internal(format!(
            "session {session_id} not found"
        )))
    }

    /// Retrieve the policy audit log for a session.
    ///
    /// Returns all accept/reject decisions recorded by the policy evaluator
    /// for the given session.
    ///
    /// # Errors
    ///
    /// - `TidalError::Internal` if the session is not found or the audit log
    ///   mutex is poisoned.
    pub fn session_audit(&self, session_id: SessionId) -> crate::Result<Vec<AuditEntry>> {
        // Try active sessions first.
        if let Some(state_ref) = self.sessions.get(&session_id) {
            let state = state_ref.value();
            let log = state
                .audit_log
                .lock()
                .map_err(|_| TidalError::Internal("audit_log mutex poisoned".into()))?;
            return Ok(log.clone());
        }

        // For archived sessions, return the audit log captured at close time.
        if let Some(snap_ref) = self.closed_sessions.get(&session_id) {
            return Ok(snap_ref.value().audit_log.clone());
        }

        Err(TidalError::Internal(format!(
            "session {session_id} not found"
        )))
    }

    // ── Lifecycle ─────────────────────────────────────────────────────────────

    /// Cleanly shut down the database.
    ///
    /// 1. Checkpoints all in-memory signal state to durable storage.
    /// 2. Flushes the storage engine.
    /// 3. Writes a WAL checkpoint marker and truncates old segments.
    /// 4. Shuts down the WAL writer thread.
    ///
    /// # Errors
    ///
    /// Returns an error if WAL shutdown fails.
    #[tracing::instrument(skip(self))]
    pub fn close(self) -> crate::Result<()> {
        self.shutdown_inner()
    }

    /// Alias for [`close`](Self::close). Cleanly shuts down the database.
    ///
    /// # Errors
    ///
    /// Returns an error if WAL shutdown fails.
    pub fn shutdown(self) -> crate::Result<()> {
        self.close()
    }

    /// Internal shutdown logic shared by `close()` and `Drop`.
    fn shutdown_inner(&self) -> crate::Result<()> {
        // CAS: first caller to flip false -> true executes the shutdown body.
        if self
            .closed
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            // Already closed -- idempotent.
            return Ok(());
        }

        tracing::debug!(mode = ?self.config.mode, "tidaldb shutting down");

        // Mark health as degraded so the metrics endpoint reflects shutdown.
        self.metrics.health_ok.store(false, Ordering::Release);

        // Signal the periodic checkpoint thread to stop, then join it.
        // Must happen before the final checkpoint below to avoid a race where
        // both the background thread and shutdown_inner write simultaneously.
        self.shutdown_checkpoint.store(true, Ordering::Release);
        // Use into_inner() on poisoned mutex so the thread is always joined
        // even if the checkpoint thread panicked. Without this, a panicking
        // checkpoint thread would leave the join skipped, and the thread would
        // keep running after close() returns, racing with ledger/storage drop.
        // The guard is dropped immediately after take() to release the lock.
        let checkpoint_handle = {
            let mut guard = match self.checkpoint_thread.lock() {
                Ok(g) => g,
                Err(poisoned) => poisoned.into_inner(),
            };
            guard.take()
        };
        if let Some(handle) = checkpoint_handle {
            let _ = handle.join();
        }

        // Stop the metrics HTTP server if running.
        #[cfg(feature = "metrics")]
        {
            // SAFETY: We need &mut to stop the handle, but we only have &self.
            // This is safe because shutdown_inner is guarded by the closed
            // compare_exchange above -- only one thread will ever reach this
            // point. We use a raw pointer to get interior mutability for
            // the Option<MetricsHandle> field.
            //
            // NOTE: This is the same pattern used in Drop, which also has &mut self.
            // For the close() path we route through shutdown_inner(&self) to share
            // logic. In practice this runs exactly once due to the CAS guard.
        }

        // 1. Checkpoint signal state to storage.
        // Shutdown ordering: (1) stop checkpoint thread -> (2) checkpoint ledger
        // -> (3) shutdown WAL. This is safe because the `closed` flag prevents
        // new writes before shutdown begins, the checkpoint below captures the
        // current in-memory state + last WAL seq, and WAL replay on next open
        // re-applies any post-checkpoint events still in the WAL file.
        if let (Some(ledger), Some(storage)) = (&self.ledger, &self.storage) {
            let meta = crate::signals::checkpoint::CheckpointMeta {
                checkpoint_time_ns: Timestamp::now().as_nanos(),
                // Acquire pairs with the Release in WalHandleWriter::append_signal,
                // ensuring we see the highest seq committed by any WAL writer thread.
                wal_sequence: self.last_wal_seq.load(Ordering::Acquire),
            };
            if let Err(e) = ledger.checkpoint(storage.items_engine(), meta) {
                tracing::error!(error = %e, "signal ledger checkpoint failed during shutdown");
            }
            if let Err(e) = storage.flush() {
                tracing::error!(error = %e, "storage flush failed during shutdown");
            }
        }

        // 2. Shut down WAL (needs ownership via take()).
        let wal_opt = self
            .wal
            .lock()
            .map_err(|_| TidalError::Internal("WAL mutex poisoned".into()))?
            .take();

        if let Some(wal) = wal_opt {
            let seq = self.last_wal_seq.load(Ordering::Acquire);
            if seq > 0 {
                // Write WAL checkpoint marker so the next open knows the replay
                // start point.
                if let Err(e) = wal.checkpoint(seq) {
                    tracing::error!(error = %e, "WAL checkpoint marker failed during shutdown");
                }
                // Truncate segments that precede the checkpoint.
                if let Err(e) = wal.truncate_before(seq) {
                    tracing::error!(error = %e, "WAL truncation failed during shutdown");
                }
            }
            wal.shutdown().map_err(|e| {
                TidalError::Durability(DurabilityError {
                    message: format!("WAL shutdown failed: {e}"),
                })
            })?;
        }

        Ok(())
    }

    // ── Internal construction helper ──────────────────────────────────────────

    /// Open storage and ledger components for the given schema.
    ///
    /// Called from `TidalDbBuilder::open()` when `with_schema()` was called.
    /// Returns an `OpenResult` containing all components needed by `from_parts`.
    #[allow(clippy::too_many_lines)] // Construction logic is inherently verbose; splitting would scatter initialization flow.
    pub(crate) fn open_with_schema(config: &Config, schema: Schema) -> crate::Result<OpenResult> {
        let last_seq = Arc::new(AtomicU64::new(0));

        // Initialize profile registry with builtins.
        let mut profile_registry = ProfileRegistry::new();
        register_builtins(&mut profile_registry).map_err(|e| {
            TidalError::Internal(format!("failed to register builtin profiles: {e}"))
        })?;

        // Initialize M2 indexes (empty -- populated as items are written).
        let category_index = BitmapIndex::new("category");
        let format_index = BitmapIndex::new("format");
        let creator_index = BitmapIndex::new("creator");
        let tag_index = BitmapIndex::new("tags");
        let duration_index = RangeIndex::new("duration");
        let created_at_index = RangeIndex::new("created_at");
        let universe = RoaringBitmap::new();
        let embedding_registry = EmbeddingSlotRegistry::new();

        let schema_def = schema.clone();

        match config.mode {
            StorageMode::Ephemeral => {
                let storage = StorageBox::Memory {
                    items: InMemoryBackend::default(),
                    users: InMemoryBackend::default(),
                    creators: InMemoryBackend::default(),
                };
                let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));

                // Read preference vector dimensionality from the schema.
                let pref_dim = schema_def
                    .embedding_slots()
                    .first()
                    .map_or(128, |s| s.dimensions);

                Ok(OpenResult {
                    storage,
                    ledger,
                    wal: None,
                    last_seq,
                    profile_registry,
                    category_index,
                    format_index,
                    creator_index,
                    tag_index,
                    duration_index,
                    created_at_index,
                    universe,
                    embedding_registry,
                    schema_def,
                    creator_items: CreatorItemsBitmap::new(),
                    user_state: UserStateIndex::new(),
                    hard_negatives: HardNegIndex::new(),
                    interaction_ledger: InteractionLedger::new(),
                    preference_vectors: PreferenceVectors::new(pref_dim),
                })
            }
            StorageMode::Persistent => {
                let data_dir = config.data_dir.as_ref().ok_or_else(|| {
                    TidalError::Internal("persistent mode requires data_dir".into())
                })?;

                // Open fjall storage.
                let fjall_storage =
                    crate::storage::FjallStorage::open(data_dir).map_err(TidalError::from)?;
                let storage = StorageBox::Fjall(fjall_storage);

                // Build WAL config.  The WAL directory sits inside data_dir
                // but `WalConfig.dir` is the *parent* of the "wal/" subdirectory.
                let wal_config = WalConfig {
                    dir: data_dir.clone(),
                    ..WalConfig::default()
                };

                let (wal, replayed_events) = WalHandle::open(wal_config).map_err(|e| {
                    TidalError::Durability(DurabilityError {
                        message: format!("WAL open failed: {e}"),
                    })
                })?;

                // Build the WAL bridge for the ledger.
                let wal_writer =
                    Box::new(WalHandleWriter::new(wal.sender(), Arc::clone(&last_seq)));

                // Construct the ledger.
                let ledger = Arc::new(SignalLedger::new(schema, wal_writer));

                // Restore signal state from the last checkpoint.
                if let Err(e) = ledger.restore(storage.items_engine()) {
                    tracing::warn!(
                        error = %e,
                        "signal ledger restore failed; starting from empty state"
                    );
                }

                // Replay WAL events that post-date the checkpoint.
                for event in replayed_events {
                    let type_id = SignalTypeId::new(u16::from(event.signal_type));
                    let entity_id = EntityId::new(event.entity_id);
                    let weight = f64::from(event.weight);
                    let timestamp = Timestamp::from_nanos(event.timestamp_nanos);
                    ledger.apply_wal_event(type_id, entity_id, weight, timestamp);
                }

                // Rebuild in-memory entity state from durable storage.
                // This reconstructs blocked/hidden/follows user state, creator-items
                // bitmaps, and interaction weights from persisted relationship edges
                // and item metadata.
                let user_state = UserStateIndex::new();
                let creator_items = CreatorItemsBitmap::new();
                let interaction_ledger = InteractionLedger::new();
                rebuild_entity_state(&storage, &user_state, &creator_items, &interaction_ledger)?;

                // Read preference vector dimensionality from the schema.
                let pref_dim = schema_def
                    .embedding_slots()
                    .first()
                    .map_or(128, |s| s.dimensions);

                Ok(OpenResult {
                    storage,
                    ledger,
                    wal: Some(wal),
                    last_seq,
                    profile_registry,
                    category_index,
                    format_index,
                    creator_index,
                    tag_index,
                    duration_index,
                    created_at_index,
                    universe,
                    embedding_registry,
                    schema_def,
                    creator_items,
                    user_state,
                    hard_negatives: HardNegIndex::new(),
                    interaction_ledger,
                    preference_vectors: PreferenceVectors::new(pref_dim),
                })
            }
        }
    }
}

impl Drop for TidalDb {
    fn drop(&mut self) {
        // Stop metrics HTTP server if still running.
        #[cfg(feature = "metrics")]
        if let Some(ref mut handle) = self.metrics_handle {
            handle.stop();
        }

        if let Err(e) = self.shutdown_inner() {
            tracing::error!(error = %e, "error during tidaldb shutdown");
        }
    }
}

impl std::fmt::Debug for TidalDb {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TidalDb")
            .field("mode", &self.config.mode)
            .field("closed", &self.closed.load(Ordering::Relaxed))
            .finish_non_exhaustive()
    }
}