limiteron 0.3.0-rc.7

// Copyright (c) 2026 Kirky.X🌠
// SPDX-License-Identifier: MIT
//! Governor 主控制器 - 重构版本
//!
//! 流量控制的核心控制器，重构后具有更好的模块化设计：
//! - 使用专门的并行封禁检查器提高性能
//! - 使用 RuleBuilder 构建规则和决策链
//! - 使用 StatsManager 管理统计信息
//! - 集成 L1 本地缓存层提高热点访问性能
//! - 简化核心逻辑，提高可维护性
//! - 保持向后兼容性

use crate::config::{ConfigChangeRecord, ConfigHistory, FlowControlConfig};
use crate::decision_chain::DecisionChain;
use crate::error::Decision;
use crate::error::LimiteronError;
#[cfg(feature = "fallback")]
use crate::error::RejectionMetadata;
#[cfg(feature = "fallback")]
use crate::fallback::FallbackManager;
use crate::i18n::t;
#[cfg(feature = "fallback")]
use crate::l1_cache::IslandFallbackStrategy;
#[cfg(feature = "fallback")]
use crate::l1_cache::IslandModeConfig;
use crate::l1_cache::{CacheableDecision, L1Cache, L1CacheConfig, RateLimitCacheKey};

/// 决策结果维度标签（prometheus 指标与 telemetry span 共用同一判定；
/// "error" 形态仅链执行失败产生，纯决策无该值）
#[cfg(any(feature = "prometheus", feature = "telemetry"))]
fn decision_outcome(decision: &Decision) -> &'static str {
    match decision {
        Decision::Allowed(_) => "allowed",
        Decision::Rejected(_) => "rejected",
        Decision::Banned(_) => "banned",
    }
}

/// 日志指纹:仅暴露长度,不含任何标识符内容(CodeQL cleartext-logging 结构性断源)。
fn log_fingerprint(key: &str) -> String {
    format!("<{} chars>", key.len())
}
use crate::matchers::{IdentifierExtractor, RequestContext, RuleMatcher};
use crate::rules::{RuleBuilder, StatsManager, StatsSnapshot};
// storage module removed as part of direct-inheritance refactoring
// Use dbnexus traits directly instead
// Re-exported from storage module for compatibility
#[cfg(all(feature = "ban-manager", not(feature = "parallel-checker")))]
use crate::error::BanInfo;
#[cfg(all(feature = "ban-manager", not(feature = "parallel-checker")))]
use crate::storage::BanTarget;
use crate::storage::{BanStorage, Storage};
use dashmap::DashMap;
#[cfg(feature = "parallel-checker")]
use log::warn;
use log::{debug, error, info, trace};
use std::sync::Arc;
use tokio::sync::RwLock;

// Conditional imports for optional features
#[cfg(feature = "ban-manager")]
use crate::BanSource;
#[cfg(feature = "ban-manager")]
use crate::ban::BanManager;
#[cfg(feature = "circuit-breaker")]
use crate::circuit::CircuitBreaker;
#[cfg(feature = "audit-log")]
use crate::logging::AuditLogger;
#[cfg(any(
    feature = "parallel-checker",
    feature = "ban-manager",
    feature = "multi-tenant"
))]
use crate::matchers::Identifier;
#[cfg(feature = "prometheus")]
use crate::telemetry::Metrics;
#[cfg(feature = "telemetry")]
use crate::telemetry::Tracer;

/// Governor 统计信息
///
/// 保持向后兼容性的统计信息结构体。
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct GovernorStats {
    /// 总请求数
    pub total_requests: u64,
    /// 允许的请求数
    pub allowed_requests: u64,
    /// 拒绝的请求数
    pub rejected_requests: u64,
    /// 封禁的请求数
    pub banned_requests: u64,
    /// 错误数
    pub error_count: u64,
    /// 最后更新时间
    pub last_updated: Option<chrono::DateTime<chrono::Utc>>,
}

impl From<StatsSnapshot> for GovernorStats {
    fn from(snapshot: StatsSnapshot) -> Self {
        Self {
            total_requests: snapshot.total_requests,
            allowed_requests: snapshot.allowed_requests,
            rejected_requests: snapshot.rejected_requests,
            banned_requests: snapshot.banned_requests,
            error_count: snapshot.error_count,
            last_updated: snapshot.last_updated,
        }
    }
}

/// Governor 主控制器
///
/// 重构后的 Governor，具有更清晰的职责分离和更好的性能。
/// 孤岛 ConservativeQuota 的本地固定窗口计数器
///
/// 降级期为请求提供有界放行：窗口内至多放行 max_requests 个，
/// 超出拒绝（带窗口翻转时间作为 retry_after）。纯逻辑可测。
#[cfg(feature = "fallback")]
pub(crate) struct ConservativeQuotaCounter {
    window_start_secs: u64,
    count: u64,
}

#[cfg(feature = "fallback")]
impl ConservativeQuotaCounter {
    fn try_acquire(&mut self, max_requests: u64, window_secs: u64, now_secs: u64) -> bool {
        let window_secs = window_secs.max(1);
        if now_secs >= self.window_start_secs.saturating_add(window_secs) {
            self.window_start_secs = now_secs;
            self.count = 0;
        }
        if self.count >= max_requests {
            return false;
        }
        self.count += 1;
        true
    }
}

pub struct Governor {
    /// 配置
    config: Arc<RwLock<FlowControlConfig>>,

    /// 存储后端
    storage: Arc<dyn Storage>,

    /// 封禁存储
    ban_storage: Arc<dyn BanStorage>,

    /// 共享配额账本（storage-backed QuotaLimiter 的可选后端）
    quota_storage: Option<Arc<dyn crate::storage::QuotaStorage>>,

    /// 孤岛 ConservativeQuota 本地计数器状态（惰性初始化）
    #[cfg(feature = "fallback")]
    conservative_quota: parking_lot::Mutex<Option<ConservativeQuotaCounter>>,

    /// 封禁管理器
    #[cfg(feature = "ban-manager")]
    ban_manager: Arc<BanManager>,

    /// 并行封禁检查器
    #[cfg(feature = "parallel-checker")]
    parallel_ban_checker: Arc<crate::storage::ParallelBanChecker>,

    /// 决策链
    decision_chain: Arc<RwLock<DecisionChain>>,

    /// 规则匹配器
    rule_matcher: Arc<RwLock<RuleMatcher>>,

    /// 规则对应的决策链
    rule_chains: Arc<RwLock<DashMap<String, DecisionChain>>>,

    /// 标识符提取器
    identifier_extractor: Arc<dyn IdentifierExtractor>,

    /// 熔断器
    #[cfg(feature = "circuit-breaker")]
    circuit_breaker: Arc<CircuitBreaker>,

    /// 审计日志记录器
    #[cfg(feature = "audit-log")]
    audit_logger: Arc<RwLock<Option<Arc<AuditLogger>>>>,

    /// 配置历史记录
    config_history: Arc<RwLock<ConfigHistory>>,

    /// 统计管理器
    stats: StatsManager,

    /// 本地缓存（用于缓存热点限流结果）
    l1_cache: L1Cache<CacheableDecision>,

    /// 是否启用 缓存
    l1_cache_enabled: std::sync::atomic::AtomicBool,

    /// 降级管理器（可选，feature-gated）
    #[cfg(feature = "fallback")]
    fallback_manager: Option<Arc<FallbackManager>>,

    /// 事件发射器（可选，feature-gated）
    #[cfg(feature = "event-system")]
    event_emitter: Option<Arc<crate::events::EventEmitter>>,

    /// 指标收集器（可选，feature-gated）
    #[cfg(feature = "prometheus")]
    metrics: Option<Arc<Metrics>>,

    /// per-rule/per-limiter 维度指标开关（默认开启；标签基数随规则与
    /// 限流器数量增长，超大规模配置可关闭仅保留全局计数）
    #[cfg(feature = "prometheus")]
    per_rule_metrics: bool,

    /// 追踪器（可选，feature-gated）
    #[cfg(feature = "telemetry")]
    tracer: Option<Arc<Tracer>>,

    /// 租户解析器（可选，feature-gated `multi-tenant`）
    ///
    /// 配置后，决策键（缓存键/事件键/封禁键）以 tenant+key 复合键计算，
    /// 实现存储/封禁与按标识符键控的限流器按租户隔离；未配置时行为与现状
    /// 逐位一致。注意：规则链中的配额限流器按 `rule.id` 键控
    /// （`QuotaLimiter::with_storage` 的 storage key），其用量不随命名空间
    /// 隔离——租户级额度需用 `QuotaController` 的租户键 API。
    #[cfg(feature = "multi-tenant")]
    tenant_resolver: Option<Arc<dyn crate::tenant::TenantResolver>>,

    /// 优雅关闭令牌：取消时通知所有后台任务退出
    shutdown_token: tokio_util::sync::CancellationToken,

    /// 配置热重载 watcher 的取消令牌：集成方经 `register_config_watcher_token`
    /// 注册（`watch_and_reload` / `watch_and_reload_with_applier` 返回的令牌，
    /// 支持同时监听多个配置文件），shutdown 统一取消
    config_watcher_tokens: parking_lot::Mutex<Vec<tokio_util::sync::CancellationToken>>,

    /// 关闭时状态快照落盘目录（None = 不落盘；循仓库 data/ 目录惯例由调用方指定）
    shutdown_snapshot_dir: Option<std::path::PathBuf>,

    /// 是否已关闭（幂等性保证）
    is_shutdown: std::sync::atomic::AtomicBool,

    /// 已导出的规则内限流器拒绝计数快照（上次导出值，增量求基）——
    /// 各链「已导出版本号」下沉为链内原子量（`DecisionChain`），稳态
    /// 版本比较零锁；本锁仅拒绝路径的合并落账持有——监控 feature 门控
    #[cfg(feature = "prometheus")]
    exported_rejections: parking_lot::Mutex<ExportedRejections>,

    /// 可控时钟：贯通给拥有 `with_clock` 的限流器类型（生产默认
    /// `SystemClock`，测试可注入 `MockClock` 使窗口/令牌补充等
    /// 时间行为确定可测）。无时钟类型（gcra/concurrency/quota_limiter
    /// 等）保持各自时间源，不受此字段影响。
    clock: Arc<dyn crate::clock::Clock>,
}

/// per-limiter 拒绝增量导出的 Governor 侧状态（监控 feature 门控）
#[cfg(feature = "prometheus")]
#[derive(Default)]
struct ExportedRejections {
    /// rule_id → (node_name → 上次导出的累计拒绝值)
    ///
    /// 外层按规则分桶：导出循环按节点迭代，内层 `get_mut(&str)` 命中时
    /// 零分配（平铺 `(rule, node)` 键需每次克隆 rule_id，拒绝即导出的
    /// flood 场景是热路径分配点）
    per_rule: std::collections::HashMap<String, std::collections::HashMap<String, u64>>,
}

#[cfg(feature = "prometheus")]
impl ExportedRejections {
    /// 清空导出快照（热更新换装后新链计数从零重计，快照必须同步归零）
    fn clear(&mut self) {
        self.per_rule.clear();
    }
}

/// Governor 构建器
///
/// 用于链式配置 Governor 实例。
///
/// # 示例
///
/// ```rust,no_run
/// use limiteron::Governor;
/// use limiteron::storage::{MemoryStorage, MemoryBanStorage};
/// use std::sync::Arc;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
///     let storage: Arc<dyn limiteron::storage::Storage> = MemoryStorage::create_storage();
///     let ban_storage: Arc<dyn limiteron::storage::BanStorage> = MemoryBanStorage::create_ban_storage();
///
///     let governor = Governor::builder()
///         .with_storage(storage)
///         .with_ban_storage(ban_storage)
///         .build()
///         .await
///         .unwrap();
///     Ok(())
/// }
/// ```
#[derive(Clone, Default)]
#[allow(clippy::type_complexity)]
pub struct GovernorBuilder {
    config: Option<FlowControlConfig>,
    storage: Option<Arc<dyn Storage>>,
    ban_storage: Option<Arc<dyn BanStorage>>,
    quota_storage: Option<Arc<dyn crate::storage::QuotaStorage>>,
    identifier_extractor: Option<Arc<dyn crate::matchers::IdentifierExtractor>>,
    #[cfg(feature = "circuit-breaker")]
    circuit_breaker: Option<Arc<CircuitBreaker>>,
    #[cfg(feature = "audit-log")]
    audit_logger: Option<Arc<crate::logging::AuditLogger>>,
    #[cfg(feature = "prometheus")]
    metrics: Option<Arc<Metrics>>,
    #[cfg(feature = "prometheus")]
    per_rule_metrics: bool,
    #[cfg(feature = "telemetry")]
    tracer: Option<Arc<Tracer>>,
    #[cfg(feature = "parallel-checker")]
    parallel_ban_checker: Option<Arc<crate::storage::ParallelBanChecker>>,
    /// 缓存配置
    l1_cache_config: Option<L1CacheConfig>,
    /// 是否启用 缓存
    l1_cache_enabled: bool,
    /// 降级管理器（可选，feature-gated）
    #[cfg(feature = "fallback")]
    fallback_manager: Option<Arc<FallbackManager>>,
    /// 事件发射器（可选）
    #[cfg(feature = "event-system")]
    event_emitter: Option<Arc<crate::events::EventEmitter>>,
    /// 自定义匹配器注册表（可选）：提供后，配置中的 `Custom` 匹配器
    /// 在构建期从注册表解析并真实参与运行时求值（E1）
    custom_matcher_registry: Option<Arc<crate::matchers::custom::CustomMatcherRegistry>>,
    /// 租户解析器（可选）
    #[cfg(feature = "multi-tenant")]
    tenant_resolver: Option<Arc<dyn crate::tenant::TenantResolver>>,
    /// 关闭时状态快照落盘目录（可选，默认不落盘）
    shutdown_snapshot_dir: Option<std::path::PathBuf>,
    /// 可控时钟（可选，缺省 `SystemClock`；见 `with_clock`）
    clock: Option<Arc<dyn crate::clock::Clock>>,
}

impl GovernorBuilder {
    /// 创建新的 GovernorBuilder
    pub fn new() -> Self {
        Self {
            config: None,
            storage: None,
            ban_storage: None,
            quota_storage: None,
            identifier_extractor: None,
            #[cfg(feature = "circuit-breaker")]
            circuit_breaker: None,
            #[cfg(feature = "audit-log")]
            audit_logger: None,
            #[cfg(feature = "prometheus")]
            metrics: None,
            #[cfg(feature = "prometheus")]
            per_rule_metrics: true,
            #[cfg(feature = "telemetry")]
            tracer: None,
            #[cfg(feature = "parallel-checker")]
            parallel_ban_checker: None,
            l1_cache_config: None,
            l1_cache_enabled: true, // 默认启用 缓存（负缓存语义：仅缓存拒绝/封禁决策）
            #[cfg(feature = "fallback")]
            fallback_manager: None,
            #[cfg(feature = "event-system")]
            event_emitter: None,
            custom_matcher_registry: None,
            #[cfg(feature = "multi-tenant")]
            tenant_resolver: None,
            shutdown_snapshot_dir: None,
            clock: None,
        }
    }

    /// 设置租户解析器（feature `multi-tenant`）
    ///
    /// 配置后，Governor 决策键以 tenant+key 复合键计算：
    /// 缓存键、事件键、封禁键均带租户命名空间前缀，
    /// 存储/配额/封禁按租户隔离。未配置时行为与现状逐位一致。
    #[cfg(feature = "multi-tenant")]
    pub fn with_tenant_resolver(
        mut self,
        resolver: Arc<dyn crate::tenant::TenantResolver>,
    ) -> Self {
        self.tenant_resolver = Some(resolver);
        self
    }

    /// 设置关闭时状态快照的落盘目录
    ///
    /// 快照以 JSON 写入 `<dir>/governor-shutdown-snapshot.json`（原子替换写），
    /// 循仓库 `data/` 目录惯例由调用方显式指定路径；库默认不落盘，
    /// 避免未经允许的文件系统副作用（只读文件系统等场景）。
    pub fn with_shutdown_snapshot_dir(mut self, dir: impl Into<std::path::PathBuf>) -> Self {
        self.shutdown_snapshot_dir = Some(dir.into());
        self
    }

    /// 注入自定义匹配器注册表（E1）
    ///
    /// 提供后，配置中的 `Custom` 匹配器在 `build()` 时从注册表解析：
    /// 已注册的匹配器真实参与运行时求值，未注册的仍为恒不匹配占位
    /// 并在构建期告警。
    pub fn with_custom_matcher_registry(
        mut self,
        registry: Arc<crate::matchers::custom::CustomMatcherRegistry>,
    ) -> Self {
        self.custom_matcher_registry = Some(registry);
        self
    }

    /// 设置流量控制配置
    pub fn with_config(mut self, config: FlowControlConfig) -> Self {
        self.config = Some(config);
        self
    }

    /// 设置存储后端
    pub fn with_storage(mut self, storage: Arc<dyn Storage>) -> Self {
        self.storage = Some(storage);
        self
    }

    /// 设置封禁存储后端
    pub fn with_ban_storage(mut self, ban_storage: Arc<dyn BanStorage>) -> Self {
        self.ban_storage = Some(ban_storage);
        self
    }

    /// 设置共享配额账本（storage-backed 配额限流）
    ///
    /// 注入后，配置中的 Quota 规则以 `QuotaStorage` 为真账：
    /// 跨实例额度一致、进程重启不清零。未注入保持纯内存配额（单实例语义）。
    pub fn with_quota_storage(
        mut self,
        quota_storage: Arc<dyn crate::storage::QuotaStorage>,
    ) -> Self {
        self.quota_storage = Some(quota_storage);
        self
    }

    /// 设置可控时钟（缺省 `SystemClock`）
    ///
    /// 时钟贯通给拥有 `with_clock` 的限流器类型（TokenBucket、SlidingWindow、
    /// FixedWindow、SlidingWindowLog、LeakyBucket、PriorityQueue、AdmissionControl），
    /// 注入 `MockClock` 后窗口滑动/令牌补充等时间行为可在测试中确定驱动；
    /// **无时钟类型（gcra/concurrency/quota_limiter 等）保持各自时间源，
    /// 不受此时钟影响**。
    pub fn with_clock(mut self, clock: Arc<dyn crate::clock::Clock>) -> Self {
        self.clock = Some(clock);
        self
    }

    /// 设置标识符提取器
    pub fn with_identifier_extractor(
        mut self,
        extractor: Arc<dyn crate::matchers::IdentifierExtractor>,
    ) -> Self {
        self.identifier_extractor = Some(extractor);
        self
    }

    /// 设置熔断器
    #[cfg(feature = "circuit-breaker")]
    pub fn with_circuit_breaker(mut self, circuit_breaker: Arc<CircuitBreaker>) -> Self {
        self.circuit_breaker = Some(circuit_breaker);
        self
    }

    /// 设置审计日志记录器
    #[cfg(feature = "audit-log")]
    pub fn with_audit_logger(mut self, audit_logger: Arc<crate::logging::AuditLogger>) -> Self {
        self.audit_logger = Some(audit_logger);
        self
    }

    /// 设置指标收集器
    #[cfg(feature = "prometheus")]
    pub fn with_metrics(mut self, metrics: Arc<Metrics>) -> Self {
        self.metrics = Some(metrics);
        self
    }

    /// 设置 per-rule/per-limiter 维度指标开关（默认开启）
    ///
    /// 关闭后 Governor 仅记录全局计数（requests_total 等），不再导出
    /// `flowguard_rule_checks_total` / `flowguard_rule_limiter_rejections_total`
    /// 维度序列——标签基数随规则与限流器数量增长，超大规模配置可关闭。
    #[cfg(feature = "prometheus")]
    pub fn with_per_rule_metrics(mut self, enabled: bool) -> Self {
        self.per_rule_metrics = enabled;
        self
    }

    /// 设置追踪器
    #[cfg(feature = "telemetry")]
    pub fn with_tracer(mut self, tracer: Arc<Tracer>) -> Self {
        self.tracer = Some(tracer);
        self
    }

    /// 设置 缓存配置
    pub fn with_l1_cache_config(mut self, config: L1CacheConfig) -> Self {
        self.l1_cache_config = Some(config);
        self
    }

    /// 启用或禁用 缓存
    pub fn with_l1_cache_enabled(mut self, enabled: bool) -> Self {
        self.l1_cache_enabled = enabled;
        self
    }

    /// 设置降级管理器
    ///
    /// 当设置降级管理器后，Governor 会在存储层故障时自动触发孤岛模式，
    /// 并在 check 流程中使用降级策略。
    ///
    /// # 参数
    /// - `fallback_manager`: 降级管理器实例
    #[cfg(feature = "fallback")]
    pub fn with_fallback_manager(mut self, fallback_manager: Arc<FallbackManager>) -> Self {
        self.fallback_manager = Some(fallback_manager);
        self
    }

    /// 设置事件发射器
    #[cfg(feature = "event-system")]
    pub fn with_event_emitter(mut self, emitter: Arc<crate::events::EventEmitter>) -> Self {
        self.event_emitter = Some(emitter);
        self
    }

