wp_knowledge/

runtime.rs

use std::collections::{HashMap, HashSet, hash_map::DefaultHasher};
use std::hash::{Hash, Hasher};
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock, RwLock};
use std::time::{Duration, Instant};

/// 默认 provider 在命名注册表中的键名，与配置层 [`crate::loader::DEFAULT_SQLDB_NAME`] 保持一致。
pub const DEFAULT_PROVIDER_NAME: &str = "default";

use crate::error::{KnowReason, KnowledgeResult};
use async_trait::async_trait;
use lru::LruCache;
use orion_error::conversion::ToStructError;
use tokio::task;
use wp_log::{debug_kdb, warn_kdb};
use wp_model_core::model::{DataField, DataType, Value};

use crate::loader::ProviderKind;
use crate::mem::RowData;
use crate::telemetry::{
    CacheLayer, CacheOutcome, CacheTelemetryEvent, QueryTelemetryEvent, ReloadOutcome,
    ReloadTelemetryEvent, telemetry, telemetry_enabled,
};

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct DatasourceId(pub String);

impl DatasourceId {
    pub fn from_seed(kind: ProviderKind, seed: &str) -> Self {
        let mut hasher = DefaultHasher::new();
        seed.hash(&mut hasher);
        let kind_str = match kind {
            ProviderKind::SqliteAuthority => "sqlite",
            ProviderKind::Postgres => "postgres",
            ProviderKind::Mysql => "mysql",
            ProviderKind::Redis => "redis",
        };
        Self(format!("{kind_str}:{:016x}", hasher.finish()))
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Generation(pub u64);

#[derive(Debug, Clone)]
pub enum QueryMode {
    Many,
    FirstRow,
}

#[derive(Debug, Clone, Copy)]
pub enum CachePolicy {
    Bypass,
    UseGlobal,
    UseCallScope,
}

#[derive(Debug, Clone)]
pub enum QueryValue {
    Null,
    Bool(bool),
    Int(i64),
    Float(f64),
    Text(String),
}

#[derive(Debug, Clone)]
pub struct QueryParam {
    pub name: String,
    pub value: QueryValue,
}

#[derive(Debug, Clone)]
pub struct QueryRequest {
    pub sql: String,
    pub params: Vec<QueryParam>,
    pub mode: QueryMode,
    pub cache_policy: CachePolicy,
}

impl QueryRequest {
    pub fn many(
        sql: impl Into<String>,
        params: Vec<QueryParam>,
        cache_policy: CachePolicy,
    ) -> Self {
        Self {
            sql: sql.into(),
            params,
            mode: QueryMode::Many,
            cache_policy,
        }
    }

    pub fn first_row(
        sql: impl Into<String>,
        params: Vec<QueryParam>,
        cache_policy: CachePolicy,
    ) -> Self {
        Self {
            sql: sql.into(),
            params,
            mode: QueryMode::FirstRow,
            cache_policy,
        }
    }
}

#[derive(Debug, Clone)]
pub enum QueryResponse {
    Rows(Vec<RowData>),
    Row(RowData),
}

impl QueryResponse {
    pub fn into_rows(self) -> Vec<RowData> {
        match self {
            QueryResponse::Rows(rows) => rows,
            QueryResponse::Row(row) => vec![row],
        }
    }

    pub fn into_row(self) -> RowData {
        match self {
            QueryResponse::Rows(rows) => rows.into_iter().next().unwrap_or_default(),
            QueryResponse::Row(row) => row,
        }
    }
}

#[async_trait]
pub trait ProviderExecutor: Send + Sync {
    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>>;
    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>>;
    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData>;
    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData>;

    async fn query_async(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
        self.query(sql)
    }

    async fn query_fields_async(
        &self,
        sql: &str,
        params: &[DataField],
    ) -> KnowledgeResult<Vec<RowData>> {
        self.query_fields(sql, params)
    }

    async fn query_row_async(&self, sql: &str) -> KnowledgeResult<RowData> {
        self.query_row(sql)
    }

    async fn query_named_fields_async(
        &self,
        sql: &str,
        params: &[DataField],
    ) -> KnowledgeResult<RowData> {
        self.query_named_fields(sql, params)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum QueryModeTag {
    Many,
    FirstRow,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct ResultCacheKey {
    pub datasource_id: DatasourceId,
    pub generation: Generation,
    pub query_hash: u64,
    pub params_hash: u64,
    pub mode: QueryModeTag,
}

pub struct ProviderHandle {
    pub provider: Arc<dyn ProviderExecutor>,
    pub datasource_id: DatasourceId,
    pub generation: Generation,
    pub kind: ProviderKind,
}

#[derive(Debug, Clone)]
pub struct RuntimeSnapshot {
    pub provider_kind: Option<ProviderKind>,
    pub datasource_id: Option<DatasourceId>,
    pub generation: Option<Generation>,
    pub result_cache_enabled: bool,
    pub result_cache_len: usize,
    pub result_cache_capacity: usize,
    pub result_cache_ttl_ms: u64,
    pub metadata_cache_len: usize,
    pub metadata_cache_capacity: usize,
    pub result_cache_hits: u64,
    pub result_cache_misses: u64,
    pub metadata_cache_hits: u64,
    pub metadata_cache_misses: u64,
    pub local_cache_hits: u64,
    pub local_cache_misses: u64,
    pub reload_successes: u64,
    pub reload_failures: u64,
}

#[derive(Debug, Clone)]
pub struct MetadataCacheScope {
    pub datasource_id: DatasourceId,
    pub generation: Generation,
}

#[derive(Debug, Clone, Copy)]
pub struct ResultCacheConfig {
    pub enabled: bool,
    pub capacity: usize,
    pub ttl: Duration,
}

impl Default for ResultCacheConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            capacity: 1024,
            ttl: Duration::from_millis(30_000),
        }
    }
}

#[derive(Debug, Clone)]
struct CachedQueryResponse {
    response: Arc<QueryResponse>,
    cached_at: Instant,
}

// ---------------------------------------------------------------------------
// Redis result cache types
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) enum RedisCmdTag {
    BfExists,
    HGet,
    Get,
    SetExists,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) struct RedisCacheKey {
    pub generation: u64,
    pub cmd_tag: RedisCmdTag,
    pub key_hash: u64,
    pub args_hash: u64,
}

#[derive(Debug, Clone)]
pub(crate) enum CachedRedisValue {
    Bool(bool),
    OptString(Option<String>),
}

#[derive(Debug, Clone)]
struct CachedRedisEntry {
    value: CachedRedisValue,
    cached_at: Instant,
    ttl_ms: u64, // 0 = no TTL (generation-only)
}

pub struct KnowledgeRuntime {
    provider: RwLock<Option<Arc<ProviderHandle>>>,
    /// 命名 provider 注册表：`name -> handle`。默认 provider 也会以 [`DEFAULT_PROVIDER_NAME`] 注册。
    named_providers: RwLock<HashMap<String, Arc<ProviderHandle>>>,
    /// 命名 provider 注册表版本号：任何安装/清理都会递增，用于路由 memo 失效。
    named_provider_epoch: AtomicU64,
    next_generation: AtomicU64,
    provider_epoch: AtomicU64,
    current_generation_value: AtomicU64,
    result_cache_config: RwLock<ResultCacheConfig>,
    result_cache_enabled: AtomicBool,
    result_cache_ttl_ms: AtomicU64,
    result_cache: RwLock<LruCache<ResultCacheKey, CachedQueryResponse>>,
    result_cache_hits: AtomicU64,
    result_cache_misses: AtomicU64,
    metadata_cache_hits: AtomicU64,
    metadata_cache_misses: AtomicU64,
    local_cache_hits: AtomicU64,
    local_cache_misses: AtomicU64,
    reload_successes: AtomicU64,
    reload_failures: AtomicU64,
    redis_cache: RwLock<LruCache<RedisCacheKey, CachedRedisEntry>>,
    redis_cache_hits: AtomicU64,
    redis_cache_misses: AtomicU64,
    redis_global_enabled: AtomicBool,
}

impl KnowledgeRuntime {
    pub fn new(result_cache_capacity: usize) -> Self {
        let config = ResultCacheConfig {
            capacity: result_cache_capacity.max(1),
            ..ResultCacheConfig::default()
        };
        let capacity = NonZeroUsize::new(config.capacity).expect("non-zero capacity");
        Self {
            provider: RwLock::new(None),
            named_providers: RwLock::new(HashMap::new()),
            named_provider_epoch: AtomicU64::new(0),
            next_generation: AtomicU64::new(0),
            provider_epoch: AtomicU64::new(0),
            current_generation_value: AtomicU64::new(0),
            result_cache_config: RwLock::new(config),
            result_cache_enabled: AtomicBool::new(config.enabled),
            result_cache_ttl_ms: AtomicU64::new(config.ttl.as_millis() as u64),
            result_cache: RwLock::new(LruCache::new(capacity)),
            result_cache_hits: AtomicU64::new(0),
            result_cache_misses: AtomicU64::new(0),
            metadata_cache_hits: AtomicU64::new(0),
            metadata_cache_misses: AtomicU64::new(0),
            local_cache_hits: AtomicU64::new(0),
            local_cache_misses: AtomicU64::new(0),
            reload_successes: AtomicU64::new(0),
            reload_failures: AtomicU64::new(0),
            redis_cache: RwLock::new(LruCache::new(capacity)),
            redis_cache_hits: AtomicU64::new(0),
            redis_cache_misses: AtomicU64::new(0),
            redis_global_enabled: AtomicBool::new(true),
        }
    }

