reddb_server/runtime/

statement_frame.rs

use std::cell::RefCell;
use std::collections::HashSet;
use std::rc::Rc;
use std::sync::Arc;

use super::execution_context::capture_current_snapshot;
use super::impl_core::{
    collections_referenced, current_auth_identity, current_connection_id, current_tenant,
    has_with_prefix, intent_lock_modes_for, peek_top_level_as_of_with_table,
    query_has_volatile_builtin, query_is_ask_statement, ConfigSnapshotGuard, CurrentSnapshotGuard,
    KvStoreGuard, SecretStoreGuard, SnapshotContext, TxLocalTenantGuard,
};
use super::{RedDBRuntime, RuntimeQueryResult, RuntimeResultCacheEntry};
use crate::api::{RedDBError, RedDBResult};
use crate::auth::Role;
use crate::storage::transaction::snapshot::{Snapshot, SnapshotRetentionGuard, Xid};
use reddb_rql::ast::QueryExpr;
use reddb_rql::modes::{detect_mode, parse_multi, QueryMode};

/// Coarse privilege classification for a statement, computed once at
/// frame-build time from the SQL text. Mirrors the three-role auth
/// model (`Role::Read < Role::Write < Role::Admin`) so the frame can
/// answer "can this identity run this statement?" without re-walking
/// the parsed `QueryExpr` at every call site.
///
/// `None` means the statement does not touch the privilege gate at
/// all (transaction control, SET, SHOW). Such statements must remain
/// runnable under any authenticated identity.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Privilege {
    /// Read-only data access (SELECT, EXPLAIN, SHOW). Satisfied by
    /// any role from `Role::Read` upward.
    Read,
    /// Mutation of user data or schema author DDL (INSERT, UPDATE,
    /// DELETE, CREATE/ALTER/DROP TABLE, CREATE MIGRATION). Requires
    /// at least `Role::Write`.
    Write,
    /// Authority statements — GRANT, REVOKE, ALTER USER, APPLY /
    /// ROLLBACK MIGRATION, IAM policy mutation. Requires `Role::Admin`.
    Admin,
    /// Statement does not consult the privilege gate (BEGIN, COMMIT,
    /// ROLLBACK, SET, SHOW with no data exposure). Always permitted
    /// for any authenticated identity.
    None,
}

impl Privilege {
    /// `true` iff `role` is sufficient to execute a statement carrying
    /// this required privilege. Encodes the standard `Read ⊆ Write ⊆
    /// Admin` containment used by the auth fallback path.
    pub(crate) fn is_satisfied_by(self, role: Role) -> bool {
        match self {
            Self::None => true,
            Self::Read => role.can_read(),
            Self::Write => role.can_write(),
            Self::Admin => role.can_admin(),
        }
    }
}

/// Coarse lock intent for a statement, computed once at frame-build
/// time. Maps onto the storage-layer's `LockMode` matrix downstream
/// but stays decoupled here so the runtime can answer "does this
/// statement need the lock manager at all?" without a `use storage::`
/// at every call site.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum LockIntent {
    /// No collection-level lock needed (transaction control, SET,
    /// SHOW, EXPLAIN). The lock-acquisition path can short-circuit.
    None,
    /// Reader-style intent: SELECT, joins, graph / queue / search
    /// reads. Maps to `(IS, IS)` at the storage layer.
    Shared,
    /// Writer- or DDL-style intent: INSERT/UPDATE/DELETE (`(IX, IX)`)
    /// and CREATE/ALTER/DROP (`(IX, X)`). Both are surfaced as
    /// `Exclusive` at this granularity — call sites that need the
    /// finer distinction still consult `intent_lock_modes_for`.
    Exclusive,
}

/// Small, stable Interface that *represents* a read statement's
/// execution context. Every read caller that needs to know "under
/// what scope / identity / snapshot am I running, and is there an
/// AS OF floor in effect?" consults this trait — never the
/// underlying thread-locals or runtime fields directly.
///
/// The deletion test: removing this trait would force the four
/// concerns it exposes back into ad-hoc lookups at every read
/// callsite (`current_tenant()`, `current_auth_identity()`,
/// `capture_current_snapshot()`, AS OF re-parsing). The trait
/// concentrates them in one place so future changes (per-statement
/// logging, audit, scope policy) have a single seam to extend.
pub(crate) trait ReadFrame {
    /// Effective tenant scope for the statement after WITHIN /
    /// SET LOCAL TENANT / SET TENANT resolution. `None` means
    /// "no tenant bound" (RLS deny-default applies).
    fn effective_scope(&self) -> Option<&str>;

    /// Authenticated identity observed at frame-build time, if any.
    /// Returns `(username, role)` so callers can render audit lines
    /// or feed RLS policy lookups without re-reading thread-locals.
    fn identity(&self) -> Option<(&str, Role)>;

    /// MVCC snapshot the statement reads against. For autocommit
    /// this is a fresh snapshot; inside an active transaction it
    /// is the txn's snapshot; under AS OF it is the resolved
    /// historical xid.
    fn snapshot(&self) -> &Snapshot;

    /// Complete visibility context, including the reader's savepoint writes.
    fn snapshot_context(&self, runtime: &RedDBRuntime) -> SnapshotContext {
        SnapshotContext {
            snapshot: self.snapshot().clone(),
            manager: runtime.inner.transaction_state.snapshot_manager(),
            own_xids: HashSet::new(),
            requires_index_fallback: true,
            serializable_reader: None,
        }
    }

    /// AS OF xid floor when AS OF was applied for this statement,
    /// `None` for live reads. Useful for downstream callers that
    /// want to gate behaviour on historical-read mode without
    /// re-parsing the query.
    fn as_of_floor(&self) -> Option<Xid>;

    /// Stable result-cache key for the statement (already mixes
    /// effective tenant + identity).
    fn cache_key(&self) -> &str;

    /// Whether the statement is safe to serve from / populate the
    /// result cache. Combines two underlying signals:
    ///
    ///   * the query does not call a volatile builtin (e.g. `NOW()`,
    ///     `RANDOM()`, `UUID()`), which would change between calls,
    ///   * the connection is not inside an active transaction with
    ///     uncommitted writes that other readers shouldn't observe.
    ///
    /// SELECT cache callsites (read + write) consult this method
    /// instead of re-deriving safety from globals or poking the
    /// frame's private fields. Removing it would force every cache
    /// callsite to re-run `query_has_volatile_builtin` plus
    /// `result_cache_safe(conn_id)` inline.
    fn should_cache_result(&self) -> bool;

    /// Coarse privilege class the statement requires, computed once
    /// at frame-build time from the SQL prefix. Read/write dispatch
    /// sites consult this instead of re-classifying the parsed
    /// `QueryExpr` inline at every callsite.
    ///
    /// Removing this method would force every privilege gate to
    /// recompute the (action, resource) classification from the
    /// parsed expression and re-check the role hierarchy inline.
    fn required_privilege(&self) -> Privilege;

    /// Coarse collection-level lock intent the statement implies.
    /// `None` lets the lock-acquisition path short-circuit without
    /// touching the lock manager.
    ///
    /// Removing this method would force the lock-acquisition path
    /// to always invoke `intent_lock_modes_for` (which itself walks
    /// the parsed expression) even for transaction-control / SET /
    /// SHOW statements that need no collection lock at all.
    fn lock_intent(&self) -> LockIntent;

    /// Set of collection ids the calling identity is allowed to
    /// observe under the active `(tenant, role)` scope. Computed once
    /// at frame-build time via the `AuthStore` visible-collections
    /// cache (see `auth::scope_cache`) and used by `AuthorizedSearch`
    /// to pre-filter SEARCH SIMILAR / SEARCH CONTEXT candidate sets
    /// before any similarity score is computed (issue #119).
    ///
    /// `None` means the frame was built without an auth store wired —
    /// embedded / single-tenant tests run that way. AI search call
    /// sites refuse to proceed with `None`, which is the deny-default
    /// the issue requires; pure SELECT paths fall back to the existing
    /// per-row RLS gate.
    fn visible_collections(&self) -> Option<&std::collections::HashSet<String>>;
}

/// Cheap first-word classification of a SQL statement, used at
/// frame-build time to derive `Privilege` + `LockIntent` without
/// re-parsing the query. Matches the keywords that the legacy
/// inline checks in `RedDBRuntime::check_query_privilege` and
/// `intent_lock_modes_for` already key on.
/// Whether [`statement_kind`] reads `query` as a mutating statement.
/// The Request seam uses it to keep its pre-write commit-policy check off
/// read paths.
pub(crate) fn statement_is_write(query: &str) -> bool {
    statement_kind(query) == "write"
}

fn statement_kind(query: &str) -> &'static str {
    let trimmed = query.trim_start();
    // Skip a leading line / block comment so the classifier doesn't
    // misread `/* ... */ SELECT ...` as an unknown statement.
    let trimmed = if let Some(rest) = trimmed.strip_prefix("--") {
        rest.split_once('\n')
            .map(|(_, r)| r)
            .unwrap_or("")
            .trim_start()
    } else {
        trimmed
    };
    let mut tokens = trimmed.split(|c: char| c.is_whitespace() || c == '(' || c == ';');
    let first = tokens.next().unwrap_or("");
    let second = tokens.next().unwrap_or("");
    if first.eq_ignore_ascii_case("KV") {
        if second.eq_ignore_ascii_case("GET")
            || second.eq_ignore_ascii_case("LIST")
            || second.eq_ignore_ascii_case("WATCH")
        {
            return "read";
        }
        return "write";
    }
    if first.eq_ignore_ascii_case("VAULT") {
        if second.eq_ignore_ascii_case("GET")
            || second.eq_ignore_ascii_case("REVEAL")
            || second.eq_ignore_ascii_case("UNSEAL")
            || second.eq_ignore_ascii_case("LIST")
            || second.eq_ignore_ascii_case("WATCH")
            || second.eq_ignore_ascii_case("HISTORY")
        {
            return "read";
        }
        return "write";
    }
    // ASCII-uppercase compare without allocating: SQL keywords are ASCII.
    let mut buf = [0u8; 16];
    let bytes = first.as_bytes();
    let n = bytes.len().min(buf.len());
    for i in 0..n {
        buf[i] = bytes[i].to_ascii_uppercase();
    }
    match &buf[..n] {
        b"SELECT" | b"WITH" | b"SHOW" | b"EXPLAIN" | b"DESCRIBE" | b"DESC" | b"RANK"
        | b"APPROX" | b"APPROXIMATE" | b"ZRANK" | b"ZRANGE" | b"LIST" | b"WATCH" | b"GET"
        | b"HISTORY" | b"REVEAL" | b"UNSEAL" => "read",
        b"INSERT" | b"UPDATE" | b"DELETE" | b"UPSERT" | b"MERGE" | b"COPY" | b"TRUNCATE" => "write",
        b"CREATE" | b"ALTER" | b"DROP" | b"PROMOTE" | b"REINDEX" | b"VACUUM" | b"ANALYZE" => "ddl",
        b"GRANT" | b"REVOKE" => "admin",
        b"BEGIN" | b"START" | b"COMMIT" | b"ROLLBACK" | b"SAVEPOINT" | b"RELEASE" | b"END"
        | b"SET" | b"RESET" | b"PREPARE" | b"EXECUTE" | b"DEALLOCATE" | b"USE" => "control",
        b"PUT" | b"INCR" | b"DECR" | b"ADD" | b"ROTATE" | b"PURGE" | b"INVALIDATE" => "write",
        _ => "unknown",
    }
}

