reddb_server/runtime/

impl_kv.rs

//! KV DSL command execution and KvAtomicOps module.
//!
//! Handles `KV PUT key = value [EXPIRE n] [TAGS [...]] [IF NOT EXISTS]`,
//! `KV GET key`, and `KV DELETE key`.

use crate::application::ports::RuntimeEntityPort;
use crate::catalog::CollectionModel;
use crate::storage::unified::{Metadata, MetadataValue};

use super::impl_config_secret::show_secrets_allows_key;
use super::impl_core::{current_auth_identity, current_connection_id, current_tenant};
use super::*;

mod transactions;

#[derive(Clone)]
pub(super) struct PendingVaultWrite {
    collection: String,
    entity: crate::storage::UnifiedEntity,
    metadata: Metadata,
    tenant: Option<String>,
    key: String,
    base_id: Option<crate::storage::EntityId>,
    writer_xid: u64,
    applied: bool,
}

/// Default collection name for bare-key KV operations.
pub const KV_DEFAULT_COLLECTION: &str = "kv_default";

fn vault_master_key_ref(collection: &str) -> String {
    format!("red.vault.{collection}.master_key")
}

pub(super) fn keyed_model_name(model: crate::catalog::CollectionModel) -> &'static str {
    match model {
        crate::catalog::CollectionModel::Kv => "kv",
        crate::catalog::CollectionModel::Vault => "vault",
        crate::catalog::CollectionModel::Config => "config",
        _ => "collection",
    }
}

#[derive(Debug, Clone)]
struct VaultEntry {
    id: crate::storage::EntityId,
    key: String,
    value: reddb_types::Value,
    metadata: Metadata,
    created_at: u64,
    updated_at: u64,
    sequence_id: u64,
    version: i64,
    tombstone: bool,
    op: String,
    tenant: Option<String>,
}

impl super::keyed_spine::KeyedVersion for VaultEntry {
    fn key(&self) -> &str {
        &self.key
    }

    fn version(&self) -> i64 {
        self.version
    }
}

impl VaultEntry {
    fn from_keyed_row(
        version: super::keyed_spine::KeyedRowVersion,
        metadata: Metadata,
        created_at: u64,
        updated_at: u64,
        sequence_id: u64,
        tenant: Option<String>,
    ) -> Self {
        Self {
            id: version.id,
            key: version.key,
            value: version.value,
            metadata,
            created_at,
            updated_at,
            sequence_id,
            version: version.version,
            tombstone: version.tombstone,
            op: version.op,
            tenant,
        }
    }
}

/// Atomic KV operations interface — the seam that transports and drivers depend on.
///
/// All three verbs delegate to the runtime's existing `create_kv` / `get_kv` /
/// `delete_kv` plumbing; this struct adds the auto-create and upsert logic.
pub struct KvAtomicOps<'a> {
    runtime: &'a RedDBRuntime,
    tenant: Option<String>,
}

impl<'a> KvAtomicOps<'a> {
    pub fn new(runtime: &'a RedDBRuntime) -> Self {
        Self {
            runtime,
            tenant: current_tenant(),
        }
    }

    fn for_tenant(runtime: &'a RedDBRuntime, tenant: Option<String>) -> Self {
        Self { runtime, tenant }
    }

    fn vault_row_in_scope(&self, row: &crate::storage::RowData) -> bool {
        let tenant = row
            .get_field("_vault_tenant")
            .and_then(|value| match value {
                Value::Text(tenant) => Some(tenant.as_ref()),
                _ => None,
            });
        tenant == self.tenant.as_deref()
    }

    /// Insert or update a KV entry. Auto-creates the collection when needed.
    ///
    /// Insert or update a KV entry. Returns `(created: bool, id: EntityId)`.
    pub fn set(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
        value: reddb_types::Value,
        ttl_ms: Option<u64>,
        if_not_exists: bool,
    ) -> RedDBResult<(bool, crate::storage::EntityId)> {
        self.set_with_tags_for_model(model, collection, key, value, ttl_ms, &[], if_not_exists)
    }

    pub fn set_with_tags(
        &self,
        collection: &str,
        key: &str,
        value: reddb_types::Value,
        ttl_ms: Option<u64>,
        tags: &[String],
        if_not_exists: bool,
    ) -> RedDBResult<(bool, crate::storage::EntityId)> {
        self.set_with_tags_for_model(
            crate::catalog::CollectionModel::Kv,
            collection,
            key,
            value,
            ttl_ms,
            tags,
            if_not_exists,
        )
    }

    fn set_with_tags_for_model(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
        value: reddb_types::Value,
        ttl_ms: Option<u64>,
        tags: &[String],
        if_not_exists: bool,
    ) -> RedDBResult<(bool, crate::storage::EntityId)> {
        self.set_with_tags_and_metadata_for_model(
            model,
            collection,
            key,
            value,
            ttl_ms,
            tags,
            if_not_exists,
            Vec::new(),
        )
    }

    pub fn set_with_tags_and_metadata(
        &self,
        collection: &str,
        key: &str,
        value: reddb_types::Value,
        ttl_ms: Option<u64>,
        tags: &[String],
        if_not_exists: bool,
        metadata: Vec<(String, MetadataValue)>,
    ) -> RedDBResult<(bool, crate::storage::EntityId)> {
        self.set_with_tags_and_metadata_for_model(
            crate::catalog::CollectionModel::Kv,
            collection,
            key,
            value,
            ttl_ms,
            tags,
            if_not_exists,
            metadata,
        )
    }

    fn set_with_tags_and_metadata_for_model(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
        value: reddb_types::Value,
        ttl_ms: Option<u64>,
        tags: &[String],
        if_not_exists: bool,
        mut metadata: Vec<(String, MetadataValue)>,
    ) -> RedDBResult<(bool, crate::storage::EntityId)> {
        // PUT, INCR and CAS serialize replacement of the same physical key.
        let vault_model = model == crate::catalog::CollectionModel::Vault;
        if vault_model {
            self.runtime
                .check_vault_capability("vault:write", collection, key)
                .map_err(RedDBError::Query)?;
        }
        let lock_key = if vault_model {
            format!("vault/{}/{}", self.tenant.as_deref().unwrap_or(""), key)
        } else {
            key.to_string()
        };
        let rmw_lock = self.runtime.inner.rmw_locks.lock_for(collection, &lock_key);
        let _rmw_guard = rmw_lock.lock();
        let vault_store = self.runtime.inner.db.store();
        let _vault_guard = vault_model.then(|| vault_store.vault_publication_lock.lock());
        let _snapshot_guard = vault_model.then(|| {
            super::execution_context::CurrentSnapshotGuard::install(
                self.runtime.vault_write_snapshot(),
            )
        });

        self.ensure_keyed_collection(model, collection)?;

        if model == crate::catalog::CollectionModel::Vault {
            let latest = self.get_vault_entry(collection, key)?;
            let was_present = latest.as_ref().is_some_and(|entry| !entry.tombstone);
            if if_not_exists && was_present {
                return Ok((false, latest.expect("checked present").id));
            }
            let entry =
                self.append_vault_version_unlocked(collection, key, value, "put", false, tags)?;
            self.runtime.record_kv_watch_event(
                if latest.is_some() {
                    crate::replication::cdc::ChangeOperation::Update
                } else {
                    crate::replication::cdc::ChangeOperation::Insert
                },
                collection,
                key,
                entry.id.raw(),
                latest.as_ref().map(vault_entry_metadata_json),
                Some(vault_entry_metadata_json(&entry)),
            );
            return Ok((!was_present, entry.id));
        }

        let existing = self.get_entry(model, collection, key)?;
        let was_present = existing.is_some();

        if if_not_exists && was_present {
            let (_, id) = existing.unwrap();
            self.runtime.inner.kv_stats.incr_puts();
            return Ok((false, id));
        }

        let before = existing
            .as_ref()
            .map(|(value, _)| reddb_types::Value::to_json(value));
        let op = if was_present {
            crate::replication::cdc::ChangeOperation::Update
        } else {
            crate::replication::cdc::ChangeOperation::Insert
        };
        let after = Some(reddb_types::Value::to_json(&value));

        // Versioned KV retains MVCC history: the prior visible version is
        // tombstoned (set_xmax) *after* the new version is created, so both
        // coexist physically — exactly like a table-row update. Non-versioned
        // KV publishes the replacement and TTL in one durable WAL batch.
        let versioned = self.is_versioned_collection(model, collection)?;
        // Capture the prior *visible* version (if any) before the write so we
        // can tombstone it. On a re-PUT after a delete (`was_present` false)
        // there is no visible prior, but earlier versions still exist
        // physically — see the xmin restamp below.
        let prior_versioned_entity = if versioned && was_present {
            self.get_entity(model, collection, key)?
        } else {
            None
        };
        let atomic = super::kv_write::atomic_available(self.runtime, collection)?;
        let previous = atomic
            .then(|| existing.as_ref().map(|(_, id)| *id))
            .flatten();
        if was_present && !versioned && !atomic {
            self.delete_unlocked(model, collection, key)?;
        }

        if let Some(ttl_metadata) = ttl_metadata(ttl_ms) {
            metadata.extend(ttl_metadata.fields);
        }
        if let Some(tags_metadata) = kv_tags_metadata(tags) {
            metadata.push(tags_metadata);
        }

        let output = super::kv_write::create(
            self.runtime,
            crate::application::entity::CreateKvInput {
                collection: collection.to_string(),
                key: key.to_string(),
                value,
                metadata,
            },
            previous,
        )?;

        // Every versioned PUT stamps the new version's xmin from ONE monotonic
        // xid (begin()/commit() or the active tx xid) — NOT the autocommit
        // pool, whose batched reservation is not monotonic against commit
        // root_xids and would let AS OF resolve the wrong version. The prior
        // visible version (if any) is tombstoned with that SAME xid, so
        // old.xmax == new.xmin and exactly one version is visible to any
        // snapshot (mirrors `apply_loaded_sql_update_row_core`). Restamping
        // unconditionally also fixes the re-PUT-after-delete case, where no
        // visible prior exists but the pool xmin would still scramble ordering
        // against existing tombstoned versions.
        if versioned {
            let version_xid = match self.runtime.current_xid() {
                Some(xid) => xid,
                None => {
                    let mgr = self.runtime.snapshot_manager();
                    let xid = mgr.begin();
                    mgr.commit(xid);
                    xid
                }
            };
            if let Some(new_entity) = output.entity.clone() {
                self.restamp_xmin(collection, new_entity, version_xid)?;
            }
            if let Some(prior) = prior_versioned_entity {
                // First-committer-wins (ADR 0014): inside an explicit
                // transaction, defer the prior version's tombstone to the
                // commit-time conflict check by recording it as a pending
                // versioned update — exactly like a table-row versioned
                // UPDATE (`persist_applied_entity_mutations`). The check
                // rejects this txn's COMMIT if the prior version was
                // already tombstoned by a concurrent committed writer.
                // Autocommit (`current_xid()` is None) commits eagerly and
                // records nothing, matching the table path.
                let previous_xmax = prior.xmax;
                let old_id = prior.id;
                self.tombstone_version(collection, prior, version_xid)?;
                if self.runtime.current_xid().is_some() {
                    self.runtime.record_pending_versioned_update(
                        crate::runtime::impl_core::current_connection_id(),
                        collection,
                        old_id,
                        output.id,
                        version_xid,
                        previous_xmax,
                    );
                }
            }
        }
        if model == crate::catalog::CollectionModel::Kv {
            self.runtime
                .inner
                .kv_tag_index
                .replace(collection, key, output.id, tags);
        }

        if model == crate::catalog::CollectionModel::Kv {
            if let Some(previous) = previous {
                self.runtime.record_kv_watch_event(
                    crate::replication::cdc::ChangeOperation::Delete,
                    collection,
                    key,
                    previous.raw(),
                    before.clone(),
                    None,
                );
                self.runtime.inner.kv_stats.incr_deletes();
            }
            self.runtime
                .record_kv_watch_event(op, collection, key, output.id.raw(), before, after);
        }

        if model == crate::catalog::CollectionModel::Kv {
            self.runtime.inner.kv_stats.incr_puts();
        }
        Ok((!was_present, output.id))
    }

    /// Retrieve a KV value by key. Returns `None` when not found.
    pub fn get(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
    ) -> RedDBResult<Option<reddb_types::Value>> {
        if model == CollectionModel::Vault {
            self.runtime
                .check_vault_capability("vault:read_metadata", collection, key)
                .map_err(RedDBError::Query)?;
            return Ok(self
                .get_vault_entry(collection, key)?
                .filter(|entry| !entry.tombstone)
                .map(|entry| entry.value));
        }
        let result = self.get_entry(model, collection, key)?;
        if model == crate::catalog::CollectionModel::Kv {
            self.runtime.inner.kv_stats.incr_gets();
        }
        Ok(result.map(|(v, _)| v))
    }

    /// Delete a KV entry. Returns `true` if the key existed.
    pub fn delete(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
    ) -> RedDBResult<bool> {
        let lock = (model != CollectionModel::Vault)
            .then(|| self.runtime.inner.rmw_locks.lock_for(collection, key));
        let _guard = lock.as_ref().map(|lock| lock.lock());
        if model == CollectionModel::Kv {
            self.runtime
                .check_collection_privilege(reddb_rql::ast::PolicyAction::Delete, collection)?;
            let Some(entity) = self.get_entity(model, collection, key)? else {
                return Ok(false);
            };
            if !crate::runtime::rls_injection::entity_passes_rls_action(
                self.runtime,
                collection,
                reddb_rql::ast::PolicyAction::Delete,
                &entity,
            ) {
                return Ok(false);
            }
        }
        self.delete_unlocked(model, collection, key)
    }

    fn delete_unlocked(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
    ) -> RedDBResult<bool> {
        self.ensure_declared_model(model, collection)?;
        if model == CollectionModel::Vault {
            self.runtime
                .check_vault_capability("vault:write", collection, key)
                .map_err(RedDBError::Query)?;
            let existed = self
                .get_vault_entry(collection, key)?
                .is_some_and(|entry| !entry.tombstone);
            self.append_vault_version(collection, key, Value::Null, "delete", true, &[])?;
            return Ok(existed);
        }

        // Versioned KV deletes tombstone the prior visible version
        // (set_xmax) instead of reclaiming it physically, so history and
        // time-travel reads survive — VACUUM reclaims the version later.
        if self.is_versioned_collection(model, collection)? {
            let Some(entity) = self.get_entity(model, collection, key)? else {
                return Ok(false);
            };
            let id = entity.id;
            let value = kv_value_from_entity(&entity);
            let xid = self.runtime.current_xid().unwrap_or_else(|| {
                let mgr = self.runtime.snapshot_manager();
                let xid = mgr.begin();
                mgr.commit(xid);
                xid
            });
            // First-committer-wins (ADR 0014): inside an explicit
            // transaction, defer this tombstone to the commit-time
            // conflict check by recording it as a pending tombstone —
            // exactly like a table-row DELETE (`delete_entities_batch`).
            // The check rejects this txn's COMMIT if the version was
            // already tombstoned by a concurrent committed writer.
            // Autocommit (`current_xid()` is None) commits eagerly and
            // records nothing, matching the table path.
            let previous_xmax = entity.xmax;
            self.tombstone_version(collection, entity, xid)?;
            if self.runtime.current_xid().is_some() {
                self.runtime.record_pending_tombstone(
                    crate::runtime::impl_core::current_connection_id(),
                    collection,
                    id,
                    xid,
                    previous_xmax,
                );
            }
            self.runtime.inner.kv_tag_index.remove(collection, key);
            self.runtime.record_kv_watch_event(
                crate::replication::cdc::ChangeOperation::Delete,
                collection,
                key,
                id.raw(),
                value.as_ref().map(reddb_types::Value::to_json),
                None,
            );
            self.runtime.inner.kv_stats.incr_deletes();
            return Ok(true);
        }

        let found = self.get_entry(model, collection, key)?;
        if let Some((value, id)) = found {
            let store = self.runtime.inner.db.store();
            let deleted = store
                .delete(collection, id)
                .map_err(|err| RedDBError::Internal(err.to_string()))?;
            if deleted {
                store.context_index().remove_entity(id);
                if model == crate::catalog::CollectionModel::Kv {
                    self.runtime.inner.kv_tag_index.remove(collection, key);
                    self.runtime.record_kv_watch_event(
                        crate::replication::cdc::ChangeOperation::Delete,
                        collection,
                        key,
                        id.raw(),
                        Some(reddb_types::Value::to_json(&value)),
                        None,
                    );
                    self.runtime.inner.kv_stats.incr_deletes();
                }
            }
            Ok(deleted)
        } else {
            Ok(false)
        }
    }

    fn delete_existing_for_replace_unlocked(
        &self,
        collection: &str,
        id: crate::storage::EntityId,
    ) -> RedDBResult<()> {
        let store = self.runtime.inner.db.store();
        let deleted = store
            .delete(collection, id)
            .map_err(|err| RedDBError::Internal(err.to_string()))?;
        if deleted {
            store.context_index().remove_entity(id);
        }
        Ok(())
    }

    /// Tombstone a KV version by stamping `xmax = xid` and persisting it.
    ///
    /// The physical row stays alive (history); VACUUM reclaims it once no
    /// snapshot can see it. Used by versioned PUT (old version) and
    /// versioned DELETE.
    fn tombstone_version(
        &self,
        collection: &str,
        mut entity: crate::storage::UnifiedEntity,
        xid: crate::storage::transaction::snapshot::Xid,
    ) -> RedDBResult<()> {
        let store = self.runtime.inner.db.store();
        let Some(manager) = store.get_collection(collection) else {
            return Ok(());
        };
        let id = entity.id;
        entity.set_xmax(xid);
        manager
            .update(entity.clone())
            .map_err(|err| RedDBError::Internal(format!("{err:?}")))?;
        store
            .persist_entities_to_pager(collection, std::slice::from_ref(&entity))
            .map_err(|err| RedDBError::Internal(err.to_string()))?;
        // The tombstoned version drops out of the live context index but
        // stays physically resident for AS OF reads until VACUUM.
        store.context_index().remove_entity(id);
        Ok(())
    }