    /// 构建 Governor 实例
    ///
    /// # 返回
    ///
    /// * `Ok(Governor)` - 构建成功
    /// * `Err(LimiteronError)` - 构建失败（配置错误或依赖缺失）
    ///
    pub async fn build(self) -> Result<Governor, LimiteronError> {
        let config = self.config.unwrap_or_default();

        // 校验配置
        config.validate().map_err(LimiteronError::ConfigError)?;

        // 获取存储后端（必需依赖）
        let storage = self
            .storage
            .ok_or_else(|| LimiteronError::DependencyError("storage is required".to_string()))?;
        let ban_storage = self.ban_storage.ok_or_else(|| {
            LimiteronError::DependencyError("ban_storage is required".to_string())
        })?;

        // 创建封禁管理器
        #[cfg(feature = "ban-manager")]
        let ban_manager = {
            use crate::ban::{BanManager, BanManagerConfig};
            let config = BanManagerConfig::default();
            BanManager::with_dependencies(ban_storage.clone(), config)
                .await
                .map(Arc::new)?
        };

        // 创建并行封禁检查器（parallel-checker 蕴含 ban-manager，BanManager 必可用）
        #[cfg(feature = "parallel-checker")]
        let parallel_ban_checker = self.parallel_ban_checker.unwrap_or_else(|| {
            Arc::new(crate::storage::ParallelBanChecker::new(ban_manager.clone()))
        });

        // 创建标识符提取器（如果未提供）
        let identifier_extractor = if let Some(extractor) = self.identifier_extractor {
            extractor
        } else {
            Arc::new(
                crate::matchers::CompositeExtractor::builder()
                    .add_extractor(Box::new(crate::matchers::UserIdExtractor::from_header(
                        "X-User-Id",
                    )))
                    .add_extractor(Box::new(crate::matchers::IpExtractor::builder().build()))
                    .add_extractor(Box::new(crate::matchers::ApiKeyExtractor::from_header(
                        "X-API-Key",
                    )))
                    .build(),
            )
        };

        // 使用 RuleBuilder 创建规则匹配器（E1：提供注册表时 Custom
        // 匹配器真实参与求值）
        let rules = if let Some(registry) = &self.custom_matcher_registry {
            RuleBuilder::build_rules_with_registry(&config, registry).await?
        } else {
            RuleBuilder::build_rules(&config)?
        };
        let rule_matcher = Arc::new(tokio::sync::RwLock::new(
            crate::matchers::RuleMatcher::with_dependencies(rules),
        ));

        // 创建决策链
        let decision_chain = Arc::new(tokio::sync::RwLock::new(
            crate::decision_chain::DecisionChain::with_dependencies(vec![]),
        ));

        // 创建熔断器
        #[cfg(feature = "circuit-breaker")]
        let circuit_breaker = self.circuit_breaker.unwrap_or_else(|| {
            Arc::new(CircuitBreaker::with_dependencies(
                crate::circuit::CircuitBreakerConfig::default(),
            ))
        });

        // 创建审计日志记录器
        #[cfg(feature = "audit-log")]
        let audit_logger = self
            .audit_logger
            .map(|logger| Arc::new(tokio::sync::RwLock::new(Some(logger))))
            .unwrap_or_else(|| Arc::new(tokio::sync::RwLock::new(None)));

        // 使用 RuleBuilder 创建规则对应的决策链
        let clock: Arc<dyn crate::clock::Clock> = self
            .clock
            .clone()
            .unwrap_or_else(|| Arc::new(crate::clock::SystemClock));
        let rule_chains_map = RuleBuilder::build_rule_chains_with_quota_storage(
            &config,
            self.quota_storage.as_ref(),
            clock.clone(),
        )?;
        let rule_chains = Arc::new(tokio::sync::RwLock::new(rule_chains_map));

        // 创建 缓存
        let l1_cache_config = self
            .l1_cache_config
            .unwrap_or_else(|| L1CacheConfig::new(std::time::Duration::from_secs(60), 10_000));
        let l1_cache = L1Cache::with_config(l1_cache_config).await.map_err(|e| {
            LimiteronError::DependencyError(format!("Failed to create L1Cache: {}", e))
        })?;
        let l1_cache_enabled = self.l1_cache_enabled;

        // 集成降级管理器 (feature-gated)
        #[cfg(feature = "fallback")]
        let fallback_manager = self.fallback_manager.clone();

        // 如果提供了 fallback_manager，注册孤岛模式回调
        #[cfg(feature = "fallback")]
        if let Some(ref fm) = self.fallback_manager {
            let l1_cache_ref = l1_cache.clone();

            // 注册孤岛模式回调（直接 await 确保注册完成）
            fm.register_island_mode_callback(std::sync::Arc::new(move |is_island| {
                if is_island {
                    // 进入孤岛模式：配置 缓存的孤岛降级策略
                    let island_config =
                        IslandModeConfig::new(IslandFallbackStrategy::LocalDecision);
                    l1_cache_ref.enable_island_mode(island_config);
                } else {
                    // 退出孤岛模式
                    l1_cache_ref.disable_island_mode();
                }
            }))
            .await;

            log::info!(
                target: "governor",
                "{}",
                t("governor-island-callback-registered", &[])
            );
        }

        Ok(Governor {
            config: Arc::new(tokio::sync::RwLock::new(config)),
            storage,
            ban_storage,
            quota_storage: self.quota_storage,
            #[cfg(feature = "fallback")]
            conservative_quota: parking_lot::Mutex::new(None),
            #[cfg(feature = "ban-manager")]
            ban_manager,
            #[cfg(feature = "parallel-checker")]
            parallel_ban_checker,
            #[cfg(feature = "circuit-breaker")]
            circuit_breaker,
            decision_chain,
            rule_matcher,
            rule_chains,
            identifier_extractor,
            #[cfg(feature = "audit-log")]
            audit_logger,
            config_history: Arc::new(tokio::sync::RwLock::new(ConfigHistory::new(100))),
            stats: StatsManager::new(),
            l1_cache,
            l1_cache_enabled: std::sync::atomic::AtomicBool::new(l1_cache_enabled),
            #[cfg(feature = "fallback")]
            fallback_manager,
            #[cfg(feature = "event-system")]
            event_emitter: self.event_emitter,
            #[cfg(feature = "prometheus")]
            metrics: self.metrics,
            #[cfg(feature = "prometheus")]
            per_rule_metrics: self.per_rule_metrics,
            #[cfg(feature = "telemetry")]
            tracer: self.tracer,
            #[cfg(feature = "multi-tenant")]
            tenant_resolver: self.tenant_resolver,
            shutdown_token: tokio_util::sync::CancellationToken::new(),
            config_watcher_tokens: parking_lot::Mutex::new(Vec::new()),
            shutdown_snapshot_dir: self.shutdown_snapshot_dir.clone(),
            is_shutdown: std::sync::atomic::AtomicBool::new(false),
            #[cfg(feature = "prometheus")]
            exported_rejections: parking_lot::Mutex::new(ExportedRejections::default()),
            clock,
        })
    }
}

impl Governor {
    /// 创建 GovernorBuilder 用于链式配置
    pub fn builder() -> GovernorBuilder {
        GovernorBuilder::new()
    }