/// Refuse a pre-parsed expression that carries a top-level `AS OF`.
///
/// `AS OF` is resolved from the statement text (`AsOfSpec` -> commit /
/// branch / tag / timestamp -> xid) and `TableQuery::as_of` is consumed
/// nowhere in the runtime or scan path. An expression-only frame
/// therefore cannot honour it, and installing a live snapshot for a
/// historical read would answer with current rows presented as history.
/// Fail loudly instead; callers that have the text (`execute_query`,
/// `execute_prepared_query`, `execute_query_with_params`) resolve it
/// correctly. Mirrors `peek_top_level_as_of_with_table`, which likewise
/// only recognises a top-level `AS OF`.
fn reject_unresolvable_as_of(expr: &QueryExpr) -> RedDBResult<()> {
    let QueryExpr::Table(table) = expr else {
        return Ok(());
    };
    if table.as_of.is_none() {
        return Ok(());
    }
    Err(RedDBError::InvalidOperation(
        "AS OF cannot be resolved from a pre-parsed expression — \
         execute the statement text (execute_query / execute_prepared_query / \
         execute_query_with_params) so the historical snapshot is resolved"
            .to_string(),
    ))
}

fn classify_privilege(query: &str) -> Privilege {
    match statement_kind(query) {
        "read" => Privilege::Read,
        "write" => Privilege::Write,
        // DDL is gated at `Role::Write` in the legacy fallback (see
        // `RedDBRuntime::check_query_privilege` for CreateTable et al.),
        // so it classifies as Write here. APPLY / ROLLBACK MIGRATION and
        // GRANT / REVOKE upgrade to Admin via finer checks at the call
        // site — the frame surfaces only the coarse class.
        "ddl" => Privilege::Write,
        "admin" => Privilege::Admin,
        _ => Privilege::None,
    }
}

fn classify_lock_intent(query: &str) -> LockIntent {
    match statement_kind(query) {
        "read" => LockIntent::Shared,
        "write" | "ddl" => LockIntent::Exclusive,
        _ => LockIntent::None,
    }
}

pub(super) struct StatementExecutionFrame {
    tx_local_tenant: Option<Option<String>>,
    snapshot: Snapshot,
    snapshot_retention: RefCell<Option<SnapshotRetentionGuard>>,
    own_xids: HashSet<Xid>,
    serializable_reader: Option<Xid>,
    cache_key: String,
    is_volatile_query: bool,
    cache_safe: bool,
    /// Effective tenant captured at frame-build time after WITHIN /
    /// SET LOCAL TENANT / SET TENANT resolution. Stored on the frame
    /// so the `ReadFrame` Interface can return a borrow without
    /// re-touching the thread-local stack.
    effective_scope: Option<String>,
    /// Auth identity captured at frame-build time. `None` for
    /// embedded / anonymous callers.
    identity: Option<(String, Role)>,
    /// `Some(xid)` when AS OF resolved to a historical xid; `None`
    /// for live reads.
    as_of_floor: Option<Xid>,
    /// True when the statement snapshot can require tuple versions that
    /// current secondary indexes no longer contain.
    requires_index_fallback: bool,
    /// Privilege class required by the statement, derived from the
    /// SQL text at frame-build time. Read/write dispatch sites
    /// consult this instead of re-classifying the parsed expression.
    required_privilege: Privilege,
    /// Collection-level lock intent the statement implies. The
    /// lock-acquisition path short-circuits when this is `None`.
    lock_intent: LockIntent,
    /// Set of collection ids the active `(tenant, role)` scope is
    /// allowed to observe. Computed at frame-build time via the
    /// `AuthStore` visibility cache and consumed by `AuthorizedSearch`
    /// to gate SEARCH SIMILAR / SEARCH CONTEXT candidate sets before
    /// scoring (issue #119). `None` when no auth store is wired
    /// (embedded test mode) — AI search refuses on `None`.
    visible_collections: Option<HashSet<String>>,
    /// Per-owner buffer arena for query-result row chunks (#885). Owned
    /// by the frame because the frame already owns the query lifecycle;
    /// lent to the row-streaming path (`execute_runtime_table_query_in`)
    /// so chunk buffers are reused across the statement's chunk-fetches
    /// instead of allocated fresh per chunk. Reclaimed when the frame
    /// drops at statement end — no `thread_local!` scratch, which would
    /// be unsound under tokio's work-stealing runtime.
    row_arena: Rc<RefCell<super::query_exec::RowBufferArena>>,
}

pub(super) struct StatementFrameGuards {
    _tx_local_guard: TxLocalTenantGuard,
    _config_snapshot_guard: ConfigSnapshotGuard,
    _secret_store_guard: SecretStoreGuard,
    _kv_store_guard: KvStoreGuard,
    _snapshot_guard: CurrentSnapshotGuard,
}

pub(super) struct PreparedStatement {
    pub(super) expr: QueryExpr,
    pub(super) mode: QueryMode,
}

/// What a `StatementExecutionFrame` is built from.
///
/// The frame derives the statement snapshot (including any `AS OF`
/// floor), the coarse privilege class, the lock intent, the volatility
/// flag, and the result-cache key from its input. Feeding it anything
/// other than the statement itself makes every one of those describe a
/// different statement than the one that runs, so the input is named
/// explicitly here rather than being "some string".
#[derive(Clone, Copy)]
pub(super) enum StatementIdentity<'a> {
    /// The statement's SQL text. All five derivations are exact.
    Text(&'a str),
    /// A pre-parsed expression with no accompanying text — the nested
    /// in-statement callers (CTE prelude, materialized-view refresh,
    /// inline graph TVF) hold an expression the outer statement already
    /// parsed.
    ///
    /// Text-derived facts are not guessed from a kind label: there is no
    /// stable cache key (a label would collide across every statement of
    /// the same kind), and the coarse privilege / lock short-circuits are
    /// left at `None` so the deep grant engine stays the gate. `AS OF` is
    /// rejected up front by `build` because it cannot be resolved without
    /// the text.
    Expr(&'a QueryExpr),
}

impl<'a> StatementIdentity<'a> {
    /// SQL text when the frame was built from text, `""` otherwise. Used
    /// for the text-derived classifications, all of which degrade to the
    /// deliberately inert `unknown` / `None` classes on the expr path.
    fn text(&self) -> &'a str {
        match self {
            Self::Text(query) => query,
            Self::Expr(_) => "",
        }
    }
}

impl StatementExecutionFrame {
    pub(super) fn build(
        runtime: &RedDBRuntime,
        identity_source: StatementIdentity<'_>,
    ) -> RedDBResult<Self> {
        let query = identity_source.text();
        if let StatementIdentity::Expr(expr) = identity_source {
            reject_unresolvable_as_of(expr)?;
        }
        let conn_id = current_connection_id();
        let tx_local_tenant = runtime.inner.transaction_state.local_tenant(conn_id);
        let tx_context = runtime.inner.transaction_state.context(conn_id);
        let own_xids = own_transaction_xids(tx_context.as_ref());
        let serializable_reader = tx_context
            .as_ref()
            .filter(|ctx| {
                ctx.isolation == crate::storage::transaction::IsolationLevel::Serializable
            })
            .map(|ctx| ctx.xid);
        let (snapshot, as_of_floor, snapshot_retention) =
            runtime.statement_snapshot(query, identity_source)?;
        let requires_index_fallback = as_of_floor.is_some() || tx_context.is_some();
        let (cache_key, is_volatile_query) = match identity_source {
            StatementIdentity::Text(_) => (
                result_cache_key(query),
                query_has_volatile_builtin(query) || query_is_ask_statement(query),
            ),
            // No text means no stable key, so the frame opts out of the
            // result cache on both sides: an empty key is never read from
            // or written to, and `is_volatile_query` is set so
            // `should_cache_result` refuses regardless of the key.
            StatementIdentity::Expr(_) => (String::new(), true),
        };
        let cache_safe = runtime.result_cache_safe(conn_id);
        // Capture identity + effective scope under the same
        // thread-local view that the cache key was built from, so
        // the Interface and the cache key agree on what "this
        // statement" means.
        let effective_scope = current_tenant();
        let identity = current_auth_identity();

        // Coarse classification of the statement, computed once from
        // the SQL prefix so downstream callers don't re-derive it
        // from the parsed `QueryExpr` at every privilege / lock site.
        // On the expr path `query` is empty, so both classify as the
        // inert `None` class: the coarse gate can only deny, never
        // allow (see `check_query_privilege`), so declining to
        // short-circuit leaves the deep grant engine as the gate.
        let required_privilege = classify_privilege(query);
        let lock_intent = classify_lock_intent(query);

        // Issue #119: resolve the visible-collections set for the
        // active (tenant, role) scope. Only meaningful when an auth
        // store is wired *and* an identity was captured — embedded
        // anonymous callers fall back to `None`, and AI search call
        // sites refuse on `None`.
        let visible_collections = match (runtime.inner.auth_store.read().clone(), identity.as_ref())
        {
            (Some(store), Some((principal, role))) => {
                let collections = runtime.inner.db.store().list_collections();
                Some(store.visible_collections_for_scope(
                    effective_scope.as_deref(),
                    *role,
                    principal,
                    &collections,
                ))
            }
            _ => None,
        };

        Ok(Self {
            tx_local_tenant,
            snapshot,
            snapshot_retention: RefCell::new(Some(snapshot_retention)),
            own_xids,
            serializable_reader,
            cache_key,
            is_volatile_query,
            cache_safe,
            effective_scope,
            identity,
            as_of_floor,
            requires_index_fallback,
            required_privilege,
            lock_intent,
            visible_collections,
            row_arena: Rc::new(RefCell::new(super::query_exec::RowBufferArena::new())),
        })
    }

    /// Lend the frame's per-owner row-buffer arena (#885) to the
    /// row-streaming path. Returns a cloned `Rc` handle; the frame remains
    /// the owner and the arena is reclaimed when the frame drops at
    /// statement end.
    pub(super) fn row_arena(&self) -> Rc<RefCell<super::query_exec::RowBufferArena>> {
        Rc::clone(&self.row_arena)
    }

    pub(super) fn install(&self, runtime: &RedDBRuntime) -> StatementFrameGuards {
        StatementFrameGuards {
            _tx_local_guard: TxLocalTenantGuard::install(self.tx_local_tenant.clone()),
            _config_snapshot_guard: ConfigSnapshotGuard::install(
                Arc::clone(&runtime.inner.db),
                runtime.inner.auth_store.read().clone(),
            ),
            _secret_store_guard: SecretStoreGuard::install(runtime),
            _kv_store_guard: KvStoreGuard::install(runtime.inner.auth_store.read().clone()),
            _snapshot_guard: CurrentSnapshotGuard::install(SnapshotContext {
                snapshot: self.snapshot.clone(),
                manager: runtime.inner.transaction_state.snapshot_manager(),
                own_xids: self.own_xids.clone(),
                requires_index_fallback: self.requires_index_fallback,
                serializable_reader: self.serializable_reader,
            }),
        }
    }