    pub fn install_provider<F>(
        &self,
        kind: ProviderKind,
        datasource_id: DatasourceId,
        build: F,
    ) -> KnowledgeResult<Generation>
    where
        F: FnOnce(Generation) -> KnowledgeResult<Arc<dyn ProviderExecutor>>,
    {
        self.install_provider_named(DEFAULT_PROVIDER_NAME, kind, datasource_id, build, true)
    }

    /// 安装一个 provider：注册到命名注册表（`name -> handle`），并在 `is_default`
    /// 时同时设置为默认 provider。共享同一个单调递增 generation 计数器，
    /// 保证重载后的结果缓存键不与旧数据冲突。
    pub fn install_provider_named<F>(
        &self,
        name: &str,
        kind: ProviderKind,
        datasource_id: DatasourceId,
        build: F,
        is_default: bool,
    ) -> KnowledgeResult<Generation>
    where
        F: FnOnce(Generation) -> KnowledgeResult<Arc<dyn ProviderExecutor>>,
    {
        let generation = Generation(self.next_generation.fetch_add(1, Ordering::SeqCst) + 1);
        let previous = self
            .provider
            .read()
            .ok()
            .and_then(|guard| guard.as_ref().cloned());
        debug_kdb!(
            "[kdb] reload provider start name={} kind={kind:?} datasource_id={} target_generation={} previous_generation={}",
            name,
            datasource_id.0,
            generation.0,
            previous
                .as_ref()
                .map(|handle| handle.generation.0.to_string())
                .unwrap_or_else(|| "none".to_string())
        );
        let provider = match build(generation) {
            Ok(provider) => provider,
            Err(err) => {
                self.reload_failures.fetch_add(1, Ordering::Relaxed);
                warn_kdb!(
                    "[kdb] reload provider failed name={} kind={kind:?} datasource_id={} target_generation={} err={}",
                    name,
                    datasource_id.0,
                    generation.0,
                    err
                );
                if telemetry_enabled() {
                    telemetry().on_reload(&ReloadTelemetryEvent {
                        outcome: ReloadOutcome::Failure,
                        provider_kind: kind.clone(),
                    });
                }
                return Err(err);
            }
        };
        debug_kdb!(
            "[kdb] install provider name={} kind={kind:?} datasource_id={} generation={}",
            name,
            datasource_id.0,
            generation.0
        );
        let kind_for_handle = kind.clone();
        let datasource_id_for_handle = datasource_id.clone();
        let handle = Arc::new(ProviderHandle {
            provider,
            datasource_id: datasource_id_for_handle,
            generation,
            kind: kind_for_handle,
        });
        {
            let mut guard = self
                .named_providers
                .write()
                .expect("runtime named provider lock poisoned");
            guard.insert(name.to_string(), handle.clone());
        }
        self.named_provider_epoch.fetch_add(1, Ordering::AcqRel);
        if is_default {
            self.provider_epoch.fetch_add(1, Ordering::AcqRel);
            {
                let mut guard = self
                    .provider
                    .write()
                    .expect("runtime provider lock poisoned");
                *guard = Some(handle);
            }
            self.current_generation_value
                .store(generation.0, Ordering::Release);
            self.provider_epoch.fetch_add(1, Ordering::Release);
        }
        self.reload_successes.fetch_add(1, Ordering::Relaxed);
        if telemetry_enabled() {
            telemetry().on_reload(&ReloadTelemetryEvent {
                outcome: ReloadOutcome::Success,
                provider_kind: kind.clone(),
            });
        }
        debug_kdb!(
            "[kdb] reload provider success name={} kind={kind:?} datasource_id={} generation={}",
            name,
            datasource_id.0,
            generation.0
        );
        Ok(generation)
    }