    /// 开箱即用：创建使用默认配置的 Governor
    ///
    /// 此方法使用内存存储作为默认依赖，无需外部配置即可运行。
    /// 适用于快速原型、测试或独立使用场景。
    ///
    /// **注意**：默认配置使用内存存储，不适用于多实例生产环境。
    /// 对于生产环境，建议使用 `builder()` 或 `with_dependencies()` 方法
    /// 配合持久化存储（如 PostgreSQL）。
    ///
    /// # 返回
    /// - `Ok(Governor)`: 创建成功
    /// - `Err(LimiteronError)`: 初始化失败
    ///
    /// 默认配置内置一条兜底限流规则（内存 TokenBucket：容量 100、
    /// 每秒补充 10、匹配所有用户、超限拒绝），使「开箱即用」真正可用。
    /// 修复：旧实现默认空配置必然 `validate()` 失败后 `expect` panic。
    /// 需要自定义规则集时使用 `builder()`。
    ///
    /// # 示例
    ///
    /// ```rust,no_run
    /// use limiteron::Governor;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), limiteron::error::LimiteronError> {
    ///     let governor = Governor::new().await?;
    ///     // governor 现在可以用于流量控制检查
    ///     Ok(())
    /// }
    /// ```
    pub async fn new() -> Result<Self, LimiteronError> {
        use crate::config::{Action, ActionConfig, LimiterConfig, Matcher, Rule};
        use crate::storage::{MemoryBanStorage, MemoryStorage};

        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        // 默认配置：内置一条兜底限流规则（见 doc），保证 validate() 通过
        let config = FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::GlobalConfig::default(),
            rules: vec![Rule {
                id: "default".to_string(),
                name: "Default out-of-the-box rule".to_string(),
                priority: 100,
                matchers: vec![Matcher::User {
                    user_ids: vec!["*".to_string()],
                }],
                limiters: vec![LimiterConfig::TokenBucket {
                    capacity: 100,
                    refill_rate: 10,
                }],
                action: ActionConfig {
                    on_exceed: Action::Reject,
                    ban: None,
                },
            }],
        };

        Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
    }

    /// 使用依赖注入创建 Governor 实例（用于应用容器集成）
    ///
    /// # 返回
    /// - `Ok(Governor)`: 创建成功
    /// - `Err(LimiteronError)`: 依赖初始化失败（修复：不再在初始化
    ///   失败时 panic，而是把错误交还给调用方处置）
    #[allow(clippy::too_many_arguments)]
    pub async fn with_dependencies(
        config: Arc<tokio::sync::RwLock<FlowControlConfig>>,
        storage: Arc<dyn Storage>,
        ban_storage: Arc<dyn BanStorage>,
        identifier_extractor: Arc<dyn crate::matchers::IdentifierExtractor>,
        rule_matcher: Arc<tokio::sync::RwLock<crate::matchers::RuleMatcher>>,
        decision_chain: Arc<tokio::sync::RwLock<crate::decision_chain::DecisionChain>>,
        rule_chains: Arc<
            tokio::sync::RwLock<DashMap<String, crate::decision_chain::DecisionChain>>,
        >,
        #[cfg(feature = "circuit-breaker")] circuit_breaker: Arc<CircuitBreaker>,
    ) -> Result<Self, LimiteronError> {
        // 创建封禁管理器
        #[cfg(feature = "ban-manager")]
        use crate::ban::{BanManager, BanManagerConfig};

        #[cfg(feature = "ban-manager")]
        let ban_manager: Arc<BanManager> =
            BanManager::with_dependencies(ban_storage.clone(), BanManagerConfig::default())
                .await
                .map(Arc::new)?;

        // 创建并行封禁检查器
        #[cfg(feature = "parallel-checker")]
        let parallel_ban_checker: Arc<crate::storage::ParallelBanChecker> =
            Arc::new(crate::storage::ParallelBanChecker::new(ban_manager.clone()));

        Ok(Self {
            config,
            storage,
            ban_storage,
            // with_dependencies 直构路径不注入共享配额账本（保持签名兼容），
            // 需要 storage-backed 配额请走 GovernorBuilder::with_quota_storage
            quota_storage: None,
            #[cfg(feature = "fallback")]
            conservative_quota: parking_lot::Mutex::new(None),
            #[cfg(feature = "ban-manager")]
            ban_manager,
            #[cfg(feature = "parallel-checker")]
            parallel_ban_checker,
            #[cfg(feature = "circuit-breaker")]
            circuit_breaker,
            decision_chain,
            rule_matcher,
            rule_chains,
            identifier_extractor,
            #[cfg(feature = "audit-log")]
            audit_logger: Arc::new(tokio::sync::RwLock::new(None)),
            config_history: Arc::new(tokio::sync::RwLock::new(ConfigHistory::new(100))),
            stats: StatsManager::new(),
            l1_cache: L1Cache::new().await.map_err(|e| {
                LimiteronError::ConfigError(format!("Failed to create L1Cache: {}", e))
            })?,
            l1_cache_enabled: std::sync::atomic::AtomicBool::new(true),
            #[cfg(feature = "fallback")]
            fallback_manager: None,
            #[cfg(feature = "event-system")]
            event_emitter: None,
            #[cfg(feature = "prometheus")]
            metrics: None,
            #[cfg(feature = "prometheus")]
            per_rule_metrics: true,
            #[cfg(feature = "telemetry")]
            tracer: None,
            #[cfg(feature = "multi-tenant")]
            tenant_resolver: None,
            shutdown_token: tokio_util::sync::CancellationToken::new(),
            config_watcher_tokens: parking_lot::Mutex::new(Vec::new()),
            shutdown_snapshot_dir: None,
            is_shutdown: std::sync::atomic::AtomicBool::new(false),
            #[cfg(feature = "prometheus")]
            exported_rejections: parking_lot::Mutex::new(ExportedRejections::default()),
            clock: Arc::new(crate::clock::SystemClock),
        })
    }

    /// 创建新的 Governor 实例（使用显式配置）
    ///
    /// 与 `new()` 不同，此方法需要显式提供配置和存储依赖。
    /// 适用于需要自定义配置或使用持久化存储的场景。
    ///
    /// # 参数
    ///
    /// - `config`: 流量控制配置
    /// - `storage`: 存储后端
    /// - `ban_storage`: 封禁存储后端
    /// - `metrics`: 可选的指标收集器
    /// - `tracer`: 可选的追踪器
    ///
    /// # 示例
    ///
    /// ```rust,no_run
    /// use limiteron::Governor;
    /// use limiteron::config::FlowControlConfig;
    /// use limiteron::storage::MemoryStorage;
    /// use std::sync::Arc;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let config = FlowControlConfig::default();
    ///     let storage: Arc<dyn limiteron::storage::Storage> = Arc::new(MemoryStorage::new());
    ///     let ban_storage: Arc<dyn limiteron::storage::BanStorage> = Arc::new(limiteron::storage::MemoryBanStorage::new());
    ///     let _ = Governor::builder()
    ///         .with_config(config)
    ///         .with_storage(storage)
    ///         .with_ban_storage(ban_storage)
    ///         .build()
    ///         .await;
    /// }
    /// ```
    pub async fn with_storage(
        config: FlowControlConfig,
        storage: Arc<dyn Storage>,
        ban_storage: Arc<dyn BanStorage>,
        #[cfg(feature = "prometheus")] metrics: Option<Arc<Metrics>>,
        #[cfg(feature = "telemetry")] tracer: Option<Arc<Tracer>>,
    ) -> Result<Self, LimiteronError> {
        // 使用 builder 模式创建 Governor
        let builder = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_identifier_extractor(Arc::new(
                crate::matchers::CompositeExtractor::builder()
                    .add_extractor(Box::new(crate::matchers::UserIdExtractor::from_header(
                        "X-User-Id",
                    )))
                    .add_extractor(Box::new(crate::matchers::IpExtractor::builder().build()))
                    .add_extractor(Box::new(crate::matchers::ApiKeyExtractor::from_header(
                        "X-API-Key",
                    )))
                    .build(),
            ));

        // 转发可选的 metrics/tracer（之前被静默丢弃）
        #[cfg(feature = "prometheus")]
        let builder = if let Some(m) = metrics {
            builder.with_metrics(m)
        } else {
            builder
        };

        #[cfg(feature = "telemetry")]
        let builder = if let Some(t) = tracer {
            builder.with_tracer(t)
        } else {
            builder
        };

        builder.build().await
    }

    /// 从配置文件创建 Governor 实例
    ///
    /// # 参数
    /// - `config_path`: 配置文件路径（支持 YAML、TOML、JSON）
    /// - `storage`: 存储后端
    /// - `ban_storage`: 封禁存储后端
    ///
    /// # 示例
    ///
    /// ```rust,no_run
    /// use limiteron::Governor;
    /// use limiteron::storage::{MemoryStorage, MemoryBanStorage};
    /// use std::sync::Arc;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let storage: Arc<dyn limiteron::storage::Storage> = MemoryStorage::create_storage();
    ///     let ban_storage: Arc<dyn limiteron::storage::BanStorage> = MemoryBanStorage::create_ban_storage();
    ///
    ///     let governor = Governor::from_config_file(
    ///         "/path/to/config.yaml",
    ///         storage,
    ///         ban_storage,
    ///     ).await?;
    ///
    ///     Ok(())
    /// }
    /// ```
    pub async fn from_config_file<P: AsRef<std::path::Path>>(
        config_path: P,
        storage: Arc<dyn Storage>,
        ban_storage: Arc<dyn BanStorage>,
    ) -> Result<Self, LimiteronError> {
        // 使用 ConfigLoader 加载配置
        let config = crate::config::ConfigLoader::load_from_file(config_path)?;
        Self::create_with_config(config, storage, ban_storage).await
    }

    /// 从配置文件和环境变量创建 Governor 实例
    ///
    /// 环境变量可以覆盖配置文件中的值。环境变量命名规则：
    /// `LIMITERON_<SECTION>_<FIELD>`
    ///
    /// # 参数
    /// - `config_path`: 配置文件路径（支持 YAML、TOML、JSON）
    /// - `storage`: 存储后端
    /// - `ban_storage`: 封禁存储后端
    ///
    /// # 示例
    ///
    /// ```rust,no_run
    /// use limiteron::Governor;
    /// use limiteron::storage::{MemoryStorage, MemoryBanStorage};
    /// use std::sync::Arc;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let storage: Arc<dyn limiteron::storage::Storage> = MemoryStorage::create_storage();
    ///     let ban_storage: Arc<dyn limiteron::storage::BanStorage> = MemoryBanStorage::create_ban_storage();
    ///
    ///     // 设置环境变量覆盖
    ///     unsafe { std::env::set_var("LIMITERON_GLOBAL_STORAGE", "redis") };
    ///
    ///     let governor = Governor::from_config_with_env(
    ///         "/path/to/config.yaml",
    ///         storage,
    ///         ban_storage,
    ///     ).await?;
    ///
    ///     Ok(())
    /// }
    /// ```
    pub async fn from_config_with_env<P: AsRef<std::path::Path>>(
        config_path: P,
        storage: Arc<dyn Storage>,
        ban_storage: Arc<dyn BanStorage>,
    ) -> Result<Self, LimiteronError> {
        // 使用 ConfigLoader 加载配置，支持环境变量覆盖
        let config = crate::ConfigLoader::load_from_file_with_env(config_path)?;
        Self::create_with_config(config, storage, ban_storage).await
    }

    /// 根据配置创建 Governor 实例（内部辅助方法）
    ///
    /// 统一处理不同 feature 组合下的创建逻辑，避免重复的条件编译代码。
    async fn create_with_config(
        config: FlowControlConfig,
        storage: Arc<dyn Storage>,
        ban_storage: Arc<dyn BanStorage>,
    ) -> Result<Self, LimiteronError> {
        Self::with_storage(
            config,
            storage,
            ban_storage,
            #[cfg(feature = "prometheus")]
            None,
            #[cfg(feature = "telemetry")]
            None,
        )
        .await
    }

    /// 检查请求 - 简化版本使用并行检查器
    ///
    /// 该方法会首先检查 缓存，如果缓存命中则直接返回缓存结果。
    /// 如果缓存未命中，则执行完整的检查流程，并将结果缓存。
    ///
    /// 当启用了 FallbackManager 时，会在存储层故障时自动降级。
    pub async fn check(&self, context: &RequestContext) -> Result<Decision, LimiteronError> {
        #[cfg(feature = "prometheus")]
        let start_time = std::time::Instant::now();

        #[cfg(feature = "telemetry")]
        // 关键路径 span：终态裁决规则与结果以属性记录（见规则级联循环；
        // 全放行时仅记 outcome=allowed，无单一裁决规则）
        let check_span = self.tracer.as_ref().map(|t| t.start_span("governor_check"));

        #[cfg(feature = "telemetry")]
        let result = self.check_inner(context, check_span.as_ref()).await;
        #[cfg(not(feature = "telemetry"))]
        let result = self.check_inner(context).await;

        // 审计接线：决策终态落审计日志（audit-log 特性；脱敏在 AuditLogger
        // 内部完成；失败仅告警不阻塞主流程。历史教训：AuditLogger 完整实现
        // 从未被 check 流程调用——审计能力处于"有实现无接线"状态）
        #[cfg(feature = "audit-log")]
        if let Ok(decision) = &result
            && let Some(logger) = self.audit_logger.read().await.as_ref()
        {
            let (kind, reason) = match decision {
                Decision::Allowed(_) => ("allowed".to_string(), String::new()),
                Decision::Rejected(m) => ("rejected".to_string(), m.reason.clone()),
                Decision::Banned(b) => ("banned".to_string(), b.reason().to_string()),
            };
            let identifier = self
                .identifier_extractor
                .extract(context)
                .map(|i| i.key())
                .unwrap_or_default();
            let logger = logger.clone();
            tokio::spawn(async move {
                logger.log_decision(identifier, kind, reason, None).await;
            });
        }

        // 记录指标（Rule 12：失败必须显性化 — 错误也记录）
        #[cfg(feature = "prometheus")]
        if let Some(ref metrics) = self.metrics {
            let duration = start_time.elapsed();
            let allowed = matches!(&result, Ok(Decision::Allowed(_)));
            metrics.record_check(duration, allowed);
            if matches!(&result, Ok(Decision::Banned(_))) {
                metrics.record_ban();
            }
            if result.is_err() {
                metrics.record_error("check_error");
            }
        }

        #[cfg(feature = "telemetry")]
        if let Some(s) = check_span {
            s.finish();
        }

        result
    }

    /// 内部检查分发（不含 metrics/tracer 包装，由 check() 统一处理）
    async fn check_inner(
        &self,
        context: &RequestContext,
        #[cfg(feature = "telemetry")] check_span: Option<&crate::telemetry::Span>,
    ) -> Result<Decision, LimiteronError> {
        // 如果启用了 FallbackManager，使用降级包装的检查逻辑
        #[cfg(feature = "fallback")]
        if let Some(ref fallback_mgr) = self.fallback_manager {
            #[cfg(feature = "telemetry")]
            let result = self
                .check_with_fallback(context, fallback_mgr, check_span)
                .await;
            #[cfg(not(feature = "telemetry"))]
            let result = self.check_with_fallback(context, fallback_mgr).await;
            return result;
        }

        // 否则直接执行检查
        #[cfg(feature = "telemetry")]
        let result = self.check_internal(context, check_span).await;
        #[cfg(not(feature = "telemetry"))]
        let result = self.check_internal(context).await;
        result
    }

    /// 内部检查逻辑（不包含降级处理）
    async fn check_internal(
        &self,
        context: &RequestContext,
        #[cfg(feature = "telemetry")] check_span: Option<&crate::telemetry::Span>,
    ) -> Result<Decision, LimiteronError> {
        self.stats.increment_total();

        // Extracted identifier
        let identifier = self.identifier_extractor.extract(context).ok_or_else(|| {
            LimiteronError::ConfigError("Failed to extract identifier".to_string())
        })?;
        trace!(
            "Extracted identifier: {}",
            log_fingerprint(&identifier.key())
        );

        // 多租户贯穿：决策键改写为 tenant+key 复合键。
        // 缓存键、封禁精确匹配键、事件键均以限定后的标识符计算，
        // 实现 缓存与租户封禁按租户隔离；未配置 resolver 时原样返回。
        #[cfg(feature = "multi-tenant")]
        let identifier = self.tenant_scoped_identifier(context, identifier);

        // 规则匹配 - 只计算一次，贯穿整个检查流程。
        // 换装原子性：matcher 与 chains 的读锁同时获取、贯穿规则解析段
        //（写侧 apply_config 同序原子持有全部写锁），杜绝「新 matcher +
        // 旧 chains」的窗口内新规则 ID 查不到节点被静默跳过放行。
        let matcher_guard = self.rule_matcher.read().await;
        let chains_guard = self.rule_chains.read().await;
        let matched_rules = {
            #[allow(clippy::disallowed_methods)]
            matcher_guard
                .match_all(context)
                .into_iter()
                .cloned()
                .collect::<Vec<_>>()
        };

        // 并行封禁检查 (仅当 parallel-checker 特性启用时)
        //
        // 必须先于 缓存读取执行：即使缓存中存在条目，被封禁的标识符也
        // 不会被缓存击中而绕过封禁检查。
        #[cfg(feature = "parallel-checker")]
        {
            // 尝试转换为 BanTarget 进行检查
            let ban_target = identifier.to_ban_target();

            if let Some(target) = ban_target {
                // 使用专门的并行封禁检查器。
                // 熔断接入：外部存储调用经 `execute` 门控——连续故障使
                // 熔断打开后，Err(CircuitOpen) 被降级为"跳过封禁检查"并
                // 告警（请求继续走限流/配额裁决，不触碰已故障的存储），
                // 避免对故障存储的持续冲击雪崩。未启用 circuit-breaker
                // 特性时行为不变（错误照常传播）。
                #[cfg(feature = "circuit-breaker")]
                let ban_info = match self
                    .circuit_breaker
                    .execute(|| self.parallel_ban_checker.check_single_target(&target))
                    .await
                {
                    Ok(info) => info,
                    Err(e) => {
                        warn!(
                            "{}",
                            t(
                                "governor-ban-check-circuit-open",
                                &[("error", e.to_string())],
                            )
                        );
                        None
                    }
                };
                #[cfg(not(feature = "circuit-breaker"))]
                let ban_info = self
                    .parallel_ban_checker
                    .check_single_target(&target)
                    .await?;

                if let Some(info) = ban_info {
                    warn!(
                        "{}",
                        t(
                            "governor-request-banned",
                            &[
                                ("user", log_fingerprint(&identifier.key()).to_string(),),
                                ("reason", info.reason().to_string()),
                            ],
                        )
                    );
                    self.stats.increment_banned();
                    return Ok(Decision::Banned(info));
                }
            }
        }

        // 尝试从 缓存获取结果
        //
        // 只缓存"拒绝/封禁"决策（负缓存，fail-closed）：任何一次请求都必须重新
        // 执行限流器与封禁检查来消耗令牌/更新状态；缓存命中只能返回非"允许"的
        // 结果，从而不会绕过限流或封禁语义。缓存键包含该标识符匹配的全部规则。
        if self.is_l1_cache_enabled() && !matched_rules.is_empty() {
            let cache_key = self.build_cache_key_multi(&identifier, &matched_rules);

            if let Ok(Some(cached_decision)) = self.l1_cache.get(&cache_key).await {
                let decision = cached_decision.to_decision();
                if !matches!(decision, Decision::Allowed(_)) {
                    trace!(
                        "L1 cache hit (reject decision): key={}",
                        log_fingerprint(&cache_key)
                    );
                    // 负缓存命中的拒绝未执行规则链，但按缓存内裁决规则计入
                    // per-rule 维度（与链上拒绝同序列）；缓存条目无规则归因时
                    // 退回独立计数，保证与全局计数可对账（requests_total 含
                    // 缓存命中）。规则被热更新移除后，其存量缓存条目在 TTL
                    // 内仍按已删 rule_id 递增 per-rule 序列（Prometheus 惰性
                    // 创建，短暂抬高基数，条目到期后自愈）
                    #[cfg(feature = "prometheus")]
                    if let Some(ref metrics) = self.metrics {
                        if self.per_rule_metrics
                            && let Some(rule_id) = cached_decision.rule_id.as_deref()
                        {
                            metrics.record_rule_outcome(rule_id, decision_outcome(&decision));
                        } else {
                            metrics.record_negative_cache_hit();
                        }
                    }

                    // 缓存裁决同样记录终态规则与结果，观测面不留盲区
                    //（不依赖 metrics 配置——追踪与指标互相独立）；缓存条目
                    // 无裁决规则归因（如旧版本写入的共享缓存条目）时不虚构
                    // rule.id，仅保留结果信号
                    #[cfg(feature = "telemetry")]
                    if let Some(span) = check_span {
                        if let Some(rule_id) = cached_decision.rule_id.as_deref() {
                            span.set_attribute("rule.id", rule_id);
                        }
                        span.set_attribute("rule.outcome", decision_outcome(&decision));
                    }
                    self.update_stats_for_decision(&Result::Ok(decision.clone()));
                    return Ok(decision);
                }
                trace!(
                    "L1 cache hit but allow decision, ignoring and re-running full check: key={}",
                    log_fingerprint(&cache_key)
                );
            }
        }

        // 继续其他检查（使用已计算的 matched_rules）

        if matched_rules.is_empty() {
            // 如果没有匹配的规则，检查默认决策链
            // 目前默认决策链为空，相当于直接允许
            let result = self.decision_chain.read().await.check().await;
            self.update_stats_for_decision(&result);
            return result;
        }

        // 有匹配的规则，按顺序执行（级联）
        // 只要有一个规则拒绝，请求就被拒绝
        // （沿用函数入口取得的 chains 读锁：与 matcher 同一快照）
        let rule_chains = &*chains_guard;

        for rule in &matched_rules {
            if let Some(chain) = rule_chains.get(&rule.id) {
                // 执行决策链
                let result = chain.check().await;

                // 规则维度结果（prometheus 指标与 telemetry span 共用一次判定；
                // "error" 仅链执行失败产生，纯决策不含该形态）
                #[cfg(any(feature = "prometheus", feature = "telemetry"))]
                let outcome: &str = match &result {
                    Ok(decision) => decision_outcome(decision),
                    Err(_) => "error",
                };

                // per-rule/per-limiter 维度指标（默认开启，builder 可关）：
                // 规则结果计数随每次链检查记录；链内限流器拒绝计数按链版本号
                // 增量导出（Prometheus Counter 单调语义）——导出版本号下沉为
                // 链内原子量，版本一致时两读一比零锁跳过，放行稳态与无新增
                // 拒绝的流量不为监控导出付互斥锁与分配成本
                #[cfg(feature = "prometheus")]
                if self.per_rule_metrics
                    && let Some(ref metrics) = self.metrics
                {
                    metrics.record_rule_outcome(&rule.id, outcome);

                    if chain.node_rejections_version() != chain.exported_rejections_version() {
                        self.export_chain_rejections(&chain, &rule.id, metrics);
                    }
                }

                match result {
                    Ok(Decision::Allowed(_)) => {
                        // 当前规则允许，继续检查下一个规则
                        continue;
                    }
                    _ => {
                        // 拒绝、封禁或错误，直接返回
                        self.update_stats_for_decision(&result);

                        // 关键路径 span 属性：终态裁决规则与结果（拒绝/封禁/
                        // 错误场景裁决规则唯一；全放行场景在循环后统一记
                        // outcome，不虚构单一裁决规则）
                        #[cfg(feature = "telemetry")]
                        if let Some(span) = check_span {
                            span.set_attribute("rule.id", &rule.id);
                            span.set_attribute("rule.outcome", outcome);
                        }

                        // 负缓存：仅缓存拒绝/封禁决策（fail-closed），"允许"决策永不入缓存
                        if self.is_l1_cache_enabled()
                            && let Ok(ref decision) = result
                        {
                            let cache_key = self.build_cache_key_multi(&identifier, &matched_rules);
                            let mut cacheable = CacheableDecision::from_decision(decision);
                            // 缓存条目携带裁决规则 ID：命中路径据此归因 per-rule 维度
                            cacheable.rule_id = Some(rule.id.clone());
                            let _ = self.l1_cache.set(cache_key, cacheable).await;
                            trace!("L1 cache updated: decision=rejected");
                        }

                        // 发射事件
                        #[cfg(feature = "event-system")]
                        {
                            if let Ok(ref decision) = result {
                                let decision_str = match decision {
                                    Decision::Banned(_) => "Banned",
                                    Decision::Rejected(_) => "Denied",
                                    _ => "Allowed",
                                };
                                self.emit_rate_limit_event(
                                    &identifier.key(),
                                    &rule.id,
                                    decision_str,
                                )
                                .await;
                            }
                        }

                        return result;
                    }
                }
            }
        }

        // 所有规则都允许
        self.stats.increment_allowed();
        let decision = Decision::allowed_default();

        // 关键路径 span：全放行无单一裁决规则，仅记录结果
        #[cfg(feature = "telemetry")]
        if let Some(span) = check_span {
            span.set_attribute("rule.outcome", "allowed");
        }

        Ok(decision)
    }

    /// 将一条链的节点拒绝增量导出到监控面（监控 feature 门控）
    ///
    /// 拒绝即触发导出，flood 场景这是热路径，临界区编排（锁序固定
    /// `exported` 互斥锁 → 链统计读锁，后者从不反向持有前者）：
    /// 1. 互斥锁内二次复核「已导出版本 ≥ 当前版本」——稳态探测的版本
    ///    只是省锁提示值，并发导出方可能已在探测之后完成同版本或更新
    ///    的合并，复核通过才继续；
    /// 2. 链统计读锁内遍历计数并同点读取版本号，计数与版本严格同序
    ///    观测，迭代零克隆（仅对有增量的节点付分配与标签查找成本）；
    /// 3. 基线 `last` 与导出版本号只在同一临界区内落账——快照与合并
    ///    串行化后，基线只会随真实新增前进，不会被迟到的旧快照拉回
    ///    （曾经锁外快照 + 锁内无条件合并的 check-then-act 交错把基线
    ///    拉回旧值，下轮误判 stale 对已导出区间静默双算）。
    #[cfg(feature = "prometheus")]
    fn export_chain_rejections(&self, chain: &DecisionChain, rule_id: &str, metrics: &Metrics) {
        let mut exported = self.exported_rejections.lock();
        if chain.exported_rejections_version() >= chain.node_rejections_version() {
            return;
        }
        let nodes = match exported.per_rule.get_mut(rule_id) {
            Some(nodes) => nodes,
            None => exported.per_rule.entry(rule_id.to_owned()).or_default(),
        };
        let version = chain.for_each_node_rejection(|node, count| match nodes.get_mut(node) {
            Some(last) => {
                let delta = count.saturating_sub(*last);
                if delta > 0 {
                    metrics.record_limiter_rejections(rule_id, node, delta);
                }
                // 基线一律跟随快照：计数回落（reset 清零）时随之下降，
                // 之后的新增拒绝才能正常进入增量
                *last = count;
            }
            None => {
                nodes.insert(node.to_owned(), count);
                metrics.record_limiter_rejections(rule_id, node, count);
            }
        });
        chain.set_exported_rejections_version(version);
    }

    /// 带降级处理的检查逻辑
    ///
    /// 使用 FallbackManager 包装检查流程，在存储层故障时自动降级。
    #[cfg(feature = "fallback")]
    async fn check_with_fallback(
        &self,
        context: &RequestContext,
        fallback_mgr: &Arc<FallbackManager>,
        #[cfg(feature = "telemetry")] check_span: Option<&crate::telemetry::Span>,
    ) -> Result<Decision, LimiteronError> {
        let context_clone = context.clone();

        fallback_mgr
            .execute_with_fallback(
                crate::fallback::ComponentType::Redis,
                || async {
                    #[cfg(feature = "telemetry")]
                    let result = self.check_internal(&context_clone, check_span).await;
                    #[cfg(not(feature = "telemetry"))]
                    let result = self.check_internal(&context_clone).await;

                    // 降级计数仅认存储类错误（分类见 note_degraded_check）
                    #[cfg(feature = "prometheus")]
                    self.note_degraded_check(&result);
                    result
                },
                || async {
                    // 降级操作：尝试仅使用 缓存
                    self.check_l1_cache_only(&context_clone).await
                },
            )
            .await
    }

    /// 检查结果落监控面：存储类错误计降级（Redis 故障窗口信号）；
    /// 配置/标识符类错误不计——可被未认证请求无门槛刷出，计入会让
    /// degraded_checks_total 被请求洪水刷高、掩盖真实存储故障
    #[cfg(all(feature = "prometheus", feature = "fallback"))]
    fn note_degraded_check(&self, result: &Result<Decision, LimiteronError>) {
        if let Err(LimiteronError::StorageError(_)) = result
            && let Some(ref metrics) = self.metrics
        {
            metrics.record_degraded();
        }
    }

    /// 仅使用 缓存的降级检查
    ///
    /// 当存储层不可用时，尝试仅从 缓存获取决策结果。
    #[cfg(feature = "fallback")]
    async fn check_l1_cache_only(
        &self,
        context: &RequestContext,
    ) -> Result<Decision, LimiteronError> {
        if !self.is_l1_cache_enabled() {
            return Err(LimiteronError::LimitError(t(
                "governor-storage-failure-no-l1",
                &[],
            )));
        }

        // 尝试从 缓存获取结果
        let identifier = self.identifier_extractor.extract(context).ok_or_else(|| {
            LimiteronError::ConfigError("Failed to extract identifier".to_string())
        })?;

        // 多租户贯穿：与常规路径一致的租户限定决策键
        #[cfg(feature = "multi-tenant")]
        let identifier = self.tenant_scoped_identifier(context, identifier);

        let matched_rules = {
            // 降级路径不解析规则链,无 matcher/chains 配对需求;换装窗口内
            // 缓存键随规则集变化 → 缓存 miss → 走孤岛策略(语义安全)
            let matcher = self.rule_matcher.read().await;
            #[allow(clippy::disallowed_methods)]
            matcher
                .match_all(context)
                .into_iter()
                .cloned()
                .collect::<Vec<_>>()
        };

        if matched_rules.is_empty() {
            // 没有规则，允许通过
            return Ok(Decision::allowed_default());
        }

        // 尝试从 缓存读取决策。
        // 历史教训：曾用 build_cache_key（仅首条规则）读,而写路径
        // check_internal 用 build_cache_key_multi（全部规则）——多规则
        // 匹配场景下降级读永远 miss,孤岛策略退化为事实 AllowAll。
        let cache_key = self.build_cache_key_multi(&identifier, &matched_rules);

        // 缓存有效命中则直接返回;损坏条目按 miss 走孤岛裁决
        // （历史教训:旧实现对未知 decision_type 默认返回 Allowed = fail-open）
        let mut valid_cached: Option<Decision> = None;
        if let Ok(Some(cached_decision)) = self.l1_cache.get(&cache_key).await {
            match cached_decision.to_decision_strict() {
                Ok(decision) => {
                    trace!(
                        "island mode - L1 cache hit: key={}",
                        log_fingerprint(&cache_key)
                    );
                    valid_cached = Some(decision);
                }
                Err(e) => {
                    log::warn!(
                        "island mode - corrupted cache entry: {e}; falling back to island strategy"
                    );
                    self.stats.increment_error();
                }
            }
        }

        if let Some(decision) = valid_cached {
            self.update_stats_for_decision(&Result::Ok(decision.clone()));
            return Ok(decision);
        }

        {
            // 缓存未命中（或损坏），根据孤岛模式策略处理。
            // 各分支均补齐请求级统计：降级路径此前不计数，
            // 孤岛/降级期间的 allowed/error 指标全部丢失。
            if self.l1_cache.is_island_mode() {
                if let Some(config) = self.l1_cache.island_config() {
                    match config.fallback_strategy {
                        IslandFallbackStrategy::AllowAll => {
                            log::warn!(
                                target: "governor",
                                "{}",
                                t("governor-island-allow-all", &[])
                            );
                            let decision = Decision::allowed_default();
                            self.update_stats_for_decision(&Ok(decision.clone()));
                            Ok(decision)
                        }
                        IslandFallbackStrategy::RejectAll => {
                            log::warn!(
                                target: "governor",
                                "{}",
                                t("governor-island-reject-all", &[])
                            );
                            self.stats.increment_error();
                            Err(LimiteronError::LimitError(t(
                                "governor-island-reject-storage-failure",
                                &[],
                            )))
                        }
                        IslandFallbackStrategy::LocalDecision => {
                            // 已在上面尝试过 缓存，未命中
                            log::warn!(
                                target: "governor",
                                "{}",
                                t("governor-island-l1-miss-conservative", &[])
                            );
                            let decision = Decision::allowed_default();
                            self.update_stats_for_decision(&Ok(decision.clone()));
                            Ok(decision)
                        }
                        IslandFallbackStrategy::ConservativeQuota {
                            max_requests,
                            window_secs,
                        } => {
                            let max_requests = u64::from(max_requests);
                            // 保守配额真实现：本地固定窗口计数器，
                            // 窗口内至多放行 max_requests 个，超出拒绝。
                            // （历史教训：曾为空壳"直接允许"——配置了
                            // 保守配额等于存储故障期全放行，语义相反。）
                            log::warn!(
                                target: "governor",
                                "{}",
                                t(
                                    "governor-island-conservative-quota",
                                    &[
                                        ("max", max_requests.to_string()),
                                        ("window", window_secs.to_string()),
                                    ],
                                )
                            );
                            let now_secs = std::time::SystemTime::now()
                                .duration_since(std::time::UNIX_EPOCH)
                                .map(|d| d.as_secs())
                                .unwrap_or(0);
                            let mut guard = self.conservative_quota.lock();
                            let counter = guard.get_or_insert_with(|| ConservativeQuotaCounter {
                                window_start_secs: now_secs,
                                count: 0,
                            });
                            let decision =
                                if counter.try_acquire(max_requests, window_secs, now_secs) {
                                    Decision::allowed_default()
                                } else {
                                    Decision::Rejected(RejectionMetadata {
                                        reason: t(
                                            "governor-island-conservative-quota",
                                            &[
                                                ("max", max_requests.to_string()),
                                                ("window", window_secs.to_string()),
                                            ],
                                        ),
                                        retry_after: window_secs.max(1),
                                        limit: max_requests,
                                        reset_at: now_secs + window_secs.max(1),
                                    })
                                };
                            drop(guard);
                            self.update_stats_for_decision(&Ok(decision.clone()));
                            Ok(decision)
                        }
                    }
                } else {
                    // 未配置孤岛模式，使用默认策略
                    let decision = Decision::allowed_default();
                    self.update_stats_for_decision(&Ok(decision.clone()));
                    Ok(decision)
                }
            } else {
                // 不在孤岛模式，返回错误
                self.stats.increment_error();
                Err(LimiteronError::LimitError(t(
                    "governor-storage-failure-cache-miss",
                    &[],
                )))
            }
        }
    }

    /// 构建缓存键（基于全部匹配规则）
    ///
    /// 键包含该标识符在本请求中匹配的**全部**规则 ID，避免同一标识符在
    /// 不同规则集合下复用彼此的缓存条目，产生错误的决策复用。
    /// 常规检查路径与降级路径（`check_l1_cache_only`）统一使用本方法。
    fn build_cache_key_multi(
        &self,
        identifier: &crate::matchers::Identifier,
        rules: &[crate::Rule],
    ) -> String {
        let rule_part = rules
            .iter()
            .map(|r| r.id.as_str())
            .collect::<Vec<_>>()
            .join("|");
        match identifier {
            crate::matchers::Identifier::UserId(user_id) => {
                RateLimitCacheKey::user_rate_limit(user_id, &rule_part)
            }
            crate::matchers::Identifier::Ip(ip) => RateLimitCacheKey::ip_rate_limit(ip, &rule_part),
            crate::matchers::Identifier::ApiKey(api_key) => {
                RateLimitCacheKey::api_key_rate_limit(api_key, &rule_part)
            }
            _ => RateLimitCacheKey::generic(&identifier.key(), &rule_part),
        }
    }

    /// 根据决策结果更新统计信息
    ///
    /// 统一处理不同决策类型的统计更新，避免重复的 match 分支。
    fn update_stats_for_decision(&self, result: &Result<Decision, LimiteronError>) {
        match result {
            Ok(Decision::Allowed(_)) => {
                self.stats.increment_allowed();
            }
            Ok(Decision::Banned(_)) => {
                self.stats.increment_banned();
            }
            Ok(Decision::Rejected(_)) => {
                self.stats.increment_rejected();
            }
            Err(_) => {
                self.stats.increment_error();
            }
        }
    }

    /// 发射限流事件（内部辅助方法）
    #[cfg(feature = "event-system")]
    async fn emit_rate_limit_event(&self, key: &str, rule_id: &str, decision: &str) {
        if let Some(ref emitter) = self.event_emitter {
            let event = crate::events::Event::new(crate::events::EventType::RateLimitTriggered {
                key: key.to_string(),
                rule_id: rule_id.to_string(),
                decision: decision.to_string(),
            });
            if let Err(e) = emitter.emit(event).await {
                error!(
                    "{}",
                    t(
                        "governor-emit-rate-limit-event-failed",
                        &[("error", e.to_string())]
                    )
                );
            }
        }
    }

    /// 并行资源检查 - 保持原有接口兼容性
    #[cfg(feature = "parallel-checker")]
    pub async fn check_resource_parallel(
        &self,
        resource: &str,
    ) -> Result<Decision, LimiteronError> {
        // 使用专门的并行封禁检查器
        let ban_info = self
            .parallel_ban_checker
            .check_user_banned(resource)
            .await?;

        match ban_info {
            Some(info) => {
                warn!(
                    "{}",
                    t(
                        "governor-resource-banned",
                        &[
                            ("resource", resource.to_string()),
                            ("reason", info.reason().to_string()),
                        ],
                    )
                );
                Ok(Decision::Banned(info))
            }
            None => Ok(Decision::allowed_default()),
        }
    }

    /// 并行资源检查 - 未启用 parallel-checker 时的存根实现
    #[cfg(not(feature = "parallel-checker"))]
    pub async fn check_resource_parallel(
        &self,
        _resource: &str,
    ) -> Result<Decision, LimiteronError> {
        #[cfg(feature = "ban-manager")]
        {
            let target = BanTarget::UserId(_resource.to_string());
            let ban_record = self.ban_manager.is_banned(&target).await?;

            if let Some(record) = ban_record {
                return Ok(Decision::Banned(BanInfo::new(
                    record.reason,
                    record.expires_at,
                    record.ban_times,
                )));
            }

            Ok(Decision::allowed_default())
        }

        #[cfg(not(feature = "ban-manager"))]
        {
            Err(LimiteronError::ConfigError(t(
                "governor-resource-ban-check-unavailable",
                &[],
            )))
        }
    }

    /// 手动Ban user
    #[cfg(feature = "ban-manager")]
    pub async fn ban_identifier(
        &self,
        identifier: &Identifier,
        reason: &str,
        source: Option<BanSource>,
    ) -> Result<(), LimiteronError> {
        debug!(
            "Ban user: {} reason: {}",
            log_fingerprint(&identifier.key()),
            reason
        );

        let ban_target = identifier.to_ban_target();

        if let Some(target) = ban_target {
            let ban_source = source.unwrap_or(BanSource::Manual {
                operator: "unknown".to_string(),
            });

            self.ban_manager
                .create_ban(
                    target,
                    reason.to_string(),
                    ban_source,
                    serde_json::json!({}),
                    None,
                )
                .await?;
            info!(
                "{}",
                t(
                    "governor-user-banned",
                    &[(
                        "user",
                        crate::logging::redact_user_id(Some(identifier.key().as_ref())).to_string(),
                    )],
                )
            );
        } else {
            return Err(LimiteronError::ValidationError(
                "Unsupported identifier type".to_string(),
            ));
        }

        Ok(())
    }

    /// 取消用户封禁
    #[cfg(feature = "ban-manager")]
    pub async fn unban_identifier(&self, identifier: &Identifier) -> Result<(), LimiteronError> {
        debug!("Unban user: {}", log_fingerprint(&identifier.key()));

        let ban_target = identifier.to_ban_target();

        if let Some(target) = ban_target {
            let unbanned = self
                .ban_manager
                .delete_ban(&target, "unknown".to_string())
                .await?;

            if unbanned {
                info!(
                    "{}",
                    t(
                        "governor-user-unbanned",
                        &[(
                            "user",
                            crate::logging::redact_user_id(Some(identifier.key().as_ref()))
                                .to_string(),
                        )],
                    )
                );
            }
            Ok(())
        } else {
            Err(LimiteronError::ValidationError(
                "Unsupported identifier type".to_string(),
            ))
        }
    }

    /// 获取配置历史
    pub async fn get_config_history(&self) -> Vec<ConfigChangeRecord> {
        self.config_history.read().await.get_records().to_vec()
    }

    // ========================================================================
    // 多租户贯穿（feature `multi-tenant`）
    // ========================================================================

    /// 从请求上下文解析租户命名空间
    ///
    /// 未配置 resolver 或解析失败时返回 `None`（决策键回退到无租户前缀的
    /// 现状行为）。
    #[cfg(feature = "multi-tenant")]
    pub fn resolve_tenant(&self, context: &RequestContext) -> Option<crate::tenant::Namespace> {
        self.tenant_resolver
            .as_ref()
            .and_then(|resolver| resolver.resolve(context))
    }

    /// 计算决策键：tenant + key 复合（公开辅助 API）
    ///
    /// - 配置 resolver 且解析到非默认租户 → `tenant:{id}:env:{env}:{identifier.key()}`
    /// - 否则（未配置 resolver / 解析失败 / 默认命名空间）→ `identifier.key()`
    ///   （与现状逐位一致）
    ///
    /// 与请求热路径的键改写（`tenant_scoped_identifier`，私有）同口径；
    /// 内部负缓存、事件与封禁键由热路径自身的改写结果贯穿，不经本方法。
    #[cfg(feature = "multi-tenant")]
    pub fn decision_key(&self, context: &RequestContext, identifier: &Identifier) -> String {
        match self.resolve_tenant(context) {
            Some(ns) if !ns.is_default() => ns.qualify_key(&identifier.key()),
            _ => identifier.key(),
        }
    }

    /// 将标识符改写为租户限定标识符（内部）
    ///
    /// 保持标识符类型不变，仅对值加命名空间前缀——下游所有按值键控的
    /// 消费点（缓存键、封禁精确匹配、事件键）自动获得租户隔离。
    /// 未配置 resolver / 解析失败 / 租户为默认 global 时原样返回。
    #[cfg(feature = "multi-tenant")]
    fn tenant_scoped_identifier(
        &self,
        context: &RequestContext,
        identifier: Identifier,
    ) -> Identifier {
        let Some(ns) = self.resolve_tenant(context) else {
            return identifier;
        };
        // 默认命名空间（global/development）不加前缀，保持与无租户部署逐位一致
        if ns.is_default() {
            return identifier;
        }
        let qualified = ns.qualify_key(identifier.as_str());
        match identifier {
            Identifier::UserId(_) => Identifier::UserId(qualified),
            Identifier::Ip(_) => Identifier::Ip(qualified),
            Identifier::Mac(_) => Identifier::Mac(qualified),
            Identifier::ApiKey(_) => Identifier::ApiKey(qualified),
            Identifier::DeviceId(_) => Identifier::DeviceId(qualified),
        }
    }

    /// 按命名空间封禁标识符（低级写入口）
    ///
    /// 非默认命名空间：以 tenant 限定的 BanTarget 写入封禁存储，仅影响该
    /// 租户内的同标识符请求；其他租户与无租户请求不受影响。
    /// 默认命名空间（global/development）：写无前缀键（全局封禁），与决策键
    /// 改写和读侧编码一致——写限定键会使记录在任何查询路径都不可见
    /// （读侧跳过默认命名空间的限定键）。
    ///
    /// 强制点说明：默认命名空间写入的全局封禁由请求热路径对默认命名空间
    /// （或无 resolver）流量强制；对非默认命名空间流量，仅
    /// [`Governor::is_identifier_banned`] 的回退查询可见（热路径对非默认
    /// 命名空间只查租户限定键，无无前缀回退）。
    ///
    /// 本方法直接写封禁存储（`ban_times = 1`、默认 3600s、`is_manual = true`），
    /// 不经 [`crate::ban::BanManager::create_ban`]：不做授权检查、不累加历史
    /// 封禁次数、不用退避时长策略，也不做面向外部输入的格式校验（tenant 前缀
    /// 由 [`crate::tenant::Namespace`] 的转义规则保证无歧义）。需要上述语义时
    /// 用 [`Governor::ban_identifier`]。
    #[cfg(all(feature = "multi-tenant", feature = "ban-manager"))]
    pub async fn ban_identifier_for_namespace(
        &self,
        namespace: &crate::tenant::Namespace,
        identifier: &Identifier,
        reason: &str,
        duration: Option<std::time::Duration>,
    ) -> Result<(), LimiteronError> {
        let Some(target) = identifier.to_ban_target() else {
            return Err(LimiteronError::ValidationError(
                "Unsupported identifier type".to_string(),
            ));
        };
        // 默认命名空间写无前缀键；非默认以租户前缀限定（保持 BanTarget
        // 变体类型不变）。`qualify_ban_target` 返回 None 的唯一成因是
        // Geo/Cidr 刻意全局生效（见 storage::ban_target_value），
        // 此时保持原 target。
        let target = if namespace.is_default() {
            target
        } else {
            crate::storage::qualify_ban_target(&target, namespace).unwrap_or(target)
        };
        let now = chrono::Utc::now();
        let duration = duration.unwrap_or(std::time::Duration::from_secs(3600));
        let record = crate::storage::BanRecord {
            target,
            ban_times: 1,
            duration,
            banned_at: now,
            expires_at: now
                + chrono::Duration::from_std(duration).unwrap_or(chrono::Duration::hours(1)),
            is_manual: true,
            reason: reason.to_string(),
        };
        self.ban_storage
            .save(&record)
            .await
            .map_err(LimiteronError::StorageError)?;
        let message_key = if namespace.is_default() {
            "governor-global-ban-applied"
        } else {
            "governor-tenant-ban-applied"
        };
        info!(
            "{}",
            t(
                message_key,
                &[
                    ("namespace", namespace.to_string()),
                    ("key", log_fingerprint(&identifier.key()).to_string(),),
                ],
            )
        );
        Ok(())
    }

    /// 租户感知的封禁检查
    ///
    /// 先查租户限定键（tenant 隔离封禁），未命中再查无前缀键（全局封禁，
    /// 如自动封禁/Geo 封禁，保持既有语义）。未配置 resolver / 解析到默认
    /// 命名空间时仅查无前缀键——默认命名空间不加前缀，写侧
    /// [`Governor::ban_identifier_for_namespace`] 同口径写无前缀键。
    #[cfg(all(feature = "multi-tenant", feature = "ban-manager"))]
    pub async fn is_identifier_banned(
        &self,
        context: &RequestContext,
        identifier: &Identifier,
    ) -> Result<Option<crate::storage::BanRecord>, LimiteronError> {
        let Some(target) = identifier.to_ban_target() else {
            return Ok(None);
        };
        if let Some(ns) = self.resolve_tenant(context)
            && !ns.is_default()
            && let Some(scoped) = crate::storage::qualify_ban_target(&target, &ns)
            && let Some(record) = self.ban_storage.is_banned(&scoped).await?
        {
            return Ok(Some(record));
        }
        Ok(self.ban_storage.is_banned(&target).await?)
    }

    /// 停止配置监视器
    ///
    /// 取消 `register_config_watcher_token` 注册的全部热重载令牌（令牌由
    /// `watch_and_reload` 系列 spawn 的 watcher 任务在 `select!` 分支消费，
    /// 收到取消即退出）。未注册令牌时为幂等 no-op。
    pub async fn stop_config_watcher(&self) -> Result<(), LimiteronError> {
        for token in self.config_watcher_tokens.lock().drain(..) {
            token.cancel();
        }
        info!("{}", t("governor-config-watcher-stopped", &[]));

        Ok(())
    }

    /// 手动配置检查
    pub async fn manual_config_check(&self) -> Result<bool, LimiteronError> {
        info!("{}", t("governor-manual-config-check", &[]));

        let _config = self.config.read().await;

        // 执行各种配置验证
        // 具体验证逻辑取决于具体的验证需求

        Ok(true)
    }

    /// 获取统计信息
    pub async fn stats(&self) -> GovernorStats {
        self.stats.snapshot().into()
    }

    /// 获取决策链统计
    pub async fn decision_chain_stats(&self) -> crate::decision_chain::ChainStats {
        self.decision_chain.read().await.stats().await
    }

    /// 获取规则匹配器统计
    pub async fn rule_matcher_stats(&self) -> crate::matchers::MatcherStats {
        self.rule_matcher.read().await.stats()
    }

    /// 重置统计信息
    pub async fn reset_stats(&self) {
        info!("{}", t("governor-stats-reset", &[]));

        self.decision_chain.write().await.reset_stats().await;
        self.rule_matcher.write().await.reset_stats();
        self.stats.reset();
        self.l1_cache.reset_stats();
    }

    /// 获取配置的原子换句柄。
    ///
    /// 返回 `Arc<RwLock<FlowControlConfig>>`，允许外部（如 confers 热重载）
    /// 原子地替换当前运行配置。替换后所有后续 `check()` 调用均使用新配置。
    pub fn config_handle(&self) -> Arc<RwLock<FlowControlConfig>> {
        self.config.clone()
    }

    /// 规则热更新：校验并原子换入新配置。
    ///
    /// 与 confers reload 的原子换装模式（`Arc<RwLock<FlowControlConfig>>`）
    /// 一致，并在同一次换装中同步重建规则匹配器与决策链，保证热更新
    /// **真实生效**（仅换 config 句柄不会改变构造期构建的 rule_matcher）。
    ///
    /// 流程：
    /// 1. 校验新配置（失败 → Err，旧配置原样保留，即 rollback 语义）；
    /// 2. 预构建新 `RuleMatcher` 与 rule→decision-chain 映射（构建失败同样拒绝）；
    /// 3. 原子换入 config + rule_matcher + rule_chains；
    /// 4. 清空 决策缓存（旧配置下的缓存决策不再可信）；
    /// 5. 记录 `ConfigChangeRecord`（`ChangeSource::Api`）。
    ///
    /// 供 Admin API `POST /api/v1/config` 与 confers watch 共同使用。
    pub async fn apply_config(
        &self,
        new_config: FlowControlConfig,
    ) -> Result<ConfigApplyReport, LimiteronError> {
        // 1. 校验（规则非空 / ID 唯一 / 匹配器与限流器合法等）
        new_config.validate().map_err(LimiteronError::ConfigError)?;

        // 2. 预构建（任何失败都在换装前发生，旧配置不受影响）。
        //    注：热更新路径不重建自定义匹配器注册表（registry 属构造期
        //    builder 注入；含 Custom 匹配器的配置请走进程重启或经注册表
        //    预注册后使用同名字段）。
        let rules = RuleBuilder::build_rules(&new_config)?;
        let new_matcher = RuleMatcher::with_dependencies(rules);
        let new_chains = RuleBuilder::build_rule_chains_with_quota_storage(
            &new_config,
            self.quota_storage.as_ref(),
            Arc::clone(&self.clock),
        )?;
        let new_hash = new_config.compute_hash();
        let new_version = new_config.version.clone();
        let rule_count = new_config.rules.len();

        // 3. 原子换装（写侧按 matcher→chains→config 固定序在**同一临界区**
        //    持有全部写锁直至换装完成；读侧 check_internal 同时持有 matcher+
        //    chains 读锁贯穿规则解析——历史教训：曾逐语句获取/释放三把写锁，
        //    窗口期并发请求读到「新 matcher+旧 chains」，新规则 ID 查不到
        //    节点被静默跳过放行，与"不存在撕裂读"的注释不符）
        let (old_version, old_hash) = {
            let mut matcher_guard = self.rule_matcher.write().await;
            let mut chains_guard = self.rule_chains.write().await;
            let mut cfg_guard = self.config.write().await;
            *matcher_guard = new_matcher;
            *chains_guard = new_chains;
            let old = std::mem::replace(&mut *cfg_guard, new_config);
            // 新链拒绝计数从零重计，增量导出快照同步清零——否则新链拒绝
            // 在累计值追平旧链快照前不会进入指标（漏报）
            #[cfg(feature = "prometheus")]
            self.exported_rejections.lock().clear();
            let old_hash = old.compute_hash();
            let old_version = old.version;
            (old_version, old_hash)
        };

        // 4. 决策缓存失效（best-effort：清空失败不阻断换装）
        if let Err(e) = self.l1_cache.clear().await {
            log::warn!("apply_config: L1 cache clear failed: {e}");
        }

        // 5. 记录配置变更历史
        self.config_history
            .write()
            .await
            .add_record(ConfigChangeRecord {
                timestamp: chrono::Utc::now(),
                old_version: Some(old_version.clone()),
                new_version: new_version.clone(),
                old_hash: Some(old_hash),
                new_hash: new_hash.clone(),
                source: crate::config::ChangeSource::Api,
                changes: vec![format!("rules={rule_count} (hot-reload)")],
            });

        info!(
            "apply_config: hot-reloaded config {} → {} (rules={})",
            old_version, new_version, rule_count
        );

        Ok(ConfigApplyReport {
            applied: true,
            old_version,
            new_version,
            config_hash: new_hash,
            rule_count,
            applied_at: chrono::Utc::now(),
        })
    }

    // ==================== 缓存相关方法 ====================

    /// 启用 缓存
    pub fn enable_l1_cache(&self) {
        self.l1_cache_enabled
            .store(true, std::sync::atomic::Ordering::Release);
        info!("{}", t("governor-l1-cache-enabled", &[]));
    }

    /// 禁用 缓存
    pub fn disable_l1_cache(&self) {
        self.l1_cache_enabled
            .store(false, std::sync::atomic::Ordering::Release);
        info!("{}", t("governor-l1-cache-disabled", &[]));
    }

    /// 检查 缓存是否启用
    pub fn is_l1_cache_enabled(&self) -> bool {
        self.l1_cache_enabled
            .load(std::sync::atomic::Ordering::Acquire)
    }

    /// 清空 缓存
    pub async fn clear_l1_cache(&self) {
        let _ = self.l1_cache.clear().await;
        info!("{}", t("governor-l1-cache-cleared", &[]));
    }

    /// 清理 缓存中的过期条目
    pub async fn evict_expired_l1_cache(&self) -> usize {
        let evicted = self.l1_cache.evict_expired().await.unwrap_or(0);
        if evicted > 0 {
            debug!("L1 cache evicted {} expired entries", evicted);
        }
        evicted
    }

    /// 使指定标识符的缓存失效
    ///
    /// # 参数
    /// - `identifier`: 标识符（如用户 ID、IP 地址等）
    pub async fn invalidate_l1_cache(&self, identifier: &str) {
        // 使所有与该标识符相关的缓存失效
        let _ = self
            .l1_cache
            .invalidate_by_prefix(&format!("rl:user:{}:", identifier))
            .await;
        let _ = self
            .l1_cache
            .invalidate_by_prefix(&format!("rl:ip:{}:", identifier))
            .await;
        let _ = self
            .l1_cache
            .invalidate_by_prefix(&format!("rl:apikey:{}:", identifier))
            .await;
        let _ = self
            .l1_cache
            .invalidate_by_prefix(&format!("rl:generic:{}:", identifier))
            .await;
        let _ = self
            .l1_cache
            .invalidate(&RateLimitCacheKey::ban_check(identifier))
            .await;
        debug!("invalidated L1 cache for identifier {}", identifier);
    }

    /// 使指定规则的缓存失效
    ///
    /// # 参数
    /// - `rule_id`: 规则 ID
    pub async fn invalidate_rule_cache(&self, rule_id: &str) {
        // 使用包含匹配移除所有与该规则相关的条目
        let _ = self
            .l1_cache
            .invalidate_containing(&format!(":{}", rule_id))
            .await;
        debug!("invalidated L1 cache for rule {}", rule_id);
    }

    /// 获取 缓存大小
    pub async fn l1_cache_size(&self) -> usize {
        self.l1_cache.len().await.unwrap_or(0)
    }

    /// 设置审计日志记录器
    #[cfg(feature = "audit-log")]
    pub async fn set_audit_logger(&self, audit_logger: Arc<AuditLogger>) {
        let mut logger = self.audit_logger.write().await;
        *logger = Some(audit_logger);

        info!("{}", t("governor-audit-logger-set", &[]));
    }

    /// 获取审计日志记录器
    #[cfg(feature = "audit-log")]
    pub async fn audit_logger(&self) -> Option<Arc<AuditLogger>> {
        let guard = self.audit_logger.read().await;
        guard.as_ref().cloned()
    }

    // ========================================================================
    // 运行时自省
    // ========================================================================

    /// 获取运行时自省快照
    ///
    /// 一次性聚合「规则 → 决策链 → 统计 → 缓存 → 健康」的结构化状态，
    /// 供 Admin API `GET /api/v1/introspect`（JSON）与排障工具消费。
    pub async fn introspect(&self) -> IntrospectionSnapshot {
        let config = self.config.read().await;
        let config_version = config.version.clone();
        let rules = config
            .rules
            .iter()
            .map(|r| RuleIntrospection {
                id: r.id.clone(),
                name: r.name.clone(),
                priority: r.priority,
                matcher_count: r.matchers.len(),
                limiters: r.limiters.iter().map(limiter_kind).collect(),
                ban_on_exceed: r.action.ban.is_some(),
            })
            .collect();
        drop(config);

        let rule_chains = self.rule_chains.read().await;
        let mut chains: Vec<ChainIntrospection> = Vec::with_capacity(rule_chains.len());
        for entry in rule_chains.iter() {
            let chain = entry.value();
            let stats = chain.stats().await;
            chains.push(ChainIntrospection {
                rule_id: entry.key().clone(),
                node_count: chain.node_count(),
                enabled_node_count: chain.enabled_node_count(),
                total_checks: stats.total_checks,
                allowed_count: stats.allowed_count,
                rejected_count: stats.rejected_count,
                error_count: stats.error_count,
                node_rejections: stats.node_rejections,
            });
        }
        drop(rule_chains);
        chains.sort_by(|a, b| a.rule_id.cmp(&b.rule_id));

        IntrospectionSnapshot {
            config_version,
            rules,
            chains,
            stats: self.stats().await,
            l1_cache: L1Introspection {
                enabled: self.is_l1_cache_enabled(),
                size: self.l1_cache_size().await,
            },
            health: HealthIntrospection::from(&self.health_status().await),
            #[cfg(feature = "multi-tenant")]
            multi_tenant_enabled: self.tenant_resolver.is_some(),
            is_shutdown: self.is_shutdown(),
        }
    }

    /// 健康检查
    ///
    /// 执行真实的存储 ping、缓存可用性检查、后台任务存活检查。
    /// 返回 `Ok(())` 表示所有关键组件健康；返回 `Err` 包含具体故障信息。
    pub async fn health_check(&self) -> Result<(), LimiteronError> {
        info!("{}", t("governor-health-check", &[]));

        let status = self.health_status().await;

        if status.healthy() {
            Ok(())
        } else {
            let mut issues = Vec::new();
            if !status.storage_healthy {
                issues.push("storage".to_string());
            }
            if !status.ban_storage_healthy {
                issues.push("ban_storage".to_string());
            }
            if !status.cache_healthy {
                issues.push("cache".to_string());
            }
            if !status.background_tasks_alive {
                issues.push("background_tasks".to_string());
            }
            Err(LimiteronError::StorageError(
                crate::error::StorageError::ConnectionError(format!(
                    "Components unhealthy: {}",
                    issues.join(", ")
                )),
            ))
        }
    }

    /// 获取详细的健康状态
    ///
    /// 返回各组件的健康状况，用于监控和诊断。
    pub async fn health_status(&self) -> HealthStatus {
        // 检查存储后端：执行一次 probe get 操作
        let storage_healthy = self.storage.get("__health_probe__").await.is_ok();

        // 检查封禁存储：执行一次 list_bans(0, 1) 操作
        let ban_storage_healthy = self.ban_storage.list_bans(false, 0, 1).await.is_ok();

        // 检查配置锁是否可读（如果 RwLock 中毒则 panic，视为健康）
        let _config_guard = self.config.read().await;

        // 缓存为内存实现，不会故障；只要 Governor 存在即视为健康
        let cache_healthy = true;

        // 检查 ban_manager（feature-gated）
        #[cfg(feature = "ban-manager")]
        {
            let _ = self.ban_manager.get_config().await;
        }

        // 检查 circuit_breaker（feature-gated）
        #[cfg(feature = "circuit-breaker")]
        {
            let _ = self.circuit_breaker.get_state().await;
        }

        // 检查 audit_logger（feature-gated）
        #[cfg(feature = "audit-log")]
        {
            let _ = self.audit_logger().await;
        }

        HealthStatus {
            storage_healthy,
            ban_storage_healthy,
            cache_healthy,
            background_tasks_alive: !self.is_shutdown.load(std::sync::atomic::Ordering::SeqCst),
        }
    }

    /// 优雅关闭 Governor
    ///
    /// 关闭编排（顺序执行，任一阶段失败仅记录 error 日志并继续后续阶段，
    /// 最终标记已关闭）：
    /// 1. 停止配置热重载 watcher（取消 `register_config_watcher_token` 注册的令牌）；
    /// 2. 取消 shutdown 令牌，通知所有订阅的后台任务退出；
    /// 3. 统计快照落盘（`with_shutdown_snapshot_dir` 显式启用时；失败仅记录
    ///    error 级日志，不阻塞停机）；
    /// 4. 停止内部后台任务：BanManager 自动解封任务（`ban-manager`）、
    ///    审计日志写入任务尽力优雅等待（`audit-log`，无法取得所有权时由其
    ///    `Drop` 兜底 abort）；
    /// 5. 清空 L1 缓存，释放缓存引用。
    ///
    /// 存储连接（dbnexus 连接池 / Redis 连接等）由各存储实现的 `Arc` 引用计数
    /// 管理：Governor 无法独占关闭共享连接，随最后一个持有者释放由底层驱动
    /// 关闭；shutdown 通过停止后台任务确保不再发起新的存储访问。
    ///
    /// 幂等：多次调用返回相同的 Ok 结果。实例被 Drop 时由 `Drop` 实现执行
    /// 同步兜底（无条件取消令牌），不重复完整编排。
    ///
    /// # 返回
    ///
    /// * `Ok(())` - 已成功关闭（或之前已关闭）
    ///
    /// # 示例
    ///
    /// ```rust,no_run
    /// use limiteron::Governor;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let governor = Governor::new().await?;
    ///     // ... 使用 governor 处理请求 ...
    ///     governor.shutdown().await?; // 优雅关闭
    ///     Ok(())
    /// }
    /// ```
    pub async fn shutdown(&self) -> Result<(), LimiteronError> {
        // 幂等性检查：使用 compare_exchange 确保只执行一次关闭逻辑
        let already_shutdown = self
            .is_shutdown
            .compare_exchange(
                false,
                true,
                std::sync::atomic::Ordering::SeqCst,
                std::sync::atomic::Ordering::SeqCst,
            )
            .is_err();

        if already_shutdown {
            info!("{}", t("governor-already-shutdown", &[]));
            return Ok(());
        }

        info!("{}", t("governor-shutdown-started", &[]));

        // 阶段 1：停止配置热重载 watcher（取消注册令牌，watcher 随即退出）；
        // 失败仅记录并继续——is_shutdown 已置位，中断编排会令牌悬空
        if let Err(e) = self.stop_config_watcher().await {
            error!("failed to stop config watcher during shutdown: {}", e);
        }

        // 阶段 2：取消所有订阅 shutdown 令牌的后台任务
        self.shutdown_token.cancel();

        // 阶段 3：状态快照落盘（趁组件可用时采集；失败不阻塞停机）
        self.persist_shutdown_snapshot();

        // 阶段 4：停止持有 JoinHandle 的内部后台任务
        #[cfg(feature = "ban-manager")]
        self.ban_manager.stop_auto_unban_task().await;
        #[cfg(feature = "audit-log")]
        self.detach_audit_logger().await;

        // 阶段 5：清空 缓存
        self.clear_l1_cache().await;

        info!("{}", t("governor-shutdown-complete", &[]));
        Ok(())
    }

    /// 将统计快照原子写入配置的落盘目录（`<dir>/governor-shutdown-snapshot.json`）
    ///
    /// 先以 `create_new`（O_EXCL）创建同目录临时文件再 rename，避免读取方
    /// 观察到半写状态；O_EXCL 不跟随已存在的路径（含预置符号链接），临时文件
    /// 冲突时退避到带 pid 后缀的名字重试，权限收紧为 0600（Unix）。失败记录
    /// error 日志但不阻塞停机（停机的首要目标是停止后台活动）。
    ///
    /// 刻意使用同步 I/O：shutdown 是一次性终态路径，`spawn_blocking` 反而引入
    /// runtime 上下文依赖（runtime 拆除期不可用），快照为 KB 级小写入。
    fn persist_shutdown_snapshot(&self) {
        let Some(dir) = self.shutdown_snapshot_dir.as_ref() else {
            return;
        };
        let snapshot = serde_json::json!({
            "captured_at": chrono::Utc::now().to_rfc3339(),
            "stats": GovernorStats::from(self.stats.snapshot()),
        });
        let payload = match serde_json::to_vec_pretty(&snapshot) {
            Ok(payload) => payload,
            Err(e) => {
                error!("failed to serialize shutdown snapshot: {}", e);
                return;
            }
        };
        if let Err(e) = std::fs::create_dir_all(dir) {
            error!("failed to create snapshot dir {}: {}", dir.display(), e);
            return;
        }
        let target = dir.join("governor-shutdown-snapshot.json");
        let mut tmp = dir.join("governor-shutdown-snapshot.json.tmp");
        match Self::create_private_tmp(&tmp, &payload) {
            Ok(()) => {}
            Err(first_err) => {
                // 名字被占用（残留/预置链接等）：退避到带 pid 后缀的独占名字
                tmp = dir.join(format!(
                    "governor-shutdown-snapshot.json.tmp.{}",
                    std::process::id()
                ));
                if let Err(e) = Self::create_private_tmp(&tmp, &payload) {
                    error!(
                        "failed to write shutdown snapshot to {}: first={} retry={}",
                        tmp.display(),
                        first_err,
                        e
                    );
                    return;
                }
            }
        }
        if let Err(e) = std::fs::rename(&tmp, &target) {
            error!(
                "failed to finalize shutdown snapshot at {}: {}",
                target.display(),
                e
            );
        }
    }

    /// 以 `create_new`（O_EXCL）独占创建临时文件并写入，Unix 下权限收紧 0600
    ///
    /// O_EXCL 语义：路径已存在（包括指向任意目标的符号链接）即失败，不跟随
    /// 链接写入，杜绝共享目录下的符号链接覆写攻击面。
    fn create_private_tmp(tmp: &std::path::Path, payload: &[u8]) -> std::io::Result<()> {
        use std::io::Write;
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(tmp)?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
        }
        file.write_all(payload)?;
        file.sync_all().ok();
        Ok(())
    }

    /// 摘除审计日志器：取回所有权后尽力优雅等待写入任务排空
    ///
    /// `Arc::try_unwrap` 成功（无 in-flight clone）时走 `AuditLogger::shutdown`
    /// 的 5s 超时等待；失败时丢弃本侧引用，由 `AuditLogger::drop` 的 abort 兜底。
    #[cfg(feature = "audit-log")]
    async fn detach_audit_logger(&self) {
        let logger = self.audit_logger.write().await.take();
        if let Some(logger) = logger.and_then(|a| Arc::try_unwrap(a).ok()) {
            logger.shutdown().await;
        }
    }

    /// 获取关闭令牌的引用（用于后台任务订阅取消信号）
    ///
    /// 后台任务可通过 `token.cancelled()` 等待取消信号。
    pub fn shutdown_token(&self) -> &tokio_util::sync::CancellationToken {
        &self.shutdown_token
    }

    /// 注册配置热重载 watcher 的取消令牌
    ///
    /// 集成方将 `watch_and_reload` / `watch_and_reload_with_applier` 返回的
    /// 令牌注册到 Governor，`shutdown`（经 `stop_config_watcher`）统一取消，
    /// 避免热重载任务在停机后仍存活。支持多个 watcher（如同时监听多个配置
    /// 文件）：多次注册的令牌全部保留，shutdown/Drop 全部取消。
    pub fn register_config_watcher_token(&self, token: tokio_util::sync::CancellationToken) {
        self.config_watcher_tokens.lock().push(token);
    }

    /// 检查 Governor 是否已关闭
    pub fn is_shutdown(&self) -> bool {
        self.is_shutdown.load(std::sync::atomic::Ordering::SeqCst)
    }
}