    pub(super) fn cache_key(&self) -> &str {
        &self.cache_key
    }

    pub(super) fn can_read_result_cache(&self) -> bool {
        // Delegates to the `ReadFrame` Interface so the volatile +
        // active-tx safety decision lives in exactly one place.
        <Self as ReadFrame>::should_cache_result(self)
    }

    pub(super) fn should_write_result_cache(&self, result: &RuntimeQueryResult) -> bool {
        // Cache-safety (volatile builtin, active-tx writes) comes from
        // the Interface; the rest are write-side payload heuristics
        // (statement shape, result size) that aren't part of the
        // safety contract.
        <Self as ReadFrame>::should_cache_result(self)
            && result.statement_type == "select"
            && result.engine != "vault"
            && result.engine != "runtime-rank"
            // `QUEUE READ` is a stateful read: a delayed message
            // (issue #722) becomes deliverable over time without a
            // producer push to invalidate the cache, so a cached empty
            // result would hide it. Skip caching entirely.
            && result.statement != "queue_group_read"
            // SCRUB is a stateful read too: the background form advances
            // a pacing cursor per tick (a cached tick would replay the
            // summary and never advance), and even a full pass must
            // re-read the file to see new corruption. Never cache.
            && result.engine != "runtime-scrub"
            && result.result.pre_serialized_json.is_none()
            // Graph-analytics TVF output (issue #802) is deterministic and
            // expensive to recompute, so it is cached at any row count. The
            // ≤5-row heuristic only bounds payload size for ordinary SELECTs.
            && (is_graph_tvf_engine(result.engine) || result.result.records.len() <= 5)
    }

    pub(super) fn read_result_cache(&self, runtime: &RedDBRuntime) -> Option<RuntimeQueryResult> {
        if runtime.resource_limits().read_execution_limits_enabled() {
            return None;
        }
        if self.can_read_result_cache() {
            runtime
                .get_result_cache_entry(self.cache_key())
                .map(|mut result| {
                    if let Some(vector) = result.result.stats.vector.as_mut() {
                        vector.cache_hit = true;
                    }
                    result
                })
        } else {
            None
        }
    }

    pub(super) fn write_result_cache(
        &self,
        runtime: &RedDBRuntime,
        result: &RuntimeQueryResult,
        scopes: HashSet<String>,
    ) {
        if self.should_write_result_cache(result) {
            runtime.put_result_cache_entry(
                self.cache_key(),
                RuntimeResultCacheEntry {
                    result: result.clone(),
                    cached_at: std::time::Instant::now(),
                    scopes,
                },
            );
        }
    }

    pub(super) fn prepare_cte(&self, query: &str) -> RedDBResult<Option<QueryExpr>> {
        // Detected via cheap prefix check so non-CTE queries skip the
        // full parse here. CTE-bearing queries bypass the plan cache
        // and result cache (rare workload — perf optimization is a
        // follow-up). Inlining substitutes every CTE reference with
        // its body as a subquery in FROM, after which the existing
        // subquery-in-FROM machinery handles execution. Recursive
        // CTEs are rejected explicitly until fixpoint execution wires
        // through the runtime.
        if !has_with_prefix(query) {
            return Ok(None);
        }
        let parsed =
            reddb_rql::parser::parse(query).map_err(|err| RedDBError::Query(err.to_string()))?;
        if parsed.with_clause.is_some() {
            let rewritten = crate::storage::query::executors::inline_ctes(parsed)
                .map_err(|err| RedDBError::Query(err.to_string()))?;
            return Ok(Some(rewritten));
        }
        // No WITH after parse (the prefix matched something else like
        // `WITHIN` that already routed elsewhere) — fall through to
        // the normal path with the original query.
        Ok(None)
    }

    pub(super) fn prepare_statement(
        &self,
        runtime: &RedDBRuntime,
        query: &str,
    ) -> RedDBResult<PreparedStatement> {
        let mode = detect_mode(query);
        if matches!(mode, QueryMode::Unknown) {
            return Err(RedDBError::Query("unable to detect query mode".to_string()));
        }

        // ── Plan cache: reuse only exact-query ASTs ──
        //
        // DML statements (INSERT/UPDATE/DELETE) almost always have unique literal
        // values, so caching them burns CPU on eviction bookkeeping (Vec::remove(0)
        // shifts the entire LRU list) with zero hit rate. Skip the cache entirely
        // Plan cache applies to statements whose shape can be
        // normalised + rebound (`UPDATE t SET x=? WHERE _entity_id=?`
        // reuses the same plan across thousands of varying literals).
        // INSERT is still bypassed — its shape changes per column set
        // and bulk paths don't go through here anyway.
        let first_word = query
            .trim()
            .split_ascii_whitespace()
            .next()
            .unwrap_or("")
            .to_ascii_uppercase();
        let is_insert = first_word == "INSERT";
        // #1370 — volatile queries ($config / $secret resolve mutable runtime
        // state at execution time) must bypass the plan cache too. A cached
        // optimized plan drops the live `$config` resolution, so a later
        // `SET CONFIG` would be ignored and the query would serve a stale value.
        // Re-parse fresh every time so the resolver runs against current state.
        let bypass_plan_cache = is_insert || self.is_volatile_query;

        // Fused normalize+extract: one byte-scan produces both the
        // cache_key AND the literal bindings. Saves a second Lexer
        // pass over the query text on every cache hit — dominant
        // cost on tight UPDATE loops that hit the same shape
        // thousands of times with varying literals.
        let (cache_key, prescan_binds) = if bypass_plan_cache {
            (String::new(), Vec::new())
        } else {
            crate::storage::query::planner::cache_key::normalize_and_extract(query)
        };

        let expr = if bypass_plan_cache {
            // Bypass plan cache for INSERT — shape varies per query.
            parse_multi(query).map_err(|err| RedDBError::Query(err.to_string()))?
        } else {
            // ── Hot path: read lock only (no writer serialization on cache hits) ──
            //
            // peek() is a non-mutating probe: no LRU promotion, no touch().
            // This lets concurrent readers proceed without blocking each other.
            // On hit we bind literals if needed and return immediately.
            // Only on miss do we drop to a write lock to parse + insert.
            let hit = {
                let plan_cache = runtime.inner.query_cache.read();
                plan_cache.peek(&cache_key).map(|cached| {
                    let parameter_count = cached.parameter_count;
                    let optimized = cached.plan.optimized.clone();
                    let exact_query = cached.exact_query.clone();
                    (parameter_count, optimized, exact_query)
                })
            };

            if let Some((parameter_count, optimized, exact_query)) = hit {
                if parameter_count > 0 {
                    // Shape hit: use the binds extracted during normalise.
                    let shape_binds = prescan_binds.clone();
                    if let Some(bound) =
                        crate::storage::query::planner::shape::bind_parameterized_query(
                            &optimized,
                            &shape_binds,
                            parameter_count,
                        )
                    {
                        bound
                    } else {
                        // A failed bind leaves placeholders in the cached AST,
                        // even for the same exact SQL. Never execute it unbound.
                        parse_multi(query).map_err(|err| RedDBError::Query(err.to_string()))?
                    }
                } else {
                    // No parameters means either there truly are no literals,
                    // or this statement type does not participate in shape
                    // parameterization (for example graph/queue commands).
                    // Reusing a normalized-cache hit across a different exact
                    // query can therefore leak stale literals into execution.
                    if exact_query.as_deref() == Some(query) {
                        optimized
                    } else {
                        parse_multi(query).map_err(|err| RedDBError::Query(err.to_string()))?
                    }
                }
            } else {
                // Cache miss — parse, parameterize, store.
                let parsed =
                    parse_multi(query).map_err(|err| RedDBError::Query(err.to_string()))?;
                let (cached_expr, parameter_count) = if let Some(prepared) =
                    crate::storage::query::planner::shape::parameterize_query_expr(&parsed)
                {
                    (prepared.shape, prepared.parameter_count)
                } else {
                    (parsed.clone(), 0)
                };
                {
                    let mut pc = runtime.inner.query_cache.write();
                    let plan = crate::storage::query::planner::QueryPlan::new(
                        parsed.clone(),
                        cached_expr,
                        Default::default(),
                    );
                    pc.insert(
                        cache_key.clone(),
                        crate::storage::query::planner::CachedPlan::new(plan)
                            .with_shape_key(cache_key.clone())
                            .with_exact_query(query.to_string())
                            .with_parameter_count(parameter_count),
                    );
                }
                parsed
            }
        };

        // Phase 5 PG parity: substitute any registered view name that
        // appears in the expression with its stored body. Runs after
        // parse and before dispatch so the SQL entrypoint gets the
        // same view resolution `execute_query_expr` already does.
        let expr = runtime.rewrite_view_refs(expr);

        Ok(PreparedStatement { expr, mode })
    }

    pub(super) fn check_query_privilege(
        &self,
        runtime: &RedDBRuntime,
        expr: &QueryExpr,
    ) -> RedDBResult<()> {
        // Frame-level coarse gate. We consult `required_privilege()`
        // (computed once at frame-build) against the captured identity
        // before the deep grant engine walks the parsed expression.
        // The coarse gate cannot ALLOW anything the grant engine would
        // deny — it only short-circuits the obvious "Role::Read tries
        // INSERT" case so a downstream caller never has to redo this
        // check inline. `Privilege::None` (transaction control / SET /
        // SHOW) flows through unchanged; the grant engine treats those
        // as bypass too.
        if let Some((username, role)) = <Self as ReadFrame>::identity(self) {
            let needed = <Self as ReadFrame>::required_privilege(self);
            if !needed.is_satisfied_by(role) {
                // Issue #205 — when the deep grant engine *also*
                // denies, we treat this as an ordinary permission
                // failure. But when an Admin-only statement reaches
                // this gate without an auth_store wired (so the deep
                // engine can't double-check), the coarse rejection is
                // the only line of defence — emit an OperatorEvent so
                // the operator notices an Admin-class statement was
                // attempted with insufficient role.
                if matches!(needed, Privilege::Admin) && runtime.inner.auth_store.read().is_none() {
                    crate::telemetry::operator_event::OperatorEvent::AuthBypass {
                        principal: username.to_string(),
                        resource: format!("statement requiring {needed:?}"),
                        detail: format!(
                            "auth_store not wired; coarse gate is sole defence (role={role:?})"
                        ),
                    }
                    .emit_global();
                }
                return Err(RedDBError::Query(format!(
                    "permission denied: principal=`{username}` role=`{role:?}` lacks {needed:?} privilege"
                )));
            }
        }
        runtime
            .check_query_privilege(expr)
            .map_err(|err| RedDBError::Query(format!("permission denied: {err}")))
    }

    pub(super) fn prepare_dispatch(
        &self,
        runtime: &RedDBRuntime,
        expr: &QueryExpr,
    ) -> RedDBResult<Option<crate::runtime::locking::LockerGuard>> {
        runtime.validate_model_operations_before_auth(expr)?;
        self.check_query_privilege(runtime, expr)?;
        self.release_snapshot_for_maintenance(expr);
        self.acquire_intent_locks(runtime, expr)
    }