    pub fn configure_result_cache(&self, enabled: bool, capacity: usize, ttl: Duration) {
        let new_config = ResultCacheConfig {
            enabled,
            capacity: capacity.max(1),
            ttl: ttl.max(Duration::from_millis(1)),
        };
        let mut should_reset_cache = false;
        {
            let mut guard = self
                .result_cache_config
                .write()
                .expect("runtime result cache config lock poisoned");
            if guard.capacity != new_config.capacity || (!new_config.enabled && guard.enabled) {
                should_reset_cache = true;
            }
            *guard = new_config;
        }
        self.result_cache_enabled
            .store(new_config.enabled, Ordering::Relaxed);
        self.result_cache_ttl_ms.store(
            new_config.ttl.as_millis().min(u128::from(u64::MAX)) as u64,
            Ordering::Relaxed,
        );

        if should_reset_cache {
            let mut cache = self
                .result_cache
                .write()
                .expect("runtime result cache lock poisoned");
            *cache = LruCache::new(
                NonZeroUsize::new(new_config.capacity).expect("non-zero result cache capacity"),
            );
        }
    }

    pub fn configure_redis_cache(&self, global_enabled: bool, capacity: usize) {
        let new_capacity =
            NonZeroUsize::new(capacity.max(1)).expect("non-zero redis cache capacity");
        if let Ok(mut cache) = self.redis_cache.write() {
            *cache = LruCache::new(new_capacity);
        }
        self.redis_global_enabled
            .store(global_enabled, Ordering::Relaxed);
    }

    pub fn current_generation(&self) -> Option<Generation> {
        let epoch_before = self.provider_epoch.load(Ordering::Acquire);
        if epoch_before % 2 == 1 {
            return self.current_generation_from_provider();
        }
        let generation = self.current_generation_value.load(Ordering::Acquire);
        let epoch_after = self.provider_epoch.load(Ordering::Acquire);
        if epoch_before != epoch_after {
            return self.current_generation_from_provider();
        }
        match generation {
            0 => None,
            generation => Some(Generation(generation)),
        }
    }

    pub fn snapshot(&self) -> RuntimeSnapshot {
        let provider = self
            .provider
            .read()
            .ok()
            .and_then(|guard| guard.as_ref().cloned());
        let result_cache_config = self
            .result_cache_config
            .read()
            .map(|guard| *guard)
            .unwrap_or_default();
        let (result_cache_len, result_cache_capacity) = self
            .result_cache
            .read()
            .map(|cache| (cache.len(), cache.cap().get()))
            .unwrap_or((0, 0));
        let (metadata_cache_len, metadata_cache_capacity) =
            crate::mem::query_util::column_metadata_cache_snapshot();
        RuntimeSnapshot {
            provider_kind: provider.as_ref().map(|handle| handle.kind.clone()),
            datasource_id: provider.as_ref().map(|handle| handle.datasource_id.clone()),
            generation: provider.as_ref().map(|handle| handle.generation),
            result_cache_enabled: result_cache_config.enabled,
            result_cache_len,
            result_cache_capacity,
            result_cache_ttl_ms: result_cache_config.ttl.as_millis() as u64,
            metadata_cache_len,
            metadata_cache_capacity,
            result_cache_hits: self.result_cache_hits.load(Ordering::Relaxed),
            result_cache_misses: self.result_cache_misses.load(Ordering::Relaxed),
            metadata_cache_hits: self.metadata_cache_hits.load(Ordering::Relaxed),
            metadata_cache_misses: self.metadata_cache_misses.load(Ordering::Relaxed),
            local_cache_hits: self.local_cache_hits.load(Ordering::Relaxed),
            local_cache_misses: self.local_cache_misses.load(Ordering::Relaxed),
            reload_successes: self.reload_successes.load(Ordering::Relaxed),
            reload_failures: self.reload_failures.load(Ordering::Relaxed),
        }
    }

    pub fn current_metadata_scope(&self) -> MetadataCacheScope {
        self.provider
            .read()
            .ok()
            .and_then(|guard| guard.as_ref().cloned())
            .map(|handle| MetadataCacheScope {
                datasource_id: handle.datasource_id.clone(),
                generation: handle.generation,
            })
            .unwrap_or_else(|| MetadataCacheScope {
                datasource_id: DatasourceId("sqlite:standalone".to_string()),
                generation: Generation(0),
            })
    }

    pub fn current_provider_kind(&self) -> Option<ProviderKind> {
        self.provider
            .read()
            .ok()
            .and_then(|guard| guard.as_ref().map(|handle| handle.kind.clone()))
    }

    /// 按名称取 provider handle；未找到时返回错误。
    pub fn provider_by_name(&self, name: &str) -> KnowledgeResult<Arc<ProviderHandle>> {
        self.named_providers
            .read()
            .expect("runtime named provider lock poisoned")
            .get(name)
            .cloned()
            .ok_or_else(|| {
                KnowReason::from_logic()
                    .to_err()
                    .with_detail(format!("knowledge provider '{name}' not initialized"))
            })
    }

    pub fn provider_exists(&self, name: &str) -> bool {
        self.named_providers
            .read()
            .ok()
            .is_some_and(|guard| guard.contains_key(name))
    }

    pub fn provider_names(&self) -> Vec<String> {
        self.named_providers
            .read()
            .ok()
            .map(|guard| guard.keys().cloned().collect())
            .unwrap_or_default()
    }

    /// 清理命名注册表中不在 `keep` 内的过期 provider。仅在成功安装后调用，
    /// 中途失败时保留旧状态。
    pub fn prune_named_providers_except(&self, keep: &HashSet<String>) {
        if let Ok(mut guard) = self.named_providers.write() {
            let before = guard.len();
            guard.retain(|name, _| keep.contains(name));
            if guard.len() != before {
                self.named_provider_epoch.fetch_add(1, Ordering::AcqRel);
            }
        }
    }

    /// 命名 provider 注册表版本号：任何安装/清理都会递增，用于路由 memo 失效判断。
    pub fn named_provider_epoch(&self) -> u64 {
        self.named_provider_epoch.load(Ordering::Acquire)
    }