impl Drop for Governor {
    /// 同步兜底关闭：无条件取消 shutdown 令牌与已注册的配置 watcher 令牌，
    /// 防止后台任务在实例销毁后仍存活。
    ///
    /// 刻意不检查 `is_shutdown` 早退：显式 shutdown 在置位标志与取消令牌之间
    /// 存在 panic 窗口，此时 Drop 必须补上取消；`CancellationToken::cancel`
    /// 幂等，重复取消为 no-op。持有 JoinHandle 的后台任务（auto-unban/审计
    /// 写入）与存储连接池由各组件自身的 `Drop` 在最后一个 `Arc` 引用释放时
    /// 兜底，此处不重复处理。
    fn drop(&mut self) {
        self.is_shutdown
            .store(true, std::sync::atomic::Ordering::SeqCst);
        self.shutdown_token.cancel();
        for token in self.config_watcher_tokens.lock().drain(..) {
            token.cancel();
        }
    }
}

/// Governor 健康状态
///
/// 描述各关键组件的健康状况，用于监控和诊断。
#[derive(Debug, Clone)]
pub struct HealthStatus {
    /// 主存储后端是否健康
    pub storage_healthy: bool,
    /// 封禁存储后端是否健康
    pub ban_storage_healthy: bool,
    /// 缓存是否健康
    pub cache_healthy: bool,
    /// 后台任务是否存活
    pub background_tasks_alive: bool,
}