    /// Restamp a freshly-created KV version's `xmin` to a monotonic version
    /// xid and persist it. `create_kv` stamps a (non-monotonic) autocommit
    /// pool xid; the versioned write path overrides it so the new version's
    /// xmin equals the old version's xmax, preserving AS OF ordering against
    /// commit root_xids.
    fn restamp_xmin(
        &self,
        collection: &str,
        mut entity: crate::storage::UnifiedEntity,
        xid: crate::storage::transaction::snapshot::Xid,
    ) -> RedDBResult<()> {
        let store = self.runtime.inner.db.store();
        let Some(manager) = store.get_collection(collection) else {
            return Ok(());
        };
        entity.set_xmin(xid);
        manager
            .update(entity.clone())
            .map_err(|err| RedDBError::Internal(format!("{err:?}")))?;
        store
            .persist_entities_to_pager(collection, std::slice::from_ref(&entity))
            .map_err(|err| RedDBError::Internal(err.to_string()))?;
        Ok(())
    }

    /// The `_ttl_ms` currently recorded for `key`, if it has one.
    ///
    /// Used to carry an existing expiry across the delete-then-create that
    /// non-versioned KV updates go through, so an operation that does not
    /// mention TTL does not silently clear it.
    fn existing_ttl_ms(&self, collection: &str, key: &str) -> RedDBResult<Option<u64>> {
        let Some(entity) = self.get_entity(crate::catalog::CollectionModel::Kv, collection, key)?
        else {
            return Ok(None);
        };
        let Some(metadata) =
            self.runtime
                .inner
                .db
                .store()
                .try_get_metadata(collection, entity.id, usize::MAX)?
        else {
            return Ok(None);
        };
        Ok(match metadata.get("_ttl_ms") {
            Some(MetadataValue::Int(ms)) if *ms >= 0 => Some(*ms as u64),
            Some(MetadataValue::Timestamp(ms)) => Some(*ms),
            _ => None,
        })
    }

    /// Atomically increment (or decrement) a counter key. Returns the new value.
    ///
    /// - Missing key initialises at `by` (Redis-compat).
    /// - Non-integer value returns an error before any mutation.
    pub fn incr(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
        by: i64,
        ttl_ms: Option<u64>,
    ) -> RedDBResult<i64> {
        if model == crate::catalog::CollectionModel::Vault {
            return Err(RedDBError::InvalidOperation(
                "VAULT INCR is not supported for sealed secrets".to_string(),
            ));
        }
        let rmw_lock = self.runtime.inner.rmw_locks.lock_for(collection, key);
        let _rmw_guard = rmw_lock.lock();
        self.ensure_kv_collection(collection)?;
        let existing = self.runtime.get_kv(collection, key)?;
        let current: i64 = match existing.as_ref() {
            None => 0,
            Some((reddb_types::Value::Integer(n), _)) => *n,
            Some((reddb_types::Value::Float(f), _)) => *f as i64,
            Some((other, _)) => {
                return Err(RedDBError::Internal(format!(
                    "INCR on non-integer value: {:?}",
                    other
                )));
            }
        };

        let next = current
            .checked_add(by)
            .ok_or_else(|| RedDBError::Internal(format!("INCR overflow: {current} + {by}")))?;

        // `KV INCR key [BY n] [EXPIRE dur]` has no TAGS clause, and EXPIRE is
        // optional, so the command cannot restate either. Non-versioned KV
        // updates by delete-then-create (that is how a TTL gets refreshed),
        // which drops whatever the previous version carried — so a plain
        // INCR silently made an expiring key permanent and cleared its tags.
        // Carry both forward unless the caller asked to change them.
        let carried_tags = self
            .runtime
            .inner
            .kv_tag_index
            .tags_for_key(collection, key);
        let carried_ttl_ms = if ttl_ms.is_none() {
            self.existing_ttl_ms(collection, key)?
        } else {
            None
        };

        let atomic = super::kv_write::atomic_available(self.runtime, collection)?;
        let previous = atomic
            .then(|| existing.as_ref().map(|(_, id)| *id))
            .flatten();
        if !atomic {
            if let Some((_, id)) = existing.as_ref() {
                self.delete_existing_for_replace_unlocked(collection, *id)?;
            }
        }

        let mut meta_vec: Vec<(String, crate::storage::unified::MetadataValue)> =
            ttl_metadata(ttl_ms.or(carried_ttl_ms))
                .map(|m| m.fields.into_iter().collect())
                .unwrap_or_default();
        if let Some(tags_metadata) = kv_tags_metadata(&carried_tags) {
            meta_vec.push(tags_metadata);
        }

        let output = super::kv_write::create(
            self.runtime,
            crate::application::entity::CreateKvInput {
                collection: collection.to_string(),
                key: key.to_string(),
                value: reddb_types::Value::Integer(next),
                metadata: meta_vec,
            },
            previous,
        )?;
        self.runtime
            .inner
            .kv_tag_index
            .replace(collection, key, output.id, &carried_tags);

        self.runtime.record_kv_watch_event(
            if existing.is_some() {
                crate::replication::cdc::ChangeOperation::Update
            } else {
                crate::replication::cdc::ChangeOperation::Insert
            },
            collection,
            key,
            output.id.raw(),
            existing
                .as_ref()
                .map(|(value, _)| reddb_types::Value::to_json(value)),
            Some(reddb_types::Value::to_json(&reddb_types::Value::Integer(
                next,
            ))),
        );

        self.runtime.inner.kv_stats.incr_incrs();
        Ok(next)
    }

    /// Compare-and-set: atomically swap `key` from `expected` to `new_value`.
    ///
    /// Returns `(ok, current)`:
    /// - `ok = true`  → swap applied; `current` is the value *before* the swap.
    /// - `ok = false` → swap skipped; `current` holds the actual current value.
    ///
    /// `expected = None` means the caller expects the key to be *absent* (create-if-absent).
    pub fn cas(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
        expected: Option<&reddb_types::Value>,
        new_value: reddb_types::Value,
        ttl_ms: Option<u64>,
    ) -> RedDBResult<(bool, Option<reddb_types::Value>)> {
        if model == crate::catalog::CollectionModel::Vault {
            return Err(RedDBError::InvalidOperation(
                "VAULT CAS is not supported for sealed secrets".to_string(),
            ));
        }
        let rmw_lock = self.runtime.inner.rmw_locks.lock_for(collection, key);
        let _rmw_guard = rmw_lock.lock();
        self.ensure_kv_collection(collection)?;
        let existing = self.runtime.get_kv(collection, key)?;
        let current = existing.as_ref().map(|(v, _)| v.clone());

        let matches = match (&current, expected) {
            (None, None) => true,
            (Some(cur), Some(exp)) => cur == exp,
            _ => false,
        };

        if !matches {
            self.runtime.inner.kv_stats.incr_cas_conflict();
            return Ok((false, current));
        }

        // `KV CAS key EXPECT <v> SET <v> [EXPIRE dur]` has no TAGS clause and
        // an optional EXPIRE, so — exactly as with INCR — the delete-then-
        // create below would drop a TTL and tags the caller never asked to
        // change. Carry both across the swap.
        let carried_tags = self
            .runtime
            .inner
            .kv_tag_index
            .tags_for_key(collection, key);
        let carried_ttl_ms = if ttl_ms.is_none() {
            self.existing_ttl_ms(collection, key)?
        } else {
            None
        };

        let atomic = super::kv_write::atomic_available(self.runtime, collection)?;
        let previous = atomic
            .then(|| existing.as_ref().map(|(_, id)| *id))
            .flatten();
        if !atomic {
            if let Some((_, id)) = existing.as_ref() {
                self.delete_existing_for_replace_unlocked(collection, *id)?;
            }
        }

        let mut meta_vec: Vec<(String, crate::storage::unified::MetadataValue)> =
            ttl_metadata(ttl_ms.or(carried_ttl_ms))
                .map(|m| m.fields.into_iter().collect())
                .unwrap_or_default();
        if let Some(tags_metadata) = kv_tags_metadata(&carried_tags) {
            meta_vec.push(tags_metadata);
        }

        let output = super::kv_write::create(
            self.runtime,
            crate::application::entity::CreateKvInput {
                collection: collection.to_string(),
                key: key.to_string(),
                value: new_value.clone(),
                metadata: meta_vec,
            },
            previous,
        )?;
        self.runtime
            .inner
            .kv_tag_index
            .replace(collection, key, output.id, &carried_tags);

        self.runtime.record_kv_watch_event(
            if current.is_some() {
                crate::replication::cdc::ChangeOperation::Update
            } else {
                crate::replication::cdc::ChangeOperation::Insert
            },
            collection,
            key,
            output.id.raw(),
            current.as_ref().map(reddb_types::Value::to_json),
            Some(reddb_types::Value::to_json(&new_value)),
        );

        self.runtime.inner.kv_stats.incr_cas_success();
        Ok((true, current))
    }

    pub fn invalidate_tags(&self, collection: &str, tags: &[String]) -> RedDBResult<usize> {
        self.runtime
            .check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        self.runtime.check_kv_invalidate_policy(collection)?;
        self.ensure_kv_collection(collection)?;
        let entries = self
            .runtime
            .inner
            .kv_tag_index
            .entries_for_tags(collection, tags);
        if entries.is_empty() {
            return Ok(0);
        }

        let store = self.runtime.inner.db.store();
        let mut removed = 0usize;
        for (key, id) in entries {
            let Some(entity) = store.get(collection, id) else {
                continue;
            };
            if !crate::runtime::rls_injection::entity_passes_rls_action(
                self.runtime,
                collection,
                reddb_rql::ast::PolicyAction::Delete,
                &entity,
            ) {
                continue;
            }
            let before = kv_value_from_entity(&entity);
            let deleted = store
                .delete(collection, id)
                .map_err(|err| RedDBError::Internal(err.to_string()))?;
            if deleted {
                store.context_index().remove_entity(id);
                self.runtime.inner.kv_tag_index.remove(collection, &key);
                self.runtime.record_kv_watch_event(
                    crate::replication::cdc::ChangeOperation::Delete,
                    collection,
                    &key,
                    id.raw(),
                    before.as_ref().map(reddb_types::Value::to_json),
                    None,
                );
                removed += 1;
            }
        }
        if removed > 0 {
            self.runtime.inner.kv_stats.incr_deletes();
        }
        Ok(removed)
    }

    pub fn tags_for_key(&self, collection: &str, key: &str) -> Vec<String> {
        self.runtime
            .inner
            .kv_tag_index
            .tags_for_key(collection, key)
    }

    /// Auto-create a KV collection if it does not exist yet.
    fn ensure_kv_collection(&self, collection: &str) -> RedDBResult<()> {
        self.ensure_keyed_collection(crate::catalog::CollectionModel::Kv, collection)
    }

    fn ensure_keyed_collection(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
    ) -> RedDBResult<()> {
        let store = self.runtime.inner.db.store();
        if store.get_collection(collection).is_some() {
            return self.ensure_declared_model(model, collection);
        }
        if model != crate::catalog::CollectionModel::Kv {
            return Err(RedDBError::NotFound(format!(
                "{} collection '{collection}' does not exist",
                keyed_model_name(model)
            )));
        }
        // Check config gate: red.config.kv.default_collection (default = true).
        let auto_create = self
            .runtime
            .config_bool("red.config.kv.default_collection", true);
        if !auto_create {
            return Err(RedDBError::NotFound(format!(
                "kv collection '{collection}' does not exist and auto-create is disabled \
                 (red.config.kv.default_collection = false)"
            )));
        }
        store
            .create_collection(collection)
            .map_err(|err| RedDBError::Internal(err.to_string()))?;
        self.runtime
            .inner
            .db
            .save_collection_contract(kv_collection_contract(collection))
            .map_err(|err| RedDBError::Internal(err.to_string()))?;
        Ok(())
    }

    fn get_entry(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
    ) -> RedDBResult<Option<(reddb_types::Value, crate::storage::EntityId)>> {
        let resolver =
            super::kv_read::KvReadResolver::for_keyed_model(self.runtime, model, collection)?;
        Ok(resolver
            .record(key)?
            .and_then(|record| record.value.map(|value| (value, record.id))))
    }

    /// Whether `collection` retains MVCC history (opted into versioning).
    ///
    /// Versioned KV reads must do MVCC version-selection; non-versioned
    /// collections (config `red_config`, the secrets vault) keep the
    /// last-writer-wins fast path. Internal `red_*` collections are
    /// never versioned, so this stays cheap for them.
    fn is_versioned_collection(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
    ) -> RedDBResult<bool> {
        // Only plain KV ever opts into versioning; the vault and config
        // tiers are non-versioned by construction.
        super::kv_read::is_versioned(self.runtime, model, collection)
    }

    fn get_entity(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
        key: &str,
    ) -> RedDBResult<Option<crate::storage::UnifiedEntity>> {
        super::kv_read::KvReadResolver::for_keyed_model(self.runtime, model, collection)?
            .owned_entity(key)
    }

    fn latest_kv_entries(
        &self,
        collection: &str,
        prefix: Option<&str>,
    ) -> RedDBResult<Vec<(String, crate::storage::UnifiedEntity)>> {
        self.ensure_declared_model(crate::catalog::CollectionModel::Kv, collection)?;
        let store = self.runtime.inner.db.store();
        let Some(manager) = store.get_collection(collection) else {
            return Ok(Vec::new());
        };
        let _publication = manager
            .has_backing()
            .then(|| store.vault_publication_lock.lock());
        // Versioned KV: per logical key, keep the snapshot-visible
        // version with the greatest xmin; keys whose every version is
        // tombstoned (no visible version) drop out entirely. Mirrors
        // the per-key version-selection used by `get_entity`.
        let versioned =
            self.is_versioned_collection(crate::catalog::CollectionModel::Kv, collection)?;
        let snapshot = crate::runtime::impl_core::capture_current_snapshot();
        let mut entries: std::collections::BTreeMap<String, crate::storage::UnifiedEntity> =
            std::collections::BTreeMap::new();
        super::function_budget::try_scan(
            |visit| {
                if versioned {
                    manager.try_scan_for_each(snapshot.as_ref(), usize::MAX, visit)
                } else {
                    manager.try_for_each_entity(usize::MAX, visit)
                }
            },
            |entity| {
                let Some(reddb_types::Value::Text(key)) =
                    entity.data.as_row().and_then(|row| row.get_field("key"))
                else {
                    return true;
                };
                if prefix.is_some_and(|prefix| !key.starts_with(prefix)) {
                    return true;
                }
                let should_replace = match entries.get(key.as_ref()) {
                    Some(existing) if versioned => entity.xmin >= existing.xmin,
                    Some(existing) => entity.id.raw() >= existing.id.raw(),
                    None => true,
                };
                if should_replace {
                    entries.insert(key.to_string(), entity.clone());
                }
                true
            },
        )?;
        Ok(entries.into_iter().collect())
    }

    fn get_vault_entry(&self, collection: &str, key: &str) -> RedDBResult<Option<VaultEntry>> {
        self.record_vault_read(collection, Some(key));
        let mut pending = None;
        self.visit_pending_vault_entries(collection, Some(key), |entry| pending = Some(entry));
        if let Some(entry) = pending {
            return Ok(Some(entry));
        }
        let mut latest: Option<VaultEntry> = None;
        self.visit_vault_versions(collection, key, |entry| {
            super::keyed_spine::select_latest_version(&mut latest, entry);
        })?;
        Ok(latest)
    }

    fn get_vault_entry_version(
        &self,
        collection: &str,
        key: &str,
        version: i64,
    ) -> RedDBResult<Option<VaultEntry>> {
        let mut selected = None;
        self.visit_vault_versions(collection, key, |entry| {
            if entry.version == version {
                selected = Some(entry);
            }
        })?;
        Ok(selected)
    }

    fn vault_versions(&self, collection: &str, key: &str) -> RedDBResult<Vec<VaultEntry>> {
        let mut versions = Vec::new();
        self.visit_vault_versions(collection, key, |entry| versions.push(entry))?;
        Ok(versions)
    }

    /// Point selectors retain only their chosen entry. Explicit history callers
    /// can collect the same fallible stream; every version remains in storage.
    fn visit_vault_versions(
        &self,
        collection: &str,
        key: &str,
        mut visit: impl FnMut(VaultEntry),
    ) -> RedDBResult<()> {
        self.record_vault_read(collection, Some(key));
        self.ensure_declared_model(crate::catalog::CollectionModel::Vault, collection)?;
        let store = self.runtime.inner.db.store();
        let manager = store.get_collection(collection);
        let snapshot = self.runtime.vault_read_snapshot();
        let _publication = manager
            .as_ref()
            .filter(|manager| manager.has_backing())
            .map(|_| store.vault_publication_lock.lock());
        let records = manager.as_ref().map(|manager| {
            super::keyed_spine::records_with_metadata(manager, Some(snapshot), Some(key), |entity| {
                entity.data.as_row().is_some_and(|row| self.vault_row_in_scope(row)
                    && matches!(row.get_field("key"), Some(Value::Text(stored)) if stored.as_ref() == key))
            })
        }).transpose()?.into_iter().flatten();
        let mut found = false;
        for record in records {
            let (entity, metadata) = record?;
            let crate::storage::EntityData::Row(ref row) = entity.data else {
                continue;
            };
            if !self.vault_row_in_scope(row) {
                continue;
            }
            let Some(version) =
                super::keyed_spine::row_version(entity.id, row, entity.sequence_id as i64)
            else {
                continue;
            };
            if version.key != key {
                continue;
            }
            let metadata = metadata.unwrap_or_default();
            found = true;
            visit(VaultEntry::from_keyed_row(
                version,
                metadata,
                entity.created_at,
                entity.updated_at,
                entity.sequence_id,
                self.tenant.clone(),
            ));
        }
        if !found {
            if let Some(entry) = self.legacy_vault_entry(collection, key)? {
                visit(entry);
            }
        }
        self.visit_pending_vault_entries(collection, Some(key), visit);
        Ok(())
    }