    pub fn record_result_cache_hit(&self) {
        self.result_cache_hits.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_result_cache_miss(&self) {
        self.result_cache_misses.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_metadata_cache_hit(&self) {
        self.metadata_cache_hits.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_metadata_cache_miss(&self) {
        self.metadata_cache_misses.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_local_cache_hit(&self) {
        self.local_cache_hits.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_local_cache_miss(&self) {
        self.local_cache_misses.fetch_add(1, Ordering::Relaxed);
    }

    pub fn execute(&self, req: &QueryRequest) -> KnowledgeResult<QueryResponse> {
        let handle = self.current_handle()?;
        self.execute_with_handle(&handle, req)
    }

    /// 对指定命名 provider 执行查询。
    pub fn execute_for(&self, name: &str, req: &QueryRequest) -> KnowledgeResult<QueryResponse> {
        let handle = self.provider_by_name(name)?;
        self.execute_with_handle(&handle, req)
    }

    fn execute_with_handle(
        &self,
        handle: &Arc<ProviderHandle>,
        req: &QueryRequest,
    ) -> KnowledgeResult<QueryResponse> {
        let use_global_cache =
            matches!(req.cache_policy, CachePolicy::UseGlobal) && self.result_cache_enabled();
        if use_global_cache && let Some(hit) = self.fetch_result_cache(handle, req) {
            self.record_result_cache_hit();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Hit,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
            debug_kdb!(
                "[kdb] global result cache hit kind={:?} generation={}",
                handle.kind,
                handle.generation.0
            );
            return Ok(hit);
        }
        if use_global_cache {
            self.record_result_cache_miss();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Miss,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
            debug_kdb!(
                "[kdb] global result cache miss kind={:?} generation={}",
                handle.kind,
                handle.generation.0
            );
        }

        let params = params_to_fields(&req.params);
        let mode_tag = query_mode_tag(&req.mode);
        let started = Instant::now();
        debug_kdb!(
            "[kdb] execute query kind={:?} generation={} mode={:?} cache_policy={:?}",
            handle.kind,
            handle.generation.0,
            req.mode,
            req.cache_policy
        );
        let response = match match req.mode {
            QueryMode::Many => {
                if params.is_empty() {
                    handle.provider.query(&req.sql).map(QueryResponse::Rows)
                } else {
                    handle
                        .provider
                        .query_fields(&req.sql, &params)
                        .map(QueryResponse::Rows)
                }
            }
            QueryMode::FirstRow => {
                if params.is_empty() {
                    handle.provider.query_row(&req.sql).map(QueryResponse::Row)
                } else {
                    handle
                        .provider
                        .query_named_fields(&req.sql, &params)
                        .map(QueryResponse::Row)
                }
            }
        } {
            Ok(response) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: mode_tag,
                        success: true,
                        elapsed: started.elapsed(),
                    });
                }
                response
            }
            Err(err) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: mode_tag,
                        success: false,
                        elapsed: started.elapsed(),
                    });
                }
                return Err(err);
            }
        };

        if use_global_cache {
            self.save_result_cache(handle, req, response.clone());
            debug_kdb!(
                "[kdb] global result cache store kind={:?} generation={}",
                handle.kind,
                handle.generation.0
            );
        }

        Ok(response)
    }

    pub fn execute_first_row_fields(
        &self,
        sql: &str,
        params: &[DataField],
        cache_policy: CachePolicy,
    ) -> KnowledgeResult<RowData> {
        let handle = self.current_handle()?;
        self.execute_first_row_fields_with_handle(&handle, sql, params, cache_policy)
    }

    /// 对指定命名 provider 执行首行查询。
    pub fn execute_first_row_fields_for(
        &self,
        name: &str,
        sql: &str,
        params: &[DataField],
        cache_policy: CachePolicy,
    ) -> KnowledgeResult<RowData> {
        let handle = self.provider_by_name(name)?;
        self.execute_first_row_fields_with_handle(&handle, sql, params, cache_policy)
    }

    fn execute_first_row_fields_with_handle(
        &self,
        handle: &Arc<ProviderHandle>,
        sql: &str,
        params: &[DataField],
        cache_policy: CachePolicy,
    ) -> KnowledgeResult<RowData> {
        let use_global_cache =
            matches!(cache_policy, CachePolicy::UseGlobal) && self.result_cache_enabled();
        if use_global_cache
            && let Some(hit) = self.fetch_result_cache_by_key(result_cache_key_fields(
                handle,
                sql,
                params,
                QueryModeTag::FirstRow,
            ))
        {
            self.record_result_cache_hit();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Hit,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
            return Ok(hit.into_row());
        }
        if use_global_cache {
            self.record_result_cache_miss();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Miss,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
        }

        let started = Instant::now();
        let row = if params.is_empty() {
            handle.provider.query_row(sql)
        } else {
            handle.provider.query_named_fields(sql, params)
        };
        let row = match row {
            Ok(row) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: QueryModeTag::FirstRow,
                        success: true,
                        elapsed: started.elapsed(),
                    });
                }
                row
            }
            Err(err) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: QueryModeTag::FirstRow,
                        success: false,
                        elapsed: started.elapsed(),
                    });
                }
                return Err(err);
            }
        };

        if use_global_cache {
            self.save_result_cache_by_key(
                result_cache_key_fields(handle, sql, params, QueryModeTag::FirstRow),
                QueryResponse::Row(row.clone()),
            );
        }

        Ok(row)
    }

    pub async fn execute_async(&self, req: &QueryRequest) -> KnowledgeResult<QueryResponse> {
        let handle = self.current_handle()?;
        if matches!(handle.kind, ProviderKind::SqliteAuthority) {
            let handle = handle.clone();
            let req = req.clone();
            return task::spawn_blocking(move || runtime().execute_with_handle(&handle, &req))
                .await
                .map_err(|err| {
                    KnowReason::from_logic()
                        .to_err()
                        .with_detail(format!("knowledge async sqlite query join failed: {err}"))
                })?;
        }
        self.execute_async_with_handle(&handle, req).await
    }

    pub async fn execute_for_async(
        &self,
        name: &str,
        req: &QueryRequest,
    ) -> KnowledgeResult<QueryResponse> {
        let handle = self.provider_by_name(name)?;
        self.execute_async_with_handle(&handle, req).await
    }

    async fn execute_async_with_handle(
        &self,
        handle: &Arc<ProviderHandle>,
        req: &QueryRequest,
    ) -> KnowledgeResult<QueryResponse> {
        let use_global_cache =
            matches!(req.cache_policy, CachePolicy::UseGlobal) && self.result_cache_enabled();
        if use_global_cache && let Some(hit) = self.fetch_result_cache(handle, req) {
            self.record_result_cache_hit();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Hit,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
            return Ok(hit);
        }
        if use_global_cache {
            self.record_result_cache_miss();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Miss,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
        }

        let params = params_to_fields(&req.params);
        let mode_tag = query_mode_tag(&req.mode);
        let started = Instant::now();
        let response = match req.mode {
            QueryMode::Many => {
                if params.is_empty() {
                    handle
                        .provider
                        .query_async(&req.sql)
                        .await
                        .map(QueryResponse::Rows)
                } else {
                    handle
                        .provider
                        .query_fields_async(&req.sql, &params)
                        .await
                        .map(QueryResponse::Rows)
                }
            }
            QueryMode::FirstRow => {
                if params.is_empty() {
                    handle
                        .provider
                        .query_row_async(&req.sql)
                        .await
                        .map(QueryResponse::Row)
                } else {
                    handle
                        .provider
                        .query_named_fields_async(&req.sql, &params)
                        .await
                        .map(QueryResponse::Row)
                }
            }
        };
        let response = match response {
            Ok(response) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: mode_tag,
                        success: true,
                        elapsed: started.elapsed(),
                    });
                }
                response
            }
            Err(err) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: mode_tag,
                        success: false,
                        elapsed: started.elapsed(),
                    });
                }
                return Err(err);
            }
        };

        if use_global_cache {
            self.save_result_cache(handle, req, response.clone());
        }

        Ok(response)
    }

    pub async fn execute_first_row_fields_async(
        &self,
        sql: &str,
        params: &[DataField],
        cache_policy: CachePolicy,
    ) -> KnowledgeResult<RowData> {
        let handle = self.current_handle()?;
        if matches!(handle.kind, ProviderKind::SqliteAuthority) {
            let handle = handle.clone();
            let sql = sql.to_string();
            let params = params.to_vec();
            return task::spawn_blocking(move || {
                runtime().execute_first_row_fields_with_handle(&handle, &sql, &params, cache_policy)
            })
            .await
            .map_err(|err| {
                KnowReason::from_logic().to_err().with_detail(format!(
                    "knowledge async sqlite first-row query join failed: {err}"
                ))
            })?;
        }
        self.execute_first_row_fields_async_with_handle(&handle, sql, params, cache_policy)
            .await
    }

    pub async fn execute_first_row_fields_for_async(
        &self,
        name: &str,
        sql: &str,
        params: &[DataField],
        cache_policy: CachePolicy,
    ) -> KnowledgeResult<RowData> {
        let handle = self.provider_by_name(name)?;
        self.execute_first_row_fields_async_with_handle(&handle, sql, params, cache_policy)
            .await
    }

    async fn execute_first_row_fields_async_with_handle(
        &self,
        handle: &Arc<ProviderHandle>,
        sql: &str,
        params: &[DataField],
        cache_policy: CachePolicy,
    ) -> KnowledgeResult<RowData> {
        let use_global_cache =
            matches!(cache_policy, CachePolicy::UseGlobal) && self.result_cache_enabled();
        if use_global_cache
            && let Some(hit) = self.fetch_result_cache_by_key(result_cache_key_fields(
                handle,
                sql,
                params,
                QueryModeTag::FirstRow,
            ))
        {
            self.record_result_cache_hit();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Hit,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
            return Ok(hit.into_row());
        }
        if use_global_cache {
            self.record_result_cache_miss();
            if telemetry_enabled() {
                telemetry().on_cache(&CacheTelemetryEvent {
                    layer: CacheLayer::Result,
                    outcome: CacheOutcome::Miss,
                    provider_kind: Some(handle.kind.clone()),
                });
            }
        }

        let started = Instant::now();
        let row = if params.is_empty() {
            handle.provider.query_row_async(sql).await
        } else {
            handle.provider.query_named_fields_async(sql, params).await
        };
        let row = match row {
            Ok(row) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: QueryModeTag::FirstRow,
                        success: true,
                        elapsed: started.elapsed(),
                    });
                }
                row
            }
            Err(err) => {
                if telemetry_enabled() {
                    telemetry().on_query(&QueryTelemetryEvent {
                        provider_kind: handle.kind.clone(),
                        mode: QueryModeTag::FirstRow,
                        success: false,
                        elapsed: started.elapsed(),
                    });
                }
                return Err(err);
            }
        };

        if use_global_cache {
            self.save_result_cache_by_key(
                result_cache_key_fields(handle, sql, params, QueryModeTag::FirstRow),
                QueryResponse::Row(row.clone()),
            );
        }

        Ok(row)
    }

    fn current_handle(&self) -> KnowledgeResult<Arc<ProviderHandle>> {
        self.provider
            .read()
            .expect("runtime provider lock poisoned")
            .clone()
            .ok_or_else(|| {
                KnowReason::from_logic()
                    .to_err()
                    .with_detail("knowledge provider not initialized")
            })
    }

    fn current_generation_from_provider(&self) -> Option<Generation> {
        self.provider
            .read()
            .ok()
            .and_then(|guard| guard.as_ref().map(|handle| handle.generation))
    }

    fn fetch_result_cache(
        &self,
        handle: &ProviderHandle,
        req: &QueryRequest,
    ) -> Option<QueryResponse> {
        self.fetch_result_cache_by_key(result_cache_key(handle, req))
    }

    fn fetch_result_cache_by_key(&self, key: ResultCacheKey) -> Option<QueryResponse> {
        if !self.result_cache_enabled() {
            return None;
        }
        let cached = self
            .result_cache
            .read()
            .ok()
            .and_then(|cache| cache.peek(&key).cloned())?;
        if cached.cached_at.elapsed() > self.result_cache_ttl() {
            if let Ok(mut cache) = self.result_cache.write() {
                let _ = cache.pop(&key);
            }
            return None;
        }
        Some((*cached.response).clone())
    }

    fn save_result_cache(
        &self,
        handle: &ProviderHandle,
        req: &QueryRequest,
        response: QueryResponse,
    ) {
        self.save_result_cache_by_key(result_cache_key(handle, req), response);
    }

    fn save_result_cache_by_key(&self, key: ResultCacheKey, response: QueryResponse) {
        if let Ok(mut cache) = self.result_cache.write() {
            cache.put(
                key,
                CachedQueryResponse {
                    response: Arc::new(response),
                    cached_at: Instant::now(),
                },
            );
        }
    }

    // -----------------------------------------------------------------------
    // Redis result cache
    // -----------------------------------------------------------------------

    fn redis_cache_enabled(&self) -> bool {
        self.result_cache_enabled.load(Ordering::Acquire)
    }

    #[allow(dead_code)]
    fn redis_cache_ttl(&self) -> Duration {
        Duration::from_millis(self.result_cache_ttl_ms.load(Ordering::Acquire))
    }

    fn fetch_redis_cache(&self, key: &RedisCacheKey) -> Option<CachedRedisEntry> {
        if !self.redis_cache_enabled() {
            return None;
        }
        let entry = self
            .redis_cache
            .read()
            .ok()
            .and_then(|cache| cache.peek(key).cloned())?;
        // Check TTL expiry
        if entry.ttl_ms > 0 && entry.cached_at.elapsed() > Duration::from_millis(entry.ttl_ms) {
            if let Ok(mut cache) = self.redis_cache.write() {
                let _ = cache.pop(key);
            }
            return None;
        }
        Some(entry)
    }

    fn save_redis_cache(&self, key: RedisCacheKey, entry: CachedRedisEntry) {
        if !self.redis_cache_enabled() {
            return;
        }
        if let Ok(mut cache) = self.redis_cache.write() {
            cache.put(key, entry);
        }
    }

    pub(crate) fn redis_cache_get(&self, ck: &RedisCacheKey) -> Option<CachedRedisValue> {
        if !self.redis_global_enabled.load(Ordering::Relaxed) {
            return None;
        }
        let entry = self.fetch_redis_cache(ck)?;
        self.redis_cache_hits.fetch_add(1, Ordering::Relaxed);
        Some(entry.value)
    }

    pub(crate) fn redis_cache_put(&self, ck: RedisCacheKey, value: CachedRedisValue) {
        self.redis_cache_put_with_ttl(ck, value, 0);
    }

    pub(crate) fn redis_cache_put_with_ttl(
        &self,
        ck: RedisCacheKey,
        value: CachedRedisValue,
        ttl_ms: u64,
    ) {
        if !self.redis_global_enabled.load(Ordering::Relaxed) {
            return;
        }
        self.redis_cache_misses.fetch_add(1, Ordering::Relaxed);
        self.save_redis_cache(
            ck,
            CachedRedisEntry {
                value,
                cached_at: Instant::now(),
                ttl_ms,
            },
        );
    }

    #[allow(dead_code)]
    fn clear_redis_cache(&self) {
        if let Ok(mut cache) = self.redis_cache.write() {
            cache.clear();
        }
    }

    #[inline]
    fn result_cache_enabled(&self) -> bool {
        self.result_cache_enabled.load(Ordering::Relaxed)
    }

    #[inline]
    fn result_cache_ttl(&self) -> Duration {
        Duration::from_millis(self.result_cache_ttl_ms.load(Ordering::Relaxed))
    }
}