impl HealthStatus {
    /// 所有关键组件是否健康
    pub fn healthy(&self) -> bool {
        self.storage_healthy
            && self.ban_storage_healthy
            && self.cache_healthy
            && self.background_tasks_alive
    }
}

// ============================================================================
// 自省快照类型（serde Serialize 供 Admin API JSON 输出）
// ============================================================================

/// 规则自省摘要
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RuleIntrospection {
    /// 规则 ID
    pub id: String,
    /// 规则名
    pub name: String,
    /// 优先级
    pub priority: u16,
    /// 匹配器数量
    pub matcher_count: usize,
    /// 限流器类型列表
    pub limiters: Vec<String>,
    /// 超限是否联动封禁
    pub ban_on_exceed: bool,
}

/// 决策链自省摘要（含运行统计）
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ChainIntrospection {
    /// 规则 ID
    pub rule_id: String,
    /// 节点数
    pub node_count: usize,
    /// 启用节点数
    pub enabled_node_count: usize,
    /// 总检查次数
    pub total_checks: u64,
    /// 允许次数
    pub allowed_count: u64,
    /// 拒绝次数
    pub rejected_count: u64,
    /// 错误次数
    pub error_count: u64,
    /// 各节点拒绝计数
    pub node_rejections: Vec<(String, u64)>,
}

/// 缓存自省摘要
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub struct L1Introspection {
    /// 是否启用
    pub enabled: bool,
    /// 当前条目数
    pub size: usize,
}

/// 健康自省摘要
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub struct HealthIntrospection {
    pub storage_healthy: bool,
    pub ban_storage_healthy: bool,
    pub cache_healthy: bool,
    pub background_tasks_alive: bool,
}

impl From<&HealthStatus> for HealthIntrospection {
    fn from(s: &HealthStatus) -> Self {
        Self {
            storage_healthy: s.storage_healthy,
            ban_storage_healthy: s.ban_storage_healthy,
            cache_healthy: s.cache_healthy,
            background_tasks_alive: s.background_tasks_alive,
        }
    }
}

/// 配置热更新结果报告
///
/// [`Governor::apply_config`] 的返回值，供 Admin API JSON 输出与审计。
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ConfigApplyReport {
    /// 是否已换入（成功时恒为 true；失败走 Err 路径）
    pub applied: bool,
    /// 换装前的配置版本
    pub old_version: String,
    /// 换入的新配置版本
    pub new_version: String,
    /// 新配置哈希（`FlowControlConfig::compute_hash`）
    pub config_hash: String,
    /// 新配置规则数
    pub rule_count: usize,
    /// 换装完成时间（UTC）
    pub applied_at: chrono::DateTime<chrono::Utc>,
}

/// Governor 运行时自省快照
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct IntrospectionSnapshot {
    /// 配置版本
    pub config_version: String,
    /// 规则清单
    pub rules: Vec<RuleIntrospection>,
    /// 决策链清单（按规则 ID 排序）
    pub chains: Vec<ChainIntrospection>,
    /// 聚合统计
    pub stats: GovernorStats,
    /// 缓存状态
    pub l1_cache: L1Introspection,
    /// 组件健康
    pub health: HealthIntrospection,
    /// 是否启用多租户（multi-tenant feature）
    #[cfg(feature = "multi-tenant")]
    pub multi_tenant_enabled: bool,
    /// 是否已关闭
    pub is_shutdown: bool,
}

/// 限流器配置类型名（自省输出用）
fn limiter_kind(l: &crate::config::LimiterConfig) -> String {
    use crate::config::LimiterConfig as L;
    match l {
        L::TokenBucket { .. } => "token_bucket".to_string(),
        L::SlidingWindow { .. } => "sliding_window".to_string(),
        L::FixedWindow { .. } => "fixed_window".to_string(),
        L::Quota { .. } => "quota".to_string(),
        L::Concurrency { .. } => "concurrency".to_string(),
        L::LeakyBucket { .. } => "leaky_bucket".to_string(),
        L::SlidingWindowLog { .. } => "sliding_window_log".to_string(),
        L::PriorityQueue { .. } => "priority_queue".to_string(),
        L::AdmissionControl { .. } => "admission_control".to_string(),
        L::Custom { name, .. } => format!("custom:{name}"),
    }
}

// ============================================================================
// Governor Construction Patterns Tests
// ============================================================================

#[cfg(test)]
#[allow(clippy::field_reassign_with_default)]
mod governor_construction_tests {
    use super::*;
    use crate::config::{Action, ActionConfig, FlowControlConfig, LimiterConfig, Matcher, Rule};
    use crate::error::StorageError;
    use crate::storage::{BanHistory, BanRecord, BanTarget, MemoryBanStorage, MemoryStorage};
    use async_trait::async_trait;