    fn latest_vault_entries(
        &self,
        collection: &str,
        prefix: Option<&str>,
    ) -> RedDBResult<Vec<VaultEntry>> {
        // Listing protects the tenant's collection, including absent keys.
        self.record_vault_read(collection, None);
        self.ensure_declared_model(crate::catalog::CollectionModel::Vault, collection)?;
        let store = self.runtime.inner.db.store();
        let manager = store.get_collection(collection);
        let snapshot = self.runtime.vault_read_snapshot();
        let _publication = manager
            .as_ref()
            .filter(|manager| manager.has_backing())
            .map(|_| store.vault_publication_lock.lock());
        let records = manager
            .as_ref()
            .map(|manager| {
                super::keyed_spine::records_with_metadata(manager, Some(snapshot), None, |entity| {
                    entity
                        .data
                        .as_row()
                        .is_some_and(|row| self.vault_row_in_scope(row)
                            && prefix.is_none_or(|prefix| matches!(row.get_field("key"), Some(Value::Text(key)) if key.starts_with(prefix))))
                })
            })
            .transpose()?
            .into_iter()
            .flatten();
        let mut latest = std::collections::BTreeMap::new();
        for record in records {
            let (entity, metadata) = record?;
            let crate::storage::EntityData::Row(ref row) = entity.data else {
                continue;
            };
            if !self.vault_row_in_scope(row) {
                continue;
            }
            let Some(version) =
                super::keyed_spine::row_version(entity.id, row, entity.sequence_id as i64)
            else {
                continue;
            };
            let metadata = metadata.unwrap_or_default();
            let entry = VaultEntry::from_keyed_row(
                version,
                metadata,
                entity.created_at,
                entity.updated_at,
                entity.sequence_id,
                self.tenant.clone(),
            );
            super::keyed_spine::retain_latest_version(&mut latest, entry);
        }
        if collection == "red.vault" && self.tenant.is_none() {
            if let Some(auth) = self.runtime.inner.auth_store.read().clone() {
                for key in auth.vault_kv_keys() {
                    if prefix.is_some_and(|prefix| !key.starts_with(prefix)) {
                        continue;
                    }
                    if !latest.contains_key(&key) {
                        if let Some(entry) = self.legacy_vault_entry(collection, &key)? {
                            super::keyed_spine::retain_latest_version(&mut latest, entry);
                        }
                    }
                }
            }
        }
        let mut pending_keys = std::collections::HashSet::new();
        self.visit_pending_vault_entries(collection, None, |entry| {
            if prefix.is_some_and(|prefix| !entry.key.starts_with(prefix)) {
                return;
            }
            // Any pending version shadows the base for this tenant/key, even
            // when an imported base carried a higher version number.
            if pending_keys.insert(entry.key.clone()) {
                latest.remove(&entry.key);
            }
            super::keyed_spine::retain_latest_version(&mut latest, entry);
        });
        Ok(latest
            .into_values()
            .filter(|entry| prefix.is_none_or(|prefix| entry.key.starts_with(prefix)))
            .collect())
    }

    fn legacy_vault_entry(&self, collection: &str, key: &str) -> RedDBResult<Option<VaultEntry>> {
        if collection != "red.vault" || self.tenant.is_some() || !show_secrets_allows_key(key) {
            return Ok(None);
        }
        let Some(value) = self.runtime.vault_kv_get(key) else {
            return Ok(None);
        };
        self.runtime.provision_default_vault_key()?;
        Ok(Some(VaultEntry {
            id: crate::storage::EntityId::new(0),
            key: key.into(),
            value: self.runtime.seal_vault_value_for_tenant(
                collection,
                Value::text(value),
                None,
            )?,
            metadata: Metadata::default(),
            created_at: 0,
            updated_at: 0,
            sequence_id: 0,
            version: 0,
            tombstone: false,
            op: "legacy".into(),
            tenant: None,
        }))
    }

    fn append_vault_version(
        &self,
        collection: &str,
        key: &str,
        value: reddb_types::Value,
        op: &str,
        tombstone: bool,
        tags: &[String],
    ) -> RedDBResult<VaultEntry> {
        let lock_key = format!("vault/{}/{}", self.tenant.as_deref().unwrap_or(""), key);
        let lock = self.runtime.inner.rmw_locks.lock_for(collection, &lock_key);
        let _guard = lock.lock();
        let vault_store = self.runtime.inner.db.store();
        let _vault_guard = vault_store.vault_publication_lock.lock();
        let _snapshot_guard = super::execution_context::CurrentSnapshotGuard::install(
            self.runtime.vault_write_snapshot(),
        );
        self.append_vault_version_unlocked(collection, key, value, op, tombstone, tags)
    }

    fn append_vault_version_unlocked(
        &self,
        collection: &str,
        key: &str,
        value: reddb_types::Value,
        op: &str,
        tombstone: bool,
        tags: &[String],
    ) -> RedDBResult<VaultEntry> {
        self.ensure_declared_model(crate::catalog::CollectionModel::Vault, collection)?;
        let previous = self.get_vault_entry(collection, key)?;
        let version = previous
            .as_ref()
            .map(|entry| entry.version)
            .unwrap_or(0)
            .checked_add(1)
            .ok_or_else(|| RedDBError::Query("vault version overflow".into()))?;
        let stored_value = if tombstone {
            reddb_types::Value::Null
        } else {
            self.runtime
                .seal_vault_value_for_tenant(collection, value, self.tenant.as_deref())?
        };
        let now = current_unix_ms() as i64;
        let mut fields = vec![
            ("key".to_string(), reddb_types::Value::text(key.to_string())),
            ("value".to_string(), stored_value),
            ("version".to_string(), reddb_types::Value::Integer(version)),
            (
                "tombstone".to_string(),
                reddb_types::Value::Boolean(tombstone),
            ),
            ("op".to_string(), reddb_types::Value::text(op.to_string())),
            (
                "created_at_ms".to_string(),
                reddb_types::Value::Integer(now),
            ),
        ];
        if let Some(tenant) = &self.tenant {
            fields.push(("_vault_tenant".into(), Value::text(tenant.clone())));
        }
        let mut row = crate::storage::RowData::new(Vec::new());
        row.named = Some(fields.into_iter().collect());
        let entity = crate::storage::UnifiedEntity::new(
            self.runtime.inner.db.store().next_entity_id(),
            crate::storage::EntityKind::TableRow {
                table: std::sync::Arc::from(collection),
                row_id: 0,
            },
            crate::storage::EntityData::Row(row),
        );
        self.stage_or_store_vault_version(
            collection,
            key,
            entity,
            Metadata::with_fields(vault_tags_metadata(tags)),
            previous.as_ref().map(|entry| entry.id),
        )
    }

    fn purge_vault_versions(&self, collection: &str, key: &str) -> RedDBResult<usize> {
        self.runtime.reject_vault_transaction()?;
        let lock_key = format!("vault/{}/{}", self.tenant.as_deref().unwrap_or(""), key);
        let lock = self.runtime.inner.rmw_locks.lock_for(collection, &lock_key);
        let _guard = lock.lock();
        let vault_store = self.runtime.inner.db.store();
        let _vault_guard = vault_store.vault_publication_lock.lock();
        let _snapshot_guard = super::execution_context::CurrentSnapshotGuard::install(
            self.runtime.vault_write_snapshot(),
        );
        self.ensure_declared_model(crate::catalog::CollectionModel::Vault, collection)?;
        let versions = self.vault_versions(collection, key)?;
        let store = self.runtime.inner.db.store();
        let mut purged = 0usize;
        for entry in versions {
            if store
                .delete(collection, entry.id)
                .map_err(|err| RedDBError::Internal(err.to_string()))?
            {
                store.context_index().remove_entity(entry.id);
                purged += 1;
            }
        }
        if collection == "red.vault" && self.tenant.is_none() && show_secrets_allows_key(key) {
            if let Some(auth) = self.runtime.inner.auth_store.read().clone() {
                if auth
                    .vault_kv_try_delete(key)
                    .map_err(|err| RedDBError::Query(err.to_string()))?
                {
                    purged += 1;
                }
            }
        }
        Ok(purged)
    }

    fn ensure_declared_model(
        &self,
        model: crate::catalog::CollectionModel,
        collection: &str,
    ) -> RedDBResult<()> {
        super::kv_read::KvReadResolver::for_keyed_model(self.runtime, model, collection).map(|_| ())
    }
}

impl RedDBRuntime {
    /// Offline dump records retain the native encrypted values, tenant markers,
    /// versions and metadata. They are never formatted as query results.
    pub fn export_vault_collection_records(&self, collection: &str) -> RedDBResult<Vec<String>> {
        let vault_store = self.inner.db.store();
        let _vault_guard = vault_store.vault_publication_lock.lock();
        let store = self.inner.db.store();
        let contract = self
            .inner
            .db
            .collection_contract(collection)
            .ok_or_else(|| RedDBError::NotFound(collection.into()))?;
        if contract.declared_model != CollectionModel::Vault {
            return Err(RedDBError::Query(
                "native vault export requires a vault collection".into(),
            ));
        }
        let manager = store
            .get_collection(collection)
            .ok_or_else(|| RedDBError::NotFound(collection.into()))?;
        let mut records = Vec::new();
        for record in super::keyed_spine::records_with_metadata(&manager, None, None, |_| true)? {
            let (entity, metadata) = record?;
            records.push((
                entity.id,
                hex::encode(crate::storage::UnifiedStore::serialize_entity_record(
                    &entity,
                    metadata.as_ref(),
                    store.format_version(),
                )),
            ));
        }
        records.sort_unstable_by_key(|(id, _)| id.raw());
        Ok(records.into_iter().map(|(_, encoded)| encoded).collect())
    }

    /// Restore an offline native Vault dump into an empty collection. Validate
    /// every encrypted row against the imported keys before writing any rows.
    pub fn import_vault_collection_records(
        &self,
        collection: &str,
        records: &[String],
        format_version: u32,
    ) -> RedDBResult<usize> {
        self.reject_vault_transaction()?;
        let vault_store = self.inner.db.store();
        let _vault_guard = vault_store.vault_publication_lock.lock();
        let store = self.inner.db.store();
        if collection.is_empty() || (collection.starts_with("red.") && collection != "red.vault") {
            return Err(RedDBError::Query("invalid vault import collection".into()));
        }
        if let Some(contract) = self.inner.db.collection_contract(collection) {
            if contract.declared_model != CollectionModel::Vault {
                return Err(RedDBError::Query(
                    "cannot restore vault records into another collection model".into(),
                ));
            }
        }
        if store
            .get_collection(collection)
            .is_some_and(|manager| manager.count() != 0)
        {
            return Err(RedDBError::Query(
                "vault restore requires an empty collection".into(),
            ));
        }
        let auth = self.inner.auth_store.read().clone().ok_or_else(|| {
            RedDBError::Query("vault restore requires imported vault keys".into())
        })?;
        if !auth.is_vault_backed() || auth.vault_secret_key().is_none() {
            return Err(RedDBError::Query(
                "vault restore requires imported vault keys".into(),
            ));
        }
        if format_version > store.format_version() {
            return Err(RedDBError::Query(
                "vault dump uses a newer storage format".into(),
            ));
        }
        let mut decoded = Vec::with_capacity(records.len());
        for record in records {
            let bytes = hex::decode(record)
                .map_err(|_| RedDBError::Query("invalid native vault dump record".into()))?;
            let (mut entity, metadata) =
                crate::storage::UnifiedStore::deserialize_vault_entity_record(
                    &bytes,
                    format_version,
                )
                .map_err(|error| RedDBError::Query(error.to_string()))?;
            let crate::storage::EntityData::Row(row) = &entity.data else {
                return Err(RedDBError::Query("vault dump requires rows".into()));
            };
            let Some(version) =
                super::keyed_spine::row_version(entity.id, row, entity.sequence_id as i64)
            else {
                return Err(RedDBError::Query("invalid vault version record".into()));
            };
            if !version.tombstone {
                let tenant = row
                    .get_field("_vault_tenant")
                    .and_then(|value| match value {
                        Value::Text(tenant) => Some(tenant.as_ref()),
                        _ => None,
                    });
                self.unseal_vault_value_for_tenant(collection, &version.value, tenant)?;
            }
            entity.id = crate::storage::EntityId::new(0);
            // Transaction ids belong to the source runtime. Restored Vault
            // versions are committed maintenance data in the destination.
            entity.set_xmin(0);
            entity.set_xmax(0);
            entity.kind = crate::storage::EntityKind::TableRow {
                table: std::sync::Arc::from(collection),
                row_id: 0,
            };
            decoded.push((entity, metadata));
        }
        if store.get_collection(collection).is_none() {
            store
                .create_collection(collection)
                .map_err(|error| RedDBError::Internal(error.to_string()))?;
        }
        let mut contract = kv_collection_contract(collection);
        contract.declared_model = CollectionModel::Vault;
        self.inner
            .db
            .save_collection_contract(contract)
            .map_err(|error| RedDBError::Internal(error.to_string()))?;
        self.inner
            .db
            .persist_metadata()
            .map_err(|error| RedDBError::Internal(error.to_string()))?;
        let count = decoded.len();
        for (entity, metadata) in decoded {
            let id = store
                .insert(collection, entity)
                .map_err(|error| RedDBError::Internal(error.to_string()))?;
            if let Some(metadata) = metadata {
                store
                    .set_metadata(collection, id, metadata)
                    .map_err(|error| RedDBError::Internal(error.to_string()))?;
            }
        }
        self.invalidate_result_cache();
        Ok(count)
    }

    pub(crate) fn reject_vault_transaction(&self) -> RedDBResult<()> {
        if self
            .inner
            .transaction_state
            .in_transaction(current_connection_id())
        {
            return Err(RedDBError::Query(
                "vault administration cannot run inside a transaction; execute after COMMIT or ROLLBACK"
                    .into(),
            ));
        }
        Ok(())
    }

    fn ensure_default_vault(&self) -> RedDBResult<()> {
        self.provision_default_vault_key()?;
        let lock = self.inner.rmw_locks.lock_for("red.vault", "initialize");
        let _guard = lock.lock();
        let store = self.inner.db.store();
        if store.get_collection("red.vault").is_none() {
            store
                .create_collection("red.vault")
                .map_err(|err| RedDBError::Internal(err.to_string()))?;
            let mut contract = kv_collection_contract("red.vault");
            contract.declared_model = CollectionModel::Vault;
            self.inner
                .db
                .save_collection_contract(contract)
                .map_err(|err| RedDBError::Internal(err.to_string()))?;
            self.inner
                .db
                .persist_metadata()
                .map_err(|err| RedDBError::Internal(err.to_string()))?;
        }
        Ok(())
    }

    fn provision_default_vault_key(&self) -> RedDBResult<()> {
        let auth = self.inner.auth_store.read().clone().ok_or_else(|| {
            RedDBError::Query("secret storage requires an enabled, unsealed vault".into())
        })?;
        if !auth.is_vault_backed() {
            return Err(RedDBError::Query(
                "secret storage requires an enabled, unsealed vault".into(),
            ));
        }
        auth.ensure_vault_secret_key_result()
            .map_err(|err| RedDBError::Query(err.to_string()))?;
        Ok(())
    }

    pub(crate) fn set_user_secret(&self, key: &str, value: Option<String>) -> RedDBResult<()> {
        self.check_vault_capability("vault:write", "red.vault", key)
            .map_err(RedDBError::Query)?;
        self.ensure_default_vault()?;
        let command = match value {
            Some(value) => reddb_rql::ast::KvCommand::Put {
                model: CollectionModel::Vault,
                collection: "red.vault".into(),
                key: key.into(),
                value: Value::text(value),
                ttl_ms: None,
                tags: Vec::new(),
                if_not_exists: false,
            },
            None => reddb_rql::ast::KvCommand::Delete {
                model: CollectionModel::Vault,
                collection: "red.vault".into(),
                key: key.into(),
            },
        };
        self.execute_kv_command("", &command)?;
        self.invalidate_result_cache();
        Ok(())
    }

    /// SQL references and Vault commands resolve the same entries. Legacy root
    /// KV entries are a platform-only read fallback until rewritten or deleted.
    pub(crate) fn query_secret_value(&self, path: &str) -> RedDBResult<Option<Value>> {
        let (collection, key) = path.split_once('/').unwrap_or(("red.vault", path));
        let (collection, key, owner_tenant) = if collection == "platform" {
            let (collection, key) = key.split_once('.').ok_or_else(|| {
                RedDBError::Query("platform secret expects platform.<vault>.<path>".into())
            })?;
            (
                if collection == "default" {
                    "red.vault"
                } else {
                    collection
                },
                key,
                None,
            )
        } else {
            (collection, key, current_tenant())
        };
        if collection == "red.vault" && !show_secrets_allows_key(key) {
            return Ok(None);
        }
        let ops = KvAtomicOps::for_tenant(self, owner_tenant.clone());
        if let Err(reason) = self.check_vault_capability_in_tenant(
            "vault:use",
            collection,
            key,
            owner_tenant.as_deref(),
        ) {
            self.audit_vault_lifecycle(
                "use",
                collection,
                key,
                crate::runtime::audit_log::Outcome::Denied,
                &reason,
                None,
            );
            self.emit_vault_control_event(
                crate::runtime::control_events::EventKind::VaultUse,
                crate::runtime::control_events::Outcome::Denied,
                "vault:use",
                collection,
                key,
                &reason,
                None,
                Vec::new(),
            )?;
            // SQL references hide denied values as NULL. Audit persistence and
            // physical read failures still propagate through their fallible paths.
            return Ok(None);
        }
        let entry = ops.get_vault_entry(collection, key)?;
        let value = match entry.as_ref() {
            Some(entry) if entry.tombstone => None,
            // SQL references retain the nullable policy while Vault is sealed.
            // Physical reads above and unsealed decode failures below still
            // propagate; VAULT REVEAL keeps its explicit sealed error.
            Some(_)
                if !self
                    .inner
                    .auth_store
                    .read()
                    .as_ref()
                    .is_some_and(|auth| auth.is_vault_backed()) =>
            {
                None
            }
            Some(entry) => Some(self.unseal_vault_value_for_tenant(
                collection,
                &entry.value,
                owner_tenant.as_deref(),
            )?),
            None if collection == "red.vault" && owner_tenant.is_none() => {
                self.vault_kv_get(key).map(Value::text)
            }
            None => None,
        };
        self.audit_vault_lifecycle(
            "use",
            collection,
            key,
            crate::runtime::audit_log::Outcome::Success,
            "ok",
            entry.as_ref(),
        );
        self.emit_vault_control_event(
            crate::runtime::control_events::EventKind::VaultUse,
            crate::runtime::control_events::Outcome::Allowed,
            "vault:use",
            collection,
            key,
            "ok",
            entry.as_ref(),
            Vec::new(),
        )?;
        Ok(value)
    }