pub fn runtime() -> &'static KnowledgeRuntime {
    static RUNTIME: OnceLock<KnowledgeRuntime> = OnceLock::new();
    RUNTIME.get_or_init(|| KnowledgeRuntime::new(1024))
}

#[cfg(test)]
pub(crate) struct RuntimeTestGuard(tokio::sync::Mutex<()>);

#[cfg(test)]
impl RuntimeTestGuard {
    pub(crate) fn lock(&self) -> Result<tokio::sync::MutexGuard<'_, ()>, std::convert::Infallible> {
        Ok(self.0.blocking_lock())
    }

    pub(crate) async fn lock_async(&self) -> tokio::sync::MutexGuard<'_, ()> {
        self.0.lock().await
    }
}

#[cfg(test)]
pub(crate) fn runtime_test_guard() -> &'static RuntimeTestGuard {
    static GUARD: OnceLock<RuntimeTestGuard> = OnceLock::new();
    GUARD.get_or_init(|| RuntimeTestGuard(tokio::sync::Mutex::new(())))
}

fn result_cache_key(handle: &ProviderHandle, req: &QueryRequest) -> ResultCacheKey {
    ResultCacheKey {
        datasource_id: handle.datasource_id.clone(),
        generation: handle.generation,
        query_hash: stable_hash(&req.sql),
        params_hash: stable_params_hash(&req.params),
        mode: match req.mode {
            QueryMode::Many => QueryModeTag::Many,
            QueryMode::FirstRow => QueryModeTag::FirstRow,
        },
    }
}