    pub(super) fn release_snapshot_for_maintenance(&self, expr: &QueryExpr) {
        if matches!(
            expr,
            QueryExpr::MaintenanceCommand(reddb_rql::ast::MaintenanceCommand::Vacuum { .. })
        ) {
            // VACUUM inspects physical versions rather than reading its own
            // statement snapshot. Its temporary pin would otherwise inhibit
            // reclamation, especially with unused reserved autocommit xids.
            // Outer/nested reader frames keep their independent pins.
            self.snapshot_retention.borrow_mut().take();
        }
    }

    pub(super) fn acquire_intent_locks(
        &self,
        runtime: &RedDBRuntime,
        expr: &QueryExpr,
    ) -> RedDBResult<Option<crate::runtime::locking::LockerGuard>> {
        use crate::runtime::lock_manager::LockMode;
        use crate::runtime::locking::{LockerGuard, Resource};

        if !runtime.config_bool("concurrency.locking.enabled", true) {
            return Ok(None);
        }
        // Frame-level short-circuit: if the statement carries no lock
        // intent (transaction control, SET, SHOW), skip the lock
        // manager entirely instead of letting `intent_lock_modes_for`
        // walk the parsed expression to reach the same conclusion.
        if matches!(<Self as ReadFrame>::lock_intent(self), LockIntent::None) {
            return Ok(None);
        }
        let Some((global_mode, coll_mode)) = intent_lock_modes_for(expr) else {
            return Ok(None);
        };
        // A failed acquire (timeout, deadlock) used to be discarded and the
        // statement ran unlocked (#2376). Fail it instead; the client retries.
        let acquire = |guard: &mut LockerGuard, resource: Resource, mode: LockMode| {
            guard.acquire(resource.clone(), mode).map_err(|err| {
                RedDBError::Query(format!(
                    "could not acquire {mode:?} lock on {resource:?}: {err}; retry the statement"
                ))
            })
        };
        let mut guard = LockerGuard::new(runtime.inner.lock_manager.clone());
        acquire(&mut guard, Resource::Global, global_mode)?;
        // DDL takes X on its own collection, but every DDL also rewrites
        // shared catalog state the collection lock does not cover: the
        // system collections, the physical metadata and a full-store
        // `persist()`. Serialise all DDL on the catalog (#2376).
        if coll_mode == LockMode::Exclusive {
            acquire(&mut guard, Resource::Catalog, LockMode::Exclusive)?;
        }
        for collection in collections_referenced(expr) {
            acquire(&mut guard, Resource::Collection(collection), coll_mode)?;
        }
        Ok(Some(guard))
    }
}

impl ReadFrame for StatementExecutionFrame {
    fn snapshot_context(&self, runtime: &RedDBRuntime) -> SnapshotContext {
        SnapshotContext {
            snapshot: self.snapshot.clone(),
            manager: runtime.inner.transaction_state.snapshot_manager(),
            own_xids: self.own_xids.clone(),
            requires_index_fallback: self.requires_index_fallback,
            serializable_reader: self.serializable_reader,
        }
    }

    fn effective_scope(&self) -> Option<&str> {
        self.effective_scope.as_deref()
    }

    fn identity(&self) -> Option<(&str, Role)> {
        self.identity.as_ref().map(|(u, r)| (u.as_str(), *r))
    }

    fn snapshot(&self) -> &Snapshot {
        &self.snapshot
    }

    fn as_of_floor(&self) -> Option<Xid> {
        self.as_of_floor
    }

    fn cache_key(&self) -> &str {
        &self.cache_key
    }

    fn should_cache_result(&self) -> bool {
        !self.is_volatile_query && self.cache_safe
    }

    fn required_privilege(&self) -> Privilege {
        self.required_privilege
    }

    fn lock_intent(&self) -> LockIntent {
        self.lock_intent
    }

    fn visible_collections(&self) -> Option<&HashSet<String>> {
        self.visible_collections.as_ref()
    }
}

/// Lightweight `ReadFrame` carrier used by AI command entry points
/// (`SEARCH SIMILAR`, `SEARCH CONTEXT`, `ASK`).
///
/// Issue #119 calls this struct `EffectiveScope`. It bundles the
/// `(tenant, identity, role, visible_collections, snapshot)` tuple so
/// every AI runtime entry can pass *one* value to `AuthorizedSearch`
/// instead of re-reading thread-locals at every call site.
///
/// Built via `RedDBRuntime::ai_scope()` which sources tenant + identity
/// from the per-statement thread-locals (identical to how
/// `StatementExecutionFrame::build` derives them) and resolves
/// `visible_collections` via the `AuthStore` cache.
pub struct EffectiveScope {
    pub(crate) captured_snapshot: Option<SnapshotContext>,
    pub(crate) tenant: Option<String>,
    pub(crate) identity: Option<(String, Role)>,
    pub(crate) snapshot: Snapshot,
    pub(crate) visible_collections: Option<HashSet<String>>,
}

impl EffectiveScope {
    /// Capability check used by the AI runtime (`runtime/ai/ner.rs`)
    /// to gate LLM-backed NER calls behind `ai:ner:read`.
    ///
    /// Placeholder for now: always returns `false`. The auth engine's
    /// capability matrix is future work; until it lands, every routed
    /// LLM-NER call denies at the gate and `extract_tokens_routed`'s
    /// heuristic fallback fires (see `ask_pipeline::extract_tokens_routed`).
    /// Documented in code so the wire-up is a one-line change once
    /// the auth engine learns capabilities.
    pub fn has_capability(&self, _capability: &str) -> bool {
        false
    }
}

impl ReadFrame for EffectiveScope {
    fn snapshot_context(&self, runtime: &RedDBRuntime) -> SnapshotContext {
        self.captured_snapshot
            .clone()
            .unwrap_or_else(|| SnapshotContext {
                snapshot: self.snapshot.clone(),
                manager: runtime.inner.transaction_state.snapshot_manager(),
                own_xids: HashSet::new(),
                requires_index_fallback: true,
                serializable_reader: None,
            })
    }

    fn effective_scope(&self) -> Option<&str> {
        self.tenant.as_deref()
    }
    fn identity(&self) -> Option<(&str, Role)> {
        self.identity.as_ref().map(|(u, r)| (u.as_str(), *r))
    }
    fn snapshot(&self) -> &Snapshot {
        &self.snapshot
    }
    fn as_of_floor(&self) -> Option<Xid> {
        None
    }
    fn cache_key(&self) -> &str {
        ""
    }
    fn should_cache_result(&self) -> bool {
        false
    }
    fn required_privilege(&self) -> Privilege {
        Privilege::Read
    }
    fn lock_intent(&self) -> LockIntent {
        LockIntent::Shared
    }
    fn visible_collections(&self) -> Option<&HashSet<String>> {
        self.visible_collections.as_ref()
    }
}

fn own_transaction_xids(
    ctx: Option<&crate::storage::transaction::snapshot::TxnContext>,
) -> HashSet<Xid> {
    let mut set = HashSet::new();
    if let Some(ctx) = ctx {
        set.insert(ctx.xid);
        for (_, sub) in &ctx.savepoints {
            set.insert(*sub);
        }
        for sub in &ctx.released_sub_xids {
            set.insert(*sub);
        }
    }
    set
}

impl RedDBRuntime {
    /// Build the AI command `EffectiveScope` from the current
    /// statement thread-locals + auth store.
    ///
    /// Returns `None` for embedded callers (no auth store, no
    /// identity) — `AuthorizedSearch` treats `None` as deny-default.
    pub(crate) fn ai_scope(&self) -> EffectiveScope {
        let tenant = super::impl_core::current_tenant();
        let identity = super::impl_core::current_auth_identity();
        let captured_snapshot = super::execution_context::capture_current_snapshot();
        let snapshot = captured_snapshot.as_ref().map_or_else(
            || self.current_snapshot(),
            |context| context.snapshot.clone(),
        );
        let visible_collections = match (self.inner.auth_store.read().clone(), identity.as_ref()) {
            (Some(store), Some((principal, role))) => {
                let collections = self.inner.db.store().list_collections();
                Some(store.visible_collections_for_scope(
                    tenant.as_deref(),
                    *role,
                    principal,
                    &collections,
                ))
            }
            _ => None,
        };
        EffectiveScope {
            captured_snapshot,
            tenant,
            identity,
            snapshot,
            visible_collections,
        }
    }
}

/// Test fixtures for callers that need to drive `ReadFrame` without
/// booting a runtime. Lives behind `cfg(test)` and `pub(crate)` so it
/// only leaks across module boundaries inside the crate.
#[cfg(test)]
pub(crate) mod test_support {
    use super::{LockIntent, Privilege, ReadFrame};
    use crate::auth::Role;
    use crate::storage::transaction::snapshot::{Snapshot, Xid};
    use std::collections::HashSet;

    /// A `ReadFrame` impl with hand-set fields. Used by
    /// `authorized_search` tests to assert the deny-default and
    /// scope-trim behaviour without going through frame construction.
    pub(crate) struct FakeReadFrame {
        pub tenant: Option<String>,
        pub identity: Option<(String, Role)>,
        pub snapshot: Snapshot,
        pub visible: Option<HashSet<String>>,
    }

    impl FakeReadFrame {
        pub(crate) fn without_scope() -> Self {
            Self {
                tenant: None,
                identity: None,
                snapshot: Snapshot {
                    xid: 0,
                    in_progress: HashSet::new(),
                },
                visible: None,
            }
        }

        pub(crate) fn with_visible(visible: HashSet<String>) -> Self {
            Self {
                tenant: Some("acme".to_string()),
                identity: Some(("alice".to_string(), Role::Read)),
                snapshot: Snapshot {
                    xid: 0,
                    in_progress: HashSet::new(),
                },
                visible: Some(visible),
            }
        }
    }

    impl ReadFrame for FakeReadFrame {
        fn effective_scope(&self) -> Option<&str> {
            self.tenant.as_deref()
        }
        fn identity(&self) -> Option<(&str, Role)> {
            self.identity.as_ref().map(|(u, r)| (u.as_str(), *r))
        }
        fn snapshot(&self) -> &Snapshot {
            &self.snapshot
        }
        fn as_of_floor(&self) -> Option<Xid> {
            None
        }
        fn cache_key(&self) -> &str {
            ""
        }
        fn should_cache_result(&self) -> bool {
            false
        }
        fn required_privilege(&self) -> Privilege {
            Privilege::Read
        }
        fn lock_intent(&self) -> LockIntent {
            LockIntent::Shared
        }
        fn visible_collections(&self) -> Option<&HashSet<String>> {
            self.visible.as_ref()
        }
    }
}