    pub(crate) fn show_user_secrets(&self, prefix: Option<&str>) -> RedDBResult<UnifiedResult> {
        self.provision_default_vault_key()?;
        let ops = KvAtomicOps::new(self);
        let mut result = UnifiedResult::with_columns(vec![
            "collection".into(),
            "key".into(),
            "value".into(),
            "status".into(),
        ]);
        let mut default_keys = std::collections::HashSet::new();
        for collection in self.inner.db.store().list_collections() {
            let model = self
                .inner
                .db
                .collection_contract(&collection)
                .map(|contract| contract.declared_model);
            if model != Some(CollectionModel::Vault) {
                continue;
            }
            for entry in ops.latest_vault_entries(&collection, None)? {
                if collection == "red.vault" {
                    default_keys.insert(entry.key.clone());
                }
                let display_key = if collection == "red.vault" {
                    entry.key.clone()
                } else {
                    format!("{collection}.{}", entry.key)
                };
                if entry.tombstone
                    || !show_secrets_allows_key(&entry.key)
                    || prefix.is_some_and(|prefix| !display_key.starts_with(prefix))
                    || self
                        .check_vault_capability("vault:read_metadata", &collection, &entry.key)
                        .is_err()
                {
                    continue;
                }
                let mut record = UnifiedRecord::new();
                record.set("collection", Value::text(collection.clone()));
                record.set("key", Value::text(display_key));
                record.set("value", Value::text("***"));
                record.set("status", Value::text("active"));
                result.push(record);
            }
        }
        if current_tenant().is_none() {
            if let Some(auth) = self.inner.auth_store.read().clone() {
                for key in auth.vault_kv_keys() {
                    if default_keys.contains(&key)
                        || !show_secrets_allows_key(&key)
                        || prefix.is_some_and(|prefix| !key.starts_with(prefix))
                        || self
                            .check_vault_capability("vault:read_metadata", "red.vault", &key)
                            .is_err()
                    {
                        continue;
                    }
                    let mut record = UnifiedRecord::new();
                    record.set("collection", Value::text("red.vault"));
                    record.set("key", Value::text(key));
                    record.set("value", Value::text("***"));
                    record.set("status", Value::text("active"));
                    result.push(record);
                }
            }
        }
        result
            .records
            .sort_by_key(|record| record.get("key").map(Value::display_string));
        Ok(result)
    }
    pub(crate) fn seal_vault_value(
        &self,
        collection: &str,
        value: reddb_types::Value,
    ) -> RedDBResult<reddb_types::Value> {
        self.seal_vault_value_for_tenant(collection, value, current_tenant().as_deref())
    }

    fn seal_vault_value_for_tenant(
        &self,
        collection: &str,
        value: Value,
        tenant: Option<&str>,
    ) -> RedDBResult<Value> {
        let key = self.vault_encryption_key(collection)?;
        let plaintext = value.to_bytes();
        let nonce_bytes = crate::auth::store::random_bytes(12);
        let mut nonce = [0u8; 12];
        nonce.copy_from_slice(&nonce_bytes[..12]);
        let aad = vault_value_aad(collection, tenant);
        let ciphertext =
            crate::crypto::aes_gcm::aes256_gcm_encrypt(&key, &nonce, aad.as_bytes(), &plaintext);
        let mut payload = Vec::with_capacity(12 + ciphertext.len());
        payload.extend_from_slice(&nonce);
        payload.extend_from_slice(&ciphertext);
        Ok(reddb_types::Value::Secret(payload))
    }

    fn vault_key_available(&self, collection: &str) -> bool {
        self.vault_encryption_key(collection).is_ok()
    }

    fn vault_encryption_key(&self, collection: &str) -> RedDBResult<[u8; 32]> {
        let auth_store = self.inner.auth_store.read().clone().ok_or_else(|| {
            RedDBError::Query("vault sealed_unavailable: no key provider is configured".to_string())
        })?;
        if !auth_store.is_vault_backed() {
            return Err(RedDBError::Query(
                "vault sealed_unavailable: key provider is sealed".to_string(),
            ));
        }

        if let Some(hex_key) = auth_store.vault_kv_get(&vault_master_key_ref(collection)) {
            return decode_vault_key(&hex_key);
        }
        auth_store.vault_secret_key().ok_or_else(|| {
            RedDBError::Query("vault sealed_unavailable: cluster vault key is missing".to_string())
        })
    }

    fn unseal_vault_value(
        &self,
        collection: &str,
        sealed: &reddb_types::Value,
    ) -> RedDBResult<reddb_types::Value> {
        self.unseal_vault_value_for_tenant(collection, sealed, current_tenant().as_deref())
    }

    fn unseal_vault_value_for_tenant(
        &self,
        collection: &str,
        sealed: &Value,
        tenant: Option<&str>,
    ) -> RedDBResult<Value> {
        let reddb_types::Value::Secret(payload) = sealed else {
            return Err(RedDBError::Query(
                "vault unseal failed: stored value is not sealed".to_string(),
            ));
        };
        if payload.len() < 12 {
            return Err(RedDBError::Query(
                "vault unseal failed: sealed payload is truncated".to_string(),
            ));
        }
        let engine_limit =
            usize::try_from(self.memory_accounting().budget().resolved_bytes).unwrap_or(usize::MAX);
        let limit_bytes = self
            .inner
            .db
            .store()
            .get_collection(collection)
            .and_then(|manager| manager.backing_read_bytes_max())
            .map_or(engine_limit, |limit| limit.min(engine_limit));
        let limit_error = |required_bytes| {
            RedDBError::from(crate::storage::unified::StoreError::ReadLimitExceeded {
                required_bytes,
                limit_bytes,
            })
        };
        // A reveal can retain both the latest entry and its selected version.
        // Admit those cipher copies plus the plaintext buffer before decrypting.
        // This is a per-value allowance, not global/transport admission.
        let cipher_bytes = payload.len().saturating_mul(2);
        let crypto_bytes = cipher_bytes.saturating_add(payload.len());
        if crypto_bytes > limit_bytes {
            return Err(limit_error(crypto_bytes));
        }
        let key = self.vault_encryption_key(collection)?;
        let mut nonce = [0u8; 12];
        nonce.copy_from_slice(&payload[..12]);
        let aad = vault_value_aad(collection, tenant);
        let plaintext = crate::crypto::aes_gcm::aes256_gcm_decrypt(
            &key,
            &nonce,
            aad.as_bytes(),
            &payload[12..],
        )
        .map_err(|_| RedDBError::Query("vault unseal failed: decryption failed".to_string()))?;
        let (value, consumed) =
            reddb_types::Value::from_bytes_with_limit(&plaintext, limit_bytes - cipher_bytes)
                .map_err(|error| match error {
                    reddb_types::ValueError::ReadLimitExceeded { required_bytes, .. } => {
                        limit_error(required_bytes.saturating_add(cipher_bytes))
                    }
                    other => RedDBError::Query(format!(
                        "vault unseal failed: bad plaintext value: {other}"
                    )),
                })?;
        if consumed != plaintext.len() {
            return Err(RedDBError::Query(
                "vault unseal failed: trailing plaintext bytes".to_string(),
            ));
        }
        Ok(value)
    }

    /// Internal peek of a vault entry's unsealed value, bypassing audit and
    /// policy checks. Used by `SecretRefGuard` to detect chained `secret_ref`
    /// indirection at config write time without emitting vault read events.
    /// Returns `Ok(None)` if the entry is absent, tombstoned, or its sealed
    /// payload cannot be decrypted with the available key material.
    pub(crate) fn peek_vault_unsealed(
        &self,
        collection: &str,
        key: &str,
    ) -> RedDBResult<Option<reddb_types::Value>> {
        let ops = KvAtomicOps::new(self);
        let Some(entry) = ops.get_vault_entry(collection, key)? else {
            return Ok(None);
        };
        if entry.tombstone {
            return Ok(None);
        }
        Ok(self.unseal_vault_value(collection, &entry.value).ok())
    }

    fn vault_target_resource(collection: &str, key: &str) -> String {
        if collection == "red.vault" {
            return format!("red.vault/{}", key.to_ascii_lowercase());
        }
        format!("{collection}.{key}")
    }

    fn current_vault_actor() -> String {
        current_auth_identity()
            .map(|(principal, _)| principal)
            .unwrap_or_else(|| "anonymous".to_string())
    }

    fn vault_request_id() -> String {
        let conn_id = current_connection_id();
        if conn_id == 0 {
            "embedded".to_string()
        } else {
            format!("conn-{conn_id}")
        }
    }

    pub(crate) fn check_vault_capability(
        &self,
        action: &str,
        collection: &str,
        key: &str,
    ) -> Result<(), String> {
        self.check_vault_capability_in_tenant(action, collection, key, current_tenant().as_deref())
    }

    fn check_vault_capability_in_tenant(
        &self,
        action: &str,
        collection: &str,
        key: &str,
        owner_tenant: Option<&str>,
    ) -> Result<(), String> {
        let Some(auth_store) = self.inner.auth_store.read().clone() else {
            return Ok(());
        };
        let Some((principal, role)) = current_auth_identity() else {
            if !auth_store.iam_authorization_enabled() {
                return Ok(());
            }
            return Err(
                "IAM authorization is enabled; vault capability check requires an authenticated principal"
                    .to_string(),
            );
        };
        let tenant = current_tenant();
        let principal_id = crate::auth::UserId::from_parts(tenant.as_deref(), &principal);
        let mut resource = crate::auth::policies::ResourceRef::new(
            "vault",
            Self::vault_target_resource(collection, key),
        );
        if let Some(owner_tenant) = owner_tenant {
            resource = resource.with_tenant(owner_tenant.to_string());
        } else if tenant.is_some() {
            resource.name = format!("tenant/platform/{}", resource.name);
        }
        let ctx = crate::auth::policies::EvalContext {
            principal_tenant: tenant.clone(),
            current_tenant: tenant,
            peer_ip: None,
            mfa_present: false,
            now_ms: crate::utils::now_unix_millis() as u128,
            principal_is_admin_role: role == crate::auth::Role::Admin,
            principal_is_platform_scoped: principal_id.tenant.is_none(),
        };
        let allowed = if owner_tenant != ctx.principal_tenant.as_deref() {
            // Legacy admin role is authority inside its own tenant, never an
            // implicit platform or cross-tenant secret grant.
            auth_store.check_policy_authz(&principal_id, action, &resource, &ctx)
        } else {
            auth_store.check_policy_authz_with_role(&principal_id, action, &resource, &ctx, role)
        };
        if allowed {
            Ok(())
        } else {
            Err(format!(
                "principal=`{}` action=`{}` resource=`vault:{}` denied by IAM policy",
                principal,
                action,
                Self::vault_target_resource(collection, key)
            ))
        }
    }

    fn check_system_vault_capability(
        &self,
        action: &str,
        collection: &str,
        key: &str,
    ) -> Result<(), String> {
        self.check_vault_capability(action, collection, key)
    }

    fn audit_vault_unseal(
        &self,
        collection: &str,
        key: &str,
        outcome: crate::runtime::audit_log::Outcome,
        reason: &str,
        entry: Option<&VaultEntry>,
    ) {
        let actor = Self::current_vault_actor();
        let request_id = Self::vault_request_id();
        let mut builder = crate::runtime::audit_log::AuditEvent::builder("vault/reveal")
            .principal(actor.clone())
            .source(crate::runtime::audit_log::AuditAuthSource::Password)
            .resource(format!(
                "vault:{}",
                Self::vault_target_resource(collection, key)
            ))
            .outcome(outcome)
            .correlation_id(request_id.clone())
            .fields([
                crate::runtime::audit_log::AuditFieldEscaper::field("actor", actor),
                crate::runtime::audit_log::AuditFieldEscaper::field("collection", collection),
                crate::runtime::audit_log::AuditFieldEscaper::field("key", key),
                crate::runtime::audit_log::AuditFieldEscaper::field(
                    "target",
                    Self::vault_target_resource(collection, key),
                ),
                crate::runtime::audit_log::AuditFieldEscaper::field("reason", reason),
                crate::runtime::audit_log::AuditFieldEscaper::field("request_id", request_id),
                crate::runtime::audit_log::AuditFieldEscaper::field(
                    "connection_id",
                    current_connection_id(),
                ),
            ]);
        if let Some(tenant) = current_tenant() {
            builder = builder.tenant(tenant);
        }
        if let Some(entry) = entry {
            builder = builder.fields([
                crate::runtime::audit_log::AuditFieldEscaper::field("entity_id", entry.id.raw()),
                crate::runtime::audit_log::AuditFieldEscaper::field(
                    "sequence_id",
                    entry.sequence_id,
                ),
            ]);
        }
        self.audit_log().record_event(builder.build());
    }

    fn audit_vault_lifecycle(
        &self,
        operation: &str,
        collection: &str,
        key: &str,
        outcome: crate::runtime::audit_log::Outcome,
        reason: &str,
        entry: Option<&VaultEntry>,
    ) {
        let actor = Self::current_vault_actor();
        let request_id = Self::vault_request_id();
        let mut builder =
            crate::runtime::audit_log::AuditEvent::builder(format!("vault/{operation}"))
                .principal(actor.clone())
                .source(crate::runtime::audit_log::AuditAuthSource::Password)
                .resource(format!(
                    "vault:{}",
                    Self::vault_target_resource(collection, key)
                ))
                .outcome(outcome)
                .correlation_id(request_id.clone())
                .fields([
                    crate::runtime::audit_log::AuditFieldEscaper::field("actor", actor),
                    crate::runtime::audit_log::AuditFieldEscaper::field("collection", collection),
                    crate::runtime::audit_log::AuditFieldEscaper::field("key", key),
                    crate::runtime::audit_log::AuditFieldEscaper::field(
                        "target",
                        Self::vault_target_resource(collection, key),
                    ),
                    crate::runtime::audit_log::AuditFieldEscaper::field("reason", reason),
                    crate::runtime::audit_log::AuditFieldEscaper::field("request_id", request_id),
                    crate::runtime::audit_log::AuditFieldEscaper::field(
                        "connection_id",
                        current_connection_id(),
                    ),
                ]);
        if let Some(tenant) = current_tenant() {
            builder = builder.tenant(tenant);
        }
        if let Some(entry) = entry {
            builder = builder.fields([
                crate::runtime::audit_log::AuditFieldEscaper::field("entity_id", entry.id.raw()),
                crate::runtime::audit_log::AuditFieldEscaper::field("version", entry.version),
                crate::runtime::audit_log::AuditFieldEscaper::field(
                    "sequence_id",
                    entry.sequence_id,
                ),
            ]);
        }
        self.audit_log().record_event(builder.build());
    }

    fn emit_vault_control_event(
        &self,
        kind: crate::runtime::control_events::EventKind,
        outcome: crate::runtime::control_events::Outcome,
        action: &'static str,
        collection: &str,
        key: &str,
        reason: &str,
        entry: Option<&VaultEntry>,
        extra_fields: Vec<(String, crate::runtime::control_events::Sensitivity)>,
    ) -> RedDBResult<()> {
        use crate::runtime::control_events::{
            ActorRef, ControlEvent, ControlEventCtx, ControlEventLedger, Sensitivity,
        };
        use std::borrow::Cow;

        let tenant = current_tenant();
        let principal = current_auth_identity().map(|(principal, _)| principal);
        let actor_user = principal
            .as_ref()
            .map(|principal| crate::auth::UserId::from_parts(tenant.as_deref(), principal));
        let request_id = Self::vault_request_id();
        let actor = actor_user
            .as_ref()
            .map(ActorRef::User)
            .unwrap_or(ActorRef::Anonymous);
        let ctx = ControlEventCtx {
            actor,
            scope: tenant.as_ref().map(|scope| Cow::Borrowed(scope.as_str())),
            request_id: Some(Cow::Borrowed(request_id.as_str())),
            trace_id: None,
        };

        let target = Self::vault_target_resource(collection, key);
        let mut fields = std::collections::HashMap::new();
        fields.insert("path".to_string(), Sensitivity::raw(target.clone()));
        fields.insert("collection".to_string(), Sensitivity::raw(collection));
        fields.insert("key".to_string(), Sensitivity::raw(key));
        fields.insert(
            "connection_id".to_string(),
            Sensitivity::raw(current_connection_id().to_string()),
        );
        if let Some(entry) = entry {
            fields.insert(
                "entity_id".to_string(),
                Sensitivity::raw(entry.id.raw().to_string()),
            );
            fields.insert(
                "sequence_id".to_string(),
                Sensitivity::raw(entry.sequence_id.to_string()),
            );
            fields.insert(
                "version".to_string(),
                Sensitivity::raw(entry.version.to_string()),
            );
            fields.insert("op".to_string(), Sensitivity::raw(entry.op.clone()));
            fields.insert(
                "tombstone".to_string(),
                Sensitivity::raw(entry.tombstone.to_string()),
            );
            if !entry.tombstone {
                fields.insert(
                    "fingerprint".to_string(),
                    Sensitivity::raw(vault_fingerprint(&entry.value)),
                );
            }
            fields.insert(
                "tags".to_string(),
                Sensitivity::raw(format!("{:?}", vault_tags_value(&entry.metadata))),
            );
        }
        for (key, value) in extra_fields {
            fields.insert(key, value);
        }

        let event = ControlEvent {
            kind,
            outcome,
            action: Cow::Borrowed(action),
            resource: Some(format!("vault:{target}")),
            reason: Some(reason.to_string()),
            matched_policy_id: None,
            fields,
        };
        let ledger = self.inner.control_event_ledger.read();
        match ledger.emit(&ctx, event) {
            Ok(_) => Ok(()),
            Err(err) if self.inner.control_event_config.require_persistence() => {
                Err(RedDBError::Internal(err.to_string()))
            }
            Err(_) => Ok(()),
        }
    }