    fn create_valid_test_config() -> FlowControlConfig {
        FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![Rule {
                id: "test_rule".to_string(),
                name: "Test Rule".to_string(),
                priority: 100,
                matchers: vec![Matcher::User {
                    user_ids: vec!["*".to_string()],
                }],
                limiters: vec![LimiterConfig::TokenBucket {
                    capacity: 100,
                    refill_rate: 10,
                }],
                action: ActionConfig {
                    on_exceed: Action::Reject,
                    ban: None,
                },
            }],
        }
    }

    #[tokio::test]
    async fn test_governor_new_with_valid_config() {
        // Governor::new() requires valid config with at least one rule
        // Create a valid config for testing
        let config = create_valid_test_config();

        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
        assert_eq!(stats.allowed_requests, 0);
    }

    #[tokio::test]
    async fn test_governor_builder_with_storage() {
        // Test Governor::builder() with explicit storage
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let config = create_valid_test_config();

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
    }

    #[tokio::test]
    async fn test_governor_with_storage_explicit() {
        // Test Governor::with_storage() with explicit configuration
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let config = create_valid_test_config();

        #[cfg(feature = "prometheus")]
        let metrics: Option<Arc<Metrics>> = None;
        #[cfg(feature = "telemetry")]
        let tracer: Option<Arc<Tracer>> = None;

        let governor = Governor::with_storage(
            config,
            storage,
            ban_storage,
            #[cfg(feature = "prometheus")]
            metrics,
            #[cfg(feature = "telemetry")]
            tracer,
        )
        .await
        .expect("Governor creation should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
    }

    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_governor_with_memory_ban_storage() {
        // Verify MemoryBanStorage is properly integrated
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        // Health check should work
        let result = governor.health_check().await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_governor_check_with_valid_request() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.user_id = Some("test_user".to_string());

        let result = governor.check(&ctx).await;
        // check() may return error if no matching rule found - just verify it doesn't panic
        let _ = result;
    }

    #[tokio::test]
    async fn test_governor_stats_after_requests() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.user_id = Some("test_user".to_string());

        // Make several requests
        for _ in 0..5 {
            let _ = governor.check(&ctx).await;
        }

        let stats = governor.stats().await;
        // Stats should reflect the requests (allowed or rejected depends on rule matching)
        assert_eq!(stats.total_requests, 5);
    }

    #[tokio::test]
    async fn test_governor_health_check() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        // 真实 health_check：MemoryStorage 应健康
        let result = governor.health_check().await;
        assert!(
            result.is_ok(),
            "health_check should succeed with healthy MemoryStorage: {:?}",
            result.err()
        );
    }

    /// 验证 health_status() 返回真实的组件状态（非硬编码 true）
    #[tokio::test]
    async fn test_governor_health_status_real() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let status = governor.health_status().await;

        // MemoryStorage 的 probe get 应成功（返回 Ok(None)）
        assert!(
            status.storage_healthy,
            "storage_healthy should be true for MemoryStorage"
        );
        // MemoryBanStorage 的 list_bans 应成功
        assert!(
            status.ban_storage_healthy,
            "ban_storage_healthy should be true for MemoryBanStorage"
        );
        // 缓存为内存实现，应健康
        assert!(status.cache_healthy, "cache_healthy should be true");
        // 无后台任务时视为存活
        assert!(
            status.background_tasks_alive,
            "background_tasks_alive should be true"
        );
        // 整体应健康
        assert!(status.healthy(), "all components should be healthy");
    }

    /// 验证 shutdown() 幂等性：连续两次调用均返回 Ok
    #[tokio::test]
    async fn test_shutdown_idempotent() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        // 关闭前：is_shutdown 应为 false
        assert!(
            !governor.is_shutdown(),
            "Governor should not be shutdown initially"
        );

        // 第一次 shutdown
        let result1 = governor.shutdown().await;
        assert!(
            result1.is_ok(),
            "first shutdown should succeed: {:?}",
            result1.err()
        );
        assert!(
            governor.is_shutdown(),
            "Governor should be shutdown after first call"
        );

        // 第二次 shutdown（幂等）
        let result2 = governor.shutdown().await;
        assert!(
            result2.is_ok(),
            "second shutdown should be idempotent: {:?}",
            result2.err()
        );
        assert!(governor.is_shutdown(), "Governor should still be shutdown");
    }

    /// 验证 shutdown 后 health_status 反映关闭状态
    #[tokio::test]
    async fn test_shutdown_affects_health_status() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        // 关闭前：background_tasks_alive 应为 true
        let status_before = governor.health_status().await;
        assert!(
            status_before.background_tasks_alive,
            "background_tasks_alive should be true before shutdown"
        );

        // shutdown
        governor.shutdown().await.unwrap();

        // 关闭后：background_tasks_alive 应为 false
        let status_after = governor.health_status().await;
        assert!(
            !status_after.background_tasks_alive,
            "background_tasks_alive should be false after shutdown"
        );
    }

    /// 验证 shutdown_token 可被外部订阅
    #[tokio::test]
    async fn test_shutdown_token_cancellation() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let token = governor.shutdown_token();
        assert!(
            !token.is_cancelled(),
            "token should not be cancelled initially"
        );

        governor.shutdown().await.unwrap();

        assert!(
            token.is_cancelled(),
            "token should be cancelled after shutdown"
        );
    }

    /// 验证 shutdown 将统计快照落盘到指定目录（JSON，含制造的真实计数）
    #[tokio::test]
    async fn test_shutdown_persists_stats_snapshot() {
        let dir = tempfile::tempdir().expect("tempdir should be created");
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_shutdown_snapshot_dir(dir.path())
            .build()
            .await
            .expect("Governor build should succeed");

        // 制造确定性计数（直接驱动 StatsManager，避免依赖规则匹配）
        governor.stats.increment_total();
        governor.stats.increment_total();
        governor.stats.increment_allowed();
        governor.stats.increment_rejected();

        governor.shutdown().await.unwrap();

        let snapshot_path = dir.path().join("governor-shutdown-snapshot.json");
        let raw = std::fs::read_to_string(&snapshot_path)
            .expect("shutdown snapshot should be written to the configured dir");
        let parsed: serde_json::Value =
            serde_json::from_str(&raw).expect("snapshot should be valid JSON");

        let stats = &parsed["stats"];
        assert_eq!(
            stats["total_requests"].as_u64(),
            Some(2),
            "total_requests should be captured in snapshot"
        );
        assert_eq!(
            stats["allowed_requests"].as_u64(),
            Some(1),
            "allowed_requests should be captured in snapshot"
        );
        assert_eq!(
            stats["rejected_requests"].as_u64(),
            Some(1),
            "rejected_requests should be captured in snapshot"
        );
        let captured_at_valid = parsed["captured_at"]
            .as_str()
            .is_some_and(|ts| chrono::DateTime::parse_from_rfc3339(ts).is_ok());
        assert!(
            captured_at_valid,
            "captured_at should be a valid RFC3339 timestamp"
        );
    }

    /// 验证未配置快照目录时 shutdown 不写任何文件且仍成功
    #[tokio::test]
    async fn test_shutdown_without_snapshot_dir_writes_nothing() {
        let dir = tempfile::tempdir().expect("tempdir should be created");
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        governor.shutdown().await.unwrap();

        assert!(
            dir.path()
                .read_dir()
                .expect("dir should exist")
                .next()
                .is_none(),
            "no snapshot file should be written when snapshot dir is not configured"
        );
    }

    /// 验证快照目录不存在时 shutdown 自动创建并落盘
    #[tokio::test]
    async fn test_shutdown_creates_missing_snapshot_dir() {
        let dir = tempfile::tempdir().expect("tempdir should be created");
        let nested = dir.path().join("metrics").join("snapshots");
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_shutdown_snapshot_dir(&nested)
            .build()
            .await
            .expect("Governor build should succeed");

        governor.shutdown().await.unwrap();

        assert!(
            nested.join("governor-shutdown-snapshot.json").exists(),
            "missing snapshot dir should be created and snapshot written"
        );
    }

    /// 验证临时文件名被占用时退避到带 pid 后缀的独占名字，且不破坏占用文件
    ///
    /// 模拟共享目录下的残留/预置冲突：create_new（O_EXCL）对已存在的 .tmp
    /// 失败后应换名重试，快照仍落盘，原占用文件内容保持原样。
    #[cfg(unix)]
    #[tokio::test]
    async fn test_shutdown_snapshot_tmp_conflict_falls_back() {
        let dir = tempfile::tempdir().expect("tempdir should be created");
        let tmp_path = dir.path().join("governor-shutdown-snapshot.json.tmp");
        std::fs::write(&tmp_path, b"occupied").expect("placeholder should be written");

        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_shutdown_snapshot_dir(dir.path())
            .build()
            .await
            .expect("Governor build should succeed");

        governor.shutdown().await.unwrap();

        let snapshot_path = dir.path().join("governor-shutdown-snapshot.json");
        assert!(
            snapshot_path.exists(),
            "snapshot should be written via pid-suffixed tmp despite conflict"
        );
        let occupied = std::fs::read(&tmp_path).expect("placeholder should remain");
        assert_eq!(
            occupied, b"occupied",
            "conflicting file must not be touched"
        );
    }

    /// 验证 shutdown 取消已注册的配置 watcher 令牌
    #[tokio::test]
    async fn test_shutdown_cancels_registered_config_watcher() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let watcher_token = tokio_util::sync::CancellationToken::new();
        governor.register_config_watcher_token(watcher_token.clone());

        governor.shutdown().await.unwrap();

        assert!(
            watcher_token.is_cancelled(),
            "registered config watcher token should be cancelled by shutdown"
        );
    }

    /// 验证多 watcher 场景：多次注册的令牌在 shutdown 时全部取消，不覆盖丢失
    #[tokio::test]
    async fn test_shutdown_cancels_all_registered_watchers() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let first = tokio_util::sync::CancellationToken::new();
        let second = tokio_util::sync::CancellationToken::new();
        let third = tokio_util::sync::CancellationToken::new();
        governor.register_config_watcher_token(first.clone());
        governor.register_config_watcher_token(second.clone());
        governor.register_config_watcher_token(third.clone());

        governor.shutdown().await.unwrap();

        assert!(
            first.is_cancelled(),
            "first watcher token should be cancelled"
        );
        assert!(
            second.is_cancelled(),
            "second watcher token should be cancelled"
        );
        assert!(
            third.is_cancelled(),
            "third watcher token should be cancelled"
        );
    }

    /// 验证 Drop 取消全部已注册的 watcher 令牌
    #[tokio::test]
    async fn test_drop_cancels_all_registered_watchers() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let first = tokio_util::sync::CancellationToken::new();
        let second = tokio_util::sync::CancellationToken::new();
        {
            let governor = Governor::builder()
                .with_config(config)
                .with_storage(storage)
                .with_ban_storage(ban_storage)
                .build()
                .await
                .expect("Governor build should succeed");
            governor.register_config_watcher_token(first.clone());
            governor.register_config_watcher_token(second.clone());
        }

        assert!(
            first.is_cancelled(),
            "first watcher token cancelled on drop"
        );
        assert!(
            second.is_cancelled(),
            "second watcher token cancelled on drop"
        );
    }

    /// 验证 stop_config_watcher 单独调用即取消注册的令牌，且重复调用幂等
    #[tokio::test]
    async fn test_stop_config_watcher_cancels_registered_token() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let watcher_token = tokio_util::sync::CancellationToken::new();
        governor.register_config_watcher_token(watcher_token.clone());

        governor.stop_config_watcher().await.unwrap();
        assert!(
            watcher_token.is_cancelled(),
            "stop_config_watcher should cancel the registered token"
        );

        // 重复调用：令牌已取走，应幂等返回 Ok
        governor.stop_config_watcher().await.unwrap();
    }

    /// 验证 Drop 触发关闭路径：取消 shutdown 令牌与已注册的 watcher 令牌
    #[tokio::test]
    async fn test_drop_cancels_shutdown_tokens() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let watcher_token = tokio_util::sync::CancellationToken::new();
        let shutdown_token;
        {
            let governor = Governor::builder()
                .with_config(config)
                .with_storage(storage)
                .with_ban_storage(ban_storage)
                .build()
                .await
                .expect("Governor build should succeed");
            governor.register_config_watcher_token(watcher_token.clone());
            shutdown_token = governor.shutdown_token().clone();
            assert!(!shutdown_token.is_cancelled());
        }
        // governor 在此 drop：无显式 shutdown 时应兜底取消所有令牌

        assert!(
            shutdown_token.is_cancelled(),
            "shutdown token should be cancelled on drop"
        );
        assert!(
            watcher_token.is_cancelled(),
            "registered config watcher token should be cancelled on drop"
        );
    }

    /// 验证显式 shutdown 后再 Drop 不逆操作、不 panic（幂等兜底）
    #[tokio::test]
    async fn test_drop_after_explicit_shutdown_is_noop() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let shutdown_token;
        {
            let governor = Governor::builder()
                .with_config(config)
                .with_storage(storage)
                .with_ban_storage(ban_storage)
                .build()
                .await
                .expect("Governor build should succeed");
            shutdown_token = governor.shutdown_token().clone();
            governor.shutdown().await.unwrap();
            assert!(shutdown_token.is_cancelled());
        }

        assert!(
            shutdown_token.is_cancelled(),
            "drop after explicit shutdown should keep token cancelled"
        );
    }

    /// 验证 shutdown 停止 BanManager 的自动解封后台任务（feature `ban-manager`）
    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_shutdown_stops_auto_unban_task() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        // builder 构造 BanManager 默认启用自动解封（enable_auto_unban=true）
        assert!(
            governor.ban_manager.is_auto_unban_running().await,
            "auto-unban task should be running before shutdown"
        );

        governor.shutdown().await.unwrap();

        assert!(
            !governor.ban_manager.is_auto_unban_running().await,
            "auto-unban task should be stopped after shutdown"
        );
    }

    /// 验证 shutdown 摘除审计日志器：尽力优雅等待写入任务，
    /// 无法取得所有权时依赖 AuditLogger::drop 兜底（feature `audit-log`）
    #[cfg(feature = "audit-log")]
    #[tokio::test]
    async fn test_shutdown_detaches_audit_logger() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let audit_logger = Arc::new(
            crate::logging::AuditLogger::new(crate::logging::AuditLogConfig::default()).await,
        );

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_audit_logger(audit_logger)
            .build()
            .await
            .expect("Governor build should succeed");

        assert!(
            governor.audit_logger.read().await.is_some(),
            "audit logger should be attached before shutdown"
        );

        governor.shutdown().await.unwrap();

        assert!(
            governor.audit_logger.read().await.is_none(),
            "audit logger should be detached after shutdown"
        );
    }

    /// 验证外部仍持有审计器引用时的摘除分支（Arc::try_unwrap 失败路径）：
    /// 槽位清空、不 panic，优雅排空不可行时由 AuditLogger::drop 兜底
    #[cfg(feature = "audit-log")]
    #[tokio::test]
    async fn test_shutdown_detaches_audit_logger_with_external_ref() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let external_ref = Arc::new(
            crate::logging::AuditLogger::new(crate::logging::AuditLogConfig::default()).await,
        );
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_audit_logger(external_ref.clone())
            .build()
            .await
            .expect("Governor build should succeed");

        governor.shutdown().await.unwrap();

        assert!(
            governor.audit_logger.read().await.is_none(),
            "audit logger slot should be cleared even when ownership is shared"
        );
        // 外部引用仍存活：drop 它以触发 AuditLogger 的 abort 兜底（不应 panic）
        drop(external_ref);
    }

    // ========================================================================
    // Mock 存储实现（用于测试 health_check 错误路径）
    // ========================================================================

    /// 所有操作均返回 Err 的 Storage mock，用于测试 health_check 错误分支
    struct FailingStorage;

    #[async_trait]
    impl Storage for FailingStorage {
        async fn get(&self, _key: &str) -> Result<Option<String>, StorageError> {
            Err(StorageError::ConnectionError(
                "mock storage failure".to_string(),
            ))
        }
        async fn set(
            &self,
            _key: &str,
            _value: &str,
            _ttl: Option<u64>,
        ) -> Result<(), StorageError> {
            Err(StorageError::ConnectionError(
                "mock storage failure".to_string(),
            ))
        }
        async fn delete(&self, _key: &str) -> Result<(), StorageError> {
            Err(StorageError::ConnectionError(
                "mock storage failure".to_string(),
            ))
        }
    }

    /// 所有操作均返回 Err 的 BanStorage mock，用于测试 health_check 错误分支
    struct FailingBanStorage;

    #[async_trait]
    impl BanStorage for FailingBanStorage {
        async fn is_banned(&self, _target: &BanTarget) -> Result<Option<BanRecord>, StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        async fn save(&self, _record: &BanRecord) -> Result<(), StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        async fn get_history(
            &self,
            _target: &BanTarget,
        ) -> Result<Option<BanHistory>, StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        async fn increment_ban_times(&self, _target: &BanTarget) -> Result<u64, StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        async fn get_ban_times(&self, _target: &BanTarget) -> Result<u64, StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        async fn remove_ban(&self, _target: &BanTarget) -> Result<(), StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        async fn cleanup_expired_bans(&self) -> Result<u64, StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        async fn list_bans(
            &self,
            _active_only: bool,
            _offset: u64,
            _limit: u64,
        ) -> Result<Vec<BanRecord>, StorageError> {
            Err(StorageError::ConnectionError(
                "mock ban storage failure".to_string(),
            ))
        }
        fn as_any(&self) -> &dyn std::any::Any {
            self
        }
    }

    // ========================================================================
    // health_check 错误路径测试
    // ========================================================================

    /// 验证 health_check() 在存储不可用时返回 Err
    #[tokio::test]
    async fn test_health_check_returns_err_when_storage_unhealthy() {
        let storage: Arc<dyn Storage> = Arc::new(FailingStorage);
        let ban_storage: Arc<dyn BanStorage> = Arc::new(FailingBanStorage);

        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let result = governor.health_check().await;
        assert!(
            result.is_err(),
            "health_check should return Err when storage is unhealthy"
        );
    }

    /// 验证 health_check() 在 shutdown 后返回 Err（background_tasks_alive = false）
    #[tokio::test]
    async fn test_health_check_returns_err_after_shutdown() {
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        // shutdown 前：health_check 应返回 Ok
        let result_before = governor.health_check().await;
        assert!(
            result_before.is_ok(),
            "health_check should pass before shutdown"
        );

        governor.shutdown().await.unwrap();

        // shutdown 后：health_check 应返回 Err（background_tasks_alive = false）
        let result_after = governor.health_check().await;
        assert!(
            result_after.is_err(),
            "health_check should return Err after shutdown (background_tasks not alive)"
        );
    }

    /// 验证 health_check() 错误消息包含具体不健康组件名称
    #[tokio::test]
    async fn test_health_check_error_message_contains_component_names() {
        let storage: Arc<dyn Storage> = Arc::new(FailingStorage);
        let ban_storage: Arc<dyn BanStorage> = Arc::new(FailingBanStorage);

        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let err = governor.health_check().await.unwrap_err();
        let err_msg = match err {
            LimiteronError::StorageError(StorageError::ConnectionError(msg)) => msg,
            other => panic!("expected StorageError::ConnectionError, got {other:?}"),
        };
        assert!(
            err_msg.contains("storage"),
            "error message should mention 'storage' component: {err_msg}"
        );
        assert!(
            err_msg.contains("ban_storage"),
            "error message should mention 'ban_storage' component: {err_msg}"
        );
        assert!(
            err_msg.contains("Components unhealthy"),
            "error message should have descriptive prefix: {err_msg}"
        );
    }

    /// 验证 health_status() 正确报告各组件健康状况
    #[tokio::test]
    async fn test_health_status_reports_unhealthy_components() {
        let storage: Arc<dyn Storage> = Arc::new(FailingStorage);
        let ban_storage: Arc<dyn BanStorage> = Arc::new(FailingBanStorage);

        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let status = governor.health_status().await;
        assert!(
            !status.storage_healthy,
            "storage_healthy should be false with FailingStorage"
        );
        assert!(
            !status.ban_storage_healthy,
            "ban_storage_healthy should be false with FailingBanStorage"
        );
        assert!(
            status.cache_healthy,
            "cache_healthy should always be true (in-memory, cannot fail)"
        );
        assert!(
            status.background_tasks_alive,
            "background_tasks_alive should be true before shutdown"
        );
        assert!(
            !status.healthy(),
            "overall healthy() should be false when any component is down"
        );
    }

    /// 验证 HealthStatus::healthy() 在所有组件健康时返回 true
    #[test]
    fn test_health_status_healthy_all_true() {
        let status = HealthStatus {
            storage_healthy: true,
            ban_storage_healthy: true,
            cache_healthy: true,
            background_tasks_alive: true,
        };
        assert!(status.healthy(), "all true should be healthy");
    }

    /// 验证 HealthStatus::healthy() 在任一组件不健康时返回 false
    #[test]
    fn test_health_status_healthy_any_false() {
        // 测试每个字段单独为 false 的情况
        let cases = [
            (
                "storage",
                HealthStatus {
                    storage_healthy: false,
                    ban_storage_healthy: true,
                    cache_healthy: true,
                    background_tasks_alive: true,
                },
            ),
            (
                "ban_storage",
                HealthStatus {
                    storage_healthy: true,
                    ban_storage_healthy: false,
                    cache_healthy: true,
                    background_tasks_alive: true,
                },
            ),
            (
                "cache",
                HealthStatus {
                    storage_healthy: true,
                    ban_storage_healthy: true,
                    cache_healthy: false,
                    background_tasks_alive: true,
                },
            ),
            (
                "background_tasks",
                HealthStatus {
                    storage_healthy: true,
                    ban_storage_healthy: true,
                    cache_healthy: true,
                    background_tasks_alive: false,
                },
            ),
        ];
        for (name, status) in cases {
            assert!(
                !status.healthy(),
                "healthy() should be false when {name} is false"
            );
        }
    }

    #[tokio::test]
    async fn test_governor_new_with_defaults() {
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");
        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
    }

    #[test]
    fn test_governor_builder_validation_empty_config() {
        let result = std::thread::spawn(|| {
            let rt = tokio::runtime::Runtime::new().unwrap();
            rt.block_on(async {
                let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
                let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());
                Governor::builder()
                    .with_storage(storage)
                    .with_ban_storage(ban_storage)
                    .build()
                    .await
            })
        });
        let result = result.join().unwrap();
        assert!(result.is_err());
    }

    /// 回归：Governor::new() 不再因空默认配置 expect panic，
    /// 而是内置一条可用的兜底规则并返回 Ok；兜底规则实际生效
    /// （同 user 超过 TokenBucket 容量后被拒绝）。
    #[tokio::test]
    async fn test_governor_new_works_with_builtin_default_rule() {
        let governor = Governor::new().await.expect("new() should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);

        // 内置兜底规则匹配所有用户（User: ["*"]），TokenBucket 容量 100；
        // 默认标识符提取器读取 X-User-Id 请求头
        let ctx = RequestContext::new().with_header("X-User-Id", "out-of-box-user");
        for _ in 0..100 {
            let result = governor.check(&ctx).await.expect("check should not error");
            assert!(matches!(result, Decision::Allowed(_)));
        }
        // 第 101 个请求应被内置兜底规则限流
        let result = governor.check(&ctx).await.expect("check should not error");
        assert!(
            matches!(result, Decision::Rejected(_)),
            "the 101st request must be limited by the built-in fallback rule"
        );
    }

    #[tokio::test]
    async fn test_governor_builder_with_l1_cache_config() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        use crate::l1_cache::L1CacheConfig;
        let l1_config = L1CacheConfig {
            default_ttl: std::time::Duration::from_secs(60),
            max_size: 10000,
            ..Default::default()
        };

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_l1_cache_config(l1_config)
            .build()
            .await
            .expect("Governor build should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
    }

    #[tokio::test]
    async fn test_governor_config_validation_missing_rules() {
        let config = FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![],
        };
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let result = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await;

        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_governor_multiple_requests_stats() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.user_id = Some("user_a".to_string());

        for _ in 0..3 {
            let _ = governor.check(&ctx).await;
        }

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 3);

        let mut ctx_b = RequestContext::default();
        ctx_b.user_id = Some("user_b".to_string());
        let _ = governor.check(&ctx_b).await;

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 4);
    }

    #[tokio::test]
    async fn test_governor_with_storage_reuse() {
        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let governor = Governor::with_storage(
            config,
            storage.clone(),
            ban_storage.clone(),
            #[cfg(feature = "prometheus")]
            None,
            #[cfg(feature = "telemetry")]
            None,
        )
        .await
        .expect("Governor creation should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
    }

    // ============================================================================
    // Builder Error Paths
    // ============================================================================

    #[tokio::test]
    async fn test_builder_missing_storage() {
        let result = Governor::builder()
            .with_config(create_valid_test_config())
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await;
        let err = result
            .err()
            .expect("expected DependencyError for missing storage");
        match err {
            LimiteronError::DependencyError(msg) => assert!(msg.contains("storage")),
            _ => panic!("expected DependencyError"),
        }
    }

    #[tokio::test]
    async fn test_builder_missing_ban_storage() {
        let result = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .build()
            .await;
        let err = result
            .err()
            .expect("expected DependencyError for missing ban_storage");
        match err {
            LimiteronError::DependencyError(msg) => assert!(msg.contains("ban_storage")),
            _ => panic!("expected DependencyError"),
        }
    }

    // ============================================================================
    // check() Code Paths - Identifier Extraction
    // ============================================================================

    #[tokio::test]
    async fn test_check_identifier_extraction_failure() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let ctx = RequestContext::default();
        let result = governor.check(&ctx).await;
        match result {
            Err(LimiteronError::ConfigError(msg)) => {
                assert!(msg.contains("Failed to extract identifier"))
            }
            other => panic!("Expected ConfigError, got: {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_check_total_incremented_on_extraction_failure() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let ctx = RequestContext::default();
        let _ = governor.check(&ctx).await;

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 1);
        assert_eq!(stats.allowed_requests, 0);
        assert_eq!(stats.rejected_requests, 0);
        assert_eq!(stats.banned_requests, 0);
    }

    // ============================================================================
    // check() Code Paths - No Matched Rules (fallthrough to default decision chain)
    // ============================================================================

    fn create_non_matching_config() -> FlowControlConfig {
        FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![Rule {
                id: "specific_rule".to_string(),
                name: "Specific Rule".to_string(),
                priority: 100,
                matchers: vec![Matcher::User {
                    user_ids: vec!["specific_user".to_string()],
                }],
                limiters: vec![LimiterConfig::TokenBucket {
                    capacity: 100,
                    refill_rate: 10,
                }],
                action: ActionConfig {
                    on_exceed: Action::Reject,
                    ban: None,
                },
            }],
        }
    }

    #[tokio::test]
    async fn test_check_no_matched_rules_falls_to_default_chain() {
        let config = create_non_matching_config();
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("192.168.1.1".to_string());

        let result = governor.check(&ctx).await;
        match result {
            Ok(Decision::Allowed(_)) => {}
            other => panic!("Expected Allowed via default chain, got: {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_check_no_matched_rules_updates_stats() {
        let config = create_non_matching_config();
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("192.168.1.1".to_string());

        let _ = governor.check(&ctx).await;
        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 1);
        assert_eq!(stats.allowed_requests, 1);
    }

    // ============================================================================
    // check() Code Paths - Rule Accepts (TokenBucket has capacity)
    // ============================================================================

    #[tokio::test]
    async fn test_check_rule_allowed() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.1".to_string());

        let result = governor.check(&ctx).await;
        match result {
            Ok(Decision::Allowed(_)) => {}
            other => panic!("Expected Allowed, got: {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_check_allowed_increments_stats() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.2".to_string());

        let _ = governor.check(&ctx).await;
        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 1);
        assert_eq!(stats.allowed_requests, 1);
    }

    // ============================================================================
    // check() Code Paths - Rule Rejects (TokenBucket exhausted)
    // ============================================================================

    fn create_small_capacity_config() -> FlowControlConfig {
        FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![Rule {
                id: "small_bucket".to_string(),
                name: "Small Bucket Rule".to_string(),
                priority: 100,
                matchers: vec![Matcher::User {
                    user_ids: vec!["*".to_string()],
                }],
                limiters: vec![LimiterConfig::TokenBucket {
                    capacity: 3,
                    refill_rate: 1,
                }],
                action: ActionConfig {
                    on_exceed: Action::Reject,
                    ban: None,
                },
            }],
        }
    }

    #[tokio::test]
    async fn test_check_rule_accepted_then_rejected() {
        let config = create_small_capacity_config();
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.3".to_string());

        for _ in 0..3 {
            let result = governor.check(&ctx).await;
            assert!(
                result.is_ok(),
                "Expected Ok within capacity, got: {:?}",
                result
            );
        }

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 3);
        assert_eq!(stats.allowed_requests, 3);
    }

    #[tokio::test]
    async fn test_check_rejected_updates_rejected_stats() {
        let config = create_small_capacity_config();
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.4".to_string());

        for _ in 0..4 {
            let _ = governor.check(&ctx).await;
        }

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 4);
    }

    #[tokio::test]
    async fn test_check_multiple_users_isolated() {
        let config = create_small_capacity_config();
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx_a = RequestContext::default();
        ctx_a.client_ip = Some("10.0.0.10".to_string());

        let mut ctx_b = RequestContext::default();
        ctx_b.client_ip = Some("10.0.0.11".to_string());

        for _ in 0..3 {
            let _ = governor.check(&ctx_a).await;
        }
        governor.check(&ctx_a).await.expect("check should succeed");
        governor.check(&ctx_b).await.expect("check should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 5);
    }

    // ============================================================================
    // Cache Methods
    // ============================================================================

    #[tokio::test]
    async fn test_l1_cache_enable_disable() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        assert!(governor.is_l1_cache_enabled());

        governor.disable_l1_cache();
        assert!(!governor.is_l1_cache_enabled());

        governor.enable_l1_cache();
        assert!(governor.is_l1_cache_enabled());
    }

    #[tokio::test]
    async fn test_l1_cache_disabled_does_not_cache() {
        let config = create_small_capacity_config();
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_l1_cache_enabled(false)
            .build()
            .await
            .expect("Governor build should succeed");

        assert!(!governor.is_l1_cache_enabled());

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.5".to_string());

        for _ in 0..3 {
            let result = governor.check(&ctx).await;
            assert!(result.is_ok());
        }
        let result = governor.check(&ctx).await;
        match result {
            Ok(Decision::Rejected(_)) => {}
            other => panic!(
                "Expected Rejected after exhaust with L1 disabled, got: {:?}",
                other
            ),
        }
    }

    #[tokio::test]
    async fn test_l1_cache_clear_and_size() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let size = governor.l1_cache_size().await;
        assert_eq!(size, 0);

        let _ = governor.clear_l1_cache().await;
        assert_eq!(governor.l1_cache_size().await, 0);
    }

    #[tokio::test]
    async fn test_evict_expired_l1_cache() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let evicted = governor.evict_expired_l1_cache().await;
        assert_eq!(evicted, 0);
    }

    // ============================================================================
    // Stats and Reset
    // ============================================================================

    #[tokio::test]
    async fn test_reset_stats() {
        let config = create_small_capacity_config();
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.6".to_string());

        let _ = governor.check(&ctx).await;
        let stats_before = governor.stats().await;
        assert_eq!(stats_before.total_requests, 1);
        assert_eq!(stats_before.allowed_requests, 1);

        governor.reset_stats().await;

        let stats_after = governor.stats().await;
        assert_eq!(stats_after.total_requests, 0);
        assert_eq!(stats_after.allowed_requests, 0);
        assert_eq!(stats_after.rejected_requests, 0);
        assert_eq!(stats_after.banned_requests, 0);
        assert_eq!(stats_after.error_count, 0);
    }

    // ============================================================================
    // Chain and Matcher Stats
    // ============================================================================

    #[tokio::test]
    async fn test_decision_chain_stats() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let stats = governor.decision_chain_stats().await;
        assert_eq!(stats.allowed_count, 0);
        assert_eq!(stats.rejected_count, 0);
        assert_eq!(stats.error_count, 0);
    }

    #[tokio::test]
    async fn test_rule_matcher_stats() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let stats = governor.rule_matcher_stats().await;
        assert_eq!(stats.total_matches, 0);
        assert_eq!(stats.total_mismatches, 0);
    }

    // ============================================================================
    // Config and Administrative Methods
    // ============================================================================

    #[tokio::test]
    async fn test_get_config_history() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let history = governor.get_config_history().await;
        assert!(history.is_empty());
    }

    #[tokio::test]
    async fn test_manual_config_check_returns_ok() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let result = governor.manual_config_check().await;
        assert!(result.is_ok());
        match result {
            Ok(val) => assert!(val),
            _ => unreachable!(),
        }
    }

    #[tokio::test]
    async fn test_stop_config_watcher() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let result = governor.stop_config_watcher().await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_health_check_returns_ok() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let result = governor.health_check().await;
        assert!(result.is_ok());
    }

    // ============================================================================
    // Ban/Unban Operations (feature-gated)
    // ============================================================================

    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_ban_identifier_user_id() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let identifier = crate::matchers::Identifier::UserId("test_user".to_string());
        let result = governor
            .ban_identifier(&identifier, "test ban reason", None)
            .await;
        assert!(
            result.is_ok(),
            "ban_identifier should succeed: {:?}",
            result
        );
    }

    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_ban_identifier_ip() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let identifier = crate::matchers::Identifier::Ip("192.168.1.1".to_string());
        let result = governor
            .ban_identifier(&identifier, "test IP ban", None)
            .await;
        assert!(
            result.is_ok(),
            "ban_identifier for IP should succeed: {:?}",
            result
        );
    }

    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_ban_identifier_unsupported_type() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let identifier = crate::matchers::Identifier::ApiKey("test_key".to_string());
        let result = governor.ban_identifier(&identifier, "test ban", None).await;
        match result {
            Err(LimiteronError::ValidationError(msg)) => {
                assert!(msg.contains("Unsupported identifier type"))
            }
            other => panic!("Expected ValidationError, got: {:?}", other),
        }
    }

    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_unban_identifier() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let identifier = crate::matchers::Identifier::UserId("unban_test_user".to_string());

        let ban_result = governor.ban_identifier(&identifier, "temp ban", None).await;
        assert!(ban_result.is_ok(), "ban should succeed");

        let unban_result = governor.unban_identifier(&identifier).await;
        assert!(
            unban_result.is_ok(),
            "unban should succeed: {:?}",
            unban_result
        );
    }

    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_unban_identifier_unsupported_type() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let identifier = crate::matchers::Identifier::ApiKey("test_key".to_string());
        let result = governor.unban_identifier(&identifier).await;
        match result {
            Err(LimiteronError::ValidationError(msg)) => {
                assert!(msg.contains("Unsupported identifier type"))
            }
            other => panic!("Expected ValidationError, got: {:?}", other),
        }
    }

    // ============================================================================
    // GovernorBuilder - Full configuration chain
    // ============================================================================

    #[tokio::test]
    async fn test_governor_builder_full_chain() {
        use crate::l1_cache::L1CacheConfig;

        let config = create_valid_test_config();
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let l1_config = L1CacheConfig {
            default_ttl: std::time::Duration::from_secs(30),
            max_size: 5000,
            ..Default::default()
        };

        let governor = Governor::builder()
            .with_config(config)
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .with_l1_cache_config(l1_config)
            .with_l1_cache_enabled(true)
            .with_identifier_extractor(Arc::new(
                crate::matchers::CompositeExtractor::builder()
                    .add_extractor(Box::new(crate::matchers::UserIdExtractor::from_header(
                        "X-User-Id",
                    )))
                    .build(),
            ))
            .build()
            .await
            .expect("Governor build with full chain should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
        assert!(governor.is_l1_cache_enabled());
    }

    // ============================================================================
    // Cache Invalidation Methods
    // ============================================================================

    #[tokio::test]
    async fn test_invalidate_l1_cache() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        governor.invalidate_l1_cache("test_user").await;
        governor.invalidate_rule_cache("test_rule").await;

        assert_eq!(governor.l1_cache_size().await, 0);
    }

    // ============================================================================
    // From/Into Implementations
    // ============================================================================

    #[test]
    fn test_governor_stats_from_snapshot() {
        let snapshot = crate::rules::StatsSnapshot {
            total_requests: 100,
            allowed_requests: 80,
            rejected_requests: 15,
            banned_requests: 3,
            error_count: 2,
            last_updated: Some(chrono::Utc::now()),
        };
        let stats: GovernorStats = snapshot.into();
        assert_eq!(stats.total_requests, 100);
        assert_eq!(stats.allowed_requests, 80);
        assert_eq!(stats.rejected_requests, 15);
        assert_eq!(stats.banned_requests, 3);
        assert_eq!(stats.error_count, 2);
        assert!(stats.last_updated.is_some());
    }

    // ============================================================================
    // Trait Implementations (Default, Debug, Clone)
    // ============================================================================

    #[test]
    fn test_governor_stats_default_impl() {
        let stats = GovernorStats::default();
        assert_eq!(stats.total_requests, 0);
        assert_eq!(stats.allowed_requests, 0);
        assert_eq!(stats.rejected_requests, 0);
        assert_eq!(stats.banned_requests, 0);
        assert_eq!(stats.error_count, 0);
        assert!(stats.last_updated.is_none());
    }

    #[test]
    fn test_governor_stats_debug_format() {
        let stats = GovernorStats::default();
        let debug = format!("{:?}", stats);
        assert!(debug.contains("total_requests"));
        assert!(debug.contains("allowed_requests"));
        assert!(debug.contains("rejected_requests"));
        assert!(debug.contains("banned_requests"));
        assert!(debug.contains("error_count"));
        assert!(debug.contains("last_updated"));
    }

    #[test]
    fn test_governor_stats_clone() {
        let mut stats = GovernorStats::default();
        stats.total_requests = 42;
        let cloned = stats.clone();
        assert_eq!(cloned.total_requests, 42);
    }

    // ============================================================================
    // Cache - pub(crate) Stats
    // ============================================================================

    #[tokio::test]
    async fn test_l1_cache_stats() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        // 测试模块为 Governor 子模块，直访私有字段（原 cfg(test) 方法已删）
        let stats = governor.l1_cache.stats().await;
        assert_eq!(stats.total_lookups, 0);
        assert_eq!(stats.hits, 0);
        assert_eq!(stats.misses, 0);
        assert_eq!(stats.current_size, 0);
    }

    // ============================================================================
    // Governor::with_dependencies() Constructor
    // ============================================================================

    #[tokio::test]
    async fn test_with_dependencies_constructor() {
        let config = Arc::new(tokio::sync::RwLock::new(create_valid_test_config()));
        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let extractor: Arc<dyn crate::matchers::IdentifierExtractor> = Arc::new(
            crate::matchers::CompositeExtractor::builder()
                .add_extractor(Box::new(crate::matchers::UserIdExtractor::from_header(
                    "X-User-Id",
                )))
                .build(),
        );

        let config_value = config.read().await.clone();
        let rules = RuleBuilder::build_rules(&config_value).expect("build_rules should succeed");
        let rule_matcher = Arc::new(tokio::sync::RwLock::new(
            crate::matchers::RuleMatcher::with_dependencies(rules),
        ));

        let decision_chain = Arc::new(tokio::sync::RwLock::new(
            crate::decision_chain::DecisionChain::with_dependencies(vec![]),
        ));

        let chain_map = RuleBuilder::build_rule_chains(&config_value)
            .expect("build_rule_chains should succeed");
        let rule_chains = Arc::new(tokio::sync::RwLock::new(chain_map));

        #[cfg(feature = "circuit-breaker")]
        let circuit_breaker = {
            use crate::circuit::{CircuitBreaker, CircuitBreakerConfig};
            Arc::new(CircuitBreaker::with_dependencies(
                CircuitBreakerConfig::default(),
            ))
        };

        let governor = Governor::with_dependencies(
            config,
            storage,
            ban_storage,
            extractor,
            rule_matcher,
            decision_chain,
            rule_chains,
            #[cfg(feature = "circuit-breaker")]
            circuit_breaker,
        )
        .await
        .expect("with_dependencies should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
    }

    // ============================================================================
    // build_cache_key / build_cache_key_multi - All Identifier Types
    // ============================================================================

    // build_cache_key 仅存在于 fallback 特性下（降级路径专用）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_build_cache_key_all_types() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let rule = |id: &str| crate::Rule {
            id: id.to_string(),
            name: id.to_string(),
            priority: 1,
            condition: Box::new(crate::MatchCondition::User(vec![])),
            enabled: true,
        };
        let rules = vec![rule("rule_1")];

        let user_id = crate::matchers::Identifier::UserId("user123".to_string());
        let key = governor.build_cache_key_multi(&user_id, &rules);
        assert_eq!(key, "rl:user:user123:rule_1");

        let ip = crate::matchers::Identifier::Ip("192.168.1.1".to_string());
        let key = governor.build_cache_key_multi(&ip, &rules);
        assert_eq!(key, "rl:ip:192.168.1.1:rule_1");

        let api_key = crate::matchers::Identifier::ApiKey("key123".to_string());
        let key = governor.build_cache_key_multi(&api_key, &rules);
        assert_eq!(key, "rl:apikey:key123:rule_1");

        let mac = crate::matchers::Identifier::Mac("AA:BB:CC:DD:EE:FF".to_string());
        let key = governor.build_cache_key_multi(&mac, &rules);
        assert_eq!(key, "rl:generic:mac:AA:BB:CC:DD:EE:FF:rule_1");

        let device = crate::matchers::Identifier::DeviceId("device-001".to_string());
        let key = governor.build_cache_key_multi(&device, &rules);
        assert_eq!(key, "rl:generic:device_id:device-001:rule_1");
    }

    /// 基于全部匹配规则构建缓存键（生产路径，任何特性组合下都存在）
    #[tokio::test]
    async fn test_build_cache_key_multi_all_types() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("Governor build should succeed");

        let rule = |id: &str| crate::Rule {
            id: id.to_string(),
            name: id.to_string(),
            priority: 1,
            condition: Box::new(crate::MatchCondition::User(vec![])),
            enabled: true,
        };
        let rules = vec![rule("rule_a"), rule("rule_b")];

        let user_id = crate::matchers::Identifier::UserId("user123".to_string());
        let key = governor.build_cache_key_multi(&user_id, &rules);
        assert_eq!(key, "rl:user:user123:rule_a|rule_b");

        let ip = crate::matchers::Identifier::Ip("192.168.1.1".to_string());
        let key = governor.build_cache_key_multi(&ip, &rules);
        assert_eq!(key, "rl:ip:192.168.1.1:rule_a|rule_b");
    }

    // ============================================================================
    // Governor::from_config_file
    // ============================================================================

    #[tokio::test]
    async fn test_governor_from_config_file_json() {
        let config_json = r#"{
            "version": "0.1.0",
            "global": {
                "storage": "memory",
                "cache": "memory",
                "metrics": "none"
            },
            "rules": [{
                "id": "test_rule",
                "name": "Test Rule",
                "priority": 100,
                "matchers": [{"type": "User", "user_ids": ["*"]}],
                "limiters": [{"type": "TokenBucket", "capacity": 100, "refill_rate": 10}],
                "action": {"on_exceed": "reject"}
            }]
        }"#;

        let dir = std::env::temp_dir().join(format!("limiteron_cfg_{}", std::process::id()));
        let _ = std::fs::create_dir_all(&dir);
        let config_path = dir.join("test_config.json");
        std::fs::write(&config_path, config_json).expect("write config file");

        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let result = Governor::from_config_file(&config_path, storage, ban_storage).await;
        assert!(
            result.is_ok(),
            "from_config_file failed: {:?}",
            result.err()
        );

        let _ = std::fs::remove_file(&config_path);
        let _ = std::fs::remove_dir(&dir);
    }

    // ============================================================================
    // Governor::from_config_with_env
    // ============================================================================

    #[tokio::test]
    async fn test_governor_from_config_with_env_json() {
        let config_json = r#"{
            "version": "0.1.0",
            "global": {
                "storage": "memory",
                "cache": "memory",
                "metrics": "none"
            },
            "rules": [{
                "id": "test_rule_env",
                "name": "Test Rule Env",
                "priority": 100,
                "matchers": [{"type": "User", "user_ids": ["*"]}],
                "limiters": [{"type": "TokenBucket", "capacity": 100, "refill_rate": 10}],
                "action": {"on_exceed": "reject"}
            }]
        }"#;

        let dir = std::env::temp_dir().join(format!("limiteron_cfg_env_{}", std::process::id()));
        let _ = std::fs::create_dir_all(&dir);
        let config_path = dir.join("test_env_config.json");
        std::fs::write(&config_path, config_json).expect("write config file");

        let storage: Arc<dyn Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn BanStorage> = Arc::new(MemoryBanStorage::new());

        let result = Governor::from_config_with_env(&config_path, storage, ban_storage).await;
        assert!(
            result.is_ok(),
            "from_config_with_env failed: {:?}",
            result.err()
        );

        let _ = std::fs::remove_file(&config_path);
        let _ = std::fs::remove_dir(&dir);
    }

    #[cfg(feature = "parallel-checker")]
    #[tokio::test]
    async fn test_check_resource_parallel_banned() {
        use crate::matchers::Identifier;
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // Ban a user first
        let user_id = Identifier::UserId("banned_resource_user".to_string());
        governor
            .ban_identifier(&user_id, "test ban for parallel", None)
            .await
            .expect("ban should succeed");

        // Check resource parallel should detect the ban
        let result = governor
            .check_resource_parallel("banned_resource_user")
            .await;
        match result {
            Ok(Decision::Banned(_)) => {}
            other => panic!("Expected Banned for banned user, got: {:?}", other),
        }
    }

    // ============================================================================
    // Health Check - feature-gated sub-checks with all features enabled
    // ============================================================================

    #[tokio::test]
    async fn test_health_check_with_all_features() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // Under --features full, this exercises:
        //   #[cfg(feature = "ban-manager")]  -> ban_manager.get_config()
        //   #[cfg(feature = "circuit-breaker")] -> circuit_breaker.get_state()
        //   #[cfg(feature = "audit-log")] -> audit_logger()
        let result = governor.health_check().await;
        assert!(result.is_ok(), "health_check should succeed: {:?}", result);
    }
}