impl RedDBRuntime {
    /// Resolve the snapshot for the current statement, returning
    /// the snapshot itself and (when AS OF is in effect) the
    /// resolved xid floor. The floor is the same xid carried inside
    /// `Snapshot.xid` for AS OF reads — exposing it separately lets
    /// the `ReadFrame` Interface tell "live read" from "historical
    /// read" without inferring from `in_progress.is_empty()`.
    ///
    /// A live read nested inside an already-installed frame inherits that
    /// frame's snapshot instead of minting a fresh one — but only when it
    /// is genuinely *part of* that statement, which is exactly what
    /// [`StatementIdentity::Expr`] means. A CTE prelude, a
    /// materialized-view refresh, or an inline graph TVF is one statement
    /// and must observe one snapshot; re-minting under autocommit / Read
    /// Committed takes a new high-water mark, so the nodes and edges of a
    /// single graph TVF could otherwise be materialized under two
    /// different snapshots.
    ///
    /// A nested `Text` statement is a different thing entirely: a driver
    /// loop issuing statement after statement through `execute_query`.
    /// Those are sequential autocommit statements and each must see its
    /// predecessors' commits. Inheriting there froze the loop's view —
    /// `apply_batched` (`impl_migrations.rs`) drives `APPLY MIGRATION …
    /// BATCH N ROWS` by re-running the body with `LIMIT N` until fewer
    /// than `N` rows are affected, so a frozen snapshot re-selected the
    /// same rows forever and the migration never converged. Twenty-five
    /// rows at `BATCH 7` should finish in four iterations; it hung
    /// unbounded instead.
    ///
    /// `AS OF` still wins over both — it names its own snapshot explicitly.
    fn statement_snapshot(
        &self,
        query: &str,
        identity: StatementIdentity<'_>,
    ) -> RedDBResult<(Snapshot, Option<Xid>, SnapshotRetentionGuard)> {
        match peek_top_level_as_of_with_table(query) {
            Some((spec, Some(table))) => {
                if !table.starts_with("red_")
                    && !table.starts_with("red.")
                    && !self.vcs_is_versioned(&table)?
                {
                    return Err(RedDBError::InvalidConfig(format!(
                        "AS OF requires a versioned collection — \
                         `{table}` has not opted in. \
                         Call vcs.set_versioned(\"{table}\", true) first."
                    )));
                }
                let xid = self.vcs_resolve_as_of(spec)?;
                let (snapshot, retention) = self.snapshot_manager().retain_snapshot(Snapshot {
                    xid,
                    in_progress: HashSet::new(),
                });
                Ok((snapshot, Some(xid), retention))
            }
            Some((spec, None)) => {
                let xid = self.vcs_resolve_as_of(spec)?;
                let (snapshot, retention) = self.snapshot_manager().retain_snapshot(Snapshot {
                    xid,
                    in_progress: HashSet::new(),
                });
                Ok((snapshot, Some(xid), retention))
            }
            None => match (identity, capture_current_snapshot()) {
                (StatementIdentity::Expr(_), Some(outer)) => {
                    let (snapshot, retention) =
                        self.snapshot_manager().retain_snapshot(outer.snapshot);
                    Ok((snapshot, None, retention))
                }
                _ => {
                    let (snapshot, retention) = self
                        .inner
                        .transaction_state
                        .retained_current_snapshot(current_connection_id());
                    Ok((snapshot, None, retention))
                }
            },
        }
    }

    fn result_cache_safe(&self, conn_id: u64) -> bool {
        let has_active_xids = self
            .inner
            .transaction_state
            .snapshot_manager()
            .oldest_active_xid()
            .is_some();
        let in_own_tx = self.inner.transaction_state.in_transaction(conn_id);
        !has_active_xids && !in_own_tx
    }
}

/// Whether a result's `engine` tag is one of the graph-analytics TVF
/// executors (issue #802). Graph-collection (`louvain(g)`) and inline
/// (`louvain(nodes => …, edges => …)`) forms both produce deterministic
/// algorithm output that is cached regardless of row count.
fn is_graph_tvf_engine(engine: &str) -> bool {
    matches!(engine, "runtime-graph-tvf" | "runtime-graph-tvf-inline")
}