    pub(crate) fn resolve_vault_secret_value(
        &self,
        collection: &str,
        key: &str,
    ) -> RedDBResult<Value> {
        let ops = KvAtomicOps::new(self);
        let entry = ops.get_vault_entry(collection, key)?;
        if let Err(reason) = self.check_vault_capability("vault:reveal", collection, key) {
            self.audit_vault_unseal(
                collection,
                key,
                crate::runtime::audit_log::Outcome::Denied,
                &reason,
                entry.as_ref(),
            );
            self.emit_vault_control_event(
                crate::runtime::control_events::EventKind::VaultReveal,
                crate::runtime::control_events::Outcome::Denied,
                "vault:reveal",
                collection,
                key,
                &reason,
                entry.as_ref(),
                Vec::new(),
            )?;
            return Err(RedDBError::Query(reason));
        }
        let Some(entry) = entry else {
            let reason = "not_found";
            self.audit_vault_unseal(
                collection,
                key,
                crate::runtime::audit_log::Outcome::Denied,
                reason,
                None,
            );
            self.emit_vault_control_event(
                crate::runtime::control_events::EventKind::VaultReveal,
                crate::runtime::control_events::Outcome::Denied,
                "vault:reveal",
                collection,
                key,
                reason,
                None,
                Vec::new(),
            )?;
            return Err(RedDBError::NotFound(format!(
                "vault secret '{}.{}' not found",
                collection, key
            )));
        };
        if entry.tombstone {
            let reason = "deleted";
            self.audit_vault_unseal(
                collection,
                key,
                crate::runtime::audit_log::Outcome::Denied,
                reason,
                Some(&entry),
            );
            self.emit_vault_control_event(
                crate::runtime::control_events::EventKind::VaultReveal,
                crate::runtime::control_events::Outcome::Denied,
                "vault:reveal",
                collection,
                key,
                reason,
                Some(&entry),
                Vec::new(),
            )?;
            return Err(RedDBError::NotFound(format!(
                "vault secret '{}.{}' is deleted",
                collection, key
            )));
        }
        match self.unseal_vault_value(collection, &entry.value) {
            Ok(value) => {
                self.audit_vault_unseal(
                    collection,
                    key,
                    crate::runtime::audit_log::Outcome::Success,
                    "ok",
                    Some(&entry),
                );
                self.emit_vault_control_event(
                    crate::runtime::control_events::EventKind::VaultReveal,
                    crate::runtime::control_events::Outcome::Allowed,
                    "vault:reveal",
                    collection,
                    key,
                    "ok",
                    Some(&entry),
                    Vec::new(),
                )?;
                Ok(value)
            }
            Err(err) => {
                let reason = err.to_string();
                self.audit_vault_unseal(
                    collection,
                    key,
                    crate::runtime::audit_log::Outcome::Error,
                    &reason,
                    Some(&entry),
                );
                self.emit_vault_control_event(
                    crate::runtime::control_events::EventKind::VaultReveal,
                    crate::runtime::control_events::Outcome::Error,
                    "vault:reveal",
                    collection,
                    key,
                    &reason,
                    Some(&entry),
                    Vec::new(),
                )?;
                Err(err)
            }
        }
    }

    /// Dispatch a `KV PUT / GET / DELETE` command.
    pub fn execute_kv_command(
        &self,
        raw_query: &str,
        cmd: &reddb_rql::ast::KvCommand,
    ) -> RedDBResult<RuntimeQueryResult> {
        use reddb_rql::ast::KvCommand;

        let ops = KvAtomicOps::new(self);

        match cmd {
            KvCommand::Put {
                model,
                collection,
                key,
                value,
                ttl_ms,
                tags,
                if_not_exists,
            } => {
                if *model == crate::catalog::CollectionModel::Vault {
                    self.check_system_vault_capability("vault:write", collection, key)
                        .map_err(RedDBError::Query)?;
                }
                self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
                let (created, id) = ops.set_with_tags_for_model(
                    *model,
                    collection,
                    key,
                    value.clone(),
                    *ttl_ms,
                    tags,
                    *if_not_exists,
                )?;

                let mut result = UnifiedResult::with_columns(vec![
                    "ok".into(),
                    "collection".into(),
                    "key".into(),
                    "id".into(),
                    "created".into(),
                    "tags".into(),
                ]);
                let mut record = UnifiedRecord::new();
                record.set("ok", Value::Boolean(true));
                record.set("collection", Value::text(collection.clone()));
                record.set("key", Value::text(key.clone()));
                record.set("id", Value::Integer(id.raw() as i64));
                record.set("created", Value::Boolean(created));
                record.set("tags", kv_tags_value(tags));
                result.push(record);

                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: if *model == crate::catalog::CollectionModel::Vault {
                        "vault_put"
                    } else {
                        "kv_put"
                    },
                    engine: if *model == crate::catalog::CollectionModel::Vault {
                        "vault"
                    } else {
                        "kv"
                    },
                    result,
                    affected_rows: 1,
                    statement_type: if created { "insert" } else { "update" },
                    bookmark: None,
                    notice: None,
                })
            }
            KvCommand::InvalidateTags { collection, tags } => {
                let invalidated = ops.invalidate_tags(collection, tags)?;

                let mut result = UnifiedResult::with_columns(vec![
                    "ok".into(),
                    "collection".into(),
                    "invalidated".into(),
                    "tags".into(),
                ]);
                let mut record = UnifiedRecord::new();
                record.set("ok", Value::Boolean(true));
                record.set("collection", Value::text(collection.clone()));
                record.set("invalidated", Value::Integer(invalidated as i64));
                record.set("tags", kv_tags_value(tags));
                result.push(record);

                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "kv_invalidate_tags",
                    engine: "kv",
                    result,
                    affected_rows: invalidated as u64,
                    statement_type: "delete",
                    bookmark: None,
                    notice: None,
                })
            }

            KvCommand::Rotate {
                collection,
                key,
                value,
                tags,
            } => {
                self.check_system_vault_capability("vault:write", collection, key)
                    .map_err(RedDBError::Query)?;
                self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
                let entry = ops.append_vault_version(
                    collection,
                    key,
                    value.clone(),
                    "rotate",
                    false,
                    tags,
                )?;
                self.record_kv_watch_event(
                    crate::replication::cdc::ChangeOperation::Update,
                    collection,
                    key,
                    entry.id.raw(),
                    None,
                    Some(vault_entry_metadata_json(&entry)),
                );
                self.audit_vault_lifecycle(
                    "rotate",
                    collection,
                    key,
                    crate::runtime::audit_log::Outcome::Success,
                    "ok",
                    Some(&entry),
                );
                self.emit_vault_control_event(
                    crate::runtime::control_events::EventKind::VaultRotate,
                    crate::runtime::control_events::Outcome::Allowed,
                    "vault:rotate",
                    collection,
                    key,
                    "ok",
                    Some(&entry),
                    Vec::new(),
                )?;
                Ok(vault_write_result(
                    raw_query,
                    "vault_rotate",
                    "update",
                    collection,
                    key,
                    &entry,
                    1,
                ))
            }

            KvCommand::List {
                model,
                collection,
                prefix,
                limit,
                offset,
                as_json,
            } => {
                if *model == crate::catalog::CollectionModel::Vault {
                    if *as_json {
                        return Err(RedDBError::Query(
                            "LIST VAULT AS JSON is not supported; vault list only exposes metadata"
                                .to_string(),
                        ));
                    }
                    let mut entries = ops.latest_vault_entries(collection, prefix.as_deref())?;
                    entries.sort_by(|left, right| left.key.cmp(&right.key));
                    let mut result = UnifiedResult::with_columns(vec![
                        "collection".into(),
                        "key".into(),
                        "version".into(),
                        "fingerprint".into(),
                        "tags".into(),
                        "created_at".into(),
                        "updated_at".into(),
                        "status".into(),
                        "tombstone".into(),
                        "op".into(),
                    ]);
                    let mut visible = Vec::new();
                    for entry in entries {
                        match self.check_vault_capability(
                            "vault:read_metadata",
                            collection,
                            &entry.key,
                        ) {
                            Ok(()) => {
                                self.emit_vault_control_event(
                                    crate::runtime::control_events::EventKind::VaultMetadataRead,
                                    crate::runtime::control_events::Outcome::Allowed,
                                    "vault:read_metadata",
                                    collection,
                                    &entry.key,
                                    "ok",
                                    Some(&entry),
                                    Vec::new(),
                                )?;
                                visible.push(entry);
                            }
                            Err(reason) => {
                                self.emit_vault_control_event(
                                    crate::runtime::control_events::EventKind::VaultMetadataRead,
                                    crate::runtime::control_events::Outcome::Denied,
                                    "vault:read_metadata",
                                    collection,
                                    &entry.key,
                                    &reason,
                                    Some(&entry),
                                    Vec::new(),
                                )?;
                            }
                        }
                    }
                    for entry in visible
                        .into_iter()
                        .skip(*offset)
                        .take(limit.unwrap_or(usize::MAX))
                    {
                        push_vault_metadata_record(&mut result, collection, &entry.key, &entry);
                    }
                    Ok(RuntimeQueryResult {
                        query: raw_query.to_string(),
                        mode: reddb_rql::modes::QueryMode::Sql,
                        statement: "vault_list",
                        engine: "vault",
                        result,
                        affected_rows: 0,
                        statement_type: "select",
                        bookmark: None,
                        notice: None,
                    })
                } else {
                    let mut result = UnifiedResult::with_columns(vec![
                        "rid".into(),
                        "collection".into(),
                        "kind".into(),
                        "tenant".into(),
                        "created_at".into(),
                        "updated_at".into(),
                        "key".into(),
                        "value".into(),
                        "tags".into(),
                    ]);
                    let entries = ops.latest_kv_entries(collection, prefix.as_deref())?;
                    if *as_json {
                        let mut tree = crate::json::Value::Object(crate::json::Map::new());
                        for (key, entity) in entries
                            .into_iter()
                            .skip(*offset)
                            .take(limit.unwrap_or(usize::MAX))
                        {
                            let Some(value) = kv_value_from_entity(&entity) else {
                                continue;
                            };
                            let relative = match prefix {
                                Some(pfx) if key == *pfx => "",
                                Some(pfx) => key
                                    .strip_prefix(pfx.as_str())
                                    .map(|tail| tail.strip_prefix('.').unwrap_or(tail))
                                    .unwrap_or(key.as_str()),
                                None => key.as_str(),
                            };
                            insert_kv_json_path(
                                &mut tree,
                                relative,
                                reddb_types::Value::to_json(&value),
                            );
                        }
                        Ok(kv_list_json_result(raw_query, collection, prefix, tree))
                    } else {
                        for (key, entity) in entries
                            .into_iter()
                            .skip(*offset)
                            .take(limit.unwrap_or(usize::MAX))
                        {
                            let mut record = UnifiedRecord::new();
                            record.set("rid", Value::UnsignedInteger(entity.id.raw()));
                            record.set("collection", Value::text(collection.clone()));
                            record.set("kind", Value::text("kv"));
                            record.set("tenant", Value::Null);
                            record.set("created_at", Value::UnsignedInteger(entity.created_at));
                            record.set("updated_at", Value::UnsignedInteger(entity.updated_at));
                            record.set("key", Value::text(key.clone()));
                            record.set(
                                "value",
                                kv_value_from_entity(&entity).unwrap_or(reddb_types::Value::Null),
                            );
                            record.set("tags", kv_tags_value(&ops.tags_for_key(collection, &key)));
                            result.push(record);
                        }
                        Ok(RuntimeQueryResult {
                            query: raw_query.to_string(),
                            mode: reddb_rql::modes::QueryMode::Sql,
                            statement: "kv_list",
                            engine: "kv",
                            result,
                            affected_rows: 0,
                            statement_type: "select",
                            bookmark: None,
                            notice: None,
                        })
                    }
                }
            }

            KvCommand::History { collection, key } => {
                let latest = ops.get_vault_entry(collection, key)?;
                if let Err(reason) =
                    self.check_vault_capability("vault:read_metadata", collection, key)
                {
                    self.emit_vault_control_event(
                        crate::runtime::control_events::EventKind::VaultMetadataRead,
                        crate::runtime::control_events::Outcome::Denied,
                        "vault:read_metadata",
                        collection,
                        key,
                        &reason,
                        latest.as_ref(),
                        Vec::new(),
                    )?;
                    return Err(RedDBError::Query(reason));
                }
                let versions =
                    super::keyed_spine::history_versions(ops.vault_versions(collection, key)?);
                let result = vault_history_result(collection, key, &versions);
                self.emit_vault_control_event(
                    crate::runtime::control_events::EventKind::VaultMetadataRead,
                    crate::runtime::control_events::Outcome::Allowed,
                    "vault:read_metadata",
                    collection,
                    key,
                    "ok",
                    latest.as_ref(),
                    Vec::new(),
                )?;
                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "vault_history",
                    engine: "vault",
                    result,
                    affected_rows: 0,
                    statement_type: "select",
                    bookmark: None,
                    notice: None,
                })
            }

            KvCommand::Purge { collection, key } => {
                let entry = ops.get_vault_entry(collection, key)?;
                if let Err(reason) = self.check_vault_capability("vault:purge", collection, key) {
                    self.audit_vault_lifecycle(
                        "purge",
                        collection,
                        key,
                        crate::runtime::audit_log::Outcome::Denied,
                        &reason,
                        entry.as_ref(),
                    );
                    self.emit_vault_control_event(
                        crate::runtime::control_events::EventKind::VaultPurge,
                        crate::runtime::control_events::Outcome::Denied,
                        "vault:purge",
                        collection,
                        key,
                        &reason,
                        entry.as_ref(),
                        Vec::new(),
                    )?;
                    return Err(RedDBError::Query(reason));
                }
                self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
                let purged = ops.purge_vault_versions(collection, key)?;
                self.audit_vault_lifecycle(
                    "purge",
                    collection,
                    key,
                    crate::runtime::audit_log::Outcome::Success,
                    "ok",
                    entry.as_ref(),
                );
                self.emit_vault_control_event(
                    crate::runtime::control_events::EventKind::VaultPurge,
                    crate::runtime::control_events::Outcome::Allowed,
                    "vault:purge",
                    collection,
                    key,
                    "ok",
                    entry.as_ref(),
                    vec![(
                        "purged".to_string(),
                        crate::runtime::control_events::Sensitivity::raw(purged.to_string()),
                    )],
                )?;
                let mut result = UnifiedResult::with_columns(vec![
                    "ok".into(),
                    "collection".into(),
                    "key".into(),
                    "purged".into(),
                ]);
                let mut record = UnifiedRecord::new();
                record.set("ok", Value::Boolean(true));
                record.set("collection", Value::text(collection.clone()));
                record.set("key", Value::text(key.clone()));
                record.set("purged", Value::Integer(purged as i64));
                result.push(record);
                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "vault_purge",
                    engine: "vault",
                    result,
                    affected_rows: purged as u64,
                    statement_type: "delete",
                    bookmark: None,
                    notice: None,
                })
            }

            KvCommand::Get {
                model,
                collection,
                key,
            } => {
                if *model == crate::catalog::CollectionModel::Vault {
                    let entry = ops.get_vault_entry(collection, key)?;
                    if let Err(reason) =
                        self.check_vault_capability("vault:read_metadata", collection, key)
                    {
                        self.emit_vault_control_event(
                            crate::runtime::control_events::EventKind::VaultMetadataRead,
                            crate::runtime::control_events::Outcome::Denied,
                            "vault:read_metadata",
                            collection,
                            key,
                            &reason,
                            entry.as_ref(),
                            Vec::new(),
                        )?;
                        return Err(RedDBError::Query(reason));
                    }
                    let key_available = self.vault_key_available(collection);
                    let result =
                        vault_metadata_result(collection, key, entry.as_ref(), key_available);
                    self.emit_vault_control_event(
                        crate::runtime::control_events::EventKind::VaultMetadataRead,
                        crate::runtime::control_events::Outcome::Allowed,
                        "vault:read_metadata",
                        collection,
                        key,
                        "ok",
                        entry.as_ref(),
                        Vec::new(),
                    )?;
                    return Ok(RuntimeQueryResult {
                        query: raw_query.to_string(),
                        mode: reddb_rql::modes::QueryMode::Sql,
                        statement: "vault_get",
                        engine: "vault",
                        result,
                        affected_rows: 0,
                        statement_type: "select",
                        bookmark: None,
                        notice: None,
                    });
                }

                let entry =
                    super::kv_read::KvReadResolver::for_keyed_model(self, *model, collection)?
                        .record(key)?;
                if *model == crate::catalog::CollectionModel::Kv {
                    self.inner.kv_stats.incr_gets();
                }
                let mut result = UnifiedResult::with_columns(vec![
                    "rid".into(),
                    "collection".into(),
                    "kind".into(),
                    "tenant".into(),
                    "created_at".into(),
                    "updated_at".into(),
                    "key".into(),
                    "value".into(),
                    "tags".into(),
                ]);
                let mut record = UnifiedRecord::new();
                if let Some(entry) = entry.as_ref() {
                    record.set("rid", Value::UnsignedInteger(entry.id.raw()));
                    record.set("created_at", Value::UnsignedInteger(entry.created_at));
                    record.set("updated_at", Value::UnsignedInteger(entry.updated_at));
                } else {
                    record.set("rid", Value::Null);
                    record.set("created_at", Value::Null);
                    record.set("updated_at", Value::Null);
                }
                record.set("collection", Value::text(collection.clone()));
                record.set("kind", Value::text(keyed_model_name(*model).to_string()));
                record.set("tenant", Value::Null);
                record.set("key", Value::text(key.clone()));
                record.set(
                    "value",
                    entry.and_then(|entry| entry.value).unwrap_or(Value::Null),
                );
                record.set("tags", kv_tags_value(&ops.tags_for_key(collection, key)));
                result.push(record);

                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "kv_get",
                    engine: "kv",
                    result,
                    affected_rows: 0,
                    statement_type: "select",
                    bookmark: None,
                    notice: None,
                })
            }
            KvCommand::Watch {
                model,
                collection,
                key,
                prefix,
                from_lsn,
            } => {
                let watch_key = if *prefix {
                    format!("{key}.*")
                } else {
                    key.clone()
                };
                let endpoint = match from_lsn {
                    Some(lsn) => format!(
                        "/collections/{collection}/{}/{watch_key}/watch?since_lsn={lsn}",
                        keyed_model_name(*model)
                    ),
                    None => format!(
                        "/collections/{collection}/{}/{watch_key}/watch",
                        keyed_model_name(*model)
                    ),
                };
                let mut result = UnifiedResult::with_columns(vec![
                    "collection".into(),
                    "key".into(),
                    "prefix".into(),
                    "from_lsn".into(),
                    "watch_url".into(),
                    "streaming".into(),
                ]);
                let mut record = UnifiedRecord::new();
                record.set("collection", Value::text(collection.clone()));
                record.set("key", Value::text(watch_key));
                record.set("prefix", Value::Boolean(*prefix));
                record.set(
                    "from_lsn",
                    from_lsn
                        .map(Value::UnsignedInteger)
                        .unwrap_or(reddb_types::Value::Null),
                );
                record.set("watch_url", Value::text(endpoint));
                record.set("streaming", Value::Boolean(true));
                result.push(record);

                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "kv_watch",
                    engine: keyed_model_name(*model),
                    result,
                    affected_rows: 0,
                    statement_type: "stream",
                    bookmark: None,
                    notice: None,
                })
            }

            KvCommand::Reveal {
                collection,
                key,
                version,
            } => {
                let latest = ops.get_vault_entry(collection, key)?;
                let entry = match version {
                    Some(version) => ops.get_vault_entry_version(collection, key, *version)?,
                    None => latest.clone(),
                };
                let action = match (version, latest.as_ref()) {
                    (Some(requested), Some(latest)) if *requested == latest.version => {
                        "vault:reveal"
                    }
                    (Some(_), _) => "vault:reveal_history",
                    _ => "vault:reveal",
                };
                let event_kind = crate::runtime::control_events::EventKind::VaultReveal;
                if let Err(reason) = self.check_vault_capability(action, collection, key) {
                    self.audit_vault_unseal(
                        collection,
                        key,
                        crate::runtime::audit_log::Outcome::Denied,
                        &reason,
                        entry.as_ref(),
                    );
                    self.emit_vault_control_event(
                        event_kind,
                        crate::runtime::control_events::Outcome::Denied,
                        action,
                        collection,
                        key,
                        &reason,
                        entry.as_ref(),
                        Vec::new(),
                    )?;
                    return Err(RedDBError::Query(reason));
                }
                let Some(entry) = entry else {
                    let reason = "not_found";
                    self.audit_vault_unseal(
                        collection,
                        key,
                        crate::runtime::audit_log::Outcome::Denied,
                        reason,
                        None,
                    );
                    self.emit_vault_control_event(
                        event_kind,
                        crate::runtime::control_events::Outcome::Denied,
                        action,
                        collection,
                        key,
                        reason,
                        None,
                        Vec::new(),
                    )?;
                    return Err(RedDBError::NotFound(format!(
                        "vault secret '{}.{}' not found",
                        collection, key
                    )));
                };
                if entry.tombstone {
                    let reason = "deleted";
                    self.audit_vault_unseal(
                        collection,
                        key,
                        crate::runtime::audit_log::Outcome::Denied,
                        reason,
                        Some(&entry),
                    );
                    self.emit_vault_control_event(
                        event_kind,
                        crate::runtime::control_events::Outcome::Denied,
                        action,
                        collection,
                        key,
                        reason,
                        Some(&entry),
                        Vec::new(),
                    )?;
                    return Err(RedDBError::NotFound(format!(
                        "vault secret '{}.{}' is deleted",
                        collection, key
                    )));
                }
                match self.unseal_vault_value(collection, &entry.value) {
                    Ok(value) => {
                        self.audit_vault_unseal(
                            collection,
                            key,
                            crate::runtime::audit_log::Outcome::Success,
                            "ok",
                            Some(&entry),
                        );
                        self.emit_vault_control_event(
                            event_kind,
                            crate::runtime::control_events::Outcome::Allowed,
                            action,
                            collection,
                            key,
                            "ok",
                            Some(&entry),
                            Vec::new(),
                        )?;
                        let mut result = UnifiedResult::with_columns(vec![
                            "collection".into(),
                            "key".into(),
                            "value".into(),
                        ]);
                        let mut record = UnifiedRecord::new();
                        record.set("collection", Value::text(collection.clone()));
                        record.set("key", Value::text(key.clone()));
                        record.set("value", value);
                        result.push(record);
                        Ok(RuntimeQueryResult {
                            query: raw_query.to_string(),
                            mode: reddb_rql::modes::QueryMode::Sql,
                            statement: "vault_reveal",
                            engine: "vault",
                            result,
                            affected_rows: 0,
                            statement_type: "select",
                            bookmark: None,
                            notice: None,
                        })
                    }
                    Err(err) => {
                        let reason = err.to_string();
                        self.audit_vault_unseal(
                            collection,
                            key,
                            crate::runtime::audit_log::Outcome::Error,
                            &reason,
                            Some(&entry),
                        );
                        self.emit_vault_control_event(
                            event_kind,
                            crate::runtime::control_events::Outcome::Error,
                            action,
                            collection,
                            key,
                            &reason,
                            Some(&entry),
                            Vec::new(),
                        )?;
                        Err(err)
                    }
                }
            }

            KvCommand::Incr {
                model,
                collection,
                key,
                by,
                ttl_ms,
            } => {
                self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
                let new_value = ops.incr(*model, collection, key, *by, *ttl_ms)?;

                let mut result = UnifiedResult::with_columns(vec![
                    "ok".into(),
                    "collection".into(),
                    "key".into(),
                    "value".into(),
                ]);
                let mut record = UnifiedRecord::new();
                record.set("ok", Value::Boolean(true));
                record.set("collection", Value::text(collection.clone()));
                record.set("key", Value::text(key.clone()));
                record.set("value", Value::Integer(new_value));
                result.push(record);

                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "kv_incr",
                    engine: "kv",
                    result,
                    affected_rows: 1,
                    statement_type: "update",
                    bookmark: None,
                    notice: None,
                })
            }

            KvCommand::Cas {
                model,
                collection,
                key,
                expected,
                new_value,
                ttl_ms,
            } => {
                self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
                let (ok, current) = ops.cas(
                    *model,
                    collection,
                    key,
                    expected.as_ref(),
                    new_value.clone(),
                    *ttl_ms,
                )?;

                let mut result = UnifiedResult::with_columns(vec![
                    "ok".into(),
                    "collection".into(),
                    "key".into(),
                    "current".into(),
                ]);
                let mut record = UnifiedRecord::new();
                record.set("ok", Value::Boolean(ok));
                record.set("collection", Value::text(collection.clone()));
                record.set("key", Value::text(key.clone()));
                record.set("current", current.unwrap_or(reddb_types::Value::Null));
                result.push(record);

                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "kv_cas",
                    engine: "kv",
                    result,
                    affected_rows: if ok { 1 } else { 0 },
                    statement_type: "update",
                    bookmark: None,
                    notice: None,
                })
            }

            KvCommand::Delete {
                model,
                collection,
                key,
            } => {
                if *model == crate::catalog::CollectionModel::Vault {
                    self.check_system_vault_capability("vault:write", collection, key)
                        .map_err(RedDBError::Query)?;
                    self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
                    let entry = ops.append_vault_version(
                        collection,
                        key,
                        Value::Null,
                        "delete",
                        true,
                        &[],
                    )?;
                    self.record_kv_watch_event(
                        crate::replication::cdc::ChangeOperation::Delete,
                        collection,
                        key,
                        entry.id.raw(),
                        None,
                        Some(vault_entry_metadata_json(&entry)),
                    );
                    self.audit_vault_lifecycle(
                        "delete",
                        collection,
                        key,
                        crate::runtime::audit_log::Outcome::Success,
                        "ok",
                        Some(&entry),
                    );
                    return Ok(vault_write_result(
                        raw_query,
                        "vault_delete",
                        "delete",
                        collection,
                        key,
                        &entry,
                        1,
                    ));
                }
                self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
                let deleted = ops.delete(*model, collection, key)?;

                let mut result = UnifiedResult::with_columns(vec![
                    "ok".into(),
                    "collection".into(),
                    "key".into(),
                    "deleted".into(),
                ]);
                let mut record = UnifiedRecord::new();
                record.set("ok", Value::Boolean(true));
                record.set("collection", Value::text(collection.clone()));
                record.set("key", Value::text(key.clone()));
                record.set("deleted", Value::Boolean(deleted));
                result.push(record);

                Ok(RuntimeQueryResult {
                    query: raw_query.to_string(),
                    mode: reddb_rql::modes::QueryMode::Sql,
                    statement: "kv_delete",
                    engine: "kv",
                    result,
                    affected_rows: if deleted { 1 } else { 0 },
                    statement_type: "delete",
                    bookmark: None,
                    notice: None,
                })
            }
        }
    }

    pub fn vault_watch_events_since(
        &self,
        collection: &str,
        key: &str,
        since_lsn: u64,
        max_count: usize,
    ) -> Vec<crate::replication::cdc::KvWatchEvent> {
        self.kv_watch_events_since(collection, key, since_lsn, max_count)
            .into_iter()
            .filter(|event| {
                vault_watch_in_scope(event, current_tenant().as_deref())
                    && self
                        .check_vault_capability(
                            "vault:read_metadata",
                            &event.collection,
                            &event.key,
                        )
                        .is_ok()
            })
            .map(vault_filter_watch_event)
            .collect()
    }

    pub fn vault_watch_events_since_prefix(
        &self,
        collection: &str,
        prefix: &str,
        since_lsn: u64,
        max_count: usize,
    ) -> Vec<crate::replication::cdc::KvWatchEvent> {
        self.kv_watch_events_since_prefix(collection, prefix, since_lsn, max_count)
            .into_iter()
            .filter(|event| {
                vault_watch_in_scope(event, current_tenant().as_deref())
                    && self
                        .check_vault_capability(
                            "vault:read_metadata",
                            &event.collection,
                            &event.key,
                        )
                        .is_ok()
            })
            .map(vault_filter_watch_event)
            .collect()
    }

    fn check_kv_invalidate_policy(&self, collection: &str) -> RedDBResult<()> {
        let auth_store = match self.inner.auth_store.read().clone() {
            Some(store) => store,
            None => return Ok(()),
        };
        let (username, role) = match crate::runtime::impl_core::current_auth_identity() {
            Some(identity) => identity,
            None => return Ok(()),
        };
        if role < crate::auth::Role::Write {
            return Err(RedDBError::Query(format!(
                "principal=`{username}` role=`{role:?}` cannot invalidate KV tags"
            )));
        }
        if !auth_store.iam_authorization_enabled() {
            return Ok(());
        }

        let tenant = crate::runtime::impl_core::current_tenant();
        let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
        let mut resource =
            crate::auth::policies::ResourceRef::new("kv".to_string(), collection.to_string());
        if let Some(tenant) = tenant.clone() {
            resource = resource.with_tenant(tenant);
        }
        let ctx = crate::auth::policies::EvalContext {
            principal_tenant: tenant.clone(),
            current_tenant: tenant,
            peer_ip: None,
            mfa_present: false,
            now_ms: current_unix_ms(),
            principal_is_admin_role: role == crate::auth::Role::Admin,
            principal_is_platform_scoped: principal.tenant.is_none(),
        };
        if auth_store.check_policy_authz_with_role(
            &principal,
            "kv:invalidate",
            &resource,
            &ctx,
            role,
        ) {
            Ok(())
        } else {
            Err(RedDBError::Query(format!(
                "principal=`{username}` action=`kv:invalidate` resource=`kv:{collection}` denied by IAM policy"
            )))
        }
    }
}