// ============================================================================
// Feature-Gated Code Paths
// ============================================================================

#[cfg(test)]
// 部分测试助手仅被 feature 门控用例引用（默认组合下无调用方），保留放行。
#[allow(clippy::field_reassign_with_default, dead_code, unused_imports)]
mod governor_feature_gated_tests {
    use super::*;
    use crate::storage::{MemoryBanStorage, MemoryStorage};
    use std::sync::Arc;
    use std::time::Duration;

    fn create_valid_test_config() -> FlowControlConfig {
        FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![crate::config::types::Rule {
                id: "test_rule".to_string(),
                name: "Test Rule".to_string(),
                priority: 100,
                matchers: vec![crate::config::types::Matcher::User {
                    user_ids: vec!["*".to_string()],
                }],
                limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                    capacity: 100,
                    refill_rate: 10,
                }],
                action: crate::config::types::ActionConfig {
                    on_exceed: crate::config::types::Action::Reject,
                    ban: None,
                },
            }],
        }
    }

    // ============================================================================
    // Builder Feature-Gated Methods
    // ============================================================================

    #[cfg(feature = "circuit-breaker")]
    #[tokio::test]
    async fn test_builder_with_circuit_breaker() {
        use crate::circuit::{CircuitBreaker, CircuitBreakerConfig};

        let cb = Arc::new(CircuitBreaker::with_dependencies(
            CircuitBreakerConfig::default(),
        ));
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_circuit_breaker(cb)
            .build()
            .await
            .expect("build with circuit_breaker should succeed");

        assert!(governor.health_check().await.is_ok());
    }

    #[cfg(feature = "audit-log")]
    #[tokio::test]
    async fn test_builder_with_audit_logger() {
        use crate::logging::AuditLogger;

        let logger = Arc::new(AuditLogger::default().await);
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_audit_logger(logger)
            .build()
            .await
            .expect("build with audit_logger should succeed");

        assert!(governor.health_check().await.is_ok());
    }

    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_builder_with_metrics() {
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build with metrics should succeed");

        // check() 前计数器应为 0
        assert_eq!(metrics.requests_total.get(), 0.0);

        let mut ctx = RequestContext::default();
        ctx.user_id = Some("test_user".to_string());
        let _ = governor.check(&ctx).await;

        // check() 后 requests_total 应被记录（Rule 9：验证有意义的副作用）
        assert!(
            metrics.requests_total.get() >= 1.0,
            "metrics.requests_total should be recorded after check()"
        );
    }

    /// per-rule/per-limiter 维度指标接线验证：规则命中后
    /// flowguard_rule_checks_total（rule/outcome 标签）被记录，且
    /// flowguard_rule_limiter_rejections_total 随链内拒绝增量导出
    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_per_rule_metrics_wiring() {
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build should succeed");

        // 提取器/匹配器走 X-User-Id 头（ctx.user_id 字段不在默认提取链上）
        let ctx = RequestContext::new()
            .with_header("X-User-Id", "u_per_rule")
            .with_method("GET");
        // 配置规则 TokenBucket capacity=100：多次消费均放行（allowed 维度）
        for _ in 0..3 {
            let _ = governor.check(&ctx).await;
        }
        let output = metrics.gather();
        assert!(
            output.contains("flowguard_rule_checks_total"),
            "dimension metrics must be registered in the registry: {output}"
        );
        assert!(
            output.contains("rule=\"test_rule\""),
            "there should be a test_rule label series: {output}"
        );
        assert!(
            output.contains("outcome=\"allowed\""),
            "there should be an allowed outcome series: {output}"
        );
    }

    /// 负缓存命中路径的 per-rule 归因验证：缓存命中的拒绝按缓存内裁决
    /// 规则计入 flowguard_rule_checks_total（与链上拒绝同序列），可归因
    /// 命中不重复计入 negative_cache_hits_total
    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_negative_cache_hit_counts_per_rule_metrics() {
        use crate::telemetry::Metrics;

        // capacity=1、refill_rate=1/s：同秒连发首个放行其后拒绝并入负缓存
        let config = FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![crate::config::types::Rule {
                id: "test_rule".to_string(),
                name: "Test Rule".to_string(),
                priority: 100,
                matchers: vec![crate::config::types::Matcher::User {
                    user_ids: vec!["*".to_string()],
                }],
                limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                    capacity: 1,
                    refill_rate: 1,
                }],
                action: crate::config::types::ActionConfig {
                    on_exceed: crate::config::types::Action::Reject,
                    ban: None,
                },
            }],
        };

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(config)
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build should succeed");

        let ctx = RequestContext::new()
            .with_header("X-User-Id", "u_neg_cache")
            .with_method("GET");
        // 1 放行（链上）+ 2 拒绝：第 2 次链上拒绝并写缓存，第 3 次负缓存命中
        for _ in 0..3 {
            let _ = governor.check(&ctx).await;
        }

        let output = metrics.gather();
        assert!(
            output
                .contains("flowguard_rule_checks_total{outcome=\"rejected\",rule=\"test_rule\"} 2"),
            "rejections from negative-cache hits must be counted into the per-rule series per the decision rule (1 on-chain + 1 cache hit): {output}"
        );
        assert_eq!(
            metrics.negative_cache_hits_total.get(),
            0,
            "a rule-attributable hit must not be counted twice into negative_cache_hits_total"
        );
    }

    /// 维度指标开关关闭验证：with_per_rule_metrics(false) 后 check 不再
    /// 导出 rule 维度序列（全局计数仍工作）
    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_per_rule_metrics_disabled_opt_out() {
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .with_per_rule_metrics(false)
            .build()
            .await
            .expect("build should succeed");

        let ctx = RequestContext::new()
            .with_header("X-User-Id", "u_opt_out")
            .with_method("GET");
        let _ = governor.check(&ctx).await;

        let output = metrics.gather();
        assert!(
            !output.contains("flowguard_rule_checks_total{"),
            "there should be no rule dimension series after shutdown: {output}"
        );
        // 全局计数仍工作
        assert!(metrics.requests_total.get() >= 1.0);
    }

    /// 增量导出幂等性钉子：导出基线（per-node last/落账版本号）必须与
    /// 合并串行化。曾经锁外快照 + 锁内无条件合并的 check-then-act 交错
    /// 把基线拉回旧值，下轮误判 stale 对已导出区间静默双算——同版本
    /// 重复导出必须整体跳过，新增拒绝后只导出真实增量
    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_incremental_export_idempotent_and_exact() {
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(FlowControlConfig {
                version: "0.1.0".to_string(),
                global: crate::config::types::GlobalConfig::default(),
                rules: vec![crate::config::types::Rule {
                    id: "test_rule".to_string(),
                    name: "Test Rule".to_string(),
                    priority: 100,
                    matchers: vec![crate::config::types::Matcher::User {
                        user_ids: vec!["*".to_string()],
                    }],
                    limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                        capacity: 1,
                        refill_rate: 1,
                    }],
                    action: crate::config::types::ActionConfig {
                        on_exceed: crate::config::types::Action::Reject,
                        ban: None,
                    },
                }],
            })
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build should succeed");

        let chain = governor
            .rule_chains
            .read()
            .await
            .get("test_rule")
            .map(|entry| entry.value().clone())
            .expect("test_rule chain should exist");

        let exported_total = |m: &Metrics| -> u64 {
            use prometheus::core::Collector;
            m.rule_limiter_rejections_total
                .collect()
                .iter()
                .map(|family| {
                    family
                        .get_metric()
                        .iter()
                        .map(|metric| metric.get_counter().value() as u64)
                        .sum::<u64>()
                })
                .sum()
        };

        // 链上直接制造拒绝（绕过 Governor 负缓存）：首查耗尽令牌，其后拒绝
        let _ = chain.check().await.unwrap();
        assert_eq!(
            chain.stats_sync().rejected_count,
            0,
            "the first check must exhaust the tokens and allow"
        );
        chain.check().await.unwrap();
        chain.check().await.unwrap();
        let actual = chain.stats_sync().node_rejections[0].1;

        // 线程一视角：全量导出
        governor.export_chain_rejections(&chain, "test_rule", &metrics);
        assert_eq!(
            exported_total(&metrics),
            actual,
            "the first export must exactly equal the true reject count"
        );
        assert_eq!(
            chain.exported_rejections_version(),
            chain.node_rejections_version(),
            "the exported ledger version must catch up with the chain version"
        );

        // 线程二迟到闯入：持合并前的旧观测重复导出（并发导出方已落账同
        // 版本的交错形态），必须整体跳过
        governor.export_chain_rejections(&chain, "test_rule", &metrics);
        assert_eq!(
            exported_total(&metrics),
            actual,
            "repeated exports of the same version must not double count"
        );

        // 新增真实拒绝后仅导出新增 delta
        chain.check().await.unwrap();
        let actual = chain.stats_sync().node_rejections[0].1;
        governor.export_chain_rejections(&chain, "test_rule", &metrics);
        assert_eq!(
            exported_total(&metrics),
            actual,
            "the next export must exactly cover the new rejections"
        );
    }

    /// 并发导出压力钉子：多任务并发触发同一链的导出，累计导出量必须
    /// 恰等于链上真实拒绝数——不双算不漏算
    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_concurrent_exports_never_double_count() {
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        let governor = Arc::new(
            Governor::builder()
                .with_config(FlowControlConfig {
                    version: "0.1.0".to_string(),
                    global: crate::config::types::GlobalConfig::default(),
                    rules: vec![crate::config::types::Rule {
                        id: "test_rule".to_string(),
                        name: "Test Rule".to_string(),
                        priority: 100,
                        matchers: vec![crate::config::types::Matcher::User {
                            user_ids: vec!["*".to_string()],
                        }],
                        limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                            capacity: 1,
                            refill_rate: 1,
                        }],
                        action: crate::config::types::ActionConfig {
                            on_exceed: crate::config::types::Action::Reject,
                            ban: None,
                        },
                    }],
                })
                .with_storage(Arc::new(MemoryStorage::new()))
                .with_ban_storage(Arc::new(MemoryBanStorage::new()))
                .with_metrics(metrics.clone())
                .build()
                .await
                .expect("build should succeed"),
        );

        let chain = governor
            .rule_chains
            .read()
            .await
            .get("test_rule")
            .map(|entry| entry.value().clone())
            .expect("test_rule chain should exist");

        // 耗尽令牌，此后链上检查全部拒绝
        let _ = chain.check().await.unwrap();

        let tasks = (0..8)
            .map(|_| {
                let governor = governor.clone();
                let chain = chain.clone();
                let metrics = metrics.clone();
                tokio::spawn(async move {
                    for _ in 0..50 {
                        let _ = chain.check().await;
                        governor.export_chain_rejections(&chain, "test_rule", &metrics);
                    }
                })
            })
            .collect::<Vec<_>>();
        for task in tasks {
            task.await.unwrap();
        }

        let actual: u64 = chain
            .stats_sync()
            .node_rejections
            .iter()
            .map(|(_, count)| count)
            .sum();
        let exported: u64 = {
            use prometheus::core::Collector;
            metrics
                .rule_limiter_rejections_total
                .collect()
                .iter()
                .map(|family| {
                    family
                        .get_metric()
                        .iter()
                        .map(|metric| metric.get_counter().value() as u64)
                        .sum::<u64>()
                })
                .sum()
        };
        assert_eq!(
            exported, actual,
            "concurrent exports must not double count or miss"
        );
    }

    /// 负缓存无归因命中（缓存条目 rule_id=None，如旧版本写入的共享缓存
    /// 条目）的防御路径：span 不虚构 rule.id，但保留 rule.outcome 结果
    /// 信号，观测面不留全盲 span；无归因命中退独立计数
    #[cfg(all(test, feature = "prometheus", feature = "telemetry", feature = "otlp"))]
    #[tokio::test]
    async fn test_negative_cache_hit_without_rule_id_records_outcome() {
        use crate::l1_cache::CacheableDecision;
        use crate::matchers::Identifier;
        use crate::telemetry::Metrics;
        use crate::telemetry::otlp::{InMemoryTransport, OtlpSpanExporter};

        let transport = Arc::new(InMemoryTransport::new());
        let exporter = OtlpSpanExporter::new(
            "limiteron-test",
            transport.clone(),
            "http://mock-collector:4318/v1/traces",
        );
        let (sink, worker) = OtlpSpanExporter::spawn_worker(exporter, 64);
        tokio::spawn(worker.run());

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(FlowControlConfig {
                version: "0.1.0".to_string(),
                global: crate::config::types::GlobalConfig::default(),
                rules: vec![crate::config::types::Rule {
                    id: "test_rule".to_string(),
                    name: "Test Rule".to_string(),
                    priority: 100,
                    matchers: vec![crate::config::types::Matcher::User {
                        user_ids: vec!["*".to_string()],
                    }],
                    limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                        capacity: 100,
                        refill_rate: 10,
                    }],
                    action: crate::config::types::ActionConfig {
                        on_exceed: crate::config::types::Action::Reject,
                        ban: None,
                    },
                }],
            })
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .with_tracer(Arc::new(Tracer::with_otlp_sink(true, Some(Arc::new(sink)))))
            .build()
            .await
            .expect("build should succeed");

        // 预置无归因（rule_id=None）的拒绝缓存条目，命中后走防御路径；
        // 缓存键按 check 同源路径经 matcher 计算
        let ctx = RequestContext::new()
            .with_header("X-User-Id", "u_noattr")
            .with_method("GET");
        let matched_rules: Vec<_> = governor
            .rule_matcher
            .read()
            .await
            .match_all(&ctx)
            .into_iter()
            .cloned()
            .collect();
        assert!(!matched_rules.is_empty(), "the rule should match");
        let identifier = Identifier::UserId("u_noattr".to_string());
        let cache_key = governor.build_cache_key_multi(&identifier, &matched_rules);
        governor
            .l1_cache
            .set(cache_key, CacheableDecision::rejected("cached deny"))
            .await
            .expect("seed cache should succeed");

        let decision = governor.check(&ctx).await.expect("check should succeed");
        assert!(
            matches!(decision, Decision::Rejected(_)),
            "should hit the cached rejection: {decision:?}"
        );
        assert_eq!(
            metrics.negative_cache_hits_total.get(),
            1,
            "an unattributable hit should fall back to an independent count"
        );

        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        let span = loop {
            let requests = transport.requests();
            if let Some(request) = requests.first() {
                break request.1["resourceSpans"][0]["scopeSpans"][0]["spans"][0].clone();
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "no OTLP export received within 5s"
            );
            tokio::time::sleep(Duration::from_millis(10)).await;
        };

        let attribute = |key: &str| -> Option<String> {
            span["attributes"]
                .as_array()?
                .iter()
                .find(|a| a["key"] == key)
                .map(|a| {
                    a["value"]["stringValue"]
                        .as_str()
                        .unwrap_or_default()
                        .to_string()
                })
        };
        assert_eq!(
            attribute("rule.id"),
            None,
            "an unattributable hit must not fabricate rule.id"
        );
        assert_eq!(
            attribute("rule.outcome").as_deref(),
            Some("rejected"),
            "an unattributable hit must keep the result signal"
        );
    }

    /// 降级计数接线验证：配置 FallbackManager 后主检查失败走降级闭包，
    /// 但配置类错误（无标识符）不计 degraded——可被未认证请求无门槛刷出，
    /// 非存储健康度信号；未配置 fallback 或主检查成功时不计降级
    #[cfg(all(feature = "prometheus", feature = "fallback"))]
    #[tokio::test]
    async fn test_degraded_metrics_wiring() {
        use crate::fallback::FallbackManager;
        use crate::telemetry::Metrics;
        use oxcache::Cache;

        // 未配置 fallback_manager：主路径执行，degraded 不增
        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build should succeed");
        let ctx = RequestContext::new()
            .with_header("X-User-Id", "u_degraded")
            .with_method("GET");
        let _ = governor.check(&ctx).await;
        assert_eq!(
            metrics.degraded_checks_total.get(),
            0,
            "success on the main path must not count as degraded"
        );

        // 配置 fallback_manager + 无标识符请求：降级闭包执行但错误为配置类，
        // 不计入 degraded（避免未认证洪水刷高存储健康度告警）
        let cache: Cache<String, String> = Cache::builder()
            .capacity(10000)
            .ttl(Duration::from_secs(60))
            .build()
            .await
            .unwrap();
        let metrics_fb = Arc::new(Metrics::new());
        let governor_fb = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics_fb.clone())
            .with_fallback_manager(Arc::new(FallbackManager::new(Arc::new(cache))))
            .build()
            .await
            .expect("build with fallback should succeed");
        let ctx_no_id = RequestContext::new().with_method("GET");
        let _ = governor_fb.check(&ctx_no_id).await;
        assert_eq!(
            metrics_fb.degraded_checks_total.get(),
            0,
            "a config-class error triggering degradation must not count into degraded_checks_total"
        );
    }

    /// 降级分类判定验证：存储类错误计降级，配置类错误与成功结果不计
    /// （分类器直接驱动——链内限流器为本地实现，端到端注入存储错误
    /// 需分布式后端，不可在单测稳定复现）
    #[cfg(all(feature = "prometheus", feature = "fallback"))]
    #[tokio::test]
    async fn test_degraded_error_classification() {
        use crate::error::StorageError;
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build should succeed");

        governor.note_degraded_check(&Err(LimiteronError::StorageError(
            StorageError::ConnectionError("redis down".to_string()),
        )));
        assert_eq!(
            metrics.degraded_checks_total.get(),
            1,
            "a storage-class error must count as degraded"
        );

        governor.note_degraded_check(&Err(LimiteronError::ConfigError(
            "no identifier".to_string(),
        )));
        governor.note_degraded_check(&Ok(Decision::allowed_default()));
        assert_eq!(
            metrics.degraded_checks_total.get(),
            1,
            "config-class errors and successful results must not count as degraded"
        );
    }

    /// 热更新后增量导出对齐验证：apply_config 重建决策链后链内拒绝计数
    /// 从零重计，导出快照同步清零，新链拒绝立即进入指标；若快照未清零，
    /// 新链累计在追平旧快照前会被增量语义吞掉（漏报）
    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_per_rule_limiter_rejections_reset_on_hot_reload() {
        use crate::telemetry::Metrics;

        let make_config = || FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![crate::config::types::Rule {
                id: "test_rule".to_string(),
                name: "Test Rule".to_string(),
                priority: 100,
                matchers: vec![crate::config::types::Matcher::User {
                    user_ids: vec!["*".to_string()],
                }],
                // capacity=1、refill_rate=1/s：同毫秒连发请求令牌不恢复，
                // 首个放行其后拒绝
                limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                    capacity: 1,
                    refill_rate: 1,
                }],
                action: crate::config::types::ActionConfig {
                    on_exceed: crate::config::types::Action::Reject,
                    ban: None,
                },
            }],
        };

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(make_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build should succeed");

        let ctx = RequestContext::new()
            .with_header("X-User-Id", "u_hot_reload")
            .with_method("GET");
        // 旧链：1 放行 + 1 拒绝（第 3 次命中拒绝负缓存短路返回，不再
        // 经过链，因而不计入链检查维度指标）→ 导出 1
        for _ in 0..3 {
            let _ = governor.check(&ctx).await;
        }
        let counter = metrics
            .rule_limiter_rejections_total
            .with_label_values(&["test_rule", "test_rule_limiter_0"]);
        assert_eq!(
            counter.get(),
            1,
            "1 on-chain rejection from the old chain must export as 1"
        );

        // 热更新换新链（同规则 ID）：链内计数归零，快照同步清零
        governor
            .apply_config(make_config())
            .await
            .expect("hot reload should succeed");

        // 新链：1 放行 + 1 拒绝 → 导出应增长到 2（快照未清零则停在 1）
        for _ in 0..2 {
            let _ = governor.check(&ctx).await;
        }
        assert_eq!(
            counter.get(),
            2,
            "after a hot swap, the new chain's rejections must export immediately (the snapshot is cleared with the swap)"
        );
    }

    #[cfg(feature = "telemetry")]
    #[tokio::test]
    async fn test_builder_with_tracer() {
        use crate::telemetry::Tracer;

        let tracer = Arc::new(Tracer::new(true));
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_tracer(tracer.clone())
            .build()
            .await
            .expect("build with tracer should succeed");

        // 验证 tracer 被存储且启用
        assert!(tracer.is_enabled());

        let mut ctx = RequestContext::default();
        ctx.user_id = Some("test_user".to_string());
        // check() 应正常完成（tracer span start/finish 不 panic）
        let _ = governor.check(&ctx).await;
    }

    /// 验证 with_storage() 转发 metrics 参数到 Governor（之前被 #[allow(unused_variables)] 静默丢弃）
    #[cfg(feature = "prometheus")]
    #[tokio::test]
    async fn test_with_storage_forwards_metrics() {
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        #[cfg(feature = "telemetry")]
        let tracer: Option<Arc<crate::telemetry::Tracer>> = None;

        let governor = Governor::with_storage(
            create_valid_test_config(),
            Arc::new(MemoryStorage::new()),
            Arc::new(MemoryBanStorage::new()),
            Some(metrics.clone()),
            #[cfg(feature = "telemetry")]
            tracer,
        )
        .await
        .expect("with_storage should succeed");

        let mut ctx = RequestContext::default();
        ctx.user_id = Some("test_user".to_string());
        let _ = governor.check(&ctx).await;

        assert!(
            metrics.requests_total.get() >= 1.0,
            "with_storage should forward metrics to Governor (was silently dropped before fix)"
        );
    }

    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_builder_with_fallback_manager() {
        use crate::fallback::FallbackManager;
        use oxcache::Cache;

        let cache: Cache<String, String> = Cache::builder()
            .capacity(10000)
            .ttl(Duration::from_secs(60))
            .build()
            .await
            .unwrap();
        let fm = Arc::new(FallbackManager::new(Arc::new(cache)));
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_fallback_manager(fm)
            .build()
            .await
            .expect("build with fallback_manager should succeed");

        let stats = governor.stats().await;
        assert_eq!(stats.total_requests, 0);
    }

    // ============================================================================
    // check() with Fallback Manager
    // Covers: check() dispatch to check_with_fallback, check_with_fallback body
    // ============================================================================

    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_with_fallback_manager() {
        use crate::fallback::FallbackManager;
        use oxcache::Cache;

        let cache: Cache<String, String> = Cache::builder()
            .capacity(10000)
            .ttl(Duration::from_secs(60))
            .build()
            .await
            .unwrap();
        let fm = Arc::new(FallbackManager::new(Arc::new(cache)));
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_fallback_manager(fm)
            .build()
            .await
            .expect("build should succeed");

        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.100".to_string());
        let result = governor.check(&ctx).await;
        assert!(
            result.is_ok(),
            "check with fallback should succeed: {:?}",
            result
        );
    }

    /// 当 check_internal 因标识符提取失败而返回 Err 时，降级逻辑调用
    /// check_l1_cache_only，后者也因标识符提取失败而返回 Err。
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_with_fallback_no_identifier() {
        use crate::fallback::FallbackManager;
        use oxcache::Cache;

        let cache: Cache<String, String> = Cache::builder()
            .capacity(10000)
            .ttl(Duration::from_secs(60))
            .build()
            .await
            .unwrap();
        let fm = Arc::new(FallbackManager::new(Arc::new(cache)));
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_fallback_manager(fm)
            .build()
            .await
            .expect("build should succeed");

        // 无标识符请求：check_internal 失败 → 降级到 check_l1_cache_only
        let ctx = RequestContext::default();
        let result = governor.check(&ctx).await;
        assert!(result.is_err());
    }

    /// 缓存禁用时，check_l1_cache_only 返回 Err。
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_with_fallback_l1_cache_disabled() {
        use crate::fallback::FallbackManager;
        use oxcache::Cache;

        let cache: Cache<String, String> = Cache::builder()
            .capacity(10000)
            .ttl(Duration::from_secs(60))
            .build()
            .await
            .unwrap();
        let fm = Arc::new(FallbackManager::new(Arc::new(cache)));
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_l1_cache_enabled(false)
            .with_fallback_manager(fm)
            .build()
            .await
            .expect("build should succeed");

        // 无标识符请求：check_internal 失败 → 降级到 check_l1_cache_only
        // → 缓存禁用，返回 Err
        let ctx = RequestContext::default();
        let result = governor.check(&ctx).await;
        assert!(result.is_err());
    }

    // ============================================================================
    // check() - Parallel ban check integration in check_internal
    // With parallel-checker + ban-manager, every check() with a valid identifier
    // goes through the parallel ban checker code path.
    // ============================================================================

    #[cfg(feature = "parallel-checker")]
    #[tokio::test]
    async fn test_check_parallel_ban_checker_path() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // IP-based request triggers to_ban_target -> Some -> parallel check
        let mut ctx = RequestContext::default();
        ctx.client_ip = Some("10.0.0.200".to_string());
        let result = governor.check(&ctx).await;
        assert!(
            result.is_ok(),
            "check with parallel ban checker should succeed: {:?}",
            result
        );
    }

    // ============================================================================
    // check_resource_parallel (parallel-checker feature gate)
    // ============================================================================

    #[cfg(feature = "parallel-checker")]
    #[tokio::test]
    async fn test_check_resource_parallel_unbanned() {
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        let result = governor.check_resource_parallel("test_user_123").await;
        match result {
            Ok(Decision::Allowed(_)) => {}
            other => panic!("Expected Allowed for unbanned user, got: {:?}", other),
        }
    }

    // ============================================================================
    // Audit Logger Set/Get (audit-log feature gate)
    // ============================================================================

    #[cfg(feature = "audit-log")]
    #[tokio::test]
    async fn test_audit_logger_set_and_get() {
        use crate::logging::AuditLogger;

        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // Initially no logger is set
        let initial = governor.audit_logger().await;
        assert!(initial.is_none());

        // Set a logger and verify it's returned
        let logger = Arc::new(AuditLogger::default().await);
        governor.set_audit_logger(logger.clone()).await;
        let retrieved = governor.audit_logger().await;
        assert!(retrieved.is_some(), "audit_logger should be Some after set");
    }

    // ============================================================================
    // check_l1_cache_only - 直接测试各分支（fallback feature gate）
    // 覆盖 lines 943-1023 的多个分支
    // ============================================================================

    /// 创建使用 IP matcher 的测试配置（匹配所有 IP）
    fn create_ip_test_config() -> FlowControlConfig {
        FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![crate::config::types::Rule {
                id: "ip_rule".to_string(),
                name: "IP Rule".to_string(),
                priority: 100,
                matchers: vec![crate::config::types::Matcher::Ip {
                    ip_ranges: vec!["0.0.0.0/0".to_string()],
                }],
                limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                    capacity: 100,
                    refill_rate: 10,
                }],
                action: crate::config::types::ActionConfig {
                    on_exceed: crate::config::types::Action::Reject,
                    ban: None,
                },
            }],
        }
    }

    /// 创建带指定 user_ids 的 User matcher 配置（用于"无匹配规则"测试）
    fn create_user_test_config(user_ids: Vec<String>) -> FlowControlConfig {
        FlowControlConfig {
            version: "0.1.0".to_string(),
            global: crate::config::types::GlobalConfig::default(),
            rules: vec![crate::config::types::Rule {
                id: "user_rule".to_string(),
                name: "User Rule".to_string(),
                priority: 100,
                matchers: vec![crate::config::types::Matcher::User { user_ids }],
                limiters: vec![crate::config::types::LimiterConfig::TokenBucket {
                    capacity: 100,
                    refill_rate: 10,
                }],
                action: crate::config::types::ActionConfig {
                    on_exceed: crate::config::types::Action::Reject,
                    ban: None,
                },
            }],
        }
    }

    /// 构造带 IP 请求的 RequestContext
    fn create_ip_request_context(ip: &str) -> RequestContext {
        let mut ctx = RequestContext::default();
        ctx.client_ip = Some(ip.to_string());
        ctx
    }

    /// check_l1_cache_only: 缓存禁用时返回 Err（覆盖 lines 943-946）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_disabled_returns_err() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_l1_cache_enabled(false)
            .build()
            .await
            .expect("build should succeed");

        let ctx = create_ip_request_context("10.0.0.1");
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            LimiteronError::LimitError(msg) => {
                assert!(
                    msg.contains("L1 缓存未启用") || msg.contains("L1 cache not enabled"),
                    "unexpected: {msg}"
                );
            }
            other => panic!("expected LimitError, got: {:?}", other),
        }
    }

    /// check_l1_cache_only: identifier 提取失败时返回 Err（覆盖 lines 950-952）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_identifier_extraction_failure() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // 默认 RequestContext 无 client_ip/user_id/api_key，identifier 提取失败
        let ctx = RequestContext::default();
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            LimiteronError::ConfigError(msg) => {
                assert!(
                    msg.contains("Failed to extract identifier"),
                    "unexpected: {}",
                    msg
                );
            }
            other => panic!("expected ConfigError, got: {:?}", other),
        }
    }

    /// check_l1_cache_only: 无匹配规则时返回 Ok(Allowed)（覆盖 lines 964-966）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_no_matched_rules() {
        // User matcher 只匹配特定 user_id，但请求是 IP（无 user_id header）
        let governor = Governor::builder()
            .with_config(create_user_test_config(vec!["specific_user".to_string()]))
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        let ctx = create_ip_request_context("10.0.0.1");
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
        match result.unwrap() {
            Decision::Allowed(_) => {}
            other => panic!("expected Allowed, got: {:?}", other),
        }
    }

    /// check_l1_cache_only: cache miss + 非 island mode → Err（覆盖 lines 970, 973, 980, 1015-1018）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_cache_miss_no_island() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // 确保不在 island mode
        assert!(!governor.l1_cache.is_island_mode());

        let ctx = create_ip_request_context("10.0.0.1");
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            LimiteronError::LimitError(msg) => {
                assert!(
                    msg.contains("降级缓存未命中") || msg.contains("degraded cache miss"),
                    "unexpected: {msg}"
                );
            }
            other => panic!("expected LimitError, got: {:?}", other),
        }
    }

    /// check_l1_cache_only: cache miss + island AllowAll → Ok(Allowed)（覆盖 lines 982-987）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_island_allow_all() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        let island_config = IslandModeConfig::new(IslandFallbackStrategy::AllowAll);
        governor.l1_cache.enable_island_mode(island_config);
        assert!(governor.l1_cache.is_island_mode());

        let ctx = create_ip_request_context("10.0.0.1");
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
        match result.unwrap() {
            Decision::Allowed(_) => {}
            other => panic!("expected Allowed, got: {:?}", other),
        }
    }

    /// check_l1_cache_only: cache miss + island RejectAll → Err（覆盖 lines 990-992）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_island_reject_all() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        let island_config = IslandModeConfig::new(IslandFallbackStrategy::RejectAll);
        governor.l1_cache.enable_island_mode(island_config);

        let ctx = create_ip_request_context("10.0.0.1");
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            LimiteronError::LimitError(msg) => {
                assert!(
                    msg.contains("拒绝请求") || msg.contains("reject"),
                    "unexpected: {msg}"
                );
            }
            other => panic!("expected LimitError, got: {:?}", other),
        }
    }

    /// check_l1_cache_only: cache miss + island LocalDecision → Ok(Allowed)（覆盖 lines 997-998）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_island_local_decision() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        let island_config = IslandModeConfig::new(IslandFallbackStrategy::LocalDecision);
        governor.l1_cache.enable_island_mode(island_config);

        let ctx = create_ip_request_context("10.0.0.1");
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
    }

    /// check_l1_cache_only: cache miss + island ConservativeQuota → Ok(Allowed)（覆盖 lines 1005, 1008）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_island_conservative_quota() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        let island_config = IslandModeConfig::new(IslandFallbackStrategy::ConservativeQuota {
            max_requests: 10,
            window_secs: 60,
        });
        governor.l1_cache.enable_island_mode(island_config);

        let ctx = create_ip_request_context("10.0.0.1");
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
    }

    /// check_l1_cache_only: cache hit → 返回 cached decision（覆盖 lines 973-978）
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_cache_hit() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // 负缓存语义：check() 的"允许"结果永不写入 缓存（只缓存拒绝/封禁决策）。
        // 先验证允许请求不会填充缓存。
        let ctx = create_ip_request_context("10.0.0.1");
        let _ = governor.check(&ctx).await;

        // 允许决策未入缓存 → check_l1_cache_only（非 island mode）应未命中并返回 Err
        let miss = governor.check_l1_cache_only(&ctx).await;
        assert!(miss.is_err(), "expected miss, got: {:?}", miss);

        // 再验证负缓存命中：手动放入"拒绝"决策（键格式与 check_l1_cache_only 一致），
        // check_l1_cache_only 应命中并返回 Rejected。
        use crate::l1_cache::CacheableDecision;
        let cache_key = "rl:ip:10.0.0.1:ip_rule".to_string();
        let rejected = CacheableDecision::rejected("rate limit exceeded");
        governor
            .l1_cache
            .set(cache_key, rejected)
            .await
            .expect("cache set should succeed");

        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
        match result.unwrap() {
            Decision::Rejected(_) => {}
            other => panic!("expected Rejected, got: {:?}", other),
        }
    }

    /// 覆盖 update_stats_for_decision 中 Banned 分支 (line 1050)
    /// 通过在 缓存中手动放入 Banned 决策，然后调用 check_l1_cache_only 触发
    #[cfg(feature = "fallback")]
    #[tokio::test]
    async fn test_check_l1_cache_only_cached_banned_updates_stats() {
        use crate::error::BanInfo;
        use crate::l1_cache::CacheableDecision;

        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // 手动构建 cache_key（与 check_l1_cache_only 中 build_cache_key 一致）
        let ctx = create_ip_request_context("10.0.0.50");
        let cache_key = "rl:ip:10.0.0.50:ip_rule".to_string();

        // 在 缓存中放入 Banned 决策
        let ban_info = BanInfo::new("test ban".to_string(), chrono::Utc::now(), 1);
        let banned_decision = CacheableDecision::banned(&ban_info);
        governor
            .l1_cache
            .set(cache_key, banned_decision)
            .await
            .expect("cache set should succeed");

        // 调用 check_l1_cache_only，应命中缓存并返回 Banned
        // 这会触发 update_stats_for_decision 的 Banned 分支 (line 1050)
        let result = governor.check_l1_cache_only(&ctx).await;
        assert!(result.is_ok(), "expected Ok, got: {:?}", result.err());
        match result.unwrap() {
            Decision::Banned(_) => {}
            other => panic!("expected Banned, got: {:?}", other),
        }
    }

    /// 覆盖 update_stats_for_decision 中 Err 分支 (line 1056)
    /// 直接调用私有方法传入 Err 结果
    #[tokio::test]
    async fn test_update_stats_for_decision_err_branch() {
        let governor = Governor::builder()
            .with_config(create_ip_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .build()
            .await
            .expect("build should succeed");

        // 直接调用 update_stats_for_decision 传入 Err
        let err_result: Result<Decision, LimiteronError> =
            Err(LimiteronError::LimitError("test error".to_string()));
        governor.update_stats_for_decision(&err_result);
        // 验证不会 panic 即可（stats.increment_error 内部是原子操作）
    }

    // ============================================================================
    // metrics feature 门控计数测试
    // metrics 隐含 prometheus，启用后 governor check() 中 allow/reject/ban
    // 三点指标记录自动激活
    // ============================================================================

    #[cfg(feature = "metrics")]
    #[tokio::test]
    async fn test_metrics_feature_records_check_metrics() {
        use crate::telemetry::Metrics;

        let metrics = Arc::new(Metrics::new());
        let governor = Governor::builder()
            .with_config(create_valid_test_config())
            .with_storage(Arc::new(MemoryStorage::new()))
            .with_ban_storage(Arc::new(MemoryBanStorage::new()))
            .with_metrics(metrics.clone())
            .build()
            .await
            .expect("build with metrics should succeed");

        // 执行一次 check — 触发 allow 路径
        let ctx = RequestContext {
            user_id: Some("metrics_test_user".to_string()),
            ip: Some("127.0.0.1".to_string()),
            path: "/test".to_string(),
            method: "GET".to_string(),
            ..Default::default()
        };
        let _ = governor.check(&ctx).await;

        // 验证 metrics 已记录（requests_total > 0 表示 record_check 被调用）
        assert!(
            metrics.requests_total.get() > 0.0,
            "metrics should record at least one check via requests_total counter"
        );
    }

    // 与被测类型同门控：ConservativeQuotaCounter 仅在 fallback 下编译，
    // 测试模块缺该门控会使默认 feature 的测试二进制编译失败
    #[cfg(all(test, feature = "fallback"))]
    mod conservative_quota_counter_tests {
        use super::*;

        #[test]
        fn test_conservative_quota_counter_bounds_window() {
            // 保守配额计数器：窗口内有界放行，翻转重置
            let mut c = ConservativeQuotaCounter {
                window_start_secs: 100,
                count: 0,
            };
            for i in 0..5u64 {
                assert!(
                    c.try_acquire(5, 60, 100 + i),
                    "request {} should be allowed",
                    i + 1
                );
            }
            assert!(
                !c.try_acquire(5, 60, 105),
                "over-limit within the window must be rejected"
            );
            // 窗口翻转：重置后恢复放行
            assert!(
                c.try_acquire(5, 60, 160),
                "the window flip should reset and allow"
            );
            // 窗口回拨（时钟异常）：不重置窗口、计数继续有界累计
            //（window_start 只前推，回拨不产生新窗口容量）
            assert!(
                c.try_acquire(5, 60, 120),
                "a clock turn-back does not reset the window, capacity stays bounded (2/5)"
            );
        }
    }
}