fn result_cache_key(query: &str) -> String {
    let tenant = current_tenant().unwrap_or_default();
    let auth = current_auth_identity()
        .map(|(user, role)| format!("{}|{:?}", user, role))
        .unwrap_or_default();
    if tenant.is_empty() && auth.is_empty() {
        query.to_string()
    } else {
        format!("{query}\u{001e}{tenant}\u{001e}{auth}")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::api::RedDBOptions;
    use crate::runtime::impl_core::{
        clear_current_auth_identity, clear_current_tenant, set_current_auth_identity,
        set_current_tenant,
    };
    use crate::runtime::RedDBRuntime;

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

    enum HeldFrame {
        Ordinary,
        Nested,
        Historical,
    }

    fn verify_frame_survives_vacuum(delete: bool, kind: HeldFrame) {
        use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};

        reset_thread_locals();
        clear_current_connection_id();
        let runtime = fresh_runtime();
        let execute = |connection, sql: &str| {
            set_current_connection_id(connection);
            let result = runtime.execute_query(sql);
            clear_current_connection_id();
            result.expect("valid frame retention fixture statement")
        };
        execute(1, "CREATE TABLE frame_rows (id TEXT, n INTEGER)");
        execute(1, "INSERT INTO frame_rows (id, n) VALUES ('row', 1)");
        let historical_xid = runtime.current_snapshot().xid;
        if matches!(kind, HeldFrame::Historical) {
            runtime
                .vcs_set_versioned("frame_rows", true)
                .expect("enable historical reads");
        }
        execute(2, "BEGIN");
        let query = if matches!(kind, HeldFrame::Historical) {
            format!("SELECT id, n FROM frame_rows AS OF SNAPSHOT {historical_xid}")
        } else {
            "SELECT id, n FROM frame_rows".to_string()
        };
        let frame = StatementExecutionFrame::build(&runtime, StatementIdentity::Text(&query))
            .expect("frame captures its snapshot");
        let frame = if matches!(kind, HeldFrame::Nested) {
            let guards = frame.install(&runtime);
            let expr = QueryExpr::Table(reddb_rql::ast::TableQuery::new("frame_rows"));
            let nested = StatementExecutionFrame::build(&runtime, StatementIdentity::Expr(&expr))
                .expect("nested frame inherits the protected snapshot");
            drop(guards);
            drop(frame);
            nested
        } else {
            frame
        };
        // Pause after capture and before install/scan: GC must also protect
        // this gap, rather than starting retention only at TLS installation.
        execute(
            2,
            if delete {
                "DELETE FROM frame_rows WHERE id = 'row'"
            } else {
                "UPDATE frame_rows SET n = 2 WHERE id = 'row'"
            },
        );
        execute(2, "COMMIT");
        execute(1, "VACUUM frame_rows");
        {
            let _guards = frame.install(&runtime);
            let records = super::super::record_search::scan_runtime_table_source_records_limited(
                &runtime.db(),
                "frame_rows",
                None,
            )
            .expect("captured frame scan succeeds after vacuum");
            assert_eq!(
                records.len(),
                1,
                "vacuum reclaimed an active frame's visible row"
            );
            assert_eq!(
                records[0]
                    .get("n")
                    .expect("selected value exists")
                    .to_string(),
                "1"
            );
        }
        drop(frame);
        execute(1, "VACUUM frame_rows");
        let physical = runtime
            .db()
            .store()
            .get_collection("frame_rows")
            .expect("fixture collection exists")
            .query_all(|_| true)
            .len();
        assert_eq!(
            physical,
            usize::from(!delete),
            "completed frame leaked its retention pin"
        );
        clear_current_connection_id();
    }

    #[test]
    fn autocommit_frame_retains_update_until_scan_completes() {
        verify_frame_survives_vacuum(false, HeldFrame::Ordinary);
    }

    #[test]
    fn autocommit_frame_retains_delete_until_scan_completes() {
        verify_frame_survives_vacuum(true, HeldFrame::Ordinary);
    }

    #[test]
    fn nested_frame_retains_history_after_parent_frame_drops() {
        verify_frame_survives_vacuum(false, HeldFrame::Nested);
    }

    #[test]
    fn historical_frame_retains_history_until_scan_completes() {
        verify_frame_survives_vacuum(true, HeldFrame::Historical);
    }

    /// Ensure thread-local state from a prior test can't leak into
    /// the next one — tests in the same binary share the thread.
    fn reset_thread_locals() {
        clear_current_tenant();
        clear_current_auth_identity();
    }

    #[test]
    fn autocommit_select_takes_live_snapshot() {
        reset_thread_locals();
        let rt = fresh_runtime();
        let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Text("SELECT 1"))
            .expect("frame builds for SELECT 1");

        // The inclusive bound excludes the next unallocated writer xid.
        // A pristine manager therefore has a valid empty snapshot at zero.
        let f: &dyn ReadFrame = &frame;
        assert!(f.as_of_floor().is_none(), "live read has no AS OF floor");
        assert_eq!(f.snapshot().xid, rt.snapshot_manager().peek_next_xid() - 1);
        assert!(!f.snapshot().sees(rt.snapshot_manager().peek_next_xid(), 0));
    }

    #[test]
    fn autocommit_snapshot_does_not_lose_rows_after_index_publication() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE frame_index_race (id INT PRIMARY KEY, status TEXT)")
            .expect("create fixture");
        rt.execute_query("INSERT INTO frame_index_race (id, status) VALUES (1, 'old'), (2, 'old')")
            .expect("insert fixture");
        rt.execute_query("CREATE INDEX frame_status_hash ON frame_index_race (status) USING HASH")
            .expect("create index");
        let sql = "SELECT id FROM frame_index_race WHERE status = 'old'";
        let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Text(sql))
            .expect("capture autocommit snapshot");
        rt.execute_query("UPDATE frame_index_race SET status = 'new' WHERE id = 1")
            .expect("publish changed index entry");
        let reddb_rql::ast::QueryExpr::Table(query) =
            reddb_rql::modes::parse_multi(sql).expect("parse")
        else {
            panic!("expected table query");
        };
        let _guards = frame.install(&rt);
        let records =
            super::super::query_exec::table::execute_runtime_canonical_table_query_indexed(
                &rt.db(),
                &query,
                Some(rt.index_store_ref()),
            )
            .expect("read captured snapshot");
        let mut ids = records
            .iter()
            .map(|record| record.get("id").expect("id").to_string())
            .collect::<Vec<_>>();
        ids.sort();
        assert_eq!(
            ids,
            ["1", "2"],
            "current index omitted a visible historical row"
        );
    }

    #[test]
    fn captured_dml_snapshot_finds_rows_after_index_publication() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE frame_dml_race (id INT PRIMARY KEY, status TEXT)")
            .expect("create fixture");
        rt.execute_query("INSERT INTO frame_dml_race (id, status) VALUES (1, 'old'), (2, 'old')")
            .expect("insert fixture");
        rt.execute_query("CREATE INDEX frame_dml_status ON frame_dml_race (status) USING HASH")
            .expect("create index");
        let sql = "DELETE FROM frame_dml_race WHERE status = 'old'";
        let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Text(sql))
            .expect("capture DML snapshot");
        rt.execute_query("UPDATE frame_dml_race SET status = 'new' WHERE id = 1")
            .expect("publish new indexed value");
        let reddb_rql::ast::QueryExpr::Table(query) =
            reddb_rql::modes::parse_multi("SELECT id FROM frame_dml_race WHERE status = 'old'")
                .expect("parse filter")
        else {
            panic!("expected table query")
        };
        let filter = reddb_rql::sql_lowering::effective_table_filter(&query).expect("filter");
        let _guards = frame.install(&rt);
        let ids = super::super::dml_target_scan::DmlTargetScan::new(
            &rt,
            "frame_dml_race",
            Some(&filter),
            None,
        )
        .find_target_ids()
        .expect("find DML targets");
        assert_eq!(
            ids.len(),
            2,
            "DML target scan must discover both snapshot-visible rows"
        );
    }

    #[test]
    fn captured_snapshot_survives_indexed_update_and_delete_for_range_queries() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE frame_range_race (id INT PRIMARY KEY, score INT)")
            .expect("create fixture");
        rt.execute_query("INSERT INTO frame_range_race (id, score) VALUES (1, 10), (2, 20)")
            .expect("insert fixture");
        rt.execute_query("CREATE INDEX frame_score_range ON frame_range_race (score)")
            .expect("create index");
        let sql = "SELECT id FROM frame_range_race WHERE score >= 10 AND score <= 20 ORDER BY id";
        let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Text(sql))
            .expect("capture read snapshot");
        rt.execute_query("UPDATE frame_range_race SET score = 99 WHERE id = 1")
            .expect("update row");
        rt.execute_query("DELETE FROM frame_range_race WHERE id = 2")
            .expect("delete row");
        let reddb_rql::ast::QueryExpr::Table(query) =
            reddb_rql::modes::parse_multi(sql).expect("parse")
        else {
            panic!("expected table query");
        };
        let _guards = frame.install(&rt);
        let records =
            super::super::query_exec::table::execute_runtime_canonical_table_query_indexed(
                &rt.db(),
                &query,
                Some(rt.index_store_ref()),
            )
            .expect("read captured snapshot");
        let ids = records
            .iter()
            .map(|record| record.get("id").expect("id").to_string())
            .collect::<Vec<_>>();
        assert_eq!(ids, ["1", "2"]);
    }

    #[test]
    fn dml_target_scan_does_not_retry_inside_stored_function_budget() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::Arc;
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE dml_budget_guard (id INT, status TEXT)")
            .expect("create fixture");
        rt.execute_query("INSERT INTO dml_budget_guard (id, status) VALUES (1, 'old')")
            .expect("insert fixture");
        rt.execute_query("CREATE INDEX dml_budget_status ON dml_budget_guard (status) USING HASH")
            .expect("create index");
        let frame = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("DELETE FROM dml_budget_guard WHERE status = 'old'"),
        )
        .expect("capture snapshot");
        let reddb_rql::ast::QueryExpr::Table(query) =
            reddb_rql::modes::parse_multi("SELECT id FROM dml_budget_guard WHERE status = 'old'")
                .expect("parse filter")
        else {
            panic!("expected table query")
        };
        let filter = reddb_rql::sql_lowering::effective_table_filter(&query).expect("filter");
        let calls = Arc::new(AtomicUsize::new(0));
        let observed = Arc::clone(&calls);
        *rt.index_store_ref().mutation_hook.lock() = Some(Arc::new(move |collection, phase| {
            if collection == "dml_budget_guard"
                && phase == super::super::index_store::MutationTestPhase::DmlIndexLookupStarted
            {
                observed.fetch_add(1, Ordering::SeqCst);
            }
        }));
        let _budget = super::super::function_budget::Scope::enter(1_000, 10_000)
            .expect("stored function budget");
        let _guards = frame.install(&rt);
        let ids = super::super::dml_target_scan::DmlTargetScan::new(
            &rt,
            "dml_budget_guard",
            Some(&filter),
            None,
        )
        .find_target_ids()
        .expect("find DML targets");
        assert_eq!(ids.len(), 1);
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "bounded stored functions must not retry indexed discovery"
        );
        *rt.index_store_ref().mutation_hook.lock() = None;
    }

    #[test]
    fn dml_index_lookup_retries_when_publication_races_target_selection() {
        use std::sync::{mpsc, Arc, Mutex};
        use std::time::Duration;
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE dml_lookup_race (id INT PRIMARY KEY, status TEXT)")
            .expect("create fixture");
        rt.execute_query("INSERT INTO dml_lookup_race (id, status) VALUES (1, 'old'), (2, 'old')")
            .expect("insert fixture");
        rt.execute_query("CREATE INDEX dml_lookup_status ON dml_lookup_race (status) USING HASH")
            .expect("create index");
        let frame = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("DELETE FROM dml_lookup_race WHERE status = 'old'"),
        )
        .expect("capture snapshot");
        let reddb_rql::ast::QueryExpr::Table(query) =
            reddb_rql::modes::parse_multi("SELECT id FROM dml_lookup_race WHERE status = 'old'")
                .expect("parse filter")
        else {
            panic!("expected table query")
        };
        let filter = reddb_rql::sql_lowering::effective_table_filter(&query).expect("filter");
        let (started_tx, started_rx) = mpsc::channel();
        let (published_tx, published_rx) = mpsc::channel();
        let published_rx = Mutex::new(published_rx);
        let first = std::sync::atomic::AtomicBool::new(true);
        *rt.index_store_ref().mutation_hook.lock() = Some(Arc::new(move |collection, phase| {
            if collection == "dml_lookup_race"
                && phase == super::super::index_store::MutationTestPhase::DmlIndexLookupStarted
                && first.swap(false, std::sync::atomic::Ordering::SeqCst)
            {
                started_tx.send(()).expect("notify writer");
                published_rx
                    .lock()
                    .unwrap()
                    .recv_timeout(Duration::from_secs(10))
                    .expect("writer published");
            }
        }));
        std::thread::scope(|scope| {
            let writer_rt = &rt;
            let writer = scope.spawn(move || {
                started_rx
                    .recv_timeout(Duration::from_secs(10))
                    .expect("lookup started");
                writer_rt
                    .execute_query("UPDATE dml_lookup_race SET status = 'new' WHERE id = 1")
                    .expect("publish update");
                published_tx.send(()).expect("release lookup");
            });
            let limits = super::super::resource_limits::ResourceLimits {
                max_query_duration: Some(Duration::from_secs(30)),
                ..Default::default()
            };
            let slots = std::sync::atomic::AtomicU64::new(0);
            let _budget = super::super::query_budget::Scope::enter(&limits, &slots)
                .expect("query deadline scope");
            let _guards = frame.install(&rt);
            let ids = super::super::dml_target_scan::DmlTargetScan::new(
                &rt,
                "dml_lookup_race",
                Some(&filter),
                None,
            )
            .find_target_ids()
            .expect("find DML targets");
            writer.join().expect("writer thread");
            assert_eq!(
                ids.len(),
                2,
                "racing publication must not omit a snapshot-visible row"
            );
        });
        *rt.index_store_ref().mutation_hook.lock() = None;
    }

    #[test]
    fn replica_replay_preserves_snapshots_across_insert_update_and_delete() {
        use crate::replication::cdc::{change_record_from_entity, ChangeOperation};
        use crate::replication::logical::{ApplyMode, LogicalChangeApplier};
        reset_thread_locals();
        let source = fresh_runtime();
        let replica = fresh_runtime();
        for rt in [&source, &replica] {
            rt.execute_query("CREATE TABLE replica_snapshot (id INT, status TEXT)")
                .expect("schema");
        }
        replica
            .execute_query(
                "CREATE INDEX replica_snapshot_status ON replica_snapshot (status) USING HASH",
            )
            .expect("index");
        let sql = "SELECT id FROM replica_snapshot WHERE status = 'old'";
        let capture = || {
            StatementExecutionFrame::build(&replica, StatementIdentity::Text(sql)).expect("capture")
        };
        let read = |frame: &StatementExecutionFrame| {
            let reddb_rql::ast::QueryExpr::Table(query) =
                reddb_rql::modes::parse_multi(sql).expect("parse")
            else {
                panic!("table")
            };
            let _guards = frame.install(&replica);
            super::super::query_exec::table::execute_runtime_canonical_table_query_indexed(
                &replica.db(),
                &query,
                Some(replica.index_store_ref()),
            )
            .expect("snapshot read")
            .len()
        };
        let empty = capture();
        source
            .execute_query("INSERT INTO replica_snapshot (id, status) VALUES (1, 'old')")
            .expect("insert");
        let applier = LogicalChangeApplier::new(0);
        let replay = |lsn, op| {
            let entity = source
                .db()
                .store()
                .get_collection("replica_snapshot")
                .unwrap()
                .query_all(|entity| entity.xmax == 0)
                .pop()
                .unwrap();
            applier
                .apply(
                    &replica.db(),
                    &change_record_from_entity(
                        lsn,
                        1,
                        op,
                        "replica_snapshot",
                        "table",
                        &entity,
                        12,
                        None,
                    ),
                    ApplyMode::Replica,
                )
                .expect("replay");
        };
        replay(1, ChangeOperation::Insert);
        assert_eq!(
            read(&empty),
            0,
            "replay must not backdate an insert into an earlier snapshot"
        );
        let inserted = capture();
        source
            .execute_query("UPDATE replica_snapshot SET status = 'new' WHERE id = 1")
            .expect("update");
        replay(2, ChangeOperation::Update);
        assert_eq!(read(&inserted), 1, "old version remains visible");
        assert_eq!(read(&capture()), 0, "fresh snapshot uses the updated index");
        // A snapshot of all rows must retain the deleted version as well.
        let all_sql = "SELECT id FROM replica_snapshot";
        let before_delete =
            StatementExecutionFrame::build(&replica, StatementIdentity::Text(all_sql)).unwrap();
        replay(3, ChangeOperation::Delete);
        let reddb_rql::ast::QueryExpr::Table(query) =
            reddb_rql::modes::parse_multi(all_sql).unwrap()
        else {
            panic!("table")
        };
        {
            let _guards = before_delete.install(&replica);
            assert_eq!(
                super::super::query_exec::table::execute_runtime_canonical_table_query_indexed(
                    &replica.db(),
                    &query,
                    Some(replica.index_store_ref())
                )
                .unwrap()
                .len(),
                1
            );
        }
        assert_eq!(
            replica.execute_query(all_sql).unwrap().result.records.len(),
            0
        );
    }

    #[test]
    fn autocommit_index_reads_keep_old_rows_while_another_transaction_is_open() {
        use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};
        reset_thread_locals();
        let rt = fresh_runtime();
        let execute = |connection, sql: &str| {
            set_current_connection_id(connection);
            let result = rt.execute_query(sql);
            clear_current_connection_id();
            result.expect("statement")
        };
        execute(1, "CREATE TABLE active_writer_rows (id INT, status TEXT)");
        execute(
            1,
            "INSERT INTO active_writer_rows (id, status) VALUES (1, 'old')",
        );
        execute(
            1,
            "CREATE INDEX active_writer_status ON active_writer_rows (status) USING HASH",
        );
        execute(2, "BEGIN");
        execute(
            2,
            "UPDATE active_writer_rows SET status = 'new' WHERE id = 1",
        );
        assert_eq!(
            execute(1, "SELECT id FROM active_writer_rows WHERE status = 'old'")
                .result
                .records
                .len(),
            1
        );
        assert_eq!(
            execute(1, "SELECT id FROM active_writer_rows WHERE status = 'new'")
                .result
                .records
                .len(),
            0
        );
        execute(2, "ROLLBACK");
        assert_eq!(
            execute(1, "SELECT id FROM active_writer_rows WHERE status = 'old'")
                .result
                .records
                .len(),
            1
        );
    }

    #[test]
    fn frame_captures_identity_and_scope() {
        reset_thread_locals();
        set_current_tenant("acme".to_string());
        set_current_auth_identity("alice".to_string(), Role::Write);

        let rt = fresh_runtime();
        let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Text("SELECT 1"))
            .expect("frame builds");
        let f: &dyn ReadFrame = &frame;

        assert_eq!(f.effective_scope(), Some("acme"));
        let id = f.identity().expect("identity captured");
        assert_eq!(id.0, "alice");
        assert!(matches!(id.1, Role::Write));

        // Cache key mixes scope + identity so two callers under
        // different tenants never share a cache slot.
        assert!(
            f.cache_key().contains("acme") && f.cache_key().contains("alice"),
            "cache key folds in scope + identity, got {:?}",
            f.cache_key()
        );

        reset_thread_locals();
    }

    #[test]
    fn as_of_rejects_non_versioned_user_collection() {
        reset_thread_locals();
        let rt = fresh_runtime();

        // `not_versioned` is a plain user collection — the frame
        // builder must reject AS OF until the caller opts in via
        // `vcs.set_versioned`.
        let err = match StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("SELECT * FROM not_versioned AS OF COMMIT 'deadbeef'"),
        ) {
            Err(e) => e,
            Ok(_) => panic!("AS OF on non-versioned user collection rejected"),
        };

        let msg = format!("{err}");
        assert!(
            msg.contains("AS OF requires a versioned collection"),
            "expected AS OF rejection, got: {msg}"
        );
    }

    /// End-to-end proof that the SELECT path consumes a `ReadFrame`.
    ///
    /// Sets a tenant + identity via the public thread-local API the
    /// runtime uses for ambient scope, drives a real `SELECT` through
    /// `execute_query`, then inspects the result cache that the SELECT
    /// path populates via `frame.cache_key()`. The key only carries
    /// the tenant + identity *because* it was built through the frame —
    /// reverting the wiring to inline `current_tenant()` /
    /// `current_auth_identity()` reads would still pass this test, but
    /// dropping the frame entirely (so the SELECT path stopped touching
    /// `cache_key`) would break it.
    #[test]
    fn select_path_routes_through_frame_cache_key() {
        reset_thread_locals();
        set_current_tenant("acme".to_string());
        set_current_auth_identity("alice".to_string(), Role::Read);

        let rt = fresh_runtime();
        let result = rt
            .execute_query("SELECT 1")
            .expect("SELECT 1 executes under tenant=acme/identity=alice");
        assert_eq!(result.statement_type, "select");

        // The textual SELECT path builds a frame and
        // writes its result through `frame.cache_key()`. That key folds
        // tenant + identity in via `result_cache_key`, so finding "acme"
        // and "alice" inside any cached key proves the frame was the
        // seam used.
        let cache = rt.inner.result_cache.read();
        let any_keyed_with_scope = cache
            .0
            .keys()
            .any(|k| k.contains("acme") && k.contains("alice"));
        assert!(
            any_keyed_with_scope,
            "expected at least one result-cache key carrying tenant+identity, \
             got keys: {:?}",
            cache.0.keys().collect::<Vec<_>>()
        );

        reset_thread_locals();
    }

    /// A SELECT that calls a volatile builtin (here:
    /// `pg_advisory_unlock`, the volatile token the runtime currently
    /// recognises in `query_has_volatile_builtin`) must NOT populate
    /// the result cache. Any caller hitting the cache after this would
    /// see a stale answer for an inherently-volatile query, so the
    /// SELECT path gates writes through `frame.should_cache_result()`.
    ///
    /// Deletion test: removing `ReadFrame::should_cache_result`, or
    /// reverting the SELECT path to skip its safety gate, would let
    /// the result cache silently absorb this statement and break the
    /// assertion below.
    #[test]
    fn volatile_select_does_not_populate_result_cache() {
        reset_thread_locals();
        let rt = fresh_runtime();

        // Frame-level invariant: the volatile-builtin signal collapses
        // `should_cache_result` to false even for an autocommit /
        // out-of-tx connection.
        let frame = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("SELECT pg_advisory_unlock(1)"),
        )
        .expect("frame");
        let f: &dyn ReadFrame = &frame;
        assert!(
            !f.should_cache_result(),
            "volatile builtin must disable result-cache safety"
        );

        // End-to-end: drive the volatile SELECT through `execute_query`
        // and confirm no entry was stamped under its cache key. Other
        // entries from prior tests sharing the binary may exist, so we
        // assert specifically on this query's key.
        let _ = rt
            .execute_query("SELECT pg_advisory_unlock(1)")
            .expect("volatile SELECT executes");
        let cache = rt.inner.result_cache.read();
        let key = result_cache_key("SELECT pg_advisory_unlock(1)");
        assert!(
            !cache.0.contains_key(&key),
            "volatile SELECT must not populate result cache, found key {key:?} in {:?}",
            cache.0.keys().collect::<Vec<_>>()
        );

        reset_thread_locals();
    }

    #[test]
    fn blob_cache_backend_populates_blob_path_without_legacy_write() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.inner
            .db
            .store()
            .set_config_tree("runtime.result_cache.backend", &crate::json!("blob_cache"));

        let result = rt.execute_query("SELECT 1").expect("SELECT 1 executes");
        assert_eq!(result.statement_type, "select");

        let key = result_cache_key("SELECT 1");
        assert!(
            rt.inner
                .result_blob_cache
                .get("runtime.result_cache", &key)
                .is_some(),
            "blob backend should stamp the Blob Cache path"
        );
        assert!(rt.inner.result_blob_entries.read().0.contains_key(&key));
        assert!(
            !rt.inner.result_cache.read().0.contains_key(&key),
            "blob backend should not write the legacy map"
        );
    }

    #[test]
    fn blob_cache_backend_keeps_volatile_select_out_of_blob_path() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.inner
            .db
            .store()
            .set_config_tree("runtime.result_cache.backend", &crate::json!("blob_cache"));

        let _ = rt
            .execute_query("SELECT pg_advisory_unlock(1)")
            .expect("volatile SELECT executes");
        let key = result_cache_key("SELECT pg_advisory_unlock(1)");
        assert!(
            rt.inner
                .result_blob_cache
                .get("runtime.result_cache", &key)
                .is_none(),
            "volatile SELECT must not populate blob result cache"
        );
        assert!(!rt.inner.result_blob_entries.read().0.contains_key(&key));
    }

    #[test]
    fn shadow_backend_dual_writes_and_reports_no_divergence_on_equal_results() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.inner
            .db
            .store()
            .set_config_tree("runtime.result_cache.backend", &crate::json!("shadow"));

        let first = rt.execute_query("SELECT 1").expect("first SELECT");
        let second = rt.execute_query("SELECT 1").expect("cached SELECT");
        assert_eq!(first.result.len(), second.result.len());

        let key = result_cache_key("SELECT 1");
        assert!(rt.inner.result_cache.read().0.contains_key(&key));
        assert!(rt.inner.result_blob_entries.read().0.contains_key(&key));
        assert_eq!(rt.result_cache_shadow_divergences(), 0);
        assert_eq!(
            crate::runtime::METRIC_CACHE_SHADOW_DIVERGENCE_TOTAL,
            "cache_shadow_divergence_total"
        );
    }

    #[test]
    fn as_of_on_red_collection_records_floor() {
        reset_thread_locals();
        let rt = fresh_runtime();

        // `red_*` collections always allow AS OF. The frame should
        // resolve to a concrete xid and surface it via the Interface.
        let frame = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("SELECT * FROM red_commits AS OF SNAPSHOT 1"),
        )
        .expect("AS OF SNAPSHOT 1 on red_commits resolves");

        let f: &dyn ReadFrame = &frame;
        assert_eq!(
            f.as_of_floor(),
            Some(1),
            "AS OF SNAPSHOT 1 records xid=1 as the floor"
        );
        assert_eq!(f.snapshot().xid, 1);
        assert!(
            f.snapshot().in_progress.is_empty(),
            "historical reads have no in-progress set"
        );
    }

    /// The frame classifies common SQL prefixes into the coarse
    /// `Privilege` / `LockIntent` buckets at build time. This test
    /// pins the mapping so a regression that silently re-routes
    /// (e.g. INSERT classified as Read) surfaces here, not at a
    /// downstream privilege gate.
    #[test]
    fn frame_classifies_privilege_and_lock_intent_from_prefix() {
        reset_thread_locals();
        let rt = fresh_runtime();

        let cases = [
            ("SELECT 1", Privilege::Read, LockIntent::Shared),
            ("LIST KV settings", Privilege::Read, LockIntent::Shared),
            (
                "KV GET settings.feature",
                Privilege::Read,
                LockIntent::Shared,
            ),
            ("VAULT LIST secrets", Privilege::Read, LockIntent::Shared),
            (
                "INSERT INTO t (id) VALUES (1)",
                Privilege::Write,
                LockIntent::Exclusive,
            ),
            (
                "KV PUT settings.feature = 'on'",
                Privilege::Write,
                LockIntent::Exclusive,
            ),
            (
                "VAULT PUT secrets.api = 'x'",
                Privilege::Write,
                LockIntent::Exclusive,
            ),
            (
                "UPDATE t SET x = 1 WHERE id = 1",
                Privilege::Write,
                LockIntent::Exclusive,
            ),
            (
                "DELETE FROM t WHERE id = 1",
                Privilege::Write,
                LockIntent::Exclusive,
            ),
            (
                "CREATE TABLE foo (id INT)",
                Privilege::Write,
                LockIntent::Exclusive,
            ),
            ("BEGIN", Privilege::None, LockIntent::None),
            ("COMMIT", Privilege::None, LockIntent::None),
            ("SET timezone = 'UTC'", Privilege::None, LockIntent::None),
        ];

        for (q, want_priv, want_lock) in cases {
            let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Text(q))
                .unwrap_or_else(|e| panic!("frame builds for {q:?}: {e}"));
            let f: &dyn ReadFrame = &frame;
            assert_eq!(f.required_privilege(), want_priv, "privilege for {q:?}");
            assert_eq!(f.lock_intent(), want_lock, "lock intent for {q:?}");
        }
    }

    /// Deletion-test for `ReadFrame::required_privilege`: a SELECT
    /// driven through `execute_query` under an identity whose role
    /// doesn't satisfy the frame's coarse `Read` privilege gets
    /// denied with the frame's signal.
    ///
    /// We test the gate by classifying an INSERT (which the frame
    /// reports as `Privilege::Write`) under `Role::Read` — the only
    /// pair the legacy fallback would also reject, but here the
    /// rejection comes through `frame.check_query_privilege` BEFORE
    /// the parsed-expression walker runs. Removing
    /// `required_privilege` (or the `is_satisfied_by` consult inside
    /// `check_query_privilege`) would force the deny path back to the
    /// inline `RedDBRuntime::check_query_privilege` walker — but the
    /// auth_store gate up there is bypassed when no auth_store is
    /// wired (embedded test mode), so this test would FLIP from
    /// denied to permitted and break the assertion below.
    #[test]
    fn insert_under_read_role_denied_via_frame_privilege() {
        reset_thread_locals();
        set_current_auth_identity("alice".to_string(), Role::Read);

        let rt = fresh_runtime();
        // Bypass parser by reaching into the frame directly: the
        // frame derives privilege from the SQL prefix without
        // needing an auth_store wired up. Driving end-to-end via
        // `execute_query` would also reject (no table `t`), but for
        // a different reason — we want to pin the privilege seam.
        let frame = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("INSERT INTO t (id) VALUES (1)"),
        )
        .expect("frame builds for INSERT");
        let f: &dyn ReadFrame = &frame;
        assert_eq!(
            f.required_privilege(),
            Privilege::Write,
            "INSERT classified as Write"
        );
        let id = f.identity().expect("identity captured");
        assert!(
            !f.required_privilege().is_satisfied_by(id.1),
            "Role::Read does not satisfy Privilege::Write — frame must deny"
        );

        // End-to-end: the frame's `check_query_privilege` sees the
        // (Read role, Write privilege) mismatch and denies before
        // dispatch. We drive a synthetic `QueryExpr::Table` because
        // the SELECT/INSERT parser would happen to also fail, and we
        // want the failure to come from the privilege seam.
        use reddb_rql::ast::{QueryExpr, TableQuery};
        let expr = QueryExpr::Table(TableQuery::new("t"));
        let err = frame
            .check_query_privilege(&rt, &expr)
            .expect_err("denied via frame's coarse privilege gate");
        let msg = format!("{err}");
        assert!(
            msg.contains("permission denied") && msg.contains("Write"),
            "expected frame-level Write deny, got: {msg}"
        );

        reset_thread_locals();
    }

    /// End-to-end proof that the frame-owned row-buffer arena (#885) is
    /// wired into the SELECT path and produces observable results
    /// byte-identical to the per-request-allocation baseline.
    ///
    /// A table with more rows than the streaming high-water mark
    /// (`DEFAULT_HIGH_WATER_MARK`) forces the `execute_runtime_table_query_in`
    /// path to assemble many chunks, each leasing/recycling the frame
    /// arena's single chunk buffer. Driving it through `execute_query`
    /// (which builds a `StatementExecutionFrame` and lends its arena)
    /// must return every inserted row, in order — exactly what the
    /// allocate-per-chunk path returned. A bug in the arena wiring
    /// (dropped rows, bled rows, mis-ordering) would surface here.
    #[test]
    fn large_select_through_frame_arena_returns_all_rows_in_order() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE big (id INT)")
            .expect("create table");

        // > DEFAULT_HIGH_WATER_MARK (1024) rows so the streaming channel
        // spans multiple chunks and the arena buffer is reused.
        const N: usize = 2_500;
        for start in (0..N).step_by(250) {
            let end = (start + 250).min(N);
            let values = (start..end)
                .map(|i| format!("({i})"))
                .collect::<Vec<_>>()
                .join(", ");
            rt.execute_query(&format!("INSERT INTO big (id) VALUES {values}"))
                .unwrap_or_else(|err| panic!("insert rows {start}..{end}: {err:?}"));
        }

        let result = rt
            .execute_query("SELECT id FROM big ORDER BY id")
            .expect("large SELECT executes through the frame arena path");
        assert_eq!(result.statement_type, "select");
        assert_eq!(
            result.result.records.len(),
            N,
            "every inserted row streams back through the arena-backed channel"
        );
        for (i, record) in result.result.records.iter().enumerate() {
            assert_eq!(
                record.get("id"),
                Some(&reddb_types::Value::Integer(i as i64)),
                "row {i} is byte-identical to the per-request-allocation baseline"
            );
        }

        reset_thread_locals();
    }

    /// Transport adapters may decode their wire-specific parameter value
    /// shapes, but SQL parsing/binding must stay behind the runtime's
    /// statement entrypoint. This pins the deeper seam introduced for
    /// parameterized query execution. HTTP, JSON-RPC, RedWire, and gRPC hand
    /// a `QueryRequest` to the Request module, as does PG wire through the
    /// shared prepared registry. All paths install a real
    /// `StatementExecutionFrame` before dispatch.
    ///
    /// The Request transports reach the runtime entrypoint one layer deeper:
    /// `QueryRequestExecutor` calls `execute_query_with_params` on their
    /// behalf. Their delegation target is the Request seam, so that is what
    /// is pinned here — the "no binding in the adapter" half is unchanged
    /// for all.
    #[test]
    fn parameterized_transport_adapters_delegate_binding_to_runtime() {
        let manifest_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        let adapters = [
            ("src/server/handlers_query.rs", "QueryRequestExecutor"),
            ("src/rpc_stdio.rs", "QueryRequestExecutor"),
            ("src/wire/redwire/session.rs", "QueryRequestExecutor"),
            ("src/wire/postgres/server.rs", "QueryRequestExecutor"),
            ("src/grpc.rs", "QueryRequestExecutor"),
        ];

        for (relative, delegation) in adapters {
            let path = manifest_dir.join(relative);
            let text = std::fs::read_to_string(&path)
                .unwrap_or_else(|err| panic!("read {}: {err}", path.display()));
            assert!(
                text.contains(delegation),
                "{relative} should delegate parameterized query execution through {delegation}"
            );
            assert!(
                !text.contains("user_params::bind"),
                "{relative} must not bind SQL params in the transport adapter"
            );
        }

        // The Request module is the only place RedWire's delegation may
        // bottom out: it must still reach the runtime entrypoint itself.
        let seam = std::fs::read_to_string(manifest_dir.join("src/runtime/query_request.rs"))
            .expect("read src/runtime/query_request.rs");
        assert!(
            seam.contains("execute_query_with_params"),
            "the Request module must execute text SQL through the runtime entrypoint"
        );
    }

    /// Deletion-test for `ReadFrame::lock_intent`: a transaction
    /// control statement carries `LockIntent::None` and the
    /// `acquire_intent_locks` path returns `None` without consulting
    /// `intent_lock_modes_for`. Removing the method (or its consult
    /// site in `acquire_intent_locks`) would force the lock-mode
    /// helper to walk a fabricated parsed expression to reach the
    /// same conclusion — but the assertion that no guard is allocated
    /// for a `BEGIN` frame would still hold, so we additionally pin
    /// the classifier mapping above to make the deletion observable.
    #[test]
    fn control_statement_skips_intent_locks_via_frame() {
        reset_thread_locals();
        let rt = fresh_runtime();

        let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Text("BEGIN"))
            .expect("frame builds for BEGIN");
        let f: &dyn ReadFrame = &frame;
        assert_eq!(f.lock_intent(), LockIntent::None);

        // Drive `acquire_intent_locks` against a fabricated SELECT
        // expression that WOULD normally yield `(IS, IS)`; the frame's
        // `lock_intent() == None` short-circuit must still suppress
        // the guard.
        use reddb_rql::ast::{QueryExpr, TableQuery};
        let expr = QueryExpr::Table(TableQuery::new("t"));
        let guard = frame
            .acquire_intent_locks(&rt, &expr)
            .expect("no lock is attempted");
        assert!(
            guard.is_none(),
            "BEGIN frame's lock_intent=None must short-circuit lock acquisition"
        );
    }

    /// Nested sub-execution inside one statement (CTE prelude,
    /// materialized-view refresh, inline graph TVF) must read the
    /// enclosing statement's snapshot. Autocommit is the case that used
    /// to break: `current_snapshot()` mints a fresh high-water mark per
    /// call, so an inline graph TVF could materialize its nodes and its
    /// edges under two different snapshots within one statement.
    #[test]
    fn nested_frame_inherits_outer_statement_snapshot() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE nested_snap (id INT)")
            .expect("create table");

        let outer = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("SELECT * FROM nested_snap"),
        )
        .expect("outer frame builds");
        let _outer_guards = outer.install(&rt);

        // A write between the two builds moves the high-water mark, so a
        // re-minting nested frame would land on a different xid.
        rt.execute_query("INSERT INTO nested_snap (id) VALUES (1)")
            .expect("concurrent insert");

        let expr = QueryExpr::Table(reddb_rql::ast::TableQuery::new("nested_snap"));
        let nested = StatementExecutionFrame::build(&rt, StatementIdentity::Expr(&expr))
            .expect("nested frame builds");

        assert_eq!(
            ReadFrame::snapshot(&nested).xid,
            ReadFrame::snapshot(&outer).xid,
            "nested execution must inherit the outer statement's snapshot"
        );
    }

    /// The other side of the same rule, and the one that matters more:
    /// a nested statement built from *text* is a statement in its own
    /// right, not part of the enclosing one, so it must mint a fresh
    /// snapshot and observe everything committed before it.
    ///
    /// `apply_batched` (`impl_migrations.rs`) is the load-bearing caller:
    /// it drives `APPLY MIGRATION … BATCH N ROWS` by re-running the body
    /// with `LIMIT N` through `execute_query` until fewer than `N` rows
    /// are affected. Inheriting the outer `APPLY MIGRATION` statement's
    /// snapshot froze that loop's view, so every iteration re-selected
    /// the same rows and the migration never converged — 25 rows at
    /// `BATCH 7` hung instead of finishing in four iterations.
    #[test]
    fn nested_text_statement_sees_commits_made_since_the_outer_frame() {
        reset_thread_locals();
        let rt = fresh_runtime();
        rt.execute_query("CREATE TABLE nested_text_snap (id INT)")
            .expect("create table");

        let outer = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("SELECT * FROM nested_text_snap"),
        )
        .expect("outer frame builds");
        let _outer_guards = outer.install(&rt);

        rt.execute_query("INSERT INTO nested_text_snap (id) VALUES (1)")
            .expect("concurrent insert");

        let nested = StatementExecutionFrame::build(
            &rt,
            StatementIdentity::Text("SELECT * FROM nested_text_snap"),
        )
        .expect("nested frame builds");

        assert_ne!(
            ReadFrame::snapshot(&nested).xid,
            ReadFrame::snapshot(&outer).xid,
            "a nested text statement must mint a fresh snapshot, not inherit \
             the enclosing statement's frozen view"
        );
    }

    /// A frame with no statement text must not silently answer an
    /// `AS OF` read with live data. `TableQuery::as_of` is consumed
    /// nowhere in the runtime or scan path, so the only correct
    /// behaviours are "resolve it" (text path) or "refuse".
    #[test]
    fn expr_frame_rejects_as_of_instead_of_reading_live() {
        reset_thread_locals();
        let rt = fresh_runtime();

        let mut table = reddb_rql::ast::TableQuery::new("red_commits");
        table.as_of = Some(reddb_rql::ast::AsOfClause::Snapshot(1));
        let expr = QueryExpr::Table(table);

        let err = match StatementExecutionFrame::build(&rt, StatementIdentity::Expr(&expr)) {
            Err(err) => err,
            Ok(_) => panic!("AS OF without statement text must be refused, not read live"),
        };
        let msg = format!("{err}");
        assert!(
            msg.contains("AS OF cannot be resolved from a pre-parsed expression"),
            "expected a loud AS OF refusal, got: {msg}"
        );
    }

    /// The expression path has no stable text key, so it must opt out of
    /// the result cache entirely. A kind label (`"table"`, `"insert"`)
    /// would key every statement of the same kind to one cache slot.
    #[test]
    fn expr_frame_opts_out_of_the_result_cache() {
        reset_thread_locals();
        let rt = fresh_runtime();

        let expr = QueryExpr::Table(reddb_rql::ast::TableQuery::new("t"));
        let frame = StatementExecutionFrame::build(&rt, StatementIdentity::Expr(&expr))
            .expect("expr frame builds");

        assert!(frame.cache_key().is_empty(), "expr frame has no cache key");
        assert!(
            !frame.can_read_result_cache(),
            "expr frame must never serve from the result cache"
        );
    }
}