fn ttl_metadata(ttl_ms: Option<u64>) -> Option<Metadata> {
    let ttl_ms = ttl_ms?;
    Some(Metadata::with_fields(
        [(
            "_ttl_ms".to_string(),
            if ttl_ms <= i64::MAX as u64 {
                MetadataValue::Int(ttl_ms as i64)
            } else {
                MetadataValue::Timestamp(ttl_ms)
            },
        )]
        .into_iter()
        .collect(),
    ))
}

fn vault_write_result(
    raw_query: &str,
    statement: &'static str,
    statement_type: &'static str,
    collection: &str,
    key: &str,
    entry: &VaultEntry,
    affected_rows: u64,
) -> RuntimeQueryResult {
    let mut result = UnifiedResult::with_columns(vec![
        "ok".into(),
        "collection".into(),
        "key".into(),
        "version".into(),
        "fingerprint".into(),
        "tombstone".into(),
        "op".into(),
        "id".into(),
    ]);
    let mut record = UnifiedRecord::new();
    record.set("ok", Value::Boolean(true));
    record.set("collection", Value::text(collection.to_string()));
    record.set("key", Value::text(key.to_string()));
    record.set("version", Value::Integer(entry.version));
    if entry.tombstone {
        record.set("fingerprint", Value::Null);
    } else {
        record.set("fingerprint", Value::text(vault_fingerprint(&entry.value)));
    }
    record.set("tombstone", Value::Boolean(entry.tombstone));
    record.set("op", Value::text(entry.op.clone()));
    record.set("id", Value::Integer(entry.id.raw() as i64));
    result.push(record);
    RuntimeQueryResult {
        query: raw_query.to_string(),
        mode: reddb_rql::modes::QueryMode::Sql,
        statement,
        engine: "vault",
        result,
        affected_rows,
        statement_type,
        bookmark: None,
        notice: None,
    }
}

fn vault_history_result(collection: &str, key: &str, versions: &[VaultEntry]) -> UnifiedResult {
    let mut result = UnifiedResult::with_columns(vec![
        "collection".into(),
        "key".into(),
        "version".into(),
        "fingerprint".into(),
        "tags".into(),
        "created_at".into(),
        "updated_at".into(),
        "status".into(),
        "tombstone".into(),
        "op".into(),
    ]);
    for entry in versions {
        push_vault_metadata_record(&mut result, collection, key, entry);
    }
    result
}

fn push_vault_metadata_record(
    result: &mut UnifiedResult,
    collection: &str,
    key: &str,
    entry: &VaultEntry,
) {
    let mut record = UnifiedRecord::new();
    record.set("collection", Value::text(collection.to_string()));
    record.set("key", Value::text(key.to_string()));
    record.set("version", Value::Integer(entry.version));
    if entry.tombstone {
        record.set("fingerprint", Value::Null);
        record.set("status", Value::text("deleted"));
    } else {
        record.set("fingerprint", Value::text(vault_fingerprint(&entry.value)));
        record.set("status", Value::text("sealed"));
    }
    record.set("tags", vault_tags_value(&entry.metadata));
    record.set("created_at", Value::TimestampMs(entry.created_at as i64));
    record.set("updated_at", Value::TimestampMs(entry.updated_at as i64));
    record.set("tombstone", Value::Boolean(entry.tombstone));
    record.set("op", Value::text(entry.op.clone()));
    result.push(record);
}

fn vault_metadata_result(
    collection: &str,
    key: &str,
    entry: Option<&VaultEntry>,
    key_available: bool,
) -> UnifiedResult {
    let mut result = UnifiedResult::with_columns(vec![
        "collection".into(),
        "key".into(),
        "version".into(),
        "fingerprint".into(),
        "tags".into(),
        "created_at".into(),
        "updated_at".into(),
        "value".into(),
        "status".into(),
        "tombstone".into(),
        "op".into(),
    ]);
    let mut record = UnifiedRecord::new();
    record.set("collection", Value::text(collection.to_string()));
    record.set("key", Value::text(key.to_string()));
    match entry {
        Some(entry) => {
            record.set("version", Value::Integer(entry.version));
            if entry.tombstone {
                record.set("fingerprint", Value::Null);
            } else {
                record.set("fingerprint", Value::text(vault_fingerprint(&entry.value)));
            }
            record.set("tags", vault_tags_value(&entry.metadata));
            record.set("created_at", Value::TimestampMs(entry.created_at as i64));
            record.set("updated_at", Value::TimestampMs(entry.updated_at as i64));
            record.set("value", Value::text("***"));
            record.set(
                "status",
                Value::text(if entry.tombstone {
                    "deleted"
                } else if key_available {
                    "sealed"
                } else {
                    "sealed_unavailable"
                }),
            );
            record.set("tombstone", Value::Boolean(entry.tombstone));
            record.set("op", Value::text(entry.op.clone()));
        }
        None => {
            record.set("version", Value::Null);
            record.set("fingerprint", Value::Null);
            record.set("tags", Value::Array(Vec::new()));
            record.set("created_at", Value::Null);
            record.set("updated_at", Value::Null);
            record.set("value", Value::text(""));
            record.set("status", Value::text("missing"));
            record.set("tombstone", Value::Boolean(false));
            record.set("op", Value::Null);
        }
    }
    result.push(record);
    result
}

fn vault_fingerprint(value: &Value) -> String {
    match value {
        Value::Secret(payload) => crate::utils::to_hex(&crate::crypto::sha256::sha256(payload)),
        other => crate::utils::to_hex(&crate::crypto::sha256::sha256(&other.to_bytes())),
    }
}

fn vault_entry_metadata_json(entry: &VaultEntry) -> crate::json::Value {
    let mut object = crate::json::Map::new();
    object.insert(
        "key".to_string(),
        crate::json::Value::String(entry.key.clone()),
    );
    object.insert(
        "version".to_string(),
        crate::json::Value::Number(entry.version as f64),
    );
    object.insert(
        "fingerprint".to_string(),
        if entry.tombstone {
            crate::json::Value::Null
        } else {
            crate::json::Value::String(vault_fingerprint(&entry.value))
        },
    );
    object.insert(
        "tenant".to_string(),
        entry
            .tenant
            .as_ref()
            .map(|tenant| crate::json::Value::String(tenant.clone()))
            .unwrap_or(crate::json::Value::Null),
    );
    object.insert("tags".to_string(), vault_tags_json(&entry.metadata));
    object.insert(
        "actor".to_string(),
        crate::json::Value::String(RedDBRuntime::current_vault_actor()),
    );
    object.insert(
        "sequence_id".to_string(),
        crate::json::Value::Number(entry.sequence_id as f64),
    );
    object.insert(
        "tombstone".to_string(),
        crate::json::Value::Bool(entry.tombstone),
    );
    object.insert(
        "op".to_string(),
        crate::json::Value::String(entry.op.clone()),
    );
    crate::json::Value::Object(object)
}

fn vault_tags_json(metadata: &Metadata) -> crate::json::Value {
    match vault_tags_value(metadata) {
        Value::Array(values) => crate::json::Value::Array(
            values
                .into_iter()
                .filter_map(|value| match value {
                    Value::Text(tag) => Some(crate::json::Value::String(tag.to_string())),
                    _ => None,
                })
                .collect(),
        ),
        _ => crate::json::Value::Array(Vec::new()),
    }
}

fn vault_tags_metadata(tags: &[String]) -> std::collections::HashMap<String, MetadataValue> {
    [(
        "tags".to_string(),
        MetadataValue::Array(
            tags.iter()
                .map(|tag| MetadataValue::String(tag.clone()))
                .collect(),
        ),
    )]
    .into_iter()
    .collect()
}

fn vault_filter_watch_event(
    mut event: crate::replication::cdc::KvWatchEvent,
) -> crate::replication::cdc::KvWatchEvent {
    event.before = event.before.and_then(vault_metadata_json_only);
    event.after = event.after.and_then(vault_metadata_json_only);
    event
}

fn vault_metadata_json_only(value: crate::json::Value) -> Option<crate::json::Value> {
    let object = value.as_object()?;
    let mut out = crate::json::Map::new();
    for field in [
        "key",
        "version",
        "fingerprint",
        "tags",
        "actor",
        "sequence_id",
        "tombstone",
        "op",
    ] {
        if let Some(value) = object.get(field) {
            out.insert(field.to_string(), value.clone());
        }
    }
    Some(crate::json::Value::Object(out))
}