fn result_cache_key_fields(
    handle: &ProviderHandle,
    sql: &str,
    params: &[DataField],
    mode: QueryModeTag,
) -> ResultCacheKey {
    ResultCacheKey {
        datasource_id: handle.datasource_id.clone(),
        generation: handle.generation,
        query_hash: stable_hash(sql),
        params_hash: stable_field_params_hash(params),
        mode,
    }
}

fn query_mode_tag(mode: &QueryMode) -> QueryModeTag {
    match mode {
        QueryMode::Many => QueryModeTag::Many,
        QueryMode::FirstRow => QueryModeTag::FirstRow,
    }
}

fn stable_hash(value: &str) -> u64 {
    let mut hasher = DefaultHasher::new();
    value.hash(&mut hasher);
    hasher.finish()
}

fn stable_params_hash(params: &[QueryParam]) -> u64 {
    let mut hasher = DefaultHasher::new();
    for param in params {
        param.name.hash(&mut hasher);
        match &param.value {
            QueryValue::Null => 0u8.hash(&mut hasher),
            QueryValue::Bool(value) => {
                1u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            QueryValue::Int(value) => {
                2u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            QueryValue::Float(value) => {
                3u8.hash(&mut hasher);
                value.to_bits().hash(&mut hasher);
            }
            QueryValue::Text(value) => {
                4u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
        }
    }
    hasher.finish()
}

fn stable_field_params_hash(params: &[DataField]) -> u64 {
    let mut hasher = DefaultHasher::new();
    for field in params {
        field.get_name().hash(&mut hasher);
        match field.get_value() {
            Value::Null | Value::Ignore(_) => 0u8.hash(&mut hasher),
            Value::Bool(value) => {
                1u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            Value::Digit(value) => {
                2u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            Value::Float(value) => {
                3u8.hash(&mut hasher);
                value.to_bits().hash(&mut hasher);
            }
            Value::Chars(value) => {
                4u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            Value::Symbol(value) => {
                5u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            Value::Time(value) => {
                6u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            Value::Hex(value) => {
                7u8.hash(&mut hasher);
                value.to_string().hash(&mut hasher);
            }
            Value::IpNet(value) => {
                8u8.hash(&mut hasher);
                value.to_string().hash(&mut hasher);
            }
            Value::IpAddr(value) => {
                9u8.hash(&mut hasher);
                value.hash(&mut hasher);
            }
            Value::Obj(value) => {
                10u8.hash(&mut hasher);
                format!("{:?}", value).hash(&mut hasher);
            }
            Value::Array(value) => {
                11u8.hash(&mut hasher);
                format!("{:?}", value).hash(&mut hasher);
            }
            Value::Domain(value) => {
                12u8.hash(&mut hasher);
                value.0.hash(&mut hasher);
            }
            Value::Url(value) => {
                13u8.hash(&mut hasher);
                value.0.hash(&mut hasher);
            }
            Value::Email(value) => {
                14u8.hash(&mut hasher);
                value.0.hash(&mut hasher);
            }
            Value::IdCard(value) => {
                15u8.hash(&mut hasher);
                value.0.hash(&mut hasher);
            }
            Value::MobilePhone(value) => {
                16u8.hash(&mut hasher);
                value.0.hash(&mut hasher);
            }
            Value::BigUint(value) => {
                17u8.hash(&mut hasher);
                value.to_string().hash(&mut hasher);
            }
        }
    }
    hasher.finish()
}

pub fn fields_to_params(params: &[DataField]) -> Vec<QueryParam> {
    params
        .iter()
        .map(|field| {
            let value = match field.get_value() {
                Value::Null | Value::Ignore(_) => QueryValue::Null,
                Value::Bool(value) => QueryValue::Bool(*value),
                Value::Digit(value) => QueryValue::Int(*value),
                Value::Float(value) => QueryValue::Float(*value),
                Value::Chars(value) => QueryValue::Text(value.to_string()),
                Value::Symbol(value) => QueryValue::Text(value.to_string()),
                Value::Time(value) => QueryValue::Text(value.to_string()),
                Value::Hex(value) => QueryValue::Text(value.to_string()),
                Value::IpNet(value) => QueryValue::Text(value.to_string()),
                Value::IpAddr(value) => QueryValue::Text(value.to_string()),
                Value::Obj(value) => QueryValue::Text(format!("{:?}", value)),
                Value::Array(value) => QueryValue::Text(format!("{:?}", value)),
                Value::Domain(value) => QueryValue::Text(value.0.to_string()),
                Value::Url(value) => QueryValue::Text(value.0.to_string()),
                Value::Email(value) => QueryValue::Text(value.0.to_string()),
                Value::IdCard(value) => QueryValue::Text(value.0.to_string()),
                Value::MobilePhone(value) => QueryValue::Text(value.0.to_string()),
                // 任意精度整数：以十进制文本传递，消费端按数值语义处理
                Value::BigUint(value) => QueryValue::Text(value.to_string()),
            };
            QueryParam {
                name: field.get_name().to_string(),
                value,
            }
        })
        .collect()
}

pub fn params_to_fields(params: &[QueryParam]) -> Vec<DataField> {
    params
        .iter()
        .map(|param| match &param.value {
            QueryValue::Null => {
                DataField::new(DataType::default(), param.name.clone(), Value::Null)
            }
            QueryValue::Bool(value) => {
                DataField::new(DataType::default(), param.name.clone(), Value::Bool(*value))
            }
            QueryValue::Int(value) => DataField::from_digit(param.name.clone(), *value),
            QueryValue::Float(value) => DataField::from_float(param.name.clone(), *value),
            QueryValue::Text(value) => DataField::from_chars(param.name.clone(), value.clone()),
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use std::sync::Arc;
    use wp_model_core::model::Value;

    struct TestProvider {
        value: &'static str,
    }

    #[async_trait]
    impl ProviderExecutor for TestProvider {
        fn query(&self, _sql: &str) -> KnowledgeResult<Vec<RowData>> {
            Ok(vec![vec![DataField::from_chars("value", self.value)]])
        }

        fn query_fields(&self, _sql: &str, _params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
            self.query("")
        }

        fn query_row(&self, _sql: &str) -> KnowledgeResult<RowData> {
            Ok(vec![DataField::from_chars("value", self.value)])
        }

        fn query_named_fields(
            &self,
            _sql: &str,
            _params: &[DataField],
        ) -> KnowledgeResult<RowData> {
            self.query_row("")
        }
    }

    #[test]
    fn query_param_hash_is_stable() {
        let params = vec![
            QueryParam {
                name: ":id".to_string(),
                value: QueryValue::Int(7),
            },
            QueryParam {
                name: ":name".to_string(),
                value: QueryValue::Text("abc".to_string()),
            },
        ];
        assert_eq!(stable_params_hash(&params), stable_params_hash(&params));
    }

    #[test]
    fn fields_to_params_preserves_raw_chars_value() {
        let fields = [DataField::from_chars(
            ":name".to_string(),
            "令狐冲".to_string(),
        )];
        let params = fields_to_params(&fields);
        assert_eq!(params.len(), 1);
        match &params[0].value {
            QueryValue::Text(value) => assert_eq!(value, "令狐冲"),
            other => panic!("unexpected param value: {other:?}"),
        }
        let roundtrip = params_to_fields(&params);
        assert!(matches!(roundtrip[0].get_value(), Value::Chars(_)));
    }

    #[test]
    fn fields_to_params_biguint_is_text() {
        use num_bigint::BigUint;
        use std::str::FromStr;

        let fields = [DataField::new(
            wp_model_core::model::DataType::BigInt,
            ":ip_num",
            Value::BigUint(BigUint::from_str("382824323044708348099391746388336347272").unwrap()),
        )];
        // 任意精度整数以十进制文本传递
        let params = fields_to_params(&fields);
        assert_eq!(params.len(), 1);
        match &params[0].value {
            QueryValue::Text(value) => {
                assert_eq!(value, "382824323044708348099391746388336347272")
            }
            other => panic!("unexpected param value: {other:?}"),
        }
        // 参数 hash 稳定可重复（缓存键）
        assert_eq!(
            stable_field_params_hash(&fields),
            stable_field_params_hash(&fields)
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn sqlite_async_bridge_keeps_captured_handle_after_reload() {
        let _guard = runtime_test_guard().lock_async().await;
        runtime()
            .install_provider(
                ProviderKind::SqliteAuthority,
                DatasourceId("sqlite:old".to_string()),
                |_generation| Ok(Arc::new(TestProvider { value: "old" })),
            )
            .expect("install old provider");
        let old_handle = runtime().current_handle().expect("current old handle");

        runtime()
            .install_provider(
                ProviderKind::SqliteAuthority,
                DatasourceId("sqlite:new".to_string()),
                |_generation| Ok(Arc::new(TestProvider { value: "new" })),
            )
            .expect("install new provider");

        let req = QueryRequest::first_row("SELECT value", Vec::new(), CachePolicy::Bypass);
        let row = task::spawn_blocking(move || runtime().execute_with_handle(&old_handle, &req))
            .await
            .expect("join sqlite bridge")
            .expect("execute old handle")
            .into_row();
        assert_eq!(row[0].to_string(), "chars(old)");
    }

    #[test]
    fn named_providers_are_registered_and_queryable() {
        let _guard = runtime_test_guard()
            .lock()
            .expect("named provider test guard");
        runtime()
            .install_provider_named(
                "geo",
                ProviderKind::Postgres,
                DatasourceId::from_seed(ProviderKind::Postgres, "geo"),
                |_generation| Ok(Arc::new(TestProvider { value: "geo-value" })),
                false,
            )
            .expect("install geo");
        runtime()
            .install_provider_named(
                "asset",
                ProviderKind::Postgres,
                DatasourceId::from_seed(ProviderKind::Postgres, "asset"),
                |_generation| {
                    Ok(Arc::new(TestProvider {
                        value: "asset-value",
                    }))
                },
                true,
            )
            .expect("install asset (default)");

        assert!(runtime().provider_exists("geo"));
        assert!(runtime().provider_exists("asset"));
        assert!(!runtime().provider_exists("nope"));
        let names = runtime().provider_names();
        assert!(names.contains(&"geo".to_string()));
        assert!(names.contains(&"asset".to_string()));

        let req = QueryRequest::first_row("SELECT value", Vec::new(), CachePolicy::Bypass);
        let geo_row = runtime()
            .execute_for("geo", &req)
            .expect("execute geo")
            .into_row();
        assert_eq!(geo_row[0].to_string(), "chars(geo-value)");

        let row = runtime()
            .execute_first_row_fields_for("asset", "SELECT value", &[], CachePolicy::Bypass)
            .expect("execute asset");
        assert_eq!(row[0].to_string(), "chars(asset-value)");

        let default_row = runtime()
            .execute(&req)
            .expect("execute default (asset)")
            .into_row();
        assert_eq!(default_row[0].to_string(), "chars(asset-value)");
    }

    #[test]
    fn prune_named_providers_except_keeps_only_selected() {
        let _guard = runtime_test_guard().lock().expect("prune test guard");
        runtime()
            .install_provider_named(
                "a",
                ProviderKind::Postgres,
                DatasourceId::from_seed(ProviderKind::Postgres, "a"),
                |_generation| Ok(Arc::new(TestProvider { value: "a" })),
                false,
            )
            .expect("install a");
        runtime()
            .install_provider_named(
                "b",
                ProviderKind::Postgres,
                DatasourceId::from_seed(ProviderKind::Postgres, "b"),
                |_generation| Ok(Arc::new(TestProvider { value: "b" })),
                false,
            )
            .expect("install b");

        let keep: HashSet<String> = ["a".to_string()].into_iter().collect();
        runtime().prune_named_providers_except(&keep);
        assert!(runtime().provider_exists("a"));
        assert!(!runtime().provider_exists("b"));
    }

    #[test]
    fn provider_by_name_unknown_returns_error() {
        let err = runtime()
            .provider_by_name("nope")
            .err()
            .expect("unknown provider name should error");
        assert!(err.to_string().contains("not initialized"));
    }

    #[test]
    fn execute_first_row_fields_for_unknown_returns_error() {
        let err = runtime()
            .execute_first_row_fields_for("nope", "SELECT value", &[], CachePolicy::Bypass)
            .expect_err("unknown provider name");
        assert!(err.to_string().contains("not initialized"));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn named_provider_async_dispatch_works() {
        let _guard = runtime_test_guard().lock_async().await;
        runtime()
            .install_provider_named(
                "geo",
                ProviderKind::Postgres,
                DatasourceId::from_seed(ProviderKind::Postgres, "geo"),
                |_generation| Ok(Arc::new(TestProvider { value: "geo-async" })),
                false,
            )
            .expect("install geo");

        let req = QueryRequest::first_row("SELECT value", Vec::new(), CachePolicy::Bypass);
        let row = runtime()
            .execute_for_async("geo", &req)
            .await
            .expect("async geo")
            .into_row();
        assert_eq!(row[0].to_string(), "chars(geo-async)");

        let row = runtime()
            .execute_first_row_fields_for_async("geo", "SELECT value", &[], CachePolicy::Bypass)
            .await
            .expect("async geo first row");
        assert_eq!(row[0].to_string(), "chars(geo-async)");
    }

    // -----------------------------------------------------------------------
    // Redis cache unit tests
    // -----------------------------------------------------------------------

    fn redis_ck(cmd: RedisCmdTag, generation: u64, key: &str, args: &[&str]) -> RedisCacheKey {
        let mut hasher = DefaultHasher::new();
        key.hash(&mut hasher);
        let key_hash = hasher.finish();
        let mut hasher = DefaultHasher::new();
        for arg in args {
            arg.hash(&mut hasher);
        }
        let args_hash = hasher.finish();
        RedisCacheKey {
            generation,
            cmd_tag: cmd,
            key_hash,
            args_hash,
        }
    }

    #[test]
    fn redis_cache_hit_and_miss() {
        let rt = KnowledgeRuntime::new(64);
        rt.configure_redis_cache(true, 64);

        let ck = redis_ck(RedisCmdTag::Get, 1, "user:1", &[]);
        // First access — miss
        assert!(rt.redis_cache_get(&ck).is_none());
        // Store
        rt.redis_cache_put(ck.clone(), CachedRedisValue::Bool(true));
        // Second access — hit
        let val = rt.redis_cache_get(&ck).expect("should hit cache");
        assert!(matches!(val, CachedRedisValue::Bool(true)));
    }

    #[test]
    fn redis_cache_global_enabled_access() {
        let rt = KnowledgeRuntime::new(64);
        rt.configure_redis_cache(true, 64);

        let ck = redis_ck(RedisCmdTag::Get, 1, "k", &[]);
        rt.redis_cache_put(ck.clone(), CachedRedisValue::Bool(true));
        assert!(rt.redis_cache_get(&ck).is_some());
    }

    #[test]
    fn redis_cache_global_disabled_blocks_all() {
        let rt = KnowledgeRuntime::new(64);
        rt.configure_redis_cache(false, 64);

        let ck = redis_ck(RedisCmdTag::BfExists, 1, "any_key", &["item"]);
        rt.redis_cache_put(ck.clone(), CachedRedisValue::Bool(true));
        // Global disabled — no reads
        assert!(rt.redis_cache_get(&ck).is_none());
    }

    #[test]
    fn redis_cache_ttl_expiry() {
        let rt = KnowledgeRuntime::new(64);
        rt.configure_redis_cache(true, 64);

        let ck = redis_ck(RedisCmdTag::BfExists, 1, "k", &["item"]);
        // Store with 1ms TTL
        rt.redis_cache_put_with_ttl(
            ck.clone(),
            CachedRedisValue::Bool(true),
            1, // 1ms TTL
        );
        // Immediately valid
        assert!(rt.redis_cache_get(&ck).is_some());
        // Wait for TTL to expire
        std::thread::sleep(std::time::Duration::from_millis(5));
        // Should be expired
        assert!(rt.redis_cache_get(&ck).is_none());
    }

    #[test]
    fn redis_cache_no_ttl_never_expires() {
        let rt = KnowledgeRuntime::new(64);
        rt.configure_redis_cache(true, 64);

        let ck = redis_ck(RedisCmdTag::Get, 1, "k", &[]);
        // Store with ttl = 0 (generation-only)
        rt.redis_cache_put(ck.clone(), CachedRedisValue::Bool(true));
        // Should be valid
        assert!(rt.redis_cache_get(&ck).is_some());
        std::thread::sleep(std::time::Duration::from_millis(5));
        // Still valid — 0 means no TTL
        assert!(rt.redis_cache_get(&ck).is_some());
    }

    #[test]
    fn redis_cache_generation_isolation() {
        let rt = KnowledgeRuntime::new(64);
        rt.configure_redis_cache(true, 64);

        let ck_gen1 = redis_ck(RedisCmdTag::BfExists, 1, "key", &["item"]);
        let ck_gen2 = redis_ck(RedisCmdTag::BfExists, 2, "key", &["item"]);

        // Store with generation 1
        rt.redis_cache_put(ck_gen1.clone(), CachedRedisValue::Bool(false));

        // Same key but generation 2 — miss
        assert!(rt.redis_cache_get(&ck_gen2).is_none());
        // Generation 1 — hit
        assert!(rt.redis_cache_get(&ck_gen1).is_some());
    }
}