#[cfg(all(test, feature = "multi-tenant"))]
/// Governor 的租户限定封禁与限流器标签契约测试（自带 import，独立于其它测试模块）。
mod governor_tenant_ban_tests {
    use super::{Governor, RequestContext};
    use super::{create_valid_test_config_for_tenant, limiter_kind};
    use crate::config::{LimiterConfig, QuotaType};
    use crate::storage::{MemoryBanStorage, MemoryStorage};
    use std::sync::Arc;

    async fn make_governor() -> Governor {
        let storage: Arc<dyn crate::storage::Storage> = Arc::new(MemoryStorage::new());
        let ban_storage: Arc<dyn crate::storage::BanStorage> = Arc::new(MemoryBanStorage::new());
        Governor::builder()
            .with_config(create_valid_test_config_for_tenant())
            .with_storage(storage)
            .with_ban_storage(ban_storage)
            .build()
            .await
            .expect("Governor build should succeed")
    }

    /// limiter_kind 覆盖全部 LimiterConfig 变体且标签互不相同；新增变体漏分支即失败。
    #[test]
    fn test_limiter_kind_covers_every_variant() {
        let cases: Vec<(LimiterConfig, &str)> = vec![
            (
                LimiterConfig::TokenBucket {
                    capacity: 10,
                    refill_rate: 1,
                },
                "token_bucket",
            ),
            (
                LimiterConfig::SlidingWindow {
                    window_size: "1m".into(),
                    max_requests: 5,
                },
                "sliding_window",
            ),
            (
                LimiterConfig::FixedWindow {
                    window_size: "1m".into(),
                    max_requests: 5,
                },
                "fixed_window",
            ),
            (
                LimiterConfig::Quota {
                    quota_type: QuotaType::Count,
                    limit: 5,
                    window: "1d".into(),
                    alert_threshold: None,
                    overdraft: None,
                },
                "quota",
            ),
            (
                LimiterConfig::Concurrency { max_concurrent: 2 },
                "concurrency",
            ),
            (
                LimiterConfig::LeakyBucket {
                    capacity: 4,
                    leak_rate: 2,
                },
                "leaky_bucket",
            ),
            (
                LimiterConfig::SlidingWindowLog {
                    window_size: "1m".into(),
                    max_requests: 3,
                },
                "sliding_window_log",
            ),
            (
                LimiterConfig::PriorityQueue {
                    window_size: "1m".into(),
                    total_per_window: 10,
                    level_weights: vec![1],
                    default_priority: None,
                },
                "priority_queue",
            ),
            (
                LimiterConfig::AdmissionControl {
                    max_concurrent: 2,
                    max_per_second: 5,
                },
                "admission_control",
            ),
            (
                LimiterConfig::Custom {
                    name: "mylim".into(),
                    config: serde_json::Value::Null,
                },
                "custom:mylim",
            ),
        ];
        let mut kinds: Vec<String> = Vec::new();
        for (cfg, expected) in &cases {
            let got = limiter_kind(cfg);
            assert_eq!(got, *expected, "limiter_kind mapping drift");
            kinds.push(got);
        }
        let mut uniq = kinds.clone();
        uniq.sort();
        uniq.dedup();
        assert_eq!(
            uniq.len(),
            cases.len(),
            "limiter type labels must be mutually distinct"
        );
    }

    /// 租户限定封禁键契约：非默认命名空间加前缀落库、默认命名空间保持全局键、
    /// 不支持封禁的标识符显式报错（不得静默跳过）。
    ///
    /// 被测方法经 `all(multi-tenant, ban-manager)` 门控，测试须同门控。
    #[cfg(feature = "ban-manager")]
    #[tokio::test]
    async fn test_ban_identifier_namespace_scoping() {
        use super::LimiteronError;
        use crate::matchers::Identifier;
        use crate::tenant::Namespace;
        use std::time::Duration;

        let governor = make_governor().await;
        let ns = Namespace::new("t1", "prod");
        let user = Identifier::UserId("u1".to_string());
        let ctx = RequestContext::default();

        // 无 resolver → 标识符不被限定
        assert_eq!(
            governor.tenant_scoped_identifier(&ctx, user.clone()).key(),
            user.key()
        );
        // 非默认命名空间的限定键含租户标识且不同于原键
        let qualified = ns.qualify_key("u1");
        assert_ne!(qualified, "u1");
        assert!(
            qualified.contains("t1"),
            "the qualified key must carry the tenant id: {qualified}"
        );

        // 按命名空间封禁 → 经 is_identifier_banned 可见（全局键路径）
        governor
            .ban_identifier_for_namespace(&ns, &user, "abuse", None)
            .await
            .expect("UserId is bannable");

        // 默认命名空间 → 全局键（无前缀），显式 duration 路径
        governor
            .ban_identifier_for_namespace(
                &Namespace::default(),
                &user,
                "global",
                Some(Duration::from_secs(60)),
            )
            .await
            .expect("default namespace path");
        // 已知不一致（早于 rc.5，提交 1cadcff 引入，本测试发现）：
        // tenant_scoped_identifier/is_identifier_banned 视默认命名空间为「不加前缀」，
        // 而 qualify_ban_target 无条件加前缀，故此处写入的键查不到。修复需评估既有
        // 存储键迁移影响，登记为独立工单，不在本发布轮次内改语义。

        // 不支持封禁的标识符必须硬报错
        for id in [
            Identifier::ApiKey("k".to_string()),
            Identifier::DeviceId("d".to_string()),
        ] {
            let err = governor
                .ban_identifier_for_namespace(&ns, &id, "x", None)
                .await
                .expect_err("unsupported identifier must fail loudly");
            assert!(
                matches!(err, LimiteronError::ValidationError(_)),
                "should be ValidationError: {err:?}"
            );
        }

        // 未封禁主体返回 None（不得误判）
        assert!(
            governor
                .is_identifier_banned(&ctx, &Identifier::UserId("nobody".to_string()))
                .await
                .expect("ban lookup")
                .is_none()
        );
    }

    /// 未配置 resolver 时 resolve_tenant 必须返回 None（不得凭空产生租户归属）。
    #[tokio::test]
    async fn test_resolve_tenant_without_resolver_is_none() {
        let governor = make_governor().await;
        assert!(
            governor
                .resolve_tenant(&RequestContext::default())
                .is_none()
        );
    }
}

#[cfg(all(test, feature = "multi-tenant"))]
fn create_valid_test_config_for_tenant() -> crate::config::FlowControlConfig {
    use crate::config::types::{Action, ActionConfig, GlobalConfig, Matcher, Rule};
    crate::config::FlowControlConfig {
        version: "0.1.0".to_string(),
        global: GlobalConfig::default(),
        rules: vec![Rule {
            id: "tenant_ban_rule".to_string(),
            name: "Tenant Ban Rule".to_string(),
            priority: 100,
            matchers: vec![Matcher::User {
                user_ids: vec!["*".to_string()],
            }],
            limiters: vec![limiter_cfg_token_bucket()],
            action: ActionConfig {
                on_exceed: Action::Reject,
                ban: None,
            },
        }],
    }
}

#[cfg(all(test, feature = "multi-tenant"))]
fn limiter_cfg_token_bucket() -> crate::config::LimiterConfig {
    crate::config::LimiterConfig::TokenBucket {
        capacity: 100,
        refill_rate: 10,
    }
}