fn vault_tags_value(metadata: &Metadata) -> Value {
    match metadata.get("tags") {
        Some(MetadataValue::Array(values)) => Value::Array(
            values
                .iter()
                .filter_map(|value| match value {
                    MetadataValue::String(tag) => Some(Value::text(tag.clone())),
                    _ => None,
                })
                .collect(),
        ),
        Some(MetadataValue::String(tag)) if !tag.is_empty() => {
            Value::Array(vec![Value::text(tag.clone())])
        }
        _ => Value::Array(Vec::new()),
    }
}

fn decode_vault_key(hex_key: &str) -> RedDBResult<[u8; 32]> {
    let bytes = hex::decode(hex_key)
        .map_err(|_| RedDBError::Query("vault sealed_unavailable: bad key material".to_string()))?;
    let key: [u8; 32] = bytes.try_into().map_err(|_| {
        RedDBError::Query("vault sealed_unavailable: bad key material length".to_string())
    })?;
    Ok(key)
}

fn kv_tags_metadata(tags: &[String]) -> Option<(String, MetadataValue)> {
    if tags.is_empty() {
        return None;
    }
    let values = tags
        .iter()
        .map(|tag| MetadataValue::String(tag.clone()))
        .collect();
    Some(("_kv_tags".to_string(), MetadataValue::Array(values)))
}

impl RedDBRuntime {
    pub(crate) fn rehydrate_kv_tags(&self) -> RedDBResult<()> {
        let db = self.db();
        let store = db.store();
        for contract in db.collection_contracts() {
            if contract.declared_model != CollectionModel::Kv {
                continue;
            }
            let Some(manager) = store.get_collection(&contract.name) else {
                continue;
            };
            let resolver = super::kv_read::KvReadResolver::for_keyed_model(
                self,
                CollectionModel::Kv,
                &contract.name,
            )?;
            for entity in manager.scan(None, |_| true) {
                let Some(reddb_types::Value::Text(key)) =
                    entity.data.as_row().and_then(|row| row.get_field("key"))
                else {
                    continue;
                };
                let Some(metadata) = manager.get_metadata(entity.id) else {
                    continue;
                };
                let Some(MetadataValue::Array(tags)) = metadata.fields.get("_kv_tags") else {
                    continue;
                };
                if resolver
                    .record(key)?
                    .is_none_or(|record| record.id != entity.id)
                {
                    continue;
                }
                let tags: Vec<_> = tags
                    .iter()
                    .filter_map(|tag| match tag {
                        MetadataValue::String(tag) => Some(tag.clone()),
                        _ => None,
                    })
                    .collect();
                self.inner
                    .kv_tag_index
                    .replace(&contract.name, key, entity.id, &tags);
            }
        }
        Ok(())
    }
}

fn kv_tags_value(tags: &[String]) -> Value {
    let json = crate::json::Value::Array(
        tags.iter()
            .map(|tag| crate::json::Value::String(tag.clone()))
            .collect(),
    );
    Value::Json(crate::json::to_vec(&json).unwrap_or_default())
}

pub(super) fn kv_value_from_entity(entity: &crate::storage::UnifiedEntity) -> Option<Value> {
    if let crate::storage::EntityData::Row(ref row) = entity.data {
        return row.get_field("value").cloned();
    }
    None
}

/// Whether a KV row entity carries the logical `key`.
fn kv_entity_has_key(entity: &crate::storage::UnifiedEntity, key: &str) -> bool {
    if let crate::storage::EntityData::Row(ref row) = entity.data {
        if let Some(reddb_types::Value::Text(ref k)) = row.get_field("key") {
            return &**k == key;
        }
    }
    false
}

fn insert_kv_json_path(root: &mut crate::json::Value, path: &str, value: crate::json::Value) {
    let segments: Vec<&str> = path
        .split('.')
        .filter(|segment| !segment.is_empty())
        .collect();
    insert_kv_json_segments(root, &segments, value);
}

fn insert_kv_json_segments(
    root: &mut crate::json::Value,
    segments: &[&str],
    value: crate::json::Value,
) {
    if segments.is_empty() {
        *root = value;
        return;
    }

    if !matches!(root, crate::json::Value::Object(_)) {
        *root = crate::json::Value::Object(crate::json::Map::new());
    }

    let crate::json::Value::Object(map) = root else {
        return;
    };
    if segments.len() == 1 {
        map.insert(segments[0].to_string(), value);
        return;
    }
    let entry = map
        .entry(segments[0].to_string())
        .or_insert_with(|| crate::json::Value::Object(crate::json::Map::new()));
    insert_kv_json_segments(entry, &segments[1..], value);
}

fn kv_list_json_result(
    raw_query: &str,
    collection: &str,
    prefix: &Option<String>,
    value: crate::json::Value,
) -> RuntimeQueryResult {
    let mut result =
        UnifiedResult::with_columns(vec!["collection".into(), "prefix".into(), "value".into()]);
    let mut record = UnifiedRecord::new();
    record.set("collection", Value::text(collection.to_string()));
    record.set(
        "prefix",
        prefix
            .as_ref()
            .map(|prefix| Value::text(prefix.clone()))
            .unwrap_or(Value::Null),
    );
    record.set("value", Value::Json(value.to_string_compact().into_bytes()));
    result.push(record);
    RuntimeQueryResult {
        query: raw_query.to_string(),
        mode: reddb_rql::modes::QueryMode::Sql,
        statement: "kv_list_json",
        engine: "kv",
        result,
        affected_rows: 0,
        statement_type: "select",
        bookmark: None,
        notice: None,
    }
}

fn kv_collection_contract(name: &str) -> crate::physical::CollectionContract {
    let now = current_unix_ms();
    crate::physical::CollectionContract {
        name: name.to_string(),
        declared_model: crate::catalog::CollectionModel::Kv,
        schema_mode: crate::catalog::SchemaMode::Dynamic,
        origin: crate::physical::ContractOrigin::Implicit,
        version: 1,
        created_at_unix_ms: now,
        updated_at_unix_ms: now,
        default_ttl_ms: None,
        vector_dimension: None,
        vector_metric: None,
        context_index_fields: Vec::new(),
        declared_columns: Vec::new(),
        table_def: None,
        timestamps_enabled: false,
        context_index_enabled: false,
        metrics_raw_retention_ms: None,
        metrics_rollup_policies: Vec::new(),
        metrics_tenant_identity: None,
        metrics_namespace: None,
        append_only: false,
        subscriptions: Vec::new(),
        analytics_config: Vec::new(),
        session_key: None,
        session_gap_ms: None,
        retention_duration_ms: None,
        analytical_storage: None,

        ai_policy: None,
    }
}

fn current_unix_ms() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
}

#[cfg(test)]
mod tests {
    use crate::api::RedDBOptions;
    use crate::application::ports::RuntimeEntityPort;
    use crate::auth::Role;
    use crate::catalog::CollectionModel;
    use crate::runtime::execution_context::{with_snapshot_bundle, SnapshotBundle};
    use crate::runtime::RedDBRuntime;
    use reddb_types::Value;

    fn rt() -> RedDBRuntime {
        RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("in-memory runtime")
    }

    #[test]
    fn keyed_overwrite_and_metadata_share_one_durable_publication() {
        let directory = tempfile::tempdir().expect("directory");
        let path = directory.path().join("atomic.rdb");
        {
            let runtime = RedDBRuntime::with_options(
                RedDBOptions::persistent(&path)
                    .with_durability_mode(crate::api::DurabilityMode::Strict),
            )
            .expect("runtime");
            let ops = super::KvAtomicOps::new(&runtime);
            ops.set(
                CollectionModel::Kv,
                "keys",
                "item",
                Value::Integer(1),
                None,
                false,
            )
            .expect("seed");
            let before = crate::storage::EmbeddedRdbArtifact::open(&path).expect("before");
            ops.set_with_tags(
                "keys",
                "item",
                Value::Integer(2),
                Some(600_000),
                &["daily".into()],
                false,
            )
            .expect("replace");
            let after = crate::storage::EmbeddedRdbArtifact::open(&path).expect("after");
            assert_eq!(
                after.selected_superblock.generation,
                before.selected_superblock.generation + 1
            );
            assert_eq!(
                runtime.db().store().get_collection("keys").unwrap().count(),
                1
            );
        }
        let runtime = RedDBRuntime::with_options(RedDBOptions::persistent(&path)).expect("reopen");
        let ops = super::KvAtomicOps::new(&runtime);
        assert_eq!(
            ops.get(CollectionModel::Kv, "keys", "item").expect("read"),
            Some(Value::Integer(2))
        );
        assert_eq!(
            ops.existing_ttl_ms("keys", "item").expect("TTL read"),
            Some(600_000)
        );
        assert_eq!(
            runtime.inner.kv_tag_index.tags_for_key("keys", "item"),
            vec!["daily".to_string()]
        );
    }

    #[test]
    fn kv_get_projection_preserves_metadata_tags_and_missing_envelope() {
        let runtime = rt();
        let ops = super::KvAtomicOps::new(&runtime);
        let value = reddb_types::Value::Array(vec![reddb_types::Value::Integer(7)]);
        let (_, id) = ops
            .set_with_tags(
                "projected",
                "item",
                value.clone(),
                None,
                &["daily".into()],
                false,
            )
            .expect("seed");
        let db = runtime.db();
        let store = db.store();
        let manager = store.get_collection("projected").expect("collection");
        let entity = manager.get(id).expect("stored row");
        manager.force_seal().expect("seal row");
        let result = runtime.execute_query("KV GET projected.item").expect("GET");
        let record = &result.result.records[0];
        assert_eq!(record.iter_fields().count(), 9);
        for (field, expected) in [
            ("rid", reddb_types::Value::UnsignedInteger(id.raw())),
            (
                "created_at",
                reddb_types::Value::UnsignedInteger(entity.created_at),
            ),
            (
                "updated_at",
                reddb_types::Value::UnsignedInteger(entity.updated_at),
            ),
            ("collection", reddb_types::Value::text("projected")),
            ("kind", reddb_types::Value::text("kv")),
            ("tenant", reddb_types::Value::Null),
            ("key", reddb_types::Value::text("item")),
            ("value", value),
            ("tags", super::kv_tags_value(&["daily".into()])),
        ] {
            assert_eq!(record.get(field), Some(&expected), "{field}");
        }
        let missing = runtime
            .execute_query("KV GET projected.absent")
            .expect("missing GET");
        let record = &missing.result.records[0];
        for field in ["rid", "created_at", "updated_at", "value", "tenant"] {
            assert_eq!(
                record.get(field),
                Some(&reddb_types::Value::Null),
                "{field}"
            );
        }
        assert_eq!(record.get("tags"), Some(&super::kv_tags_value(&[])));
    }

    #[test]
    fn kv_key_lookup_rebuilds_after_reopen() {
        let directory = tempfile::tempdir().expect("temporary database");
        let path = directory.path().join("keys.rdb");
        {
            let runtime =
                RedDBRuntime::with_options(RedDBOptions::persistent(&path)).expect("open");
            for statement in [
                "KV PUT indexed.alpha = 'first'",
                "KV PUT indexed.beta = 'second'",
                "KV PUT indexed.alpha = 'updated'",
                "KV DELETE indexed.beta",
                "KV PUT indexed.gamma = 'third'",
            ] {
                runtime.execute_query(statement).expect("KV mutation");
            }
        }
        let runtime = RedDBRuntime::with_options(RedDBOptions::persistent(&path)).expect("reopen");
        let ops = super::KvAtomicOps::new(&runtime);
        for (key, expected) in [
            ("alpha", Some("updated")),
            ("beta", None),
            ("gamma", Some("third")),
            ("absent", None),
        ] {
            assert_eq!(
                ops.get(CollectionModel::Kv, "indexed", key).expect("read"),
                expected.map(reddb_types::Value::text)
            );
        }
        ops.set(
            CollectionModel::Kv,
            "indexed",
            "beta",
            reddb_types::Value::text("reused"),
            None,
            false,
        )
        .expect("reuse deleted key");
        assert_eq!(
            ops.get(CollectionModel::Kv, "indexed", "beta")
                .expect("read reused key"),
            Some(reddb_types::Value::text("reused"))
        );
    }

    #[test]
    fn concurrent_put_if_not_exists_has_exactly_one_winner() {
        // `IF NOT EXISTS` is the primitive callers build locks and once-only
        // initialisation on, so "did I insert it?" must be true for exactly
        // one racer. It reads the key, decides, then writes — the same
        // read-modify-write shape as `incr`, and it needs the same per-key
        // lock; without one, every thread could observe "absent" and claim
        // the insert.
        const THREADS: usize = 8;

        let runtime = std::sync::Arc::new(rt());
        let barrier = std::sync::Arc::new(std::sync::Barrier::new(THREADS));
        let winners = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let mut handles = Vec::new();

        for i in 0..THREADS {
            let runtime = std::sync::Arc::clone(&runtime);
            let barrier = std::sync::Arc::clone(&barrier);
            let winners = std::sync::Arc::clone(&winners);
            handles.push(std::thread::spawn(move || {
                let ops = super::KvAtomicOps::new(&runtime);
                barrier.wait();
                let (inserted, _) = ops
                    .set_with_tags_and_metadata_for_model(
                        CollectionModel::Kv,
                        "kv_default",
                        "once",
                        reddb_types::Value::Integer(i as i64),
                        None,
                        &[],
                        true,
                        Vec::new(),
                    )
                    .expect("put should succeed");
                if inserted {
                    winners.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            }));
        }
        for handle in handles {
            handle.join().expect("worker should finish");
        }

        assert_eq!(
            winners.load(std::sync::atomic::Ordering::Relaxed),
            1,
            "exactly one racer may report having inserted the key"
        );
    }

    #[test]
    fn incr_missing_key_initialises_at_by() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        let v = ops
            .incr(CollectionModel::Kv, "kv_default", "missing", 5, None)
            .unwrap();
        assert_eq!(v, 5);
    }

    #[test]
    fn kv_runtime_stats_count_public_ops() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);

        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "profile",
            reddb_types::Value::text("alice"),
            None,
            false,
        )
        .unwrap();
        ops.get(CollectionModel::Kv, "kv_default", "profile")
            .unwrap();
        ops.delete(CollectionModel::Kv, "kv_default", "profile")
            .unwrap();
        ops.incr(CollectionModel::Kv, "kv_default", "hits", 1, None)
            .unwrap();
        ops.cas(
            CollectionModel::Kv,
            "kv_default",
            "profile",
            None,
            reddb_types::Value::text("created"),
            None,
        )
        .unwrap();
        ops.cas(
            CollectionModel::Kv,
            "kv_default",
            "profile",
            Some(&reddb_types::Value::text("different")),
            reddb_types::Value::text("ignored"),
            None,
        )
        .unwrap();

        let stats = r.stats().kv;
        assert_eq!(stats.puts, 1);
        assert_eq!(stats.gets, 1);
        assert_eq!(stats.deletes, 1);
        assert_eq!(stats.incrs, 1);
        assert_eq!(stats.cas_success, 1);
        assert_eq!(stats.cas_conflict, 1);
    }

    #[test]
    fn kv_invalidate_tags_removes_matching_entries_only() {
        let r = rt();

        r.execute_query("KV PUT sessions.blob = 'payload' TAGS [user:42, org:7]")
            .unwrap();

        let miss = r
            .execute_query("INVALIDATE TAGS [org:99] FROM sessions")
            .unwrap();
        assert_eq!(miss.affected_rows, 0);
        assert!(matches!(
            r.execute_query("KV GET sessions.blob")
                .unwrap()
                .result
                .records[0]
                .get("value"),
            Some(reddb_types::Value::Text(value)) if &**value == "payload"
        ));

        let hit = r
            .execute_query("INVALIDATE TAGS [user:42] FROM sessions")
            .unwrap();
        assert_eq!(hit.affected_rows, 1);
        assert!(matches!(
            r.execute_query("KV GET sessions.blob")
                .unwrap()
                .result
                .records[0]
                .get("value"),
            Some(reddb_types::Value::Null)
        ));
    }

    #[test]
    fn authenticated_kv_watch_and_deletes_apply_row_level_security() {
        let runtime = rt();
        let ops = super::KvAtomicOps::new(&runtime);
        for key in ["visible", "hidden"] {
            ops.set_with_tags(
                "guarded_cache",
                key,
                Value::text(key),
                None,
                &["shared".to_string()],
                false,
            )
            .expect("seed key");
        }
        for query in [
            "CREATE POLICY watch_visible ON guarded_cache FOR SELECT USING (key='visible')",
            "CREATE POLICY delete_visible ON guarded_cache FOR DELETE USING (key='visible')",
            "ALTER TABLE guarded_cache ENABLE ROW LEVEL SECURITY",
        ] {
            runtime.execute_query(query).expect(query);
        }

        let bundle = SnapshotBundle {
            snapshot: None,
            auth: Some(("alice".to_string(), Role::Write)),
            tenant: None,
        };
        with_snapshot_bundle(&bundle, || {
            let events = runtime.kv_watch_events_since_prefix("guarded_cache", "", 0, 100);
            assert!(!events.is_empty());
            assert!(events.iter().all(|event| event.key == "visible"));

            assert!(!runtime
                .delete_kv("guarded_cache", "hidden")
                .expect("hidden delete"));
            assert_eq!(
                ops.invalidate_tags("guarded_cache", &["shared".to_string()])
                    .expect("invalidate visible rows"),
                1
            );
        });

        assert_eq!(
            ops.get(CollectionModel::Kv, "guarded_cache", "visible")
                .expect("visible key"),
            None
        );
        assert_eq!(
            ops.get(CollectionModel::Kv, "guarded_cache", "hidden")
                .expect("hidden key"),
            Some(Value::text("hidden"))
        );
    }

    #[test]
    fn kv_runtime_stats_count_watch_streams_and_events() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        assert_eq!(r.stats().kv.watch_streams_active, 0);

        {
            let mut stream = r.kv_watch_subscribe("kv_default", "watched", None);
            assert_eq!(r.stats().kv.watch_streams_active, 1);

            ops.set(
                CollectionModel::Kv,
                "kv_default",
                "watched",
                reddb_types::Value::Integer(1),
                None,
                false,
            )
            .unwrap();
            let event = stream.poll_next().expect("watch event");
            assert_eq!(event.key, "watched");
            assert_eq!(r.stats().kv.watch_events_emitted, 1);

            stream.record_drop_count(3);
            assert_eq!(r.stats().kv.watch_drops, 3);
        }

        assert_eq!(r.stats().kv.watch_streams_active, 0);
    }

    #[test]
    fn incr_preserves_ttl_and_tags_it_was_not_asked_to_change() {
        // `KV INCR key [BY n] [EXPIRE dur]` has no TAGS clause and EXPIRE is
        // optional, so a plain INCR cannot restate either. It is implemented
        // as delete-then-create (that is how non-versioned KV refreshes TTL),
        // which drops whatever the previous version carried — so an INCR
        // silently made an expiring key permanent and cleared its tags.
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);

        ops.set_with_tags_and_metadata_for_model(
            CollectionModel::Kv,
            "kv_default",
            "hits",
            reddb_types::Value::Integer(1),
            Some(60_000),
            &["daily".to_string()],
            false,
            Vec::new(),
        )
        .expect("seed put");

        ops.incr(CollectionModel::Kv, "kv_default", "hits", 1, None)
            .expect("incr");

        assert_eq!(
            r.inner.kv_tag_index.tags_for_key("kv_default", "hits"),
            vec!["daily".to_string()],
            "INCR must not drop the key's tags"
        );
        let entity = ops
            .get_entity(CollectionModel::Kv, "kv_default", "hits")
            .expect("lookup")
            .expect("key still present");
        let metadata = r
            .inner
            .db
            .store()
            .get_metadata("kv_default", entity.id)
            .expect("kv entry carries metadata");
        assert!(
            metadata.get("_ttl_ms").is_some(),
            "INCR without EXPIRE must not make an expiring key permanent; metadata was {:?}",
            metadata
        );
    }

    #[test]
    fn cas_preserves_ttl_and_tags_it_was_not_asked_to_change() {
        // `KV CAS key EXPECT <v> SET <v> [EXPIRE dur]` has the same shape as
        // INCR — no TAGS clause, optional EXPIRE — and the same
        // delete-then-create underneath, so it had the same defect.
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);

        ops.set_with_tags_and_metadata_for_model(
            CollectionModel::Kv,
            "kv_default",
            "flag",
            reddb_types::Value::Integer(1),
            Some(60_000),
            &["daily".to_string()],
            false,
            Vec::new(),
        )
        .expect("seed put");

        let (swapped, _) = ops
            .cas(
                CollectionModel::Kv,
                "kv_default",
                "flag",
                Some(&reddb_types::Value::Integer(1)),
                reddb_types::Value::Integer(2),
                None,
            )
            .expect("cas");
        assert!(swapped, "expected value matched, so the swap must apply");

        assert_eq!(
            r.inner.kv_tag_index.tags_for_key("kv_default", "flag"),
            vec!["daily".to_string()],
            "CAS must not drop the key's tags"
        );
        let entity = ops
            .get_entity(CollectionModel::Kv, "kv_default", "flag")
            .expect("lookup")
            .expect("key still present");
        let metadata = r
            .inner
            .db
            .store()
            .get_metadata("kv_default", entity.id)
            .expect("kv entry carries metadata");
        assert!(
            metadata.get("_ttl_ms").is_some(),
            "CAS without EXPIRE must not make an expiring key permanent; metadata was {:?}",
            metadata
        );
    }

    #[test]
    fn incr_existing_integer_accumulates() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        ops.incr(CollectionModel::Kv, "kv_default", "ctr", 1, None)
            .unwrap();
        ops.incr(CollectionModel::Kv, "kv_default", "ctr", 1, None)
            .unwrap();
        let v = ops
            .incr(CollectionModel::Kv, "kv_default", "ctr", 1, None)
            .unwrap();
        assert_eq!(v, 3);
    }

    #[test]
    fn decr_via_negative_by() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        ops.incr(CollectionModel::Kv, "kv_default", "stock", 10, None)
            .unwrap();
        let v = ops
            .incr(CollectionModel::Kv, "kv_default", "stock", -3, None)
            .unwrap();
        assert_eq!(v, 7);
    }

    #[test]
    fn concurrent_incr_single_key_is_atomic() {
        const THREADS: usize = 8;
        const ITERS: usize = 1000;

        let runtime = std::sync::Arc::new(rt());
        let barrier = std::sync::Arc::new(std::sync::Barrier::new(THREADS));
        let mut handles = Vec::new();

        for _ in 0..THREADS {
            let runtime = std::sync::Arc::clone(&runtime);
            let barrier = std::sync::Arc::clone(&barrier);
            handles.push(std::thread::spawn(move || {
                let ops = super::KvAtomicOps::new(&runtime);
                barrier.wait();
                for _ in 0..ITERS {
                    ops.incr(CollectionModel::Kv, "kv_default", "counter", 1, None)
                        .unwrap();
                }
            }));
        }

        for handle in handles {
            handle.join().expect("worker should finish");
        }

        let ops = super::KvAtomicOps::new(&runtime);
        assert_eq!(
            ops.get(CollectionModel::Kv, "kv_default", "counter")
                .unwrap(),
            Some(reddb_types::Value::Integer((THREADS * ITERS) as i64))
        );
    }

    #[test]
    fn incr_on_string_value_returns_error() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "name",
            reddb_types::Value::text("alice"),
            None,
            false,
        )
        .unwrap();
        let err = ops
            .incr(CollectionModel::Kv, "kv_default", "name", 1, None)
            .unwrap_err();
        assert!(err.to_string().contains("non-integer"));
    }

    // --- CAS tests ---

    #[test]
    fn cas_matching_value_succeeds() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "lock",
            reddb_types::Value::text("free"),
            None,
            false,
        )
        .unwrap();
        let (ok, prev) = ops
            .cas(
                CollectionModel::Kv,
                "kv_default",
                "lock",
                Some(&reddb_types::Value::text("free")),
                reddb_types::Value::text("held"),
                None,
            )
            .unwrap();
        assert!(ok);
        assert_eq!(prev, Some(reddb_types::Value::text("free")));
        // Value actually changed.
        assert_eq!(
            ops.get(CollectionModel::Kv, "kv_default", "lock").unwrap(),
            Some(reddb_types::Value::text("held"))
        );
    }

    #[test]
    fn concurrent_cas_allows_one_success_per_round() {
        const THREADS: usize = 8;
        const ROUNDS: usize = 100;

        let runtime = std::sync::Arc::new(rt());
        let ops = super::KvAtomicOps::new(&runtime);
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "cas_counter",
            reddb_types::Value::Integer(0),
            None,
            false,
        )
        .unwrap();

        let start_round = std::sync::Arc::new(std::sync::Barrier::new(THREADS));
        let finish_round = std::sync::Arc::new(std::sync::Barrier::new(THREADS));
        let successes = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let mut handles = Vec::new();

        for _ in 0..THREADS {
            let runtime = std::sync::Arc::clone(&runtime);
            let start_round = std::sync::Arc::clone(&start_round);
            let finish_round = std::sync::Arc::clone(&finish_round);
            let successes = std::sync::Arc::clone(&successes);
            handles.push(std::thread::spawn(move || {
                let ops = super::KvAtomicOps::new(&runtime);
                for round in 0..ROUNDS {
                    start_round.wait();
                    let (ok, _) = ops
                        .cas(
                            CollectionModel::Kv,
                            "kv_default",
                            "cas_counter",
                            Some(&reddb_types::Value::Integer(round as i64)),
                            reddb_types::Value::Integer((round + 1) as i64),
                            None,
                        )
                        .unwrap();
                    if ok {
                        successes.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    }
                    finish_round.wait();
                }
            }));
        }

        for handle in handles {
            handle.join().expect("worker should finish");
        }

        assert_eq!(successes.load(std::sync::atomic::Ordering::SeqCst), ROUNDS);
        assert_eq!(
            ops.get(CollectionModel::Kv, "kv_default", "cas_counter")
                .unwrap(),
            Some(reddb_types::Value::Integer(ROUNDS as i64))
        );
    }

    #[test]
    fn cas_mismatching_value_fails() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "lock",
            reddb_types::Value::text("free"),
            None,
            false,
        )
        .unwrap();
        let (ok, current) = ops
            .cas(
                CollectionModel::Kv,
                "kv_default",
                "lock",
                Some(&reddb_types::Value::text("held")),
                reddb_types::Value::text("worker-7"),
                None,
            )
            .unwrap();
        assert!(!ok);
        assert_eq!(current, Some(reddb_types::Value::text("free")));
        // Value unchanged.
        assert_eq!(
            ops.get(CollectionModel::Kv, "kv_default", "lock").unwrap(),
            Some(reddb_types::Value::text("free"))
        );
    }

    #[test]
    fn cas_expect_null_on_missing_key_creates() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        let (ok, prev) = ops
            .cas(
                CollectionModel::Kv,
                "kv_default",
                "new_key",
                None,
                reddb_types::Value::text("created"),
                None,
            )
            .unwrap();
        assert!(ok);
        assert_eq!(prev, None);
        assert_eq!(
            ops.get(CollectionModel::Kv, "kv_default", "new_key")
                .unwrap(),
            Some(reddb_types::Value::text("created"))
        );
    }

    #[test]
    fn cas_expect_null_on_existing_key_fails() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "taken",
            reddb_types::Value::text("worker-1"),
            None,
            false,
        )
        .unwrap();
        let (ok, current) = ops
            .cas(
                CollectionModel::Kv,
                "kv_default",
                "taken",
                None,
                reddb_types::Value::text("worker-2"),
                None,
            )
            .unwrap();
        assert!(!ok);
        assert_eq!(current, Some(reddb_types::Value::text("worker-1")));
    }

    #[test]
    fn cas_via_sql_roundtrip() {
        let r = rt();
        // Seed value.
        r.execute_query("KV PUT lock = 'free'").unwrap();
        // CAS: free → held — should succeed.
        let res = r
            .execute_query("KV CAS lock EXPECT 'free' SET 'held'")
            .unwrap();
        let row = &res.result.records[0];
        assert_eq!(row.get("ok"), Some(&reddb_types::Value::Boolean(true)));
        // CAS: free → held again — should fail (value is now 'held').
        let res2 = r
            .execute_query("KV CAS lock EXPECT 'free' SET 'held'")
            .unwrap();
        let row2 = &res2.result.records[0];
        assert_eq!(row2.get("ok"), Some(&reddb_types::Value::Boolean(false)));
    }

    #[test]
    fn cas_expect_null_via_sql() {
        let r = rt();
        let res = r
            .execute_query("KV CAS singleton EXPECT NULL SET 'first'")
            .unwrap();
        let row = &res.result.records[0];
        assert_eq!(row.get("ok"), Some(&reddb_types::Value::Boolean(true)));
        // Second call must fail.
        let res2 = r
            .execute_query("KV CAS singleton EXPECT NULL SET 'second'")
            .unwrap();
        let row2 = &res2.result.records[0];
        assert_eq!(row2.get("ok"), Some(&reddb_types::Value::Boolean(false)));
    }

    #[test]
    fn incr_via_sql_roundtrip() {
        let r = rt();
        let res = r.execute_query("KV INCR hits").unwrap();
        let row = &res.result.records[0];
        assert_eq!(row.get("value"), Some(&reddb_types::Value::Integer(1)));
        let res2 = r.execute_query("KV INCR hits BY 4").unwrap();
        let row2 = &res2.result.records[0];
        assert_eq!(row2.get("value"), Some(&reddb_types::Value::Integer(5)));
    }

    #[test]
    fn concurrent_self_referential_update_is_atomic() {
        const THREADS: usize = 8;
        const ITERS: usize = 100;

        let runtime = std::sync::Arc::new(rt());
        runtime
            .execute_query("CREATE TABLE counters (id INT, n INT)")
            .unwrap();
        runtime
            .execute_query("INSERT INTO counters (id, n) VALUES (1, 0)")
            .unwrap();

        let barrier = std::sync::Arc::new(std::sync::Barrier::new(THREADS));
        let mut handles = Vec::new();
        for _ in 0..THREADS {
            let runtime = std::sync::Arc::clone(&runtime);
            let barrier = std::sync::Arc::clone(&barrier);
            handles.push(std::thread::spawn(move || {
                barrier.wait();
                for _ in 0..ITERS {
                    runtime
                        .execute_query("UPDATE counters SET n = n + 1 WHERE id = 1")
                        .unwrap();
                }
            }));
        }

        for handle in handles {
            handle.join().expect("worker should finish");
        }

        let selected = runtime
            .execute_query("SELECT n FROM counters WHERE id = 1")
            .unwrap();
        assert_eq!(
            selected.result.records[0].get("n"),
            Some(&reddb_types::Value::Integer((THREADS * ITERS) as i64))
        );
    }

    #[test]
    fn watch_stream_delivers_key_events_in_lsn_order() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        let mut stream = r.kv_watch_subscribe("kv_default", "seq", None);

        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "seq",
            reddb_types::Value::Integer(1),
            None,
            false,
        )
        .unwrap();
        ops.incr(CollectionModel::Kv, "kv_default", "seq", 1, None)
            .unwrap();
        ops.delete(CollectionModel::Kv, "kv_default", "seq")
            .unwrap();
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "seq",
            reddb_types::Value::Integer(9),
            None,
            false,
        )
        .unwrap();

        let mut events = Vec::new();
        while let Some(event) = stream.poll_next() {
            events.push(event);
            if events.len() == 4 {
                break;
            }
        }

        assert_eq!(events.len(), 4);
        assert_eq!(
            events[0].op,
            crate::replication::cdc::ChangeOperation::Insert
        );
        assert_eq!(
            events[1].op,
            crate::replication::cdc::ChangeOperation::Update
        );
        assert_eq!(
            events[2].op,
            crate::replication::cdc::ChangeOperation::Delete
        );
        assert_eq!(
            events[3].op,
            crate::replication::cdc::ChangeOperation::Insert
        );
        assert!(events.windows(2).all(|pair| pair[0].lsn < pair[1].lsn));
    }

    #[test]
    fn watch_prefix_stream_delivers_matching_events_only() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        let mut stream = r.kv_watch_subscribe_prefix("kv_default", "acct:", None);

        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "acct:1",
            reddb_types::Value::Integer(1),
            None,
            false,
        )
        .unwrap();
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "session:1",
            reddb_types::Value::Integer(2),
            None,
            false,
        )
        .unwrap();
        ops.set(
            CollectionModel::Kv,
            "kv_default",
            "acct:2",
            reddb_types::Value::Integer(3),
            None,
            false,
        )
        .unwrap();

        let first = stream.poll_next().expect("first prefix event");
        let second = stream.poll_next().expect("second prefix event");
        assert_eq!(first.key, "acct:1");
        assert_eq!(second.key, "acct:2");
        assert!(stream.poll_next().is_none());
    }

    #[test]
    fn watch_stream_resume_from_lsn_delivers_missed_events_without_duplicates() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        let mut stream = r.kv_watch_subscribe("kv_default", "resume", None);

        let mut last_seen_lsn = 0;
        for value in 0..5 {
            ops.set(
                CollectionModel::Kv,
                "kv_default",
                "resume",
                reddb_types::Value::Integer(value),
                None,
                false,
            )
            .unwrap();
            last_seen_lsn = stream.poll_next().expect("initial event").lsn;
        }
        drop(stream);

        for value in 5..55 {
            ops.set(
                CollectionModel::Kv,
                "kv_default",
                "resume",
                reddb_types::Value::Integer(value),
                None,
                false,
            )
            .unwrap();
        }

        let expected_lsns: Vec<u64> = r
            .kv_watch_events_since("kv_default", "resume", last_seen_lsn, 200)
            .into_iter()
            .map(|event| event.lsn)
            .collect();
        assert!(!expected_lsns.is_empty());

        let mut resumed = r.kv_watch_subscribe("kv_default", "resume", Some(last_seen_lsn));
        let mut lsns = Vec::new();
        while let Some(event) = resumed.poll_next() {
            lsns.push(event.lsn);
            if lsns.len() == expected_lsns.len() {
                break;
            }
        }

        assert_eq!(lsns, expected_lsns);
        assert!(lsns.iter().all(|lsn| *lsn > last_seen_lsn));
        assert!(lsns.windows(2).all(|pair| pair[0] < pair[1]));
        assert!(resumed.poll_next().is_none());
    }

    #[test]
    fn watch_stream_slow_consumer_drops_oldest_buffered_events() {
        let r = rt();
        let ops = super::KvAtomicOps::new(&r);
        let mut stream = r.kv_watch_subscribe("kv_default", "slow", None);

        for value in 0..1_200 {
            ops.set(
                CollectionModel::Kv,
                "kv_default",
                "slow",
                reddb_types::Value::Integer(value),
                None,
                false,
            )
            .unwrap();
        }

        let event = stream.poll_next().expect("tail event after drops");
        assert!(event.lsn > 1);
        assert!(event.dropped_event_count > 0);
        assert_eq!(stream.dropped_event_count(), event.dropped_event_count);
        assert_eq!(r.stats().kv.watch_drops, event.dropped_event_count);
    }

    #[test]
    fn watch_stream_idle_timeout_closes_subscription() {
        let r = rt();
        r.execute_query("SET CONFIG red.config.kv.watch.idle_timeout_ms = 1")
            .unwrap();

        let mut stream = r.kv_watch_subscribe("kv_default", "idle", None);
        assert_eq!(r.stats().kv.watch_streams_active, 1);
        std::thread::sleep(std::time::Duration::from_millis(5));

        assert!(stream.poll_next().is_none());
        assert_eq!(r.stats().kv.watch_streams_active, 0);
    }

    #[test]
    fn watch_stream_does_not_emit_rolled_back_put() {
        let r = rt();
        let mut stream = r.kv_watch_subscribe("kv_default", "rollback_key", None);

        r.execute_query("BEGIN").unwrap();
        r.execute_query("KV PUT rollback_key = 'dirty'").unwrap();
        r.execute_query("ROLLBACK").unwrap();

        assert!(stream.poll_next().is_none());
    }
}

fn vault_value_aad(collection: &str, tenant: Option<&str>) -> String {
    match tenant {
        Some(tenant) => format!("reddb.vault.{collection}.tenant/{}", hex::encode(tenant)),
        None => format!("reddb.vault.{collection}"),
    }
}

fn vault_watch_in_scope(
    event: &crate::replication::cdc::KvWatchEvent,
    tenant: Option<&str>,
) -> bool {
    let stored_tenant = event
        .after
        .as_ref()
        .or(event.before.as_ref())
        .and_then(|metadata| metadata.get("tenant"))
        .and_then(|value| value.as_str());
    stored_tenant == tenant
}